CREATE OR REPLACE PROCEDURE FACT_PLAY_IOS_LEGACY_UPSERT_PROC(start_p_date DATE, start_p_hour INTEGER, end_p_date DATE, end_p_hour INTEGER)
    RETURNS STRING
    LANGUAGE SQL
    AS
    $$
    DECLARE
        -- This procedure will generally be used to upsert one hour of data, i.e. start_p_date = end_p_date and start_p_hour = end_p_hour.
        -- To account for song sessions that may start in the previous hour or end in the next, we automatically create a +/- 1 hour buffer
        -- around the provided start/end times.
        start_ts_unbuffered TIMESTAMP_NTZ := CREATE_HOUR_TS(:start_p_date::STRING, :start_p_hour);
        end_ts_unbuffered TIMESTAMP_NTZ := CREATE_HOUR_TS(:end_p_date::STRING, :end_p_hour);

        start_ts_buffered TIMESTAMP_NTZ := DATEADD(hour, -1, start_ts_unbuffered);
        end_ts_buffered TIMESTAMP_NTZ := DATEADD(hour, 1, end_ts_unbuffered);

        start_p_date_buffered DATE := DATE(start_ts_buffered);
        start_p_hour_buffered INTEGER := HOUR(start_ts_buffered);
        end_p_date_buffered DATE := DATE(end_ts_buffered);
        end_p_hour_buffered INTEGER := HOUR(end_ts_buffered);
    BEGIN
        -- Deduplicate iOS audio player events (legacy) and write to temp table
        CREATE OR REPLACE TEMPORARY TABLE audio_player_events_deduplicated AS (
            SELECT * FROM (
                SELECT
                    *
                    , ROW_NUMBER() OVER (
                        PARTITION BY
                            user_id
                            , clip_id
                            , action_name
                            , FLOOR(EXTRACT(epoch_millisecond FROM client_timestamp::TIMESTAMP_NTZ(9)) / 200) * 200 -- 200ms precision
                            , is_playing
                            , is_user_song_owner
                            , os_name
                        ORDER BY server_timestamp ASC
                    ) AS rn
                FROM APP_AUDIO_ACTIONS
                WHERE 1=1
                    AND p_date BETWEEN :start_p_date_buffered AND :end_p_date_buffered
                    AND IFF(
                        :end_p_date_buffered > :start_p_date_buffered,
                        -- Multi-day case
                        (p_date = :start_p_date_buffered AND p_hour >= :start_p_hour_buffered) OR
                        (p_date > :start_p_date_buffered AND p_date < :end_p_date_buffered) OR  -- Middle days (all hours)
                        (p_date = :end_p_date_buffered AND p_hour <= :end_p_hour_buffered),
                        -- Single day case
                        p_hour BETWEEN :start_p_hour_buffered AND :end_p_hour_buffered
                    )
                    AND user_id IS NOT NULL
                    AND LOWER(os_name) IN ('ios', 'ipados')
                    AND action_name IN (
                        'PlayNewSong'
                        , 'PlaySongTapped'
                        , 'PauseSongTapped'
                        , 'RestartSongTapped'
                        , 'PlayNextSong'
                        , 'PlayPreviousSong'
                        , 'AutoplayNextSong'
                        , 'SeekFinished'
                        , 'EndOfQueue'
                    )
            ) WHERE rn = 1
        );

        -- Create ordered actions with previous action tracking
        CREATE OR REPLACE TEMPORARY TABLE ordered_actions AS (
            SELECT
                user_id AS user_uid
                , clip_id
                , client_timestamp::TIMESTAMP_NTZ(9) AS client_timestamp
                , server_timestamp::TIMESTAMP_NTZ(9) AS server_timestamp
                , action_name
                , is_playing
                , is_user_song_owner
                , is_streaming
                , LOWER(os_name) AS platform
                , app_version
                , LEAD(action_name) OVER (
                    PARTITION BY user_id
                    ORDER BY client_timestamp
                ) AS next_action_name
                , LEAD(clip_id) OVER (
                    PARTITION BY user_id
                    ORDER BY client_timestamp
                ) AS next_clip_id
                , LEAD(client_timestamp) OVER (
                    PARTITION BY user_id
                    ORDER BY client_timestamp
                ) AS next_client_timestamp
                , LEAD(is_playing) OVER (
                    PARTITION BY user_id
                    ORDER BY client_timestamp
                ) AS next_is_playing
            FROM
                audio_player_events_deduplicated
            WHERE 1=1
                AND p_date BETWEEN :start_p_date_buffered AND :end_p_date_buffered
                AND IFF(
                    :end_p_date_buffered > :start_p_date_buffered,
                    -- Multi-day case
                    (p_date = :start_p_date_buffered AND p_hour >= :start_p_hour_buffered) OR
                    (p_date > :start_p_date_buffered AND p_date < :end_p_date_buffered) OR  -- Middle days (all hours)
                    (p_date = :end_p_date_buffered AND p_hour <= :end_p_hour_buffered),
                    -- Single day case
                    p_hour BETWEEN :start_p_hour_buffered AND :end_p_hour_buffered
                )
                AND user_id IS NOT NULL
                AND action_name IN (
                    'PlayNewSong'
                    , 'PlaySongTapped'
                    , 'PauseSongTapped'
                    , 'RestartSongTapped'
                    , 'PlayNextSong'
                    , 'PlayPreviousSong'
                    , 'AutoplayNextSong'
                    , 'SeekFinished'
                    , 'EndOfQueue'
                )
        );

        -- Assign play durations based on action name
        CREATE OR REPLACE TEMPORARY TABLE actions_with_play_durations AS (
            SELECT
                *
                -- Calculate play duration for this audio event
                , CASE
                    ---------- NO PLAY DURATION ----------
                    -- If this is the last audio event for the user (i.e. the next action is null), assign no play duration
                    WHEN next_action_name IS NULL THEN 0
                    -- If the next action happened more than 10 minutes after this action, assign no play duration.
                    -- This is only an issue in iOS legacy, where we do not have song sessions and use consecutive events across time to determine song sessions
                    WHEN DATEDIFF('minute', client_timestamp, next_client_timestamp) > 10 THEN 0
                    ---------- SAME SONG ----------
                    -- If the next action is a pause on the same song as the current action,
                    -- then assign a play duration based on timestamp differences between this action and the next action
                    WHEN next_clip_id = clip_id AND next_action_name = 'PauseSongTapped'
                        THEN DATEDIFF('second', client_timestamp, next_client_timestamp)
                    -- If the next action is a seek, restart, forward/backward, end of queue, or autoplay next song,
                    -- then assign a play duration based on timestamp differences between this action and the next action
                    WHEN (
                        next_clip_id = clip_id
                        AND next_is_playing
                        AND next_action_name IN (
                            'SeekFinished'
                            , 'RestartSongTapped'
                            , 'AutoplayNextSong'
                            , 'PlayPreviousSong'
                            , 'PlayNextSong'
                            , 'EndOfQueue'
                        )
                    ) THEN DATEDIFF('second', client_timestamp, next_client_timestamp)
                    ---------- DIFFERENT SONG ----------
                    -- If the next action is to play a new song and the current action is playing a song,
                    -- then assign a play duration based on timestamp differences between this action and the next action
                    WHEN next_action_name = 'PlayNewSong' AND action_name IN ('PlayNewSong', 'PlaySongTapped')
                        THEN DATEDIFF('second', client_timestamp, next_client_timestamp)
                    -- Otherwise, no play duration
                    ELSE 0
                END AS play_duration_seconds
                -- Song session index: incremented for events that start a new song session, partitioned by user
                , SUM(CASE WHEN action_name = 'PlayNewSong' THEN 1 ELSE 0 END)
                    OVER (PARTITION BY user_uid ORDER BY client_timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)
                    AS song_session_index
            FROM
                ordered_actions
        );

        -- Aggregate play data by user and song session
        CREATE OR REPLACE TEMPORARY TABLE song_sessions AS (
            SELECT
                apd.user_uid
                , apd.clip_id
                , du.user_id
                , apd.song_session_index
                , SUM(GREATEST(0, apd.play_duration_seconds)) AS play_duration_seconds -- Exclude negative play durations
                , BOOLOR_AGG(apd.is_user_song_owner) AS is_user_song_owner
                , BOOLOR_AGG(apd.is_streaming) AS is_play_during_generation
                , ANY_VALUE(apd.platform) AS platform
                , ANY_VALUE(apd.app_version) AS app_version
                , MIN(apd.client_timestamp) AS play_start_ts_utc
                , MAX(apd.client_timestamp) AS play_end_ts_utc
                , DATEDIFF('second', MIN(apd.client_timestamp), MAX(apd.client_timestamp)) AS song_session_duration_seconds
                , MIN_BY(apd.action_name, apd.client_timestamp) AS play_start_reason
                , CASE
                    WHEN MAX_BY(apd.action_name, apd.client_timestamp) = 'PauseSongTapped' THEN 'click_pause'
                    WHEN MAX_BY(apd.action_name, apd.client_timestamp) = 'AutoplayNextSong' THEN 'auto_play'
                    WHEN MAX_BY(apd.action_name, apd.client_timestamp) = 'EndOfQueue' THEN 'end_of_queue'
                    WHEN MAX_BY(apd.action_name, apd.client_timestamp) = 'PlayPreviousSong' THEN 'click_backward'
                    WHEN MAX_BY(apd.action_name, apd.client_timestamp) = 'PlayNextSong' THEN 'click_forward'
                    ELSE MAX_BY(apd.action_name, apd.client_timestamp)
                END AS play_end_reason
                , (
                    -- If the user seeked or restarted at any point, then the playback is not continuous
                    NOT BOOLOR_AGG(
                        apd.action_name IN (
                            'SeekFinished'
                            , 'RestartSongTapped'
                        )
                    )
                    -- If the user paused *and* resumed, then the playback is not continuous
                    AND NOT (
                        BOOLOR_AGG(apd.action_name = 'PauseSongTapped')
                        AND BOOLOR_AGG(apd.action_name IN ('PlaySongTapped', 'RestartSongTapped'))
                    )
                ) AS is_play_continuous
                , ARRAY_SLICE(ARRAY_AGG(
                    OBJECT_CONSTRUCT(
                        'action_name', apd.action_name,
                        'client_timestamp', apd.client_timestamp,
                        'server_timestamp', apd.server_timestamp,
                        'is_playing', apd.is_playing,
                        'play_duration_seconds', apd.play_duration_seconds
                    )
                ) WITHIN GROUP (ORDER BY apd.client_timestamp), 0, 30) AS audio_player_events
                , DATE(MIN_BY(apd.server_timestamp::TIMESTAMP_NTZ(9), apd.client_timestamp::TIMESTAMP_NTZ(9))) AS p_date
                , HOUR(MIN_BY(apd.server_timestamp::TIMESTAMP_NTZ(9), apd.client_timestamp::TIMESTAMP_NTZ(9))) AS p_hour
            FROM actions_with_play_durations apd
            LEFT JOIN DIM_USER du ON apd.user_uid = du.user_uid
            GROUP BY apd.user_uid
                , apd.clip_id
                , du.user_id
                , apd.song_session_index
            HAVING
                MIN_BY(apd.action_name, apd.client_timestamp) = 'PlayNewSong'
                AND SUM(apd.play_duration_seconds) > 0
        );

        -- Upsert iOS legacy plays
        MERGE INTO FACT_PLAY fp
        USING (
            SELECT
                MD5_TO_UUID(CONCAT(ss.user_uid, ss.clip_id, ss.play_start_ts_utc)) AS id
                , ss.user_id
                , ss.user_uid
                , ss.clip_id
                , NULL AS song_session_id

                -- Playback
                , ss.play_start_ts_utc
                , ss.play_end_ts_utc
                , LEAST(ss.song_session_duration_seconds, 6 * 60 * 60) AS song_session_duration_seconds -- Cap at 6 hours
                , LEAST(CEIL(LEAST(ss.play_duration_seconds, 2 * dc.duration_sec)), 3 * 60 * 60) AS play_duration_seconds -- Cap play_duration_seconds at 2x the clip duration and 3 hours max
                -- Legacy data does not support unique play duration
                , NULL AS unique_play_duration_seconds
                , NULL AS unique_play_percent
                -- If the previous song session ended recently (within 10 seconds) and ended due to AutoplayNextSong, then the play_start_reason is 'auto_play'
                , CASE
                    WHEN
                        DATEDIFF('second', ss.play_start_ts_utc, LAG(ss.play_end_ts_utc) OVER (PARTITION BY ss.user_uid ORDER BY ss.song_session_index)) <= 10
                        AND LAG(ss.play_end_reason) OVER (PARTITION BY ss.user_uid ORDER BY ss.song_session_index) = 'AutoplayNextSong'
                    THEN 'auto_play'
                    ELSE ss.play_start_reason
                END AS play_start_reason
                , CASE
                    -- If the next song session starts soon (within 10 seconds) and started due to previous/next, then the play_end_reason is based on those actions
                    WHEN
                        DATEDIFF('second', ss.play_end_ts_utc, LEAD(ss.play_start_ts_utc) OVER (PARTITION BY ss.user_uid ORDER BY ss.song_session_index)) <= 10
                        AND LEAD(ss.play_start_reason) OVER (PARTITION BY ss.user_uid ORDER BY ss.song_session_index) IN ('click_backward', 'click_forward')
                    THEN LEAD(ss.play_start_reason) OVER (PARTITION BY ss.user_uid ORDER BY ss.song_session_index)
                    -- If the next song session starts soon (within 10 seconds) and started due to PlayNewSong, then the play_end_reason is 'click_new_song'
                    WHEN
                        DATEDIFF('second', ss.play_end_ts_utc, LEAD(ss.play_start_ts_utc) OVER (PARTITION BY ss.user_uid ORDER BY ss.song_session_index)) <= 10
                        AND LEAD(ss.play_start_reason) OVER (PARTITION BY ss.user_uid ORDER BY ss.song_session_index) = 'PlayNewSong'
                    THEN 'click_new_song'
                    -- Otherwise, use the play_end_reason
                    ELSE ss.play_end_reason
                END AS play_end_reason
                , ss.is_play_continuous
                , NULL AS is_manually_started
                , OBJECT_CONSTRUCT(
                    'v1',
                    CASE
                        WHEN NOT ss.is_user_song_owner THEN TRUE  -- Not their song = always high intent
                        WHEN ss.is_user_song_owner AND
                             COALESCE(
                                 ss.audio_player_events[0]:server_timestamp::TIMESTAMP_NTZ(9),
                                 CREATE_HOUR_TS(ss.p_date::STRING, ss.p_hour)
                             ) >= dc.created_at + INTERVAL '2 HOUR'
                        THEN TRUE  -- Their song + 2+ hours after creation = high intent
                        ELSE FALSE  -- Their song + <2 hours after creation = not high intent
                    END
                ) AS is_high_intent_listening
                , ss.audio_player_events

                -- Clip info
                , COALESCE(dc.duration_sec, 0) AS clip_duration_seconds
                , ss.is_user_song_owner
                , ss.is_play_during_generation

                -- Context info (not available in legacy data)
                , NULL AS previous_song_session_id
                , NULL AS play_page_type
                , NULL AS play_context_type
                , NULL AS play_context_id
                , NULL AS play_context_metadata
                , NULL AS origin_context_id
                , NULL AS origin_item_type
                , NULL AS origin_song_id
                , NULL AS origin_item_metadata
                , NULL AS browser_history
                , NULL AS is_song_autoplay_queue
                , NULL AS audio_player_metadata

                -- Device metadata
                , NULL AS session_id
                , ss.platform
                , ss.app_version
                , NULL AS is_mobile
                , NULL AS country
                , NULL AS city
                , ss.p_date
                , ss.p_hour
            FROM song_sessions ss
            LEFT JOIN
                dim_clip dc
                ON dc.clip_id = ss.clip_id
            WHERE 1=1
                AND ss.user_uid IS NOT NULL
                AND ss.clip_id IS NOT NULL

                -- Only include records with actual play time (cap at 3 hours)
                AND ss.play_duration_seconds > 0
                AND ss.play_duration_seconds <= 3 * 60 * 60
        ) source
        ON fp.id = source.id
        AND fp.p_date BETWEEN :start_p_date_buffered AND :end_p_date_buffered
        AND IFF(
            :end_p_date_buffered > :start_p_date_buffered,
            -- Multi-day case
            (fp.p_date = :start_p_date_buffered AND fp.p_hour >= :start_p_hour_buffered) OR
            (fp.p_date > :start_p_date_buffered AND fp.p_date < :end_p_date_buffered) OR
            (fp.p_date = :end_p_date_buffered AND fp.p_hour <= :end_p_hour_buffered),
            -- Single day case
            fp.p_hour BETWEEN :start_p_hour_buffered AND :end_p_hour_buffered
        )
        WHEN MATCHED AND (
            -- Only update if the source record's play_end_ts_utc is greater than or equal to the existing one
            -- This prevents overwriting more complete sessions with older data
            source.play_end_ts_utc >= fp.play_end_ts_utc
        ) THEN
            UPDATE SET
                user_id = source.user_id
                , user_uid = source.user_uid
                , clip_id = source.clip_id
                , song_session_id = source.song_session_id
                , play_start_ts_utc = source.play_start_ts_utc
                , play_end_ts_utc = source.play_end_ts_utc
                , song_session_duration_seconds = source.song_session_duration_seconds
                , play_duration_seconds = source.play_duration_seconds
                , unique_play_duration_seconds = source.unique_play_duration_seconds
                , unique_play_percent = source.unique_play_percent
                , play_start_reason = source.play_start_reason
                , play_end_reason = source.play_end_reason
                , is_play_continuous = source.is_play_continuous
                , is_manually_started = source.is_manually_started
                , is_high_intent_listening = source.is_high_intent_listening
                , audio_player_events = source.audio_player_events
                , clip_duration_seconds = source.clip_duration_seconds
                , is_user_song_owner = source.is_user_song_owner
                , is_play_during_generation = source.is_play_during_generation
                , previous_song_session_id = source.previous_song_session_id
                , play_page_type = source.play_page_type
                , play_context_type = source.play_context_type
                , play_context_id = source.play_context_id
                , play_context_metadata = source.play_context_metadata
                , origin_context_id = source.origin_context_id
                , origin_item_type = source.origin_item_type
                , origin_song_id = source.origin_song_id
                , origin_item_metadata = source.origin_item_metadata
                , browser_history = source.browser_history
                , is_song_autoplay_queue = source.is_song_autoplay_queue
                , audio_player_metadata = source.audio_player_metadata
                , session_id = source.session_id
                , platform = source.platform
                , app_version = source.app_version
                , is_mobile = source.is_mobile
                , country = source.country
                , city = source.city
                , p_date = source.p_date
                , p_hour = source.p_hour
        WHEN NOT MATCHED THEN
            INSERT (
                id
                , user_id
                , user_uid
                , clip_id
                , song_session_id
                , play_start_ts_utc
                , play_end_ts_utc
                , song_session_duration_seconds
                , play_duration_seconds
                , unique_play_duration_seconds
                , unique_play_percent
                , play_start_reason
                , play_end_reason
                , is_play_continuous
                , is_manually_started
                , is_high_intent_listening
                , audio_player_events
                , clip_duration_seconds
                , is_user_song_owner
                , is_play_during_generation
                , previous_song_session_id
                , play_page_type
                , play_context_type
                , play_context_id
                , play_context_metadata
                , origin_context_id
                , origin_item_type
                , origin_song_id
                , origin_item_metadata
                , browser_history
                , is_song_autoplay_queue
                , audio_player_metadata
                , session_id
                , platform
                , app_version
                , is_mobile
                , country
                , city
                , p_date
                , p_hour
            ) VALUES (
                source.id
                , source.user_id
                , source.user_uid
                , source.clip_id
                , source.song_session_id
                , source.play_start_ts_utc
                , source.play_end_ts_utc
                , source.song_session_duration_seconds
                , source.play_duration_seconds
                , source.unique_play_duration_seconds
                , source.unique_play_percent
                , source.play_start_reason
                , source.play_end_reason
                , source.is_play_continuous
                , source.is_manually_started
                , source.is_high_intent_listening
                , source.audio_player_events
                , source.clip_duration_seconds
                , source.is_user_song_owner
                , source.is_play_during_generation
                , source.previous_song_session_id
                , source.play_page_type
                , source.play_context_type
                , source.play_context_id
                , source.play_context_metadata
                , source.origin_context_id
                , source.origin_item_type
                , source.origin_song_id
                , source.origin_item_metadata
                , source.browser_history
                , source.is_song_autoplay_queue
                , source.audio_player_metadata
                , source.session_id
                , source.platform
                , source.app_version
                , source.is_mobile
                , source.country
                , source.city
                , source.p_date
                , source.p_hour
            );

    RETURN 'SUCCESS';
    END;
$$;
