homeio
PublicForked from doctor-io/homeio
operations.ts
import "server-only";
import { randomUUID } from "node:crypto";
import { copyFile, readFile, rm, writeFile } from "node:fs/promises";
import path from "node:path";
import {
cleanupComposeDataOnUninstall,
getComposeRuntimeInfo,
materializeInlineStackFiles,
runComposeDown,
runComposePull,
runComposeRestart,
runComposeStart,
runComposeStop,
runComposeUp,
sanitizeStackName,
} from "@/lib/server/modules/docker/compose-runner";
import { invalidateInstalledAppsCache } from "@/lib/server/modules/apps/service";
import { findStoreCatalogTemplateByAppId } from "@/lib/server/modules/store/catalog";
import {
findCustomStoreTemplateByAppId,
isCustomStoreTemplate,
} from "@/lib/server/modules/store/custom-apps";
import {
extractPrimaryServiceWithName,
parseComposeFile,
} from "@/lib/server/modules/docker/compose-parser";
import {
createStoreOperation,
deleteInstalledStackByAppId,
findInstalledStackByAppId,
findStackByWebUiPort,
findStoreOperationById,
markStaleOperationsAsError,
pruneFinishedOperations,
updateStackUpdateStatus,
updateStoreOperation,
upsertInstalledStack,
} from "@/lib/server/modules/apps/stacks-repository";
import { resolveInstalledComposePath } from "@/lib/server/modules/apps/installed-compose-path";
import { logServerAction } from "@/lib/server/logging/logger";
import { serverEnv } from "@/lib/server/env";
import type {
InstalledStackConfig,
StoreOperation,
StoreOperationAction,
StoreOperationEvent,
StoreOperationStatus,
} from "@/lib/shared/contracts/apps";
import { isTagUpdateAvailable } from "@/lib/server/modules/store/update-check";
import { createNotification } from "@/lib/server/modules/notifications/service";
import yaml from "js-yaml";
type OperationParams = {
appId: string;
action: StoreOperationAction;
displayName?: string;
env?: Record<string, string>;
webUiPort?: number;
composeSource?: string;
removeVolumes?: boolean;
resetToCatalog?: boolean;
};
type OperationSubscriber = (event: StoreOperationEvent) => void;
type ComposeDocument = {
services?: Record<string, Record<string, unknown>>;
[key: string]: unknown;
};
const operationSubscribers = new Map<string, Set<OperationSubscriber>>();
const latestOperationEvent = new Map<string, StoreOperationEvent>();
const IS_TEST_ENV = process.env.NODE_ENV === "test" || process.env.VITEST === "true";
/**
* How long after a terminal event we keep the cached payload in
* `latestOperationEvent` so SSE clients that connect right after
* completion still receive it without a DB round-trip. After this
* window the entry is dropped to keep memory bounded on long-lived
* servers (especially Pi). The timeout is unref'd so it never blocks
* graceful shutdown.
*/
const FINISHED_EVENT_TTL_MS = IS_TEST_ENV ? 1_000 : 60_000;
/**
* In-memory guard for concurrent operations on the same app.
* Prevents two simultaneous requests from launching duplicate operations.
* Cleared on operation completion (success or error).
*/
const activeOperationsByApp = new Set<string>();
export class StoreOperationError extends Error {
constructor(
message: string,
public readonly code: "operation_conflict" | "operation_limit_reached",
public readonly statusCode: number,
) {
super(message);
this.name = "StoreOperationError";
}
}
/** Exposed only for unit tests — clears the in-memory guard between test cases. */
export function _resetActiveOperationsForTesting() {
activeOperationsByApp.clear();
}
// On startup, mark any DB operations that were left in "queued"/"running" state
// (from a previous crash or hard restart) as errors so the UI doesn't show them
// as stuck forever.
if (!IS_TEST_ENV) {
void markStaleOperationsAsError().catch(() => undefined);
}
const DEPLOYMENT_HEALTH_TIMEOUT_MS = IS_TEST_ENV ? 25 : 25_000;
const DEPLOYMENT_HEALTH_POLL_MS = IS_TEST_ENV ? 1 : 2_000;
const WEB_UI_PROBE_TIMEOUT_MS = IS_TEST_ENV ? 25 : 2_000;
const WEB_UI_PROBE_ATTEMPTS = 3;
const WEB_UI_PROBE_DELAY_MS = IS_TEST_ENV ? 1 : 750;
function clampPercent(value: number) {
return Math.max(0, Math.min(100, Math.round(value)));
}
// Runs `docker compose pull` and advances progress incrementally while it runs.
// Progress goes from startPercent up to (endPercent - 2), then the caller emits endPercent.
async function pullImagesWithProgress(
composeInput: { composePath: string; envPath: string; stackName: string },
opts: {
startPercent: number;
endPercent: number;
onProgress: (percent: number) => Promise<void>;
},
) {
let currentPercent = opts.startPercent;
let done = false;
const cap = opts.endPercent - 2;
const interval = setInterval(() => {
if (done || currentPercent >= cap) return;
currentPercent = Math.min(cap, currentPercent + 1);
void opts.onProgress(currentPercent);
}, IS_TEST_ENV ? 1 : 2_000);
// unref so an in-flight pull doesn't keep the Node.js event loop
// alive during graceful shutdown (SIGTERM). clearInterval still runs
// in the finally block below.
interval.unref?.();
try {
await runComposePull(composeInput);
} finally {
done = true;
clearInterval(interval);
}
}
function assertValidPort(port: number) {
if (!Number.isInteger(port)) {
throw new Error("webUiPort must be an integer");
}
if (port < 1024 || port > 65535) {
throw new Error("webUiPort must be between 1024 and 65535");
}
}
function mergeEnv(
allowedKeys: string[],
defaults: Record<string, string>,
existing: Record<string, string>,
overrides: Record<string, string>,
) {
const allowed = new Set(allowedKeys);
const unknownKeys = Object.keys(overrides).filter((key) => !allowed.has(key));
if (unknownKeys.length > 0) {
throw new Error(
`Unsupported env key(s): ${unknownKeys.join(", ")}. Only template-defined env keys are allowed.`,
);
}
const merged: Record<string, string> = {
...defaults,
...existing,
};
for (const [key, value] of Object.entries(overrides)) {
merged[key] = String(value);
}
return merged;
}
function withComposeInterpolationEnv(env: Record<string, string>, appId: string) {
return {
AppID: appId,
...env,
};
}
function parseFirstPublishedPort(ports: string[] | undefined): number | null {
if (!ports || ports.length === 0) return null;
for (const mapping of ports) {
const [mappingPart] = mapping.split("/");
const parts = mappingPart.split(":").filter(Boolean);
const published = parts.length >= 2 ? parts[parts.length - 2] : parts[0];
if (!published) continue;
const parsed = Number.parseInt(published, 10);
if (Number.isInteger(parsed)) return parsed;
}
return null;
}
function parseComposeDocument(content: string, sourceLabel: string): ComposeDocument {
let parsed: unknown;
try {
parsed = yaml.load(content);
} catch {
throw new Error(`Invalid compose source (${sourceLabel})`);
}
if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) {
throw new Error(`Invalid compose source (${sourceLabel})`);
}
return parsed as ComposeDocument;
}
function resolveComposeServices(
document: ComposeDocument,
sourceLabel: string,
): Record<string, Record<string, unknown>> {
const services = document.services;
if (!services || typeof services !== "object" || Array.isArray(services)) {
throw new Error(`Invalid compose source (${sourceLabel})`);
}
const normalized: Record<string, Record<string, unknown>> = {};
for (const [name, value] of Object.entries(services)) {
if (!value || typeof value !== "object" || Array.isArray(value)) {
throw new Error(`Invalid compose source (${sourceLabel})`);
}
normalized[name] = value;
}
if (Object.keys(normalized).length === 0) {
throw new Error(`Invalid compose source (${sourceLabel})`);
}
return normalized;
}
function haveEqualServiceSet(
left: Record<string, Record<string, unknown>>,
right: Record<string, Record<string, unknown>>,
) {
const leftNames = Object.keys(left).sort((a, b) => a.localeCompare(b));
const rightNames = Object.keys(right).sort((a, b) => a.localeCompare(b));
if (leftNames.length !== rightNames.length) return false;
return leftNames.every((name, index) => name === rightNames[index]);
}
async function resolveStoreComposeSource(appId: string) {
const template =
(await findStoreCatalogTemplateByAppId(appId)) ??
(await findCustomStoreTemplateByAppId(appId));
if (!template) {
throw new Error(`Template not found for appId "${appId}"`);
}
if (isCustomStoreTemplate(template)) {
return template.composeContent;
}
return readFile(template.composePath, "utf8");
}
function sleep(ms: number) {
return new Promise((resolve) => {
setTimeout(resolve, ms);
});
}
function buildLocalWebUiUrl(input: {
webUiPort: number | null;
scheme?: string | null;
index?: string | null;
}) {
if (!Number.isInteger(input.webUiPort) || input.webUiPort === null) {
return null;
}
const scheme = typeof input.scheme === "string" && input.scheme.trim().length > 0
? input.scheme.trim()
: "http";
const index = typeof input.index === "string" && input.index.trim().length > 0
? input.index.trim()
: "/";
try {
return new URL(index, `${scheme}://127.0.0.1:${input.webUiPort}`).toString();
} catch {
return null;
}
}
async function probeWebUiReachability(url: string) {
for (let attempt = 0; attempt < WEB_UI_PROBE_ATTEMPTS; attempt += 1) {
try {
const response = await fetch(url, {
signal: AbortSignal.timeout(WEB_UI_PROBE_TIMEOUT_MS),
cache: "no-store",
});
if (response.status >= 200 && response.status < 500) {
return true;
}
} catch {
// Retry within the bounded health-check window.
}
if (attempt < WEB_UI_PROBE_ATTEMPTS - 1) {
await sleep(WEB_UI_PROBE_DELAY_MS);
}
}
return false;
}
async function waitForDeploymentHealth(input: {
operationId: string;
appId: string;
action: StoreOperationAction;
composePath: string;
envPath: string;
stackName: string;
webUiUrl?: string | null;
}) {
const deadline = Date.now() + DEPLOYMENT_HEALTH_TIMEOUT_MS;
let lastLifecycleStatus: "running" | "paused" | "restarting" | "stopped" | "unknown" = "unknown";
let lastMessage: string | null = null;
while (Date.now() < deadline) {
const runtime = await getComposeRuntimeInfo({
composePath: input.composePath,
envPath: input.envPath,
stackName: input.stackName,
});
lastLifecycleStatus = runtime.lifecycleStatus;
const statusMessage =
runtime.lifecycleStatus === "restarting"
? "Containers are restarting"
: runtime.lifecycleStatus === "paused"
? "Containers are paused"
: runtime.lifecycleStatus === "unknown"
? "Waiting for docker compose runtime status"
: "Waiting for containers to become healthy";
if (runtime.status === "running") {
if (!input.webUiUrl) {
return {
webUiReachable: null,
};
}
const webUiReachable = await probeWebUiReachability(input.webUiUrl);
if (!webUiReachable) {
await patchOperationAndEmit({
operationId: input.operationId,
appId: input.appId,
action: input.action,
status: "running",
eventType: "operation.step",
progressPercent: 98,
step: "health-check",
message: "Web UI not reachable yet",
});
}
return {
webUiReachable,
};
}
if (statusMessage !== lastMessage) {
await patchOperationAndEmit({
operationId: input.operationId,
appId: input.appId,
action: input.action,
status: "running",
eventType: "operation.step",
progressPercent: 96,
step: "health-check",
message: statusMessage,
});
lastMessage = statusMessage;
}
await sleep(DEPLOYMENT_HEALTH_POLL_MS);
}
if (lastLifecycleStatus === "restarting") {
throw new Error("Deployment failed because containers are restarting");
}
if (lastLifecycleStatus === "paused") {
throw new Error("Deployment failed because containers are paused");
}
if (lastLifecycleStatus === "stopped") {
throw new Error("Deployment failed because containers never reached a running state");
}
throw new Error("Deployment failed because runtime status could not be confirmed");
}
function emitEvent(event: StoreOperationEvent) {
latestOperationEvent.set(event.operationId, event);
const subscribers = operationSubscribers.get(event.operationId);
if (!subscribers || subscribers.size === 0) return;
for (const subscriber of subscribers) {
subscriber(event);
}
}
async function patchOperationAndEmit(input: {
operationId: string;
appId: string;
action: StoreOperationAction;
status: StoreOperationStatus;
eventType: StoreOperationEvent["type"];
progressPercent: number;
step: string;
message?: string;
image?: string;
dockerStatus?: string;
progressDetail?: StoreOperationEvent["progressDetail"];
errorMessage?: string | null;
markStarted?: boolean;
markFinished?: boolean;
}) {
const progressPercent = clampPercent(input.progressPercent);
await updateStoreOperation(input.operationId, {
status: input.status,
progressPercent,
currentStep: input.step,
errorMessage: input.errorMessage,
markStarted: input.markStarted,
markFinished: input.markFinished,
});
if (input.eventType !== "operation.pull.progress" || input.status === "error") {
logServerAction({
level: input.status === "error" ? "error" : "info",
layer: "service",
action: "store.apps.operation",
status: input.status === "error" ? "error" : "info",
message: input.message,
meta: {
operationId: input.operationId,
appId: input.appId,
action: input.action,
eventType: input.eventType,
step: input.step,
progressPercent,
},
});
}
emitEvent({
type: input.eventType,
operationId: input.operationId,
appId: input.appId,
action: input.action,
status: input.status,
progressPercent,
step: input.step,
message: input.message,
image: input.image,
dockerStatus: input.dockerStatus,
progressDetail: input.progressDetail,
timestamp: new Date().toISOString(),
});
}
async function runInstallOrRedeployOperation(
operationId: string,
params: OperationParams,
existingStack: InstalledStackConfig | null,
) {
const template =
(await findStoreCatalogTemplateByAppId(params.appId)) ??
(await findCustomStoreTemplateByAppId(params.appId));
if (!template) {
throw new Error(`Template not found for appId "${params.appId}"`);
}
if (params.action === "redeploy" && !existingStack) {
throw new Error(`Cannot redeploy "${params.appId}" because it is not installed`);
}
let composeSourceInput = params.composeSource;
if (
existingStack &&
!params.resetToCatalog &&
(params.action === "install" || params.action === "redeploy") &&
(typeof composeSourceInput !== "string" || composeSourceInput.trim().length === 0)
) {
try {
composeSourceInput = await readFile(
resolveInstalledComposePath({
appId: existingStack.appId,
composePath: existingStack.composePath,
stackName: existingStack.stackName,
}),
"utf8",
);
} catch {
// Keep catalog/template compose fallback when installed compose is unavailable.
}
}
const hasComposeSource = typeof composeSourceInput === "string" &&
composeSourceInput.trim().length > 0;
const requestedEnv = params.env ?? {};
const templateComposeSource = hasComposeSource ? null : await resolveStoreComposeSource(params.appId);
let composeSourcePrimary:
| ReturnType<typeof extractPrimaryServiceWithName>
| null = null;
if (hasComposeSource) {
const parsedCompose = parseComposeFile(composeSourceInput as string);
if (!parsedCompose) {
throw new Error("Invalid compose source");
}
composeSourcePrimary = extractPrimaryServiceWithName(parsedCompose, params.appId);
if (!composeSourcePrimary) {
throw new Error("Invalid compose source");
}
}
let effectiveEnv: Record<string, string>;
if (hasComposeSource) {
const composeEnv = composeSourcePrimary?.service.environment ?? {};
effectiveEnv = {
...(existingStack?.env ?? {}),
...composeEnv,
...requestedEnv,
};
} else {
const allowedEnvKeys = template.env.map((definition) => definition.name);
const templateDefaults = Object.fromEntries(
template.env
.filter((definition) => typeof definition.default === "string")
.map((definition) => [definition.name, definition.default as string]),
);
effectiveEnv = mergeEnv(
allowedEnvKeys,
templateDefaults,
existingStack?.env ?? {},
requestedEnv,
);
}
effectiveEnv = withComposeInterpolationEnv(effectiveEnv, params.appId);
const composeSourcePort = hasComposeSource
? parseFirstPublishedPort(composeSourcePrimary?.service.ports)
: null;
const requestedPort =
typeof params.webUiPort === "number"
? params.webUiPort
: existingStack?.webUiPort ?? composeSourcePort ?? template.port ?? null;
if (requestedPort !== null) {
assertValidPort(requestedPort);
const occupied = await findStackByWebUiPort(requestedPort, {
excludeAppId: params.appId,
});
if (occupied) {
throw new Error(`webUiPort ${requestedPort} is already used by "${occupied.appId}"`);
}
}
const stackName =
existingStack?.stackName ?? sanitizeStackName(params.appId, params.displayName ?? template.name);
const isFreshInstall = params.action === "install" && !existingStack;
const storageMappingStrategy = isFreshInstall
? "app_target_path"
: "legacy_named_source";
await patchOperationAndEmit({
operationId,
appId: params.appId,
action: params.action,
status: "running",
eventType: "operation.step",
progressPercent: 8,
step: "render",
message: "Materializing compose files",
});
const materialized = await materializeInlineStackFiles({
appId: params.appId,
stackName,
composeContent: hasComposeSource ? (composeSourceInput as string) : (templateComposeSource as string),
env: effectiveEnv,
webUiPort: requestedPort,
storageMappingStrategy,
});
await patchOperationAndEmit({
operationId,
appId: params.appId,
action: params.action,
status: "running",
eventType: "operation.step",
progressPercent: 15,
step: "pull-images",
message: "Pulling images",
});
await pullImagesWithProgress(
{
composePath: materialized.composePath,
envPath: materialized.envPath,
stackName: materialized.stackName,
},
{
startPercent: 15,
endPercent: 80,
onProgress: async (percent) => {
await patchOperationAndEmit({
operationId,
appId: params.appId,
action: params.action,
status: "running",
eventType: "operation.step",
progressPercent: percent,
step: "pull-images",
message: "Pulling images",
});
},
},
);
await patchOperationAndEmit({
operationId,
appId: params.appId,
action: params.action,
status: "running",
eventType: "operation.step",
progressPercent: 85,
step: "compose-up",
message: "Applying docker compose up -d",
});
await runComposeUp({
composePath: materialized.composePath,
envPath: materialized.envPath,
stackName: materialized.stackName,
});
await patchOperationAndEmit({
operationId,
appId: params.appId,
action: params.action,
status: "running",
eventType: "operation.step",
progressPercent: 95,
step: "health-check",
message: "Finalizing deployment",
});
const deploymentHealth = await waitForDeploymentHealth({
operationId,
appId: params.appId,
action: params.action,
composePath: materialized.composePath,
envPath: materialized.envPath,
stackName: materialized.stackName,
webUiUrl: buildLocalWebUiUrl({
webUiPort: materialized.webUiPort,
scheme: template.scheme,
index: template.index,
}),
});
if (deploymentHealth.webUiReachable === false) {
logServerAction({
level: "warn",
layer: "service",
action: "store.apps.operation.health-check",
status: "error",
message: "Web UI probe did not respond before deployment completed",
meta: {
operationId,
appId: params.appId,
action: params.action,
},
});
}
await upsertInstalledStack({
appId: params.appId,
templateName: template.templateName,
stackName: materialized.stackName,
composePath: materialized.composePath,
status: "installed",
webUiPort: materialized.webUiPort,
env: effectiveEnv,
markInstalledAt: true,
});
}
async function runUpdateOperation(
operationId: string,
params: OperationParams,
existingStack: InstalledStackConfig | null,
) {
if (!existingStack || existingStack.status === "not_installed") {
throw new Error(`Cannot update "${params.appId}" because it is not installed`);
}
const resolvedComposePath = resolveInstalledComposePath({
appId: existingStack.appId,
composePath: existingStack.composePath,
stackName: existingStack.stackName,
});
const composeInput = {
composePath: resolvedComposePath,
envPath: resolveEnvPathFromComposePath(resolvedComposePath),
stackName: existingStack.stackName,
};
await patchOperationAndEmit({
operationId,
appId: params.appId,
action: params.action,
status: "running",
eventType: "operation.step",
progressPercent: 8,
step: "load-installed-compose",
message: "Loading installed compose source",
});
let installedComposeSource: string;
try {
installedComposeSource = await readFile(resolvedComposePath, "utf8");
} catch {
throw new Error("Installed compose source is unavailable");
}
await patchOperationAndEmit({
operationId,
appId: params.appId,
action: params.action,
status: "running",
eventType: "operation.step",
progressPercent: 16,
step: "load-store-compose",
message: "Loading store compose source",
});
const storeComposeSource = await resolveStoreComposeSource(params.appId);
const installedDocument = parseComposeDocument(installedComposeSource, "installed");
const storeDocument = parseComposeDocument(storeComposeSource, "store");
const installedServices = resolveComposeServices(installedDocument, "installed");
const storeServices = resolveComposeServices(storeDocument, "store");
if (!haveEqualServiceSet(installedServices, storeServices)) {
throw new Error("Installed compose does not match store services");
}
let updatedImageCount = 0;
for (const serviceName of Object.keys(installedServices)) {
const installedService = installedServices[serviceName];
const storeService = storeServices[serviceName];
const storeImage = storeService.image;
if (typeof storeImage !== "string" || storeImage.trim().length === 0) {
continue;
}
const normalizedStoreImage = storeImage.trim();
if (installedService.image !== normalizedStoreImage) {
installedService.image = normalizedStoreImage;
updatedImageCount += 1;
}
}
if (updatedImageCount === 0) {
await patchOperationAndEmit({
operationId,
appId: params.appId,
action: params.action,
status: "running",
eventType: "operation.step",
progressPercent: 95,
step: "up-to-date",
message: "Installed compose images are already up to date",
});
return;
}
const nextComposeSource = yaml.dump(installedDocument, {
lineWidth: -1,
noRefs: true,
sortKeys: false,
});
const backupPath = `${resolvedComposePath}.bak`;
let backupPrepared = false;
try {
await patchOperationAndEmit({
operationId,
appId: params.appId,
action: params.action,
status: "running",
eventType: "operation.step",
progressPercent: 24,
step: "write-compose",
message: "Updating compose image references",
});
await copyFile(resolvedComposePath, backupPath);
backupPrepared = true;
await writeFile(resolvedComposePath, nextComposeSource, "utf8");
await patchOperationAndEmit({
operationId,
appId: params.appId,
action: params.action,
status: "running",
eventType: "operation.step",
progressPercent: 32,
step: "pull-images",
message: "Pulling updated images",
});
await pullImagesWithProgress(
composeInput,
{
startPercent: 32,
endPercent: 83,
onProgress: async (percent) => {
await patchOperationAndEmit({
operationId,
appId: params.appId,
action: params.action,
status: "running",
eventType: "operation.step",
progressPercent: percent,
step: "pull-images",
message: "Pulling updated images",
});
},
},
);
await patchOperationAndEmit({
operationId,
appId: params.appId,
action: params.action,
status: "running",
eventType: "operation.step",
progressPercent: 85,
step: "compose-up",
message: "Applying docker compose up -d",
});
await runComposeUp(composeInput);
await patchOperationAndEmit({
operationId,
appId: params.appId,
action: params.action,
status: "running",
eventType: "operation.step",
progressPercent: 95,
step: "health-check",
message: "Finalizing update",
});
const updateTemplate =
(await findStoreCatalogTemplateByAppId(params.appId)) ??
(await findCustomStoreTemplateByAppId(params.appId));
const deploymentHealth = await waitForDeploymentHealth({
operationId,
appId: params.appId,
action: params.action,
composePath: composeInput.composePath,
envPath: composeInput.envPath,
stackName: composeInput.stackName,
webUiUrl: buildLocalWebUiUrl({
webUiPort: existingStack.webUiPort,
scheme: updateTemplate?.scheme,
index: updateTemplate?.index,
}),
});
if (deploymentHealth.webUiReachable === false) {
logServerAction({
level: "warn",
layer: "service",
action: "store.apps.operation.health-check",
status: "error",
message: "Web UI probe did not respond before update completed",
meta: {
operationId,
appId: params.appId,
action: params.action,
},
});
}
} catch (error) {
if (backupPrepared) {
try {
await copyFile(backupPath, resolvedComposePath);
await runComposeUp(composeInput);
} catch (recoveryError) {
logServerAction({
level: "error",
layer: "service",
action: "store.apps.operation.rollback",
status: "error",
message: "Failed to restore previous compose after update error",
error: recoveryError,
meta: {
operationId,
appId: params.appId,
},
});
}
}
throw error;
} finally {
if (backupPrepared) {
try {
await rm(backupPath, { force: true });
} catch {
// Best-effort cleanup.
}
}
}
await upsertInstalledStack({
appId: existingStack.appId,
templateName: existingStack.templateName,
stackName: existingStack.stackName,
composePath: resolvedComposePath,
status: "installed",
webUiPort: existingStack.webUiPort,
env: existingStack.env,
});
}
async function runUninstallOperation(operationId: string, params: OperationParams) {
const stack = await findInstalledStackByAppId(params.appId);
if (!stack || stack.status === "not_installed") {
await patchOperationAndEmit({
operationId,
appId: params.appId,
action: params.action,
status: "running",
eventType: "operation.step",
progressPercent: 90,
step: "noop",
message: "Application already uninstalled",
});
await deleteInstalledStackByAppId(params.appId);
return;
}
await patchOperationAndEmit({
operationId,
appId: params.appId,
action: params.action,
status: "running",
eventType: "operation.step",
progressPercent: 35,
step: "compose-down",
message: "Running docker compose down",
});
await runComposeDown({
composePath: stack.composePath,
envPath: stack.composePath.replace(/docker-compose\.yml$/, ".env"),
stackName: stack.stackName,
removeVolumes: params.removeVolumes,
});
await patchOperationAndEmit({
operationId,
appId: params.appId,
action: params.action,
status: "running",
eventType: "operation.step",
progressPercent: params.removeVolumes ? 70 : 90,
step: "cleanup",
message: params.removeVolumes
? "Removing app data"
: "Removing installed stack files",
});
await cleanupComposeDataOnUninstall({
composePath: stack.composePath,
removeVolumes: params.removeVolumes,
});
await patchOperationAndEmit({
operationId,
appId: params.appId,
action: params.action,
status: "running",
eventType: "operation.step",
progressPercent: 95,
step: "cleanup",
message: "Removing installed stack record",
});
await deleteInstalledStackByAppId(params.appId);
}
function resolveEnvPathFromComposePath(composePath: string) {
return path.join(path.dirname(composePath), ".env");
}
async function runRuntimeLifecycleOperation(operationId: string, params: OperationParams) {
const stack = await findInstalledStackByAppId(params.appId);
if (!stack || stack.status === "not_installed") {
throw new Error(`Cannot ${params.action} "${params.appId}" because it is not installed`);
}
const composeInput = {
composePath: stack.composePath,
envPath: resolveEnvPathFromComposePath(stack.composePath),
stackName: stack.stackName,
};
await patchOperationAndEmit({
operationId,
appId: params.appId,
action: params.action,
status: "running",
eventType: "operation.step",
progressPercent: 40,
step: "compose-action",
message: `Running compose ${params.action}`,
});
if (params.action === "start") {
await runComposeStart(composeInput);
} else if (params.action === "stop") {
await runComposeStop(composeInput);
} else {
await runComposeRestart(composeInput);
}
await patchOperationAndEmit({
operationId,
appId: params.appId,
action: params.action,
status: "running",
eventType: "operation.step",
progressPercent: 85,
step: "status-refresh",
message: "Refreshing runtime status",
});
await getComposeRuntimeInfo(composeInput);
}
async function runCheckUpdatesOperation(operationId: string, params: OperationParams) {
const stack = await findInstalledStackByAppId(params.appId);
if (!stack || stack.status === "not_installed") {
throw new Error(`Cannot check updates for "${params.appId}" because it is not installed`);
}
await patchOperationAndEmit({
operationId,
appId: params.appId,
action: params.action,
status: "running",
eventType: "operation.step",
progressPercent: 45,
step: "inspect-images",
message: "Checking Docker image updates",
});
const [catalogTemplate, customTemplate] = await Promise.all([
findStoreCatalogTemplateByAppId(params.appId),
findCustomStoreTemplateByAppId(params.appId),
]);
const template = catalogTemplate ?? customTemplate;
const composePath = resolveInstalledComposePath({
appId: stack.appId,
composePath: stack.composePath,
stackName: stack.stackName,
});
let installedImage: string | null = null;
try {
const composeContent = await readFile(composePath, "utf8");
const parsed = parseComposeFile(composeContent);
if (parsed) {
installedImage = extractPrimaryServiceWithName(parsed, params.appId)?.service.image ?? null;
}
} catch {
// ignore read errors
}
const updateAvailable = isTagUpdateAvailable(installedImage, template?.image);
await updateStackUpdateStatus({
appId: params.appId,
isUpToDate: !updateAvailable,
});
await patchOperationAndEmit({
operationId,
appId: params.appId,
action: params.action,
status: "running",
eventType: "operation.step",
progressPercent: 90,
step: "updates-resolved",
message: updateAvailable ? "Updates available" : "Already up to date",
});
}
async function executeStoreOperation(operationId: string, params: OperationParams) {
await patchOperationAndEmit({
operationId,
appId: params.appId,
action: params.action,
status: "running",
eventType: "operation.started",
progressPercent: 1,
step: "start",
markStarted: true,
message: "Operation started",
});
try {
const existingStack = await findInstalledStackByAppId(params.appId);
if (params.action === "install" || params.action === "redeploy") {
await runInstallOrRedeployOperation(operationId, params, existingStack);
} else if (params.action === "update") {
await runUpdateOperation(operationId, params, existingStack);
} else if (params.action === "uninstall") {
await runUninstallOperation(operationId, params);
} else if (params.action === "check-updates") {
await runCheckUpdatesOperation(operationId, params);
} else {
await runRuntimeLifecycleOperation(operationId, params);
}
// Mark app as up-to-date after successful install/redeploy/update
if (
params.action === "install" ||
params.action === "redeploy" ||
params.action === "update"
) {
await updateStackUpdateStatus({
appId: params.appId,
isUpToDate: true,
});
}
invalidateInstalledAppsCache();
await patchOperationAndEmit({
operationId,
appId: params.appId,
action: params.action,
status: "success",
eventType: "operation.completed",
progressPercent: 100,
step: "completed",
markFinished: true,
message: "Operation completed",
errorMessage: null,
});
const appLabel = params.displayName ?? params.appId;
if (params.action === "install") {
void createNotification({ title: "App Installed", body: `${appLabel} installed successfully`, kind: "success" }).catch(() => undefined);
} else if (params.action === "update") {
void createNotification({ title: "App Updated", body: `${appLabel} updated successfully`, kind: "success" }).catch(() => undefined);
} else if (params.action === "uninstall") {
void createNotification({ title: "App Uninstalled", body: `${appLabel} was uninstalled`, kind: "info" }).catch(() => undefined);
} else if (params.action === "redeploy") {
void createNotification({ title: "App Redeployed", body: `${appLabel} redeployed successfully`, kind: "success" }).catch(() => undefined);
}
} catch (error) {
const message = error instanceof Error ? error.message : "Operation failed";
await patchOperationAndEmit({
operationId,
appId: params.appId,
action: params.action,
status: "error",
eventType: "operation.failed",
progressPercent: 100,
step: "failed",
markFinished: true,
message,
errorMessage: message,
});
const appLabel = params.displayName ?? params.appId;
if (params.action === "install" || params.action === "update" || params.action === "uninstall" || params.action === "redeploy") {
void createNotification({ title: "Operation Failed", body: `Failed to ${params.action} ${appLabel}: ${message}`, kind: "error" }).catch(() => undefined);
}
} finally {
// Always release the guard so future operations on this app can proceed.
activeOperationsByApp.delete(params.appId);
// Drop the cached terminal event after a short grace window so any
// SSE client that connects right at completion still sees it from
// memory. After the TTL, the entry is purged to keep memory bounded
// on long-lived servers (Pi).
const purgeTimer = setTimeout(() => {
latestOperationEvent.delete(operationId);
}, FINISHED_EVENT_TTL_MS);
purgeTimer.unref?.();
}
}
export async function startStoreOperation(params: OperationParams) {
if (activeOperationsByApp.has(params.appId)) {
throw new StoreOperationError(
`App ${params.appId} already has an active operation in progress. Wait for it to complete before starting another.`,
"operation_conflict",
409,
);
}
if (activeOperationsByApp.size >= serverEnv.STORE_MAX_CONCURRENT_OPERATIONS) {
throw new StoreOperationError(
"Too many app operations are already running. Wait for one to complete before starting another.",
"operation_limit_reached",
429,
);
}
activeOperationsByApp.add(params.appId);
// Prune stale finished rows non-blocking so the table doesn't grow unbounded.
void pruneFinishedOperations().catch(() => undefined);
const operationId = randomUUID();
try {
await createStoreOperation({
id: operationId,
appId: params.appId,
action: params.action,
status: "queued",
progressPercent: 0,
currentStep: "queued",
startedAt: false,
});
} catch (error) {
activeOperationsByApp.delete(params.appId);
throw error;
}
logServerAction({
layer: "service",
action: "store.apps.operation",
status: "start",
message: "Queued store operation",
meta: {
operationId,
appId: params.appId,
action: params.action,
},
});
queueMicrotask(() => {
void executeStoreOperation(operationId, params);
});
return {
operationId,
};
}
export async function getStoreOperation(operationId: string): Promise<StoreOperation | null> {
return findStoreOperationById(operationId);
}
export function getLatestStoreOperationEvent(operationId: string) {
return latestOperationEvent.get(operationId) ?? null;
}
export function subscribeToStoreOperation(operationId: string, callback: OperationSubscriber) {
const existing = operationSubscribers.get(operationId) ?? new Set<OperationSubscriber>();
existing.add(callback);
operationSubscribers.set(operationId, existing);
return () => {
const subscribers = operationSubscribers.get(operationId);
if (!subscribers) return;
subscribers.delete(callback);
if (subscribers.size === 0) {
operationSubscribers.delete(operationId);
}
};
}