INSERT INTO {{ stg_hook_session_backend_events_agg_table_name }} (
    user_id
    , user_uid
    , hook_id
    , hook_session_id
    , event_timestamp
    , event_type
    , event_count
    , platform
    , p_date
    , p_hour
)

WITH backend_events_raw AS (
    -- Backend events: likes, dislikes, comments, follows, playlist adds, reports

    -- Hook likes
    SELECT
        user_id
        , hook_id
        , NULL AS hook_session_id  -- Future: when logged directly
        , recommendation_item_id
        , session_id
        , timestamp::TIMESTAMP_NTZ AS event_timestamp
        , 'like' AS event_type
        , 1 AS event_count
        , platform
        , p_date
        , p_hour
    FROM suno_prod.{{ dbt_stg_schema }}.stg_backend_events__hook_like
    WHERE 1=1
        AND IFF(
            '{{ buffered_partition_start_date }}' = '{{ buffered_partition_end_date }}',
            -- Partition hours are within the same day
            p_date = '{{ buffered_partition_start_date }}' AND p_hour >= {{ buffered_partition_start_hour }} AND p_hour <  {{ buffered_partition_end_hour }},
            -- Partition hours are across multiple days
            (
                (p_date = '{{ buffered_partition_start_date }}' AND p_hour >= {{ buffered_partition_start_hour }})
                OR
                (p_date > '{{ buffered_partition_start_date }}' AND p_date < '{{ buffered_partition_end_date }}')
                OR
                (p_date = '{{ buffered_partition_end_date }}' AND p_hour <  {{ buffered_partition_end_hour }})
            )
        )

    UNION ALL

    -- Hook undo likes
    SELECT
        user_id
        , hook_id
        , NULL AS hook_session_id
        , recommendation_item_id
        , session_id
        , timestamp::TIMESTAMP_NTZ AS event_timestamp
        , 'undo_like' AS event_type
        , 1 AS event_count
        , platform
        , p_date
        , p_hour
    FROM suno_prod.{{ dbt_stg_schema }}.stg_backend_events__hook_undo_like
    WHERE 1=1
        AND IFF(
            '{{ buffered_partition_start_date }}' = '{{ buffered_partition_end_date }}',
            p_date = '{{ buffered_partition_start_date }}' AND p_hour >= {{ buffered_partition_start_hour }} AND p_hour <  {{ buffered_partition_end_hour }},
            (
                (p_date = '{{ buffered_partition_start_date }}' AND p_hour >= {{ buffered_partition_start_hour }})
                OR
                (p_date > '{{ buffered_partition_start_date }}' AND p_date < '{{ buffered_partition_end_date }}')
                OR
                (p_date = '{{ buffered_partition_end_date }}' AND p_hour <  {{ buffered_partition_end_hour }})
            )
        )

    UNION ALL

    -- Hook dislikes
    SELECT
        user_id
        , hook_id
        , NULL AS hook_session_id
        , recommendation_item_id
        , session_id
        , timestamp::TIMESTAMP_NTZ AS event_timestamp
        , 'dislike' AS event_type
        , 1 AS event_count
        , platform
        , p_date
        , p_hour
    FROM suno_prod.{{ dbt_stg_schema }}.stg_backend_events__hook_dislike
    WHERE 1=1
        AND IFF(
            '{{ buffered_partition_start_date }}' = '{{ buffered_partition_end_date }}',
            p_date = '{{ buffered_partition_start_date }}' AND p_hour >= {{ buffered_partition_start_hour }} AND p_hour <  {{ buffered_partition_end_hour }},
            (
                (p_date = '{{ buffered_partition_start_date }}' AND p_hour >= {{ buffered_partition_start_hour }})
                OR
                (p_date > '{{ buffered_partition_start_date }}' AND p_date < '{{ buffered_partition_end_date }}')
                OR
                (p_date = '{{ buffered_partition_end_date }}' AND p_hour <  {{ buffered_partition_end_hour }})
            )
        )

    UNION ALL

    -- Hook undo dislikes
    SELECT
        user_id
        , hook_id
        , NULL AS hook_session_id
        , recommendation_item_id
        , session_id
        , timestamp::TIMESTAMP_NTZ AS event_timestamp
        , 'undo_dislike' AS event_type
        , 1 AS event_count
        , platform
        , p_date
        , p_hour
    FROM suno_prod.{{ dbt_stg_schema }}.stg_backend_events__hook_undo_dislike
    WHERE 1=1
        AND IFF(
            '{{ buffered_partition_start_date }}' = '{{ buffered_partition_end_date }}',
            p_date = '{{ buffered_partition_start_date }}' AND p_hour >= {{ buffered_partition_start_hour }} AND p_hour <  {{ buffered_partition_end_hour }},
            (
                (p_date = '{{ buffered_partition_start_date }}' AND p_hour >= {{ buffered_partition_start_hour }})
                OR
                (p_date > '{{ buffered_partition_start_date }}' AND p_date < '{{ buffered_partition_end_date }}')
                OR
                (p_date = '{{ buffered_partition_end_date }}' AND p_hour <  {{ buffered_partition_end_hour }})
            )
        )

    UNION ALL

    -- Hook comments
    SELECT
        user_id
        , hook_id
        , NULL AS hook_session_id
        , recommendation_item_id
        , session_id
        , timestamp::TIMESTAMP_NTZ AS event_timestamp
        , 'comment' AS event_type
        , 1 AS event_count
        , platform
        , p_date
        , p_hour
    FROM suno_prod.{{ dbt_stg_schema }}.stg_backend_events__hook_comment
    WHERE 1=1
        AND IFF(
            '{{ buffered_partition_start_date }}' = '{{ buffered_partition_end_date }}',
            p_date = '{{ buffered_partition_start_date }}' AND p_hour >= {{ buffered_partition_start_hour }} AND p_hour <  {{ buffered_partition_end_hour }},
            (
                (p_date = '{{ buffered_partition_start_date }}' AND p_hour >= {{ buffered_partition_start_hour }})
                OR
                (p_date > '{{ buffered_partition_start_date }}' AND p_date < '{{ buffered_partition_end_date }}')
                OR
                (p_date = '{{ buffered_partition_end_date }}' AND p_hour <  {{ buffered_partition_end_hour }})
            )
        )

    UNION ALL

    -- Profile follows (hook-related)
    SELECT
        user_id
        , hook_id  -- This should be followed_user_hook_id in staging table
        , NULL AS hook_session_id
        , recommendation_item_id
        , session_id
        , timestamp::TIMESTAMP_NTZ AS event_timestamp
        , 'follow' AS event_type
        , 1 AS event_count
        , platform
        , p_date
        , p_hour
    FROM suno_prod.{{ dbt_stg_schema }}.stg_backend_events__profile_follow_hooks
    WHERE 1=1
        AND IFF(
            '{{ buffered_partition_start_date }}' = '{{ buffered_partition_end_date }}',
            p_date = '{{ buffered_partition_start_date }}' AND p_hour >= {{ buffered_partition_start_hour }} AND p_hour <  {{ buffered_partition_end_hour }},
            (
                (p_date = '{{ buffered_partition_start_date }}' AND p_hour >= {{ buffered_partition_start_hour }})
                OR
                (p_date > '{{ buffered_partition_start_date }}' AND p_date < '{{ buffered_partition_end_date }}')
                OR
                (p_date = '{{ buffered_partition_end_date }}' AND p_hour <  {{ buffered_partition_end_hour }})
            )
        )

    UNION ALL

    -- Profile unfollows (hook-related)
    SELECT
        user_id
        , hook_id  -- This should be unfollowed_user_hook_id in staging table
        , NULL AS hook_session_id
        , recommendation_item_id
        , session_id
        , timestamp::TIMESTAMP_NTZ AS event_timestamp
        , 'unfollow' AS event_type
        , 1 AS event_count
        , platform
        , p_date
        , p_hour
    FROM suno_prod.{{ dbt_stg_schema }}.stg_backend_events__profile_unfollow_hooks
    WHERE 1=1
        AND IFF(
            '{{ buffered_partition_start_date }}' = '{{ buffered_partition_end_date }}',
            p_date = '{{ buffered_partition_start_date }}' AND p_hour >= {{ buffered_partition_start_hour }} AND p_hour <  {{ buffered_partition_end_hour }},
            (
                (p_date = '{{ buffered_partition_start_date }}' AND p_hour >= {{ buffered_partition_start_hour }})
                OR
                (p_date > '{{ buffered_partition_start_date }}' AND p_date < '{{ buffered_partition_end_date }}')
                OR
                (p_date = '{{ buffered_partition_end_date }}' AND p_hour <  {{ buffered_partition_end_hour }})
            )
        )

    UNION ALL

    -- Playlist add clip (hook-related)
    SELECT
        user_id
        , hook_id
        , NULL AS hook_session_id
        , recommendation_item_id
        , session_id
        , timestamp::TIMESTAMP_NTZ AS event_timestamp
        , 'playlist_add' AS event_type
        , 1 AS event_count
        , platform
        , p_date
        , p_hour
    FROM suno_prod.{{ dbt_stg_schema }}.stg_backend_events__playlist_add_clip_hooks
    WHERE 1=1
        AND IFF(
            '{{ buffered_partition_start_date }}' = '{{ buffered_partition_end_date }}',
            p_date = '{{ buffered_partition_start_date }}' AND p_hour >= {{ buffered_partition_start_hour }} AND p_hour <  {{ buffered_partition_end_hour }},
            (
                (p_date = '{{ buffered_partition_start_date }}' AND p_hour >= {{ buffered_partition_start_hour }})
                OR
                (p_date > '{{ buffered_partition_start_date }}' AND p_date < '{{ buffered_partition_end_date }}')
                OR
                (p_date = '{{ buffered_partition_end_date }}' AND p_hour <  {{ buffered_partition_end_hour }})
            )
        )

    UNION ALL

    -- Hook reports
    SELECT
        user_id
        , hook_id
        , NULL AS hook_session_id
        , recommendation_item_id
        , session_id
        , timestamp::TIMESTAMP_NTZ AS event_timestamp
        , 'report' AS event_type
        , 1 AS event_count
        , platform
        , p_date
        , p_hour
    FROM suno_prod.{{ dbt_stg_schema }}.stg_backend_events__hook_report
    WHERE 1=1
        AND IFF(
            '{{ buffered_partition_start_date }}' = '{{ buffered_partition_end_date }}',
            p_date = '{{ buffered_partition_start_date }}' AND p_hour >= {{ buffered_partition_start_hour }} AND p_hour <  {{ buffered_partition_end_hour }},
            (
                (p_date = '{{ buffered_partition_start_date }}' AND p_hour >= {{ buffered_partition_start_hour }})
                OR
                (p_date > '{{ buffered_partition_start_date }}' AND p_date < '{{ buffered_partition_end_date }}')
                OR
                (p_date = '{{ buffered_partition_end_date }}' AND p_hour <  {{ buffered_partition_end_hour }})
            )
        )

    UNION ALL

    -- Hook hide creator
    SELECT
        user_id
        , hook_id
        , NULL AS hook_session_id
        , recommendation_item_id
        , session_id
        , timestamp::TIMESTAMP_NTZ AS event_timestamp
        , 'hide_creator' AS event_type
        , 1 AS event_count
        , platform
        , p_date
        , p_hour
    FROM suno_prod.{{ dbt_stg_schema }}.stg_backend_events__hook_hide_creator
    WHERE 1=1
        AND IFF(
            '{{ buffered_partition_start_date }}' = '{{ buffered_partition_end_date }}',
            p_date = '{{ buffered_partition_start_date }}' AND p_hour >= {{ buffered_partition_start_hour }} AND p_hour <  {{ buffered_partition_end_hour }},
            (
                (p_date = '{{ buffered_partition_start_date }}' AND p_hour >= {{ buffered_partition_start_hour }})
                OR
                (p_date > '{{ buffered_partition_start_date }}' AND p_date < '{{ buffered_partition_end_date }}')
                OR
                (p_date = '{{ buffered_partition_end_date }}' AND p_hour <  {{ buffered_partition_end_hour }})
            )
        )

    UNION ALL

    -- Hook share
    SELECT
        user_id
        , hook_id
        , NULL AS hook_session_id
        , recommendation_item_id
        , session_id
        , timestamp::TIMESTAMP_NTZ AS event_timestamp
        , 'share' AS event_type
        , 1 AS event_count
        , platform
        , p_date
        , p_hour
    FROM suno_prod.{{ dbt_stg_schema }}.stg_backend_events__hook_share
    WHERE 1=1
        AND IFF(
            '{{ buffered_partition_start_date }}' = '{{ buffered_partition_end_date }}',
            p_date = '{{ buffered_partition_start_date }}' AND p_hour >= {{ buffered_partition_start_hour }} AND p_hour <  {{ buffered_partition_end_hour }},
            (
                (p_date = '{{ buffered_partition_start_date }}' AND p_hour >= {{ buffered_partition_start_hour }})
                OR
                (p_date > '{{ buffered_partition_start_date }}' AND p_date < '{{ buffered_partition_end_date }}')
                OR
                (p_date = '{{ buffered_partition_end_date }}' AND p_hour <  {{ buffered_partition_end_hour }})
            )
        )
),

-- Backend events with temporal join to get correct hook_session_id
backend_events_final AS (
    SELECT
        ber.user_id
        , hp.user_uid
        , ber.hook_id
        -- Future-proof: use direct hook_session_id if available, else temporal join
        , CASE
            WHEN ber.hook_session_id IS NOT NULL
            THEN ber.hook_session_id
            ELSE hp.hook_session_id
        END AS hook_session_id
        , ber.event_timestamp
        , ber.event_type
        , ber.event_count
        , ber.platform
        , ber.p_date
        , ber.p_hour
    FROM backend_events_raw ber
    LEFT JOIN {{ fact_hook_play_table_name }} hp ON (
        -- Join on recommendation_item_id (much cleaner than temporal join)
        ber.recommendation_item_id = hp.recommendation_item_id
        AND ber.user_id = hp.user_id
        AND ber.hook_id = hp.hook_id
        -- Only do recommendation_item_id join when direct hook_session_id is not available
        AND ber.hook_session_id IS NULL
        -- Filter to partition window to avoid full table scan
        AND IFF(
            '{{ buffered_partition_start_date }}' = '{{ buffered_partition_end_date }}',
            hp.p_date = '{{ buffered_partition_start_date }}' AND hp.p_hour >= {{ buffered_partition_start_hour }} AND hp.p_hour <  {{ buffered_partition_end_hour }},
            (
                (hp.p_date = '{{ buffered_partition_start_date }}' AND hp.p_hour >= {{ buffered_partition_start_hour }})
                OR
                (hp.p_date > '{{ buffered_partition_start_date }}' AND hp.p_date < '{{ buffered_partition_end_date }}')
                OR
                (hp.p_date = '{{ buffered_partition_end_date }}' AND hp.p_hour <  {{ buffered_partition_end_hour }})
            )
        )
    )
    -- Only include events that have either direct hook_session_id or successful recommendation_item_id join
    WHERE (ber.hook_session_id IS NOT NULL OR hp.hook_session_id IS NOT NULL)
    -- TODO: Remove recommendation_item_id join logic once hook_session_id is logged directly in backend events
    -- Use ROW_NUMBER to handle multiple hook sessions per recommendation_item_id (pick first temporally)
    QUALIFY CASE
        WHEN ber.hook_session_id IS NOT NULL THEN 1
        ELSE ROW_NUMBER() OVER (
            PARTITION BY ber.recommendation_item_id, ber.user_id, ber.hook_id
            ORDER BY hp.play_start_ts_utc
        )
    END = 1
)

SELECT
    user_id
    , user_uid
    , hook_id
    , hook_session_id
    , event_timestamp
    , event_type
    , event_count
    , platform
    , p_date
    , p_hour
FROM backend_events_final
WHERE IFF(
    '{{ partition_start_date }}' = '{{ partition_end_date }}',
    -- Target partition hours are within the same day
    p_date = '{{ partition_start_date }}' AND p_hour >= {{ partition_start_hour }} AND p_hour < {{ partition_end_hour }},
    -- Target partition hours are across multiple days
    (
        (p_date = '{{ partition_start_date }}' AND p_hour >= {{ partition_start_hour }})
        OR
        (p_date > '{{ partition_start_date }}' AND p_date < '{{ partition_end_date }}')
        OR
        (p_date = '{{ partition_end_date }}' AND p_hour < {{ partition_end_hour }})
    )
);
