1import { pipe } from "effect/Function";
2import type { AuthSessionState, EnvironmentId, ServerConfig } from "@t3tools/contracts";
3import * as Schema from "effect/Schema";
4import * as Effect from "effect/Effect";
5import * as Option from "effect/Option";
6import * as Stream from "effect/Stream";
7import * as SubscriptionRef from "effect/SubscriptionRef";
8import { HttpClient } from "effect/http";
9import { AsyncResult, Atom } from "effect/reactivity";
10
11import * as RemoteEnvironmentAuthorization from "../authorization/service.ts";
12import * as EnvironmentRegistry from "../connection/registry.ts";
13import type { PreparedConnection } from "../connection/model.ts";
14import * as EnvironmentSupervisor from "../connection/supervisor.ts";
15import * as ManagedRelay from "../relay/managedRelay.ts";
16import type { RemoteEnvironmentRequestError } from "../rpc/http.ts";
17import type { RpcSession } from "../rpc/session.ts";
18import { safeErrorLogAttributes } from "../errors/safeLog.ts";
19import { executeAuthenticatedEnvironmentHttpRequest } from "./environmentHttpAuth.ts";
20import { followStreamInEnvironment } from "./environmentStreams.ts";
21
22/** @public Required to name the error in consumers' inferred session results. */
23export class SessionHttpClientUnavailable extends Schema.TaggedError<SessionHttpClientUnavailable>()(
24 "SessionHttpClientUnavailable",
25 {},
26) {}
27
28function initialConfigOption<E>(
29 initialConfig: Effect.Effect<ServerConfig, E>,
30): Effect.Effect<Option.Option<ServerConfig>> {
31 return initialConfig.pipe(
32 Effect.asSome,
33 Effect.catch((error) =>
34 Effect.logWarning("Could not load the initial environment configuration.").pipe(
35 Effect.annotateLogs({ ...safeErrorLogAttributes(error) }),
36 Effect.as(Option.none<ServerConfig>()),
37 ),
38 ),
39 );
40}
41
42const DEFAULT_SESSION_STATE_TIMEOUT_MS = 6_000;
43const MAX_SESSION_STATE_TIMEOUT_MS = 60_000;
44const SESSION_STATE_RETRY_BASE_MS = 2_000;
45const SESSION_STATE_RETRY_MAX_MS = 60_000;
46
47export const fetchEnvironmentSessionState = Effect.fn(
48 "clientRuntime.state.fetchEnvironmentSessionState",
49)(function* (input: {
50 readonly prepared: PreparedConnection;
51 readonly signer: Option.Option<ManagedRelay.ManagedRelayDpopSigner["Service"]>;
52 readonly remoteAuthorization?: Option.Option<
53 RemoteEnvironmentAuthorization.RemoteEnvironmentAuthorization["Service"]
54 >;
55 readonly timeoutMs?: number;
56}) {
57 const request = (
58 timeoutMs: number,
59 ): Effect.Effect<AuthSessionState, RemoteEnvironmentRequestError, HttpClient.HttpClient> =>
60 executeAuthenticatedEnvironmentHttpRequest({
61 ...input,
62 group: "auth",
63 method: "GET",
64 url: (urls) => urls.session(),
65 timeoutMs,
66 request: ({ client, headers }) => client.session({ headers }),
67 // This endpoint returns 200 with authenticated:false for expired credentials.
68 isUnauthorizedResponse: (response) => !response.authenticated,
69 }).pipe(
70 // A timeout says nothing about grants; failing would show a busy host as read-only.
71 Effect.catchTags({
72 RemoteEnvironmentAuthTimeoutError: (error) => {
73 const nextTimeoutMs = Math.min(timeoutMs * 2, MAX_SESSION_STATE_TIMEOUT_MS);
74 return nextTimeoutMs > timeoutMs ? request(nextTimeoutMs) : Effect.fail(error);
75 },
76 }),
77 );
78 return yield* request(input.timeoutMs ?? DEFAULT_SESSION_STATE_TIMEOUT_MS);
79});
80
81function makeEnvironmentSessionAtoms<R, E>(
82 runtime: Atom.AtomRuntime<EnvironmentRegistry.EnvironmentRegistry | R, E>,
83) {
84 const initialConfigAtom = Atom.family((environmentId: EnvironmentId) =>
85 runtime.atom(
86 followStreamInEnvironment(
87 environmentId,
88 Stream.unwrap(
89 EnvironmentSupervisor.EnvironmentSupervisor.pipe(
90 Effect.map((supervisor) =>
91 SubscriptionRef.changes(supervisor.session).pipe(
92 Stream.mapEffect(
93 Option.match({
94 onNone: () => Effect.succeed(Option.none<ServerConfig>()),
95 onSome: (session) => initialConfigOption(session.initialConfig),
96 }),
97 ),
98 ),
99 ),
100 ),
101 ),
102 ),
103 { initialValue: Option.none() },
104 ),
105 );
106
107 // This is only the bootstrap config captured when a transport session is
108 // established. Consumers that need current provider/settings state must use
109 // createServerEnvironmentAtoms(...).configValueAtom instead.
110 const initialConfigValueAtom = Atom.family((environmentId: EnvironmentId) =>
111 Atom.make((get): ServerConfig | null =>
112 pipe(
113 get(initialConfigAtom(environmentId)),
114 AsyncResult.value,
115 Option.flatten,
116 Option.getOrNull,
117 ),
118 ).pipe(Atom.withLabel(`environment-config-value:${environmentId}`)),
119 );
120
121 const preparedConnectionAtom = Atom.family((environmentId: EnvironmentId) =>
122 runtime.atom(
123 followStreamInEnvironment(
124 environmentId,
125 Stream.unwrap(
126 EnvironmentSupervisor.EnvironmentSupervisor.pipe(
127 Effect.map((supervisor) => SubscriptionRef.changes(supervisor.prepared)),
128 ),
129 ),
130 ),
131 { initialValue: Option.none<PreparedConnection>() },
132 ),
133 );
134
135 const preparedConnectionValueAtom = Atom.family((environmentId: EnvironmentId) =>
136 Atom.make((get) =>
137 AsyncResult.getOrElse(get(preparedConnectionAtom(environmentId)), () =>
138 Option.none<PreparedConnection>(),
139 ),
140 ).pipe(Atom.withLabel(`environment-prepared-connection:${environmentId}`)),
141 );
142
143 const transportSessionAtom = Atom.family((environmentId: EnvironmentId) =>
144 runtime.atom(
145 followStreamInEnvironment(
146 environmentId,
147 Stream.unwrap(
148 EnvironmentSupervisor.EnvironmentSupervisor.pipe(
149 Effect.map((supervisor) => SubscriptionRef.changes(supervisor.session)),
150 ),
151 ),
152 ),
153 { initialValue: Option.none() },
154 ),
155 );
156
157 const sessionStateAtom = Atom.family((environmentId: EnvironmentId) => {
158 let consecutiveFailures = 0;
159 let backoffConnection: RpcSession | null = null;
160 return runtime
161 .atom((get) => {
162 const prepared = Option.getOrNull(get(preparedConnectionValueAtom(environmentId)));
163 // Each established connection fetches anew, and a failure retries while it stays up,
164 // so a fetch that failed mid-reconnect cannot outlive the socket that recovered.
165 const connection = AsyncResult.getOrElse(get(transportSessionAtom(environmentId)), () =>
166 Option.none(),
167 );
168 if (prepared === null || Option.isNone(connection)) {
169 return Effect.never;
170 }
171 if (connection.value !== backoffConnection) {
172 backoffConnection = connection.value;
173 consecutiveFailures = 0;
174 }
175 return Effect.gen(function* () {
176 const signer = yield* Effect.serviceOption(ManagedRelay.ManagedRelayDpopSigner);
177 const remoteAuthorization = yield* Effect.serviceOption(
178 RemoteEnvironmentAuthorization.RemoteEnvironmentAuthorization,
179 );
180 const client = yield* Effect.serviceOption(HttpClient.HttpClient);
181 if (Option.isNone(client)) return yield* new SessionHttpClientUnavailable();
182 return yield* fetchEnvironmentSessionState({
183 prepared,
184 signer,
185 remoteAuthorization,
186 }).pipe(Effect.provideService(HttpClient.HttpClient, client.value));
187 }).pipe(
188 Effect.tap(() =>
189 Effect.sync(() => {
190 consecutiveFailures = 0;
191 }),
192 ),
193 Effect.tapError(() => {
194 const delayMs = Math.min(
195 SESSION_STATE_RETRY_BASE_MS * 2 ** consecutiveFailures,
196 SESSION_STATE_RETRY_MAX_MS,
197 );
198 consecutiveFailures += 1;
199 return Effect.sleep(delayMs).pipe(
200 Effect.andThen(Effect.sync(() => get.refreshSelf())),
201 Effect.forkScoped,
202 );
203 }),
204 );
205 })
206 .pipe(
207 Atom.swr({ staleTime: 30_000, revalidateOnMount: true }),
208 Atom.setIdleTTL(5 * 60_000),
209 Atom.withLabel(`environment-session-state:${environmentId}`),
210 );
211 });
212
213 const sessionStateValueAtom = Atom.family((environmentId: EnvironmentId) =>
214 Atom.make(
215 (get): AuthSessionState | null =>
216 AsyncResult.getOrElse(get(sessionStateAtom(environmentId)), () => null) ?? null,
217 ).pipe(Atom.withLabel(`environment-session-state-value:${environmentId}`)),
218 );
219
220 return {
221 initialConfigAtom,
222 initialConfigValueAtom,
223 preparedConnectionAtom,
224 preparedConnectionValueAtom,
225 sessionStateAtom,
226 sessionStateValueAtom,
227 };
228}
229
230const sessionAtomsByRuntime = new WeakMap<object, ReturnType<typeof makeEnvironmentSessionAtoms>>();
231
232export function createEnvironmentSessionAtoms<R, E>(
233 runtime: Atom.AtomRuntime<EnvironmentRegistry.EnvironmentRegistry | R, E>,
234): ReturnType<typeof makeEnvironmentSessionAtoms<R, E>> {
235 const existing = sessionAtomsByRuntime.get(runtime);
236 if (existing) return existing as ReturnType<typeof makeEnvironmentSessionAtoms<R, E>>;
237 const atoms = makeEnvironmentSessionAtoms(runtime);
238 sessionAtomsByRuntime.set(runtime, atoms);
239 return atoms;
240}