1import {
2 EnvironmentId,
3 HubDirectory,
4 HubInbox,
5 HubInboxCommand,
6 HubInboxEnvironment,
7} from "@t3tools/contracts";
8import * as Context from "effect/Context";
9import * as Effect from "effect/Effect";
10import * as Equal from "effect/Equal";
11import * as Layer from "effect/Layer";
12import * as Option from "effect/Option";
13import * as Schema from "effect/Schema";
14import * as Semaphore from "effect/Semaphore";
15import * as SubscriptionRef from "effect/SubscriptionRef";
16import { HttpBody, HttpClient } from "effect/http";
17
18import {
19 BearerConnectionCredential,
20 BearerConnectionProfile,
21 BearerConnectionRegistration,
22 HubConnectionCredential,
23 HostedConnectionCredential,
24} from "./catalog.ts";
25import {
26 BearerConnectionTarget,
27 ConnectionBlockedError,
28 ConnectionTransientError,
29} from "./model.ts";
30import type { ConnectionAttemptError } from "./model.ts";
31import { preparePairingRegistration, type PairingConnectionInput } from "./onboarding.ts";
32import * as CredentialStore from "./credentialStore.ts";
33import * as ProfileStore from "./profileStore.ts";
34import * as Registry from "./registry.ts";
35import { connectionRouteId, connectionRoutes } from "./routes.ts";
36import { orchestrationProtocolCompatibilityError } from "./compatibility.ts";
37import { fetchRemoteEnvironmentDescriptor } from "../environment/descriptor.ts";
38import { deriveWsBaseUrl, environmentEndpointUrl } from "../environment/endpoint.ts";
39import * as ClientCapabilities from "../platform/capabilities.ts";
40
41const HUB_PROFILE_ID = "self-hosted-server";
42const isConnectionAttemptError = Schema.is(
43 Schema.Union([ConnectionBlockedError, ConnectionTransientError]),
44);
45const encodeInboxCommand = Schema.encodeEffect(Schema.toCodecJson(HubInboxCommand));
46const decodeInbox = Schema.decodeUnknownEffect(Schema.toCodecJson(HubInbox));
47const decodeInboxEnvironment = Schema.decodeUnknownEffect(Schema.toCodecJson(HubInboxEnvironment));
48
49export interface HubServerConnection {
50 readonly environmentId: EnvironmentId;
51 readonly label: string;
52 readonly httpBaseUrl: string;
53 readonly authorization?: "t3-connect" | "hosted";
54}
55
56export interface HubConnectionState {
57 readonly connection: HubServerConnection | null;
58 readonly directory: HubDirectory | null;
59 readonly inbox: HubInbox;
60 readonly refreshing: boolean;
61 readonly error: string | null;
62}
63
64export const EMPTY_HUB_CONNECTION_STATE: HubConnectionState = {
65 connection: null,
66 directory: null,
67 inbox: { environments: [] },
68 refreshing: false,
69 error: null,
70};
71
72export class HubConnection extends Context.Service<
73 HubConnection,
74 {
75 readonly state: SubscriptionRef.SubscriptionRef<HubConnectionState>;
76 readonly connect: (
77 input: PairingConnectionInput,
78 ) => Effect.Effect<void, ConnectionAttemptError>;
79 readonly connectAuthenticated: (input: {
80 readonly httpBaseUrl: string;
81 }) => Effect.Effect<void, ConnectionAttemptError>;
82 readonly headers: Effect.Effect<Record<string, string>, ConnectionAttemptError>;
83 readonly refresh: Effect.Effect<void, ConnectionAttemptError>;
84 readonly refreshInbox: Effect.Effect<void, ConnectionAttemptError>;
85 readonly organize: (
86 input: HubInboxCommand,
87 ) => Effect.Effect<Option.Option<{ readonly sequence: number }>, ConnectionAttemptError>;
88 readonly disconnect: Effect.Effect<void, ConnectionAttemptError>;
89 }
90>()("@t3tools/client-runtime/connection/hub/HubConnection") {}
91
92const hubCredentialId = (environmentId: EnvironmentId) => `hub-session:${environmentId}`;
93const hubRoutePrefix = (environmentId: EnvironmentId) => `hub:${environmentId}:`;
94
95export const make = Effect.gen(function* () {
96 const profiles = yield* ProfileStore.ConnectionProfileStore;
97 const credentials = yield* CredentialStore.ConnectionCredentialStore;
98 const registry = yield* Registry.EnvironmentRegistry;
99 const http = yield* HttpClient.HttpClient;
100 const presentation = yield* ClientCapabilities.ClientPresentation;
101 const lock = yield* Semaphore.make(1);
102 const saved = Option.getOrNull(yield* profiles.get(HUB_PROFILE_ID));
103 const connection = saved?._tag === "BearerConnectionProfile" ? saved : null;
104 const state = yield* SubscriptionRef.make<HubConnectionState>({
105 ...EMPTY_HUB_CONNECTION_STATE,
106 connection,
107 });
108 let inboxCache:
109 | {
110 readonly url: string;
111 readonly token: string;
112 readonly etag: string | undefined;
113 readonly inbox: HubInbox;
114 }
115 | undefined;
116
117 const currentCredential = Effect.fn("HubConnection.currentCredential")(function* (
118 profile: HubServerConnection,
119 ) {
120 const credential = Option.getOrNull(
121 yield* credentials.get(hubCredentialId(profile.environmentId)),
122 );
123 if (
124 credential?._tag !== "BearerConnectionCredential" &&
125 credential?._tag !== "HostedConnectionCredential"
126 ) {
127 return yield* new ConnectionBlockedError({
128 reason: "authentication",
129 detail: "Sign in to this server to restore access.",
130 });
131 }
132 return credential;
133 });
134
135 const headersFor = (credential: BearerConnectionCredential | HostedConnectionCredential) =>
136 credential._tag === "BearerConnectionCredential"
137 ? Effect.succeed({ authorization: `Bearer ${credential.token}` } as Record<string, string>)
138 : Effect.gen(function* () {
139 const access = yield* Effect.serviceOption(ClientCapabilities.HostedAuthAccess);
140 if (Option.isNone(access))
141 return yield* new ConnectionBlockedError({
142 reason: "authentication",
143 detail: "Sign in to this server to reconnect.",
144 });
145 return yield* access.value.headers(credential.serverUrl);
146 });
147
148 const requestInbox = Effect.fn("HubConnection.inbox")(
149 function* (
150 profile: HubServerConnection,
151 credential: BearerConnectionCredential | HostedConnectionCredential,
152 ) {
153 const url = environmentEndpointUrl(profile.httpBaseUrl, "/api/hub/inbox");
154 const headers = yield* headersFor(credential);
155 const cacheIdentity = headers.authorization ?? headers["x-csrf-token"] ?? "cookie";
156 const cached =
157 inboxCache?.url === url && inboxCache.token === cacheIdentity ? inboxCache : undefined;
158 const response = yield* http.get(url, {
159 headers: {
160 ...headers,
161 ...(cached?.etag === undefined ? {} : { "if-none-match": cached.etag }),
162 },
163 });
164 if (response.status === 304 && cached?.etag !== undefined) return cached.inbox;
165 if (response.status === 401 || response.status === 403) {
166 return yield* new ConnectionBlockedError({
167 reason: "authentication",
168 detail: "Sign in to this server to restore access.",
169 });
170 }
171 if (response.status !== 200) {
172 return yield* new ConnectionTransientError({
173 reason: "remote-unavailable",
174 detail: "Couldn't load chats from this server. Try again.",
175 });
176 }
177 const inbox = yield* response.json.pipe(Effect.flatMap(decodeInbox));
178 inboxCache = { url, token: cacheIdentity, etag: response.headers.etag, inbox };
179 return inbox;
180 },
181 Effect.timeout("15 seconds"),
182 Effect.mapError((error) =>
183 isConnectionAttemptError(error)
184 ? error
185 : new ConnectionTransientError({
186 reason: "remote-unavailable",
187 detail: "Couldn't load chats from this server. Try again.",
188 }),
189 ),
190 );
191
192 const requestDirectory = Effect.fn("HubConnection.directory")(
193 function* (
194 profile: HubServerConnection,
195 credential: BearerConnectionCredential | HostedConnectionCredential,
196 ) {
197 const response = yield* http
198 .get(environmentEndpointUrl(profile.httpBaseUrl, "/api/hub/environments"), {
199 headers: yield* headersFor(credential),
200 })
201 .pipe(
202 Effect.mapError(
203 () =>
204 new ConnectionTransientError({
205 reason: "remote-unavailable",
206 detail: "Couldn't reach this server. Check the address and try again.",
207 }),
208 ),
209 );
210 if (response.status === 401 || response.status === 403) {
211 return yield* new ConnectionBlockedError({
212 reason: "authentication",
213 detail: "Sign in to this server to restore access.",
214 });
215 }
216 if (response.status === 404) {
217 return yield* new ConnectionBlockedError({
218 reason: "unsupported",
219 detail: "This server can't share other machines. Pair it as a single machine.",
220 });
221 }
222 if (response.status !== 200) {
223 return yield* new ConnectionTransientError({
224 reason: "remote-unavailable",
225 detail: "The server is unavailable. Try again.",
226 });
227 }
228 const directory = yield* response.json.pipe(
229 Effect.flatMap(Schema.decodeUnknownEffect(HubDirectory)),
230 Effect.mapError(
231 () =>
232 new ConnectionBlockedError({
233 reason: "configuration",
234 detail: "The server returned an invalid machine list.",
235 }),
236 ),
237 );
238 if (directory.hubEnvironmentId !== profile.environmentId) {
239 return yield* new ConnectionBlockedError({
240 reason: "configuration",
241 detail: "This address reaches a different server. Check the address and pair again.",
242 });
243 }
244 const seen = new Set<EnvironmentId>();
245 for (const machine of directory.environments) {
246 const expected = environmentEndpointUrl(
247 profile.httpBaseUrl,
248 `/environments/${encodeURIComponent(machine.environmentId)}/`,
249 );
250 if (
251 seen.has(machine.environmentId) ||
252 machine.descriptor.environmentId !== machine.environmentId ||
253 machine.httpBaseUrl !== expected
254 ) {
255 return yield* new ConnectionBlockedError({
256 reason: "configuration",
257 detail: "The server returned an invalid machine address.",
258 });
259 }
260 seen.add(machine.environmentId);
261 }
262 return directory;
263 },
264 Effect.timeoutOrElse({
265 duration: "15 seconds",
266 orElse: () =>
267 Effect.fail(
268 new ConnectionTransientError({
269 reason: "timeout",
270 detail: "The server took too long to respond. Try again.",
271 }),
272 ),
273 }),
274 );
275
276 const removeRoutes = Effect.fn("HubConnection.removeRoutes")(function* (
277 environmentId: EnvironmentId,
278 keep: ReadonlySet<string> = new Set(),
279 ) {
280 const entries = yield* SubscriptionRef.get(registry.entries);
281 for (const [id, entry] of entries) {
282 for (const route of connectionRoutes(entry)) {
283 const routeId = connectionRouteId(route.target);
284 if (routeId.startsWith(hubRoutePrefix(environmentId)) && !keep.has(routeId)) {
285 yield* registry.removeRoute(id, routeId, { retainDrafts: true });
286 }
287 }
288 }
289 });
290
291 const reconcile = Effect.fn("HubConnection.reconcile")(function* (
292 profile: HubServerConnection,
293 directory: HubDirectory,
294 ) {
295 const keep = new Set<string>();
296 for (const machine of directory.environments) {
297 const connectionId = `${hubRoutePrefix(profile.environmentId)}${machine.environmentId}`;
298 keep.add(connectionId);
299 const registration = new BearerConnectionRegistration({
300 target: new BearerConnectionTarget({
301 connectionId,
302 environmentId: machine.environmentId,
303 label: machine.label,
304 }),
305 profile: new BearerConnectionProfile({
306 connectionId,
307 environmentId: machine.environmentId,
308 label: machine.label,
309 httpBaseUrl: machine.httpBaseUrl,
310 wsBaseUrl: deriveWsBaseUrl(machine.httpBaseUrl),
311 }),
312 credential: new HubConnectionCredential({
313 connectionId: hubCredentialId(profile.environmentId),
314 }),
315 });
316 const entry = (yield* SubscriptionRef.get(registry.entries)).get(machine.environmentId);
317 const unchanged =
318 entry !== undefined &&
319 connectionRoutes(entry).some(
320 (route) =>
321 Equal.equals(route.target, registration.target) &&
322 Option.isSome(route.profile) &&
323 Equal.equals(route.profile.value, registration.profile),
324 );
325 if (!unchanged) yield* registry.register(registration);
326 const current = (yield* SubscriptionRef.get(registry.entries)).get(machine.environmentId);
327 if (
328 current !== undefined &&
329 connectionRoutes(current).every((route) => connectionRouteId(route.target) === connectionId)
330 ) {
331 yield* registry.setCompatibility(
332 machine.environmentId,
333 orchestrationProtocolCompatibilityError(machine.descriptor),
334 );
335 }
336 }
337 yield* removeRoutes(profile.environmentId, keep);
338 });
339
340 const reportError = (error: { readonly message: string }) =>
341 SubscriptionRef.update(state, (current) => ({ ...current, error: error.message }));
342 const finish = SubscriptionRef.update(state, (current) => ({ ...current, refreshing: false }));
343 const mapPersistence = (error: { readonly message: string }) =>
344 new ConnectionBlockedError({ reason: "configuration", detail: error.message });
345
346 const refresh = lock.withPermits(1)(
347 Effect.gen(function* () {
348 const current = yield* SubscriptionRef.get(state);
349 if (current.connection === null) return;
350 yield* SubscriptionRef.update(state, (value) => ({
351 ...value,
352 refreshing: true,
353 error: null,
354 }));
355 const credential = yield* currentCredential(current.connection);
356 const directory = yield* requestDirectory(current.connection, credential);
357 yield* reconcile(current.connection, directory).pipe(Effect.mapError(mapPersistence));
358 const inbox = yield* requestInbox(current.connection, credential);
359 yield* SubscriptionRef.update(state, (value) => ({ ...value, directory, inbox }));
360 }).pipe(Effect.tapError(reportError), Effect.ensuring(finish)),
361 );
362
363 const connect = (input: PairingConnectionInput) =>
364 lock.withPermits(1)(
365 Effect.gen(function* () {
366 yield* SubscriptionRef.update(state, (value) => ({
367 ...value,
368 refreshing: true,
369 error: null,
370 }));
371 const registration = yield* preparePairingRegistration(input).pipe(
372 Effect.provideService(ClientCapabilities.ClientPresentation, presentation),
373 Effect.provideService(HttpClient.HttpClient, http),
374 Effect.mapError((error) =>
375 error._tag === "ConnectionBlockedError" && error.reason === "authentication"
376 ? new ConnectionBlockedError({
377 reason: "authentication",
378 detail:
379 "The pairing code is invalid or expired. Create a new code on the server.",
380 })
381 : error,
382 ),
383 );
384 const profile = registration.profile;
385 const credential = registration.credential;
386 if (credential._tag !== "BearerConnectionCredential")
387 return yield* new ConnectionBlockedError({
388 reason: "configuration",
389 detail: "The pairing credential is invalid.",
390 });
391 const directory = yield* requestDirectory(profile, credential);
392 inboxCache = undefined;
393 const inbox = yield* requestInbox(profile, credential);
394 const previous = (yield* SubscriptionRef.get(state)).connection;
395 yield* credentials.put(hubCredentialId(profile.environmentId), credential);
396 yield* reconcile(profile, directory).pipe(Effect.mapError(mapPersistence));
397 yield* profiles.put(
398 new BearerConnectionProfile({ ...profile, connectionId: HUB_PROFILE_ID }),
399 );
400 yield* SubscriptionRef.set(state, {
401 connection: profile,
402 directory,
403 inbox,
404 refreshing: false,
405 error: null,
406 });
407 for (const machine of directory.environments) {
408 yield* registry.retryNow(machine.environmentId);
409 }
410 if (previous !== null && previous.environmentId !== profile.environmentId) {
411 yield* removeRoutes(previous.environmentId).pipe(Effect.mapError(mapPersistence));
412 yield* credentials.remove(hubCredentialId(previous.environmentId));
413 }
414 }).pipe(Effect.tapError(reportError), Effect.ensuring(finish)),
415 );
416
417 const connectAuthenticated = (input: { readonly httpBaseUrl: string }) =>
418 lock.withPermits(1)(
419 Effect.gen(function* () {
420 const descriptor = yield* fetchRemoteEnvironmentDescriptor(input).pipe(
421 Effect.provideService(HttpClient.HttpClient, http),
422 Effect.mapError(mapPersistence),
423 );
424 const credential = new HostedConnectionCredential({ serverUrl: input.httpBaseUrl });
425 const profile = new BearerConnectionProfile({
426 connectionId: HUB_PROFILE_ID,
427 authorization: "hosted",
428 environmentId: descriptor.environmentId,
429 label: descriptor.label,
430 httpBaseUrl: input.httpBaseUrl,
431 wsBaseUrl: deriveWsBaseUrl(input.httpBaseUrl),
432 });
433 const directory = yield* requestDirectory(profile, credential);
434 inboxCache = undefined;
435 const inbox = yield* requestInbox(profile, credential);
436 const previous = (yield* SubscriptionRef.get(state)).connection;
437 const credentialId = hubCredentialId(profile.environmentId);
438 const previousCredential = yield* credentials.get(credentialId);
439 let credentialWritten = false;
440 yield* Effect.gen(function* () {
441 yield* credentials.put(credentialId, credential);
442 credentialWritten = true;
443 yield* reconcile(profile, directory).pipe(Effect.mapError(mapPersistence));
444 yield* profiles.put(profile);
445 }).pipe(
446 Effect.tapError(() =>
447 !credentialWritten
448 ? Effect.void
449 : Option.isSome(previousCredential)
450 ? credentials.put(credentialId, previousCredential.value)
451 : credentials.remove(credentialId),
452 ),
453 );
454
455 yield* SubscriptionRef.set(state, {
456 connection: profile,
457 directory,
458 inbox,
459 refreshing: false,
460 error: null,
461 });
462 if (previous !== null && previous.environmentId !== profile.environmentId) {
463 yield* removeRoutes(previous.environmentId).pipe(Effect.mapError(mapPersistence));
464 yield* credentials.remove(hubCredentialId(previous.environmentId));
465 }
466 }).pipe(Effect.tapError(reportError), Effect.ensuring(finish)),
467 );
468
469 const disconnect = lock.withPermits(1)(
470 Effect.gen(function* () {
471 const current = (yield* SubscriptionRef.get(state)).connection;
472 if (current === null) return;
473 yield* removeRoutes(current.environmentId).pipe(Effect.mapError(mapPersistence));
474 yield* profiles.remove(HUB_PROFILE_ID);
475 yield* credentials.remove(hubCredentialId(current.environmentId));
476 inboxCache = undefined;
477 yield* SubscriptionRef.set(state, EMPTY_HUB_CONNECTION_STATE);
478 }).pipe(Effect.tapError(reportError)),
479 );
480
481 const refreshInbox = lock.withPermit(
482 Effect.gen(function* () {
483 const current = yield* SubscriptionRef.get(state);
484 if (current.connection === null) return;
485 const credential = yield* currentCredential(current.connection);
486 const inbox = yield* requestInbox(current.connection, credential);
487 yield* SubscriptionRef.update(state, (value) => ({ ...value, inbox, error: null }));
488 }).pipe(Effect.tapError(reportError)),
489 );
490
491 const organize = (input: HubInboxCommand) =>
492 lock.withPermit(
493 Effect.gen(function* () {
494 const current = yield* SubscriptionRef.get(state);
495 if (
496 current.connection === null ||
497 !current.directory?.environments.some(
498 (entry) => entry.environmentId === input.environmentId,
499 )
500 )
501 return Option.none();
502 const credential = yield* currentCredential(current.connection);
503 const encoded = yield* encodeInboxCommand(input);
504 const response = yield* http.post(
505 environmentEndpointUrl(current.connection.httpBaseUrl, "/api/hub/inbox"),
506 {
507 headers: yield* headersFor(credential),
508 body: HttpBody.jsonUnsafe(encoded),
509 },
510 );
511 if (response.status === 401 || response.status === 403) {
512 return yield* new ConnectionBlockedError({
513 reason: "authentication",
514 detail:
515 "This server connection cannot change chats. Pair again with access to change threads.",
516 });
517 }
518 if (response.status !== 200) {
519 return yield* new ConnectionTransientError({
520 reason: "remote-unavailable",
521 detail: "Couldn't save this chat change on the server. Try again.",
522 });
523 }
524 const entry = yield* response.json.pipe(Effect.flatMap(decodeInboxEnvironment));
525 if (entry.environmentId !== input.environmentId) {
526 return yield* new ConnectionBlockedError({
527 reason: "configuration",
528 detail: "The server returned a different machine. Refresh and try again.",
529 });
530 }
531 inboxCache = undefined;
532 yield* SubscriptionRef.update(state, (value) => ({
533 ...value,
534 inbox: {
535 environments: [
536 ...value.inbox.environments.filter(
537 (item) => item.environmentId !== entry.environmentId,
538 ),
539 entry,
540 ],
541 },
542 error: null,
543 }));
544 return Option.some({ sequence: entry.snapshot.snapshotSequence });
545 }).pipe(
546 Effect.timeout("15 seconds"),
547 Effect.mapError((error) =>
548 isConnectionAttemptError(error)
549 ? error
550 : new ConnectionTransientError({
551 reason: "remote-unavailable",
552 detail: "Couldn't save this chat change on the server. Try again.",
553 }),
554 ),
555 ),
556 );
557
558 const headers = Effect.gen(function* () {
559 const connection = (yield* SubscriptionRef.get(state)).connection;
560 if (connection === null)
561 return yield* new ConnectionBlockedError({
562 reason: "configuration",
563 detail: "Connect your server to open archived chats.",
564 });
565 return yield* headersFor(yield* currentCredential(connection));
566 });
567
568 return HubConnection.of({
569 state,
570 connect,
571 connectAuthenticated,
572 headers,
573 refresh,
574 refreshInbox,
575 organize,
576 disconnect,
577 });
578});
579
580export const layer = Layer.effect(HubConnection, make);