feat: criticality worker init (wip)

Signed-off-by: Mouad BANI <mouad-mb@outlook.com>

Author: mbani01

backend/src/osspckgs/migrations/V1780394591__packages_universe_graph_signals.sql

@@ -0,0 +1,6 @@
+-- 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);

backend/src/osspckgs/migrations/V1780416481__rank_packages_universe_v2.sql

@@ -0,0 +1,110 @@
+-- Replaces the placeholder rank_packages_universe() with the full ADR formula.
+--
+-- Formula per ADR-0001 Criticality scoring methodology:
+--   score = w_centrality  * pct_rank( centrality_score                   ) within ecosystem
+--         + w_transitive  * pct_rank( LN(1 + transitive_dependent_count) ) within ecosystem
+--         + w_dep_pkgs    * pct_rank( LN(1 + dependent_packages_count)   ) within ecosystem
+--         + w_dep_repos   * pct_rank( LN(1 + dependent_repos_count)      ) within ecosystem
+--         + w_downloads   * pct_rank( LN(1 + downloads_30d)              ) within ecosystem
+--
+-- All five weights must sum to 1.0 — caller's responsibility.
+-- centrality_score skips LN(): PageRank is already bounded in (0,1].
+-- COALESCE(x, 0) gives missing signals the floor percentile rather than an error.
+-- ROW_NUMBER() (not RANK()) keeps each ecosystem's critical set exactly at top-N.
+
+CREATE OR REPLACE FUNCTION rank_packages_universe(
+    weight_centrality           numeric,
+    weight_transitive           numeric,
+    weight_dependent_packages   numeric,
+    weight_dependent_repos      numeric,
+    weight_downloads            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
+    -- ── Step 1: score ──────────────────────────────────────────────────────────
+    WITH percentile_scores AS (
+        SELECT
+            id,
+            (
+              weight_centrality * PERCENT_RANK() OVER (
+                  PARTITION BY ecosystem ORDER BY COALESCE(centrality_score, 0))
+
+            + weight_transitive * PERCENT_RANK() OVER (
+                  PARTITION BY ecosystem ORDER BY LN(1 + COALESCE(transitive_dependent_count, 0)))
+
+            + weight_dependent_packages * PERCENT_RANK() OVER (
+                  PARTITION BY ecosystem ORDER BY LN(1 + COALESCE(dependent_packages_count, 0)))
+
+            + weight_dependent_repos * PERCENT_RANK() OVER (
+                  PARTITION BY ecosystem ORDER BY LN(1 + COALESCE(dependent_repos_count, 0)))
+
+            + weight_downloads * PERCENT_RANK() OVER (
+                  PARTITION BY ecosystem ORDER BY LN(1 + COALESCE(downloads_30d, 0)))
+            )::numeric(10, 4) AS new_score
+        FROM packages_universe
+    )
+    UPDATE packages_universe pu
+       SET criticality_score = ps.new_score,
+           last_rank_pass_at = NOW()
+      FROM percentile_scores ps
+     WHERE pu.id = ps.id
+       AND pu.criticality_score IS DISTINCT FROM ps.new_score;
+
+    GET DIAGNOSTICS n_scored = ROW_COUNT;
+
+    -- ── Step 2: rank + flag ────────────────────────────────────────────────────
+    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
+    ),
+    flagged 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 = f.r,
+           is_critical       = f.new_is_critical
+      FROM flagged f
+     WHERE pu.id = f.id
+       AND (
+             pu.rank_in_ecosystem IS DISTINCT FROM f.r
+          OR pu.is_critical       IS DISTINCT FROM f.new_is_critical
+       );
+
+    GET DIAGNOSTICS n_ranked = ROW_COUNT;
+
+    -- ── Step 3: propagate to packages ─────────────────────────────────────────
+    UPDATE packages p
+       SET criticality_score = pu.criticality_score,
+           is_critical       = pu.is_critical,
+           last_rank_pass_at = NOW()
+      FROM packages_universe pu
+     WHERE p.purl      = pu.purl
+       AND p.ecosystem = pu.ecosystem
+       AND (
+             p.criticality_score IS DISTINCT FROM pu.criticality_score
+          OR p.is_critical       IS DISTINCT FROM pu.is_critical
+       );
+
+    GET DIAGNOSTICS n_propagated = ROW_COUNT;
+
+    RETURN QUERY SELECT n_scored, n_ranked, n_propagated;
+END;
+$$;

services/apps/packages_worker/src/criticality/activities.ts

@@ -0,0 +1,54 @@
+import { Context } from '@temporalio/activity'
+
+import { getServiceChildLogger } from '@crowd/logging'
+
+import { getPackagesDb } from '../db'
+import { buildGraph, computePageRank } from './graph'
+import { loadDirectEdges, mergeCentralityScores } from './queries'
+import { CentralityInput, CentralityResult } from './types'
+
+const log = getServiceChildLogger('criticality')
+
+function getConfig() {
+  return {
+    damping:     Number(process.env.CRITICALITY_PAGERANK_DAMPING)                  || 0.85,
+    maxIter:     parseInt(process.env.CRITICALITY_PAGERANK_MAX_ITER   ?? '', 10)   || 100,
+    convergence: Number(process.env.CRITICALITY_PAGERANK_CONVERGENCE)              || 1e-6,
+  }
+}
+
+export async function criticalityComputePageRank(input: CentralityInput): Promise<CentralityResult> {
+  const { ecosystem } = input
+  const { damping, maxIter, convergence } = getConfig()
+  const start = Date.now()
+  const qx = await getPackagesDb()
+
+  // ── Step 1: build CSR graph
+  const edges = await loadDirectEdges(qx, ecosystem)
+  const edgeCount = edges.length
+  const graph = buildGraph(edges)
+  edges.length = 0  // release JS edge objects — CSR holds all graph data
+  log.info({ ecosystem, nodeCount: graph.N, edgeCount }, 'graph loaded')
+
+  // ── Step 2 & 3: PageRank
+  const { scores, iterations } = computePageRank(
+    graph, damping, maxIter, convergence,
+    (iter, delta) => {
+      try { Context.current().heartbeat({ ecosystem, iter, delta }) } catch { /* standalone */ }
+    },
+  )
+  log.info({ ecosystem, iterations, nodeCount: graph.N }, 'PageRank converged')
+
+  // ── Step 4: merge centrality_score into packages_universe
+  const batch = Array.from(graph.nodeIndex.entries()).map(([packageId, idx]) => ({
+    packageId,
+    centralityScore: scores[idx],
+  }))
+
+  const CHUNK = 10_000
+  for (let i = 0; i < batch.length; i += CHUNK) {
+    await mergeCentralityScores(qx, batch.slice(i, i + CHUNK))
+  }
+
+  return { ecosystem, nodeCount: graph.N, edgeCount, iterations, durationMs: Date.now() - start }
+}

services/apps/packages_worker/src/criticality/graph.ts

@@ -0,0 +1,103 @@
+import { DirectEdge } from './queries'
+
+export interface Graph {
+  nodeIndex: Map<number, number>
+  nodeIds:   number[]
+  numDeps:   Int32Array  // how many packages each node depends on (divides its vote)
+  rowPtr:    Int32Array  // rowPtr[v] = start of v's dependents in colData
+  colData:   Int32Array  // flat array of dependent node indices
+  N:         number
+}
+
+export function buildGraph(edges: DirectEdge[]): Graph {
+  // Pass 0: assign contiguous indices
+  const nodeIndex = new Map<number, number>()
+  const nodeIds: number[] = []
+
+  for (const { packageId, dependsOnId } of edges) {
+    if (!nodeIndex.has(packageId))   { nodeIndex.set(packageId,   nodeIds.length); nodeIds.push(packageId)   }
+    if (!nodeIndex.has(dependsOnId)) { nodeIndex.set(dependsOnId, nodeIds.length); nodeIds.push(dependsOnId) }
+  }
+
+  const N = nodeIds.length
+
+  // Pass 1: count
+  const depCount = new Int32Array(N)
+  const numDeps  = new Int32Array(N)
+
+  for (const { packageId, dependsOnId } of edges) {
+    depCount[nodeIndex.get(dependsOnId)!]++
+    numDeps[nodeIndex.get(packageId)!]++
+  }
+
+  // Build rowPtr (prefix sum of depCount)
+  const rowPtr = new Int32Array(N + 1)
+  for (let i = 0; i < N; i++) rowPtr[i + 1] = rowPtr[i] + depCount[i]
+
+  // Pass 2: fill colData
+  const colData = new Int32Array(edges.length)
+  const fillIdx = new Int32Array(N)
+
+  for (const { packageId, dependsOnId } of edges) {
+    const v = nodeIndex.get(dependsOnId)!
+    colData[rowPtr[v] + fillIdx[v]++] = nodeIndex.get(packageId)!
+  }
+
+  return { nodeIndex, nodeIds, numDeps, rowPtr, colData, N }
+}
+
+export function computePageRank(
+  { numDeps, rowPtr, colData, N }: Graph,
+  damping     = 0.85,
+  maxIter     = 100,
+  convergence = 1e-6,
+  onIteration?: (iter: number, delta: number) => void,
+): { scores: Float64Array; iterations: number } {
+  const teleportation = (1 - damping) / N   // base score every node gets regardless of links
+  let scores          = new Float64Array(N).fill(1 / N)  // start: equal weight for all nodes
+  let next            = new Float64Array(N)               // scratch buffer, swapped each iteration
+  let iters           = 0
+
+  for (iters = 1; iters <= maxIter; iters++) {
+
+    // Every node starts with the teleportation floor
+    next.fill(teleportation)
+
+    // Each node v collects votes from packages that depend on it.
+    // numDeps[dependent] is always >= 1 here — only packages with at least one
+    // outgoing edge appear in colData, so division by zero cannot occur.
+    // Dangling nodes (numDeps = 0) never appear in colData; their score
+    // accumulates but never redistributes. This is acceptable because scores
+    // are used for relative ranking via pct_rank(), not as absolute values.
+    for (let v = 0; v < N; v++) {
+      let incoming = 0
+      for (let j = rowPtr[v]; j < rowPtr[v + 1]; j++) {
+        const dependent = colData[j]
+        incoming += scores[dependent] / numDeps[dependent]
+      }
+      next[v] = teleportation + damping * incoming
+    }
+
+    // L1 delta: total change in scores across all nodes
+    let delta = 0
+    for (let i = 0; i < N; i++) delta += Math.abs(next[i] - scores[i])
+
+    ;[scores, next] = [next, scores]  // swap buffers — no data copy
+    onIteration?.(iters, delta)
+
+    if (delta < convergence) break    // scores have stabilised
+  }
+
+  return { scores, iterations: iters }
+}
+
+export function getDependents(graph: Graph, packageId: number): number[] {
+  const idx = graph.nodeIndex.get(packageId)
+  if (idx === undefined) return []
+  const { rowPtr, colData, nodeIds } = graph
+  const result: number[] = []
+  for (let j = rowPtr[idx]; j < rowPtr[idx + 1]; j++) {
+    result.push(nodeIds[colData[j]])
+  }
+  return result
+}

services/apps/packages_worker/src/criticality/queries.ts

@@ -0,0 +1,47 @@
+import { QueryExecutor } from '@crowd/data-access-layer/src/queryExecutor'
+
+export interface DirectEdge {
+  packageId: number
+  dependsOnId: number
+}
+
+
+// Unique direct package→package edges for the ecosystem.
+// DISTINCT deduplicates multi-version rows; parallel hint scoped to the transaction.
+export async function loadDirectEdges(qx: QueryExecutor, ecosystem: string): Promise<DirectEdge[]> {
+  return qx.tx(async (tx) => {
+    await tx.result(`SET LOCAL max_parallel_workers_per_gather = 4`)
+    return tx.select(
+      `SELECT DISTINCT pd.package_id    AS "packageId",
+                       pd.depends_on_id AS "dependsOnId"
+         FROM package_dependencies pd
+         JOIN packages p
+           ON p.id        = pd.package_id
+          AND p.ecosystem = $/ecosystem/
+        WHERE pd.dependency_kind = 'direct'`,
+      { ecosystem },
+    )
+  })
+}
+
+// Bulk-update centrality_score on packages_universe rows by joining through packages.
+// Uses unnest — one parameterised query regardless of row count, no string interpolation.
+// Isolated packages (not in the graph) remain NULL; rank_packages_universe() treats
+// NULL as 0 via COALESCE. Idempotent — safe for Temporal retries.
+export async function mergeCentralityScores(
+  qx: QueryExecutor,
+  rows: Array<{ packageId: number; centralityScore: number }>,
+): Promise<void> {
+  if (rows.length === 0) return
+  await qx.result(
+    `UPDATE packages_universe pu
+        SET centrality_score = v.score
+       FROM unnest($(packageIds)::bigint[], $(scores)::numeric[]) AS v(package_id, score)
+       JOIN packages p ON p.id = v.package_id
+      WHERE pu.purl = p.purl`,
+    {
+      packageIds: rows.map(r => r.packageId),
+      scores:     rows.map(r => r.centralityScore),
+    },
+  )
+}