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

SELECT
    cph.clip_id
    , ANY_VALUE(cph.clip_creator_user_id) AS clip_creator_user_id
    , MIN(cph.first_play_start_ts_utc) AS first_play_start_ts_utc
    , MAX(cph.last_play_start_ts_utc) AS last_play_start_ts_utc

    -- Plays from creator
    , MIN(CASE WHEN cph.is_user_song_owner THEN cph.first_play_start_ts_utc END) AS creator_first_play_start_ts_utc
    , MAX(CASE WHEN cph.is_user_song_owner THEN cph.last_play_start_ts_utc END) AS creator_last_play_start_ts_utc
    , MAX(CASE WHEN cph.is_user_song_owner THEN cph.play_cnt_0s END) AS creator_play_cnt_0s
    , MAX(CASE WHEN cph.is_user_song_owner THEN cph.play_cnt_5s END) AS creator_play_cnt_5s
    , MAX(CASE WHEN cph.is_user_song_owner THEN cph.play_cnt_25pct END) AS creator_play_cnt_25pct
    , MAX(CASE WHEN cph.is_user_song_owner THEN cph.play_cnt_50pct END) AS creator_play_cnt_50pct
    , MAX(CASE WHEN cph.is_user_song_owner THEN cph.play_cnt_90pct END) AS creator_play_cnt_90pct
    , MAX(CASE WHEN cph.is_user_song_owner THEN cph.play_duration_seconds_0s END) AS creator_play_duration_seconds_0s
    , MAX(CASE WHEN cph.is_user_song_owner THEN cph.play_duration_seconds_5s END) AS creator_play_duration_seconds_5s
    , MAX(CASE WHEN cph.is_user_song_owner THEN cph.play_duration_seconds_25pct END) AS creator_play_duration_seconds_25pct
    , MAX(CASE WHEN cph.is_user_song_owner THEN cph.play_duration_seconds_50pct END) AS creator_play_duration_seconds_50pct
    , MAX(CASE WHEN cph.is_user_song_owner THEN cph.play_duration_seconds_90pct END) AS creator_play_duration_seconds_90pct

    -- Plays from non-creators
    , MIN(CASE WHEN NOT cph.is_user_song_owner THEN cph.first_play_start_ts_utc END) AS non_creator_first_play_start_ts_utc
    , MAX(CASE WHEN NOT cph.is_user_song_owner THEN cph.last_play_start_ts_utc END) AS non_creator_last_play_start_ts_utc
    , COUNT_IF(NOT cph.is_user_song_owner AND cph.play_cnt_0s > 0) AS n_non_creator_listeners_0s
    , COUNT_IF(NOT cph.is_user_song_owner AND cph.play_cnt_5s > 0) AS n_non_creator_listeners_5s
    , COUNT_IF(NOT cph.is_user_song_owner AND cph.play_cnt_25pct > 0) AS n_non_creator_listeners_25pct
    , COUNT_IF(NOT cph.is_user_song_owner AND cph.play_cnt_50pct > 0) AS n_non_creator_listeners_50pct
    , COUNT_IF(NOT cph.is_user_song_owner AND cph.play_cnt_90pct > 0) AS n_non_creator_listeners_90pct
    , SUM(CASE WHEN NOT cph.is_user_song_owner THEN cph.play_cnt_0s END) AS non_creator_play_cnt_0s
    , SUM(CASE WHEN NOT cph.is_user_song_owner THEN cph.play_cnt_5s END) AS non_creator_play_cnt_5s
    , SUM(CASE WHEN NOT cph.is_user_song_owner THEN cph.play_cnt_25pct END) AS non_creator_play_cnt_25pct
    , SUM(CASE WHEN NOT cph.is_user_song_owner THEN cph.play_cnt_50pct END) AS non_creator_play_cnt_50pct
    , SUM(CASE WHEN NOT cph.is_user_song_owner THEN cph.play_cnt_90pct END) AS non_creator_play_cnt_90pct
    , SUM(CASE WHEN NOT cph.is_user_song_owner THEN cph.play_duration_seconds_0s END) AS non_creator_play_duration_seconds_0s
    , SUM(CASE WHEN NOT cph.is_user_song_owner THEN cph.play_duration_seconds_5s END) AS non_creator_play_duration_seconds_5s
    , SUM(CASE WHEN NOT cph.is_user_song_owner THEN cph.play_duration_seconds_25pct END) AS non_creator_play_duration_seconds_25pct
    , SUM(CASE WHEN NOT cph.is_user_song_owner THEN cph.play_duration_seconds_50pct END) AS non_creator_play_duration_seconds_50pct
    , SUM(CASE WHEN NOT cph.is_user_song_owner THEN cph.play_duration_seconds_90pct END) AS non_creator_play_duration_seconds_90pct

    , MIN(cph.p_date) AS p_date
    , MIN(cph.p_hour) AS p_hour
FROM {{ ref('int_agg_user_clip_plays') }} cph
WHERE 1=1
{% if is_incremental() %}
    -- Only recompute for clips that have had plays since the last run
    AND cph.clip_id IN (
        SELECT DISTINCT clip_id
        FROM {{ ref('int_agg_user_clip_plays') }}
        WHERE last_play_start_ts_utc > (SELECT MAX(last_play_start_ts_utc) FROM {{ this }})
    )
{% endif %}
GROUP BY
    cph.clip_id
