CREATE OR REPLACE TEMPORARY TABLE {{ sessions_table_name }} AS (
    SELECT
        -- Primary key and identifiers
        user_id AS user_uid
        , hook_id
        , hook_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 hook_session_duration_seconds
        , SUM(GREATEST(0, play_duration_seconds)) AS play_duration_seconds -- Exclude negative play durations
        , CASE
            WHEN MIN_BY(action_name, client_timestamp::TIMESTAMP_NTZ(9)) = 'PlayNewHook' THEN 'play_new'
            WHEN MIN_BY(action_name, client_timestamp::TIMESTAMP_NTZ(9)) = 'PlayHook' THEN 'click_play'
            WHEN MIN_BY(action_name, client_timestamp::TIMESTAMP_NTZ(9)) = 'ScrollDownPlayNewHook' THEN 'scroll_down'
            WHEN MIN_BY(action_name, client_timestamp::TIMESTAMP_NTZ(9)) = 'ScrollUpPlayNewHook' THEN 'scroll_up'
            WHEN MIN_BY(action_name, client_timestamp::TIMESTAMP_NTZ(9)) = 'AutoRepeatPlayHook' THEN 'auto_repeat'
            WHEN MIN_BY(action_name, client_timestamp::TIMESTAMP_NTZ(9)) = 'CloseOmniPlayerPlayHook' THEN 'close_omni_player'
            WHEN MIN_BY(action_name, client_timestamp::TIMESTAMP_NTZ(9)) = 'TapTabPlayHook' THEN 'tap_tab'
            WHEN MIN_BY(action_name, client_timestamp::TIMESTAMP_NTZ(9)) = 'CloseCreateHookPlayHook' THEN 'close_create_hook'
            WHEN MIN_BY(action_name, client_timestamp::TIMESTAMP_NTZ(9)) = 'CloseCreateSongPlayHook' THEN 'close_create_song'
            WHEN MIN_BY(action_name, client_timestamp::TIMESTAMP_NTZ(9)) = 'ForegroundAppPlayHook' THEN 'foreground_app'
            WHEN MIN_BY(action_name, client_timestamp::TIMESTAMP_NTZ(9)) = 'TapBackPlayHook' THEN 'tap_back'
            WHEN MIN_BY(action_name, client_timestamp::TIMESTAMP_NTZ(9)) = 'OpenDeeplinkPlayHook' THEN 'open_deeplink'
            WHEN MIN_BY(action_name, client_timestamp::TIMESTAMP_NTZ(9)) IN ('CloseCoverPlayHook', 'CloseExtendPlayHook') THEN 'close_remix'
            ELSE MIN_BY(action_name, client_timestamp::TIMESTAMP_NTZ(9))
        END AS hook_start_reason
        , CASE
            WHEN MAX_BY(action_name, client_timestamp::TIMESTAMP_NTZ(9)) = 'PauseHook' THEN 'click_pause'
            WHEN MAX_BY(action_name, client_timestamp::TIMESTAMP_NTZ(9)) = 'ScrollDownPauseHook' THEN 'scroll_down'
            WHEN MAX_BY(action_name, client_timestamp::TIMESTAMP_NTZ(9)) = 'ScrollUpPauseHook' THEN 'scroll_up'
            WHEN MAX_BY(action_name, client_timestamp::TIMESTAMP_NTZ(9)) = 'TapSongPillPauseHook' THEN 'tap_song_pill'
            WHEN MAX_BY(action_name, client_timestamp::TIMESTAMP_NTZ(9)) = 'TapCreateHookPauseHook' THEN 'tap_create_hook'
            WHEN MAX_BY(action_name, client_timestamp::TIMESTAMP_NTZ(9)) = 'TapCreateSongPauseHook' THEN 'tap_create_song'
            WHEN MAX_BY(action_name, client_timestamp::TIMESTAMP_NTZ(9)) = 'TapTabPauseHook' THEN 'tap_tab'
            WHEN MAX_BY(action_name, client_timestamp::TIMESTAMP_NTZ(9)) = 'TapProfilePauseHook' THEN 'tap_profile'
            WHEN MAX_BY(action_name, client_timestamp::TIMESTAMP_NTZ(9)) = 'BackgroundAppPauseHook' THEN 'background_app'
            WHEN MAX_BY(action_name, client_timestamp::TIMESTAMP_NTZ(9)) = 'TapBackPauseHook' THEN 'tap_back'
            WHEN MAX_BY(action_name, client_timestamp::TIMESTAMP_NTZ(9)) = 'OpenDeeplinkPauseHook' THEN 'open_deeplink'
            WHEN MAX_BY(action_name, client_timestamp::TIMESTAMP_NTZ(9)) IN ('TapCoverPauseHook', 'TapExtendPauseHook') THEN 'tap_remix'
            ELSE MAX_BY(action_name, client_timestamp::TIMESTAMP_NTZ(9))
        END AS hook_end_reason
        , (
            -- If the user seeked at any point, then the playback is not continuous
            NOT BOOLOR_AGG(
                action_name IN (
                    'SeekHookPauseHook', 'SeekHookPlayHook'
                )
            )
            -- If the user paused *and* resumed/repeated, then the playback is not continuous
            AND NOT (
                BOOLOR_AGG(action_name = 'PauseHook')
                AND BOOLOR_AGG(action_name IN ('PlayHook', 'AutoRepeatPlaySameHook'))
            )
        ) AS is_play_continuous
        , COUNT_IF(action_name = 'AutoRepeatPlaySameHook') AS repeat_count
        , MIN_BY(action_name, client_timestamp::TIMESTAMP_NTZ(9)) IN (
            'PlayHook', 'ScrollDownPlayNewHook', 'ScrollUpPlayNewHook', 'PlayNewHook'
        ) 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', start_time,
                'end_time', end_time,
                'play_duration_seconds', play_duration_seconds
            )
        ) WITHIN GROUP (ORDER BY client_timestamp::TIMESTAMP_NTZ(9), server_timestamp::TIMESTAMP_NTZ(9)), 0, 50) AS hook_player_events

        -- Hook info
        , ANY_VALUE(is_user_hook_owner) AS is_user_hook_owner

        -- Context info
        , MIN_BY(previous_hook_session_id, client_timestamp::TIMESTAMP_NTZ(9)) AS previous_hook_session_id
        , MIN_BY(previous_clip_session_id, client_timestamp::TIMESTAMP_NTZ(9)) AS previous_clip_session_id
        , MIN_BY(context_type, client_timestamp::TIMESTAMP_NTZ(9)) AS context_type
        , MIN_BY(context_id, client_timestamp::TIMESTAMP_NTZ(9)) AS context_id
        , MODE(recommendation_item_id) AS recommendation_item_id
        , OBJECT_CONSTRUCT(
            'volume', MIN_BY(volume, client_timestamp::TIMESTAMP_NTZ(9))
        ) AS hook_player_metadata

        -- Device and context info (taking the first value in the session)
        , ANY_VALUE(session_id) AS session_id
        , 'ios' 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 {{ dedup_table_name }}
    WHERE 1=1
        AND user_uid IS NOT NULL
        AND hook_session_id IS NOT NULL
        AND hook_session_id != ''
    GROUP BY
        user_uid
        , hook_id
        , hook_session_id
)
