spb/cancerindex

1import { sql } from 'drizzle-orm';2import type { Database } from '@cancerindex/database';3import { proposeDrugMerges } from './drug-duplicates.js';45export const DRUG_PIPELINE_FORMULA_VERSION = 'ci-drug-pipeline-v1';67/** Stages, least to most advanced (funnel order on /pipeline). `withdrawn` sits outside the funnel. */8export const PIPELINE_STAGES = ['phase_not_stated', 'phase1', 'phase2', 'phase3', 'phase4', 'approved', 'withdrawn'] as const;9export type PipelineStage = (typeof PIPELINE_STAGES)[number];1011/** drug_approvals.status values that count as a current market authorization. */12export const PIPELINE_APPROVED_STATUSES = ['approved', 'accelerated', 'conditional'] as const;13/** Registry statuses counted as "active" (same list as trial-intelligence / counters). */14export const PIPELINE_ACTIVE_STATUSES = ['RECRUITING', 'NOT_YET_RECRUITING', 'ENROLLING_BY_INVITATION', 'ACTIVE_NOT_RECRUITING'] as const;15/** Registry phase labels → rank. EARLY_PHASE1 and PHASE1 share rank 1 (both → stage phase1); NA → 0. */16export const PHASE_RANK: Record<string, number> = { PHASE4: 4, PHASE3: 3, PHASE2: 2, PHASE1: 1, EARLY_PHASE1: 1, NA: 0 };1718export const PIPELINE_THRESHOLDS = {19  /** cancer_hierarchy traversal depth from a top-level cancer down to trial / approval cancers (same as counters). */20  maxHierarchyDepth: 12,21} as const;2223export interface DrugPipelineResult {24  rows: number;25  drugRows: number;26  cancerRows: number;27  mergeProposals: number;28  mergeCandidates: number;29  ms: number;30}3132/**33 * Highest registry phase over a set of trials (pure, unit-tested). Returns the phase label:34 * PHASE4 > PHASE3 > PHASE2 > PHASE1 ≡ EARLY_PHASE1 (PHASE1 wins the label when both occur) > NA;35 * null when no trial carries a known phase label.36 */37export function maxPhase(phaseLists: Iterable<readonly string[]>): string | null {38  let best = -1;39  let label: string | null = null;40  let sawPhase1 = false;41  for (const phases of phaseLists) {42    for (const p of phases) {43      const r = PHASE_RANK[p];44      if (r === undefined) continue;45      if (p === 'PHASE1') sawPhase1 = true;46      if (r > best) {47        best = r;48        label = p;49      }50    }51  }52  if (best < 0) return null;53  if (best === 1) return sawPhase1 ? 'PHASE1' : 'EARLY_PHASE1';54  return label;55}5657/** Phase label from a SQL-side rank (mirrors `maxPhase`). */58export function phaseLabel(rank: number | null, hasPhase1: boolean): string | null {59  if (rank == null || rank < 0) return null;60  if (rank >= 4) return 'PHASE4';61  if (rank === 3) return 'PHASE3';62  if (rank === 2) return 'PHASE2';63  if (rank === 1) return hasPhase1 ? 'PHASE1' : 'EARLY_PHASE1';64  return 'NA';65}6667export interface StageInput {68  /** Approvals with status ∈ PIPELINE_APPROVED_STATUSES (for the cancer when scoped, any cancer when not). */69  approvedLike: number;70  /** Approvals with any other status (withdrawn, superseded). */71  withdrawnLike: number;72  /** Interventional trials linking the drug (and the cancer when scoped). */73  totalTrials: number;74  /** Output of `maxPhase` over those trials. */75  maxPhase: string | null;76}7778/**79 * Stage rule (pure, unit-tested; docs/methodology/pipeline.md):80 *   approved            any approval with status approved | accelerated | conditional81 *   withdrawn           approvals exist but all are withdrawn / superseded82 *   phase4 … phase1     otherwise, by the highest registry phase among interventional trials83 *                       (PHASE2+PHASE3 → phase3, PHASE1+PHASE2 → phase2, EARLY_PHASE1 → phase1)84 *   phase_not_stated    trials exist but none states a phase (NA / empty)85 *   null                no trials and no approvals → no row86 */87export function stageFor(input: StageInput): PipelineStage | null {88  if (input.approvedLike > 0) return 'approved';89  if (input.withdrawnLike > 0) return 'withdrawn';90  if (input.totalTrials <= 0) return null;91  switch (input.maxPhase) {92    case 'PHASE4':93      return 'phase4';94    case 'PHASE3':95      return 'phase3';96    case 'PHASE2':97      return 'phase2';98    case 'PHASE1':99    case 'EARLY_PHASE1':100      return 'phase1';101    default:102      return 'phase_not_stated';103  }104}105106type AggRow = {107  drug_id: string;108  top_id: string | null;109  total_trials: string | number | null;110  active_trials: string | number | null;111  recruiting_trials: string | number | null;112  phase3_trials: string | number | null;113  phase_rank: string | number | null;114  has_phase1: boolean | null;115  first_trial_date: string | null;116  approvals: string | number | null;117  approved_like: string | number | null;118  jurisdictions: string[] | null;119  first_approval_date: string | null;120  latest_approval_date: string | null;121};122123const n =(v: string | number | null | undefined) => (v == null ? 0 : Number(v));124125/**126 * Recompute `drug_pipeline`: one row per drug (cancer_id NULL, across all cancers) and one per127 * (drug, top-level cancer). A trial reaches a top-level cancer through `trial_conditions.cancer_id`128 * and its ancestors in `cancer_hierarchy` (depth ≤ 12); an approval through `drug_approvals.cancer_id`129 * the same way (approvals without a cancer only feed the unscoped row). Counts are over130 * interventional studies (DISTINCT trials). Set-based SQL over temp tables, one transaction, the131 * stage decided by the pure `stageFor` rule. Finally, salt-form / alias duplicates among drugs are132 * *proposed* to `entity_merges` (never merged here).133 */134export async function computeDrugPipeline(db: Database): Promise<DrugPipelineResult> {135  const t0 = Date.now();136  const depth = sql.raw(String(PIPELINE_THRESHOLDS.maxHierarchyDepth));137  const activeSql = sql.raw(`ARRAY[${PIPELINE_ACTIVE_STATUSES.map((s) => `'${s}'`).join(',')}]::text[]`);138  const approvedSql = sql.raw(`ARRAY[${PIPELINE_APPROVED_STATUSES.map((s) => `'${s}'`).join(',')}]::text[]`);139  const rankCase = sql.raw(`CASE p ${Object.entries(PHASE_RANK)140    .map(([k, v]) => `WHEN '${k}' THEN ${v}`)141    .join(' ')} ELSE NULL END`);142143  const { drugRows, cancerRows } = await db.transaction(async (tx) => {144    // Top-level cancer → every descendant (itself included), across all hierarchy types.145    await tx.execute(sql`CREATE TEMP TABLE _dp_top (top_id varchar(32), cancer_id varchar(32), PRIMARY KEY (top_id, cancer_id)) ON COMMIT DROP`);146    await tx.execute(sql`147      INSERT INTO _dp_top148      WITH RECURSIVE d AS (149        SELECT id AS top_id, id AS cancer_id, 0 AS depth FROM cancers WHERE top_level AND status = 'active'150        UNION151        SELECT d.top_id, h.child_id, d.depth + 1 FROM d JOIN cancer_hierarchy h ON h.parent_id = d.cancer_id WHERE d.depth < ${depth}152      )153      SELECT DISTINCT top_id, cancer_id FROM d`);154    // Interventional trials with their activity flags and highest phase rank.155    await tx.execute(sql`156      CREATE TEMP TABLE _dp_trial ON COMMIT DROP AS157      SELECT t.id,158        t.overall_status = ANY(${activeSql}) AS active,159        t.overall_status = 'RECRUITING' AS recruiting,160        (SELECT max(${rankCase}) FROM unnest(t.phases) AS p) AS phase_rank,161        'PHASE1' = ANY(t.phases) AS has_phase1,162        'PHASE3' = ANY(t.phases) AS has_phase3,163        CASE WHEN t.start_date ~ '^\\d{4}' THEN t.start_date END AS start_date164      FROM clinical_trials t WHERE t.study_type = 'INTERVENTIONAL'`);165    await tx.execute(sql`CREATE INDEX ON _dp_trial (id)`);166    await tx.execute(sql`167      CREATE TEMP TABLE _dp_td ON COMMIT DROP AS168      SELECT DISTINCT ti.drug_id, ti.trial_id FROM trial_interventions ti JOIN _dp_trial tr ON tr.id = ti.trial_id WHERE ti.drug_id IS NOT NULL`);169    await tx.execute(sql`170      CREATE TEMP TABLE _dp_tdc ON COMMIT DROP AS171      SELECT DISTINCT td.drug_id, td.trial_id, tp.top_id172      FROM _dp_td td JOIN trial_conditions tc ON tc.trial_id = td.trial_id AND tc.cancer_id IS NOT NULL JOIN _dp_top tp ON tp.cancer_id = tc.cancer_id`);173174    const trialAgg = (scoped: boolean) => sql`175      SELECT x.drug_id, ${scoped ? sql`x.top_id` : sql`NULL::varchar`} AS top_id,176        count(*) AS total_trials,177        count(*) FILTER (WHERE tr.active) AS active_trials,178        count(*) FILTER (WHERE tr.recruiting) AS recruiting_trials,179        count(*) FILTER (WHERE tr.has_phase3) AS phase3_trials,180        max(tr.phase_rank) AS phase_rank,181        bool_or(tr.has_phase1) AS has_phase1,182        min(tr.start_date) AS first_trial_date183      FROM ${scoped ? sql`_dp_tdc` : sql`_dp_td`} x JOIN _dp_trial tr ON tr.id = x.trial_id184      GROUP BY x.drug_id${scoped ? sql`, x.top_id` : sql``}`;185    const approvalAgg = (scoped: boolean) => sql`186      SELECT a.drug_id, ${scoped ? sql`tp.top_id` : sql`NULL::varchar`} AS top_id,187        count(*) AS approvals,188        count(*) FILTER (WHERE a.status = ANY(${approvedSql})) AS approved_like,189        array_agg(DISTINCT a.jurisdiction ORDER BY a.jurisdiction) AS jurisdictions,190        min(a.approval_date) FILTER (WHERE a.status = ANY(${approvedSql}) AND a.approval_date IS NOT NULL) AS first_approval_date,191        max(a.approval_date) FILTER (WHERE a.status = ANY(${approvedSql}) AND a.approval_date IS NOT NULL) AS latest_approval_date192      FROM drug_approvals a ${scoped ? sql`JOIN _dp_top tp ON tp.cancer_id = a.cancer_id` : sql``}193      GROUP BY a.drug_id${scoped ? sql`, tp.top_id` : sql``}`;194    const combined = (scoped: boolean) => sql`195      SELECT COALESCE(t.drug_id, a.drug_id) AS drug_id, COALESCE(t.top_id, a.top_id) AS top_id,196        t.total_trials, t.active_trials, t.recruiting_trials, t.phase3_trials, t.phase_rank, t.has_phase1, t.first_trial_date,197        a.approvals, a.approved_like, a.jurisdictions, a.first_approval_date, a.latest_approval_date198      FROM (${trialAgg(scoped)}) t FULL OUTER JOIN (${approvalAgg(scoped)}) a ON a.drug_id = t.drug_id ${scoped ? sql`AND a.top_id = t.top_id` : sql``}199      WHERE EXISTS (SELECT 1 FROM drugs d WHERE d.id = COALESCE(t.drug_id, a.drug_id))`;200201    const unscoped = (await tx.execute<AggRow>(combined(false))) as unknown as AggRow[];202    const scoped = (await tx.execute<AggRow>(combined(true))) as unknown as AggRow[];203204    await tx.execute(sql`DELETE FROM drug_pipeline`);205    let drugRows = 0;206    let cancerRows = 0;207    const batch: Array<Record<string, unknown>> = [];208    const push = (r: AggRow, scope: 'all' | 'top_level_cancer') => {209      const approvedLike = n(r.approved_like);210      const approvals = n(r.approvals);211      const total = n(r.total_trials);212      const rank = r.phase_rank == null ? null : Number(r.phase_rank);213      const mp = total > 0 ? phaseLabel(rank, !!r.has_phase1) : null;214      const stage = stageFor({ approvedLike, withdrawnLike: approvals - approvedLike, totalTrials: total, maxPhase: mp });215      if (!stage) return;216      batch.push({217        drug_id: r.drug_id,218        cancer_id: r.top_id,219        stage,220        max_phase: mp,221        active_trials: n(r.active_trials),222        recruiting_trials: n(r.recruiting_trials),223        phase3_trials: n(r.phase3_trials),224        total_trials: total,225        approvals,226        // text[] inside unnest cannot carry a nested array → serialized as JSON text, expanded in the INSERT below.227        jurisdictions: JSON.stringify(r.jurisdictions ?? []),228        first_approval_date: r.first_approval_date,229        latest_approval_date: r.latest_approval_date,230        first_trial_date: r.first_trial_date,231        formula_version: DRUG_PIPELINE_FORMULA_VERSION,232        inputs: JSON.stringify({233          scope,234          approvedLike,235          withdrawnLike: approvals - approvedLike,236          phaseRank: rank,237          approvedStatuses: PIPELINE_APPROVED_STATUSES,238          activeStatuses: PIPELINE_ACTIVE_STATUSES,239          ancestorMapping: `trial_conditions.cancer_id / drug_approvals.cancer_id → top-level ancestor via cancer_hierarchy (depth ≤ ${PIPELINE_THRESHOLDS.maxHierarchyDepth})`,240          stageRule: 'approved > withdrawn > highest registry phase (PHASE4, PHASE3, PHASE2, PHASE1/EARLY_PHASE1) > phase_not_stated',241        }),242      });243      if (scope === 'all') drugRows++;244      else cancerRows++;245    };246    for (const r of unscoped) push(r, 'all');247    for (const r of scoped) if (r.top_id) push(r, 'top_level_cancer');248    // Insert in batches of 1,000 rows.249    for (let i = 0; i < batch.length; i += 1000) await flushBatch(tx, batch.slice(i, i + 1000));250    return { drugRows, cancerRows };251  });252253  const merges = await proposeDrugMerges(db);254  return { rows: drugRows + cancerRows, drugRows, cancerRows, mergeProposals: merges.inserted, mergeCandidates: merges.candidates, ms: Date.now() - t0 };255}256257async function flushBatch(tx: Pick<Database, 'execute'>, rows: Array<Record<string, unknown>>): Promise<void> {258  if (!rows.length) return;259  const col = (k: string) => sql.param(rows.map((r) => r[k]));260  await tx.execute(sql`261    INSERT INTO drug_pipeline (drug_id, cancer_id, stage, max_phase, active_trials, recruiting_trials, phase3_trials, total_trials, approvals, jurisdictions, first_approval_date, latest_approval_date, first_trial_date, formula_version, inputs)262    SELECT u.drug_id, u.cancer_id, u.stage, u.max_phase, u.active_trials, u.recruiting_trials, u.phase3_trials, u.total_trials, u.approvals,263      COALESCE(ARRAY(SELECT jsonb_array_elements_text(u.jurisdictions_json::jsonb)), '{}'::text[]), u.first_approval_date, u.latest_approval_date, u.first_trial_date, u.formula_version, u.inputs::jsonb264    FROM unnest(265      ${col('drug_id')}::varchar[], ${col('cancer_id')}::varchar[], ${col('stage')}::text[], ${col('max_phase')}::text[],266      ${col('active_trials')}::int[], ${col('recruiting_trials')}::int[], ${col('phase3_trials')}::int[], ${col('total_trials')}::int[], ${col('approvals')}::int[],267      ${col('jurisdictions')}::text[], ${col('first_approval_date')}::text[], ${col('latest_approval_date')}::text[], ${col('first_trial_date')}::text[],268      ${col('formula_version')}::text[], ${col('inputs')}::text[]269    ) AS u(drug_id, cancer_id, stage, max_phase, active_trials, recruiting_trials, phase3_trials, total_trials, approvals, jurisdictions_json, first_approval_date, latest_approval_date, first_trial_date, formula_version, inputs)`);270}271