{%- set incremental_predicate = partition_filter_daily('DBT_INTERNAL_DEST.p_date') -%}
{{
    config(
        materialized='incremental',
        incremental_strategy='delete+insert',
        incremental_predicates=[incremental_predicate],
        unique_key=['user_id', 'user_uid', 'platform', 'p_date'],
        cluster_by=['p_date'],
        snowflake_warehouse='FACT_PLAY_BACKFILL_SMALL',
    )
}}

WITH historical_activity_aggregated AS (
    SELECT
        user_id
        , user_uid
        , subscription_tier
        , platform
        , clip_total_count
        , gen_clip_total_count
        , play_count
        , play_duration_seconds
        , unique_play_duration_seconds
        , play_breakdown
        , hook_create_count
        , hook_play_count
        , hook_play_duration_seconds
        , hook_unique_play_duration_seconds
        , hook_play_breakdown
        , is_high_intent_listening
        , high_intent_listening_breakdown
        , is_active
        , engagement_flags
        , total_following_cnt
        , total_follower_cnt
        , net_likes_received_count

        -- Previous day activity using LAG functions
        , LAG(is_active, 1) OVER (PARTITION BY user_id, platform ORDER BY p_date) AS is_prev_day_active
        , LAG(engagement_flags:clip_create::BOOLEAN, 1) OVER (PARTITION BY user_id, platform ORDER BY p_date) AS is_prev_day_clip_create
        , LAG(engagement_flags:clip_listen::BOOLEAN, 1) OVER (PARTITION BY user_id, platform ORDER BY p_date) AS is_prev_day_clip_listen
        , LAG(engagement_flags:clip_listen_own::BOOLEAN, 1) OVER (PARTITION BY user_id, platform ORDER BY p_date) AS is_prev_day_clip_listen_own
        , LAG(engagement_flags:clip_listen_others::BOOLEAN, 1) OVER (PARTITION BY user_id, platform ORDER BY p_date) AS is_prev_day_clip_listen_others
        , LAG(engagement_flags:hook_create::BOOLEAN, 1) OVER (PARTITION BY user_id, platform ORDER BY p_date) AS is_prev_day_hook_create
        , LAG(engagement_flags:hook_listen::BOOLEAN, 1) OVER (PARTITION BY user_id, platform ORDER BY p_date) AS is_prev_day_hook_listen
        , LAG(engagement_flags:hook_listen_own::BOOLEAN, 1) OVER (PARTITION BY user_id, platform ORDER BY p_date) AS is_prev_day_hook_listen_own
        , LAG(engagement_flags:hook_listen_others::BOOLEAN, 1) OVER (PARTITION BY user_id, platform ORDER BY p_date) AS is_prev_day_hook_listen_others

        -- L7D activity using window frames (including current day)
        , MAX(is_active) OVER (PARTITION BY user_id, platform ORDER BY p_date ROWS BETWEEN 6 PRECEDING AND CURRENT ROW) AS is_7d_active
        , MAX(engagement_flags:clip_create::BOOLEAN) OVER (PARTITION BY user_id, platform ORDER BY p_date ROWS BETWEEN 6 PRECEDING AND CURRENT ROW) AS is_7d_clip_create
        , MAX(engagement_flags:clip_listen::BOOLEAN) OVER (PARTITION BY user_id, platform ORDER BY p_date ROWS BETWEEN 6 PRECEDING AND CURRENT ROW) AS is_7d_clip_listen
        , MAX(engagement_flags:clip_listen_own::BOOLEAN) OVER (PARTITION BY user_id, platform ORDER BY p_date ROWS BETWEEN 6 PRECEDING AND CURRENT ROW) AS is_7d_clip_listen_own
        , MAX(engagement_flags:clip_listen_others::BOOLEAN) OVER (PARTITION BY user_id, platform ORDER BY p_date ROWS BETWEEN 6 PRECEDING AND CURRENT ROW) AS is_7d_clip_listen_others
        , MAX(engagement_flags:hook_create::BOOLEAN) OVER (PARTITION BY user_id, platform ORDER BY p_date ROWS BETWEEN 6 PRECEDING AND CURRENT ROW) AS is_7d_hook_create
        , MAX(engagement_flags:hook_listen::BOOLEAN) OVER (PARTITION BY user_id, platform ORDER BY p_date ROWS BETWEEN 6 PRECEDING AND CURRENT ROW) AS is_7d_hook_listen
        , MAX(engagement_flags:hook_listen_own::BOOLEAN) OVER (PARTITION BY user_id, platform ORDER BY p_date ROWS BETWEEN 6 PRECEDING AND CURRENT ROW) AS is_7d_hook_listen_own
        , MAX(engagement_flags:hook_listen_others::BOOLEAN) OVER (PARTITION BY user_id, platform ORDER BY p_date ROWS BETWEEN 6 PRECEDING AND CURRENT ROW) AS is_7d_hook_listen_others

        -- Previous 7 days activity using window frames (excluding current day)
        , MAX(is_active) OVER (PARTITION BY user_id, platform ORDER BY p_date ROWS BETWEEN 13 PRECEDING AND 7 PRECEDING) AS is_prev_7d_active
        , MAX(engagement_flags:clip_create::BOOLEAN) OVER (PARTITION BY user_id, platform ORDER BY p_date ROWS BETWEEN 13 PRECEDING AND 7 PRECEDING) AS is_prev_7d_clip_create
        , MAX(engagement_flags:clip_listen::BOOLEAN) OVER (PARTITION BY user_id, platform ORDER BY p_date ROWS BETWEEN 13 PRECEDING AND 7 PRECEDING) AS is_prev_7d_clip_listen
        , MAX(engagement_flags:clip_listen_own::BOOLEAN) OVER (PARTITION BY user_id, platform ORDER BY p_date ROWS BETWEEN 13 PRECEDING AND 7 PRECEDING) AS is_prev_7d_clip_listen_own
        , MAX(engagement_flags:clip_listen_others::BOOLEAN) OVER (PARTITION BY user_id, platform ORDER BY p_date ROWS BETWEEN 13 PRECEDING AND 7 PRECEDING) AS is_prev_7d_clip_listen_others
        , MAX(engagement_flags:hook_create::BOOLEAN) OVER (PARTITION BY user_id, platform ORDER BY p_date ROWS BETWEEN 13 PRECEDING AND 7 PRECEDING) AS is_prev_7d_hook_create
        , MAX(engagement_flags:hook_listen::BOOLEAN) OVER (PARTITION BY user_id, platform ORDER BY p_date ROWS BETWEEN 13 PRECEDING AND 7 PRECEDING) AS is_prev_7d_hook_listen
        , MAX(engagement_flags:hook_listen_own::BOOLEAN) OVER (PARTITION BY user_id, platform ORDER BY p_date ROWS BETWEEN 13 PRECEDING AND 7 PRECEDING) AS is_prev_7d_hook_listen_own
        , MAX(engagement_flags:hook_listen_others::BOOLEAN) OVER (PARTITION BY user_id, platform ORDER BY p_date ROWS BETWEEN 13 PRECEDING AND 7 PRECEDING) AS is_prev_7d_hook_listen_others

        -- L28D activity using window frames (including current day)
        , MAX(is_active) OVER (PARTITION BY user_id, platform ORDER BY p_date ROWS BETWEEN 27 PRECEDING AND CURRENT ROW) AS is_28d_active
        , MAX(engagement_flags:clip_create::BOOLEAN) OVER (PARTITION BY user_id, platform ORDER BY p_date ROWS BETWEEN 27 PRECEDING AND CURRENT ROW) AS is_28d_clip_create
        , MAX(engagement_flags:clip_listen::BOOLEAN) OVER (PARTITION BY user_id, platform ORDER BY p_date ROWS BETWEEN 27 PRECEDING AND CURRENT ROW) AS is_28d_clip_listen
        , MAX(engagement_flags:clip_listen_own::BOOLEAN) OVER (PARTITION BY user_id, platform ORDER BY p_date ROWS BETWEEN 27 PRECEDING AND CURRENT ROW) AS is_28d_clip_listen_own
        , MAX(engagement_flags:clip_listen_others::BOOLEAN) OVER (PARTITION BY user_id, platform ORDER BY p_date ROWS BETWEEN 27 PRECEDING AND CURRENT ROW) AS is_28d_clip_listen_others
        , MAX(engagement_flags:hook_create::BOOLEAN) OVER (PARTITION BY user_id, platform ORDER BY p_date ROWS BETWEEN 27 PRECEDING AND CURRENT ROW) AS is_28d_hook_create
        , MAX(engagement_flags:hook_listen::BOOLEAN) OVER (PARTITION BY user_id, platform ORDER BY p_date ROWS BETWEEN 27 PRECEDING AND CURRENT ROW) AS is_28d_hook_listen
        , MAX(engagement_flags:hook_listen_own::BOOLEAN) OVER (PARTITION BY user_id, platform ORDER BY p_date ROWS BETWEEN 27 PRECEDING AND CURRENT ROW) AS is_28d_hook_listen_own
        , MAX(engagement_flags:hook_listen_others::BOOLEAN) OVER (PARTITION BY user_id, platform ORDER BY p_date ROWS BETWEEN 27 PRECEDING AND CURRENT ROW) AS is_28d_hook_listen_others

        , p_date
        , days_since_user_joined
    FROM
        {{ ref('int_agg_user_platform_daily_spine_base') }} base
    WHERE 1=1
        AND p_date >= DATEADD(day, -27, '{{ var('partition_start_date', '1900-01-01') }}')
        AND p_date < '{{ var('partition_end_date', '9999-12-31') }}'
)

SELECT
    user_id
    , user_uid
    , subscription_tier
    , platform
    , clip_total_count
    , gen_clip_total_count
    , play_count
    , play_duration_seconds
    , unique_play_duration_seconds
    , play_breakdown
    , hook_create_count
    , hook_play_count
    , hook_play_duration_seconds
    , hook_unique_play_duration_seconds
    , hook_play_breakdown
    , is_high_intent_listening
    , high_intent_listening_breakdown
    , is_active
    , engagement_flags
    , total_following_cnt
    , total_follower_cnt
    , net_likes_received_count

    -- Build historical activity object using calculated window functions
    , OBJECT_CONSTRUCT(
        'prev_day', OBJECT_CONSTRUCT(
            'is_active', is_prev_day_active
            , 'is_clip_create', is_prev_day_clip_create
            , 'is_clip_listen', is_prev_day_clip_listen
            , 'is_clip_listen_own', is_prev_day_clip_listen_own
            , 'is_clip_listen_others', is_prev_day_clip_listen_others
            , 'is_hook_create', is_prev_day_hook_create
            , 'is_hook_listen', is_prev_day_hook_listen
            , 'is_hook_listen_own', is_prev_day_hook_listen_own
            , 'is_hook_listen_others', is_prev_day_hook_listen_others
        )
        , '7d', OBJECT_CONSTRUCT(
            'is_active', is_7d_active
            , 'is_clip_create', is_7d_clip_create
            , 'is_clip_listen', is_7d_clip_listen
            , 'is_clip_listen_own', is_7d_clip_listen_own
            , 'is_clip_listen_others', is_7d_clip_listen_others
            , 'is_hook_create', is_7d_hook_create
            , 'is_hook_listen', is_7d_hook_listen
            , 'is_hook_listen_own', is_7d_hook_listen_own
            , 'is_hook_listen_others', is_7d_hook_listen_others
        )
        , 'prev_7d', OBJECT_CONSTRUCT(
            'is_active', is_prev_7d_active
            , 'is_clip_create', is_prev_7d_clip_create
            , 'is_clip_listen', is_prev_7d_clip_listen
            , 'is_clip_listen_own', is_prev_7d_clip_listen_own
            , 'is_clip_listen_others', is_prev_7d_clip_listen_others
            , 'is_hook_create', is_prev_7d_hook_create
            , 'is_hook_listen', is_prev_7d_hook_listen
            , 'is_hook_listen_own', is_prev_7d_hook_listen_own
            , 'is_hook_listen_others', is_prev_7d_hook_listen_others
        )
        , '28d', OBJECT_CONSTRUCT(
            'is_active', is_28d_active
            , 'is_clip_create', is_28d_clip_create
            , 'is_clip_listen', is_28d_clip_listen
            , 'is_clip_listen_own', is_28d_clip_listen_own
            , 'is_clip_listen_others', is_28d_clip_listen_others
            , 'is_hook_create', is_28d_hook_create
            , 'is_hook_listen', is_28d_hook_listen
            , 'is_hook_listen_own', is_28d_hook_listen_own
            , 'is_hook_listen_others', is_28d_hook_listen_others
        )
    ) AS historical_activity

    , p_date
    , days_since_user_joined

FROM historical_activity_aggregated
WHERE 1=1
    AND {{ partition_filter_daily('p_date') }}
