name: http-over-kafka

services:
  kafka:
    image: apache/kafka:3.9.1 # same image as internal/kafkatest
    ports: ["9092:9092"]
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      # HOST for processes on the laptop (go run, tests with KAFKA_BROKERS),
      # DOCKER for containers in this compose project.
      KAFKA_LISTENERS: HOST://:9092,DOCKER://:29092,CONTROLLER://:9093
      KAFKA_ADVERTISED_LISTENERS: HOST://localhost:9092,DOCKER://kafka:29092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: HOST:PLAINTEXT,DOCKER:PLAINTEXT,CONTROLLER:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: DOCKER
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@localhost:9093
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      # Topics are provisioned by the binaries (kafkaenv.EnsureTopics): a
      # missing one must fail, not appear with 1 MiB max.message.bytes.
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: "false"
    healthcheck:
      test: ["CMD", "/opt/kafka/bin/kafka-broker-api-versions.sh", "--bootstrap-server", "localhost:9092"]
      interval: 2s
      timeout: 10s
      retries: 30

  orders:
    build: { context: ., args: { CMD: orders } }
    ports: ["8081:8081"]
    environment:
      ADDR: ":8081"

  payments:
    build: { context: ., args: { CMD: payments } }
    ports: ["8082:8082"]
    environment:
      ADDR: ":8082"
      PAYMENTS_DELAY: ${PAYMENTS_DELAY:-0s}

  # gateway and bridge join here; they read KAFKA_BROKERS=kafka:29092 and
  # Keys (D10): each process gets its own private key file and the public
  # trusted-keys.env, never another role's private key.

  # What application A calls. Routed by Host: curl -H 'Host: orders' localhost:8080/orders
  # Token: go run ./cmd/devtoken -app checkout
  gateway:
    build: { context: ., args: { CMD: gateway } }
    depends_on: { kafka: { condition: service_healthy } }
    restart: on-failure
    ports: ["8080:8080", "9080:9080"] # API, admin (/healthz, /readyz)
    env_file: [deploy/gateway-signing.env, deploy/trusted-keys.env, deploy/gateway.env]
    environment:
      KAFKA_BROKERS: kafka:29092
      HOK_GATEWAY_INSTANCE: gateway-1 # stable: names http.responses.gateway-1
      HOK_SERVICES: orders=http://orders:8081,payments=http://payments:8082
      HOK_REQUEST_TIMEOUT: ${HOK_REQUEST_TIMEOUT:-10s}

  # Provider side, one per service: http.requests.<svc> → B over plain HTTP →
  # http.responses.<gateway> + http.results.<svc>, dedup state in
  # http.bridge-state.<svc>. Scale with --scale: replicas share a consumer group.
  bridge-orders:
    build: { context: ., args: { CMD: bridge } }
    depends_on: { kafka: { condition: service_healthy } }
    restart: on-failure
    env_file: [deploy/bridge-signing.env, deploy/trusted-keys.env]
    environment:
      KAFKA_BROKERS: kafka:29092
      HOK_SERVICE: orders
      HOK_UPSTREAM: http://orders:8081

  bridge-payments:
    build: { context: ., args: { CMD: bridge } }
    depends_on: { kafka: { condition: service_healthy } }
    restart: on-failure
    env_file: [deploy/bridge-signing.env, deploy/trusted-keys.env]
    environment:
      KAFKA_BROKERS: kafka:29092
      HOK_SERVICE: payments
      HOK_UPSTREAM: http://payments:8082

  # Domain events for orders: http.results.orders → orders.events, or
  # http.event-failures.orders when the mapping cannot be evaluated.
  # HOK_DERIVER_INSTANCE must stay stable per replica (it names the transactional id).
  deriver-orders:
    build: { context: ., args: { CMD: deriver } }
    depends_on: { kafka: { condition: service_healthy } }
    restart: on-failure
    env_file: deploy/trusted-keys.env
    environment:
      KAFKA_BROKERS: kafka:29092
      HOK_SERVICE: orders
      HOK_DERIVER_INSTANCE: deriver-orders-1

  # What a visitor opens: http://localhost:8090. Buttons call the gateway as
  # application checkout would; the page streams what Kafka committed. A
  # client and a reader: the dev IdP key (to mint caller tokens, like
  # cmd/devtoken) and public keys, no signing key. Bound to localhost because
  # it mints tokens for whoever reaches it.
  playground:
    build: { context: ., args: { CMD: playground } }
    depends_on:
      kafka: { condition: service_healthy }
      gateway: { condition: service_started }
    restart: on-failure
    ports: ["127.0.0.1:8090:8090"]
    env_file: [deploy/dev-idp.env, deploy/trusted-keys.env]
    environment:
      KAFKA_BROKERS: kafka:29092
      HOK_PLAYGROUND_GATEWAY: http://gateway:8080
      HOK_PLAYGROUND_GATEWAY_ADMIN: http://gateway:9080
      HOK_PLAYGROUND_PUBLIC_GATEWAY: http://localhost:8080 # written in curl commands

  # Independent reader of every http.requests.* / http.results.* topic:
  # `docker compose logs -f audit` shows the mutation stream as JSON lines.
  audit:
    build: { context: ., args: { CMD: audit } }
    depends_on: { kafka: { condition: service_healthy } }
    restart: on-failure
    env_file: deploy/trusted-keys.env
    environment:
      KAFKA_BROKERS: kafka:29092

  # Conduktor Console on http://localhost:8088, to browse the topics this stack
  # writes (http.requests.*, http.results.*, orders.events). Dev login below.
  console-db:
    image: postgres:17
    environment:
      POSTGRES_DB: console
      POSTGRES_USER: console
      POSTGRES_PASSWORD: console
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U console -d console"]
      interval: 2s
      timeout: 5s
      retries: 30

  console:
    image: conduktor/conduktor-console:1.47.2
    depends_on:
      console-db: { condition: service_healthy }
      kafka: { condition: service_healthy }
    ports: ["8088:8080"]
    environment:
      CDK_DATABASE_URL: postgresql://console:console@console-db:5432/console
      CDK_ORGANIZATION_NAME: http-over-kafka
      CDK_ADMIN_EMAIL: admin@example.com
      CDK_ADMIN_PASSWORD: Admin_1234!
      CDK_CLUSTERS_0_ID: local
      CDK_CLUSTERS_0_NAME: http-over-kafka
      CDK_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:29092
      CDK_CLUSTERS_0_COLOR: "#4F46E5"
    healthcheck:
      test: ["CMD-SHELL", "curl -sf http://localhost:8080/api/health/ready || exit 1"]
      interval: 5s
      timeout: 5s
      retries: 60