CREATE OR REPLACE PROCEDURE FACT_PLAY_IOS_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 omniplayer events and write to temp table
        CREATE OR REPLACE TEMPORARY TABLE omniplayer_events_deduplicated AS (
            SELECT * FROM (
                SELECT
                    *
                    , ROW_NUMBER() OVER (
                        PARTITION BY
                            user_id
                            , anonymous_id
                            , session_id
                            , TRY_PARSE_JSON(context):songSessionId::VARCHAR
                            , element_id
                            , category
                            , action_name
                            , action_type
                            , ARRAY_SORT(TRY_PARSE_JSON(context)) -- Sort the array to deduplicate
                            , FLOOR(EXTRACT(epoch_millisecond FROM client_timestamp::TIMESTAMP_NTZ(9)) / 200) * 200 -- 200ms precision
                        ORDER BY server_timestamp ASC
                    ) AS rn
                FROM
                    app_event
                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 os_name = 'iOS'
                    AND category = 'omni_player'
                    AND context IS NOT NULL
                    AND user_id IS NOT NULL
                    AND TRY_PARSE_JSON(context):songSessionId::VARCHAR IS NOT NULL
            ) WHERE rn = 1
        );

        -- Unique play segments
        CREATE OR REPLACE TEMPORARY TABLE unique_ios_play_segments AS (
            WITH

            -- Sort play segments by start time and end time, and get cumulative min start time and max end time in the song session
            sorted_play_segments AS (
                SELECT
                    user_id AS user_uid
                    , element_id AS clip_id
                    , TRY_PARSE_JSON(context):songSessionId::VARCHAR AS song_session_id
                    , TRY_PARSE_JSON(context):startTime::DOUBLE AS start_time
                    , TRY_PARSE_JSON(context):endTime::DOUBLE AS end_time
                    , TRY_PARSE_JSON(context):playDuration::DOUBLE AS play_duration_seconds
                    , MIN(start_time) OVER (
                        PARTITION BY song_session_id
                        ORDER BY start_time, end_time
                        ROWS BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING
                    ) AS cumulative_min_start_time
                    , MAX(end_time) OVER (
                        PARTITION BY song_session_id
                        ORDER BY start_time, end_time
                        ROWS BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING
                    ) AS cumulative_max_end_time
                FROM
                    omniplayer_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 element_id IS NOT NULL
                    AND TRY_PARSE_JSON(context):songSessionId::VARCHAR IS NOT NULL
                    AND TRY_PARSE_JSON(context):startTime::DOUBLE IS NOT NULL
                    AND TRY_PARSE_JSON(context):endTime::DOUBLE IS NOT NULL
                    AND TRY_PARSE_JSON(context):playDuration::DOUBLE BETWEEN 0 AND 1 * 60 * 60 -- Exclude play segments that are longer than 1 hour
                    AND TRY_PARSE_JSON(context):startTime::DOUBLE < TRY_PARSE_JSON(context):endTime::DOUBLE -- start time should be before end time
                    AND TRY_PARSE_JSON(context):startTime::DOUBLE >= 0  -- Filter out negative start times
                    AND TRY_PARSE_JSON(context):endTime::DOUBLE >= 0    -- Filter out negative end times
                ORDER BY
                    TRY_PARSE_JSON(context):songSessionId::VARCHAR
                    , TRY_PARSE_JSON(context):startTime::DOUBLE
                    , TRY_PARSE_JSON(context):endTime::DOUBLE
            )

            -- Filter out play segments that are subsumed by previous play segments
            , non_subsumed_play_segments AS (
                SELECT
                    user_uid
                    , clip_id
                    , song_session_id
                    , start_time
                    , end_time
                    , CASE
                        WHEN start_time > LAG(end_time, 1, -1) OVER (
                            PARTITION BY user_uid, clip_id, song_session_id
                            ORDER BY start_time, end_time
                        ) THEN 1
                        ELSE 0
                    END AS is_new_group
                FROM
                    sorted_play_segments
                WHERE 1=1
                    AND (
                        cumulative_min_start_time IS NULL -- First row in session
                        OR start_time < cumulative_min_start_time -- Starts before any previous range
                        OR end_time > cumulative_max_end_time -- Ends after any previous range
                    )
            )

            -- Assign group IDs
            , grouped_play_segments AS (
                SELECT
                    user_uid
                    , clip_id
                    , song_session_id
                    , start_time
                    , end_time
                    , SUM(is_new_group) OVER (
                        PARTITION BY user_uid, clip_id, song_session_id
                        ORDER BY start_time, end_time
                    ) AS group_id
                FROM
                    non_subsumed_play_segments
            )

            -- Summarize play groups
            , grouped_play_segments_summary AS (
                SELECT
                    user_uid
                    , clip_id
                    , song_session_id
                    , group_id
                    , MIN(start_time) AS grouped_segment_start_time
                    , MAX(end_time) AS grouped_segment_end_time
                FROM
                    grouped_play_segments
                GROUP BY
                    user_uid
                    , clip_id
                    , song_session_id
                    , group_id
            )

            SELECT
                user_uid
                , clip_id
                , song_session_id
                , SUM(grouped_segment_end_time - grouped_segment_start_time) AS unique_play_duration_seconds
            FROM
                grouped_play_segments_summary
            GROUP BY
                user_uid
                , clip_id
                , song_session_id
        );

        -- Create song sessions temporary table
        CREATE OR REPLACE TEMPORARY TABLE song_sessions AS (
            SELECT
                -- Primary key and identifiers
                user_id AS user_uid
                , MODE(element_id) AS clip_id
                , TRY_PARSE_JSON(context):songSessionId::VARCHAR AS song_session_id

                -- Playback
                , MIN(client_timestamp::TIMESTAMP_NTZ(9)) AS play_start_ts_utc
                , MAX(client_timestamp::TIMESTAMP_NTZ(9)) AS play_end_ts_utc
                , DATEDIFF('second', MIN(client_timestamp::TIMESTAMP_NTZ(9)), MAX(client_timestamp::TIMESTAMP_NTZ(9))) AS song_session_duration_seconds
                , CASE
                    -- For iOS app versions 1.23.0 to 1.29.0, the play duration was recorded as an updating value, so we use the last value to measure the play duration
                    WHEN SPLIT(MIN_BY(app_version, client_timestamp::TIMESTAMP_NTZ(9)), '.')[1] = '1'
                        AND SPLIT(MIN_BY(app_version, client_timestamp::TIMESTAMP_NTZ(9)), '.')[2]::INTEGER BETWEEN 23 AND 29
                        THEN GREATEST(0, MAX_BY(TRY_PARSE_JSON(context):playDuration::DOUBLE, client_timestamp::TIMESTAMP_NTZ(9)))
                    -- For all other app versions, we use the sum of the play durations
                    ELSE SUM(GREATEST(0, TRY_PARSE_JSON(context):playDuration::DOUBLE))
                END AS play_duration_seconds -- Exclude negative play durations
                , CASE
                    WHEN MIN_BY(action_name, client_timestamp::TIMESTAMP_NTZ(9)) = 'PlayNewSong' THEN 'click_song'
                    WHEN MIN_BY(action_name, client_timestamp::TIMESTAMP_NTZ(9)) = 'ForwardPlayNewSong' THEN 'click_forward'
                    WHEN MIN_BY(action_name, client_timestamp::TIMESTAMP_NTZ(9)) = 'BackwardPlayNewSong' THEN 'click_backward'
                    WHEN MIN_BY(action_name, client_timestamp::TIMESTAMP_NTZ(9)) = 'BackwardRepeatSong' THEN 'click_backward_repeat'
                    WHEN MIN_BY(action_name, client_timestamp::TIMESTAMP_NTZ(9)) IN ('AutoRepeatPlayNewSong', 'AutoPlayNewSong') THEN 'auto_play'
                    WHEN MIN_BY(action_name, client_timestamp::TIMESTAMP_NTZ(9)) = 'AutoRepeatPlaySong' THEN 'auto_repeat'
                    ELSE MIN_BY(action_name, client_timestamp::TIMESTAMP_NTZ(9))
                END AS play_start_reason
                , CASE
                    WHEN MAX_BY(action_name, client_timestamp::TIMESTAMP_NTZ(9)) = 'SongEnd' THEN 'song_end'
                    WHEN MAX_BY(action_name, client_timestamp::TIMESTAMP_NTZ(9)) = 'PauseSong' THEN 'click_pause'
                    WHEN MAX_BY(action_name, client_timestamp::TIMESTAMP_NTZ(9)) = 'PlayNewSongPauseSong' THEN 'click_new_song'
                    WHEN MAX_BY(action_name, client_timestamp::TIMESTAMP_NTZ(9)) = 'BackwardPauseSong' THEN 'click_backward'
                    WHEN MAX_BY(action_name, client_timestamp::TIMESTAMP_NTZ(9)) = 'ForwardPausePreSong' THEN 'click_forward'
                    ELSE MAX_BY(action_name, client_timestamp::TIMESTAMP_NTZ(9))
                END AS play_end_reason
                , (
                    -- If the user seeked or backwards repeated at any point, then the playback is not continuous
                    NOT BOOLOR_AGG(
                        action_name IN (
                            'SeekProgressBarPauseSong', 'SeekProgressBarPlaySong',
                            'SeekProgressBarPauseSongInMediaSession', 'SeekProgressBarPlaySongInMediaSession',
                            'BackwardRepeatSong'
                        )
                    )
                    -- If the user paused *and* resumed/repeated, then the playback is not continuous
                    AND NOT (
                        BOOLOR_AGG(action_name = 'PauseSong')
                        AND BOOLOR_AGG(action_name IN ('PlaySong', 'BackwardRepeatSong'))
                    )
                ) AS is_play_continuous
                , MIN_BY(action_name, client_timestamp::TIMESTAMP_NTZ(9)) IN (
                    'PlayNewSong', 'ForwardPlayNewSong', 'BackwardPlayNewSong', 'BackwardRepeatSong'
                ) AS is_manually_started
                , ARRAY_SLICE(ARRAY_AGG(
                    OBJECT_CONSTRUCT(
                        'action_name', action_name,
                        'client_timestamp', client_timestamp::TIMESTAMP_NTZ(9),
                        'server_timestamp', server_timestamp::TIMESTAMP_NTZ(9),
                        'start_time', TRY_PARSE_JSON(context):startTime::DOUBLE,
                        'end_time', TRY_PARSE_JSON(context):endTime::DOUBLE,
                        'play_duration_seconds', TRY_PARSE_JSON(context):playDuration::DOUBLE
                    )
                ) WITHIN GROUP (ORDER BY client_timestamp::TIMESTAMP_NTZ(9)), 0, 30) AS audio_player_events

                -- Clip info
                , ANY_VALUE(TRY_PARSE_JSON(context):isUserSongOwner::BOOLEAN) AS is_user_song_owner

                -- Context info
                , MIN_BY(TRY_PARSE_JSON(context):previousSongSessionId::VARCHAR, client_timestamp::TIMESTAMP_NTZ(9)) AS previous_song_session_id
                , ANY_VALUE(session_id) AS session_id
                , LOWER(MIN_BY(os_name, client_timestamp::TIMESTAMP_NTZ(9))) AS platform
                , MIN_BY(app_version, client_timestamp::TIMESTAMP_NTZ(9)) AS app_version
                , MIN_BY(country, client_timestamp::TIMESTAMP_NTZ(9)) AS country
                , MIN_BY(city, client_timestamp::TIMESTAMP_NTZ(9)) AS city
                , DATE(MIN_BY(server_timestamp::TIMESTAMP_NTZ(9), client_timestamp::TIMESTAMP_NTZ(9))) AS p_date
                , HOUR(MIN_BY(server_timestamp::TIMESTAMP_NTZ(9), client_timestamp::TIMESTAMP_NTZ(9))) AS p_hour
            FROM
                omniplayer_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 TRY_PARSE_JSON(context):songSessionId::VARCHAR IS NOT NULL
            GROUP BY
                user_id
                , TRY_PARSE_JSON(context):songSessionId::VARCHAR
            HAVING
                NOT (COUNT(*) = 1 AND MIN_BY(action_name, client_timestamp::TIMESTAMP_NTZ(9)) = 'PlayNewSong') -- Exclude song sessions that are a single PlayNewSong event https://linear.app/sunomusic/issue/IOS-1411/check-for-double-counting-of-play-count-due-to-double-firing
        );

        -- Upsert iOS plays
        MERGE INTO FACT_PLAY fp
        USING (
            SELECT
                MD5_TO_UUID(CONCAT(ss.user_uid, ss.clip_id, ss.song_session_id)) AS id
                , du.user_id
                , ss.user_uid
                , ss.clip_id
                , ss.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(ss.play_duration_seconds), 3 * 60 * 60) AS play_duration_seconds -- Cap at 3 hours
                , COALESCE(CEIL(ups.unique_play_duration_seconds), 0) AS unique_play_duration_seconds
                , CASE
                    WHEN dc.duration_sec = 0 THEN NULL
                    ELSE LEAST(DIV0(CEIL(ups.unique_play_duration_seconds), dc.duration_sec), 1)
                END AS unique_play_percent
                , ss.play_start_reason
                , ss.play_end_reason
                , ss.is_play_continuous
                , ss.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
                , dc.duration_sec AS clip_duration_seconds
                , ss.is_user_song_owner
                , NULL AS is_play_during_generation

                -- Context info
                , ss.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
                , ss.session_id
                , ss.platform
                , ss.app_version
                , NULL AS is_mobile
                , ss.country
                , ss.city

                -- Partition fields
                , ss.p_date
                , ss.p_hour
            FROM
                song_sessions ss
            LEFT JOIN
                dim_user du
                ON du.user_uid = ss.user_uid
            LEFT JOIN
                unique_ios_play_segments ups
                ON ups.user_uid = ss.user_uid
                AND ups.clip_id = ss.clip_id
                AND ups.song_session_id = ss.song_session_id
            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
                AND ss.song_session_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;
$$;
