feat(messaging): add NATS JetStream event bus with durable consumers

.env.example

@@ -18,6 +18,17 @@ REDIS_ADDR=redis:6379
 REDIS_PASSWORD=
 REDIS_DB=0
 
+# --- Event transport ---
+# Which messaging transport carries inter-service events. Redis remains the
+# system of record (sessions, matchmaking queue, leaderboard, room cache)
+# regardless of this setting; EVENT_BUS only swaps the event-delivery layer.
+#   nats  -> NATS JetStream: durable, replayable, at-least-once (default in compose)
+#   redis -> Redis Pub/Sub:  fire-and-forget, at-most-once (original behavior)
+EVENT_BUS=nats
+NATS_URL=nats://nats:4222
+# Bounded worker pool size for the JetStream consumers (per service).
+EVENT_WORKERS=8
+
 # --- Service URLs (as seen from the API gateway) ---
 PLAYER_SERVICE_URL=http://player:8081
 MATCHMAKING_SERVICE_URL=http://matchmaking:8082

cmd/leaderboard/main.go

@@ -43,7 +43,15 @@ func main() {
 	}
 
 	m := metrics.New(cfg.ServiceName)
-	bus := events.NewRedisBus(rdb, log)
+	bus, err := events.NewBus(events.Config{
+		Transport:   cfg.EventBus,
+		DurableName: cfg.ServiceName,
+		Workers:     cfg.EventWorkers,
+	}, rdb, cfg.NATSURL, m, log)
+	if err != nil {
+		log.Fatal("failed to init event bus", zap.Error(err))
+	}
+	defer func() { _ = bus.Close() }()
 	svc := leaderboard.NewService(leaderboard.NewStore(rdb), bus, m, log)
 	handler := leaderboard.NewHandler(svc)
 

cmd/matchmaking/main.go

@@ -43,7 +43,15 @@ func main() {
 	}
 
 	m := metrics.New(cfg.ServiceName)
-	bus := events.NewRedisBus(rdb, log)
+	bus, err := events.NewBus(events.Config{
+		Transport:   cfg.EventBus,
+		DurableName: cfg.ServiceName,
+		Workers:     cfg.EventWorkers,
+	}, rdb, cfg.NATSURL, m, log)
+	if err != nil {
+		log.Fatal("failed to init event bus", zap.Error(err))
+	}
+	defer func() { _ = bus.Close() }()
 	svc := matchmaking.NewService(
 		matchmaking.NewQueue(rdb),
 		matchmaking.NewRoomStore(rdb, cfg.RoomTTL),

cmd/websocket/main.go

@@ -51,7 +51,16 @@ func main() {
 
 	ctx, stop := server.ShutdownContext()
 	defer stop()
-	bridge := wsgateway.NewBridge(hub, events.NewRedisBus(rdb, log), log)
+	bus, err := events.NewBus(events.Config{
+		Transport:   cfg.EventBus,
+		DurableName: cfg.ServiceName,
+		Workers:     cfg.EventWorkers,
+	}, rdb, cfg.NATSURL, m, log)
+	if err != nil {
+		log.Fatal("failed to init event bus", zap.Error(err))
+	}
+	defer func() { _ = bus.Close() }()
+	bridge := wsgateway.NewBridge(hub, bus, log)
 	go func() {
 		if err := bridge.Run(ctx); err != nil {
 			log.Fatal("event bridge failed", zap.Error(err))

docker-compose.yml

@@ -10,6 +10,13 @@ x-service-env: &service-env
   REDIS_ADDR: redis:6379
   REDIS_PASSWORD: ${REDIS_PASSWORD:-}
   ALLOWED_ORIGINS: ${ALLOWED_ORIGINS:-*}
+  # Event transport. Redis stays the system of record (sessions, queue,
+  # leaderboard, room cache); NATS JetStream carries inter-service events with
+  # durability, replay and at-least-once delivery. Set EVENT_BUS=redis to fall
+  # back to the original Pub/Sub transport — no code change, just config.
+  EVENT_BUS: ${EVENT_BUS:-nats}
+  NATS_URL: ${NATS_URL:-nats://nats:4222}
+  EVENT_WORKERS: ${EVENT_WORKERS:-8}
   # Distributed tracing: every service ships spans to the OTel Collector, which
   # forwards them to Jaeger. Set OTEL_ENABLED=false to turn tracing off entirely.
   OTEL_ENABLED: ${OTEL_ENABLED:-true}
@@ -48,6 +55,25 @@ services:
       timeout: 3s
       retries: 10
 
+  # NATS with JetStream enabled (-js): the durable event transport. Streams are
+  # created idempotently by the services on boot. Storage is persisted to a
+  # volume so events (and consumer cursors) survive container restarts —
+  # enabling replay and at-least-once redelivery. :8222 serves the monitoring
+  # endpoint used for the healthcheck.
+  nats:
+    image: nats:2.10-alpine
+    command: ["-js", "-sd", "/data", "-m", "8222"]
+    ports:
+      - "4222:4222" # client
+      - "8222:8222" # monitoring
+    volumes:
+      - natsdata:/data
+    healthcheck:
+      test: ["CMD", "wget", "-qO-", "http://localhost:8222/healthz"]
+      interval: 5s
+      timeout: 3s
+      retries: 10
+
   gateway:
     build:
       context: .
@@ -113,6 +139,8 @@ services:
     depends_on:
       redis:
         condition: service_healthy
+      nats:
+        condition: service_healthy
       otel-collector:
         condition: service_started
     healthcheck:
@@ -129,6 +157,8 @@ services:
     depends_on:
       redis:
         condition: service_healthy
+      nats:
+        condition: service_healthy
       otel-collector:
         condition: service_started
     healthcheck:
@@ -147,6 +177,8 @@ services:
     depends_on:
       redis:
         condition: service_healthy
+      nats:
+        condition: service_healthy
       otel-collector:
         condition: service_started
     healthcheck:
@@ -209,3 +241,4 @@ services:
 
 volumes:
   pgdata:
+  natsdata: