-- Step 3: Create song sessions from web audio player actions
-- Aggregates deduplicated events into play sessions
-- IMPORTANT: Uses INNER JOIN with events_per_song_session to filter out sessions with 1000+ events

CREATE OR REPLACE TEMPORARY TABLE {{ web_play_sessions_table_name }} AS (
    SELECT
        -- Primary key and identifiers
        wapad.user_id AS user_uid
        , CASE
            WHEN MIN_BY(wapad.context_type, wapad.client_timestamp::TIMESTAMP_NTZ(9)) = 'studio_new_timeline'
                THEN MIN_BY(wapad.context_id, wapad.client_timestamp::TIMESTAMP_NTZ(9)) -- For studio timeline clips, we use the studio project ID as the clip ID
            ELSE MODE(wapad.song_id) -- Sometimes there are multiple song_ids in a session (logging issues), so we use the mode
            END AS clip_id
        , wapad.song_session_id

        -- Playback
        , MIN(wapad.client_timestamp::TIMESTAMP_NTZ(9)) AS play_start_ts_utc
        , MAX(wapad.client_timestamp::TIMESTAMP_NTZ(9)) AS play_end_ts_utc
        , DATEDIFF('second', MIN(wapad.client_timestamp::TIMESTAMP_NTZ(9)), MAX(wapad.client_timestamp::TIMESTAMP_NTZ(9))) AS song_session_duration_seconds
        , SUM(GREATEST(0, wapad.play_duration)) AS play_duration_seconds -- Exclude negative play durations
        , CASE
            WHEN MIN_BY(wapad.action_index, wapad.client_timestamp::TIMESTAMP_NTZ(9)) > 0 THEN 'missing_first_event'
            WHEN MIN_BY(wapad.action_name, wapad.client_timestamp::TIMESTAMP_NTZ(9)) = 'PlayNewSong' THEN 'click_song'
            WHEN MIN_BY(wapad.action_name, wapad.client_timestamp::TIMESTAMP_NTZ(9)) = 'ForwardPlayNewSong' THEN 'click_forward'
            WHEN MIN_BY(wapad.action_name, wapad.client_timestamp::TIMESTAMP_NTZ(9)) = 'BackwardPlayNewSong' THEN 'click_backward'
            WHEN MIN_BY(wapad.action_name, wapad.client_timestamp::TIMESTAMP_NTZ(9)) = 'BackwardRepeatSong' THEN 'click_backward_repeat'
            WHEN MIN_BY(wapad.action_name, wapad.client_timestamp::TIMESTAMP_NTZ(9)) = 'ForwardRepeatSong' THEN 'click_forward_repeat'
            WHEN MIN_BY(wapad.action_name, wapad.client_timestamp::TIMESTAMP_NTZ(9)) IN ('AutoRepeatPlayNewSong', 'AutoPlayNewSong') THEN 'auto_play'
            WHEN MIN_BY(wapad.action_name, wapad.client_timestamp::TIMESTAMP_NTZ(9)) = 'AutoRepeatPlaySong' THEN 'auto_repeat'
            ELSE MIN_BY(wapad.action_name, wapad.client_timestamp::TIMESTAMP_NTZ(9))
        END AS play_start_reason
        , CASE
            WHEN MAX_BY(wapad.action_name, wapad.client_timestamp::TIMESTAMP_NTZ(9)) = 'SongEnd' THEN 'song_end'
            WHEN MAX_BY(wapad.action_name, wapad.client_timestamp::TIMESTAMP_NTZ(9)) = 'PauseSong' THEN 'click_pause'
            WHEN MAX_BY(wapad.action_name, wapad.client_timestamp::TIMESTAMP_NTZ(9)) = 'PlayNewSongPauseSong' THEN 'click_new_song'
            WHEN MAX_BY(wapad.action_name, wapad.client_timestamp::TIMESTAMP_NTZ(9)) = 'BackwardPauseSong' THEN 'click_backward'
            WHEN MAX_BY(wapad.action_name, wapad.client_timestamp::TIMESTAMP_NTZ(9)) IN ('ForwardPausePreSong', 'ForwardPauseSong') THEN 'click_forward'
            ELSE MAX_BY(wapad.action_name, wapad.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(
                wapad.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(wapad.action_name = 'PauseSong')
                AND BOOLOR_AGG(wapad.action_name IN ('PlaySong', 'BackwardRepeatSong'))
            )
        ) AS is_play_continuous
        , MIN_BY(wapad.action_name, wapad.client_timestamp::TIMESTAMP_NTZ(9)) IN (
            'PlayNewSong', 'ForwardPlayNewSong', 'BackwardPlayNewSong', 'BackwardRepeatSong'
        ) AS is_manually_started
        , ARRAY_SLICE(ARRAY_AGG(
            OBJECT_CONSTRUCT(
                'action_name', wapad.action_name,
                'action_index', wapad.action_index,
                'client_timestamp', wapad.client_timestamp::TIMESTAMP_NTZ(9),
                'server_timestamp', wapad.server_timestamp::TIMESTAMP_NTZ(9),
                'start_time', wapad.start_time,
                'end_time', wapad.end_time,
                'play_duration_seconds', wapad.play_duration
            )
        ) WITHIN GROUP (ORDER BY wapad.client_timestamp::TIMESTAMP_NTZ(9)), 0, 30) AS audio_player_events

        -- Clip info
        , MAX_BY(wapad.song_length, wapad.client_timestamp::TIMESTAMP_NTZ(9)) AS clip_duration_seconds -- Get last song length due to bug where first song_length is sometimes the previous song session's song length
        , BOOLOR_AGG(wapad.is_user_song_owner) AS is_user_song_owner -- Sometimes this is null, so we use a boolean aggregation

        -- Context info
        , MIN_BY(wapad.previous_song_session_id, wapad.client_timestamp::TIMESTAMP_NTZ(9)) AS previous_song_session_id
        , MIN_BY(wapad.context_type, wapad.client_timestamp::TIMESTAMP_NTZ(9)) AS play_context_type
        , MIN_BY(wapad.context_id, wapad.client_timestamp::TIMESTAMP_NTZ(9)) AS play_context_id
        , CASE
            WHEN MIN_BY(wapad.click_source_url, wapad.client_timestamp::TIMESTAMP_NTZ(9)) = '/' THEN 'home'
            WHEN STARTSWITH(SPLIT(MIN_BY(wapad.click_source_url, wapad.client_timestamp::TIMESTAMP_NTZ(9)), '/')[1], '@') THEN 'artist'
            ELSE SPLIT(MIN_BY(wapad.click_source_url, wapad.client_timestamp::TIMESTAMP_NTZ(9)), '/')[1]
        END AS play_page_type
        , OBJECT_CONSTRUCT(
            'surface_type', MIN_BY(TRY_PARSE_JSON(wapad.metadata):surfaceType::VARCHAR, wapad.client_timestamp::TIMESTAMP_NTZ(9)),
            'surface_id', MIN_BY(TRY_PARSE_JSON(wapad.metadata):surfaceId::VARCHAR, wapad.client_timestamp::TIMESTAMP_NTZ(9)),
            'click_source_url', MIN_BY(wapad.click_source_url, wapad.client_timestamp::TIMESTAMP_NTZ(9))
        ) AS play_context_metadata
        -- For browser history, we want browser history at the start of the song session
        , ARRAY_SLICE(MIN_BY(TRY_PARSE_JSON(wapad.metadata):browserHistory::VARIANT, wapad.client_timestamp::TIMESTAMP_NTZ(9)), 0, 30) AS browser_history
        , MIN_BY(wapad.is_song_autoplay_queue, wapad.client_timestamp::TIMESTAMP_NTZ(9)) AS is_song_autoplay_queue
        , OBJECT_CONSTRUCT(
            'volume', MIN_BY(wapad.volume, wapad.client_timestamp::TIMESTAMP_NTZ(9)),
            'is_auto_play_on', MIN_BY(wapad.is_auto_play_on, wapad.client_timestamp::TIMESTAMP_NTZ(9)),
            'is_repeat_on', MIN_BY(wapad.is_repeat_on, wapad.client_timestamp::TIMESTAMP_NTZ(9)),
            'is_shuffle_on', MIN_BY(wapad.is_shuffle_on, wapad.client_timestamp::TIMESTAMP_NTZ(9))
        ) AS audio_player_metadata

        -- Use min_by to capture the first value in this song session, i.e. where the song session started
        , ANY_VALUE(wapad.session_id) AS session_id
        , MIN_BY(wapad.is_mobile, wapad.client_timestamp::TIMESTAMP_NTZ(9)) AS is_mobile
        , MIN_BY(wapad.country, wapad.client_timestamp::TIMESTAMP_NTZ(9)) AS country
        , MIN_BY(wapad.city, wapad.client_timestamp::TIMESTAMP_NTZ(9)) AS city

        -- Partition fields
        , DATE(MIN_BY(wapad.server_timestamp::TIMESTAMP_NTZ(9), wapad.client_timestamp::TIMESTAMP_NTZ(9))) AS p_date
        , HOUR(MIN_BY(wapad.server_timestamp::TIMESTAMP_NTZ(9), wapad.client_timestamp::TIMESTAMP_NTZ(9))) AS p_hour
    FROM
        {{ web_audio_player_actions_dedup_table_name }} wapad
    INNER JOIN
        {{ web_events_per_song_session_table_name }} epss
        ON epss.song_session_id = wapad.song_session_id
        AND epss.n_events < 1000 -- Exclude song sessions with more than 1000 events; 0.0025% of song sessions fit this criteria. This is to prevent storage overflow
    WHERE 1=1
        AND wapad.song_session_id IS NOT NULL
    GROUP BY
        wapad.user_id
        , wapad.song_session_id
    HAVING
        COUNT(*) >= 1 -- At least 1 event in session
);
