feat: wire WebSocket event stream end-to-end (#1288)

bjcoombs · web-flow · commit 200080dde77d · 2026-02-27T18:24:00.000Z
* feat: wire WebSocket event stream end-to-end

Backend: Wire event stream components (OutboxEventSource + LocalFanOut)
in the unified binary, reusing the existing FA GORM connection for outbox
polling. The event router starts in the background and the WebSocket
handler is registered on GET /ws/events via the gateway server option.

Frontend: Replace the stub useEffect in useEventStream with an actual
WebSocket connection that connects to /ws/events, subscribes to all
channels, and delivers events through the existing handleEvent callback
for React Query cache invalidation. Includes exponential backoff
reconnection (max 5 attempts) and clean teardown on unmount.

Vite: Add /ws/events WebSocket proxy for the dev server so the
frontend can reach the backend during local development.

* fix: guard against negative EVENT_STREAM_BUFFER_SIZE

Clamp negative values to the default (256) to avoid a panic in
NewLocalFanOut when the env var is misconfigured.

* fix: default EVENT_STREAM_ENABLED to false and add reconnect jitter

Change EVENT_STREAM_ENABLED default from true to false so operators must
explicitly opt-in. Add random jitter to WebSocket reconnect backoff to
prevent thundering herd when backend restarts.

---------

Co-authored-by: Ben Coombs &lt;bjcoombs@users.noreply.github.com&gt;
diff --git a/cmd/meridian/main.go b/cmd/meridian/main.go
@@ -87,6 +87,8 @@ import (
 
 	// Gateway
 	"github.com/meridianhub/meridian/services/gateway"
+	"github.com/meridianhub/meridian/services/gateway/eventstream"
+	"github.com/meridianhub/meridian/services/gateway/eventstream/adapters"
 
 	// Shared platform
 	masterbootstrap "github.com/meridianhub/meridian/internal/bootstrap"
@@ -283,12 +285,20 @@ func run(logger *slog.Logger, grpcPort, httpPort int) error {
 	// ─── Start Gateway HTTP Server ───────────────────────────────────────
 
 	platformDSN := replaceDSNDatabase(baseDSN, "meridian_platform")
-	gwServer, err := wireGateway(grpcPort, httpPort, platformDSN, conns.gormDB("tenant"), logger)
+
+	eventRouter, extraGWOpts := wireEventStream(conns.gormDB("financial-accounting"), logger)
+
+	gwServer, err := wireGateway(grpcPort, httpPort, platformDSN, conns.gormDB("tenant"), logger, extraGWOpts...)
 	if err != nil {
 		return fmt.Errorf("gateway init: %w", err)
 	}
 
 	gatewayErrors := make(chan error, 1)
+
+	// Start event router in background (consumes from EventSource and publishes to FanOut).
+	routerCancel := startEventRouter(ctx, eventRouter, logger)
+	defer routerCancel()
+
 	go func() {
 		if err := gwServer.Start(context.Background()); err != nil {
 			gatewayErrors <- err
@@ -870,7 +880,7 @@ var serviceNames = []string{
 // wireGateway creates the gateway HTTP server with the Vanguard transcoder
 // routing REST/JSON, Connect, and gRPC-Web requests to the shared gRPC server
 // running on grpcPort.
-func wireGateway(grpcPort, httpPort int, databaseURL string, tenantDB *gorm.DB, logger *slog.Logger) (*gateway.Server, error) {
+func wireGateway(grpcPort, httpPort int, databaseURL string, tenantDB *gorm.DB, logger *slog.Logger, extraOpts ...gateway.ServerOption) (*gateway.Server, error) {
 	grpcTarget := fmt.Sprintf("localhost:%d", grpcPort)
 
 	authConfig := gateway.LoadAuthConfig()
@@ -931,9 +941,66 @@ func wireGateway(grpcPort, httpPort int, databaseURL string, tenantDB *gorm.DB,
 		return nil, fmt.Errorf("failed to create tenant resolver: %w", err)
 	}
 
+	// Append caller-provided options (e.g., event stream handler).
+	opts = append(opts, extraOpts...)
+
 	return gateway.NewServer(config, logger, tenantResolver, opts...), nil
 }
 
+// ─── Event Stream Wiring ─────────────────────────────────────────────────────
+
+// startEventRouter launches the event router in a background goroutine and
+// returns a cancel function. If router is nil the returned cancel is a no-op.
+func startEventRouter(ctx context.Context, router *eventstream.Router, logger *slog.Logger) context.CancelFunc {
+	if router == nil {
+		return func() {}
+	}
+	routerCtx, cancel := context.WithCancel(ctx)
+	go func() {
+		if err := router.Start(routerCtx); err != nil {
+			logger.Error("event router error", "error", err)
+		}
+	}()
+	logger.Info("event router started")
+	return cancel
+}
+
+// wireEventStream conditionally builds event stream components and returns the
+// Router (for lifecycle management) and any gateway ServerOptions that should be
+// applied. When EVENT_STREAM_ENABLED is false or unset, both return values are
+// nil/empty. Set EVENT_STREAM_ENABLED=true to enable.
+func wireEventStream(faDB *gorm.DB, logger *slog.Logger) (*eventstream.Router, []gateway.ServerOption) {
+	if !env.GetEnvAsBool("EVENT_STREAM_ENABLED", false) {
+		return nil, nil
+	}
+	router, wsHandler := buildUnifiedEventStreamComponents(faDB, logger)
+	logger.Info("event stream components initialized (outbox source, local fan-out)")
+	return router, []gateway.ServerOption{gateway.WithEventStreamHandler(wsHandler)}
+}
+
+// buildUnifiedEventStreamComponents creates event stream components for the unified binary.
+// Uses OutboxEventSource (polls shared outbox table) and LocalFanOut (in-process channels).
+// This matches dev/CI mode from the standalone gateway.
+//
+// The db parameter should be the financial-accounting GORM connection where
+// outbox events are written via OutboxPublisher. The connection is shared with
+// the service — no separate cleanup is needed.
+func buildUnifiedEventStreamComponents(db *gorm.DB, logger *slog.Logger) (*eventstream.Router, *eventstream.Handler) {
+	pollInterval := env.GetEnvAsDuration("OUTBOX_POLL_INTERVAL", 500*time.Millisecond)
+	source := adapters.NewOutboxEventSource(db, pollInterval, logger)
+
+	bufferSize := env.GetEnvAsInt("EVENT_STREAM_BUFFER_SIZE", 256)
+	if bufferSize < 0 {
+		bufferSize = 256
+	}
+	fanOut := adapters.NewLocalFanOut(bufferSize)
+
+	router := eventstream.NewRouter(source, fanOut)
+	handler := eventstream.NewHandler(router, logger)
+
+	return router, handler
+}
+
 // ─── Per-Service Database Connections ────────────────────────────────────────
 
 // replaceDSNDatabase replaces the database name in a PostgreSQL/CockroachDB DSN URL.
diff --git a/frontend/src/hooks/use-event-stream.ts b/frontend/src/hooks/use-event-stream.ts
@@ -1,10 +1,9 @@
-import { useEffect, useState, useCallback } from 'react'
+import { useEffect, useState, useCallback, useRef } from 'react'
 import { useQueryClient } from '@tanstack/react-query'
 import { useTenantSlug } from './use-tenant-context'
 
 /**
  * Represents a domain event from the backend event stream.
- * This stub supports the Phase 4 WebSocket integration.
  *
  * @example
  * {
@@ -56,10 +55,35 @@ export const EVENT_QUERY_MAP: Record<string, (event: DomainEvent) => unknown[][]
   ],
 }
 
+/** Wire format of a message received from the gateway WebSocket endpoint. */
+interface ServerMessage {
+  type: 'event' | 'subscribed' | 'error' | 'system'
+  subscription_id?: string
+  channel?: string
+  event?: {
+    event_id: string
+    event_type: string
+    aggregate_id?: string
+    aggregate_type?: string
+    tenant_id: string
+    correlation_id?: string
+    causation_id?: string
+    timestamp: string
+    payload: Record<string, unknown>
+  }
+  error_code?: string
+  error_message?: string
+  system_message?: string
+}
+
+const MAX_RECONNECT_ATTEMPTS = 5
+const BASE_RECONNECT_DELAY_MS = 1000
+
 /**
- * React hook for consuming real-time domain events.
- * This is a stub implementation that provides the complete interface for Phase 3 component development.
- * The actual WebSocket connection will be implemented in Phase 4 (PRD-025).
+ * React hook for consuming real-time domain events via WebSocket.
+ *
+ * Connects to the gateway's /ws/events endpoint and delivers events to
+ * the caller through the onEvent callback and React Query cache invalidation.
  *
  * Enforces tenant isolation - events from other tenants are silently ignored.
  *
@@ -72,44 +96,20 @@ export const EVENT_QUERY_MAP: Record<string, (event: DomainEvent) => unknown[][]
  *   onEvent: (event) => console.log('Event received:', event),
  *   autoInvalidate: true
  * })
- *
- * @phase 3 (Stub for Phase 4 WebSocket integration)
  */
 export function useEventStream(options: UseEventStreamOptions = {}) {
   const { autoInvalidate = true, onEvent, eventTypes } = options
   const queryClient = useQueryClient()
   const tenantSlug = useTenantSlug()
 
-  // Phase 4: setConnected will be called from WebSocket onopen/onclose handlers
-  // eslint-disable-next-line @typescript-eslint/no-unused-vars
   const [connected, setConnected] = useState(false)
   const [lastEvent, setLastEvent] = useState<DomainEvent | null>(null)
 
-  // Phase 4: WebSocket connection setup
-  // This useEffect will be implemented in Phase 4 (PRD-025) to connect to:
-  // const ws = new WebSocket(`wss://${tenantSlug}.api.meridian.io/events`)
-  // The connection will:
-  // - Set connected = true on open
-  // - Call handleEvent on message with decoded event data
-  // - Set error and connected = false on close/error
-  // - Clean up WebSocket on unmount
-  useEffect(() => {
-    // STUB: Phase 4 implementation
-    // For now, this hook provides the interface without connection logic.
-    // Phase 4 will establish a WebSocket connection here, call _handleEvent
-    // on messages, and set connected = true on open.
-
-    return () => {
-      // Phase 4: Cleanup WebSocket connection
-    }
-  }, [tenantSlug])
-
   /**
    * Internal handler for processing received events.
    * Validates tenant isolation, applies event type filtering, fires callbacks, and triggers cache invalidation.
-   * Prefixed with _ as it's unused until Phase 4 WebSocket integration.
    */
-  const _handleEvent = useCallback(
+  const handleEvent = useCallback(
     (event: DomainEvent) => {
       // Enforce tenant isolation - ignore events from other tenants
       if (event.tenantSlug !== tenantSlug) {
@@ -138,11 +138,91 @@ export function useEventStream(options: UseEventStreamOptions = {}) {
     [eventTypes, onEvent, autoInvalidate, queryClient, tenantSlug]
   )
 
-  // Ensure _handleEvent is referenced for Phase 4
-  void _handleEvent
+  // Stable ref for handleEvent so the WebSocket effect does not re-run when
+  // the callback identity changes.
+  const handleEventRef = useRef(handleEvent)
+  handleEventRef.current = handleEvent
+
+  useEffect(() => {
+    if (!tenantSlug) return
+
+    let ws: WebSocket | null = null
+    let reconnectTimeout: ReturnType<typeof setTimeout> | null = null
+    let reconnectAttempts = 0
+    let intentionalClose = false
+
+    const connect = () => {
+      const wsProtocol = window.location.protocol === 'https:' ? 'wss:' : 'ws:'
+      const wsUrl = `${wsProtocol}//${window.location.host}/ws/events`
+
+      ws = new WebSocket(wsUrl)
+
+      ws.onopen = () => {
+        setConnected(true)
+        reconnectAttempts = 0
+
+        // Subscribe to all channels; server-side auth controls access.
+        // Client-side eventTypes filtering is applied in handleEvent.
+        const subscribeMsg = {
+          type: 'subscribe',
+          id: crypto.randomUUID(),
+          channels: ['*'],
+        }
+        ws?.send(JSON.stringify(subscribeMsg))
+      }
+
+      ws.onmessage = (event) => {
+        try {
+          const msg: ServerMessage = JSON.parse(event.data)
+          if (msg.type === 'event' && msg.event) {
+            const domainEvent: DomainEvent = {
+              eventType: msg.event.event_type,
+              tenantSlug: msg.event.tenant_id,
+              payload: msg.event.payload ?? {},
+              timestamp: new Date(msg.event.timestamp),
+            }
+            handleEventRef.current(domainEvent)
+          }
+        } catch {
+          // Malformed message - silently ignore
+        }
+      }
+
+      ws.onclose = () => {
+        setConnected(false)
+
+        // Reconnect with exponential backoff + jitter unless intentionally closed
+        if (!intentionalClose && reconnectAttempts < MAX_RECONNECT_ATTEMPTS) {
+          const baseDelay = BASE_RECONNECT_DELAY_MS * Math.pow(2, reconnectAttempts)
+          const delay = baseDelay + Math.random() * baseDelay
+          reconnectTimeout = setTimeout(() => {
+            reconnectAttempts++
+            connect()
+          }, delay)
+        }
+      }
+
+      ws.onerror = () => {
+        // onclose will fire after onerror; reconnect logic lives there.
+        ws?.close()
+      }
+    }
+
+    connect()
+
+    return () => {
+      intentionalClose = true
+      if (reconnectTimeout) clearTimeout(reconnectTimeout)
+      if (ws) {
+        ws.onclose = null // Prevent reconnect on intentional close
+        ws.close()
+      }
+      setConnected(false)
+    }
+  }, [tenantSlug])
 
   return {
-    /** Whether the WebSocket is currently connected (Phase 4) */
+    /** Whether the WebSocket is currently connected */
     connected,
     /** The last event received, or null if no events yet */
     lastEvent,
diff --git a/frontend/vite.config.ts b/frontend/vite.config.ts
@@ -16,6 +16,7 @@ export default defineConfig({
   },
   server: {
     proxy: {
+      '/ws/events': { target: 'http://localhost:8090', ws: true },
       '/meridian.': { target: 'http://localhost:8090', changeOrigin: true },
       '/v1': { target: 'http://localhost:8090', changeOrigin: true },
     },
@@ -24,6 +25,7 @@ export default defineConfig({
     port: 5173,
     strictPort: true,
     proxy: {
+      '/ws/events': { target: 'http://localhost:8090', ws: true },
       '/meridian.': { target: 'http://localhost:8090', changeOrigin: true },
       '/v1': { target: 'http://localhost:8090', changeOrigin: true },
     },
