-- ============================================================
-- INIT SCRIPT: ClickHouse Real-time Analytics
-- Tables for streaming data from Flink
-- ============================================================

-- Real-time order metrics (5-min windows from Flink)
CREATE TABLE IF NOT EXISTS realtime_order_metrics (
    window_start        DateTime,
    window_end          DateTime,
    order_count         UInt32,
    total_revenue       Decimal64(2),
    total_freight       Decimal64(2),
    avg_order_value     Decimal64(2),
    unique_customers    UInt32,
    payment_credit_card UInt32,
    payment_boleto      UInt32,
    payment_voucher     UInt32,
    payment_debit_card  UInt32
) ENGINE = MergeTree()
ORDER BY (window_start)
TTL window_start + INTERVAL 90 DAY;

-- Real-time delivery tracking
CREATE TABLE IF NOT EXISTS realtime_delivery_metrics (
    window_start        DateTime,
    window_end          DateTime,
    delivered_count     UInt32,
    avg_delivery_days   Float32,
    late_count          UInt32,
    late_percentage     Float32,
    max_delay_days      Int32
) ENGINE = MergeTree()
ORDER BY (window_start)
TTL window_start + INTERVAL 90 DAY;

-- Review sentiment stream
CREATE TABLE IF NOT EXISTS realtime_review_stream (
    event_time          DateTime,
    review_id           String,
    order_id            String,
    review_score        UInt8,
    has_text            UInt8,
    comment_length      UInt32
) ENGINE = MergeTree()
ORDER BY (event_time, review_id)
TTL event_time + INTERVAL 90 DAY;

-- Anomaly alerts
CREATE TABLE IF NOT EXISTS anomaly_alerts (
    detected_at         DateTime DEFAULT now(),
    alert_type          String,
    metric_name         String,
    window_start        DateTime,
    window_end          DateTime,
    current_value       Float64,
    baseline_value      Float64,
    deviation_pct       Float64,
    z_score             Float64,
    severity            String,      -- 'warning', 'critical'
    investigated        UInt8 DEFAULT 0,
    investigation_notes String DEFAULT ''
) ENGINE = MergeTree()
ORDER BY (detected_at, alert_type);

-- Raw events log (for replay/debugging)
CREATE TABLE IF NOT EXISTS raw_events_log (
    event_time          DateTime,
    event_type          String,
    topic               String,
    order_id            String,
    payload             String       -- JSON string
) ENGINE = MergeTree()
ORDER BY (event_time, event_type)
TTL event_time + INTERVAL 30 DAY;

-- Materialized view: Hourly aggregation
CREATE TABLE IF NOT EXISTS hourly_order_summary (
    hour                DateTime,
    order_count         UInt32,
    total_revenue       Decimal64(2),
    avg_order_value     Decimal64(2)
) ENGINE = SummingMergeTree()
ORDER BY (hour);

-- Kafka engine table (connects directly to Kafka topic)
-- Uncomment when Kafka is running:
/*
CREATE TABLE IF NOT EXISTS kafka_orders_raw (
    order_id            String,
    customer_id         String,
    order_status        String,
    event_type          String,
    event_time          String
) ENGINE = Kafka()
SETTINGS
    kafka_broker_list = 'kafka:29092',
    kafka_topic_list = 'ecom.orders.created',
    kafka_group_name = 'clickhouse_consumer',
    kafka_format = 'JSONEachRow',
    kafka_num_consumers = 1;
*/
