1import {
2 AuthAccessWriteScope,
3 AuthAccessReadScope,
4 AuthAdministrativeScopes,
5 AuthAccessTokenResult,
6 AuthEnvironmentScopes,
7 AuthSessionState,
8 AuthStandardClientScopes,
9 sessionGrantsScope,
10 EnvironmentId,
11 ExecutionEnvironmentDescriptor,
12 HubError,
13 type AuthEnvironmentScope,
14 type HubDirectory,
15 type HubManagementRequest,
16} from "@t3tools/contracts";
17import * as Context from "effect/Context";
18import * as Clock from "effect/Clock";
19import * as Deferred from "effect/Deferred";
20import * as Effect from "effect/Effect";
21import * as FileSystem from "effect/FileSystem";
22import * as Layer from "effect/Layer";
23import * as Option from "effect/Option";
24import * as Schema from "effect/Schema";
25import * as Redacted from "effect/Redacted";
26import * as Semaphore from "effect/Semaphore";
27import {
28 FetchHttpClient,
29 HttpBody,
30 HttpClient,
31 HttpClientRequest,
32 HttpClientResponse,
33} from "effect/http";
34
35import type { AuthenticatedSession } from "../auth/EnvironmentAuth.ts";
36import * as ServerSecretStore from "../auth/ServerSecretStore.ts";
37import * as ServerEnvironment from "./ServerEnvironment.ts";
38import * as OutboundHub from "./OutboundHub.ts";
39import * as HostedAuthConfig from "../auth/HostedAuthConfig.ts";
40
41const isHubError = Schema.is(HubError);
42const descriptorsEqual = Schema.toEquivalence(ExecutionEnvironmentDescriptor);
43
44const Machine = Schema.Struct({
45 url: Schema.String,
46 token: Schema.NonEmptyString,
47 descriptor: ExecutionEnvironmentDescriptor,
48 scopes: AuthEnvironmentScopes,
49});
50type Machine = typeof Machine.Type;
51const Configuration = Schema.Struct({
52 publicUrl: Schema.String,
53 machines: Schema.Array(Machine),
54});
55const jsonConfiguration = Schema.fromJsonString(Configuration);
56
57export class Hub extends Context.Service<
58 Hub,
59 {
60 readonly directory: (
61 session: AuthenticatedSession,
62 options?: { readonly refresh?: boolean },
63 ) => Effect.Effect<HubDirectory, HubError>;
64 readonly manage: (input: HubManagementRequest) => Effect.Effect<void, HubError>;
65 readonly descriptor: (
66 id: EnvironmentId,
67 ) => Effect.Effect<ExecutionEnvironmentDescriptor, HubError>;
68 readonly authorize: (
69 id: EnvironmentId,
70 session: AuthenticatedSession,
71 ) => Effect.Effect<ReadonlyArray<AuthEnvironmentScope>, HubError>;
72 readonly connect: (
73 id: EnvironmentId,
74 session: AuthenticatedSession,
75 ) => Effect.Effect<
76 {
77 readonly url: string;
78 readonly token: string;
79 readonly invalidated: Effect.Effect<void>;
80 },
81 HubError
82 >;
83 }
84>()("t3/environment/Hub") {}
85
86const normalizeUrl = (value: string) =>
87 Effect.try({
88 try: () => {
89 const url = new URL(value);
90 if (
91 !["http:", "https:"].includes(url.protocol) ||
92 url.username ||
93 url.password ||
94 url.search ||
95 url.hash
96 ) {
97 throw new Error("Expected an HTTP base URL without credentials, query, or fragment.");
98 }
99 url.pathname = `${url.pathname.replace(/\/+$/, "")}/`;
100 return url.href;
101 },
102 catch: (cause) => new HubError({ reason: "address", cause: Redacted.make(cause) }),
103 });
104
105const grantScopes = (machine: Machine, session: AuthenticatedSession) =>
106 AuthStandardClientScopes.filter(
107 (scope) => machine.scopes.includes(scope) && session.scopes.includes(scope),
108 );
109
110/** @public Service construction is part of the canonical Effect module API. */
111export const make = Effect.gen(function* () {
112 const secrets = yield* ServerSecretStore.ServerSecretStore;
113 const files = yield* FileSystem.FileSystem;
114 const client = yield* HttpClient.HttpClient;
115 const environment = yield* ServerEnvironment.ServerEnvironment;
116 const outbound = yield* OutboundHub.OutboundHub;
117 const hosted = yield* HostedAuthConfig.HostedAuthConfig;
118 const lock = yield* Semaphore.make(1);
119 const stored = yield* secrets
120 .get("hub")
121 .pipe(
122 Effect.mapError((cause) => new HubError({ reason: "storage", cause: Redacted.make(cause) })),
123 );
124 let configuration = Option.isSome(stored)
125 ? yield* Schema.decodeEffect(jsonConfiguration)(new TextDecoder().decode(stored.value)).pipe(
126 Effect.mapError(
127 (cause) => new HubError({ reason: "storage", cause: Redacted.make(cause) }),
128 ),
129 )
130 : ({ publicUrl: "", machines: [] } satisfies typeof Configuration.Type);
131 const removals = new Map<EnvironmentId, Deferred.Deferred<void>>();
132 for (const machine of configuration.machines) {
133 removals.set(machine.descriptor.environmentId, yield* Deferred.make<void>());
134 }
135 const credentials = new Map<string, { token: string; expiresAt: number }>();
136 const publicUrl = () =>
137 Option.isSome(hosted) ? `${hosted.value.publicUrl}/` : configuration.publicUrl;
138
139 const request = <A>(
140 machine: Pick<Machine, "url" | "token">,
141 path: string,
142 schema: Schema.Decoder<A>,
143 body?: HttpBody.HttpBody,
144 ) =>
145 Effect.gen(function* () {
146 let outgoing = HttpClientRequest.make(body ? "POST" : "GET")(new URL(path, machine.url).href);
147 if (machine.token) outgoing = HttpClientRequest.bearerToken(outgoing, machine.token);
148 if (body) outgoing = HttpClientRequest.setBody(outgoing, body);
149 const response = yield* client
150 .execute(outgoing)
151 .pipe(Effect.provideService(FetchHttpClient.RequestInit, { redirect: "manual" }));
152 if (response.status !== 200)
153 return yield* new HubError({
154 reason: response.status === 401 || response.status === 403 ? "credential" : "upstream",
155 });
156 return yield* HttpClientResponse.schemaBodyJson(schema)(response).pipe(
157 Effect.mapError(
158 (cause) =>
159 new HubError({
160 reason: path === ".well-known/t3/environment" ? "identity" : "upstream",
161 cause: Redacted.make(cause),
162 }),
163 ),
164 );
165 }).pipe(
166 Effect.timeout("10 seconds"),
167 Effect.mapError((cause) =>
168 isHubError(cause)
169 ? cause
170 : new HubError({ reason: "upstream", cause: Redacted.make(cause) }),
171 ),
172 Effect.scoped,
173 );
174
175 const find = (id: EnvironmentId): Effect.Effect<Machine, HubError> => {
176 const machine = configuration.machines.find((value) => value.descriptor.environmentId === id);
177 return machine ? Effect.succeed(machine) : Effect.fail(new HubError({ reason: "not-found" }));
178 };
179 const persist = (next: typeof Configuration.Type) =>
180 Schema.encodeEffect(jsonConfiguration)(next).pipe(
181 Effect.flatMap((json) => secrets.set("hub", new TextEncoder().encode(json))),
182 Effect.mapError((cause) => new HubError({ reason: "storage", cause: Redacted.make(cause) })),
183 Effect.tap(() =>
184 Effect.sync(() => {
185 configuration = next;
186 }),
187 ),
188 );
189
190 const refreshDescriptor = (id: EnvironmentId) =>
191 Effect.gen(function* () {
192 const remote = (yield* outbound.enrollments).find(
193 (value) => value.descriptor.environmentId === id,
194 );
195 if (remote) return remote.descriptor;
196 const machine = yield* find(id);
197 const descriptor = yield* request(
198 { ...machine, token: "" },
199 ".well-known/t3/environment",
200 ExecutionEnvironmentDescriptor,
201 ).pipe(
202 Effect.catchTags({
203 HubError: (error) =>
204 error.reason === "upstream" ? Effect.succeed(machine.descriptor) : Effect.fail(error),
205 }),
206 );
207 if (descriptor.environmentId !== id) return yield* new HubError({ reason: "identity" });
208 if (!descriptorsEqual(descriptor, machine.descriptor)) {
209 yield* lock.withPermit(
210 Effect.gen(function* () {
211 const current = yield* find(id);
212 if (current.url !== machine.url) return;
213 yield* persist({
214 ...configuration,
215 machines: configuration.machines.map((value) =>
216 value.descriptor.environmentId === id ? { ...value, descriptor } : value,
217 ),
218 });
219 }),
220 );
221 }
222 return descriptor;
223 });
224
225 return Hub.of({
226 directory: (session, options) =>
227 Effect.gen(function* () {
228 const baseUrl = publicUrl();
229 if (!baseUrl) return yield* new HubError({ reason: "configuration" });
230 const registered = yield* outbound.enrollments;
231 return {
232 hubEnvironmentId: yield* environment.getEnvironmentId,
233 environments: yield* Effect.forEach(
234 [
235 ...configuration.machines.filter(
236 (machine) =>
237 !registered.some(
238 (value) => value.descriptor.environmentId === machine.descriptor.environmentId,
239 ),
240 ),
241 ...registered.map((value) => ({
242 descriptor: value.descriptor,
243 scopes: AuthStandardClientScopes,
244 })),
245 ],
246 (machine) =>
247 Effect.gen(function* () {
248 const descriptor =
249 options?.refresh === false
250 ? machine.descriptor
251 : yield* refreshDescriptor(machine.descriptor.environmentId);
252 return {
253 environmentId: machine.descriptor.environmentId,
254 label: descriptor.label,
255 descriptor,
256 httpBaseUrl: new URL(
257 `environments/${encodeURIComponent(machine.descriptor.environmentId)}/`,
258 baseUrl,
259 ).href,
260 scopes: AuthStandardClientScopes.filter(
261 (scope) => machine.scopes.includes(scope) && session.scopes.includes(scope),
262 ),
263 };
264 }),
265 { concurrency: "unbounded" },
266 ),
267 };
268 }),
269 descriptor: refreshDescriptor,
270 authorize: (id, session) =>
271 Effect.gen(function* () {
272 if ((yield* outbound.enrollments).some((value) => value.descriptor.environmentId === id)) {
273 const scopes = AuthStandardClientScopes.filter((scope) => session.scopes.includes(scope));
274 if (scopes.length === 0) return yield* new HubError({ reason: "credential" });
275 return scopes;
276 }
277 return yield* find(id).pipe(
278 Effect.flatMap((machine) => {
279 const scopes = grantScopes(machine, session);
280 return scopes.length > 0
281 ? Effect.succeed(scopes)
282 : Effect.fail(new HubError({ reason: "credential" }));
283 }),
284 );
285 }),
286 manage: (input) =>
287 lock.withPermit(
288 Effect.gen(function* () {
289 if (input.type === "hub.status") return;
290 if (input.type === "hub.configure") {
291 const publicUrl = yield* normalizeUrl(input.publicUrl);
292 yield* persist({ ...configuration, publicUrl });
293 return;
294 }
295 if (!publicUrl()) return yield* new HubError({ reason: "configuration" });
296 if (input.type === "hub.remove") {
297 if (
298 (yield* outbound.enrollments).some(
299 (value) => value.descriptor.environmentId === input.environmentId,
300 )
301 ) {
302 yield* outbound.remove(input.environmentId);
303 return;
304 }
305 yield* find(input.environmentId);
306 yield* persist({
307 ...configuration,
308 machines: configuration.machines.filter(
309 (machine) => machine.descriptor.environmentId !== input.environmentId,
310 ),
311 });
312 const removal = removals.get(input.environmentId);
313 if (removal) yield* Deferred.succeed(removal, undefined);
314 removals.delete(input.environmentId);
315 for (const key of credentials.keys())
316 if (key.startsWith(`${input.environmentId}:`)) credentials.delete(key);
317 return;
318 }
319 const url = yield* normalizeUrl(input.url);
320 const token = (yield* files
321 .readFileString(input.tokenFile)
322 .pipe(
323 Effect.mapError(
324 (cause) => new HubError({ reason: "storage", cause: Redacted.make(cause) }),
325 ),
326 )).trim();
327 const descriptor = yield* request(
328 { url, token: "" },
329 ".well-known/t3/environment",
330 ExecutionEnvironmentDescriptor,
331 );
332 if (
333 descriptor.environmentId !== input.environmentId ||
334 descriptor.environmentId === (yield* environment.getEnvironmentId)
335 ) {
336 return yield* new HubError({ reason: "identity" });
337 }
338 const state = yield* request({ url, token }, "api/auth/session", AuthSessionState);
339 if (
340 !sessionGrantsScope(state, AuthAccessReadScope) ||
341 !sessionGrantsScope(state, AuthAccessWriteScope)
342 ) {
343 return yield* new HubError({ reason: "credential" });
344 }
345 yield* persist({
346 ...configuration,
347 machines: [
348 ...configuration.machines.filter(
349 (machine) => machine.descriptor.environmentId !== input.environmentId,
350 ),
351 {
352 url,
353 token,
354 descriptor,
355 scopes: AuthAdministrativeScopes.filter((scope) =>
356 sessionGrantsScope(state, scope),
357 ),
358 },
359 ],
360 });
361 const removal = removals.get(input.environmentId);
362 if (removal) yield* Deferred.succeed(removal, undefined);
363 removals.set(input.environmentId, yield* Deferred.make<void>());
364 for (const key of credentials.keys())
365 if (key.startsWith(`${input.environmentId}:`)) credentials.delete(key);
366 }),
367 ),
368 connect: (id, session) =>
369 lock.withPermit(
370 Effect.gen(function* () {
371 if (
372 (yield* outbound.enrollments).some((value) => value.descriptor.environmentId === id)
373 ) {
374 return yield* outbound.connect(id, session);
375 }
376 const machine = yield* find(id);
377 const scopes = grantScopes(machine, session);
378 if (scopes.length === 0) return yield* new HubError({ reason: "credential" });
379 const key = `${id}:${session.sessionId}:${scopes.toSorted().join(" ")}`;
380 const now = yield* Clock.currentTimeMillis;
381 for (const [key, credential] of credentials) {
382 if (credential.expiresAt <= now + 5_000) credentials.delete(key);
383 }
384 const cached = credentials.get(key);
385 let token = cached && cached.expiresAt > now + 5_000 ? cached.token : undefined;
386 if (!token) {
387 // A changed endpoint must prove the enrolled identity before receiving its private grant.
388 const descriptor = yield* request(
389 { ...machine, token: "" },
390 ".well-known/t3/environment",
391 ExecutionEnvironmentDescriptor,
392 );
393 if (descriptor.environmentId !== id) return yield* new HubError({ reason: "identity" });
394 const issued = yield* request(
395 machine,
396 "api/auth/delegated-session",
397 AuthAccessTokenResult,
398 HttpBody.jsonUnsafe({ scopes, label: "Self-hosted server client" }),
399 );
400 if (
401 issued.token_type !== "Bearer" ||
402 issued.scope.split(" ").some((scope) => !scopes.some((granted) => granted === scope))
403 ) {
404 return yield* new HubError({ reason: "credential" });
405 }
406 token = issued.access_token;
407 credentials.set(key, { token, expiresAt: now + issued.expires_in * 1_000 });
408 }
409 const removal = removals.get(id);
410 if (!removal) return yield* new HubError({ reason: "not-found" });
411 return { url: machine.url, token, invalidated: Deferred.await(removal) };
412 }),
413 ),
414 });
415});
416
417export const layer = Layer.effect(Hub, make);