1import type {
2 RelayClientEnvironmentRecord,
3 RelayEnvironmentStatusResponse,
4} from "@t3tools/contracts/relay";
5import type { EnvironmentId } from "@t3tools/contracts";
6import {
7 RelayEnvironmentConnectScope,
8 RelayEnvironmentStatusScope,
9} from "@t3tools/contracts/relay";
10import { decodeRelayJwt } from "@t3tools/shared/relayJwt";
11import * as Cause from "effect/Cause";
12import * as Clock from "effect/Clock";
13import * as Data from "effect/Data";
14import * as Effect from "effect/Effect";
15import * as Option from "effect/Option";
16import * as Schema from "effect/Schema";
17import * as Stream from "effect/Stream";
18import { AsyncResult, Atom, AtomRegistry } from "effect/reactivity";
19
20import { findErrorTraceId } from "../errors/errorTrace.ts";
21import * as ManagedRelay from "./managedRelay.ts";
22import { relayProtectedErrorMessage } from "./errorPresentation.ts";
23
24const DEFAULT_STALE_TIME_MS = 15_000;
25const DEFAULT_IDLE_TTL_MS = 5 * 60_000;
26const CLERK_TOKEN_EXPIRY_SKEW_MS = 5_000;
27const isManagedRelayRequestFailedError = Schema.is(ManagedRelay.ManagedRelayRequestFailedError);
28
29export interface ManagedRelaySession {
30 readonly accountId: string;
31 readonly readClerkToken: () => Effect.Effect<string | null, ManagedRelaySessionError>;
32}
33
34export interface ManagedRelaySessionInput {
35 readonly accountId: string;
36 readonly readClerkToken: () => Promise<string | null>;
37}
38
39interface ManagedRelaySessionControl {
40 readonly updateReadClerkToken: (
41 readClerkToken: ManagedRelaySessionInput["readClerkToken"],
42 ) => void;
43}
44
45export interface ManagedRelaySnapshotState<A> {
46 readonly data: A | null;
47 readonly error: string | null;
48 readonly errorTraceId: string | null;
49 readonly isPending: boolean;
50}
51
52export interface ManagedRelayQueryEvent {
53 readonly operation: "environments" | "devices" | "environment-status";
54 readonly stage: "clerk-token" | "relay-request" | "validation";
55 readonly phase: "start" | "success" | "failure";
56 readonly accountId: string;
57 readonly environmentId?: string;
58 readonly message?: string;
59 readonly traceId?: string | null;
60}
61
62export class ManagedRelaySessionError extends Data.TaggedError("ManagedRelaySessionError")<{
63 readonly message: string;
64 readonly cause?: unknown;
65}> {}
66
67export class ManagedRelaySnapshotError extends Data.TaggedError("ManagedRelaySnapshotError")<{
68 readonly message: string;
69}> {}
70
71export const managedRelaySessionAtom = Atom.make<ManagedRelaySession | null>(null).pipe(
72 Atom.keepAlive,
73 Atom.withLabel("managed-relay:session"),
74);
75
76const managedRelaySessionControls = new WeakMap<ManagedRelaySession, ManagedRelaySessionControl>();
77
78export function createManagedRelaySession(input: ManagedRelaySessionInput): ManagedRelaySession {
79 let cachedToken: { readonly token: string; readonly expiresAtMillis: number } | null = null;
80 let pendingToken: Promise<string | null> | null = null;
81 let readClerkToken = input.readClerkToken;
82 let tokenProviderGeneration = 0;
83
84 const readCachedClerkToken = async (nowMillis: number): Promise<string | null> => {
85 if (cachedToken && cachedToken.expiresAtMillis > nowMillis + CLERK_TOKEN_EXPIRY_SKEW_MS) {
86 return cachedToken.token;
87 }
88 if (pendingToken) {
89 return await pendingToken;
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 try {
102 const expiresAtSeconds = decodeRelayJwt(token).exp;
103 cachedToken =
104 typeof expiresAtSeconds === "number"
105 ? { token, expiresAtMillis: expiresAtSeconds * 1_000 }
106 : null;
107 } catch {
108 cachedToken = null;
109 }
110 return token;
111 });
112 pendingToken = operation;
113 try {
114 return await operation;
115 } finally {
116 if (pendingToken === operation) {
117 pendingToken = null;
118 }
119 }
120 };
121
122 const session: ManagedRelaySession = {
123 accountId: input.accountId,
124 readClerkToken: Effect.fn("clientRuntime.managedRelaySession.readClerkToken")(function* () {
125 const nowMillis = yield* Clock.currentTimeMillis;
126 return yield* Effect.tryPromise({
127 try: () => readCachedClerkToken(nowMillis),
128 catch: (cause) =>
129 new ManagedRelaySessionError({
130 message: "Could not obtain the T3 Connect session token.",
131 cause,
132 }),
133 });
134 }),
135 };
136 managedRelaySessionControls.set(session, {
137 updateReadClerkToken: (nextReadClerkToken) => {
138 readClerkToken = nextReadClerkToken;
139 tokenProviderGeneration += 1;
140 pendingToken = null;
141 },
142 });
143 return session;
144}
145
146export function setManagedRelaySession(
147 registry: AtomRegistry.AtomRegistry,
148 input: ManagedRelaySessionInput | null,
149): void {
150 const current = registry.get(managedRelaySessionAtom);
151 if (input === null) {
152 if (current !== null) {
153 registry.set(managedRelaySessionAtom, null);
154 }
155 return;
156 }
157 if (current?.accountId === input.accountId) {
158 const control = managedRelaySessionControls.get(current);
159 if (control) {
160 // Clerk can replace its token reader during routine same-account refreshes.
161 // Keep the session stable so those refreshes do not invalidate queries or reconnect leases.
162 control.updateReadClerkToken(input.readClerkToken);
163 return;
164 }
165 }
166 registry.set(managedRelaySessionAtom, createManagedRelaySession(input));
167}
168
169export function managedRelayAccountChanges(
170 registry: AtomRegistry.AtomRegistry,
171): Stream.Stream<string | null> {
172 return AtomRegistry.toStream(registry, managedRelaySessionAtom).pipe(
173 Stream.map((session) => session?.accountId ?? null),
174 Stream.changes,
175 Stream.drop(1),
176 );
177}
178
179function readSessionClerkToken(
180 session: ManagedRelaySession,
181): Effect.Effect<string, ManagedRelaySessionError> {
182 return session.readClerkToken().pipe(
183 Effect.filterOrFail(
184 (token): token is string => Boolean(token),
185 () =>
186 new ManagedRelaySessionError({
187 message: "The T3 Connect session token is unavailable.",
188 }),
189 ),
190 );
191}
192
193/** Removes an environment from the signed-in account without contacting that environment. */
194export const deregisterManagedRelayEnvironment = Effect.fn(
195 "clientRuntime.managedRelaySession.deregisterEnvironment",
196)(function* (
197 registry: AtomRegistry.AtomRegistry,
198 input: { readonly accountId: string; readonly environmentId: EnvironmentId },
199) {
200 const session = registry.get(managedRelaySessionAtom);
201 if (!session || session.accountId !== input.accountId) {
202 return yield* new ManagedRelaySessionError({
203 message: "Sign in to T3 Connect before deregistering an environment.",
204 });
205 }
206 const clerkToken = yield* readSessionClerkToken(session);
207 const relay = yield* ManagedRelay.ManagedRelayClient;
208 yield* relay.unlinkEnvironment({ clerkToken, environmentId: input.environmentId });
209});
210
211function requireClerkToken(
212 get: Atom.AtomContext,
213 accountId: string,
214): Effect.Effect<string, ManagedRelaySessionError> {
215 const session = get(managedRelaySessionAtom);
216 if (!session || session.accountId !== accountId) {
217 return Effect.fail(
218 new ManagedRelaySessionError({
219 message: "Sign in to T3 Connect before loading relay data.",
220 }),
221 );
222 }
223 return readSessionClerkToken(session);
224}
225
226function statusKey(input: {
227 readonly accountId: string;
228 readonly environment: RelayClientEnvironmentRecord;
229}): string {
230 return JSON.stringify(input);
231}
232
233function parseStatusKey(key: string): {
234 readonly accountId: string;
235 readonly environment: RelayClientEnvironmentRecord;
236} {
237 return JSON.parse(key) as {
238 readonly accountId: string;
239 readonly environment: RelayClientEnvironmentRecord;
240 };
241}
242
243function endpointMatches(
244 left: RelayClientEnvironmentRecord["endpoint"],
245 right: RelayClientEnvironmentRecord["endpoint"],
246): boolean {
247 return (
248 left.httpBaseUrl === right.httpBaseUrl &&
249 left.wsBaseUrl === right.wsBaseUrl &&
250 left.providerKind === right.providerKind
251 );
252}
253
254function validateEnvironmentStatus(
255 environment: RelayClientEnvironmentRecord,
256 status: RelayEnvironmentStatusResponse,
257): Effect.Effect<RelayEnvironmentStatusResponse, ManagedRelaySnapshotError> {
258 if (status.environmentId !== environment.environmentId) {
259 return Effect.fail(
260 new ManagedRelaySnapshotError({
261 message: "Relay returned status for a different environment.",
262 }),
263 );
264 }
265 if (!endpointMatches(status.endpoint, environment.endpoint)) {
266 return Effect.fail(
267 new ManagedRelaySnapshotError({
268 message: "Relay returned status for a different endpoint.",
269 }),
270 );
271 }
272 if (status.descriptor && status.descriptor.environmentId !== environment.environmentId) {
273 return Effect.fail(
274 new ManagedRelaySnapshotError({
275 message: "Relay returned status descriptor for a different environment.",
276 }),
277 );
278 }
279 return Effect.succeed(status);
280}
281
282export function readManagedRelaySnapshotState<A>(
283 result: AsyncResult.AsyncResult<A, unknown>,
284): ManagedRelaySnapshotState<A> {
285 let error: string | null = null;
286 let errorTraceId: string | null = null;
287 if (result._tag === "Failure") {
288 const cause = Cause.squash(result.cause);
289 error =
290 isManagedRelayRequestFailedError(cause) && cause.relayError
291 ? relayProtectedErrorMessage(cause.relayError)
292 : cause instanceof Error
293 ? cause.message
294 : "Could not load T3 Connect data.";
295 errorTraceId = findErrorTraceId(cause);
296 }
297 return {
298 data: Option.getOrNull(AsyncResult.value(result)),
299 error,
300 errorTraceId,
301 isPending: result.waiting,
302 };
303}
304
305export function createManagedRelayQueryManager(
306 runtime: Atom.AtomRuntime<ManagedRelay.ManagedRelayClient>,
307 options?: {
308 readonly staleTimeMs?: number;
309 readonly idleTtlMs?: number;
310 readonly onQueryEvent?: (event: ManagedRelayQueryEvent) => void;
311 },
312) {
313 const staleTime = options?.staleTimeMs ?? DEFAULT_STALE_TIME_MS;
314 const idleTtl = options?.idleTtlMs ?? DEFAULT_IDLE_TTL_MS;
315 const observe = <A, E, R>(
316 input: Omit<ManagedRelayQueryEvent, "phase" | "message" | "traceId">,
317 effect: Effect.Effect<A, E, R>,
318 ): Effect.Effect<A, E, R> =>
319 Effect.gen(function* () {
320 options?.onQueryEvent?.({ ...input, phase: "start" });
321 return yield* effect.pipe(
322 Effect.onExit((exit) =>
323 Effect.sync(() => {
324 if (exit._tag === "Success") {
325 options?.onQueryEvent?.({ ...input, phase: "success" });
326 return;
327 }
328 const error = Cause.squash(exit.cause);
329 options?.onQueryEvent?.({
330 ...input,
331 phase: "failure",
332 message: error instanceof Error ? error.message : String(error),
333 traceId: findErrorTraceId(error),
334 });
335 }),
336 ),
337 );
338 });
339
340 const environmentsAtom = Atom.family((accountId: string) =>
341 runtime
342 .atom((get) =>
343 Effect.gen(function* () {
344 const base = { operation: "environments" as const, accountId };
345 const clerkToken = yield* observe(
346 { ...base, stage: "clerk-token" },
347 requireClerkToken(get, accountId),
348 );
349 const relay = yield* ManagedRelay.ManagedRelayClient;
350 return yield* observe(
351 { ...base, stage: "relay-request" },
352 relay.listEnvironments({ clerkToken }),
353 );
354 }),
355 )
356 .pipe(
357 Atom.swr({ staleTime, revalidateOnMount: true }),
358 Atom.setIdleTTL(idleTtl),
359 Atom.withLabel(`managed-relay:environments:${accountId}`),
360 ),
361 );
362
363 const devicesAtom = Atom.family((accountId: string) =>
364 runtime
365 .atom((get) =>
366 Effect.gen(function* () {
367 const base = { operation: "devices" as const, accountId };
368 const clerkToken = yield* observe(
369 { ...base, stage: "clerk-token" },
370 requireClerkToken(get, accountId),
371 );
372 const relay = yield* ManagedRelay.ManagedRelayClient;
373 return yield* observe(
374 { ...base, stage: "relay-request" },
375 relay.listDevices({ clerkToken }),
376 );
377 }),
378 )
379 .pipe(
380 Atom.swr({ staleTime, revalidateOnMount: true }),
381 Atom.setIdleTTL(idleTtl),
382 Atom.withLabel(`managed-relay:devices:${accountId}`),
383 ),
384 );
385
386 const environmentStatusAtom = Atom.family((key: string) => {
387 const { accountId, environment } = parseStatusKey(key);
388 return runtime
389 .atom((get) =>
390 Effect.gen(function* () {
391 const base = {
392 operation: "environment-status" as const,
393 accountId,
394 environmentId: environment.environmentId,
395 };
396 const clerkToken = yield* observe(
397 { ...base, stage: "clerk-token" },
398 requireClerkToken(get, accountId),
399 );
400 const relay = yield* ManagedRelay.ManagedRelayClient;
401 const status = yield* observe(
402 { ...base, stage: "relay-request" },
403 relay.getEnvironmentStatus({
404 clerkToken,
405 scopes: [RelayEnvironmentStatusScope, RelayEnvironmentConnectScope],
406 environmentId: environment.environmentId,
407 }),
408 );
409 return yield* observe(
410 { ...base, stage: "validation" },
411 validateEnvironmentStatus(environment, status),
412 );
413 }),
414 )
415 .pipe(
416 Atom.swr({ staleTime, revalidateOnMount: true }),
417 Atom.setIdleTTL(idleTtl),
418 Atom.withLabel(`managed-relay:environment-status:${key}`),
419 );
420 });
421
422 return {
423 environmentsAtom,
424 devicesAtom,
425 environmentStatusAtom: (input: {
426 readonly accountId: string;
427 readonly environment: RelayClientEnvironmentRecord;
428 }) => environmentStatusAtom(statusKey(input)),
429 refreshEnvironments(registry: AtomRegistry.AtomRegistry, accountId: string): void {
430 registry.refresh(environmentsAtom(accountId));
431 },
432 refreshDevices(registry: AtomRegistry.AtomRegistry, accountId: string): void {
433 registry.refresh(devicesAtom(accountId));
434 },
435 refreshEnvironmentStatus(
436 registry: AtomRegistry.AtomRegistry,
437 input: {
438 readonly accountId: string;
439 readonly environment: RelayClientEnvironmentRecord;
440 },
441 ): void {
442 registry.refresh(environmentStatusAtom(statusKey(input)));
443 },
444 };
445}