---
# DragonflyDB (Redis-compatible in-memory datastore)
apiVersion: v1
kind: Service
metadata:
  name: dragonfly
spec:
  selector:
    app: dragonfly
  ports:
    - protocol: TCP
      port: 6379
      targetPort: 6379
  type: ClusterIP
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: dragonfly
spec:
  replicas: 1
  selector:
    matchLabels:
      app: dragonfly
  template:
    metadata:
      labels:
        app: dragonfly
    spec:
      containers:
      - name: dragonfly
        image: docker.dragonflydb.io/dragonflydb/dragonfly:latest
        ports:
        - containerPort: 6379
          name: dragonfly
        args:
        - --logtostderr
        - --alsologtostderr=false
        - --cache_mode=true
        resources:
          requests:
            memory: "256Mi"
            cpu: "100m"
          limits:
            memory: "512Mi"
            cpu: "500m"
---
# Qdrant (Vector database for RAG)
apiVersion: v1
kind: Service
metadata:
  name: qdrant
spec:
  selector:
    app: qdrant
  ports:
    - name: http
      protocol: TCP
      port: 6333
      targetPort: 6333
    - name: grpc
      protocol: TCP
      port: 6334
      targetPort: 6334
  type: ClusterIP
---
apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: qdrant
spec:
  serviceName: qdrant
  replicas: 1
  selector:
    matchLabels:
      app: qdrant
  template:
    metadata:
      labels:
        app: qdrant
    spec:
      containers:
      - name: qdrant
        image: qdrant/qdrant:latest
        ports:
        - containerPort: 6333
          name: http
        - containerPort: 6334
          name: grpc
        resources:
          requests:
            memory: "512Mi"
            cpu: "200m"
          limits:
            memory: "1Gi"
            cpu: "1000m"
        volumeMounts:
        - name: qdrant-data
          mountPath: /qdrant/storage
  volumeClaimTemplates:
  - metadata:
      name: qdrant-data
    spec:
      accessModes: ["ReadWriteOnce"]
      storageClassName: dev-ephemeral
      resources:
        requests:
          storage: 10Gi
---
# Kafka (KRaft mode - no Zookeeper needed)
# Using apache/kafka:3.9.0 instead of confluentinc/cp-kafka because:
# - cp-kafka's entrypoint script has issues with KRaft configuration
# - apache/kafka allows explicit command configuration
# - For production, use Strimzi operator (see kafka/ directory)
apiVersion: v1
kind: Service
metadata:
  name: kafka
spec:
  selector:
    app: kafka
  ports:
    - name: broker
      protocol: TCP
      port: 9092
      targetPort: 9092
    - name: controller
      protocol: TCP
      port: 9093
      targetPort: 9093
  type: ClusterIP
---
apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: kafka
spec:
  serviceName: kafka
  replicas: 1
  selector:
    matchLabels:
      app: kafka
  template:
    metadata:
      labels:
        app: kafka
    spec:
      containers:
      - name: kafka
        image: apache/kafka:3.9.0
        ports:
        - containerPort: 9092
          name: broker
        - containerPort: 9093
          name: controller
        command:
        - sh
        - -c
        - |
          CLUSTER_ID="dexorder-dev-cluster"
          LOG_DIR="/var/lib/kafka/data"

          # Ensure log directory exists
          mkdir -p $LOG_DIR

          # Create temporary config with custom log.dirs for formatting
          cp /opt/kafka/config/kraft/server.properties /tmp/server.properties
          echo "log.dirs=$LOG_DIR" >> /tmp/server.properties

          # Format storage if not already formatted
          if [ ! -f $LOG_DIR/meta.properties ]; then
            echo "Formatting Kafka storage with cluster ID: $CLUSTER_ID"
            /opt/kafka/bin/kafka-storage.sh format -t $CLUSTER_ID -c /tmp/server.properties
          else
            echo "Kafka storage already formatted, skipping format step"
          fi

          # Start Kafka server
          /opt/kafka/bin/kafka-server-start.sh /opt/kafka/config/kraft/server.properties \
            --override node.id=1 \
            --override process.roles=broker,controller \
            --override listeners=PLAINTEXT://:9092,CONTROLLER://:9093 \
            --override advertised.listeners=PLAINTEXT://kafka:9092 \
            --override controller.quorum.voters=1@kafka:9093 \
            --override controller.listener.names=CONTROLLER \
            --override listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT \
            --override log.dirs=$LOG_DIR \
            --override offsets.topic.replication.factor=1 \
            --override transaction.state.log.replication.factor=1 \
            --override transaction.state.log.min.isr=1
        env: []
        volumeMounts:
        - name: kafka-data
          mountPath: /var/lib/kafka/data
  volumeClaimTemplates:
  - metadata:
      name: kafka-data
    spec:
      accessModes: ["ReadWriteOnce"]
      storageClassName: dev-ephemeral
      resources:
        requests:
          storage: 5Gi
---
# PostgreSQL (for Iceberg catalog metadata)
apiVersion: v1
kind: Service
metadata:
  name: postgres
spec:
  selector:
    app: postgres
  ports:
    - protocol: TCP
      port: 5432
      targetPort: 5432
  type: ClusterIP
---
apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: postgres
spec:
  serviceName: postgres
  replicas: 1
  selector:
    matchLabels:
      app: postgres
  template:
    metadata:
      labels:
        app: postgres
    spec:
      containers:
      - name: postgres
        image: postgres:15
        ports:
        - containerPort: 5432
        env:
        - name: POSTGRES_USER
          value: postgres
        - name: POSTGRES_PASSWORD
          valueFrom:
            secretKeyRef:
              name: postgres-secret
              key: password
        - name: POSTGRES_DB
          value: iceberg
        volumeMounts:
        - name: postgres-data
          mountPath: /var/lib/postgresql/data
  volumeClaimTemplates:
  - metadata:
      name: postgres-data
    spec:
      accessModes: ["ReadWriteOnce"]
      storageClassName: dev-ephemeral
      resources:
        requests:
          storage: 2Gi
---
# MinIO (S3-compatible object storage)
apiVersion: v1
kind: Service
metadata:
  name: minio
spec:
  selector:
    app: minio
  ports:
    - name: api
      protocol: TCP
      port: 9000
      targetPort: 9000
    - name: console
      protocol: TCP
      port: 9001
      targetPort: 9001
  type: ClusterIP
---
apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: minio
spec:
  serviceName: minio
  replicas: 1
  selector:
    matchLabels:
      app: minio
  template:
    metadata:
      labels:
        app: minio
    spec:
      containers:
      - name: minio
        image: minio/minio:latest
        args:
        - server
        - /data
        - --console-address
        - ":9001"
        ports:
        - containerPort: 9000
          name: api
        - containerPort: 9001
          name: console
        env:
        - name: MINIO_ROOT_USER
          valueFrom:
            secretKeyRef:
              name: minio-secret
              key: root-user
        - name: MINIO_ROOT_PASSWORD
          valueFrom:
            secretKeyRef:
              name: minio-secret
              key: root-password
        volumeMounts:
        - name: minio-data
          mountPath: /data
  volumeClaimTemplates:
  - metadata:
      name: minio-data
    spec:
      accessModes: ["ReadWriteOnce"]
      storageClassName: dev-ephemeral
      resources:
        requests:
          storage: 10Gi
---
# MinIO bucket initialization job
apiVersion: batch/v1
kind: Job
metadata:
  name: minio-init-buckets
spec:
  ttlSecondsAfterFinished: 100
  template:
    spec:
      restartPolicy: OnFailure
      containers:
      - name: create-buckets
        image: minio/mc:latest
        command:
        - sh
        - -c
        - |
          echo "Waiting for MinIO to be ready..."
          until mc alias set minio http://minio:9000 $MINIO_ROOT_USER $MINIO_ROOT_PASSWORD; do
            sleep 2
          done

          echo "Creating warehouse bucket..."
          mc mb minio/warehouse --ignore-existing

          echo "Buckets initialized successfully"
        env:
        - name: MINIO_ROOT_USER
          valueFrom:
            secretKeyRef:
              name: minio-secret
              key: root-user
        - name: MINIO_ROOT_PASSWORD
          valueFrom:
            secretKeyRef:
              name: minio-secret
              key: root-password
---
# Iceberg REST Catalog
apiVersion: v1
kind: Service
metadata:
  name: iceberg-catalog
spec:
  selector:
    app: iceberg-catalog
  ports:
    - protocol: TCP
      port: 8181
      targetPort: 8181
  type: ClusterIP
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: iceberg-catalog
spec:
  replicas: 1
  selector:
    matchLabels:
      app: iceberg-catalog
  template:
    metadata:
      labels:
        app: iceberg-catalog
    spec:
      initContainers:
      - name: wait-for-postgres
        image: busybox:1.36
        command: ['sh', '-c', 'until nc -z postgres 5432; do echo waiting for postgres; sleep 2; done;']
      - name: wait-for-minio
        image: busybox:1.36
        command: ['sh', '-c', 'until nc -z minio 9000; do echo waiting for minio; sleep 2; done;']
      containers:
      - name: iceberg-catalog
        image: tabulario/iceberg-rest:latest
        ports:
        - containerPort: 8181
        env:
        - name: CATALOG_WAREHOUSE
          value: "s3://warehouse/"
        - name: CATALOG_IO__IMPL
          value: "org.apache.iceberg.aws.s3.S3FileIO"
        - name: CATALOG_S3_ENDPOINT
          value: "http://minio:9000"
        - name: CATALOG_S3_ACCESS__KEY__ID
          valueFrom:
            secretKeyRef:
              name: minio-secret
              key: root-user
        - name: CATALOG_S3_SECRET__ACCESS__KEY
          valueFrom:
            secretKeyRef:
              name: minio-secret
              key: root-password
        - name: CATALOG_S3_PATH__STYLE__ACCESS
          value: "true"
        - name: AWS_REGION
          value: "us-east-1"
---
# Flink JobManager
apiVersion: v1
kind: Service
metadata:
  name: flink-jobmanager
spec:
  selector:
    app: flink-jobmanager
  ports:
    - name: rpc
      protocol: TCP
      port: 6123
      targetPort: 6123
    - name: ui
      protocol: TCP
      port: 8081
      targetPort: 8081
    - name: zmq-market-data
      protocol: TCP
      port: 5558
      targetPort: 5558
    - name: zmq-notif-pull
      protocol: TCP
      port: 5561
      targetPort: 5561
  type: ClusterIP
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: flink-jobmanager
spec:
  replicas: 1
  selector:
    matchLabels:
      app: flink-jobmanager
  template:
    metadata:
      labels:
        app: flink-jobmanager
    spec:
      initContainers:
      - name: wait-for-kafka
        image: busybox:1.36
        command: ['sh', '-c', 'until nc -z kafka 9092; do echo waiting for kafka; sleep 2; done;']
      - name: wait-for-iceberg-catalog
        image: busybox:1.36
        command: ['sh', '-c', 'until nc -z iceberg-catalog 8181; do echo waiting for iceberg-catalog; sleep 2; done;']
      containers:
      - name: flink-jobmanager
        image: dexorder/flink:latest
        imagePullPolicy: Never
        args: ["standalone-job", "--job-classname", "com.dexorder.flink.TradingFlinkApp"]
        ports:
        - containerPort: 6123
          name: rpc
        - containerPort: 8081
          name: ui
        - containerPort: 5558
          name: zmq-market-data
        - containerPort: 5561
          name: zmq-notif-pull
        env:
        - name: JOB_MANAGER_RPC_ADDRESS
          value: flink-jobmanager
        - name: AWS_REGION
          value: us-east-1
        - name: AWS_ACCESS_KEY_ID
          valueFrom:
            secretKeyRef:
              name: flink-secrets
              key: minio-access-key
        - name: AWS_SECRET_ACCESS_KEY
          valueFrom:
            secretKeyRef:
              name: flink-secrets
              key: minio-secret-key
        volumeMounts:
        - name: flink-config
          mountPath: /etc/config/config.yaml
          subPath: config.yaml
        - name: flink-secrets
          mountPath: /etc/secrets
      volumes:
      - name: flink-config
        configMap:
          name: flink-config
      - name: flink-secrets
        secret:
          secretName: flink-secrets
---
# Flink TaskManager
apiVersion: apps/v1
kind: Deployment
metadata:
  name: flink-taskmanager
spec:
  replicas: 1
  selector:
    matchLabels:
      app: flink-taskmanager
  template:
    metadata:
      labels:
        app: flink-taskmanager
    spec:
      initContainers:
      - name: wait-for-jobmanager
        image: busybox:1.36
        command: ['sh', '-c', 'until nc -z flink-jobmanager 6123; do echo waiting for jobmanager; sleep 2; done;']
      containers:
      - name: flink-taskmanager
        image: dexorder/flink:latest
        imagePullPolicy: Never
        args: ["taskmanager"]
        env:
        - name: JOB_MANAGER_RPC_ADDRESS
          value: flink-jobmanager
        - name: AWS_REGION
          value: us-east-1
        - name: AWS_ACCESS_KEY_ID
          valueFrom:
            secretKeyRef:
              name: flink-secrets
              key: minio-access-key
        - name: AWS_SECRET_ACCESS_KEY
          valueFrom:
            secretKeyRef:
              name: flink-secrets
              key: minio-secret-key
        volumeMounts:
        - name: flink-config
          mountPath: /etc/config/config.yaml
          subPath: config.yaml
        - name: flink-secrets
          mountPath: /etc/secrets
      volumes:
      - name: flink-config
        configMap:
          name: flink-config
      - name: flink-secrets
        secret:
          secretName: flink-secrets
---
# Relay (ZMQ router)
apiVersion: v1
kind: Service
metadata:
  name: relay
spec:
  selector:
    app: relay
  ports:
    - name: work-queue
      protocol: TCP
      port: 5555
      targetPort: 5555
    - name: responses
      protocol: TCP
      port: 5556
      targetPort: 5556
    - name: market-data
      protocol: TCP
      port: 5558
      targetPort: 5558
    - name: client-requests
      protocol: TCP
      port: 5559
      targetPort: 5559
  type: ClusterIP
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: relay
spec:
  replicas: 1
  selector:
    matchLabels:
      app: relay
  template:
    metadata:
      labels:
        app: relay
    spec:
      containers:
      - name: relay
        image: dexorder/relay:latest
        imagePullPolicy: Never
        ports:
        - containerPort: 5555
          name: work-queue
        - containerPort: 5556
          name: responses
        - containerPort: 5558
          name: market-data
        - containerPort: 5559
          name: client-requests
        env:
        - name: RUST_LOG
          value: relay=info
        - name: CONFIG_PATH
          value: /config/config.yaml
        volumeMounts:
        - name: relay-config
          mountPath: /config
      volumes:
      - name: relay-config
        configMap:
          name: relay-config
---
# Ingestor (CCXT data fetcher)
apiVersion: apps/v1
kind: Deployment
metadata:
  name: ingestor
spec:
  replicas: 1
  selector:
    matchLabels:
      app: ingestor
  template:
    metadata:
      labels:
        app: ingestor
    spec:
      initContainers:
      - name: wait-for-relay
        image: busybox:1.36
        command: ['sh', '-c', 'until nc -z relay 5555; do echo waiting for relay; sleep 2; done;']
      - name: wait-for-kafka
        image: busybox:1.36
        command: ['sh', '-c', 'until nc -z kafka 9092; do echo waiting for kafka; sleep 2; done;']
      containers:
      - name: ingestor
        image: dexorder/ingestor:latest
        imagePullPolicy: Never
        env:
        - name: LOG_LEVEL
          value: info
        - name: CONFIG_PATH
          value: /config/config.yaml
        volumeMounts:
        - name: ingestor-config
          mountPath: /config
        - name: ingestor-secrets
          mountPath: /secrets
      volumes:
      - name: ingestor-config
        configMap:
          name: ingestor-config
      - name: ingestor-secrets
        secret:
          secretName: ingestor-secrets