{{
  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 DIM_CLIP_INSERT_PROC procedure logic
-- It reads from S3 parquet files via the raw external table and transforms the metadata
-- to match the exact structure of the production dim_clip table
-- This replaces the Snowflake task/procedure with dbt-dagster orchestration
--
-- Additionally, this model merges lyrics (lyrics_at_gen and lyrics_final)
-- from the ddb_clip_meta_heavy table, replicating the DIM_CLIP_LYRICS_UPSERT_PROC procedure logic.

WITH raw_data AS (
    SELECT * FROM {{ ref('raw_s3_bots_generatedclip_insert') }}
    WHERE {{ partition_filter_hourly('P_DATE', 'P_HOUR') }}
)

, parsed_data AS (
    SELECT
        *
        , metadata_json AS metadata
    FROM raw_data
)

, base_clip_data AS (
    SELECT
        -- Primary identifiers (matching DIM_CLIP_INSERT_PROC)
        id AS clip_id
        , created_at::TIMESTAMP_TZ AS created_at
        , user_id
        , request_id
        , s3_id
        , status
        , batch_index AS display_index
        , title AS initial_title
        , time_used AS time_used_sec
        , prompt_text AS lyrics_at_gen_base
        , model_name
        , false AS is_deleted
        , false AS is_audio_watermark
        , null::VARCHAR AS clip_uid
        , metadata
        , Date(created_at::TIMESTAMP) AS p_date
        , Hour(created_at::TIMESTAMP) AS p_hour
        , Iff(metadata:source IS null, 'web', metadata:source) AS platform
        , metadata:type AS type
        , metadata:task AS task
        , metadata:priority AS priority
        , metadata:duration AS duration_sec
        , metadata:refund_credits AS is_refund_credits
        , metadata:is_bot AS is_bot
        , metadata:credit_cost AS credit_cost
        , metadata:gpt_description_prompt AS simple_prompt
        , metadata:prompt AS lyrics_final_base
        , Iff(metadata:gpt_description_prompt IS NOT null AND metadata:gpt_description_prompt != 'null', true, false)
            AS is_simple
        , metadata:make_instrumental AS is_make_instrumental
        , metadata:tags AS style_tags
        , metadata:negative_tags AS negative_style_tags
        , metadata:experiment AS request_models
        , metadata:param_experiment AS experiment_inf_param
        , metadata:experiment_version AS experiment_version
        , metadata:history AS history
        , metadata:concat_history AS concat_history
        , metadata:error_type AS error_type
        , metadata:error_message AS error_message
        , metadata:has_vocal AS upload_has_vocal
        , metadata:edit_session_id AS edit_session_id
        , metadata:edited_clip_id AS edited_clip_id
        , Object_construct(
            'stem'
            , metadata:stem_from_id
            , 'extend'
            , Object_construct(
                'continue_at', metadata:continue_at
                , 'continued_from_prompt', metadata:continued_from_prompt
                , 'continued_aligned_prompt', metadata:continued_aligned_prompt
            )
            , 'infill'
            , Object_construct(
                'infill_context_start_s', metadata:infill_context_start_s
                , 'infill_context_end_s', metadata:infill_context_end_s
                , 'infill_start_s', metadata:infill_start_s
                , 'infill_end_s', metadata:infill_end_s
                , 'include_future_s', metadata:include_future_s
            )
            , 'trim'
            , Object_construct(
                'crop_start_time', metadata:crop_start_time
                , 'crop_end_time', metadata:crop_end_time
            )
            , 'artist_consistency'
            , metadata:persona_id
            , 'covers'
            , metadata:cover_clip_id
            , 'video_upload'
            , Object_construct(
                'video_to_song_description', metadata:video_to_song_description
                , 'video_to_song_video_upload_id', metadata:video_to_song_video_upload_id
            )
        ) AS task_details
        , metadata:creation_source AS creation_source
        , metadata:augmented_tags AS augmented_tags
        , metadata:free_quota_category AS free_quota_category
        , metadata:lang AS lyrics_language
        , metadata:lyrics_model AS simple_lyrics_model
        , metadata:created_session_id AS created_session_id
        , metadata:is_audio_watermarked AS is_audio_watermarked
        -- Additional fields that may exist in table but not in insert procedure
        , metadata:user_tier AS user_tier  -- Default value from table definition
        , metadata:create_mode AS create_mode
        , metadata:combined_from_clip_ids AS combined_from_clip_ids  -- May be populated later
        , metadata:web_client_pathname AS web_client_pathname
    FROM parsed_data
)

, lyrics_data AS (
    SELECT
        id AS clip_id
        , prompt_text AS lyrics_at_gen_from_ddb
        , prompt AS lyrics_final_from_ddb
    FROM {{ source('snowflake', 'ddb_clip_meta_heavy') }}
    WHERE p_date IN (SELECT DISTINCT base_clip_data.p_date FROM base_clip_data)
)

SELECT
    -- All base columns except the temporary lyrics columns
    base.clip_id
    , base.p_date
    , base.p_hour
    , base.created_at
    , base.user_id
    , base.request_id
    , base.s3_id
    , base.status
    , base.platform
    , base.display_index
    , base.initial_title
    , base.time_used_sec
    , base.type
    , base.task
    , base.priority
    , base.duration_sec
    , base.is_refund_credits
    , base.is_bot
    , base.credit_cost
    , base.simple_prompt
    , base.is_simple
    , base.is_make_instrumental
    , base.style_tags
    , base.negative_style_tags
    , base.request_models
    , base.model_name
    , base.experiment_inf_param
    , base.experiment_version
    , base.history
    , base.concat_history
    , base.error_type
    , base.error_message
    , base.upload_has_vocal
    , base.edit_session_id
    , base.edited_clip_id
    , base.task_details
    , base.creation_source
    , base.augmented_tags
    , base.free_quota_category
    , base.lyrics_language
    , base.simple_lyrics_model
    , base.created_session_id
    , base.is_audio_watermarked
    , base.user_tier
    , base.is_deleted
    , base.create_mode
    , base.combined_from_clip_ids
    , base.is_audio_watermark
    , base.clip_uid
    , base.web_client_pathname
    , base.metadata
    -- Merge lyrics: prefer ddb_clip_meta_heavy if available, otherwise use base values
    , Coalesce(lyrics.lyrics_at_gen_from_ddb, base.lyrics_at_gen_base) AS lyrics_at_gen
    , Coalesce(lyrics.lyrics_final_from_ddb, base.lyrics_final_base) AS lyrics_final
FROM base_clip_data AS base
LEFT JOIN lyrics_data AS lyrics
    ON base.clip_id = lyrics.clip_id
