diff --git a/app/server/modules/backups/backup-executor.ts b/app/server/modules/backups/backup-executor.ts index fa45a472..60c9918c 100644 --- a/app/server/modules/backups/backup-executor.ts +++ b/app/server/modules/backups/backup-executor.ts @@ -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) => { diff --git a/apps/agent/package.json b/apps/agent/package.json new file mode 100644 index 00000000..c3bfdbb2 --- /dev/null +++ b/apps/agent/package.json @@ -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" + } +} 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..ec09d033 --- /dev/null +++ b/apps/agent/src/commands/backup-cancel.ts @@ -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(); + }); +}; diff --git a/apps/agent/src/commands/backup-run.test.ts b/apps/agent/src/commands/backup-run.test.ts new file mode 100644 index 00000000..70fc44ce --- /dev/null +++ b/apps/agent/src/commands/backup-run.test.ts @@ -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 = () => { + let resolve!: (value: T) => void; + const promise = new Promise((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(); + const setRunningJobGate = createDeferred(); + 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((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({ + 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({ + jobId: "job-1", + scheduleId: "schedule-1", + }); + + const runPromise = Effect.runPromise(handleBackupRunCommand(context, runPayload)); + + try { + const returnedBeforeRegistration = await Promise.race([ + runPromise.then(() => true), + new Promise((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 }); + } +}); diff --git a/apps/agent/src/commands/backup-run.ts b/apps/agent/src/commands/backup-run.ts new file mode 100644 index 00000000..a99b8f9f --- /dev/null +++ b/apps/agent/src/commands/backup-run.ts @@ -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(); + + 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..2030ffe5 --- /dev/null +++ b/apps/agent/src/commands/heartbeat-ping.ts @@ -0,0 +1,12 @@ +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) => { + return 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..9948d00e --- /dev/null +++ b/apps/agent/src/controller-session.ts @@ -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(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) => { + 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..1b545406 --- /dev/null +++ b/apps/agent/src/index.ts @@ -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 | 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(); +} diff --git a/apps/agent/tsconfig.json b/apps/agent/tsconfig.json new file mode 100644 index 00000000..a39e28eb --- /dev/null +++ b/apps/agent/tsconfig.json @@ -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 + } +} diff --git a/bun.lock b/bun.lock index cda43919..717d99ff 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": "^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=="], 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..e6c03119 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 { 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, - }; }; 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",