docker-compose.yml

services:
  postgres:
    image: postgres:15
    container_name: postgres
    restart: always
    environment:
      POSTGRES_USER: "${POSTGRES_USER}"
      POSTGRES_PASSWORD: "${POSTGRES_PASSWORD}"
      POSTGRES_DB: "${POSTGRES_DB}"
    command: >
      postgres
      -c wal_level=logical
      -c max_replication_slots=5
      -c max_wal_senders=5
    volumes:
      - postgres_data:/var/lib/postgresql/data
    ports:
      - "${POSTGRES_PORT}:5432"
    networks:
      - data-platform
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U ${POSTGRES_USER} -d ${POSTGRES_DB}"]
      interval: 5s
      timeout: 5s
      retries: 5

  db-seeder:
    build:
      context: ./python_services/db_seeder
    container_name: db-seeder
    depends_on:
      postgres:
        condition: service_healthy
    environment:
      POSTGRES_HOST: postgres
      POSTGRES_PORT: "${POSTGRES_PORT}"
      POSTGRES_DB: "${POSTGRES_DB}"
      POSTGRES_USER: "${POSTGRES_USER}"
      POSTGRES_PASSWORD: "${POSTGRES_PASSWORD}"
      USERS_CSV: "${USERS_CSV}"
      PRODUCTS_CSV: "${PRODUCTS_CSV}"
      TRANSACTIONS_CSV: "${TRANSACTIONS_CSV}"
    volumes:
      - ./python_services/db_seeder/data:/app/data:ro
    command: ["python", "insert_data.py"]
    networks:
      - data-platform

  zookeeper:
    image: confluentinc/cp-zookeeper:7.5.0
    container_name: zookeeper
    environment:
      ZOOKEEPER_CLIENT_PORT: "${ZOOKEEPER_CLIENT_PORT}"
      ZOOKEEPER_TICK_TIME: "${ZOOKEEPER_TICK_TIME}"
    networks:
      - data-platform
    healthcheck:
      test: ["CMD-SHELL", "echo ruok | nc localhost ${ZOOKEEPER_CLIENT_PORT}"]
      interval: 5s
      timeout: 5s
      retries: 5

  kafka:
    image: confluentinc/cp-kafka:7.5.0
    container_name: kafka
    ports:
      - "9092:9092"
    depends_on:
      zookeeper:
        condition: service_healthy
    environment:
      KAFKA_BROKER_ID: "${KAFKA_BROKER_ID}"
      KAFKA_ZOOKEEPER_CONNECT: "${KAFKA_ZOOKEEPER_CONNECT}"
      KAFKA_LISTENERS: "${KAFKA_LISTENERS}"
      KAFKA_ADVERTISED_LISTENERS: "${KAFKA_ADVERTISED_LISTENERS}"
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: "${KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR}"
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: "${KAFKA_AUTO_CREATE_TOPICS_ENABLE}"
      KAFKA_INTER_BROKER_LISTENER_NAME: "${KAFKA_INTER_BROKER_LISTENER_NAME}"
    networks:
      - data-platform
    healthcheck:
      test: ["CMD-SHELL", "kafka-topics --bootstrap-server kafka:9092 --list || exit 1"]
      interval: 10s
      timeout: 10s
      retries: 5

  schema-registry:
    image: confluentinc/cp-schema-registry:7.5.0
    container_name: schema-registry
    depends_on:
      kafka:
        condition: service_healthy
    ports:
      - "8081:8081"
    environment:
      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: "${SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS}"
      SCHEMA_REGISTRY_HOST_NAME: "${SCHEMA_REGISTRY_HOST_NAME}"
      SCHEMA_REGISTRY_LISTENERS: "${SCHEMA_REGISTRY_LISTENERS}"
    networks:
      - data-platform
    healthcheck:
      test: ["CMD-SHELL", "curl -f http://localhost:8081/subjects || exit 1"]
      interval: 10s
      timeout: 10s
      retries: 5

  kafka-connect:
    image: debezium/connect:2.5
    container_name: kafka-connect
    depends_on:
      kafka:
        condition: service_healthy
      schema-registry:
        condition: service_healthy
    ports:
      - "8083:8083"
    volumes:
      - ./debezium/plugins:/kafka/connect/libs
    environment:
      BOOTSTRAP_SERVERS: "${BOOTSTRAP_SERVERS}"
      GROUP_ID: "${GROUP_ID}"
      CONFIG_STORAGE_TOPIC: "${CONFIG_STORAGE_TOPIC}"
      OFFSET_STORAGE_TOPIC: "${OFFSET_STORAGE_TOPIC}"
      STATUS_STORAGE_TOPIC: "${STATUS_STORAGE_TOPIC}"
      CONFIG_STORAGE_REPLICATION_FACTOR: "${CONFIG_STORAGE_REPLICATION_FACTOR}"
      OFFSET_STORAGE_REPLICATION_FACTOR: "${OFFSET_STORAGE_REPLICATION_FACTOR}"
      STATUS_STORAGE_REPLICATION_FACTOR: "${STATUS_STORAGE_REPLICATION_FACTOR}"
      KEY_CONVERTER: "${KEY_CONVERTER}"
      VALUE_CONVERTER: "${VALUE_CONVERTER}"
      CONNECT_KEY_CONVERTER_SCHEMA_REGISTRY_URL: "${CONNECT_KEY_CONVERTER_SCHEMA_REGISTRY_URL}"
      CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL: "${CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL}"
      KAFKA_CONNECT_PLUGINS_DIR: "${KAFKA_CONNECT_PLUGINS_DIR}"
    networks:
      - data-platform
    healthcheck:
      test: ["CMD-SHELL", "curl -f http://localhost:8083/connectors || exit 1"]
      interval: 15s
      timeout: 10s
      retries: 10
      start_period: 30s
    restart: unless-stopped

  connector-register:
    image: curlimages/curl:7.85.0
    container_name: connector-register
    depends_on:
      kafka-connect:
        condition: service_healthy
    entrypoint:
      - /bin/sh
      - -c
      - |
        echo "Waiting 30s for Kafka Connect..." && sleep 30 && \
        echo "Checking Kafka Connect availability..." && \
        curl -f http://kafka-connect:8083/connectors && \
        echo "Registering Debezium connector..." && \
        curl -X POST http://kafka-connect:8083/connectors \
            -H "Content-Type: application/json" \
            --data @/register_debezium.json && \
        echo "Connector successfully registered" || \
        echo "Registration failed or connector already exists"
    volumes:
      - ./debezium/register_debezium.json:/register_debezium.json:ro
    networks:
      - data-platform
    restart: "no"
  
  kafka-consumer:
    build:
      context: ./python_services/kafka_consumer
    container_name: kafka-consumer
    depends_on:
      kafka-connect:
        condition: service_healthy
      schema-registry:
        condition: service_healthy
    environment:
      BOOTSTRAP_SERVERS: "${BOOTSTRAP_SERVERS_CONSUMER}"
      SCHEMA_REGISTRY_URL: "${SCHEMA_REGISTRY_URL}"
      TOPICS: "${TOPICS}"
      GROUP_ID: "${GROUP_ID_CONSUMER}"
    volumes:
      - ./python_services/kafka_consumer:/app
    networks:
      - data-platform
    command: ["python", "consumer.py"]
    restart: on-failure

  minio:
    image: minio/minio:latest
    container_name: minio
    ports:
      - "9005:9000"
      - "9006:9001"
    environment:
      MINIO_ROOT_USER: "${MINIO_ROOT_USER}"
      MINIO_ROOT_PASSWORD: "${MINIO_ROOT_PASSWORD}"
    command:
      - server
      - /data
      - --console-address
      - ":9001"
    volumes:
      - minio_data:/data
    networks:
      - data-platform
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:9000/minio/health/live"]
      interval: 10s
      timeout: 5s
      retries: 5

  minio-setup:
    image: minio/mc:latest
    container_name: minio-setup
    depends_on:
      minio:
        condition: service_healthy
    entrypoint: >
      /bin/sh -c "
        mc alias set local ${SPARK_MINIO_ENDPOINT} ${MINIO_ROOT_USER} ${MINIO_ROOT_PASSWORD} && \
        mc mb --ignore-existing local/delta-bucket && \
        mc mb --ignore-existing local/delta-checkpoint
      "
    networks:
      - data-platform

  spark:
    image: bitnami/spark:3.3.3
    container_name: spark
    user: root
    depends_on:
      minio:
        condition: service_healthy
      kafka:
        condition: service_healthy
      schema-registry:
        condition: service_healthy
    networks:
      - data-platform
    volumes:
      - ./spark:/opt/spark-apps
    command: bash -c "pip install requests && \
      /opt/bitnami/spark/bin/spark-submit --master local[*] \
      --repositories https://packages.confluent.io/maven/,https://repo1.maven.org/maven2/ \
      --packages io.delta:delta-core_2.12:2.2.0,org.apache.spark:spark-sql-kafka-0-10_2.12:3.3.3,org.apache.spark:spark-avro_2.12:3.3.3,za.co.absa:abris_2.12:6.4.0 \
      --conf spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension \
      --conf spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog \
      --conf spark.hadoop.fs.s3a.endpoint=${SPARK_MINIO_ENDPOINT} \
      --conf spark.hadoop.fs.s3a.access.key=${SPARK_MINIO_ACCESS_KEY} \
      --conf spark.hadoop.fs.s3a.secret.key=${SPARK_MINIO_SECRET_KEY} \
      --conf spark.hadoop.fs.s3a.path.style.access=${SPARK_MINIO_PATH_STYLE_ACCESS} \
      --conf spark.hadoop.fs.s3a.impl=${SPARK_MINIO_IMPL} \
      /opt/spark-apps/spark_job.py"
    
volumes:
  postgres_data:
    external: false
  minio_data:
    external: false

networks:
  data-platform:
    driver: bridge