feat: add dispatcher integration for transcription fan-out (#45)

* feat: add dispatcher integration for transcription fan-out

* docs: update dispatcher configuration options

Author: rpurdel

.beads/metadata.json

@@ -0,0 +1,4 @@
+{
+  "database": "beads.db",
+  "jsonl_export": "issues.jsonl"
+}

DISPATCHER_INTEGRATION.md

@@ -94,30 +94,47 @@ containerClientWs.addEventListener('message', (event) => {
 
 ## Configuration
 
-### 1. Enable Dispatcher in Worker Config
+### 1. Add Service Binding
 
-Edit `wrangler-container.jsonc`:
+Add the `TRANSCRIPTION_DISPATCHER` service binding to your `wrangler.jsonc`:
 
 ```jsonc
 {
   "services": [
     {
       "binding": "TRANSCRIPTION_DISPATCHER",
-      "service": "your-dispatcher-worker-name",
-      "environment": "production"
+      "service": "transcription-dispatcher"
     }
   ]
 }
 ```
 
-### 2. Connect with Dispatcher Enabled
+### 2. Enable Dispatcher
+
+The dispatcher can be enabled in two ways:
+
+**Option A: Environment Variable (recommended for global enable)**
+
+Add `USE_DISPATCHER` to your wrangler.jsonc vars:
+
+```jsonc
+{
+  "vars": {
+    "USE_DISPATCHER": "true"
+  }
+}
+```
+
+**Option B: Query Parameter (per-request control)**
 
 Add `useDispatcher=true` query parameter:
 
 ```
-wss://your-worker.workers.dev/transcribe?sessionId=test&transcribe=true&sendBack=true&useDispatcher=true
+wss://your-worker.workers.dev/transcribe?sessionId=test&useDispatcher=true
 ```
 
+**Precedence:** Query parameter overrides the environment variable. If neither is set, dispatcher is disabled by default.
+
 ### 3. Implement Dispatcher Worker
 
 Your dispatcher worker must implement the `dispatch()` RPC method:
@@ -242,15 +259,24 @@ interface DispatcherTranscriptionMessage {
 ### Without Dispatcher
 
 ```bash
-# Media server receives transcripts directly
-wscat -c "wss://your-worker.workers.dev/transcribe?sessionId=test&transcribe=true&sendBack=true"
+# Media server receives transcripts directly (dispatcher disabled)
+wscat -c "wss://your-worker.workers.dev/transcribe?sessionId=test&sendBack=true"
 ```
 
-### With Dispatcher
+### With Dispatcher (via query param)
 
 ```bash
 # Media server receives transcripts + dispatcher gets notified
-wscat -c "wss://your-worker.workers.dev/transcribe?sessionId=test&transcribe=true&sendBack=true&useDispatcher=true"
+wscat -c "wss://your-worker.workers.dev/transcribe?sessionId=test&sendBack=true&useDispatcher=true"
+```
+
+### With Dispatcher (via env var)
+
+If `USE_DISPATCHER=true` is set in wrangler.jsonc, all connections will dispatch:
+
+```bash
+# Dispatcher enabled globally via env var
+wscat -c "wss://your-worker.workers.dev/transcribe?sessionId=test&sendBack=true"
 ```
 
 Check dispatcher logs:

worker/env.d.ts

@@ -1,12 +1,35 @@
 // Environment types for the Cloudflare Worker
 
+import type { TranscriptionDispatcher } from './index';
+
 interface Env {
 	// Durable Object binding for the container
 	TRANSCRIBER: DurableObjectNamespace;
 
+	// Service bindings
+	TRANSCRIPTION_DISPATCHER?: Service<TranscriptionDispatcher>;
+
+	// Durable Object for auto-scaling
+	CONTAINER_COORDINATOR: DurableObjectNamespace;
+
 	// Environment variables
 	OPENAI_API_KEY: string;
 	OPENAI_MODEL?: string;
+	GEMINI_API_KEY?: string;
+	DEEPGRAM_API_KEY?: string;
+	DEEPGRAM_MODEL?: string;
+	DEEPGRAM_DETECT_LANGUAGE?: string;
+	DEEPGRAM_INCLUDE_LANGUAGE?: string;
+	DEEPGRAM_PUNCTUATE?: string;
+	DEEPGRAM_ENCODING?: string;
+	PROVIDERS_PRIORITY?: string;
 	FORCE_COMMIT_TIMEOUT?: string;
 	DEBUG?: string;
+	ROUTING_MODE?: string;
+	CONTAINER_POOL_SIZE?: string;
+	MAX_CONNECTIONS_PER_CONTAINER?: string;
+	MIN_CONTAINERS?: string;
+	SCALE_DOWN_IDLE_TIME?: string;
+	TRANSLATION_MIXING_MODE?: string;
+	USE_DISPATCHER?: string;
 }

worker/index.ts

@@ -33,14 +33,16 @@ export interface TranscriptionDispatcher extends WorkerEntrypoint<Env> {
  * Transcription message from container
  */
 interface TranscriptionMessage {
-	type: 'transcription' | 'interim_transcription';
+	type: 'transcription-result';
+	is_interim: boolean;
 	participant: {
 		id?: string;
 	};
 	transcript: Array<{
 		text: string;
 	}>;
 	timestamp: number;
+	language?: string;
 }
 
 /**
@@ -168,6 +170,112 @@ async function selectContainerInstance(request: Request, env: Env): Promise<stri
 	}
 }
 
+/**
+ * Handle WebSocket connection with dispatcher interception.
+ * Creates a proxy between client and container, dispatching transcriptions asynchronously.
+ */
+async function handleWebSocketWithDispatcher(
+	request: Request,
+	container: ReturnType<typeof getContainer>,
+	env: Env,
+	ctx: ExecutionContext,
+	sessionId: string,
+): Promise<Response> {
+	// Create WebSocket pair for the client
+	const clientPair = new WebSocketPair();
+	const [clientWs, serverWs] = Object.values(clientPair);
+
+	// Accept the server side of the client connection
+	serverWs.accept();
+
+	// Forward the upgrade request to the container
+	const containerResponse = await container.fetch(request);
+
+	if (containerResponse.status !== 101 || !containerResponse.webSocket) {
+		// Container didn't upgrade - return error to client
+		serverWs.close(1011, 'Container failed to upgrade WebSocket');
+		return containerResponse;
+	}
+
+	const containerWs = containerResponse.webSocket;
+	containerWs.accept();
+
+	const dispatcher = env.TRANSCRIPTION_DISPATCHER!;
+
+	// Pipe: client → container (upstream, no interception needed)
+	serverWs.addEventListener('message', (event) => {
+		if (containerWs.readyState === WebSocket.READY_STATE_OPEN) {
+			containerWs.send(event.data);
+		}
+	});
+
+	// Pipe: container → client (downstream, intercept for dispatcher)
+	containerWs.addEventListener('message', (event) => {
+		// Forward to client immediately
+		if (serverWs.readyState === WebSocket.READY_STATE_OPEN) {
+			serverWs.send(event.data);
+		}
+
+		// Dispatch transcriptions asynchronously (non-blocking)
+		if (typeof event.data === 'string') {
+			try {
+				const data = JSON.parse(event.data) as TranscriptionMessage;
+				if (data.type === 'transcription-result' && !data.is_interim) {
+					const dispatcherMessage: DispatcherTranscriptionMessage = {
+						sessionId,
+						endpointId: data.participant?.id || 'unknown',
+						text: data.transcript.map((t) => t.text).join(' '),
+						timestamp: data.timestamp,
+						language: data.language,
+					};
+
+					// Fire and forget - don't block the client
+					dispatcher
+						.dispatch(dispatcherMessage)
+						.then((response) => {
+							if (!response.success || response.errors) {
+								console.error('Dispatcher error:', {
+									message: response.message,
+									errors: response.errors,
+								});
+							}
+						})
+						.catch((error) => {
+							const msg = error instanceof Error ? error.message : String(error);
+							console.error('Dispatcher RPC failed:', msg);
+						});
+				}
+			} catch {
+				// Not JSON or parse error - ignore, still forwarded to client
+			}
+		}
+	});
+
+	// Handle close events
+	serverWs.addEventListener('close', (event) => {
+		containerWs.close(event.code, event.reason);
+	});
+
+	containerWs.addEventListener('close', (event) => {
+		serverWs.close(event.code, event.reason);
+	});
+
+	// Handle errors
+	serverWs.addEventListener('error', () => {
+		containerWs.close(1011, 'Client WebSocket error');
+	});
+
+	containerWs.addEventListener('error', () => {
+		serverWs.close(1011, 'Container WebSocket error');
+	});
+
+	// Return the client WebSocket
+	return new Response(null, {
+		status: 101,
+		webSocket: clientWs,
+	});
+}
+
 export default {
 	async fetch(request: Request, env: Env, ctx: ExecutionContext): Promise<Response> {
 		const url = new URL(request.url);
@@ -185,7 +293,11 @@ export default {
 			}
 		}
 
-		const useDispatcher = url.searchParams.get('useDispatcher') === 'true';
+		// Check query param first, fall back to env var
+		const useDispatcherParam = url.searchParams.get('useDispatcher');
+		const useDispatcher = useDispatcherParam !== null
+			? useDispatcherParam === 'true'
+			: env.USE_DISPATCHER === 'true';
 		const sessionId = url.searchParams.get('sessionId') || 'unknown';
 
 		// Select which container instance to use based on routing strategy
@@ -194,13 +306,9 @@ export default {
 		// Get the container instance
 		const container = getContainer(env.TRANSCRIBER, containerInstanceId);
 
-	// Start the container and wait for ports to be ready
-	// This is required for the fetch to work properly
-	await container.startAndWaitForPorts();
-
-		// For now, just forward all requests directly to the container
-		// TODO: Implement WebSocket interception for dispatcher fan-out
-		// This requires more complex WebSocket handling with Cloudflare Containers
+		// Start the container and wait for ports to be ready
+		// This is required for the fetch to work properly
+		await container.startAndWaitForPorts();
 
 		// Report connection tracking for autoscale mode
 		const routingMode = env.ROUTING_MODE || 'session';
@@ -220,13 +328,14 @@ export default {
 						console.error('Failed to report connection opened:', error);
 					}),
 			);
+		}
 
-			// Note: We can't easily detect when WebSocket closes from here
-			// The container would need to report back, or we'd need WebSocket interception
-			// For now, rely on scale-down idle timeout
+		// If dispatcher is enabled and this is a WebSocket upgrade, intercept the connection
+		if (useDispatcher && upgradeHeader === 'websocket' && env.TRANSCRIPTION_DISPATCHER) {
+			return handleWebSocketWithDispatcher(request, container, env, ctx, sessionId);
 		}
 
-		// Forward request directly to container
+		// Forward request directly to container (pass-through mode)
 		return container.fetch(request);
 
 	},