import { defaultInstanceIdForDriver, ProviderDriverKind, ServerProviderUpdateError, type ProviderInstanceId, type ServerProvider, type ServerProviderUpdatedPayload, type ServerProviderUpdateState, } from "@t3tools/contracts"; import * as Cause from "effect/Cause"; import * as Context from "effect/Context"; import * as Data from "effect/Data"; import * as DateTime from "effect/DateTime"; import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Ref from "effect/Ref"; import * as Schema from "effect/Schema"; import { HttpClient } from "effect/unstable/http"; import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"; import { ProviderRegistry } from "./Services/ProviderRegistry.ts"; import { makeProviderMaintenanceCommandCoordinator } from "./providerMaintenanceCommandCoordinator.ts"; import { enrichProviderSnapshotWithVersionAdvisory } from "./providerMaintenance.ts"; import type { ProviderMaintenanceCapabilities } from "./providerMaintenance.ts"; import { collectUint8StreamText } from "../stream/collectUint8StreamText.ts"; const isServerProviderUpdateError = Schema.is(ServerProviderUpdateError); const UPDATE_TIMEOUT_MS = 5 * 60_000; const UPDATE_OUTPUT_MAX_BYTES = 10_000; export interface ProviderMaintenanceCommandResult { readonly stdout: string; readonly stderr: string; readonly exitCode: number | null; readonly timedOut: boolean; readonly stdoutTruncated: boolean; readonly stderrTruncated: boolean; } export interface ProviderMaintenanceRunnerShape { readonly updateProvider: ( target: | ProviderDriverKind | { readonly provider: ProviderDriverKind; readonly instanceId?: ProviderInstanceId | undefined; }, ) => Effect.Effect; } export class ProviderMaintenanceRunner extends Context.Service< ProviderMaintenanceRunner, ProviderMaintenanceRunnerShape >()("t3/provider/providerMaintenanceRunner") {} class ProviderMaintenanceCommandError extends Data.TaggedError("ProviderMaintenanceCommandError")<{ readonly message: string; readonly cause?: unknown; }> {} interface VerifiedProviderRefresh { readonly providers: ReadonlyArray; readonly verifiedProviders: ReadonlyArray; } const nowIso = Effect.map(DateTime.now, DateTime.formatIso); const runProviderMaintenanceCommandWithSpawner = Effect.fn("ProviderMaintenanceRunner.runCommand")( function* (input: { readonly spawner: ChildProcessSpawner.ChildProcessSpawner["Service"]; readonly command: string; readonly args: ReadonlyArray; }) { const collectCommandResult = Effect.fn("ProviderMaintenanceRunner.collectCommandResult")( function* () { const child = yield* input.spawner .spawn(ChildProcess.make(input.command, [...input.args])) .pipe( Effect.mapError( (cause) => new ProviderMaintenanceCommandError({ message: `Failed to run update command ${input.command}: ${cause.message}`, cause, }), ), ); yield* Effect.addFinalizer(() => child.kill().pipe(Effect.ignore)); const [stdout, stderr, exitCode] = yield* Effect.all( [ collectUint8StreamText({ stream: child.stdout, maxBytes: UPDATE_OUTPUT_MAX_BYTES, }), collectUint8StreamText({ stream: child.stderr, maxBytes: UPDATE_OUTPUT_MAX_BYTES, }), child.exitCode, ], { concurrency: "unbounded" }, ).pipe( Effect.mapError( (cause) => new ProviderMaintenanceCommandError({ message: cause instanceof Error ? cause.message : "Update command failed to run.", cause, }), ), ); return { stdout: stdout.text, stderr: stderr.text, exitCode: Number(exitCode), timedOut: false, stdoutTruncated: stdout.truncated, stderrTruncated: stderr.truncated, } satisfies ProviderMaintenanceCommandResult; }, ); return yield* collectCommandResult().pipe( Effect.scoped, Effect.timeoutOption(Duration.millis(UPDATE_TIMEOUT_MS)), Effect.map((result) => Option.match(result, { onSome: (value) => value, onNone: () => ({ stdout: "", stderr: "", exitCode: null, timedOut: true, stdoutTruncated: false, stderrTruncated: false, }) satisfies ProviderMaintenanceCommandResult, }), ), ); }, ); function trimNullable(value: string): string | null { const trimmed = value.trim(); return trimmed.length > 0 ? trimmed : null; } function truncateText(value: string, maxLength: number): string { return value.length <= maxLength ? value : value.slice(0, maxLength); } function commandOutput(result: ProviderMaintenanceCommandResult): string | null { const output = trimNullable([result.stderr, result.stdout].filter(Boolean).join("\n\n")); if (!output) { return null; } return truncateText(output, UPDATE_OUTPUT_MAX_BYTES); } function failureMessage(result: ProviderMaintenanceCommandResult): string { if (result.timedOut) { return "Update timed out."; } if (result.exitCode !== null && result.exitCode !== 0) { return `Update command exited with code ${result.exitCode}.`; } return "Update command failed."; } function isOutdatedProvider(provider: ServerProvider | undefined): boolean { return provider?.versionAdvisory?.status === "behind_latest"; } function makeUpdateState(input: { readonly status: ServerProviderUpdateState["status"]; readonly startedAt: string | null; readonly finishedAt: string | null; readonly message: string | null; readonly output?: string | null; }): ServerProviderUpdateState { return { status: input.status, startedAt: input.startedAt, finishedAt: input.finishedAt, message: input.message, output: input.output ?? null, }; } export const make = Effect.fn("ProviderMaintenanceRunner.make")(function* () { const providerRegistry = yield* ProviderRegistry; const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; const httpClient = yield* HttpClient.HttpClient; const runMaintenanceCommand = (command: string, args: ReadonlyArray) => runProviderMaintenanceCommandWithSpawner({ spawner, command, args, }); const commandCoordinator = yield* makeProviderMaintenanceCommandCoordinator({ makeAlreadyRunningError: () => new ServerProviderUpdateError({ provider: ProviderDriverKind.make("unknown"), reason: "An update is already running for this provider.", }), }); const verifyRefreshedProvider = ( provider: ProviderDriverKind, maintenanceCapabilities: ProviderMaintenanceCapabilities, instanceId: ProviderInstanceId, ): Effect.Effect => providerRegistry.getProviders.pipe( Effect.map((providers) => { const instanceIds: Array = []; for (const candidate of providers) { if (candidate.driver === provider && candidate.instanceId === instanceId) { instanceIds.push(candidate.instanceId); } } return instanceIds; }), Effect.flatMap((instanceIds) => instanceIds.length === 0 ? providerRegistry.refreshInstance(instanceId) : Effect.forEach( instanceIds, (instanceId) => providerRegistry.refreshInstance(instanceId), { concurrency: "unbounded", discard: true, }, ).pipe(Effect.andThen(providerRegistry.getProviders)), ), Effect.flatMap((providers) => { const refreshedProviders = providers.filter( (candidate) => candidate.driver === provider && candidate.instanceId === instanceId, ); if (refreshedProviders.length === 0) { return Effect.succeed({ providers, verifiedProviders: [], }); } return Effect.forEach( refreshedProviders, (refreshedProvider) => enrichProviderSnapshotWithVersionAdvisory( refreshedProvider, maintenanceCapabilities, ).pipe(Effect.provideService(HttpClient.HttpClient, httpClient)), { concurrency: "unbounded", }, ).pipe( Effect.map( (verifiedProviders): VerifiedProviderRefresh => ({ providers, verifiedProviders, }), ), Effect.catchCause((cause) => Effect.logWarning("Provider post-update version verification failed", { provider, cause: Cause.pretty(cause), }).pipe( Effect.as({ providers, verifiedProviders: refreshedProviders, }), ), ), ); }), ); const updateProvider: ProviderMaintenanceRunnerShape["updateProvider"] = Effect.fn( "ProviderMaintenanceRunner.updateProvider", )(function* (target) { const provider = typeof target === "string" ? target : target.provider; const instanceId = typeof target === "string" ? defaultInstanceIdForDriver(provider) : (target.instanceId ?? defaultInstanceIdForDriver(provider)); const targetKey = `instance:${instanceId}`; const capabilities = yield* providerRegistry.getProviderMaintenanceCapabilitiesForInstance( instanceId, provider, ); const update = capabilities.update; if (!update) { return yield* new ServerProviderUpdateError({ provider, reason: "This provider does not support one-click updates.", }); } const setUpdateState = (state: ServerProviderUpdateState | null) => providerRegistry.setProviderMaintenanceActionState({ instanceId, action: "update", state, }); const setQueuedState = setUpdateState( makeUpdateState({ status: "queued", startedAt: null, finishedAt: null, message: "Waiting for another provider update to finish.", }), ).pipe(Effect.asVoid); const runProviderUpdate = Effect.fn("ProviderMaintenanceRunner.runProviderUpdate")( function* () { const finish = (state: ServerProviderUpdateState) => setUpdateState(state).pipe(Effect.map((providers) => ({ providers }))); const startedAtRef = yield* Ref.make(null); const runCommandAndVerify = Effect.fn("ProviderMaintenanceRunner.runCommandAndVerify")( function* () { const startedAt = yield* nowIso; yield* Ref.set(startedAtRef, startedAt); yield* setUpdateState( makeUpdateState({ status: "running", startedAt, finishedAt: null, message: "Updating provider.", }), ); const result = yield* runMaintenanceCommand(update.executable, update.args); const finishedAt = yield* nowIso; if (result.timedOut || result.exitCode !== 0) { return yield* finish( makeUpdateState({ status: "failed", startedAt, finishedAt, message: failureMessage(result), output: commandOutput(result), }), ); } const { verifiedProviders } = yield* verifyRefreshedProvider( provider, capabilities, instanceId, ); const couldNotVerify = verifiedProviders.length === 0; const stillOutdated = couldNotVerify || verifiedProviders.some((verifiedProvider) => isOutdatedProvider(verifiedProvider)); return yield* finish( makeUpdateState({ status: stillOutdated ? "unchanged" : "succeeded", startedAt, finishedAt, message: couldNotVerify ? "Update command completed, but T3 Code could not verify the provider version." : stillOutdated ? "Update command completed, but T3 Code still detects an outdated provider version." : "Provider updated.", output: commandOutput(result), }), ); }, ); const recordFailedUpdate = Effect.fn("ProviderMaintenanceRunner.recordFailedUpdate")( function* (cause: Cause.Cause) { const failure = Cause.squash(cause); const startedAt = yield* Ref.get(startedAtRef); return yield* finish( makeUpdateState({ status: "failed", startedAt, finishedAt: yield* nowIso, message: failure instanceof Error ? failure.message : "Update command failed.", output: null, }), ); }, ); return yield* runCommandAndVerify().pipe(Effect.catchCause(recordFailedUpdate)); }, ); return yield* commandCoordinator .withCommandLock({ targetKey, lockKey: update.lockKey, onQueued: setQueuedState, run: runProviderUpdate(), }) .pipe( Effect.mapError((error) => isServerProviderUpdateError(error) ? new ServerProviderUpdateError({ provider, reason: error.reason, }) : error, ), ); }); return ProviderMaintenanceRunner.of({ updateProvider, }); }); export const layer = Layer.effect(ProviderMaintenanceRunner, make());