{{ config(materialized='view') }}

WITH base AS (
  SELECT
    CAST(cus_cust_id AS STRING)                           AS cus_cust_id,
    LOWER(TRIM(CAST(warehouse_id AS STRING)))             AS warehouse_id_norm,
    CAST(inventory_id AS STRING)                          AS inventory_id,
    CAST(last_inb_status AS STRING)                       AS last_inb_status,
    SAFE_CAST(min_appointment_datetime_tz AS TIMESTAMP)   AS min_appointment_ts,
    SAFE_CAST(min_arrival_datetime_tz     AS TIMESTAMP)   AS min_arrival_ts
  FROM {{ source('meli_case_raw','raw_inbound_operations_ext') }}
),

joined AS (
  SELECT
    b.*,
    f.facility_id,
    f.facility_type,
    f.site_id                                   AS site_id_facilities,
    s.cus_cust_id_int                           AS seller_id_norm,
    s.seller_state,
    d.site_id                                   AS site_id_fallback
  FROM base b
  LEFT JOIN {{ ref('stg_facilities') }} f
    ON LOWER(TRIM(f.facility_id)) = b.warehouse_id_norm
  LEFT JOIN {{ ref('stg_sellers') }} s
    ON SAFE_CAST(b.cus_cust_id AS INT64) = s.cus_cust_id_int
  LEFT JOIN {{ ref('dim_site_from_code') }} d
    ON d.country_prefix = UPPER(SUBSTR(b.warehouse_id_norm, 1, 2))
)

SELECT
  cus_cust_id,
  warehouse_id_norm                                    AS warehouse_id,
  COALESCE(site_id_facilities, site_id_fallback)       AS site_id,
  facility_type,
  seller_state,
  min_appointment_ts,
  min_arrival_ts,
  DATE(min_appointment_ts)                             AS appointment_date,
  EXTRACT(HOUR FROM min_appointment_ts)                AS slot_hour,
  EXTRACT(DAYOFWEEK FROM min_appointment_ts)           AS dow,       
  TIMESTAMP_DIFF(min_arrival_ts, min_appointment_ts, MINUTE) AS arrival_delay_min,
  CASE WHEN TIMESTAMP_DIFF(min_arrival_ts, min_appointment_ts, MINUTE) <= 0
       THEN 1 ELSE 0 END                               AS on_time_flag,
  inventory_id,
  last_inb_status
FROM joined
