version: "3.8"

networks:
  app-tier:
    driver: bridge

services:
  web:
    build: .
    container_name: "web"
    command: python main.py
    volumes:
      - .:/app
    ports:
      - "8000:8000"
    env_file:
      - .env
    restart: always
    environment:
      - KAFKA_BOOTSTRAP_SERVERS=kafka:9202
    networks:
      - app-tier

  mongo:
    image: mongo:latest
    container_name: mongo
    command: mongod --noauth
    env_file:
      - .env
    volumes:
      - mongo:/data/db
    ports:
      - "6000:27017"
    restart: always
    networks:
      - app-tier

  zookeeper:
    image: 'bitnami/zookeeper:latest'
    ports:
      - '2181:2181'
    environment:
      - ALLOW_ANONYMOUS_LOGIN=yes
    networks:
      - app-tier

  kafka:
    image: 'bitnami/kafka:latest'
    ports:
      - '9092:9092'
    environment:
      - KAFKA_BROKER_ID=1
      - AUTO_CREATE_TOPICS=true
      - KAFKA_ADVERTISED_HOST_NAME=kafka
      - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092
      - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://kafka:9092
      - KAFKA_CFG_ZOOKEEPER_CONNECT=zookeeper:2181
      - ALLOW_PLAINTEXT_LISTENER=yes
    depends_on:
      - zookeeper
    networks:
      - app-tier

  kafka-ui:
    image: provectuslabs/kafka-ui
    container_name: kafka-ui
    ports:
      - '8081:8080'
    environment:
      KAFKA_CLUSTERS_0_NAME: "local"
      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:9092
      KAFKA_CLUSTERS_0_ZOOKEEPER: zookeeper:2181
      KAFKA_CLUSTERS_0_READONLY: "yes"
    depends_on:
      - kafka
      - zookeeper
    networks:
      - app-tier


volumes:
  mongo:


# .env
HOST=0.0.0.0
PORT=8000
RELOAD=1
MONGO_INITDB_DATABASE=mongowhite
DATABASE_URL=mongodb://mongo
CLIENT_ID=inraddgleubfds4ohodjyk8y11wevm
CLIENT_SECRET_KEY=h2nr4gntb4tey3wcymsdpri52hg9wl
KAFKA_BOOTSTRAP_SERVERS=kafka:9202


# app.py
@app.on_event("startup")
async def create_topic_if_not_exists(kafka_cfg=settings.kafka_settings):
    bootstrap_servers = kafka_cfg.KAFKA_BOOTSTRAP_SERVERS
    client = KafkaClient(bootstrap_servers=bootstrap_servers)
    admin_client = KafkaAdminClient(bootstrap_servers=bootstrap_servers)

    future = client.cluster.request_update()
    client.poll(future=future)
    metadata = client.cluster

    if kafka_cfg.KAFKA_TOPIC not in metadata.topics():
        print('Topic creation..')
        topic = NewTopic(name=kafka_cfg.KAFKA_TOPIC, num_partitions=2, replication_factor=1)
        admin_client.create_topics(new_topics=[topic], validate_only=False)