feat: repos collaboration signal [CM-1332]

backend/src/osspckgs/migrations/V1785167431__collaboration_signal.sql

@@ -0,0 +1,5 @@
+ALTER TABLE repo_activity_snapshot ADD COLUMN IF NOT EXISTS external_prs_opened_12m int;
+ALTER TABLE repo_activity_snapshot ADD COLUMN IF NOT EXISTS external_prs_merged_12m int;
+
+ALTER TABLE repos ADD COLUMN IF NOT EXISTS collaboration_score int;
+ALTER TABLE repos ADD COLUMN IF NOT EXISTS collaboration_tier text;

services/apps/packages_worker/src/enricher/computeMedians.ts

@@ -28,6 +28,7 @@ export interface PrNode {
   mergedAt: string | null
   closedAt: string | null
   author: { login: string } | null
+  authorAssociation: string | null
   comments: { nodes: ResponseNode[] }
   reviews: { nodes: ResponseNode[] }
 }
@@ -104,3 +105,19 @@ export function computeIssueMedians(issues: IssueNode[]): {
     medianTimeToFirstResponseHours: toIntHours(median(firstResponseHours)),
   }
 }
+
+const MAINTAINER_ASSOCIATIONS = new Set(['OWNER', 'MEMBER', 'COLLABORATOR'])
+
+export function computeExternalPrCounts(prs: PrNode[]): {
+  externalPrsOpened: number
+  externalPrsMerged: number
+} {
+  let externalPrsOpened = 0
+  let externalPrsMerged = 0
+  for (const pr of prs) {
+    if (pr.authorAssociation && MAINTAINER_ASSOCIATIONS.has(pr.authorAssociation)) continue
+    externalPrsOpened++
+    if (pr.mergedAt != null) externalPrsMerged++
+  }
+  return { externalPrsOpened, externalPrsMerged }
+}

services/apps/packages_worker/src/enricher/fetchActivitySnapshot.ts

@@ -1,6 +1,12 @@
 import { getServiceChildLogger } from '@crowd/logging'
 
-import { IssueNode, PrNode, computeIssueMedians, computePrMedians } from './computeMedians'
+import {
+  IssueNode,
+  PrNode,
+  computeExternalPrCounts,
+  computeIssueMedians,
+  computePrMedians,
+} from './computeMedians'
 import { FetchError, RepoActivitySnapshot } from './types'
 
 const log = getServiceChildLogger('fetch-activity-snapshot')
@@ -170,6 +176,7 @@ const PR_PAGE_QUERY = `
           mergedAt
           closedAt
           author { login }
+          authorAssociation
           comments(first: ${RESPONSES_PER_NODE}) { nodes { createdAt author { login } } }
           reviews(first: ${RESPONSES_PER_NODE})  { nodes { createdAt author { login } } }
         }
@@ -336,6 +343,7 @@ export async function fetchActivitySnapshot(
   )
 
   const prMedians = computePrMedians(prResult.nodes)
+  const externalPrCounts = computeExternalPrCounts(prResult.nodes)
   const issueMedians = computeIssueMedians(issueResult.nodes)
   const issuesOpenedLast6m = issueResult.nodes.filter(
     (n) => new Date(n.createdAt) >= since6m,
@@ -358,6 +366,8 @@ export async function fetchActivitySnapshot(
     prsClosedUnmerged12m: prResult.nodes.filter((n) => n.closedAt && !n.mergedAt).length,
     prMedianTimeToMergeHours: prMedians.medianTimeToMergeHours,
     prMedianTimeToFirstResponseHours: prMedians.medianTimeToFirstResponseHours,
+    externalPrsOpened12m: externalPrCounts.externalPrsOpened,
+    externalPrsMerged12m: externalPrCounts.externalPrsMerged,
     issuesOpenedLast12m: issueResult.nodes.length,
     issuesClosedLast12m: issueResult.nodes.filter((n) => n.closedAt).length,
     issuesOpenedLast6m,

services/apps/packages_worker/src/enricher/runEnrichmentLoop.ts

@@ -13,6 +13,7 @@ import {
   RepoActivitySnapshot,
   RepoWellKnownFilesUpdate,
 } from './types'
+import { updateCollaborationSignal } from './updateCollaborationSignal'
 import { bulkUpdateEnrichedRepos, markReposSkipped } from './updateEnrichedRepos'
 import { bulkUpsertRepoActivitySnapshot } from './updateRepoActivitySnapshot'
 import { bulkUpsertRepoWellKnownFiles } from './updateWellKnownFiles'
@@ -534,6 +535,17 @@ export async function runEnrichmentLoop(
       },
       `All repos processed — sleeping ${config.idleSleepSec}s`,
     )
+
+    try {
+      const updated = await updateCollaborationSignal(qx)
+      log.info({ updated }, 'Collaboration signal recomputed')
+    } catch (err) {
+      log.error(
+        { errMsg: (err as Error).message },
+        'Collaboration signal pass failed — will retry next sweep',
+      )
+    }
+
     await new Promise((r) => setTimeout(r, config.idleSleepSec * 1000))
   }
 

services/apps/packages_worker/src/enricher/types.ts

@@ -50,6 +50,8 @@ export interface RepoActivitySnapshot {
   prsClosedUnmerged12m: number | null
   prMedianTimeToMergeHours: number | null
   prMedianTimeToFirstResponseHours: number | null
+  externalPrsOpened12m: number | null
+  externalPrsMerged12m: number | null
   issuesOpenedLast12m: number | null
   issuesClosedLast12m: number | null
   issuesOpenedLast6m: number | null

services/apps/packages_worker/src/enricher/updateCollaborationSignal.ts

@@ -0,0 +1,91 @@
+import { QueryExecutor } from '@crowd/data-access-layer/src/queryExecutor'
+
+export async function updateCollaborationSignal(qx: QueryExecutor): Promise<number> {
+  return qx.result(
+    `
+    WITH critical_repos AS (
+      SELECT DISTINCT pr.repo_id
+      FROM package_repos pr
+      JOIN packages p ON p.id = pr.package_id AND p.is_critical
+    ),
+    advisory_repos AS (
+      SELECT
+        pr.repo_id,
+        BOOL_OR(EXISTS (
+          SELECT 1 FROM advisory_affected_ranges rg
+          WHERE rg.advisory_package_id = ap.id AND rg.fixed_version IS NOT NULL
+        ))                                                        AS any_fixed,
+        BOOL_OR(a.published_at < NOW() - INTERVAL '90 days')      AS any_old
+      FROM advisory_packages ap
+      JOIN advisories a     ON a.id = ap.advisory_id
+      JOIN package_repos pr ON pr.package_id = ap.package_id
+      WHERE ap.package_id IS NOT NULL
+      GROUP BY pr.repo_id
+    ),
+    components AS (
+      SELECT
+        cr.repo_id,
+        (r.archived IS TRUE OR r.disabled IS TRUE)                AS inactive,
+        CASE
+          WHEN s.issues_opened_last_12m >= 5
+           AND s.issue_median_time_to_first_response_hours IS NOT NULL THEN
+            CASE
+              WHEN s.issue_median_time_to_first_response_hours <= 72  THEN 1.0
+              WHEN s.issue_median_time_to_first_response_hours <= 336 THEN 0.5
+              ELSE 0.0
+            END
+        END                                                       AS comp_r,
+        CASE
+          WHEN s.external_prs_opened_12m IS NOT NULL THEN
+            CASE
+              WHEN s.external_prs_opened_12m >= 3
+              THEN s.external_prs_merged_12m::numeric / s.external_prs_opened_12m
+            END
+          WHEN s.prs_opened_last_12m >= 3
+          THEN s.prs_merged_last_12m::numeric / s.prs_opened_last_12m
+        END                                                       AS comp_m,
+        CASE
+          WHEN ar.any_fixed THEN 1.0
+          WHEN ar.any_old   THEN 0.0
+        END                                                       AS comp_a
+      FROM critical_repos cr
+      JOIN repos r ON r.id = cr.repo_id
+      LEFT JOIN repo_activity_snapshot s ON s.repo_id = cr.repo_id
+      LEFT JOIN advisory_repos ar        ON ar.repo_id = cr.repo_id
+    ),
+    scored AS (
+      SELECT
+        repo_id,
+        inactive,
+        ROUND(
+          (COALESCE(comp_r, 0) + COALESCE(comp_m, 0) + COALESCE(comp_a, 0))
+          / NULLIF((comp_r IS NOT NULL)::int + (comp_m IS NOT NULL)::int
+                 + (comp_a IS NOT NULL)::int, 0)
+          * 100
+        )::int                                                    AS score
+      FROM components
+    ),
+    final AS (
+      SELECT
+        repo_id,
+        CASE WHEN inactive THEN NULL ELSE score END               AS new_score,
+        CASE
+          WHEN inactive        THEN 'inactive'
+          WHEN score IS NULL   THEN 'unknown'
+          WHEN score >= 70     THEN 'responsive'
+          WHEN score >= 40     THEN 'mixed'
+          ELSE                      'unresponsive'
+        END                                                       AS new_tier
+      FROM scored
+    )
+    UPDATE repos r
+    SET collaboration_score = f.new_score,
+        collaboration_tier  = f.new_tier
+    FROM final f
+    WHERE r.id = f.repo_id
+      AND (r.collaboration_score, r.collaboration_tier)
+          IS DISTINCT FROM (f.new_score, f.new_tier)
+    `,
+    {},
+  )
+}

services/apps/packages_worker/src/enricher/updateRepoActivitySnapshot.ts

@@ -23,6 +23,8 @@ export async function bulkUpsertRepoActivitySnapshot(
         prs_closed_unmerged_12m,
         pr_median_time_to_merge_hours,
         pr_median_time_to_first_response_hours,
+        external_prs_opened_12m,
+        external_prs_merged_12m,
         issues_opened_last_12m,
         issues_closed_last_12m,
         issues_opened_last_6m,
@@ -43,6 +45,8 @@ export async function bulkUpsertRepoActivitySnapshot(
         (j->>'prsClosedUnmerged12m')::int,
         (j->>'prMedianTimeToMergeHours')::int,
         (j->>'prMedianTimeToFirstResponseHours')::int,
+        (j->>'externalPrsOpened12m')::int,
+        (j->>'externalPrsMerged12m')::int,
         (j->>'issuesOpenedLast12m')::int,
         (j->>'issuesClosedLast12m')::int,
         (j->>'issuesOpenedLast6m')::int,
@@ -62,6 +66,8 @@ export async function bulkUpsertRepoActivitySnapshot(
         prs_closed_unmerged_12m                   = EXCLUDED.prs_closed_unmerged_12m,
         pr_median_time_to_merge_hours             = EXCLUDED.pr_median_time_to_merge_hours,
         pr_median_time_to_first_response_hours    = EXCLUDED.pr_median_time_to_first_response_hours,
+        external_prs_opened_12m                   = EXCLUDED.external_prs_opened_12m,
+        external_prs_merged_12m                   = EXCLUDED.external_prs_merged_12m,
         issues_opened_last_12m                    = EXCLUDED.issues_opened_last_12m,
         issues_closed_last_12m                    = EXCLUDED.issues_closed_last_12m,
         issues_opened_last_6m                     = EXCLUDED.issues_opened_last_6m,