fix: address review on the folded-migration round

Six findings from Copilot + Cursor on commit df89604, all valid
per the sub-agent eval pass. Three structural fixes, three doc/code
hygiene.

Code:

- workflows.ts: split proxyActivities into separate configs for sync
  and derive. Sync keeps the 5-min heartbeatTimeout (matches the
  per-1000-record heartbeat in the sync activity). Derive has no
  heartbeatTimeout — it's a single tight loop with no per-page
  heartbeat, and sharing the sync config would silently cancel the
  activity at Tier 2 scale (~600-700k packages). 1-hour
  startToCloseTimeout gives generous headroom over the ~5-min observed
  derive duration. Cursor caught this (CRITICAL).
- parseOsvRecord.ts: when an OSV record provides only an enumerated
  versions[] list (no ranges[]), convert each version into a degenerate
  range {introduced: v, fixed: null, last_affected: v} so isInRange
  matches exact-version vulnerability. Previously we kept the
  advisory_package row but wrote zero ranges, silently breaking the
  version-in-range check downstream. Copilot caught it (HIGH). Two new
  tests cover the conversion and the ranges-wins-over-versions case
  when both are present.
- upsertAdvisory.ts: scope the per-pass DELETE to rows where
  range_raw IS NULL AND unaffected_raw IS NULL. The deps.dev BQ ingest
  worker (future) will write rows with only the raw text columns
  populated; OSV's unconditional DELETE would wipe them on every
  pass. Copilot caught it (HIGH).

Cleanup:

- types.ts: drop the unused OsvConfig interface. Leftover from the
  pre-Temporal standalone-bin shape; the activity reads env on demand
  via requirePositiveInt in activities.ts. Misleading + dead code.
- V1779710880: rewrite the comment above advisory_affected_ranges.
  The previous comment said the structured boundary columns "are
  populated by a future range-parsing workstream" — that's no longer
  true; OSV ingest writes them now. The new comment captures the
  OSV-vs-deps.dev split and the DELETE-filter contract that keeps the
  two sources from clobbering each other.
- PR body: rewritten to describe the single-consolidated-migration
  shape instead of the three-additive-migrations shape the earlier
  draft assumed.

All 72 unit tests pass. Schema reapplied locally from a fresh
volume to confirm correctness.

Signed-off-by: Joan Reyero <joan@reyero.io>

Author: joanreyero

backend/src/osspckgs/migrations/V1779710880__initial_schema.sql

@@ -614,10 +614,14 @@ CREATE INDEX ON advisory_packages (package_id)
 WHERE
     package_id IS NOT NULL;
 
--- Version ranges affected by an advisory per package.
+-- Version ranges affected by an advisory per package. Populated by the OSV
+-- ingest worker (packages_worker/src/osv) using introduced_version /
+-- fixed_version / last_affected. range_raw / unaffected_raw are reserved
+-- for the deps.dev BQ ingest worker (future): that worker writes the raw
+-- range strings without parsing into structured boundaries. The OSV upsert
+-- path only deletes rows where range_raw / unaffected_raw are both NULL,
+-- so deps.dev rows are not clobbered when OSV re-syncs.
 -- COALESCE prevents silent duplicates when introduced_version is NULL.
--- BQ-sourced rows populate range_raw / unaffected_raw only; introduced/fixed/last_affected
--- are populated by a future range-parsing workstream.
 CREATE TABLE advisory_affected_ranges (
     id bigserial PRIMARY KEY,
     advisory_package_id bigint NOT NULL REFERENCES advisory_packages (id),

services/apps/packages_worker/src/osv/__tests__/parseOsvRecord.test.ts

@@ -202,6 +202,46 @@ describe('parseOsvRecord — range flattening', () => {
     expect(out.packages).toEqual([])
   })
 
+  it('converts versions[] to discrete ranges when ranges[] is empty', () => {
+    // Some OSV records list exact vulnerable versions without a structured
+    // range. Without the conversion the advisory_package would land in the DB
+    // with zero ranges and deriveCriticalFlag would never flag the package.
+    const r = baseRecord({
+      affected: [
+        {
+          package: { ecosystem: 'npm', name: 'pkg' },
+          versions: ['1.0.0', '1.0.1', '1.0.2'],
+        },
+      ],
+    })
+    const out = parseOsvRecord(r, ALLOW)
+    expect(out.packages).toHaveLength(1)
+    expect(out.packages[0].ranges).toEqual([
+      { introducedVersion: '1.0.0', fixedVersion: null, lastAffected: '1.0.0' },
+      { introducedVersion: '1.0.1', fixedVersion: null, lastAffected: '1.0.1' },
+      { introducedVersion: '1.0.2', fixedVersion: null, lastAffected: '1.0.2' },
+    ])
+  })
+
+  it('ignores versions[] when ranges[] is non-empty (avoids redundant rows)', () => {
+    // If OSV provides both, ranges[] is the source of truth — versions[] is
+    // typically a subset enumeration. Avoid writing the same vulnerability
+    // window twice (once as a structured range, once as a per-version range).
+    const r = baseRecord({
+      affected: [
+        {
+          package: { ecosystem: 'npm', name: 'pkg' },
+          ranges: [{ type: 'SEMVER', events: [{ introduced: '1.0.0' }, { fixed: '2.0.0' }] }],
+          versions: ['1.5.0'],
+        },
+      ],
+    })
+    const out = parseOsvRecord(r, ALLOW)
+    expect(out.packages[0].ranges).toEqual([
+      { introducedVersion: '1.0.0', fixedVersion: '2.0.0', lastAffected: null },
+    ])
+  })
+
   it('merges multiple affected[] entries for the same package', () => {
     // Some OSV records list the same (ecosystem, name) multiple times to express
     // disjoint range sets. We collapse them under one advisory_package row.

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

@@ -115,7 +115,17 @@ export function parseOsvRecord(
     if (!allowedEcosystems.has(ecosystem)) continue
 
     const ranges = flattenRanges(affected)
-    if (ranges.length === 0 && (affected.versions ?? []).length === 0) {
+    // If OSV only provides an enumerated `versions[]` list (no `ranges[]`),
+    // convert each exact version into a degenerate range (introduced=v,
+    // last_affected=v) so isInRange matches when `latest_version === v`.
+    // Without this we'd record the advisory_package but no range rows, and
+    // deriveCriticalFlag would silently fail to flag the package.
+    if (ranges.length === 0 && (affected.versions ?? []).length > 0) {
+      for (const v of affected.versions ?? []) {
+        ranges.push({ introducedVersion: v, fixedVersion: null, lastAffected: v })
+      }
+    }
+    if (ranges.length === 0) {
       // No usable affected range at all; skip this package entry but keep going.
       continue
     }

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

@@ -1,14 +1,3 @@
-// Worker config, read from env in src/config.ts.
-export interface OsvConfig {
-  bulkBaseUrl: string
-  ecosystems: string[]
-  syncIntervalHours: number
-  idleSleepSec: number
-  tmpDir: string
-  batchSize: number
-  deriveBatchSize: number
-}
-
 // ---- Raw OSV JSON (only the fields we read) ----
 // Reference: https://ossf.github.io/osv-schema/
 

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

@@ -85,8 +85,18 @@ async function upsertOne(qx: QueryExecutor, record: NormalizedRecord): Promise<v
     )
     const advisoryPackageId = advisoryPackageRow.id as number
 
+    // Only delete OSV-derived rows: rows with at least one of
+    // introduced/fixed/last_affected populated AND no deps.dev-source raw text
+    // columns. The deps.dev BQ worker (future) is expected to populate
+    // range_raw / unaffected_raw on rows of its own; we must not wipe those
+    // on every OSV pass.
     await qx.result(
-      `DELETE FROM advisory_affected_ranges WHERE advisory_package_id = $(advisoryPackageId)`,
+      `
+      DELETE FROM advisory_affected_ranges
+      WHERE advisory_package_id = $(advisoryPackageId)
+        AND range_raw IS NULL
+        AND unaffected_raw IS NULL
+      `,
       { advisoryPackageId },
     )
 

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

@@ -9,7 +9,14 @@ import type * as activities from './activities'
 // §`has_critical_vulnerability` semantics the derive step runs after every
 // ingest pass so packages added between schedule firings are at most one
 // cycle stale and self-heal on the next run.
-const { osvSyncEcosystem, osvDeriveCriticalFlag } = proxyActivities<typeof activities>({
+//
+// Sync and derive use separate proxyActivities configs because their
+// heartbeat shape differs: sync emits one heartbeat per ~1000 records (see
+// activities.ts) so a 5-minute heartbeatTimeout is the right liveness signal;
+// derive is a single tight loop over packages with no per-page heartbeat,
+// so it relies on startToCloseTimeout only — sharing the sync heartbeat
+// config would silently cancel the derive activity at Tier 2 scale.
+const { osvSyncEcosystem } = proxyActivities<typeof activities>({
   // npm sync alone is ~1 hour today (N+1 upsert path, see deferred review
   // comment on upsertAdvisory.ts). Maven is ~5 minutes. We give each
   // per-ecosystem activity a generous 2-hour ceiling.
@@ -27,6 +34,20 @@ const { osvSyncEcosystem, osvDeriveCriticalFlag } = proxyActivities<typeof activ
   },
 })
 
+const { osvDeriveCriticalFlag } = proxyActivities<typeof activities>({
+  // Paged scan over packages (~600-700k at Tier 2 scale). The whole derive
+  // pass runs in ~5 minutes on the current dataset; we give it 1 hour of
+  // headroom for the table to grow. No heartbeatTimeout — the activity does
+  // not heartbeat, so adding one would cancel the activity silently before
+  // startToCloseTimeout fires.
+  startToCloseTimeout: '1 hour',
+  retry: {
+    initialInterval: '30 seconds',
+    backoffCoefficient: 2,
+    maximumAttempts: 3,
+  },
+})
+
 export interface OsvSyncWorkflowInput {
   // OSV ecosystem labels in OSV's canonical case ('npm', 'Maven', 'PyPI', …).
   // Same list is used by the per-ecosystem download loop, where OSV's bucket