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.scoped(
189 Effect.gen(function* () {
190 const harness = yield* makeHarness();
191 yield* Effect.gen(function* () {
192 const discovery = yield* RelayEnvironmentDiscovery.RelayEnvironmentDiscovery;
193 const atoms = createRelayEnvironmentDiscoveryAtoms(
194 Atom.runtime(
195 Layer.succeed(RelayEnvironmentDiscovery.RelayEnvironmentDiscovery, discovery),
196 ),
197 );
198 const registry = AtomRegistry.make();
199 yield* Effect.addFinalizer(() => Effect.sync(() => registry.dispose()));
200
201 const refresh = atoms.refresh.run(registry, undefined);
202 yield* SubscriptionRef.changes(discovery.state).pipe(
203 Stream.filter((state) => state.environments.size === environments.length),
204 Stream.runHead,
205 );
206 const coalescedRefresh = atoms.refresh.run(registry, undefined);
207 yield* Ref.set(harness.listedEnvironments, [environments[1]!]);
208 const mutationRefresh = atoms.refreshAfterMutation.run(registry, undefined);
209
210 const requests = yield* Ref.get(harness.statusRequests);
211 for (const environment of environments) {
212 yield* Deferred.succeed(
213 requests.get(environment.environmentId)!,
214 status(environment, "online"),
215 );
216 }
217 const results = yield* Effect.promise(() =>
218 Promise.all([refresh, coalescedRefresh, mutationRefresh]),
219 );
220 for (const result of results) {
221 expect(result._tag).toBe("Success");
222 }
223 expect(yield* Ref.get(harness.listCalls)).toBe(2);
224 const complete = yield* SubscriptionRef.get(discovery.state);
225 expect([...complete.environments.keys()]).toEqual([environments[1]!.environmentId]);
226 expect(complete.refreshing).toBe(false);
227 }).pipe(Effect.provide(harness.layer));
228 }),
229 ),
230 );
231
232 it.effect("publishes each environment status as soon as that lookup completes", () =>
233 Effect.gen(function* () {
234 const harness = yield* makeHarness();
235 yield* Effect.gen(function* () {
236 const discovery = yield* RelayEnvironmentDiscovery.RelayEnvironmentDiscovery;
237 const refreshFiber = yield* Effect.forkChild(discovery.refresh);
238
239 const checking = yield* SubscriptionRef.changes(discovery.state).pipe(
240 Stream.filter((state) => state.environments.size === 2),
241 Stream.runHead,
242 Effect.map(Option.getOrThrow),
243 );
244 expect(
245 [...checking.environments.values()].every((entry) => entry.availability === "checking"),
246 ).toBe(true);
247
248 const requests = yield* Ref.get(harness.statusRequests);
249 yield* Deferred.succeed(
250 requests.get(environments[1]!.environmentId)!,
251 status(environments[1]!, "online"),
252 );
253
254 const partiallyResolved = yield* SubscriptionRef.changes(discovery.state).pipe(
255 Stream.filter(
256 (state) =>
257 state.environments.get(environments[1]!.environmentId)?.availability === "online",
258 ),
259 Stream.runHead,
260 Effect.map(Option.getOrThrow),
261 );
262 expect(
263 partiallyResolved.environments.get(environments[0]!.environmentId)?.availability,
264 ).toBe("checking");
265
266 yield* Deferred.succeed(
267 requests.get(environments[0]!.environmentId)!,
268 status(environments[0]!, "offline"),
269 );
270 yield* Fiber.join(refreshFiber);
271
272 const complete = yield* SubscriptionRef.get(discovery.state);
273 expect(complete.environments.get(environments[0]!.environmentId)?.availability).toBe(
274 "offline",
275 );
276 expect(complete.refreshing).toBe(false);
277 }).pipe(Effect.provide(harness.layer));
278 }),
279 );
280
281 it.effect("reports an environment as unregistered only once the relay list has loaded", () =>
282 Effect.gen(function* () {
283 const harness = yield* makeHarness();
284 yield* Effect.gen(function* () {
285 const discovery = yield* RelayEnvironmentDiscovery.RelayEnvironmentDiscovery;
286 const listed = environments[0]!.environmentId;
287 const deregistered = "deregistered-environment";
288 const before = yield* SubscriptionRef.get(discovery.state);
289 expect(RelayEnvironmentDiscovery.relayEnvironmentRegistration(before, deregistered)).toBe(
290 "unknown",
291 );
292
293 const requests = yield* Ref.get(harness.statusRequests);
294 for (const environment of environments) {
295 yield* Deferred.succeed(
296 requests.get(environment.environmentId)!,
297 status(environment, "online"),
298 );
299 }
300 yield* discovery.refresh;
301
302 const after = yield* SubscriptionRef.get(discovery.state);
303 expect(RelayEnvironmentDiscovery.relayEnvironmentRegistration(after, listed)).toBe(
304 "registered",
305 );
306 expect(RelayEnvironmentDiscovery.relayEnvironmentRegistration(after, deregistered)).toBe(
307 "unregistered",
308 );
309 }).pipe(Effect.provide(harness.layer));
310 }),
311 );
312
313 it.effect(
314 "preserves discovered rows while offline and refreshes after connectivity returns",
315 () =>
316 Effect.gen(function* () {
317 const harness = yield* makeHarness();
318 yield* Effect.gen(function* () {
319 const discovery = yield* RelayEnvironmentDiscovery.RelayEnvironmentDiscovery;
320 const requests = yield* Ref.get(harness.statusRequests);
321 for (const environment of environments) {
322 yield* Deferred.succeed(
323 requests.get(environment.environmentId)!,
324 status(environment, "online"),
325 );
326 }
327 yield* discovery.refresh;
328
329 const offlineFiber = yield* SubscriptionRef.changes(discovery.state).pipe(
330 Stream.filter((state) => state.offline),
331 Stream.runHead,
332 Effect.forkChild,
333 );
334 yield* SubscriptionRef.set(harness.networkStatus, "offline");
335 yield* Fiber.join(offlineFiber);
336 expect((yield* SubscriptionRef.get(discovery.state)).environments.size).toBe(2);
337
338 yield* SubscriptionRef.set(harness.networkStatus, "online");
339 yield* Deferred.await(harness.secondListCall);
340 expect(yield* Ref.get(harness.listCalls)).toBe(2);
341 }).pipe(Effect.provide(harness.layer));
342 }),
343 );
344
345 it.effect("publishes listing failures without rejecting the refresh command", () =>
346 Effect.gen(function* () {
347 const networkStatus = yield* SubscriptionRef.make<NetworkStatus>("online");
348 const client = ManagedRelay.ManagedRelayClient.of({
349 relayUrl: "https://relay.example.test",
350 listEnvironments: () =>
351 Effect.fail(
352 new ManagedRelay.ManagedRelayRequestTimeoutError({
353 activity: "Relay environment listing",
354 timeoutMs: ManagedRelay.MANAGED_RELAY_REQUEST_TIMEOUT_MS,
355 traceId: null,
356 }),
357 ),
358 getEnvironmentStatus: () => Effect.die("unused"),
359 listDevices: () => Effect.die("unused"),
360 createEnvironmentLinkChallenge: () => Effect.die("unused"),
361 linkEnvironment: () => Effect.die("unused"),
362 unlinkEnvironment: () => Effect.die("unused"),
363 getT3ChatLink: () => Effect.die("unused"),
364 createT3ChatLinkRequest: () => Effect.die("unused"),
365 unlinkT3Chat: () => Effect.die("unused"),
366 connectEnvironment: () => Effect.die("unused"),
367 registerDevice: () => Effect.die("unused"),
368 unregisterDevice: () => Effect.die("unused"),
369 registerLiveActivity: () => Effect.die("unused"),
370 getAgentActivitySnapshot: () => Effect.die("unused"),
371 resetTokenCache: Effect.void,
372 } satisfies ManagedRelay.ManagedRelayClient["Service"]);
373 const layer = RelayEnvironmentDiscovery.layer.pipe(
374 Layer.provide(
375 Layer.mergeAll(
376 Layer.succeed(ManagedRelay.ManagedRelayClient, client),
377 Layer.succeed(ClientCapabilities.CloudSession, {
378 identity: Effect.succeedSome({ accountId: "account-1" }),
379 clerkToken: Effect.succeed("clerk-token"),
380 }),
381 Layer.succeed(Connectivity.Connectivity, {
382 status: SubscriptionRef.get(networkStatus),
383 changes: SubscriptionRef.changes(networkStatus),
384 }),
385 Layer.succeed(
386 ConnectionWakeups.ConnectionWakeups,
387 ConnectionWakeups.ConnectionWakeups.of({ changes: Stream.never }),
388 ),
389 ),
390 ),
391 );
392
393 yield* Effect.gen(function* () {
394 const discovery = yield* RelayEnvironmentDiscovery.RelayEnvironmentDiscovery;
395 yield* discovery.refresh;
396
397 const state = yield* SubscriptionRef.get(discovery.state);
398 expect(state.refreshing).toBe(false);
399 expect(Option.getOrThrow(state.error)).toMatchObject({
400 _tag: "ConnectionTransientError",
401 reason: "timeout",
402 message: `Relay environment listing timed out. ${NETWORK_BLOCKING_HINT}`,
403 });
404 }).pipe(Effect.provide(layer));
405 }),
406 );
407
408 it.effect("clears previously discovered rows when a refresh fails", () =>
409 Effect.gen(function* () {
410 const harness = yield* makeHarness();
411 yield* Effect.gen(function* () {
412 const discovery = yield* RelayEnvironmentDiscovery.RelayEnvironmentDiscovery;
413 const requests = yield* Ref.get(harness.statusRequests);
414 for (const environment of environments) {
415 yield* Deferred.succeed(
416 requests.get(environment.environmentId)!,
417 status(environment, "online"),
418 );
419 }
420 yield* discovery.refresh;
421 expect((yield* SubscriptionRef.get(discovery.state)).environments.size).toBe(2);
422
423 yield* Ref.set(
424 harness.listFailure,
425 new ManagedRelay.ManagedRelayRequestFailedError({
426 action: "list relay-managed environments",
427 cause: new Error("Relay request failed."),
428 }),
429 );
430 yield* discovery.refresh;
431
432 const failed = yield* SubscriptionRef.get(discovery.state);
433 expect(failed.environments.size).toBe(0);
434 expect(Option.isSome(failed.error)).toBe(true);
435 expect(
436 RelayEnvironmentDiscovery.relayEnvironmentRegistration(
437 failed,
438 environments[0]!.environmentId,
439 ),
440 ).toBe("unknown");
441 }).pipe(Effect.provide(harness.layer));
442 }),
443 );
444
445 it.effect("refreshes proactively when credentials change before any manual refresh", () =>
446 Effect.gen(function* () {
447 const harness = yield* makeHarness();
448 yield* Effect.gen(function* () {
449 const discovery = yield* RelayEnvironmentDiscovery.RelayEnvironmentDiscovery;
450 const requests = yield* Ref.get(harness.statusRequests);
451 for (const environment of environments) {
452 yield* Deferred.succeed(
453 requests.get(environment.environmentId)!,
454 status(environment, "online"),
455 );
456 }
457
458 // Let the scoped wakeup subscription start before emitting, mirroring
459 // a real sign-in which always happens long after service start.
460 yield* Effect.yieldNow;
461
462 yield* harness.wake("credentials-changed");
463 const populated = yield* SubscriptionRef.changes(discovery.state).pipe(
464 Stream.filter(
465 (state) => state.environments.size === environments.length && !state.refreshing,
466 ),
467 Stream.runHead,
468 Effect.map(Option.getOrThrow),
469 );
470 expect(Option.isNone(populated.error)).toBe(true);
471 expect(yield* Ref.get(harness.listCalls)).toBe(1);
472 }).pipe(Effect.provide(harness.layer), Effect.scoped);
473 }),
474 );
475
476 it.effect("settles to a clean empty state when refreshed while signed out", () =>
477 Effect.gen(function* () {
478 const harness = yield* makeHarness();
479 yield* Effect.gen(function* () {
480 const discovery = yield* RelayEnvironmentDiscovery.RelayEnvironmentDiscovery;
481 yield* Ref.set(harness.clerkToken, null);
482 yield* discovery.refresh;
483
484 const state = yield* SubscriptionRef.get(discovery.state);
485 expect(state.environments.size).toBe(0);
486 expect(state.refreshing).toBe(false);
487 expect(Option.isNone(state.error)).toBe(true);
488 expect(
489 RelayEnvironmentDiscovery.relayEnvironmentRegistration(
490 state,
491 environments[0]!.environmentId,
492 ),
493 ).toBe("unknown");
494 }).pipe(Effect.provide(harness.layer));
495 }),
496 );
497
498 it.effect("does not republish stale rows after sign-out invalidates an in-flight refresh", () =>
499 Effect.gen(function* () {
500 const harness = yield* makeHarness();
501 yield* Effect.gen(function* () {
502 const discovery = yield* RelayEnvironmentDiscovery.RelayEnvironmentDiscovery;
503 const refreshFiber = yield* Effect.forkChild(discovery.refresh);
504 yield* SubscriptionRef.changes(discovery.state).pipe(
505 Stream.filter((state) => state.environments.size === environments.length),
506 Stream.runHead,
507 );
508
509 yield* Ref.set(harness.clerkToken, null);
510 yield* harness.wake("credentials-changed");
511 yield* SubscriptionRef.changes(discovery.state).pipe(
512 Stream.filter((state) => state.environments.size === 0),
513 Stream.runHead,
514 );
515
516 const requests = yield* Ref.get(harness.statusRequests);
517 for (const environment of environments) {
518 yield* Deferred.succeed(
519 requests.get(environment.environmentId)!,
520 status(environment, "online"),
521 );
522 }
523 yield* Fiber.join(refreshFiber);
524 yield* Effect.yieldNow;
525
526 const signedOut = yield* SubscriptionRef.get(discovery.state);
527 expect(signedOut.environments.size).toBe(0);
528 expect(
529 RelayEnvironmentDiscovery.relayEnvironmentRegistration(
530 signedOut,
531 environments[0]!.environmentId,
532 ),
533 ).toBe("unknown");
534 }).pipe(Effect.provide(harness.layer), Effect.scoped);
535 }),
536 );
537});