Open History to see the last commit
1import { EnvironmentId } from "@t3tools/contracts";
2import {
3 type RelayClientDeviceRecord,
4 type RelayClientEnvironmentRecord,
5} from "@t3tools/contracts/relay";
6import { describe, expect, it } from "@effect/vitest";
7import * as Effect from "effect/Effect";
8import * as Fiber from "effect/Fiber";
9import * as Layer from "effect/Layer";
10import * as Logger from "effect/Logger";
11import * as TestClock from "effect/testing/TestClock";
12import { Atom, AtomRegistry } from "effect/reactivity";
13import { afterEach, vi } from "vite-plus/test";
14
15import * as ManagedRelay from "./managedRelay.ts";
16import {
17 createManagedRelayQueryManager,
18 createManagedRelaySession,
19 deregisterManagedRelayEnvironment,
20 ManagedRelaySessionError,
21 managedRelaySessionAtom,
22 setManagedRelaySession,
23} from "./managedRelayState.ts";
24
25let registry = AtomRegistry.make();
26
27const environment = {
28 environmentId: EnvironmentId.make("environment-1"),
29 label: "Main environment",
30 endpoint: {
31 httpBaseUrl: "https://environment.example.test",
32 wsBaseUrl: "wss://environment.example.test",
33 providerKind: "cloudflare_tunnel",
34 },
35 linkedAt: "2026-06-01T00:00:00.000Z",
36} satisfies RelayClientEnvironmentRecord;
37
38const device = {
39 deviceId: "device-1",
40 label: "Julius iPhone",
41 platform: "ios",
42 iosMajorVersion: 18,
43 appVersion: null,
44 notifications: {
45 enabled: true,
46 notifyOnApproval: true,
47 notifyOnInput: true,
48 notifyOnCompletion: true,
49 notifyOnFailure: true,
50 },
51 liveActivities: {
52 enabled: true,
53 },
54 updatedAt: "2026-06-01T00:00:00.000Z",
55} satisfies RelayClientDeviceRecord;
56
57function resetRegistry() {
58 registry.dispose();
59 registry = AtomRegistry.make();
60}
61
62function createClient(overrides?: Partial<ManagedRelay.ManagedRelayClient["Service"]>) {
63 return ManagedRelay.ManagedRelayClient.of({
64 relayUrl: "https://relay.example.test",
65 listEnvironments: () => Effect.succeed([environment]),
66 listDevices: () => Effect.succeed([device]),
67 createEnvironmentLinkChallenge: () => Effect.die("unused"),
68 linkEnvironment: () => Effect.die("unused"),
69 unlinkEnvironment: () => Effect.die("unused"),
70 getT3ChatLink: () => Effect.die("unused"),
71 createT3ChatLinkRequest: () => Effect.die("unused"),
72 unlinkT3Chat: () => Effect.die("unused"),
73 getEnvironmentStatus: () =>
74 Effect.succeed({
75 environmentId: environment.environmentId,
76 endpoint: environment.endpoint,
77 status: "online",
78 checkedAt: "2026-06-01T00:00:00.000Z",
79 }),
80 connectEnvironment: () => Effect.die("unused"),
81 registerDevice: () => Effect.die("unused"),
82 unregisterDevice: () => Effect.die("unused"),
83 registerLiveActivity: () => Effect.die("unused"),
84 getAgentActivitySnapshot: () => Effect.die("unused"),
85 resetTokenCache: Effect.void,
86 ...overrides,
87 });
88}
89
90function setSession() {
91 setManagedRelaySession(registry, {
92 accountId: "account-1",
93 readClerkToken: () => Promise.resolve("clerk-token"),
94 });
95}
96
97function clerkToken(expiresAtSeconds: number): string {
98 const encode = (value: unknown) =>
99 btoa(JSON.stringify(value)).replaceAll("+", "-").replaceAll("/", "_").replace(/=+$/u, "");
100 return `${encode({ alg: "RS256" })}.${encode({ exp: expiresAtSeconds })}.c2lnbmF0dXJl`;
101}
102
103describe("createManagedRelayQueryManager", () => {
104 afterEach(resetRegistry);
105
106 it.effect("logs a failed device listing once however many readers mount it", () =>
107 Effect.gen(function* () {
108 const logs: Array<unknown> = [];
109 const runtime = Atom.runtime(
110 Layer.mergeAll(
111 Layer.succeed(
112 ManagedRelay.ManagedRelayClient,
113 createClient({
114 listDevices: () =>
115 Effect.fail(
116 new ManagedRelay.ManagedRelayRequestFailedError({
117 action: "list relay client devices",
118 cause: new Error("relay down"),
119 }),
120 ),
121 }),
122 ),
123 Logger.layer([Logger.make(({ message }) => logs.push(message))], {
124 mergeWithExisting: false,
125 }),
126 ),
127 );
128 const manager = createManagedRelayQueryManager(runtime);
129 setSession();
130 const atom = manager.devicesAtom("account-1");
131
132 const failed = Promise.withResolvers<void>();
133 const readers = [0, 1].map(() =>
134 registry.subscribe(
135 atom,
136 (result) => {
137 if (result._tag === "Failure") failed.resolve();
138 },
139 { immediate: true },
140 ),
141 );
142 yield* Effect.promise(() => failed.promise);
143 for (const unsubscribe of readers) unsubscribe();
144
145 expect(logs).toEqual([["[t3-cloud] Relay device listing failed"]]);
146 }),
147 );
148
149 it.effect("rejects deregistration after the account changes", () =>
150 Effect.gen(function* () {
151 const unlinkEnvironment = vi.fn(() => Effect.succeed({ ok: true }));
152 setSession();
153
154 const error = yield* deregisterManagedRelayEnvironment(registry, {
155 accountId: "previous-account",
156 environmentId: environment.environmentId,
157 }).pipe(
158 Effect.provideService(ManagedRelay.ManagedRelayClient, createClient({ unlinkEnvironment })),
159 Effect.flip,
160 );
161
162 expect(error).toBeInstanceOf(ManagedRelaySessionError);
163 expect(unlinkEnvironment).not.toHaveBeenCalled();
164 }),
165 );
166
167 it.effect("deduplicates concurrent Clerk token reads and reuses the token until JWT expiry", () =>
168 Effect.gen(function* () {
169 const token = clerkToken(4_102_444_800);
170 let resolveToken!: (value: string) => void;
171 const readClerkToken = vi.fn(
172 () =>
173 new Promise<string>((resolve) => {
174 resolveToken = resolve;
175 }),
176 );
177 const session = createManagedRelaySession({
178 accountId: "account-1",
179 readClerkToken,
180 });
181
182 const readsFiber = yield* Effect.all([session.readClerkToken(), session.readClerkToken()], {
183 concurrency: "unbounded",
184 }).pipe(Effect.forkChild);
185 yield* Effect.yieldNow;
186 expect(readClerkToken).toHaveBeenCalledTimes(1);
187
188 resolveToken(token);
189 expect(yield* Fiber.join(readsFiber)).toEqual([token, token]);
190 expect(yield* session.readClerkToken()).toBe(token);
191 expect(readClerkToken).toHaveBeenCalledTimes(1);
192 }),
193 );
194
195 it.effect("replaces a Clerk token read that has not answered after 4 seconds", () =>
196 Effect.gen(function* () {
197 const token = clerkToken(4_102_444_800);
198 const readClerkToken = vi
199 .fn<() => Promise<string | null>>()
200 .mockReturnValueOnce(new Promise(() => {}))
201 .mockResolvedValue(token);
202 const session = createManagedRelaySession({ accountId: "account-1", readClerkToken });
203
204 const readsFiber = yield* Effect.all([session.readClerkToken(), session.readClerkToken()], {
205 concurrency: "unbounded",
206 }).pipe(Effect.forkChild);
207 yield* TestClock.adjust("3999 millis");
208 expect(readClerkToken).toHaveBeenCalledTimes(1);
209
210 yield* TestClock.adjust("1 millis");
211 expect(yield* Fiber.join(readsFiber)).toEqual([token, token]);
212 expect(readClerkToken).toHaveBeenCalledTimes(2);
213 }),
214 );
215
216 it.effect("fails a Clerk token read that never answers after three tries", () =>
217 Effect.gen(function* () {
218 const readClerkToken = vi.fn(() => new Promise<string | null>(() => {}));
219 const session = createManagedRelaySession({ accountId: "account-1", readClerkToken });
220
221 const readFiber = yield* session.readClerkToken().pipe(Effect.flip, Effect.forkChild);
222 yield* TestClock.adjust("12 seconds");
223
224 expect((yield* Fiber.join(readFiber)).message).toBe(
225 "Timed out obtaining the T3 Connect session token.",
226 );
227 expect(readClerkToken).toHaveBeenCalledTimes(3);
228 }),
229 );
230
231 it.effect("updates the token provider without replacing a same-account session", () =>
232 Effect.gen(function* () {
233 const firstRead = vi.fn(() => Promise.resolve<string | null>(null));
234 setManagedRelaySession(registry, {
235 accountId: "account-1",
236 readClerkToken: firstRead,
237 });
238 const firstSession = registry.get(managedRelaySessionAtom);
239 expect(firstSession).not.toBeNull();
240 expect(yield* firstSession!.readClerkToken()).toBeNull();
241
242 const secondRead = vi.fn(() => Promise.resolve<string | null>("refreshed-token"));
243 setManagedRelaySession(registry, {
244 accountId: "account-1",
245 readClerkToken: secondRead,
246 });
247
248 expect(registry.get(managedRelaySessionAtom)).toBe(firstSession);
249 expect(yield* firstSession!.readClerkToken()).toBe("refreshed-token");
250 expect(firstRead).toHaveBeenCalledTimes(1);
251 expect(secondRead).toHaveBeenCalledTimes(1);
252 }),
253 );
254
255 it.effect("does not pin a refreshed session to an older pending token read", () =>
256 Effect.gen(function* () {
257 let resolveFirst!: (token: string) => void;
258 setManagedRelaySession(registry, {
259 accountId: "account-1",
260 readClerkToken: () =>
261 new Promise<string>((resolve) => {
262 resolveFirst = resolve;
263 }),
264 });
265 const session = registry.get(managedRelaySessionAtom);
266 const firstRead = yield* session!.readClerkToken().pipe(Effect.forkChild);
267 yield* Effect.yieldNow;
268
269 setManagedRelaySession(registry, {
270 accountId: "account-1",
271 readClerkToken: () => Promise.resolve("refreshed-token"),
272 });
273
274 expect(yield* session!.readClerkToken()).toBe("refreshed-token");
275 resolveFirst("older-token");
276 expect(yield* Fiber.join(firstRead)).toBe("older-token");
277 }),
278 );
279});