fix: dedup advisory_affected_ranges via diff-based upsert (CM-1258)

backend/src/osspckgs/migrations/V1784567585__advisory_affected_ranges_soft_delete.sql

@@ -0,0 +1,5 @@
+ALTER TABLE advisory_affected_ranges ADD COLUMN IF NOT EXISTS deleted_at TIMESTAMPTZ;
+
+CREATE INDEX IF NOT EXISTS advisory_affected_ranges_live_idx
+   ON advisory_affected_ranges (advisory_package_id)
+   WHERE deleted_at IS NULL;

docs/adr/0001-oss-packages-design-decisions.md

@@ -223,7 +223,7 @@ Five sub-workers run concurrently (npm, Maven, OSV, GitHub, Docker Hub), all wri
 | `versions`                            | Append-only via `INSERT … ON CONFLICT DO NOTHING`. Yanked/deprecated status is a separate targeted `UPDATE (is_yanked = true) WHERE …`.                                                                                                                     |
 | `repos`                               | Registry workers (npm, Maven) do **not** write `repos` enrichment metadata. They INSERT a minimal `repos(url, host)` row — `url` (canonical) and `host` (coarse classification) are both derived from the declared repository URL — solely to create the FK target their `package_repos` link needs. `owner`/`name`/`stars`/`description` and all other metadata stay NULL and remain enricher-owned; existing rows are never updated by registry workers. The GitHub enricher — triggered when `repos.last_synced_at IS NULL` — upserts `repos` with metadata. Docker Hub worker adds `docker_*` columns on top. |
 | `package_repos`                       | Composite PK `(package_id, repo_url)`. Each `source` value ('declared', 'deps_dev', 'heuristic', 'manual') is a separate row — sources do not overwrite each other.                                                                                         |
-| `advisories`                          | Upsert on `osv_id`. OSV is the single source of truth; no other worker writes to this table.                                                                                                                                                                |
+| `advisories`                          | Upsert on `osv_id`. OSV is the source of truth for ongoing updates; deps.dev also writes this table at backfill / new-package time (see §Source of truth: deps.dev backfill vs registries / OSV) — OSV's writes are expected to overwrite deps.dev's over time, not the other way around.                                                                                                                                                                |
 | `maintainers` / `package_maintainers` | `maintainers`: upsert on `(ecosystem, username)`, never deleted — the identity history is preserved. `package_maintainers`: reflects the **current** link set — the npm worker replaces a package's links each ingest (delete + reinsert), so prior link rows are not retained.                                                                                                                                                                                     |
 | `downloads_daily`                     | Append-only time-series. Each `(package_id, date)` row is written once. npm and Maven workers own disjoint rows by ecosystem. Historical timelines are preserved — workers do not overwrite past dates.                                                     |
 | `downloads_last_30d`                  | Upsert on `(purl, end_date)`. Written by the weekly ranking worker only. The cached `packages.downloads_last_30d` column must be updated in the same pass.                                                                                                  |
@@ -364,6 +364,26 @@ Local verification against the live OSV dataset (2026-05-28) showed the multi-ra
 
 ---
 
+### `advisory_affected_ranges` delete/dedup strategy
+
+Two duplication problems surfaced in Tinybird for `advisory_affected_ranges` (CM-1258):
+
+**Problem A — zombie rows from hard-delete + reinsert.** The OSV worker's per-sync flow (`upsertAdvisory.ts` → `deleteOsvOnlyRanges` in `services/libs/data-access-layer/src/packages/osv.ts`) hard-deletes all OSV-owned ranges for an advisory_package and reinserts the deduped set via plain `INSERT` (no `ON CONFLICT`), so an unchanged range gets a brand-new PK every sync. `advisory_affected_ranges` is Sequin-replicated with `REPLICA IDENTITY FULL` into a Tinybird `ReplacingMergeTree(updatedAt)` datasource that has no sign/`is_deleted` column and no delete-shaped ingest path — the DELETE never propagates downstream, only the new-PK reinsert does. Net effect: every sync produces one zombie row per range, even when nothing changed.
+
+**Problem B — deps.dev raw rows and OSV structured rows never collide.** deps.dev writes `range_raw`/`unaffected_raw`; OSV writes `introduced_version`/`fixed_version`/`last_affected`. Per §`advisory_affected_ranges` uniqueness scope above, the unique index is on the structured columns only, so a deps.dev row (structured columns NULL) never conflicts with the equivalent OSV row — both survive as separate rows for the same real-world range, even though §Source of truth already says OSV should overwrite deps.dev's data over time.
+
+**Decided fix — diff-based upsert with soft-delete, not sweep-and-reinsert:**
+
+- Add a `deleted_at` column to `advisory_affected_ranges` (soft-delete, same pattern as `V1783036800__security_contacts_soft_delete.sql`).
+- Per advisory_package, per sync: read existing live OSV-owned tuples, diff against the new deduped tuple set. Tuple in both → skip (no write, no event). Tuple only in new → `INSERT ... ON CONFLICT (advisory_package_id, COALESCE(introduced_version,''), COALESCE(fixed_version,''), COALESCE(last_affected,'')) DO UPDATE SET updated_at = NOW(), deleted_at = NULL`. Tuple only in old → soft-delete (`SET deleted_at = NOW()`), not hard-delete. An unchanged advisory_package now produces zero writes and zero Sequin/Tinybird events per sync.
+- Soft-delete (not hard-delete) is required because the unique key is the value tuple itself: when OSV corrects a range (e.g. narrows `fixed_version`), the corrected tuple is a different key, so the old tuple's row has no successor row to collapse into — without a `deleted_at` flag it would sit forever as a false-positive vulnerable-range match.
+- After OSV writes its structured ranges for an advisory_package, soft-delete any live deps.dev-owned row (`range_raw IS NOT NULL OR unaffected_raw IS NOT NULL`) for that same advisory_package — this is the §Source of truth overwrite rule, made to actually take effect now that raw and structured rows are reconciled instead of coexisting.
+- Tinybird: add `deletedAt` to `advisoryAffectedRanges.datasource` and filter `deletedAt IS NULL` in `ossPackages_enriched.pipe`'s join, alongside the existing `FINAL` modifier (`FINAL` only collapses rows sharing a sorting key — it does not merge two distinct PKs claiming the same real-world range, which is why this needs an explicit filter rather than relying on `FINAL` alone).
+
+**Decided**: 2026-07-20
+
+---
+
 ### Per-source ingestion strategies
 
 #### npm
@@ -576,6 +596,7 @@ The writer of a `downloads_last_30d` row must also update the cached `packages.d
 - 2026-05-29 — clarified `packages_universe` import semantics (one-time backfill + weekly snapshot-diff incrementals; the ranking job updates score/flag columns in place). Added §Source of truth: deps.dev backfill vs registries / OSV with lifecycle ownership rules and the agreed `package_source_log` provenance table (`(package_id, source)` PK; `columns` array tracks `table.column` paths each source writes).
 - 2026-05-29 — added §Criticality scoring methodology (graph signals — transitive dependent count and PageRank centrality; per-ecosystem percentile-rank formula in `[0, 1]`; floor + ceiling tier budget policy; `package_criticality_spotlight` table).
 - 2026-06-08 — retired `packages_universe` table. Signals (`downloads_last_30d`, `centrality_score`, `rank_in_ecosystem`) migrated onto `packages`. Both `rank_packages_universe()` overloads dropped; replaced by `rank_packages()` operating directly on `packages`, scoped to ecosystems present in `critical_top_n_by_ecosystem` JSONB. §Downloads timeline by tier simplified (Tier 2/3 split removed). §Write semantics `packages_universe` row removed.
+- 2026-07-20 (CM-1258) — corrected stale §Write semantics `advisories` row (deps.dev also writes this table; OSV overwrites over time per §Source of truth, it isn't the sole writer). Added §`advisory_affected_ranges` delete/dedup strategy: diff-based upsert + soft-delete (`deleted_at`) replaces hard-delete + reinsert to stop zombie rows in Tinybird, and OSV now supersedes deps.dev's raw rows on overlap so the two sources stop coexisting as duplicates.
 ---
 
 ## Note on promotion to production

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/osspckgs/api.ts

@@ -1009,7 +1009,7 @@ export async function getAdvisoriesByPackageId(
         ${ADVISORY_RESOLUTION_EXPR} AS resolution
       FROM advisory_packages ap
       JOIN advisories a ON a.id = ap.advisory_id
-      LEFT JOIN advisory_affected_ranges ar ON ar.advisory_package_id = ap.id
+      LEFT JOIN advisory_affected_ranges ar ON ar.advisory_package_id = ap.id AND ar.deleted_at IS NULL
       JOIN packages p ON p.id = ap.package_id
       WHERE ap.package_id = $(packageId)::bigint
       GROUP BY a.osv_id, a.severity, a.is_critical, p.latest_version
@@ -1117,7 +1117,7 @@ export async function getAdvisoriesByPurls(
     FROM packages p
     LEFT JOIN advisory_packages ap ON ap.package_id = p.id
     LEFT JOIN advisories a ON a.id = ap.advisory_id
-    LEFT JOIN advisory_affected_ranges ar ON ar.advisory_package_id = ap.id
+    LEFT JOIN advisory_affected_ranges ar ON ar.advisory_package_id = ap.id AND ar.deleted_at IS NULL
     WHERE p.purl = ANY($(purls))
     GROUP BY p.purl, p.latest_version, a.osv_id, a.severity, a.is_critical
     ORDER BY

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

@@ -139,42 +139,107 @@ 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, already a clean live OSV row -> untouched.
+// Tuple only in new, or matching a live deps.dev raw row on the same key ->
+// upserted (INSERT ON CONFLICT; clears deleted_at in case it was previously
+// soft-deleted, and clears range_raw/unaffected_raw to reclaim the row from
+// deps.dev — OSV wins on key overlap per ADR-0001 §Write semantics). 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,
+      range_raw = NULL,
+      unaffected_raw = NULL
+    WHERE advisory_affected_ranges.deleted_at IS NOT NULL
+       OR advisory_affected_ranges.range_raw IS NOT NULL
+       OR advisory_affected_ranges.unaffected_raw 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(),
+        updated_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(),
+        updated_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 },
   )
 }
 
@@ -232,7 +297,7 @@ export async function getRangesForPackages(qx: QueryExecutor, ids: number[]): Pr
       a.osv_id
     FROM advisory_packages ap
     JOIN advisories a ON a.id = ap.advisory_id
-    JOIN advisory_affected_ranges ar ON ar.advisory_package_id = ap.id
+    JOIN advisory_affected_ranges ar ON ar.advisory_package_id = ap.id AND ar.deleted_at IS NULL
     WHERE ap.package_id IN ($(ids:csv))
       AND (a.is_critical = TRUE OR a.osv_id LIKE 'MAL-%')
     `,

services/libs/tinybird/datasources/advisoryAffectedRanges.datasource

@@ -10,6 +10,7 @@ DESCRIPTION >
     - `rangeRaw` is the raw AffectedVersions string from the deps.dev BigQuery source (empty string if OSV-sourced).
     - `unaffectedRaw` is the raw UnaffectedVersions string from deps.dev BigQuery (empty string if OSV-sourced).
     - `createdAt` and `updatedAt` are row-level audit timestamps for Tinybird watermark-based sync.
+    - `deletedAt` is set when Postgres soft-deletes the row (stale OSV tuple or superseded deps.dev raw row); NULL means live. `FINAL` alone doesn't collapse this — a soft-deleted row and its live replacement are distinct PKs, so consumers must also filter `deletedAt IS NULL` (see ADR-0001 §`advisory_affected_ranges` delete/dedup strategy, CM-1258).
 
 SCHEMA >
     `id` UInt64 `json:$.record.id`,
@@ -20,7 +21,8 @@ SCHEMA >
     `rangeRaw` String `json:$.record.range_raw` DEFAULT '',
     `unaffectedRaw` String `json:$.record.unaffected_raw` DEFAULT '',
     `createdAt` DateTime64(3) `json:$.record.created_at`,
-    `updatedAt` DateTime64(3) `json:$.record.updated_at`
+    `updatedAt` DateTime64(3) `json:$.record.updated_at`,
+    `deletedAt` Nullable(DateTime64(3)) `json:$.record.deleted_at`
 
 ENGINE ReplacingMergeTree
 ENGINE_PARTITION_KEY toYear(createdAt)

services/libs/tinybird/pipes/ossPackages_enriched.pipe

@@ -93,6 +93,7 @@ SQL >
     INNER JOIN advisoryPackages AS ap FINAL ON ap.packageId = p.id
     INNER JOIN advisories AS a FINAL ON a.id = ap.advisoryId
     INNER JOIN advisoryAffectedRanges AS aar FINAL ON aar.advisoryPackageId = ap.id
+        AND aar.deletedAt IS NULL
     LEFT JOIN
         (SELECT packageId, number, publishedAt AS introducedAt FROM versions FINAL) AS vi
         ON vi.packageId = p.id