1import {
2 EnvironmentId,
3 type RelayClientInstallProgressEvent,
4 type ServerConfig,
5 type ServerLifecycleStreamEvent,
6 WS_METHODS,
7} from "@t3tools/contracts";
8import { describe, expect, it } from "@effect/vitest";
9import * as Cause from "effect/Cause";
10import * as Deferred from "effect/Deferred";
11import * as Effect from "effect/Effect";
12import * as Fiber from "effect/Fiber";
13import * as Option from "effect/Option";
14import * as Queue from "effect/Queue";
15import * as Ref from "effect/Ref";
16import * as Stream from "effect/Stream";
17import * as TestClock from "effect/testing/TestClock";
18import * as SubscriptionRef from "effect/SubscriptionRef";
19import { RpcClientError } from "effect/rpc";
20
21import {
22 AVAILABLE_CONNECTION_STATE,
23 PrimaryConnectionTarget,
24 type PreparedConnection,
25 type SupervisorConnectionState,
26} from "../connection/model.ts";
27import * as EnvironmentSupervisor from "../connection/supervisor.ts";
28import * as RpcSession from "../rpc/session.ts";
29import type { WsRpcProtocolClient } from "../rpc/protocol.ts";
30import { runStream, subscribe, subscribeDynamicWithSession, refreshVcsBranch } from "./client.ts";
31
32const TARGET = new PrimaryConnectionTarget({
33 environmentId: EnvironmentId.make("environment-1"),
34 label: "Test environment",
35 httpBaseUrl: "https://environment.example.test",
36 wsBaseUrl: "wss://environment.example.test",
37});
38
39const INSTALL_CHECKING: RelayClientInstallProgressEvent = {
40 type: "progress",
41 stage: "checking",
42};
43const INSTALL_DOWNLOADING: RelayClientInstallProgressEvent = {
44 type: "progress",
45 stage: "downloading",
46};
47
48function session(client: WsRpcProtocolClient): RpcSession.RpcSession {
49 return {
50 client,
51 initialConfig: Effect.never,
52 subscribeServerConfig: (input = {}) => client.subscribeServerConfig(input),
53 ready: Effect.void,
54 probe: Effect.void,
55 inboundFrames: Effect.succeed(0),
56 closed: Effect.never,
57 };
58}
59
60const makeHarness = Effect.fn("TestEnvironmentRpc.makeHarness")(function* () {
61 const state = yield* SubscriptionRef.make<SupervisorConnectionState>(AVAILABLE_CONNECTION_STATE);
62 const activeSession = yield* SubscriptionRef.make<Option.Option<RpcSession.RpcSession>>(
63 Option.none(),
64 );
65 const prepared = yield* SubscriptionRef.make<Option.Option<PreparedConnection>>(Option.none());
66 const retryCount = yield* Ref.make(0);
67 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
68 target: TARGET,
69 state,
70 session: activeSession,
71 prepared,
72 connect: Effect.void,
73 disconnect: Effect.void,
74 retryNow: Ref.update(retryCount, (count) => count + 1),
75 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
76 return {
77 activeSession,
78 retryCount,
79 supervisor,
80 };
81});
82
83describe("environment RPC", () => {
84 it.effect("binds finite streaming commands to one active session", () =>
85 Effect.gen(function* () {
86 const firstEvents = yield* Queue.unbounded<RelayClientInstallProgressEvent>();
87 const secondEvents = yield* Queue.unbounded<RelayClientInstallProgressEvent>();
88 const firstClient = {
89 [WS_METHODS.cloudInstallRelayClient]: () => Stream.fromQueue(firstEvents),
90 } as unknown as WsRpcProtocolClient;
91 const secondClient = {
92 [WS_METHODS.cloudInstallRelayClient]: () => Stream.fromQueue(secondEvents),
93 } as unknown as WsRpcProtocolClient;
94 const { activeSession, supervisor } = yield* makeHarness();
95
96 yield* SubscriptionRef.set(activeSession, Option.some(session(firstClient)));
97 const resultFiber = yield* runStream(WS_METHODS.cloudInstallRelayClient, {}).pipe(
98 Stream.take(2),
99 Stream.runCollect,
100 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
101 Effect.forkChild,
102 );
103 yield* Effect.yieldNow;
104
105 yield* Queue.offer(firstEvents, INSTALL_CHECKING);
106 yield* SubscriptionRef.set(activeSession, Option.some(session(secondClient)));
107 yield* Queue.offer(secondEvents, INSTALL_DOWNLOADING);
108 yield* Queue.offer(firstEvents, INSTALL_DOWNLOADING);
109
110 expect(yield* Fiber.join(resultFiber)).toEqual([INSTALL_CHECKING, INSTALL_DOWNLOADING]);
111 }),
112 );
113
114 it.effect("switches durable subscriptions when the supervisor replaces the session", () =>
115 Effect.gen(function* () {
116 const subscriptions: string[] = [];
117 const subscriptionChanges = yield* Queue.make<void>();
118 const firstClient = {
119 [WS_METHODS.subscribeTerminalMetadata]: () => {
120 subscriptions.push("first");
121 Queue.offerUnsafe(subscriptionChanges, undefined);
122 return Stream.never;
123 },
124 } as unknown as WsRpcProtocolClient;
125 const secondClient = {
126 [WS_METHODS.subscribeTerminalMetadata]: () => {
127 subscriptions.push("second");
128 Queue.offerUnsafe(subscriptionChanges, undefined);
129 return Stream.never;
130 },
131 } as unknown as WsRpcProtocolClient;
132 const { activeSession, retryCount, supervisor } = yield* makeHarness();
133 const awaitSubscriptions = Effect.fn("TestEnvironmentRpc.awaitSubscriptions")(function* (
134 count: number,
135 ) {
136 for (;;) {
137 if (subscriptions.length >= count) {
138 return;
139 }
140 yield* Queue.take(subscriptionChanges);
141 }
142 });
143
144 const subscriptionFiber = yield* subscribe(WS_METHODS.subscribeTerminalMetadata, {}).pipe(
145 Stream.runDrain,
146 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
147 Effect.forkChild,
148 );
149 yield* SubscriptionRef.set(activeSession, Option.some(session(firstClient)));
150 yield* awaitSubscriptions(1);
151 yield* SubscriptionRef.set(activeSession, Option.some(session(secondClient)));
152 yield* awaitSubscriptions(2);
153 yield* Fiber.interrupt(subscriptionFiber);
154
155 expect(subscriptions).toEqual(["first", "second"]);
156 expect(yield* Ref.get(retryCount)).toBe(0);
157 }),
158 );
159
160 it.effect("keeps the producer session on an old value buffered across a session switch", () =>
161 Effect.gen(function* () {
162 const firstSubscribed = yield* Deferred.make<void>();
163 const secondSubscribed = yield* Deferred.make<void>();
164 const firstValueBlocked = yield* Deferred.make<void>();
165 const releaseFirstValue = yield* Deferred.make<void>();
166 const firstValue = { source: "first", index: 1 } as unknown as ServerLifecycleStreamEvent;
167 const bufferedFirstValue = {
168 source: "first",
169 index: 2,
170 } as unknown as ServerLifecycleStreamEvent;
171 const secondValue = { source: "second", index: 1 } as unknown as ServerLifecycleStreamEvent;
172 const firstClient = {
173 [WS_METHODS.subscribeServerLifecycle]: () =>
174 Stream.fromEffect(Deferred.succeed(firstSubscribed, undefined)).pipe(
175 Stream.drain,
176 Stream.concat(Stream.fromIterable([firstValue, bufferedFirstValue])),
177 Stream.concat(Stream.never),
178 ),
179 } as unknown as WsRpcProtocolClient;
180 const secondClient = {
181 [WS_METHODS.subscribeServerLifecycle]: () =>
182 Stream.fromEffect(Deferred.succeed(secondSubscribed, undefined)).pipe(
183 Stream.drain,
184 Stream.concat(Stream.make(secondValue)),
185 Stream.concat(Stream.never),
186 ),
187 } as unknown as WsRpcProtocolClient;
188 const firstSession = session(firstClient);
189 const secondSession = session(secondClient);
190 const { activeSession, supervisor } = yield* makeHarness();
191
192 const resultFiber = yield* subscribeDynamicWithSession(
193 WS_METHODS.subscribeServerLifecycle,
194 () => Effect.succeed({}),
195 ).pipe(
196 Stream.mapEffect(([producerSession, value]) =>
197 value === firstValue
198 ? Deferred.succeed(firstValueBlocked, undefined).pipe(
199 Effect.andThen(Deferred.await(releaseFirstValue)),
200 Effect.as([producerSession, value] as const),
201 )
202 : Effect.succeed([producerSession, value] as const),
203 ),
204 Stream.take(3),
205 Stream.runCollect,
206 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
207 Effect.forkChild,
208 );
209
210 yield* SubscriptionRef.set(activeSession, Option.some(firstSession));
211 yield* Deferred.await(firstSubscribed);
212 yield* Deferred.await(firstValueBlocked);
213 yield* SubscriptionRef.set(activeSession, Option.some(secondSession));
214 yield* Deferred.await(secondSubscribed);
215 yield* Deferred.succeed(releaseFirstValue, undefined);
216
217 const result = yield* Fiber.join(resultFiber);
218 expect(result).toEqual([
219 [firstSession, firstValue],
220 [firstSession, bufferedFirstValue],
221 [secondSession, secondValue],
222 ]);
223 }),
224 );
225
226 it.effect("keeps durable subscriptions alive across a transport failure and new session", () =>
227 Effect.gen(function* () {
228 const subscriptions: string[] = [];
229 const subscriptionChanges = yield* Queue.make<void>();
230 const firstClient = {
231 [WS_METHODS.subscribeTerminalMetadata]: () => {
232 subscriptions.push("first");
233 Queue.offerUnsafe(subscriptionChanges, undefined);
234 return Stream.fail(
235 new RpcClientError.RpcClientError({
236 reason: new RpcClientError.RpcClientDefect({
237 message: "socket closed",
238 cause: new Error("socket closed"),
239 }),
240 }),
241 );
242 },
243 } as unknown as WsRpcProtocolClient;
244 const secondClient = {
245 [WS_METHODS.subscribeTerminalMetadata]: () => {
246 subscriptions.push("second");
247 Queue.offerUnsafe(subscriptionChanges, undefined);
248 return Stream.never;
249 },
250 } as unknown as WsRpcProtocolClient;
251 const { activeSession, retryCount, supervisor } = yield* makeHarness();
252
253 const subscriptionFiber = yield* subscribe(WS_METHODS.subscribeTerminalMetadata, {}).pipe(
254 Stream.runDrain,
255 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
256 Effect.forkChild,
257 );
258 yield* SubscriptionRef.set(activeSession, Option.some(session(firstClient)));
259 while (subscriptions.length < 1) {
260 yield* Queue.take(subscriptionChanges);
261 }
262 yield* SubscriptionRef.set(activeSession, Option.none());
263 yield* SubscriptionRef.set(activeSession, Option.some(session(secondClient)));
264
265 while (subscriptions.length < 2) {
266 yield* Queue.take(subscriptionChanges);
267 }
268 yield* Fiber.interrupt(subscriptionFiber);
269
270 expect(subscriptions).toEqual(["first", "second"]);
271 expect(yield* Ref.get(retryCount)).toBe(0);
272 }),
273 );
274
275 it.effect("surfaces domain subscription failures without reconnecting", () =>
276 Effect.gen(function* () {
277 const domainError = new Error("terminal subscription rejected");
278 const client = {
279 [WS_METHODS.subscribeTerminalMetadata]: () => Stream.fail(domainError),
280 } as unknown as WsRpcProtocolClient;
281 const { activeSession, retryCount, supervisor } = yield* makeHarness();
282
283 yield* SubscriptionRef.set(activeSession, Option.some(session(client)));
284 const error = yield* subscribe(WS_METHODS.subscribeTerminalMetadata, {}).pipe(
285 Stream.runDrain,
286 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
287 Effect.flip,
288 );
289
290 expect(error).toBe(domainError);
291 expect(yield* Ref.get(retryCount)).toBe(0);
292 }),
293 );
294
295 it.effect("keeps handled domain failures dormant until a replacement session arrives", () =>
296 Effect.gen(function* () {
297 const domainError = new Error("terminal subscription rejected");
298 const subscriptions: string[] = [];
299 const subscriptionChanges = yield* Queue.make<void>();
300 const observedFailures: Error[] = [];
301 const failureObserved = yield* Deferred.make<void>();
302 const firstClient = {
303 [WS_METHODS.subscribeTerminalMetadata]: () => {
304 subscriptions.push("first");
305 Queue.offerUnsafe(subscriptionChanges, undefined);
306 return Stream.fail(domainError);
307 },
308 } as unknown as WsRpcProtocolClient;
309 const secondClient = {
310 [WS_METHODS.subscribeTerminalMetadata]: () => {
311 subscriptions.push("second");
312 Queue.offerUnsafe(subscriptionChanges, undefined);
313 return Stream.never;
314 },
315 } as unknown as WsRpcProtocolClient;
316 const { activeSession, retryCount, supervisor } = yield* makeHarness();
317
318 yield* SubscriptionRef.set(activeSession, Option.some(session(firstClient)));
319 const subscriptionFiber = yield* subscribe(
320 WS_METHODS.subscribeTerminalMetadata,
321 {},
322 {
323 onExpectedFailure: (cause) =>
324 Effect.sync(() => {
325 observedFailures.push(Cause.squash(cause) as Error);
326 Deferred.doneUnsafe(failureObserved, Effect.void);
327 }),
328 },
329 ).pipe(
330 Stream.runDrain,
331 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
332 Effect.forkChild,
333 );
334 yield* Deferred.await(failureObserved);
335
336 expect(subscriptions).toEqual(["first"]);
337 expect(observedFailures).toEqual([domainError]);
338
339 yield* SubscriptionRef.set(activeSession, Option.some(session(secondClient)));
340 while (subscriptions.length < 2) {
341 yield* Queue.take(subscriptionChanges);
342 }
343 yield* Fiber.interrupt(subscriptionFiber);
344
345 expect(subscriptions).toEqual(["first", "second"]);
346 expect(yield* Ref.get(retryCount)).toBe(0);
347 }),
348 );
349
350 it.effect.each(["input", "stream"] as const)(
351 "reports %s subscription defects and resubscribes on the next session",
352 (where) =>
353 Effect.gen(function* () {
354 const defect = new Error("subscription invariant failed");
355 let expectedFailureCount = 0;
356 let inputs = 0;
357 let streams = 0;
358 const observedDefects: unknown[] = [];
359 const defectObserved = yield* Deferred.make<void>();
360 const resubscribed = yield* Deferred.make<void>();
361 const firstClient = {
362 [WS_METHODS.subscribeTerminalMetadata]: () => {
363 streams += 1;
364 return where === "stream" ? Stream.die(defect) : Stream.never;
365 },
366 } as unknown as WsRpcProtocolClient;
367 const secondClient = {
368 [WS_METHODS.subscribeTerminalMetadata]: () => {
369 Deferred.doneUnsafe(resubscribed, Effect.void);
370 return Stream.never;
371 },
372 } as unknown as WsRpcProtocolClient;
373 const { activeSession, supervisor } = yield* makeHarness();
374
375 yield* SubscriptionRef.set(activeSession, Option.some(session(firstClient)));
376 const subscriptionFiber = yield* subscribeDynamicWithSession(
377 WS_METHODS.subscribeTerminalMetadata,
378 () =>
379 Effect.suspend(() => {
380 inputs += 1;
381 return where === "input" && inputs === 1 ? Effect.die(defect) : Effect.succeed({});
382 }),
383 {
384 onDefect: (cause) =>
385 Effect.sync(() => {
386 observedDefects.push(Cause.squash(cause));
387 Deferred.doneUnsafe(defectObserved, Effect.void);
388 }),
389 onExpectedFailure: () =>
390 Effect.sync(() => {
391 expectedFailureCount += 1;
392 }),
393 retryExpectedFailureAfter: "250 millis",
394 },
395 ).pipe(
396 Stream.runDrain,
397 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
398 Effect.forkChild,
399 );
400 yield* Deferred.await(defectObserved);
401
402 expect(inputs).toBe(1);
403 expect(streams).toBe(where === "input" ? 0 : 1);
404 expect(expectedFailureCount).toBe(0);
405 expect(observedDefects).toEqual([defect]);
406
407 // Let the defect settle before the next session arrives, as it does on a real device.
408 yield* TestClock.adjust("1 second");
409 expect(subscriptionFiber.pollUnsafe()).toBeUndefined();
410 yield* SubscriptionRef.set(activeSession, Option.some(session(secondClient)));
411 yield* Deferred.await(resubscribed);
412 yield* Fiber.interrupt(subscriptionFiber);
413
414 expect(inputs).toBe(2);
415 }),
416 );
417});
418
419it.effect.each([true, false])(
420 "refreshes branches using the connected server capability (%s)",
421 (supported) =>
422 Effect.gen(function* () {
423 const { activeSession, supervisor } = yield* makeHarness();
424 const inputs: unknown[] = [];
425 const client = {
426 [supported ? WS_METHODS.vcsRefreshBranch : WS_METHODS.vcsRefreshStatus]: (
427 input: unknown,
428 ) => {
429 inputs.push(input);
430 return Effect.void;
431 },
432 } as unknown as WsRpcProtocolClient;
433 yield* SubscriptionRef.set(
434 activeSession,
435 Option.some({
436 ...session(client),
437 initialConfig: Effect.succeed({
438 environment: { capabilities: supported ? { vcsBranchStatus: true } : {} },
439 } as ServerConfig),
440 }),
441 );
442 expect(
443 yield* refreshVcsBranch({ cwd: "/repo" }).pipe(
444 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
445 ),
446 ).toBeUndefined();
447 expect(inputs).toEqual(supported ? [{ cwd: "/repo" }] : []);
448 }),
449);
450
451it.effect("binds branch capability selection and execution to one session", () =>
452 Effect.gen(function* () {
453 const { activeSession, supervisor } = yield* makeHarness();
454 const configRequested = yield* Deferred.make<void>();
455 const config = yield* Deferred.make<ServerConfig>();
456 const executed = yield* Deferred.make<void>();
457 const original = {
458 [WS_METHODS.vcsRefreshBranch]: () =>
459 Deferred.succeed(executed, undefined).pipe(Effect.asVoid),
460 } as unknown as WsRpcProtocolClient;
461 const replacement = {
462 [WS_METHODS.vcsRefreshStatus]: () => Effect.succeed({ refName: "replacement" }),
463 } as unknown as WsRpcProtocolClient;
464 yield* SubscriptionRef.set(
465 activeSession,
466 Option.some({
467 ...session(original),
468 initialConfig: Deferred.succeed(configRequested, undefined).pipe(
469 Effect.andThen(Deferred.await(config)),
470 ),
471 }),
472 );
473 const refresh = yield* refreshVcsBranch({ cwd: "/repo" }).pipe(
474 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
475 Effect.forkChild,
476 );
477 yield* Deferred.await(configRequested);
478 yield* SubscriptionRef.set(activeSession, Option.some(session(replacement)));
479 yield* Deferred.succeed(config, {
480 environment: { capabilities: { vcsBranchStatus: true } },
481 } as ServerConfig);
482 expect(yield* Fiber.join(refresh)).toBeUndefined();
483 yield* Deferred.await(executed);
484 }),
485);