danvics/zerobyte
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

feat(agent): add standalone agent runtime (#761)

Browse source 
* feat(agent): add standalone agent runtime

* fix(backups): bridge local executor to Effect restic API

* fix(agent): add Bun and DOM types to agent tsconfig

* refactor: wrap backup error in a tagged effect error

* fix: pr feedbacks
This commit is contained in:
Nico 2026-04-08 20:47:15 +02:00 committed by GitHub
parent 3169627b79
commit c371676ad0
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
18 changed files with 846 additions and 185 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

37
app/server/modules/backups/backup-executor.ts
View file

@ -1,3 +1,4 @@
import { Effect } from "effect";
import { restic } from "../../core/restic";
import type { BackupSchedule, Repository, Volume } from "../../db/schema";
import type { ResticBackupOutputDto, ResticBackupProgressDto } from "@zerobyte/core/restic";
@ -55,24 +56,28 @@ export const backupExecutor = {
const volumePath = getVolumePath(volume);
const backupOptions = createBackupOptions(schedule, volumePath, signal);
const result = await restic.backup(repository.config, volumePath, {
...backupOptions,
compressionMode: repository.compressionMode ?? "auto",
organizationId,
onProgress,
});
const execution = await Effect.runPromise(
restic
.backup(repository.config, volumePath, {
...backupOptions,
compressionMode: repository.compressionMode ?? "auto",
organizationId,
onProgress,
})
.pipe(
Effect.map((result) => ({ success: true as const, result })),
Effect.catchAll((error) => Effect.succeed({ success: false as const, error })),
),
);
return {
status: "completed",
exitCode: result.exitCode,
result: result.result,
warningDetails: result.warningDetails,
} satisfies BackupExecutionResult;
if (!execution.success) {
throw execution.error;
}
const { exitCode, result, warningDetails } = execution.result;
return { status: "completed", exitCode, result, warningDetails };
} catch (error) {
return {
status: "failed",
error,
} satisfies BackupExecutionResult;
return { status: "failed", error };
}
},
cancel: (scheduleId: number) => {

20
apps/agent/package.json Normal file
View file

@ -0,0 +1,20 @@
{
"name": "agent",
"private": true,
"type": "module",
"module": "src/index.ts",
"scripts": {
"tsc": "tsc --noEmit"
},
"dependencies": {
"@zerobyte/contracts": "workspace:*",
"@zerobyte/core": "workspace:*",
"effect": "^3.18.4"
},
"devDependencies": {
"@types/bun": "latest"
},
"peerDependencies": {
"typescript": "^6"
}
}

71
apps/agent/src/__tests__/controller-session.test.ts Normal file
View file

@ -0,0 +1,71 @@
import { afterEach, expect, mock, spyOn, test } from "bun:test";
import { Effect } from "effect";
import waitForExpect from "wait-for-expect";
import { fromPartial } from "@total-typescript/shoehorn";
import { createControllerMessage, parseAgentMessage } from "@zerobyte/contracts/agent-protocol";
import * as resticServer from "@zerobyte/core/restic/server";
import { createControllerSession } from "../controller-session";
afterEach(() => {
mock.restore();
});
test("emits backup.failed when a backup command hits a restic error", async () => {
spyOn(resticServer, "createRestic").mockReturnValue(
fromPartial({
backup: () => Effect.fail("source path missing"),
}),
);
const outboundMessages: string[] = [];
const session = createControllerSession(
fromPartial({
send: (message: string) => {
outboundMessages.push(message);
},
}),
);
try {
session.onOpen();
session.onMessage(
createControllerMessage("backup.run", {
jobId: "job-1",
scheduleId: "schedule-1",
organizationId: "org-1",
sourcePath: "/tmp/missing-source",
repositoryConfig: {
backend: "local",
path: "/tmp/test-repository",
},
options: {},
runtime: {
password: "password",
cacheDir: "/tmp/restic-cache",
passFile: "/tmp/restic-pass",
defaultExcludes: [],
},
}),
);
await waitForExpect(() => {
const failedMessage = outboundMessages
.map((message) => parseAgentMessage(message))
.find((message) => message?.success && message.data.type === "backup.failed");
expect(failedMessage?.success).toBe(true);
if (!failedMessage || !failedMessage.success || failedMessage.data.type !== "backup.failed") {
return;
}
expect(failedMessage.data.payload).toEqual({
jobId: "job-1",
scheduleId: "schedule-1",
error: "source path missing",
errorDetails: "source path missing",
});
});
} finally {
session.close();
}
});

21
apps/agent/src/commands/backup-cancel.ts Normal file
View file

@ -0,0 +1,21 @@
import { Effect } from "effect";
import { type BackupCancelPayload } from "@zerobyte/contracts/agent-protocol";
import { logger } from "@zerobyte/core/node";
import type { ControllerCommandContext } from "../context";
export const handleBackupCancelCommand = (context: ControllerCommandContext, payload: BackupCancelPayload) => {
return Effect.gen(function* () {
const running = yield* context.getRunningJob(payload.jobId);
if (!running) {
logger.warn(`Backup ${payload.jobId} is not running`);
return;
}
if (running.scheduleId !== payload.scheduleId) {
logger.warn(`Ignoring cancel for backup ${payload.jobId} due to schedule mismatch ${payload.scheduleId}`);
return;
}
running.abortController.abort();
});
};

117
apps/agent/src/commands/backup-run.test.ts Normal file
View file

@ -0,0 +1,117 @@
import { afterEach, expect, mock, spyOn, test } from "bun:test";
import { Effect } from "effect";
import waitForExpect from "wait-for-expect";
import { fromPartial } from "@total-typescript/shoehorn";
import { parseAgentMessage, type BackupCancelPayload, type BackupRunPayload } from "@zerobyte/contracts/agent-protocol";
import * as resticServer from "@zerobyte/core/restic/server";
import { handleBackupCancelCommand } from "./backup-cancel";
import { handleBackupRunCommand } from "./backup-run";
import type { ControllerCommandContext, RunningJob } from "../context";
const createDeferred = <T>() => {
let resolve!: (value: T) => void;
const promise = new Promise<T>((resolvePromise) => {
resolve = resolvePromise;
});
return { promise, resolve };
};
afterEach(() => {
mock.restore();
});
test("waits for running-job registration before returning to the processor loop", async () => {
const outboundMessages: string[] = [];
const runningJobs = new Map<string, RunningJob>();
const setRunningJobGate = createDeferred<void>();
const backupGate = createDeferred<{ exitCode: number; result: null; warningDetails: null }>();
let registeredAbortController: AbortController | undefined;
spyOn(resticServer, "createRestic").mockReturnValue(
fromPartial({
backup: () =>
Effect.async<{ exitCode: number; result: null; warningDetails: null }, never>((resume) => {
void backupGate.promise.then((result) => {
resume(Effect.succeed(result));
});
}),
}),
);
const context: ControllerCommandContext = {
getRunningJob: (jobId) => Effect.succeed(runningJobs.get(jobId)),
setRunningJob: (jobId, job) =>
Effect.async<void, never>((resume) => {
void setRunningJobGate.promise.then(() => {
runningJobs.set(jobId, job);
registeredAbortController = job.abortController;
resume(Effect.void);
});
}),
deleteRunningJob: (jobId) =>
Effect.sync(() => {
runningJobs.delete(jobId);
}),
offerOutbound: (message) =>
Effect.sync(() => {
outboundMessages.push(message);
return true;
}),
};
const runPayload = fromPartial<BackupRunPayload>({
jobId: "job-1",
scheduleId: "schedule-1",
organizationId: "org-1",
sourcePath: "/tmp/source",
repositoryConfig: {
backend: "local",
path: "/tmp/repository",
},
options: {},
runtime: {
password: "password",
cacheDir: "/tmp/restic-cache",
passFile: "/tmp/restic-pass",
defaultExcludes: [],
},
});
const cancelPayload = fromPartial<BackupCancelPayload>({
jobId: "job-1",
scheduleId: "schedule-1",
});
const runPromise = Effect.runPromise(handleBackupRunCommand(context, runPayload));
try {
const returnedBeforeRegistration = await Promise.race([
runPromise.then(() => true),
new Promise<false>((resolve) => {
setTimeout(() => resolve(false), 0);
}),
]);
expect(returnedBeforeRegistration).toBe(false);
setRunningJobGate.resolve(undefined);
await runPromise;
await Effect.runPromise(handleBackupCancelCommand(context, cancelPayload));
expect(registeredAbortController?.signal.aborted).toBe(true);
backupGate.resolve({ exitCode: 0, result: null, warningDetails: null });
await waitForExpect(() => {
const cancelledMessage = outboundMessages
.map((message) => parseAgentMessage(message))
.find((message) => message?.success && message.data.type === "backup.cancelled");
expect(cancelledMessage?.success).toBe(true);
expect(runningJobs.has("job-1")).toBe(false);
});
} finally {
setRunningJobGate.resolve(undefined);
backupGate.resolve({ exitCode: 0, result: null, warningDetails: null });
}
});

115
apps/agent/src/commands/backup-run.ts Normal file
View file

@ -0,0 +1,115 @@
import { Effect, Runtime } from "effect";
import { createAgentMessage, type BackupRunPayload } from "@zerobyte/contracts/agent-protocol";
import { logger } from "@zerobyte/core/node";
import { type ResticDeps } from "@zerobyte/core/restic";
import { createRestic } from "@zerobyte/core/restic/server";
import { toErrorDetails, toMessage } from "@zerobyte/core/utils";
import type { ControllerCommandContext } from "../context";
export const handleBackupRunCommand = (context: ControllerCommandContext, payload: BackupRunPayload) => {
return Effect.gen(function* () {
const existing = yield* context.getRunningJob(payload.jobId);
if (existing) {
yield* context.offerOutbound(
createAgentMessage("backup.failed", {
jobId: payload.jobId,
scheduleId: payload.scheduleId,
error: "Backup job is already running",
}),
);
return;
}
logger.info(`Starting backup ${payload.jobId} for schedule ${payload.scheduleId}`);
const abortController = new AbortController();
yield* context.setRunningJob(payload.jobId, { scheduleId: payload.scheduleId, abortController });
yield* Effect.fork(
Effect.gen(function* () {
const sendCancelled = () => {
return context.offerOutbound(
createAgentMessage("backup.cancelled", {
jobId: payload.jobId,
scheduleId: payload.scheduleId,
message: "Backup was cancelled",
}),
);
};
yield* context.offerOutbound(
createAgentMessage("backup.started", {
jobId: payload.jobId,
scheduleId: payload.scheduleId,
}),
);
const deps: ResticDeps = {
resolveSecret: async (encrypted) => encrypted,
getOrganizationResticPassword: async () => payload.runtime.password,
resticCacheDir: payload.runtime.cacheDir,
resticPassFile: payload.runtime.passFile,
defaultExcludes: payload.runtime.defaultExcludes,
hostname: payload.runtime.hostname,
};
const restic = createRestic(deps);
const runtime = yield* Effect.runtime<never>();
yield* restic
.backup(payload.repositoryConfig, payload.sourcePath, {
organizationId: payload.organizationId,
...payload.options,
signal: abortController.signal,
onProgress: (progress) => {
void Runtime.runPromise(
runtime,
context.offerOutbound(
createAgentMessage("backup.progress", {
jobId: payload.jobId,
scheduleId: payload.scheduleId,
progress,
}),
),
).catch((error) => {
logger.error(`Failed to send backup progress update: ${toMessage(error)}`);
});
},
})
.pipe(
Effect.matchEffect({
onSuccess: (result) => {
if (abortController.signal.aborted) {
return sendCancelled();
}
return context.offerOutbound(
createAgentMessage("backup.completed", {
jobId: payload.jobId,
scheduleId: payload.scheduleId,
exitCode: result.exitCode,
result: result.result,
warningDetails: result.warningDetails ?? undefined,
}),
);
},
onFailure: (error) => {
if (abortController.signal.aborted) {
return sendCancelled();
}
return context.offerOutbound(
createAgentMessage("backup.failed", {
jobId: payload.jobId,
scheduleId: payload.scheduleId,
error: toMessage(error),
errorDetails: toErrorDetails(error),
}),
);
},
}),
Effect.ensuring(context.deleteRunningJob(payload.jobId)),
);
}),
);
}).pipe(Effect.asVoid);
};

12
apps/agent/src/commands/heartbeat-ping.ts Normal file
View file

@ -0,0 +1,12 @@
import { createAgentMessage, type ControllerMessage } from "@zerobyte/contracts/agent-protocol";
import type { ControllerCommandContext } from "../context";
type HeartbeatPingPayload = Extract<ControllerMessage, { type: "heartbeat.ping" }>["payload"];
export const handleHeartbeatPingCommand = (context: ControllerCommandContext, payload: HeartbeatPingPayload) => {
return context.offerOutbound(
createAgentMessage("heartbeat.pong", {
sentAt: payload.sentAt,
}),
);
};

19
apps/agent/src/commands/index.ts Normal file
View file

@ -0,0 +1,19 @@
import type { ControllerMessage } from "@zerobyte/contracts/agent-protocol";
import { handleBackupCancelCommand } from "./backup-cancel";
import { handleBackupRunCommand } from "./backup-run";
import type { ControllerCommandContext } from "../context";
import { handleHeartbeatPingCommand } from "./heartbeat-ping";
export const handleControllerCommand = (context: ControllerCommandContext, message: ControllerMessage) => {
switch (message.type) {
case "backup.run": {
return handleBackupRunCommand(context, message.payload);
}
case "backup.cancel": {
return handleBackupCancelCommand(context, message.payload);
}
case "heartbeat.ping": {
return handleHeartbeatPingCommand(context, message.payload);
}
}
};

14
apps/agent/src/context.ts Normal file
View file

@ -0,0 +1,14 @@
import type { AgentWireMessage } from "@zerobyte/contracts/agent-protocol";
import type { Effect } from "effect";
export type RunningJob = {
scheduleId: string;
abortController: AbortController;
};
export type ControllerCommandContext = {
getRunningJob: (jobId: string) => Effect.Effect<RunningJob | undefined, never, never>;
setRunningJob: (jobId: string, job: RunningJob) => Effect.Effect<void, never, never>;
deleteRunningJob: (jobId: string) => Effect.Effect<void, never, never>;
offerOutbound: (message: AgentWireMessage) => Effect.Effect<boolean, never, never>;
};

121
apps/agent/src/controller-session.ts Normal file
View file

@ -0,0 +1,121 @@
import { Effect, Fiber, Queue, Ref } from "effect";
import {
createAgentMessage,
parseControllerMessage,
type AgentWireMessage,
type ControllerWireMessage,
} from "@zerobyte/contracts/agent-protocol";
import { logger } from "@zerobyte/core/node";
import { toMessage } from "@zerobyte/core/utils";
import { handleControllerCommand } from "./commands";
import type { ControllerCommandContext, RunningJob } from "./context";
export type ControllerSession = {
onOpen: () => void;
onMessage: (data: unknown) => void;
close: () => void;
};
export const createControllerSession = (ws: WebSocket): ControllerSession => {
const outboundQueue = Effect.runSync(Queue.bounded<AgentWireMessage>(64));
const inboundQueue = Effect.runSync(Queue.bounded<ControllerWireMessage>(64));
const runningJobsRef = Effect.runSync(Ref.make<Map<string, RunningJob>>(new Map()));
const getRunningJob = (jobId: string) => Ref.get(runningJobsRef).pipe(Effect.map((map) => map.get(jobId)));
const setRunningJob = (jobId: string, job: RunningJob) => {
return Ref.update(runningJobsRef, (current) => {
const next = new Map(current);
next.set(jobId, job);
return next;
});
};
const abortRunningJobs = Effect.gen(function* () {
const runningJobs = yield* Ref.modify(runningJobsRef, (current) => [current, new Map()]);
yield* Effect.sync(() => {
for (const runningJob of runningJobs.values()) {
runningJob.abortController.abort();
}
});
});
const deleteRunningJob = (jobId: string) => {
return Ref.update(runningJobsRef, (current) => {
const next = new Map(current);
next.delete(jobId);
return next;
});
};
const offerOutbound = (message: AgentWireMessage) => {
return Queue.offer(outboundQueue, message);
};
const offerInbound = (message: ControllerWireMessage) => {
return Queue.offer(inboundQueue, message);
};
const commandContext: ControllerCommandContext = {
getRunningJob,
setRunningJob,
deleteRunningJob,
offerOutbound,
};
const writerFiber = Effect.runFork(
Effect.forever(
Effect.gen(function* () {
const message = yield* Queue.take(outboundQueue);
yield* Effect.sync(() => {
try {
ws.send(message);
} catch (error) {
logger.error(`Failed to send controller message: ${toMessage(error)}`);
}
});
}),
),
);
const processorFiber = Effect.runFork(
Effect.forever(
Effect.gen(function* () {
const data = yield* Queue.take(inboundQueue);
const parsed = parseControllerMessage(data);
if (parsed === null) {
logger.warn("Agent received invalid JSON");
return;
}
if (!parsed.success) {
logger.warn(`Agent received an invalid message: ${parsed.error.message}`);
return;
}
yield* handleControllerCommand(commandContext, parsed.data);
}),
),
);
return {
onOpen: () => {
void Effect.runPromise(offerOutbound(createAgentMessage("agent.ready", { agentId: "" }))).catch((error) => {
logger.error(`Failed to queue ready message: ${toMessage(error)}`);
});
},
onMessage: (data) => {
void Effect.runPromise(offerInbound(data as ControllerWireMessage)).catch((error) => {
logger.error(`Failed to queue inbound message: ${toMessage(error)}`);
});
},
close: () => {
void Effect.runPromise(abortRunningJobs).catch(() => {});
void Effect.runPromise(Fiber.interrupt(writerFiber)).catch(() => {});
void Effect.runPromise(Fiber.interrupt(processorFiber)).catch(() => {});
void Effect.runPromise(Queue.shutdown(outboundQueue)).catch(() => {});
void Effect.runPromise(Queue.shutdown(inboundQueue)).catch(() => {});
},
};
};

67
apps/agent/src/index.ts Normal file
View file

@ -0,0 +1,67 @@
import { logger } from "@zerobyte/core/node";
import { createControllerSession, type ControllerSession } from "./controller-session";
const controllerUrl = process.env.ZEROBYTE_CONTROLLER_URL;
const agentToken = process.env.ZEROBYTE_AGENT_TOKEN;
const RECONNECT_DELAY_MS = 1000;
class Agent {
private ws: WebSocket | null = null;
private controllerSession: ControllerSession | null = null;
private reconnectTimeout: ReturnType<typeof setTimeout> | null = null;
private scheduleReconnect() {
if (this.reconnectTimeout) {
return;
}
this.reconnectTimeout = setTimeout(() => {
this.reconnectTimeout = null;
this.connect();
}, RECONNECT_DELAY_MS);
}
connect() {
if (this.ws) {
return;
}
if (!controllerUrl) {
throw new Error("Env variable ZEROBYTE_CONTROLLER_URL is not set");
}
if (!agentToken) {
throw new Error("Env variable ZEROBYTE_AGENT_TOKEN is not set");
}
const url = new URL(controllerUrl);
url.searchParams.set("token", agentToken);
this.ws = new WebSocket(url.toString());
this.controllerSession = createControllerSession(this.ws);
this.ws.onopen = () => {
logger.info("Agent connected to controller");
this.controllerSession?.onOpen();
};
this.ws.onmessage = (event) => {
this.controllerSession?.onMessage(event.data);
};
this.ws.onclose = () => {
this.controllerSession?.close();
this.controllerSession = null;
this.ws = null;
logger.info("Agent disconnected from controller");
this.scheduleReconnect();
};
this.ws.onerror = (error) => {
logger.error("Agent encountered an error:", error);
};
}
}
if (import.meta.main) {
const agent = new Agent();
agent.connect();
}

31
apps/agent/tsconfig.json Normal file
View file

@ -0,0 +1,31 @@
{
"compilerOptions": {
"plugins": [{ "name": "@effect/language-service" }],
// Environment setup & latest features
"lib": ["DOM", "ESNext"],
"target": "ESNext",
"module": "Preserve",
"moduleDetection": "force",
"jsx": "react-jsx",
"types": ["bun", "node"],
"allowJs": true,
// Bundler mode
"moduleResolution": "bundler",
"allowImportingTsExtensions": true,
"verbatimModuleSyntax": true,
"noEmit": true,
// Best practices
"strict": true,
"skipLibCheck": true,
"noFallthroughCasesInSwitch": true,
"noUncheckedIndexedAccess": true,
"noImplicitOverride": true,
// Some stricter flags (disabled by default)
"noUnusedLocals": false,
"noUnusedParameters": false,
"noPropertyAccessFromIndexSignature": false
}
}

20
bun.lock
View file

@ -47,6 +47,7 @@
"dither-plugin": "^1.1.1",
"dotenv": "^17.3.1",
"drizzle-orm": "^1.0.0-beta.16-ea816b6",
"effect": "^3.18.4",
"es-toolkit": "^1.45.1",
"hono": "^4.12.8",
"hono-openapi": "^1.3.0",
@ -74,6 +75,7 @@
},
"devDependencies": {
"@babel/preset-typescript": "^7.28.5",
"@effect/language-service": "^0.84.2",
"@faker-js/faker": "^10.3.0",
"@happy-dom/global-registrator": "^20.8.4",
"@hey-api/openapi-ts": "^0.94.4",
@ -115,6 +117,20 @@
"wait-for-expect": "^4.0.0",
},
},
"apps/agent": {
"name": "agent",
"dependencies": {
"@zerobyte/contracts": "workspace:*",
"@zerobyte/core": "workspace:*",
"effect": "^3.18.4",
},
"devDependencies": {
"@types/bun": "latest",
},
"peerDependencies": {
"typescript": "^6",
},
},
"packages/contracts": {
"name": "@zerobyte/contracts",
"dependencies": {
@ -277,6 +293,8 @@
"@drizzle-team/brocli": ["@drizzle-team/brocli@0.11.0", "", {}, "sha512-hD3pekGiPg0WPCCGAZmusBBJsDqGUR66Y452YgQsZOnkdQ7ViEPKuyP4huUGEZQefp8g34RRodXYmJ2TbCH+tg=="],
"@effect/language-service": ["@effect/language-service@0.84.3", "", { "bin": { "effect-language-service": "cli.js" } }, "sha512-zpxi6rLCwst/cBQd7ElwDvt36Y6Jvz8v6bCLnNiOL6OXvdLmqjOFWyzWZdMh92vvBQA/aVKhfIAAOP3o4wKt0A=="],
"@electric-sql/pglite": ["@electric-sql/pglite@0.3.15", "", {}, "sha512-Cj++n1Mekf9ETfdc16TlDi+cDDQF0W7EcbyRHYOAeZdsAe8M/FJg18itDTSwyHfar2WIezawM9o0EKaRGVKygQ=="],
"@electric-sql/pglite-socket": ["@electric-sql/pglite-socket@0.0.20", "", { "peerDependencies": { "@electric-sql/pglite": "0.3.15" }, "bin": { "pglite-server": "dist/scripts/server.js" } }, "sha512-J5nLGsicnD9wJHnno9r+DGxfcZWh+YJMCe0q/aCgtG6XOm9Z7fKeite8IZSNXgZeGltSigM9U/vAWZQWdgcSFg=="],
@ -1065,6 +1083,8 @@
"acorn": ["acorn@8.16.0", "", { "bin": { "acorn": "bin/acorn" } }, "sha512-UVJyE9MttOsBQIDKw1skb9nAwQuR5wuGD3+82K6JgJlm/Y+KI92oNsMNGZCYdDsVtRHSak0pcV5Dno5+4jh9sw=="],
"agent": ["agent@workspace:apps/agent"],
"agent-base": ["agent-base@7.1.4", "", {}, "sha512-MnA+YT8fwfJPgBx3m60MNqakm30XOkyIoH1y6huTQvC0PwZG7ki8NacLBcrPbNoo8vEZy7Jpuk7+jMO+CUovTQ=="],
"ansi-colors": ["ansi-colors@4.1.3", "", {}, "sha512-/6w/C21Pm1A7aZitlI5Ni/2J6FFQN8i1Cvz3kHABAAbw93v/NlvKdVOqz7CCWz/3iv/JplRSEEZ83XION15ovw=="],

2
package.json
View file

@ -72,6 +72,7 @@
"dither-plugin": "^1.1.1",
"dotenv": "^17.3.1",
"drizzle-orm": "^1.0.0-beta.16-ea816b6",
"effect": "^3.18.4",
"es-toolkit": "^1.45.1",
"hono": "^4.12.8",
"hono-openapi": "^1.3.0",
@ -99,6 +100,7 @@
},
"devDependencies": {
"@babel/preset-typescript": "^7.28.5",
"@effect/language-service": "^0.84.2",
"@faker-js/faker": "^10.3.0",
"@happy-dom/global-registrator": "^20.8.4",
"@hey-api/openapi-ts": "^0.94.4",

41
packages/core/src/restic/commands/__tests__/backup.test.ts
View file

@ -1,4 +1,5 @@
import { afterEach, describe, expect, test, vi } from "vitest";
import { Effect } from "effect";
import * as cleanupModule from "../../helpers/cleanup-temporary-keys";
import * as spawnModule from "../../../utils/spawn";
import { ResticError } from "../../error";
@ -62,7 +63,7 @@ const setup = ({ spawnResult = {}, onSpawnCall }: SetupOptions = {}) => {
let capturedArgs: string[] = [];
vi.spyOn(cleanupModule, "cleanupTemporaryKeys").mockImplementation(() => Promise.resolve());
vi.spyOn(spawnModule, "safeSpawn").mockImplementation((params) => {
vi.spyOn(spawnModule, "safeSpawn").mockImplementation((params: SafeSpawnParams) => {
capturedArgs = params.args;
return Promise.resolve(onSpawnCall?.(params)).then(() => ({
exitCode: 0,
@ -87,6 +88,9 @@ const setup = ({ spawnResult = {}, onSpawnCall }: SetupOptions = {}) => {
};
};
const runBackup = (...args: Parameters<typeof backup>) => Effect.runPromise(backup(...args));
const runBackupError = (...args: Parameters<typeof backup>) => Effect.runPromise(Effect.flip(backup(...args)));
afterEach(() => {
vi.restoreAllMocks();
});
@ -95,7 +99,7 @@ describe("backup command", () => {
describe("argument construction", () => {
test("passes source path as positional arg when no include list is given", async () => {
const { getArgs, hasFlag } = setup();
await backup(config, "/mnt/data", { organizationId: "org-1" }, mockDeps);
await runBackup(config, "/mnt/data", { organizationId: "org-1" }, mockDeps);
expect(getArgs()).toContain("/mnt/data");
expect(hasFlag("--files-from")).toBe(false);
@ -105,7 +109,7 @@ describe("backup command", () => {
const { getArgs } = setup();
const source = "--help";
await backup(config, source, { organizationId: "org-1" }, mockDeps);
await runBackup(config, source, { organizationId: "org-1" }, mockDeps);
const separatorIndex = getArgs().indexOf("--");
expect(separatorIndex).toBeGreaterThan(-1);
@ -132,7 +136,7 @@ describe("backup command", () => {
},
});
await backup(
await runBackup(
config,
"/mnt/data",
{
@ -152,7 +156,7 @@ describe("backup command", () => {
test("always includes DEFAULT_EXCLUDES as --exclude args", async () => {
const { getOptionValues } = setup();
await backup(config, "/mnt/data", { organizationId: "org-1" }, mockDeps);
await runBackup(config, "/mnt/data", { organizationId: "org-1" }, mockDeps);
expect(getOptionValues("--exclude").length).toBeGreaterThan(0);
});
@ -161,7 +165,7 @@ describe("backup command", () => {
describe("exit code handling", () => {
test("returns parsed result on exit code 0", async () => {
setup();
const { result, exitCode } = await backup(config, "/mnt/data", { organizationId: "org-1" }, mockDeps);
const { result, exitCode } = await runBackup(config, "/mnt/data", { organizationId: "org-1" }, mockDeps);
expect(exitCode).toBe(0);
expect(result?.snapshot_id).toBe("abcd1234");
@ -169,7 +173,7 @@ describe("backup command", () => {
test("returns result without throwing on exit code 3 (partial read errors)", async () => {
setup({ spawnResult: { exitCode: 3 } });
const { result, exitCode } = await backup(config, "/mnt/data", { organizationId: "org-1" }, mockDeps);
const { result, exitCode } = await runBackup(config, "/mnt/data", { organizationId: "org-1" }, mockDeps);
expect(exitCode).toBe(3);
expect(result).not.toBeNull();
@ -178,15 +182,14 @@ describe("backup command", () => {
test("throws ResticError on non-zero, non-3 exit codes", async () => {
setup({ spawnResult: { exitCode: 1, summary: "", error: "fatal error" } });
await expect(backup(config, "/mnt/data", { organizationId: "org-1" }, mockDeps)).rejects.toBeInstanceOf(
ResticError,
);
const error = await runBackupError(config, "/mnt/data", { organizationId: "org-1" }, mockDeps);
expect(error).toBeInstanceOf(ResticError);
});
test("preserves the exit code inside the thrown ResticError", async () => {
setup({ spawnResult: { exitCode: 12, summary: "", error: "wrong password" } });
const error = await backup(config, "/mnt/data", { organizationId: "org-1" }, mockDeps).catch((e) => e);
const error = await runBackupError(config, "/mnt/data", { organizationId: "org-1" }, mockDeps);
expect(error).toBeInstanceOf(ResticError);
expect((error as ResticError).code).toBe(12);
});
@ -204,7 +207,7 @@ describe("backup command", () => {
},
});
const error = await backup(config, "/mnt/data", { organizationId: "org-1" }, mockDeps).catch((e) => e);
const error = await runBackupError(config, "/mnt/data", { organizationId: "org-1" }, mockDeps);
expect(error).toBeInstanceOf(ResticError);
expect((error as ResticError).summary).toBe("Command failed: An error occurred while executing the command.");
expect((error as ResticError).details).toBe(
@ -219,7 +222,7 @@ describe("backup command", () => {
spawnResult: { exitCode: 130, summary: "", error: "" },
});
const { result, exitCode, warningDetails } = await backup(
const { result, exitCode, warningDetails } = await runBackup(
config,
"/mnt/data",
{
@ -238,7 +241,7 @@ describe("backup command", () => {
describe("output parsing", () => {
test("returns a fully parsed summary object on valid output", async () => {
setup();
const { result } = await backup(config, "/mnt/data", { organizationId: "org-1" }, mockDeps);
const { result } = await runBackup(config, "/mnt/data", { organizationId: "org-1" }, mockDeps);
expect(result).toMatchObject({
message_type: "summary",
@ -249,14 +252,14 @@ describe("backup command", () => {
test("returns { result: null } when summary line is not valid JSON", async () => {
setup({ spawnResult: { summary: "not-json" } });
const { result } = await backup(config, "/mnt/data", { organizationId: "org-1" }, mockDeps);
const { result } = await runBackup(config, "/mnt/data", { organizationId: "org-1" }, mockDeps);
expect(result).toBeNull();
});
test("returns { result: null } when summary JSON does not satisfy the schema", async () => {
setup({ spawnResult: { summary: JSON.stringify({ message_type: "summary" }) } });
const { result } = await backup(config, "/mnt/data", { organizationId: "org-1" }, mockDeps);
const { result } = await runBackup(config, "/mnt/data", { organizationId: "org-1" }, mockDeps);
expect(result).toBeNull();
});
@ -267,7 +270,7 @@ describe("backup command", () => {
const progressUpdates: unknown[] = [];
setup({ onSpawnCall: (params) => params.onStdout?.(VALID_PROGRESS_LINE) });
await backup(
await runBackup(
config,
"/mnt/data",
{
@ -294,7 +297,7 @@ describe("backup command", () => {
});
await expect(
backup(config, "/mnt/data", { organizationId: "org-1", onProgress: () => {} }, mockDeps),
runBackup(config, "/mnt/data", { organizationId: "org-1", onProgress: () => {} }, mockDeps),
).resolves.toBeDefined();
});
@ -304,7 +307,7 @@ describe("backup command", () => {
onSpawnCall: (params) => params.onStdout?.(JSON.stringify({ message_type: "verbose_status", action: "scan" })),
});
await backup(
await runBackup(
config,
"/mnt/data",
{

321
packages/core/src/restic/commands/backup.ts
View file

@ -1,6 +1,7 @@
import fs from "node:fs/promises";
import os from "node:os";
import path from "node:path";
import { Data, Effect } from "effect";
import { throttle } from "es-toolkit";
import type { CompressionMode, RepositoryConfig } from "../schemas";
import { type ResticBackupProgressDto, resticBackupOutputSchema, resticBackupProgressSchema } from "../restic-dto";
@ -12,8 +13,14 @@ import { validateCustomResticParams } from "../helpers/validate-custom-params";
import { ResticError } from "../error";
import { logger, safeSpawn } from "../../node";
import type { ResticDeps } from "../types";
import { toMessage } from "../../utils";
export const backup = async (
class ResticBackupCommandError extends Data.TaggedError("ResticBackupCommandError")<{
cause: unknown;
message: string;
}> {}
export const backup = (
config: RepositoryConfig,
source: string,
options: {
@ -31,183 +38,197 @@ export const backup = async (
},
deps: ResticDeps,
) => {
const repoUrl = buildRepoUrl(config);
const env = await buildEnv(config, options.organizationId, deps);
return Effect.tryPromise({
try: async () => {
const repoUrl = buildRepoUrl(config);
const env = await buildEnv(config, options.organizationId, deps);
const args: string[] = ["--repo", repoUrl, "backup", "--compression", options.compressionMode ?? "auto"];
const args: string[] = ["--repo", repoUrl, "backup", "--compression", options.compressionMode ?? "auto"];
if (options.oneFileSystem) {
args.push("--one-file-system");
}
if (options.oneFileSystem) {
args.push("--one-file-system");
}
if (deps.hostname) {
args.push("--host", deps.hostname);
}
if (deps.hostname) {
args.push("--host", deps.hostname);
}
if (options.tags && options.tags.length > 0) {
for (const tag of options.tags) {
args.push("--tag", tag);
}
}
if (options.tags && options.tags.length > 0) {
for (const tag of options.tags) {
args.push("--tag", tag);
}
}
let includeFile: string | null = null;
let rawIncludeFile: string | null = null;
const usesSourceArg =
(!options.includePaths || options.includePaths.length === 0) &&
(!options.includePatterns || options.includePatterns.length === 0);
let includeFile: string | null = null;
let rawIncludeFile: string | null = null;
const usesSourceArg =
(!options.includePaths || options.includePaths.length === 0) &&
(!options.includePatterns || options.includePatterns.length === 0);
if (options.includePatterns?.length) {
const tmp = await fs.mkdtemp(path.join(os.tmpdir(), "zerobyte-restic-include-"));
includeFile = path.join(tmp, "include.txt");
if (options.includePatterns?.length) {
const tmp = await fs.mkdtemp(path.join(os.tmpdir(), "zerobyte-restic-include-"));
includeFile = path.join(tmp, "include.txt");
await fs.writeFile(includeFile, options.includePatterns.join("\n"), "utf-8");
await fs.writeFile(includeFile, options.includePatterns.join("\n"), "utf-8");
args.push("--files-from", includeFile);
}
args.push("--files-from", includeFile);
}
if (options.includePaths?.length) {
const tmp = await fs.mkdtemp(path.join(os.tmpdir(), "zerobyte-restic-include-raw-"));
rawIncludeFile = path.join(tmp, "include.raw");
if (options.includePaths?.length) {
const tmp = await fs.mkdtemp(path.join(os.tmpdir(), "zerobyte-restic-include-raw-"));
rawIncludeFile = path.join(tmp, "include.raw");
await fs.writeFile(rawIncludeFile, Buffer.from(`${options.includePaths.join("\0")}\0`, "utf-8"));
await fs.writeFile(rawIncludeFile, Buffer.from(`${options.includePaths.join("\0")}\0`, "utf-8"));
args.push("--files-from-raw", rawIncludeFile);
}
args.push("--files-from-raw", rawIncludeFile);
}
for (const exclude of deps.defaultExcludes) {
args.push("--exclude", exclude);
}
for (const exclude of deps.defaultExcludes) {
args.push("--exclude", exclude);
}
let excludeFile: string | null = null;
if (options.exclude?.length) {
const tmp = await fs.mkdtemp(path.join(os.tmpdir(), "zerobyte-restic-exclude-"));
excludeFile = path.join(tmp, "exclude.txt");
let excludeFile: string | null = null;
if (options.exclude?.length) {
const tmp = await fs.mkdtemp(path.join(os.tmpdir(), "zerobyte-restic-exclude-"));
excludeFile = path.join(tmp, "exclude.txt");
await fs.writeFile(excludeFile, options.exclude.join("\n"), "utf-8");
await fs.writeFile(excludeFile, options.exclude.join("\n"), "utf-8");
args.push("--exclude-file", excludeFile);
}
args.push("--exclude-file", excludeFile);
}
if (options.excludeIfPresent?.length) {
for (const filename of options.excludeIfPresent) {
args.push("--exclude-if-present", filename);
}
}
if (options.excludeIfPresent?.length) {
for (const filename of options.excludeIfPresent) {
args.push("--exclude-if-present", filename);
}
}
if (options.customResticParams?.length) {
const validationError = validateCustomResticParams(options.customResticParams);
if (validationError) {
throw new Error(`Invalid customResticParams: ${validationError}`);
}
for (const param of options.customResticParams) {
const tokens = param.trim().split(/\s+/).filter(Boolean);
args.push(...tokens);
}
}
if (options.customResticParams?.length) {
const validationError = validateCustomResticParams(options.customResticParams);
if (validationError) {
throw new Error(`Invalid customResticParams: ${validationError}`);
}
for (const param of options.customResticParams) {
const tokens = param.trim().split(/\s+/).filter(Boolean);
args.push(...tokens);
}
}
addCommonArgs(args, env, config);
addCommonArgs(args, env, config);
if (usesSourceArg) {
args.push("--", source);
}
if (usesSourceArg) {
args.push("--", source);
}
const logData = throttle((data: string) => {
logger.info(data.trim());
}, 5000);
const stderrLines: string[] = [];
const logData = throttle((data: string) => {
logger.info(data.trim());
}, 5000);
const stderrLines: string[] = [];
const streamProgress = throttle((data: string) => {
if (options.onProgress) {
const streamProgress = throttle((data: string) => {
if (options.onProgress) {
try {
const jsonData = JSON.parse(data);
if (jsonData.message_type !== "status") {
return;
}
const progressResult = resticBackupProgressSchema.safeParse(jsonData);
if (progressResult.success) {
options.onProgress(progressResult.data);
} else {
logger.error(progressResult.error.message);
}
} catch {
// Ignore JSON parse errors for non-JSON lines
}
}
}, 1000);
logger.debug(`Executing: restic ${args.join(" ")}`);
const res = await safeSpawn({
command: "restic",
args,
env,
signal: options.signal,
onStdout: (data) => {
logData(data);
if (options.onProgress) {
streamProgress(data);
}
},
onStderr: (error) => {
const line = error.trim();
if (line.length > 0) {
stderrLines.push(line);
logger.error(`restic stderr: ${line}`);
}
},
});
if (includeFile) {
await fs.unlink(includeFile).catch(() => {});
}
if (rawIncludeFile) {
await fs.unlink(rawIncludeFile).catch(() => {});
}
if (excludeFile) {
await fs.unlink(excludeFile).catch(() => {});
}
await cleanupTemporaryKeys(env, deps);
if (options.signal?.aborted) {
logger.warn("Restic backup was aborted by signal.");
return { result: null, exitCode: res.exitCode, warningDetails: "Backup was stopped by the user" };
}
if (res.exitCode === 3) {
logger.error(`Restic backup encountered read errors: ${res.error}`);
}
if (res.exitCode !== 0 && res.exitCode !== 3) {
logger.error(`Restic backup failed: ${res.error}`);
logger.error(`Command executed: restic ${args.join(" ")}`);
throw new ResticError(res.exitCode, stderrLines.join("\n") || res.stderr || res.error);
}
const lastLine = res.summary.trim();
let summaryLine: unknown = {};
try {
const jsonData = JSON.parse(data);
if (jsonData.message_type !== "status") {
return;
}
const progressResult = resticBackupProgressSchema.safeParse(jsonData);
if (progressResult.success) {
options.onProgress(progressResult.data);
} else {
logger.error(progressResult.error.message);
}
summaryLine = JSON.parse(lastLine ?? "{}");
} catch {
// Ignore JSON parse errors for non-JSON lines
logger.warn("Failed to parse restic backup output JSON summary.", lastLine);
summaryLine = {};
}
}
}, 1000);
logger.debug(`Executing: restic ${args.join(" ")}`);
const res = await safeSpawn({
command: "restic",
args,
env,
signal: options.signal,
onStdout: (data) => {
logData(data);
if (options.onProgress) {
streamProgress(data);
logger.debug(`Restic backup output last line: ${JSON.stringify(summaryLine)}`);
const result = resticBackupOutputSchema.safeParse(summaryLine);
if (!result.success) {
logger.error(`Restic backup output validation failed: ${result.error.message}`);
return {
result: null,
exitCode: res.exitCode,
warningDetails: stderrLines.length > 0 ? stderrLines.join("\n") : null,
};
}
return {
result: result.data,
exitCode: res.exitCode,
warningDetails: stderrLines.length > 0 ? stderrLines.join("\n") : null,
};
},
onStderr: (error) => {
const line = error.trim();
if (line.length > 0) {
stderrLines.push(line);
logger.error(`restic stderr: ${line}`);
catch: (error) => {
if (error instanceof ResticError) {
return error;
}
return new ResticBackupCommandError({
cause: error,
message: toMessage(error),
});
},
});
if (includeFile) {
await fs.unlink(includeFile).catch(() => {});
}
if (rawIncludeFile) {
await fs.unlink(rawIncludeFile).catch(() => {});
}
if (excludeFile) {
await fs.unlink(excludeFile).catch(() => {});
}
await cleanupTemporaryKeys(env, deps);
if (options.signal?.aborted) {
logger.warn("Restic backup was aborted by signal.");
return { result: null, exitCode: res.exitCode, warningDetails: "Backup was stopped by the user" };
}
if (res.exitCode === 3) {
logger.error(`Restic backup encountered read errors: ${res.error}`);
}
if (res.exitCode !== 0 && res.exitCode !== 3) {
logger.error(`Restic backup failed: ${res.error}`);
logger.error(`Command executed: restic ${args.join(" ")}`);
throw new ResticError(res.exitCode, stderrLines.join("\n") || res.stderr || res.error);
}
const lastLine = res.summary.trim();
let summaryLine: unknown = {};
try {
summaryLine = JSON.parse(lastLine ?? "{}");
} catch {
logger.warn("Failed to parse restic backup output JSON summary.", lastLine);
summaryLine = {};
}
logger.debug(`Restic backup output last line: ${JSON.stringify(summaryLine)}`);
const result = resticBackupOutputSchema.safeParse(summaryLine);
if (!result.success) {
logger.error(`Restic backup output validation failed: ${result.error.message}`);
return {
result: null,
exitCode: res.exitCode,
warningDetails: stderrLines.length > 0 ? stderrLines.join("\n") : null,
};
}
return {
result: result.data,
exitCode: res.exitCode,
warningDetails: stderrLines.length > 0 ? stderrLines.join("\n") : null,
};
};

1
packages/core/tsconfig.json
View file

@ -1,5 +1,6 @@
{
"compilerOptions": {
"plugins": [{ "name": "@effect/language-service" }],
// Environment setup & latest features
"lib": ["ESNext"],
"target": "ESNext",

1
tsconfig.json
View file

@ -1,6 +1,7 @@
{
"include": ["app/**/*"],
"compilerOptions": {
"plugins": [{ "name": "@effect/language-service" }],
"lib": ["DOM", "DOM.Iterable", "ES2023"],
"types": ["bun", "node", "vite/client"],
"target": "ES2022",