1// @effect-diagnostics globalTimers:off - imperative WebSocket callbacks own and clear their response deadline.
2import {
3 SPEECH_STREAM_MAX_CHUNK_BYTES,
4 SPEECH_STREAM_MAX_QUEUED_BYTES,
5 SpeechStreamEvent,
6 type SpeechStreamText,
7 type SpeechTranscriptionOptions,
8} from "@t3tools/contracts";
9import * as Schema from "effect/Schema";
10
11const decodeEvent = Schema.decodeUnknownSync(Schema.fromJsonString(SpeechStreamEvent));
12
13/** Audio stays bounded while the environment acknowledges a small send window. */
14export function openSpeechStream(input: {
15 readonly url: string;
16 readonly options: SpeechTranscriptionOptions;
17 readonly signal: AbortSignal;
18 readonly canSend: () => boolean;
19 readonly onText: (text: SpeechStreamText) => void;
20 readonly onError: (error: Error) => void;
21 readonly createSocket?: (url: string) => WebSocket;
22}) {
23 input.signal.throwIfAborted();
24 const socket = (input.createSocket ?? ((url) => new WebSocket(url)))(input.url);
25 let resolveFinal!: (text: string) => void;
26 let rejectFinal!: (error: Error) => void;
27 const final = new Promise<string>((resolve, reject) => {
28 resolveFinal = resolve;
29 rejectFinal = reject;
30 });
31 // A capture failure can precede stop(), when nobody is awaiting the final result yet.
32 void final.catch(() => undefined);
33 let queue: Uint8Array<ArrayBuffer>[] = [];
34 let queuedBytes = 0;
35 const inFlight: number[] = [];
36 let inFlightBytes = 0;
37 let finishing = false;
38 let finishSent = false;
39 let closed = false;
40 let isReady = false;
41 let revision = -1;
42 let timer: ReturnType<typeof setTimeout>;
43 const cleanup = () => {
44 clearTimeout(timer);
45 input.signal.removeEventListener("abort", abort);
46 queue = [];
47 queuedBytes = 0;
48 };
49 const fail = (error: Error, notify = true) => {
50 if (closed) return;
51 closed = true;
52 cleanup();
53 rejectFinal(error);
54 socket.close();
55 if (notify) input.onError(error);
56 };
57 const armTimeout = () => {
58 clearTimeout(timer);
59 timer = setTimeout(() => fail(new Error("Live transcription stopped responding.")), 120_000);
60 };
61 const abort = () => fail(new Error("Voice transcription was cancelled."), false);
62 const pump = () => {
63 if (!isReady || closed || finishSent) return;
64 if (!input.canSend()) return fail(new Error("Voice consent was revoked."));
65 while (inFlight.length < 2 && queue.length) {
66 const chunk = queue.shift()!;
67 inFlight.push(chunk.byteLength);
68 inFlightBytes += chunk.byteLength;
69 queuedBytes -= chunk.byteLength;
70 socket.send(chunk);
71 armTimeout();
72 }
73 if (finishing && !queue.length && !inFlight.length) {
74 finishSent = true;
75 socket.send(JSON.stringify({ type: "finish" }));
76 armTimeout();
77 }
78 };
79 socket.addEventListener("open", () => {
80 if (!input.canSend()) return fail(new Error("Voice consent was revoked."));
81 if (!closed) socket.send(JSON.stringify({ type: "start", options: input.options }));
82 });
83 socket.addEventListener("message", (event) => {
84 if (closed) return;
85 if (!input.canSend()) return fail(new Error("Voice consent was revoked."));
86 try {
87 const message = decodeEvent(event.data);
88 switch (message.type) {
89 case "ready":
90 if (isReady) throw new Error("Unexpected speech stream response.");
91 isReady = true;
92 clearTimeout(timer);
93 pump();
94 break;
95 case "update":
96 if (!inFlight.length || message.revision < revision)
97 throw new Error("Unexpected speech stream response.");
98 revision = message.revision;
99 inFlightBytes -= inFlight.shift()!;
100 clearTimeout(timer);
101 if (message.text) input.onText(message.text);
102 pump();
103 if (inFlight.length) armTimeout();
104 break;
105 case "finished":
106 if (!finishSent) throw new Error("Unexpected speech stream response.");
107 closed = true;
108 cleanup();
109 resolveFinal(message.text);
110 socket.close();
111 break;
112 case "error":
113 fail(new Error(message.message));
114 }
115 } catch (error) {
116 fail(error instanceof Error ? error : new Error("Invalid speech stream response."));
117 }
118 });
119 socket.addEventListener("close", () =>
120 fail(new Error("The voice connection closed. Please record again.")),
121 );
122 socket.addEventListener("error", () =>
123 fail(new Error("Could not connect to live transcription.")),
124 );
125 input.signal.addEventListener("abort", abort, { once: true });
126 armTimeout();
127 if (input.signal.aborted) abort();
128 return {
129 feed: (pcm: Float32Array) => {
130 if (closed || finishing) return;
131 const bytes = new Uint8Array(pcm.buffer, pcm.byteOffset, pcm.byteLength);
132 if (queuedBytes + inFlightBytes + bytes.byteLength > SPEECH_STREAM_MAX_QUEUED_BYTES) {
133 fail(new Error("Voice recordings are limited to five minutes."));
134 return;
135 }
136 let offset = 0;
137 while (offset < bytes.length) {
138 const tail = queue.at(-1);
139 const available = tail ? SPEECH_STREAM_MAX_CHUNK_BYTES - tail.byteLength : 0;
140 if (tail && available > 0) {
141 const appended = Math.min(available, bytes.length - offset);
142 const merged = new Uint8Array(tail.byteLength + appended);
143 merged.set(tail);
144 merged.set(bytes.subarray(offset, offset + appended), tail.byteLength);
145 queue[queue.length - 1] = merged;
146 queuedBytes += appended;
147 offset += appended;
148 continue;
149 }
150 const chunk = bytes.slice(offset, offset + SPEECH_STREAM_MAX_CHUNK_BYTES);
151 queue.push(chunk);
152 queuedBytes += chunk.byteLength;
153 offset += chunk.byteLength;
154 }
155 try {
156 pump();
157 } catch {
158 fail(new Error("Could not send microphone audio."));
159 }
160 },
161 finish: () => {
162 finishing = true;
163 try {
164 pump();
165 } catch {
166 fail(new Error("Could not finish live transcription."));
167 }
168 return final;
169 },
170 cancel: abort,
171 };
172}