1import * as Context from "effect/Context";
2import * as Effect from "effect/Effect";
3import * as Exit from "effect/Exit";
4import * as Fiber from "effect/Fiber";
5import * as Layer from "effect/Layer";
6import * as Scope from "effect/Scope";
7import * as HttpClient from "effect/http/HttpClient";
8
9import type { ConnectionCatalogEntry, ConnectionRoute } from "./catalog.ts";
10import type {
11 ConnectionAttemptError,
12 ConnectionAttemptStage,
13 PreparedConnection,
14} from "./model.ts";
15import { ConnectionTransientError } from "./model.ts";
16import * as ConnectionResolver from "./resolver.ts";
17import { connectionRoutes, routeEntry, routeHttpBaseUrl } from "./routes.ts";
18import * as RpcSession from "../rpc/session.ts";
19import { fetchRemoteEnvironmentDescriptor } from "../environment/descriptor.ts";
20
21export type ConnectionDriverProgress =
22 | {
23 readonly stage: "preparing";
24 }
25 | {
26 readonly stage: Exclude<ConnectionAttemptStage, "preparing">;
27 readonly prepared: PreparedConnection;
28 };
29
30export interface EnvironmentConnectionLease {
31 readonly prepared: PreparedConnection;
32 readonly session: RpcSession.RpcSession;
33}
34
35/**
36 * The result of an unauthenticated reachability check. T3 Connect and SSH
37 * routes have no cheap check, so they are "unchecked".
38 */
39export type RouteCheck = "answered" | "silent" | "unchecked";
40
41/** How long a direct route has to answer before it counts as unreachable from here. */
42const ROUTE_CHECK_TIMEOUT_MS = 2_500;
43
44export class ConnectionDriver extends Context.Service<
45 ConnectionDriver,
46 {
47 readonly connect: (
48 entry: ConnectionCatalogEntry,
49 reportProgress: (progress: ConnectionDriverProgress) => Effect.Effect<void>,
50 ) => Effect.Effect<EnvironmentConnectionLease, ConnectionAttemptError, Scope.Scope>;
51 /** Whether a direct route answers as the entry's environment, without credentials. */
52 readonly checkRoute: (
53 entry: ConnectionCatalogEntry,
54 route: ConnectionRoute,
55 ) => Effect.Effect<RouteCheck>;
56 /**
57 * Whether a direct route answers and accepts this client's credential,
58 * without opening a socket. Switching to a route that fails this would
59 * drop a working connection for nothing.
60 */
61 readonly preflight: (
62 entry: ConnectionCatalogEntry,
63 route: ConnectionRoute,
64 ) => Effect.Effect<boolean>;
65 }
66>()("@t3tools/client-runtime/connection/driver/ConnectionDriver") {}
67
68/**
69 * Connects over the first route, in preference order, that is worth trying.
70 * Every route is checked at once, but a route only waits for its own check,
71 * so a reachable LAN address connects without waiting on a silent tailnet
72 * one. A silent route is skipped on the first pass so a LAN address from
73 * another network costs one short check, not a connection timeout. A route
74 * that fails to connect moves on to the next: a signed-out T3 Connect must
75 * not hide a working LAN. Silent routes are tried last, since a check is not
76 * proof.
77 *
78 * The reported error is a transient one when any route failed transiently,
79 * so the supervisor keeps retrying a route that may come back; a blocked
80 * error is reported only when every attempted route was blocked.
81 */
82export const connectOverRoutes = Effect.fn("ConnectionDriver.connectOverRoutes")(function* <R>(
83 entry: ConnectionCatalogEntry,
84 checkRoute: (route: ConnectionRoute) => Effect.Effect<RouteCheck>,
85 connectRoute: (
86 route: ConnectionRoute,
87 ) => Effect.Effect<EnvironmentConnectionLease, ConnectionAttemptError, R | Scope.Scope>,
88) {
89 const routes = connectionRoutes(entry);
90 const checks =
91 routes.length === 1
92 ? []
93 : yield* Effect.forEach(routes, (route) => Effect.forkChild(checkRoute(route)));
94 const attemptScope = yield* Scope.Scope;
95 let transient: ConnectionAttemptError | undefined;
96 let blocked: ConnectionAttemptError | undefined;
97 // Each route gets its own scope so a half-open session closes before the next try.
98 const attempt = Effect.fnUntraced(function* (route: ConnectionRoute) {
99 const routeScope = yield* Scope.fork(attemptScope);
100 const result = yield* connectRoute(route).pipe(
101 Scope.provide(routeScope),
102 Effect.onExit((exit) => (Exit.isSuccess(exit) ? Effect.void : Scope.close(routeScope, exit))),
103 Effect.result,
104 );
105 if (result._tag === "Failure") {
106 if (result.failure._tag === "ConnectionTransientError") transient ??= result.failure;
107 else blocked ??= result.failure;
108 }
109 return result;
110 });
111 const silent: Array<ConnectionRoute> = [];
112 for (const [index, route] of routes.entries()) {
113 const check = checks.length === 0 ? "unchecked" : yield* Fiber.join(checks[index]!);
114 if (check === "silent") {
115 silent.push(route);
116 continue;
117 }
118 const result = yield* attempt(route);
119 if (result._tag === "Success") return result.success;
120 // An incompatible server is the same server on every route.
121 if (result.failure.reason === "unsupported") return yield* result.failure;
122 }
123 for (const route of silent) {
124 const result = yield* attempt(route);
125 if (result._tag === "Success") return result.success;
126 if (result.failure.reason === "unsupported") return yield* result.failure;
127 }
128 return yield* (
129 transient ??
130 blocked ??
131 new ConnectionTransientError({
132 reason: "endpoint-unavailable",
133 detail: `${entry.target.label} did not answer on any saved route.`,
134 })
135 );
136});
137
138/** @public Service construction is part of the canonical Effect module API. */
139export const make = Effect.gen(function* () {
140 const resolver = yield* ConnectionResolver.ConnectionResolver;
141 const sessions = yield* RpcSession.RpcSessionFactory;
142 const httpClient = yield* HttpClient.HttpClient;
143
144 const checkRoute = (entry: ConnectionCatalogEntry, route: ConnectionRoute) => {
145 const httpBaseUrl = routeHttpBaseUrl(route);
146 if (httpBaseUrl === null) return Effect.succeed<RouteCheck>("unchecked");
147 // The descriptor is public, so this sends no credential to whatever
148 // answers at a saved LAN address on a different network.
149 return fetchRemoteEnvironmentDescriptor({
150 httpBaseUrl,
151 timeoutMs: ROUTE_CHECK_TIMEOUT_MS,
152 }).pipe(
153 Effect.map((descriptor): RouteCheck =>
154 descriptor.environmentId === entry.target.environmentId ? "answered" : "silent",
155 ),
156 Effect.orElseSucceed((): RouteCheck => "silent"),
157 Effect.provideService(HttpClient.HttpClient, httpClient),
158 Effect.withSpan("ConnectionDriver.checkRoute", {
159 attributes: { "connection.target.kind": route.target._tag },
160 }),
161 );
162 };
163
164 const connect = Effect.fn("ConnectionDriver.connect")(function* (
165 entry: ConnectionCatalogEntry,
166 reportProgress: (progress: ConnectionDriverProgress) => Effect.Effect<void>,
167 ) {
168 const target = entry.target;
169 yield* Effect.annotateCurrentSpan({
170 "connection.environment.id": target.environmentId,
171 "connection.target.kind": target._tag,
172 "connection.route.count": connectionRoutes(entry).length,
173 });
174 yield* reportProgress({ stage: "preparing" });
175 return yield* connectOverRoutes(
176 entry,
177 (route) => checkRoute(entry, route),
178 Effect.fnUntraced(function* (route) {
179 const prepared = yield* resolver.prepare(routeEntry(entry, route));
180 yield* reportProgress({ stage: "opening", prepared });
181 const session = yield* sessions.connect(
182 prepared,
183 resolver.prepare(routeEntry(entry, route)),
184 );
185 yield* reportProgress({ stage: "synchronizing", prepared });
186 yield* session.ready;
187 return { prepared, session } satisfies EnvironmentConnectionLease;
188 }),
189 );
190 });
191
192 const preflight = (entry: ConnectionCatalogEntry, route: ConnectionRoute) =>
193 checkRoute(entry, route).pipe(
194 Effect.flatMap((check) =>
195 check === "answered"
196 ? resolver.prepare(routeEntry(entry, route)).pipe(
197 Effect.as(true),
198 Effect.orElseSucceed(() => false),
199 )
200 : Effect.succeed(false),
201 ),
202 Effect.withSpan("ConnectionDriver.preflight"),
203 );
204
205 return ConnectionDriver.of({ connect, checkRoute, preflight });
206});
207
208export const layer = Layer.effect(ConnectionDriver, make);