1import type { ServerSelfUpdateOutcome } from "@t3tools/contracts";
2import * as HostProcess from "@t3tools/shared/HostProcess";
3import * as Context from "effect/Context";
4import * as Cause from "effect/Cause";
5import * as Effect from "effect/Effect";
6import * as FileSystem from "effect/FileSystem";
7import * as Layer from "effect/Layer";
8import * as Option from "effect/Option";
9import * as Path from "effect/Path";
10import * as Schema from "effect/Schema";
11
12import packageJson from "../../package.json" with { type: "json" };
13import { pinnedRuntimePaths, pinnedRuntimeVersionsDir } from "./pinnedRuntime.ts";
14import {
15 decodeServiceLauncherContext,
16 decodeServiceLauncherParentMessage,
17 SERVICE_LAUNCHER_CONTEXT_ENV,
18 type ServiceLauncherChildMessage,
19 type ServiceLauncherParentMessage,
20 type SupportedLauncherProtocol,
21} from "./serviceProtocol.ts";
22
23export class ServiceLauncherClientError extends Schema.TaggedError<ServiceLauncherClientError>()(
24 "ServiceLauncherClientError",
25 {
26 operation: Schema.Literals([
27 "decode-context",
28 "version-mismatch",
29 "unmanaged",
30 "send",
31 "disconnect",
32 "timeout",
33 ]),
34 cause: Schema.optional(Schema.Defect()),
35 },
36) {
37 override get message(): string {
38 switch (this.operation) {
39 case "decode-context":
40 return "The service launcher supplied invalid startup context.";
41 case "version-mismatch":
42 return "The service launcher started a different t3 version.";
43 case "unmanaged":
44 return "This server is not managed by the launcher.";
45 case "send":
46 return "Could not send a request to the service launcher.";
47 case "disconnect":
48 return "The service launcher disconnected before acknowledging the request.";
49 case "timeout":
50 return "The service launcher did not respond within 30 seconds.";
51 }
52 }
53}
54
55export class ServiceLauncherRejectedError extends Schema.TaggedError<ServiceLauncherRejectedError>()(
56 "ServiceLauncherRejectedError",
57 {
58 targetVersion: Schema.String,
59 reason: Schema.String,
60 },
61) {
62 override get message(): string {
63 return this.reason;
64 }
65}
66
67interface ServiceLauncherProcess {
68 readonly connected: boolean;
69 readonly send: (
70 message: ServiceLauncherChildMessage,
71 callback?: (error: Error | null) => void,
72 ) => boolean;
73 readonly on: (
74 event: "message" | "disconnect",
75 listener: (...args: ReadonlyArray<unknown>) => void,
76 ) => void;
77 readonly off: (
78 event: "message" | "disconnect",
79 listener: (...args: ReadonlyArray<unknown>) => void,
80 ) => void;
81}
82
83export const ServiceLauncherHostProcess = Context.Reference<ServiceLauncherProcess>(
84 "t3/cloud/serviceLauncherHostProcess",
85 {
86 defaultValue: () => ({
87 connected: process.connected && process.send !== undefined,
88 send: (message, callback) => {
89 if (process.send === undefined) return false;
90 return callback === undefined ? process.send(message) : process.send(message, callback);
91 },
92 on: (event, listener) => {
93 process.on(event, listener);
94 },
95 off: (event, listener) => {
96 process.off(event, listener);
97 },
98 }),
99 },
100);
101
102export class ServiceLauncherClient extends Context.Service<
103 ServiceLauncherClient,
104 {
105 readonly managed: boolean;
106 /** The supervising launcher's protocol. A remote update never replaces the launcher,
107 so it can be older than this runtime's own. */
108 readonly protocol: SupportedLauncherProtocol | undefined;
109 readonly requestUpdate: (input: {
110 readonly targetVersion: string;
111 readonly dbPath: string;
112 }) => Effect.Effect<string, ServiceLauncherClientError | ServiceLauncherRejectedError>;
113 /** Owned `t3` symlinks move to the new runtime only when this start commits the trial. */
114 readonly prepareTrial: (input: {
115 readonly baseDir: string;
116 }) => Effect.Effect<ServerSelfUpdateOutcome | undefined, ServiceLauncherClientError>;
117 }
118>()("t3/cloud/serviceLauncherClient") {}
119
120const resolveStartup = Effect.fn("cloud.service_launcher_client.resolve_startup")(
121 function* (options?: { readonly currentVersion?: string }) {
122 const host = yield* ServiceLauncherHostProcess;
123 const environment = yield* HostProcess.Environment;
124 const currentVersion = options?.currentVersion ?? packageJson.version;
125 // IPC channels are not inherited; without one, launcher context belongs to an ancestor.
126 const rawContext = host.connected ? environment[SERVICE_LAUNCHER_CONTEXT_ENV] : undefined;
127 const context = rawContext === undefined ? undefined : decodeServiceLauncherContext(rawContext);
128
129 if (rawContext !== undefined && context === undefined) {
130 return yield* new ServiceLauncherClientError({ operation: "decode-context" });
131 }
132 if (context !== undefined && context.childVersion !== currentVersion) {
133 return yield* new ServiceLauncherClientError({ operation: "version-mismatch" });
134 }
135
136 const managed = context !== undefined;
137
138 return { host, context, managed };
139 },
140);
141
142export const resolveServiceLauncherMode = Effect.fn("cloud.service_launcher_client.resolve_mode")(
143 function* () {
144 const { managed } = yield* resolveStartup();
145 return { managed };
146 },
147);
148
149export const repointLaunchers = Effect.fn("cloud.service_launcher_client.repoint_launchers")(
150 function* (input: { readonly baseDir: string; readonly targetVersion: string }) {
151 const fs = yield* FileSystem.FileSystem;
152 const path = yield* Path.Path;
153 const environment = yield* HostProcess.Environment;
154 const platform = yield* HostProcess.Platform;
155 if (platform !== "linux" && platform !== "darwin") return;
156 const installBinDir =
157 environment["T3CODE_INSTALL_BIN_DIR"] ||
158 (environment["HOME"] ? path.join(environment["HOME"], ".local", "bin") : undefined);
159 const candidates = new Set<string>();
160 if (installBinDir !== undefined) candidates.add(path.join(installBinDir, "t3"));
161 for (const directory of (environment["PATH"] ?? "").split(":")) {
162 if (directory.length === 0) continue;
163 const candidate = path.join(directory, "t3");
164 if (
165 yield* fs.exists(candidate).pipe(
166 Effect.catch((cause) =>
167 Effect.logWarning("Could not check the t3 command on PATH", {
168 launchedAs: candidate,
169 cause,
170 }).pipe(Effect.as(false)),
171 ),
172 )
173 ) {
174 candidates.add(candidate);
175 break;
176 }
177 }
178 const versionsDir = pinnedRuntimeVersionsDir(path, input.baseDir);
179 const target = pinnedRuntimePaths(path, input.baseDir, input.targetVersion, platform);
180 for (const launchedAs of candidates) {
181 yield* Effect.scoped(
182 Effect.gen(function* () {
183 const linkTarget = yield* fs.readLink(launchedAs).pipe(
184 Effect.asSome,
185 Effect.catchIf(
186 (error) =>
187 error.reason._tag === "NotFound" ||
188 (error.reason.cause as NodeJS.ErrnoException | undefined)?.code === "EINVAL",
189 () => Effect.succeedNone,
190 ),
191 );
192 if (Option.isNone(linkTarget)) return;
193 const relative = path.relative(
194 versionsDir,
195 path.resolve(path.dirname(launchedAs), linkTarget.value),
196 );
197 if (
198 relative.length === 0 ||
199 relative === ".." ||
200 relative.startsWith(`..${path.sep}`) ||
201 path.isAbsolute(relative)
202 )
203 return;
204 const temporary = yield* fs.makeTempDirectoryScoped({
205 directory: path.dirname(launchedAs),
206 prefix: ".t3-launcher-",
207 });
208 const link = path.join(temporary, "t3");
209 yield* fs.symlink(target.entryPath, link);
210 yield* fs.rename(link, launchedAs);
211 }),
212 ).pipe(
213 Effect.catchCause((cause) =>
214 Cause.hasInterrupts(cause)
215 ? Effect.interrupt
216 : Effect.logWarning("Could not move the t3 command to the updated version", {
217 launchedAs,
218 cause,
219 }),
220 ),
221 );
222 }
223 },
224);
225
226/** Completes when the launcher asks this server to stop, which Windows cannot signal. */
227export const awaitServiceLauncherStop = Effect.fn("cloud.service_launcher_client.await_stop")(
228 function* (options?: { readonly currentVersion?: string }) {
229 const startup = yield* resolveStartup(options).pipe(Effect.option);
230 if (Option.isNone(startup) || !startup.value.managed) return yield* Effect.never;
231 const { host } = startup.value;
232 return yield* Effect.callback<void>((resume) => {
233 const onMessage = (...args: ReadonlyArray<unknown>) => {
234 if (decodeServiceLauncherParentMessage(args[0])?.type === "stop") resume(Effect.void);
235 };
236 host.on("message", onMessage);
237 return Effect.sync(() => host.off("message", onMessage));
238 });
239 },
240);
241
242export const make = Effect.fn("cloud.service_launcher_client.make")(function* (options?: {
243 readonly currentVersion?: string;
244}) {
245 const { host, context, managed } = yield* resolveStartup(options);
246 // prepareTrial runs later, outside whatever context built the client.
247 const fs = yield* FileSystem.FileSystem;
248 const path = yield* Path.Path;
249 const environment = yield* HostProcess.Environment;
250 const platform = yield* HostProcess.Platform;
251
252 const exchange = (
253 message: ServiceLauncherChildMessage,
254 accept: (reply: ServiceLauncherParentMessage) => boolean,
255 ) =>
256 Effect.callback<ServiceLauncherParentMessage, ServiceLauncherClientError>((resume) => {
257 if (!managed) {
258 resume(Effect.fail(new ServiceLauncherClientError({ operation: "unmanaged" })));
259 return;
260 }
261
262 let settled = false;
263 const cleanup = () => {
264 host.off("message", onMessage);
265 host.off("disconnect", onDisconnect);
266 };
267 const settle = (
268 effect: Effect.Effect<ServiceLauncherParentMessage, ServiceLauncherClientError>,
269 ) => {
270 if (settled) return;
271 settled = true;
272 cleanup();
273 resume(effect);
274 };
275 const onMessage = (...args: ReadonlyArray<unknown>) => {
276 const reply = decodeServiceLauncherParentMessage(args[0]);
277 if (reply !== undefined && accept(reply)) settle(Effect.succeed(reply));
278 };
279 const onDisconnect = () =>
280 settle(Effect.fail(new ServiceLauncherClientError({ operation: "disconnect" })));
281
282 host.on("message", onMessage);
283 host.on("disconnect", onDisconnect);
284 try {
285 host.send(message, (error) => {
286 if (error !== null) {
287 settle(
288 Effect.fail(new ServiceLauncherClientError({ operation: "send", cause: error })),
289 );
290 }
291 });
292 } catch (cause) {
293 settle(Effect.fail(new ServiceLauncherClientError({ operation: "send", cause })));
294 }
295
296 return Effect.sync(cleanup);
297 }).pipe(
298 Effect.timeoutOrElse({
299 duration: "30 seconds",
300 orElse: () => Effect.fail(new ServiceLauncherClientError({ operation: "timeout" })),
301 }),
302 );
303
304 const requestUpdate = (input: { readonly targetVersion: string; readonly dbPath: string }) =>
305 exchange(
306 { type: "request-update", ...input },
307 (reply) => reply.type === "update-accepted" || reply.type === "update-rejected",
308 ).pipe(
309 Effect.flatMap((reply) =>
310 reply.type === "update-accepted"
311 ? Effect.succeed(reply.updateId)
312 : reply.type === "update-rejected"
313 ? Effect.fail(
314 new ServiceLauncherRejectedError({
315 targetVersion: input.targetVersion,
316 reason: reply.reason,
317 }),
318 )
319 : Effect.die("service launcher returned an impossible update response"),
320 ),
321 );
322
323 const pending = context?.update?.status === "pending" ? context.update : undefined;
324 const outcome =
325 context?.update === undefined || context.update.status === "pending"
326 ? undefined
327 : context.update;
328 const prepareTrial = (input: { readonly baseDir: string }) =>
329 pending !== undefined
330 ? exchange(
331 { type: "prepared", updateId: pending.id },
332 (reply) => reply.type === "committed" && reply.updateId === pending.id,
333 ).pipe(
334 Effect.flatMap((reply) =>
335 reply.type !== "committed"
336 ? Effect.die("service launcher returned an impossible prepared response")
337 : repointLaunchers({
338 baseDir: input.baseDir,
339 targetVersion: pending.targetVersion,
340 }).pipe(
341 Effect.provideService(FileSystem.FileSystem, fs),
342 Effect.provideService(Path.Path, path),
343 Effect.provideService(HostProcess.Environment, environment),
344 Effect.provideService(HostProcess.Platform, platform),
345 Effect.as({
346 id: pending.id,
347 fromVersion: pending.fromVersion,
348 targetVersion: pending.targetVersion,
349 status: "committed" as const,
350 }),
351 ),
352 ),
353 )
354 : Effect.succeed(outcome);
355
356 return ServiceLauncherClient.of({
357 managed,
358 protocol: context?.protocol,
359 requestUpdate,
360 prepareTrial,
361 });
362});
363
364export const layer = Layer.effect(ServiceLauncherClient, make());