1import {
2 AuthMcpClientAccess,
3 AuthMcpRegistrationError,
4 AuthMcpTokenError,
5 EnvironmentHttpApi,
6 EnvironmentId,
7 type ExecutionEnvironmentCapabilities,
8 MCP_PEER_LINK_SOFTWARE_ID,
9 type PeerLink,
10 type PeerLinkCreateInput,
11 PeerLinkError,
12 type PeerLinkSummary,
13} from "@t3tools/contracts";
14import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
15import * as Cause from "effect/Cause";
16import * as Context from "effect/Context";
17import * as Digest from "effect/crypto/Digest";
18import * as SecureRandom from "effect/crypto/SecureRandom";
19import * as DateTime from "effect/DateTime";
20import * as Duration from "effect/Duration";
21import * as Effect from "effect/Effect";
22import * as Base64Url from "effect/encoding/Base64Url";
23import * as Hex from "effect/encoding/Hex";
24import * as Layer from "effect/Layer";
25import * as Option from "effect/Option";
26import * as Predicate from "effect/Predicate";
27import * as Redacted from "effect/Redacted";
28import * as Result from "effect/Result";
29import * as Schedule from "effect/Schedule";
30import * as Schema from "effect/Schema";
31import * as SqlClient from "effect/sql/SqlClient";
32import * as FetchHttpClient from "effect/http/FetchHttpClient";
33import * as HttpClient from "effect/http/HttpClient";
34import * as HttpClientError from "effect/http/HttpClientError";
35import * as HttpApi from "effect/http-api/HttpApi";
36import * as HttpApiClient from "effect/http-api/HttpApiClient";
37import { isLocalLoopbackHost, isTailnetHost } from "@t3tools/shared/hostClassification";
38
39import * as ServerSecretStore from "../auth/ServerSecretStore.ts";
40import * as ServerEnvironment from "../environment/ServerEnvironment.ts";
41import * as ServerSettings from "../serverSettings.ts";
42
43const PeerHttpApi = HttpApi.make(EnvironmentHttpApi.identifier)
44 .add(EnvironmentHttpApi.groups.metadata)
45 .add(EnvironmentHttpApi.groups.mcpOAuth)
46 .annotateMerge(EnvironmentHttpApi.annotations);
47
48/** The redirect registered for a link. Nothing ever listens there: the code comes back in the response. */
49const LINK_REDIRECT_URI = "http://127.0.0.1/t3code-peer-link";
50/**
51 * How long a peer's descriptor or MCP handshake may take before a route counts
52 * as unreachable. A tailnet or relayed route to a machine that is waking up takes longer.
53 */
54export const probeTimeout = (origin: string) =>
55 isLocalLoopbackHost(new URL(origin).hostname) ? Duration.seconds(3) : Duration.seconds(10);
56/** How long a listing waits on a link's addresses; a real call still gives a waking peer `probeTimeout`. */
57const LIST_PROBE_TIMEOUT = Duration.seconds(3);
58
59/** Node's TLS verification code in a failed request's cause chain, if any. */
60export const certificateErrorCode = (error: unknown) => {
61 let current = error;
62 for (let depth = 0; depth < 6 && Predicate.isObject(current); depth++) {
63 if (
64 Predicate.hasProperty(current, "code") &&
65 typeof current.code === "string" &&
66 /CERT|SELF_SIGNED|UNABLE_TO_(GET|VERIFY)|INVALID_(CA|PURPOSE)|PATH_LENGTH_EXCEEDED|HOSTNAME_MISMATCH/.test(
67 current.code,
68 )
69 )
70 return current.code;
71 current = Predicate.hasProperty(current, "cause") ? current.cause : undefined;
72 }
73 return undefined;
74};
75
76/** A usable link: where to reach the peer and the token to send there. */
77export interface ResolvedPeerLink {
78 readonly link: PeerLink;
79 /** The first address that answered as this peer. */
80 readonly url: string;
81 readonly token: string;
82}
83
84export const LINKED_ENVIRONMENTS_OFF =
85 "Linked environments is off on this machine. Turn it on in Settings → Labs.";
86
87/**
88 * Links from this environment to others. Linking signs this environment in to
89 * the peer's `/mcp` as an outside agent with a pairing code from the peer, the
90 * same MCP OAuth flow an outside agent uses, without a browser. The peer's
91 * session token is kept in the server secret store; the table keeps the rest.
92 */
93export class PeerLinks extends Context.Service<
94 PeerLinks,
95 {
96 /** Whether the Linked environments beta (Settings → Labs) is on here. */
97 readonly enabled: Effect.Effect<boolean>;
98 readonly link: (input: PeerLinkCreateInput) => Effect.Effect<PeerLink, PeerLinkError>;
99 readonly list: Effect.Effect<ReadonlyArray<PeerLinkSummary>, PeerLinkError>;
100 /** The stored link, without asking the peer whether it answers. */
101 readonly get: (
102 environmentId: EnvironmentId,
103 ) => Effect.Effect<Option.Option<PeerLink>, PeerLinkError>;
104 /** Forgets the link here. The peer still lists the session until it is revoked there. */
105 readonly unlink: (environmentId: EnvironmentId) => Effect.Effect<boolean, PeerLinkError>;
106 /** An address that answers as the linked peer, with its token, ready to call. */
107 readonly resolve: (
108 environmentId: EnvironmentId,
109 ) => Effect.Effect<ResolvedPeerLink, PeerLinkError>;
110 /** Records the outcome of a call, so listing shows whether the peer answered last time. */
111 readonly recordOutcome: (
112 environmentId: EnvironmentId,
113 outcome: { readonly error: string | null },
114 ) => Effect.Effect<void>;
115 }
116>()("t3/peer/PeerLinks") {}
117
118interface PeerLinkRow {
119 readonly environment_id: string;
120 readonly secret_name: string;
121 readonly label: string;
122 readonly urls_json: string;
123 readonly access: string;
124 readonly linked_at: string;
125 readonly expires_at: string;
126 readonly last_reached_at: string | null;
127 readonly last_error: string | null;
128 readonly fenced: number;
129}
130
131const decodeUrls = Schema.decodeUnknownEffect(Schema.fromJsonString(Schema.Array(Schema.String)));
132const encodeUrls = Schema.encodeEffect(Schema.fromJsonString(Schema.Array(Schema.String)));
133const decodeAccess = Schema.decodeUnknownEffect(AuthMcpClientAccess);
134
135/** Fails with a fixed storage error that keeps its cause; the log gets the details. */
136const failStorage = (cause: unknown) =>
137 Effect.logWarning("Peer link storage failed", cause).pipe(
138 Effect.andThen(
139 Effect.fail(
140 new PeerLinkError({
141 reason: "storage",
142 message:
143 "Linked environments could not be read or saved here. The server log has details.",
144 cause,
145 }),
146 ),
147 ),
148 );
149
150/**
151 * The base URL a peer is reached at, normalised to its origin. The pairing
152 * code and then the link's token travel to it, so plain http is accepted only
153 * where the network already encrypts: this machine, or a Tailscale address.
154 */
155const peerOrigin = (url: string) => {
156 const parsed = URL.parse(url);
157 if (parsed === null || (parsed.protocol !== "https:" && parsed.protocol !== "http:")) {
158 return Effect.fail(
159 new PeerLinkError({
160 reason: "unreachable",
161 message: `${url} is not an http or https address.`,
162 }),
163 );
164 }
165 if (
166 parsed.protocol === "http:" &&
167 !isLocalLoopbackHost(parsed.hostname) &&
168 !isTailnetHost(parsed.hostname)
169 ) {
170 return Effect.fail(
171 new PeerLinkError({
172 reason: "unreachable",
173 message: `${url} is plain http, which would send the link's credentials unencrypted. Use its https address, or its Tailscale address.`,
174 }),
175 );
176 }
177 return Effect.succeed(parsed.origin);
178};
179
180/** The system error under a failed connection, such as ECONNREFUSED or ENOTFOUND. */
181const connectionCode = (cause: unknown) =>
182 [cause, Predicate.hasProperty(cause, "cause") ? cause.cause : undefined]
183 .map((error) => (Predicate.hasProperty(error, "code") ? error.code : undefined))
184 .find(Predicate.isString);
185
186/** Why `origin` did not answer as T3 Code, short enough for a toast. Never echoes a body. */
187const probeFailure = (
188 origin: string,
189 error: HttpClientError.HttpClientError | Schema.SchemaError | Cause.TimeoutError,
190 timeout: Duration.Duration,
191) => {
192 const unreachable = (message: string) => new PeerLinkError({ reason: "unreachable", message });
193 const certificate = certificateErrorCode(error);
194 if (certificate !== undefined) {
195 return unreachable(`${origin} presented a certificate this machine rejected (${certificate}).`);
196 }
197 if (Cause.isTimeoutError(error)) {
198 return unreachable(`${origin} did not answer within ${Duration.toSeconds(timeout)} seconds.`);
199 }
200 if (HttpClientError.isHttpClientError(error) && error.reason._tag === "TransportError") {
201 const code = connectionCode(error.reason.cause);
202 return unreachable(
203 code === undefined ? `${origin} did not answer.` : `${origin} did not answer (${code}).`,
204 );
205 }
206 return new PeerLinkError({
207 reason: "not_a_t3_environment",
208 message: `${origin} answered, but not as a T3 Code environment this version understands.`,
209 });
210};
211
212/** Whether a host is this machine's own loopback. Tests serve their peers there, so they narrow it. */
213export const LoopbackHost = Context.Reference<(host: string) => boolean>(
214 "t3/peer/PeerLinks/LoopbackHost",
215 { defaultValue: () => isLocalLoopbackHost },
216);
217
218/** Whether a peer holds a linked agent to its limits and keeps its work within its link. */
219const linkLimitsHold = (peer: { readonly capabilities: ExecutionEnvironmentCapabilities }) =>
220 peer.capabilities.mcpModeLimitHeader === true && peer.capabilities.linkFence === true;
221
222const make = Effect.gen(function* () {
223 const sql = yield* SqlClient.SqlClient;
224 const secrets = yield* ServerSecretStore.ServerSecretStore;
225 const random = yield* SecureRandom.SecureRandom;
226 const digest = yield* Digest.Digest;
227 const httpClient = yield* HttpClient.HttpClient;
228 const environment = yield* ServerEnvironment.ServerEnvironment;
229 const serverSettings = yield* ServerSettings.ServerSettingsService;
230 const isOwnLoopback = yield* LoopbackHost;
231 const enabled = serverSettings.getSettings.pipe(
232 Effect.map((settings) => settings.linkedEnvironmentsEnabled),
233 Effect.orElseSucceed(() => false),
234 );
235
236 /** A fresh name for one link's token; the row records it. */
237 const newSecretName = random.randomBytes(16).pipe(
238 Effect.map((bytes) => `peer-link-${Hex.encode(bytes)}`),
239 Effect.orDie,
240 );
241
242 // Only the origin `peerOrigin` checked may receive a pairing code or token:
243 // a redirect could forward them anywhere, including plain http.
244 const peerHttpClient = HttpClient.transform(httpClient, (response) =>
245 Effect.provideService(response, FetchHttpClient.RequestInit, { redirect: "manual" }),
246 );
247
248 const clientFor = (origin: string) =>
249 HttpApiClient.make(PeerHttpApi, { baseUrl: origin }).pipe(
250 Effect.provideService(HttpClient.HttpClient, peerHttpClient),
251 );
252
253 /** The environment answering at `origin`, or why nothing there answers as T3 Code. */
254 const probe = (origin: string) => {
255 const timeout = probeTimeout(origin);
256 return clientFor(origin).pipe(
257 Effect.flatMap((client) => client.metadata.descriptor()),
258 // A brief network drop must not list a live peer as down; the timeout still bounds both tries.
259 Effect.retry({
260 times: 1,
261 schedule: Schedule.spaced("1 second"),
262 while: (error) => certificateErrorCode(error) === undefined,
263 }),
264 Effect.timeout(timeout),
265 // No cause: a decode error can carry the response body to the client.
266 Effect.mapError((error) => probeFailure(origin, error, timeout)),
267 );
268 };
269
270 /** Whether `url` answers as `environmentId`, counting any failure as no. */
271 const answersAs = Effect.fn("PeerLinks.answersAs")(function* (
272 url: string,
273 environmentId: EnvironmentId,
274 ) {
275 const found = yield* probe(url).pipe(Effect.option);
276 return Option.filter(found, (peer) => peer.environmentId === environmentId);
277 });
278
279 const toLink = Effect.fn("PeerLinks.toLink")(function* (row: PeerLinkRow) {
280 return {
281 environmentId: EnvironmentId.make(row.environment_id),
282 label: row.label,
283 urls: yield* decodeUrls(row.urls_json),
284 access: yield* decodeAccess(row.access),
285 linkedAt: DateTime.makeUnsafe(row.linked_at),
286 // Its token predates link fencing, so the peer would not fence its work:
287 // like an expired link, only a new pairing code brings it back.
288 expiresAt: DateTime.makeUnsafe(row.fenced === 1 ? row.expires_at : row.linked_at),
289 lastReachedAt: row.last_reached_at === null ? null : DateTime.makeUnsafe(row.last_reached_at),
290 lastError: row.last_error,
291 } satisfies PeerLink;
292 }, Effect.catch(failStorage));
293
294 const readRows = sql<PeerLinkRow>`
295 SELECT * FROM peer_environment_links ORDER BY label, environment_id
296 `.pipe(NodeSqliteClient.readOnly, Effect.catch(failStorage));
297
298 const readRow = (environmentId: EnvironmentId) =>
299 sql<PeerLinkRow>`
300 SELECT * FROM peer_environment_links WHERE environment_id = ${environmentId}
301 `.pipe(
302 NodeSqliteClient.readOnly,
303 Effect.catch(failStorage),
304 Effect.map((rows) => Option.fromNullishOr(rows[0])),
305 );
306
307 /** Signs in to the peer's `/mcp` with a pairing code: register, decide, exchange. */
308 const signIn = Effect.fn("PeerLinks.signIn")(
309 function* (origin: string, input: PeerLinkCreateInput, clientName: string) {
310 const client = yield* clientFor(origin);
311 // Only the peer's own refusals are rejections. A dropped connection or
312 // an answer this version cannot read is classified below.
313 const registered = yield* client.mcpOAuth
314 .register({
315 payload: {
316 client_name: clientName,
317 redirect_uris: [LINK_REDIRECT_URI],
318 software_id: MCP_PEER_LINK_SOFTWARE_ID,
319 },
320 })
321 .pipe(
322 // A plain Schema.Error has no tag to catch by.
323 Effect.catchIf(Schema.is(AuthMcpRegistrationError), (error) =>
324 Effect.fail(
325 new PeerLinkError({
326 reason: "pairing_rejected",
327 message: `The peer refused to register this environment: ${error.message}`,
328 }),
329 ),
330 ),
331 );
332 const verifier = Base64Url.encode(yield* random.randomBytes(32));
333 const challenge = Base64Url.encode(
334 yield* digest.digest(Digest.sha256, new TextEncoder().encode(verifier)).pipe(Effect.orDie),
335 );
336 const state = Base64Url.encode(yield* random.randomBytes(16));
337 const resource = `${origin}/mcp`;
338 const decided = yield* client.mcpOAuth
339 .decision({
340 headers: {},
341 payload: {
342 authorization: {
343 response_type: "code",
344 client_id: registered.client_id,
345 redirect_uri: LINK_REDIRECT_URI,
346 code_challenge: challenge,
347 code_challenge_method: "S256",
348 state: Redacted.make(state),
349 resource,
350 },
351 decision: { _tag: "pairing-code", access: input.access, code: input.pairingCode },
352 },
353 })
354 .pipe(
355 Effect.catchTags({
356 AuthMcpApprovalError: (error) =>
357 Effect.fail(
358 new PeerLinkError({ reason: "pairing_rejected", message: error.message }),
359 ),
360 }),
361 );
362 const redirect = URL.parse(decided.redirectTo);
363 if (redirect === null) {
364 return yield* new PeerLinkError({
365 reason: "incompatible",
366 message: `${origin} answered the sign-in with an address this version cannot read. Update both environments.`,
367 });
368 }
369 const code = redirect.searchParams.get("code");
370 if (code === null || redirect.searchParams.get("state") !== state) {
371 return yield* new PeerLinkError({
372 reason: "pairing_rejected",
373 message:
374 redirect.searchParams.get("error_description") ??
375 "The peer did not approve the link. Check the pairing code and try again.",
376 });
377 }
378 return yield* client.mcpOAuth
379 .token({
380 payload: {
381 grant_type: "authorization_code",
382 code: Redacted.make(code),
383 redirect_uri: LINK_REDIRECT_URI,
384 client_id: registered.client_id,
385 code_verifier: Redacted.make(verifier),
386 resource,
387 },
388 })
389 .pipe(
390 Effect.catchIf(Schema.is(AuthMcpTokenError), (error) =>
391 Effect.fail(
392 new PeerLinkError({
393 reason: "pairing_rejected",
394 message: `The peer refused the sign-in: ${error.message}`,
395 }),
396 ),
397 ),
398 );
399 },
400 (effect, origin) =>
401 effect.pipe(
402 // Endpoint errors are mapped above; what is left is the transport or an
403 // answer this version cannot decode.
404 Effect.catchTags({
405 SchemaError: (cause) =>
406 Effect.fail(
407 new PeerLinkError({
408 reason: "incompatible",
409 message: `${origin} answered in a shape this version does not understand. Update both environments.`,
410 cause,
411 }),
412 ),
413 HttpClientError: (cause) =>
414 Effect.fail(
415 new PeerLinkError({
416 reason: "unreachable",
417 message: `${origin} stopped answering while linking.`,
418 cause,
419 }),
420 ),
421 }),
422 ),
423 );
424
425 /**
426 * The addresses that answer as `environmentId`, in the order given, with
427 * what the first one answered. Addresses this environment would not send
428 * credentials to are skipped; when none answers as it, the failure says why
429 * for each.
430 */
431 const answeringAs = Effect.fn("PeerLinks.answeringAs")(function* (
432 urls: ReadonlyArray<string>,
433 environmentId: EnvironmentId,
434 name: string,
435 ) {
436 const checked = yield* Effect.forEach(
437 urls,
438 (url) =>
439 Effect.gen(function* () {
440 const origin = yield* peerOrigin(url);
441 // Loopback is this machine, where anything may answer as the one picked.
442 if (isOwnLoopback(new URL(origin).hostname)) {
443 return yield* new PeerLinkError({
444 reason: "unreachable",
445 message: `${origin} is this machine, not ${name}.`,
446 });
447 }
448 const descriptor = yield* probe(origin);
449 if (descriptor.environmentId !== environmentId) {
450 return yield* new PeerLinkError({
451 reason: "unreachable",
452 message: `${origin} answered as another environment, ${descriptor.label}.`,
453 });
454 }
455 return { origin, descriptor };
456 }).pipe(Effect.result),
457 { concurrency: "unbounded" },
458 );
459 const matching = checked.flatMap((result) =>
460 Result.isSuccess(result) ? [result.success] : [],
461 );
462 const first = matching[0];
463 if (first === undefined) {
464 const reasons = checked.flatMap((result) =>
465 Result.isFailure(result) ? [result.failure.message] : [],
466 );
467 return yield* new PeerLinkError({
468 reason: "unreachable",
469 message: `This environment could not reach ${name}: ${reasons.join(" ")}`,
470 });
471 }
472 // Each address is kept once, even when two spellings share an origin.
473 const origins = [...new Set(matching.map((found) => found.origin))];
474 return { primary: first.origin, peer: first.descriptor, origins };
475 });
476
477 /** The peer at `url`, with every address checked for credentials first. */
478 const answeringAt = Effect.fn("PeerLinks.answeringAt")(function* (urls: ReadonlyArray<string>) {
479 const origins = yield* Effect.forEach(urls, peerOrigin);
480 const primary = origins[0]!;
481 return { primary, peer: yield* probe(primary), origins };
482 });
483
484 const link: PeerLinks["Service"]["link"] = Effect.fn("PeerLinks.link")(function* (input) {
485 if (!(yield* enabled)) {
486 return yield* new PeerLinkError({ reason: "disabled", message: LINKED_ENVIRONMENTS_OFF });
487 }
488 const urls = [input.url, ...(input.alternateUrls ?? [])];
489 // The user picked a machine: link through the addresses that answer as
490 // it, so one that now answers as another never receives the pairing code.
491 const { primary, peer, origins } =
492 input.expectedEnvironmentId === undefined
493 ? yield* answeringAt(urls)
494 : yield* answeringAs(
495 urls,
496 input.expectedEnvironmentId,
497 input.expectedLabel ?? input.expectedEnvironmentId,
498 );
499 // Anything can answer as a linked machine, so an address alone renews its
500 // link only where the link already reaches it, and never moves it.
501 if (input.expectedEnvironmentId === undefined) {
502 const row = yield* readRow(peer.environmentId);
503 const existing = Option.isSome(row) ? yield* toLink(row.value) : undefined;
504 if (
505 existing !== undefined &&
506 DateTime.isGreaterThan(existing.expiresAt, yield* DateTime.now) &&
507 !existing.urls.includes(primary)
508 ) {
509 return yield* new PeerLinkError({
510 reason: "pairing_rejected",
511 message: `${primary} answers as ${existing.label}, which is already linked at another address. If ${existing.label} moved, forget that link first.`,
512 });
513 }
514 }
515 const self = yield* environment.getDescriptor;
516 if (peer.environmentId === self.environmentId) {
517 return yield* new PeerLinkError({
518 reason: "self",
519 message: "That address is this environment.",
520 });
521 }
522 if (!linkLimitsHold(peer)) {
523 return yield* new PeerLinkError({
524 reason: "incompatible",
525 message: `${peer.label} runs a T3 Code version that cannot keep a linked agent's limits. Update it, then link again.`,
526 });
527 }
528 const issued = yield* signIn(primary, input, `T3 Code · ${self.label}`);
529 const now = yield* DateTime.now;
530 const linked: PeerLink = {
531 environmentId: peer.environmentId,
532 label: peer.label,
533 urls: origins,
534 access: input.access,
535 linkedAt: now,
536 expiresAt: DateTime.add(now, { seconds: issued.expires_in }),
537 lastReachedAt: now,
538 lastError: null,
539 };
540 // The token goes under a name of its own before the row points at it,
541 // so the row and its token are replaced together.
542 const secretName = yield* newSecretName;
543 yield* secrets
544 .set(secretName, new TextEncoder().encode(issued.access_token))
545 .pipe(Effect.catch(failStorage));
546 // A token no row names could never be removed.
547 const previous = yield* Effect.gen(function* () {
548 const urlsJson = yield* encodeUrls(origins).pipe(Effect.catch(failStorage));
549 const previous = yield* readRow(peer.environmentId);
550 yield* sql`
551 INSERT INTO peer_environment_links (
552 environment_id, secret_name, label, urls_json, access, linked_at, expires_at,
553 last_reached_at, last_error, fenced
554 ) VALUES (
555 ${linked.environmentId}, ${secretName}, ${linked.label}, ${urlsJson}, ${linked.access},
556 ${DateTime.formatIso(linked.linkedAt)}, ${DateTime.formatIso(linked.expiresAt)},
557 ${DateTime.formatIso(now)}, NULL, 1
558 )
559 ON CONFLICT (environment_id) DO UPDATE SET
560 secret_name = excluded.secret_name,
561 fenced = excluded.fenced,
562 label = excluded.label,
563 urls_json = excluded.urls_json,
564 access = excluded.access,
565 linked_at = excluded.linked_at,
566 expires_at = excluded.expires_at,
567 last_reached_at = excluded.last_reached_at,
568 last_error = NULL
569 `.pipe(Effect.catch(failStorage));
570 return previous;
571 }).pipe(Effect.onError(() => secrets.remove(secretName).pipe(Effect.ignore)));
572 // Linking again replaces the old token. A racing link may leave its own
573 // behind; it is unused either way.
574 if (Option.isSome(previous) && previous.value.secret_name !== secretName) {
575 yield* secrets.remove(previous.value.secret_name).pipe(Effect.ignore);
576 }
577 return linked;
578 });
579
580 /** A peer's label follows its machine name, so the link takes the one its own address answers with. */
581 const withLabel = (link: PeerLink, label: string) =>
582 label === link.label
583 ? Effect.succeed(link)
584 : sql`
585 UPDATE peer_environment_links SET label = ${label}
586 WHERE environment_id = ${link.environmentId}
587 `.pipe(Effect.ignoreCause({ log: true }), Effect.as({ ...link, label }));
588
589 const statusOf = Effect.fn("PeerLinks.statusOf")(function* (
590 link: PeerLink,
591 now: DateTime.Utc,
592 ): Effect.fn.Return<PeerLinkSummary> {
593 if (DateTime.isLessThanOrEqualTo(link.expiresAt, now)) return { ...link, status: "expired" };
594 // A race over no addresses never settles.
595 if (link.urls.length === 0) return { ...link, status: "unreachable" };
596 const answered = yield* Effect.raceAll(
597 link.urls.map((url) =>
598 answersAs(url, link.environmentId).pipe(Effect.flatMap(Effect.fromOption)),
599 ),
600 ).pipe(Effect.timeoutOption(LIST_PROBE_TIMEOUT), Effect.orElseSucceed(Option.none));
601 if (Option.isNone(answered)) return { ...link, status: "unreachable" };
602 return { ...(yield* withLabel(link, answered.value.label)), status: "reachable" };
603 });
604
605 const list: PeerLinks["Service"]["list"] = Effect.gen(function* () {
606 const links = yield* Effect.forEach(yield* readRows, toLink);
607 const now = yield* DateTime.now;
608 return yield* Effect.forEach(links, (link) => statusOf(link, now), {
609 concurrency: "unbounded",
610 });
611 });
612
613 const get: PeerLinks["Service"]["get"] = (environmentId) =>
614 readRow(environmentId).pipe(
615 Effect.flatMap(
616 Option.match({
617 onNone: () => Effect.succeedNone,
618 onSome: (row) => Effect.asSome(toLink(row)),
619 }),
620 ),
621 );
622
623 const unlink: PeerLinks["Service"]["unlink"] = Effect.fn("PeerLinks.unlink")(
624 function* (environmentId) {
625 const deleted = yield* sql<{ secret_name: string }>`
626 DELETE FROM peer_environment_links WHERE environment_id = ${environmentId}
627 RETURNING secret_name
628 `.pipe(Effect.catch(failStorage));
629 for (const row of deleted) {
630 yield* secrets.remove(row.secret_name).pipe(Effect.catch(failStorage));
631 }
632 return deleted.length > 0;
633 },
634 );
635
636 const resolve: PeerLinks["Service"]["resolve"] = Effect.fn("PeerLinks.resolve")(
637 function* (environmentId) {
638 const row = yield* readRow(environmentId);
639 if (Option.isNone(row)) {
640 return yield* new PeerLinkError({
641 reason: "unknown_link",
642 message: `This environment is not linked to ${environmentId}. t3_environment_links lists the linked ones.`,
643 });
644 }
645 const link = yield* toLink(row.value);
646 if (DateTime.isLessThanOrEqualTo(link.expiresAt, yield* DateTime.now)) {
647 return yield* new PeerLinkError({
648 reason: "expired",
649 message: `The link to ${link.label} expired. Link it again with a new pairing code from ${link.label}.`,
650 });
651 }
652 const stored = yield* secrets.get(row.value.secret_name).pipe(Effect.catch(failStorage));
653 if (Option.isNone(stored)) {
654 return yield* new PeerLinkError({
655 reason: "unknown_link",
656 message: `The link to ${link.label} lost its credential. Link it again.`,
657 });
658 }
659 // The token goes only to an address that still answers as this peer,
660 // so one that now answers as another machine never receives it.
661 let fenced = true;
662 let label = link.label;
663 const url = yield* Effect.findFirst(link.urls, (candidate) =>
664 answersAs(candidate, environmentId).pipe(
665 Effect.map((found) => {
666 if (Option.isNone(found)) return false;
667 fenced = linkLimitsHold(found.value);
668 label = found.value.label;
669 return true;
670 }),
671 ),
672 );
673 if (Option.isNone(url)) {
674 return yield* new PeerLinkError({
675 reason: "unreachable",
676 message: `${link.label} did not answer at ${link.urls.join(", ")}.`,
677 });
678 }
679 // A link made before the peer had to fence linked work, or a peer moved
680 // back to such a version, would carry work past those limits.
681 if (!fenced) {
682 return yield* new PeerLinkError({
683 reason: "incompatible",
684 message: `${link.label} runs a T3 Code version that cannot keep a linked agent's limits. Update it, then link again.`,
685 });
686 }
687 return {
688 link: yield* withLabel(link, label),
689 url: url.value,
690 token: new TextDecoder().decode(stored.value),
691 };
692 },
693 );
694
695 const recordOutcome: PeerLinks["Service"]["recordOutcome"] = (environmentId, outcome) =>
696 DateTime.now.pipe(
697 Effect.flatMap((now) =>
698 outcome.error === null
699 ? sql`
700 UPDATE peer_environment_links
701 SET last_reached_at = ${DateTime.formatIso(now)}, last_error = NULL
702 WHERE environment_id = ${environmentId}
703 `
704 : sql`
705 UPDATE peer_environment_links SET last_error = ${outcome.error}
706 WHERE environment_id = ${environmentId}
707 `,
708 ),
709 Effect.asVoid,
710 Effect.ignoreCause({ log: true }),
711 );
712
713 return PeerLinks.of({
714 enabled,
715 link,
716 list,
717 get,
718 unlink,
719 resolve,
720 recordOutcome,
721 });
722});
723
724export const layer = Layer.effect(PeerLinks, make);