src/customer_intelligence/etl/warehouse_transform.sql

-- Staging → Warehouse transforms
-- Executed in order by OlistLoader.run_warehouse_transform()
-- All statements are idempotent via ON CONFLICT DO UPDATE

-- ── 1. UPSERT dim_customers ────────────────────────────────────────────────
INSERT INTO warehouse.dim_customers (
    customer_id,
    customer_unique_id,
    zip_code_prefix,
    city,
    state,
    lat,
    lng
)
SELECT
    c.customer_id,
    c.customer_unique_id,
    c.customer_zip_code_prefix,
    INITCAP(c.customer_city)            AS city,
    UPPER(c.customer_state)             AS state,
    g.geolocation_lat::DOUBLE PRECISION AS lat,
    g.geolocation_lng::DOUBLE PRECISION AS lng
FROM staging.stg_customers c
LEFT JOIN staging.stg_geolocation g
    ON c.customer_zip_code_prefix = g.geolocation_zip_code_prefix
ON CONFLICT (customer_id) DO UPDATE SET
    customer_unique_id  = EXCLUDED.customer_unique_id,
    zip_code_prefix     = EXCLUDED.zip_code_prefix,
    city                = EXCLUDED.city,
    state               = EXCLUDED.state,
    lat                 = EXCLUDED.lat,
    lng                 = EXCLUDED.lng,
    _updated_at         = now();

-- ── 2. UPSERT dim_products ─────────────────────────────────────────────────
INSERT INTO warehouse.dim_products (
    product_id,
    category_name_pt,
    category_name_en,
    weight_g,
    length_cm,
    height_cm,
    width_cm
)
SELECT
    p.product_id,
    p.product_category_name,
    t.product_category_name_english,
    p.product_weight_g::INTEGER,
    p.product_length_cm::INTEGER,
    p.product_height_cm::INTEGER,
    p.product_width_cm::INTEGER
FROM staging.stg_products p
LEFT JOIN staging.stg_product_category_translations t
    ON p.product_category_name = t.product_category_name
ON CONFLICT (product_id) DO UPDATE SET
    category_name_pt = EXCLUDED.category_name_pt,
    category_name_en = EXCLUDED.category_name_en,
    weight_g         = EXCLUDED.weight_g,
    length_cm        = EXCLUDED.length_cm,
    height_cm        = EXCLUDED.height_cm,
    width_cm         = EXCLUDED.width_cm,
    _updated_at      = now();

-- ── 3. UPSERT dim_sellers ──────────────────────────────────────────────────
INSERT INTO warehouse.dim_sellers (
    seller_id,
    zip_code_prefix,
    city,
    state,
    lat,
    lng
)
SELECT
    s.seller_id,
    s.seller_zip_code_prefix,
    INITCAP(s.seller_city)              AS city,
    UPPER(s.seller_state)               AS state,
    g.geolocation_lat::DOUBLE PRECISION AS lat,
    g.geolocation_lng::DOUBLE PRECISION AS lng
FROM staging.stg_sellers s
LEFT JOIN staging.stg_geolocation g
    ON s.seller_zip_code_prefix = g.geolocation_zip_code_prefix
ON CONFLICT (seller_id) DO UPDATE SET
    zip_code_prefix = EXCLUDED.zip_code_prefix,
    city            = EXCLUDED.city,
    state           = EXCLUDED.state,
    lat             = EXCLUDED.lat,
    lng             = EXCLUDED.lng,
    _updated_at     = now();

-- ── 4. INSERT fact_orders (order-item grain) ────────────────────────────────
-- Aggregate one payment row per order (sum values, keep first payment_type)
WITH order_payments AS (
    SELECT
        order_id,
        SUM(payment_value::NUMERIC)         AS payment_value,
        MIN(payment_type)                   AS payment_type,
        MAX(payment_installments::INTEGER)  AS payment_installments
    FROM staging.stg_order_payments
    GROUP BY order_id
),
order_reviews AS (
    SELECT DISTINCT ON (order_id)
        order_id,
        review_score::SMALLINT AS review_score
    FROM staging.stg_order_reviews
    WHERE review_score ~ '^\d+$'
    ORDER BY order_id, review_answer_timestamp DESC
)
INSERT INTO warehouse.fact_orders (
    order_id,
    order_item_id,
    customer_key,
    product_key,
    seller_key,
    order_date_id,
    approved_date_id,
    delivered_date_id,
    estimated_delivery_id,
    order_status,
    price,
    freight_value,
    payment_value,
    payment_type,
    payment_installments,
    review_score,
    days_to_delivery,
    delivery_delay_days,
    is_late
)
SELECT
    o.order_id,
    oi.order_item_id::SMALLINT,
    dc.customer_key,
    dp.product_key,
    ds.seller_key,
    -- Dates (cast timestamp text to date)
    DATE(o.order_purchase_timestamp::TIMESTAMP)                 AS order_date_id,
    CASE WHEN o.order_approved_at <> '' THEN DATE(o.order_approved_at::TIMESTAMP) END,
    CASE WHEN o.order_delivered_customer_date <> ''
         THEN DATE(o.order_delivered_customer_date::TIMESTAMP) END,
    CASE WHEN o.order_estimated_delivery_date <> ''
         THEN DATE(o.order_estimated_delivery_date::TIMESTAMP) END,
    o.order_status,
    oi.price::NUMERIC,
    oi.freight_value::NUMERIC,
    op.payment_value,
    op.payment_type,
    op.payment_installments::SMALLINT,
    orv.review_score,
    -- Derived delivery metrics
    CASE
        WHEN o.order_delivered_customer_date <> '' AND o.order_approved_at <> ''
        THEN DATE(o.order_delivered_customer_date::TIMESTAMP)
             - DATE(o.order_approved_at::TIMESTAMP)
    END AS days_to_delivery,
    CASE
        WHEN o.order_delivered_customer_date <> '' AND o.order_estimated_delivery_date <> ''
        THEN DATE(o.order_delivered_customer_date::TIMESTAMP)
             - DATE(o.order_estimated_delivery_date::TIMESTAMP)
    END AS delivery_delay_days,
    CASE
        WHEN o.order_delivered_customer_date <> '' AND o.order_estimated_delivery_date <> ''
        THEN DATE(o.order_delivered_customer_date::TIMESTAMP)
             > DATE(o.order_estimated_delivery_date::TIMESTAMP)
        ELSE NULL
    END AS is_late
FROM staging.stg_orders o
JOIN staging.stg_order_items oi USING (order_id)
JOIN warehouse.dim_customers dc ON o.customer_id = dc.customer_id
JOIN warehouse.dim_products dp ON oi.product_id = dp.product_id
JOIN warehouse.dim_sellers ds ON oi.seller_id = ds.seller_id
LEFT JOIN order_payments op ON o.order_id = op.order_id
LEFT JOIN order_reviews orv ON o.order_id = orv.order_id
ON CONFLICT (order_id, order_item_id) DO UPDATE SET
    order_status        = EXCLUDED.order_status,
    price               = EXCLUDED.price,
    freight_value       = EXCLUDED.freight_value,
    payment_value       = EXCLUDED.payment_value,
    payment_type        = EXCLUDED.payment_type,
    payment_installments = EXCLUDED.payment_installments,
    review_score        = EXCLUDED.review_score,
    days_to_delivery    = EXCLUDED.days_to_delivery,
    delivery_delay_days = EXCLUDED.delivery_delay_days,
    is_late             = EXCLUDED.is_late;

-- ── 5. Update dim_customers aggregates from fact_orders ────────────────────
UPDATE warehouse.dim_customers dc
SET
    first_order_date = agg.first_order_date,
    last_order_date  = agg.last_order_date,
    total_orders     = agg.total_orders,
    total_revenue    = agg.total_revenue,
    _updated_at      = now()
FROM (
    SELECT
        customer_key,
        MIN(order_date_id)          AS first_order_date,
        MAX(order_date_id)          AS last_order_date,
        COUNT(DISTINCT order_id)    AS total_orders,
        SUM(price + freight_value)  AS total_revenue
    FROM warehouse.fact_orders
    GROUP BY customer_key
) agg
WHERE dc.customer_key = agg.customer_key;

-- ── 6. Update dim_sellers aggregates from fact_orders ──────────────────────
UPDATE warehouse.dim_sellers ds
SET
    total_orders     = agg.total_orders,
    avg_review_score = agg.avg_review_score,
    on_time_rate     = agg.on_time_rate,
    _updated_at      = now()
FROM (
    SELECT
        seller_key,
        COUNT(DISTINCT order_id)                AS total_orders,
        AVG(review_score)                       AS avg_review_score,
        AVG(CASE WHEN is_late = false THEN 1.0
                 WHEN is_late = true  THEN 0.0
                 ELSE NULL END)                 AS on_time_rate
    FROM warehouse.fact_orders
    GROUP BY seller_key
) agg
WHERE ds.seller_key = agg.seller_key