craftvia/src/server/backup/worker.ts

// ── BullMQ-Worker der Backup-/DSGVO-Ops-Queue ────────────────────────────────
//
// Analog zum Mail-Worker (SEC1). Nimmt Jobs aus `backup-ops` und führt sie über
// `processBackupJob` aus. Concurrency 1: Restore ist destruktiv und langlaufend
// (Mandant-Sperre, Pre-Restore-Snapshot) — Jobs laufen bewusst seriell, nicht
// parallel. Kein automatisches Retry (attempts=1 in queue.ts): eine Wiederholung
// eines destruktiven Restore ist eine bewusste Betreiber-Entscheidung.

import { Worker, type Job } from "bullmq";
import { BACKUP_QUEUE, type BackupOpsJob } from "./job";
import { closeBackupQueues, getBackupConnection, getBackupDeadLetterQueue } from "./queue";
import { processBackupJob } from "./ops";

export function startBackupWorker(): Worker<BackupOpsJob> {
  const connection = getBackupConnection();
  if (!connection) {
    throw new Error("REDIS_URL ist nicht gesetzt — ohne Redis gibt es keinen Backup-Worker-Betrieb.");
  }

  const worker = new Worker<BackupOpsJob>(
    BACKUP_QUEUE,
    async (job: Job<BackupOpsJob>) => {
      await processBackupJob(job.data);
    },
    { connection, concurrency: 1 },
  );

  worker.on("failed", async (job, err) => {
    if (!job) return;
    console.error(`[backup] Job ${job.id} (${job.data.kind}) fehlgeschlagen: ${err.message}`);
    // attempts=1 → jeder Fehlschlag ist endgültig: Dead-Letter.
    await getBackupDeadLetterQueue()
      ?.add("dead", { job: job.data, error: err.message })
      .catch(() => {});
    console.error(`[backup] ALARM — Job ${job.id} in die Dead-Letter-Queue verschoben.`);
  });

  worker.on("completed", (job) => {
    console.info(`[backup] Job ${job.id} (${job.data.kind}) fertig — Mandant ${job.data.tenantId}.`);
  });

  return worker;
}

/** Sauberes Herunterfahren von Worker und Redis. */
export async function shutdownBackupWorker(worker: Worker | null): Promise<void> {
  await worker?.close();
  await closeBackupQueues();
}