CREATE OR REPLACE PROCEDURE PROD.USER_SESSION_STATE_HOURLY_INSERT_PROC(
    p_date DATE,
    p_hour number,
    geo_table_name STRING
)
RETURNS STRING
LANGUAGE SQL
AS 
$$
DECLARE 
    row_count INT;
    duplicate_count INT;
    start_time TIMESTAMP;
    query_sql string;
BEGIN
    start_time := CURRENT_TIMESTAMP();
    insert into USER_SESSION_STATE
    with a as (
        select 
            user_id,
            session_id, 
            parse_json(properties) as jsonstr,
            jsonstr:ip as ip,
            jsonstr: anonymous_id as anonymous_id,
            jsonstr:platform as platform,
            jsonstr:user_agent as user_agent,
            created_at,
            expire_time,
            p_date,
            p_hour
        from rds_session_history 
        where p_date = :p_date and session_type = 2 and p_hour = :p_hour
    )
    select 
        user_id,
        session_id,
        ip,
        anonymous_id,
        platform,
        user_agent,
        HASH(user_agent) as HASHED_USER_AGENT_ID,
        created_at as session_start_ts_utc,
        expire_time as session_end_ts_utc,
        datediff('second', created_at, expire_time) as session_duration_seconds,
        '' as country,
        '' as city,
        created_at as first_auth_check_ts_utc,
        created_at as last_auth_check_ts_utc,
        0 as auth_check_duration_seconds,
        p_date,
        p_hour,
        CASE WHEN jsonstr:user_agent IS NOT NULL THEN GET_MD5_ID(CAST(jsonstr:user_agent AS VARCHAR)) ELSE NULL END as md5_hashed_user_agent,
        IS_USER_AGENT_MOBILE(user_agent) as is_mobile
    from a;

    -- Call stored procedure to create a temporary Geo Info Table
    CALL CREATE_GEO_INFO_TEMP_TABLE_PROC(:geo_table_name, 
    (SELECT ARRAY_AGG(DISTINCT ip) 
        FROM PROD.USER_SESSION_STATE
        WHERE p_date = :p_date and p_hour = :p_hour)
    );

    -- Merge the temporary geo table with the USER_SESSION_STATE
    MERGE INTO PROD.USER_SESSION_STATE AS target
    USING (
        SELECT "ip" AS ip, "country" AS country, "city" AS city FROM IDENTIFIER(:geo_table_name)
    ) as source
    ON (target.ip = source.ip AND target.p_date = :p_date and target.p_hour = :p_hour)
    WHEN MATCHED THEN
        UPDATE SET 
            target.country = source.country,
            target.city = source.city;

    -- Drop the temporary geo table after merge
    DROP TABLE IDENTIFIER(:geo_table_name);

    -- Update the auth_check
    MERGE INTO PROD.USER_SESSION_STATE AS target
    USING (
        SELECT
            session_id,
            max(timestamp) as latest_auth_ts
        FROM bk_auth_event
        WHERE p_date = :p_date and p_hour = :p_hour
        GROUP BY session_id
    ) as source
    ON (target.session_id = source.session_id)
    WHEN MATCHED THEN
        UPDATE SET
            target.last_auth_check_ts_utc = source.latest_auth_ts,
            target.auth_check_duration_seconds = datediff('second', target.first_auth_check_ts_utc, source.latest_auth_ts);

    -- Count sessions with duplicates
    SELECT COUNT(*) 
        INTO :duplicate_count 
    FROM (
        SELECT session_id, COUNT(*) as cnt
        FROM user_session_state 
        GROUP BY session_id
        HAVING COUNT(*) > 1
    );

    -- Check for duplicates
    IF (:duplicate_count > 0) THEN
        select post_to_slack('#data-alerts', 'USER_SESSION_STATE_HOURLY_INSERT: Duplicate session_id found');
    END IF;
    
     -- Count number of rows for the given p_date
     SELECT COUNT(*) 
        INTO :row_count 
    FROM PROD.USER_SESSION_STATE 
    WHERE p_date = :p_date and p_hour = :p_hour;
    call TASK_MONITOR_INSERT_PROC('USER_SESSION_STATE_HOURLY_INSERT', :start_time, :row_count, 'SUCCESS', 'HOURLY');
    RETURN 'Data Inserted';
END;
$$;
