CREATE OR REPLACE TASK SUNO_PROD.PROD.TRENDING_MONTHLY_INSERT
    WAREHOUSE = SUNO_PROD_LARGE
    SCHEDULE = 'USING CRON 0 0 * * 0 UTC'
AS DECLARE
        stage_path string;
        p_date date;
        p_hour string;
        query_sql string;
        row_count number;
        start_time TIMESTAMP;
    BEGIN
        start_time := CURRENT_TIMESTAMP();

        -- Insert new records
        INSERT INTO SUNO_PROD.PROD.TRENDING_MONTHLY (
            CLIP_ID,
            VERIFIED_LIKES,
            QUALITY_MULTIPLIER,
            NORMALIZED_LISTEN_DURATION,
            SCORE,
            INFERRED_LANGUAGE,
            P_DATE
        )
        WITH UserLikeHistory AS (
            SELECT 
                user_uid,
                principal_object_value as clip_id,
                COUNT(*) as like_count
            FROM suno_prod.prod.web_user_event 
            WHERE action_name = 'SongCardLikeClicked'
            AND server_timestamp >= DATEADD(DAY, -32, CURRENT_DATE)
            GROUP BY user_uid, clip_id
        ),
        UserPlayCount AS (
            SELECT 
                user_id, 
                song_id AS clip_id,
                COUNT(*) as play_count
            FROM suno_prod.prod.web_audio_player_actions 
            WHERE 
                server_timestamp > DATEADD(DAY, -32, CURRENT_DATE)
                AND NOT is_user_song_owner
                AND ACTION_NAME in ('PlaySong', 'AutoRepeatPlaySong', 'BackwardPlayNewSong', 'PlayNewSong', 'ForwardPlayNewSong','AutoRepeatPlayNewSong', 'AutoPlayNewSong', 'BackwardPlaySong', 'PlaySongInMediaSession')
            GROUP BY user_id, clip_id
        ),
        UserListenDuration AS (
            SELECT 
                user_id, 
                song_id AS clip_id,
                SUM(play_duration) AS total_play_duration
            FROM suno_prod.prod.web_audio_player_actions 
            WHERE 
                server_timestamp > DATEADD(DAY, -32, CURRENT_DATE)
                AND NOT is_user_song_owner
            GROUP BY user_id, clip_id
        ),
        CombinedUserActivity AS (
            SELECT 
                COALESCE(l.user_uid, p.user_id) AS user_id,
                COALESCE(l.clip_id, p.clip_id) AS clip_id,
                l.like_count,
                p.play_count
            FROM UserLikeHistory l
            FULL OUTER JOIN UserPlayCount p
                ON l.user_uid = p.user_id 
                AND l.clip_id = p.clip_id
        ),
        CombinedWithListening AS (
            SELECT 
                c.user_id,
                c.clip_id,
                c.like_count,
                c.play_count,
                d.total_play_duration,
                CASE WHEN (d.total_play_duration IS NOT NULL AND d.total_play_duration > 5) THEN 1 ELSE 0 END AS has_listening_history
            FROM CombinedUserActivity c
            LEFT JOIN UserListenDuration d
                ON c.user_id = d.user_id 
                AND c.clip_id = d.clip_id
        ),
        AggregatedClipMetrics AS (
            SELECT 
                clip_id,
                COUNT(DISTINCT CASE WHEN like_count IS NOT NULL THEN user_id END) AS total_likes,
                COUNT(DISTINCT CASE WHEN play_count IS NOT NULL THEN user_id END) AS users_who_played,
                COUNT(DISTINCT CASE WHEN like_count IS NOT NULL AND has_listening_history = 1 THEN user_id END) AS verified_likes,
                ROUND(AVG(CASE WHEN has_listening_history = 1 THEN total_play_duration ELSE NULL END), 2) AS avg_listen_duration,
                SUM(play_count) AS total_plays
            FROM CombinedWithListening
            GROUP BY clip_id
        ),
        ClipDuration AS (
            SELECT 
                a.*,
                CAST(try_parse_json(c.metadata):duration::VARCHAR AS FLOAT) as duration,
                c.play_count as total_play_count,
                c.created_at
            FROM AggregatedClipMetrics a
            LEFT JOIN suno_prod.prod.clip c ON a.clip_id = c.id
            WHERE c.is_public = TRUE
        ),
        ClipRecQualityScore AS (
            SELECT *
            FROM suno_prod.prod.public_clip_rec_quality_score_daily
            QUALIFY ROW_NUMBER() OVER (
                PARTITION BY clip_id 
                ORDER BY p_date DESC
            ) = 1
        ),
        ClipWithQualityScore AS (
            SELECT
                c.*,
                q.quality_multiplier
            FROM ClipDuration c
            LEFT JOIN ClipRecQualityScore q ON q.clip_id = c.clip_id
        ),
        ClipMetricsWithDuration AS (
            SELECT
                c.clip_id,
                c.quality_multiplier,
                c.verified_likes,
                c.total_likes,
                c.avg_listen_duration,
                c.total_play_count,
                c.duration,
                c.created_at,
                -- Calculate normalized average duration
                ROUND(COALESCE(c.avg_listen_duration, 0) / GREATEST(LEAST(COALESCE(c.duration, 120), 720), 120) * 240, 2) AS normalized_avg_listen_duration,
                -- Calculate raw duration ratio
                ROUND(COALESCE(c.avg_listen_duration, 0) / GREATEST(LEAST(COALESCE(c.duration, 120), 720), 120) * 240 / 64, 2) AS raw_duration_ratio
            FROM ClipWithQualityScore c
        )
        SELECT 
            m.clip_id,
            m.verified_likes,
            m.quality_multiplier,
            m.normalized_avg_listen_duration,
            -- Calculate score as duration_ratio * verified_likes
            ROUND(GREATEST(0, 2 * (1 / (1 + EXP(-1.5 * (m.raw_duration_ratio - 1))) - 0.24)) * m.verified_likes, 2) AS score,
            e.inferred_language,
            CURRENT_DATE AS p_date
        FROM ClipMetricsWithDuration m
        LEFT JOIN suno_prod.prod.rds_clip_extra e ON m.clip_id = e.clip_id
        WHERE m.created_at >= DATEADD(DAY, -32, CURRENT_DATE)
        ORDER BY score DESC;

        row_count := SQLROWCOUNT;

        call TASK_MONITOR_INSERT_PROC('TRENDING_MONTHLY_INSERT', :start_time, :row_count, 'SUCCESS', 'HOURLY');
    end;
