--IMPORT
WITH messages AS (
    SELECT * FROM {{ ref('stg_crisp__messages') }}
    ORDER BY conversation_id, created_datetime, message_id
),

conversations AS (
    SELECT * FROM {{ ref('stg_crisp__conversations') }}
),

--LOGIC
mark_first_dialog_message AS (
    SELECT
        CASE 
            WHEN LAG(event_type) OVER (ORDER BY conversation_id, created_datetime, message_id) = 'state:resolved' 
            THEN 1 
            ELSE 0 
        END AS first_dialog_message_flag,
        *

    FROM messages
),

message_with_dialog_id AS (
    SELECT 
        (SUM(first_dialog_message_flag) OVER (ORDER BY conversation_id, created_datetime, message_id ROWS UNBOUNDED PRECEDING))+1 AS dialog_id,
        *

    FROM mark_first_dialog_message
    ORDER BY conversation_id, created_datetime
),

--FINAL
message_with_dialog_id_get_user_id AS (
    SELECT 
        message_with_dialog_id.dialog_id,
        message_with_dialog_id.message_id,

        message_with_dialog_id.conversation_id,

        CASE
            WHEN message_with_dialog_id.message_from_side = 'operator' THEN message_with_dialog_id.user_id
            WHEN message_with_dialog_id.message_from_side = 'user' and conversations.user_id IS NOT NULL THEN conversations.user_id
            WHEN message_with_dialog_id.message_from_side = 'user' and conversations.user_id IS NULL THEN null
            ELSE 'ERROR please contact DATA team'
        END AS user_id,

        message_with_dialog_id.user_name,
        message_with_dialog_id.message_from_side,

        CASE
            WHEN message_with_dialog_id.message_from_side = 'user' AND message_with_dialog_id.user_id IS NOT NULL THEN 'user'
            WHEN message_with_dialog_id.message_from_side = 'operator' AND message_with_dialog_id.user_id IS NOT NULL THEN 'operator'
            WHEN message_with_dialog_id.message_from_side = 'operator' AND message_with_dialog_id.user_id IS NULL THEN 'bot'
            ELSE 'ERROR please contact DATA team'
        END AS message_generated_by,
        
        message_with_dialog_id.message_generated_platform,
        message_with_dialog_id.message_delivered_platform,
        message_with_dialog_id.message_read_platform,
        message_with_dialog_id.message_format,
        message_with_dialog_id.event_type,
        message_with_dialog_id.event_message,
        message_with_dialog_id.message,
        message_with_dialog_id.is_edited_message,
        message_with_dialog_id.attachment_type,
        message_with_dialog_id.attachment_name,
        message_with_dialog_id.attachment_url,
        message_with_dialog_id.created_datetime,
        '{{ modules.datetime.datetime.now(modules.pytz.timezone("Asia/Kuala_Lumpur")) }}' AS _dbt_ran_datetime

    FROM message_with_dialog_id
    LEFT JOIN conversations
        ON (message_with_dialog_id.user_id = conversations.conversation_id)    --The user_id will become conversation_id if message_from_side is user

),

-- FINAL
final__int_crisp__messages_generate_dialog_ids AS (
    SELECT
        -- ids
        dialog_id,
        message_id,
        conversation_id,
        user_id,

        -- dimensions
        user_name,
        message_from_side,
        message_generated_by,
        message_generated_platform,
        message_delivered_platform,
        message_read_platform,
        message_format,
        event_type,
        event_message,
        message,
        is_edited_message,
        attachment_type,
        attachment_name,
        attachment_url,

        -- measures

        -- date/times
        created_datetime,

        -- metadata
        _dbt_ran_datetime

    FROM message_with_dialog_id_get_user_id
)

SELECT * FROM final__int_crisp__messages_generate_dialog_ids