CREATE OR REPLACE PROCEDURE AGG_PLAY_INFO_HOURLY_UPSERT_PROC(start_p_date DATE, start_p_hour INTEGER, end_p_date DATE, end_p_hour INTEGER)
    RETURNS STRING
    LANGUAGE SQL
    AS
    $$
    BEGIN
        -- This procedure upserts the agg_play_info_hourly table for a given hour range.
        -- For its scheduled hourly run, it operates on a 2-hour delay to ensure the source data from fact_play is settled.
        -- It can also be used for backfills of arbitrary time ranges.

        MERGE INTO suno_prod.prod.agg_play_info_hourly apih
        USING (
            WITH fact_play_agg AS (
                SELECT
                    COALESCE(MAX(fp.user_id), MAX(du.user_id)) AS user_id
                    , fp.user_uid
                    , fp.clip_id
                    , BOOLOR_AGG(fp.is_user_song_owner) AS is_user_song_owner
                    , fp.platform
                    , threshold.play_duration_threshold
                    , SUM(CASE
                        WHEN fp.play_duration_seconds >= threshold.play_duration_threshold
                            THEN fp.play_duration_seconds
                        ELSE 0
                    END) AS play_duration_sec
                    , COUNT(CASE
                        WHEN fp.play_duration_seconds >= threshold.play_duration_threshold
                            THEN 1
                        END
                    ) AS play_cnt

                    -- High Intent Definition Versions:
                    -- v1: Play starts >2 hours after creation
                    , OBJECT_CONSTRUCT(
                        'v1', OBJECT_CONSTRUCT(
                            'play_duration_sec', SUM(CASE
                                WHEN fp.audio_player_events[0]:server_timestamp >= DATEADD('MINUTE', 120, dc.created_at)
                                    AND fp.play_duration_seconds >= threshold.play_duration_threshold
                                    THEN fp.play_duration_seconds
                                ELSE 0
                                END
                            ),
                            'play_cnt', COUNT(CASE
                                WHEN fp.audio_player_events[0]:server_timestamp >= DATEADD('MINUTE', 120, dc.created_at)
                                    AND fp.play_duration_seconds >= threshold.play_duration_threshold
                                    THEN 1
                                END
                            )
                        )
                    ) AS high_intent_listening

                    , fp.p_date
                    , fp.p_hour
                FROM
                    suno_prod.prod.fact_play fp
                LEFT JOIN
                    suno_prod.prod.dim_user du
                    ON fp.user_uid = du.user_uid
                CROSS JOIN (
                    SELECT
                        *
                    FROM
                        VALUES (0), (5)
                        AS t(play_duration_threshold)
                ) threshold
                LEFT JOIN
                    suno_prod.prod.dim_clip dc
                    ON fp.clip_id = dc.clip_id
                WHERE 1=1
                    AND fp.p_date BETWEEN :start_p_date AND :end_p_date
                    AND IFF(
                        :end_p_date > :start_p_date,
                        -- Multi-day case
                        (fp.p_date = :start_p_date AND fp.p_hour >= :start_p_hour) OR
                        (fp.p_date > :start_p_date AND fp.p_date < :end_p_date) OR
                        (fp.p_date = :end_p_date AND fp.p_hour <= :end_p_hour),
                        -- Single day case
                        fp.p_hour BETWEEN :start_p_hour AND :end_p_hour
                    )
                GROUP BY
                    fp.user_uid
                    , fp.clip_id
                    , fp.platform
                    , threshold.play_duration_threshold
                    , fp.p_date
                    , fp.p_hour
                HAVING
                    play_duration_sec > 0 OR play_cnt > 0 -- Exclude rows with no plays (typically for higher thresholds)
            )
            SELECT
                user_id
                , user_uid
                , clip_id
                , is_user_song_owner
                , platform
                , play_duration_threshold
                , play_duration_sec
                , play_cnt
                , high_intent_listening
                , p_date
                , p_hour
            FROM
                fact_play_agg
        ) source
        ON apih.user_uid = source.user_uid
        AND apih.clip_id = source.clip_id
        AND apih.platform = source.platform
        AND apih.play_duration_threshold = source.play_duration_threshold
        AND apih.p_date = source.p_date
        AND apih.p_hour = source.p_hour
        WHEN MATCHED THEN
            UPDATE SET
                user_id = source.user_id
                , is_user_song_owner = source.is_user_song_owner
                , play_duration_sec = source.play_duration_sec
                , play_cnt = source.play_cnt
                , high_intent_listening = source.high_intent_listening
        WHEN NOT MATCHED THEN
            INSERT (
                user_id
                , user_uid
                , clip_id
                , is_user_song_owner
                , platform
                , play_duration_threshold
                , play_duration_sec
                , play_cnt
                , high_intent_listening
                , p_date
                , p_hour
            )
            VALUES (
                source.user_id
                , source.user_uid
                , source.clip_id
                , source.is_user_song_owner
                , source.platform
                , source.play_duration_threshold
                , source.play_duration_sec
                , source.play_cnt
                , source.high_intent_listening
                , source.p_date
                , source.p_hour
            );

        RETURN 'SUCCESS';

    END;
    $$;
