diff --git a/apps/agent/package.json b/apps/agent/package.json new file mode 100644 index 00000000..40361ac1 --- /dev/null +++ b/apps/agent/package.json @@ -0,0 +1,20 @@ +{ + "name": "agent", + "private": true, + "type": "module", + "module": "index.ts", + "scripts": { + "tsc": "tsc --noEmit" + }, + "dependencies": { + "@zerobyte/contracts": "workspace:*", + "@zerobyte/core": "workspace:*", + "effect": "^3.18.4" + }, + "devDependencies": { + "@types/bun": "latest" + }, + "peerDependencies": { + "typescript": "^5" + } +} diff --git a/apps/agent/src/__tests__/controller-session.test.ts b/apps/agent/src/__tests__/controller-session.test.ts new file mode 100644 index 00000000..87d1fdbc --- /dev/null +++ b/apps/agent/src/__tests__/controller-session.test.ts @@ -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(); + } +}); diff --git a/apps/agent/src/commands/backup-cancel.ts b/apps/agent/src/commands/backup-cancel.ts new file mode 100644 index 00000000..c36c21e9 --- /dev/null +++ b/apps/agent/src/commands/backup-cancel.ts @@ -0,0 +1,20 @@ +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) => + 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(); + }); diff --git a/apps/agent/src/commands/backup-run.ts b/apps/agent/src/commands/backup-run.ts new file mode 100644 index 00000000..0719cd4c --- /dev/null +++ b/apps/agent/src/commands/backup-run.ts @@ -0,0 +1,112 @@ +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) => + Effect.fork( + 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 }); + + 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(); + + 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); diff --git a/apps/agent/src/commands/heartbeat-ping.ts b/apps/agent/src/commands/heartbeat-ping.ts new file mode 100644 index 00000000..7012141f --- /dev/null +++ b/apps/agent/src/commands/heartbeat-ping.ts @@ -0,0 +1,11 @@ +import { createAgentMessage, type ControllerMessage } from "@zerobyte/contracts/agent-protocol"; +import type { ControllerCommandContext } from "../context"; + +type HeartbeatPingPayload = Extract["payload"]; + +export const handleHeartbeatPingCommand = (context: ControllerCommandContext, payload: HeartbeatPingPayload) => + context.offerOutbound( + createAgentMessage("heartbeat.pong", { + sentAt: payload.sentAt, + }), + ); diff --git a/apps/agent/src/commands/index.ts b/apps/agent/src/commands/index.ts new file mode 100644 index 00000000..a17360df --- /dev/null +++ b/apps/agent/src/commands/index.ts @@ -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); + } + } +}; diff --git a/apps/agent/src/context.ts b/apps/agent/src/context.ts new file mode 100644 index 00000000..4daa68bc --- /dev/null +++ b/apps/agent/src/context.ts @@ -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; + setRunningJob: (jobId: string, job: RunningJob) => Effect.Effect; + deleteRunningJob: (jobId: string) => Effect.Effect; + offerOutbound: (message: AgentWireMessage) => Effect.Effect; +}; diff --git a/apps/agent/src/controller-session.ts b/apps/agent/src/controller-session.ts new file mode 100644 index 00000000..c1b45b9d --- /dev/null +++ b/apps/agent/src/controller-session.ts @@ -0,0 +1,126 @@ +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(64)); + const inboundQueue = Effect.runSync(Queue.bounded(64)); + const runningJobsRef = Effect.runSync(Ref.make>(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) => { + if (typeof data !== "string") { + logger.warn("Agent received a non-text message"); + return; + } + + 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(() => {}); + }, + }; +}; diff --git a/apps/agent/src/index.ts b/apps/agent/src/index.ts new file mode 100644 index 00000000..1e5b004d --- /dev/null +++ b/apps/agent/src/index.ts @@ -0,0 +1,47 @@ +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; + +class Agent { + private ws: WebSocket | null = null; + private controllerSession: ControllerSession | null = null; + + connect() { + 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.ws.onerror = (error) => { + logger.error("Agent encountered an error:", error); + }; + } +} + +const agent = new Agent(); +agent.connect(); diff --git a/apps/agent/tsconfig.json b/apps/agent/tsconfig.json new file mode 100644 index 00000000..4845a354 --- /dev/null +++ b/apps/agent/tsconfig.json @@ -0,0 +1,30 @@ +{ + "compilerOptions": { + "plugins": [{ "name": "@effect/language-service" }], + // Environment setup & latest features + "lib": ["ESNext"], + "target": "ESNext", + "module": "Preserve", + "moduleDetection": "force", + "jsx": "react-jsx", + "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 + } +} diff --git a/bun.lock b/bun.lock index cda43919..410a1230 100644 --- a/bun.lock +++ b/bun.lock @@ -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": "^5", + }, + }, "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=="], diff --git a/package.json b/package.json index c0903187..f32054fe 100644 --- a/package.json +++ b/package.json @@ -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", diff --git a/packages/core/src/restic/commands/__tests__/backup.test.ts b/packages/core/src/restic/commands/__tests__/backup.test.ts index fdf7aca8..d1345f0f 100644 --- a/packages/core/src/restic/commands/__tests__/backup.test.ts +++ b/packages/core/src/restic/commands/__tests__/backup.test.ts @@ -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) => Effect.runPromise(backup(...args)); +const runBackupError = (...args: Parameters) => 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", { diff --git a/packages/core/src/restic/commands/backup.ts b/packages/core/src/restic/commands/backup.ts index 1c4f3eeb..88949707 100644 --- a/packages/core/src/restic/commands/backup.ts +++ b/packages/core/src/restic/commands/backup.ts @@ -1,6 +1,7 @@ import fs from "node:fs/promises"; import os from "node:os"; import path from "node:path"; +import { Effect } from "effect"; import { throttle } from "es-toolkit"; import type { CompressionMode, RepositoryConfig } from "../schemas"; import { type ResticBackupProgressDto, resticBackupOutputSchema, resticBackupProgressSchema } from "../restic-dto"; @@ -13,7 +14,7 @@ import { ResticError } from "../error"; import { logger, safeSpawn } from "../../node"; import type { ResticDeps } from "../types"; -export const backup = async ( +export const backup = ( config: RepositoryConfig, source: string, options: { @@ -30,184 +31,188 @@ export const backup = async ( customResticParams?: string[]; }, deps: ResticDeps, -) => { - const repoUrl = buildRepoUrl(config); - const env = await buildEnv(config, options.organizationId, deps); +) => + 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); - } - }, - onStderr: (error) => { - const line = error.trim(); - if (line.length > 0) { - stderrLines.push(line); - logger.error(`restic stderr: ${line}`); + 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, + }; }, + catch: (error) => (error instanceof Error ? error : new Error(String(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, - }; -}; diff --git a/packages/core/tsconfig.json b/packages/core/tsconfig.json index 846638e1..4ac5f0a9 100644 --- a/packages/core/tsconfig.json +++ b/packages/core/tsconfig.json @@ -1,5 +1,6 @@ { "compilerOptions": { + "plugins": [{ "name": "@effect/language-service" }], // Environment setup & latest features "lib": ["ESNext"], "target": "ESNext", diff --git a/tsconfig.json b/tsconfig.json index 8bde7300..0aa3d75d 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -1,6 +1,7 @@ { "include": ["app/**/*"], "compilerOptions": { + "plugins": [{ "name": "@effect/language-service" }], "lib": ["DOM", "DOM.Iterable", "ES2023"], "types": ["bun", "node", "vite/client"], "target": "ES2022",