{{
    config(
        materialized = 'incremental'
        , incremental_strategy = 'merge'
        , unique_key = ['user_id', 'clip_id', 'is_user_song_owner']
        , cluster_by = ['p_date', 'p_hour']
        , snowflake_warehouse = 'DBT_DEV_LARGE'
        , tags = ['exclude_from_dagster=true']
    )
}}

WITH

-- Get all-time plays for clips that have had a play since the last run
all_plays AS (
    SELECT
        fp.user_id
        , fp.clip_id
        , dc.user_id AS clip_creator_user_id
        , fp.song_session_id
        , fp.is_user_song_owner
        , fp.play_start_ts_utc
        , fp.play_duration_seconds
        , fp.unique_play_percent
        , CASE
            WHEN LOWER(fp.play_context_type) LIKE '%song_recommendations%' THEN 'song_recommendations'
            WHEN LOWER(fp.play_context_type) LIKE '%featured_feed%' THEN 'featured_feed'
            WHEN LOWER(fp.play_context_type) LIKE '%feed%' THEN 'feed'
            WHEN LOWER(fp.play_context_type) LIKE '%studio%' THEN 'studio'
            WHEN LOWER(fp.play_context_type) LIKE '%song_referenced_clip%' THEN 'song_referenced_clip'
            WHEN LOWER(fp.play_context_type) LIKE '%shortcut%' THEN 'shortcut'
            WHEN LOWER(fp.play_context_type) LIKE '%playlist%' THEN 'playlist'
            WHEN LOWER(fp.play_context_type) LIKE '%workspace%' THEN 'workspace'
            WHEN LOWER(fp.play_context_type) LIKE '%search%' THEN 'search'
            WHEN LOWER(fp.play_context_type) LIKE '%profile%' THEN 'profile'
            WHEN LOWER(fp.play_context_type) LIKE '%contest%' THEN 'contest'
            WHEN LOWER(fp.play_context_type) LIKE '%history%' THEN 'history'
            WHEN LOWER(fp.play_context_type) LIKE '%radio%' THEN 'radio'
            WHEN LOWER(fp.play_context_type) LIKE '%persona%' THEN 'persona'
        END AS play_context_type_grouped
        , dc.p_date
        , dc.p_hour
    FROM
        {{ ref('stg_cleaned__fact_play') }} fp
    LEFT JOIN
        {{ ref('dim_clip') }} dc
        ON fp.clip_id = dc.clip_id
    WHERE 1=1
        AND fp.p_date >= '2025-04-01' -- TODO (@junseo): this is currently the earliest date of plays we have in fact_play
        {% if is_incremental() %}
        -- We need to reprocess all data where the last play was after our last run
        -- Include the max timestamp so that we reprocess song play sessions that have updated since last run (i.e. song sessions at hour boundaries)
        AND fp.play_start_ts_utc >= (
            SELECT MAX(last_play_start_ts_utc)
            FROM {{ this }}
        )
        {% endif %}
)

, new_plays_agg AS (
    SELECT
        user_id
        , clip_id
        , clip_creator_user_id
        , is_user_song_owner
        , MIN(play_start_ts_utc) AS first_play_start_ts_utc
        , MAX(play_start_ts_utc) AS last_play_start_ts_utc
        , COUNT(DISTINCT p_date) AS n_unique_days_listened
        , COUNT(*) AS play_cnt_0s
        , COUNT_IF(play_duration_seconds >= 5) AS play_cnt_5s
        , COUNT_IF(unique_play_percent >= 0.25) AS play_cnt_25pct
        , COUNT_IF(unique_play_percent >= 0.5) AS play_cnt_50pct
        , COUNT_IF(unique_play_percent >= 0.9) AS play_cnt_90pct
        , SUM(play_duration_seconds) AS play_duration_seconds_0s
        , SUM(IFF(play_duration_seconds >= 5, play_duration_seconds, 0)) AS play_duration_seconds_5s
        , SUM(IFF(unique_play_percent >= 0.25, play_duration_seconds, 0)) AS play_duration_seconds_25pct
        , SUM(IFF(unique_play_percent >= 0.5, play_duration_seconds, 0)) AS play_duration_seconds_50pct
        , SUM(IFF(unique_play_percent >= 0.9, play_duration_seconds, 0)) AS play_duration_seconds_90pct
        , ARRAY_DISTINCT(ARRAY_AGG(play_context_type_grouped)) AS unique_play_contexts
        , ARRAY_SIZE(ARRAY_DISTINCT(ARRAY_AGG(play_context_type_grouped))) AS n_unique_play_contexts
        , ANY_VALUE(p_date) AS p_date  -- Use clip creation date for partitioning
        , ANY_VALUE(p_hour) AS p_hour  -- Use clip creation hour for partitioning
    FROM
        all_plays
    GROUP BY 1, 2, 3, 4
)

{% if is_incremental() %}
, existing_plays_agg AS (
    SELECT
        user_id
        , clip_id
        , clip_creator_user_id
        , is_user_song_owner
        , first_play_start_ts_utc
        , last_play_start_ts_utc
        , n_unique_days_listened
        , play_cnt_0s
        , play_cnt_5s
        , play_cnt_25pct
        , play_cnt_50pct
        , play_cnt_90pct
        , play_duration_seconds_0s
        , play_duration_seconds_5s
        , play_duration_seconds_25pct
        , play_duration_seconds_50pct
        , play_duration_seconds_90pct
        , unique_play_contexts
        , n_unique_play_contexts
        , p_date
        , p_hour
    FROM
        {{ this }}
)

SELECT
    COALESCE(existing.user_id, new.user_id) AS user_id
    , COALESCE(existing.clip_id, new.clip_id) AS clip_id
    , COALESCE(existing.clip_creator_user_id, new.clip_creator_user_id) AS clip_creator_user_id
    , COALESCE(existing.is_user_song_owner, new.is_user_song_owner) AS is_user_song_owner
    , LEAST(
        COALESCE(existing.first_play_start_ts_utc, '1900-01-01'::timestamp_ntz),
        COALESCE(new.first_play_start_ts_utc, '1900-01-01'::timestamp_ntz)
    ) AS first_play_start_ts_utc
    , GREATEST(
        COALESCE(existing.last_play_start_ts_utc, '1900-01-01'::timestamp_ntz),
        COALESCE(new.last_play_start_ts_utc, '1900-01-01'::timestamp_ntz)
    ) AS last_play_start_ts_utc
    , COALESCE(existing.n_unique_days_listened, 0) + COALESCE(new.n_unique_days_listened, 0) AS n_unique_days_listened
    , COALESCE(existing.play_cnt_0s, 0) + COALESCE(new.play_cnt_0s, 0) AS play_cnt_0s
    , COALESCE(existing.play_cnt_5s, 0) + COALESCE(new.play_cnt_5s, 0) AS play_cnt_5s
    , COALESCE(existing.play_cnt_25pct, 0) + COALESCE(new.play_cnt_25pct, 0) AS play_cnt_25pct
    , COALESCE(existing.play_cnt_50pct, 0) + COALESCE(new.play_cnt_50pct, 0) AS play_cnt_50pct
    , COALESCE(existing.play_cnt_90pct, 0) + COALESCE(new.play_cnt_90pct, 0) AS play_cnt_90pct
    , COALESCE(existing.play_duration_seconds_0s, 0) + COALESCE(new.play_duration_seconds_0s, 0) AS play_duration_seconds_0s
    , COALESCE(existing.play_duration_seconds_5s, 0) + COALESCE(new.play_duration_seconds_5s, 0) AS play_duration_seconds_5s
    , COALESCE(existing.play_duration_seconds_25pct, 0) + COALESCE(new.play_duration_seconds_25pct, 0) AS play_duration_seconds_25pct
    , COALESCE(existing.play_duration_seconds_50pct, 0) + COALESCE(new.play_duration_seconds_50pct, 0) AS play_duration_seconds_50pct
    , COALESCE(existing.play_duration_seconds_90pct, 0) + COALESCE(new.play_duration_seconds_90pct, 0) AS play_duration_seconds_90pct
    , ARRAY_DISTINCT(ARRAY_CAT(existing.unique_play_contexts, new.unique_play_contexts)) AS unique_play_contexts
    , ARRAY_SIZE(ARRAY_DISTINCT(ARRAY_CAT(existing.unique_play_contexts, new.unique_play_contexts))) AS n_unique_play_contexts
    , COALESCE(existing.p_date, new.p_date) AS p_date
    , COALESCE(existing.p_hour, new.p_hour) AS p_hour
FROM
    existing_plays_agg existing
FULL OUTER JOIN new_plays_agg new
    ON existing.user_id = new.user_id
    AND existing.clip_id = new.clip_id
    AND existing.is_user_song_owner = new.is_user_song_owner
{% else %}
SELECT * FROM new_plays_agg
{% endif %}
