1import { EnvironmentId } from "@t3tools/contracts";
2import type {
3 RelayClientEnvironmentRecord,
4 RelayEnvironmentStatusResponse,
5} from "@t3tools/contracts/relay";
6import { describe, expect, it } from "@effect/vitest";
7import * as Deferred from "effect/Deferred";
8import * as Effect from "effect/Effect";
9import * as Fiber from "effect/Fiber";
10import * as Layer from "effect/Layer";
11import * as Option from "effect/Option";
12import * as Ref from "effect/Ref";
13import * as Stream from "effect/Stream";
14import * as SubscriptionRef from "effect/SubscriptionRef";
15import { Atom, AtomRegistry } from "effect/reactivity";
16
17import * as ManagedRelay from "./managedRelay.ts";
18import * as ClientCapabilities from "../platform/capabilities.ts";
19import * as Connectivity from "../connection/connectivity.ts";
20import { ConnectionBlockedError, type NetworkStatus } from "../connection/model.ts";
21import * as ConnectionWakeups from "../connection/wakeups.ts";
22import * as RelayEnvironmentDiscovery from "./discovery.ts";
23import { NETWORK_BLOCKING_HINT } from "../errors/network.ts";
24import { createRelayEnvironmentDiscoveryAtoms } from "../state/relayDiscovery.ts";
25
26const environments = [
27 {
28 environmentId: EnvironmentId.make("environment-1"),
29 label: "Environment One",
30 endpoint: {
31 httpBaseUrl: "https://one.example.test",
32 wsBaseUrl: "wss://one.example.test",
33 providerKind: "cloudflare_tunnel",
34 },
35 linkedAt: "2026-06-01T00:00:00.000Z",
36 },
37 {
38 environmentId: EnvironmentId.make("environment-2"),
39 label: "Environment Two",
40 endpoint: {
41 httpBaseUrl: "https://two.example.test",
42 wsBaseUrl: "wss://two.example.test",
43 providerKind: "cloudflare_tunnel",
44 },
45 linkedAt: "2026-06-01T00:00:00.000Z",
46 },
47] satisfies ReadonlyArray<RelayClientEnvironmentRecord>;
48
49function status(
50 environment: RelayClientEnvironmentRecord,
51 value: "online" | "offline",
52): RelayEnvironmentStatusResponse {
53 return {
54 environmentId: environment.environmentId,
55 endpoint: environment.endpoint,
56 status: value,
57 checkedAt: "2026-06-01T00:00:00.000Z",
58 };
59}
60
61const makeHarness = Effect.fn("RelayDiscoveryTest.makeHarness")(function* () {
62 const networkStatus = yield* SubscriptionRef.make<NetworkStatus>("online");
63 const listCalls = yield* Ref.make(0);
64 const listedEnvironments =
65 yield* Ref.make<ReadonlyArray<RelayClientEnvironmentRecord>>(environments);
66 const listFailure = yield* Ref.make<ManagedRelay.ManagedRelayClientError | null>(null);
67 const secondListCall = yield* Deferred.make<void>();
68 const clerkToken = yield* Ref.make<string | null>("clerk-token");
69 const sessionIdentity = { accountId: "account-1" };
70 const wakeups = yield* SubscriptionRef.make<{
71 readonly sequence: number;
72 readonly reason: "application-active" | "credentials-changed";
73 }>({
74 sequence: 0,
75 reason: "application-active",
76 });
77 const statusRequests = yield* Ref.make(
78 new Map<
79 string,
80 Deferred.Deferred<RelayEnvironmentStatusResponse, ManagedRelay.ManagedRelayClientError>
81 >(),
82 );
83 for (const environment of environments) {
84 const request = yield* Deferred.make<
85 RelayEnvironmentStatusResponse,
86 ManagedRelay.ManagedRelayClientError
87 >();
88 yield* Ref.update(statusRequests, (current) => {
89 const next = new Map(current);
90 next.set(environment.environmentId, request);
91 return next;
92 });
93 }
94
95 const client = ManagedRelay.ManagedRelayClient.of({
96 relayUrl: "https://relay.example.test",
97 listEnvironments: () =>
98 Effect.gen(function* () {
99 const count = yield* Ref.updateAndGet(listCalls, (current) => current + 1);
100 if (count >= 2) {
101 yield* Deferred.succeed(secondListCall, undefined);
102 }
103 const failure = yield* Ref.get(listFailure);
104 if (failure) {
105 return yield* failure;
106 }
107 return yield* Ref.get(listedEnvironments);
108 }),
109 getEnvironmentStatus: ({ environmentId }) =>
110 Ref.get(statusRequests).pipe(
111 Effect.flatMap((requests) => Deferred.await(requests.get(environmentId)!)),
112 ),
113 listDevices: () => Effect.die("unused"),
114 createEnvironmentLinkChallenge: () => Effect.die("unused"),
115 linkEnvironment: () => Effect.die("unused"),
116 unlinkEnvironment: () => Effect.die("unused"),
117 getT3ChatLink: () => Effect.die("unused"),
118 createT3ChatLinkRequest: () => Effect.die("unused"),
119 unlinkT3Chat: () => Effect.die("unused"),
120 connectEnvironment: () => Effect.die("unused"),
121 registerDevice: () => Effect.die("unused"),
122 unregisterDevice: () => Effect.die("unused"),
123 registerLiveActivity: () => Effect.die("unused"),
124 getAgentActivitySnapshot: () => Effect.die("unused"),
125 resetTokenCache: Effect.void,
126 } satisfies ManagedRelay.ManagedRelayClient["Service"]);
127 const connectivity = Connectivity.Connectivity.of({
128 status: SubscriptionRef.get(networkStatus),
129 changes: SubscriptionRef.changes(networkStatus),
130 });
131 const layer = RelayEnvironmentDiscovery.layer.pipe(
132 Layer.provide(
133 Layer.mergeAll(
134 Layer.succeed(ManagedRelay.ManagedRelayClient, client),
135 Layer.succeed(
136 ClientCapabilities.CloudSession,
137 ClientCapabilities.CloudSession.of({
138 identity: Ref.get(clerkToken).pipe(
139 Effect.map((token) =>
140 token === null ? Option.none() : Option.some(sessionIdentity),
141 ),
142 ),
143 clerkToken: Ref.get(clerkToken).pipe(
144 Effect.filterOrFail(
145 (token) => token !== null,
146 () =>
147 new ConnectionBlockedError({
148 reason: "authentication",
149 detail: "Signed out.",
150 }),
151 ),
152 ),
153 }),
154 ),
155 Layer.succeed(Connectivity.Connectivity, connectivity),
156 Layer.succeed(
157 ConnectionWakeups.ConnectionWakeups,
158 ConnectionWakeups.ConnectionWakeups.of({
159 changes: SubscriptionRef.changes(wakeups).pipe(
160 Stream.drop(1),
161 Stream.map((event) => event.reason),
162 ),
163 }),
164 ),
165 ),
166 ),
167 );
168
169 return {
170 layer,
171 listCalls,
172 listedEnvironments,
173 listFailure,
174 clerkToken,
175 networkStatus,
176 secondListCall,
177 statusRequests,
178 wake: (reason: "application-active" | "credentials-changed") =>
179 SubscriptionRef.update(wakeups, (event) => ({
180 sequence: event.sequence + 1,
181 reason,
182 })),
183 };
184});
185
186describe("RelayEnvironmentDiscovery", () => {
187 it.effect("fetches a fresh list after removal while an older refresh is checking statuses", () =>
188 Effect.gen(function* () {
189 const harness = yield* makeHarness();
190 yield* Effect.gen(function* () {
191 const discovery = yield* RelayEnvironmentDiscovery.RelayEnvironmentDiscovery;
192 const atoms = createRelayEnvironmentDiscoveryAtoms(
193 Atom.runtime(
194 Layer.succeed(RelayEnvironmentDiscovery.RelayEnvironmentDiscovery, discovery),
195 ),
196 );
197 const registry = AtomRegistry.make();
198 yield* Effect.addFinalizer(() => Effect.sync(() => registry.dispose()));
199
200 const refresh = atoms.refresh.run(registry, undefined);
201 yield* SubscriptionRef.changes(discovery.state).pipe(
202 Stream.filter((state) => state.environments.size === environments.length),
203 Stream.runHead,
204 );
205 const coalescedRefresh = atoms.refresh.run(registry, undefined);
206 yield* Ref.set(harness.listedEnvironments, [environments[1]!]);
207 const mutationRefresh = atoms.refreshAfterMutation.run(registry, undefined);
208
209 const requests = yield* Ref.get(harness.statusRequests);
210 for (const environment of environments) {
211 yield* Deferred.succeed(
212 requests.get(environment.environmentId)!,
213 status(environment, "online"),
214 );
215 }
216 const results = yield* Effect.promise(() =>
217 Promise.all([refresh, coalescedRefresh, mutationRefresh]),
218 );
219 for (const result of results) {
220 expect(result._tag).toBe("Success");
221 }
222 expect(yield* Ref.get(harness.listCalls)).toBe(2);
223 const complete = yield* SubscriptionRef.get(discovery.state);
224 expect([...complete.environments.keys()]).toEqual([environments[1]!.environmentId]);
225 expect(complete.refreshing).toBe(false);
226 }).pipe(Effect.provide(harness.layer));
227 }),
228 );
229
230 it.effect("publishes each environment status as soon as that lookup completes", () =>
231 Effect.gen(function* () {
232 const harness = yield* makeHarness();
233 yield* Effect.gen(function* () {
234 const discovery = yield* RelayEnvironmentDiscovery.RelayEnvironmentDiscovery;
235 const refreshFiber = yield* Effect.forkChild(discovery.refresh);
236
237 const checking = yield* SubscriptionRef.changes(discovery.state).pipe(
238 Stream.filter((state) => state.environments.size === 2),
239 Stream.runHead,
240 Effect.map(Option.getOrThrow),
241 );
242 expect(
243 [...checking.environments.values()].every((entry) => entry.availability === "checking"),
244 ).toBe(true);
245
246 const requests = yield* Ref.get(harness.statusRequests);
247 yield* Deferred.succeed(
248 requests.get(environments[1]!.environmentId)!,
249 status(environments[1]!, "online"),
250 );
251
252 const partiallyResolved = yield* SubscriptionRef.changes(discovery.state).pipe(
253 Stream.filter(
254 (state) =>
255 state.environments.get(environments[1]!.environmentId)?.availability === "online",
256 ),
257 Stream.runHead,
258 Effect.map(Option.getOrThrow),
259 );
260 expect(
261 partiallyResolved.environments.get(environments[0]!.environmentId)?.availability,
262 ).toBe("checking");
263
264 yield* Deferred.succeed(
265 requests.get(environments[0]!.environmentId)!,
266 status(environments[0]!, "offline"),
267 );
268 yield* Fiber.join(refreshFiber);
269
270 const complete = yield* SubscriptionRef.get(discovery.state);
271 expect(complete.environments.get(environments[0]!.environmentId)?.availability).toBe(
272 "offline",
273 );
274 expect(complete.refreshing).toBe(false);
275 }).pipe(Effect.provide(harness.layer));
276 }),
277 );
278
279 it.effect("reports an environment as unregistered only once the relay list has loaded", () =>
280 Effect.gen(function* () {
281 const harness = yield* makeHarness();
282 yield* Effect.gen(function* () {
283 const discovery = yield* RelayEnvironmentDiscovery.RelayEnvironmentDiscovery;
284 const listed = environments[0]!.environmentId;
285 const deregistered = "deregistered-environment";
286 const before = yield* SubscriptionRef.get(discovery.state);
287 expect(RelayEnvironmentDiscovery.relayEnvironmentRegistration(before, deregistered)).toBe(
288 "unknown",
289 );
290
291 const requests = yield* Ref.get(harness.statusRequests);
292 for (const environment of environments) {
293 yield* Deferred.succeed(
294 requests.get(environment.environmentId)!,
295 status(environment, "online"),
296 );
297 }
298 yield* discovery.refresh;
299
300 const after = yield* SubscriptionRef.get(discovery.state);
301 expect(RelayEnvironmentDiscovery.relayEnvironmentRegistration(after, listed)).toBe(
302 "registered",
303 );
304 expect(RelayEnvironmentDiscovery.relayEnvironmentRegistration(after, deregistered)).toBe(
305 "unregistered",
306 );
307 }).pipe(Effect.provide(harness.layer));
308 }),
309 );
310
311 it.effect(
312 "preserves discovered rows while offline and refreshes after connectivity returns",
313 () =>
314 Effect.gen(function* () {
315 const harness = yield* makeHarness();
316 yield* Effect.gen(function* () {
317 const discovery = yield* RelayEnvironmentDiscovery.RelayEnvironmentDiscovery;
318 const requests = yield* Ref.get(harness.statusRequests);
319 for (const environment of environments) {
320 yield* Deferred.succeed(
321 requests.get(environment.environmentId)!,
322 status(environment, "online"),
323 );
324 }
325 yield* discovery.refresh;
326
327 const offlineFiber = yield* SubscriptionRef.changes(discovery.state).pipe(
328 Stream.filter((state) => state.offline),
329 Stream.runHead,
330 Effect.forkChild,
331 );
332 yield* SubscriptionRef.set(harness.networkStatus, "offline");
333 yield* Fiber.join(offlineFiber);
334 expect((yield* SubscriptionRef.get(discovery.state)).environments.size).toBe(2);
335
336 yield* SubscriptionRef.set(harness.networkStatus, "online");
337 yield* Deferred.await(harness.secondListCall);
338 expect(yield* Ref.get(harness.listCalls)).toBe(2);
339 }).pipe(Effect.provide(harness.layer));
340 }),
341 );
342
343 it.effect("publishes listing failures without rejecting the refresh command", () =>
344 Effect.gen(function* () {
345 const networkStatus = yield* SubscriptionRef.make<NetworkStatus>("online");
346 const client = ManagedRelay.ManagedRelayClient.of({
347 relayUrl: "https://relay.example.test",
348 listEnvironments: () =>
349 Effect.fail(
350 new ManagedRelay.ManagedRelayRequestTimeoutError({
351 activity: "Relay environment listing",
352 timeoutMs: ManagedRelay.MANAGED_RELAY_REQUEST_TIMEOUT_MS,
353 traceId: null,
354 }),
355 ),
356 getEnvironmentStatus: () => Effect.die("unused"),
357 listDevices: () => Effect.die("unused"),
358 createEnvironmentLinkChallenge: () => Effect.die("unused"),
359 linkEnvironment: () => Effect.die("unused"),
360 unlinkEnvironment: () => Effect.die("unused"),
361 getT3ChatLink: () => Effect.die("unused"),
362 createT3ChatLinkRequest: () => Effect.die("unused"),
363 unlinkT3Chat: () => Effect.die("unused"),
364 connectEnvironment: () => Effect.die("unused"),
365 registerDevice: () => Effect.die("unused"),
366 unregisterDevice: () => Effect.die("unused"),
367 registerLiveActivity: () => Effect.die("unused"),
368 getAgentActivitySnapshot: () => Effect.die("unused"),
369 resetTokenCache: Effect.void,
370 } satisfies ManagedRelay.ManagedRelayClient["Service"]);
371 const layer = RelayEnvironmentDiscovery.layer.pipe(
372 Layer.provide(
373 Layer.mergeAll(
374 Layer.succeed(ManagedRelay.ManagedRelayClient, client),
375 Layer.succeed(ClientCapabilities.CloudSession, {
376 identity: Effect.succeedSome({ accountId: "account-1" }),
377 clerkToken: Effect.succeed("clerk-token"),
378 }),
379 Layer.succeed(Connectivity.Connectivity, {
380 status: SubscriptionRef.get(networkStatus),
381 changes: SubscriptionRef.changes(networkStatus),
382 }),
383 Layer.succeed(
384 ConnectionWakeups.ConnectionWakeups,
385 ConnectionWakeups.ConnectionWakeups.of({ changes: Stream.never }),
386 ),
387 ),
388 ),
389 );
390
391 yield* Effect.gen(function* () {
392 const discovery = yield* RelayEnvironmentDiscovery.RelayEnvironmentDiscovery;
393 yield* discovery.refresh;
394
395 const state = yield* SubscriptionRef.get(discovery.state);
396 expect(state.refreshing).toBe(false);
397 expect(Option.getOrThrow(state.error)).toMatchObject({
398 _tag: "ConnectionTransientError",
399 reason: "timeout",
400 message: `Relay environment listing timed out. ${NETWORK_BLOCKING_HINT}`,
401 });
402 }).pipe(Effect.provide(layer));
403 }),
404 );
405
406 it.effect("clears previously discovered rows when a refresh fails", () =>
407 Effect.gen(function* () {
408 const harness = yield* makeHarness();
409 yield* Effect.gen(function* () {
410 const discovery = yield* RelayEnvironmentDiscovery.RelayEnvironmentDiscovery;
411 const requests = yield* Ref.get(harness.statusRequests);
412 for (const environment of environments) {
413 yield* Deferred.succeed(
414 requests.get(environment.environmentId)!,
415 status(environment, "online"),
416 );
417 }
418 yield* discovery.refresh;
419 expect((yield* SubscriptionRef.get(discovery.state)).environments.size).toBe(2);
420
421 yield* Ref.set(
422 harness.listFailure,
423 new ManagedRelay.ManagedRelayRequestFailedError({
424 action: "list relay-managed environments",
425 cause: new Error("Relay request failed."),
426 }),
427 );
428 yield* discovery.refresh;
429
430 const failed = yield* SubscriptionRef.get(discovery.state);
431 expect(failed.environments.size).toBe(0);
432 expect(Option.isSome(failed.error)).toBe(true);
433 expect(
434 RelayEnvironmentDiscovery.relayEnvironmentRegistration(
435 failed,
436 environments[0]!.environmentId,
437 ),
438 ).toBe("unknown");
439 }).pipe(Effect.provide(harness.layer));
440 }),
441 );
442
443 it.effect("refreshes proactively when credentials change before any manual refresh", () =>
444 Effect.gen(function* () {
445 const harness = yield* makeHarness();
446 yield* Effect.gen(function* () {
447 const discovery = yield* RelayEnvironmentDiscovery.RelayEnvironmentDiscovery;
448 const requests = yield* Ref.get(harness.statusRequests);
449 for (const environment of environments) {
450 yield* Deferred.succeed(
451 requests.get(environment.environmentId)!,
452 status(environment, "online"),
453 );
454 }
455
456 // Let the scoped wakeup subscription start before emitting, mirroring
457 // a real sign-in which always happens long after service start.
458 yield* Effect.yieldNow;
459
460 yield* harness.wake("credentials-changed");
461 const populated = yield* SubscriptionRef.changes(discovery.state).pipe(
462 Stream.filter(
463 (state) => state.environments.size === environments.length && !state.refreshing,
464 ),
465 Stream.runHead,
466 Effect.map(Option.getOrThrow),
467 );
468 expect(Option.isNone(populated.error)).toBe(true);
469 expect(yield* Ref.get(harness.listCalls)).toBe(1);
470 }).pipe(Effect.provide(harness.layer), Effect.scoped);
471 }),
472 );
473
474 it.effect("settles to a clean empty state when refreshed while signed out", () =>
475 Effect.gen(function* () {
476 const harness = yield* makeHarness();
477 yield* Effect.gen(function* () {
478 const discovery = yield* RelayEnvironmentDiscovery.RelayEnvironmentDiscovery;
479 yield* Ref.set(harness.clerkToken, null);
480 yield* discovery.refresh;
481
482 const state = yield* SubscriptionRef.get(discovery.state);
483 expect(state.environments.size).toBe(0);
484 expect(state.refreshing).toBe(false);
485 expect(Option.isNone(state.error)).toBe(true);
486 expect(
487 RelayEnvironmentDiscovery.relayEnvironmentRegistration(
488 state,
489 environments[0]!.environmentId,
490 ),
491 ).toBe("unknown");
492 }).pipe(Effect.provide(harness.layer));
493 }),
494 );
495
496 it.effect("does not republish stale rows after sign-out invalidates an in-flight refresh", () =>
497 Effect.gen(function* () {
498 const harness = yield* makeHarness();
499 yield* Effect.gen(function* () {
500 const discovery = yield* RelayEnvironmentDiscovery.RelayEnvironmentDiscovery;
501 const refreshFiber = yield* Effect.forkChild(discovery.refresh);
502 yield* SubscriptionRef.changes(discovery.state).pipe(
503 Stream.filter((state) => state.environments.size === environments.length),
504 Stream.runHead,
505 );
506
507 yield* Ref.set(harness.clerkToken, null);
508 yield* harness.wake("credentials-changed");
509 yield* SubscriptionRef.changes(discovery.state).pipe(
510 Stream.filter((state) => state.environments.size === 0),
511 Stream.runHead,
512 );
513
514 const requests = yield* Ref.get(harness.statusRequests);
515 for (const environment of environments) {
516 yield* Deferred.succeed(
517 requests.get(environment.environmentId)!,
518 status(environment, "online"),
519 );
520 }
521 yield* Fiber.join(refreshFiber);
522 yield* Effect.yieldNow;
523
524 const signedOut = yield* SubscriptionRef.get(discovery.state);
525 expect(signedOut.environments.size).toBe(0);
526 expect(
527 RelayEnvironmentDiscovery.relayEnvironmentRegistration(
528 signedOut,
529 environments[0]!.environmentId,
530 ),
531 ).toBe("unknown");
532 }).pipe(Effect.provide(harness.layer), Effect.scoped);
533 }),
534 );
535});