Open History to see the last commit
1import type { EnvironmentId } from "@t3tools/contracts";
2import { decodeRelayJwt } from "@t3tools/shared/relayJwt";
3import * as Cause from "effect/Cause";
4import * as Clock from "effect/Clock";
5import * as Effect from "effect/Effect";
6import * as Option from "effect/Option";
7import * as Schema from "effect/Schema";
8import * as Stream from "effect/Stream";
9import { AsyncResult, Atom, AtomRegistry } from "effect/reactivity";
10
11import { findErrorTraceId } from "../errors/errorTrace.ts";
12import * as ManagedRelay from "./managedRelay.ts";
13import { relayProtectedErrorMessage } from "./errorPresentation.ts";
14
15const DEFAULT_STALE_TIME_MS = 15_000;
16const DEFAULT_IDLE_TTL_MS = 5 * 60_000;
17const CLERK_TOKEN_EXPIRY_SKEW_MS = 5_000;
18// Clerk retries a failed token request itself, but waits 3 s before the first
19// retry and longer before each next one. A fresh request usually answers at
20// once, so a slow read is replaced instead of waited out.
21const CLERK_TOKEN_READ_TIMEOUT_MS = 4_000;
22const CLERK_TOKEN_READ_ATTEMPTS = 3;
23const isManagedRelayRequestFailedError = Schema.is(ManagedRelay.ManagedRelayRequestFailedError);
24
25export interface ManagedRelaySession {
26 readonly accountId: string;
27 readonly readClerkToken: () => Effect.Effect<string | null, ManagedRelaySessionError>;
28}
29
30export interface ManagedRelaySessionInput {
31 readonly accountId: string;
32 readonly readClerkToken: () => Promise<string | null>;
33}
34
35interface ManagedRelaySessionControl {
36 readonly updateReadClerkToken: (
37 readClerkToken: ManagedRelaySessionInput["readClerkToken"],
38 ) => void;
39}
40
41export interface ManagedRelaySnapshotState<A> {
42 readonly data: A | null;
43 readonly error: string | null;
44 readonly errorTraceId: string | null;
45 readonly isPending: boolean;
46}
47
48export interface ManagedRelayQueryEvent {
49 readonly operation: "environments" | "devices" | "t3chat-link";
50 readonly stage: "clerk-token" | "relay-request";
51 readonly phase: "start" | "success" | "failure";
52 readonly accountId: string;
53 readonly environmentId?: string;
54 readonly message?: string;
55 readonly traceId?: string | null;
56}
57
58export class ManagedRelaySessionError extends Schema.TaggedError<ManagedRelaySessionError>()(
59 "ManagedRelaySessionError",
60 {
61 message: Schema.String,
62 cause: Schema.optional(Schema.Defect()),
63 },
64) {}
65
66export const managedRelaySessionAtom = Atom.make<ManagedRelaySession | null>(null).pipe(
67 Atom.keepAlive,
68 Atom.withLabel("managed-relay:session"),
69);
70
71const managedRelaySessionControls = new WeakMap<ManagedRelaySession, ManagedRelaySessionControl>();
72
73export function createManagedRelaySession(input: ManagedRelaySessionInput): ManagedRelaySession {
74 let cachedToken: { readonly token: string; readonly expiresAtMillis: number } | null = null;
75 let pendingToken: {
76 readonly operation: Promise<string | null>;
77 readonly startedAtMillis: number;
78 } | null = null;
79 let readClerkToken = input.readClerkToken;
80 let tokenProviderGeneration = 0;
81
82 const readCachedClerkToken = async (nowMillis: number): Promise<string | null> => {
83 if (cachedToken && cachedToken.expiresAtMillis > nowMillis + CLERK_TOKEN_EXPIRY_SKEW_MS) {
84 return cachedToken.token;
85 }
86 // A read that outlived its timeout is left running, since it still fills
87 // the cache, but later readers start a fresh one instead of joining it.
88 if (pendingToken && nowMillis - pendingToken.startedAtMillis < CLERK_TOKEN_READ_TIMEOUT_MS) {
89 return await pendingToken.operation;
90 }
91
92 const operationGeneration = tokenProviderGeneration;
93 const operation = readClerkToken().then((token) => {
94 if (operationGeneration !== tokenProviderGeneration) {
95 return token;
96 }
97 if (!token) {
98 cachedToken = null;
99 return null;
100 }
101 const expiresAtSeconds = Effect.runSync(
102 decodeRelayJwt(token).pipe(
103 Effect.map((claims) => claims.exp),
104 Effect.orElseSucceed(() => undefined),
105 ),
106 );
107 cachedToken =
108 typeof expiresAtSeconds === "number"
109 ? { token, expiresAtMillis: expiresAtSeconds * 1_000 }
110 : null;
111 return token;
112 });
113 const pending = { operation, startedAtMillis: nowMillis };
114 pendingToken = pending;
115 try {
116 return await operation;
117 } finally {
118 if (pendingToken === pending) {
119 pendingToken = null;
120 }
121 }
122 };
123
124 const session: ManagedRelaySession = {
125 accountId: input.accountId,
126 readClerkToken: Effect.fn("clientRuntime.managedRelaySession.readClerkToken")(function* () {
127 for (let attempt = 0; attempt < CLERK_TOKEN_READ_ATTEMPTS; attempt += 1) {
128 const nowMillis = yield* Clock.currentTimeMillis;
129 const token = yield* Effect.tryPromise({
130 try: () => readCachedClerkToken(nowMillis),
131 catch: (cause) =>
132 new ManagedRelaySessionError({
133 message: "Could not obtain the T3 Connect session token.",
134 cause,
135 }),
136 }).pipe(Effect.timeoutOption(CLERK_TOKEN_READ_TIMEOUT_MS));
137 if (Option.isSome(token)) return token.value;
138 }
139 return yield* new ManagedRelaySessionError({
140 message: "Timed out obtaining the T3 Connect session token.",
141 });
142 }),
143 };
144 managedRelaySessionControls.set(session, {
145 updateReadClerkToken: (nextReadClerkToken) => {
146 readClerkToken = nextReadClerkToken;
147 tokenProviderGeneration += 1;
148 pendingToken = null;
149 },
150 });
151 return session;
152}
153
154export function setManagedRelaySession(
155 registry: AtomRegistry.AtomRegistry,
156 input: ManagedRelaySessionInput | null,
157): void {
158 const current = registry.get(managedRelaySessionAtom);
159 if (input === null) {
160 if (current !== null) {
161 registry.set(managedRelaySessionAtom, null);
162 }
163 return;
164 }
165 if (current?.accountId === input.accountId) {
166 const control = managedRelaySessionControls.get(current);
167 if (control) {
168 // Clerk can replace its token reader during routine same-account refreshes.
169 // Keep the session stable so those refreshes do not invalidate queries or reconnect leases.
170 control.updateReadClerkToken(input.readClerkToken);
171 return;
172 }
173 }
174 registry.set(managedRelaySessionAtom, createManagedRelaySession(input));
175}
176
177export function managedRelayAccountChanges(
178 registry: AtomRegistry.AtomRegistry,
179): Stream.Stream<string | null> {
180 return AtomRegistry.toStream(registry, managedRelaySessionAtom).pipe(
181 Stream.map((session) => session?.accountId ?? null),
182 Stream.changes,
183 Stream.drop(1),
184 );
185}
186
187function readSessionClerkToken(
188 session: ManagedRelaySession,
189): Effect.Effect<string, ManagedRelaySessionError> {
190 return session.readClerkToken().pipe(
191 Effect.filterOrFail(
192 (token): token is string => Boolean(token),
193 () =>
194 new ManagedRelaySessionError({
195 message: "The T3 Connect session token is unavailable.",
196 }),
197 ),
198 );
199}
200
201function requireSessionClerkToken(
202 registry: AtomRegistry.AtomRegistry,
203 accountId: string,
204 action: string,
205): Effect.Effect<string, ManagedRelaySessionError> {
206 const session = registry.get(managedRelaySessionAtom);
207 if (!session || session.accountId !== accountId) {
208 return Effect.fail(
209 new ManagedRelaySessionError({ message: `Sign in to T3 Connect before ${action}.` }),
210 );
211 }
212 return readSessionClerkToken(session);
213}
214
215export const deregisterManagedRelayEnvironment = Effect.fn(
216 "clientRuntime.managedRelaySession.deregisterEnvironment",
217)(function* (
218 registry: AtomRegistry.AtomRegistry,
219 input: { readonly accountId: string; readonly environmentId: EnvironmentId },
220) {
221 const clerkToken = yield* requireSessionClerkToken(
222 registry,
223 input.accountId,
224 "deregistering an environment",
225 );
226 const relay = yield* ManagedRelay.ManagedRelayClient;
227 yield* relay.unlinkEnvironment({ clerkToken, environmentId: input.environmentId });
228});
229
230/** Returns the T3 Chat page URL where the signed-in user confirms the link. */
231export const createManagedRelayT3ChatLinkRequest = Effect.fn(
232 "clientRuntime.managedRelaySession.createT3ChatLinkRequest",
233)(function* (registry: AtomRegistry.AtomRegistry, input: { readonly accountId: string }) {
234 const clerkToken = yield* requireSessionClerkToken(registry, input.accountId, "linking T3 Chat");
235 const relay = yield* ManagedRelay.ManagedRelayClient;
236 return (yield* relay.createT3ChatLinkRequest({ clerkToken })).url;
237});
238
239export const unlinkManagedRelayT3Chat = Effect.fn("clientRuntime.managedRelaySession.unlinkT3Chat")(
240 function* (registry: AtomRegistry.AtomRegistry, input: { readonly accountId: string }) {
241 const clerkToken = yield* requireSessionClerkToken(
242 registry,
243 input.accountId,
244 "unlinking T3 Chat",
245 );
246 const relay = yield* ManagedRelay.ManagedRelayClient;
247 yield* relay.unlinkT3Chat({ clerkToken });
248 },
249);
250
251function requireClerkToken(
252 get: Atom.AtomContext,
253 accountId: string,
254): Effect.Effect<string, ManagedRelaySessionError> {
255 const session = get(managedRelaySessionAtom);
256 if (!session || session.accountId !== accountId) {
257 return Effect.fail(
258 new ManagedRelaySessionError({
259 message: "Sign in to T3 Connect before loading relay data.",
260 }),
261 );
262 }
263 return readSessionClerkToken(session);
264}
265
266function describeManagedRelayFailure(cause: Cause.Cause<unknown>): {
267 readonly message: string;
268 readonly traceId: string | null;
269} {
270 const error = Cause.squash(cause);
271 return {
272 message:
273 isManagedRelayRequestFailedError(error) && error.relayError
274 ? relayProtectedErrorMessage(error.relayError)
275 : error instanceof Error
276 ? error.message
277 : "Could not load T3 Connect data.",
278 traceId: findErrorTraceId(error),
279 };
280}
281
282export function readManagedRelaySnapshotState<A>(
283 result: AsyncResult.AsyncResult<A, unknown>,
284): ManagedRelaySnapshotState<A> {
285 const failure = result._tag === "Failure" ? describeManagedRelayFailure(result.cause) : null;
286 return {
287 data: AsyncResult.getOrElse(result, () => null),
288 error: failure?.message ?? null,
289 errorTraceId: failure?.traceId ?? null,
290 isPending: result.waiting,
291 };
292}
293
294// Logged where the listing runs so each failed fetch reports once, however many
295// components read the atom.
296const logListingFailure =
297 (label: string) =>
298 <A, E, R>(self: Effect.Effect<A, E, R>): Effect.Effect<A, E, R> =>
299 self.pipe(
300 Effect.tapCause((cause) => {
301 if (Cause.hasInterruptsOnly(cause)) return Effect.void;
302 const { message, traceId } = describeManagedRelayFailure(cause);
303 return Effect.logError(`[t3-cloud] Relay ${label} listing failed`).pipe(
304 Effect.annotateLogs({ message, traceId }),
305 );
306 }),
307 );
308
309export function createManagedRelayQueryManager(
310 runtime: Atom.AtomRuntime<ManagedRelay.ManagedRelayClient>,
311 options?: {
312 readonly staleTimeMs?: number;
313 readonly idleTtlMs?: number;
314 readonly onQueryEvent?: (event: ManagedRelayQueryEvent) => void;
315 },
316) {
317 const staleTime = options?.staleTimeMs ?? DEFAULT_STALE_TIME_MS;
318 const idleTtl = options?.idleTtlMs ?? DEFAULT_IDLE_TTL_MS;
319 const observe = Effect.fn("managedRelayState.observe")(function* <A, E, R>(
320 input: Omit<ManagedRelayQueryEvent, "phase" | "message" | "traceId">,
321 effect: Effect.Effect<A, E, R>,
322 ): Effect.fn.Return<A, E, R> {
323 options?.onQueryEvent?.({ ...input, phase: "start" });
324 return yield* effect.pipe(
325 Effect.onExit((exit) =>
326 Effect.sync(() => {
327 if (exit._tag === "Success") {
328 options?.onQueryEvent?.({ ...input, phase: "success" });
329 return;
330 }
331 const error = Cause.squash(exit.cause);
332 options?.onQueryEvent?.({
333 ...input,
334 phase: "failure",
335 message: error instanceof Error ? error.message : String(error),
336 traceId: findErrorTraceId(error),
337 });
338 }),
339 ),
340 );
341 });
342
343 const environmentsAtom = Atom.family((accountId: string) =>
344 runtime
345 .atom((get) =>
346 Effect.gen(function* () {
347 const base = { operation: "environments" as const, accountId };
348 const clerkToken = yield* observe(
349 { ...base, stage: "clerk-token" },
350 requireClerkToken(get, accountId),
351 );
352 const relay = yield* ManagedRelay.ManagedRelayClient;
353 return yield* observe(
354 { ...base, stage: "relay-request" },
355 relay.listEnvironments({ clerkToken }),
356 );
357 }).pipe(logListingFailure("environment")),
358 )
359 .pipe(
360 Atom.swr({ staleTime, revalidateOnMount: true }),
361 Atom.setIdleTTL(idleTtl),
362 Atom.withLabel(`managed-relay:environments:${accountId}`),
363 ),
364 );
365
366 const devicesAtom = Atom.family((accountId: string) =>
367 runtime
368 .atom((get) =>
369 Effect.gen(function* () {
370 const base = { operation: "devices" as const, accountId };
371 const clerkToken = yield* observe(
372 { ...base, stage: "clerk-token" },
373 requireClerkToken(get, accountId),
374 );
375 const relay = yield* ManagedRelay.ManagedRelayClient;
376 return yield* observe(
377 { ...base, stage: "relay-request" },
378 relay.listDevices({ clerkToken }),
379 );
380 }).pipe(logListingFailure("device")),
381 )
382 .pipe(
383 Atom.swr({ staleTime, revalidateOnMount: true }),
384 Atom.setIdleTTL(idleTtl),
385 Atom.withLabel(`managed-relay:devices:${accountId}`),
386 ),
387 );
388
389 const t3ChatLinkAtom = Atom.family((accountId: string) =>
390 runtime
391 .atom((get) =>
392 Effect.gen(function* () {
393 const base = { operation: "t3chat-link" as const, accountId };
394 const clerkToken = yield* observe(
395 { ...base, stage: "clerk-token" },
396 requireClerkToken(get, accountId),
397 );
398 const relay = yield* ManagedRelay.ManagedRelayClient;
399 return yield* observe(
400 { ...base, stage: "relay-request" },
401 relay.getT3ChatLink({ clerkToken }),
402 );
403 }),
404 )
405 .pipe(
406 Atom.swr({ staleTime, revalidateOnMount: true }),
407 Atom.setIdleTTL(idleTtl),
408 Atom.withLabel(`managed-relay:t3chat-link:${accountId}`),
409 ),
410 );
411
412 return {
413 environmentsAtom,
414 devicesAtom,
415 t3ChatLinkAtom,
416 refreshEnvironments(registry: AtomRegistry.AtomRegistry, accountId: string): void {
417 registry.refresh(environmentsAtom(accountId));
418 },
419 refreshDevices(registry: AtomRegistry.AtomRegistry, accountId: string): void {
420 registry.refresh(devicesAtom(accountId));
421 },
422 refreshT3ChatLink(registry: AtomRegistry.AtomRegistry, accountId: string): void {
423 registry.refresh(t3ChatLinkAtom(accountId));
424 },
425 };
426}