move topics to code

Author: geclos

.env.example

@@ -79,10 +79,6 @@ LAT_KAFKA_SSL=false
 # 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

apps/workers/src/server.ts

@@ -46,7 +46,6 @@ const initializeWorkers = async () => {
   // Create Redpanda publisher (async initialization)
   const eventsPublisher = await createRedpandaEventsPublisher({
     kafka: kafkaClient,
-    config: kafkaConfig,
   })
 
   // Create outbox consumer (polls Postgres outbox and publishes to Redpanda)
@@ -62,7 +61,7 @@ const initializeWorkers = async () => {
   // Create Redpanda event consumer (consumes from Redpanda and processes events)
   const redpandaConsumer = createRedpandaEventsConsumer({
     kafka: kafkaClient,
-    config: kafkaConfig,
+    groupId: kafkaConfig.groupId,
   })
 
   // Start consumers

docker-compose.yml

@@ -63,7 +63,7 @@ services:
     volumes:
       - redpanda_data:/var/lib/redpanda/data
     healthcheck:
-      test: ["CMD-SHELL", "rpk cluster health --brokers localhost:29092"]
+      test: ["CMD-SHELL", "curl -sf http://localhost:9644/v1/brokers || exit 1"]
       interval: 5s
       timeout: 10s
       retries: 12
@@ -75,12 +75,12 @@ services:
       redpanda:
         condition: service_healthy
     command:
-      - rpk
       - topic
       - create
       - domain-events
       - domain-events-dlq
-      - --brokers=redpanda:29092
+      - -X
+      - brokers=redpanda:29092
       - --partitions=1
       - --replicas=1
     restart: "no"

packages/platform/queue-redpanda/src/config.test.ts

@@ -6,8 +6,6 @@ const ENV_KEYS = [
   "LAT_KAFKA_BROKERS",
   "LAT_KAFKA_CLIENT_ID",
   "LAT_KAFKA_CONSUMER_GROUP_ID",
-  "LAT_KAFKA_EVENTS_TOPIC",
-  "LAT_KAFKA_DLQ_TOPIC",
   "LAT_KAFKA_SSL",
   "LAT_KAFKA_SASL_USERNAME",
   "LAT_KAFKA_SASL_PASSWORD",
@@ -17,8 +15,6 @@ const BASE_ENV: Record<string, string> = {
   LAT_KAFKA_BROKERS: "localhost:9092,localhost:9093",
   LAT_KAFKA_CLIENT_ID: "test-client",
   LAT_KAFKA_CONSUMER_GROUP_ID: "test-group",
-  LAT_KAFKA_EVENTS_TOPIC: "domain-events",
-  LAT_KAFKA_DLQ_TOPIC: "domain-events-dlq",
 }
 
 function setEnv(vars: Record<string, string>) {
@@ -43,8 +39,6 @@ describe("loadKafkaConfig", () => {
     expect(config.brokers).toEqual(["localhost:9092", "localhost:9093"])
     expect(config.clientId).toBe("test-client")
     expect(config.groupId).toBe("test-group")
-    expect(config.eventsTopic).toBe("domain-events")
-    expect(config.dlqTopic).toBe("domain-events-dlq")
     expect(config.ssl).toBeUndefined()
     expect(config.sasl).toBeUndefined()
   })

packages/platform/queue-redpanda/src/config.ts

@@ -11,8 +11,6 @@ export const loadKafkaConfig = (): Effect.Effect<KafkaConfig, KafkaSaslWithoutTl
     const brokers = yield* parseEnv("LAT_KAFKA_BROKERS", "string")
     const clientId = yield* parseEnv("LAT_KAFKA_CLIENT_ID", "string")
     const groupId = yield* parseEnv("LAT_KAFKA_CONSUMER_GROUP_ID", "string")
-    const eventsTopic = yield* parseEnv("LAT_KAFKA_EVENTS_TOPIC", "string")
-    const dlqTopic = yield* parseEnv("LAT_KAFKA_DLQ_TOPIC", "string")
 
     // Optional: SSL/TLS encryption
     const ssl = yield* parseEnvOptional("LAT_KAFKA_SSL", "boolean")
@@ -34,8 +32,6 @@ export const loadKafkaConfig = (): Effect.Effect<KafkaConfig, KafkaSaslWithoutTl
       clientId,
       brokers: brokers.split(","),
       groupId,
-      eventsTopic,
-      dlqTopic,
       ssl,
       sasl:
         saslUsername && saslPassword

packages/platform/queue-redpanda/src/consumer.ts

@@ -2,11 +2,11 @@ import type { EventEnvelope } from "@domain/events"
 import { Effect } from "effect"
 import type { EachMessagePayload } from "kafkajs"
 import { z } from "zod"
-import type { KafkaConfig } from "./types.ts"
+import { Topics } from "./topics.ts"
 
 export interface RedpandaEventsConsumerConfig {
-  readonly config: KafkaConfig
   readonly kafka: import("kafkajs").Kafka
+  readonly groupId: string
 }
 
 export const DomainEventSchema = z.object({
@@ -34,15 +34,15 @@ const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms))
 
 export const createRedpandaEventsConsumer = (config: RedpandaEventsConsumerConfig) => {
   const consumer = config.kafka.consumer({
-    groupId: config.config.groupId,
+    groupId: config.groupId,
   })
 
   let isRunning = false
 
   const start = async (handler: EventHandler): Promise<void> => {
     await consumer.connect()
     await consumer.subscribe({
-      topic: config.config.eventsTopic,
+      topic: Topics.domainEvents,
     })
 
     isRunning = true
@@ -104,7 +104,7 @@ export const createRedpandaEventsConsumer = (config: RedpandaEventsConsumerConfi
   return {
     start,
     stop,
-    pause: () => consumer.pause([{ topic: config.config.eventsTopic }]),
-    resume: () => consumer.resume([{ topic: config.config.eventsTopic }]),
+    pause: () => consumer.pause([{ topic: Topics.domainEvents }]),
+    resume: () => consumer.resume([{ topic: Topics.domainEvents }]),
   }
 }

packages/platform/queue-redpanda/src/index.ts

@@ -4,6 +4,9 @@ import { RedpandaQueueAdapterTag } from "./types.ts"
 
 export { createKafkaClient, createKafkaClientEffect } from "./client.ts"
 
+export { Topics } from "./topics.ts"
+export type { TopicName } from "./topics.ts"
+
 export {
   createRedpandaEventsPublisher,
   createRedpandaEventsPublisherEffect,

packages/platform/queue-redpanda/src/producer.ts

@@ -1,8 +1,8 @@
 import type { EventEnvelope, EventsPublisher } from "@domain/events"
 import { Data, Effect } from "effect"
 import type { Kafka } from "kafkajs"
+import { Topics } from "./topics.ts"
 import { KafkaClientError } from "./types.ts"
-import type { KafkaConfig } from "./types.ts"
 
 export class RedpandaProducerError extends Data.TaggedError("RedpandaProducerError")<{
   readonly cause: unknown
@@ -11,7 +11,6 @@ export class RedpandaProducerError extends Data.TaggedError("RedpandaProducerErr
 
 export interface RedpandaEventsPublisherConfig {
   readonly kafka: Kafka
-  readonly config: KafkaConfig
 }
 
 export const mapEnvelopeToMessage = (envelope: EventEnvelope) => {
@@ -63,7 +62,7 @@ export const createRedpandaEventsPublisherEffect = (
         yield* Effect.tryPromise({
           try: () =>
             producer.send({
-              topic: config.config.eventsTopic,
+              topic: Topics.domainEvents,
               acks: -1,
               messages: [message],
             }),

packages/platform/queue-redpanda/src/topics.ts

@@ -0,0 +1,6 @@
+export const Topics = {
+  domainEvents: "domain-events",
+  domainEventsDlq: "domain-events-dlq",
+} as const
+
+export type TopicName = (typeof Topics)[keyof typeof Topics]

packages/platform/queue-redpanda/src/types.ts

@@ -9,8 +9,6 @@ export interface KafkaConfig {
   readonly clientId: string
   readonly brokers: string[]
   readonly groupId: string
-  readonly eventsTopic: string
-  readonly dlqTopic: string
   readonly ssl: boolean | undefined
   readonly sasl: { readonly mechanism: "plain"; readonly username: string; readonly password: string } | undefined
 }