incus-backup-ui/agent/src/jobs.js

import crypto from 'node:crypto';
import { mkdirSync } from 'node:fs';
import { DatabaseSync } from 'node:sqlite';
import path from 'node:path';
import { config } from './config.js';

const jobs = new Map();
const locks = new Map();
const maxJobs = 100;

mkdirSync(path.dirname(config.agentDatabasePath), { recursive: true, mode: 0o700 });
const db = new DatabaseSync(config.agentDatabasePath);

migrate();
loadPersistedJobs();

export function createJob(type, vmName) {
  if (locks.has(vmName)) {
    const activeJob = jobs.get(locks.get(vmName));
    const error = new Error(`A ${activeJob?.type || 'job'} job is already active for ${vmName}.`);
    error.status = 409;
    throw error;
  }

  const id = `job_${crypto.randomBytes(8).toString('hex')}`;
  const now = new Date().toISOString();
  const job = {
    id,
    type,
    vmName,
    status: 'queued',
    startedAt: now,
    finishedAt: null,
    currentStep: 'Queued',
    progress: {
      percent: 0,
      detail: 'Queued',
      currentBytes: 0,
      totalBytes: null,
    },
    logs: [],
    error: null,
  };

  jobs.set(id, job);
  locks.set(vmName, id);
  persistJob(job);
  trimJobs();
  return job;
}

export function getJob(id) {
  return jobs.get(id) || null;
}

export function listJobs() {
  return [...jobs.values()].sort((a, b) => b.startedAt.localeCompare(a.startedAt));
}

export function activeJobForVm(vmName) {
  const id = locks.get(vmName);
  return id ? getJob(id) : null;
}

export function latestJobForVm(vmName) {
  return listJobs().find((job) => job.vmName === vmName) || null;
}

export function setJobRunning(job, step) {
  job.status = 'running';
  setJobStep(job, step);
}

export function setJobStep(job, step) {
  job.currentStep = step;
  job.progress = {
    ...job.progress,
    detail: step,
  };
  appendJobLog(job, `==> ${step}`);
  persistJob(job);
}

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)),
  };
  persistJob(job);
}

export function appendJobLog(job, line) {
  if (!line) return;
  job.logs.push(...String(line).split('\n').filter(Boolean));
  if (job.logs.length > 1000) {
    job.logs = job.logs.slice(-1000);
  }
  persistJob(job);
}

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);
  persistJob(job);
}

function trimJobs() {
  const allJobs = listJobs();
  for (const job of allJobs.slice(maxJobs)) {
    if (job.status !== 'running' && job.status !== 'queued') {
      jobs.delete(job.id);
      db.prepare('DELETE FROM jobs WHERE id = ?').run(job.id);
    }
  }
}

function migrate() {
  db.exec(`
    PRAGMA journal_mode = WAL;
    CREATE TABLE IF NOT EXISTS jobs (
      id TEXT PRIMARY KEY,
      vm_name TEXT NOT NULL,
      type TEXT NOT NULL,
      status TEXT NOT NULL,
      started_at TEXT NOT NULL,
      finished_at TEXT,
      updated_at TEXT NOT NULL,
      payload TEXT NOT NULL
    );
  `);
}

function loadPersistedJobs() {
  const now = new Date().toISOString();
  const rows = db.prepare('SELECT payload FROM jobs ORDER BY started_at DESC LIMIT ?').all(maxJobs);
  for (const row of rows.reverse()) {
    const job = JSON.parse(row.payload);
    if (['queued', 'running'].includes(job.status)) {
      job.status = 'failed';
      job.finishedAt = now;
      job.currentStep = 'Failed';
      job.progress = {
        ...job.progress,
        detail: 'Failed',
      };
      job.error = 'Agent restarted while this job was active.';
      job.logs = [...(job.logs || []), 'Agent restarted while this job was active; job marked failed.'].slice(-1000);
    }
    jobs.set(job.id, job);
    persistJob(job);
  }
  trimJobs();
}

function persistJob(job) {
  db.prepare(`
    INSERT INTO jobs (id, vm_name, type, status, started_at, finished_at, updated_at, payload)
    VALUES (?, ?, ?, ?, ?, ?, ?, ?)
    ON CONFLICT(id) DO UPDATE SET
      vm_name = excluded.vm_name,
      type = excluded.type,
      status = excluded.status,
      started_at = excluded.started_at,
      finished_at = excluded.finished_at,
      updated_at = excluded.updated_at,
      payload = excluded.payload
  `).run(
    job.id,
    job.vmName,
    job.type,
    job.status,
    job.startedAt,
    job.finishedAt,
    new Date().toISOString(),
    JSON.stringify(job),
  );
}