1import type {
2 RelayClientEnvironmentRecord,
3 RelayEnvironmentStatusResponse,
4} from "@t3tools/contracts/relay";
5import { decodeRelayJwt } from "@t3tools/shared/relayJwt";
6import {
7 RelayEnvironmentConnectScope,
8 RelayEnvironmentStatusScope,
9} from "@t3tools/contracts/relay";
10import * as Context from "effect/Context";
11import * as Effect from "effect/Effect";
12import * as Layer from "effect/Layer";
13import * as Option from "effect/Option";
14import * as Ref from "effect/Ref";
15import * as Semaphore from "effect/Semaphore";
16import * as Stream from "effect/Stream";
17import * as SubscriptionRef from "effect/SubscriptionRef";
18
19import * as ManagedRelay from "./managedRelay.ts";
20import * as ClientCapabilities from "../platform/capabilities.ts";
21import * as Connectivity from "../connection/connectivity.ts";
22import { mapManagedRelayError } from "../connection/errors.ts";
23import { ConnectionBlockedError, type ConnectionAttemptError } from "../connection/model.ts";
24import * as ConnectionWakeups from "../connection/wakeups.ts";
25
26export type RelayEnvironmentAvailability = "checking" | "online" | "offline" | "error";
27
28export interface RelayDiscoveredEnvironment {
29 readonly environment: RelayClientEnvironmentRecord;
30 readonly availability: RelayEnvironmentAvailability;
31 readonly status: Option.Option<RelayEnvironmentStatusResponse>;
32 readonly error: Option.Option<ConnectionAttemptError>;
33}
34
35export interface RelayEnvironmentDiscoveryState {
36 readonly environments: ReadonlyMap<string, RelayDiscoveredEnvironment>;
37 // Only a loaded account list distinguishes a missing registration from an unknown one.
38 readonly listed: boolean;
39 readonly refreshing: boolean;
40 readonly offline: boolean;
41 readonly error: Option.Option<ConnectionAttemptError>;
42}
43
44export class RelayEnvironmentDiscovery extends Context.Service<
45 RelayEnvironmentDiscovery,
46 {
47 readonly state: SubscriptionRef.SubscriptionRef<RelayEnvironmentDiscoveryState>;
48 readonly refresh: Effect.Effect<void>;
49 }
50>()("@t3tools/client-runtime/relay/discovery/RelayEnvironmentDiscovery") {}
51
52export const EMPTY_RELAY_ENVIRONMENT_DISCOVERY_STATE: RelayEnvironmentDiscoveryState = {
53 environments: new Map(),
54 listed: false,
55 refreshing: false,
56 offline: false,
57 error: Option.none(),
58};
59
60export type RelayEnvironmentRegistration = "registered" | "unregistered" | "unknown";
61
62export function relayEnvironmentRegistration(
63 state: RelayEnvironmentDiscoveryState,
64 environmentId: string,
65): RelayEnvironmentRegistration {
66 if (state.environments.has(environmentId)) return "registered";
67 return state.listed ? "unregistered" : "unknown";
68}
69
70function validateStatus(
71 environment: RelayClientEnvironmentRecord,
72 status: RelayEnvironmentStatusResponse,
73): Effect.Effect<RelayEnvironmentStatusResponse, ConnectionAttemptError> {
74 if (status.environmentId !== environment.environmentId) {
75 return Effect.fail(
76 new ConnectionBlockedError({
77 reason: "configuration",
78 detail: "Relay returned status for a different environment.",
79 }),
80 );
81 }
82 if (
83 status.endpoint.httpBaseUrl !== environment.endpoint.httpBaseUrl ||
84 status.endpoint.wsBaseUrl !== environment.endpoint.wsBaseUrl ||
85 status.endpoint.providerKind !== environment.endpoint.providerKind
86 ) {
87 return Effect.fail(
88 new ConnectionBlockedError({
89 reason: "configuration",
90 detail: "Relay returned status for a different environment endpoint.",
91 }),
92 );
93 }
94 if (
95 status.descriptor !== undefined &&
96 status.descriptor.environmentId !== environment.environmentId
97 ) {
98 return Effect.fail(
99 new ConnectionBlockedError({
100 reason: "configuration",
101 detail: "Relay returned a descriptor for a different environment.",
102 }),
103 );
104 }
105 return Effect.succeed(status);
106}
107
108const relayAccountId = Effect.fn("clientRuntime.discovery.relayAccountId")(function* (
109 clerkToken: string,
110) {
111 const claims = yield* decodeRelayJwt(clerkToken).pipe(Effect.option);
112 const sub = Option.isSome(claims) ? claims.value.sub : undefined;
113 return typeof sub === "string" && sub.length > 0 ? Option.some(sub) : Option.none<string>();
114});
115
116/** @public */
117export const make = Effect.fn("RelayEnvironmentDiscovery.make")(function* () {
118 const relay = yield* ManagedRelay.ManagedRelayClient;
119 const session = yield* ClientCapabilities.CloudSession;
120 const connectivity = yield* Connectivity.Connectivity;
121 const wakeups = yield* ConnectionWakeups.ConnectionWakeups;
122 const state = yield* SubscriptionRef.make(EMPTY_RELAY_ENVIRONMENT_DISCOVERY_STATE);
123 const refreshLock = yield* Semaphore.make(1);
124 const hasRefreshed = yield* Ref.make(false);
125 const accountGeneration = yield* Ref.make(0);
126 const activeAccountId = yield* Ref.make<Option.Option<string>>(Option.none());
127 const refreshGeneration = yield* Ref.make(0);
128 const offlineReportFingerprints = yield* Ref.make<ReadonlyMap<string, string>>(new Map());
129
130 const clearOfflineReport = Effect.fn("RelayEnvironmentDiscovery.clearOfflineReport")(function* (
131 environmentId: string,
132 ) {
133 yield* Ref.update(offlineReportFingerprints, (current) => {
134 if (!current.has(environmentId)) {
135 return current;
136 }
137 const next = new Map(current);
138 next.delete(environmentId);
139 return next;
140 });
141 });
142
143 const updateEnvironment = Effect.fn("RelayEnvironmentDiscovery.updateEnvironment")(function* (
144 generation: number,
145 environmentId: string,
146 update: (current: RelayDiscoveredEnvironment) => RelayDiscoveredEnvironment,
147 ) {
148 if ((yield* Ref.get(accountGeneration)) !== generation) {
149 return;
150 }
151 yield* SubscriptionRef.update(state, (current) => {
152 const entry = current.environments.get(environmentId);
153 if (entry === undefined) {
154 return current;
155 }
156 const environments = new Map(current.environments);
157 environments.set(environmentId, update(entry));
158 return { ...current, environments };
159 });
160 });
161
162 const refreshStatus = Effect.fn("RelayEnvironmentDiscovery.refreshStatus")(function* (
163 generation: number,
164 clerkToken: string,
165 environment: RelayClientEnvironmentRecord,
166 ) {
167 const result = yield* relay
168 .getEnvironmentStatus({
169 clerkToken,
170 scopes: [RelayEnvironmentStatusScope, RelayEnvironmentConnectScope],
171 environmentId: environment.environmentId,
172 })
173 .pipe(
174 Effect.mapError(mapManagedRelayError),
175 Effect.flatMap((status) => validateStatus(environment, status)),
176 Effect.result,
177 );
178
179 if (result._tag === "Success") {
180 if (result.success.status === "offline") {
181 const fingerprint = `${result.success.endpoint.httpBaseUrl}\n${result.success.error ?? ""}`;
182 const shouldReport = yield* Ref.modify(offlineReportFingerprints, (current) => {
183 if (current.get(environment.environmentId) === fingerprint) {
184 return [false, current];
185 }
186 return [true, new Map(current).set(environment.environmentId, fingerprint)];
187 });
188 if (shouldReport) {
189 yield* Effect.logWarning("Relay environment health check reported offline", {
190 environmentId: result.success.environmentId,
191 endpoint: result.success.endpoint.httpBaseUrl,
192 message: result.success.error,
193 traceId: result.success.traceId,
194 });
195 }
196 } else {
197 yield* clearOfflineReport(environment.environmentId);
198 }
199 yield* updateEnvironment(generation, environment.environmentId, (current) => ({
200 ...current,
201 availability: result.success.status,
202 status: Option.some(result.success),
203 error: Option.none(),
204 }));
205 return;
206 }
207
208 yield* clearOfflineReport(environment.environmentId);
209 yield* updateEnvironment(generation, environment.environmentId, (current) => ({
210 ...current,
211 availability: "error",
212 error: Option.some(result.failure),
213 }));
214 });
215
216 const refresh = refreshLock.withPermits(1)(
217 Effect.gen(function* () {
218 yield* Ref.set(hasRefreshed, true);
219 if ((yield* connectivity.status) === "offline") {
220 yield* SubscriptionRef.update(state, (current) => ({
221 ...current,
222 refreshing: false,
223 offline: true,
224 }));
225 return;
226 }
227
228 let generation = yield* Ref.get(accountGeneration);
229 yield* Ref.set(refreshGeneration, generation);
230 yield* SubscriptionRef.set(state, {
231 environments: new Map(),
232 listed: false,
233 refreshing: true,
234 offline: false,
235 error: Option.none(),
236 });
237
238 // Sign-out triggers credentials-changed refreshes and must settle to an empty list.
239 // Only session-level missing credentials are benign; relay-side auth failures must surface.
240 const tokenResult = yield* Effect.result(session.clerkToken);
241 if (tokenResult._tag === "Failure") {
242 const failure = tokenResult.failure;
243 if (failure._tag === "ConnectionBlockedError" && failure.reason === "authentication") {
244 if ((yield* Ref.get(accountGeneration)) !== generation) {
245 return;
246 }
247 yield* SubscriptionRef.update(state, (current) => ({
248 ...current,
249 refreshing: false,
250 }));
251 return;
252 }
253 return yield* failure;
254 }
255 const clerkToken = tokenResult.success;
256 if ((yield* Ref.get(accountGeneration)) !== generation) {
257 return;
258 }
259 const accountId = yield* relayAccountId(clerkToken);
260 const previousAccountId = yield* Ref.get(activeAccountId);
261 if (
262 Option.isSome(previousAccountId) &&
263 (!Option.isSome(accountId) || previousAccountId.value !== accountId.value)
264 ) {
265 generation = yield* Ref.updateAndGet(accountGeneration, (current) => current + 1);
266 yield* Ref.set(refreshGeneration, generation);
267 }
268 yield* Ref.set(activeAccountId, accountId);
269
270 const environments = yield* relay
271 .listEnvironments({ clerkToken })
272 .pipe(Effect.mapError(mapManagedRelayError));
273 if ((yield* Ref.get(accountGeneration)) !== generation) {
274 return;
275 }
276 const next = new Map<string, RelayDiscoveredEnvironment>();
277 for (const environment of environments) {
278 next.set(environment.environmentId, {
279 environment,
280 availability: "checking",
281 status: Option.none(),
282 error: Option.none(),
283 });
284 }
285 yield* SubscriptionRef.update(state, (current) => ({
286 ...current,
287 environments: next,
288 listed: true,
289 }));
290
291 yield* Effect.forEach(
292 environments,
293 (environment) => refreshStatus(generation, clerkToken, environment),
294 {
295 concurrency: "unbounded",
296 discard: true,
297 },
298 );
299 if ((yield* Ref.get(accountGeneration)) !== generation) {
300 return;
301 }
302 yield* SubscriptionRef.update(state, (current) => ({
303 ...current,
304 refreshing: false,
305 }));
306 }).pipe(
307 Effect.catch((error) =>
308 Effect.gen(function* () {
309 const generation = yield* Ref.get(refreshGeneration);
310 if ((yield* Ref.get(accountGeneration)) !== generation) {
311 return;
312 }
313 yield* SubscriptionRef.update(state, (current) => ({
314 ...current,
315 refreshing: false,
316 error: Option.some(error),
317 }));
318 }),
319 ),
320 ),
321 );
322
323 yield* connectivity.changes.pipe(
324 Stream.changes,
325 Stream.runForEach((networkStatus) =>
326 networkStatus === "offline"
327 ? SubscriptionRef.update(state, (current) => ({
328 ...current,
329 refreshing: false,
330 offline: true,
331 }))
332 : Ref.get(hasRefreshed).pipe(
333 Effect.flatMap((shouldRefresh) => (shouldRefresh ? refresh : Effect.void)),
334 ),
335 ),
336 Effect.forkScoped,
337 );
338 yield* wakeups.changes.pipe(
339 Stream.runForEach((reason) =>
340 reason === "credentials-changed"
341 ? Effect.gen(function* () {
342 yield* Ref.update(accountGeneration, (current) => current + 1);
343 yield* Ref.set(activeAccountId, Option.none());
344 yield* Ref.set(offlineReportFingerprints, new Map());
345 yield* SubscriptionRef.set(state, EMPTY_RELAY_ENVIRONMENT_DISCOVERY_STATE);
346 yield* refresh.pipe(Effect.forkScoped);
347 })
348 : Effect.void,
349 ),
350 Effect.forkScoped,
351 );
352
353 return RelayEnvironmentDiscovery.of({ state, refresh });
354});
355
356export const layer = Layer.effect(RelayEnvironmentDiscovery, make());