CREATE OR REPLACE TASK HOOK_BACKEND_EVENTS_HOURLY_INSERT
    WAREHOUSE = SUNO_PROD_OTHER_HOURLY_X_SMALL
    AFTER BK_NONREALTIME_EVENT_HOURLY_INSERT
    AS DECLARE
        start_time TIMESTAMP;
        row_count NUMBER;
        p_date DATE;
        p_hour NUMBER;
        BEGIN
            start_time := CURRENT_TIMESTAMP();
            p_date := DATE(SYSDATE() - INTERVAL '1 HOUR');
            p_hour := HOUR(SYSDATE() - INTERVAL '1 HOUR');
            
            INSERT INTO HOOK_BACKEND_EVENTS (
                HOOK_ID,
                USER_ID,
                HOOK_EVENT_NAME,
                RECOMMENDATION_ID,
                HOOK_STATUS,
                PAYLOAD_STATUS,
                VIDEO_S3_ID,
                IS_TEST,
                STAFF_REVIEW_STATUS,
                TIMESTAMP,
                SOURCE,
                P_DATE,
                P_HOUR
            )
            SELECT
                -- Extract hook_id from properties (present in all hook events)
                PROPERTIES:hook_id::VARCHAR AS HOOK_ID,
                
                -- Map base event fields
                USER_ID::VARCHAR AS USER_ID,
                NAME AS HOOK_EVENT_NAME,
                
                -- Extract optional properties fields, handling different event types
                PROPERTIES:recommendation_item_id::VARCHAR AS RECOMMENDATION_ID,
                PROPERTIES:hook_status::VARCHAR AS HOOK_STATUS,
                PROPERTIES:payload_status::VARCHAR AS PAYLOAD_STATUS,
                PROPERTIES:video_s3_id::VARCHAR AS VIDEO_S3_ID,
                PROPERTIES:is_test::BOOLEAN AS IS_TEST,
                PROPERTIES:staff_review_status::VARCHAR AS STAFF_REVIEW_STATUS,
                
                -- Map timestamp and source
                TIMESTAMP,
                SOURCE,
                P_DATE,
                P_HOUR
            FROM BK_NONREALTIME_EVENT
            WHERE p_date = :p_date 
              AND p_hour = :p_hour 
              AND NAME LIKE 'Hook%';  -- Filter for all hook events
            
            -- Get row count for monitoring
            SELECT COUNT(*)
            INTO :row_count
            FROM HOOK_BACKEND_EVENTS
            WHERE p_date = :p_date AND p_hour = :p_hour;
            
            -- Log task completion for monitoring
            CALL TASK_MONITOR_INSERT_PROC('HOOK_BACKEND_EVENTS_HOURLY_INSERT', :start_time, :row_count, 'SUCCESS', 'HOURLY');
        END;
