1import { expect, it } from "@effect/vitest";
2import * as NodeServices from "@effect/platform-node/NodeServices";
3import {
4 ORCHESTRATION_PROTOCOL_QUERY_PARAM,
5 ORCHESTRATION_PROTOCOL_VERSION,
6 WS_METHODS,
7 type DesktopTerminalRequest,
8 type DesktopTerminalResponse,
9} from "@t3tools/contracts";
10import * as Effect from "effect/Effect";
11import * as Deferred from "effect/Deferred";
12import * as Layer from "effect/Layer";
13import * as Option from "effect/Option";
14import * as Queue from "effect/Queue";
15import * as Stream from "effect/Stream";
16import * as ServerConfig from "../config.ts";
17import * as ServerEnvironment from "../environment/ServerEnvironment.ts";
18import * as EnvironmentAuth from "../auth/EnvironmentAuth.ts";
19import * as SessionStore from "../auth/SessionStore.ts";
20import * as ServerSecretStore from "../auth/ServerSecretStore.ts";
21import * as SqlitePersistence from "../persistence/Sqlite.ts";
22import * as TerminalManager from "./Manager.ts";
23import * as DesktopTerminalRpc from "./DesktopTerminalRpc.ts";
24
25const config = ServerConfig.layerTest(process.cwd(), { prefix: "t3-terminal-pipe-test-" });
26const auth = EnvironmentAuth.layer.pipe(
27 Layer.provideMerge(SqlitePersistence.layerMemory),
28 Layer.provide(ServerSecretStore.layer),
29 Layer.provideMerge(ServerEnvironment.layerIdentity),
30 Layer.provide(config),
31);
32
33it.layer(NodeServices.layer)("DesktopTerminalRpc", (it) => {
34 it.effect("port close releases the running RPC, stream subscription and session connection", () =>
35 Effect.gen(function* () {
36 const requests = yield* Queue.unbounded<DesktopTerminalRequest>();
37 const responses = yield* Queue.unbounded<DesktopTerminalResponse>();
38 const unsubscribed = yield* Deferred.make<void>();
39 const sessions = yield* SessionStore.SessionStore;
40 const identity = yield* ServerEnvironment.ServerEnvironmentIdentity;
41 const environmentId = yield* identity.getEnvironmentId;
42 const session = yield* sessions.issue({ scopes: ["terminal:read"] });
43 const ticket = yield* sessions.issueWebSocketToken(session.sessionId);
44 const manager = Layer.mock(TerminalManager.TerminalManager, {
45 observeStream: (input, listener) =>
46 listener({
47 type: "output",
48 threadId: input.threadId,
49 terminalId: input.terminalId,
50 data: "ready",
51 }).pipe(
52 Effect.as(() => {
53 Deferred.doneUnsafe(unsubscribed, Effect.void);
54 }),
55 ),
56 });
57 yield* DesktopTerminalRpc.run(Stream.fromQueue(requests), (response) =>
58 Queue.offer(responses, response).pipe(Effect.asVoid),
59 ).pipe(Effect.provide(manager), Effect.forkScoped);
60 expect((yield* Queue.take(responses)).type).toBe("ready");
61 const url = `ws://127.0.0.1/ws?${ORCHESTRATION_PROTOCOL_QUERY_PARAM}=${ORCHESTRATION_PROTOCOL_VERSION}&wsTicket=${ticket.token}`;
62 yield* Queue.offer(requests, { type: "open", id: "port-1", environmentId, url });
63 yield* Queue.offer(requests, {
64 type: "message",
65 id: "port-1",
66 data: JSON.stringify({
67 _tag: "Request",
68 id: "1",
69 tag: WS_METHODS.terminalObserve,
70 payload: { threadId: "thread", terminalId: "term" },
71 headers: [],
72 }),
73 });
74 const response = yield* Queue.take(responses);
75 expect(response.type).toBe("message");
76 expect(
77 (yield* sessions.listActive()).find((value) => value.sessionId === session.sessionId)
78 ?.connected,
79 ).toBe(true);
80 yield* Queue.offer(requests, { type: "close", id: "port-1" });
81 yield* Deferred.await(unsubscribed);
82 const closed = yield* Stream.fromQueue(responses).pipe(
83 Stream.filter((response) => response.type === "close"),
84 Stream.runHead,
85 );
86 expect(Option.getOrThrow(closed)).toEqual({ type: "close", id: "port-1" });
87 expect(
88 (yield* sessions.listActive()).find((value) => value.sessionId === session.sessionId)
89 ?.connected,
90 ).toBe(false);
91
92 yield* Queue.offer(requests, {
93 type: "open",
94 id: "wrong-environment",
95 environmentId: "other-environment",
96 url,
97 });
98 expect(yield* Queue.take(responses)).toEqual({ type: "close", id: "wrong-environment" });
99 yield* Queue.offer(requests, {
100 type: "open",
101 id: "invalid-ticket",
102 environmentId,
103 url: url.replace(ticket.token, "invalid"),
104 });
105 expect(yield* Queue.take(responses)).toEqual({ type: "close", id: "invalid-ticket" });
106 yield* Queue.offer(requests, { type: "open", id: "read-only", environmentId, url });
107 yield* Queue.offer(requests, {
108 type: "message",
109 id: "read-only",
110 data: JSON.stringify({
111 _tag: "Request",
112 id: "2",
113 tag: WS_METHODS.terminalWrite,
114 payload: { threadId: "thread", terminalId: "term", data: "j" },
115 headers: [],
116 }),
117 });
118 const denied = yield* Queue.take(responses);
119 expect(denied.type).toBe("message");
120 if (denied.type === "message") expect(denied.data).toContain("EnvironmentAuthorizationError");
121 yield* sessions.revoke(session.sessionId);
122 const revoked = yield* Stream.fromQueue(responses).pipe(
123 Stream.filter((response) => response.type === "close"),
124 Stream.runHead,
125 );
126 expect(Option.getOrThrow(revoked)).toEqual({ type: "close", id: "read-only" });
127 }).pipe(Effect.provide(auth), Effect.scoped),
128 );
129});