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