fix: audit packagist writes inside their own transaction

Signed-off-by: anilb <epipav@gmail.com>

Author: epipav

services/apps/packages_worker/src/packagist/__tests__/downloads.test.ts

@@ -1,17 +1,24 @@
 import { beforeEach, describe, expect, it, vi } from 'vitest'
 
-import { insertLast30dDownloadIfAbsent } from '@crowd/data-access-layer/src/packages'
+import {
+  insertLast30dDownloadIfAbsent,
+  logAuditFieldChanges,
+} from '@crowd/data-access-layer/src/packages'
 import type { QueryExecutor } from '@crowd/data-access-layer/src/queryExecutor'
 
 import { monthlyWindowFor, persistPackagist30dWindow } from '../downloads'
 
 vi.mock('@crowd/data-access-layer/src/packages', () => ({
   insertLast30dDownloadIfAbsent: vi.fn().mockResolvedValue([]),
+  logAuditFieldChanges: vi.fn(),
 }))
 
 const mockWindow = vi.mocked(insertLast30dDownloadIfAbsent)
+const mockAudit = vi.mocked(logAuditFieldChanges)
 
-const qx = {} as QueryExecutor
+const qx = {
+  tx: vi.fn((cb: (t: QueryExecutor) => Promise<string[]>) => cb(qx)),
+} as unknown as QueryExecutor
 const PURL = 'pkg:composer/monolog/monolog'
 
 beforeEach(() => {
@@ -45,30 +52,37 @@ describe('monthlyWindowFor', () => {
 // is never overwritten — the value is the observation closest to the boundary.
 // insertLast30dDownloadIfAbsent does the presence check and the write atomically, so
 // there is no separate "does it exist" call to assert on here — a race is exercised
-// directly against Postgres in downloadsLast30d's own DAL-level coverage.
+// directly against Postgres in downloadsLast30d's own DAL-level coverage. The audit
+// record shares the same transaction as the write.
 describe('persistPackagist30dWindow', () => {
-  it('writes the window with the mirror and returns the changed fields when the month has no row yet', async () => {
+  it('writes the window with the mirror, audits it, and returns the changed fields when the month has no row yet', async () => {
     mockWindow.mockResolvedValue(['downloads_last_30d.count', 'packages.downloads_last_30d'])
 
     const changedFields = await persistPackagist30dWindow(qx, PURL, 300, '2026-07-01')
 
     expect(mockWindow).toHaveBeenCalledWith(qx, PURL, '2026-06-01', '2026-07-01', 300, true)
+    expect(mockAudit).toHaveBeenCalledWith(qx, 'packagist', PURL, [
+      'downloads_last_30d.count',
+      'packages.downloads_last_30d',
+    ])
     expect(changedFields).toEqual(['downloads_last_30d.count', 'packages.downloads_last_30d'])
   })
 
-  it('returns no changed fields when a window for the month already exists', async () => {
+  it('returns no changed fields and does not audit when a window for the month already exists', async () => {
     mockWindow.mockResolvedValue([])
 
     const changedFields = await persistPackagist30dWindow(qx, PURL, 300, '2026-07-15')
 
     expect(mockWindow).toHaveBeenCalledWith(qx, PURL, '2026-06-01', '2026-07-01', 300, true)
+    expect(mockAudit).toHaveBeenCalledWith(qx, 'packagist', PURL, [])
     expect(changedFields).toEqual([])
   })
 
   it('writes nothing and returns no changed fields when the registry reported no monthly count', async () => {
     const changedFields = await persistPackagist30dWindow(qx, PURL, null, '2026-07-01')
 
     expect(mockWindow).not.toHaveBeenCalled()
+    expect(mockAudit).not.toHaveBeenCalled()
     expect(changedFields).toEqual([])
   })
 })

services/apps/packages_worker/src/packagist/__tests__/ingest.test.ts

@@ -72,7 +72,9 @@ const mockFetchList = vi.mocked(fetchPackagistPackageList)
 const mockParseList = vi.mocked(parsePackagistPackageList)
 const mockSeedInsert = vi.mocked(insertPackagistPackages)
 
-const qx = {} as QueryExecutor
+const qx = {
+  tx: vi.fn((cb: (t: QueryExecutor) => Promise<void>) => cb(qx)),
+} as unknown as QueryExecutor
 const PURL = 'pkg:composer/monolog/monolog'
 const RUN_DATE = '2026-07-15'
 const SCHEDULED_AT = '2026-07-15T00:00:00.000Z'
@@ -111,7 +113,7 @@ describe('ingestOnePackagistMetadata', () => {
     })
   }
 
-  it('audits phase 1 immediately and phase 2 separately, then marks scanned with the fresh Last-Modified', async () => {
+  it('persists phase 1 immediately and phase 2 separately, then marks scanned with the fresh Last-Modified', async () => {
     happyMocks()
 
     await ingestOnePackagistMetadata(qx, candidate, SCHEDULED_AT)
@@ -120,10 +122,9 @@ describe('ingestOnePackagistMetadata', () => {
     expect(mockFetchP2).toHaveBeenCalledWith('monolog/monolog', 'Tue, 30 Jun 2026 00:00:00 GMT')
     expect(mockExpand).toHaveBeenCalledWith(minified)
     expect(mockPersistMetadata).toHaveBeenCalledWith(qx, PURL, expanded)
-    // Two separate audit rows, not one merged call — phase 1 is logged as soon as it's
-    // committed, before the p2 fetch (which can throw) ever runs.
-    expect(mockAudit).toHaveBeenNthCalledWith(1, qx, 'packagist', PURL, ['packages.description'])
-    expect(mockAudit).toHaveBeenNthCalledWith(2, qx, 'packagist', PURL, ['versions.number'])
+    // Each phase audits its own writes atomically inside its own persist* call (see
+    // persistPackageInfo.test.ts / persistMetadata.test.ts) — phase 1 is committed and
+    // audited before the p2 fetch (which can throw) ever runs.
     expect(mockMarkMetadata).toHaveBeenCalledWith(
       qx,
       PURL,
@@ -209,8 +210,9 @@ describe('ingestOnePackagistMetadata', () => {
     )
   })
 
-  it('audits phase-1 changes even when the p2 fetch gives up', async () => {
-    // the dynamic-endpoint writes are committed by then — their audit rows must not be dropped
+  it('still persists (and self-audits) phase 1 even when the p2 fetch gives up', async () => {
+    // the dynamic-endpoint writes are committed and audited by persistPackagistPackageInfo
+    // itself before the p2 fetch (which can throw) ever runs — see persistPackageInfo.test.ts
     vi.useFakeTimers()
     mockFetchStats.mockResolvedValue(statsJson as never)
     mockPersistInfo.mockResolvedValue({ found: true, changedFields: ['packages.description'] })
@@ -224,7 +226,7 @@ describe('ingestOnePackagistMetadata', () => {
     await vi.runAllTimersAsync()
     await p
 
-    expect(mockAudit).toHaveBeenCalledWith(qx, 'packagist', PURL, ['packages.description'])
+    expect(mockPersistInfo).toHaveBeenCalledWith(qx, PURL, expect.anything())
     expect(mockMarkMetadata).toHaveBeenCalledWith(
       qx,
       PURL,
@@ -233,32 +235,30 @@ describe('ingestOnePackagistMetadata', () => {
     )
   })
 
-  it('throws on a transient p2 result without marking scanned, but still audits phase 1', async () => {
+  it('throws on a transient p2 result without marking scanned, but phase 1 already persisted', async () => {
     mockFetchStats.mockResolvedValue(statsJson as never)
     mockPersistInfo.mockResolvedValue({ found: true, changedFields: ['packages.description'] })
     mockFetchP2.mockResolvedValue({ kind: 'TRANSIENT', message: 'HTTP 502' } as never)
 
     await expect(ingestOnePackagistMetadata(qx, candidate, SCHEDULED_AT)).rejects.toThrow()
     expect(mockMarkMetadata).not.toHaveBeenCalled()
-    // phase-1 writes are already committed when the throw happens — a retry re-runs
-    // phase 1 idempotently and reports no changes, so this audit event can't be deferred
-    expect(mockAudit).toHaveBeenCalledWith(qx, 'packagist', PURL, ['packages.description'])
+    // phase-1 writes (and their audit row) are already committed by
+    // persistPackagistPackageInfo when the throw happens — a retry re-runs phase 1
+    // idempotently and reports no changes, so nothing here can lose that audit event
+    expect(mockPersistInfo).toHaveBeenCalledWith(qx, PURL, expect.anything())
   })
 })
 
 // The monthly downloads-30d lane: dynamic fetch, window row only.
 describe('ingestOnePackagist30dWindow', () => {
-  it('persists the observed rolling window, audits the change, and marks the run processed', async () => {
+  it('persists the observed rolling window (self-audited) and marks the run processed', async () => {
     mockFetchStats.mockResolvedValue(statsJson as never)
     mockPersist30d.mockResolvedValue(['downloads_last_30d.count', 'packages.downloads_last_30d'])
 
     await ingestOnePackagist30dWindow(qx, PURL, RUN_DATE, SCHEDULED_AT)
 
     expect(mockPersist30d).toHaveBeenCalledWith(qx, PURL, 300, RUN_DATE)
-    expect(mockAudit).toHaveBeenCalledWith(qx, 'packagist', PURL, [
-      'downloads_last_30d.count',
-      'packages.downloads_last_30d',
-    ])
+    // persistPackagist30dWindow audits its own write atomically — see downloads.test.ts
     expect(mockMark30d).toHaveBeenCalledWith(
       qx,
       PURL,

services/apps/packages_worker/src/packagist/__tests__/persistMetadata.test.ts

@@ -2,6 +2,7 @@ import { beforeEach, describe, expect, it, vi } from 'vitest'
 
 import {
   getPackagistPackageIdsByNames,
+  logAuditFieldChanges,
   reconcileVersionDependencies,
   updatePackagistVersionAggregates,
   upsertPackagistVersions,
@@ -16,12 +17,14 @@ vi.mock('@crowd/data-access-layer/src/packages', () => ({
   upsertPackagistVersions: vi.fn(),
   getPackagistPackageIdsByNames: vi.fn(),
   reconcileVersionDependencies: vi.fn().mockResolvedValue([]),
+  logAuditFieldChanges: vi.fn(),
 }))
 
 const mockAggregates = vi.mocked(updatePackagistVersionAggregates)
 const mockVersions = vi.mocked(upsertPackagistVersions)
 const mockIds = vi.mocked(getPackagistPackageIdsByNames)
 const mockDeps = vi.mocked(reconcileVersionDependencies)
+const mockAudit = vi.mocked(logAuditFieldChanges)
 
 const qx = {
   tx: vi.fn((cb: (t: QueryExecutor) => Promise<void>) => cb(qx)),
@@ -120,6 +123,13 @@ describe('persistPackagistMetadata', () => {
     expect(result.unresolvedDependencyTargets).toBe(0)
     // dependency-edge changes must reach the audit log alongside aggregates/versions
     expect(result.changedFields).toContain('package_dependencies.depends_on_id')
+    // audited atomically inside the same transaction as the writes above
+    expect(mockAudit).toHaveBeenCalledWith(
+      qx,
+      'packagist',
+      PURL,
+      expect.arrayContaining(['packages.latest_version', 'package_dependencies.depends_on_id']),
+    )
   })
 
   it('skips and counts dependency targets that do not resolve to a packages row', async () => {
@@ -170,6 +180,7 @@ describe('persistPackagistMetadata', () => {
     expect(mockVersions).not.toHaveBeenCalled()
     expect(mockIds).not.toHaveBeenCalled()
     expect(mockDeps).not.toHaveBeenCalled()
+    expect(mockAudit).not.toHaveBeenCalled()
   })
 
   it('handles a dev-branches-only package: aggregates written, no version or dependency writes', async () => {

services/apps/packages_worker/src/packagist/__tests__/persistPackageInfo.test.ts

@@ -2,6 +2,7 @@ import { beforeEach, describe, expect, it, vi } from 'vitest'
 
 import {
   getOrCreateRepoByUrl,
+  logAuditFieldChanges,
   removeDeclaredPackageRepo,
   updatePackagistPackageStats,
   upsertPackageMaintainers,
@@ -18,13 +19,15 @@ vi.mock('@crowd/data-access-layer/src/packages', () => ({
   getOrCreateRepoByUrl: vi.fn(),
   upsertPackageRepo: vi.fn().mockResolvedValue([]),
   removeDeclaredPackageRepo: vi.fn().mockResolvedValue([]),
+  logAuditFieldChanges: vi.fn(),
 }))
 
 const mockUpdate = vi.mocked(updatePackagistPackageStats)
 const mockMaintainers = vi.mocked(upsertPackageMaintainers)
 const mockRepoGet = vi.mocked(getOrCreateRepoByUrl)
 const mockRepoLink = vi.mocked(upsertPackageRepo)
 const mockRepoRemove = vi.mocked(removeDeclaredPackageRepo)
+const mockAudit = vi.mocked(logAuditFieldChanges)
 
 const qx = {
   tx: vi.fn((cb: (t: QueryExecutor) => Promise<void>) => cb(qx)),
@@ -82,6 +85,13 @@ describe('persistPackagistPackageInfo', () => {
     expect(result.changedFields).toContain('packages.description')
     // maintainer changes must reach the audit log too, matching the npm/pypi paths
     expect(result.changedFields).toContain('maintainers.display_name')
+    // audited atomically inside the same transaction as the writes above
+    expect(mockAudit).toHaveBeenCalledWith(
+      qx,
+      'packagist',
+      PURL,
+      expect.arrayContaining(['packages.description', 'maintainers.display_name']),
+    )
   })
 
   it('skips maintainers for a non-critical package but still links the repo', async () => {
@@ -187,6 +197,7 @@ describe('persistPackagistPackageInfo', () => {
     expect(result).toEqual({ found: false, changedFields: [] })
     expect(mockRepoGet).not.toHaveBeenCalled()
     expect(mockMaintainers).not.toHaveBeenCalled()
+    expect(mockAudit).not.toHaveBeenCalled()
   })
 
   it('strips NUL bytes from the description before writing (Postgres rejects them)', async () => {

services/apps/packages_worker/src/packagist/activities.ts

@@ -147,9 +147,10 @@ export async function ingestOnePackagistMetadata(
   }
 
   const stats = normalizePackagistStats(info.value.package)
-  const persistedInfo = await persistPackagistPackageInfo(qx, candidate.purl, stats)
-  // Audit phase 1 immediately: those writes are already committed
-  await logAuditFieldChanges(qx, WORKER, candidate.purl, persistedInfo.changedFields)
+  // persistPackagistPackageInfo audits its own writes atomically, inside the same
+  // transaction — phase 1 is committed-and-audited before the p2 fetch (which can
+  // throw) ever runs.
+  await persistPackagistPackageInfo(qx, candidate.purl, stats)
 
   // Phase 2: p2 endpoint
   const p2 = await fetchWithFastRetry(
@@ -172,21 +173,20 @@ export async function ingestOnePackagistMetadata(
   }
 
   let lastModified: string | null = null
-  let phase2ChangedFields: string[] = []
   if (!isP2NotModified(p2.value)) {
     const expanded = expandComposerMetadata(p2.value.minifiedVersions)
+    // persistPackagistMetadata audits its own writes atomically, inside the same
+    // transaction as the aggregate/version/dependency writes.
     const persistResult = await persistPackagistMetadata(qx, candidate.purl, expanded)
     if (persistResult.unresolvedDependencyTargets > 0) {
       log.debug(
         { purl: candidate.purl, unresolved: persistResult.unresolvedDependencyTargets },
         'packagist dependency targets not found in packages — edges skipped',
       )
     }
-    phase2ChangedFields = persistResult.changedFields
     lastModified = p2.value.lastModified
   }
 
-  await logAuditFieldChanges(qx, WORKER, candidate.purl, phase2ChangedFields)
   await markPackagistMetadataScanned(
     qx,
     candidate.purl,
@@ -219,13 +219,9 @@ export async function ingestOnePackagist30dWindow(
     return
   }
 
-  const changedFields = await persistPackagist30dWindow(
-    qx,
-    purl,
-    info.value.package.downloads?.monthly ?? null,
-    runDate,
-  )
-  await logAuditFieldChanges(qx, WORKER, purl, changedFields)
+  // persistPackagist30dWindow audits its own write atomically, inside the same
+  // transaction as the insert-if-absent.
+  await persistPackagist30dWindow(qx, purl, info.value.package.downloads?.monthly ?? null, runDate)
   await markPackagist30dProcessed(qx, purl, { status: 'success', attempts: info.attempts })
 }
 
@@ -258,10 +254,15 @@ export async function ingestOnePackagistDailyDownload(
 
   const daily = info.value.package.downloads?.daily
   if (typeof daily === 'number') {
-    const changedFields = await insertDailyDownloads(qx, candidate.packageId, [
-      { day: runDate, downloads: daily },
-    ])
-    await logAuditFieldChanges(qx, WORKER, candidate.purl, changedFields)
+    // Insert + audit share one transaction so a failed audit insert can never leave a
+    // committed row unaudited — a retry would hit ON CONFLICT DO NOTHING and report no
+    // changes, permanently losing the audit event otherwise.
+    await qx.tx(async (t) => {
+      const changedFields = await insertDailyDownloads(t, candidate.packageId, [
+        { day: runDate, downloads: daily },
+      ])
+      await logAuditFieldChanges(t, WORKER, candidate.purl, changedFields)
+    })
   }
   await markPackagistDailyProcessed(qx, candidate.purl, {
     status: 'success',

services/apps/packages_worker/src/packagist/downloads.ts

@@ -1,6 +1,11 @@
-import { insertLast30dDownloadIfAbsent } from '@crowd/data-access-layer/src/packages'
+import {
+  insertLast30dDownloadIfAbsent,
+  logAuditFieldChanges,
+} from '@crowd/data-access-layer/src/packages'
 import type { QueryExecutor } from '@crowd/data-access-layer/src/queryExecutor'
 
+const WORKER = 'packagist'
+
 // Compute the monthly window for observed rolling 30d downloads.
 // endDate = first of the run month; startDate = endDate minus 30 days.
 export function monthlyWindowFor(runDate: string): { startDate: string; endDate: string } {
@@ -21,8 +26,8 @@ export function monthlyWindowFor(runDate: string): { startDate: string; endDate:
 // the value is the observation closest to the boundary. insertLast30dDownloadIfAbsent
 // does the presence check and the insert (+ mirror) atomically in one statement, so
 // concurrent runs for the same purl+month can't both pass a check-then-insert gap and
-// have the later one clobber the earlier one's count. Returns the changed fields (like
-// the sibling persist*/upsert* functions) — the caller audits them.
+// have the later one clobber the earlier one's count. The audit record shares the same
+// transaction, so a failed audit insert can never leave a committed window unaudited.
 export async function persistPackagist30dWindow(
   qx: QueryExecutor,
   purl: string,
@@ -32,5 +37,16 @@ export async function persistPackagist30dWindow(
   if (monthly == null) return []
 
   const { startDate, endDate } = monthlyWindowFor(runDate)
-  return insertLast30dDownloadIfAbsent(qx, purl, startDate, endDate, monthly, true)
+  return qx.tx(async (t) => {
+    const changedFields = await insertLast30dDownloadIfAbsent(
+      t,
+      purl,
+      startDate,
+      endDate,
+      monthly,
+      true,
+    )
+    await logAuditFieldChanges(t, WORKER, purl, changedFields)
+    return changedFields
+  })
 }