diff --git a/.dockerignore b/.dockerignore index 09292fa2..ed763eb8 100644 --- a/.dockerignore +++ b/.dockerignore @@ -11,6 +11,7 @@ !**/components.json !app/** +!apps/agent/** !packages/** !public/** diff --git a/Dockerfile b/Dockerfile index 2feb66be..ee1bd691 100644 --- a/Dockerfile +++ b/Dockerfile @@ -64,6 +64,8 @@ COPY --from=deps /deps/shoutrrr /usr/local/bin/shoutrrr COPY ./package.json ./bun.lock ./ COPY ./packages/core/package.json ./packages/core/package.json +COPY ./packages/contracts/package.json ./packages/contracts/package.json +COPY ./apps/agent/package.json ./apps/agent/package.json RUN bun install --frozen-lockfile --ignore-scripts @@ -86,11 +88,14 @@ WORKDIR /app COPY ./package.json ./bun.lock ./ COPY ./packages/core/package.json ./packages/core/package.json +COPY ./packages/contracts/package.json ./packages/contracts/package.json +COPY ./apps/agent/package.json ./apps/agent/package.json RUN bun install --frozen-lockfile COPY . . RUN bun run build +RUN bun build apps/agent/src/index.ts --outfile .output/agent/index.mjs --target bun FROM base AS production diff --git a/app/server/modules/agents/__tests__/controller-agent-session.test.ts b/app/server/modules/agents/__tests__/controller-agent-session.test.ts new file mode 100644 index 00000000..c2a35dd5 --- /dev/null +++ b/app/server/modules/agents/__tests__/controller-agent-session.test.ts @@ -0,0 +1,121 @@ +import { expect, mock, test } from "bun:test"; +import { createAgentMessage } from "@zerobyte/contracts/agent-protocol"; +import { createControllerAgentSession } from "../controller-agent-session"; + +const createSocket = () => { + return { + data: { id: "connection-1", agentId: "local", organizationId: null, agentName: "Local Agent" }, + send: mock(() => undefined), + } as unknown as Parameters[0]; +}; + +test("close emits a synthetic backup.cancelled for a started backup", () => { + const onBackupCancelled = mock(() => undefined); + const session = createControllerAgentSession(createSocket(), { + onBackupCancelled, + }); + + session.handleMessage( + createAgentMessage("backup.started", { + jobId: "job-1", + scheduleId: "schedule-1", + }), + ); + + session.close(); + + expect(onBackupCancelled).toHaveBeenCalledTimes(1); + expect(onBackupCancelled).toHaveBeenCalledWith({ + jobId: "job-1", + scheduleId: "schedule-1", + message: + "The connection to the backup agent was lost while this backup was running. Restart the backup to ensure it completes.", + }); +}); + +test("close does not emit a synthetic backup.cancelled after a terminal event", () => { + for (const testCase of [ + { + jobId: "job-1", + scheduleId: "schedule-1", + terminalMessage: createAgentMessage("backup.completed", { + jobId: "job-1", + scheduleId: "schedule-1", + exitCode: 0, + result: null, + }), + expectedCancelledCalls: 0, + }, + { + jobId: "job-2", + scheduleId: "schedule-2", + terminalMessage: createAgentMessage("backup.failed", { + jobId: "job-2", + scheduleId: "schedule-2", + error: "backup failed", + }), + expectedCancelledCalls: 0, + }, + { + jobId: "job-3", + scheduleId: "schedule-3", + terminalMessage: createAgentMessage("backup.cancelled", { + jobId: "job-3", + scheduleId: "schedule-3", + message: "Backup was cancelled", + }), + expectedCancelledCalls: 1, + }, + ]) { + const onBackupCancelled = mock(() => undefined); + const session = createControllerAgentSession(createSocket(), { + onBackupCancelled, + }); + + session.handleMessage( + createAgentMessage("backup.started", { + jobId: testCase.jobId, + scheduleId: testCase.scheduleId, + }), + ); + session.handleMessage(testCase.terminalMessage); + session.close(); + + expect(onBackupCancelled).toHaveBeenCalledTimes(testCase.expectedCancelledCalls); + } +}); + +test("close emits a synthetic backup.cancelled for a queued backup", () => { + const onBackupCancelled = mock(() => undefined); + const session = createControllerAgentSession(createSocket(), { + onBackupCancelled, + }); + + session.sendBackup({ + jobId: "job-queued", + scheduleId: "schedule-queued", + organizationId: "org-1", + sourcePath: "/tmp/source", + repositoryConfig: { + backend: "local", + path: "/tmp/repository", + }, + options: {}, + runtime: { + password: "password", + cacheDir: "/tmp/cache", + passFile: "/tmp/pass", + defaultExcludes: [], + }, + }); + + session.close(); + + expect(onBackupCancelled).toHaveBeenCalledTimes(1); + expect(onBackupCancelled).toHaveBeenCalledWith({ + jobId: "job-queued", + scheduleId: "schedule-queued", + message: + "The connection to the backup agent was lost before this backup started. Restart the backup to ensure it completes.", + }); +}); diff --git a/app/server/modules/agents/agent-tokens.ts b/app/server/modules/agents/agent-tokens.ts new file mode 100644 index 00000000..febb9d94 --- /dev/null +++ b/app/server/modules/agents/agent-tokens.ts @@ -0,0 +1,12 @@ +import { cryptoUtils } from "~/server/utils/crypto"; + +export const deriveLocalAgentToken = async () => { + return cryptoUtils.deriveSecret("zerobyte:local-agent-token"); +}; + +export const validateAgentToken = async (token: string) => { + const localToken = await deriveLocalAgentToken(); + if (token === localToken) { + return { agentId: "local", organizationId: null, agentName: "local" }; + } +}; diff --git a/app/server/modules/agents/agents-manager.ts b/app/server/modules/agents/agents-manager.ts new file mode 100644 index 00000000..9df9ce71 --- /dev/null +++ b/app/server/modules/agents/agents-manager.ts @@ -0,0 +1,348 @@ +import { type ChildProcess, spawn } from "node:child_process"; +import { existsSync } from "node:fs"; +import path from "node:path"; +import { Effect, Exit, Ref, Scope } from "effect"; +import { logger } from "@zerobyte/core/node"; +import type { + BackupCancelPayload, + BackupCancelledPayload, + BackupCompletedPayload, + BackupFailedPayload, + BackupProgressPayload, + BackupRunPayload, + BackupStartedPayload, +} from "@zerobyte/contracts/agent-protocol"; +import { config } from "../../core/config"; +import { validateAgentToken, deriveLocalAgentToken } from "./agent-tokens"; +import { + createControllerAgentSession, + type AgentConnectionData, + type ControllerAgentSession, +} from "./controller-agent-session"; + +type AgentBackupEventContext = { + agentId: string; + agentName: string; + payload: + | BackupStartedPayload + | BackupProgressPayload + | BackupCompletedPayload + | BackupFailedPayload + | BackupCancelledPayload; +}; + +export type AgentBackupEventHandlers = { + onBackupStarted?: (context: AgentBackupEventContext & { payload: BackupStartedPayload }) => void; + onBackupProgress?: (context: AgentBackupEventContext & { payload: BackupProgressPayload }) => void; + onBackupCompleted?: (context: AgentBackupEventContext & { payload: BackupCompletedPayload }) => void; + onBackupFailed?: (context: AgentBackupEventContext & { payload: BackupFailedPayload }) => void; + onBackupCancelled?: (context: AgentBackupEventContext & { payload: BackupCancelledPayload }) => void; +}; + +type AgentManagerRuntime = ReturnType; +type AgentRuntimeState = { + agentManager: AgentManagerRuntime; + localAgent: ChildProcess | null; +}; + +type ProcessWithAgentRuntime = NodeJS.Process & { + __zerobyteAgentRuntime?: AgentRuntimeState; +}; + +const getAgentRuntimeState = () => { + const runtimeProcess = process as ProcessWithAgentRuntime; + const existingRuntime = runtimeProcess.__zerobyteAgentRuntime; + + if (existingRuntime) { + return existingRuntime; + } + + const runtime = { + agentManager: createAgentManagerRuntime(), + localAgent: null, + }; + + runtimeProcess.__zerobyteAgentRuntime = runtime; + return runtime; +}; + +const getAgentManagerRuntime = () => getAgentRuntimeState().agentManager; + +export const spawnLocalAgent = async () => { + await stopLocalAgent(); + + const sourceEntryPoint = path.join(process.cwd(), "apps", "agent", "src", "index.ts"); + const productionEntryPoint = path.join(process.cwd(), ".output", "agent", "index.mjs"); + + if (config.__prod__ && !existsSync(productionEntryPoint)) { + throw new Error(`Local agent entrypoint not found at ${productionEntryPoint}`); + } + + const agentEntryPoint = config.__prod__ ? productionEntryPoint : sourceEntryPoint; + const agentToken = await deriveLocalAgentToken(); + const args = config.__prod__ ? ["run", agentEntryPoint] : ["run", "--watch", agentEntryPoint]; + + const runtime = getAgentRuntimeState(); + const agentProcess = spawn("bun", args, { + env: { + PATH: process.env.PATH, + ZEROBYTE_CONTROLLER_URL: "ws://localhost:3001", + ZEROBYTE_AGENT_TOKEN: agentToken, + }, + stdio: ["ignore", "pipe", "pipe"], + }); + + runtime.localAgent = agentProcess; + + agentProcess.stdout?.on("data", (data: Buffer) => { + const line = data.toString().trim(); + if (line) logger.info(`[agent] ${line}`); + }); + + agentProcess.stderr?.on("data", (data: Buffer) => { + const line = data.toString().trim(); + if (line) logger.error(`[agent] ${line}`); + }); + + agentProcess.on("exit", (code, signal) => { + if (runtime.localAgent === agentProcess) { + runtime.localAgent = null; + } + logger.info(`Agent process exited with code ${code} and signal ${signal}`); + }); +}; + +export const stopLocalAgent = async () => { + const runtime = getAgentRuntimeState(); + if (!runtime.localAgent) { + return; + } + + const agentProcess = runtime.localAgent; + runtime.localAgent = null; + + if (agentProcess.exitCode !== null || agentProcess.signalCode !== null) { + return; + } + + const exited = new Promise((resolve) => { + agentProcess.once("exit", () => { + resolve(); + }); + }); + + agentProcess.kill(); + await exited; +}; + +const createAgentManagerRuntime = () => { + const sessionsRef = Effect.runSync(Ref.make>(new Map())); + const backupHandlersRef = Effect.runSync(Ref.make({})); + let runtimeScope: Scope.CloseableScope | null = null; + + const getSessions = () => Effect.runSync(Ref.get(sessionsRef)); + const getBackupHandlers = () => Effect.runSync(Ref.get(backupHandlersRef)); + const setSessions = (sessions: Map) => { + Effect.runSync(Ref.set(sessionsRef, sessions)); + }; + const setBackupHandlers = (handlers: AgentBackupEventHandlers) => { + Effect.runSync(Ref.set(backupHandlersRef, handlers)); + }; + + const closeAllSessions = () => { + const sessions = getSessions(); + for (const session of sessions.values()) { + session.close(); + } + setSessions(new Map()); + }; + + const getSession = (agentId: string) => getSessions().get(agentId); + + const setSession = (agentId: string, session: ControllerAgentSession) => { + const existingSession = getSession(agentId); + if (existingSession) { + existingSession.close(); + } + + const nextSessions = new Map(getSessions()); + nextSessions.set(agentId, session); + setSessions(nextSessions); + }; + + const removeSession = (agentId: string, connectionId: string) => { + const session = getSession(agentId); + if (!session || session.connectionId !== connectionId) { + return; + } + + session.close(); + const nextSessions = new Map(getSessions()); + nextSessions.delete(agentId); + setSessions(nextSessions); + }; + + const handleBackupStarted = (ws: Bun.ServerWebSocket, payload: BackupStartedPayload) => { + getBackupHandlers().onBackupStarted?.({ agentId: ws.data.agentId, agentName: ws.data.agentName, payload }); + }; + + const handleBackupProgress = (ws: Bun.ServerWebSocket, payload: BackupProgressPayload) => { + getBackupHandlers().onBackupProgress?.({ agentId: ws.data.agentId, agentName: ws.data.agentName, payload }); + }; + + const handleBackupCompleted = (ws: Bun.ServerWebSocket, payload: BackupCompletedPayload) => { + getBackupHandlers().onBackupCompleted?.({ agentId: ws.data.agentId, agentName: ws.data.agentName, payload }); + }; + + const handleBackupFailed = (ws: Bun.ServerWebSocket, payload: BackupFailedPayload) => { + getBackupHandlers().onBackupFailed?.({ agentId: ws.data.agentId, agentName: ws.data.agentName, payload }); + }; + + const handleBackupCancelled = (ws: Bun.ServerWebSocket, payload: BackupCancelledPayload) => { + getBackupHandlers().onBackupCancelled?.({ agentId: ws.data.agentId, agentName: ws.data.agentName, payload }); + }; + + const acquireServer = Effect.acquireRelease( + Effect.sync(() => + Bun.serve({ + port: 3001, + async fetch(req, srv) { + const url = new URL(req.url); + const token = url.searchParams.get("token"); + + if (!token) { + return new Response("Missing token", { status: 401 }); + } + + const result = await validateAgentToken(token); + if (!result) { + return new Response("Invalid or revoked token", { status: 401 }); + } + + const upgraded = srv.upgrade(req, { + data: { + id: Bun.randomUUIDv7(), + agentId: result.agentId, + organizationId: result.organizationId, + agentName: result.agentName, + }, + }); + if (upgraded) return undefined; + return new Response("WebSocket upgrade failed", { status: 400 }); + }, + websocket: { + open: (ws) => { + setSession( + ws.data.agentId, + createControllerAgentSession(ws, { + onBackupStarted: (payload) => handleBackupStarted(ws, payload), + onBackupProgress: (payload) => handleBackupProgress(ws, payload), + onBackupCompleted: (payload) => handleBackupCompleted(ws, payload), + onBackupFailed: (payload) => handleBackupFailed(ws, payload), + onBackupCancelled: (payload) => handleBackupCancelled(ws, payload), + }), + ); + logger.info(`Agent "${ws.data.agentName}" (${ws.data.agentId}) connected on ${ws.data.id}`); + }, + message: (ws, data) => { + if (typeof data !== "string") { + logger.warn(`Ignoring non-text message from agent ${ws.data.agentId}`); + return; + } + + const session = getSession(ws.data.agentId); + if (!session || session.connectionId !== ws.data.id) { + logger.warn(`No active session for agent ${ws.data.agentId} on ${ws.data.id}`); + return; + } + + session.handleMessage(data); + }, + close: (ws) => { + removeSession(ws.data.agentId, ws.data.id); + logger.info(`Agent "${ws.data.agentName}" (${ws.data.agentId}) disconnected`); + }, + }, + }), + ), + (server) => + Effect.sync(() => { + closeAllSessions(); + void server.stop(true); + }), + ); + + const stop = () => { + if (!runtimeScope) { + return; + } + + logger.info("Stopping Agent Manager..."); + const scope = runtimeScope; + runtimeScope = null; + Effect.runSync(Scope.close(scope, Exit.succeed(undefined))); + }; + + const start = () => { + if (runtimeScope) { + stop(); + } + + logger.info("Starting Agent Manager..."); + const scope = Effect.runSync(Scope.make()); + + try { + const server = Effect.runSync(Scope.extend(acquireServer, scope)); + runtimeScope = scope; + logger.info(`Agent Manager listening on port ${server.port}`); + } catch (error) { + Effect.runSync(Scope.close(scope, Exit.fail(error))); + throw error; + } + }; + + return { + start, + sendBackup: (agentId: string, payload: BackupRunPayload) => { + const session = getSession(agentId); + + if (!session) { + logger.warn(`Cannot send backup command. Agent ${agentId} is not connected.`); + return false; + } + + if (!session.isReady()) { + logger.warn(`Cannot send backup command. Agent ${agentId} is not ready.`); + return false; + } + + const jobId = session.sendBackup(payload); + logger.info(`Sent backup command ${jobId} to agent ${agentId} for schedule ${payload.scheduleId}`); + return true; + }, + cancelBackup: (agentId: string, payload: BackupCancelPayload) => { + const session = getSession(agentId); + + if (!session) { + logger.warn(`Cannot cancel backup command. Agent ${agentId} is not connected.`); + return false; + } + + session.sendBackupCancel(payload); + logger.info(`Sent backup cancel for command ${payload.jobId} to agent ${agentId}`); + return true; + }, + setBackupEventHandlers: (handlers: AgentBackupEventHandlers) => { + setBackupHandlers(handlers); + }, + getBackupEventHandlers: () => getBackupHandlers(), + stop, + }; +}; + +export const agentManager = getAgentManagerRuntime(); + +export const stopAgentRuntime = async () => { + getAgentManagerRuntime().stop(); + await stopLocalAgent(); +}; diff --git a/app/server/modules/agents/controller-agent-session.ts b/app/server/modules/agents/controller-agent-session.ts new file mode 100644 index 00000000..900aee72 --- /dev/null +++ b/app/server/modules/agents/controller-agent-session.ts @@ -0,0 +1,247 @@ +import { Effect, Fiber, Queue, Ref } from "effect"; +import { + createControllerMessage, + parseAgentMessage, + type AgentMessage, + type BackupCancelledPayload, + type BackupCompletedPayload, + type BackupFailedPayload, + type BackupProgressPayload, + type BackupRunPayload, + type BackupCancelPayload, + type BackupStartedPayload, + type ControllerWireMessage, +} from "@zerobyte/contracts/agent-protocol"; +import { logger } from "@zerobyte/core/node"; +import { toMessage } from "@zerobyte/core/utils"; + +export type AgentConnectionData = { + id: string; + agentId: string; + organizationId: string | null; + agentName: string; +}; + +type AgentSocket = Bun.ServerWebSocket; + +type SessionState = { + isReady: boolean; + lastSeenAt: number | null; + lastPongAt: number | null; +}; + +type ControllerAgentSessionHandlers = { + onBackupStarted?: (payload: BackupStartedPayload) => void; + onBackupProgress?: (payload: BackupProgressPayload) => void; + onBackupCompleted?: (payload: BackupCompletedPayload) => void; + onBackupFailed?: (payload: BackupFailedPayload) => void; + onBackupCancelled?: (payload: BackupCancelledPayload) => void; +}; + +export type ControllerAgentSession = { + readonly connectionId: string; + handleMessage: (data: string) => void; + sendBackup: (payload: BackupRunPayload) => string; + sendBackupCancel: (payload: BackupCancelPayload) => void; + isReady: () => boolean; + close: () => void; +}; + +export const createControllerAgentSession = ( + socket: AgentSocket, + handlers: ControllerAgentSessionHandlers = {}, +): ControllerAgentSession => { + const outboundQueue = Effect.runSync(Queue.bounded(64)); + const activeBackupJobs = Effect.runSync(Ref.make>(new Map())); + const pendingBackupJobs = Effect.runSync(Ref.make>(new Map())); + const state = Effect.runSync( + Ref.make({ + isReady: false, + lastSeenAt: null, + lastPongAt: null, + }), + ); + + const offerOutbound = (message: ControllerWireMessage) => { + void Effect.runPromise(Queue.offer(outboundQueue, message)).catch((error) => { + logger.error(`Failed to queue outbound message for agent ${socket.data.agentId}: ${toMessage(error)}`); + }); + }; + + const updateState = (update: (current: SessionState) => SessionState) => { + Effect.runSync(Ref.update(state, update)); + }; + + const setActiveBackupJob = (jobId: string, scheduleId: string) => { + Effect.runSync( + Ref.update(activeBackupJobs, (current) => { + const next = new Map(current); + next.set(jobId, scheduleId); + return next; + }), + ); + }; + + const setPendingBackupJob = (jobId: string, scheduleId: string) => { + Effect.runSync( + Ref.update(pendingBackupJobs, (current) => { + const next = new Map(current); + next.set(jobId, scheduleId); + return next; + }), + ); + }; + + const deleteActiveBackupJob = (jobId: string) => { + Effect.runSync( + Ref.update(activeBackupJobs, (current) => { + const next = new Map(current); + next.delete(jobId); + return next; + }), + ); + }; + + const deletePendingBackupJob = (jobId: string) => { + Effect.runSync( + Ref.update(pendingBackupJobs, (current) => { + const next = new Map(current); + next.delete(jobId); + return next; + }), + ); + }; + + const writerFiber = Effect.runFork( + Effect.forever( + Effect.gen(function* () { + const message = yield* Queue.take(outboundQueue); + yield* Effect.sync(() => { + try { + socket.send(message); + } catch (error) { + logger.error( + `Failed to send message to agent ${socket.data.agentId} on ${socket.data.id}: ${toMessage(error)}`, + ); + } + }); + }), + ), + ); + + const heartbeatFiber = Effect.runFork( + Effect.forever( + Effect.gen(function* () { + yield* Effect.sleep("15 seconds"); + yield* Queue.offer( + outboundQueue, + createControllerMessage("heartbeat.ping", { + sentAt: Date.now(), + }), + ); + }), + ), + ); + + const handleAgentMessage = (message: AgentMessage) => { + updateState((current) => ({ ...current, lastSeenAt: Date.now() })); + + switch (message.type) { + case "agent.ready": { + updateState((current) => ({ ...current, isReady: true })); + logger.info(`Agent "${socket.data.agentName}" (${socket.data.agentId}) is ready`); + break; + } + case "backup.started": { + deletePendingBackupJob(message.payload.jobId); + setActiveBackupJob(message.payload.jobId, message.payload.scheduleId); + logger.info( + `Backup ${message.payload.jobId} started on agent ${socket.data.agentId} for schedule ${message.payload.scheduleId}`, + ); + handlers.onBackupStarted?.(message.payload); + break; + } + case "backup.progress": { + handlers.onBackupProgress?.(message.payload); + break; + } + case "backup.completed": { + deletePendingBackupJob(message.payload.jobId); + deleteActiveBackupJob(message.payload.jobId); + handlers.onBackupCompleted?.(message.payload); + break; + } + case "backup.failed": { + deletePendingBackupJob(message.payload.jobId); + deleteActiveBackupJob(message.payload.jobId); + handlers.onBackupFailed?.(message.payload); + break; + } + case "backup.cancelled": { + deletePendingBackupJob(message.payload.jobId); + deleteActiveBackupJob(message.payload.jobId); + handlers.onBackupCancelled?.(message.payload); + break; + } + case "heartbeat.pong": { + updateState((current) => ({ ...current, lastPongAt: message.payload.sentAt })); + break; + } + } + }; + + return { + connectionId: socket.data.id, + handleMessage: (data: string) => { + const parsed = parseAgentMessage(data); + + if (parsed === null) { + logger.warn(`Invalid JSON from agent ${socket.data.agentId}`); + return; + } + + if (!parsed.success) { + logger.warn(`Invalid agent message from ${socket.data.agentId}: ${parsed.error.message}`); + return; + } + + handleAgentMessage(parsed.data); + }, + sendBackup: (payload) => { + setPendingBackupJob(payload.jobId, payload.scheduleId); + offerOutbound(createControllerMessage("backup.run", payload)); + return payload.jobId; + }, + sendBackupCancel: (payload) => { + offerOutbound(createControllerMessage("backup.cancel", payload)); + }, + isReady: () => Effect.runSync(Ref.get(state)).isReady, + close: () => { + updateState((current) => ({ ...current, isReady: false })); + const pendingJobs = Effect.runSync(Ref.get(pendingBackupJobs)); + Effect.runSync(Ref.set(pendingBackupJobs, new Map())); + for (const [jobId, scheduleId] of pendingJobs) { + handlers.onBackupCancelled?.({ + jobId, + scheduleId, + message: + "The connection to the backup agent was lost before this backup started. Restart the backup to ensure it completes.", + }); + } + + const activeJobs = Effect.runSync(Ref.get(activeBackupJobs)); + Effect.runSync(Ref.set(activeBackupJobs, new Map())); + for (const [jobId, scheduleId] of activeJobs) { + handlers.onBackupCancelled?.({ + jobId, + scheduleId, + message: + "The connection to the backup agent was lost while this backup was running. Restart the backup to ensure it completes.", + }); + } + void Effect.runPromise(Fiber.interrupt(writerFiber)).catch(() => {}); + void Effect.runPromise(Fiber.interrupt(heartbeatFiber)).catch(() => {}); + void Effect.runPromise(Queue.shutdown(outboundQueue)).catch(() => {}); + }, + }; +}; diff --git a/app/server/modules/backups/__tests__/backups.execution.test.ts b/app/server/modules/backups/__tests__/backups.service.execution.test.ts similarity index 69% rename from app/server/modules/backups/__tests__/backups.execution.test.ts rename to app/server/modules/backups/__tests__/backups.service.execution.test.ts index 2aafe12c..e8c021d1 100644 --- a/app/server/modules/backups/__tests__/backups.execution.test.ts +++ b/app/server/modules/backups/__tests__/backups.service.execution.test.ts @@ -1,7 +1,6 @@ import waitForExpect from "wait-for-expect"; import { afterEach, describe, expect, test, vi } from "vitest"; import { backupsService } from "../backups.service"; -import { backupsExecutionService } from "../backups.execution"; import { createTestVolume } from "~/test/helpers/volume"; import { createTestBackupSchedule } from "~/test/helpers/backup"; import { createTestRepository } from "~/test/helpers/repository"; @@ -13,8 +12,13 @@ import * as spawnModule from "@zerobyte/core/node"; import type { SafeSpawnParams } from "@zerobyte/core/node"; import { restic } from "~/server/core/restic"; import { NotFoundError, BadRequestError } from "http-errors-enhanced"; +import { fromAny } from "@total-typescript/shoehorn"; +import { scheduleQueries } from "../backups.queries"; import { repositoriesService } from "~/server/modules/repositories/repositories.service"; import { repoMutex } from "~/server/core/repository-mutex"; +import { agentManager } from "~/server/modules/agents/agents-manager"; +import { createAgentBackupMocks } from "~/test/helpers/agent-mock"; +import { getScheduleByIdOrShortId } from "../helpers/backup-schedule-lookups"; const setup = () => { const resticBackupMock = vi.fn((_: SafeSpawnParams) => @@ -22,6 +26,7 @@ const setup = () => { ); const resticForgetMock = vi.fn(() => Promise.resolve({ success: true, data: null })); const resticCopyMock = vi.fn(() => Promise.resolve({ success: true, output: "" })); + const { sendBackupMock, cancelBackupMock } = createAgentBackupMocks(resticBackupMock); const refreshStatsMock = vi.fn(() => Promise.resolve({ total_size: 0, @@ -37,12 +42,16 @@ const setup = () => { vi.spyOn(restic, "forget").mockImplementation(resticForgetMock); vi.spyOn(restic, "copy").mockImplementation(resticCopyMock); vi.spyOn(repositoriesService, "refreshRepositoryStats").mockImplementation(refreshStatsMock); + vi.spyOn(agentManager, "sendBackup").mockImplementation(sendBackupMock); + vi.spyOn(agentManager, "cancelBackup").mockImplementation(cancelBackupMock); vi.spyOn(context, "getOrganizationId").mockReturnValue(TEST_ORG_ID); return { resticBackupMock, resticForgetMock, resticCopyMock, + sendBackupMock, + cancelBackupMock, refreshStatsMock, }; }; @@ -63,7 +72,7 @@ describe("backup execution - validation failures", () => { }); // act - const result = await backupsExecutionService.validateBackupExecution(schedule.id); + const result = await backupsService.validateBackupExecution(schedule.id); // assert expect(result.type).toBe("failure"); @@ -74,10 +83,70 @@ describe("backup execution - validation failures", () => { expect(resticBackupMock).not.toHaveBeenCalled(); }); + test("should fail backup when volume does not exist", async () => { + // arrange + setup(); + const volume = await createTestVolume(); + const repository = await createTestRepository(); + const schedule = await createTestBackupSchedule({ + volumeId: volume.id, + repositoryId: repository.id, + }); + + const hydratedSchedule = await scheduleQueries.findById(schedule.id, TEST_ORG_ID); + expect(hydratedSchedule).toBeDefined(); + const scheduleWithoutVolume = { + ...hydratedSchedule, + volume: null, + }; + vi.spyOn(scheduleQueries, "findById").mockResolvedValueOnce(fromAny(scheduleWithoutVolume)); + + // act + const result = await backupsService.validateBackupExecution(schedule.id); + + // assert + expect(result.type).toBe("failure"); + if (result.type === "failure") { + expect(result.error).toBeInstanceOf(NotFoundError); + expect(result.error.message).toBe("Volume not found"); + expect(result.partialContext?.schedule).toBeDefined(); + } + }); + + test("should fail backup when repository does not exist", async () => { + // arrange + setup(); + const volume = await createTestVolume(); + const repository = await createTestRepository(); + const schedule = await createTestBackupSchedule({ + volumeId: volume.id, + repositoryId: repository.id, + }); + + const hydratedSchedule = await scheduleQueries.findById(schedule.id, TEST_ORG_ID); + expect(hydratedSchedule).toBeDefined(); + const scheduleWithoutRepository = { + ...hydratedSchedule, + repository: null, + }; + vi.spyOn(scheduleQueries, "findById").mockResolvedValueOnce(fromAny(scheduleWithoutRepository)); + + // act + const result = await backupsService.validateBackupExecution(schedule.id); + + // assert + expect(result.type).toBe("failure"); + if (result.type === "failure") { + expect(result.error).toBeInstanceOf(NotFoundError); + expect(result.error.message).toBe("Repository not found"); + expect(result.partialContext?.schedule).toBeDefined(); + expect(result.partialContext?.volume).toBeDefined(); + } + }); test("should fail backup when schedule does not exist", async () => { setup(); // act - const result = await backupsExecutionService.validateBackupExecution(99999); + const result = await backupsService.validateBackupExecution(99999); // assert expect(result.type).toBe("failure"); @@ -108,9 +177,9 @@ describe("stop backup", () => { }); }); - await backupsExecutionService.executeBackup(schedule.id); + await backupsService.executeBackup(schedule.id); - const updatedSchedule = await backupsService.getScheduleById(schedule.id); + const updatedSchedule = await getScheduleByIdOrShortId(schedule.id); expect(updatedSchedule.lastBackupStatus).toBe("warning"); expect(updatedSchedule.lastBackupError).toBe("error: open /mnt/data/private.db: permission denied"); }); @@ -136,15 +205,63 @@ describe("stop backup", () => { }); }); - await backupsExecutionService.executeBackup(schedule.id); + await backupsService.executeBackup(schedule.id); - const updatedSchedule = await backupsService.getScheduleById(schedule.id); + const updatedSchedule = await getScheduleByIdOrShortId(schedule.id); expect(updatedSchedule.lastBackupStatus).toBe("error"); expect(updatedSchedule.lastBackupError).toBe( "Permissions 0755 for '/tmp/zerobyte-ssh-key' are too open.\nThis private key will be ignored.", ); }); + test("should block forget on the same repository until the active backup completes", async () => { + const { resticBackupMock, resticForgetMock, sendBackupMock } = setup(); + const volume = await createTestVolume(); + const repository = await createTestRepository(); + const schedule = await createTestBackupSchedule({ + volumeId: volume.id, + repositoryId: repository.id, + retentionPolicy: { keepHourly: 24 }, + }); + + let completeBackup: (() => void) | undefined; + resticBackupMock.mockImplementationOnce( + () => + new Promise((resolve) => { + completeBackup = () => resolve({ exitCode: 0, summary: generateBackupOutput(), error: "" }); + }), + ); + + const backupPromise = backupsService.executeBackup(schedule.id); + + await waitForExpect(() => { + expect(sendBackupMock).toHaveBeenCalledTimes(1); + }); + + let forgetFinished = false; + const forgetPromise = backupsService.runForget(schedule.id).finally(() => { + forgetFinished = true; + }); + + await new Promise((resolve) => setTimeout(resolve, 50)); + + expect(resticForgetMock).not.toHaveBeenCalled(); + expect(forgetFinished).toBe(false); + + expect(completeBackup).toBeDefined(); + completeBackup?.(); + + await backupPromise; + await forgetPromise; + + expect(resticForgetMock).toHaveBeenCalled(); + expect(resticForgetMock).toHaveBeenCalledWith( + repository.config, + expect.objectContaining({ keepHourly: 24 }), + expect.objectContaining({ tag: schedule.shortId, organizationId: TEST_ORG_ID }), + ); + }); + test("should stop a running backup", async () => { // arrange const { resticBackupMock } = setup(); @@ -172,19 +289,19 @@ describe("stop backup", () => { }); }); - const executePromise = backupsExecutionService.executeBackup(schedule.id); + const executePromise = backupsService.executeBackup(schedule.id); await waitForExpect(async () => { - const runningSchedule = await backupsService.getScheduleById(schedule.id); + const runningSchedule = await getScheduleByIdOrShortId(schedule.id); expect(runningSchedule.lastBackupStatus).toBe("in_progress"); }); // act - await backupsExecutionService.stopBackup(schedule.id); + await backupsService.stopBackup(schedule.id); await executePromise; // assert - const updatedSchedule = await backupsService.getScheduleById(schedule.id); + const updatedSchedule = await getScheduleByIdOrShortId(schedule.id); expect(updatedSchedule.lastBackupStatus).toBe("warning"); expect(updatedSchedule.lastBackupError).toBe("Backup was stopped by the user"); }); @@ -198,36 +315,25 @@ describe("stop backup", () => { repositoryId: repository.id, }); - vi.spyOn(repoMutex, "acquireShared").mockImplementation((_repositoryId, _operation, signal) => { - return new Promise((_, reject) => { - if (signal?.aborted) { - reject(signal.reason instanceof Error ? signal.reason : new Error("Operation aborted")); - return; - } + const releaseLock = await repoMutex.acquireExclusive(repository.id, "test"); + const executePromise = backupsService.executeBackup(schedule.id); - signal?.addEventListener( - "abort", - () => { - reject(signal.reason instanceof Error ? signal.reason : new Error("Operation aborted")); - }, - { once: true }, - ); + try { + await waitForExpect(async () => { + const queuedSchedule = await getScheduleByIdOrShortId(schedule.id); + expect(queuedSchedule.lastBackupStatus).toBe("in_progress"); }); - }); - const executePromise = backupsExecutionService.executeBackup(schedule.id); + expect(resticBackupMock).not.toHaveBeenCalled(); - await waitForExpect(async () => { - const queuedSchedule = await backupsService.getScheduleById(schedule.id); - expect(queuedSchedule.lastBackupStatus).toBe("in_progress"); - }); + await backupsService.stopBackup(schedule.id); + } finally { + releaseLock(); + } - expect(resticBackupMock).not.toHaveBeenCalled(); - - await backupsExecutionService.stopBackup(schedule.id); await executePromise; - const updatedSchedule = await backupsService.getScheduleById(schedule.id); + const updatedSchedule = await getScheduleByIdOrShortId(schedule.id); expect(updatedSchedule.lastBackupStatus).toBe("warning"); expect(updatedSchedule.lastBackupError).toBe("Backup was stopped by the user"); expect(resticBackupMock).not.toHaveBeenCalled(); @@ -247,20 +353,40 @@ describe("stop backup", () => { }); // act & assert - await expect(backupsExecutionService.stopBackup(schedule.id)).rejects.toThrow( + await expect(backupsService.stopBackup(schedule.id)).rejects.toThrow( "No backup is currently running for this schedule", ); - const updatedSchedule = await backupsService.getScheduleById(schedule.id); + const updatedSchedule = await getScheduleByIdOrShortId(schedule.id); expect(updatedSchedule.lastBackupAt).toBe(previousLastBackupAt); expect(updatedSchedule.lastBackupStatus).toBe("warning"); expect(updatedSchedule.lastBackupError).toBe("Backup was stopped by the user"); }); + test("should reset a stuck in_progress status even when no backup is running", async () => { + // arrange + setup(); + const volume = await createTestVolume(); + const repository = await createTestRepository(); + const schedule = await createTestBackupSchedule({ + volumeId: volume.id, + repositoryId: repository.id, + lastBackupStatus: "in_progress", + }); + + // act + await backupsService.stopBackup(schedule.id).catch(() => {}); + + // assert + const updatedSchedule = await getScheduleByIdOrShortId(schedule.id); + expect(updatedSchedule.lastBackupStatus).toBe("warning"); + expect(updatedSchedule.lastBackupError).toBe("Backup was stopped by the user"); + }); + test("should throw NotFoundError when schedule does not exist", async () => { setup(); // act & assert - await expect(backupsExecutionService.stopBackup(99999)).rejects.toThrow("Backup schedule not found"); + await expect(backupsService.stopBackup(99999)).rejects.toThrow("Backup schedule not found"); }); }); @@ -281,7 +407,7 @@ describe("retention policy - runForget", () => { }); // act - await backupsExecutionService.runForget(schedule.id); + await backupsService.runForget(schedule.id); // assert expect(resticForgetMock).toHaveBeenCalledWith( @@ -310,7 +436,7 @@ describe("retention policy - runForget", () => { }); // act & assert - await expect(backupsExecutionService.runForget(schedule.id)).rejects.toThrow( + await expect(backupsService.runForget(schedule.id)).rejects.toThrow( "No retention policy configured for this schedule", ); }); @@ -318,7 +444,7 @@ describe("retention policy - runForget", () => { test("should throw NotFoundError when schedule does not exist", async () => { setup(); // act & assert - await expect(backupsExecutionService.runForget(99999)).rejects.toThrow("Backup schedule not found"); + await expect(backupsService.runForget(99999)).rejects.toThrow("Backup schedule not found"); }); test("should throw NotFoundError when repository does not exist", async () => { @@ -331,9 +457,7 @@ describe("retention policy - runForget", () => { }); // act & assert - await expect(backupsExecutionService.runForget(schedule.id, "non-existent-repo")).rejects.toThrow( - "Repository not found", - ); + await expect(backupsService.runForget(schedule.id, "non-existent-repo")).rejects.toThrow("Repository not found"); }); }); @@ -352,7 +476,7 @@ describe("mirror operations", () => { await createTestBackupScheduleMirror(schedule.id, mirrorRepository.id); // act - await backupsExecutionService.copyToMirrors(schedule.id, sourceRepository, null); + await backupsService.copyToMirrors(schedule.id, sourceRepository, null); // assert expect(resticCopyMock).toHaveBeenCalledWith( @@ -379,7 +503,7 @@ describe("mirror operations", () => { await createTestBackupScheduleMirror(schedule.id, mirrorRepository.id, { enabled: false }); // act - await backupsExecutionService.copyToMirrors(schedule.id, sourceRepository, null); + await backupsService.copyToMirrors(schedule.id, sourceRepository, null); // assert expect(resticCopyMock).not.toHaveBeenCalled(); @@ -399,7 +523,7 @@ describe("mirror operations", () => { const mirror = await createTestBackupScheduleMirror(schedule.id, mirrorRepository.id); // act - await backupsExecutionService.copyToMirrors(schedule.id, sourceRepository, null); + await backupsService.copyToMirrors(schedule.id, sourceRepository, null); // assert const mirrors = await backupsService.getMirrors(schedule.id); @@ -430,7 +554,7 @@ describe("mirror operations", () => { }); // act - await backupsExecutionService.copyToMirrors(schedule.id, sourceRepository, null); + await backupsService.copyToMirrors(schedule.id, sourceRepository, null); // assert const mirrors = await backupsService.getMirrors(schedule.id); @@ -457,7 +581,7 @@ describe("mirror operations", () => { resticCopyMock.mockImplementationOnce(() => Promise.reject(new Error("Copy failed"))); // act - await backupsExecutionService.copyToMirrors(schedule.id, sourceRepository, null); + await backupsService.copyToMirrors(schedule.id, sourceRepository, null); // assert const mirrors = await backupsService.getMirrors(schedule.id); @@ -485,7 +609,7 @@ describe("mirror operations", () => { resticCopyMock.mockImplementation(() => Promise.resolve({ success: true, output: "" })); // act - await backupsExecutionService.copyToMirrors(schedule.id, sourceRepository, schedule.retentionPolicy); + await backupsService.copyToMirrors(schedule.id, sourceRepository, schedule.retentionPolicy); await waitForExpect(() => { expect(resticCopyMock).toHaveBeenCalled(); @@ -516,7 +640,7 @@ describe("mirror operations", () => { resticForgetMock.mockClear(); // act - await backupsExecutionService.copyToMirrors(schedule.id, sourceRepository, schedule.retentionPolicy); + await backupsService.copyToMirrors(schedule.id, sourceRepository, schedule.retentionPolicy); await waitForExpect(() => { expect(resticCopyMock).toHaveBeenCalled(); @@ -559,10 +683,10 @@ describe("mirror operations", () => { ); resticCopyMock.mockImplementation(() => Promise.resolve({ success: true, output: "" })); - const firstCopyPromise = backupsExecutionService.copyToMirrors(firstSchedule.id, sourceRepository, null); + const firstCopyPromise = backupsService.copyToMirrors(firstSchedule.id, sourceRepository, null); await firstCopyStarted; - const secondCopyPromise = backupsExecutionService.copyToMirrors(secondSchedule.id, sourceRepository, null); + const secondCopyPromise = backupsService.copyToMirrors(secondSchedule.id, sourceRepository, null); try { const secondCopyState = await Promise.race<"resolved" | "timeout">([ diff --git a/app/server/modules/backups/__tests__/backups.service.test.ts b/app/server/modules/backups/__tests__/backups.service.test.ts index 357e5051..351bc224 100644 --- a/app/server/modules/backups/__tests__/backups.service.test.ts +++ b/app/server/modules/backups/__tests__/backups.service.test.ts @@ -12,11 +12,14 @@ import { db } from "~/server/db/db"; import { backupScheduleMirrorsTable, repositoriesTable, volumesTable } from "~/server/db/schema"; import { TEST_ORG_ID } from "~/test/helpers/organization"; import * as context from "~/server/core/request-context"; -import { backupsExecutionService } from "../backups.execution"; import { repositoriesService } from "~/server/modules/repositories/repositories.service"; +import { agentManager } from "~/server/modules/agents/agents-manager"; +import { createAgentBackupMocks } from "~/test/helpers/agent-mock"; +import { getScheduleByIdOrShortId } from "../helpers/backup-schedule-lookups"; const setup = () => { - const resticBackupMock = vi.fn(() => Promise.resolve({ exitCode: 0, summary: "", error: "" })); + const resticBackupMock = vi.fn((_: unknown) => Promise.resolve({ exitCode: 0, summary: "", error: "" })); + const { sendBackupMock, cancelBackupMock } = createAgentBackupMocks(resticBackupMock); const refreshStatsMock = vi.fn(() => Promise.resolve({ total_size: 0, @@ -29,10 +32,14 @@ const setup = () => { ); vi.spyOn(spawnModule, "safeSpawn").mockImplementation(resticBackupMock); vi.spyOn(repositoriesService, "refreshRepositoryStats").mockImplementation(refreshStatsMock); + vi.spyOn(agentManager, "sendBackup").mockImplementation(sendBackupMock); + vi.spyOn(agentManager, "cancelBackup").mockImplementation(cancelBackupMock); vi.spyOn(context, "getOrganizationId").mockReturnValue(TEST_ORG_ID); return { resticBackupMock, + sendBackupMock, + cancelBackupMock, refreshStatsMock, }; }; @@ -59,10 +66,10 @@ describe("execute backup", () => { ); // act - await backupsExecutionService.executeBackup(schedule.id); + await backupsService.executeBackup(schedule.id); // assert - const updatedSchedule = await backupsService.getScheduleById(schedule.id); + const updatedSchedule = await getScheduleByIdOrShortId(schedule.id); expect(updatedSchedule.nextBackupAt).not.toBeNull(); const nextBackupAt = new Date(updatedSchedule.nextBackupAt ?? 0); @@ -84,7 +91,7 @@ describe("execute backup", () => { }); // act - await backupsExecutionService.executeBackup(schedule.id); + await backupsService.executeBackup(schedule.id); // assert expect(resticBackupMock).not.toHaveBeenCalled(); @@ -106,11 +113,11 @@ describe("execute backup", () => { ); // act - await backupsExecutionService.executeBackup(schedule.id, true); + await backupsService.executeBackup(schedule.id, true); // assert expect(resticBackupMock).toHaveBeenCalled(); - const updatedSchedule = await backupsService.getScheduleById(schedule.id); + const updatedSchedule = await getScheduleByIdOrShortId(schedule.id); expect(updatedSchedule.lastBackupStatus).toBe("success"); expect(updatedSchedule.lastBackupAt).not.toBeNull(); }); @@ -132,10 +139,10 @@ describe("execute backup", () => { ); // act - await backupsExecutionService.executeBackup(schedule.id, true); + await backupsService.executeBackup(schedule.id, true); // assert - const updatedSchedule = await backupsService.getScheduleById(schedule.id); + const updatedSchedule = await getScheduleByIdOrShortId(schedule.id); expect(updatedSchedule.nextBackupAt).toBeNull(); }); @@ -155,13 +162,13 @@ describe("execute backup", () => { }); // act - void backupsExecutionService.executeBackup(schedule.id); + void backupsService.executeBackup(schedule.id); await waitForExpect(() => { expect(resticBackupMock).toHaveBeenCalledTimes(1); }); - await backupsExecutionService.executeBackup(schedule.id); + await backupsService.executeBackup(schedule.id); // assert expect(resticBackupMock).toHaveBeenCalledTimes(1); @@ -182,10 +189,10 @@ describe("execute backup", () => { ); // act - await backupsExecutionService.executeBackup(schedule.id); + await backupsService.executeBackup(schedule.id); // assert - const updatedSchedule = await backupsService.getScheduleById(schedule.id); + const updatedSchedule = await getScheduleByIdOrShortId(schedule.id); expect(updatedSchedule.lastBackupStatus).toBe("warning"); }); @@ -204,10 +211,10 @@ describe("execute backup", () => { ); // act - await backupsExecutionService.executeBackup(schedule.id); + await backupsService.executeBackup(schedule.id); // assert - const updatedSchedule = await backupsService.getScheduleById(schedule.id); + const updatedSchedule = await getScheduleByIdOrShortId(schedule.id); expect(updatedSchedule.lastBackupStatus).toBe("error"); }); }); @@ -229,7 +236,7 @@ describe("getSchedulesToExecute", () => { }); // act - const schedulesToExecute = await backupsExecutionService.getSchedulesToExecute(); + const schedulesToExecute = await backupsService.getSchedulesToExecute(); // assert expect(schedulesToExecute).toContain(schedule.id); @@ -246,7 +253,7 @@ describe("getScheduleByIdOrShortId", () => { repositoryId: repository.id, }); - const found = await backupsService.getScheduleByIdOrShortId(String(schedule.id)); + const found = await getScheduleByIdOrShortId(String(schedule.id)); expect(found.id).toBe(schedule.id); expect(found.shortId).toBe(schedule.shortId); @@ -261,7 +268,7 @@ describe("getScheduleByIdOrShortId", () => { repositoryId: repository.id, }); - const found = await backupsService.getScheduleByIdOrShortId(schedule.shortId); + const found = await getScheduleByIdOrShortId(schedule.shortId); expect(found.id).toBe(schedule.id); expect(found.shortId).toBe(schedule.shortId); @@ -274,10 +281,8 @@ describe("getScheduleByIdOrShortId", () => { organizationId: otherOrgId, }); - await expect(backupsService.getScheduleByIdOrShortId(schedule.shortId)).rejects.toThrow( - "Backup schedule not found", - ); - await expect(backupsService.getScheduleByIdOrShortId(schedule.id)).rejects.toThrow("Backup schedule not found"); + await expect(getScheduleByIdOrShortId(schedule.shortId)).rejects.toThrow("Backup schedule not found"); + await expect(getScheduleByIdOrShortId(schedule.id)).rejects.toThrow("Backup schedule not found"); }); }); diff --git a/app/server/modules/backups/backup-executor.ts b/app/server/modules/backups/backup-executor.ts index fa45a472..cf1418b8 100644 --- a/app/server/modules/backups/backup-executor.ts +++ b/app/server/modules/backups/backup-executor.ts @@ -1,11 +1,18 @@ -import { restic } from "../../core/restic"; -import type { BackupSchedule, Repository, Volume } from "../../db/schema"; -import type { ResticBackupOutputDto, ResticBackupProgressDto } from "@zerobyte/core/restic"; -import { createBackupOptions } from "./backup.helpers"; +import type { BackupSchedule, Volume, Repository } from "../../db/schema"; +import { logger } from "@zerobyte/core/node"; +import { resticDeps } from "../../core/restic"; +import type { ResticBackupOutputDto } from "@zerobyte/core/restic"; +import type { BackupProgressPayload, BackupRunPayload } from "@zerobyte/contracts/agent-protocol"; +import { agentManager } from "../agents/agents-manager"; import { getVolumePath } from "../volumes/helpers"; +import { decryptRepositoryConfig } from "../repositories/repository-config-secrets"; +import { createBackupOptions } from "./backup.helpers"; + +const LOCAL_AGENT_ID = "local"; type BackupExecutionRequest = { scheduleId: number; + jobId: string; schedule: BackupSchedule; volume: Volume; repository: Repository; @@ -14,7 +21,14 @@ type BackupExecutionRequest = { onProgress: (progress: BackupExecutionProgress) => void; }; -export type BackupExecutionProgress = ResticBackupProgressDto; +type ActiveBackupExecution = { + scheduleId: number; + scheduleShortId: string; + onProgress: (progress: BackupExecutionProgress) => void; + resolve: (result: BackupExecutionResult) => void; +}; + +export type BackupExecutionProgress = BackupProgressPayload["progress"]; export type BackupExecutionResult = | { @@ -29,15 +43,135 @@ export type BackupExecutionResult = } | { status: "failed"; - error: unknown; + error: string; } | { status: "cancelled"; message?: string; }; +const activeExecutionsByJobId = new Map(); +const activeExecutionJobIdsByScheduleId = new Map(); +const requestedCancellationsByScheduleId = new Set(); const activeControllersByScheduleId = new Map(); +const createBackupRunPayload = async ({ + jobId, + schedule, + volume, + repository, + organizationId, +}: BackupExecutionRequest): Promise => { + const sourcePath = getVolumePath(volume); + const { signal: _, ...options } = createBackupOptions(schedule, sourcePath); + const repositoryConfig = await decryptRepositoryConfig(repository.config); + const encryptedResticPassword = await resticDeps.getOrganizationResticPassword(organizationId); + const resticPassword = await resticDeps.resolveSecret(encryptedResticPassword); + + return { + jobId, + scheduleId: schedule.shortId, + organizationId, + sourcePath, + repositoryConfig, + options: { + ...options, + compressionMode: repository.compressionMode ?? "auto", + }, + runtime: { + password: resticPassword, + cacheDir: resticDeps.resticCacheDir, + passFile: resticDeps.resticPassFile, + defaultExcludes: resticDeps.defaultExcludes, + hostname: resticDeps.hostname, + }, + }; +}; + +const clearActiveExecution = (jobId: string) => { + const activeExecution = activeExecutionsByJobId.get(jobId); + if (!activeExecution) { + return null; + } + + activeExecutionsByJobId.delete(jobId); + activeExecutionJobIdsByScheduleId.delete(activeExecution.scheduleId); + return activeExecution; +}; + +const getActiveExecution = (jobId: string, scheduleId: string, eventName: string, executorId: string) => { + const activeExecution = activeExecutionsByJobId.get(jobId); + if (!activeExecution) { + logger.warn(`Received ${eventName} for unknown job ${jobId} from executor ${executorId}`); + return null; + } + + if (activeExecution.scheduleShortId !== scheduleId) { + logger.warn( + `Ignoring ${eventName} for job ${jobId} due to schedule mismatch ${scheduleId} from executor ${executorId}`, + ); + return null; + } + + return activeExecution; +}; + +agentManager.setBackupEventHandlers({ + onBackupStarted: ({ agentId, payload }) => { + getActiveExecution(payload.jobId, payload.scheduleId, "backup.started", agentId); + }, + onBackupProgress: ({ agentId, payload }) => { + const activeExecution = getActiveExecution(payload.jobId, payload.scheduleId, "backup.progress", agentId); + if (!activeExecution) { + return; + } + + activeExecution.onProgress(payload.progress); + }, + onBackupCompleted: ({ agentId, payload }) => { + const activeExecution = getActiveExecution(payload.jobId, payload.scheduleId, "backup.completed", agentId); + if (!activeExecution) { + return; + } + + requestedCancellationsByScheduleId.delete(activeExecution.scheduleId); + clearActiveExecution(payload.jobId); + activeExecution.resolve({ + status: "completed", + exitCode: payload.exitCode, + result: payload.result, + warningDetails: payload.warningDetails ?? null, + }); + }, + onBackupFailed: ({ agentId, payload }) => { + const activeExecution = getActiveExecution(payload.jobId, payload.scheduleId, "backup.failed", agentId); + if (!activeExecution) { + return; + } + + requestedCancellationsByScheduleId.delete(activeExecution.scheduleId); + clearActiveExecution(payload.jobId); + activeExecution.resolve({ + status: "failed", + error: payload.errorDetails ?? payload.error, + }); + }, + onBackupCancelled: ({ agentId, payload }) => { + const activeExecution = getActiveExecution(payload.jobId, payload.scheduleId, "backup.cancelled", agentId); + if (!activeExecution) { + return; + } + + const wasRequested = requestedCancellationsByScheduleId.has(activeExecution.scheduleId); + requestedCancellationsByScheduleId.delete(activeExecution.scheduleId); + clearActiveExecution(payload.jobId); + activeExecution.resolve({ + status: "cancelled", + message: wasRequested ? undefined : payload.message, + }); + }, +}); + export const backupExecutor = { track: (scheduleId: number) => { const abortController = new AbortController(); @@ -49,39 +183,69 @@ export const backupExecutor = { activeControllersByScheduleId.delete(scheduleId); } }, - execute: async (params: BackupExecutionRequest): Promise => { - const { schedule, volume, repository, organizationId, signal, onProgress } = params; - try { - 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, + execute: async (request: Omit) => { + const jobId = Bun.randomUUIDv7(); + const completion = new Promise((resolve) => { + activeExecutionsByJobId.set(jobId, { + scheduleId: request.scheduleId, + scheduleShortId: request.schedule.shortId, + onProgress: request.onProgress, + resolve, }); + activeExecutionJobIdsByScheduleId.set(request.scheduleId, jobId); + }); - return { - status: "completed", - exitCode: result.exitCode, - result: result.result, - warningDetails: result.warningDetails, - } satisfies BackupExecutionResult; + try { + if (request.signal.aborted) { + throw request.signal.reason || new Error("Operation aborted"); + } + + const payload = await createBackupRunPayload({ ...request, jobId }); + + if (request.signal.aborted) { + throw request.signal.reason || new Error("Operation aborted"); + } + + if (!agentManager.sendBackup(LOCAL_AGENT_ID, payload)) { + requestedCancellationsByScheduleId.delete(request.scheduleId); + clearActiveExecution(jobId); + return { + status: "unavailable", + error: new Error("Local backup agent is not connected"), + } satisfies BackupExecutionResult; + } + + return completion; } catch (error) { - return { - status: "failed", - error, - } satisfies BackupExecutionResult; + requestedCancellationsByScheduleId.delete(request.scheduleId); + clearActiveExecution(jobId); + throw error; } }, cancel: (scheduleId: number) => { const abortController = activeControllersByScheduleId.get(scheduleId); - if (!abortController) { + if (abortController) { + abortController.abort(); + } + + const jobId = activeExecutionJobIdsByScheduleId.get(scheduleId); + if (!jobId) { + return abortController !== undefined; + } + + const activeExecution = activeExecutionsByJobId.get(jobId); + if (!activeExecution) { + activeExecutionJobIdsByScheduleId.delete(scheduleId); + requestedCancellationsByScheduleId.delete(scheduleId); return false; } - abortController.abort(); + requestedCancellationsByScheduleId.add(scheduleId); + agentManager.cancelBackup(LOCAL_AGENT_ID, { + jobId, + scheduleId: activeExecution.scheduleShortId, + }); + return true; }, }; diff --git a/app/server/modules/backups/backups.execution.ts b/app/server/modules/backups/backups.execution.ts deleted file mode 100644 index f85388bf..00000000 --- a/app/server/modules/backups/backups.execution.ts +++ /dev/null @@ -1,11 +0,0 @@ -import { backupsService } from "./backups.service"; - -export const backupsExecutionService = { - executeBackup: backupsService.executeBackup, - validateBackupExecution: backupsService.validateBackupExecution, - getSchedulesToExecute: backupsService.getSchedulesToExecute, - stopBackup: backupsService.stopBackup, - runForget: backupsService.runForget, - copyToMirrors: backupsService.copyToMirrors, - getBackupProgress: backupsService.getBackupProgress, -}; diff --git a/app/server/modules/backups/backups.service.ts b/app/server/modules/backups/backups.service.ts index 3b9ff793..cd234ba3 100644 --- a/app/server/modules/backups/backups.service.ts +++ b/app/server/modules/backups/backups.service.ts @@ -37,14 +37,6 @@ const listSchedules = async () => { return schedules.filter((schedule) => schedule.volume && schedule.repository); }; -const getScheduleById = async (scheduleId: number) => { - return getScheduleByIdOrShortId(scheduleId); -}; - -const getScheduleByShortId = async (shortId: ShortId) => { - return getScheduleByIdOrShortId(shortId); -}; - const createSchedule = async (data: CreateBackupScheduleBody) => { const organizationId = getOrganizationId(); if (data.cronExpression && !isValidCron(data.cronExpression)) { @@ -477,9 +469,6 @@ const stopBackup = async (scheduleId: number) => { export const backupsService = { listSchedules, - getScheduleById, - getScheduleByShortId, - getScheduleByIdOrShortId, createSchedule, updateSchedule, deleteSchedule, diff --git a/app/server/modules/lifecycle/__tests__/shutdown.test.ts b/app/server/modules/lifecycle/__tests__/shutdown.test.ts new file mode 100644 index 00000000..f60839c5 --- /dev/null +++ b/app/server/modules/lifecycle/__tests__/shutdown.test.ts @@ -0,0 +1,58 @@ +import { afterEach, describe, expect, mock, spyOn, test } from "bun:test"; +import { Scheduler } from "../../../core/scheduler"; +import * as agentsManagerModule from "../../agents/agents-manager"; +import * as backendModule from "../../backends/backend"; +import type { VolumeBackend } from "../../backends/backend"; +import { createTestVolume } from "~/test/helpers/volume"; + +const loadShutdownModule = async () => { + const moduleUrl = new URL("../shutdown.ts", import.meta.url); + moduleUrl.searchParams.set("test", crypto.randomUUID()); + return import(moduleUrl.href); +}; + +afterEach(() => { + mock.restore(); +}); + +describe("shutdown", () => { + test("stops the agent runtime before unmounting mounted volumes", async () => { + const events: string[] = []; + const stopScheduler = mock(async () => { + events.push("scheduler.stop"); + }); + const stopAgentRuntime = mock(async () => { + events.push("agents.stop"); + }); + const unmountVolume = mock(async () => { + events.push("backend.unmount"); + return { status: "unmounted" as const }; + }); + + await createTestVolume({ + name: "Shutdown test volume", + config: { + backend: "directory", + path: "/Applications", + }, + status: "mounted", + }); + + spyOn(Scheduler, "stop").mockImplementation(stopScheduler); + spyOn(agentsManagerModule, "stopAgentRuntime").mockImplementation(stopAgentRuntime); + spyOn(backendModule, "createVolumeBackend").mockImplementation( + () => + ({ + mount: async () => ({ status: "mounted" as const }), + unmount: unmountVolume, + checkHealth: async () => ({ status: "mounted" as const }), + }) satisfies VolumeBackend, + ); + + const { shutdown } = await loadShutdownModule(); + + await shutdown(); + + expect(events).toEqual(["scheduler.stop", "agents.stop", "backend.unmount"]); + }); +}); diff --git a/app/server/modules/lifecycle/bootstrap.ts b/app/server/modules/lifecycle/bootstrap.ts index bb8f44ea..0866be58 100644 --- a/app/server/modules/lifecycle/bootstrap.ts +++ b/app/server/modules/lifecycle/bootstrap.ts @@ -1,4 +1,5 @@ import { runDbMigrations } from "../../db/db"; +import { agentManager, spawnLocalAgent } from "../agents/agents-manager"; import { runMigrations } from "./migrations"; import { startup } from "./startup"; @@ -7,6 +8,8 @@ let bootstrapPromise: Promise | undefined; const runBootstrap = async () => { await runDbMigrations(); await runMigrations(); + agentManager.start(); + await spawnLocalAgent(); await startup(); }; diff --git a/app/server/modules/lifecycle/shutdown.ts b/app/server/modules/lifecycle/shutdown.ts index 3bb76073..78c0ef21 100644 --- a/app/server/modules/lifecycle/shutdown.ts +++ b/app/server/modules/lifecycle/shutdown.ts @@ -2,9 +2,11 @@ import { Scheduler } from "../../core/scheduler"; import { db } from "../../db/db"; import { logger } from "@zerobyte/core/node"; import { createVolumeBackend } from "../backends/backend"; +import { stopAgentRuntime } from "../agents/agents-manager"; export const shutdown = async () => { await Scheduler.stop(); + await stopAgentRuntime(); const volumes = await db.query.volumesTable.findMany({ where: { status: "mounted" }, diff --git a/app/server/plugins/bootstrap.ts b/app/server/plugins/bootstrap.ts index 4dbc7c9e..c661205f 100644 --- a/app/server/plugins/bootstrap.ts +++ b/app/server/plugins/bootstrap.ts @@ -2,8 +2,20 @@ import { definePlugin } from "nitro"; import { bootstrapApplication } from "../modules/lifecycle/bootstrap"; import { logger } from "@zerobyte/core/node"; import { toMessage } from "../utils/errors"; +import { stopAgentRuntime } from "../modules/agents/agents-manager"; + +type ProcessWithAgentCloseHook = NodeJS.Process & { + __zerobyteAgentRuntimeCloseHookRegistered?: boolean; +}; + +export default definePlugin(async (nitroApp) => { + const runtimeProcess = process as ProcessWithAgentCloseHook; + + if (!runtimeProcess.__zerobyteAgentRuntimeCloseHookRegistered) { + nitroApp.hooks.hook("close", stopAgentRuntime); + runtimeProcess.__zerobyteAgentRuntimeCloseHookRegistered = true; + } -export default definePlugin(async () => { await bootstrapApplication().catch((err) => { logger.error(`Bootstrap failed: ${toMessage(err)}`); process.exit(1); diff --git a/app/test/helpers/agent-mock.ts b/app/test/helpers/agent-mock.ts new file mode 100644 index 00000000..bee28505 --- /dev/null +++ b/app/test/helpers/agent-mock.ts @@ -0,0 +1,105 @@ +import { mock } from "bun:test"; +import { fromAny } from "@total-typescript/shoehorn"; +import { agentManager } from "~/server/modules/agents/agents-manager"; + +export const createAgentBackupMocks = ( + resticBackupMock: (params: never) => Promise<{ + exitCode: number; + summary: string; + error: string; + stderr?: string; + }>, +) => { + const runningJobs = new Map(); + + const sendBackupMock = mock((_agentId: string, payload: { jobId: string; scheduleId: string }) => { + const handlers = agentManager.getBackupEventHandlers(); + + runningJobs.set(payload.jobId, { scheduleId: payload.scheduleId, cancelled: false }); + + handlers.onBackupStarted?.({ + agentId: "local", + agentName: "local", + payload: { jobId: payload.jobId, scheduleId: payload.scheduleId }, + }); + + void (async () => { + const stderrLines: string[] = []; + const result = await resticBackupMock( + fromAny({ + onStderr: (line: string) => { + stderrLines.push(line); + }, + }), + ); + const running = runningJobs.get(payload.jobId); + if (!running || running.cancelled) { + return; + } + + if (result.exitCode === 0 || result.exitCode === 3) { + let parsedResult: Record | null = null; + if (result.summary) { + try { + parsedResult = JSON.parse(result.summary) as Record; + } catch { + parsedResult = null; + } + } + + handlers.onBackupCompleted?.({ + agentId: "local", + agentName: "local", + payload: { + jobId: payload.jobId, + scheduleId: payload.scheduleId, + exitCode: result.exitCode, + result: fromAny(parsedResult), + warningDetails: stderrLines.join("\n") || undefined, + }, + }); + } else { + const resultWithStderr = result as typeof result & { stderr?: string }; + const errorDetails = stderrLines.join("\n") || resultWithStderr.stderr || result.error; + + handlers.onBackupFailed?.({ + agentId: "local", + agentName: "local", + payload: { + jobId: payload.jobId, + scheduleId: payload.scheduleId, + error: result.error || `Backup failed with code ${result.exitCode}`, + errorDetails, + }, + }); + } + + runningJobs.delete(payload.jobId); + })().catch(() => {}); + + return true; + }); + + const cancelBackupMock = mock((_agentId: string, payload: { jobId: string; scheduleId: string }) => { + const running = runningJobs.get(payload.jobId); + if (!running) { + return false; + } + + running.cancelled = true; + const handlers = agentManager.getBackupEventHandlers(); + handlers.onBackupCancelled?.({ + agentId: "local", + agentName: "local", + payload: { + jobId: payload.jobId, + scheduleId: payload.scheduleId, + message: "Backup was stopped by user", + }, + }); + runningJobs.delete(payload.jobId); + return true; + }); + + return { sendBackupMock, cancelBackupMock }; +}; diff --git a/apps/agent/.gitignore b/apps/agent/.gitignore new file mode 100644 index 00000000..a14702c4 --- /dev/null +++ b/apps/agent/.gitignore @@ -0,0 +1,34 @@ +# dependencies (bun install) +node_modules + +# output +out +dist +*.tgz + +# code coverage +coverage +*.lcov + +# logs +logs +_.log +report.[0-9]_.[0-9]_.[0-9]_.[0-9]_.json + +# dotenv environment variable files +.env +.env.development.local +.env.test.local +.env.production.local +.env.local + +# caches +.eslintcache +.cache +*.tsbuildinfo + +# IntelliJ based IDEs +.idea + +# Finder (MacOS) folder config +.DS_Store diff --git a/apps/agent/package.json b/apps/agent/package.json new file mode 100644 index 00000000..3751fd6a --- /dev/null +++ b/apps/agent/package.json @@ -0,0 +1,17 @@ +{ + "name": "agent", + "private": true, + "type": "module", + "module": "index.ts", + "dependencies": { + "@zerobyte/contracts": "workspace:*", + "@zerobyte/core": "workspace:*", + "effect": "^3.18.4" + }, + "devDependencies": { + "@types/bun": "latest" + }, + "peerDependencies": { + "typescript": "^5" + } +} diff --git a/apps/agent/src/commands/backup-cancel.ts b/apps/agent/src/commands/backup-cancel.ts new file mode 100644 index 00000000..2dec65dd --- /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.sync(() => { + const running = 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..a545c6c1 --- /dev/null +++ b/apps/agent/src/commands/backup-run.ts @@ -0,0 +1,108 @@ +import { Effect } 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 = context.getRunningJob(payload.jobId); + if (existing) { + 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(); + context.setRunningJob(payload.jobId, { scheduleId: payload.scheduleId, abortController }); + + 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); + + try { + const result = yield* Effect.tryPromise(() => + restic.backup(payload.repositoryConfig, payload.sourcePath, { + organizationId: payload.organizationId, + ...payload.options, + signal: abortController.signal, + onProgress: (progress) => { + context.offerOutbound( + createAgentMessage("backup.progress", { + jobId: payload.jobId, + scheduleId: payload.scheduleId, + progress, + }), + ); + }, + }), + ); + + if (abortController.signal.aborted) { + context.offerOutbound( + createAgentMessage("backup.cancelled", { + jobId: payload.jobId, + scheduleId: payload.scheduleId, + message: "Backup was cancelled", + }), + ); + return; + } + + context.offerOutbound( + createAgentMessage("backup.completed", { + jobId: payload.jobId, + scheduleId: payload.scheduleId, + exitCode: result.exitCode, + result: result.result, + warningDetails: result.warningDetails ?? undefined, + }), + ); + } catch (error) { + if (abortController.signal.aborted) { + context.offerOutbound( + createAgentMessage("backup.cancelled", { + jobId: payload.jobId, + scheduleId: payload.scheduleId, + message: "Backup was cancelled", + }), + ); + return; + } + + context.offerOutbound( + createAgentMessage("backup.failed", { + jobId: payload.jobId, + scheduleId: payload.scheduleId, + error: toMessage(error), + errorDetails: toErrorDetails(error), + }), + ); + } finally { + 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..1e7d34f8 --- /dev/null +++ b/apps/agent/src/commands/heartbeat-ping.ts @@ -0,0 +1,14 @@ +import { Effect } from "effect"; +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) => + Effect.sync(() => { + 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..87b76506 --- /dev/null +++ b/apps/agent/src/context.ts @@ -0,0 +1,13 @@ +import type { AgentWireMessage } from "@zerobyte/contracts/agent-protocol"; + +export type RunningJob = { + scheduleId: string; + abortController: AbortController; +}; + +export type ControllerCommandContext = { + getRunningJob: (jobId: string) => RunningJob | undefined; + setRunningJob: (jobId: string, job: RunningJob) => void; + deleteRunningJob: (jobId: string) => void; + offerOutbound: (message: AgentWireMessage) => void; +}; diff --git a/apps/agent/src/controller-session.ts b/apps/agent/src/controller-session.ts new file mode 100644 index 00000000..b2e43163 --- /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"; + +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) => Effect.runSync(Ref.get(runningJobsRef)).get(jobId); + + const setRunningJob = (jobId: string, job: { scheduleId: string; abortController: AbortController }) => { + Effect.runSync( + Ref.update(runningJobsRef, (current) => { + const next = new Map(current); + next.set(jobId, job); + return next; + }), + ); + }; + + const deleteRunningJob = (jobId: string) => { + Effect.runSync( + Ref.update(runningJobsRef, (current) => { + const next = new Map(current); + next.delete(jobId); + return next; + }), + ); + }; + + const offerOutbound = (message: AgentWireMessage) => { + void Effect.runPromise(Queue.offer(outboundQueue, message)).catch((error) => { + logger.error(`Failed to queue outbound controller message: ${toMessage(error)}`); + }); + }; + + const offerInbound = (message: ControllerWireMessage) => { + void Effect.runPromise(Queue.offer(inboundQueue, message)).catch((error) => { + logger.error(`Failed to queue inbound controller message: ${toMessage(error)}`); + }); + }; + + const commandContext = { + 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: () => { + offerOutbound(createAgentMessage("agent.ready", { agentId: "" })); + }, + onMessage: (data) => { + if (typeof data !== "string") { + logger.warn("Agent received a non-text message"); + return; + } + + offerInbound(data as ControllerWireMessage); + }, + close: () => { + const runningJobs = Effect.runSync(Ref.get(runningJobsRef)); + for (const running of runningJobs.values()) { + running.abortController.abort(); + } + Effect.runSync(Ref.set(runningJobsRef, new Map())); + 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..146fe4ed --- /dev/null +++ b/apps/agent/tsconfig.json @@ -0,0 +1,29 @@ +{ + "compilerOptions": { + // 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 764156eb..fe8c350e 100644 --- a/bun.lock +++ b/bun.lock @@ -34,6 +34,7 @@ "@tanstack/react-router": "^1.168.1", "@tanstack/react-router-ssr-query": "^1.166.10", "@tanstack/react-start": "^1.167.1", + "@zerobyte/contracts": "workspace:*", "@zerobyte/core": "workspace:*", "better-auth": "^1.5.5", "class-variance-authority": "^0.7.1", @@ -46,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", @@ -53,7 +55,7 @@ "http-errors-enhanced": "^4.0.2", "input-otp": "^1.4.2", "isbot": "^5.1.36", - "lucide-react": "^1.0.1", + "lucide-react": "^0.577.0", "next-themes": "^0.4.6", "qrcode.react": "^4.2.0", "react": "^19.2.4", @@ -114,6 +116,32 @@ "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": { + "@zerobyte/core": "workspace:*", + }, + "devDependencies": { + "@types/bun": "latest", + }, + "peerDependencies": { + "typescript": "^5", + }, + }, "packages/core": { "name": "@zerobyte/core", "devDependencies": { @@ -1044,12 +1072,16 @@ "@xmldom/xmldom": ["@xmldom/xmldom@0.8.11", "", {}, "sha512-cQzWCtO6C8TQiYl1ruKNn2U6Ao4o4WBBcbL61yJl84x+j5sOWWFU9X7DpND8XZG3daDppSsigMdfAIl2upQBRw=="], + "@zerobyte/contracts": ["@zerobyte/contracts@workspace:packages/contracts"], + "@zerobyte/core": ["@zerobyte/core@workspace:packages/core"], "abort-controller": ["abort-controller@3.0.0", "", { "dependencies": { "event-target-shim": "^5.0.0" } }, "sha512-h8lQ8tacZYnR3vNQTgibj+tODHI5/+l06Au2Pcriv/Gmet0eaj4TwWH41sO9wnHDiQsEj19q0drzdWdeAHtweg=="], "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=="], @@ -1544,7 +1576,7 @@ "lru.min": ["lru.min@1.1.4", "", {}, "sha512-DqC6n3QQ77zdFpCMASA1a3Jlb64Hv2N2DciFGkO/4L9+q/IpIAuRlKOvCXabtRW6cQf8usbmM6BE/TOPysCdIA=="], - "lucide-react": ["lucide-react@1.0.1", "", { "peerDependencies": { "react": "^16.5.1 || ^17.0.0 || ^18.0.0 || ^19.0.0" } }, "sha512-lih7tKEczCYOQjVEzpFuxEuNzlwf+1yhvlMlEkGWJM3va8Pugv8bYXc/pRtcjPncaP7k84X0Pt/71ufxvqEPtQ=="], + "lucide-react": ["lucide-react@0.577.0", "", { "peerDependencies": { "react": "^16.5.1 || ^17.0.0 || ^18.0.0 || ^19.0.0" } }, "sha512-4LjoFv2eEPwYDPg/CUdBJQSDfPyzXCRrVW1X7jrx/trgxnxkHFjnVZINbzvzxjN70dxychOfg+FTYwBiS3pQ5A=="], "luxon": ["luxon@3.7.2", "", {}, "sha512-vtEhXh/gNjI9Yg1u4jX/0YVPMvxzHuGgCm6tC5kZyb08yjGWGnqAjGJvcXbqQR2P3MyMEFnRbpcdFS6PBcLqew=="], diff --git a/package.json b/package.json index f17d48f0..946ce7f9 100644 --- a/package.json +++ b/package.json @@ -13,7 +13,7 @@ "start": "bun run .output/server/index.mjs", "preview": "bunx --bun vite preview", "cli:dev": "bun run app/server/cli/main.ts", - "cli": "ZEROBYTE_CLI=1 bun .output/server/_ssr/ssr.mjs", + "cli": "ZEROBYTE_CLI=1 bun .output/server/_ssr/index.mjs", "tsc": "tsc --noEmit && turbo run tsc", "start:dev": "docker compose down && docker compose up --build zerobyte-dev", "start:prod": "docker compose down && docker compose up --build zerobyte-prod", @@ -59,6 +59,7 @@ "@tanstack/react-router": "^1.168.1", "@tanstack/react-router-ssr-query": "^1.166.10", "@tanstack/react-start": "^1.167.1", + "@zerobyte/contracts": "workspace:*", "@zerobyte/core": "workspace:*", "better-auth": "^1.5.5", "class-variance-authority": "^0.7.1", @@ -71,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", @@ -78,7 +80,7 @@ "http-errors-enhanced": "^4.0.2", "input-otp": "^1.4.2", "isbot": "^5.1.36", - "lucide-react": "^1.0.1", + "lucide-react": "^0.577.0", "next-themes": "^0.4.6", "qrcode.react": "^4.2.0", "react": "^19.2.4", diff --git a/packages/contracts/.gitignore b/packages/contracts/.gitignore new file mode 100644 index 00000000..a14702c4 --- /dev/null +++ b/packages/contracts/.gitignore @@ -0,0 +1,34 @@ +# dependencies (bun install) +node_modules + +# output +out +dist +*.tgz + +# code coverage +coverage +*.lcov + +# logs +logs +_.log +report.[0-9]_.[0-9]_.[0-9]_.[0-9]_.json + +# dotenv environment variable files +.env +.env.development.local +.env.test.local +.env.production.local +.env.local + +# caches +.eslintcache +.cache +*.tsbuildinfo + +# IntelliJ based IDEs +.idea + +# Finder (MacOS) folder config +.DS_Store diff --git a/packages/contracts/README.md b/packages/contracts/README.md new file mode 100644 index 00000000..108c34c6 --- /dev/null +++ b/packages/contracts/README.md @@ -0,0 +1,15 @@ +# contracts + +To install dependencies: + +```bash +bun install +``` + +To run: + +```bash +bun run index.ts +``` + +This project was created using `bun init` in bun v1.3.10. [Bun](https://bun.com) is a fast all-in-one JavaScript runtime. diff --git a/packages/contracts/package.json b/packages/contracts/package.json new file mode 100644 index 00000000..ca203f5e --- /dev/null +++ b/packages/contracts/package.json @@ -0,0 +1,24 @@ +{ + "name": "@zerobyte/contracts", + "private": true, + "type": "module", + "exports": { + "./agent-protocol": { + "types": "./src/agent-protocol.ts", + "import": "./src/agent-protocol.ts", + "default": "./src/agent-protocol.ts" + } + }, + "scripts": { + "tsc": "tsc --noEmit" + }, + "dependencies": { + "@zerobyte/core": "workspace:*" + }, + "devDependencies": { + "@types/bun": "latest" + }, + "peerDependencies": { + "typescript": "^5" + } +} diff --git a/packages/contracts/src/agent-protocol.ts b/packages/contracts/src/agent-protocol.ts new file mode 100644 index 00000000..06b2f775 --- /dev/null +++ b/packages/contracts/src/agent-protocol.ts @@ -0,0 +1,219 @@ +import { z } from "zod"; +import { safeJsonParse } from "@zerobyte/core/utils"; +import { + repositoryConfigSchema, + resticBackupOutputSchema, + resticBackupProgressSchema, + type CompressionMode, +} from "@zerobyte/core/restic"; + +const compressionModeSchema = z.enum(["off", "auto", "max"]) satisfies z.ZodType; + +const backupExecutionOptionsSchema = z + .object({ + tags: z.array(z.string()).optional(), + oneFileSystem: z.boolean().optional(), + exclude: z.array(z.string()).optional(), + excludeIfPresent: z.array(z.string()).optional(), + includePaths: z.array(z.string()).optional(), + includePatterns: z.array(z.string()).optional(), + customResticParams: z.array(z.string()).optional(), + compressionMode: compressionModeSchema.optional(), + }) + .strict(); + +const backupRuntimeSchema = z + .object({ + password: z.string(), + cacheDir: z.string(), + passFile: z.string(), + defaultExcludes: z.array(z.string()), + hostname: z.string().optional(), + }) + .strict(); + +const backupRunSchema = z + .object({ + type: z.literal("backup.run"), + payload: z + .object({ + jobId: z.string(), + scheduleId: z.string(), + organizationId: z.string(), + sourcePath: z.string(), + repositoryConfig: repositoryConfigSchema, + options: backupExecutionOptionsSchema, + runtime: backupRuntimeSchema, + }) + .strict(), + }) + .strict(); + +const backupCancelSchema = z + .object({ + type: z.literal("backup.cancel"), + payload: z.object({ jobId: z.string(), scheduleId: z.string() }).strict(), + }) + .strict(); + +const heartbeatPingSchema = z + .object({ + type: z.literal("heartbeat.ping"), + payload: z.object({ sentAt: z.number() }), + }) + .strict(); + +const agentReadySchema = z + .object({ + type: z.literal("agent.ready"), + payload: z.object({ agentId: z.string() }), + }) + .strict(); + +const backupStartedSchema = z + .object({ + type: z.literal("backup.started"), + payload: z.object({ jobId: z.string(), scheduleId: z.string() }), + }) + .strict(); + +const backupProgressSchema = z + .object({ + type: z.literal("backup.progress"), + payload: z + .object({ + jobId: z.string(), + scheduleId: z.string(), + progress: resticBackupProgressSchema, + }) + .strict(), + }) + .strict(); + +const backupCompletedSchema = z + .object({ + type: z.literal("backup.completed"), + payload: z + .object({ + jobId: z.string(), + scheduleId: z.string(), + exitCode: z.number(), + result: resticBackupOutputSchema.nullable(), + warningDetails: z.string().optional(), + }) + .strict(), + }) + .strict(); + +const backupFailedSchema = z + .object({ + type: z.literal("backup.failed"), + payload: z + .object({ + jobId: z.string(), + scheduleId: z.string(), + error: z.string(), + errorDetails: z.string().optional(), + }) + .strict(), + }) + .strict(); + +const backupCancelledSchema = z + .object({ + type: z.literal("backup.cancelled"), + payload: z + .object({ + jobId: z.string(), + scheduleId: z.string(), + message: z.string().optional(), + }) + .strict(), + }) + .strict(); + +const heartbeatPongSchema = z + .object({ + type: z.literal("heartbeat.pong"), + payload: z.object({ sentAt: z.number() }), + }) + .strict(); + +const controllerMessageSchema = z.discriminatedUnion("type", [ + backupRunSchema, + backupCancelSchema, + heartbeatPingSchema, +]); +const agentMessageSchema = z.discriminatedUnion("type", [ + agentReadySchema, + backupStartedSchema, + backupProgressSchema, + backupCompletedSchema, + backupFailedSchema, + backupCancelledSchema, + heartbeatPongSchema, +]); + +export type BackupRunPayload = z.infer["payload"]; +export type BackupCancelPayload = z.infer["payload"]; +export type BackupStartedPayload = z.infer["payload"]; +export type BackupProgressPayload = z.infer["payload"]; +export type BackupCompletedPayload = z.infer["payload"]; +export type BackupFailedPayload = z.infer["payload"]; +export type BackupCancelledPayload = z.infer["payload"]; +export type ControllerMessage = z.infer; +export type AgentMessage = z.infer; + +type Brand = TValue & { + readonly __brand: TBrand; +}; + +export type ControllerWireMessage = Brand; +export type AgentWireMessage = Brand; + +type PayloadForMessage = Extract< + TMessage, + { type: TType } +>["payload"]; + +const parseJsonMessage = (data: string) => safeJsonParse(data); + +export const parseControllerMessage = (data: ControllerWireMessage) => { + const parsed = parseJsonMessage(data); + if (parsed === null) { + return null; + } + + return controllerMessageSchema.safeParse(parsed); +}; + +export const parseAgentMessage = (data: string) => { + const parsed = parseJsonMessage(data); + if (parsed === null) { + return null; + } + + return agentMessageSchema.safeParse(parsed); +}; + +export const createControllerMessage = ( + type: TType, + payload: PayloadForMessage, +) => + JSON.stringify( + controllerMessageSchema.parse({ + type, + payload, + }), + ) as ControllerWireMessage; + +export const createAgentMessage = ( + type: TType, + payload: PayloadForMessage, +) => + JSON.stringify( + agentMessageSchema.parse({ + type, + payload, + }), + ) as AgentWireMessage; diff --git a/packages/contracts/tsconfig.json b/packages/contracts/tsconfig.json new file mode 100644 index 00000000..146fe4ed --- /dev/null +++ b/packages/contracts/tsconfig.json @@ -0,0 +1,29 @@ +{ + "compilerOptions": { + // 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 + } +}