-- Upsert user reaction data from RDS to Snowflake
-- This query inserts new records and updates existing ones based on ID and partition

MERGE INTO {user_reaction_table_name} AS target
USING 
(
    SELECT 
        $1:id as id,
        $1:updated_at as updated_at,
        $1:play_count as play_count,
        $1:skip_count as skip_count,
        $1:flagged as flagged,
        $1:flagged_reason as flagged_reason,
        $1:reaction_type as reaction_type,
        $1:clip_id as clip_id,
        $1:user_id as user_id,
        $1:feature as feature,
        date(to_timestamp(updated_at)) as p_date,
        hour(to_timestamp(updated_at)) as p_hour
    FROM {stage_path} (file_format => 'BASIC_PARQUET_LOADER', pattern => ('.*.parquet'))
) AS source
ON (target.id = source.id)
WHEN MATCHED THEN
    UPDATE SET 
        target.updated_at = source.updated_at,
        target.play_count = source.play_count,
        target.skip_count = source.skip_count,
        target.flagged = source.flagged,
        target.flagged_reason = source.flagged_reason,
        target.reaction_type = source.reaction_type,
        target.clip_id = source.clip_id,
        target.user_id = source.user_id,
        target.feature = source.feature,
        target.p_date = source.p_date,
        target.p_hour = source.p_hour
WHEN NOT MATCHED THEN
    INSERT (id, updated_at, play_count, skip_count, flagged, flagged_reason, reaction_type, clip_id, user_id, feature, p_date, p_hour)
    VALUES (source.id, source.updated_at, source.play_count, source.skip_count, source.flagged, source.flagged_reason, source.reaction_type, source.clip_id, source.user_id, source.feature, source.p_date, source.p_hour);

