fix: preserve last good advisory DB on failed refresh

src/advisory/osv-sync.ts

@@ -1,10 +1,12 @@
 import fs from "node:fs";
 import os from "node:os";
 import path from "node:path";
+import { randomUUID } from "node:crypto";
 import { unzipSync } from "fflate";
 import type { FetchLike, OsvVuln } from "../types.js";
 import { LocalAdvisoryDatabase } from "./local-db.js";
-import { pluralize } from "../utils/string.js";
+import { ensureSentence, pluralize } from "../utils/string.js";
+import { cleanupAbandonedStagingDirectories, recordStagingOwner, stagingDirectoryPrefix } from "./staging-directory.js";
 
 const DEFAULT_OSV_NPM_DUMP_URL = "https://storage.googleapis.com/osv-vulnerabilities/npm/all.zip";
 export const ADVISORY_DB_STALE_AFTER_MS = 7 * 24 * 60 * 60 * 1000;
@@ -20,8 +22,20 @@ export type SyncOsvAdvisoriesResult = {
   advisoryCount: number;
   dbPath: string;
   sourceUrl: string;
+  warnings?: string[];
 };
 
+export class AdvisorySyncError extends Error {
+  constructor(
+    message: string,
+    readonly databaseState: "unchanged" | "replaced" | "recovery-required",
+    cause: unknown,
+  ) {
+    super(message, { cause });
+    this.name = "AdvisorySyncError";
+  }
+}
+
 export type SyncProgressEvent =
   | {
       phase: "init";
@@ -73,26 +87,140 @@ export function getDefaultAdvisoryDbPath(outputPath?: string): string {
   return path.join(os.homedir(), ".cache", "cve-lite", "advisories.db");
 }
 
-export async function syncOsvAdvisories(
-  options: SyncOsvAdvisoriesOptions = {},
-): Promise<SyncOsvAdvisoriesResult> {
-  const fetchImpl = options.fetchImpl ?? fetch;
-  const onProgress = options.onProgress;
-  const sourceUrl = options.sourceUrl ?? DEFAULT_OSV_NPM_DUMP_URL;
-  const dbPath = getDefaultAdvisoryDbPath(options.outputPath);
+/**
+ * Replaces the advisory database after a successful sync. Node's rename replaces
+ * existing files on Windows too, but an open or locked target can cause EPERM.
+ * If replacement is refused, move the old database aside only after the new one
+ * is complete and restore it if the second rename fails.
+ */
+function replaceAdvisoryDatabase(tempPath: string, dbPath: string): string | undefined {
+  try {
+    fs.renameSync(tempPath, dbPath);
+    return;
+  } catch (error) {
+    const code = (error as NodeJS.ErrnoException).code;
+    if (code !== "EEXIST" && code !== "EPERM") throw error;
+  }
+
+  const backupPath = `${dbPath}.backup-${process.pid}-${randomUUID()}`;
+  let movedExisting = false;
+  try {
+    fs.renameSync(dbPath, backupPath);
+    movedExisting = true;
+    fs.renameSync(tempPath, dbPath);
+  } catch (error) {
+    if (!movedExisting) throw error;
+
+    restoreAdvisoryDatabase(backupPath, dbPath, error);
+  }
+
+  return cleanupBackup(backupPath, "The refreshed database was installed.");
+}
+
+function cleanupBackup(backupPath: string, databaseMessage: string): string | undefined {
+  try {
+    fs.rmSync(backupPath, { force: true });
+  } catch (error) {
+    const message = error instanceof Error ? error.message : String(error);
+    return `Could not remove advisory database backup ${backupPath}: ${ensureSentence(message)} `
+      + `${databaseMessage} Resolve the filesystem error and remove ${backupPath} manually.`;
+  }
+}
+
+function restoreAdvisoryDatabase(backupPath: string, dbPath: string, installError: unknown): never {
+  const originalMessage = installError instanceof Error ? installError.message : String(installError);
+  try {
+    // Hardlink creation is atomic and refuses an existing target. An exists-check
+    // followed by rename can overwrite a database installed by another sync.
+    fs.linkSync(backupPath, dbPath);
+  } catch (restoreError) {
+    const restoreMessage = restoreError instanceof Error ? restoreError.message : String(restoreError);
+    throw new AdvisorySyncError(
+      `Failed to install refreshed advisory database: ${ensureSentence(originalMessage)} `
+        + `The previous database could not be restored and remains at ${backupPath}: ${ensureSentence(restoreMessage)} `
+        + `No existing database at ${dbPath} was overwritten. Close any program using these files and inspect both databases. `
+        + `After resolving the filesystem error, rename ${backupPath} to ${dbPath} only if the target is absent; do not overwrite a newer database.`,
+      "recovery-required",
+      restoreError,
+    );
+  }
+  const warning = cleanupBackup(backupPath, "The previous database was restored.");
+  if (warning) throw new AdvisorySyncError(`${ensureSentence(originalMessage)} ${warning}`, "unchanged", installError);
+  throw installError;
+}
+
+function rejectDirectoryOutput(dbPath: string): void {
+  try {
+    if (!fs.lstatSync(dbPath).isDirectory()) return;
+  } catch (error) {
+    if ((error as NodeJS.ErrnoException).code === "ENOENT") return;
+    throw error;
+  }
+  throw new Error(`Advisory database output ${dbPath} is a directory. Choose a database file path, for example ${path.join(dbPath, "advisories.db")}.`);
+}
+
+type AdvisoryArchiveEntry = readonly [name: string, bytes: Uint8Array];
+
+async function ingestAdvisories(
+  db: LocalAdvisoryDatabase,
+  advisoryEntries: readonly AdvisoryArchiveEntry[],
+  sourceUrl: string,
+  onProgress?: (event: SyncProgressEvent) => void,
+): Promise<number> {
+  const parsedVulns: OsvVuln[] = [];
+  let advisoryCount = 0;
+  const progressInterval = 250;
+
+  for (const [, bytes] of advisoryEntries) {
+    const text = Buffer.from(bytes).toString("utf8");
+    const vuln = JSON.parse(text) as OsvVuln;
+    if (!vuln.id) continue;
+    if (!Array.isArray(vuln.affected) || vuln.affected.length === 0) continue;
+
+    // OSV's /v1/querybatch endpoint excludes withdrawn advisories at query
+    // time; mirror that here so offline scans don't surface false positives
+    // from advisories that have been retracted.
+    if (vuln.withdrawn) continue;
+
+    parsedVulns.push(vuln);
+    advisoryCount += 1;
+
+    if (
+      advisoryCount === 1 ||
+      advisoryCount % progressInterval === 0 ||
+      advisoryCount === advisoryEntries.length
+    ) {
+      onProgress?.({
+        phase: "ingest",
+        processedAdvisories: advisoryCount,
+        totalAdvisories: advisoryEntries.length,
+        message: `Processing advisory records: ${advisoryCount} / ${advisoryEntries.length}`,
+      });
+      await yieldToEventLoop();
+    }
+  }
 
+  db.bulkUpsertVulnerabilities(parsedVulns);
+  db.setMetadata({
+    lastSyncAt: new Date().toISOString(),
+    sourceUrl,
+  });
+
+  return advisoryCount;
+}
+
+async function downloadAdvisoryArchive(
+  sourceUrl: string,
+  dbPath: string,
+  fetchImpl: FetchLike,
+  onProgress?: SyncOsvAdvisoriesOptions["onProgress"],
+): Promise<Uint8Array> {
   onProgress?.({
     phase: "init",
     sourceUrl,
     dbPath,
     message: `Sync initiated. Downloading advisory dump from ${sourceUrl}`,
   });
-
-  fs.mkdirSync(path.dirname(dbPath), { recursive: true });
-  if (fs.existsSync(dbPath)) {
-    fs.rmSync(dbPath, { force: true });
-  }
-
   const response = await fetchImpl(sourceUrl);
   if (!response.ok) {
     throw new Error(`OSV dump download failed: ${response.status} ${response.statusText}`);
@@ -111,7 +239,6 @@ export async function syncOsvAdvisories(
           : `Downloading advisory dump: ${formatBytes(bytesReceived)}`,
     });
   });
-
   onProgress?.({
     phase: "download-complete",
     bytesReceived: zippedBytes.byteLength,
@@ -122,7 +249,15 @@ export async function syncOsvAdvisories(
         : `Download complete: ${formatBytes(zippedBytes.byteLength)}`,
   });
   await yieldToEventLoop();
+  return zippedBytes;
+}
 
+async function buildAdvisoryDatabase(
+  zippedBytes: Uint8Array,
+  tempDbPath: string,
+  sourceUrl: string,
+  onProgress?: SyncOsvAdvisoriesOptions["onProgress"],
+): Promise<number> {
   const archiveEntries = unzipSync(zippedBytes);
   const advisoryEntries = Object.entries(archiveEntries).filter(([entryName]) => entryName.endsWith(".json"));
   onProgress?.({
@@ -132,72 +267,93 @@ export async function syncOsvAdvisories(
     message: `Archive loaded. Processing ${advisoryEntries.length} advisory ${pluralize(advisoryEntries.length, "record")}`,
   });
   await yieldToEventLoop();
-  const db = new LocalAdvisoryDatabase(dbPath);
+  const db = new LocalAdvisoryDatabase(tempDbPath);
+  let advisoryCount: number;
+  try {
+    advisoryCount = await ingestAdvisories(db, advisoryEntries, sourceUrl, onProgress);
+  } finally {
+    db.close();
+  }
+  if (advisoryCount === 0) {
+    throw new Error(
+      `OSV dump from ${sourceUrl} contained no usable advisory records. `
+        + `Check the source URL, then run "cve-lite advisories sync" again.`,
+    );
+  }
+  return advisoryCount;
+}
 
+/** Removes staged files, returning a warning for the caller to report. */
+function cleanupTempDirectory(tempDir: string): string | undefined {
   try {
-    const parsedVulns: OsvVuln[] = [];
-    let advisoryCount = 0;
-    const progressInterval = 250;
-
-    for (const [, bytes] of advisoryEntries) {
-      const text = Buffer.from(bytes).toString("utf8");
-      const vuln = JSON.parse(text) as OsvVuln;
-      if (!vuln.id) {
-        continue;
-      }
-
-      // OSV's /v1/querybatch endpoint excludes withdrawn advisories at query
-      // time; mirror that here so offline scans don't surface false positives
-      // from advisories that have been retracted.
-      if (vuln.withdrawn) {
-        continue;
-      }
-
-      parsedVulns.push(vuln);
-      advisoryCount += 1;
-
-      if (
-        advisoryCount === 1 ||
-        advisoryCount % progressInterval === 0 ||
-        advisoryCount === advisoryEntries.length
-      ) {
-        onProgress?.({
-          phase: "ingest",
-          processedAdvisories: advisoryCount,
-          totalAdvisories: advisoryEntries.length,
-          message: `Processing advisory records: ${advisoryCount} / ${advisoryEntries.length}`,
-        });
-        await yieldToEventLoop();
-      }
-    }
+    fs.rmSync(tempDir, { recursive: true, force: true });
+  } catch (error) {
+    const message = error instanceof Error ? error.message : String(error);
+    return `Could not remove temporary advisory sync directory ${tempDir}: ${ensureSentence(message)} `
+      + `Resolve the filesystem error and remove ${tempDir} manually.`;
+  }
+}
 
-    db.bulkUpsertVulnerabilities(parsedVulns);
+function buildSyncError(
+  refreshError: unknown,
+  databaseInstalled: boolean,
+  cleanupWarning?: string,
+): AdvisorySyncError {
+  const message = refreshError instanceof Error ? refreshError.message : String(refreshError);
+  const databaseState = refreshError instanceof AdvisorySyncError
+    ? refreshError.databaseState
+    : databaseInstalled ? "replaced" : "unchanged";
+  return new AdvisorySyncError(
+    cleanupWarning ? `${ensureSentence(message)} ${cleanupWarning}` : message,
+    databaseState,
+    refreshError,
+  );
+}
 
-    db.setMetadata({
-      lastSyncAt: new Date().toISOString(),
-      sourceUrl,
-    });
+export async function syncOsvAdvisories(
+  options: SyncOsvAdvisoriesOptions = {},
+): Promise<SyncOsvAdvisoriesResult> {
+  const onProgress = options.onProgress;
+  const sourceUrl = options.sourceUrl ?? DEFAULT_OSV_NPM_DUMP_URL;
+  const dbPath = getDefaultAdvisoryDbPath(options.outputPath);
+  rejectDirectoryOutput(dbPath);
+  let tempDir: string | undefined;
+  let databaseInstalled = false;
+  let refreshError: unknown;
+  let cleanupWarning: string | undefined;
+  let replacementWarning: string | undefined;
+  let syncResult: SyncOsvAdvisoriesResult | undefined;
+  const startupWarnings: string[] = [];
 
+  try {
+    fs.mkdirSync(path.dirname(dbPath), { recursive: true });
+    startupWarnings.push(...cleanupAbandonedStagingDirectories(dbPath));
+    tempDir = fs.mkdtempSync(stagingDirectoryPrefix(dbPath));
+    recordStagingOwner(tempDir, dbPath);
+    const tempDbPath = path.join(tempDir, path.basename(dbPath));
+    const zippedBytes = await downloadAdvisoryArchive(sourceUrl, dbPath, options.fetchImpl ?? fetch, onProgress);
+    const advisoryCount = await buildAdvisoryDatabase(zippedBytes, tempDbPath, sourceUrl, onProgress);
+    replacementWarning = replaceAdvisoryDatabase(tempDbPath, dbPath);
+    databaseInstalled = true;
     onProgress?.({
       phase: "complete",
       advisoryCount,
       dbPath,
       message: `Sync complete. Stored ${advisoryCount} advisory ${pluralize(advisoryCount, "record")} in ${dbPath}`,
     });
-
-    return { advisoryCount, dbPath, sourceUrl };
+    syncResult = { advisoryCount, dbPath, sourceUrl };
   } catch (error) {
-    if (fs.existsSync(dbPath)) {
-      fs.rmSync(dbPath, { force: true });
-    }
-    throw error;
+    refreshError = error;
   } finally {
-    try {
-      db.close();
-    } catch {
-      // ignore close errors during sync cleanup
-    }
+    if (tempDir !== undefined) cleanupWarning = cleanupTempDirectory(tempDir);
+  }
+
+  const warnings = [...startupWarnings, replacementWarning, cleanupWarning].filter((warning): warning is string => warning !== undefined);
+  if (syncResult === undefined) {
+    throw buildSyncError(refreshError, databaseInstalled, warnings.join(" ") || undefined);
   }
+  if (warnings.length > 0) syncResult.warnings = warnings;
+  return syncResult;
 }
 
 async function readResponseBytes(