feat: add packages ranking workflow and dependent_counts row-count guard (CM-1222) (#4217)

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

Author: themarolt

backend/src/osspckgs/migrations/V1781514035__ranking-job-kind.sql

@@ -0,0 +1,5 @@
+-- Drop CHECK constraints on job_kind and sync_mode so new kinds can be added
+-- without a migration every time.
+ALTER TABLE osspckgs_ingest_jobs
+  DROP CONSTRAINT osspckgs_ingest_jobs_job_kind_check,
+  DROP CONSTRAINT osspckgs_ingest_jobs_sync_mode_check;

pnpm-lock.yaml

@@ -218,12 +218,18 @@ function multiselect {
 SELECTED_SERVICES=()
 
 function select_services() {
-    # Get all services
+    # Get all deployable services from builder definitions
     SERVICES=()
-    while IFS= read -r file; do
-        filename=$(basename "$file" .yaml)
-        SERVICES+=("$filename")
-    done < <(find "$CLI_HOME/services" -name '*.yaml')
+    for env_file in "$CLI_HOME/builders/"*.env; do
+        services_line=$(grep "^SERVICES=" "$env_file" 2>/dev/null || true)
+        if [[ -n "$services_line" ]]; then
+            services_val="${services_line#SERVICES=}"
+            services_val="${services_val//\"/}"
+            for svc in $services_val; do
+                SERVICES+=("$svc")
+            done
+        fi
+    done
 
     # Use fzf if available, fall back to built-in multiselect
     if command -v fzf &>/dev/null; then
@@ -251,7 +257,7 @@ function select_services() {
 
 function deploy_staging() {
     REPOSITORY="CrowdDotDev/crowd.dev"
-    WORKFLOW_FILE="lf-staging-deploy.yaml"
+    WORKFLOW_FILE="lf-oracle-staging-deploy.yaml"
     CURRENT_BRANCH=$(git branch --show-current)
     SERVICES="$(
         IFS=" "
@@ -263,7 +269,7 @@ function deploy_staging() {
 
 function deploy_production() {
     REPOSITORY="CrowdDotDev/crowd.dev"
-    WORKFLOW_FILE="lf-production-deploy.yaml"
+    WORKFLOW_FILE="lf-oracle-production-deploy.yaml"
     CURRENT_BRANCH=$(git branch --show-current)
     SERVICES="$(
         IFS=" "

scripts/cli

@@ -50,6 +50,7 @@
     "@crowd/data-access-layer": "workspace:*",
     "@crowd/database": "workspace:*",
     "@crowd/logging": "workspace:*",
+    "@crowd/slack": "workspace:*",
     "@dsnp/parquetjs": "^1.7.0",
     "@google-cloud/bigquery": "^8.3.1",
     "@google-cloud/storage": "7.19.0",

services/apps/packages_worker/package.json

@@ -13,3 +13,4 @@ export {
 export * from './deps-dev/activities'
 export { osvSyncEcosystem, osvDeriveCriticalFlag } from './osv/activities'
 export { processMavenCriticalBatch, processMavenNonCriticalBatch } from './maven/activities'
+export { criticalityComputePageRank, rankPackages } from './criticality/activities'

services/apps/packages_worker/src/activities.ts

@@ -1,5 +1,6 @@
 import { Context } from '@temporalio/activity'
 
+import { createIngestJob, markJobStatus } from '@crowd/data-access-layer'
 import { getServiceChildLogger } from '@crowd/logging'
 
 import { getPackagesDb } from '../db'
@@ -63,3 +64,34 @@ export async function criticalityComputePageRank(
 
   return { ecosystem, nodeCount: graph.N, edgeCount, iterations, durationMs: Date.now() - start }
 }
+
+export async function rankPackages(): Promise<{ scoredRows: number; rankedRows: number }> {
+  const qx = await getPackagesDb()
+
+  // On retry, a pending row from the prior attempt may already exist — reuse it.
+  // Do NOT reuse a done row: it belongs to a previous bootstrap run and ranking must re-execute.
+  const existing = await qx.selectOneOrNone(
+    `SELECT id FROM osspckgs_ingest_jobs
+     WHERE job_kind = 'ranking' AND status = 'pending'
+     ORDER BY id DESC LIMIT 1`,
+  )
+
+  const jobId = existing?.id ?? (await createIngestJob(qx, 'ranking', 'ranking', null))
+  try {
+    const [result] = await qx.select(`SELECT * FROM rank_packages()`)
+    const scoredRows = Number(result.scored_rows ?? 0)
+    const rankedRows = Number(result.ranked_rows ?? 0)
+    await markJobStatus(qx, jobId, 'done', {
+      rowCountPg: scoredRows,
+      tableRowCounts: { scored: scoredRows, ranked: rankedRows },
+      finishedAt: new Date(),
+    })
+    return { scoredRows, rankedRows }
+  } catch (err) {
+    await markJobStatus(qx, jobId, 'failed', {
+      errorMessage: (err as Error).message,
+      finishedAt: new Date(),
+    })
+    throw err
+  }
+}

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

@@ -0,0 +1,12 @@
+import { proxyActivities } from '@temporalio/workflow'
+
+import type * as critActivities from './activities'
+
+const { rankPackages } = proxyActivities<typeof critActivities>({
+  startToCloseTimeout: '30 minutes',
+  retry: { maximumAttempts: 2 },
+})
+
+export async function rankPackagesWorkflow(): Promise<void> {
+  await rankPackages()
+}

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

@@ -0,0 +1,55 @@
+import { getLastCompletedJobRowCount as dalGetLastCompletedJobRowCount } from '@crowd/data-access-layer'
+import { SlackChannel, SlackPersona, sendSlackNotification } from '@crowd/slack'
+
+import { getPackagesDb } from '../../db'
+
+export interface CheckDependentCountsGuardInput {
+  currentRowCount: number
+  snapshotDate: string
+}
+
+export interface CheckDependentCountsGuardOutput {
+  ok: boolean
+  prevRowCount: number | null
+  dropPct: number | null
+}
+
+const DROP_THRESHOLD = 0.05
+
+export async function checkDependentCountsGuard(
+  input: CheckDependentCountsGuardInput,
+): Promise<CheckDependentCountsGuardOutput> {
+  const qx = await getPackagesDb()
+  const prevRowCount = await dalGetLastCompletedJobRowCount(qx, 'dependent_counts')
+
+  if (prevRowCount === null || prevRowCount <= 0) {
+    return { ok: true, prevRowCount, dropPct: null }
+  }
+
+  const dropPct = (prevRowCount - input.currentRowCount) / prevRowCount
+
+  if (dropPct >= DROP_THRESHOLD) {
+    sendSlackNotification(
+      SlackChannel.CDP_CRITICAL_ALERTS,
+      SlackPersona.CRITICAL_ALERTER,
+      ':warning: dependent_counts row count anomaly detected',
+      [
+        {
+          title: 'Snapshot',
+          text: input.snapshotDate,
+        },
+        {
+          title: 'Row count',
+          text: `${input.currentRowCount.toLocaleString()} (prev max: ${prevRowCount.toLocaleString()}, drop: ${(dropPct * 100).toFixed(1)}%)`,
+        },
+        {
+          title: 'Action',
+          text: 'Ingest aborted — existing `dependent_count` values preserved. Check deps.dev `Dependents` table for this snapshot date.',
+        },
+      ],
+    )
+    return { ok: false, prevRowCount, dropPct }
+  }
+
+  return { ok: true, prevRowCount, dropPct }
+}

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

@@ -8,5 +8,6 @@ export * from './listParquetFiles'
 export * from './gcsParquetToStaging'
 export * from './mergeStagingToTable'
 export * from './getLastSnapshot'
+export * from './checkDependentCountsGuard'
 export * from './probePartitionExists'
 export * from './resolveSnapshotDate'

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

@@ -14,8 +14,8 @@ WITH purl_map AS (
 )
 SELECT
   pm.purl                                                                                                  AS purl,
-  COUNT(DISTINCT IF(d.MinimumDepth = 1, CONCAT(d.Dependent.System, ':', d.Dependent.Name), NULL))         AS dependent_count,
-  COUNT(DISTINCT IF(d.MinimumDepth > 1, CONCAT(d.Dependent.System, ':', d.Dependent.Name), NULL))         AS transitive_dependent_count,
+  COUNT(DISTINCT IF(d.MinimumDepth = 1 AND d.Dependent.Name NOT LIKE '%>%', CONCAT(d.Dependent.System, ':', d.Dependent.Name), NULL))         AS dependent_count,
+  COUNT(DISTINCT IF(d.MinimumDepth > 1 AND d.Dependent.Name NOT LIKE '%>%', CONCAT(d.Dependent.System, ':', d.Dependent.Name), NULL))         AS transitive_dependent_count,
   COUNT(DISTINCT pvp.ProjectName)                                                                          AS dependent_repos_count
 FROM \`bigquery-public-data.deps_dev_v1.Dependents\` d
 JOIN purl_map pm ON pm.System = d.System AND pm.Name = d.Name