fix: make packagist 30d window insert atomic

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

Author: epipav

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

@@ -1,27 +1,21 @@
 import { beforeEach, describe, expect, it, vi } from 'vitest'
 
-import {
-  getExistingLast30dEndDates,
-  upsertLast30dDownload,
-} from '@crowd/data-access-layer/src/packages'
+import { insertLast30dDownloadIfAbsent } 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', () => ({
-  getExistingLast30dEndDates: vi.fn().mockResolvedValue([]),
-  upsertLast30dDownload: vi.fn().mockResolvedValue([]),
+  insertLast30dDownloadIfAbsent: vi.fn().mockResolvedValue([]),
 }))
 
-const mockExistingWindows = vi.mocked(getExistingLast30dEndDates)
-const mockWindow = vi.mocked(upsertLast30dDownload)
+const mockWindow = vi.mocked(insertLast30dDownloadIfAbsent)
 
 const qx = {} as QueryExecutor
 const PURL = 'pkg:composer/monolog/monolog'
 
 beforeEach(() => {
   vi.clearAllMocks()
-  mockExistingWindows.mockResolvedValue([])
 })
 
 // Observed rolling window anchored on the 1st of the run month,
@@ -49,30 +43,31 @@ describe('monthlyWindowFor', () => {
 // The monthly downloads-30d lane: one window row per purl per month, mirrored to
 // packages.downloads_last_30d (npm parity). A window already recorded for the month
 // 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.
 describe('persistPackagist30dWindow', () => {
   it('writes the window with the mirror 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(mockExistingWindows).toHaveBeenCalledWith(qx, PURL, '2026-07-01', '2026-07-01')
     expect(mockWindow).toHaveBeenCalledWith(qx, PURL, '2026-06-01', '2026-07-01', 300, true)
     expect(changedFields).toEqual(['downloads_last_30d.count', 'packages.downloads_last_30d'])
   })
 
-  it('does not overwrite an existing window for the month, and returns no changed fields', async () => {
-    mockExistingWindows.mockResolvedValue(['2026-07-01'])
+  it('returns no changed fields when a window for the month already exists', async () => {
+    mockWindow.mockResolvedValue([])
 
     const changedFields = await persistPackagist30dWindow(qx, PURL, 300, '2026-07-15')
 
-    expect(mockWindow).not.toHaveBeenCalled()
+    expect(mockWindow).toHaveBeenCalledWith(qx, PURL, '2026-06-01', '2026-07-01', 300, true)
     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(mockExistingWindows).not.toHaveBeenCalled()
     expect(mockWindow).not.toHaveBeenCalled()
     expect(changedFields).toEqual([])
   })

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

@@ -1,7 +1,4 @@
-import {
-  getExistingLast30dEndDates,
-  upsertLast30dDownload,
-} from '@crowd/data-access-layer/src/packages'
+import { insertLast30dDownloadIfAbsent } from '@crowd/data-access-layer/src/packages'
 import type { QueryExecutor } from '@crowd/data-access-layer/src/queryExecutor'
 
 // Compute the monthly window for observed rolling 30d downloads.
@@ -21,8 +18,11 @@ export function monthlyWindowFor(runDate: string): { startDate: string; endDate:
 
 // One window row per purl per month, mirrored to packages.downloads_last_30d
 // (npm parity). A window already recorded for the month is never overwritten —
-// the value is the observation closest to the boundary. Returns the changed
-// fields (like the sibling persist*/upsert* functions) — the caller audits them.
+// 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.
 export async function persistPackagist30dWindow(
   qx: QueryExecutor,
   purl: string,
@@ -32,9 +32,5 @@ export async function persistPackagist30dWindow(
   if (monthly == null) return []
 
   const { startDate, endDate } = monthlyWindowFor(runDate)
-  const existing = await getExistingLast30dEndDates(qx, purl, endDate, endDate)
-  if (existing.length === 0) {
-    return upsertLast30dDownload(qx, purl, startDate, endDate, monthly, true)
-  }
-  return []
+  return insertLast30dDownloadIfAbsent(qx, purl, startDate, endDate, monthly, true)
 }

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

@@ -133,6 +133,48 @@ export async function markLast30dHistoryBackfilled(
   )
 }
 
+// Atomic "first observation wins" insert: the window row and its packages mirror are
+// written by one statement via ON CONFLICT (purl, end_date) DO NOTHING, so concurrent
+// callers for the same purl+month (overlapping Temporal retries, or two lanes racing)
+// can never both pass a check-then-insert gap and have the later one clobber the
+// earlier one's count — only the writer that actually inserts the row also mirrors.
+export async function insertLast30dDownloadIfAbsent(
+  qx: QueryExecutor,
+  purl: string,
+  startDate: string,
+  endDate: string,
+  count: number,
+  mirrorToPackages: boolean,
+): Promise<string[]> {
+  const row: { inserted: boolean; mirrored: boolean } = await qx.selectOne(
+    `WITH ins AS (
+       INSERT INTO downloads_last_30d (purl, start_date, end_date, count, created_at, updated_at)
+       VALUES ($(purl), $(startDate)::date, $(endDate)::date, $(count), NOW(), NOW())
+       ON CONFLICT (purl, end_date) DO NOTHING
+       RETURNING purl, count
+     ),
+     pkg AS (
+       UPDATE packages p
+          SET downloads_last_30d = ins.count,
+              last_synced_at     = NOW()
+         FROM ins
+        WHERE p.purl = ins.purl
+          AND $(mirrorToPackages)
+          AND p.downloads_last_30d IS DISTINCT FROM ins.count
+        RETURNING p.id
+     )
+     SELECT
+       EXISTS (SELECT 1 FROM ins) AS inserted,
+       EXISTS (SELECT 1 FROM pkg) AS mirrored`,
+    { purl, startDate, endDate, count, mirrorToPackages },
+  )
+
+  if (!row.inserted) return []
+  const changed = ['downloads_last_30d.start_date', 'downloads_last_30d.count']
+  if (row.mirrored) changed.push('packages.downloads_last_30d')
+  return changed
+}
+
 export async function upsertLast30dDownload(
   qx: QueryExecutor,
   purl: string,