feat: deps.dev package import (#4158)

Signed-off-by: Uroš Marolt <uros@marolt.me>

Author: themarolt

backend/.env.dist.local

@@ -170,6 +170,7 @@ CROWD_TINYBIRD_BASE_URL=http://localhost:7181/
 # Auth0
 CROWD_AUTH0_ISSUER_BASE_URLS=
 CROWD_AUTH0_AUDIENCE=
+
 # packages DB (osspckgs)
 CROWD_PACKAGES_DB_READ_HOST=localhost
 CROWD_PACKAGES_DB_WRITE_HOST=localhost
@@ -184,6 +185,10 @@ ENRICHER_BATCH_SIZE=100
 ENRICHER_REPO_UPDATE_INTERVAL_HOURS=24
 ENRICHER_IDLE_SLEEP_SEC=60
 
+OSSPCKGS_GCP_PROJECT=
+OSSPCKGS_GCS_BUCKET=
+OSSPCKGS_GCP_CREDENTIALS_B64=
+
 # osv-sync (Temporal-scheduled; see services/apps/packages_worker/src/osv/schedule.ts)
 # OSV_ECOSYSTEMS uses OSV's canonical bucket case (npm lowercase, Maven titlecase) because
 # the bucket URL <BASE>/<ecosystem>/all.zip is case-sensitive (Maven/all.zip exists,

backend/src/osspckgs/migrations/V1779710880__initial_schema.sql

@@ -1,3 +1,6 @@
+-- Staging schema — used by gcsParquetToStaging activity for temporary unlogged tables.
+CREATE SCHEMA IF NOT EXISTS staging;
+
 -- ============================================================
 -- DOMAIN 1: UNIVERSE (Tier 3 → Tier 2 ranking input)
 -- ============================================================
@@ -130,7 +133,12 @@ CREATE TABLE versions (
     -- Nullable for same reason: yanked status comes from registry-specific workers, not deps.dev.
     is_yanked bool,
     is_prerelease bool NOT NULL DEFAULT FALSE,
-    license text, -- SPDX where available; can differ per version
+    -- Denormalized from packages for fast deps merge resolution.
+    -- Allows resolving (ecosystem, namespace, name, number) → version_id in one index lookup.
+    namespace text,
+    name      text NOT NULL,
+    licenses text[], -- SPDX array, deterministically sorted; can differ per version
+    download_count bigint, -- per-version where available (npm, crates)
     last_synced_at timestamptz NOT NULL DEFAULT NOW(),
     PRIMARY KEY (id, package_id),
     UNIQUE (package_id, number)
@@ -239,6 +247,8 @@ CREATE INDEX ON versions (package_id)
 WHERE
     is_latest;
 
+CREATE INDEX ON versions (ecosystem, COALESCE(namespace, ''), name, number);
+
 -- ============================================================
 -- PACKAGE DEPENDENCIES — PARTITION BY HASH(depends_on_id)
 --
@@ -658,7 +668,8 @@ CREATE UNIQUE INDEX ON advisory_affected_ranges (
     COALESCE(last_affected, '')
 );
 
-CREATE INDEX ON advisory_affected_ranges (advisory_package_id);
+-- advisory_package_id prefix lookups are served by the UNIQUE index on
+-- (advisory_package_id, introduced_version, fixed_version) — no separate index needed.
 
 -- ============================================================
 -- MAINTAINERS
@@ -844,8 +855,9 @@ BEGIN
     -- ranking effectively reduces to:
     --   LN(1 + dependent_repos_count)     * weight_dependent_repos
     -- + LN(1 + dependent_packages_count)  * weight_dependent_packages
-    UPDATE packages_universe SET last_rank_pass_at = NOW();
-
+    --
+    -- last_rank_pass_at is set at INSERT time in rankPackagesUniverse activity (TRUNCATE + INSERT
+    -- before each call), so no separate full-table UPDATE needed here.
     WITH new_scores AS (
         SELECT
             id,
@@ -910,3 +922,56 @@ BEGIN
     RETURN QUERY SELECT n_scored, n_ranked, n_propagated;
 END;
 $$;
+
+-- ============================================================
+-- INGEST JOB TRACKING
+-- Tracks each BQ → GCS → Postgres ingest run per job_kind.
+-- snapshot_at = SnapshotAt date used as watermark for incremental diff.
+-- ============================================================
+CREATE TABLE osspckgs_ingest_jobs (
+    id              bigserial PRIMARY KEY,
+    job_kind        text NOT NULL CHECK (job_kind IN (
+                        'packages', 'versions', 'package_dependencies',
+                        'repos', 'package_repos',
+                        'advisories', 'advisory_packages',
+                        'dependent_counts'
+                    )),
+    status          text NOT NULL CHECK (status IN (
+                        'pending', 'exporting', 'exported',
+                        'loading', 'merging', 'done', 'failed', 'cleaned'
+                    )),
+    sync_mode       text NOT NULL DEFAULT 'incremental'
+                        CHECK (sync_mode IN ('full', 'incremental')),
+    snapshot_at             date,  -- committed watermark: promoted from provisional unconditionally on 'done' (including 0-row quiet windows)
+    provisional_snapshot_at date,  -- set at job creation; promoted to snapshot_at when job reaches 'done'
+    gcs_prefix      text,            -- gs://bucket/packages/2026-05-26T00-00-00Z/
+    row_count_bq    bigint,
+    row_count_staging bigint,        -- rows loaded into staging table from GCS parquet files
+    row_count_pg    bigint,          -- total rows inserted into final table(s) after merge
+    table_row_counts  jsonb,         -- per-table inserted row counts, e.g. {"packages": 5000000}
+    bq_bytes_billed bigint,          -- totalBytesProcessed from BQ (cost metric, not GCS export size)
+    bq_job_id       text,            -- GCP BigQuery job ID (project:location.jobId)
+    bq_stats        jsonb,           -- full BQ job statistics: bytesProcessed, bytesBilled, slotMs, cacheHit, etc.
+    bq_cost_usd     numeric(12, 8) GENERATED ALWAYS AS (
+                        ROUND(COALESCE(bq_bytes_billed, 0)::numeric / 1000000000000.0 * 5.0, 8)
+                    ) STORED,        -- estimated BQ cost at $5/TB on-demand pricing
+    export_name     text,            -- named export group (e.g. "cargo-may-2026") for --export-name bootstrap
+    error_message   text,
+    started_at      timestamptz NOT NULL DEFAULT NOW(),
+    finished_at     timestamptz,
+    cleaned_at      timestamptz
+);
+
+CREATE INDEX ON osspckgs_ingest_jobs (job_kind, started_at DESC);
+
+CREATE INDEX ON osspckgs_ingest_jobs (status)
+WHERE status NOT IN ('done', 'cleaned');
+
+CREATE INDEX ON osspckgs_ingest_jobs (job_kind, snapshot_at DESC)
+WHERE status = 'done';
+
+CREATE INDEX ON osspckgs_ingest_jobs (bq_job_id)
+WHERE bq_job_id IS NOT NULL;
+
+CREATE INDEX ON osspckgs_ingest_jobs (job_kind, export_name)
+WHERE export_name IS NOT NULL;