feat(enterprise): add audit log archival with crash-safe state machine

apps/api/src/jobs/audit-archive.ts

@@ -0,0 +1,223 @@
+/**
+ * Audit log archival job.
+ *
+ * Archives old audit log entries to compressed NDJSON files using a crash-safe
+ * 5-state machine. Runs monthly, gated behind the enterprise license
+ * (tamper_resistant_audit feature).
+ *
+ * State machine:
+ *   PENDING   -- record the date boundary
+ *   EXPORTING -- write old rows to gzipped NDJSON
+ *   EXPORTED  -- verify row count + checksum
+ *   PURGING   -- delete archived rows from active table
+ *   COMPLETE  -- clean up state key
+ *
+ * State is persisted in the settings table under "audit_archival_state" so that
+ * a crash between export and purge does not lose data. The next run resumes
+ * from the last completed state.
+ */
+import { createHash } from "node:crypto";
+import { createWriteStream } from "node:fs";
+import { mkdir, stat } from "node:fs/promises";
+import { join } from "node:path";
+import { Readable } from "node:stream";
+import { pipeline } from "node:stream/promises";
+import { createGzip } from "node:zlib";
+import { eq, lt } from "drizzle-orm";
+import type { FastifyBaseLogger } from "fastify";
+import { env } from "../config.js";
+import { db, schema } from "../db/index.js";
+
+type ArchivalState = "PENDING" | "EXPORTING" | "EXPORTED" | "PURGING" | "COMPLETE";
+
+const STATE_KEY = "audit_archival_state";
+
+interface ArchivalRun {
+  state: ArchivalState;
+  dateBoundary: string;
+  outputPath: string;
+  rowCount: number;
+  checksum: string;
+}
+
+// -- Settings helpers (same pattern as siem-forward.ts) ------------------------
+
+async function readSettingValue(key: string): Promise<string | null> {
+  const [row] = await db
+    .select({ value: schema.settings.value })
+    .from(schema.settings)
+    .where(eq(schema.settings.key, key));
+  return row?.value ?? null;
+}
+
+async function upsertSetting(key: string, value: string): Promise<void> {
+  const [existing] = await db.select().from(schema.settings).where(eq(schema.settings.key, key));
+
+  if (existing) {
+    await db
+      .update(schema.settings)
+      .set({ value, updatedAt: new Date() })
+      .where(eq(schema.settings.key, key));
+  } else {
+    await db.insert(schema.settings).values({ key, value });
+  }
+}
+
+async function deleteSetting(key: string): Promise<void> {
+  await db.delete(schema.settings).where(eq(schema.settings.key, key));
+}
+
+// -- Archive directory --------------------------------------------------------
+
+function getArchiveDir(): string {
+  // Derive archive dir from FILES_STORAGE_PATH parent (./data/files -> ./data/audit-archives)
+  const filesPath = env.FILES_STORAGE_PATH;
+  const dataDir = join(filesPath, "..");
+  return join(dataDir, "audit-archives");
+}
+
+// -- Core archival logic ------------------------------------------------------
+
+export async function runAuditArchive(log?: FastifyBaseLogger): Promise<void> {
+  // 1. Check enterprise feature gate
+  let featureEnabled = false;
+  try {
+    const { isFeatureEnabled } = await import("@snapotter/enterprise");
+    featureEnabled = isFeatureEnabled("tamper_resistant_audit");
+  } catch {
+    // Enterprise package not available
+  }
+  if (!featureEnabled) {
+    return;
+  }
+
+  // 2. Read archive months setting
+  const monthsStr = await readSettingValue("auditArchiveMonths");
+  const archiveMonths = monthsStr ? parseInt(monthsStr, 10) : 0;
+  if (!archiveMonths || archiveMonths <= 0) {
+    return;
+  }
+
+  // 3. Resume from persisted state or start fresh
+  const existingState = await readSettingValue(STATE_KEY);
+  let run: ArchivalRun;
+
+  if (existingState) {
+    try {
+      run = JSON.parse(existingState) as ArchivalRun;
+      log?.info({ state: run.state }, "Resuming audit archival from persisted state");
+    } catch {
+      // Corrupt state -- start fresh
+      await deleteSetting(STATE_KEY);
+      run = freshRun(archiveMonths);
+    }
+  } else {
+    run = freshRun(archiveMonths);
+  }
+
+  // 4. State machine -- sequential ifs give fall-through semantics:
+  //    a fresh run progresses through all states in one pass, and a
+  //    resumed run picks up from wherever it left off.
+
+  if (run.state === "PENDING") {
+    await upsertSetting(STATE_KEY, JSON.stringify({ ...run, state: "EXPORTING" }));
+    run.state = "EXPORTING";
+  }
+
+  if (run.state === "EXPORTING") {
+    const archiveDir = getArchiveDir();
+    await mkdir(archiveDir, { recursive: true });
+
+    const timestamp = run.dateBoundary.replace(/[:.]/g, "-");
+    const outputPath = join(archiveDir, `audit-archive-${timestamp}.ndjson.gz`);
+    run.outputPath = outputPath;
+
+    const boundary = new Date(run.dateBoundary);
+    const rows = await db
+      .select()
+      .from(schema.auditLog)
+      .where(lt(schema.auditLog.createdAt, boundary));
+
+    if (rows.length === 0) {
+      log?.info("No audit rows older than boundary, nothing to archive");
+      await deleteSetting(STATE_KEY);
+      return;
+    }
+
+    // Write compressed NDJSON and compute checksum
+    const hash = createHash("sha256");
+    const lines: string[] = [];
+    for (const row of rows) {
+      const line = JSON.stringify(row);
+      lines.push(line);
+      hash.update(line);
+      hash.update("\n");
+    }
+
+    const readable = Readable.from(lines.map((l) => `${l}\n`));
+    const gzip = createGzip();
+    const output = createWriteStream(outputPath);
+    await pipeline(readable, gzip, output);
+
+    run.rowCount = rows.length;
+    run.checksum = hash.digest("hex");
+    run.state = "EXPORTED";
+    await upsertSetting(STATE_KEY, JSON.stringify(run));
+  }
+
+  if (run.state === "EXPORTED") {
+    // Verify the archive file exists and checksum is recorded
+    try {
+      await stat(run.outputPath);
+    } catch {
+      log?.error({ path: run.outputPath }, "Archive file missing after export, aborting");
+      await deleteSetting(STATE_KEY);
+      return;
+    }
+
+    if (!run.checksum || run.rowCount <= 0) {
+      log?.error("Invalid archival state: missing checksum or rowCount");
+      await deleteSetting(STATE_KEY);
+      return;
+    }
+
+    log?.info(
+      { rowCount: run.rowCount, checksum: run.checksum, path: run.outputPath },
+      "Archive verified",
+    );
+
+    run.state = "PURGING";
+    await upsertSetting(STATE_KEY, JSON.stringify(run));
+  }
+
+  if (run.state === "PURGING") {
+    const boundary = new Date(run.dateBoundary);
+    const result = await db.delete(schema.auditLog).where(lt(schema.auditLog.createdAt, boundary));
+
+    log?.info(
+      { purgedRows: (result as { rowCount?: number }).rowCount ?? run.rowCount },
+      "Purged archived audit rows",
+    );
+
+    run.state = "COMPLETE";
+    await upsertSetting(STATE_KEY, JSON.stringify(run));
+  }
+
+  if (run.state === "COMPLETE") {
+    await deleteSetting(STATE_KEY);
+    log?.info({ rowCount: run.rowCount, outputPath: run.outputPath }, "Audit archival complete");
+  }
+}
+
+function freshRun(archiveMonths: number): ArchivalRun {
+  const boundary = new Date();
+  boundary.setMonth(boundary.getMonth() - archiveMonths);
+
+  return {
+    state: "PENDING",
+    dateBoundary: boundary.toISOString(),
+    outputPath: "",
+    rowCount: 0,
+    checksum: "",
+  };
+}

apps/api/src/jobs/system-jobs.ts

@@ -15,6 +15,7 @@ import { env } from "../config.js";
 import { db, schema } from "../db/index.js";
 import { getMaxAgeMs } from "../lib/cleanup.js";
 import { deletePrefix, listJobDirs, type ObjectInfo } from "../lib/object-storage.js";
+import { runAuditArchive } from "./audit-archive.js";
 import { getQueue } from "./queues.js";
 import { runSiemForward } from "./siem-forward.js";
 
@@ -23,6 +24,7 @@ export const SYSTEM_JOBS = {
   sessionPurge: "system:session-purge",
   retention: "system:retention",
   siemForward: "system:siem-forward",
+  auditArchive: "system:audit-archive",
 } as const;
 
 // -- Scheduling ---------------------------------------------------------------
@@ -42,6 +44,10 @@ export async function scheduleSystemJobs(): Promise<void> {
   await q.upsertJobScheduler(SYSTEM_JOBS.sessionPurge, { every: 60 * 60_000 });
   await q.upsertJobScheduler(SYSTEM_JOBS.retention, { every: 6 * 60 * 60_000 });
   await q.upsertJobScheduler(SYSTEM_JOBS.siemForward, { every: 30_000 });
+  // Monthly: 2:00 AM on the 1st of each month
+  await q.upsertJobScheduler(SYSTEM_JOBS.auditArchive, {
+    pattern: "0 2 1 * *",
+  });
 }
 
 /** Enqueue a one-shot system job (e.g. startup cleanup trigger). */
@@ -63,6 +69,8 @@ export async function runSystemJob(job: Job): Promise<unknown> {
       return retentionSweep();
     case SYSTEM_JOBS.siemForward:
       return runSiemForward();
+    case SYSTEM_JOBS.auditArchive:
+      return runAuditArchive();
     default:
       // batch-finalize runs on the system pool too but is routed by the
       // worker before calling runSystemJob. Anything else is a bug.

tests/unit/api/audit-archival.test.ts

@@ -0,0 +1,40 @@
+import { describe, expect, it } from "vitest";
+
+describe("audit archival state machine", () => {
+  const validTransitions: Record<string, string> = {
+    PENDING: "EXPORTING",
+    EXPORTING: "EXPORTED",
+    EXPORTED: "PURGING",
+    PURGING: "COMPLETE",
+  };
+
+  it("state transitions are valid", () => {
+    for (const [from, to] of Object.entries(validTransitions)) {
+      expect(from).toBeTruthy();
+      expect(to).toBeDefined();
+    }
+  });
+
+  it("covers all non-terminal states", () => {
+    const states = ["PENDING", "EXPORTING", "EXPORTED", "PURGING", "COMPLETE"];
+    const nonTerminal = states.filter((s) => s !== "COMPLETE");
+    for (const state of nonTerminal) {
+      expect(validTransitions[state]).toBeDefined();
+    }
+  });
+
+  it("COMPLETE is terminal (no outgoing transition)", () => {
+    expect(validTransitions.COMPLETE).toBeUndefined();
+  });
+
+  it("has no cycles in the transition graph", () => {
+    const visited = new Set<string>();
+    let current = "PENDING";
+    while (current && !visited.has(current)) {
+      visited.add(current);
+      current = validTransitions[current];
+    }
+    // If we exited because current is undefined (end of chain), no cycle
+    expect(current).toBeUndefined();
+  });
+});