spb/cancerindex

1import 'server-only';2import { run, sql, safe } from '@/lib/db';34/**5 * "What changed in cancer" (SPEC §42-43, §62): every item is a dated record already in the database6 * (an approval row, a registered study, a ranking row, an ingest run). Nothing is generated; the7 * page only orders and groups existing evidence by recency.8 */910export interface PulseApproval {11  id: number;12  approval_date: string;13  authority: string;14  jurisdiction: string;15  status: string;16  approval_type: string | null;17  indication: string;18  drug_id: string;19  drug_slug: string;20  drug_name: string;21  cancer_id: string | null;22  cancer_slug: string | null;23  cancer_name: string | null;24  tumor_agnostic: boolean;25  source_slug: string;26  provenance_id: number;27}28export async function recentApprovals(days = 90, limit = 40): Promise<PulseApproval[]> {29  return safe(30    () =>31      run<PulseApproval>(sql`32        SELECT a.id, a.approval_date, a.authority, a.jurisdiction, a.status, a.approval_type, a.indication, a.tumor_agnostic, a.provenance_id,33               d.id AS drug_id, d.slug AS drug_slug, d.name AS drug_name, c.id AS cancer_id, c.slug AS cancer_slug, c.canonical_name AS cancer_name, s.slug AS source_slug34        FROM drug_approvals a JOIN drugs d ON d.id = a.drug_id LEFT JOIN cancers c ON c.id = a.cancer_id JOIN sources s ON s.id = a.source_id35        WHERE a.approval_date ~ '^\\d{4}-\\d{2}-\\d{2}$' AND a.approval_date::date >= current_date - ${days}::int AND a.approval_date::date <= current_date36        ORDER BY a.approval_date DESC, a.id DESC LIMIT ${limit}`),37    [] as PulseApproval[],38  );39}4041export interface PulseTrial {42  id: string;43  nct_id: string;44  brief_title: string;45  acronym: string | null;46  phases: string[];47  overall_status: string | null;48  first_posted_date: string | null;49  enrollment_count: number | null;50  lead_sponsor: string | null;51  lead_sponsor_class: string | null;52  countries: string[];53  cancer_slug: string | null;54  cancer_name: string | null;55  updated_at: Date | string;56}57/** Interventional Phase III studies first posted in the window and recruiting now, with their first mapped cancer. */58export async function newPhase3Recruiting(days = 30, limit = 25): Promise<PulseTrial[]> {59  return safe(60    () =>61      run<PulseTrial>(sql`62        SELECT t.id, t.nct_id, t.brief_title, t.acronym, t.phases, t.overall_status, t.first_posted_date, t.enrollment_count, t.lead_sponsor, t.lead_sponsor_class, t.countries, t.updated_at,63               m.slug AS cancer_slug, m.canonical_name AS cancer_name64        FROM clinical_trials t65        LEFT JOIN LATERAL (66          SELECT c.slug, c.canonical_name FROM trial_conditions tc JOIN cancers c ON c.id = tc.cancer_id67          WHERE tc.trial_id = t.id ORDER BY (tc.match_type = 'PROBABILISTIC'), c.depth, c.canonical_name LIMIT 1) m ON true68        WHERE t.study_type = 'INTERVENTIONAL' AND 'PHASE3' = ANY(t.phases) AND t.overall_status = 'RECRUITING'69          AND t.first_posted_date ~ '^\\d{4}-\\d{2}-\\d{2}$' AND t.first_posted_date::date >= current_date - ${days}::int70        ORDER BY t.first_posted_date DESC, t.nct_id LIMIT ${limit}`),71    [] as PulseTrial[],72  );73}7475export interface PhasePulse {76  phase: string;77  current: number;78  previous: number;79}80/** New oncology studies per phase: last `days` vs the preceding `days` (from trial_pulse, cancer_id NULL = all studies). */81export async function trialPulseByPhase(days = 30): Promise<{ rows: PhasePulse[]; asOf: Date | string | null }> {82  const rows = await safe(83    () =>84      run<{ phase: string; current: string; previous: string; as_of: Date | string | null }>(sql`85        SELECT phase,86               sum(new_trials) FILTER (WHERE day > current_date - ${days}::int) AS current,87               sum(new_trials) FILTER (WHERE day <= current_date - ${days}::int AND day > current_date - ${days * 2}::int) AS previous,88               max(updated_at) AS as_of89        FROM trial_pulse WHERE cancer_id IS NULL AND day > current_date - ${days * 2}::int90        GROUP BY phase ORDER BY CASE phase WHEN 'ALL' THEN 0 WHEN 'EARLY_PHASE1' THEN 1 WHEN 'PHASE1' THEN 2 WHEN 'PHASE2' THEN 3 WHEN 'PHASE3' THEN 4 WHEN 'PHASE4' THEN 5 ELSE 9 END`),91    [] as Array<{ phase: string; current: string; previous: string; as_of: Date | string | null }>,92  );93  return { rows: rows.map((r) => ({ phase: r.phase, current: Number(r.current ?? 0), previous: Number(r.previous ?? 0) })), asOf: rows[0]?.as_of ?? null };94}9596export interface CancerPulse {97  id: string;98  slug: string;99  canonical_name: string;100  current: number;101  previous: number;102}103/** Top-level cancers with the most newly registered studies in the window (any phase), with the previous window for comparison. */104export async function newTrialsByCancer(days = 30, limit = 10): Promise<CancerPulse[]> {105  return safe(106    () =>107      run<CancerPulse>(sql`108        SELECT c.id, c.slug, c.canonical_name,109               coalesce(sum(p.new_trials) FILTER (WHERE p.day > current_date - ${days}::int), 0)::int AS current,110               coalesce(sum(p.new_trials) FILTER (WHERE p.day <= current_date - ${days}::int), 0)::int AS previous111        FROM trial_pulse p JOIN cancers c ON c.id = p.cancer_id112        WHERE p.phase = 'ALL' AND c.top_level AND c.status = 'active' AND p.day > current_date - ${days * 2}::int113        GROUP BY c.id, c.slug, c.canonical_name ORDER BY current DESC, c.canonical_name LIMIT ${limit}`),114    [] as CancerPulse[],115  );116}117118export interface RankingMove {119  metric_slug: string;120  metric_name: string;121  unit: string;122  scope_key: string;123  cancer_id: string;124  slug: string;125  canonical_name: string;126  rank: number;127  previous_rank: number;128  value: number;129  eligible_entities: number;130  generated_at: Date | string;131}132/** Largest rank changes between the current snapshot and the previous one for the same metric and scope (§131). */133export async function rankingMoves(minDelta = 3, limit = 20): Promise<RankingMove[]> {134  return safe(135    () =>136      run<RankingMove>(sql`137        SELECT r.metric_slug, m.name AS metric_name, r.unit, r.scope_key, r.cancer_id, c.slug, c.canonical_name, r.rank, r.previous_rank, r.value, r.eligible_entities, s.generated_at138        FROM rankings r JOIN ranking_snapshots s ON s.id = r.snapshot_id JOIN cancers c ON c.id = r.cancer_id JOIN metric_definitions m ON m.slug = r.metric_slug139        WHERE s.is_current AND r.previous_rank IS NOT NULL AND abs(r.previous_rank - r.rank) >= ${minDelta} AND s.entity_level = 'top'140        ORDER BY abs(r.previous_rank - r.rank) DESC, s.generated_at DESC, r.metric_slug, r.rank LIMIT ${limit}`),141    [] as RankingMove[],142  );143}144145export interface PulsePublication {146  id: string;147  pmid: string | null;148  doi: string | null;149  title: string;150  journal: string | null;151  pub_date: string | null;152  publication_types: string[];153  retracted: boolean;154}155/** Most recently published records indexed (PubMed), favouring trials, reviews and meta-analyses. */156export async function recentPublications(days = 60, limit = 15): Promise<PulsePublication[]> {157  return safe(158    () =>159      run<PulsePublication>(sql`160        SELECT id, pmid, doi, title, journal, pub_date, publication_types, retracted FROM publications161        WHERE pub_date ~ '^\\d{4}-\\d{2}-\\d{2}$' AND pub_date::date >= current_date - ${days}::int AND pub_date::date <= current_date AND NOT retracted162        ORDER BY (publication_types && ARRAY['Randomized Controlled Trial','Clinical Trial, Phase III','Meta-Analysis','Systematic Review','Clinical Trial']::text[]) DESC, pub_date DESC LIMIT ${limit}`),163    [] as PulsePublication[],164  );165}166167export interface IngestRow {168  id: string;169  connector_id: string;170  source_name: string | null;171  mode: string;172  status: string;173  started_at: Date | string;174  finished_at: Date | string | null;175  duration_ms: number | null;176  records_fetched: number;177  records_created: number;178  records_updated: number;179  records_unchanged: number;180  records_rejected: number;181  dataset_version: string | null;182  anomaly: string | null;183}184/** Ingest runs of the last `days` days (all statuses), most recent first (§62 data update log). */185export async function recentIngests(days = 30, limit = 200): Promise<IngestRow[]> {186  return safe(187    () =>188      run<IngestRow>(sql`189        SELECT r.id, r.connector_id, s.name AS source_name, r.mode, r.status, r.started_at, r.finished_at, r.duration_ms, r.records_fetched, r.records_created, r.records_updated, r.records_unchanged, r.records_rejected, r.dataset_version, r.anomaly190        FROM ingest_runs r LEFT JOIN sources s ON s.slug = r.connector_id191        WHERE r.started_at >= now() - (${days}::text || ' days')::interval AND r.mode NOT IN ('dry_run','probe')192        ORDER BY r.started_at DESC LIMIT ${limit}`),193    [] as IngestRow[],194  );195}196197export interface ConnectorState {198  connector_id: string;199  source_name: string | null;200  category: string | null;201  status: string | null;202  health: string;203  last_success_at: Date | string | null;204  last_attempt_at: Date | string | null;205  paused: boolean;206  schedule: string | null;207  last_dataset_version: string | null;208  record_count: number;209}210/** One line per connector: health, last success, schedule, latest dataset version, source records held. */211export async function connectorStates(): Promise<ConnectorState[]> {212  return safe(213    () =>214      run<ConnectorState>(sql`215        SELECT s.slug AS connector_id, s.name AS source_name, s.category, s.status, coalesce(cc.health, 'unknown') AS health, cc.last_success_at, cc.last_attempt_at, coalesce(cc.paused, false) AS paused,216               s.manifest->>'schedule' AS schedule,217               (SELECT r.dataset_version FROM ingest_runs r WHERE r.connector_id = s.slug AND r.status IN ('succeeded','partial') ORDER BY r.started_at DESC LIMIT 1) AS last_dataset_version,218               (SELECT count(*) FROM source_records sr WHERE sr.source_id = s.id)::int AS record_count219        FROM sources s LEFT JOIN connector_cursors cc ON cc.connector_id = s.slug220        ORDER BY s.status = 'active' DESC, s.tier, s.name`),221    [] as ConnectorState[],222  );223}224225/** Recent change events other than bulk creations (approval_added, alias_added, deprecated, merged…). */226export interface PulseEvent {227  id: number;228  entity_type: string;229  entity_id: string;230  kind: string;231  summary: string;232  created_at: Date | string;233  entity_name: string | null;234  entity_href: string | null;235}236export async function recentEvents(days = 30, limit = 40): Promise<PulseEvent[]> {237  return safe(238    () =>239      run<PulseEvent>(sql`240        SELECT e.id, e.entity_type, e.entity_id, e.kind, e.summary, e.created_at,241          CASE e.entity_type WHEN 'cancer' THEN (SELECT canonical_name FROM cancers WHERE id = e.entity_id) WHEN 'gene' THEN (SELECT symbol FROM genes WHERE id = e.entity_id) WHEN 'drug' THEN (SELECT name FROM drugs WHERE id = e.entity_id) WHEN 'trial' THEN (SELECT nct_id FROM clinical_trials WHERE id = e.entity_id) END AS entity_name,242          CASE e.entity_type WHEN 'cancer' THEN (SELECT '/cancer/' || slug FROM cancers WHERE id = e.entity_id) WHEN 'gene' THEN (SELECT '/gene/' || symbol FROM genes WHERE id = e.entity_id) WHEN 'drug' THEN (SELECT '/drug/' || slug FROM drugs WHERE id = e.entity_id) WHEN 'trial' THEN (SELECT '/trial/' || nct_id FROM clinical_trials WHERE id = e.entity_id) END AS entity_href243        FROM change_events e244        WHERE e.created_at >= now() - (${days}::text || ' days')::interval AND e.kind <> 'created'245        ORDER BY e.created_at DESC, e.id DESC LIMIT ${limit}`),246    [] as PulseEvent[],247  );248}249