1import { expect, it } from "@effect/vitest";
2import * as NodeWS from "@effect/platform-node/NodeWS";
3import * as NodeNet from "node:net";
4import * as Context from "effect/Context";
5import * as Effect from "effect/Effect";
6import * as Fiber from "effect/Fiber";
7import * as Layer from "effect/Layer";
8import { HttpRouter, HttpServer } from "effect/http";
9import * as Socket from "effect/socket/Socket";
10import * as NodeHttpServer from "@effect/platform-node/NodeHttpServer";
11import {
12 AuthOrchestrationOperateScope,
13 AuthSessionId,
14 LOCAL_DEVICE_HOST_ID,
15} from "@t3tools/contracts";
16import * as EnvironmentAuth from "../auth/EnvironmentAuth.ts";
17import * as DeviceService from "./DeviceService.ts";
18import * as DeviceHubProxy from "./DeviceHubProxy.ts";
19
20it.effect("preserves binary input and admission frames through the real socket proxy", () =>
21 Effect.gen(function* () {
22 const hub = yield* Effect.acquireRelease(
23 Effect.sync(() => new NodeWS.WebSocketServer({ port: 0 })),
24 (server) =>
25 Effect.promise(
26 () =>
27 new Promise<void>((resolve) => {
28 for (const client of server.clients) client.terminate();
29 server.close(() => resolve());
30 }),
31 ),
32 );
33 if (!hub.address())
34 yield* Effect.promise(() => new Promise<void>((resolve) => hub.once("listening", resolve)));
35 const address = hub.address();
36 if (typeof address !== "object" || !address) throw new Error("Missing hub port");
37 hub.on("connection", (socket) => {
38 socket.send(Uint8Array.of(0x83));
39 socket.on("message", (data, binary) => socket.send(data, { binary }));
40 });
41 const services = yield* Layer.build(
42 HttpRouter.serve(DeviceHubProxy.layer, { disableListenLog: true }).pipe(
43 Layer.provideMerge(NodeHttpServer.layerTest),
44 Layer.provide(
45 Layer.succeed(EnvironmentAuth.EnvironmentAuth, {
46 authenticateWebSocketUpgrade: () =>
47 Effect.succeed({
48 sessionId: AuthSessionId.make("test"),
49 subject: "test",
50 method: "bearer-access-token",
51 scopes: [AuthOrchestrationOperateScope],
52 }),
53 } as unknown as EnvironmentAuth.EnvironmentAuth["Service"]),
54 ),
55 Layer.provide(
56 Layer.succeed(DeviceService.DeviceService, {
57 currentReadiness: () =>
58 Effect.succeed({
59 hostId: LOCAL_DEVICE_HOST_ID,
60 hub: { origin: `http://127.0.0.1:${address.port}` },
61 }),
62 } as DeviceService.DeviceService["Service"]),
63 ),
64 ),
65 );
66 const origin = HttpServer.formatAddress(
67 Context.get(services, HttpServer.HttpServer).address,
68 ).replace(/^http/, "ws");
69 const client = yield* Effect.acquireRelease(
70 Effect.sync(
71 () =>
72 new WebSocket(
73 `${origin}/api/device-hub/vendor/serve-sim/helper/ws?device=test&wsTicket=test`,
74 ),
75 ),
76 (socket) => Effect.sync(() => socket.close()),
77 );
78 client.binaryType = "arraybuffer";
79 const frames: number[][] = [];
80 const received = Promise.withResolvers<void>();
81 client.addEventListener("error", () => received.reject(new Error("Proxy failed")));
82 client.addEventListener("message", ({ data }) => {
83 frames.push(Array.from(new Uint8Array(data as ArrayBuffer)));
84 if (frames.length === 1) {
85 client.send(Uint8Array.of(0x03, 1));
86 client.send(Uint8Array.of(0x03, 2));
87 } else if (frames.length === 3) received.resolve();
88 });
89 yield* Effect.promise(() => received.promise);
90 expect(frames).toEqual([[0x83], [0x03, 1], [0x03, 2]]);
91 }),
92);
93
94it.live("closes an upstream socket that is still connecting without an unhandled error", () =>
95 Effect.gen(function* () {
96 const dialed = Promise.withResolvers<NodeNet.Socket>();
97 // Accepts the connection and never answers the WebSocket handshake.
98 const hub = yield* Effect.acquireRelease(
99 Effect.promise(
100 () =>
101 new Promise<NodeNet.Server>((resolve) => {
102 const server = NodeNet.createServer((socket) => dialed.resolve(socket));
103 server.listen(0, "127.0.0.1", () => resolve(server));
104 }),
105 ),
106 (server) =>
107 Effect.promise(() => new Promise<void>((resolve) => server.close(() => resolve()))),
108 );
109 const address = hub.address();
110 if (typeof address !== "object" || !address) throw new Error("Missing hub port");
111 // Handshake headers, as Simlock's routes send, select the `ws` package over the global WebSocket.
112 const upstream = yield* Socket.makeWebSocket(`ws://127.0.0.1:${address.port}`, {
113 headers: { authorization: "Bearer test" },
114 }).pipe(Effect.provide(DeviceHubProxy.upstreamWebSocketConstructor));
115 const reading = yield* Effect.forkChild(
116 Effect.scoped(Effect.flatMap(upstream.reader, ({ pull }) => pull)),
117 );
118 const connection = yield* Effect.promise(() => dialed.promise);
119 const abandoned = new Promise<void>((resolve) => connection.once("close", () => resolve()));
120 yield* Fiber.interrupt(reading);
121 yield* Effect.promise(() => abandoned);
122 }),
123);