CREATE OR REPLACE TASK SUNO_PROD.PROD.ML_SONG_SUMMARY_INFO_V1_HOURLY_UPSERT
    WAREHOUSE = SUNO_PROD_ML_HOURLY_LARGE
    AFTER SUNO_PROD.PROD.AGG_PLAY_INFO_HOURLY_UPSERT
    AS
    DECLARE
        p_date DATE;
        p_hour INTEGER;
        next_p_date DATE;
        next_p_hour INTEGER;
        start_time TIMESTAMP_NTZ;
        row_count INTEGER;
        delete_p_date DATE;
        delete_p_hour INTEGER;
    BEGIN
        start_time := CURRENT_TIMESTAMP();
        -- SLA for fact_play and agg_play_info_hourly is 2 hours, so we use a 2 hour buffer.
        -- e.g. An upsert job for fact_play at 3:00 updates data from 2-3, and also the sessions from 1-2 which ended between 2-3.
        -- Then this job, which runs downstream of agg_play_info_hourly, should update everything in the data up until 2:00 (i.e. p_hour = 1)
        p_date := DATE(SYSDATE() - INTERVAL '2 HOURS');
        p_hour := HOUR(SYSDATE() - INTERVAL '2 HOURS');

        CALL SUNO_PROD.PROD.ML_SONG_SUMMARY_INFO_UPSERT_PROC(:p_date, :p_hour);

        SELECT COUNT(*) INTO :row_count
        FROM SUNO_PROD.PROD.ML_SONG_SUMMARY_INFO_V1
        WHERE 1=1
            AND p_date = :p_date
            AND p_hour = :p_hour;
        CALL TASK_MONITOR_INSERT_PROC('ML_SONG_SUMMARY_INFO_V1_HOURLY_UPSERT', :start_time, :row_count, 'SUCCESS', 'HOURLY');

        -- Delete old snapshots
        -- This is to prevent the table from growing too large
        delete_p_date := DATE(DATEADD(HOUR, -3, CREATE_HOUR_TS(:p_date, :p_hour)));
        delete_p_hour := HOUR(DATEADD(HOUR, -3, CREATE_HOUR_TS(:p_date, :p_hour)));
        CALL SUNO_PROD.PROD.ML_SONG_SUMMARY_INFO_DELETE_OLD_SNAPSHOT_PROC(:delete_p_date, :delete_p_hour);

    END;
