1import { EnvironmentId } from "@t3tools/contracts";
2import {
3 RelayAuthInvalidError,
4 type RelayClientDeviceRecord,
5 type RelayClientEnvironmentRecord,
6 type RelayEnvironmentStatusResponse,
7} from "@t3tools/contracts/relay";
8import { describe, expect, it } from "@effect/vitest";
9import * as Effect from "effect/Effect";
10import * as Fiber from "effect/Fiber";
11import * as Layer from "effect/Layer";
12import * as Stream from "effect/Stream";
13import { Atom, AtomRegistry } from "effect/reactivity";
14import { afterEach, vi } from "vite-plus/test";
15
16import { DPOP_UNKNOWN_HINT } from "./errorPresentation.ts";
17import * as ManagedRelay from "./managedRelay.ts";
18import {
19 createManagedRelayQueryManager,
20 createManagedRelaySession,
21 deregisterManagedRelayEnvironment,
22 managedRelayAccountChanges,
23 type ManagedRelayQueryEvent,
24 ManagedRelaySessionError,
25 managedRelaySessionAtom,
26 readManagedRelaySnapshotState,
27 setManagedRelaySession,
28} from "./managedRelayState.ts";
29
30let registry = AtomRegistry.make();
31
32const environment = {
33 environmentId: EnvironmentId.make("environment-1"),
34 label: "Main environment",
35 endpoint: {
36 httpBaseUrl: "https://environment.example.test",
37 wsBaseUrl: "wss://environment.example.test",
38 providerKind: "cloudflare_tunnel",
39 },
40 linkedAt: "2026-06-01T00:00:00.000Z",
41} satisfies RelayClientEnvironmentRecord;
42
43const device = {
44 deviceId: "device-1",
45 label: "Julius iPhone",
46 platform: "ios",
47 iosMajorVersion: 18,
48 appVersion: null,
49 notifications: {
50 enabled: true,
51 notifyOnApproval: true,
52 notifyOnInput: true,
53 notifyOnCompletion: true,
54 notifyOnFailure: true,
55 },
56 liveActivities: {
57 enabled: true,
58 },
59 updatedAt: "2026-06-01T00:00:00.000Z",
60} satisfies RelayClientDeviceRecord;
61
62function resetRegistry() {
63 registry.dispose();
64 registry = AtomRegistry.make();
65}
66
67function createClient(overrides?: Partial<ManagedRelay.ManagedRelayClient["Service"]>) {
68 return ManagedRelay.ManagedRelayClient.of({
69 relayUrl: "https://relay.example.test",
70 listEnvironments: () => Effect.succeed([environment]),
71 listDevices: () => Effect.succeed([device]),
72 createEnvironmentLinkChallenge: () => Effect.die("unused"),
73 linkEnvironment: () => Effect.die("unused"),
74 unlinkEnvironment: () => Effect.die("unused"),
75 getEnvironmentStatus: () =>
76 Effect.succeed({
77 environmentId: environment.environmentId,
78 endpoint: environment.endpoint,
79 status: "online",
80 checkedAt: "2026-06-01T00:00:00.000Z",
81 }),
82 connectEnvironment: () => Effect.die("unused"),
83 registerDevice: () => Effect.die("unused"),
84 unregisterDevice: () => Effect.die("unused"),
85 registerLiveActivity: () => Effect.die("unused"),
86 getAgentActivitySnapshot: () => Effect.die("unused"),
87 resetTokenCache: Effect.void,
88 ...overrides,
89 });
90}
91
92function createManager(
93 overrides?: Partial<ManagedRelay.ManagedRelayClient["Service"]>,
94 onQueryEvent?: (event: ManagedRelayQueryEvent) => void,
95) {
96 const client = createClient(overrides);
97 const runtime = Atom.runtime(Layer.succeed(ManagedRelay.ManagedRelayClient, client));
98 return createManagedRelayQueryManager(runtime, {
99 staleTimeMs: 60_000,
100 ...(onQueryEvent ? { onQueryEvent } : {}),
101 });
102}
103
104function setSession() {
105 setManagedRelaySession(registry, {
106 accountId: "account-1",
107 readClerkToken: () => Promise.resolve("clerk-token"),
108 });
109}
110
111function clerkToken(expiresAtSeconds: number): string {
112 const encode = (value: unknown) =>
113 btoa(JSON.stringify(value)).replaceAll("+", "-").replaceAll("/", "_").replace(/=+$/u, "");
114 return `${encode({ alg: "none" })}.${encode({ exp: expiresAtSeconds })}.signature`;
115}
116
117describe("createManagedRelayQueryManager", () => {
118 afterEach(resetRegistry);
119
120 it.effect("deregisters an environment through the current Clerk session", () =>
121 Effect.gen(function* () {
122 const unlinkEnvironment = vi.fn(() => Effect.succeed({ ok: true }));
123 setSession();
124
125 yield* deregisterManagedRelayEnvironment(registry, {
126 accountId: "account-1",
127 environmentId: environment.environmentId,
128 }).pipe(
129 Effect.provideService(ManagedRelay.ManagedRelayClient, createClient({ unlinkEnvironment })),
130 );
131
132 expect(unlinkEnvironment).toHaveBeenCalledWith({
133 clerkToken: "clerk-token",
134 environmentId: environment.environmentId,
135 });
136 }),
137 );
138
139 it.effect("rejects deregistration after the account changes", () =>
140 Effect.gen(function* () {
141 const unlinkEnvironment = vi.fn(() => Effect.succeed({ ok: true }));
142 setSession();
143
144 const error = yield* deregisterManagedRelayEnvironment(registry, {
145 accountId: "previous-account",
146 environmentId: environment.environmentId,
147 }).pipe(
148 Effect.provideService(ManagedRelay.ManagedRelayClient, createClient({ unlinkEnvironment })),
149 Effect.flip,
150 );
151
152 expect(error).toBeInstanceOf(ManagedRelaySessionError);
153 expect(unlinkEnvironment).not.toHaveBeenCalled();
154 }),
155 );
156
157 it.effect("deduplicates concurrent Clerk token reads and reuses the token until JWT expiry", () =>
158 Effect.gen(function* () {
159 const token = clerkToken(4_102_444_800);
160 let resolveToken!: (value: string) => void;
161 const readClerkToken = vi.fn(
162 () =>
163 new Promise<string>((resolve) => {
164 resolveToken = resolve;
165 }),
166 );
167 const session = createManagedRelaySession({
168 accountId: "account-1",
169 readClerkToken,
170 });
171
172 const readsFiber = yield* Effect.all([session.readClerkToken(), session.readClerkToken()], {
173 concurrency: "unbounded",
174 }).pipe(Effect.forkChild);
175 yield* Effect.yieldNow;
176 expect(readClerkToken).toHaveBeenCalledTimes(1);
177
178 resolveToken(token);
179 expect(yield* Fiber.join(readsFiber)).toEqual([token, token]);
180 expect(yield* session.readClerkToken()).toBe(token);
181 expect(readClerkToken).toHaveBeenCalledTimes(1);
182 }),
183 );
184
185 it.effect("updates the token provider without replacing a same-account session", () =>
186 Effect.gen(function* () {
187 const firstRead = vi.fn(() => Promise.resolve<string | null>(null));
188 setManagedRelaySession(registry, {
189 accountId: "account-1",
190 readClerkToken: firstRead,
191 });
192 const firstSession = registry.get(managedRelaySessionAtom);
193 expect(firstSession).not.toBeNull();
194 expect(yield* firstSession!.readClerkToken()).toBeNull();
195
196 const secondRead = vi.fn(() => Promise.resolve<string | null>("refreshed-token"));
197 setManagedRelaySession(registry, {
198 accountId: "account-1",
199 readClerkToken: secondRead,
200 });
201
202 expect(registry.get(managedRelaySessionAtom)).toBe(firstSession);
203 expect(yield* firstSession!.readClerkToken()).toBe("refreshed-token");
204 expect(firstRead).toHaveBeenCalledTimes(1);
205 expect(secondRead).toHaveBeenCalledTimes(1);
206 }),
207 );
208
209 it.effect("does not pin a refreshed session to an older pending token read", () =>
210 Effect.gen(function* () {
211 let resolveFirst!: (token: string) => void;
212 setManagedRelaySession(registry, {
213 accountId: "account-1",
214 readClerkToken: () =>
215 new Promise<string>((resolve) => {
216 resolveFirst = resolve;
217 }),
218 });
219 const session = registry.get(managedRelaySessionAtom);
220 const firstRead = yield* session!.readClerkToken().pipe(Effect.forkChild);
221 yield* Effect.yieldNow;
222
223 setManagedRelaySession(registry, {
224 accountId: "account-1",
225 readClerkToken: () => Promise.resolve("refreshed-token"),
226 });
227
228 expect(yield* session!.readClerkToken()).toBe("refreshed-token");
229 resolveFirst("older-token");
230 expect(yield* Fiber.join(firstRead)).toBe("older-token");
231 }),
232 );
233
234 it.effect("emits credential changes only when the managed relay account changes", () =>
235 Effect.gen(function* () {
236 setManagedRelaySession(registry, {
237 accountId: "account-1",
238 readClerkToken: () => Promise.resolve("first-token"),
239 });
240 const changes = yield* managedRelayAccountChanges(registry).pipe(
241 Stream.take(2),
242 Stream.runCollect,
243 Effect.forkChild,
244 );
245 yield* Effect.promise(() =>
246 vi.waitFor(() => {
247 expect(registry.getNodes().get(managedRelaySessionAtom)?.listeners.size).toBeGreaterThan(
248 0,
249 );
250 }),
251 );
252
253 setManagedRelaySession(registry, {
254 accountId: "account-1",
255 readClerkToken: () => Promise.resolve("refreshed-token"),
256 });
257 setManagedRelaySession(registry, {
258 accountId: "account-2",
259 readClerkToken: () => Promise.resolve("second-token"),
260 });
261 setManagedRelaySession(registry, null);
262
263 expect(Array.from(yield* Fiber.join(changes))).toEqual(["account-2", null]);
264 }),
265 );
266
267 it("shares one Clerk token read across concurrent relay list and status queries", async () => {
268 const secondEnvironment = {
269 ...environment,
270 environmentId: EnvironmentId.make("environment-2"),
271 label: "Second environment",
272 endpoint: {
273 ...environment.endpoint,
274 httpBaseUrl: "https://environment-2.example.test",
275 wsBaseUrl: "wss://environment-2.example.test",
276 },
277 } satisfies RelayClientEnvironmentRecord;
278 const token = clerkToken(4_102_444_800);
279 const readClerkToken = vi.fn(() => Promise.resolve(token));
280 const manager = createManager({
281 listEnvironments: () => Effect.succeed([environment, secondEnvironment]),
282 getEnvironmentStatus: ({ environmentId }) => {
283 const current =
284 environmentId === environment.environmentId ? environment : secondEnvironment;
285 return Effect.succeed({
286 environmentId: current.environmentId,
287 endpoint: current.endpoint,
288 status: "online" as const,
289 checkedAt: "2026-06-01T00:00:00.000Z",
290 });
291 },
292 });
293 setManagedRelaySession(registry, {
294 accountId: "account-1",
295 readClerkToken,
296 });
297
298 const environmentsAtom = manager.environmentsAtom("account-1");
299 const firstStatusAtom = manager.environmentStatusAtom({
300 accountId: "account-1",
301 environment,
302 });
303 const secondStatusAtom = manager.environmentStatusAtom({
304 accountId: "account-1",
305 environment: secondEnvironment,
306 });
307 registry.get(environmentsAtom);
308 registry.get(firstStatusAtom);
309 registry.get(secondStatusAtom);
310
311 await vi.waitFor(() => {
312 expect(readManagedRelaySnapshotState(registry.get(firstStatusAtom)).data?.status).toBe(
313 "online",
314 );
315 expect(readManagedRelaySnapshotState(registry.get(secondStatusAtom)).data?.status).toBe(
316 "online",
317 );
318 });
319 expect(readClerkToken).toHaveBeenCalledTimes(1);
320 });
321
322 it("keeps environment snapshots cached and refreshes them explicitly", async () => {
323 const listEnvironments = vi.fn(() => Effect.succeed([environment]));
324 const manager = createManager({ listEnvironments });
325 setSession();
326 const atom = manager.environmentsAtom("account-1");
327
328 registry.get(atom);
329 await vi.waitFor(() => expect(listEnvironments).toHaveBeenCalledTimes(1));
330
331 registry.get(manager.environmentsAtom("account-1"));
332 expect(listEnvironments).toHaveBeenCalledTimes(1);
333
334 manager.refreshEnvironments(registry, "account-1");
335 await vi.waitFor(() => expect(listEnvironments).toHaveBeenCalledTimes(2));
336 });
337
338 it("loads device snapshots through the current account session", async () => {
339 const listDevices = vi.fn(() => Effect.succeed([device]));
340 const manager = createManager({ listDevices });
341 setSession();
342 const atom = manager.devicesAtom("account-1");
343
344 registry.get(atom);
345 await vi.waitFor(() => {
346 expect(readManagedRelaySnapshotState(registry.get(atom)).data).toEqual([device]);
347 });
348 });
349
350 it("reports token and relay request phases for environment status queries", async () => {
351 const onQueryEvent = vi.fn();
352 const manager = createManager(undefined, onQueryEvent);
353 setSession();
354 const atom = manager.environmentStatusAtom({ accountId: "account-1", environment });
355
356 registry.get(atom);
357 await vi.waitFor(() => {
358 expect(readManagedRelaySnapshotState(registry.get(atom)).data?.status).toBe("online");
359 });
360
361 expect(onQueryEvent).toHaveBeenCalledWith({
362 operation: "environment-status",
363 stage: "clerk-token",
364 phase: "start",
365 accountId: "account-1",
366 environmentId: environment.environmentId,
367 });
368 expect(onQueryEvent).toHaveBeenCalledWith(
369 expect.objectContaining({
370 operation: "environment-status",
371 stage: "relay-request",
372 phase: "success",
373 accountId: "account-1",
374 environmentId: environment.environmentId,
375 }),
376 );
377 });
378
379 it("rejects status responses for a different environment", async () => {
380 const mismatchedStatus = {
381 environmentId: EnvironmentId.make("environment-2"),
382 endpoint: environment.endpoint,
383 status: "online",
384 checkedAt: "2026-06-01T00:00:00.000Z",
385 } satisfies RelayEnvironmentStatusResponse;
386 const manager = createManager({
387 getEnvironmentStatus: () => Effect.succeed(mismatchedStatus),
388 });
389 setSession();
390 const atom = manager.environmentStatusAtom({ accountId: "account-1", environment });
391
392 registry.get(atom);
393 await vi.waitFor(() => {
394 expect(readManagedRelaySnapshotState(registry.get(atom)).error).toBe(
395 "Relay returned status for a different environment.",
396 );
397 });
398 });
399
400 it("exposes relay trace IDs alongside snapshot errors", async () => {
401 const manager = createManager({
402 getEnvironmentStatus: () =>
403 Effect.fail(
404 new ManagedRelay.ManagedRelayRequestFailedError({
405 action: "get relay environment status",
406 cause: new Error("Relay request failed."),
407 traceId: "trace-status",
408 }),
409 ),
410 });
411 setSession();
412 const atom = manager.environmentStatusAtom({ accountId: "account-1", environment });
413
414 registry.get(atom);
415 await vi.waitFor(() => {
416 expect(readManagedRelaySnapshotState(registry.get(atom))).toMatchObject({
417 error: "Could not get relay environment status.",
418 errorTraceId: "trace-status",
419 });
420 });
421 });
422
423 it("presents clock skew as one possible cause for snapshot requests from older relays", async () => {
424 const manager = createManager({
425 getEnvironmentStatus: () =>
426 Effect.fail(
427 new ManagedRelay.ManagedRelayRequestFailedError({
428 action: "get relay environment status",
429 cause: new Error("Relay request failed."),
430 relayError: new RelayAuthInvalidError({
431 code: "auth_invalid",
432 reason: "invalid_dpop",
433 traceId: "trace-status",
434 }),
435 traceId: "trace-status",
436 }),
437 ),
438 });
439 setSession();
440 const atom = manager.environmentStatusAtom({ accountId: "account-1", environment });
441
442 registry.get(atom);
443 await vi.waitFor(() => {
444 expect(readManagedRelaySnapshotState(registry.get(atom)).error).toBe(
445 `Relay rejected the DPoP proof. ${DPOP_UNKNOWN_HINT}`,
446 );
447 });
448 });
449});