added progress bar and scheduler

backend/src/executor.js

@@ -1,5 +1,5 @@
 import { spawn } from 'node:child_process';
-import { PassThrough } from 'node:stream';
+import { PassThrough, Transform } from 'node:stream';
 import { resticProcessEnv } from './config.js';
 
 export class CommandError extends Error {
@@ -74,7 +74,22 @@ export function runRestic(args, options = {}) {
   });
 }
 
-export function streamResticDumpToDd(snapshotId, filename, outputPath, log) {
+export function createProgressStream(totalBytes, onProgress) {
+  let currentBytes = 0;
+  return new Transform({
+    transform(chunk, _encoding, callback) {
+      currentBytes += chunk.length;
+      onProgress?.({
+        currentBytes,
+        totalBytes,
+        percent: totalBytes ? (currentBytes / totalBytes) * 100 : 0,
+      });
+      callback(null, chunk);
+    },
+  });
+}
+
+export function streamResticDumpToDd(snapshotId, filename, outputPath, log, progress = null) {
   return new Promise((resolve, reject) => {
     const restic = spawn('restic', ['dump', snapshotId, filename], {
       env: resticProcessEnv(),
@@ -91,7 +106,11 @@ export function streamResticDumpToDd(snapshotId, filename, outputPath, log) {
     let resticClosed = false;
     let ddClosed = false;
     const pipe = new PassThrough();
-    restic.stdout.pipe(pipe).pipe(dd.stdin);
+    let restoreStream = restic.stdout.pipe(pipe);
+    if (progress) {
+      restoreStream = restoreStream.pipe(createProgressStream(progress.totalBytes, progress.onProgress));
+    }
+    restoreStream.pipe(dd.stdin);
 
     restic.stderr.on('data', (chunk) => log?.(chunk.toString().trimEnd()));
     dd.stderr.on('data', (chunk) => log?.(chunk.toString().trimEnd()));

backend/src/index.js

@@ -5,9 +5,11 @@ import { backupRouter } from './routes/backup.js';
 import { healthRouter } from './routes/health.js';
 import { jobsRouter } from './routes/jobs.js';
 import { restoreRouter } from './routes/restore.js';
+import { schedulesRouter } from './routes/schedules.js';
 import { settingsRouter } from './routes/settings.js';
 import { snapshotsRouter } from './routes/snapshots.js';
 import { vmsRouter } from './routes/vms.js';
+import { startScheduler } from './scheduler.js';
 
 const app = express();
 
@@ -33,6 +35,7 @@ app.use('/api/snapshots', snapshotsRouter);
 app.use('/api/jobs', jobsRouter);
 app.use('/api/backup', backupRouter);
 app.use('/api/restore', restoreRouter);
+app.use('/api/schedules', schedulesRouter);
 app.use('/api/settings', settingsRouter);
 
 app.use((error, _req, res, _next) => {
@@ -43,3 +46,7 @@ app.use((error, _req, res, _next) => {
 app.listen(config.port, () => {
   console.log(`Incus backup API listening on http://localhost:${config.port}`);
 });
+
+startScheduler().catch((error) => {
+  console.error(`Failed to start scheduler: ${error.message}`);
+});

backend/src/jobs.js

@@ -22,6 +22,12 @@ export function createJob(type, vmName) {
     startedAt: now,
     finishedAt: null,
     currentStep: 'Queued',
+    progress: {
+      percent: 0,
+      detail: 'Queued',
+      currentBytes: 0,
+      totalBytes: null,
+    },
     logs: [],
     error: null,
   };
@@ -56,9 +62,22 @@ export function setJobRunning(job, step) {
 
 export function setJobStep(job, step) {
   job.currentStep = step;
+  job.progress = {
+    ...job.progress,
+    detail: step,
+  };
   appendJobLog(job, `==> ${step}`);
 }
 
+export function setJobProgress(job, progress = {}) {
+  const nextPercent = progress.percent ?? job.progress?.percent ?? 0;
+  job.progress = {
+    ...job.progress,
+    ...progress,
+    percent: Math.max(0, Math.min(100, Number(nextPercent) || 0)),
+  };
+}
+
 export function appendJobLog(job, line) {
   if (!line) return;
   job.logs.push(...String(line).split('\n').filter(Boolean));
@@ -71,6 +90,11 @@ export function finishJob(job, status, error = null) {
   job.status = status;
   job.finishedAt = new Date().toISOString();
   job.currentStep = status === 'success' ? 'Completed' : 'Failed';
+  job.progress = {
+    ...job.progress,
+    percent: status === 'success' ? 100 : job.progress?.percent || 0,
+    detail: status === 'success' ? 'Completed' : 'Failed',
+  };
   job.error = error ? String(error.message || error) : null;
   locks.delete(job.vmName);
 }

backend/src/routes/backup.js

@@ -2,8 +2,8 @@ import { Router } from 'express';
 import { finished } from 'node:stream/promises';
 import { setTimeout as delay } from 'node:timers/promises';
 import { config } from '../config.js';
-import { spawnCommand, runRestic } from '../executor.js';
-import { appendJobLog, createJob, finishJob, setJobRunning, setJobStep } from '../jobs.js';
+import { createProgressStream, spawnCommand, runRestic } from '../executor.js';
+import { appendJobLog, createJob, finishJob, setJobProgress, setJobRunning, setJobStep } from '../jobs.js';
 import { validateVmExists } from '../validators.js';
 
 export const backupRouter = Router();
@@ -11,37 +11,55 @@ export const backupRouter = Router();
 backupRouter.post('/:vmName', async (req, res, next) => {
   try {
     const vmName = req.params.vmName;
-    await validateVmExists(vmName);
-    const job = createJob('backup', vmName);
-    runBackupJob(job).catch(() => {});
+    const job = await startBackupForVm(vmName);
     res.status(202).json({ jobId: job.id, message: 'Backup job started.' });
   } catch (error) {
     next(error);
   }
 });
 
-async function runBackupJob(job) {
+export async function startBackupForVm(vmName) {
+  await validateVmExists(vmName);
+  const job = createJob('backup', vmName);
+  runBackupJob(job).catch(() => {});
+  return job;
+}
+
+export async function runBackupJob(job) {
   const timestamp = new Date().toISOString().replace(/[-:.TZ]/g, '').slice(0, 14);
   const snapshotName = `s3-backup-${timestamp}`;
   const zvol = `${config.zfsPoolName}/virtual-machines/${job.vmName}.block`;
   const snapshotDevice = `/dev/zvol/${zvol}@snapshot-${snapshotName}`;
 
   try {
+    const totalBytes = await zfsVolumeSize(zvol);
+
     setJobRunning(job, 'Creating Incus snapshot');
+    setJobProgress(job, { percent: 2 });
     await spawnCommand('incus', ['snapshot', 'create', job.vmName, snapshotName], { log: (line) => appendJobLog(job, line) });
 
     setJobStep(job, 'Making ZFS snapshot device visible');
+    setJobProgress(job, { percent: 5 });
     await spawnCommand('zfs', ['set', 'snapdev=visible', zvol], { log: (line) => appendJobLog(job, line) });
 
     setJobStep(job, 'Waiting for snapshot device');
+    setJobProgress(job, { percent: 8 });
     await new Promise((resolve) => setTimeout(resolve, 2000));
 
     setJobStep(job, 'Streaming block device to Restic');
     const snapshotStream = await deviceReadStream(snapshotDevice);
+    const progressStream = createProgressStream(totalBytes, ({ currentBytes, totalBytes: bytesTotal, percent }) => {
+      setJobProgress(job, {
+        currentBytes,
+        totalBytes: bytesTotal,
+        percent: 10 + percent * 0.78,
+        detail: `Streaming ${formatBytes(currentBytes)} of ${formatBytes(bytesTotal)}`,
+      });
+    });
     try {
       await spawnCommand('restic', ['backup', '--stdin', '--stdin-filename', `${job.vmName}.raw`, '--tag', job.vmName], {
         env: { ...process.env, ...config.resticEnv },
-        input: snapshotStream,
+        input: snapshotStream.pipe(progressStream),
         log: (line) => appendJobLog(job, line),
       });
     } finally {
@@ -52,13 +70,16 @@ async function runBackupJob(job) {
     }
 
     setJobStep(job, 'Hiding ZFS snapshot device');
+    setJobProgress(job, { percent: 90 });
     await spawnCommand('zfs', ['set', 'snapdev=hidden', zvol], { log: (line) => appendJobLog(job, line) });
     await settleUdev(job);
 
     setJobStep(job, 'Deleting temporary Incus snapshot');
+    setJobProgress(job, { percent: 93 });
     await deleteIncusSnapshotWithRetry(job, snapshotName);
 
     setJobStep(job, 'Applying Restic retention policy');
+    setJobProgress(job, { percent: 96 });
     await runRestic(['forget', '--tag', job.vmName, '--keep-daily', '7', '--prune'], {
       log: (line) => appendJobLog(job, line),
     });
@@ -76,6 +97,24 @@ async function deviceReadStream(path) {
   return createReadStream(path);
 }
 
+async function zfsVolumeSize(zvol) {
+  const result = await spawnCommand('zfs', ['get', '-Hp', '-o', 'value', 'volsize', zvol]);
+  const size = Number(result.stdout.trim());
+  return Number.isFinite(size) && size > 0 ? size : null;
+}
+
+function formatBytes(value) {
+  if (!value) return '-';
+  const units = ['B', 'KiB', 'MiB', 'GiB', 'TiB'];
+  let size = Number(value);
+  let unitIndex = 0;
+  while (size >= 1024 && unitIndex < units.length - 1) {
+    size /= 1024;
+    unitIndex += 1;
+  }
+  return `${size.toFixed(unitIndex === 0 ? 0 : 1)} ${units[unitIndex]}`;
+}
+
 async function cleanupBackup(job, zvol, snapshotName) {
   setJobStep(job, 'Running cleanup');
   await spawnCommand('zfs', ['set', 'snapdev=hidden', zvol], {

backend/src/routes/restore.js

@@ -1,7 +1,7 @@
 import { Router } from 'express';
 import { config } from '../config.js';
 import { spawnCommand, streamResticDumpToDd } from '../executor.js';
-import { appendJobLog, createJob, finishJob, setJobRunning, setJobStep } from '../jobs.js';
+import { appendJobLog, createJob, finishJob, setJobProgress, setJobRunning, setJobStep } from '../jobs.js';
 import { validateSnapshotForVm, validateVmExists } from '../validators.js';
 
 export const restoreRouter = Router();
@@ -34,29 +34,45 @@ async function runRestoreJob(job, snapshotId) {
 
   try {
     setJobRunning(job, 'Stopping VM');
+    setJobProgress(job, { percent: 3 });
     await spawnCommand('incus', ['stop', job.vmName, '--force'], {
       ignoreExitCode: true,
       log: (line) => appendJobLog(job, line),
     });
 
     setJobStep(job, 'Setting ZFS volume to device mode');
+    setJobProgress(job, { percent: 8 });
     await spawnCommand('zfs', ['set', 'volmode=dev', zvol], { log: (line) => appendJobLog(job, line) });
     volmodeDev = true;
 
     setJobStep(job, 'Settling device nodes');
+    setJobProgress(job, { percent: 12 });
     await spawnCommand('udevadm', ['trigger'], { log: (line) => appendJobLog(job, line) });
     await spawnCommand('udevadm', ['settle'], { log: (line) => appendJobLog(job, line) });
     await new Promise((resolve) => setTimeout(resolve, 2000));
 
     setJobStep(job, 'Writing Restic snapshot to block device');
-    await streamResticDumpToDd(snapshotId, `${job.vmName}.raw`, device, (line) => appendJobLog(job, line));
+    const totalBytes = await zfsVolumeSize(zvol);
+    await streamResticDumpToDd(snapshotId, `${job.vmName}.raw`, device, (line) => appendJobLog(job, line), {
+      totalBytes,
+      onProgress: ({ currentBytes, totalBytes: bytesTotal, percent }) => {
+        setJobProgress(job, {
+          currentBytes,
+          totalBytes: bytesTotal,
+          percent: 15 + percent * 0.78,
+          detail: `Writing ${formatBytes(currentBytes)} of ${formatBytes(bytesTotal)}`,
+        });
+      },
+    });
     diskWriteOk = true;
 
     setJobStep(job, 'Restoring ZFS volume mode');
+    setJobProgress(job, { percent: 94 });
     await spawnCommand('zfs', ['set', 'volmode=none', zvol], { log: (line) => appendJobLog(job, line) });
     volmodeDev = false;
 
     setJobStep(job, 'Starting VM');
+    setJobProgress(job, { percent: 97 });
     await spawnCommand('incus', ['start', job.vmName], { log: (line) => appendJobLog(job, line) });
 
     finishJob(job, 'success');
@@ -75,3 +91,21 @@ async function runRestoreJob(job, snapshotId) {
     finishJob(job, 'failed', error);
   }
 }
+
+async function zfsVolumeSize(zvol) {
+  const result = await spawnCommand('zfs', ['get', '-Hp', '-o', 'value', 'volsize', zvol]);
+  const size = Number(result.stdout.trim());
+  return Number.isFinite(size) && size > 0 ? size : null;
+}
+
+function formatBytes(value) {
+  if (!value) return '-';
+  const units = ['B', 'KiB', 'MiB', 'GiB', 'TiB'];
+  let size = Number(value);
+  let unitIndex = 0;
+  while (size >= 1024 && unitIndex < units.length - 1) {
+    size /= 1024;
+    unitIndex += 1;
+  }
+  return `${size.toFixed(unitIndex === 0 ? 0 : 1)} ${units[unitIndex]}`;
+}

backend/src/routes/schedules.js

@@ -0,0 +1,17 @@
+import { Router } from 'express';
+import { listSchedules, replaceSchedules } from '../scheduler.js';
+
+export const schedulesRouter = Router();
+
+schedulesRouter.get('/', (_req, res) => {
+  res.json(listSchedules());
+});
+
+schedulesRouter.put('/', async (req, res, next) => {
+  try {
+    const schedules = await replaceSchedules(req.body?.schedules || []);
+    res.json(schedules);
+  } catch (error) {
+    next(error);
+  }
+});

backend/src/routes/vms.js

@@ -28,5 +28,6 @@ function summarizeJob(job) {
     type: job.type,
     status: job.status,
     currentStep: job.currentStep,
+    progress: job.progress,
   };
 }

backend/src/scheduler.js

@@ -0,0 +1,103 @@
+import crypto from 'node:crypto';
+import { readFile, rename, writeFile } from 'node:fs/promises';
+import path from 'node:path';
+import { startBackupForVm } from './routes/backup.js';
+
+const schedulesPath = path.resolve(process.cwd(), 'schedules.json');
+const checkIntervalMs = 30 * 1000;
+let schedules = [];
+let timer = null;
+
+export async function startScheduler() {
+  await loadSchedules();
+  if (timer) return;
+  timer = setInterval(runDueSchedules, checkIntervalMs);
+  timer.unref?.();
+  runDueSchedules();
+}
+
+export function listSchedules() {
+  return schedules
+    .map((schedule) => ({ ...schedule }))
+    .sort((a, b) => a.vmName.localeCompare(b.vmName));
+}
+
+export async function replaceSchedules(nextSchedules) {
+  schedules = normalizeSchedules(nextSchedules);
+  await saveSchedules();
+  return listSchedules();
+}
+
+async function runDueSchedules() {
+  const now = new Date();
+  let changed = false;
+
+  for (const schedule of schedules) {
+    if (!schedule.enabled) continue;
+    if (!schedule.nextRunAt) {
+      schedule.nextRunAt = nextRunFrom(now, schedule.intervalHours);
+      changed = true;
+      continue;
+    }
+    if (new Date(schedule.nextRunAt) > now) continue;
+
+    try {
+      await startBackupForVm(schedule.vmName);
+      schedule.lastRunAt = now.toISOString();
+      schedule.lastError = '';
+      schedule.nextRunAt = nextRunFrom(now, schedule.intervalHours);
+    } catch (error) {
+      schedule.lastError = String(error.message || error);
+      schedule.nextRunAt = nextRunFrom(new Date(now.getTime() + 5 * 60 * 1000), schedule.intervalHours);
+    }
+    changed = true;
+  }
+
+  if (changed) {
+    await saveSchedules();
+  }
+}
+
+async function loadSchedules() {
+  try {
+    const content = await readFile(schedulesPath, 'utf8');
+    schedules = normalizeSchedules(JSON.parse(content));
+  } catch (error) {
+    if (error.code === 'ENOENT') {
+      schedules = [];
+      return;
+    }
+    throw error;
+  }
+}
+
+async function saveSchedules() {
+  const tempPath = `${schedulesPath}.tmp`;
+  await writeFile(tempPath, `${JSON.stringify(schedules, null, 2)}\n`, { mode: 0o600 });
+  await rename(tempPath, schedulesPath);
+}
+
+function normalizeSchedules(values) {
+  const seen = new Set();
+  return (Array.isArray(values) ? values : [])
+    .map((value) => {
+      const vmName = String(value.vmName || '').trim();
+      const intervalHours = Math.max(1, Number(value.intervalHours) || 24);
+      if (!vmName || seen.has(vmName)) return null;
+      seen.add(vmName);
+      return {
+        id: value.id || `schedule_${crypto.randomBytes(8).toString('hex')}`,
+        vmName,
+        enabled: Boolean(value.enabled),
+        intervalHours,
+        nextRunAt: value.nextRunAt || nextRunFrom(new Date(), intervalHours),
+        lastRunAt: value.lastRunAt || null,
+        lastError: value.lastError || '',
+      };
+    })
+    .filter(Boolean);
+}
+
+function nextRunFrom(date, intervalHours) {
+  return new Date(date.getTime() + intervalHours * 60 * 60 * 1000).toISOString();
+}