feat: pypi downloads ingest

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

Author: epipav

services/apps/packages_worker/src/bin/bq-dataset-ingest.ts

@@ -1,10 +1,16 @@
 import { scheduleOsspckgsBootstrap } from '../deps-dev/schedules/bootstrap'
+import {
+  schedulePypiDownloads30d,
+  schedulePypiDownloadsDaily,
+} from '../deps-dev/schedules/pypiDownloads'
 import { scheduleOsspckgsCleanup } from '../schedules/cleanup'
 import { svc } from '../service'
 
 setImmediate(async () => {
   await svc.init()
   await scheduleOsspckgsBootstrap()
   await scheduleOsspckgsCleanup()
+  await schedulePypiDownloads30d()
+  await schedulePypiDownloadsDaily()
   await svc.start()
 })

services/apps/packages_worker/src/deps-dev/README.md

@@ -58,6 +58,8 @@ The mode-specific key takes precedence over the generic key. Value must be a pos
 | `BQ_DATASET_INGEST_DEPENDENT_COUNTS_MAX_BQ_GB`     |         2000 | `dependent_counts`     |                                                           |
 | `BQ_DATASET_INGEST_SCORECARD_REPOS_MAX_BQ_GB`      |           50 | `scorecard_repos`      |                                                           |
 | `BQ_DATASET_INGEST_SCORECARD_CHECKS_MAX_BQ_GB`     |          500 | `scorecard_checks`     |                                                           |
+| `BQ_DATASET_INGEST_PYPI_DOWNLOADS_30D_MAX_BQ_GB`   |         6000 | `pypi_downloads_30d`   | Per 30-day window scan (~4.56 TB measured; set in `ingestPypiDownloads.ts`) |
+| `BQ_DATASET_INGEST_PYPI_DOWNLOADS_DAILY_MAX_BQ_GB` |         2000 | `pypi_downloads_daily` | Scales with backfill range; raise for long backfills      |
 
 The override logic lives in `src/deps-dev/activities/bqExportToGcs.ts`.
 

services/apps/packages_worker/src/deps-dev/activities/getCriticalPypiCount.ts

@@ -0,0 +1,12 @@
+import { getCriticalPypiPackageCount } from '@crowd/data-access-layer'
+
+import { getPackagesDb } from '../../db'
+
+// Count of critical PyPI packages, so the daily downloads workflow can skip its BigQuery scan
+// when there are none (the merge is scoped to is_critical, mirroring how deps.dev scopes to our
+// packages in the Postgres merge rather than pushing our package list into BigQuery).
+export async function getCriticalPypiCount(): Promise<{ count: number }> {
+  const qx = await getPackagesDb()
+  const count = await getCriticalPypiPackageCount(qx)
+  return { count }
+}

services/apps/packages_worker/src/deps-dev/activities/index.ts

@@ -1,4 +1,5 @@
 export * from './bqExportToGcs'
+export * from './getCriticalPypiCount'
 export * from './setJobStep'
 export * from './createVersionsLookup'
 export * from './managePackageDepsConstraints'

services/apps/packages_worker/src/deps-dev/queries/__tests__/pypiDownloads.test.ts

@@ -0,0 +1,179 @@
+import { describe, expect, it } from 'vitest'
+
+import {
+  PYPI_EARLIEST,
+  computeLast30dWindows,
+  defaultDailyRange,
+  utcFirstOfCurrentMonth,
+} from '../pypiDownloadsDates'
+import {
+  PYPI_DOWNLOADS_30D_KIND,
+  PYPI_DOWNLOADS_DAILY_KIND,
+  buildPypiDownloads30dMergeSql,
+  buildPypiDownloads30dSql,
+  buildPypiDownloadsDailyMergeSql,
+  buildPypiDownloadsDailySql,
+} from '../pypiDownloadsSql'
+
+// Note: PEP 503 name normalization is done in SQL (BQ `REGEXP_REPLACE(LOWER(file.project), …)`
+// and the PG merge's `REGEXP_REPLACE(LOWER(p.name), …, 'g')`), asserted in the builder tests
+// below — there is no TS-side normalizer (we don't push our package list into BigQuery).
+
+// ── Criterion 2: monthly 30-day windows (npm-identical math) ───────────────────────────────
+describe('computeLast30dWindows', () => {
+  it('PYPI_EARLIEST is the 2018-07-26 Linehaul floor', () => {
+    expect(PYPI_EARLIEST).toBe('2018-07-26')
+  })
+
+  it('with no fromDate returns only the latest bucket ending at upperEndDate', () => {
+    const w = computeLast30dWindows(null, '2026-06-01')
+    expect(w).toEqual([{ start: '2026-05-02', end: '2026-06-01', isLatest: true }])
+  })
+
+  it('enumerates a contiguous monthly series up to and including the latest', () => {
+    const w = computeLast30dWindows('2026-04-10', '2026-06-01')
+    expect(w.map((x) => x.end)).toEqual(['2026-04-01', '2026-05-01', '2026-06-01'])
+    // end_date - 30 days for each bucket
+    expect(w.map((x) => x.start)).toEqual(['2026-03-02', '2026-04-01', '2026-05-02'])
+    // exactly one latest, and it is the final bucket
+    expect(w.filter((x) => x.isLatest)).toEqual([
+      { start: '2026-05-02', end: '2026-06-01', isLatest: true },
+    ])
+  })
+
+  it('clamps the lower bound and start_date to PYPI_EARLIEST', () => {
+    const w = computeLast30dWindows('2015-01-01', '2018-09-01')
+    // 2018-07-01 bucket is skipped (its end precedes PYPI_EARLIEST)
+    expect(w.map((x) => x.end)).toEqual(['2018-08-01', '2018-09-01'])
+    // first bucket's start (would be 2018-07-02) is clamped up to the floor
+    expect(w[0]).toEqual({ start: '2018-07-26', end: '2018-08-01', isLatest: false })
+    expect(w[1].isLatest).toBe(true)
+  })
+
+  it('returns an empty array when fromDate is after upperEndDate', () => {
+    expect(computeLast30dWindows('2026-07-01', '2026-06-01')).toEqual([])
+  })
+})
+
+// ── Criterion 3: daily trailing window + first-of-month helper ─────────────────────────────
+describe('defaultDailyRange', () => {
+  it('defaults to a 2-day self-healing window [today-2, today-1]', () => {
+    expect(defaultDailyRange('2026-06-30')).toEqual({
+      startDate: '2026-06-28',
+      endDate: '2026-06-29',
+    })
+  })
+
+  it('crosses month boundaries correctly', () => {
+    expect(defaultDailyRange('2026-03-01')).toEqual({
+      startDate: '2026-02-27',
+      endDate: '2026-02-28',
+    })
+  })
+})
+
+describe('utcFirstOfCurrentMonth', () => {
+  it('returns the 1st of the month containing the reference date', () => {
+    expect(utcFirstOfCurrentMonth('2026-06-30')).toBe('2026-06-01')
+    expect(utcFirstOfCurrentMonth('2026-01-15')).toBe('2026-01-01')
+  })
+})
+
+// ── Criterion 4: 30d BigQuery aggregate query ─────────────────────────────────────────────
+describe('buildPypiDownloads30dSql', () => {
+  const sql = buildPypiDownloads30dSql({ startDate: '2026-05-02', endDate: '2026-06-01' })
+
+  it('reads from the public file_downloads table', () => {
+    expect(sql).toContain('bigquery-public-data.pypi.file_downloads')
+  })
+
+  it('filters the date range on the timestamp partition', () => {
+    expect(sql).toContain('DATE(timestamp)')
+    expect(sql).toContain('BETWEEN')
+    expect(sql).toContain('2026-05-02')
+    expect(sql).toContain('2026-06-01')
+  })
+
+  it('excludes bandersnatch mirror traffic while keeping NULL installers', () => {
+    expect(sql).toContain("COALESCE(details.installer.name, '') <> 'bandersnatch'")
+  })
+
+  it('groups by the PEP 503-normalized project and counts downloads', () => {
+    expect(sql).toContain("REGEXP_REPLACE(LOWER(file.project), r'[-_.]+', '-')")
+    expect(sql).toMatch(/COUNT\(\*\)\s+AS\s+downloads/i)
+    expect(sql).toContain('GROUP BY')
+  })
+})
+
+// ── Criterion 5: daily BigQuery aggregate query ───────────────────────────────────────────
+describe('buildPypiDownloadsDailySql', () => {
+  const sql = buildPypiDownloadsDailySql({ startDate: '2026-06-01', endDate: '2026-06-03' })
+
+  it('aggregates per project per day', () => {
+    expect(sql).toContain('bigquery-public-data.pypi.file_downloads')
+    expect(sql).toContain('DATE(timestamp) AS day')
+    expect(sql).toContain("COALESCE(details.installer.name, '') <> 'bandersnatch'")
+    expect(sql).toContain("REGEXP_REPLACE(LOWER(file.project), r'[-_.]+', '-')")
+    expect(sql).toMatch(/COUNT\(\*\)\s+AS\s+downloads/i)
+    expect(sql).toContain('GROUP BY project, day')
+  })
+
+  it('never pushes our package list into BigQuery — scoping to is_critical is the merge job', () => {
+    expect(sql).not.toContain('UNNEST')
+    expect(sql).not.toContain('is_critical')
+  })
+})
+
+// ── Criterion 6: merge SQL builders ───────────────────────────────────────────────────────
+describe('buildPypiDownloads30dMergeSql', () => {
+  it('emits only the insert when not mirroring', () => {
+    const stmts = buildPypiDownloads30dMergeSql({
+      startDate: '2026-05-02',
+      endDate: '2026-06-01',
+      mirrorToPackages: false,
+    })
+    expect(stmts).toHaveLength(1)
+    const insert = stmts[0]
+    expect(insert).toContain('INSERT INTO downloads_last_30d')
+    expect(insert).toContain('staging.pypi_downloads_30d_raw')
+    expect(insert).toContain("p.ecosystem = 'pypi'")
+    expect(insert).toContain('2026-05-02')
+    expect(insert).toContain('2026-06-01')
+    expect(insert).toContain('ON CONFLICT (purl, end_date) DO UPDATE')
+    // PG-side normalization must use the 'g' flag to replace every separator run
+    expect(insert).toContain("REGEXP_REPLACE(LOWER(p.name), '[-_.]+', '-', 'g')")
+  })
+
+  it('appends a packages mirror update when mirroring the latest window', () => {
+    const stmts = buildPypiDownloads30dMergeSql({
+      startDate: '2026-05-02',
+      endDate: '2026-06-01',
+      mirrorToPackages: true,
+    })
+    expect(stmts).toHaveLength(2)
+    expect(stmts[1]).toContain('UPDATE packages')
+    expect(stmts[1]).toContain('downloads_last_30d')
+    expect(stmts[1]).toContain('IS DISTINCT FROM')
+  })
+})
+
+describe('buildPypiDownloadsDailyMergeSql', () => {
+  it('inserts into downloads_daily scoped to critical pypi packages', () => {
+    const sql = buildPypiDownloadsDailyMergeSql()
+    expect(sql).toContain('INSERT INTO downloads_daily')
+    expect(sql).toContain('package_id')
+    expect(sql).toContain('staging.pypi_downloads_daily_raw')
+    expect(sql).toContain("p.ecosystem = 'pypi'")
+    expect(sql).toContain('p.is_critical')
+    expect(sql).toContain('ON CONFLICT (package_id, date) DO UPDATE')
+    expect(sql).toContain("REGEXP_REPLACE(LOWER(p.name), '[-_.]+', '-', 'g')")
+  })
+})
+
+// ── Criterion 7: job-kind registry ────────────────────────────────────────────────────────
+describe('PyPI downloads job kinds', () => {
+  it('exposes the two new kind identifiers', () => {
+    expect(PYPI_DOWNLOADS_30D_KIND).toBe('pypi_downloads_30d')
+    expect(PYPI_DOWNLOADS_DAILY_KIND).toBe('pypi_downloads_daily')
+  })
+})

services/apps/packages_worker/src/deps-dev/queries/pypiDownloadsDates.ts

@@ -0,0 +1,64 @@
+// Date math for the PyPI downloads workflows. Pure functions — no clock access; the
+// caller passes the reference "today" so runs are deterministic and testable.
+
+// PyPI's BigQuery download data is unreliable before this date: Linehaul under-reported
+// downloads prior to 2018-07-26 (see the official PyPI download-analysis guide). Backfills
+// are clamped to this floor so we never ingest known-bad counts.
+export const PYPI_EARLIEST = '2018-07-26'
+
+export interface Last30dWindow {
+  start: string
+  end: string
+  isLatest: boolean
+}
+
+function addDaysUTC(date: string, days: number): string {
+  const d = new Date(date + 'T00:00:00Z')
+  d.setUTCDate(d.getUTCDate() + days)
+  return d.toISOString().slice(0, 10)
+}
+
+// 1st of the calendar month (UTC) containing `today` (YYYY-MM-DD).
+export function utcFirstOfCurrentMonth(today: string): string {
+  const d = new Date(today + 'T00:00:00Z')
+  return new Date(Date.UTC(d.getUTCFullYear(), d.getUTCMonth(), 1)).toISOString().slice(0, 10)
+}
+
+// Monthly 30-day windows, identical in math to npm's computeMissingLast30dWindows:
+// end_date = 1st of a calendar month (UTC), start_date = end_date - 30 days, isLatest only
+// for the bucket whose end === upperEndDate. Walks from max(fromDate, PYPI_EARLIEST) up to
+// upperEndDate. No fromDate → only the latest bucket.
+export function computeLast30dWindows(
+  fromDate: string | null,
+  upperEndDate: string,
+): Last30dWindow[] {
+  const lower = fromDate ? (fromDate > PYPI_EARLIEST ? fromDate : PYPI_EARLIEST) : upperEndDate
+  const lowerDate = new Date(lower + 'T00:00:00Z')
+  const firstMonth = Date.UTC(lowerDate.getUTCFullYear(), lowerDate.getUTCMonth(), 1)
+  const lastMonth = new Date(upperEndDate + 'T00:00:00Z').getTime()
+  if (firstMonth > lastMonth) return []
+
+  const result: Last30dWindow[] = []
+  let m = firstMonth
+  while (m <= lastMonth) {
+    const d = new Date(m)
+    const endDate = d.toISOString().slice(0, 10) // 1st of the month
+    // Skip windows whose end precedes the Linehaul floor.
+    if (endDate >= PYPI_EARLIEST) {
+      let start = addDaysUTC(endDate, -30)
+      if (start < PYPI_EARLIEST) start = PYPI_EARLIEST
+      result.push({ start, end: endDate, isLatest: m === lastMonth })
+    }
+    m = Date.UTC(d.getUTCFullYear(), d.getUTCMonth() + 1, 1)
+  }
+  return result
+}
+
+// The daily trailing window spans 2 days so the most-recent (possibly partial) partition is
+// re-scanned once on the next run and corrected, while keeping the daily BigQuery scan small.
+const DAILY_TRAILING_DAYS = 2
+
+// Default daily trailing window: [today - 2, today - 1] inclusive (self-healing).
+export function defaultDailyRange(today: string): { startDate: string; endDate: string } {
+  return { startDate: addDaysUTC(today, -DAILY_TRAILING_DAYS), endDate: addDaysUTC(today, -1) }
+}

services/apps/packages_worker/src/deps-dev/queries/pypiDownloadsSql.ts

@@ -0,0 +1,105 @@
+export const PYPI_DOWNLOADS_30D_KIND = 'pypi_downloads_30d'
+export const PYPI_DOWNLOADS_DAILY_KIND = 'pypi_downloads_daily'
+
+// Staging tables the BQ export lands in before the merge into the final download tables.
+export const PYPI_DOWNLOADS_30D_STAGING = 'staging.pypi_downloads_30d_raw'
+export const PYPI_DOWNLOADS_DAILY_STAGING = 'staging.pypi_downloads_daily_raw'
+
+const BQ_TABLE = '`bigquery-public-data.pypi.file_downloads`'
+// PEP 503 canonical project, BigQuery dialect (REGEXP_REPLACE is global by default).
+const BQ_PROJECT_NORM = "REGEXP_REPLACE(LOWER(file.project), r'[-_.]+', '-')"
+// Keep NULL installers; only drop known mirror traffic.
+const MIRROR_EXCLUDE = "COALESCE(details.installer.name, '') <> 'bandersnatch'"
+// PEP 503 canonical name, Postgres dialect — the 'g' flag is required so EVERY separator
+// run collapses (Postgres REGEXP_REPLACE otherwise replaces only the first match).
+const PG_NAME_NORM = "REGEXP_REPLACE(LOWER(p.name), '[-_.]+', '-', 'g')"
+
+// Aggregate query: net downloads per package over [startDate, endDate], one total each.
+export function buildPypiDownloads30dSql({
+  startDate,
+  endDate,
+}: {
+  startDate: string
+  endDate: string
+}): string {
+  return `
+SELECT
+  ${BQ_PROJECT_NORM} AS project,
+  COUNT(*) AS downloads
+FROM ${BQ_TABLE}
+WHERE DATE(timestamp) BETWEEN DATE('${startDate}') AND DATE('${endDate}')
+  AND ${MIRROR_EXCLUDE}
+GROUP BY project
+`
+}
+
+// Aggregate query: per-day downloads per package over [startDate, endDate]. Like every other BQ
+// job here (see deps.dev), it exports all projects and lets the Postgres merge scope to our
+// packages (is_critical) — we never push our package list into BigQuery.
+export function buildPypiDownloadsDailySql({
+  startDate,
+  endDate,
+}: {
+  startDate: string
+  endDate: string
+}): string {
+  return `
+SELECT
+  ${BQ_PROJECT_NORM} AS project,
+  DATE(timestamp) AS day,
+  COUNT(*) AS downloads
+FROM ${BQ_TABLE}
+WHERE DATE(timestamp) BETWEEN DATE('${startDate}') AND DATE('${endDate}')
+  AND ${MIRROR_EXCLUDE}
+GROUP BY project, day
+`
+}
+
+// Merge statements for the 30d window: insert into downloads_last_30d, and (only for the
+// latest window) mirror the count onto packages.downloads_last_30d.
+export function buildPypiDownloads30dMergeSql({
+  startDate,
+  endDate,
+  mirrorToPackages,
+}: {
+  startDate: string
+  endDate: string
+  mirrorToPackages: boolean
+}): string[] {
+  const insert = `
+INSERT INTO downloads_last_30d (purl, start_date, end_date, count, created_at, updated_at)
+SELECT p.purl, DATE '${startDate}', DATE '${endDate}', s.downloads, NOW(), NOW()
+FROM ${PYPI_DOWNLOADS_30D_STAGING} s
+JOIN packages p ON p.ecosystem = 'pypi'
+  AND ${PG_NAME_NORM} = s.project
+ON CONFLICT (purl, end_date) DO UPDATE SET
+  count = EXCLUDED.count,
+  start_date = EXCLUDED.start_date,
+  updated_at = NOW()
+`
+  if (!mirrorToPackages) return [insert]
+
+  const mirror = `
+UPDATE packages p
+SET downloads_last_30d = s.downloads
+FROM ${PYPI_DOWNLOADS_30D_STAGING} s
+WHERE p.ecosystem = 'pypi'
+  AND ${PG_NAME_NORM} = s.project
+  AND p.downloads_last_30d IS DISTINCT FROM s.downloads
+`
+  return [insert, mirror]
+}
+
+// Merge statement for daily downloads into downloads_daily, scoped to critical pypi packages.
+export function buildPypiDownloadsDailyMergeSql(): string {
+  return `
+INSERT INTO downloads_daily (package_id, date, count, created_at, updated_at)
+SELECT p.id, s.day, s.downloads, NOW(), NOW()
+FROM ${PYPI_DOWNLOADS_DAILY_STAGING} s
+JOIN packages p ON p.ecosystem = 'pypi' AND p.is_critical
+  AND ${PG_NAME_NORM} = s.project
+ON CONFLICT (package_id, date) DO UPDATE SET
+  count = EXCLUDED.count,
+  updated_at = NOW()
+`
+}