-- =============================================================================
-- PostgreSQL CDC Setup for Debezium
-- =============================================================================
-- Enable logical replication wal_level=logical (set in postgresql.conf via Docker)
-- This script creates publication and initial tables for Bronze layer
-- =============================================================================

-- Create publication for CDC (idempotent)
DO $$
BEGIN
    IF NOT EXISTS (
        SELECT 1 FROM pg_publication WHERE pubname = 'dbz_publication'
    ) THEN
        CREATE PUBLICATION dbz_publication FOR TABLE
            raw_orders,
            raw_order_items,
            raw_customers,
            raw_products,
            raw_sellers,
            raw_payments,
            raw_reviews,
            raw_geolocation;
    END IF;
END $$;

-- Create replication slots (Debezium will create these automatically,
-- but pre-creating with consistent names helps idempotency)
-- Note: Slots are created by Debezium connector config 'slot.name'

-- Ensure tables have primary keys for Debezium
-- (Already defined in init-scripts/postgres/01_init_dw.sql)
-- If adding new tables, ensure:
--   1. Primary key exists
--   2. REPLICA IDENTITY FULL for tables without PK (fallback)

ALTER TABLE raw_orders REPLICA IDENTITY FULL;
ALTER TABLE raw_order_items REPLICA IDENTITY FULL;
ALTER TABLE raw_customers REPLICA IDENTITY FULL;
ALTER TABLE raw_products REPLICA IDENTITY FULL;
ALTER TABLE raw_sellers REPLICA IDENTITY FULL;
ALTER TABLE raw_payments REPLICA IDENTITY FULL;
ALTER TABLE raw_reviews REPLICA IDENTITY FULL;
ALTER TABLE raw_geolocation REPLICA IDENTITY FULL;

-- Add CDC metadata columns for tracking (optional but useful)
ALTER TABLE raw_orders ADD COLUMN IF NOT EXISTS _cdc_lsn pg_lsn DEFAULT pg_current_wal_lsn();
ALTER TABLE raw_orders ADD COLUMN IF NOT EXISTS _cdc_ts TIMESTAMPTZ DEFAULT NOW();

-- Create helper function for manual LSN tracking
CREATE OR REPLACE FUNCTION cdc_get_lsn()
RETURNS pg_lsn AS $$
BEGIN
    RETURN pg_current_wal_lsn();
END;
$$ LANGUAGE plpgsql SECURITY DEFINER;

-- Grant replication privilege to dw_user (for Debezium connection)
ALTER USER dw_user WITH REPLICATION;

-- Create monitoring view for CDC lag
CREATE OR REPLACE VIEW v_cdc_lag AS
SELECT
    slot_name,
    confirmed_flush_lsn,
    pg_current_wal_lsn() - confirmed_flush_lsn AS lag_bytes
FROM pg_replication_slots
WHERE slot_name LIKE '%slot';
