import type { ServerSelfUpdateOutcome } from "@t3tools/contracts";import { HostProcessEnvironment } from "@t3tools/shared/hostProcess";import * as Context from "effect/Context";import * as Effect from "effect/Effect";import * as Layer from "effect/Layer";import * as Schema from "effect/Schema"; import packageJson from "../../package.json" with { type: "json" };import { decodeServiceLauncherContext, decodeServiceLauncherParentMessage, SERVICE_LAUNCHER_CONTEXT_ENV, type ServiceLauncherChildMessage, type ServiceLauncherParentMessage,} from "./serviceProtocol.ts"; export class ServiceLauncherClientError extends Schema.TaggedError<ServiceLauncherClientError>()( "ServiceLauncherClientError", { operation: Schema.Literals([ "decode-context", "version-mismatch", "ipc-unavailable", "unmanaged", "send", "disconnect", "timeout", ]), cause: Schema.optional(Schema.Defect()), },) { override get message(): string { switch (this.operation) { case "decode-context": return "The service launcher supplied invalid startup context."; case "version-mismatch": return "The service launcher started a different t3 version."; case "ipc-unavailable": return "The service launcher IPC channel is unavailable."; case "unmanaged": return "This server is not managed by the launcher."; case "send": return "Could not send a request to the service launcher."; case "disconnect": return "The service launcher disconnected before acknowledging the request."; case "timeout": return "The service launcher did not respond within 30 seconds."; } }} export class ServiceLauncherRejectedError extends Schema.TaggedError<ServiceLauncherRejectedError>()( "ServiceLauncherRejectedError", { targetVersion: Schema.String, reason: Schema.String, },) { override get message(): string { return this.reason; }} interface ServiceLauncherProcess { readonly connected: boolean; readonly send: ( message: ServiceLauncherChildMessage, callback?: (error: Error | null) => void, ) => boolean; readonly on: ( event: "message" | "disconnect", listener: (...args: ReadonlyArray<unknown>) => void, ) => void; readonly off: ( event: "message" | "disconnect", listener: (...args: ReadonlyArray<unknown>) => void, ) => void;} export const ServiceLauncherHostProcess = Context.Reference<ServiceLauncherProcess>( "t3/cloud/serviceLauncherHostProcess", { defaultValue: () => ({ connected: process.connected && process.send !== undefined, send: (message, callback) => { if (process.send === undefined) return false; return callback === undefined ? process.send(message) : process.send(message, callback); }, on: (event, listener) => { process.on(event, listener); }, off: (event, listener) => { process.off(event, listener); }, }), },); export class ServiceLauncherClient extends Context.Service< ServiceLauncherClient, { readonly managed: boolean; readonly requestUpdate: (input: { readonly targetVersion: string; readonly dbPath: string; }) => Effect.Effect<string, ServiceLauncherClientError | ServiceLauncherRejectedError>; readonly prepareTrial: Effect.Effect< ServerSelfUpdateOutcome | undefined, ServiceLauncherClientError >; }>()("t3/cloud/serviceLauncherClient") {} const resolveStartup = Effect.fn("cloud.service_launcher_client.resolve_startup")( function* (options?: { readonly currentVersion?: string }) { const host = yield* ServiceLauncherHostProcess; const environment = yield* HostProcessEnvironment; const currentVersion = options?.currentVersion ?? packageJson.version; const rawContext = environment[SERVICE_LAUNCHER_CONTEXT_ENV]; const context = rawContext === undefined ? undefined : decodeServiceLauncherContext(rawContext); if (rawContext !== undefined && context === undefined) { return yield* new ServiceLauncherClientError({ operation: "decode-context" }); } if (context !== undefined && context.childVersion !== currentVersion) { return yield* new ServiceLauncherClientError({ operation: "version-mismatch" }); } const managed = context !== undefined && host.connected; if (context !== undefined && !managed) { return yield* new ServiceLauncherClientError({ operation: "ipc-unavailable" }); } return { host, context, managed }; },); export const resolveServiceLauncherMode = Effect.fn("cloud.service_launcher_client.resolve_mode")( function* () { const { managed } = yield* resolveStartup(); return { managed }; },); export const make = Effect.fn("cloud.service_launcher_client.make")(function* (options?: { readonly currentVersion?: string;}) { const { host, context, managed } = yield* resolveStartup(options); const exchange = ( message: ServiceLauncherChildMessage, accept: (reply: ServiceLauncherParentMessage) => boolean, ) => Effect.callback<ServiceLauncherParentMessage, ServiceLauncherClientError>((resume) => { if (!managed) { resume(Effect.fail(new ServiceLauncherClientError({ operation: "unmanaged" }))); return; } let settled = false; const cleanup = () => { host.off("message", onMessage); host.off("disconnect", onDisconnect); }; const settle = ( effect: Effect.Effect<ServiceLauncherParentMessage, ServiceLauncherClientError>, ) => { if (settled) return; settled = true; cleanup(); resume(effect); }; const onMessage = (...args: ReadonlyArray<unknown>) => { const reply = decodeServiceLauncherParentMessage(args[0]); if (reply !== undefined && accept(reply)) settle(Effect.succeed(reply)); }; const onDisconnect = () => settle(Effect.fail(new ServiceLauncherClientError({ operation: "disconnect" }))); host.on("message", onMessage); host.on("disconnect", onDisconnect); try { host.send(message, (error) => { if (error !== null) { settle( Effect.fail(new ServiceLauncherClientError({ operation: "send", cause: error })), ); } }); } catch (cause) { settle(Effect.fail(new ServiceLauncherClientError({ operation: "send", cause }))); } return Effect.sync(cleanup); }).pipe( Effect.timeoutOrElse({ duration: "30 seconds", orElse: () => Effect.fail(new ServiceLauncherClientError({ operation: "timeout" })), }), ); const requestUpdate = (input: { readonly targetVersion: string; readonly dbPath: string }) => exchange( { type: "request-update", ...input }, (reply) => reply.type === "update-accepted" || reply.type === "update-rejected", ).pipe( Effect.flatMap((reply) => reply.type === "update-accepted" ? Effect.succeed(reply.updateId) : reply.type === "update-rejected" ? Effect.fail( new ServiceLauncherRejectedError({ targetVersion: input.targetVersion, reason: reply.reason, }), ) : Effect.die("service launcher returned an impossible update response"), ), ); const pending = context?.update?.status === "pending" ? context.update : undefined; const outcome = context?.update === undefined || context.update.status === "pending" ? undefined : context.update; const prepareTrial = pending !== undefined ? exchange( { type: "prepared", updateId: pending.id }, (reply) => reply.type === "committed" && reply.updateId === pending.id, ).pipe( Effect.flatMap((reply) => { if (reply.type !== "committed") { return Effect.die("service launcher returned an impossible prepared response"); } return Effect.succeed({ id: pending.id, fromVersion: pending.fromVersion, targetVersion: pending.targetVersion, status: "committed" as const, }); }), ) : Effect.succeed(outcome); return ServiceLauncherClient.of({ managed, requestUpdate, prepareTrial, });}); export const layer = Layer.effect(ServiceLauncherClient, make());
import type { ServerSelfUpdateOutcome } from "@t3tools/contracts";
import { HostProcessEnvironment } from "@t3tools/shared/hostProcess";
import * as Context from "effect/Context";
import * as Effect from "effect/Effect";
import * as Layer from "effect/Layer";
import * as Schema from "effect/Schema";
import packageJson from "../../package.json" with { type: "json" };
import {
decodeServiceLauncherContext,
decodeServiceLauncherParentMessage,
SERVICE_LAUNCHER_CONTEXT_ENV,
type ServiceLauncherChildMessage,
type ServiceLauncherParentMessage,
} from "./serviceProtocol.ts";
export class ServiceLauncherClientError extends Schema.TaggedError<ServiceLauncherClientError>()(
"ServiceLauncherClientError",
{
operation: Schema.Literals([
"decode-context",
"version-mismatch",
"ipc-unavailable",
"unmanaged",
"send",
"disconnect",
"timeout",
]),
cause: Schema.optional(Schema.Defect()),
},
) {
override get message(): string {
switch (this.operation) {
case "decode-context":
return "The service launcher supplied invalid startup context.";
case "version-mismatch":
return "The service launcher started a different t3 version.";
case "ipc-unavailable":
return "The service launcher IPC channel is unavailable.";
case "unmanaged":
return "This server is not managed by the launcher.";
case "send":
return "Could not send a request to the service launcher.";
case "disconnect":
return "The service launcher disconnected before acknowledging the request.";
case "timeout":
return "The service launcher did not respond within 30 seconds.";
}
}
}
export class ServiceLauncherRejectedError extends Schema.TaggedError<ServiceLauncherRejectedError>()(
"ServiceLauncherRejectedError",
{
targetVersion: Schema.String,
reason: Schema.String,
},
) {
override get message(): string {
return this.reason;
}
}
interface ServiceLauncherProcess {
readonly connected: boolean;
readonly send: (
message: ServiceLauncherChildMessage,
callback?: (error: Error | null) => void,
) => boolean;
readonly on: (
event: "message" | "disconnect",
listener: (...args: ReadonlyArray<unknown>) => void,
) => void;
readonly off: (
event: "message" | "disconnect",
listener: (...args: ReadonlyArray<unknown>) => void,
) => void;
}
export const ServiceLauncherHostProcess = Context.Reference<ServiceLauncherProcess>(
"t3/cloud/serviceLauncherHostProcess",
{
defaultValue: () => ({
connected: process.connected && process.send !== undefined,
send: (message, callback) => {
if (process.send === undefined) return false;
return callback === undefined ? process.send(message) : process.send(message, callback);
},
on: (event, listener) => {
process.on(event, listener);
},
off: (event, listener) => {
process.off(event, listener);
},
}),
},
);
export class ServiceLauncherClient extends Context.Service<
ServiceLauncherClient,
{
readonly managed: boolean;
readonly requestUpdate: (input: {
readonly targetVersion: string;
readonly dbPath: string;
}) => Effect.Effect<string, ServiceLauncherClientError | ServiceLauncherRejectedError>;
readonly prepareTrial: Effect.Effect<
ServerSelfUpdateOutcome | undefined,
ServiceLauncherClientError
>;
}
>()("t3/cloud/serviceLauncherClient") {}
const resolveStartup = Effect.fn("cloud.service_launcher_client.resolve_startup")(
function* (options?: { readonly currentVersion?: string }) {
const host = yield* ServiceLauncherHostProcess;
const environment = yield* HostProcessEnvironment;
const currentVersion = options?.currentVersion ?? packageJson.version;
const rawContext = environment[SERVICE_LAUNCHER_CONTEXT_ENV];
const context = rawContext === undefined ? undefined : decodeServiceLauncherContext(rawContext);
if (rawContext !== undefined && context === undefined) {
return yield* new ServiceLauncherClientError({ operation: "decode-context" });
}
if (context !== undefined && context.childVersion !== currentVersion) {
return yield* new ServiceLauncherClientError({ operation: "version-mismatch" });
}
const managed = context !== undefined && host.connected;
if (context !== undefined && !managed) {
return yield* new ServiceLauncherClientError({ operation: "ipc-unavailable" });
}
return { host, context, managed };
},
);
export const resolveServiceLauncherMode = Effect.fn("cloud.service_launcher_client.resolve_mode")(
function* () {
const { managed } = yield* resolveStartup();
return { managed };
},
);
export const make = Effect.fn("cloud.service_launcher_client.make")(function* (options?: {
readonly currentVersion?: string;
}) {
const { host, context, managed } = yield* resolveStartup(options);
const exchange = (
message: ServiceLauncherChildMessage,
accept: (reply: ServiceLauncherParentMessage) => boolean,
) =>
Effect.callback<ServiceLauncherParentMessage, ServiceLauncherClientError>((resume) => {
if (!managed) {
resume(Effect.fail(new ServiceLauncherClientError({ operation: "unmanaged" })));
return;
}
let settled = false;
const cleanup = () => {
host.off("message", onMessage);
host.off("disconnect", onDisconnect);
};
const settle = (
effect: Effect.Effect<ServiceLauncherParentMessage, ServiceLauncherClientError>,
) => {
if (settled) return;
settled = true;
cleanup();
resume(effect);
};
const onMessage = (...args: ReadonlyArray<unknown>) => {
const reply = decodeServiceLauncherParentMessage(args[0]);
if (reply !== undefined && accept(reply)) settle(Effect.succeed(reply));
};
const onDisconnect = () =>
settle(Effect.fail(new ServiceLauncherClientError({ operation: "disconnect" })));
host.on("message", onMessage);
host.on("disconnect", onDisconnect);
try {
host.send(message, (error) => {
if (error !== null) {
settle(
Effect.fail(new ServiceLauncherClientError({ operation: "send", cause: error })),
);
}
});
} catch (cause) {
settle(Effect.fail(new ServiceLauncherClientError({ operation: "send", cause })));
}
return Effect.sync(cleanup);
}).pipe(
Effect.timeoutOrElse({
duration: "30 seconds",
orElse: () => Effect.fail(new ServiceLauncherClientError({ operation: "timeout" })),
}),
);
const requestUpdate = (input: { readonly targetVersion: string; readonly dbPath: string }) =>
exchange(
{ type: "request-update", ...input },
(reply) => reply.type === "update-accepted" || reply.type === "update-rejected",
).pipe(
Effect.flatMap((reply) =>
reply.type === "update-accepted"
? Effect.succeed(reply.updateId)
: reply.type === "update-rejected"
? Effect.fail(
new ServiceLauncherRejectedError({
targetVersion: input.targetVersion,
reason: reply.reason,
}),
)
: Effect.die("service launcher returned an impossible update response"),
),
);
const pending = context?.update?.status === "pending" ? context.update : undefined;
const outcome =
context?.update === undefined || context.update.status === "pending"
? undefined
: context.update;
const prepareTrial =
pending !== undefined
? exchange(
{ type: "prepared", updateId: pending.id },
(reply) => reply.type === "committed" && reply.updateId === pending.id,
).pipe(
Effect.flatMap((reply) => {
if (reply.type !== "committed") {
return Effect.die("service launcher returned an impossible prepared response");
}
return Effect.succeed({
id: pending.id,
fromVersion: pending.fromVersion,
targetVersion: pending.targetVersion,
status: "committed" as const,
});
}),
)
: Effect.succeed(outcome);
return ServiceLauncherClient.of({
managed,
requestUpdate,
prepareTrial,
});
});
export const layer = Layer.effect(ServiceLauncherClient, make());