Merge branch 'main' into feat/criticality_worker

Signed-off-by: Mouad BANI <mbani@contractor.linuxfoundation.org>

Author: mbani01

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/database/migrations/U1780555634__update-osps-catalog-id.sql

@@ -0,0 +1,3 @@
+UPDATE "securityInsightsEvaluationSuites"
+SET "catalogId" = 'OSPS_B'
+WHERE "catalogId" = 'osps-baseline-2026-02';

backend/src/database/migrations/V1780555634__update-osps-catalog-id.sql

@@ -0,0 +1,3 @@
+UPDATE "securityInsightsEvaluationSuites"
+SET "catalogId" = 'osps-baseline-2026-02'
+WHERE "catalogId" = 'OSPS_B';

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;

backend/src/osspckgs/migrations/V1780394591__packages_universe_graph_signals.sql

@@ -0,0 +1,100 @@
+-- Graph-derived signals for criticality scoring (ADR-0001 §Criticality scoring methodology).
+-- Populated by the criticality worker; NULL until first pass.
+
+ALTER TABLE packages_universe
+  ADD COLUMN IF NOT EXISTS transitive_dependent_count bigint,
+  ADD COLUMN IF NOT EXISTS centrality_score           numeric(10, 8);
+
+-- Split dependent_packages_count into direct (MinimumDepth=1) vs transitive (MinimumDepth>1).
+
+ALTER TABLE packages
+  RENAME COLUMN dependent_packages_count TO dependent_count;
+
+ALTER TABLE packages
+  ADD COLUMN IF NOT EXISTS transitive_dependent_count bigint;
+
+ALTER TABLE packages_universe
+  RENAME COLUMN dependent_packages_count TO dependent_count;
+
+-- The renamed column holds old all-dependent counts (direct + transitive combined).
+-- NULL them out so criticality scoring uses 0 for this signal until dependent_counts
+-- is re-ingested with the split values; inflated scores are worse than missing ones.
+UPDATE packages        SET dependent_count = NULL;
+UPDATE packages_universe SET dependent_count = NULL;
+
+-- rank_packages_universe() stores its body as text — renaming the column above does NOT
+-- update the function. Replace it here so it references dependent_count, not the old name.
+CREATE OR REPLACE FUNCTION rank_packages_universe(
+    weight_downloads            numeric,
+    weight_dependent_repos      numeric,
+    weight_dependent_packages   numeric,
+    log_smoothing               numeric,
+    critical_top_n_by_ecosystem jsonb
+)
+RETURNS TABLE (scored_rows int, ranked_rows int, propagated_rows int)
+LANGUAGE plpgsql
+AS $$
+DECLARE
+    n_scored     int;
+    n_ranked     int;
+    n_propagated int;
+BEGIN
+    WITH new_scores AS (
+        SELECT
+            id,
+            ( LN(log_smoothing + COALESCE(downloads_last_30d, 0))       * weight_downloads
+            + LN(log_smoothing + COALESCE(dependent_repos_count, 0))    * weight_dependent_repos
+            + LN(log_smoothing + COALESCE(dependent_count, 0))          * weight_dependent_packages
+            )::numeric(10, 4) AS new_score
+        FROM packages_universe
+    )
+    UPDATE packages_universe pu
+    SET    criticality_score = ns.new_score
+    FROM   new_scores ns
+    WHERE  pu.id = ns.id
+      AND  pu.criticality_score IS DISTINCT FROM ns.new_score;
+
+    GET DIAGNOSTICS n_scored = ROW_COUNT;
+
+    WITH ranked AS (
+        SELECT
+            id,
+            ecosystem,
+            ROW_NUMBER() OVER (
+                PARTITION BY ecosystem
+                ORDER BY criticality_score DESC NULLS LAST, id
+            ) AS r
+        FROM packages_universe
+        WHERE purl IS NOT NULL
+    ),
+    with_flag AS (
+        SELECT
+            id,
+            r,
+            COALESCE(
+                r <= (critical_top_n_by_ecosystem ->> ecosystem)::int,
+                FALSE
+            ) AS new_is_critical
+        FROM ranked
+    )
+    UPDATE packages_universe pu
+    SET    rank_in_ecosystem = wf.r,
+           is_critical       = wf.new_is_critical
+    FROM   with_flag wf
+    WHERE  pu.id = wf.id
+      AND  ( pu.rank_in_ecosystem IS DISTINCT FROM wf.r
+          OR pu.is_critical       IS DISTINCT FROM wf.new_is_critical );
+
+    GET DIAGNOSTICS n_ranked = ROW_COUNT;
+
+    UPDATE packages p
+    SET    criticality_score = pu.criticality_score
+    FROM   packages_universe pu
+    WHERE  p.purl = pu.purl
+      AND  p.criticality_score IS DISTINCT FROM pu.criticality_score;
+
+    GET DIAGNOSTICS n_propagated = ROW_COUNT;
+
+    RETURN QUERY SELECT n_scored, n_ranked, n_propagated;
+END;
+$$;

backend/src/osspckgs/migrations/V1780496100__add_skip_enrichment_to_repos.sql

@@ -0,0 +1 @@
+ALTER TABLE repos ADD COLUMN IF NOT EXISTS skip_enrichment boolean NOT NULL DEFAULT false;