fix: align packages-db writers with Tinybird ENGINE_VER semantics

joanagmaia · joanagmaia · commit e4bbce7a8d39 · 2026-06-09T14:38:40.000+01:00
Signed-off-by: Joana Maia &lt;jmaia@contractor.linuxfoundation.org&gt;
diff --git a/backend/src/osspckgs/migrations/V1781100000__rank_packages_bumps_last_synced_at.sql b/backend/src/osspckgs/migrations/V1781100000__rank_packages_bumps_last_synced_at.sql
@@ -0,0 +1,97 @@
+-- rank_packages() now bumps last_synced_at on every UPDATE that touches a
+-- DS-exported field (impact, is_critical, last_rank_pass_at). last_synced_at
+-- is the Tinybird ENGINE_VER for the packages datasource; without this bump,
+-- ReplacingMergeTree may keep an older row when criticality changes without
+-- any other write path touching the package row.
+
+CREATE OR REPLACE FUNCTION rank_packages(
+    weight_downloads            numeric DEFAULT 0.25,
+    weight_dependent_packages   numeric DEFAULT 0.25,
+    weight_transitive           numeric DEFAULT 0.50,
+    critical_top_n_by_ecosystem jsonb   DEFAULT '{"npm":400000,"go":100000,"maven":200000,"pypi":100000,"nuget":50000,"cargo":75000}'::jsonb
+)
+RETURNS TABLE(scored_rows int, ranked_rows int)
+LANGUAGE plpgsql AS $$
+DECLARE
+    n_scored int;
+    n_ranked int;
+BEGIN
+    -- Step 1: score
+    WITH percentile_scores AS (
+        SELECT
+            id,
+            (
+              weight_downloads * PERCENT_RANK() OVER (
+                  PARTITION BY ecosystem ORDER BY LOG(1 + COALESCE(downloads_last_30d, 0)))
+
+            + weight_dependent_packages * PERCENT_RANK() OVER (
+                  PARTITION BY ecosystem ORDER BY LOG(1 + COALESCE(dependent_count, 0)))
+
+            + weight_transitive * PERCENT_RANK() OVER (
+                  PARTITION BY ecosystem ORDER BY LOG(1 + COALESCE(transitive_dependent_count, 0)))
+            )::numeric(10, 4) AS new_impact
+        FROM packages
+        WHERE ecosystem IN (SELECT jsonb_object_keys(critical_top_n_by_ecosystem))
+    )
+    UPDATE packages p
+       SET impact         = ps.new_impact,
+           last_synced_at = NOW()
+      FROM percentile_scores ps
+     WHERE p.id = ps.id
+       AND p.impact IS DISTINCT FROM ps.new_impact;
+
+    GET DIAGNOSTICS n_scored = ROW_COUNT;
+
+    -- Step 2: rank + flag
+    WITH ranked AS (
+        SELECT
+            id, ecosystem,
+            ROW_NUMBER() OVER (
+                PARTITION BY ecosystem
+                ORDER BY impact DESC NULLS LAST, id
+            ) AS r
+        FROM packages
+        WHERE purl IS NOT NULL
+          AND ecosystem IN (SELECT jsonb_object_keys(critical_top_n_by_ecosystem))
+    ),
+    flagged AS (
+        SELECT
+            id, r,
+            COALESCE(
+                r <= (critical_top_n_by_ecosystem ->> ecosystem)::int,
+                FALSE
+            ) AS new_is_critical
+        FROM ranked
+    )
+    UPDATE packages p
+       SET rank_in_ecosystem = f.r,
+           is_critical       = f.new_is_critical,
+           last_synced_at    = NOW()
+      FROM flagged f
+     WHERE p.id = f.id
+       AND (
+             p.rank_in_ecosystem IS DISTINCT FROM f.r
+          OR p.is_critical       IS DISTINCT FROM f.new_is_critical
+       );
+
+    GET DIAGNOSTICS n_ranked = ROW_COUNT;
+
+    -- Step 2.5: spotlight overrides
+    UPDATE packages p
+       SET is_critical    = TRUE,
+           last_synced_at = NOW()
+      FROM package_criticality_spotlight s
+     WHERE p.ecosystem                    = s.ecosystem
+       AND (p.namespace IS NOT DISTINCT FROM s.namespace)
+       AND p.name                         = s.name
+       AND p.is_critical                  = FALSE;
+
+    -- Step 3: stamp last_rank_pass_at unconditionally
+    UPDATE packages
+       SET last_rank_pass_at = NOW(),
+           last_synced_at    = NOW()
+     WHERE ecosystem IN (SELECT jsonb_object_keys(critical_top_n_by_ecosystem));
+
+    RETURN QUERY SELECT n_scored, n_ranked;
+END;
+$$;
diff --git a/services/apps/packages_worker/src/deps-dev/workflows/ingestRepos.ts b/services/apps/packages_worker/src/deps-dev/workflows/ingestRepos.ts
@@ -58,11 +58,13 @@ const REPOS_PG_COLUMNS = [
   'open_issues',
 ]
 
+// last_synced_at is intentionally left NULL on seed — it is owned by the GitHub
+// enricher as its freshness signal. created_at / updated_at use their column defaults.
 const REPOS_MERGE_SQL = `
 INSERT INTO repos (url, raw_project_type, raw_project_name, host, owner, name,
-                   description, homepage, stars, forks, open_issues, last_synced_at)
+                   description, homepage, stars, forks, open_issues)
 SELECT s.canonical_url, s.raw_project_type, s.raw_project_name, s.host, s.owner, s.name,
-       s.description, s.homepage, s.stars, s.forks, s.open_issues, NOW()
+       s.description, s.homepage, s.stars, s.forks, s.open_issues
 FROM staging.osspckgs_repos_raw s
 ON CONFLICT (url) DO NOTHING
 `
diff --git a/services/libs/data-access-layer/src/osspckgs/repos.ts b/services/libs/data-access-layer/src/osspckgs/repos.ts
@@ -15,19 +15,20 @@ export async function findRepoIdsByUrl(
  * Inserts or updates a repo row keyed on url.
  * Uses COALESCE so richer data from other enrichers (GitHub, deps.dev) is never
  * overwritten with nulls from a partial write.
+ * `last_synced_at` is intentionally NOT touched here — that column is owned by the
+ * GitHub enricher as its freshness signal. Maven discovery only stamps updated_at.
  * Returns the repo id.
  */
 export async function upsertRepo(qx: QueryExecutor, item: IDbRepoUpsert): Promise<number> {
   const row = await qx.selectOne(
     `
-    INSERT INTO repos (url, host, owner, name, last_synced_at, updated_at)
-    VALUES ($(url), $(host), $(owner), $(name), NOW(), NOW())
+    INSERT INTO repos (url, host, owner, name, updated_at)
+    VALUES ($(url), $(host), $(owner), $(name), NOW())
     ON CONFLICT (url) DO UPDATE SET
-      host           = COALESCE(EXCLUDED.host,  repos.host),
-      owner          = COALESCE(EXCLUDED.owner, repos.owner),
-      name           = COALESCE(EXCLUDED.name,  repos.name),
-      last_synced_at = NOW(),
-      updated_at     = NOW()
+      host       = COALESCE(EXCLUDED.host,  repos.host),
+      owner      = COALESCE(EXCLUDED.owner, repos.owner),
+      name       = COALESCE(EXCLUDED.name,  repos.name),
+      updated_at = NOW()
     RETURNING id
     `,
     item,
diff --git a/services/libs/data-access-layer/src/packages/osv.ts b/services/libs/data-access-layer/src/packages/osv.ts
@@ -243,7 +243,7 @@ export async function getRangesForPackages(qx: QueryExecutor, ids: number[]): Pr
 export async function flipVulnerableFlags(qx: QueryExecutor, ids: number[]): Promise<number> {
   if (ids.length === 0) return 0
   return qx.result(
-    `UPDATE packages SET has_critical_vulnerability = TRUE
+    `UPDATE packages SET has_critical_vulnerability = TRUE, last_synced_at = NOW()
      WHERE id IN ($(ids:csv)) AND has_critical_vulnerability = FALSE`,
     { ids },
   )
@@ -252,7 +252,7 @@ export async function flipVulnerableFlags(qx: QueryExecutor, ids: number[]): Pro
 export async function clearSafeFlags(qx: QueryExecutor, ids: number[]): Promise<number> {
   if (ids.length === 0) return 0
   return qx.result(
-    `UPDATE packages SET has_critical_vulnerability = FALSE
+    `UPDATE packages SET has_critical_vulnerability = FALSE, last_synced_at = NOW()
      WHERE id IN ($(ids:csv)) AND has_critical_vulnerability = TRUE`,
     { ids },
   )
diff --git a/services/libs/tinybird/datasources/repos.datasource b/services/libs/tinybird/datasources/repos.datasource
@@ -15,9 +15,9 @@ DESCRIPTION >
     - `watchers` is the watcher count (0 if not yet enriched).
     - `openIssues` is the number of open issues (0 if not yet enriched).
     - `lastCommitAt` is the timestamp of the most recent commit (NULL if unknown).
-    - `archived` is 1 if the repository is archived, 0 otherwise (0 default until enriched).
-    - `disabled` is 1 if the repository is disabled, 0 otherwise (0 default until enriched).
-    - `isFork` is 1 if this repository is a fork of another, 0 otherwise (0 default until enriched).
+    - `archived` is 1 if the repository is archived, 0 if not, NULL until the GitHub enricher runs (deps.dev seed does not expose this).
+    - `disabled` is 1 if the repository is disabled, 0 if not, NULL until the GitHub enricher runs.
+    - `isFork` is 1 if this repository is a fork of another, 0 if not, NULL until the GitHub enricher runs.
     - `createdAt` is the repository creation date on GitHub/GitLab — a domain timestamp, not a row-insert timestamp.
     - `homepage` is the project homepage URL (empty string if not provided).
     - `rawProjectType` is the deps.dev project type string (e.g. 'GITHUB', 'GITLAB') for identity resolution.
@@ -42,9 +42,9 @@ SCHEMA >
     `watchers` Int32 `json:$.record.watchers` DEFAULT 0,
     `openIssues` Int32 `json:$.record.open_issues` DEFAULT 0,
     `lastCommitAt` Nullable(DateTime64(3)) `json:$.record.last_commit_at`,
-    `archived` UInt8 `json:$.record.archived` DEFAULT 0,
-    `disabled` UInt8 `json:$.record.disabled` DEFAULT 0,
-    `isFork` UInt8 `json:$.record.is_fork` DEFAULT 0,
+    `archived` Nullable(UInt8) `json:$.record.archived`,
+    `disabled` Nullable(UInt8) `json:$.record.disabled`,
+    `isFork` Nullable(UInt8) `json:$.record.is_fork`,
     `createdAt` Nullable(DateTime64(3)) `json:$.record.created_at`,
     `homepage` String `json:$.record.homepage` DEFAULT '',
     `rawProjectType` String `json:$.record.raw_project_type` DEFAULT '',
diff --git a/services/libs/tinybird/datasources/versions.datasource b/services/libs/tinybird/datasources/versions.datasource
@@ -9,8 +9,8 @@ DESCRIPTION >
     - `name` is the package name — denormalized from packages for fast dependency resolution without a join.
     - `number` is the version string (e.g. '1.2.3', '2.0.0-beta.1').
     - `publishedAt` is when this version was published to the registry (NULL if unknown).
-    - `isLatest` is 1 if this is the current latest stable version according to the registry, 0 otherwise (0 default until enriched).
-    - `isYanked` is 1 if this version was retracted/yanked from the registry, 0 otherwise (0 default until enriched).
+    - `isLatest` is 1 if this is the current latest stable version, 0 if not, NULL until registry workers enrich (deps.dev seed does not expose this).
+    - `isYanked` is 1 if this version was retracted/yanked from the registry, 0 if not, NULL until registry workers enrich.
     - `isPrerelease` is 1 if the version string contains a pre-release identifier (alpha, beta, rc, etc.), 0 otherwise.
     - `licenses` is an array of SPDX-normalized license identifiers for this specific version (may differ from the package-level value).
     - `downloadCount` is the per-version download count where available (npm, crates); 0 if not tracked.
@@ -25,8 +25,8 @@ SCHEMA >
     `name` String `json:$.record.name`,
     `number` String `json:$.record.number`,
     `publishedAt` Nullable(DateTime64(3)) `json:$.record.published_at`,
-    `isLatest` UInt8 `json:$.record.is_latest` DEFAULT 0,
-    `isYanked` UInt8 `json:$.record.is_yanked` DEFAULT 0,
+    `isLatest` Nullable(UInt8) `json:$.record.is_latest`,
+    `isYanked` Nullable(UInt8) `json:$.record.is_yanked`,
     `isPrerelease` UInt8 `json:$.record.is_prerelease` DEFAULT 0,
     `licenses` Array(String) `json:$.record.licenses[:]` DEFAULT [],
     `downloadCount` UInt64 `json:$.record.download_count` DEFAULT 0,
