{%- set incremental_predicate = partition_filter_hourly('DBT_INTERNAL_DEST.p_date', 'DBT_INTERNAL_DEST.p_hour') -%}

{{
    config(
        materialized='incremental',
        incremental_strategy='delete+insert',
        incremental_predicates=[incremental_predicate],
        unique_key=['user_uid', 'clip_id', 'p_date', 'p_hour', 'play_duration_threshold'],
        cluster_by=['p_date', 'p_hour']
    )
}}

SELECT
    COALESCE(MAX(fp.user_id), MAX(du.user_id)) AS user_id -- user_id is sometimes null in fact_play, so coalesce with dim_user
    , fp.user_uid
    , fp.clip_id
    , MAX(fp.is_user_song_owner) AS is_user_song_owner
    , CASE
        WHEN LOWER(fp.platform) = 'ipados' THEN 'ios'
        ELSE LOWER(fp.platform)
    END AS platform
    , threshold.play_duration_threshold
    , SUM(
        CASE
            WHEN fp.play_duration_seconds >= threshold.play_duration_threshold
            THEN fp.play_duration_seconds
        ELSE 0
        END
    ) AS play_duration_seconds
    , SUM(
        CASE
            WHEN fp.play_duration_seconds >= threshold.play_duration_threshold
            THEN fp.unique_play_duration_seconds
        ELSE 0
        END
    ) AS unique_play_duration_seconds
    , COUNT(
        CASE
            WHEN fp.play_duration_seconds >= threshold.play_duration_threshold
            THEN 1
        END
    ) AS play_count
    , OBJECT_CONSTRUCT(
        'v1', MAX(fp.is_high_intent_listening:v1::BOOLEAN)
        , 'v2', MAX(fp.is_high_intent_listening:v2::BOOLEAN)
    ) AS is_high_intent_listening
    , OBJECT_CONSTRUCT(
        'v1', OBJECT_CONSTRUCT(
            'play_duration_seconds', SUM(IFF(fp.is_high_intent_listening:v1::BOOLEAN, fp.play_duration_seconds, 0)),
            'play_count', COUNT(IFF(fp.is_high_intent_listening:v1::BOOLEAN, 1, NULL))
        ),
        'v2', OBJECT_CONSTRUCT(
            'play_duration_seconds', SUM(IFF(fp.is_high_intent_listening:v2::BOOLEAN, fp.play_duration_seconds, 0)),
            'play_count', COUNT(IFF(fp.is_high_intent_listening:v2::BOOLEAN, 1, NULL))
        )
    ) AS high_intent_listening_breakdown
    , fp.p_date
    , fp.p_hour
FROM
    {{ source('snowflake', 'fact_play_v2') }} fp
LEFT JOIN
    {{ source('snowflake', 'dim_user') }} du
    ON fp.user_uid = du.user_uid
CROSS JOIN
    (SELECT * FROM VALUES (0), (5) AS t(play_duration_threshold)) threshold
WHERE 1=1
    AND {{ partition_filter_hourly('fp.p_date', 'fp.p_hour') }}
GROUP BY
    2, 3, 5, 6, 12, 13
HAVING
    -- Exclude rows with no plays under the play duration threshold constraint
    -- e.g. for threshold = 5, exclude plays that are less than 5 seconds
    -- (otherwise they would be included with 0 play duration seconds and 0 play count)
    SUM(
        CASE
            WHEN fp.play_duration_seconds >= threshold.play_duration_threshold
            THEN fp.play_duration_seconds
        ELSE 0
        END
    ) > 0 -- Play duration seconds
    AND COUNT(
        CASE
            WHEN fp.play_duration_seconds >= threshold.play_duration_threshold
            THEN 1
        END
    ) > 0 -- Play count
