feat: s/redis/redpanda

Redis won't scale at real world scale for the ingestion pipeline, so
we've replacing it with redpanda. We keep redis for caching.

Author: geclos

.env.example

@@ -66,6 +66,23 @@ LAT_WEAVIATE_GRPC_PORT=50051
 LAT_REDIS_HOST=localhost
 LAT_REDIS_PORT=6379
 
+# Redpanda/Kafka Configuration
+LAT_KAFKA_BROKERS=localhost:9092
+LAT_KAFKA_CLIENT_ID=latitude-app
+# SECURITY: When SASL authentication is enabled (LAT_KAFKA_SASL_USERNAME/PASSWORD),
+# LAT_KAFKA_SSL must be set to true in production environments to prevent
+# transmitting credentials in plaintext.
+LAT_KAFKA_SSL=false
+# LAT_KAFKA_SASL_USERNAME=
+# LAT_KAFKA_SASL_PASSWORD=
+
+# Consumer Configuration
+LAT_KAFKA_CONSUMER_GROUP_ID=latitude-event-processors-v1
+
+# Topic Configuration
+LAT_KAFKA_EVENTS_TOPIC=domain-events
+LAT_KAFKA_DLQ_TOPIC=domain-events-dlq
+
 # Object Storage
 LAT_STORAGE_DRIVER=fs
 # Absolute path in your laptop

.gitignore

@@ -129,11 +129,3 @@ PRD.md
 progress.txt
 
 packages/cli/prompts
-
-.localstorage*
-docs/plans
-todos/
-compound-engineering.*
-
-.mcp.json
-opencode.json

AGENTS.md

@@ -888,7 +888,7 @@ Docker is used for local infrastructure services. Start them before running apps
 
 ```bash
 sudo dockerd &>/dev/null &  # if Docker daemon not already running
-sudo docker compose up -d postgres clickhouse redis mailpit
+sudo docker compose up -d postgres clickhouse redis mailpit redpanda
 ```
 
 ### Database setup

apps/workers/package.json

@@ -13,13 +13,11 @@
     "check:fix": "biome check --fix src"
   },
   "dependencies": {
-    "@platform/cache-redis": "workspace:*",
     "@platform/db-postgres": "workspace:*",
     "@platform/events-outbox": "workspace:*",
-    "@platform/queue-bullmq": "workspace:*",
+    "@platform/queue-redpanda": "workspace:*",
     "@repo/observability": "workspace:*",
     "@types/pg": "catalog:",
-    "bullmq": "catalog:",
     "dotenv": "catalog:",
     "effect": "catalog:",
     "pg": "catalog:",

apps/workers/src/clients.ts

@@ -1,18 +1,8 @@
-import type { RedisConnection } from "@platform/cache-redis"
-import { createRedisConnection } from "@platform/cache-redis"
 import { createPostgresPool } from "@platform/db-postgres"
 import type { Pool } from "pg"
 
-let redisConnectionInstance: RedisConnection | undefined
 let pgPoolInstance: Pool | undefined
 
-export const getRedisConnection = (): RedisConnection => {
-  if (!redisConnectionInstance) {
-    redisConnectionInstance = createRedisConnection()
-  }
-  return redisConnectionInstance
-}
-
 export const getPostgresPool = (maxConnections?: number): Pool => {
   if (!pgPoolInstance) {
     pgPoolInstance = createPostgresPool(maxConnections ? { maxConnections } : undefined)

apps/workers/src/server.ts

@@ -1,11 +1,16 @@
 import { existsSync } from "node:fs"
 import { fileURLToPath } from "node:url"
 import { createPollingOutboxConsumer } from "@platform/events-outbox"
-import { createBullmqEventsPublisher } from "@platform/queue-bullmq"
+import {
+  createKafkaClient,
+  createRedpandaEventsConsumer,
+  createRedpandaEventsPublisher,
+  loadKafkaConfig,
+} from "@platform/queue-redpanda"
 import { createLogger } from "@repo/observability"
 import { config as loadDotenv } from "dotenv"
-import { getPostgresPool, getRedisConnection } from "./clients.ts"
-import { createEventsWorker } from "./workers/events.ts"
+import { Effect } from "effect"
+import { getPostgresPool } from "./clients.ts"
 
 const nodeEnv = process.env.NODE_ENV || "development"
 const envFilePath = fileURLToPath(new URL(`../../../.env.${nodeEnv}`, import.meta.url))
@@ -14,31 +19,75 @@ if (existsSync(envFilePath)) {
   loadDotenv({ path: envFilePath, quiet: true })
 }
 
-const redisConnection = getRedisConnection()
 const pgPool = getPostgresPool(10)
-const { queue: eventsQueue, worker: eventsWorker } = createEventsWorker(redisConnection)
-const eventsPublisher = createBullmqEventsPublisher({ queue: eventsQueue })
-const outboxConsumer = createPollingOutboxConsumer(
-  {
-    pool: pgPool,
-    pollIntervalMs: 1000,
-    batchSize: 100,
-  },
-  eventsPublisher,
-)
 
 const logger = createLogger("workers")
 
-eventsWorker.on("ready", () => {
+// Simple event handler that logs events (placeholder for actual side effect processing)
+const eventHandler = {
+  handle: (event: { id: string; event: { name: string; organizationId: string } }): Effect.Effect<
+    void,
+    unknown,
+    never
+  > =>
+    Effect.gen(function* () {
+      yield* Effect.logInfo(
+        `Processing event ${event.id} of type ${event.event.name} for org ${event.event.organizationId}`,
+      )
+    }),
+}
+
+// Initialize workers asynchronously
+const initializeWorkers = async () => {
+  // Load Redpanda configuration
+  const kafkaConfig = Effect.runSync(loadKafkaConfig())
+  const kafkaClient = createKafkaClient(kafkaConfig)
+
+  // Create Redpanda publisher (async initialization)
+  const eventsPublisher = await createRedpandaEventsPublisher({
+    kafka: kafkaClient,
+    config: kafkaConfig,
+  })
+
+  // Create outbox consumer (polls Postgres outbox and publishes to Redpanda)
+  const outboxConsumer = createPollingOutboxConsumer(
+    {
+      pool: pgPool,
+      pollIntervalMs: 1000,
+      batchSize: 100,
+    },
+    eventsPublisher,
+  )
+
+  // Create Redpanda event consumer (consumes from Redpanda and processes events)
+  const redpandaConsumer = createRedpandaEventsConsumer({
+    kafka: kafkaClient,
+    config: kafkaConfig,
+  })
+
+  // Start consumers
   outboxConsumer.start()
+  await redpandaConsumer.start(eventHandler)
+  logger.info("workers ready - outbox consumer and Redpanda consumer started")
+
+  return { outboxConsumer, redpandaConsumer }
+}
 
-  logger.info("workers ready and outbox consumer started")
+const workersPromise = initializeWorkers().catch((error) => {
+  logger.error("Failed to initialize workers", error)
+  process.exit(1)
 })
 
+// Graceful shutdown
 process.on("SIGINT", async () => {
-  await outboxConsumer.stop()
+  logger.info("shutting down workers...")
+  try {
+    const { outboxConsumer, redpandaConsumer } = await workersPromise
+    await outboxConsumer.stop()
+    await redpandaConsumer.stop()
+  } catch (error) {
+    logger.error("Error during shutdown (workers may not have started)", error)
+  }
   await pgPool.end()
-  await eventsQueue.close()
-  await eventsWorker.close()
   process.exit(0)
 })

apps/workers/src/workers/events.ts

@@ -48,7 +48,45 @@ services:
       - "1025:1025"
       - "8025:8025"
 
+  redpanda:
+    image: docker.redpanda.com/redpandadata/redpanda:v24.3.14
+    command:
+      - redpanda
+      - start
+      - --kafka-addr=internal://0.0.0.0:29092,external://0.0.0.0:9092
+      - --advertise-kafka-addr=internal://redpanda:29092,external://localhost:9092
+      - --mode=dev-container
+      - --smp=1
+      - --default-log-level=warn
+    ports:
+      - "9092:9092"
+    volumes:
+      - redpanda_data:/var/lib/redpanda/data
+    healthcheck:
+      test: ["CMD-SHELL", "rpk cluster health --brokers localhost:29092"]
+      interval: 5s
+      timeout: 10s
+      retries: 12
+      start_period: 15s
+
+  redpanda-init:
+    image: docker.redpanda.com/redpandadata/redpanda:v24.3.14
+    depends_on:
+      redpanda:
+        condition: service_healthy
+    command:
+      - rpk
+      - topic
+      - create
+      - domain-events
+      - domain-events-dlq
+      - --brokers=redpanda:29092
+      - --partitions=1
+      - --replicas=1
+    restart: "no"
+
 volumes:
   postgres_data:
   clickhouse_data:
   weaviate-data:
+  redpanda_data: