danvics/zerobyte
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

refactor: dedupe volme backend implementations

Browse source
This commit is contained in:
Nicolas Meienberger 2026-05-06 21:33:34 +02:00 committed by Nico
parent 6b83864158
commit 89d347cbc7
10 changed files with 9 additions and 970 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
app/server/jobs/cleanup-dangling.ts
View file

@ -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";

25
app/server/modules/backends/backend.ts
View file

@ -1,23 +1,14 @@
import type { BackendStatus } from "~/schemas/volumes";
import { makeDirectoryBackend } from "../../../../apps/agent/src/volume-host/backends/directory";
import { makeNfsBackend } from "../../../../apps/agent/src/volume-host/backends/nfs";
import { makeRcloneBackend } from "../../../../apps/agent/src/volume-host/backends/rclone";
import { makeSftpBackend } from "../../../../apps/agent/src/volume-host/backends/sftp";
import { makeSmbBackend } from "../../../../apps/agent/src/volume-host/backends/smb";
import { makeWebdavBackend } from "../../../../apps/agent/src/volume-host/backends/webdav";
import type { VolumeBackend } from "../../../../apps/agent/src/volume-host/types";
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<OperationResult>;
unmount: () => Promise<OperationResult>;
checkHealth: () => Promise<OperationResult>;
};
export type { VolumeBackend };
export const createVolumeBackend = (volume: Volume, mountPath = getVolumePath(volume)): VolumeBackend => {
switch (volume.config.backend) {

53
app/server/modules/backends/directory/directory-backend.ts
View file

@ -1,53 +0,0 @@
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";
const mount = async (config: BackendConfig, _volumePath: string) => {
if (config.backend !== "directory") {
return { status: BACKEND_STATUS.error, error: "Invalid backend type" };
}
logger.info("Mounting directory volume from:", config.path);
try {
await fs.access(config.path);
const stats = await fs.stat(config.path);
if (!stats.isDirectory()) {
return { status: BACKEND_STATUS.error, error: "Path is not a directory" };
}
return { status: BACKEND_STATUS.mounted };
} catch (error) {
logger.error("Failed to mount directory volume:", error);
return { status: BACKEND_STATUS.error, error: toMessage(error) };
}
};
const unmount = async () => {
logger.info("Cannot unmount directory volume.");
return { status: BACKEND_STATUS.unmounted };
};
const checkHealth = async (config: BackendConfig) => {
if (config.backend !== "directory") {
return { status: BACKEND_STATUS.error, error: "Invalid backend type" };
}
try {
await fs.access(config.path);
return { status: BACKEND_STATUS.mounted };
} catch (error) {
logger.error("Directory health check failed:", error);
return { status: BACKEND_STATUS.error, error: toMessage(error) };
}
};
export const makeDirectoryBackend = (config: BackendConfig, volumePath: string): VolumeBackend => ({
mount: () => mount(config, volumePath),
unmount,
checkHealth: () => checkHealth(config),
});

135
app/server/modules/backends/nfs/nfs-backend.ts
View file

@ -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),
});

131
app/server/modules/backends/rclone/rclone-backend.ts
View file

@ -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),
});

212
app/server/modules/backends/sftp/sftp-backend.ts
View file

@ -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<void>((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),
});

141
app/server/modules/backends/smb/smb-backend.ts
View file

@ -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),
});

59
app/server/modules/backends/utils/__tests__/backend-utils.test.ts
View file

@ -1,59 +0,0 @@
import { afterEach, describe, expect, test, vi } from "vitest";
vi.mock("node:fs/promises", async (importOriginal) => {
const actual = await importOriginal<typeof import("node:fs/promises")>();
return {
...actual,
access: vi.fn(actual.access),
};
});
vi.mock("../../../../utils/mountinfo", async (importOriginal) => {
const actual = await importOriginal<typeof import("../../../../utils/mountinfo")>();
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();
});
});

62
app/server/modules/backends/utils/backend-utils.ts
View file

@ -1,62 +0,0 @@
import * as fs from "node:fs/promises";
import { logger } from "@zerobyte/core/node";
import { safeExec } from "@zerobyte/core/node";
import { getMountForPath } from "../../../utils/mountinfo";
export const executeMount = async (args: string[]): Promise<void> => {
const shouldBeVerbose = process.env.LOG_LEVEL === "debug" || process.env.NODE_ENV !== "production";
const hasVerboseFlag = args.some((arg) => arg === "-v" || arg.startsWith("-vv"));
const effectiveArgs = shouldBeVerbose && !hasVerboseFlag ? ["-v", ...args] : args;
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();
if (result.exitCode === 0) {
if (stdout) logger.debug(stdout);
if (stderr) logger.debug(stderr);
return;
}
if (stdout) logger.warn(stdout);
if (stderr) logger.warn(stderr);
throw new Error(`Mount command failed with exit code ${result.exitCode}: ${stderr || stdout || "unknown error"}`);
};
export const executeUnmount = async (path: string): Promise<void> => {
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());
}
if (result.exitCode !== 0) {
throw new Error(`Mount command failed with exit code ${result.exitCode}: ${stderr?.trim()}`);
}
};
export const assertMounted = async (path: string, isExpectedFilesystem: (fstype: string) => boolean) => {
try {
await fs.access(path);
} catch {
throw new Error("Volume is not mounted");
}
const mount = await getMountForPath(path);
if (!mount || mount.mountPoint !== path) {
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}).`);
}
};

159
app/server/modules/backends/webdav/webdav-backend.ts
View file

@ -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),
});