fix: diff-based upsert + soft-delete for advisory ranges (CM-1258)

Signed-off-by: Uroš Marolt <uros@marolt.me>

Author: themarolt

services/apps/packages_worker/src/osv/upsertAdvisory.ts

@@ -1,7 +1,7 @@
 import {
-  deleteOsvOnlyRanges,
   findPackageId,
-  insertAdvisoryRange,
+  reconcileOsvRanges,
+  supersedeDepsDevRanges,
   upsertAdvisory,
   upsertAdvisoryPackage,
 } from '@crowd/data-access-layer/src/packages/osv'
@@ -46,16 +46,17 @@ async function upsertOne(qx: QueryExecutor, record: NormalizedRecord): Promise<v
       packageName: entry.pkg.packageName,
     })
 
-    await deleteOsvOnlyRanges(qx, advisoryPackageId)
-
-    for (const range of dedupeRanges(entry.ranges)) {
-      await insertAdvisoryRange(qx, {
+    await reconcileOsvRanges(
+      qx,
+      advisoryPackageId,
+      dedupeRanges(entry.ranges).map((range) => ({
         advisoryPackageId,
         introducedVersion: range.introducedVersion,
         fixedVersion: range.fixedVersion,
         lastAffected: range.lastAffected,
-      })
-    }
+      })),
+    )
+    await supersedeDepsDevRanges(qx, advisoryPackageId)
   }
 }
 

services/libs/data-access-layer/src/packages/osv.ts

@@ -139,42 +139,98 @@ export async function upsertAdvisoryPackage(
   return row.id as number
 }
 
-// Delete every advisory_affected_ranges row for this advisory_package that
-// lacks deps.dev raw columns — i.e. every OSV-owned row. The deps.dev BQ
-// worker (future) is expected to populate range_raw / unaffected_raw on rows
-// of its own and never set the structured introduced/fixed/last_affected
-// columns; this predicate scopes the wipe to the OSV pipeline's rows so a
-// deps.dev row is never clobbered on resync. OSV rows where all three
-// structured columns are NULL (e.g. some MAL- "always vulnerable" ranges)
-// are still deleted here and re-inserted from the new payload below — that's
-// fine, the row is OSV-owned and idempotent.
-export async function deleteOsvOnlyRanges(
+// Diff-based upsert + soft-delete for OSV-owned advisory_affected_ranges rows.
+// Replaces the old hard-delete + reinsert sweep (which minted a new PK for
+// every unchanged range every sync, producing zombie rows once Sequin
+// replicates DELETEs into a Tinybird ReplacingMergeTree that has no
+// sign/is_deleted column — see ADR-0001 §`advisory_affected_ranges`
+// delete/dedup strategy, CM-1258).
+//
+// An advisory_package whose range set hasn't changed since the last sync now
+// produces zero writes: matching tuples are left untouched (not even an
+// `updated_at` bump), so no Sequin/Tinybird event fires for them either.
+//
+// Tuple in both old and new -> untouched. Tuple only in new -> upserted
+// (INSERT ON CONFLICT, clearing deleted_at in case it was previously
+// soft-deleted and OSV brought it back). Tuple only in old -> soft-deleted.
+// Soft-delete (not hard-delete) is required because the unique key is the
+// value tuple itself: when OSV corrects a range, the corrected tuple has no
+// successor row to collapse into, so without deleted_at the stale tuple
+// would sit forever as a false-positive vulnerable-range match.
+export async function reconcileOsvRanges(
   qx: QueryExecutor,
   advisoryPackageId: number,
+  ranges: AdvisoryRangeInsertInput[],
 ): Promise<void> {
+  const values = ranges.map((r) => ({
+    introduced_version: r.introducedVersion,
+    fixed_version: r.fixedVersion,
+    last_affected: r.lastAffected,
+  }))
+
   await qx.result(
     `
-    DELETE FROM advisory_affected_ranges
-    WHERE advisory_package_id = $(advisoryPackageId)
-      AND range_raw IS NULL
-      AND unaffected_raw IS NULL
+    INSERT INTO advisory_affected_ranges
+      (advisory_package_id, introduced_version, fixed_version, last_affected, created_at, updated_at)
+    SELECT $(advisoryPackageId), v.introduced_version, v.fixed_version, v.last_affected, NOW(), NOW()
+    FROM jsonb_to_recordset($(values)::jsonb) AS v(
+      introduced_version text,
+      fixed_version text,
+      last_affected text
+    )
+    ON CONFLICT (advisory_package_id, COALESCE(introduced_version, ''), COALESCE(fixed_version, ''), COALESCE(last_affected, ''))
+    DO UPDATE SET
+      updated_at = NOW(),
+      deleted_at = NULL
+    WHERE advisory_affected_ranges.deleted_at IS NOT NULL
     `,
-    { advisoryPackageId },
+    { advisoryPackageId, values: JSON.stringify(values) },
+  )
+
+  // Soft-delete OSV-owned live rows whose tuple isn't in the new set anymore.
+  await qx.result(
+    `
+    UPDATE advisory_affected_ranges ar
+    SET deleted_at = NOW()
+    WHERE ar.advisory_package_id = $(advisoryPackageId)
+      AND ar.deleted_at IS NULL
+      AND ar.range_raw IS NULL
+      AND ar.unaffected_raw IS NULL
+      AND NOT EXISTS (
+        SELECT 1
+        FROM jsonb_to_recordset($(values)::jsonb) AS v(
+          introduced_version text,
+          fixed_version text,
+          last_affected text
+        )
+        WHERE COALESCE(v.introduced_version, '') = COALESCE(ar.introduced_version, '')
+          AND COALESCE(v.fixed_version, '') = COALESCE(ar.fixed_version, '')
+          AND COALESCE(v.last_affected, '') = COALESCE(ar.last_affected, '')
+      )
+    `,
+    { advisoryPackageId, values: JSON.stringify(values) },
   )
 }
 
-export async function insertAdvisoryRange(
+// OSV is the source of truth for ranges over time (see ADR-0001 §Write
+// semantics `advisories` row): once OSV has written its structured ranges
+// for an advisory_package, any live deps.dev raw row for that same
+// advisory_package is superseded and should stop being treated as current.
+// Soft-deleted (not hard-deleted) for the same reason as reconcileOsvRanges —
+// no successor row exists to collapse a deps.dev tuple into.
+export async function supersedeDepsDevRanges(
   qx: QueryExecutor,
-  range: AdvisoryRangeInsertInput,
+  advisoryPackageId: number,
 ): Promise<void> {
   await qx.result(
     `
-    INSERT INTO advisory_affected_ranges
-      (advisory_package_id, introduced_version, fixed_version, last_affected, created_at, updated_at)
-    VALUES
-      ($(advisoryPackageId), $(introducedVersion), $(fixedVersion), $(lastAffected), NOW(), NOW())
+    UPDATE advisory_affected_ranges
+    SET deleted_at = NOW()
+    WHERE advisory_package_id = $(advisoryPackageId)
+      AND deleted_at IS NULL
+      AND (range_raw IS NOT NULL OR unaffected_raw IS NOT NULL)
     `,
-    range,
+    { advisoryPackageId },
   )
 }