From 296e13ba1f97bdf25afe302f99d35979f0b1392b Mon Sep 17 00:00:00 2001 From: Nicolas Meienberger Date: Sun, 22 Mar 2026 14:20:14 +0100 Subject: [PATCH] refactor: split commands in separate files --- .../agents/controller-agent-session.ts | 9 +- apps/agent/src/commands/backup-cancel.ts | 20 +++ apps/agent/src/commands/backup-run.ts | 108 +++++++++++ apps/agent/src/commands/heartbeat-ping.ts | 14 ++ apps/agent/src/commands/index.ts | 19 ++ apps/agent/src/context.ts | 13 ++ apps/agent/src/controller-session.ts | 169 ++---------------- 7 files changed, 185 insertions(+), 167 deletions(-) create mode 100644 apps/agent/src/commands/backup-cancel.ts create mode 100644 apps/agent/src/commands/backup-run.ts create mode 100644 apps/agent/src/commands/heartbeat-ping.ts create mode 100644 apps/agent/src/commands/index.ts create mode 100644 apps/agent/src/context.ts diff --git a/app/server/modules/agents/controller-agent-session.ts b/app/server/modules/agents/controller-agent-session.ts index 6ec0e0e4..4a2fe63c 100644 --- a/app/server/modules/agents/controller-agent-session.ts +++ b/app/server/modules/agents/controller-agent-session.ts @@ -13,6 +13,7 @@ import { type ControllerWireMessage, } from "@zerobyte/contracts/agent-protocol"; import { logger } from "@zerobyte/core/node"; +import { toMessage } from "@zerobyte/core/utils"; type AgentConnectionData = { id: string; @@ -29,14 +30,6 @@ type SessionState = { lastPongAt: number | null; }; -const toMessage = (error: unknown) => { - if (error instanceof Error) { - return error.message; - } - - return String(error); -}; - type ControllerAgentSessionHandlers = { onBackupStarted?: (payload: BackupStartedPayload) => void; onBackupProgress?: (payload: BackupProgressPayload) => void; 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 index f64462e3..b2e43163 100644 --- a/apps/agent/src/controller-session.ts +++ b/apps/agent/src/controller-session.ts @@ -2,30 +2,12 @@ import { Effect, Fiber, Queue, Ref } from "effect"; import { createAgentMessage, parseControllerMessage, - type BackupRunPayload, type AgentWireMessage, - type BackupCancelPayload, type ControllerWireMessage, } from "@zerobyte/contracts/agent-protocol"; import { logger } from "@zerobyte/core/node"; -import { ResticError, type ResticDeps } from "@zerobyte/core/restic"; -import { createRestic } from "@zerobyte/core/restic/server"; - -const toMessage = (error: unknown) => { - if (error instanceof Error) { - return error.message; - } - - return String(error); -}; - -const toErrorDetails = (error: unknown) => { - if (error instanceof ResticError) { - return error.details || error.summary; - } - - return toMessage(error); -}; +import { toMessage } from "@zerobyte/core/utils"; +import { handleControllerCommand } from "./commands"; export type ControllerSession = { onOpen: () => void; @@ -74,6 +56,13 @@ export const createControllerSession = (ws: WebSocket): ControllerSession => { }); }; + const commandContext = { + getRunningJob, + setRunningJob, + deleteRunningJob, + offerOutbound, + }; + const writerFiber = Effect.runFork( Effect.forever( Effect.gen(function* () { @@ -105,145 +94,7 @@ export const createControllerSession = (ws: WebSocket): ControllerSession => { return; } - switch (parsed.data.type) { - case "backup.run": { - const runBackup = async (payload: BackupRunPayload) => { - const existing = getRunningJob(payload.jobId); - if (existing) { - 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(); - setRunningJob(payload.jobId, { scheduleId: payload.scheduleId, abortController }); - - 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 = await restic.backup(payload.repositoryConfig, payload.sourcePath, { - organizationId: payload.organizationId, - tags: payload.options.tags, - oneFileSystem: payload.options.oneFileSystem, - exclude: payload.options.exclude, - excludeIfPresent: payload.options.excludeIfPresent, - includePaths: payload.options.includePaths, - includePatterns: payload.options.includePatterns, - customResticParams: payload.options.customResticParams, - compressionMode: payload.options.compressionMode, - signal: abortController.signal, - onProgress: (progress) => { - offerOutbound( - createAgentMessage("backup.progress", { - jobId: payload.jobId, - scheduleId: payload.scheduleId, - progress, - }), - ); - }, - }); - - if (abortController.signal.aborted) { - offerOutbound( - createAgentMessage("backup.cancelled", { - jobId: payload.jobId, - scheduleId: payload.scheduleId, - message: "Backup was cancelled", - }), - ); - return; - } - - 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) { - offerOutbound( - createAgentMessage("backup.cancelled", { - jobId: payload.jobId, - scheduleId: payload.scheduleId, - message: "Backup was cancelled", - }), - ); - return; - } - - offerOutbound( - createAgentMessage("backup.failed", { - jobId: payload.jobId, - scheduleId: payload.scheduleId, - error: toMessage(error), - errorDetails: toErrorDetails(error), - }), - ); - } finally { - deleteRunningJob(payload.jobId); - } - }; - - void runBackup(parsed.data.payload); - break; - } - case "backup.cancel": { - const cancelBackup = (payload: BackupCancelPayload) => { - const running = 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(); - }; - - cancelBackup(parsed.data.payload); - break; - } - case "heartbeat.ping": { - yield* Queue.offer( - outboundQueue, - createAgentMessage("heartbeat.pong", { - sentAt: parsed.data.payload.sentAt, - }), - ); - break; - } - } + yield* handleControllerCommand(commandContext, parsed.data); }), ), );