flowchart LR
    subgraph Sources["Operational Source Systems"]
        CE["Customer Events<br/>clickstream, sessions"]
        ORD["Orders<br/>cart + checkout"]
        PAY["Payments<br/>success, failed, refund"]
        INV["Inventory<br/>stock updates"]
        PROD["Product Catalog<br/>product changes"]
        SHIP["Shipments<br/>logistics events"]
        REF["Reference Data<br/>customers, products, warehouses"]
    end

    subgraph Streaming["Streaming Ingestion"]
        KAFKA["Kafka Topics"]
        CONSUMER["S3 Landing Consumer"]
    end

    subgraph Lake["Lakehouse Storage"]
        LAND["Landing Zone<br/>raw JSONL / CSV"]
        BRONZE["Bronze<br/>raw payload + metadata"]
        SILVER["Silver<br/>clean, typed, deduped"]
        QUAR["Quarantine<br/>bad records + reasons"]
        GOLD["Gold<br/>KPI-ready tables"]
    end

    subgraph Orchestration["Orchestration + Processing"]
        AF["Airflow DAGs"]
        DBX["Databricks notebooks/jobs"]
        LOCAL["Local pipeline fallback"]
        DQ["Reusable DQ framework"]
    end

    subgraph Serving["Serving + Consumption"]
        DASH["Plotly Dash<br/>executive dashboard"]
        USERS["Business users<br/>ops, finance, product"]
    end

    CE --> KAFKA
    ORD --> KAFKA
    PAY --> KAFKA
    INV --> KAFKA
    PROD --> KAFKA
    SHIP --> KAFKA
    REF --> LAND
    KAFKA --> CONSUMER --> LAND
    LAND --> BRONZE --> SILVER --> GOLD --> DASH --> USERS
    SILVER --> QUAR
    DQ --> SILVER
    DQ --> GOLD
    DQ --> QUAR
    AF --> BRONZE
    AF --> SILVER
    AF --> GOLD
    AF --> DQ
    DBX --> BRONZE
    DBX --> SILVER
    DBX --> GOLD
    LOCAL --> BRONZE
    LOCAL --> SILVER
    LOCAL --> GOLD

    style Sources fill:#E0F2FE,stroke:#0284C7,color:#0F172A
    style Streaming fill:#DCFCE7,stroke:#16A34A,color:#0F172A
    style Lake fill:#FEF3C7,stroke:#D97706,color:#0F172A
    style Orchestration fill:#FAE8FF,stroke:#A855F7,color:#0F172A
    style Serving fill:#F8FAFC,stroke:#64748B,color:#0F172A
    style QUAR fill:#FEE2E2,stroke:#DC2626,color:#0F172A
    style GOLD fill:#0F172A,stroke:#22C55E,color:#FFFFFF

flowchart TB
    A["Landing<br/>JSONL event files + reference CSVs"] --> B["Bronze"]
    B --> C["Silver"]
    C --> D["Gold"]
    C --> E["Quarantine"]
    D --> F["Dashboard"]

    B1["Preserve raw_json<br/>event_id<br/>source_topic<br/>source_file<br/>batch_id<br/>processing_date"] --> B
    C1["Parse JSON<br/>cast timestamps/numbers<br/>normalize statuses<br/>deduplicate event_id<br/>apply DQ flags"] --> C
    E1["Invalid IDs<br/>failed checks<br/>reason_code<br/>check_name<br/>source evidence"] --> E
    D1["daily_revenue<br/>product_performance<br/>customer_360<br/>inventory_health<br/>payment_reliability<br/>data_quality_scorecard"] --> D

    style A fill:#E0F2FE,stroke:#0284C7,color:#0F172A
    style B fill:#FEF3C7,stroke:#D97706,color:#0F172A
    style C fill:#DCFCE7,stroke:#16A34A,color:#0F172A
    style D fill:#0F172A,stroke:#22C55E,color:#FFFFFF
    style E fill:#FEE2E2,stroke:#DC2626,color:#0F172A
    style F fill:#F8FAFC,stroke:#64748B,color:#0F172A

stateDiagram-v2
    [*] --> RawEventReceived
    RawEventReceived --> BronzeStored: raw payload preserved
    BronzeStored --> ParsedSilverRecord: JSON parsed and standardized
    ParsedSilverRecord --> ValidRecord: schema and business checks pass
    ParsedSilverRecord --> QuarantinedRecord: required check fails
    ValidRecord --> GoldAggregation: record contributes to KPIs
    QuarantinedRecord --> DQScorecard: failed_count and reason_code recorded
    GoldAggregation --> DashboardReady
    DQScorecard --> DashboardReady
    DashboardReady --> [*]

flowchart LR
    A["streamcommerce_batch_ingestion_dag<br/>seed reference + validate landing"] --> B["streamcommerce_bronze_to_silver_dag<br/>Bronze ingest + Silver transform"]
    B --> C["streamcommerce_data_quality_dag<br/>critical checks + thresholds"]
    C --> D["streamcommerce_silver_to_gold_dag<br/>Gold KPI refresh"]
    D --> E["Dashboard refresh<br/>executive analytics"]

    C -. critical failure .-> F["Fail DAG<br/>inspect quarantine + DQ scorecard"]

    style A fill:#E0F2FE,stroke:#0284C7,color:#0F172A
    style B fill:#FEF3C7,stroke:#D97706,color:#0F172A
    style C fill:#FAE8FF,stroke:#A855F7,color:#0F172A
    style D fill:#DCFCE7,stroke:#16A34A,color:#0F172A
    style E fill:#0F172A,stroke:#22C55E,color:#FFFFFF
    style F fill:#FEE2E2,stroke:#DC2626,color:#0F172A

flowchart TB
    subgraph Local["Local Portfolio Mode"]
        LK["Docker Kafka"]
        LM["MinIO S3-compatible storage"]
        LA["Airflow containers"]
        LP["Pandas/Parquet local pipeline"]
        LD["Dash dashboard"]
    end

    subgraph Cloud["Cloud Production Pattern"]
        CK["Managed Kafka / MSK / Confluent"]
        CS["AWS S3 lakehouse bucket"]
        CA["Airflow / MWAA"]
        CD["Databricks Spark + Delta"]
        CI["Terraform infrastructure"]
    end

    LK --> CK
    LM --> CS
    LA --> CA
    LP --> CD
    LD --> CS
    CI --> CS

    style Local fill:#E0F2FE,stroke:#0284C7,color:#0F172A
    style Cloud fill:#DCFCE7,stroke:#16A34A,color:#0F172A