1import { SERVER_VERSION } from "../version.ts";
2import type { ServerSelfUpdateOutcome } from "@t3tools/contracts";
3import * as HostProcess from "@t3tools/shared/HostProcess";
4import * as Context from "effect/Context";
5import * as Effect from "effect/Effect";
6import * as Layer from "effect/Layer";
7import * as Schema from "effect/Schema";
8
9import {
10 decodeServiceLauncherContext,
11 decodeServiceLauncherParentMessage,
12 SERVICE_LAUNCHER_CONTEXT_ENV,
13 type ServiceLauncherChildMessage,
14 type ServiceLauncherParentMessage,
15} from "./serviceProtocol.ts";
16
17export class ServiceLauncherClientError extends Schema.TaggedError<ServiceLauncherClientError>()(
18 "ServiceLauncherClientError",
19 {
20 operation: Schema.Literals([
21 "decode-context",
22 "version-mismatch",
23 "ipc-unavailable",
24 "unmanaged",
25 "send",
26 "disconnect",
27 "timeout",
28 ]),
29 cause: Schema.optional(Schema.Defect()),
30 },
31) {
32 override get message(): string {
33 switch (this.operation) {
34 case "decode-context":
35 return "The service launcher supplied invalid startup context.";
36 case "version-mismatch":
37 return "The service launcher started a different t3 version.";
38 case "ipc-unavailable":
39 return "The service launcher IPC channel is unavailable.";
40 case "unmanaged":
41 return "This server is not managed by the launcher.";
42 case "send":
43 return "Could not send a request to the service launcher.";
44 case "disconnect":
45 return "The service launcher disconnected before acknowledging the request.";
46 case "timeout":
47 return "The service launcher did not respond within 30 seconds.";
48 }
49 }
50}
51
52export class ServiceLauncherRejectedError extends Schema.TaggedError<ServiceLauncherRejectedError>()(
53 "ServiceLauncherRejectedError",
54 {
55 targetVersion: Schema.String,
56 reason: Schema.String,
57 },
58) {
59 override get message(): string {
60 return this.reason;
61 }
62}
63
64interface ServiceLauncherProcess {
65 readonly connected: boolean;
66 readonly send: (
67 message: ServiceLauncherChildMessage,
68 callback?: (error: Error | null) => void,
69 ) => boolean;
70 readonly on: (
71 event: "message" | "disconnect",
72 listener: (...args: ReadonlyArray<unknown>) => void,
73 ) => void;
74 readonly off: (
75 event: "message" | "disconnect",
76 listener: (...args: ReadonlyArray<unknown>) => void,
77 ) => void;
78}
79
80export const ServiceLauncherHostProcess = Context.Reference<ServiceLauncherProcess>(
81 "t3/cloud/serviceLauncherHostProcess",
82 {
83 defaultValue: () => ({
84 connected: process.connected && process.send !== undefined,
85 send: (message, callback) => {
86 if (process.send === undefined) return false;
87 return callback === undefined ? process.send(message) : process.send(message, callback);
88 },
89 on: (event, listener) => {
90 process.on(event, listener);
91 },
92 off: (event, listener) => {
93 process.off(event, listener);
94 },
95 }),
96 },
97);
98
99export class ServiceLauncherClient extends Context.Service<
100 ServiceLauncherClient,
101 {
102 readonly managed: boolean;
103 readonly requestUpdate: (input: {
104 readonly targetVersion: string;
105 readonly dbPath: string;
106 }) => Effect.Effect<string, ServiceLauncherClientError | ServiceLauncherRejectedError>;
107 readonly prepareTrial: Effect.Effect<
108 ServerSelfUpdateOutcome | undefined,
109 ServiceLauncherClientError
110 >;
111 }
112>()("t3/cloud/serviceLauncherClient") {}
113
114const resolveStartup = Effect.fn("cloud.service_launcher_client.resolve_startup")(
115 function* (options?: { readonly currentVersion?: string }) {
116 const host = yield* ServiceLauncherHostProcess;
117 const environment = yield* HostProcess.Environment;
118 const currentVersion = options?.currentVersion ?? SERVER_VERSION;
119 const rawContext = environment[SERVICE_LAUNCHER_CONTEXT_ENV];
120 const context = rawContext === undefined ? undefined : decodeServiceLauncherContext(rawContext);
121
122 if (rawContext !== undefined && context === undefined) {
123 return yield* new ServiceLauncherClientError({ operation: "decode-context" });
124 }
125 if (context !== undefined && context.childVersion !== currentVersion) {
126 return yield* new ServiceLauncherClientError({ operation: "version-mismatch" });
127 }
128
129 const managed = context !== undefined && host.connected;
130 if (context !== undefined && !managed) {
131 return yield* new ServiceLauncherClientError({ operation: "ipc-unavailable" });
132 }
133
134 return { host, context, managed };
135 },
136);
137
138export const resolveServiceLauncherMode = Effect.fn("cloud.service_launcher_client.resolve_mode")(
139 function* () {
140 const { managed } = yield* resolveStartup();
141 return { managed };
142 },
143);
144
145export const make = Effect.fn("cloud.service_launcher_client.make")(function* (options?: {
146 readonly currentVersion?: string;
147}) {
148 const { host, context, managed } = yield* resolveStartup(options);
149
150 const exchange = (
151 message: ServiceLauncherChildMessage,
152 accept: (reply: ServiceLauncherParentMessage) => boolean,
153 ) =>
154 Effect.callback<ServiceLauncherParentMessage, ServiceLauncherClientError>((resume) => {
155 if (!managed) {
156 resume(Effect.fail(new ServiceLauncherClientError({ operation: "unmanaged" })));
157 return;
158 }
159
160 let settled = false;
161 const cleanup = () => {
162 host.off("message", onMessage);
163 host.off("disconnect", onDisconnect);
164 };
165 const settle = (
166 effect: Effect.Effect<ServiceLauncherParentMessage, ServiceLauncherClientError>,
167 ) => {
168 if (settled) return;
169 settled = true;
170 cleanup();
171 resume(effect);
172 };
173 const onMessage = (...args: ReadonlyArray<unknown>) => {
174 const reply = decodeServiceLauncherParentMessage(args[0]);
175 if (reply !== undefined && accept(reply)) settle(Effect.succeed(reply));
176 };
177 const onDisconnect = () =>
178 settle(Effect.fail(new ServiceLauncherClientError({ operation: "disconnect" })));
179
180 host.on("message", onMessage);
181 host.on("disconnect", onDisconnect);
182 try {
183 host.send(message, (error) => {
184 if (error !== null) {
185 settle(
186 Effect.fail(new ServiceLauncherClientError({ operation: "send", cause: error })),
187 );
188 }
189 });
190 } catch (cause) {
191 settle(Effect.fail(new ServiceLauncherClientError({ operation: "send", cause })));
192 }
193
194 return Effect.sync(cleanup);
195 }).pipe(
196 Effect.timeoutOrElse({
197 duration: "30 seconds",
198 orElse: () => Effect.fail(new ServiceLauncherClientError({ operation: "timeout" })),
199 }),
200 );
201
202 const requestUpdate = (input: { readonly targetVersion: string; readonly dbPath: string }) =>
203 exchange(
204 { type: "request-update", ...input },
205 (reply) => reply.type === "update-accepted" || reply.type === "update-rejected",
206 ).pipe(
207 Effect.flatMap((reply) =>
208 reply.type === "update-accepted"
209 ? Effect.succeed(reply.updateId)
210 : reply.type === "update-rejected"
211 ? Effect.fail(
212 new ServiceLauncherRejectedError({
213 targetVersion: input.targetVersion,
214 reason: reply.reason,
215 }),
216 )
217 : Effect.die("service launcher returned an impossible update response"),
218 ),
219 );
220
221 const pending = context?.update?.status === "pending" ? context.update : undefined;
222 const outcome =
223 context?.update === undefined || context.update.status === "pending"
224 ? undefined
225 : context.update;
226 const prepareTrial =
227 pending !== undefined
228 ? exchange(
229 { type: "prepared", updateId: pending.id },
230 (reply) => reply.type === "committed" && reply.updateId === pending.id,
231 ).pipe(
232 Effect.flatMap((reply) => {
233 if (reply.type !== "committed") {
234 return Effect.die("service launcher returned an impossible prepared response");
235 }
236 return Effect.succeed({
237 id: pending.id,
238 fromVersion: pending.fromVersion,
239 targetVersion: pending.targetVersion,
240 status: "committed" as const,
241 });
242 }),
243 )
244 : Effect.succeed(outcome);
245
246 return ServiceLauncherClient.of({
247 managed,
248 requestUpdate,
249 prepareTrial,
250 });
251});
252
253export const layer = Layer.effect(ServiceLauncherClient, make());