CREATE OR REPLACE TEMPORARY TABLE {{ sessions_table_name }} AS (
    SELECT
        -- Identifiers
        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
        , 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)) = 'ForegroundPlayHook' THEN 'foreground'
            WHEN MIN_BY(action_name, client_timestamp::TIMESTAMP_NTZ(9)) = 'SeekHookPlayHook' THEN 'seek'
            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)) IN ('ClosePauseHook', 'HookPauseClose') THEN 'close_hook'
            WHEN MAX_BY(action_name, client_timestamp::TIMESTAMP_NTZ(9)) = 'BackgroundPauseHook' THEN 'background'
            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)) = 'AutoRepeatPauseHook' THEN 'auto_repeat'
            WHEN MAX_BY(action_name, client_timestamp::TIMESTAMP_NTZ(9)) = 'SeekHookPauseHook' THEN 'seek'
            WHEN MAX_BY(action_name, client_timestamp::TIMESTAMP_NTZ(9)) = 'NavigatePauseHook' THEN 'navigate'
            WHEN MAX_BY(action_name, client_timestamp::TIMESTAMP_NTZ(9)) = 'PlayNewHookPauseHook' THEN 'play_new_hook'
            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', 'AutoRepeatPlayHook'))
            )
        ) AS is_play_continuous
        , COUNT_IF(action_name = 'AutoRepeatPauseHook') AS repeat_count
        , 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

        -- Misc
        , BOOLOR_AGG(is_user_hook_owner) AS is_user_hook_owner
        , ANY_VALUE(previous_hook_session_id) 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() AS hook_player_metadata

        -- Device and context info (taking the first value in the session)
        , MIN_BY(session_id, client_timestamp::TIMESTAMP_NTZ(9)) as session_id
        , 'web' as platform
        , MIN_BY(is_mobile, client_timestamp::TIMESTAMP_NTZ(9)) as is_mobile
        , MIN_BY(country, client_timestamp::TIMESTAMP_NTZ(9)) as country
        , MIN_BY(city, client_timestamp::TIMESTAMP_NTZ(9)) as city

        -- The partition for the session is defined by the *start* of the session
        , DATE(MIN(server_timestamp::TIMESTAMP_NTZ(9))) AS p_date
        , HOUR(MIN(server_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
)
