{{
    config(
        materialized = 'incremental'
        , incremental_strategy = 'delete+insert'
        , unique_key = ['clip_id']
        , cluster_by = ['p_date', 'p_hour']
        , snowflake_warehouse = 'DBT_DEV_LARGE'
        , tags = ['exclude_from_dagster=true']
    )
}}

WITH

{% if is_incremental() %}
clips_with_new_activity AS (
    SELECT DISTINCT clip_id FROM {{ ref('int_clip_play_metrics') }}
    WHERE last_play_start_ts_utc > (SELECT MAX(last_engagement_timestamp) FROM {{ this }})
    UNION
    SELECT DISTINCT clip_id FROM {{ ref('int_user_clip_like_historical') }}
    WHERE last_action_timestamp > (SELECT MAX(last_engagement_timestamp) FROM {{ this }})
    UNION
    SELECT DISTINCT clip_id FROM {{ ref('int_clip_pin_metrics') }}
    WHERE last_action_timestamp > (SELECT MAX(last_engagement_timestamp) FROM {{ this }})
    UNION
    SELECT DISTINCT clip_id FROM {{ ref('int_clip_share_metrics') }}
    WHERE last_share_timestamp > (SELECT MAX(last_engagement_timestamp) FROM {{ this }})
    UNION
    SELECT DISTINCT clip_id FROM {{ ref('int_clip_playlist_metrics') }}
    WHERE last_playlist_add_timestamp > (SELECT MAX(last_engagement_timestamp) FROM {{ this }})
),
{% endif %}

engagement_features AS (
    SELECT
        cpm.clip_id
        , ANY_VALUE(cpm.clip_creator_user_id) AS clip_creator_user_id
        , ANY_VALUE(cpm.first_play_start_ts_utc) AS first_play_start_ts_utc
        , ANY_VALUE(cpm.last_play_start_ts_utc) AS last_play_start_ts_utc
        , ANY_VALUE(cpm.creator_first_play_start_ts_utc) AS creator_first_play_start_ts_utc
        , ANY_VALUE(cpm.creator_last_play_start_ts_utc) AS creator_last_play_start_ts_utc
        , ANY_VALUE(cpm.creator_play_cnt_0s) AS creator_play_cnt_0s
        , ANY_VALUE(cpm.creator_play_cnt_5s) AS creator_play_cnt_5s
        , ANY_VALUE(cpm.creator_play_cnt_25pct) AS creator_play_cnt_25pct
        , ANY_VALUE(cpm.creator_play_cnt_50pct) AS creator_play_cnt_50pct
        , ANY_VALUE(cpm.creator_play_cnt_90pct) AS creator_play_cnt_90pct
        , ANY_VALUE(cpm.creator_play_duration_seconds_0s) AS creator_play_duration_seconds_0s
        , ANY_VALUE(cpm.creator_play_duration_seconds_5s) AS creator_play_duration_seconds_5s
        , ANY_VALUE(cpm.creator_play_duration_seconds_25pct) AS creator_play_duration_seconds_25pct
        , ANY_VALUE(cpm.creator_play_duration_seconds_50pct) AS creator_play_duration_seconds_50pct
        , ANY_VALUE(cpm.creator_play_duration_seconds_90pct) AS creator_play_duration_seconds_90pct
        , ANY_VALUE(cpm.non_creator_first_play_start_ts_utc) AS non_creator_first_play_start_ts_utc
        , ANY_VALUE(cpm.non_creator_last_play_start_ts_utc) AS non_creator_last_play_start_ts_utc
        , ANY_VALUE(cpm.n_non_creator_listeners_0s) AS n_non_creator_listeners_0s
        , ANY_VALUE(cpm.n_non_creator_listeners_5s) AS n_non_creator_listeners_5s
        , ANY_VALUE(cpm.n_non_creator_listeners_25pct) AS n_non_creator_listeners_25pct
        , ANY_VALUE(cpm.n_non_creator_listeners_50pct) AS n_non_creator_listeners_50pct
        , ANY_VALUE(cpm.n_non_creator_listeners_90pct) AS n_non_creator_listeners_90pct
        , ANY_VALUE(cpm.non_creator_play_cnt_0s) AS non_creator_play_cnt_0s
        , ANY_VALUE(cpm.non_creator_play_cnt_5s) AS non_creator_play_cnt_5s
        , ANY_VALUE(cpm.non_creator_play_cnt_25pct) AS non_creator_play_cnt_25pct
        , ANY_VALUE(cpm.non_creator_play_cnt_50pct) AS non_creator_play_cnt_50pct
        , ANY_VALUE(cpm.non_creator_play_cnt_90pct) AS non_creator_play_cnt_90pct
        , ANY_VALUE(cpm.non_creator_play_duration_seconds_0s) AS non_creator_play_duration_seconds_0s
        , ANY_VALUE(cpm.non_creator_play_duration_seconds_5s) AS non_creator_play_duration_seconds_5s
        , ANY_VALUE(cpm.non_creator_play_duration_seconds_25pct) AS non_creator_play_duration_seconds_25pct
        , ANY_VALUE(cpm.non_creator_play_duration_seconds_50pct) AS non_creator_play_duration_seconds_50pct
        , ANY_VALUE(cpm.non_creator_play_duration_seconds_90pct) AS non_creator_play_duration_seconds_90pct

        -- Engagement from creator
        , COALESCE(COUNT_IF(cph.is_user_song_owner AND iclh.net_action_value > 0), 0) AS n_creator_likes
        , COALESCE(COUNT_IF(cph.is_user_song_owner AND iclh.net_action_value > 0 AND cph.play_cnt_5s > 0), 0) AS n_creator_verified_likes
        , MAX(CASE WHEN cph.is_user_song_owner THEN iclh.has_undo_like END) AS creator_has_undo_like
        , MAX(CASE WHEN cph.is_user_song_owner THEN icp.net_action_value END) AS creator_net_pin_clip
        , MAX(CASE WHEN cph.is_user_song_owner THEN icp.has_unpin END) AS creator_has_unpin
        , ANY_VALUE(cs.n_creator_shares) AS n_creator_shares
        , ANY_VALUE(cplm.did_owner_add_to_playlist) AS did_owner_add_to_playlist
        , ANY_VALUE(cplm.n_playlists_added_by_owner) AS n_playlists_added_by_owner

        -- Engagement from non-creator
        , COALESCE(COUNT_IF(NOT cph.is_user_song_owner AND iclh.net_action_value > 0), 0) AS n_non_creator_likes
        , COALESCE(COUNT_IF(NOT cph.is_user_song_owner AND iclh.net_action_value > 0 AND cph.play_cnt_5s > 0), 0) AS n_non_creator_verified_likes
        , ANY_VALUE(cs.n_non_creator_shares) AS n_non_creator_shares
        , ANY_VALUE(cs.n_unique_non_creator_users_shared) AS n_unique_non_creator_users_shared
        , ANY_VALUE(cplm.n_non_owner_added_to_playlist) AS n_non_creator_added_to_playlist
        , ANY_VALUE(cplm.n_playlists_added_by_non_owner) AS n_playlists_added_by_non_owner

        , ANY_VALUE(cpm.p_date) AS p_date
        , ANY_VALUE(cpm.p_hour) AS p_hour

        -- Timestamps for incremental logic
        , MAX(iclh.last_action_timestamp) AS last_like_ts
        , MAX(icp.last_action_timestamp) AS last_pin_ts
        , ANY_VALUE(cs.last_share_timestamp) AS last_share_ts
        , ANY_VALUE(cplm.last_playlist_add_timestamp) AS last_playlist_add_ts

    FROM
        {{ ref('int_clip_play_metrics') }} cpm
    LEFT JOIN
        {{ ref('int_agg_user_clip_plays') }} cph
        ON cpm.clip_id = cph.clip_id
    LEFT JOIN
        {{ ref('int_user_clip_like_historical') }} iclh
        ON cph.clip_id = iclh.clip_id
        AND cph.user_id = iclh.user_id
    LEFT JOIN
        {{ ref('int_clip_pin_metrics')}} icp
        ON cph.clip_id = icp.clip_id
        AND cph.user_id = icp.user_id
    LEFT JOIN
        {{ ref('int_clip_share_metrics') }} cs
        ON cpm.clip_id = cs.clip_id
    LEFT JOIN
        {{ ref('int_clip_playlist_metrics') }} cplm
        ON cpm.clip_id = cplm.clip_id
    {% if is_incremental() %}
    WHERE cpm.clip_id IN (SELECT clip_id FROM clips_with_new_activity)
    {% endif %}
    GROUP BY
        cpm.clip_id
)

SELECT
    *
    , GREATEST(
        COALESCE(last_play_start_ts_utc, '1900-01-01'),
        COALESCE(last_share_ts, '1900-01-01'),
        COALESCE(last_playlist_add_ts, '1900-01-01'),
        COALESCE(last_like_ts, '1900-01-01'),
        COALESCE(last_pin_ts, '1900-01-01')
    ) AS last_engagement_timestamp
FROM engagement_features
