create or replace task SUNO_PROD.PROD.FACT_CLIP_STATUS_HOURLY_UPSERT
	warehouse=SUNO_PROD_X_SMAL
	schedule='USING CRON 30 * * * * UTC'
	as declare
        start_time TIMESTAMP;
        p_date date;
        p_hour string;
        row_count number;
        updated_clip_stage_path string;
        new_clip_stage_path string;
        clip_extra_stage_path string;
        clip_sql string;
        clip_extra_sql string;
        begin
            start_time := CURRENT_TIMESTAMP();
            p_date := DATE(SYSDATE() - INTERVAL '1 HOUR');
            p_hour := LPAD(HOUR(SYSDATE() - INTERVAL '1 HOUR'), 2, '0');
            updated_clip_stage_path := CONCAT('@SUNO_DATABASE_EVENTS/bots_generatedclip_update/pdate=', p_date, '/phour=', p_hour);
            new_clip_stage_path := CONCAT('@SUNO_DATABASE_EVENTS/bots_generatedclip_insert/pdate=', p_date, '/phour=', p_hour);
            clip_extra_stage_path := CONCAT('@SUNO_DATABASE_EVENTS/bots_generatedclipextra/pdate=', p_date, '/phour=', p_hour);
            clip_sql := '
            merge into fact_clip_status as target
            using
            (
                select
                    $1:id as clip_id,
                    $1:user_id as user_id,
                    $1:is_public as is_public,
                    $1:is_deleted as is_deleted,
                    $1:upvote_count as like_count,
                    $1:dislike_count as dislike_count,
                    $1:flag_count as flag_count,
                    $1:play_count as play_count_displayed,
                    date(to_timestamp($1:updated_at)) as p_date,
                    hour(to_timestamp($1:updated_at)) as p_hour
                from ' || :updated_clip_stage_path || ' (file_format => ''BASIC_PARQUET_LOADER'', pattern => (''.*''))
                union all
                select
                    $1:id as clip_id,
                    $1:user_id as user_id,
                    $1:is_public as is_public,
                    $1:is_deleted as is_deleted,
                    $1:upvote_cnt as like_count,
                    $1:dislike_count as dislike_count,
                    $1:flag_count as flag_count,
                    $1:play_count as play_count_displayed,
                    date(to_timestamp($1:updated_at)) as p_date,
                    hour(to_timestamp($1:updated_at)) as p_hour
                from ' || :new_clip_stage_path || ' (file_format => ''BASIC_PARQUET_LOADER'', pattern => (''.*''))
            ) as source
            on (target.clip_id = source.clip_id)
            when matched then
                update set
                    target.is_public = source.is_public,
                    target.is_deleted = source.is_deleted,
                    target.like_count = source.like_count,
                    target.dislike_count = source.dislike_count,
                    target.flag_count = source.flag_count,
                    target.play_count_displayed = source.play_count_displayed
            when not matched then
                insert (clip_id, user_id, is_public, is_deleted, like_count, dislike_count, flag_count, play_count_displayed, p_date, p_hour)
                values (source.clip_id, source.user_id, source.is_public, source.is_deleted, source.like_count, source.dislike_count, source.flag_count, source.play_count_displayed, source.p_date, source.p_hour);
        ';
        execute immediate :clip_sql;
        clip_extra_sql := 'merge into fact_clip_status as target
            using
            (
                select
                    $1:clip_id as clip_id,
                    $1:download_audio_count as download_audio_mp3_count,
                    $1:download_audio_wav_count as download_audio_wav_count,
                    $1:download_video_count as download_video_count,
                    $1:share_count as share_count,
                    $1:p_date as p_date,
                    $1:p_hour as p_hour
                from ' || :clip_extra_stage_path || ' (file_format => ''BASIC_PARQUET_LOADER'', pattern => (''.*''))
            ) as source
            on (target.clip_id = source.clip_id)
            when matched then
                update set
                    target.download_audio_mp3_count = source.download_audio_mp3_count,
                    target.download_audio_wav_count = source.download_audio_wav_count,
                    target.download_video_count = source.download_video_count,
                    target.share_count = source.share_count
        ';
        execute immediate :clip_sql;
        SELECT COUNT(*) 
            INTO :row_count 
        FROM PROD.fact_clip_status
        WHERE p_date = :p_date and p_hour = :p_hour;
        call TASK_MONITOR_INSERT_PROC('FACT_CLIP_STATUS_HOURLY_UPSERT', :start_time, :row_count, 'SUCCESS', 'HOURLY');
    end;