{{
  config(
    materialized = 'incremental',
    unique_key = 'clip_id',
    incremental_strategy = 'delete+insert',
    on_schema_change = 'append_new_columns',
    cluster_by = ['p_date', 'p_hour'],
    snowflake_warehouse = 'SUNO_PROD_LARGE'
  )
}}

-- This model replicates the CLIP_INITIAL_METRICS_HOURLY_UPSERT_PROC procedure logic
-- It processes clips created in a specific hour and collects initial interaction metrics
-- within a 120-minute window after clip creation. This replaces the Snowflake task/procedure
-- with dbt-dagster orchestration.
--
-- Key logic:
-- - Processes clips created in the partition hour (2 hours ago due to end_offset=-2)
-- - Collects metrics within 120-minute window after clip creation
-- - Aggregates play data from fact_play (owner plays only)
-- - Aggregates backend events from bk_nonrealtime_event (ClipLike, ClipShare, etc.)
-- - Calculates satisfaction metrics (indirect and direct)

{% set window_minutes = 120 %}
{% set partition_start_date = var('partition_start_date', '1900-01-01') %}
{% set partition_start_hour = var('partition_start_hour', 0) %}

-- Calculate timestamps and partition boundaries (matching procedure logic)
-- Procedure declares: start_ts, end_ts, start_p_date, start_p_hour, end_p_date, end_p_hour
-- We replicate this by creating SQL expressions that can be reused
{% set start_ts_expr = "TO_TIMESTAMP('" ~ partition_start_date ~ " ' || " ~ partition_start_hour ~ " || ':00:00')" %}
{% set end_ts_expr = "DATEADD('MINUTE', " ~ (window_minutes + 60) ~ ", " ~ start_ts_expr ~ ")" %}
{% set start_p_date_expr = "DATE(" ~ start_ts_expr ~ ")" %}
{% set start_p_hour_expr = "HOUR(" ~ start_ts_expr ~ ")" %}
{% set end_p_date_expr = "DATE(" ~ end_ts_expr ~ ")" %}
{% set end_p_hour_expr = "HOUR(" ~ end_ts_expr ~ ")" %}

WITH clips AS (
    SELECT
        dc.clip_id
        , dc.user_id
        , du.user_uid
        , dc.created_at::TIMESTAMP_NTZ AS clip_created_at
        , dc.model_name
        , dc.p_date
        , dc.p_hour
    FROM
        {{ ref('dim_clip') }} AS dc
    LEFT JOIN
        {{ source('snowflake', 'dim_user') }} AS du
        ON dc.user_id = du.user_id
    WHERE {{ partition_filter_hourly('dc.p_date', 'dc.p_hour') }}
-- Note: dim_clip view is built from int_dim_clip which already filters status='complete'
)

, clip_play_info AS (
    SELECT
        fp.clip_id
        , COALESCE(COUNT(*), 0) AS play_cnt
        , COALESCE(SUM(fp.play_duration_seconds), 0) AS play_duration_seconds
    FROM
        {{ source('snowflake', 'fact_play') }} AS fp
    INNER JOIN
        clips AS c
        ON fp.clip_id = c.clip_id
    WHERE
        -- Filter fact_play by partition range: from clip creation hour to window_minutes + 60 minutes later
        -- Replicate procedure logic: filter by date range and handle hour boundaries
        -- Uses variables: start_p_date, end_p_date, start_p_hour, end_p_hour (matching procedure)
        fp.p_date BETWEEN {{ start_p_date_expr }} AND {{ end_p_date_expr }}
        -- If time window spans two days, use special filtering logic on partition hours
        AND IFF(
            {{ end_p_date_expr }} > {{ start_p_date_expr }}
            , (
                fp.p_date = {{ start_p_date_expr }} AND fp.p_hour >= {{ start_p_hour_expr }}
            )
            OR (
                fp.p_date = {{ end_p_date_expr }} AND fp.p_hour BETWEEN 0 AND {{ end_p_hour_expr }}
            )
            , fp.p_hour BETWEEN {{ start_p_hour_expr }} AND {{ end_p_hour_expr }}
        )
        AND fp.play_start_ts_utc BETWEEN c.clip_created_at
        AND DATEADD('MINUTE', {{ window_minutes }}, c.clip_created_at)
        AND fp.is_user_song_owner
    GROUP BY
        1
)

, clip_backend_events AS (
    SELECT
        PARSE_JSON(bk.properties):clip_id::STRING AS clip_id
        , COALESCE(COUNT_IF(bk.name = 'ClipLike'), 0) AS n_like_events
        , COALESCE(COUNT_IF(bk.name = 'ClipUndoLike'), 0) AS n_undolike_events
        , COALESCE(COUNT_IF(bk.name = 'ClipDislike'), 0) AS n_dislike_events
        , COALESCE(COUNT_IF(bk.name = 'ClipUndoDislike'), 0) AS n_undodislike_events
        , COALESCE(COUNT_IF(bk.name = 'ClipShare'), 0) AS n_share_events
        , COALESCE(COUNT_IF(bk.name = 'ClipDownload'), 0) AS n_download_events
        , COALESCE(COUNT_IF(bk.name = 'ClipMakePublic'), 0) AS n_makepublic_events
        , COALESCE(COUNT_IF(bk.name = 'ClipMakePrivate'), 0) AS n_makeprivate_events
    FROM
        {{ source('backend_events', 'bk_nonrealtime_event') }} AS bk
    INNER JOIN
        clips AS c
        ON PARSE_JSON(bk.properties):clip_id::STRING = c.clip_id
    WHERE
        -- Filter bk_nonrealtime_event by partition range: from clip creation hour to window_minutes + 60 minutes later
        -- Same logic as clip_play_info (uses same variables)
        bk.p_date BETWEEN {{ start_p_date_expr }} AND {{ end_p_date_expr }}
        -- If time window spans two days, use special filtering logic on partition hours
        AND IFF(
            {{ end_p_date_expr }} > {{ start_p_date_expr }}
            , (
                bk.p_date = {{ start_p_date_expr }} AND bk.p_hour >= {{ start_p_hour_expr }}
            )
            OR (
                bk.p_date = {{ end_p_date_expr }} AND bk.p_hour BETWEEN 0 AND {{ end_p_hour_expr }}
            )
            , bk.p_hour BETWEEN {{ start_p_hour_expr }} AND {{ end_p_hour_expr }}
        )
        AND bk.timestamp BETWEEN c.clip_created_at AND DATEADD('MINUTE', {{ window_minutes }}, c.clip_created_at)
        AND bk.name IN (
            'ClipLike'
            , 'ClipUndoLike'
            , 'ClipDislike'
            , 'ClipUndoDislike'
            , 'ClipShare'
            , 'ClipDownload'
            , 'ClipMakePublic'
            , 'ClipMakePrivate'
        )
        AND PARSE_JSON(bk.properties):is_user_object_owner::BOOLEAN
    GROUP BY
        1
)

, tmp AS (
    SELECT
        c.clip_id
        , c.user_id
        , c.user_uid
        , c.clip_created_at
        , {{ window_minutes }} AS window_minutes
        , DATE_TRUNC('HOUR', c.clip_created_at) AS window_start_time
        , DATEADD('MINUTE', {{ window_minutes }}, DATE_TRUNC('HOUR', c.clip_created_at)) AS window_end_time
        , COALESCE(cpi.play_cnt, 0) AS play_cnt
        , COALESCE(cpi.play_duration_seconds, 0) AS play_duration_seconds
        , COALESCE(bk.n_like_events, 0) AS n_like_events
        , COALESCE(bk.n_undolike_events, 0) AS n_undolike_events
        , COALESCE(bk.n_like_events - bk.n_undolike_events, 0) AS net_like_events
        , COALESCE(bk.n_dislike_events, 0) AS n_dislike_events
        , COALESCE(bk.n_undodislike_events, 0) AS n_undodislike_events
        , COALESCE(bk.n_dislike_events - bk.n_undodislike_events, 0) AS net_dislike_events
        , COALESCE(bk.n_share_events, 0) AS n_share_events
        , COALESCE(bk.n_download_events, 0) AS n_download_events
        , COALESCE(bk.n_makepublic_events, 0) AS n_makepublic_events
        , COALESCE(bk.n_makeprivate_events, 0) AS n_makeprivate_events
        , COALESCE(bk.n_makepublic_events - bk.n_makeprivate_events, 0) AS net_public_toggle_events
        , COALESCE(
            bk.n_like_events - bk.n_undolike_events
            - (bk.n_dislike_events - bk.n_undodislike_events)
            + bk.n_share_events
            + bk.n_download_events
            + (bk.n_makepublic_events - bk.n_makeprivate_events)
            , 0
        ) AS net_positive_interactions
        , OBJECT_CONSTRUCT(
            'v1', CASE
                WHEN
                    COALESCE(cpi.play_cnt, 0) = 0
                    OR COALESCE(cpi.play_duration_seconds, 0) = 0
                    OR COALESCE(bk.n_dislike_events - bk.n_undodislike_events, 0) > 0
                    THEN false
                WHEN COALESCE(cpi.play_cnt, 0) >= 3 AND COALESCE(cpi.play_duration_seconds, 0) >= 60 THEN true
                ELSE false
            END
        ) AS is_satisfactory_indirect
        , OBJECT_CONSTRUCT(
            'v1', CASE
                WHEN
                    cpi.play_cnt = 0
                    OR cpi.play_duration_seconds = 0
                    OR COALESCE(bk.n_dislike_events - bk.n_undodislike_events, 0) > 0
                    THEN false
                WHEN (
                    (COALESCE(bk.n_like_events, 0) - COALESCE(bk.n_undolike_events, 0)) > 0
                    OR COALESCE(bk.n_share_events, 0) > 0
                    OR COALESCE(bk.n_download_events, 0) > 0
                    OR (COALESCE(bk.n_makepublic_events, 0) - COALESCE(bk.n_makeprivate_events, 0)) > 0
                ) THEN true
                ELSE false
            END
        ) AS is_satisfactory_direct
        , OBJECT_CONSTRUCT(
            'model_name', c.model_name
        ) AS metadata
        , c.p_date
        , c.p_hour
    FROM
        clips AS c
    LEFT JOIN
        clip_backend_events AS bk
        ON c.clip_id = bk.clip_id
    LEFT JOIN
        clip_play_info AS cpi
        ON c.clip_id = cpi.clip_id
)

SELECT
    clip_id
    , user_id
    , user_uid
    , clip_created_at
    , window_minutes
    , window_start_time
    , window_end_time
    , play_cnt
    , play_duration_seconds
    , n_like_events
    , n_undolike_events
    , net_like_events
    , n_dislike_events
    , n_undodislike_events
    , net_dislike_events
    , n_share_events
    , n_download_events
    , n_makepublic_events
    , n_makeprivate_events
    , net_public_toggle_events
    , net_positive_interactions
    , is_satisfactory_indirect
    , is_satisfactory_direct
    , metadata
    , p_date
    , p_hour
    , OBJECT_CONSTRUCT(
        'v1', is_satisfactory_indirect:v1 OR is_satisfactory_direct:v1
    ) AS is_satisfactory
FROM
    tmp
