diff --git a/app/server/jobs/cleanup-dangling.ts b/app/server/jobs/cleanup-dangling.ts index b89ba0bd..a1c7eef7 100644 --- a/app/server/jobs/cleanup-dangling.ts +++ b/app/server/jobs/cleanup-dangling.ts @@ -5,7 +5,7 @@ import { volumeService } from "../modules/volumes/volume.service"; import { readMountInfo } from "../utils/mountinfo"; import { getVolumePath } from "../modules/volumes/helpers"; import { logger } from "@zerobyte/core/node"; -import { executeUnmount } from "../modules/backends/utils/backend-utils"; +import { executeUnmount } from "../../../apps/agent/src/volume-host/backends/utils"; import { toMessage } from "../utils/errors"; import { VOLUME_MOUNT_BASE } from "../core/constants"; import { db } from "../db/db"; diff --git a/app/server/modules/agents/__tests__/agents-manager.backups.test.ts b/app/server/modules/agents/__tests__/agents-manager.backups.test.ts index 25830203..1718cbcf 100644 --- a/app/server/modules/agents/__tests__/agents-manager.backups.test.ts +++ b/app/server/modules/agents/__tests__/agents-manager.backups.test.ts @@ -4,7 +4,7 @@ import { fromAny, fromPartial } from "@total-typescript/shoehorn"; import { Effect } from "effect"; import { agentManager, type ProcessWithAgentRuntime } from "../agents-manager"; import type { AgentManagerRuntime } from "../controller/server"; -import type { BackupRunPayload } from "@zerobyte/contracts/agent-protocol"; +import type { BackupRunPayload, VolumeCommand } from "@zerobyte/contracts/agent-protocol"; const setAgentRuntime = (agentManagerRuntime: Partial | null) => { (process as ProcessWithAgentRuntime).__zerobyteAgentRuntime = { @@ -46,3 +46,30 @@ test("cancelBackup resolves a running backup when the cancel command cannot be d scheduleId: "schedule-1", }); }); + +test("runVolumeCommand sends the command to the selected agent", async () => { + const runVolumeCommand = vi.fn().mockResolvedValue({ + commandId: "command-1", + status: "success", + command: { name: "volume.mount", result: { status: "mounted" } }, + }); + setAgentRuntime({ runVolumeCommand }); + + const command = fromPartial({ name: "volume.mount", volume: { agentId: "agent-1" } }); + + await expect(agentManager.runVolumeCommand("agent-1", command)).resolves.toEqual({ + name: "volume.mount", + result: { status: "mounted" }, + }); + expect(runVolumeCommand).toHaveBeenCalledWith("agent-1", command); +}); + +test("runVolumeCommand fails when the selected agent is unavailable", async () => { + setAgentRuntime(null); + + const command = fromPartial({ name: "volume.mount", volume: { agentId: "agent-1" } }); + + await expect(agentManager.runVolumeCommand("agent-1", command)).rejects.toThrow( + "Volume agent agent-1 is not connected", + ); +}); diff --git a/app/server/modules/agents/agents-manager.ts b/app/server/modules/agents/agents-manager.ts index e82b8dde..962a4e03 100644 --- a/app/server/modules/agents/agents-manager.ts +++ b/app/server/modules/agents/agents-manager.ts @@ -1,5 +1,5 @@ import { logger } from "@zerobyte/core/node"; -import type { BackupRunPayload } from "@zerobyte/contracts/agent-protocol"; +import type { BackupRunPayload, VolumeCommand, VolumeCommandResult } from "@zerobyte/contracts/agent-protocol"; import { Effect } from "effect"; import { config } from "../../core/config"; import { createAgentManagerRuntime, type AgentManagerEvent } from "./controller/server"; @@ -266,6 +266,23 @@ export const agentManager = { cancelBackup: async (agentId: string, scheduleId: number) => { return requestBackupCancellation(agentId, scheduleId); }, + runVolumeCommand: async (agentId: string, command: VolumeCommand): Promise => { + const runtime = getAgentManagerRuntime(); + if (!runtime) { + throw new Error(`Volume agent ${agentId} is not connected`); + } + + const response = await Effect.runPromise(runtime.runVolumeCommand(agentId, command)); + if (!response) { + throw new Error(`Failed to send volume command ${command.name} to agent ${agentId}`); + } + + if (response.status === "error") { + throw new Error(response.error); + } + + return response.command; + }, }; export const startLocalAgent = async () => { diff --git a/app/server/modules/agents/controller/server.ts b/app/server/modules/agents/controller/server.ts index d5201fa4..995ba237 100644 --- a/app/server/modules/agents/controller/server.ts +++ b/app/server/modules/agents/controller/server.ts @@ -1,7 +1,13 @@ import { Data, Effect, Exit, Fiber, Scope } from "effect"; import { logger } from "@zerobyte/core/node"; import { toMessage } from "@zerobyte/core/utils"; -import type { AgentMessage, BackupCancelPayload, BackupRunPayload } from "@zerobyte/contracts/agent-protocol"; +import type { + AgentMessage, + BackupCancelPayload, + BackupRunPayload, + VolumeCommand, + VolumeCommandResponsePayload, +} from "@zerobyte/contracts/agent-protocol"; import { createControllerAgentSession, type AgentConnectionData, @@ -319,10 +325,30 @@ export function createAgentManagerRuntime(onEvent: (event: AgentManagerEvent) => logger.warn(`Cannot cancel backup command. Agent ${agentId} is no longer accepting commands.`); return false; } - logger.info(`Sent backup cancel for command ${payload.jobId} to agent ${agentId}`); return true; }), + runVolumeCommand: ( + agentId: string, + command: VolumeCommand, + ): Effect.Effect => + Effect.gen(function* () { + const session = getSession(agentId); + + if (!session) { + yield* logger.effect.warn(`Cannot send volume command ${command.name}. Agent ${agentId} is not connected.`); + return null; + } + + if (!(yield* session.isReady())) { + yield* logger.effect.warn(`Cannot send volume command ${command.name}. Agent ${agentId} is not ready.`); + return null; + } + + const result = yield* session.runVolumeCommand(command); + yield* logger.effect.info(`Completed volume command ${command.name} on agent ${agentId}`); + return result; + }), stop, }; } diff --git a/app/server/modules/agents/controller/session.ts b/app/server/modules/agents/controller/session.ts index c2c3d9c7..cd719efd 100644 --- a/app/server/modules/agents/controller/session.ts +++ b/app/server/modules/agents/controller/session.ts @@ -7,6 +7,8 @@ import { type BackupCancelPayload, type BackupRunPayload, type ControllerWireMessage, + type VolumeCommand, + type VolumeCommandResponsePayload, } from "@zerobyte/contracts/agent-protocol"; import { logger } from "@zerobyte/core/node"; import { toMessage } from "@zerobyte/core/utils"; @@ -27,13 +29,24 @@ type SessionState = { lastPongAt: number | null; }; -export type ControllerAgentSessionEvent = AgentMessage | { type: "agent.disconnected" }; +type PendingVolumeCommand = { + resolve: (payload: VolumeCommandResponsePayload) => void; + reject: (error: Error) => void; + timeout: ReturnType; +}; + +export type ControllerAgentSessionEvent = + | Exclude + | { + type: "agent.disconnected"; + }; export type ControllerAgentSession = { readonly connectionId: string; handleMessage: (data: string) => Effect.Effect; sendBackup: (payload: BackupRunPayload) => Effect.Effect; sendBackupCancel: (payload: BackupCancelPayload) => Effect.Effect; + runVolumeCommand: (command: VolumeCommand) => Effect.Effect; isReady: () => Effect.Effect; run: Effect.Effect; }; @@ -45,6 +58,7 @@ export const createControllerAgentSession = ( Effect.gen(function* () { let isClosed = false; const outboundQueue = yield* Queue.bounded(64); + const pendingVolumeCommands = new Map(); const state = yield* Ref.make({ isReady: false, lastSeenAt: null, @@ -63,9 +77,18 @@ export const createControllerAgentSession = ( const updateState = (update: (current: SessionState) => SessionState) => Ref.update(state, update); + const rejectPendingVolumeCommands = () => { + for (const [commandId, pending] of pendingVolumeCommands) { + clearTimeout(pending.timeout); + pending.reject(new Error(`Agent session closed before volume command ${commandId} completed`)); + } + pendingVolumeCommands.clear(); + }; + const releaseSession = Effect.gen(function* () { const disconnectedAt = Date.now(); yield* updateState((current) => ({ ...current, isReady: false, lastSeenAt: disconnectedAt })); + yield* Effect.sync(rejectPendingVolumeCommands); yield* onEvent({ type: "agent.disconnected" }); yield* Queue.shutdown(outboundQueue); @@ -129,20 +152,43 @@ export const createControllerAgentSession = ( return yield* Effect.never; }); + const handleVolumeCommandResult = (payload: VolumeCommandResponsePayload) => { + const pending = pendingVolumeCommands.get(payload.commandId); + if (!pending) { + logger.warn(`Received response for unknown volume command ${payload.commandId}`); + return; + } + + pendingVolumeCommands.delete(payload.commandId); + clearTimeout(pending.timeout); + pending.resolve(payload); + }; + const handleAgentMessage = (message: AgentMessage) => Effect.gen(function* () { - if (message.type === "agent.ready") { - const readyAt = Date.now(); - yield* updateState((current) => ({ ...current, isReady: true, lastSeenAt: readyAt })); - yield* logger.effect.info(`Agent "${socket.data.agentName}" (${socket.data.agentId}) is ready`); + switch (message.type) { + case "agent.ready": { + const readyAt = Date.now(); + yield* updateState((current) => ({ ...current, isReady: true, lastSeenAt: readyAt })); + yield* logger.effect.info(`Agent "${socket.data.agentName}" (${socket.data.agentId}) is ready`); + yield* onEvent(message); + break; + } + case "heartbeat.pong": { + const seenAt = Date.now(); + yield* updateState((current) => ({ ...current, lastSeenAt: seenAt, lastPongAt: message.payload.sentAt })); + yield* onEvent(message); + break; + } + case "volume.commandResult": { + yield* Effect.sync(() => handleVolumeCommandResult(message.payload)); + break; + } + default: { + yield* onEvent(message); + break; + } } - - if (message.type === "heartbeat.pong") { - const seenAt = Date.now(); - yield* updateState((current) => ({ ...current, lastSeenAt: seenAt, lastPongAt: message.payload.sentAt })); - } - - yield* onEvent(message); }); return { @@ -166,6 +212,37 @@ export const createControllerAgentSession = ( }, sendBackup: (payload) => offerOutbound(createControllerMessage("backup.run", payload)), sendBackupCancel: (payload) => offerOutbound(createControllerMessage("backup.cancel", payload)), + runVolumeCommand: (command) => { + return Effect.tryPromise({ + try: async () => { + const commandId = Bun.randomUUIDv7(); + const response = new Promise((resolve, reject) => { + const timeout = setTimeout(() => { + pendingVolumeCommands.delete(commandId); + reject(new Error(`Volume command ${command.name} timed out`)); + }, 60_000); + + pendingVolumeCommands.set(commandId, { resolve, reject, timeout }); + }); + + const queued = await Effect.runPromise( + offerOutbound(createControllerMessage("volume.command", { commandId, command })), + ); + + if (!queued) { + const pending = pendingVolumeCommands.get(commandId); + if (pending) { + clearTimeout(pending.timeout); + pendingVolumeCommands.delete(commandId); + } + throw new Error(`Failed to queue volume command ${command.name}`); + } + + return response; + }, + catch: (error) => (error instanceof Error ? error : new Error(toMessage(error))), + }); + }, isReady: () => Ref.get(state).pipe(Effect.map((current) => current.isReady)), run, }; diff --git a/app/server/modules/backends/backend.ts b/app/server/modules/backends/backend.ts deleted file mode 100644 index f6eb3e3a..00000000 --- a/app/server/modules/backends/backend.ts +++ /dev/null @@ -1,46 +0,0 @@ -import type { BackendStatus } from "~/schemas/volumes"; -import type { Volume } from "../../db/schema"; -import { getVolumePath } from "../volumes/helpers"; -import { makeDirectoryBackend } from "./directory/directory-backend"; -import { makeNfsBackend } from "./nfs/nfs-backend"; -import { makeRcloneBackend } from "./rclone/rclone-backend"; -import { makeSmbBackend } from "./smb/smb-backend"; -import { makeWebdavBackend } from "./webdav/webdav-backend"; -import { makeSftpBackend } from "./sftp/sftp-backend"; - -type OperationResult = { - error?: string; - status: BackendStatus; -}; - -export type VolumeBackend = { - mount: () => Promise; - unmount: () => Promise; - checkHealth: () => Promise; -}; - -export const createVolumeBackend = (volume: Volume, mountPath = getVolumePath(volume)): VolumeBackend => { - switch (volume.config.backend) { - case "nfs": { - return makeNfsBackend(volume.config, mountPath); - } - case "smb": { - return makeSmbBackend(volume.config, mountPath); - } - case "directory": { - return makeDirectoryBackend(volume.config, mountPath); - } - case "webdav": { - return makeWebdavBackend(volume.config, mountPath); - } - case "rclone": { - return makeRcloneBackend(volume.config, mountPath); - } - case "sftp": { - return makeSftpBackend(volume.config, mountPath); - } - default: { - throw new Error("Unsupported backend"); - } - } -}; diff --git a/app/server/modules/backends/nfs/nfs-backend.ts b/app/server/modules/backends/nfs/nfs-backend.ts deleted file mode 100644 index 2f0fafdc..00000000 --- a/app/server/modules/backends/nfs/nfs-backend.ts +++ /dev/null @@ -1,135 +0,0 @@ -import * as fs from "node:fs/promises"; -import * as os from "node:os"; -import { BACKEND_STATUS, type BackendConfig } from "~/schemas/volumes"; -import { OPERATION_TIMEOUT } from "../../../core/constants"; -import { toMessage } from "../../../utils/errors"; -import { logger } from "@zerobyte/core/node"; -import { getMountForPath } from "../../../utils/mountinfo"; -import { withTimeout } from "../../../utils/timeout"; -import type { VolumeBackend } from "../backend"; -import { assertMounted, executeMount, executeUnmount } from "../utils/backend-utils"; - -const mount = async (config: BackendConfig, path: string) => { - logger.debug(`Mounting volume ${path}...`); - - if (config.backend !== "nfs") { - logger.error("Provided config is not for NFS backend"); - return { - status: BACKEND_STATUS.error, - error: "Provided config is not for NFS backend", - }; - } - - if (os.platform() !== "linux") { - logger.error("NFS mounting is only supported on Linux hosts."); - return { - status: BACKEND_STATUS.error, - error: "NFS mounting is only supported on Linux hosts.", - }; - } - - const { status } = await checkHealth(path); - if (status === "mounted") { - return { status: BACKEND_STATUS.mounted }; - } - - if (status === "error") { - logger.debug(`Trying to unmount any existing mounts at ${path} before mounting...`); - await unmount(path); - } - - const run = async () => { - await fs.mkdir(path, { recursive: true }); - - const source = `${config.server}:${config.exportPath}`; - const options = [`vers=${config.version}`, `port=${config.port}`]; - if (config.version === "3") { - options.push("nolock"); - } - if (config.readOnly) { - options.push("ro"); - } - const args = ["-t", "nfs", "-o", options.join(","), source, path]; - - logger.debug(`Mounting volume ${path}...`); - logger.info(`Executing mount: mount ${args.join(" ")}`); - - try { - await executeMount(args); - } catch (error) { - logger.warn(`Initial NFS mount failed, retrying with -i flag: ${toMessage(error)}`); - // Fallback with -i flag if the first mount fails using the mount helper - await executeMount(["-i", ...args]); - } - - logger.info(`NFS volume at ${path} mounted successfully.`); - return { status: BACKEND_STATUS.mounted }; - }; - - try { - return await withTimeout(run(), OPERATION_TIMEOUT, "NFS mount"); - } catch (err) { - logger.error("Error mounting NFS volume", { error: toMessage(err) }); - return { status: BACKEND_STATUS.error, error: toMessage(err) }; - } -}; - -const unmount = async (path: string) => { - if (os.platform() !== "linux") { - logger.error("NFS unmounting is only supported on Linux hosts."); - return { - status: BACKEND_STATUS.error, - error: "NFS unmounting is only supported on Linux hosts.", - }; - } - - const run = async () => { - const mount = await getMountForPath(path); - if (!mount || mount.mountPoint !== path) { - logger.debug(`Path ${path} is not a mount point. Skipping unmount.`); - return { status: BACKEND_STATUS.unmounted }; - } - - await executeUnmount(path); - - await fs.rmdir(path).catch(() => {}); - - logger.info(`NFS volume at ${path} unmounted successfully.`); - return { status: BACKEND_STATUS.unmounted }; - }; - - try { - return await withTimeout(run(), OPERATION_TIMEOUT, "NFS unmount"); - } catch (err) { - logger.error("Error unmounting NFS volume", { - path, - error: toMessage(err), - }); - return { status: BACKEND_STATUS.error, error: toMessage(err) }; - } -}; - -const checkHealth = async (path: string) => { - const run = async () => { - await assertMounted(path, (fstype) => fstype.startsWith("nfs")); - - logger.debug(`NFS volume at ${path} is healthy and mounted.`); - return { status: BACKEND_STATUS.mounted }; - }; - - try { - return await withTimeout(run(), OPERATION_TIMEOUT, "NFS health check"); - } catch (error) { - const message = toMessage(error); - if (message !== "Volume is not mounted") { - logger.error("NFS volume health check failed:", message); - } - return { status: BACKEND_STATUS.error, error: message }; - } -}; - -export const makeNfsBackend = (config: BackendConfig, path: string): VolumeBackend => ({ - mount: () => mount(config, path), - unmount: () => unmount(path), - checkHealth: () => checkHealth(path), -}); diff --git a/app/server/modules/backends/rclone/rclone-backend.ts b/app/server/modules/backends/rclone/rclone-backend.ts deleted file mode 100644 index d4d8d241..00000000 --- a/app/server/modules/backends/rclone/rclone-backend.ts +++ /dev/null @@ -1,131 +0,0 @@ -import * as fs from "node:fs/promises"; -import * as os from "node:os"; -import { OPERATION_TIMEOUT, RCLONE_CONFIG_FILE } from "../../../core/constants"; -import { toMessage } from "../../../utils/errors"; -import { logger } from "@zerobyte/core/node"; -import { getMountForPath } from "../../../utils/mountinfo"; -import { withTimeout } from "../../../utils/timeout"; -import type { VolumeBackend } from "../backend"; -import { assertMounted, executeUnmount } from "../utils/backend-utils"; -import { BACKEND_STATUS, type BackendConfig } from "~/schemas/volumes"; -import { safeExec } from "@zerobyte/core/node"; -import { config as zbConfig } from "~/server/core/config"; - -const mount = async (config: BackendConfig, path: string) => { - logger.debug(`Mounting rclone volume ${path}...`); - - if (config.backend !== "rclone") { - logger.error("Provided config is not for rclone backend"); - return { status: BACKEND_STATUS.error, error: "Provided config is not for rclone backend" }; - } - - if (os.platform() !== "linux") { - logger.error("Rclone mounting is only supported on Linux hosts."); - return { status: BACKEND_STATUS.error, error: "Rclone mounting is only supported on Linux hosts." }; - } - - const { status } = await checkHealth(path); - if (status === "mounted") { - return { status: BACKEND_STATUS.mounted }; - } - - if (status === "error") { - logger.debug(`Trying to unmount any existing mounts at ${path} before mounting...`); - await unmount(path); - } - - const run = async () => { - await fs.mkdir(path, { recursive: true }); - - const remotePath = `${config.remote}:${config.path}`; - const args = ["mount", remotePath, path, "--daemon"]; - - if (config.readOnly) { - args.push("--read-only"); - } - - args.push("--vfs-cache-mode", "writes"); - args.push("--allow-non-empty"); - args.push("--allow-other"); - - logger.debug(`Mounting rclone volume ${path}...`); - logger.info(`Executing rclone: rclone ${args.join(" ")}`); - - const result = await safeExec({ - command: "rclone", - args, - env: { RCLONE_CONFIG: RCLONE_CONFIG_FILE }, - timeout: zbConfig.serverIdleTimeout * 1000, - }); - - if (result.exitCode !== 0) { - const errorMsg = result.stderr.toString() || result.stdout.toString() || "Unknown error"; - throw new Error(`Failed to mount rclone volume: ${errorMsg}`); - } - - logger.info(`Rclone volume at ${path} mounted successfully.`); - return { status: BACKEND_STATUS.mounted }; - }; - - try { - return await withTimeout(run(), zbConfig.serverIdleTimeout * 1000, "Rclone mount"); - } catch (error) { - const errorMsg = toMessage(error); - - logger.error("Error mounting rclone volume", { error: errorMsg }); - return { status: BACKEND_STATUS.error, error: errorMsg }; - } -}; - -const unmount = async (path: string) => { - if (os.platform() !== "linux") { - logger.error("Rclone unmounting is only supported on Linux hosts."); - return { status: BACKEND_STATUS.error, error: "Rclone unmounting is only supported on Linux hosts." }; - } - - const run = async () => { - const mount = await getMountForPath(path); - if (!mount || mount.mountPoint !== path) { - logger.debug(`Path ${path} is not a mount point. Skipping unmount.`); - return { status: BACKEND_STATUS.unmounted }; - } - - await executeUnmount(path); - await fs.rmdir(path).catch(() => {}); - - logger.info(`Rclone volume at ${path} unmounted successfully.`); - return { status: BACKEND_STATUS.unmounted }; - }; - - try { - return await withTimeout(run(), OPERATION_TIMEOUT, "Rclone unmount"); - } catch (error) { - logger.error("Error unmounting rclone volume", { path, error: toMessage(error) }); - return { status: BACKEND_STATUS.error, error: toMessage(error) }; - } -}; - -const checkHealth = async (path: string) => { - const run = async () => { - await assertMounted(path, (fstype) => fstype.includes("rclone")); - - logger.debug(`Rclone volume at ${path} is healthy and mounted.`); - return { status: BACKEND_STATUS.mounted }; - }; - - try { - return await withTimeout(run(), OPERATION_TIMEOUT, "Rclone health check"); - } catch (error) { - const message = toMessage(error); - if (message !== "Volume is not mounted") { - logger.error("Rclone volume health check failed:", message); - } - return { status: BACKEND_STATUS.error, error: message }; - } -}; - -export const makeRcloneBackend = (config: BackendConfig, path: string): VolumeBackend => ({ - mount: () => mount(config, path), - unmount: () => unmount(path), - checkHealth: () => checkHealth(path), -}); diff --git a/app/server/modules/backends/sftp/sftp-backend.ts b/app/server/modules/backends/sftp/sftp-backend.ts deleted file mode 100644 index 9fbb2587..00000000 --- a/app/server/modules/backends/sftp/sftp-backend.ts +++ /dev/null @@ -1,212 +0,0 @@ -import * as fs from "node:fs/promises"; -import * as os from "node:os"; -import * as path from "node:path"; -import { spawn } from "node:child_process"; -import { OPERATION_TIMEOUT, SSH_KEYS_DIR } from "../../../core/constants"; -import { cryptoUtils } from "../../../utils/crypto"; -import { toMessage } from "../../../utils/errors"; -import { logger, FILE_MODES, writeFileWithMode } from "@zerobyte/core/node"; -import { getMountForPath } from "../../../utils/mountinfo"; -import { withTimeout } from "../../../utils/timeout"; -import type { VolumeBackend } from "../backend"; -import { executeUnmount } from "../utils/backend-utils"; -import { BACKEND_STATUS, type BackendConfig } from "~/schemas/volumes"; - -const getPrivateKeyPath = (mountPath: string) => { - const name = path.basename(mountPath); - return path.join(SSH_KEYS_DIR, `${name}.key`); -}; - -const getKnownHostsPath = (mountPath: string) => { - const name = path.basename(mountPath); - return path.join(SSH_KEYS_DIR, `${name}.known_hosts`); -}; - -const mount = async (config: BackendConfig, mountPath: string) => { - logger.debug(`Mounting SFTP volume ${mountPath}...`); - - if (config.backend !== "sftp") { - logger.error("Provided config is not for SFTP backend"); - return { status: BACKEND_STATUS.error, error: "Provided config is not for SFTP backend" }; - } - - if (os.platform() !== "linux") { - logger.error("SFTP mounting is only supported on Linux hosts."); - return { status: BACKEND_STATUS.error, error: "SFTP mounting is only supported on Linux hosts." }; - } - - const { status } = await checkHealth(mountPath); - if (status === "mounted") { - return { status: BACKEND_STATUS.mounted }; - } - - if (status === "error") { - logger.debug(`Trying to unmount any existing mounts at ${mountPath} before mounting...`); - await unmount(mountPath); - } - - const run = async () => { - await fs.mkdir(mountPath, { recursive: true }); - await fs.mkdir(SSH_KEYS_DIR, { recursive: true }); - - const { uid, gid } = os.userInfo(); - const options = [ - "reconnect", - "ServerAliveInterval=15", - "ServerAliveCountMax=3", - "allow_other", - `uid=${uid}`, - `gid=${gid}`, - ]; - - if (config.skipHostKeyCheck) { - options.push("StrictHostKeyChecking=no", "UserKnownHostsFile=/dev/null"); - } else if (config.knownHosts) { - const knownHostsPath = getKnownHostsPath(mountPath); - await writeFileWithMode(knownHostsPath, config.knownHosts, FILE_MODES.ownerReadWrite); - options.push(`UserKnownHostsFile=${knownHostsPath}`, "StrictHostKeyChecking=yes"); - } else { - options.push("StrictHostKeyChecking=yes"); - } - - if (config.readOnly) { - options.push("ro"); - } - - if (config.port) { - options.push(`port=${config.port}`); - } - - const keyPath = getPrivateKeyPath(mountPath); - if (config.privateKey) { - const decryptedKey = await cryptoUtils.resolveSecret(config.privateKey); - let normalizedKey = decryptedKey.replace(/\r\n/g, "\n"); - if (!normalizedKey.endsWith("\n")) { - normalizedKey += "\n"; - } - await writeFileWithMode(keyPath, normalizedKey, FILE_MODES.ownerReadWrite); - options.push(`IdentityFile=${keyPath}`); - } - - const source = `${config.username}@${config.host}:${config.path || ""}`; - const args = [source, mountPath, "-o", options.join(",")]; - - logger.debug(`Mounting SFTP volume ${mountPath}...`); - - const runSshfs = async (mountArgs: string[], password?: string) => { - return new Promise((resolve, reject) => { - const child = spawn("sshfs", mountArgs, { stdio: ["pipe", "pipe", "pipe"] }); - let stdout = ""; - let stderr = ""; - - child.stdout.setEncoding("utf8"); - child.stderr.setEncoding("utf8"); - - child.stdout.on("data", (data) => { - stdout += data; - }); - - child.stderr.on("data", (data) => { - stderr += data; - }); - - child.on("error", (error) => { - reject(new Error(`Failed to start sshfs: ${error.message}`)); - }); - - child.on("close", (code) => { - if (code === 0) { - resolve(); - return; - } - - const errorMsg = stderr.trim() || stdout.trim() || "Unknown error"; - reject(new Error(`Failed to mount SFTP volume: ${errorMsg}`)); - }); - - if (password) { - child.stdin.write(password); - } - child.stdin.end(); - }); - }; - - if (config.password) { - const password = await cryptoUtils.resolveSecret(config.password); - args.push("-o", "password_stdin"); - logger.info(`Executing sshfs: sshfs ${args.join(" ")}`); - await runSshfs(args, password); - } else { - logger.info(`Executing sshfs: sshfs ${args.join(" ")}`); - await runSshfs(args); - } - - logger.info(`SFTP volume at ${mountPath} mounted successfully.`); - return { status: BACKEND_STATUS.mounted }; - }; - - try { - return await withTimeout(run(), OPERATION_TIMEOUT * 2, "SFTP mount"); - } catch (error) { - const errorMsg = toMessage(error); - logger.error("Error mounting SFTP volume", { error: errorMsg }); - return { status: BACKEND_STATUS.error, error: errorMsg }; - } -}; - -const unmount = async (mountPath: string) => { - if (os.platform() !== "linux") { - logger.error("SFTP unmounting is only supported on Linux hosts."); - return { status: BACKEND_STATUS.error, error: "SFTP unmounting is only supported on Linux hosts." }; - } - - const run = async () => { - const mount = await getMountForPath(mountPath); - if (!mount || mount.mountPoint !== mountPath) { - logger.debug(`Path ${mountPath} is not a mount point. Skipping unmount.`); - } else { - await executeUnmount(mountPath); - } - - const keyPath = getPrivateKeyPath(mountPath); - await fs.unlink(keyPath).catch(() => {}); - - const knownHostsPath = getKnownHostsPath(mountPath); - await fs.unlink(knownHostsPath).catch(() => {}); - - await fs.rmdir(mountPath).catch(() => {}); - - logger.info(`SFTP volume at ${mountPath} unmounted successfully.`); - return { status: BACKEND_STATUS.unmounted }; - }; - - try { - return await withTimeout(run(), OPERATION_TIMEOUT, "SFTP unmount"); - } catch (error) { - logger.error("Error unmounting SFTP volume", { mountPath, error: toMessage(error) }); - return { status: BACKEND_STATUS.error, error: toMessage(error) }; - } -}; - -const checkHealth = async (mountPath: string) => { - const mount = await getMountForPath(mountPath); - - if (!mount || mount.mountPoint !== mountPath) { - return { status: BACKEND_STATUS.unmounted }; - } - - if (mount.fstype !== "fuse.sshfs") { - return { - status: BACKEND_STATUS.error, - error: `Invalid filesystem type: ${mount.fstype} (expected fuse.sshfs)`, - }; - } - - return { status: BACKEND_STATUS.mounted }; -}; - -export const makeSftpBackend = (config: BackendConfig, mountPath: string): VolumeBackend => ({ - mount: () => mount(config, mountPath), - unmount: () => unmount(mountPath), - checkHealth: () => checkHealth(mountPath), -}); diff --git a/app/server/modules/backends/smb/smb-backend.ts b/app/server/modules/backends/smb/smb-backend.ts deleted file mode 100644 index a4d51a45..00000000 --- a/app/server/modules/backends/smb/smb-backend.ts +++ /dev/null @@ -1,141 +0,0 @@ -import * as fs from "node:fs/promises"; -import * as os from "node:os"; -import { OPERATION_TIMEOUT } from "../../../core/constants"; -import { cryptoUtils } from "../../../utils/crypto"; -import { toMessage } from "../../../utils/errors"; -import { logger } from "@zerobyte/core/node"; -import { getMountForPath } from "../../../utils/mountinfo"; -import { withTimeout } from "../../../utils/timeout"; -import type { VolumeBackend } from "../backend"; -import { assertMounted, executeMount, executeUnmount } from "../utils/backend-utils"; -import { BACKEND_STATUS, type BackendConfig } from "~/schemas/volumes"; - -const mount = async (config: BackendConfig, path: string) => { - logger.debug(`Mounting SMB volume ${path}...`); - - if (config.backend !== "smb") { - logger.error("Provided config is not for SMB backend"); - return { status: BACKEND_STATUS.error, error: "Provided config is not for SMB backend" }; - } - - if (os.platform() !== "linux") { - logger.error("SMB mounting is only supported on Linux hosts."); - return { status: BACKEND_STATUS.error, error: "SMB mounting is only supported on Linux hosts." }; - } - - const { status } = await checkHealth(path); - if (status === "mounted") { - return { status: BACKEND_STATUS.mounted }; - } - - if (status === "error") { - logger.debug(`Trying to unmount any existing mounts at ${path} before mounting...`); - await unmount(path); - } - - const run = async () => { - await fs.mkdir(path, { recursive: true }); - - const source = `//${config.server}/${config.share}`; - const { uid, gid } = os.userInfo(); - - const options = [`port=${config.port}`, `uid=${uid}`, `gid=${gid}`, "iocharset=utf8"]; - - if (config.guest) { - options.push("username=guest", "password="); - } else { - const password = await cryptoUtils.resolveSecret(config.password ?? ""); - const safePassword = password.replace(/\\/g, "\\\\").replace(/,/g, "\\,"); - - options.push(`username=${config.username ?? "user"}`, `password=${safePassword}`); - } - - if (config.domain) { - options.push(`domain=${config.domain}`); - } - - if (config.vers && config.vers !== "auto") { - options.push(`vers=${config.vers}`); - } - - if (config.readOnly) { - options.push("ro"); - } - - const args = ["-t", "cifs", "-o", options.join(","), source, path]; - - logger.debug(`Mounting SMB volume ${path}...`); - logger.info(`Executing SMB mount for ${source} at ${path}`); - - try { - await executeMount(args); - } catch (error) { - logger.error(`SMB mount failed: ${toMessage(error)}`); - throw error; - } - - logger.info(`SMB volume at ${path} mounted successfully.`); - return { status: BACKEND_STATUS.mounted }; - }; - - try { - return await withTimeout(run(), OPERATION_TIMEOUT, "SMB mount"); - } catch (error) { - logger.error("Error mounting SMB volume", { error: toMessage(error) }); - return { status: BACKEND_STATUS.error, error: toMessage(error) }; - } -}; - -const unmount = async (path: string) => { - if (os.platform() !== "linux") { - logger.error("SMB unmounting is only supported on Linux hosts."); - return { status: BACKEND_STATUS.error, error: "SMB unmounting is only supported on Linux hosts." }; - } - - const run = async () => { - const mount = await getMountForPath(path); - if (!mount || mount.mountPoint !== path) { - logger.debug(`Path ${path} is not a mount point. Skipping unmount.`); - return { status: BACKEND_STATUS.unmounted }; - } - - await executeUnmount(path); - - await fs.rmdir(path).catch(() => {}); - - logger.info(`SMB volume at ${path} unmounted successfully.`); - return { status: BACKEND_STATUS.unmounted }; - }; - - try { - return await withTimeout(run(), OPERATION_TIMEOUT, "SMB unmount"); - } catch (error) { - logger.error("Error unmounting SMB volume", { path, error: toMessage(error) }); - return { status: BACKEND_STATUS.error, error: toMessage(error) }; - } -}; - -const checkHealth = async (path: string) => { - const run = async () => { - await assertMounted(path, (fstype) => fstype === "cifs"); - - logger.debug(`SMB volume at ${path} is healthy and mounted.`); - return { status: BACKEND_STATUS.mounted }; - }; - - try { - return await withTimeout(run(), OPERATION_TIMEOUT, "SMB health check"); - } catch (error) { - const message = toMessage(error); - if (message !== "Volume is not mounted") { - logger.error("SMB volume health check failed:", message); - } - return { status: BACKEND_STATUS.error, error: message }; - } -}; - -export const makeSmbBackend = (config: BackendConfig, path: string): VolumeBackend => ({ - mount: () => mount(config, path), - unmount: () => unmount(path), - checkHealth: () => checkHealth(path), -}); diff --git a/app/server/modules/backends/utils/__tests__/backend-utils.test.ts b/app/server/modules/backends/utils/__tests__/backend-utils.test.ts deleted file mode 100644 index afee747c..00000000 --- a/app/server/modules/backends/utils/__tests__/backend-utils.test.ts +++ /dev/null @@ -1,59 +0,0 @@ -import { afterEach, describe, expect, test, vi } from "vitest"; - -vi.mock("node:fs/promises", async (importOriginal) => { - const actual = await importOriginal(); - - return { - ...actual, - access: vi.fn(actual.access), - }; -}); - -vi.mock("../../../../utils/mountinfo", async (importOriginal) => { - const actual = await importOriginal(); - - return { - ...actual, - getMountForPath: vi.fn(actual.getMountForPath), - }; -}); - -import * as fs from "node:fs/promises"; -import * as mountinfo from "../../../../utils/mountinfo"; -import { assertMounted } from "../backend-utils"; - -afterEach(() => { - vi.restoreAllMocks(); -}); - -describe("assertMountedFilesystem", () => { - test("throws when the path is not accessible", async () => { - vi.mocked(fs.access).mockRejectedValueOnce(new Error("missing")); - - await expect(assertMounted("/tmp/volume", (fstype) => fstype.startsWith("nfs"))).rejects.toThrow( - "Volume is not mounted", - ); - }); - - test("throws when the mount filesystem does not match", async () => { - vi.mocked(fs.access).mockResolvedValueOnce(undefined); - vi.mocked(mountinfo.getMountForPath).mockResolvedValueOnce({ - mountPoint: "/tmp/volume", - fstype: "cifs", - }); - - await expect(assertMounted("/tmp/volume", (fstype) => fstype.startsWith("nfs"))).rejects.toThrow( - "Path /tmp/volume is not mounted as correct fstype (found cifs).", - ); - }); - - test("accepts a matching mounted filesystem", async () => { - vi.mocked(fs.access).mockResolvedValueOnce(undefined); - vi.mocked(mountinfo.getMountForPath).mockResolvedValueOnce({ - mountPoint: "/tmp/volume", - fstype: "nfs4", - }); - - await expect(assertMounted("/tmp/volume", (fstype) => fstype.startsWith("nfs"))).resolves.toBeUndefined(); - }); -}); diff --git a/app/server/modules/backends/webdav/webdav-backend.ts b/app/server/modules/backends/webdav/webdav-backend.ts deleted file mode 100644 index 7885be84..00000000 --- a/app/server/modules/backends/webdav/webdav-backend.ts +++ /dev/null @@ -1,159 +0,0 @@ -import * as fs from "node:fs/promises"; -import * as os from "node:os"; -import { OPERATION_TIMEOUT } from "../../../core/constants"; -import { cryptoUtils } from "../../../utils/crypto"; -import { toMessage } from "../../../utils/errors"; -import { logger } from "@zerobyte/core/node"; -import { getMountForPath } from "../../../utils/mountinfo"; -import { withTimeout } from "../../../utils/timeout"; -import type { VolumeBackend } from "../backend"; -import { assertMounted, executeMount, executeUnmount } from "../utils/backend-utils"; -import { BACKEND_STATUS, type BackendConfig } from "~/schemas/volumes"; - -const mount = async (config: BackendConfig, path: string) => { - logger.debug(`Mounting WebDAV volume ${path}...`); - - if (config.backend !== "webdav") { - logger.error("Provided config is not for WebDAV backend"); - return { status: BACKEND_STATUS.error, error: "Provided config is not for WebDAV backend" }; - } - - if (os.platform() !== "linux") { - logger.error("WebDAV mounting is only supported on Linux hosts."); - return { status: BACKEND_STATUS.error, error: "WebDAV mounting is only supported on Linux hosts." }; - } - - const { status } = await checkHealth(path); - if (status === "mounted") { - return { status: BACKEND_STATUS.mounted }; - } - - if (status === "error") { - logger.debug(`Trying to unmount any existing mounts at ${path} before mounting...`); - await unmount(path); - } - - const run = async () => { - await fs.mkdir(path, { recursive: true }).catch((err) => { - logger.warn(`Failed to create directory ${path}: ${err.message}`); - }); - - const protocol = config.ssl ? "https" : "http"; - const defaultPort = config.ssl ? 443 : 80; - const port = config.port !== defaultPort ? `:${config.port}` : ""; - const source = `${protocol}://${config.server}${port}${config.path}`; - - const { uid, gid } = os.userInfo(); - const options = config.readOnly - ? [`uid=${uid}`, `gid=${gid}`, "file_mode=0444", "dir_mode=0555", "ro"] - : [`uid=${uid}`, `gid=${gid}`, "file_mode=0664", "dir_mode=0775"]; - - if (config.username && config.password) { - const password = await cryptoUtils.resolveSecret(config.password); - const secretsFile = "/etc/davfs2/secrets"; - const entry = [source, config.username, password].map((value) => value.replace(/[\r\n\t\s]+/g, " ")).join(" "); - const secretsContent = `${entry}\n`; - await fs.appendFile(secretsFile, secretsContent, { mode: 0o600 }); - } - - logger.debug(`Mounting WebDAV volume ${path}...`); - - const args = ["-t", "davfs", "-o", options.join(","), source, path]; - - try { - await executeMount(args); - } catch (error) { - logger.warn(`Initial WebDAV mount failed, retrying with -i flag: ${toMessage(error)}`); - // Fallback with -i flag if the first mount fails using the mount helper - await executeMount(["-i", ...args]); - } - - logger.info(`WebDAV volume at ${path} mounted successfully.`); - return { status: BACKEND_STATUS.mounted }; - }; - - try { - return await withTimeout(run(), OPERATION_TIMEOUT, "WebDAV mount"); - } catch (error) { - const errorMsg = toMessage(error); - - if (errorMsg.includes("already mounted")) { - return { status: BACKEND_STATUS.mounted }; - } - - logger.error("Error mounting WebDAV volume", { error: errorMsg }); - - if (errorMsg.includes("option") && errorMsg.includes("requires argument")) { - return { - status: BACKEND_STATUS.error, - error: "Invalid mount options. Please check your WebDAV server configuration.", - }; - } else if (errorMsg.includes("connection refused") || errorMsg.includes("Connection refused")) { - return { - status: BACKEND_STATUS.error, - error: "Cannot connect to WebDAV server. Please check the server address and port.", - }; - } else if (errorMsg.includes("unauthorized") || errorMsg.includes("Unauthorized")) { - return { - status: BACKEND_STATUS.error, - error: "Authentication failed. Please check your username and password.", - }; - } - - return { status: BACKEND_STATUS.error, error: errorMsg }; - } -}; - -const unmount = async (path: string) => { - if (os.platform() !== "linux") { - logger.error("WebDAV unmounting is only supported on Linux hosts."); - return { status: BACKEND_STATUS.error, error: "WebDAV unmounting is only supported on Linux hosts." }; - } - - const run = async () => { - const mount = await getMountForPath(path); - if (!mount || mount.mountPoint !== path) { - logger.debug(`Path ${path} is not a mount point. Skipping unmount.`); - return { status: BACKEND_STATUS.unmounted }; - } - - await executeUnmount(path); - - await fs.rmdir(path).catch(() => {}); - - logger.info(`WebDAV volume at ${path} unmounted successfully.`); - return { status: BACKEND_STATUS.unmounted }; - }; - - try { - return await withTimeout(run(), OPERATION_TIMEOUT, "WebDAV unmount"); - } catch (error) { - logger.error("Error unmounting WebDAV volume", { path, error: toMessage(error) }); - return { status: BACKEND_STATUS.error, error: toMessage(error) }; - } -}; - -const checkHealth = async (path: string) => { - const run = async () => { - await assertMounted(path, (fstype) => fstype === "fuse" || fstype === "davfs"); - - logger.debug(`WebDAV volume at ${path} is healthy and mounted.`); - return { status: BACKEND_STATUS.mounted }; - }; - - try { - return await withTimeout(run(), OPERATION_TIMEOUT, "WebDAV health check"); - } catch (error) { - const message = toMessage(error); - if (message !== "Volume is not mounted") { - logger.error("WebDAV volume health check failed:", message); - } - return { status: BACKEND_STATUS.error, error: message }; - } -}; - -export const makeWebdavBackend = (config: BackendConfig, path: string): VolumeBackend => ({ - mount: () => mount(config, path), - unmount: () => unmount(path), - checkHealth: () => checkHealth(path), -}); diff --git a/app/server/modules/lifecycle/__tests__/shutdown.test.ts b/app/server/modules/lifecycle/__tests__/shutdown.test.ts index 6e8adddf..da7aa379 100644 --- a/app/server/modules/lifecycle/__tests__/shutdown.test.ts +++ b/app/server/modules/lifecycle/__tests__/shutdown.test.ts @@ -1,7 +1,7 @@ import { afterEach, describe, expect, test, vi } from "vitest"; import { Scheduler } from "../../../core/scheduler"; -import * as backendModule from "../../backends/backend"; -import type { VolumeBackend } from "../../backends/backend"; +import * as volumeHostModule from "../../../../../apps/agent/src/volume-host"; +import type { VolumeBackend } from "../../../../../apps/agent/src/volume-host"; import * as bootstrapModule from "../bootstrap"; import { createTestVolume } from "~/test/helpers/volume"; @@ -40,7 +40,7 @@ describe("shutdown", () => { vi.spyOn(Scheduler, "stop").mockImplementation(stopScheduler); vi.spyOn(bootstrapModule, "stopApplicationRuntime").mockImplementation(stopApplicationRuntime); - vi.spyOn(backendModule, "createVolumeBackend").mockImplementation( + vi.spyOn(volumeHostModule, "createVolumeBackend").mockImplementation( () => ({ mount: async () => ({ status: "mounted" as const }), diff --git a/app/server/modules/lifecycle/shutdown.ts b/app/server/modules/lifecycle/shutdown.ts index 4deff2af..02f180c8 100644 --- a/app/server/modules/lifecycle/shutdown.ts +++ b/app/server/modules/lifecycle/shutdown.ts @@ -1,8 +1,13 @@ import { Scheduler } from "../../core/scheduler"; import { db } from "../../db/db"; import { logger } from "@zerobyte/core/node"; -import { createVolumeBackend } from "../backends/backend"; import { stopApplicationRuntime } from "./bootstrap"; +import { decryptVolumeConfig } from "../volumes/volume-config-secrets"; +import { + createVolumeBackend, + type AgentVolume, + type BackendConfig as HostBackendConfig, +} from "../../../../apps/agent/src/volume-host"; export const shutdown = async () => { await Scheduler.stop(); @@ -13,7 +18,11 @@ export const shutdown = async () => { }); for (const volume of volumes) { - const backend = createVolumeBackend(volume); + const backend = createVolumeBackend({ + ...volume, + config: (await decryptVolumeConfig(volume.config)) as HostBackendConfig, + provisioningId: volume.provisioningId ?? null, + } satisfies AgentVolume); const { status, error } = await backend.unmount(); logger.info(`Volume ${volume.name} unmount status: ${status}${error ? `, error: ${error}` : ""}`); diff --git a/app/server/modules/volumes/__tests__/volumes.service.test.ts b/app/server/modules/volumes/__tests__/volumes.service.test.ts index 7b487801..9372df16 100644 --- a/app/server/modules/volumes/__tests__/volumes.service.test.ts +++ b/app/server/modules/volumes/__tests__/volumes.service.test.ts @@ -1,19 +1,26 @@ import { afterEach, describe, expect, test, vi } from "vitest"; +const agentManagerMock = vi.hoisted(() => ({ + runVolumeCommand: vi.fn(), +})); + +vi.mock("../../agents/agents-manager", () => ({ + agentManager: agentManagerMock, +})); + import { volumeService } from "../volume.service"; import { db } from "~/server/db/db"; import { volumesTable } from "~/server/db/schema"; import { randomUUID } from "node:crypto"; -import * as fs from "node:fs/promises"; -import os from "node:os"; -import path from "node:path"; import { createTestSession } from "~/test/helpers/auth"; import { withContext } from "~/server/core/request-context"; import { asShortId } from "~/server/utils/branded"; import { createTestVolume } from "~/test/helpers/volume"; -import * as backendModule from "../../backends/backend"; +import { config } from "~/server/core/config"; afterEach(() => { + config.flags.enableLocalAgent = false; vi.restoreAllMocks(); + agentManagerMock.runVolumeCommand.mockReset(); }); describe("volumeService.getVolume", () => { @@ -98,13 +105,7 @@ describe("volumeService.getVolume", () => { describe("volumeService.listFiles security", () => { test("should reject traversal outside the volume root in listFiles", async () => { const { organizationId, user } = await createTestSession(); - const tempRoot = await fs.mkdtemp(path.join(os.tmpdir(), "zerobyte-vol-svc-")); - const volumePath = path.join(tempRoot, "vol"); - const secretPath = path.join(tempRoot, "volume-secret"); - - await fs.mkdir(volumePath, { recursive: true }); - await fs.mkdir(secretPath, { recursive: true }); - await fs.writeFile(path.join(secretPath, "secret.txt"), "top secret", "utf-8"); + agentManagerMock.runVolumeCommand.mockRejectedValue(new Error("Invalid path")); const [volume] = await db .insert(volumesTable) @@ -113,44 +114,40 @@ describe("volumeService.listFiles security", () => { name: `test-vol-${randomUUID().slice(0, 8)}`, type: "directory", status: "mounted", - config: { backend: "directory", path: volumePath }, + config: { backend: "directory", path: "/tmp/volume" }, autoRemount: true, organizationId, }) .returning(); - try { - await withContext({ organizationId, userId: user.id }, async () => { - const traversalPath = `../${path.basename(secretPath)}`; - - await expect(volumeService.listFiles(volume.shortId, traversalPath)).rejects.toThrow("Invalid path"); - }); - } finally { - await fs.rm(tempRoot, { recursive: true, force: true }); - } + await withContext({ organizationId, userId: user.id }, async () => { + await expect(volumeService.listFiles(volume.shortId, "../volume-secret")).rejects.toThrow("Invalid path"); + }); }); }); describe("volumeService.mountVolume", () => { - test("unmounts any existing mount before mounting", async () => { + test("routes unmount and mount to the owning agent before updating state", async () => { const { organizationId, user } = await createTestSession(); - const volume = await createTestVolume({ organizationId, status: "mounted" }); - const unmount = vi.fn().mockResolvedValue({ status: "unmounted" }); - const mount = vi.fn().mockResolvedValue({ status: "mounted" }); - - vi.spyOn(backendModule, "createVolumeBackend").mockImplementation(() => ({ - mount, - unmount, - checkHealth: vi.fn().mockResolvedValue({ status: "mounted" }), - })); + const volume = await createTestVolume({ organizationId, status: "mounted", agentId: "agent-1" }); + agentManagerMock.runVolumeCommand + .mockResolvedValueOnce({ name: "volume.unmount", result: { status: "unmounted" } }) + .mockResolvedValueOnce({ name: "volume.mount", result: { status: "mounted" } }); await withContext({ organizationId, userId: user.id }, async () => { const result = await volumeService.mountVolume(volume.shortId); expect(result.status).toBe("mounted"); - expect(unmount).toHaveBeenCalledOnce(); - expect(mount).toHaveBeenCalledOnce(); - expect(unmount.mock.invocationCallOrder[0]).toBeLessThan(mount.mock.invocationCallOrder[0]); + expect(agentManagerMock.runVolumeCommand).toHaveBeenNthCalledWith( + 1, + volume.agentId, + expect.objectContaining({ name: "volume.unmount", volume: expect.objectContaining({ id: volume.id }) }), + ); + expect(agentManagerMock.runVolumeCommand).toHaveBeenNthCalledWith( + 2, + volume.agentId, + expect.objectContaining({ name: "volume.mount", volume: expect.objectContaining({ id: volume.id }) }), + ); }); }); }); @@ -158,15 +155,8 @@ describe("volumeService.mountVolume", () => { describe("volumeService.ensureHealthyVolume", () => { test("returns ready when the mounted volume passes its health check", async () => { const { organizationId, user } = await createTestSession(); - const volume = await createTestVolume({ organizationId, status: "mounted" }); - const mount = vi.fn().mockResolvedValue({ status: "mounted" }); - const checkHealth = vi.fn().mockResolvedValue({ status: "mounted" }); - - vi.spyOn(backendModule, "createVolumeBackend").mockImplementation(() => ({ - mount, - unmount: vi.fn().mockResolvedValue({ status: "unmounted" }), - checkHealth, - })); + const volume = await createTestVolume({ organizationId, status: "mounted", agentId: "agent-1" }); + agentManagerMock.runVolumeCommand.mockResolvedValue({ name: "volume.checkHealth", result: { status: "mounted" } }); await withContext({ organizationId, userId: user.id }, async () => { const result = await volumeService.ensureHealthyVolume(volume.shortId); @@ -176,22 +166,21 @@ describe("volumeService.ensureHealthyVolume", () => { volume: expect.objectContaining({ id: volume.id, status: "mounted", lastError: null }), remounted: false, }); - expect(checkHealth).toHaveBeenCalledOnce(); - expect(mount).not.toHaveBeenCalled(); + expect(agentManagerMock.runVolumeCommand).toHaveBeenCalledOnce(); + expect(agentManagerMock.runVolumeCommand).toHaveBeenCalledWith( + volume.agentId, + expect.objectContaining({ name: "volume.checkHealth", volume: expect.objectContaining({ id: volume.id }) }), + ); }); }); test("auto-remounts when the mounted volume fails its health check", async () => { const { organizationId, user } = await createTestSession(); - const volume = await createTestVolume({ organizationId, status: "mounted", autoRemount: true }); - const mount = vi.fn().mockResolvedValue({ status: "mounted" }); - const checkHealth = vi.fn().mockResolvedValue({ status: "error", error: "stale mount" }); - - vi.spyOn(backendModule, "createVolumeBackend").mockImplementation(() => ({ - mount, - unmount: vi.fn().mockResolvedValue({ status: "unmounted" }), - checkHealth, - })); + const volume = await createTestVolume({ organizationId, status: "mounted", autoRemount: true, agentId: "agent-1" }); + agentManagerMock.runVolumeCommand + .mockResolvedValueOnce({ name: "volume.checkHealth", result: { status: "error", error: "stale mount" } }) + .mockResolvedValueOnce({ name: "volume.unmount", result: { status: "unmounted" } }) + .mockResolvedValueOnce({ name: "volume.mount", result: { status: "mounted" } }); await withContext({ organizationId, userId: user.id }, async () => { const result = await volumeService.ensureHealthyVolume(volume.shortId); @@ -201,8 +190,7 @@ describe("volumeService.ensureHealthyVolume", () => { volume: expect.objectContaining({ id: volume.id, status: "mounted", lastError: null }), remounted: true, }); - expect(checkHealth).toHaveBeenCalledOnce(); - expect(mount).toHaveBeenCalledOnce(); + expect(agentManagerMock.runVolumeCommand).toHaveBeenCalledTimes(3); const updatedVolume = await db.query.volumesTable.findFirst({ where: { id: volume.id } }); expect(updatedVolume?.status).toBe("mounted"); @@ -212,15 +200,16 @@ describe("volumeService.ensureHealthyVolume", () => { test("returns not ready when the health check fails and auto-remount is disabled", async () => { const { organizationId, user } = await createTestSession(); - const volume = await createTestVolume({ organizationId, status: "mounted", autoRemount: false }); - const mount = vi.fn().mockResolvedValue({ status: "mounted" }); - const checkHealth = vi.fn().mockResolvedValue({ status: "error", error: "stale mount" }); - - vi.spyOn(backendModule, "createVolumeBackend").mockImplementation(() => ({ - mount, - unmount: vi.fn().mockResolvedValue({ status: "unmounted" }), - checkHealth, - })); + const volume = await createTestVolume({ + organizationId, + status: "mounted", + autoRemount: false, + agentId: "agent-1", + }); + agentManagerMock.runVolumeCommand.mockResolvedValue({ + name: "volume.checkHealth", + result: { status: "error", error: "stale mount" }, + }); await withContext({ organizationId, userId: user.id }, async () => { const result = await volumeService.ensureHealthyVolume(volume.shortId); @@ -230,54 +219,17 @@ describe("volumeService.ensureHealthyVolume", () => { volume: expect.objectContaining({ id: volume.id, status: "error", lastError: "stale mount" }), reason: "stale mount", }); - expect(checkHealth).toHaveBeenCalledOnce(); - expect(mount).not.toHaveBeenCalled(); + expect(agentManagerMock.runVolumeCommand).toHaveBeenCalledOnce(); }); }); }); describe("volumeService.testConnection", () => { - test("uses an isolated temp mount path for backend test connections", async () => { - const mount = vi.fn().mockResolvedValue({ status: "mounted" }); - const unmount = vi.fn().mockResolvedValue({ status: "unmounted" }); - const createVolumeBackendSpy = vi.spyOn(backendModule, "createVolumeBackend").mockReturnValue({ - mount, - unmount, - checkHealth: vi.fn(), - }); - - await volumeService.testConnection({ - backend: "nfs", - server: "127.0.0.1", - exportPath: "/exports/test", - version: "4", - port: 2049, - readOnly: false, - }); - - expect(createVolumeBackendSpy).toHaveBeenCalledOnce(); - const [, mountPath] = createVolumeBackendSpy.mock.calls[0]; - expect(mountPath).toEqual(expect.stringContaining(`${path.sep}zerobyte-test-`)); - await expect(fs.access(mountPath as string)).rejects.toThrow(); - expect(mount).toHaveBeenCalledOnce(); - expect(unmount).toHaveBeenCalledOnce(); - }); - - test("does not fail when backend unmount already removed the temp mount path", async () => { - const mount = vi.fn().mockResolvedValue({ status: "mounted" }); - let mountPath: string | undefined; - const unmount = vi.fn().mockImplementation(async () => { - await fs.rm(mountPath!, { recursive: true, force: true }); - return { status: "unmounted" }; - }); - - vi.spyOn(backendModule, "createVolumeBackend").mockImplementation((_volume, tempPath) => { - mountPath = tempPath; - return { - mount, - unmount, - checkHealth: vi.fn(), - }; + test("routes test connections to the local agent", async () => { + config.flags.enableLocalAgent = true; + agentManagerMock.runVolumeCommand.mockResolvedValue({ + name: "volume.testConnection", + result: { success: true, message: "Connection successful" }, }); await expect( @@ -294,9 +246,9 @@ describe("volumeService.testConnection", () => { message: "Connection successful", }); - expect(mountPath).toEqual(expect.stringContaining(`${path.sep}zerobyte-test-`)); - await expect(fs.access(mountPath as string)).rejects.toThrow(); - expect(mount).toHaveBeenCalledOnce(); - expect(unmount).toHaveBeenCalledOnce(); + expect(agentManagerMock.runVolumeCommand).toHaveBeenCalledWith( + "local", + expect.objectContaining({ name: "volume.testConnection" }), + ); }); }); diff --git a/app/server/modules/volumes/volume-config-secrets.ts b/app/server/modules/volumes/volume-config-secrets.ts index 2f9bf1c7..1bda9efe 100644 --- a/app/server/modules/volumes/volume-config-secrets.ts +++ b/app/server/modules/volumes/volume-config-secrets.ts @@ -32,3 +32,7 @@ export const mapVolumeConfigSecrets = async ( export const encryptVolumeConfig = async (config: BackendConfig): Promise => { return await mapVolumeConfigSecrets(config, cryptoUtils.sealSecret); }; + +export const decryptVolumeConfig = async (config: BackendConfig): Promise => { + return await mapVolumeConfigSecrets(config, cryptoUtils.resolveSecret); +}; diff --git a/app/server/modules/volumes/volume.service.ts b/app/server/modules/volumes/volume.service.ts index 0c5f4ba3..91d67ca9 100644 --- a/app/server/modules/volumes/volume.service.ts +++ b/app/server/modules/volumes/volume.service.ts @@ -1,26 +1,35 @@ -import * as fs from "node:fs/promises"; -import * as os from "node:os"; -import * as path from "node:path"; import { and, eq } from "drizzle-orm"; import { BadRequestError, InternalServerError, NotFoundError } from "http-errors-enhanced"; import { db } from "../../db/db"; import { volumesTable } from "../../db/schema"; import { toMessage } from "../../utils/errors"; import { generateShortId } from "../../utils/id"; -import { getStatFs, type StatFs } from "../../utils/mountinfo"; +import type { StatFs } from "../../utils/mountinfo"; import { withTimeout } from "../../utils/timeout"; -import { createVolumeBackend } from "../backends/backend"; +import { config } from "../../core/config"; import { LOCAL_AGENT_ID } from "../agents/constants"; +import { agentManager } from "../agents/agents-manager"; import type { UpdateVolumeBody } from "./volume.dto"; -import { getVolumePath } from "./helpers"; import { logger } from "@zerobyte/core/node"; import { serverEvents } from "../../core/events"; import type { Volume } from "../../db/schema"; import { volumeConfigSchema, type BackendConfig } from "~/schemas/volumes"; import { getOrganizationId } from "~/server/core/request-context"; -import { isNodeJSErrnoException } from "~/server/utils/fs"; -import { asShortId, type ShortId } from "~/server/utils/branded"; -import { encryptVolumeConfig } from "./volume-config-secrets"; +import { type ShortId } from "~/server/utils/branded"; +import { decryptVolumeConfig, encryptVolumeConfig } from "./volume-config-secrets"; +import type { VolumeCommand, VolumeCommandResult } from "@zerobyte/contracts/agent-protocol"; +import { + createVolumeBackend, + getStatFs, + getVolumePath, + type AgentVolume, + type BackendConfig as HostBackendConfig, +} from "../../../../apps/agent/src/volume-host"; +import { + browseFilesystem as browseHostFilesystem, + listVolumeFiles, + testVolumeConnection, +} from "../../../../apps/agent/src/volume-host/operations"; type EnsureHealthyVolumeResult = | { @@ -34,6 +43,9 @@ type EnsureHealthyVolumeResult = reason: string; }; +type VolumeBackendCommandName = "volume.mount" | "volume.unmount" | "volume.checkHealth"; +type VolumeBackendCommandResult = Extract["result"]; + const listVolumes = async () => { const organizationId = getOrganizationId(); const volumes = await db.query.volumesTable.findMany({ @@ -53,6 +65,64 @@ const findVolume = async (shortId: ShortId) => { }); }; +const runVolumeCommand = async (agentId: string, command: TCommand) => { + const result = await agentManager.runVolumeCommand(agentId, command); + if (result.name !== command.name) { + throw new InternalServerError(`Unexpected agent response for ${command.name}`); + } + + return result as Extract; +}; + +const volumeForAgent = async (volume: Volume): Promise => ({ + ...volume, + config: await decryptVolumeConfig(volume.config), +}); + +const volumeForHost = async (volume: Volume): Promise => ({ + ...volume, + shortId: volume.shortId, + config: (await decryptVolumeConfig(volume.config)) as HostBackendConfig, + provisioningId: volume.provisioningId ?? null, +}); + +// TODO(agent-rollout): Remove the local host execution branch once all installs run volume operations through agents. +const shouldRunViaAgent = (volume: Volume) => volume.agentId !== LOCAL_AGENT_ID || config.flags.enableLocalAgent; + +const runVolumeBackendCommand = async ( + volume: Volume, + name: "volume.mount" | "volume.unmount" | "volume.checkHealth", +) => { + if (!shouldRunViaAgent(volume)) { + const backend = createVolumeBackend(await volumeForHost(volume)); + switch (name) { + case "volume.mount": + return backend.mount(); + case "volume.unmount": + return backend.unmount(); + case "volume.checkHealth": + return backend.checkHealth(); + } + } + + const command = await runVolumeCommand(volume.agentId, { + name, + volume: await volumeForAgent(volume), + } as VolumeCommand); + return command.result as VolumeBackendCommandResult; +}; + +const mapAgentFileError = (error: unknown) => { + const message = toMessage(error); + if (message === "Invalid path") { + throw new BadRequestError("Invalid path"); + } + if (message === "Directory not found") { + throw new NotFoundError("Directory not found"); + } + throw error; +}; + const createVolume = async (name: string, backendConfig: BackendConfig) => { const organizationId = getOrganizationId(); const trimmedName = name.trim(); @@ -80,8 +150,7 @@ const createVolume = async (name: string, backendConfig: BackendConfig) => { throw new InternalServerError("Failed to create volume"); } - const backend = createVolumeBackend(created); - const { error, status } = await backend.mount(); + const { error, status } = await runVolumeBackendCommand(created, "volume.mount"); await db .update(volumesTable) @@ -99,8 +168,7 @@ const deleteVolume = async (shortId: ShortId) => { throw new NotFoundError("Volume not found"); } - const backend = createVolumeBackend(volume); - await backend.unmount(); + await runVolumeBackendCommand(volume, "volume.unmount"); await db .delete(volumesTable) .where(and(eq(volumesTable.id, volume.id), eq(volumesTable.organizationId, organizationId))); @@ -114,9 +182,8 @@ const mountVolume = async (shortId: ShortId) => { throw new NotFoundError("Volume not found"); } - const backend = createVolumeBackend(volume); - await backend.unmount(); - const { error, status } = await backend.mount(); + await runVolumeBackendCommand(volume, "volume.unmount"); + const { error, status } = await runVolumeBackendCommand(volume, "volume.mount"); await db .update(volumesTable) @@ -138,8 +205,7 @@ const unmountVolume = async (shortId: ShortId) => { throw new NotFoundError("Volume not found"); } - const backend = createVolumeBackend(volume); - const { status, error } = await backend.unmount(); + const { status, error } = await runVolumeBackendCommand(volume, "volume.unmount"); await db .update(volumesTable) @@ -162,7 +228,15 @@ const getVolume = async (shortId: ShortId) => { let statfs: Partial = {}; if (volume.status === "mounted") { - statfs = await withTimeout(getStatFs(getVolumePath(volume)), 1000, "getStatFs").catch((error) => { + statfs = await withTimeout( + shouldRunViaAgent(volume) + ? runVolumeCommand(volume.agentId, { name: "volume.statfs", volume: await volumeForAgent(volume) }).then( + (command) => command.result, + ) + : volumeForHost(volume).then((hostVolume) => getStatFs(getVolumePath(hostVolume))), + 1000, + "volume.statfs", + ).catch((error) => { logger.warn(`Failed to get statfs for volume ${volume.name}: ${toMessage(error)}`); return {}; }); @@ -190,8 +264,7 @@ const updateVolume = async (shortId: ShortId, volumeData: UpdateVolumeBody) => { if (configChanged) { logger.debug("Unmounting existing volume before applying new config"); - const backend = createVolumeBackend(existing); - await backend.unmount(); + await runVolumeBackendCommand(existing, "volume.unmount"); } const newConfigResult = volumeConfigSchema.safeParse(volumeData.config || existing.config); @@ -219,8 +292,7 @@ const updateVolume = async (shortId: ShortId, volumeData: UpdateVolumeBody) => { } if (configChanged) { - const backend = createVolumeBackend(updated); - const { error, status } = await backend.mount(); + const { error, status } = await runVolumeBackendCommand(updated, "volume.mount"); await db .update(volumesTable) .set({ status, lastError: error ?? null, lastHealthCheck: Date.now() }) @@ -233,40 +305,12 @@ const updateVolume = async (shortId: ShortId, volumeData: UpdateVolumeBody) => { }; const testConnection = async (backendConfig: BackendConfig) => { - const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "zerobyte-test-")); - try { - const encryptedConfig = await encryptVolumeConfig(backendConfig); - - const mockVolume = { - id: 0, - shortId: asShortId("test"), - name: "test-connection", - path: tempDir, - config: encryptedConfig, - createdAt: Date.now(), - updatedAt: Date.now(), - lastHealthCheck: Date.now(), - type: encryptedConfig.backend, - status: "unmounted" as const, - lastError: null, - provisioningId: null, - autoRemount: true, - agentId: LOCAL_AGENT_ID, - organizationId: "test-org", - }; - - const backend = createVolumeBackend(mockVolume, tempDir); - const { error } = await backend.mount(); - - await backend.unmount(); - - return { - success: !error, - message: error ? toMessage(error) : "Connection successful", - }; - } finally { - await fs.rm(tempDir, { recursive: true, force: true }); + if (!config.flags.enableLocalAgent) { + return await testVolumeConnection(backendConfig as HostBackendConfig); } + + const command = await runVolumeCommand(LOCAL_AGENT_ID, { name: "volume.testConnection", backendConfig }); + return command.result; }; const checkHealth = async (shortId: ShortId) => { @@ -277,8 +321,7 @@ const checkHealth = async (shortId: ShortId) => { throw new NotFoundError("Volume not found"); } - const backend = createVolumeBackend(volume); - const { error, status } = await backend.checkHealth(); + const { error, status } = await runVolumeBackendCommand(volume, "volume.checkHealth"); if (status !== volume.status) { serverEvents.emit("volume:status_changed", { organizationId, volumeName: volume.name, status }); @@ -346,7 +389,6 @@ const ensureHealthyVolume = async (shortId: ShortId): Promise { const volume = await findVolume(shortId); @@ -359,110 +401,33 @@ const listFiles = async (shortId: ShortId, subPath?: string, offset: number = 0, throw new InternalServerError("Volume is not mounted"); } - const volumePath = getVolumePath(volume); - const requestedPath = subPath ? path.join(volumePath, subPath) : volumePath; - const normalizedPath = path.normalize(requestedPath); - const relative = path.relative(volumePath, normalizedPath); - - if (relative.startsWith("..") || path.isAbsolute(relative)) { - throw new BadRequestError("Invalid path"); - } - - const pageSize = Math.min(Math.max(limit, 1), MAX_PAGE_SIZE); - const startOffset = Math.max(offset, 0); - try { - const dirents = await fs.readdir(normalizedPath, { withFileTypes: true }); - - dirents.sort((a, b) => { - const aIsDir = a.isDirectory(); - const bIsDir = b.isDirectory(); - - if (aIsDir === bIsDir) { - return a.name.localeCompare(b.name); - } - return aIsDir ? -1 : 1; - }); - - const total = dirents.length; - const paginatedDirents = dirents.slice(startOffset, startOffset + pageSize); - - const entries = ( - await Promise.all( - paginatedDirents.map(async (dirent) => { - const fullPath = path.join(normalizedPath, dirent.name); - - try { - const stats = await fs.stat(fullPath); - const relativePath = path.relative(volumePath, fullPath); - - return { - name: dirent.name, - path: `/${relativePath}`, - type: dirent.isDirectory() ? ("directory" as const) : ("file" as const), - size: dirent.isFile() ? stats.size : undefined, - modifiedAt: stats.mtimeMs, - }; - } catch { - return null; - } - }), - ) - ).filter((e) => e !== null); - - return { - files: entries, - path: subPath || "/", - offset: startOffset, - limit: pageSize, - total, - hasMore: startOffset + pageSize < total, - }; - } catch (error) { - if (isNodeJSErrnoException(error) && error.code === "ENOENT") { - throw new NotFoundError("Directory not found"); + if (!shouldRunViaAgent(volume)) { + return await listVolumeFiles(await volumeForHost(volume), subPath, offset, limit); } + + const command = await runVolumeCommand(volume.agentId, { + name: "volume.listFiles", + volume: await volumeForAgent(volume), + subPath, + offset, + limit, + }); + return command.result; + } catch (error) { + mapAgentFileError(error); throw new InternalServerError(`Failed to list files: ${toMessage(error)}`); } }; const browseFilesystem = async (browsePath: string) => { - const normalizedPath = path.normalize(browsePath); - try { - const entries = await fs.readdir(normalizedPath, { withFileTypes: true }); + if (!config.flags.enableLocalAgent) { + return await browseHostFilesystem(browsePath); + } - const directories = await Promise.all( - entries - .filter((entry) => entry.isDirectory()) - .map(async (entry) => { - const fullPath = path.join(normalizedPath, entry.name); - - try { - const stats = await fs.stat(fullPath); - return { - name: entry.name, - path: fullPath, - type: "directory" as const, - size: undefined, - modifiedAt: stats.mtimeMs, - }; - } catch { - return { - name: entry.name, - path: fullPath, - type: "directory" as const, - size: undefined, - modifiedAt: undefined, - }; - } - }), - ); - - return { - directories: directories.sort((a, b) => a.name.localeCompare(b.name)), - path: normalizedPath, - }; + const command = await runVolumeCommand(LOCAL_AGENT_ID, { name: "filesystem.browse", path: browsePath }); + return command.result; } catch (error) { throw new InternalServerError(`Failed to browse filesystem: ${toMessage(error)}`); } diff --git a/apps/agent/src/commands/index.ts b/apps/agent/src/commands/index.ts index a17360df..6f73e0c6 100644 --- a/apps/agent/src/commands/index.ts +++ b/apps/agent/src/commands/index.ts @@ -3,6 +3,7 @@ import { handleBackupCancelCommand } from "./backup-cancel"; import { handleBackupRunCommand } from "./backup-run"; import type { ControllerCommandContext } from "../context"; import { handleHeartbeatPingCommand } from "./heartbeat-ping"; +import { handleVolumeCommand } from "./volume"; export const handleControllerCommand = (context: ControllerCommandContext, message: ControllerMessage) => { switch (message.type) { @@ -12,6 +13,9 @@ export const handleControllerCommand = (context: ControllerCommandContext, messa case "backup.cancel": { return handleBackupCancelCommand(context, message.payload); } + case "volume.command": { + return handleVolumeCommand(context, message.payload); + } case "heartbeat.ping": { return handleHeartbeatPingCommand(context, message.payload); } diff --git a/apps/agent/src/commands/volume.ts b/apps/agent/src/commands/volume.ts new file mode 100644 index 00000000..a2243097 --- /dev/null +++ b/apps/agent/src/commands/volume.ts @@ -0,0 +1,76 @@ +import { Effect } from "effect"; +import { + createAgentMessage, + type VolumeCommand, + type VolumeCommandPayload, + type VolumeCommandResult, +} from "@zerobyte/contracts/agent-protocol"; +import { toMessage } from "@zerobyte/core/utils"; +import { createVolumeBackend, getStatFs, getVolumePath, type AgentVolume, type BackendConfig } from "../volume-host"; +import { browseFilesystem, listVolumeFiles, testVolumeConnection } from "../volume-host/operations"; +import type { ControllerCommandContext } from "../context"; + +type VolumeBackedCommand = Extract; + +const asVolume = (volume: VolumeBackedCommand["volume"]): AgentVolume => ({ + ...volume, + config: volume.config as BackendConfig, + provisioningId: volume.provisioningId ?? null, +}); + +const runBackendOperation = async ( + command: Extract, + operation: "mount" | "unmount" | "checkHealth", +) => { + const backend = createVolumeBackend(asVolume(command.volume)); + return backend[operation](); +}; + +const executeVolumeCommand = async (command: VolumeCommand): Promise => { + switch (command.name) { + case "volume.mount": + return { name: command.name, result: await runBackendOperation(command, "mount") }; + case "volume.unmount": + return { name: command.name, result: await runBackendOperation(command, "unmount") }; + case "volume.checkHealth": + return { name: command.name, result: await runBackendOperation(command, "checkHealth") }; + case "volume.statfs": + return { name: command.name, result: await getStatFs(getVolumePath(asVolume(command.volume))) }; + case "volume.listFiles": + return { + name: command.name, + result: await listVolumeFiles(asVolume(command.volume), command.subPath, command.offset, command.limit), + }; + case "volume.testConnection": + return { name: command.name, result: await testVolumeConnection(command.backendConfig as BackendConfig) }; + case "filesystem.browse": + return { name: command.name, result: await browseFilesystem(command.path) }; + } +}; + +export const handleVolumeCommand = (context: ControllerCommandContext, payload: VolumeCommandPayload) => { + return Effect.promise(async () => { + try { + const command = await executeVolumeCommand(payload.command); + await Effect.runPromise( + context.offerOutbound( + createAgentMessage("volume.commandResult", { + commandId: payload.commandId, + status: "success", + command, + }), + ), + ); + } catch (error) { + await Effect.runPromise( + context.offerOutbound( + createAgentMessage("volume.commandResult", { + commandId: payload.commandId, + status: "error", + error: toMessage(error), + }), + ), + ); + } + }); +}; diff --git a/app/server/modules/backends/directory/directory-backend.ts b/apps/agent/src/volume-host/backends/directory.ts similarity index 54% rename from app/server/modules/backends/directory/directory-backend.ts rename to apps/agent/src/volume-host/backends/directory.ts index 8fa02ccf..4033cc10 100644 --- a/app/server/modules/backends/directory/directory-backend.ts +++ b/apps/agent/src/volume-host/backends/directory.ts @@ -1,12 +1,11 @@ import * as fs from "node:fs/promises"; -import { toMessage } from "../../../utils/errors"; import { logger } from "@zerobyte/core/node"; -import type { VolumeBackend } from "../backend"; -import { BACKEND_STATUS, type BackendConfig } from "~/schemas/volumes"; +import { toMessage } from "@zerobyte/core/utils"; +import type { BackendConfig, VolumeBackend } from "../types"; -const mount = async (config: BackendConfig, _volumePath: string) => { +const mount = async (config: BackendConfig, volumePath: string) => { if (config.backend !== "directory") { - return { status: BACKEND_STATUS.error, error: "Invalid backend type" }; + return { status: "error" as const, error: "Invalid backend type" }; } logger.info("Mounting directory volume from:", config.path); @@ -16,33 +15,33 @@ const mount = async (config: BackendConfig, _volumePath: string) => { const stats = await fs.stat(config.path); if (!stats.isDirectory()) { - return { status: BACKEND_STATUS.error, error: "Path is not a directory" }; + return { status: "error" as const, error: "Path is not a directory" }; } - return { status: BACKEND_STATUS.mounted }; + return { status: "mounted" as const }; } catch (error) { logger.error("Failed to mount directory volume:", error); - return { status: BACKEND_STATUS.error, error: toMessage(error) }; + return { status: "error" as const, error: toMessage(error) }; } }; const unmount = async () => { logger.info("Cannot unmount directory volume."); - return { status: BACKEND_STATUS.unmounted }; + return { status: "unmounted" as const }; }; const checkHealth = async (config: BackendConfig) => { if (config.backend !== "directory") { - return { status: BACKEND_STATUS.error, error: "Invalid backend type" }; + return { status: "error" as const, error: "Invalid backend type" }; } try { await fs.access(config.path); - return { status: BACKEND_STATUS.mounted }; + return { status: "mounted" as const }; } catch (error) { logger.error("Directory health check failed:", error); - return { status: BACKEND_STATUS.error, error: toMessage(error) }; + return { status: "error" as const, error: toMessage(error) }; } }; diff --git a/apps/agent/src/volume-host/backends/nfs.ts b/apps/agent/src/volume-host/backends/nfs.ts new file mode 100644 index 00000000..803e9a45 --- /dev/null +++ b/apps/agent/src/volume-host/backends/nfs.ts @@ -0,0 +1,112 @@ +import * as fs from "node:fs/promises"; +import * as os from "node:os"; +import { logger } from "@zerobyte/core/node"; +import { toMessage } from "@zerobyte/core/utils"; +import { OPERATION_TIMEOUT } from "../constants"; +import { withTimeout } from "../timeout"; +import { getMountForPath } from "../fs"; +import type { BackendConfig, VolumeBackend } from "../types"; +import { assertMounted, executeMount, executeUnmount } from "./utils"; + +const checkHealth = async (mountPath: string) => { + const run = async () => { + await assertMounted(mountPath, (fstype) => fstype.startsWith("nfs")); + + logger.debug(`NFS volume at ${mountPath} is healthy and mounted.`); + return { status: "mounted" as const }; + }; + + try { + return await withTimeout(run(), OPERATION_TIMEOUT, "NFS health check"); + } catch (error) { + const message = toMessage(error); + if (message !== "Volume is not mounted") { + logger.error("NFS volume health check failed:", message); + } + return { status: "error" as const, error: message }; + } +}; + +const unmount = async (mountPath: string) => { + if (os.platform() !== "linux") { + logger.error("NFS unmounting is only supported on Linux hosts."); + return { status: "error" as const, error: "NFS unmounting is only supported on Linux hosts." }; + } + + const run = async () => { + const mount = await getMountForPath(mountPath); + if (!mount || mount.mountPoint !== mountPath) { + logger.debug(`Path ${mountPath} is not a mount point. Skipping unmount.`); + return { status: "unmounted" as const }; + } + + await executeUnmount(mountPath); + await fs.rmdir(mountPath).catch(() => {}); + + logger.info(`NFS volume at ${mountPath} unmounted successfully.`); + return { status: "unmounted" as const }; + }; + + try { + return await withTimeout(run(), OPERATION_TIMEOUT, "NFS unmount"); + } catch (error) { + logger.error("Error unmounting NFS volume", { mountPath, error: toMessage(error) }); + return { status: "error" as const, error: toMessage(error) }; + } +}; + +const mount = async (config: BackendConfig, mountPath: string) => { + logger.debug(`Mounting volume ${mountPath}...`); + + if (config.backend !== "nfs") { + logger.error("Provided config is not for NFS backend"); + return { status: "error" as const, error: "Provided config is not for NFS backend" }; + } + + if (os.platform() !== "linux") { + logger.error("NFS mounting is only supported on Linux hosts."); + return { status: "error" as const, error: "NFS mounting is only supported on Linux hosts." }; + } + + const { status } = await checkHealth(mountPath); + if (status === "mounted") return { status: "mounted" as const }; + + if (status === "error") { + logger.debug(`Trying to unmount any existing mounts at ${mountPath} before mounting...`); + await unmount(mountPath); + } + + const run = async () => { + await fs.mkdir(mountPath, { recursive: true }); + const options = [`vers=${config.version}`, `port=${config.port}`]; + if (config.version === "3") options.push("nolock"); + if (config.readOnly) options.push("ro"); + const args = ["-t", "nfs", "-o", options.join(","), `${config.server}:${config.exportPath}`, mountPath]; + + logger.debug(`Mounting volume ${mountPath}...`); + logger.info(`Executing mount: mount ${args.join(" ")}`); + + try { + await executeMount(args); + } catch (error) { + logger.warn(`Initial NFS mount failed, retrying with -i flag: ${toMessage(error)}`); + await executeMount(["-i", ...args]); + } + + logger.info(`NFS volume at ${mountPath} mounted successfully.`); + return { status: "mounted" as const }; + }; + + try { + return await withTimeout(run(), OPERATION_TIMEOUT, "NFS mount"); + } catch (error) { + logger.error("Error mounting NFS volume", { error: toMessage(error) }); + return { status: "error" as const, error: toMessage(error) }; + } +}; + +export const makeNfsBackend = (config: BackendConfig, mountPath: string): VolumeBackend => ({ + mount: () => mount(config, mountPath), + unmount: () => unmount(mountPath), + checkHealth: () => checkHealth(mountPath), +}); diff --git a/apps/agent/src/volume-host/backends/rclone.ts b/apps/agent/src/volume-host/backends/rclone.ts new file mode 100644 index 00000000..011f2cf4 --- /dev/null +++ b/apps/agent/src/volume-host/backends/rclone.ts @@ -0,0 +1,124 @@ +import * as fs from "node:fs/promises"; +import * as os from "node:os"; +import { logger, safeExec } from "@zerobyte/core/node"; +import { toMessage } from "@zerobyte/core/utils"; +import { OPERATION_TIMEOUT, RCLONE_CONFIG_FILE, RCLONE_TIMEOUT } from "../constants"; +import { withTimeout } from "../timeout"; +import { getMountForPath } from "../fs"; +import type { BackendConfig, VolumeBackend } from "../types"; +import { assertMounted, executeUnmount } from "./utils"; + +const checkHealth = async (mountPath: string) => { + const run = async () => { + await assertMounted(mountPath, (fstype) => fstype.includes("rclone")); + + logger.debug(`Rclone volume at ${mountPath} is healthy and mounted.`); + return { status: "mounted" as const }; + }; + + try { + return await withTimeout(run(), OPERATION_TIMEOUT, "Rclone health check"); + } catch (error) { + const message = toMessage(error); + if (message !== "Volume is not mounted") { + logger.error("Rclone volume health check failed:", message); + } + return { status: "error" as const, error: message }; + } +}; + +const unmount = async (mountPath: string) => { + if (os.platform() !== "linux") { + logger.error("Rclone unmounting is only supported on Linux hosts."); + return { status: "error" as const, error: "Rclone unmounting is only supported on Linux hosts." }; + } + + const run = async () => { + const mount = await getMountForPath(mountPath); + if (!mount || mount.mountPoint !== mountPath) { + logger.debug(`Path ${mountPath} is not a mount point. Skipping unmount.`); + return { status: "unmounted" as const }; + } + + await executeUnmount(mountPath); + await fs.rmdir(mountPath).catch(() => {}); + + logger.info(`Rclone volume at ${mountPath} unmounted successfully.`); + return { status: "unmounted" as const }; + }; + + try { + return await withTimeout(run(), OPERATION_TIMEOUT, "Rclone unmount"); + } catch (error) { + logger.error("Error unmounting rclone volume", { mountPath, error: toMessage(error) }); + return { status: "error" as const, error: toMessage(error) }; + } +}; + +const mount = async (config: BackendConfig, mountPath: string) => { + logger.debug(`Mounting rclone volume ${mountPath}...`); + + if (config.backend !== "rclone") { + logger.error("Provided config is not for rclone backend"); + return { status: "error" as const, error: "Provided config is not for rclone backend" }; + } + + if (os.platform() !== "linux") { + logger.error("Rclone mounting is only supported on Linux hosts."); + return { status: "error" as const, error: "Rclone mounting is only supported on Linux hosts." }; + } + + const { status } = await checkHealth(mountPath); + if (status === "mounted") return { status: "mounted" as const }; + + if (status === "error") { + logger.debug(`Trying to unmount any existing mounts at ${mountPath} before mounting...`); + await unmount(mountPath); + } + + const run = async () => { + await fs.mkdir(mountPath, { recursive: true }); + const args = [ + "mount", + `${config.remote}:${config.path}`, + mountPath, + "--daemon", + "--vfs-cache-mode", + "writes", + "--allow-non-empty", + "--allow-other", + ]; + if (config.readOnly) args.push("--read-only"); + + logger.debug(`Mounting rclone volume ${mountPath}...`); + logger.info(`Executing rclone: rclone ${args.join(" ")}`); + + const result = await safeExec({ + command: "rclone", + args, + env: { RCLONE_CONFIG: RCLONE_CONFIG_FILE }, + timeout: RCLONE_TIMEOUT, + }); + if (result.exitCode !== 0) { + const errorMsg = result.stderr.toString() || result.stdout.toString() || "Unknown error"; + throw new Error(`Failed to mount rclone volume: ${errorMsg}`); + } + + logger.info(`Rclone volume at ${mountPath} mounted successfully.`); + return { status: "mounted" as const }; + }; + + try { + return await withTimeout(run(), RCLONE_TIMEOUT, "Rclone mount"); + } catch (error) { + const errorMsg = toMessage(error); + logger.error("Error mounting rclone volume", { error: errorMsg }); + return { status: "error" as const, error: errorMsg }; + } +}; + +export const makeRcloneBackend = (config: BackendConfig, mountPath: string): VolumeBackend => ({ + mount: () => mount(config, mountPath), + unmount: () => unmount(mountPath), + checkHealth: () => checkHealth(mountPath), +}); diff --git a/apps/agent/src/volume-host/backends/sftp.ts b/apps/agent/src/volume-host/backends/sftp.ts new file mode 100644 index 00000000..23ed84e5 --- /dev/null +++ b/apps/agent/src/volume-host/backends/sftp.ts @@ -0,0 +1,161 @@ +import * as fs from "node:fs/promises"; +import * as os from "node:os"; +import * as path from "node:path"; +import { spawn } from "node:child_process"; +import { FILE_MODES, logger, writeFileWithMode } from "@zerobyte/core/node"; +import { toMessage } from "@zerobyte/core/utils"; +import { OPERATION_TIMEOUT, SSH_KEYS_DIR } from "../constants"; +import { getMountForPath } from "../fs"; +import { withTimeout } from "../timeout"; +import type { BackendConfig, VolumeBackend } from "../types"; +import { executeUnmount } from "./utils"; + +const getPrivateKeyPath = (mountPath: string) => path.join(SSH_KEYS_DIR, `${path.basename(mountPath)}.key`); +const getKnownHostsPath = (mountPath: string) => path.join(SSH_KEYS_DIR, `${path.basename(mountPath)}.known_hosts`); + +const runSshfs = async (args: string[], password?: string) => + new Promise((resolve, reject) => { + const child = spawn("sshfs", args, { stdio: ["pipe", "pipe", "pipe"] }); + let stdout = ""; + let stderr = ""; + + child.stdout.setEncoding("utf8"); + child.stderr.setEncoding("utf8"); + + child.stdout.on("data", (data) => { + stdout += data; + }); + child.stderr.on("data", (data) => { + stderr += data; + }); + child.on("error", (error) => { + reject(new Error(`Failed to start sshfs: ${error.message}`)); + }); + child.on("close", (code) => { + if (code === 0) { + resolve(); + return; + } + + const errorMsg = stderr.trim() || stdout.trim() || "Unknown error"; + reject(new Error(`Failed to mount SFTP volume: ${errorMsg}`)); + }); + + if (password) child.stdin.write(password); + child.stdin.end(); + }); + +const checkHealth = async (mountPath: string) => { + const mount = await getMountForPath(mountPath); + if (!mount || mount.mountPoint !== mountPath) return { status: "unmounted" as const }; + if (mount.fstype !== "fuse.sshfs") { + return { status: "error" as const, error: `Invalid filesystem type: ${mount.fstype} (expected fuse.sshfs)` }; + } + return { status: "mounted" as const }; +}; + +const unmount = async (mountPath: string) => { + if (os.platform() !== "linux") { + logger.error("SFTP unmounting is only supported on Linux hosts."); + return { status: "error" as const, error: "SFTP unmounting is only supported on Linux hosts." }; + } + + const run = async () => { + const mount = await getMountForPath(mountPath); + if (!mount || mount.mountPoint !== mountPath) { + logger.debug(`Path ${mountPath} is not a mount point. Skipping unmount.`); + } else { + await executeUnmount(mountPath); + } + + await fs.unlink(getPrivateKeyPath(mountPath)).catch(() => {}); + await fs.unlink(getKnownHostsPath(mountPath)).catch(() => {}); + await fs.rmdir(mountPath).catch(() => {}); + + logger.info(`SFTP volume at ${mountPath} unmounted successfully.`); + return { status: "unmounted" as const }; + }; + + try { + return await withTimeout(run(), OPERATION_TIMEOUT, "SFTP unmount"); + } catch (error) { + logger.error("Error unmounting SFTP volume", { mountPath, error: toMessage(error) }); + return { status: "error" as const, error: toMessage(error) }; + } +}; + +const mount = async (config: BackendConfig, mountPath: string) => { + logger.debug(`Mounting SFTP volume ${mountPath}...`); + + if (config.backend !== "sftp") { + logger.error("Provided config is not for SFTP backend"); + return { status: "error" as const, error: "Provided config is not for SFTP backend" }; + } + + if (os.platform() !== "linux") { + logger.error("SFTP mounting is only supported on Linux hosts."); + return { status: "error" as const, error: "SFTP mounting is only supported on Linux hosts." }; + } + + const { status } = await checkHealth(mountPath); + if (status === "mounted") return { status: "mounted" as const }; + + if (status === "error") { + logger.debug(`Trying to unmount any existing mounts at ${mountPath} before mounting...`); + await unmount(mountPath); + } + + const run = async () => { + await fs.mkdir(mountPath, { recursive: true }); + await fs.mkdir(SSH_KEYS_DIR, { recursive: true }); + const { uid, gid } = os.userInfo(); + const options = [ + "reconnect", + "ServerAliveInterval=15", + "ServerAliveCountMax=3", + "allow_other", + `uid=${uid}`, + `gid=${gid}`, + ]; + + if (config.skipHostKeyCheck) { + options.push("StrictHostKeyChecking=no", "UserKnownHostsFile=/dev/null"); + } else if (config.knownHosts) { + await writeFileWithMode(getKnownHostsPath(mountPath), config.knownHosts, FILE_MODES.ownerReadWrite); + options.push(`UserKnownHostsFile=${getKnownHostsPath(mountPath)}`, "StrictHostKeyChecking=yes"); + } else { + options.push("StrictHostKeyChecking=yes"); + } + + if (config.readOnly) options.push("ro"); + if (config.port) options.push(`port=${config.port}`); + if (config.privateKey) { + let key = config.privateKey.replace(/\r\n/g, "\n"); + if (!key.endsWith("\n")) key += "\n"; + await writeFileWithMode(getPrivateKeyPath(mountPath), key, FILE_MODES.ownerReadWrite); + options.push(`IdentityFile=${getPrivateKeyPath(mountPath)}`); + } + + const args = [`${config.username}@${config.host}:${config.path || ""}`, mountPath, "-o", options.join(",")]; + if (config.password) args.push("-o", "password_stdin"); + + logger.info(`Executing sshfs: sshfs ${args.join(" ")}`); + await runSshfs(args, config.password); + + logger.info(`SFTP volume at ${mountPath} mounted successfully.`); + return { status: "mounted" as const }; + }; + + try { + return await withTimeout(run(), OPERATION_TIMEOUT * 2, "SFTP mount"); + } catch (error) { + logger.error("Error mounting SFTP volume", { error: toMessage(error) }); + return { status: "error" as const, error: toMessage(error) }; + } +}; + +export const makeSftpBackend = (config: BackendConfig, mountPath: string): VolumeBackend => ({ + mount: () => mount(config, mountPath), + unmount: () => unmount(mountPath), + checkHealth: () => checkHealth(mountPath), +}); diff --git a/apps/agent/src/volume-host/backends/smb.ts b/apps/agent/src/volume-host/backends/smb.ts new file mode 100644 index 00000000..767d27d5 --- /dev/null +++ b/apps/agent/src/volume-host/backends/smb.ts @@ -0,0 +1,124 @@ +import * as fs from "node:fs/promises"; +import * as os from "node:os"; +import { logger } from "@zerobyte/core/node"; +import { toMessage } from "@zerobyte/core/utils"; +import { OPERATION_TIMEOUT } from "../constants"; +import { withTimeout } from "../timeout"; +import { getMountForPath } from "../fs"; +import type { BackendConfig, VolumeBackend } from "../types"; +import { assertMounted, executeMount, executeUnmount } from "./utils"; + +const checkHealth = async (mountPath: string) => { + const run = async () => { + await assertMounted(mountPath, (fstype) => fstype === "cifs"); + + logger.debug(`SMB volume at ${mountPath} is healthy and mounted.`); + return { status: "mounted" as const }; + }; + + try { + return await withTimeout(run(), OPERATION_TIMEOUT, "SMB health check"); + } catch (error) { + const message = toMessage(error); + if (message !== "Volume is not mounted") { + logger.error("SMB volume health check failed:", message); + } + return { status: "error" as const, error: message }; + } +}; + +const unmount = async (mountPath: string) => { + if (os.platform() !== "linux") { + logger.error("SMB unmounting is only supported on Linux hosts."); + return { status: "error" as const, error: "SMB unmounting is only supported on Linux hosts." }; + } + + const run = async () => { + const mount = await getMountForPath(mountPath); + if (!mount || mount.mountPoint !== mountPath) { + logger.debug(`Path ${mountPath} is not a mount point. Skipping unmount.`); + return { status: "unmounted" as const }; + } + + await executeUnmount(mountPath); + await fs.rmdir(mountPath).catch(() => {}); + + logger.info(`SMB volume at ${mountPath} unmounted successfully.`); + return { status: "unmounted" as const }; + }; + + try { + return await withTimeout(run(), OPERATION_TIMEOUT, "SMB unmount"); + } catch (error) { + logger.error("Error unmounting SMB volume", { mountPath, error: toMessage(error) }); + return { status: "error" as const, error: toMessage(error) }; + } +}; + +const mount = async (config: BackendConfig, mountPath: string) => { + logger.debug(`Mounting SMB volume ${mountPath}...`); + + if (config.backend !== "smb") { + logger.error("Provided config is not for SMB backend"); + return { status: "error" as const, error: "Provided config is not for SMB backend" }; + } + + if (os.platform() !== "linux") { + logger.error("SMB mounting is only supported on Linux hosts."); + return { status: "error" as const, error: "SMB mounting is only supported on Linux hosts." }; + } + + const { status } = await checkHealth(mountPath); + if (status === "mounted") return { status: "mounted" as const }; + + if (status === "error") { + logger.debug(`Trying to unmount any existing mounts at ${mountPath} before mounting...`); + await unmount(mountPath); + } + + const run = async () => { + await fs.mkdir(mountPath, { recursive: true }); + const { uid, gid } = os.userInfo(); + const options = [`port=${config.port}`, `uid=${uid}`, `gid=${gid}`, "iocharset=utf8"]; + + if (config.guest) { + options.push("username=guest", "password="); + } else { + const safePassword = (config.password ?? "").replace(/\\/g, "\\\\").replace(/,/g, "\\,"); + options.push(`username=${config.username ?? "user"}`, `password=${safePassword}`); + } + + if (config.domain) options.push(`domain=${config.domain}`); + if (config.vers && config.vers !== "auto") options.push(`vers=${config.vers}`); + if (config.readOnly) options.push("ro"); + + const source = `//${config.server}/${config.share}`; + const args = ["-t", "cifs", "-o", options.join(","), source, mountPath]; + + logger.debug(`Mounting SMB volume ${mountPath}...`); + logger.info(`Executing SMB mount for ${source} at ${mountPath}`); + + try { + await executeMount(args); + } catch (error) { + logger.error(`SMB mount failed: ${toMessage(error)}`); + throw error; + } + + logger.info(`SMB volume at ${mountPath} mounted successfully.`); + return { status: "mounted" as const }; + }; + + try { + return await withTimeout(run(), OPERATION_TIMEOUT, "SMB mount"); + } catch (error) { + logger.error("Error mounting SMB volume", { error: toMessage(error) }); + return { status: "error" as const, error: toMessage(error) }; + } +}; + +export const makeSmbBackend = (config: BackendConfig, mountPath: string): VolumeBackend => ({ + mount: () => mount(config, mountPath), + unmount: () => unmount(mountPath), + checkHealth: () => checkHealth(mountPath), +}); diff --git a/app/server/modules/backends/utils/backend-utils.ts b/apps/agent/src/volume-host/backends/utils.ts similarity index 62% rename from app/server/modules/backends/utils/backend-utils.ts rename to apps/agent/src/volume-host/backends/utils.ts index a9e9d2b6..7a5ac091 100644 --- a/app/server/modules/backends/utils/backend-utils.ts +++ b/apps/agent/src/volume-host/backends/utils.ts @@ -1,7 +1,6 @@ import * as fs from "node:fs/promises"; -import { logger } from "@zerobyte/core/node"; -import { safeExec } from "@zerobyte/core/node"; -import { getMountForPath } from "../../../utils/mountinfo"; +import { logger, safeExec } from "@zerobyte/core/node"; +import { getMountForPath } from "../fs"; export const executeMount = async (args: string[]): Promise => { const shouldBeVerbose = process.env.LOG_LEVEL === "debug" || process.env.NODE_ENV !== "production"; @@ -10,7 +9,6 @@ export const executeMount = async (args: string[]): Promise => { logger.debug(`Executing mount ${effectiveArgs.join(" ")}`); const result = await safeExec({ command: "mount", args: effectiveArgs, timeout: 10000 }); - const stdout = result.stdout.toString().trim(); const stderr = result.stderr.toString().trim(); @@ -26,37 +24,29 @@ export const executeMount = async (args: string[]): Promise => { throw new Error(`Mount command failed with exit code ${result.exitCode}: ${stderr || stdout || "unknown error"}`); }; -export const executeUnmount = async (path: string): Promise => { - let stderr: string | undefined; - - logger.debug(`Executing umount -l ${path}`); - const result = await safeExec({ command: "umount", args: ["-l", path], timeout: 10000 }); - - stderr = result.stderr.toString(); - - if (stderr?.trim()) { - logger.warn(stderr.trim()); - } +export const executeUnmount = async (mountPath: string): Promise => { + logger.debug(`Executing umount -l ${mountPath}`); + const result = await safeExec({ command: "umount", args: ["-l", mountPath], timeout: 10000 }); + const stderr = result.stderr.toString(); + if (stderr.trim()) logger.warn(stderr.trim()); if (result.exitCode !== 0) { - throw new Error(`Mount command failed with exit code ${result.exitCode}: ${stderr?.trim()}`); + throw new Error(`Mount command failed with exit code ${result.exitCode}: ${stderr.trim()}`); } }; -export const assertMounted = async (path: string, isExpectedFilesystem: (fstype: string) => boolean) => { +export const assertMounted = async (mountPath: string, isExpectedFilesystem: (fstype: string) => boolean) => { try { - await fs.access(path); + await fs.access(mountPath); } catch { throw new Error("Volume is not mounted"); } - const mount = await getMountForPath(path); - - if (!mount || mount.mountPoint !== path) { + const mount = await getMountForPath(mountPath); + if (!mount || mount.mountPoint !== mountPath) { throw new Error("Volume is not mounted"); } - if (!isExpectedFilesystem(mount.fstype)) { - throw new Error(`Path ${path} is not mounted as correct fstype (found ${mount.fstype}).`); + throw new Error(`Path ${mountPath} is not mounted as correct fstype (found ${mount.fstype}).`); } }; diff --git a/apps/agent/src/volume-host/backends/webdav.ts b/apps/agent/src/volume-host/backends/webdav.ts new file mode 100644 index 00000000..dd3548e8 --- /dev/null +++ b/apps/agent/src/volume-host/backends/webdav.ts @@ -0,0 +1,144 @@ +import * as fs from "node:fs/promises"; +import * as os from "node:os"; +import { logger } from "@zerobyte/core/node"; +import { toMessage } from "@zerobyte/core/utils"; +import { OPERATION_TIMEOUT } from "../constants"; +import { withTimeout } from "../timeout"; +import { getMountForPath } from "../fs"; +import type { BackendConfig, VolumeBackend } from "../types"; +import { assertMounted, executeMount, executeUnmount } from "./utils"; + +const checkHealth = async (mountPath: string) => { + const run = async () => { + await assertMounted(mountPath, (fstype) => fstype === "fuse" || fstype === "davfs"); + + logger.debug(`WebDAV volume at ${mountPath} is healthy and mounted.`); + return { status: "mounted" as const }; + }; + + try { + return await withTimeout(run(), OPERATION_TIMEOUT, "WebDAV health check"); + } catch (error) { + const message = toMessage(error); + if (message !== "Volume is not mounted") { + logger.error("WebDAV volume health check failed:", message); + } + return { status: "error" as const, error: message }; + } +}; + +const unmount = async (mountPath: string) => { + if (os.platform() !== "linux") { + logger.error("WebDAV unmounting is only supported on Linux hosts."); + return { status: "error" as const, error: "WebDAV unmounting is only supported on Linux hosts." }; + } + + const run = async () => { + const mount = await getMountForPath(mountPath); + if (!mount || mount.mountPoint !== mountPath) { + logger.debug(`Path ${mountPath} is not a mount point. Skipping unmount.`); + return { status: "unmounted" as const }; + } + + await executeUnmount(mountPath); + await fs.rmdir(mountPath).catch(() => {}); + + logger.info(`WebDAV volume at ${mountPath} unmounted successfully.`); + return { status: "unmounted" as const }; + }; + + try { + return await withTimeout(run(), OPERATION_TIMEOUT, "WebDAV unmount"); + } catch (error) { + logger.error("Error unmounting WebDAV volume", { mountPath, error: toMessage(error) }); + return { status: "error" as const, error: toMessage(error) }; + } +}; + +const mount = async (config: BackendConfig, mountPath: string) => { + logger.debug(`Mounting WebDAV volume ${mountPath}...`); + + if (config.backend !== "webdav") { + logger.error("Provided config is not for WebDAV backend"); + return { status: "error" as const, error: "Provided config is not for WebDAV backend" }; + } + + if (os.platform() !== "linux") { + logger.error("WebDAV mounting is only supported on Linux hosts."); + return { status: "error" as const, error: "WebDAV mounting is only supported on Linux hosts." }; + } + + const { status } = await checkHealth(mountPath); + if (status === "mounted") return { status: "mounted" as const }; + + if (status === "error") { + logger.debug(`Trying to unmount any existing mounts at ${mountPath} before mounting...`); + await unmount(mountPath); + } + + const run = async () => { + await fs.mkdir(mountPath, { recursive: true }).catch((error) => { + logger.warn(`Failed to create directory ${mountPath}: ${toMessage(error)}`); + }); + const protocol = config.ssl ? "https" : "http"; + const defaultPort = config.ssl ? 443 : 80; + const source = `${protocol}://${config.server}${config.port !== defaultPort ? `:${config.port}` : ""}${config.path}`; + const { uid, gid } = os.userInfo(); + const options = config.readOnly + ? [`uid=${uid}`, `gid=${gid}`, "file_mode=0444", "dir_mode=0555", "ro"] + : [`uid=${uid}`, `gid=${gid}`, "file_mode=0664", "dir_mode=0775"]; + + if (config.username && config.password) { + const entry = [source, config.username, config.password] + .map((value) => value.replace(/[\r\n\t\s]+/g, " ")) + .join(" "); + await fs.appendFile("/etc/davfs2/secrets", `${entry}\n`, { mode: 0o600 }); + } + + logger.debug(`Mounting WebDAV volume ${mountPath}...`); + + const args = ["-t", "davfs", "-o", options.join(","), source, mountPath]; + try { + await executeMount(args); + } catch (error) { + logger.warn(`Initial WebDAV mount failed, retrying with -i flag: ${toMessage(error)}`); + await executeMount(["-i", ...args]); + } + + logger.info(`WebDAV volume at ${mountPath} mounted successfully.`); + return { status: "mounted" as const }; + }; + + try { + return await withTimeout(run(), OPERATION_TIMEOUT, "WebDAV mount"); + } catch (error) { + const message = toMessage(error); + if (message.includes("already mounted")) return { status: "mounted" as const }; + + logger.error("Error mounting WebDAV volume", { error: message }); + + if (message.includes("option") && message.includes("requires argument")) { + return { + status: "error" as const, + error: "Invalid mount options. Please check your WebDAV server configuration.", + }; + } + if (message.includes("connection refused") || message.includes("Connection refused")) { + return { + status: "error" as const, + error: "Cannot connect to WebDAV server. Please check the server address and port.", + }; + } + if (message.includes("unauthorized") || message.includes("Unauthorized")) { + return { status: "error" as const, error: "Authentication failed. Please check your username and password." }; + } + + return { status: "error" as const, error: message }; + } +}; + +export const makeWebdavBackend = (config: BackendConfig, mountPath: string): VolumeBackend => ({ + mount: () => mount(config, mountPath), + unmount: () => unmount(mountPath), + checkHealth: () => checkHealth(mountPath), +}); diff --git a/apps/agent/src/volume-host/constants.ts b/apps/agent/src/volume-host/constants.ts new file mode 100644 index 00000000..c11d135d --- /dev/null +++ b/apps/agent/src/volume-host/constants.ts @@ -0,0 +1,9 @@ +import * as path from "node:path"; + +export const OPERATION_TIMEOUT = 5000; +export const VOLUME_MOUNT_BASE = process.env.ZEROBYTE_VOLUMES_DIR || "/var/lib/zerobyte/volumes"; +export const SSH_KEYS_DIR = "/var/lib/zerobyte/ssh"; +export const RCLONE_CONFIG_DIR = process.env.RCLONE_CONFIG_DIR || "/root/.config/rclone"; +export const RCLONE_CONFIG_FILE = path.join(RCLONE_CONFIG_DIR, "rclone.conf"); +const serverIdleTimeout = Number(process.env.SERVER_IDLE_TIMEOUT ?? 60); +export const RCLONE_TIMEOUT = (Number.isFinite(serverIdleTimeout) ? serverIdleTimeout : 60) * 1000; diff --git a/apps/agent/src/volume-host/fs.ts b/apps/agent/src/volume-host/fs.ts new file mode 100644 index 00000000..a97a47ec --- /dev/null +++ b/apps/agent/src/volume-host/fs.ts @@ -0,0 +1,66 @@ +import * as fs from "node:fs/promises"; +import { isPathWithin } from "@zerobyte/core/utils"; + +type MountInfo = { + mountPoint: string; + fstype: string; +}; + +const unescapeMount = (value: string) => + value.replace(/\\([0-7]{3})/g, (_, oct) => String.fromCharCode(parseInt(oct, 8))); + +export const readMountInfo = async (): Promise => { + const text = await fs.readFile("/proc/self/mountinfo", "utf-8"); + const result: MountInfo[] = []; + + for (const line of text.split("\n")) { + if (!line) continue; + const sep = line.indexOf(" - "); + if (sep === -1) continue; + + const left = line.slice(0, sep).split(" "); + const right = line.slice(sep + 3).split(" "); + const mpRaw = left[4]; + const fstype = right[0]; + + if (!mpRaw || !fstype) continue; + result.push({ mountPoint: unescapeMount(mpRaw), fstype }); + } + + return result; +}; + +export const getMountForPath = async (targetPath: string): Promise => { + const mounts = await readMountInfo(); + let best: MountInfo | undefined; + + for (const mount of mounts) { + if (!isPathWithin(mount.mountPoint, targetPath)) continue; + if (!best || mount.mountPoint.length > best.mountPoint.length) { + best = mount; + } + } + + return best; +}; + +export const getStatFs = async (mountPoint: string) => { + const stat = await fs.statfs(mountPoint, { bigint: true }); + const unit = stat.bsize > 0n ? stat.bsize : 1n; + const blocks = stat.blocks > 0n ? stat.blocks : 0n; + let bfree = stat.bfree > 0n ? stat.bfree : 0n; + if (bfree > blocks) bfree = blocks; + const bavail = stat.bavail > 0n ? stat.bavail : 0n; + const max = BigInt(Number.MAX_SAFE_INTEGER); + const toNumber = (value: bigint) => (value > max ? Number.MAX_SAFE_INTEGER : Number(value)); + + return { + total: toNumber(blocks * unit), + used: toNumber((blocks - bfree) * unit), + free: toNumber(bavail * unit), + }; +}; + +export const isNodeJSErrnoException = (error: unknown): error is NodeJS.ErrnoException => { + return error instanceof Error && "code" in error; +}; diff --git a/apps/agent/src/volume-host/index.ts b/apps/agent/src/volume-host/index.ts new file mode 100644 index 00000000..d704326e --- /dev/null +++ b/apps/agent/src/volume-host/index.ts @@ -0,0 +1,31 @@ +import { makeDirectoryBackend } from "./backends/directory"; +import { makeNfsBackend } from "./backends/nfs"; +import { makeRcloneBackend } from "./backends/rclone"; +import { makeSftpBackend } from "./backends/sftp"; +import { makeSmbBackend } from "./backends/smb"; +import { makeWebdavBackend } from "./backends/webdav"; +import { getVolumePath } from "./paths"; +import type { AgentVolume, VolumeBackend } from "./types"; + +export { getStatFs, isNodeJSErrnoException } from "./fs"; +export { getVolumePath } from "./paths"; +export type { AgentVolume, BackendConfig, VolumeBackend } from "./types"; + +export const createVolumeBackend = (volume: AgentVolume, mountPath = getVolumePath(volume)): VolumeBackend => { + switch (volume.config.backend) { + case "directory": + return makeDirectoryBackend(volume.config, mountPath); + case "nfs": + return makeNfsBackend(volume.config, mountPath); + case "smb": + return makeSmbBackend(volume.config, mountPath); + case "webdav": + return makeWebdavBackend(volume.config, mountPath); + case "rclone": + return makeRcloneBackend(volume.config, mountPath); + case "sftp": + return makeSftpBackend(volume.config, mountPath); + } + + throw new Error("Unsupported backend"); +}; diff --git a/apps/agent/src/volume-host/operations.ts b/apps/agent/src/volume-host/operations.ts new file mode 100644 index 00000000..ae5feabb --- /dev/null +++ b/apps/agent/src/volume-host/operations.ts @@ -0,0 +1,156 @@ +import * as fs from "node:fs/promises"; +import * as os from "node:os"; +import * as path from "node:path"; +import { toMessage } from "@zerobyte/core/utils"; +import { createVolumeBackend, getVolumePath, isNodeJSErrnoException } from "."; +import type { AgentVolume, BackendConfig } from "./types"; + +const DEFAULT_PAGE_SIZE = 500; +const MAX_PAGE_SIZE = 500; + +export const listVolumeFiles = async ( + volume: AgentVolume, + subPath?: string, + offset: number = 0, + limit: number = DEFAULT_PAGE_SIZE, +) => { + const volumePath = getVolumePath(volume); + const requestedPath = subPath ? path.join(volumePath, subPath) : volumePath; + const normalizedPath = path.normalize(requestedPath); + const relative = path.relative(volumePath, normalizedPath); + + if (relative.startsWith("..") || path.isAbsolute(relative)) { + throw new Error("Invalid path"); + } + + const pageSize = Math.min(Math.max(limit, 1), MAX_PAGE_SIZE); + const startOffset = Math.max(offset, 0); + + try { + const dirents = await fs.readdir(normalizedPath, { withFileTypes: true }); + + dirents.sort((a, b) => { + const aIsDir = a.isDirectory(); + const bIsDir = b.isDirectory(); + + if (aIsDir === bIsDir) { + return a.name.localeCompare(b.name); + } + return aIsDir ? -1 : 1; + }); + + const total = dirents.length; + const paginatedDirents = dirents.slice(startOffset, startOffset + pageSize); + + const entries = ( + await Promise.all( + paginatedDirents.map(async (dirent) => { + const fullPath = path.join(normalizedPath, dirent.name); + + try { + const stats = await fs.stat(fullPath); + const relativePath = path.relative(volumePath, fullPath); + + return { + name: dirent.name, + path: `/${relativePath}`, + type: dirent.isDirectory() ? ("directory" as const) : ("file" as const), + size: dirent.isFile() ? stats.size : undefined, + modifiedAt: stats.mtimeMs, + }; + } catch { + return null; + } + }), + ) + ).filter((entry) => entry !== null); + + return { + files: entries, + path: subPath || "/", + offset: startOffset, + limit: pageSize, + total, + hasMore: startOffset + pageSize < total, + }; + } catch (error) { + if (isNodeJSErrnoException(error) && error.code === "ENOENT") { + throw new Error("Directory not found"); + } + if (toMessage(error) === "Invalid path") { + throw error; + } + throw new Error(`Failed to list files: ${toMessage(error)}`); + } +}; + +export const browseFilesystem = async (browsePath: string) => { + const normalizedPath = path.normalize(browsePath); + const entries = await fs.readdir(normalizedPath, { withFileTypes: true }); + + const directories = await Promise.all( + entries + .filter((entry) => entry.isDirectory()) + .map(async (entry) => { + const fullPath = path.join(normalizedPath, entry.name); + + try { + const stats = await fs.stat(fullPath); + return { + name: entry.name, + path: fullPath, + type: "directory" as const, + size: undefined, + modifiedAt: stats.mtimeMs, + }; + } catch { + return { + name: entry.name, + path: fullPath, + type: "directory" as const, + size: undefined, + modifiedAt: undefined, + }; + } + }), + ); + + return { + directories: directories.sort((a, b) => a.name.localeCompare(b.name)), + path: normalizedPath, + }; +}; + +export const testVolumeConnection = async (backendConfig: BackendConfig) => { + const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "zerobyte-test-")); + try { + const mockVolume: AgentVolume = { + id: 0, + shortId: "test", + name: "test-connection", + config: backendConfig, + createdAt: Date.now(), + updatedAt: Date.now(), + lastHealthCheck: Date.now(), + type: backendConfig.backend, + status: "unmounted", + lastError: null, + provisioningId: null, + autoRemount: true, + agentId: "local", + organizationId: "test-org", + }; + + const backend = createVolumeBackend(mockVolume, tempDir); + const { error } = await backend.mount(); + + await backend.unmount(); + + return { + success: !error, + message: error ? toMessage(error) : "Connection successful", + }; + } finally { + await fs.rm(tempDir, { recursive: true, force: true }); + } +}; diff --git a/apps/agent/src/volume-host/paths.ts b/apps/agent/src/volume-host/paths.ts new file mode 100644 index 00000000..f5a1b59e --- /dev/null +++ b/apps/agent/src/volume-host/paths.ts @@ -0,0 +1,10 @@ +import { VOLUME_MOUNT_BASE } from "./constants"; +import type { AgentVolume } from "./types"; + +export const getVolumePath = (volume: AgentVolume) => { + if (volume.config.backend === "directory") { + return volume.config.path; + } + + return `${VOLUME_MOUNT_BASE}/${volume.shortId}/_data`; +}; diff --git a/apps/agent/src/volume-host/timeout.ts b/apps/agent/src/volume-host/timeout.ts new file mode 100644 index 00000000..336ca1e0 --- /dev/null +++ b/apps/agent/src/volume-host/timeout.ts @@ -0,0 +1,21 @@ +class TimeoutError extends Error { + code = "ETIMEOUT"; + constructor(message: string) { + super(message); + this.name = "TimeoutError"; + } +} + +export const withTimeout = async (promise: Promise, ms: number, label: string): Promise => { + let timeout: ReturnType | undefined; + try { + return await Promise.race([ + promise, + new Promise((_, reject) => { + timeout = setTimeout(() => reject(new TimeoutError(`${label} timed out after ${ms}ms`)), ms); + }), + ]); + } finally { + if (timeout) clearTimeout(timeout); + } +}; diff --git a/apps/agent/src/volume-host/types.ts b/apps/agent/src/volume-host/types.ts new file mode 100644 index 00000000..72b32826 --- /dev/null +++ b/apps/agent/src/volume-host/types.ts @@ -0,0 +1,67 @@ +export type BackendStatus = "mounted" | "unmounted" | "error"; + +type BaseConfig = { backend: string; readOnly?: boolean }; + +export type BackendConfig = + | (BaseConfig & { backend: "directory"; path: string }) + | (BaseConfig & { backend: "nfs"; server: string; exportPath: string; port: number; version: "3" | "4" | "4.1" }) + | (BaseConfig & { + backend: "smb"; + server: string; + share: string; + username?: string; + password?: string; + guest?: boolean; + vers?: "1.0" | "2.0" | "2.1" | "3.0" | "auto"; + domain?: string; + port: number; + }) + | (BaseConfig & { + backend: "webdav"; + server: string; + path: string; + username?: string; + password?: string; + port: number; + ssl?: boolean; + }) + | (BaseConfig & { backend: "rclone"; remote: string; path: string }) + | (BaseConfig & { + backend: "sftp"; + host: string; + port: number; + username: string; + password?: string; + privateKey?: string; + path: string; + skipHostKeyCheck?: boolean; + knownHosts?: string; + }); + +export type AgentVolume = { + id: number; + shortId: string; + name: string; + config: BackendConfig; + createdAt: number; + updatedAt: number; + lastHealthCheck: number; + type: string; + status: BackendStatus; + lastError: string | null; + provisioningId?: string | null; + autoRemount: boolean; + agentId: string; + organizationId: string; +}; + +export type OperationResult = { + status: BackendStatus; + error?: string; +}; + +export type VolumeBackend = { + mount: () => Promise; + unmount: () => Promise; + checkHealth: () => Promise; +}; diff --git a/packages/contracts/src/agent-protocol.ts b/packages/contracts/src/agent-protocol.ts index 2cbd1c75..2be20a16 100644 --- a/packages/contracts/src/agent-protocol.ts +++ b/packages/contracts/src/agent-protocol.ts @@ -51,6 +51,111 @@ const backupCancelSchema = z.object({ payload: z.object({ jobId: z.string(), scheduleId: z.string() }), }); +const backendStatusSchema = z.enum(["mounted", "unmounted", "error"]); + +const volumeSchema = z.object({ + id: z.number(), + shortId: z.string(), + name: z.string(), + path: z.string().nullable().optional(), + config: z.record(z.string(), z.unknown()).and(z.object({ backend: z.string() })), + createdAt: z.number(), + updatedAt: z.number(), + lastHealthCheck: z.number(), + type: z.string(), + status: backendStatusSchema, + lastError: z.string().nullable(), + provisioningId: z.string().nullable().optional(), + autoRemount: z.boolean(), + agentId: z.string(), + organizationId: z.string(), +}); + +const volumeOperationResultSchema = z.object({ + status: backendStatusSchema, + error: z.string().optional(), +}); + +const statfsSchema = z.object({ + total: z.number().optional(), + used: z.number().optional(), + free: z.number().optional(), +}); + +const fileEntrySchema = z.object({ + name: z.string(), + path: z.string(), + type: z.enum(["directory", "file"]), + size: z.number().optional(), + modifiedAt: z.number().optional(), +}); + +const directoryEntrySchema = z.object({ + name: z.string(), + path: z.string(), + type: z.literal("directory"), + size: z.undefined().optional(), + modifiedAt: z.number().optional(), +}); + +const volumeCommandSchema = z.discriminatedUnion("name", [ + z.object({ name: z.literal("volume.mount"), volume: volumeSchema }), + z.object({ name: z.literal("volume.unmount"), volume: volumeSchema }), + z.object({ name: z.literal("volume.checkHealth"), volume: volumeSchema }), + z.object({ name: z.literal("volume.statfs"), volume: volumeSchema }), + z.object({ + name: z.literal("volume.listFiles"), + volume: volumeSchema, + subPath: z.string().optional(), + offset: z.number(), + limit: z.number(), + }), + z.object({ name: z.literal("volume.testConnection"), backendConfig: z.record(z.string(), z.unknown()) }), + z.object({ name: z.literal("filesystem.browse"), path: z.string() }), +]); + +const volumeCommandRequestSchema = z.object({ + type: z.literal("volume.command"), + payload: z.object({ + commandId: z.string(), + command: volumeCommandSchema, + }), +}); + +const volumeCommandResultSchema = z.discriminatedUnion("name", [ + z.object({ name: z.literal("volume.mount"), result: volumeOperationResultSchema }), + z.object({ name: z.literal("volume.unmount"), result: volumeOperationResultSchema }), + z.object({ name: z.literal("volume.checkHealth"), result: volumeOperationResultSchema }), + z.object({ name: z.literal("volume.statfs"), result: statfsSchema }), + z.object({ + name: z.literal("volume.listFiles"), + result: z.object({ + files: z.array(fileEntrySchema), + path: z.string(), + offset: z.number(), + limit: z.number(), + total: z.number(), + hasMore: z.boolean(), + }), + }), + z.object({ + name: z.literal("volume.testConnection"), + result: z.object({ success: z.boolean(), message: z.string() }), + }), + z.object({ + name: z.literal("filesystem.browse"), + result: z.object({ directories: z.array(directoryEntrySchema), path: z.string() }), + }), +]); + +const volumeCommandResponseSchema = z.object({ + type: z.literal("volume.commandResult"), + payload: z.discriminatedUnion("status", [ + z.object({ commandId: z.string(), status: z.literal("success"), command: volumeCommandResultSchema }), + z.object({ commandId: z.string(), status: z.literal("error"), error: z.string() }), + ]), +}); + const heartbeatPingSchema = z.object({ type: z.literal("heartbeat.ping"), payload: z.object({ sentAt: z.number() }), @@ -113,6 +218,7 @@ const heartbeatPongSchema = z.object({ const controllerMessageSchema = z.discriminatedUnion("type", [ backupRunSchema, backupCancelSchema, + volumeCommandRequestSchema, heartbeatPingSchema, ]); const agentMessageSchema = z.discriminatedUnion("type", [ @@ -122,6 +228,7 @@ const agentMessageSchema = z.discriminatedUnion("type", [ backupCompletedSchema, backupFailedSchema, backupCancelledSchema, + volumeCommandResponseSchema, heartbeatPongSchema, ]); @@ -132,6 +239,10 @@ export type BackupProgressPayload = z.infer["payloa export type BackupCompletedPayload = z.infer["payload"]; export type BackupFailedPayload = z.infer["payload"]; export type BackupCancelledPayload = z.infer["payload"]; +export type VolumeCommandPayload = z.infer["payload"]; +export type VolumeCommand = z.infer; +export type VolumeCommandResult = z.infer; +export type VolumeCommandResponsePayload = z.infer["payload"]; export type ControllerMessage = z.infer; export type AgentMessage = z.infer;