Feature/queue dispatch (#48)

* feat: add queue support for dispatcher, fallback to RPC

* doc: update CF integration

Author: rpurdel

CLOUDFLARE_DEPLOYMENT.md

@@ -166,37 +166,72 @@ The Worker can fan out transcriptions to a separate Transcription Dispatcher wor
 2. Container decodes audio and gets transcriptions from OpenAI
 3. Worker intercepts transcriptions and:
    - Forwards to media server via WebSocket (low latency path)
-   - Asynchronously dispatches to Dispatcher worker via RPC (doesn't block)
+   - Asynchronously dispatches to Dispatcher worker (doesn't block)
+
+**Dispatch Methods (in order of preference):**
+
+1. **Cloudflare Queue** (recommended) - Messages sent to a queue, consumed by Dispatcher
+   - Doesn't count against the 50 subrequest limit per WebSocket connection
+   - Better for high-throughput scenarios
+
+2. **Service Binding RPC** (fallback) - Direct RPC call to Dispatcher worker
+   - Each dispatch counts as a subrequest (50 limit per WebSocket lifetime)
+   - Use only for low-traffic scenarios
+
+**Setup with Queue (Recommended):**
+
+1. Deploy your Transcription Dispatcher worker with queue consumer configured
+2. Update `wrangler.jsonc` to add a queue producer:
+   ```jsonc
+   "queues": {
+     "producers": [{
+       "binding": "TRANSCRIPTION_QUEUE",
+       "queue": "transcription-dispatch-queue-staging"
+     }]
+   }
+   ```
+3. Set `USE_DISPATCHER` environment variable:
+   ```jsonc
+   "vars": {
+     "USE_DISPATCHER": "true"
+   }
+   ```
+4. Or enable per-connection with query parameter:
+   ```
+   wss://your-worker.workers.dev/transcribe?sessionId=test&transcribe=true&sendBack=true&useDispatcher=true
+   ```
+
+**Setup with RPC (Fallback):**
 
-**Setup:**
 1. Deploy your Transcription Dispatcher worker (must implement the `dispatch()` RPC method)
-2. Update `wrangler-container.jsonc` to point to your dispatcher:
+2. Update `wrangler.jsonc` to add a service binding:
    ```jsonc
    "services": [{
      "binding": "TRANSCRIPTION_DISPATCHER",
-     "service": "your-dispatcher-worker-name",
-     "environment": "production"
+     "service": "transcription-dispatcher"
    }]
    ```
-3. Connect with `useDispatcher=true`:
-   ```
-   wss://your-worker.workers.dev/transcribe?sessionId=test&transcribe=true&sendBack=true&useDispatcher=true
-   ```
+3. Enable with `USE_DISPATCHER=true` or `useDispatcher=true` query param
 
-**Dispatcher Interface:**
-Your dispatcher worker must implement:
-```typescript
-export interface TranscriptionDispatcher extends WorkerEntrypoint<Env> {
-  dispatch(message: DispatcherTranscriptionMessage): Promise<RPCResponse>;
-}
+**Environment Variables:**
+- `USE_DISPATCHER` - Set to `"true"` to enable dispatching by default (can be overridden per-connection via URL param)
 
+**Message Format:**
+```typescript
 interface DispatcherTranscriptionMessage {
   sessionId: string;
   endpointId: string;  // participant ID
   text: string;        // full transcript text
   timestamp: number;
   language?: string;
 }
+```
+
+**Dispatcher Interface (for RPC fallback):**
+```typescript
+export interface TranscriptionDispatcher extends WorkerEntrypoint<Env> {
+  dispatch(message: DispatcherTranscriptionMessage): Promise<RPCResponse>;
+}
 
 interface RPCResponse {
   success: boolean;
@@ -206,7 +241,7 @@ interface RPCResponse {
 }
 ```
 
-See the original dispatcher implementation on the `main` branch for a reference implementation.
+See the [transcription-dispatcher](https://github.com/jitsi/vo_meetings_cf-transcription-dispatcher) repository for the dispatcher implementation.
 
 ### Monitoring
 
@@ -341,7 +376,7 @@ Media Server (WebSocket)
     ↓
 Cloudflare Worker (intercepts & fans out)
     ↓                           ↓
-    ↓ (audio)         (transcripts via RPC)
+    ↓ (audio)         (transcripts via Queue/RPC)
     ↓                           ↓
 Container Instance        Transcription Dispatcher
     ↓                      (optional, parallel)
@@ -364,7 +399,7 @@ Media Server (receives transcripts)
 3. OpenAI returns transcripts → Container → Worker
 4. Worker fans out transcripts to:
    - Media server (via WebSocket, low latency)
-   - Dispatcher (via Service Binding RPC, async, optional)
+   - Dispatcher (via Queue or RPC, async, optional)
 
 Each container instance:
 - Runs the full Node.js application

worker/env.d.ts

@@ -1,14 +1,17 @@
 // Environment types for the Cloudflare Worker
 
-import type { TranscriptionDispatcher } from './index';
+import type { TranscriptionDispatcher, DispatcherTranscriptionMessage } from './index';
 
 interface Env {
 	// Durable Object binding for the container
 	TRANSCRIBER: DurableObjectNamespace;
 
-	// Service bindings
+	// Service bindings (kept for backwards compatibility, prefer queue)
 	TRANSCRIPTION_DISPATCHER?: Service<TranscriptionDispatcher>;
 
+	// Queue binding for transcription dispatch (preferred)
+	TRANSCRIPTION_QUEUE?: Queue<DispatcherTranscriptionMessage>;
+
 	// Durable Object for auto-scaling
 	CONTAINER_COORDINATOR: DurableObjectNamespace;
 

worker/index.ts

@@ -220,7 +220,9 @@ async function handleWebSocketWithDispatcher(
 	const containerWs = containerResponse.webSocket;
 	containerWs.accept();
 
-	const dispatcher = env.TRANSCRIPTION_DISPATCHER!;
+	// Prefer queue over RPC (queue doesn't count against subrequest limit)
+	const queue = env.TRANSCRIPTION_QUEUE;
+	const dispatcher = env.TRANSCRIPTION_DISPATCHER;
 
 	// Pipe: client → container (upstream, no interception needed)
 	serverWs.addEventListener('message', (event) => {
@@ -249,25 +251,33 @@ async function handleWebSocketWithDispatcher(
 						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:',
-									JSON.stringify({
-										message: response.message,
-										errors: response.errors,
-									}),
-								);
-							}
-						})
-						.catch((error) => {
+					// Use queue if available (preferred - no subrequest limit)
+					if (queue) {
+						queue.send(dispatcherMessage).catch((error) => {
 							const msg = error instanceof Error ? error.message : String(error);
-							const stack = error instanceof Error ? error.stack : undefined;
-							console.error('Dispatcher RPC failed:', msg, stack || '');
+							console.error('Queue send failed:', msg);
 						});
+					} else if (dispatcher) {
+						// Fall back to RPC (has subrequest limit)
+						dispatcher
+							.dispatch(dispatcherMessage)
+							.then((response) => {
+								if (!response.success || response.errors) {
+									console.error(
+										'Dispatcher error:',
+										JSON.stringify({
+											message: response.message,
+											errors: response.errors,
+										}),
+									);
+								}
+							})
+							.catch((error) => {
+								const msg = error instanceof Error ? error.message : String(error);
+								const stack = error instanceof Error ? error.stack : undefined;
+								console.error('Dispatcher RPC failed:', msg, stack || '');
+							});
+					}
 				}
 			} catch {
 				// Not JSON or parse error - ignore, still forwarded to client
@@ -370,7 +380,7 @@ export default {
 		}
 
 		// If dispatcher is enabled and this is a WebSocket upgrade, intercept the connection
-		if (useDispatcher && upgradeHeader === 'websocket' && env.TRANSCRIPTION_DISPATCHER) {
+		if (useDispatcher && upgradeHeader === 'websocket' && (env.TRANSCRIPTION_QUEUE || env.TRANSCRIPTION_DISPATCHER)) {
 			return handleWebSocketWithDispatcher(request, container, env, ctx, sessionId);
 		}