graph LR
    A[Source Systems] -->|CDC / Events| B[Streaming Ingestion Layer]
    B --> C[Raw Landing Zone]
    C --> D[Schema Registry]
    D --> E[Transform Layer]
    E --> F[Serving Layer]
    F --> G[BI / ML / APIs]
    H[Orchestrator] -.->|trigger| B
    H -.->|trigger| E

resource "aws_s3_bucket" "data_lake" {
  for_each = toset(["bronze", "silver", "gold"])
  bucket   = "my-platform-${each.key}-${var.environment}"

  tags = {
    Layer       = each.key
    Environment = var.environment
    ManagedBy   = "terraform"
  }
}

resource "aws_s3_bucket_versioning" "data_lake_versioning" {
  for_each = aws_s3_bucket.data_lake
  bucket   = each.value.id

  versioning_configuration {
    status = "Enabled"
  }
}

# debezium-postgres-connector.yaml
name: postgres-cdc-connector
config:
  connector.class: io.debezium.connector.postgresql.PostgresConnector
  database.hostname: prod-postgres.internal
  database.port: "5432"
  database.user: debezium_reader
  database.password: "${secrets.pg_password}"
  database.dbname: transactions
  database.server.name: prod_pg
  table.include.list: public.orders,public.customers,public.payments
  plugin.name: pgoutput
  publication.autocreate.mode: filtered
  slot.name: debezium_slot
  topic.prefix: cdc
  decimal.handling.mode: double
  heartbeat.interval.ms: "5000"

# Spark-Structured-Streaming-Job-Config
spark:
  app.name: cdc-bronze-landing
  streaming:
    trigger: processingTime=30 seconds
    checkpointLocation: s3://my-platform-bronze/checkpoints/cdc/
    outputMode: append
  kafka:
    bootstrap.servers: kafka-broker-1:9092,kafka-broker-2:9092
    subscribe: cdc.public.orders,cdc.public.customers
    startingOffsets: latest
    failOnDataLoss: "false"

# Schema mit Confluent-Schema-Registry registrieren
curl -X POST http://schema-registry:8081/subjects/orders-value/versions   -H "Content-Type: application/vnd.schemaregistry.v1+json"   -d '{
    "schema": "{"type":"record","name":"Order","fields":[{"name":"id","type":"string"},{"name":"amount","type":"double"},{"name":"currency","type":["null","string"],"default":null}]}"
  }'

# Compatibility checken bevor Schema-Change deployen
curl -X POST http://schema-registry:8081/compatibility/subjects/orders-value/versions/latest   -H "Content-Type: application/vnd.schemaregistry.v1+json"   -d @new_schema.json

# Auto-Schema-Merging in Delta-Lake enablen
spark.conf.set("spark.databricks.delta.schema.autoMerge.enabled", "true")

# Write mit mergeSchema-Option
df.write \
  .format("delta") \
  .mode("append") \
  .option("mergeSchema", "true") \
  .save("s3://my-platform-silver/orders/")

-- Delta-Lake-MERGE für Idempotent-Upserts
MERGE INTO silver.orders AS target
USING (
  SELECT * FROM bronze.orders_staging
  WHERE _ingestion_date = current_date()
) AS source
ON target.order_id = source.order_id
WHEN MATCHED AND source.updated_at > target.updated_at THEN
  UPDATE SET *
WHEN NOT MATCHED THEN
  INSERT *;

# Spark: nur spezifische Partitions overwriten
spark.conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic")

df.write \
  .format("delta") \
  .mode("overwrite") \
  .partitionBy("year", "month", "day") \
  .save("s3://my-platform-silver/events/")

# dag_config.yaml
dag_id: bronze_to_silver_orders
schedule_interval: null  # Event-driven only
default_args:
  owner: data-platform
  retries: 3
  retry_delay_minutes: 5
  email_on_failure: true

tasks:
  - id: wait_for_cdc_landing
    type: S3KeySensor
    bucket: my-platform-bronze
    prefix: orders/year={{ ds_nodash[:4] }}/
    poke_interval: 60
    timeout: 3600

  - id: run_silver_transform
    type: DatabricksRunNowOperator
    depends_on: [wait_for_cdc_landing]
    job_id: "{{ var.value.silver_orders_job_id }}"

resource "aws_cloudwatch_event_rule" "s3_landing_trigger" {
  name        = "bronze-landing-trigger"
  description = "Trigger ETL pipeline on new S3 objects"

  event_pattern = jsonencode({
    source      = ["aws.s3"]
    detail-type = ["Object Created"]
    detail = {
      bucket = {
        name = [aws_s3_bucket.data_lake["bronze"].id]
      }
      object = {
        key = [{ prefix = "orders/" }]
      }
    }
  })
}

resource "aws_cloudwatch_event_target" "trigger_step_function" {
  rule      = aws_cloudwatch_event_rule.s3_landing_trigger.name
  target_id = "TriggerEtlStateMachine"
  arn       = aws_sfn_state_machine.etl_pipeline.arn
  role_arn  = aws_iam_role.eventbridge_sfn_role.arn
}

# Kafka-DLQ-Config für deinen Consumer
spring:
  kafka:
    consumer:
      group-id: etl-silver-transform
      auto-offset-reset: earliest
    listener:
      ack-mode: manual
  cloud:
    stream:
      bindings:
        input:
          destination: cdc.public.orders
          group: etl-silver-transform
          consumer:
            max-attempts: 3
            back-off-initial-interval: 1000
            back-off-multiplier: 2.0
      rabbit:
        bindings:
          input:
            consumer:
              auto-bind-dlq: true
              dlq-ttl: 604800000  # 7 Tage

# Prometheus-Metrics exposed von typischem Spark-Job
spark_streaming_records_in_total{job="cdc-bronze-landing"} 1402847
spark_streaming_records_out_total{job="cdc-bronze-landing"} 1402831
spark_streaming_processing_lag_seconds{job="cdc-bronze-landing"} 4.2
spark_streaming_batch_duration_seconds{job="cdc-bronze-landing"} 28.1