1import * as NodeHttpServer from "@effect/platform-node/NodeHttpServer";
2import { it } from "@effect/vitest";
3import { expect } from "vite-plus/test";
4import * as Effect from "effect/Effect";
5import * as Layer from "effect/Layer";
6import * as Schema from "effect/Schema";
7import { HttpRouter, HttpServer, HttpServerRequest, HttpServerResponse } from "effect/http";
8import { SpeechStreamEvent, DEFAULT_SPEECH_TRANSCRIPTION_OPTIONS } from "@t3tools/contracts";
9import type { SpeechService, SpeechStream } from "./SpeechService.ts";
10import { runSpeechSocket } from "./stream.ts";
11import type * as NetAddress from "effect/net/NetAddress";
12
13const decodeEvent = Schema.decodeUnknownSync(Schema.fromJsonString(SpeechStreamEvent));
14
15function client(url: string) {
16 const socket = new WebSocket(url);
17 socket.addEventListener("open", () =>
18 socket.send(JSON.stringify({ type: "start", options: DEFAULT_SPEECH_TRANSCRIPTION_OPTIONS })),
19 );
20 const events: SpeechStreamEvent[] = [];
21 let waiter: ReturnType<typeof Promise.withResolvers<SpeechStreamEvent>> | undefined;
22 socket.addEventListener("message", ({ data }) => {
23 const event = decodeEvent(data);
24 if (waiter) {
25 waiter.resolve(event);
26 waiter = undefined;
27 } else events.push(event);
28 });
29 socket.addEventListener("error", () => waiter?.reject(new Error("Speech socket failed.")));
30 socket.addEventListener("close", () => waiter?.reject(new Error("Speech socket closed.")));
31 return {
32 socket,
33 next: () => {
34 const event = events.shift();
35 if (event) return Promise.resolve(event);
36 waiter = Promise.withResolvers<SpeechStreamEvent>();
37 return waiter.promise;
38 },
39 };
40}
41
42const withServer = Effect.fn("test.speech.withServer")(function* (
43 startStream: SpeechService["Service"]["startStream"],
44 run: (url: string) => Promise<void>,
45) {
46 const routes = HttpRouter.add(
47 "GET",
48 "/ws/voice",
49 Effect.gen(function* () {
50 const request = yield* HttpServerRequest.HttpServerRequest;
51 const socket = yield* request.upgrade;
52 yield* runSpeechSocket(socket, { startStream }).pipe(Effect.scoped);
53 return HttpServerResponse.empty();
54 }),
55 );
56 yield* Layer.build(HttpRouter.serve(routes, { disableLogger: true, disableListenLog: true }));
57 const server = yield* HttpServer.HttpServer;
58 const address = server.address as NetAddress.InetAddress;
59 const host = address._tag === "InetAddressV6" ? "[::1]" : "127.0.0.1";
60 const url = `ws://${host}:${address.port}/ws/voice`;
61 yield* Effect.promise(() => run(url));
62});
63
64it.live("acknowledges binary audio and returns the finalized transcript over a real socket", () => {
65 const closed = Promise.withResolvers<void>();
66 const received: number[] = [];
67 const stream: SpeechStream = {
68 feed: (bytes) =>
69 Effect.sync(() => {
70 received.push(
71 new DataView(bytes.buffer, bytes.byteOffset, bytes.byteLength).getFloat32(0, true),
72 );
73 return { revision: received.length, text: { committed: "hello ", tentative: "world" } };
74 }),
75 finish: Effect.sync(() => {
76 expect(received).toEqual([0.25, 0.5]);
77 return "hello world";
78 }),
79 };
80 return withServer(
81 () => Effect.acquireRelease(Effect.succeed(stream), () => Effect.sync(() => closed.resolve())),
82 async (url) => {
83 const connection = client(url);
84 try {
85 expect(await connection.next()).toEqual({ type: "ready" });
86 for (const value of [0.25, 0.5]) {
87 connection.socket.send(new Float32Array([value]));
88 expect(await connection.next()).toMatchObject({
89 type: "update",
90 text: { tentative: "world" },
91 });
92 }
93 connection.socket.send(JSON.stringify({ type: "finish" }));
94 expect(await connection.next()).toEqual({ type: "finished", text: "hello world" });
95 await closed.promise;
96 } finally {
97 connection.socket.close();
98 }
99 },
100 ).pipe(Effect.scoped, Effect.provide(NodeHttpServer.layerTest));
101});
102
103it.live("releases in-flight inference after disconnect", () => {
104 const feeding = Promise.withResolvers<void>();
105 const closed = Promise.withResolvers<void>();
106 const stream: SpeechStream = {
107 feed: () => Effect.sync(() => feeding.resolve()).pipe(Effect.andThen(Effect.never)),
108 finish: Effect.succeed("unexpected"),
109 };
110 return withServer(
111 () => Effect.acquireRelease(Effect.succeed(stream), () => Effect.sync(() => closed.resolve())),
112 async (url) => {
113 const connection = client(url);
114 try {
115 await connection.next();
116 connection.socket.send(new Float32Array([0.25]));
117 await feeding.promise;
118 connection.socket.close();
119 await closed.promise;
120 } finally {
121 connection.socket.close();
122 }
123 },
124 ).pipe(Effect.scoped, Effect.provide(NodeHttpServer.layerTest));
125});
126
127it.live("processes pipelined audio before the finish command", () => {
128 const firstFeed = Promise.withResolvers<void>();
129 const releaseFirst = Promise.withResolvers<void>();
130 const received: number[] = [];
131 const stream: SpeechStream = {
132 feed: (bytes) =>
133 Effect.promise(async () => {
134 const value = new DataView(bytes.buffer, bytes.byteOffset, bytes.byteLength).getFloat32(
135 0,
136 true,
137 );
138 received.push(value);
139 if (received.length === 1) {
140 firstFeed.resolve();
141 await releaseFirst.promise;
142 }
143 return { revision: received.length, text: null };
144 }),
145 finish: Effect.sync(() => {
146 expect(received).toEqual([0.25, 0.5]);
147 return "done";
148 }),
149 };
150 return withServer(
151 () => Effect.succeed(stream),
152 async (url) => {
153 const connection = client(url);
154 try {
155 expect(await connection.next()).toEqual({ type: "ready" });
156 connection.socket.send(new Float32Array([0.25]));
157 await firstFeed.promise;
158 connection.socket.send(new Float32Array([0.5]));
159 connection.socket.send(JSON.stringify({ type: "finish" }));
160 releaseFirst.resolve();
161 expect(await connection.next()).toMatchObject({ type: "update", revision: 1 });
162 expect(await connection.next()).toMatchObject({ type: "update", revision: 2 });
163 expect(await connection.next()).toEqual({ type: "finished", text: "done" });
164 } finally {
165 connection.socket.close();
166 }
167 },
168 ).pipe(Effect.scoped, Effect.provide(NodeHttpServer.layerTest));
169});