1import {
2 AuthOrchestrationOperateScope,
3 SPEECH_STREAM_PATH,
4 SPEECH_STREAM_MAX_CHUNK_BYTES,
5 SpeechStreamCommand,
6 SPEECH_MAX_OPTIONS_BYTES,
7 type SpeechStreamEvent,
8} from "@t3tools/contracts";
9import * as Effect from "effect/Effect";
10import * as Layer from "effect/Layer";
11import * as Queue from "effect/Queue";
12import * as Schema from "effect/Schema";
13import * as Scope from "effect/Scope";
14import { HttpRouter, HttpServerRequest, HttpServerResponse } from "effect/http";
15import * as Socket from "effect/socket/Socket";
16import * as EnvironmentAuth from "../auth/EnvironmentAuth.ts";
17import * as SpeechService from "./SpeechService.ts";
18
19const decodeCommand = Schema.decodeUnknownEffect(Schema.fromJsonString(SpeechStreamCommand));
20
21/** One socket owns one stream. Audio is processed and acknowledged in arrival order. */
22export const runSpeechSocket = Effect.fn("speech.runSocket")(function* (
23 socket: Socket.Socket,
24 speech: Pick<SpeechService.SpeechService["Service"], "startStream">,
25) {
26 const scope = yield* Scope.Scope;
27 const writer = yield* socket.writer;
28 const send = (event: SpeechStreamEvent) => writer.write(JSON.stringify(event));
29 let stream: SpeechService.SpeechStream | undefined;
30 let ended = false;
31 const fail = Effect.fn("speech.failSocket")(function* (message: string) {
32 if (ended) return;
33 ended = true;
34 yield* send({ type: "error", message });
35 yield* writer.write(new Socket.CloseEvent(1008, "Speech stream ended"));
36 });
37 const receive = (data: string | Uint8Array) => {
38 if (ended) return Effect.void;
39 return Effect.gen(function* () {
40 if (typeof data === "string") {
41 if (new TextEncoder().encode(data).byteLength > SPEECH_MAX_OPTIONS_BYTES)
42 return yield* fail("Invalid speech stream command.");
43 const command = yield* decodeCommand(data);
44 if (command.type === "start") {
45 if (stream) return yield* fail("Speech stream already started.");
46 stream = yield* speech.startStream(command.options).pipe(Scope.provide(scope));
47 yield* send({ type: "ready" });
48 return;
49 }
50 if (!stream) return yield* fail("Start speech transcription before finishing.");
51 const text = yield* stream.finish;
52 ended = true;
53 yield* send({ type: "finished", text });
54 yield* writer.write(new Socket.CloseEvent(1000));
55 } else {
56 if (data.byteLength > SPEECH_STREAM_MAX_CHUNK_BYTES)
57 return yield* fail("Speech audio chunk is too large.");
58 if (!stream) return yield* fail("Start speech transcription before sending audio.");
59 const update = yield* stream.feed(data);
60 yield* send({ type: "update", ...update });
61 }
62 }).pipe(
63 Effect.catch((cause) =>
64 Effect.logError("live transcription operation failed", { cause }).pipe(
65 Effect.andThen(fail("Live transcription failed. Please try recording again.")),
66 ),
67 ),
68 );
69 };
70 const reader = yield* socket.reader;
71 yield* Effect.gen(function* () {
72 const pending = yield* Queue.bounded<string | Uint8Array>(2);
73 yield* Effect.forever(Queue.take(pending).pipe(Effect.flatMap(receive))).pipe(
74 Effect.forkScoped,
75 );
76 while (true) {
77 const frames = yield* reader.pull;
78 yield* Effect.forEach(frames, (frame) => Queue.offer(pending, frame), {
79 discard: true,
80 });
81 }
82 }).pipe(
83 Effect.timeoutOrElse({
84 duration: "10 minutes",
85 // The socket read loop is already interrupted here; its writer can no longer send.
86 orElse: () => Effect.void,
87 }),
88 Effect.ignore,
89 );
90});
91
92export const speechStreamRouteLayer = Layer.unwrap(
93 Effect.gen(function* () {
94 const auth = yield* EnvironmentAuth.EnvironmentAuth;
95 const speech = yield* SpeechService.SpeechService;
96 return HttpRouter.add(
97 "GET",
98 SPEECH_STREAM_PATH,
99 Effect.gen(function* () {
100 const request = yield* HttpServerRequest.HttpServerRequest;
101 const session = yield* auth
102 .authenticateWebSocketUpgrade(request, "media")
103 .pipe(Effect.result);
104 if (session._tag === "Failure") return HttpServerResponse.empty({ status: 401 });
105 if (!session.success.scopes.includes(AuthOrchestrationOperateScope))
106 return HttpServerResponse.empty({ status: 403 });
107 const socket = yield* request.upgrade;
108 yield* runSpeechSocket(socket, speech).pipe(Effect.scoped);
109 return HttpServerResponse.empty();
110 }),
111 );
112 }),
113);