diff --git a/audio/send_audio_to_zoom_scribe_transcribe_service_js/.env.example b/audio/send_audio_to_zoom_scribe_transcribe_service_js/.env.example index c46c14f..7f66956 100644 --- a/audio/send_audio_to_zoom_scribe_transcribe_service_js/.env.example +++ b/audio/send_audio_to_zoom_scribe_transcribe_service_js/.env.example @@ -21,15 +21,15 @@ zoomWSURLForEvents= # These are separate from ZOOM_CLIENT_ID / ZOOM_CLIENT_SECRET used for RTMS. ZOOM_API_KEY= ZOOM_API_SECRET= -SCRIBE_BASE_URL=https://api.zoom.us/v2 +# Full WebSocket URL of the live transcription endpoint. +SCRIBE_LIVE_URL=wss://api.zoom.us/v2/aiservices/scribe/live -# Scribe transcription settings +# Live Scribe (real-time streaming) settings. SCRIBE_LANGUAGE=en-US -SCRIBE_WINDOW_SECONDS=10 -SCRIBE_MAX_WINDOWS=24 SCRIBE_WORD_TIME_OFFSETS=true SCRIBE_TIMESTAMPS=true SCRIBE_DIARIZATION=false SCRIBE_CHANNEL_SEPARATION=false SCRIBE_PROFANITY_FILTER=false SCRIBE_OUTPUT_FORMAT=json + diff --git a/audio/send_audio_to_zoom_scribe_transcribe_service_js/README.md b/audio/send_audio_to_zoom_scribe_transcribe_service_js/README.md index 6b6a457..b70ef9a 100644 --- a/audio/send_audio_to_zoom_scribe_transcribe_service_js/README.md +++ b/audio/send_audio_to_zoom_scribe_transcribe_service_js/README.md @@ -1,19 +1,23 @@ -# Send Audio to Zoom Scribe Transcription Service +# Send Audio to Zoom Scribe Live Transcription Service -Stream Zoom RTMS meeting audio into short WAV windows and transcribe each window with the Zoom AI Services Scribe API fast-mode endpoint. +Stream Zoom RTMS meeting audio into the Zoom AI Services Scribe **live** API — a +real-time transcription WebSocket — and log transcripts as they arrive. -> Built with `RTMSManager` and Zoom AI Services Scribe fast mode. +> Built with `RTMSManager` and Zoom AI Services Scribe live streaming. ## What This Sample Does - Receives `meeting.rtms_started` and `meeting.rtms_stopped` events through `WebhookManager` or `WebsocketManager`. - Connects RTMSManager to the meeting media stream. - Requests audio-only RTMS media as 16 kHz mono L16 mixed audio. -- Buffers RTMS PCM audio into short WAV chunks under `audio_windows/`. -- Sends each chunk to Scribe fast mode with `POST /aiservices/scribe/transcribe`. -- Logs each returned transcript to the console. +- On `meeting.rtms_started`, opens a Scribe **live** WebSocket (`/aiservices/scribe/live`). +- Forwards each RTMS PCM packet straight to the WebSocket as a binary frame (no file buffering, no resampling). +- Logs `transcription.completed` events as they stream back. +- On `meeting.rtms_stopped`, sends `session.close`, waits for the final transcript, and closes the socket. -Scribe is file-oriented, not a live streaming WebSocket. This sample is a pseudo-streaming pattern: it sends one WAV chunk every `SCRIBE_WINDOW_SECONDS` seconds, waits for the fast-mode response, and then logs the returned transcript. +Unlike the fast-mode `/transcribe` endpoint (which uploads whole audio files), the +live endpoint is a true streaming WebSocket: audio flows in continuously and +transcripts come back with low latency while the meeting is still in progress. ## Quick Start @@ -52,17 +56,15 @@ ZOOM_API_KEY= ZOOM_API_SECRET= ``` -`ZOOM_API_KEY` and `ZOOM_API_SECRET` are the Zoom AI Services / Build-platform credentials used to sign the Scribe JWT. +`ZOOM_API_KEY` and `ZOOM_API_SECRET` are the Zoom AI Services / Build-platform credentials used to sign the Scribe JWT (the same credential the fast-mode endpoint used). ## Optional Environment Variables ```env PORT=3000 WEBHOOK_PATH=/webhook -SCRIBE_BASE_URL=https://api.zoom.us/v2 +SCRIBE_LIVE_URL=wss://api.zoom.us/v2/aiservices/scribe/live SCRIBE_LANGUAGE=en-US -SCRIBE_WINDOW_SECONDS=10 -SCRIBE_MAX_WINDOWS=24 SCRIBE_WORD_TIME_OFFSETS=true SCRIBE_TIMESTAMPS=true SCRIBE_DIARIZATION=false @@ -71,36 +73,42 @@ SCRIBE_PROFANITY_FILTER=false SCRIBE_OUTPUT_FORMAT=json ``` -Recommended starting window size is `10` seconds. Lower values give faster partial results but increase upload overhead and can cut words across windows. +`SCRIBE_LIVE_URL` is the full WebSocket URL of the live transcription endpoint +(defaults to `wss://api.zoom.us/v2/aiservices/scribe/live`). ## How It Works 1. The app starts an Express server and initializes RTMSManager. 2. The webhook endpoint receives `meeting.rtms_started`. -3. RTMSManager connects to Zoom signaling/media sockets. -4. RTMS audio packets arrive as raw 16 kHz mono PCM. -5. `audioWindowBuffer.js` wraps each `SCRIBE_WINDOW_SECONDS` window in a WAV container. -6. `scribeClient.js` signs a Build-platform JWT and submits the WAV chunk to Scribe fast mode. -7. The returned transcript text is logged as `[ZoomScribe] Transcript result`. +3. `scribeClient.js` mints a Build-platform JWT and opens `wss://.../aiservices/scribe/live`, then sends `session.update` (`audio.format=pcm16`, `language`). +4. RTMSManager connects to Zoom signaling/media sockets; RTMS audio arrives as raw 16 kHz mono PCM16. +5. Each audio packet is forwarded to the WebSocket as a binary frame. Audio that arrives before the session is ready is buffered and flushed on `session.updated`. +6. The server streams back `transcription.completed` events, which are logged. +7. On `meeting.rtms_stopped`, the client sends `session.close`, waits briefly for the final transcript, logs the full meeting transcript, and closes. -## Scribe Request Shape - -The sample uses multipart upload: +## Live WebSocket Protocol ```text -POST https://api.zoom.us/v2/aiservices/scribe/transcribe -Authorization: Bearer -file= -config={"language":"en-US","word_time_offsets":true,"timestamps":true} +Connect: wss://api.zoom.us/v2/aiservices/scribe/live + Subprotocols: ["live-asr", "zoom-api-access-token."] + (the JWT is carried in the "zoom-api-access-token.*" subprotocol) + +Client -> { "type": "session.update", "audio": { "format": "pcm16" }, "language": "en-US" } +Client -> # streamed RTMS audio +Client -> { "type": "session.close" } # on meeting stop + +Server -> { "type": "session.created", "session_id": ... } +Server -> { "type": "session.updated" } # ready to receive audio +Server -> { "type": "transcription.completed", "transcript": ..., "audio_start_ms": ..., "audio_end_ms": ... } +Server -> { "type": "session.closed", "reason": ... } ``` ## Files | File | Purpose | |------|---------| -| `index.js` | Express, RTMSManager, webhook/websocket trigger, transcription queue | -| `audioWindowBuffer.js` | Converts RTMS L16 PCM chunks into WAV windows | -| `scribeClient.js` | Zoom Scribe JWT auth and fast-mode transcription client | +| `index.js` | Express, RTMSManager, webhook/websocket trigger, forwards RTMS audio to the live client | +| `scribeClient.js` | Scribe JWT auth + live streaming WebSocket client (connect, stream, event handling, cleanup) | | `.env.example` | Configuration template | ## Troubleshooting @@ -109,11 +117,11 @@ config={"language":"en-US","word_time_offsets":true,"timestamps":true} |-------|-------| | Missing credential error | Set `ZOOM_CLIENT_ID`, `ZOOM_CLIENT_SECRET`, `ZOOM_SECRET_TOKEN`, `ZOOM_API_KEY`, and `ZOOM_API_SECRET`. | | No webhook received | Confirm the public HTTPS webhook URL points to `/webhook` or your configured `WEBHOOK_PATH`. | -| No transcript text | Confirm RTMS is receiving audio and `audio_windows/` contains WAV files. | -| Scribe 401/403 | Confirm `ZOOM_API_KEY` and `ZOOM_API_SECRET` are AI Services / Build-platform credentials. | -| Slow updates | Lower `SCRIBE_WINDOW_SECONDS`, but expect more upload overhead. | +| WebSocket 401/403 on connect | Confirm `ZOOM_API_KEY`/`ZOOM_API_SECRET` are AI Services / Build-platform credentials. | +| Connects but no transcripts | Confirm RTMS is delivering audio (watch the `chunks=`/`sentBytes=` log line) and that `session.updated` was received. | ## Notes -- This sample uses mixed meeting audio. For per-participant audio, request RTMS audio multi-streams and route windows by RTMS `userId`. -- Scribe fast mode is best for short windows. For long recordings or archives, use Scribe batch jobs instead. +- This sample uses mixed meeting audio. For per-participant audio, request RTMS audio multi-streams and open one live session per RTMS `userId`. +- The live session has a server-side maximum duration; very long meetings may be closed by the server (the client logs `session.closed` with the reason). For archival transcription of long recordings, use Scribe batch jobs instead. +- RTMS L16 at 16 kHz mono matches the live API's required `pcm16` format exactly, so audio is forwarded verbatim with no resampling or WAV wrapping. diff --git a/audio/send_audio_to_zoom_scribe_transcribe_service_js/index.js b/audio/send_audio_to_zoom_scribe_transcribe_service_js/index.js index 8d65544..f523f47 100644 --- a/audio/send_audio_to_zoom_scribe_transcribe_service_js/index.js +++ b/audio/send_audio_to_zoom_scribe_transcribe_service_js/index.js @@ -7,8 +7,14 @@ import { fileURLToPath } from 'url'; import { RTMSManager } from '../../library/javascript/rtmsManager/RTMSManager.js'; import WebhookManager from '../../library/javascript/webhookManager/WebhookManager.js'; import WebsocketManager from '../../library/javascript/webSocketManager/WebsocketManager.js'; -import { AudioWindowBuffer } from './audioWindowBuffer.js'; -import { ScribeClient } from './scribeClient.js'; +import { + initializeLiveScribeSession, + sendAudioChunk, + cleanupMeeting, + closeLiveScribe, + liveScribeConfig, + activeSessionCount +} from './scribeClient.js'; const __filename = fileURLToPath(import.meta.url); const __dirname = path.dirname(__filename); @@ -70,6 +76,8 @@ const rtmsConfig = { }, mediaParams: { audio: { + // The live Scribe API expects 16 kHz mono PCM16 (LE) — request exactly that + // from RTMS so audio can be forwarded verbatim, with no resampling. contentType: MEDIA_PARAMS.MEDIA_CONTENT_TYPE_RAW_AUDIO, sampleRate: MEDIA_PARAMS.AUDIO_SAMPLE_RATE_SR_16K, channel: MEDIA_PARAMS.AUDIO_CHANNEL_MONO, @@ -82,97 +90,47 @@ const rtmsConfig = { const app = express(); const server = http.createServer(app); -const audioWindows = new AudioWindowBuffer({ - outputDir: path.join(__dirname, 'audio_windows'), - sampleRate: 16000, - channels: 1, - bitsPerSample: 16, - windowSeconds: envNumber('SCRIBE_WINDOW_SECONDS', 10), - maxWindows: envNumber('SCRIBE_MAX_WINDOWS', 24) -}); -const scribeClient = ScribeClient.fromEnv(process.env); let activeMeetingId = null; let activeStreamId = null; -let transcriptionInFlight = false; -const transcriptionQueue = []; console.log('[ZoomScribe] App Configuration:', appConfig); console.log('[ZoomScribe] RTMS Configuration:', RTMSManager.redactSecrets(rtmsConfig)); -console.log('[ZoomScribe] Scribe Configuration:', { - baseUrl: scribeClient.baseUrl, - language: scribeClient.language, - windowSeconds: audioWindows.windowSeconds, - sampleRate: audioWindows.sampleRate, - channels: audioWindows.channels -}); +console.log('[ZoomScribe] Live Scribe Configuration:', liveScribeConfig()); app.use(express.json()); app.get('/health', (req, res) => { res.json({ ok: true, + mode: 'live', activeMeetingId, activeStreamId, - queuedWindows: transcriptionQueue.length, - transcriptionInFlight + liveSessions: activeSessionCount() }); }); await RTMSManager.init(rtmsConfig); -async function transcribeWindow(window) { - const startedAt = Date.now(); - const result = await scribeClient.transcribeFile(window.filePath, { - meetingId: activeMeetingId, - streamId: activeStreamId, - window - }); - - console.log('[ZoomScribe] Transcript result:', { - fileName: window.fileName, - requestId: result.requestId, - model: result.model, - durationSec: result.durationSec, - elapsedMs: Date.now() - startedAt, - text: result.text || '(no transcript text)' - }); -} - -async function drainTranscriptionQueue() { - if (transcriptionInFlight) return; - transcriptionInFlight = true; - - try { - while (transcriptionQueue.length > 0) { - const window = transcriptionQueue.shift(); - try { - await transcribeWindow(window); - } catch (error) { - console.error('[ZoomScribe] Transcription failed:', { - fileName: window.fileName, - message: error.message - }); - } - } - } finally { - transcriptionInFlight = false; - } -} - +// Open/close the live transcription WebSocket in step with the meeting lifecycle. function updateActiveRtmsState(event, payload = {}) { if (event === 'meeting.rtms_started') { activeMeetingId = payload.meeting_uuid; activeStreamId = payload.rtms_stream_id; - audioWindows.reset(); - transcriptionQueue.length = 0; + initializeLiveScribeSession(activeMeetingId); } if (event === 'meeting.rtms_stopped') { - activeMeetingId = null; - activeStreamId = null; - audioWindows.reset(); - transcriptionQueue.length = 0; + const endingMeetingId = payload.meeting_uuid || activeMeetingId; + if (endingMeetingId) { + cleanupMeeting(endingMeetingId).catch((error) => { + console.error('[ZoomScribe] Live session cleanup failed:', error.message); + }); + } + if (endingMeetingId === activeMeetingId) { + activeMeetingId = null; + activeStreamId = null; + } } } @@ -212,29 +170,10 @@ if (appConfig.managerType === 'webhook') { console.log('[ZoomScribe] Websocket Manager initialized'); } -RTMSManager.on('audio', ({ buffer, userId, userName, timestamp, streamId }) => { - const windows = audioWindows.writeAudio(buffer, { - meetingId: activeMeetingId, - streamId, - userId: userId ?? null, - userName: userName ?? null, - timestamp: timestamp ?? Date.now() - }); - - for (const window of windows) { - console.log('[ZoomScribe] Audio window ready:', { - fileName: window.fileName, - bytes: window.size, - durationSeconds: window.durationSeconds, - sampleCount: window.sampleCount, - userName: window.userName - }); - transcriptionQueue.push(window); - } - - drainTranscriptionQueue().catch((error) => { - console.error('[ZoomScribe] Queue drain failed:', error.message); - }); +// Forward each RTMS audio packet straight to the live Scribe WebSocket. +RTMSManager.on('audio', ({ buffer, userId }) => { + if (!activeMeetingId) return; + sendAudioChunk(buffer, activeMeetingId, userId ?? 0); }); RTMSManager.on('error', (error) => { @@ -251,6 +190,7 @@ server.listen(appConfig.port, () => { process.on('SIGINT', async () => { console.log('[ZoomScribe] Shutting down...'); server.close(); + await closeLiveScribe(); await RTMSManager.stop(); process.exit(0); }); diff --git a/audio/send_audio_to_zoom_scribe_transcribe_service_js/scribeClient.js b/audio/send_audio_to_zoom_scribe_transcribe_service_js/scribeClient.js index a04cbd5..1b562ef 100644 --- a/audio/send_audio_to_zoom_scribe_transcribe_service_js/scribeClient.js +++ b/audio/send_audio_to_zoom_scribe_transcribe_service_js/scribeClient.js @@ -1,25 +1,35 @@ -import fs from 'fs/promises'; +import WebSocket from 'ws'; +import dotenv from 'dotenv'; import path from 'path'; +import { fileURLToPath } from 'url'; import { KJUR } from 'jsrsasign'; -function boolEnv(value, fallback = false) { - if (value == null || value === '') return fallback; - return ['1', 'true', 'yes', 'on'].includes(String(value).trim().toLowerCase()); -} +const __filename = fileURLToPath(import.meta.url); +const __dirname = path.dirname(__filename); -function validateRequired(value, name) { - if (!value || String(value).trim() === '') { - throw new Error(`${name} is required`); - } +// Load .env here too: ES module imports are evaluated before the importer's +// body runs, so CONFIG below must see env vars regardless of import order. +dotenv.config({ path: path.join(__dirname, '.env') }); + +const LOG = '[ZoomScribeLive]'; + +// Reconnect backoff, the pre-connect audio backlog cap, and how long to wait for +// the final transcript after asking the server to close. +const RECONNECT_DELAY_MS = 2000; +const MAX_QUEUED_AUDIO_BYTES = 5 * 1024 * 1024; +const FINALIZE_WAIT_MS = 3000; + +function requireValue(value, name) { + if (!value || String(value).trim() === '') throw new Error(`${name} is required`); } +// Zoom AI Services Scribe uses a Build-platform HS256 JWT: `iss` is ZOOM_API_KEY +// and the token is signed with ZOOM_API_SECRET. export function generateScribeJwt(apiKey, apiSecret) { - validateRequired(apiKey, 'ZOOM_API_KEY'); - validateRequired(apiSecret, 'ZOOM_API_SECRET'); - + requireValue(apiKey, 'ZOOM_API_KEY'); + requireValue(apiSecret, 'ZOOM_API_SECRET'); const iat = Math.round(Date.now() / 1000) - 30; const exp = iat + 60 * 60; - return KJUR.jws.JWS.sign( 'HS256', JSON.stringify({ alg: 'HS256', typ: 'JWT' }), @@ -28,115 +38,252 @@ export function generateScribeJwt(apiKey, apiSecret) { ); } -export function extractTranscriptText(payload) { - if (!payload) return ''; - if (typeof payload === 'string') return payload; - - const candidates = [ - payload.text, - payload.transcript, - payload.result?.text, - payload.result?.transcript, - payload.result?.summary?.text - ].filter(Boolean); - - if (candidates.length > 0) return String(candidates[0]); - - const segments = payload.result?.segments || payload.segments || payload.result?.utterances || payload.utterances; - if (Array.isArray(segments)) { - return segments - .map((segment) => segment.text || segment.transcript || segment.words?.map((word) => word.text || word.word).join(' ')) - .filter(Boolean) - .join(' ') - .trim(); - } +// Full WebSocket URL of the live transcription endpoint. +const DEFAULT_LIVE_URL = 'wss://api.zoom.us/v2/aiservices/scribe/live'; + +const CONFIG = { + apiKey: process.env.ZOOM_API_KEY, + apiSecret: process.env.ZOOM_API_SECRET, + liveUrl: process.env.SCRIBE_LIVE_URL || DEFAULT_LIVE_URL, + language: process.env.SCRIBE_LANGUAGE || 'en-US', +}; + +// meetingUuid -> session +const sessions = new Map(); + +export function liveScribeConfig() { + return { + liveUrl: CONFIG.liveUrl, + language: CONFIG.language, + }; +} - const words = payload.result?.words || payload.words; - if (Array.isArray(words)) { - return words.map((word) => word.text || word.word).filter(Boolean).join(' ').trim(); +export function activeSessionCount() { + return sessions.size; +} + +// Open a live transcription WebSocket for a meeting (called on meeting.rtms_started). +export function initializeLiveScribeSession(meetingUuid) { + if (!meetingUuid || sessions.has(meetingUuid)) return; + const session = { + meetingUuid, + ws: null, + ready: false, + stopRequested: false, + queued: [], + queuedBytes: 0, + sentBytes: 0, + sourceBytes: 0, + chunks: 0, + startedAt: Date.now(), + reconnectTimer: null, + sessionId: null, + completed: [], + closedWaiters: [], + }; + sessions.set(meetingUuid, session); + console.log(`${LOG} Initializing live session for meeting ${meetingUuid}`); + connect(session); +} + +// Stream one RTMS audio buffer (16 kHz mono PCM16 LE) as a binary WS frame. +export function sendAudioChunk(buffer, meetingUuid, userId = 0) { + if (!buffer || buffer.length === 0) return; + const session = sessions.get(meetingUuid); + if (!session || session.stopRequested) return; + + session.chunks += 1; + session.sourceBytes += buffer.length; + + if (session.ready && session.ws && session.ws.readyState === WebSocket.OPEN) { + try { + session.ws.send(buffer); + session.sentBytes += buffer.length; + } catch (error) { + console.error(`${LOG} send failed, queuing: ${error.message}`); + queueAudio(session, buffer); + } + } else { + // Not connected/ready yet (handshake + session.update in flight): buffer it. + queueAudio(session, buffer); } - return ''; + if (session.chunks % 200 === 0) { + const elapsed = ((Date.now() - session.startedAt) / 1000).toFixed(1); + console.log( + `${LOG} [${String(meetingUuid).slice(0, 8)}] chunks=${session.chunks} rtmsBytes=${session.sourceBytes} ` + + `sentBytes=${session.sentBytes} queuedBytes=${session.queuedBytes} elapsed=${elapsed}s lastUser=${userId}` + ); + } } -export class ScribeClient { - constructor(options = {}) { - this.apiKey = options.apiKey || ''; - this.apiSecret = options.apiSecret || ''; - this.baseUrl = String(options.baseUrl || 'https://api.zoom.us/v2').replace(/\/+$/, ''); - this.language = options.language || 'en-US'; - this.wordTimeOffsets = Boolean(options.wordTimeOffsets); - this.timestamps = Boolean(options.timestamps); - this.diarization = Boolean(options.diarization); - this.channelSeparation = Boolean(options.channelSeparation); - this.profanityFilter = Boolean(options.profanityFilter); - this.outputFormat = options.outputFormat || 'json'; +function queueAudio(session, buffer) { + session.queued.push(buffer); + session.queuedBytes += buffer.length; + // Bound the pre-connect backlog: drop oldest audio if we exceed the cap. + while (session.queuedBytes > MAX_QUEUED_AUDIO_BYTES && session.queued.length > 0) { + session.queuedBytes -= session.queued.shift().length; } +} - static fromEnv(env = process.env) { - return new ScribeClient({ - apiKey: env.ZOOM_API_KEY, - apiSecret: env.ZOOM_API_SECRET, - baseUrl: env.SCRIBE_BASE_URL, - language: env.SCRIBE_LANGUAGE || env.LANGUAGE || 'en-US', - wordTimeOffsets: boolEnv(env.SCRIBE_WORD_TIME_OFFSETS, true), - timestamps: boolEnv(env.SCRIBE_TIMESTAMPS, true), - diarization: boolEnv(env.SCRIBE_DIARIZATION, false), - channelSeparation: boolEnv(env.SCRIBE_CHANNEL_SEPARATION, false), - profanityFilter: boolEnv(env.SCRIBE_PROFANITY_FILTER, false), - outputFormat: env.SCRIBE_OUTPUT_FORMAT || 'json' - }); +function flushQueue(session) { + while ( + session.ready && + session.ws && + session.ws.readyState === WebSocket.OPEN && + session.queued.length > 0 + ) { + const buf = session.queued.shift(); + session.queuedBytes -= buf.length; + try { + session.ws.send(buf); + session.sentBytes += buf.length; + } catch (error) { + console.error(`${LOG} flush failed: ${error.message}`); + session.queued.unshift(buf); + session.queuedBytes += buf.length; + break; + } } +} + +function connect(session) { + if (session.stopRequested) return; - get config() { - return { - language: this.language, - word_time_offsets: this.wordTimeOffsets, - timestamps: this.timestamps, - diarization: this.diarization, - channel_separation: this.channelSeparation, - profanity_filter: this.profanityFilter, - output_format: this.outputFormat - }; + let jwt; + try { + jwt = generateScribeJwt(CONFIG.apiKey, CONFIG.apiSecret); + } catch (error) { + console.error(`${LOG} cannot mint Scribe JWT: ${error.message}`); + return; } - async transcribeFile(filePath, metadata = {}) { - const token = generateScribeJwt(this.apiKey, this.apiSecret); - const fileBuffer = await fs.readFile(filePath); - const form = new FormData(); - form.append('file', new Blob([new Uint8Array(fileBuffer)], { type: 'audio/wav' }), path.basename(filePath)); - form.append('config', JSON.stringify(this.config)); - - const response = await fetch(`${this.baseUrl}/aiservices/scribe/transcribe`, { - method: 'POST', - headers: { - Authorization: `Bearer ${token}` - }, - body: form - }); - - const responseText = await response.text(); - let payload; + // Auth is carried in the WebSocket subprotocol list: "live-asr" is the real + // subprotocol; "zoom-api-access-token." presents the credential. + console.log(`${LOG} Connecting to ${CONFIG.liveUrl} for meeting ${session.meetingUuid}`); + const ws = new WebSocket(CONFIG.liveUrl, ['live-asr', `zoom-api-access-token.${jwt}`]); + session.ws = ws; + session.ready = false; + + ws.on('open', () => { + console.log(`${LOG} Connected for meeting ${session.meetingUuid}`); try { - payload = responseText ? JSON.parse(responseText) : {}; - } catch { - payload = { raw: responseText }; + ws.send(JSON.stringify({ + type: 'session.update', + audio: { format: 'pcm16' }, + language: CONFIG.language, + })); + console.log(`${LOG} Sent session.update (lang=${CONFIG.language})`); + } catch (error) { + console.error(`${LOG} session.update send failed: ${error.message}`); } + }); - if (!response.ok) { - throw new Error(`Zoom Scribe transcription failed: ${response.status} ${responseText}`); + ws.on('message', (data, isBinary) => { + if (!isBinary) handleServerEvent(session, data); + }); + + ws.on('error', (error) => { + console.error(`${LOG} WebSocket error for meeting ${session.meetingUuid}: ${error.message}`); + }); + + ws.on('close', (code, reason) => { + session.ready = false; + console.log(`${LOG} Closed for meeting ${session.meetingUuid}: ${code} ${reason?.toString() || ''}`); + session.closedWaiters.splice(0).forEach((resolve) => resolve()); + // Reconnect only on abnormal closes while the meeting is still active. + if (!session.stopRequested && code !== 1000) { + session.reconnectTimer = setTimeout(() => connect(session), RECONNECT_DELAY_MS); } + }); +} - return { - requestId: payload.request_id || payload.requestId || null, - durationSec: payload.duration_sec ?? payload.durationSec ?? null, - model: payload.model || null, - text: extractTranscriptText(payload), - rawResult: payload, - metadata, - timestamp: Date.now() - }; +function handleServerEvent(session, raw) { + let event; + try { + event = JSON.parse(raw.toString()); + } catch { + return; + } + const tag = `[${String(session.meetingUuid).slice(0, 8)}]`; + + switch (event.type) { + case 'session.created': + session.sessionId = event.session_id || null; + console.log(`${LOG} ${tag} session.created id=${session.sessionId}`); + break; + case 'session.updated': + session.ready = true; + console.log(`${LOG} ${tag} session.updated — streaming audio`); + flushQueue(session); + break; + case 'transcription.completed': { + const text = event.transcript || ''; + if (text) session.completed.push(text); + const startSec = ((event.audio_start_ms ?? 0) / 1000).toFixed(1); + const endSec = ((event.audio_end_ms ?? 0) / 1000).toFixed(1); + console.log(`${LOG} ${tag} [${startSec}s-${endSec}s] ${text}`); + break; + } + case 'error': + console.error( + `${LOG} ${tag} server error code=${event.error?.code} ` + + `msg=${event.error?.message} fatal=${event.error?.fatal}` + ); + break; + case 'session.closed': + console.log(`${LOG} ${tag} session.closed reason=${event.reason}`); + session.closedWaiters.splice(0).forEach((resolve) => resolve()); + break; + default: + break; } } -export default ScribeClient; +// Gracefully end a meeting's live session (called on meeting.rtms_stopped). +export async function cleanupMeeting(meetingUuid) { + const session = sessions.get(meetingUuid); + if (!session) return; + + console.log( + `${LOG} Cleaning up meeting ${meetingUuid} (chunks=${session.chunks}, sentBytes=${session.sentBytes})` + ); + session.stopRequested = true; + if (session.reconnectTimer) clearTimeout(session.reconnectTimer); + + if (session.ws && session.ws.readyState === WebSocket.OPEN) { + // Ask the server to finalize; the final utterance is transcribed on close. + try { session.ws.send(JSON.stringify({ type: 'session.close' })); } catch { /* ignore */ } + await waitForClose(session, FINALIZE_WAIT_MS); + try { + if (session.ws.readyState === WebSocket.OPEN) session.ws.close(1000, 'meeting stopped'); + } catch { /* ignore */ } + } else if (session.ws) { + try { session.ws.terminate(); } catch { /* ignore */ } + } + + if (session.completed.length > 0) { + console.log(`${LOG} Final transcript for meeting ${meetingUuid}:\n${session.completed.join(' ')}`); + } + sessions.delete(meetingUuid); +} + +function waitForClose(session, timeoutMs) { + return new Promise((resolve) => { + let done = false; + const finish = () => { if (!done) { done = true; resolve(); } }; + session.closedWaiters.push(finish); + setTimeout(finish, timeoutMs); + }); +} + +// Close every active session (called on process shutdown). +export async function closeLiveScribe(meetingUuid = null) { + if (meetingUuid) { + await cleanupMeeting(meetingUuid); + return; + } + for (const uuid of [...sessions.keys()]) { + await cleanupMeeting(uuid); + } +}