1import { expect, it } from "@effect/vitest";
2import * as NodeHttpServer from "@effect/platform-node/NodeHttpServer";
3import { NodeWS } from "@effect/platform-node/NodeSocket";
4import { AuthOrchestrationReadScope, AuthSessionId, EnvironmentId } from "@t3tools/contracts";
5import * as Context from "effect/Context";
6import * as Deferred from "effect/Deferred";
7import * as Effect from "effect/Effect";
8import * as Layer from "effect/Layer";
9import { FetchHttpClient, HttpRouter, HttpServer } from "effect/http";
10import * as EnvironmentAuth from "../auth/EnvironmentAuth.ts";
11import * as SessionStore from "../auth/SessionStore.ts";
12import * as Hub from "./Hub.ts";
13import * as HubHttp from "./HubHttp.ts";
14import * as HubProxy from "./HubProxy.ts";
15
16it.effect.each(["client", "machine"] as const)(
17 "streams opaque binary frames with a private leaf grant and closes both sockets on %s revocation",
18 (invalidation) =>
19 Effect.gen(function* () {
20 const upstream = yield* Effect.acquireRelease(
21 Effect.sync(() => new NodeWS.WebSocketServer({ port: 0 })),
22 (server) =>
23 Effect.promise(
24 () =>
25 new Promise<void>((resolve) => {
26 for (const client of server.clients) client.terminate();
27 server.close(() => resolve());
28 }),
29 ),
30 );
31 if (!upstream.address())
32 yield* Effect.promise(
33 () => new Promise<void>((resolve) => upstream.once("listening", resolve)),
34 );
35 const address = upstream.address();
36 if (!address || typeof address !== "object") throw new Error("Missing test port");
37 const closedUpstream = Promise.withResolvers<void>();
38 const requests: Array<{ path?: string; authorization?: string }> = [];
39 upstream.on("connection", (socket, request) => {
40 requests.push({
41 ...(request.url ? { path: request.url } : {}),
42 ...(request.headers.authorization
43 ? { authorization: request.headers.authorization }
44 : {}),
45 });
46 socket.on("message", (data, binary) => socket.send(data, { binary }));
47 socket.on("close", () => closedUpstream.resolve());
48 socket.send(Uint8Array.of(0x83));
49 });
50 const revokedClient = yield* Deferred.make<void>();
51 const revokedMachine = yield* Deferred.make<void>();
52 const dependencies = Layer.mergeAll(
53 Layer.succeed(Hub.Hub, {
54 connect: () =>
55 Effect.succeed({
56 url: `http://127.0.0.1:${address.port}/base/`,
57 token: "leaf-private",
58 invalidated: Deferred.await(revokedMachine),
59 }),
60 } as unknown as Hub.Hub["Service"]),
61 Layer.succeed(EnvironmentAuth.EnvironmentAuth, {
62 authenticateWebSocketUpgrade: () =>
63 Effect.succeed({
64 sessionId: AuthSessionId.make("phone"),
65 subject: "phone",
66 method: "bearer-access-token",
67 scopes: [AuthOrchestrationReadScope],
68 }),
69 } as unknown as EnvironmentAuth.EnvironmentAuth["Service"]),
70 Layer.succeed(SessionStore.SessionStore, {
71 awaitInvalidation: () => Deferred.await(revokedClient),
72 } as unknown as SessionStore.SessionStore["Service"]),
73 FetchHttpClient.layer,
74 );
75 const services = yield* Layer.build(
76 HttpRouter.serve(HubHttp.layer, { disableListenLog: true }).pipe(
77 Layer.provideMerge(HubProxy.layer.pipe(Layer.provide(dependencies))),
78 Layer.provideMerge(dependencies),
79 Layer.provideMerge(NodeHttpServer.layerTest),
80 ),
81 );
82 const origin = HttpServer.formatAddress(
83 Context.get(services, HttpServer.HttpServer).address,
84 ).replace(/^http/, "ws");
85 const client = yield* Effect.acquireRelease(
86 Effect.sync(
87 () =>
88 new NodeWS.WebSocket(
89 `${origin}/environments/${EnvironmentId.make("leaf")}/ws?wsTicket=hub-private&protocolVersion=1`,
90 ),
91 ),
92 (socket) => Effect.sync(() => socket.close()),
93 );
94 client.binaryType = "arraybuffer";
95 const received = Promise.withResolvers<void>();
96 const closedClient = Promise.withResolvers<void>();
97 const frames: number[][] = [];
98 client.addEventListener("error", (event) => received.reject(event.error));
99 client.addEventListener("close", () => closedClient.resolve());
100 client.addEventListener("message", ({ data }) => {
101 frames.push(Array.from(new Uint8Array(data as ArrayBuffer)));
102 if (frames.length === 1) client.send(Uint8Array.of(0x03, 1));
103 else received.resolve();
104 });
105 yield* Effect.promise(() => received.promise);
106 expect(frames).toEqual([[0x83], [0x03, 1]]);
107 expect(requests).toEqual([
108 { path: "/base/ws?protocolVersion=1", authorization: "Bearer leaf-private" },
109 ]);
110 yield* Deferred.succeed(
111 invalidation === "client" ? revokedClient : revokedMachine,
112 undefined,
113 );
114 yield* Effect.promise(() => Promise.all([closedClient.promise, closedUpstream.promise]));
115 }),
116);