import { sql, type SQL } from 'drizzle-orm'; import type { FastifyPluginAsyncZod } from 'fastify-type-provider-zod'; import { z } from 'zod'; import { isCiId } from '@cancerindex/shared'; import { paginate } from '../lib/envelope.js'; import { NotFound } from '../lib/errors.js'; import { pageQuery } from '../lib/pagination.js'; import { AnyList, AnyRecord, camel, camelRows, num, ok, respond } from '../lib/respond.js'; import { pluck } from '../lib/sources.js'; /** * Biomarkers (SPEC §17, §52, §121). Rows are curated metadata (identity, verified NCIt concept, * anchor gene, aliases, assay conventions); every link below is DERIVED at request time with the * rules of docs/methodology/biomarkers.md (formula `biomarker-links-v1`), mirroring * apps/web/src/lib/queries/biomarkers.ts: * scope gene_ids = anchor gene ∪ measurement.genes; variant_ids = measurement.variantSlugs. * evidence CIViC ACCEPTED, type PREDICTIVE | PROGNOSTIC | DIAGNOSTIC, variant (or gene) overlap. * cancers distinct mapped cancers of that evidence. * drugs PREDICTIVE therapies ∪ PREDICTS_RESPONSE_TO edge targets from in-scope variants. * approvals of those drugs (matchedBy = drug) or whose indication text names the marker (indication). * trials interventions ∩ scope drugs AND conditions ∩ scope cancers. * literature publication_entity_edges on scope variants (molecular) or genes. */ export const BIOMARKER_LINKS_FORMULA = 'biomarker-links-v1'; const ACTIVE_STATUSES = ['RECRUITING', 'NOT_YET_RECRUITING', 'ENROLLING_BY_INVITATION', 'ACTIVE_NOT_RECRUITING']; const KINDS = ['gene_mutation', 'protein_expression', 'hormone_receptor', 'immune_marker', 'msi', 'tmb', 'hrd', 'ctdna', 'methylation', 'signature', 'cell_surface', 'other'] as const; const SCOPE_COLUMNS = sql` b.id, b.slug, b.name, b.kind, b.gene_id, b.ncit_code, b.description, b.measurement, b.created_at, b.updated_at, ARRAY(SELECT g.id FROM genes g WHERE g.id = b.gene_id OR g.symbol IN (SELECT jsonb_array_elements_text(coalesce(b.measurement->'genes', '[]'::jsonb))))::text[] AS gene_ids, ARRAY(SELECT v.id FROM variants v WHERE v.slug IN (SELECT jsonb_array_elements_text(coalesce(b.measurement->'variantSlugs', '[]'::jsonb))))::text[] AS variant_ids`; // MATERIALIZED: otherwise the scope sub-selects (45k genes / 417k variants) are re-evaluated per joined row when the CTE is inlined (22 s for MSI-H, measured 2026-09-11). const scopeCte = (where: SQL) => sql`WITH b AS MATERIALIZED (SELECT ${SCOPE_COLUMNS} FROM biomarkers b WHERE ${where})`; const EV_SCOPE = sql`e.status = 'ACCEPTED' AND e.evidence_type IN ('PREDICTIVE', 'PROGNOSTIC', 'DIAGNOSTIC') AND CASE WHEN cardinality(b.variant_ids) > 0 THEN e.variant_ids && b.variant_ids ELSE cardinality(b.gene_ids) > 0 AND e.gene_ids && b.gene_ids END`; const VAR_SCOPE = sql`CASE WHEN cardinality(b.variant_ids) > 0 THEN v.id = ANY(b.variant_ids) ELSE v.gene_id = ANY(b.gene_ids) END`; const DSET = sql`dset AS ( SELECT DISTINCT bid, drug_id FROM ( SELECT b.id AS bid, t AS drug_id FROM b JOIN civic_evidence_items e ON ${EV_SCOPE} AND e.evidence_type = 'PREDICTIVE' CROSS JOIN LATERAL unnest(e.therapy_ids) t UNION SELECT b.id, k.target_entity_id FROM b JOIN variants v ON ${VAR_SCOPE} JOIN knowledge_edges k ON k.source_entity_type = 'variant' AND k.source_entity_id = v.id AND k.target_entity_type = 'drug' AND k.relationship_type = 'PREDICTS_RESPONSE_TO' AND k.status = 'active' ) u)`; const CSET = sql`cset AS (SELECT DISTINCT b.id AS bid, e.cancer_id FROM b JOIN civic_evidence_items e ON ${EV_SCOPE} WHERE e.cancer_id IS NOT NULL)`; const TERM_MATCH = sql`EXISTS (SELECT 1 FROM jsonb_array_elements_text(coalesce(b.measurement->'indicationTerms', '[]'::jsonb)) term WHERE a.indication ILIKE '%' || term || '%')`; const ACTIVE = sql`t.overall_status = ANY(${sql.param(ACTIVE_STATUSES)}::text[])`; /** (bid, trial_id) = intervention on a scope drug AND condition in a scope cancer; MATERIALIZED so LIMIT-ed queries never scan clinical_trials row by row. */ const TSET = sql`tset AS MATERIALIZED ( SELECT DISTINCT d.bid, ti.trial_id FROM dset d JOIN trial_interventions ti ON ti.drug_id = d.drug_id JOIN trial_conditions tc ON tc.trial_id = ti.trial_id JOIN cset c ON c.bid = d.bid AND c.cancer_id = tc.cancer_id)`; const derived = { category: 'computed_metric', formulaVersion: BIOMARKER_LINKS_FORMULA, methodology: '/methodology#biomarkers', doc: 'docs/methodology/biomarkers.md' } as const; function shapeBiomarker(r: Record) { const { gene_ids, variant_ids, measurement, ...rest } = r; const m = (measurement ?? {}) as Record; return { ...camel(rest), aliases: (m.aliases as string[] | undefined) ?? [], tumorAgnosticCurated: m.tumorAgnostic === true, ncit: m.ncit ?? null, verification: m.verification ?? null, measurement: m, scope: { geneIds: gene_ids ?? [], variantIds: variant_ids ?? [] } }; } export const biomarkerRoutes: FastifyPluginAsyncZod = async (app) => { app.get( '/biomarkers', { schema: { tags: ['biomarkers'], summary: 'List canonical biomarkers with derived link counts (cancers, drugs, approvals, active trials)', querystring: z.object({ kind: z.enum(KINDS).optional().describe('Biomarker kind'), q: z.string().trim().min(1).max(80).optional().describe('Name, alias, gene symbol or NCIt code'), ...pageQuery }), response: ok(AnyList, true), }, }, async (req) => { const q = req.query; const conds = [sql`true`]; if (q.kind) conds.push(sql`b.kind = ${q.kind}`); if (q.q) { const like = `%${q.q}%`; conds.push(sql`(b.name ILIKE ${like} OR b.slug ILIKE ${like} OR b.ncit_code ILIKE ${q.q + '%'} OR EXISTS (SELECT 1 FROM jsonb_array_elements_text(coalesce(b.measurement->'aliases', '[]'::jsonb)) a WHERE a ILIKE ${like}) OR EXISTS (SELECT 1 FROM genes g WHERE (g.id = b.gene_id OR g.symbol IN (SELECT jsonb_array_elements_text(coalesce(b.measurement->'genes', '[]'::jsonb)))) AND g.symbol ILIKE ${q.q + '%'}))`); } const rows = await app.db.execute & { total: string }>(sql` ${scopeCte(sql.join(conds, sql` AND `))}, ${DSET}, ${CSET}, ${TSET} SELECT b.*, count(*) OVER() AS total, coalesce((SELECT array_agg(g.symbol ORDER BY g.symbol) FROM genes g WHERE g.id = ANY(b.gene_ids)), '{}') AS gene_symbols, (SELECT count(*) FROM cset c WHERE c.bid = b.id) AS cancers_n, (SELECT count(*) FROM dset d WHERE d.bid = b.id) AS drugs_n, (SELECT count(*) FROM drug_approvals a WHERE a.drug_id IN (SELECT d.drug_id FROM dset d WHERE d.bid = b.id) OR ${TERM_MATCH}) AS approvals_n, (SELECT count(*) FROM drug_approvals a WHERE a.tumor_agnostic AND (a.drug_id IN (SELECT d.drug_id FROM dset d WHERE d.bid = b.id) OR ${TERM_MATCH})) AS tumor_agnostic_n, (SELECT count(*) FROM tset x JOIN clinical_trials t ON t.id = x.trial_id WHERE x.bid = b.id AND ${ACTIVE}) AS active_trials_n FROM b ORDER BY b.kind, b.name LIMIT ${q.limit} OFFSET ${q.offset}`); const total = rows.length ? num(rows[0]!.total) : 0; const data = rows.map((r) => { const { total: _t, gene_symbols, cancers_n, drugs_n, approvals_n, tumor_agnostic_n, active_trials_n, ...rest } = r; return { ...shapeBiomarker(rest), geneSymbols: gene_symbols ?? [], derived: { cancers: num(cancers_n), drugs: num(drugs_n), approvals: num(approvals_n), tumorAgnosticApprovals: num(tumor_agnostic_n), activeTrials: num(active_trials_n), ...derived }, }; }); return respond(app, data, ['ncit-evs', 'hgnc', 'civic', 'clinicaltrials', 'openfda', 'health-canada-dpd'], paginate(total, q.limit, q.offset)); }, ); app.get( '/biomarkers/:slug', { schema: { tags: ['biomarkers'], summary: 'Biomarker: curated metadata + derived cancers, drugs (direction/level), approvals (tumor-agnostic flagged), trials and literature', params: z.object({ slug: z.string().min(1).describe('Biomarker slug or CI-BIO-… id') }), querystring: z.object({ trialLimit: z.coerce.number().int().min(0).max(200).default(50), publicationLimit: z.coerce.number().int().min(0).max(200).default(50) }), response: ok(AnyRecord), }, }, async (req) => { const ref = req.params.slug.trim(); const where = isCiId(ref, 'BIO') ? sql`b.id = ${ref}` : sql`b.slug = ${ref.toLowerCase()}`; const head = await app.db.execute>(sql`${scopeCte(where)} SELECT b.*, coalesce((SELECT array_agg(g.symbol ORDER BY g.symbol) FROM genes g WHERE g.id = ANY(b.gene_ids)), '{}') AS gene_symbols FROM b LIMIT 1`); const row = head[0]; if (!row) throw new NotFound('biomarker', ref); const id = row.id as string; const geneIds = (row.gene_ids as string[]) ?? []; const variantIds = (row.variant_ids as string[]) ?? []; const byId = scopeCte(sql`b.id = ${id}`); const db = app.db; const lit = variantIds.length ? { type: 'variant', ids: variantIds } : { type: 'gene', ids: geneIds }; const [genes, variants, cancers, drugs, approvals, trialCounts, trials, pubs] = await Promise.all([ geneIds.length ? db.execute>(sql`SELECT id, symbol, name, hgnc_id, is_cancer_gene FROM genes WHERE id = ANY(${sql.param(geneIds)}::text[]) ORDER BY symbol`) : Promise.resolve([]), variantIds.length ? db.execute>(sql`SELECT id, slug, name, gene_symbol, variant_type, civic_variant_id FROM variants WHERE id = ANY(${sql.param(variantIds)}::text[]) ORDER BY gene_symbol, name`) : Promise.resolve([]), db.execute>(sql` ${byId} SELECT c.id, c.slug, c.canonical_name AS name, count(*) AS n, count(*) FILTER (WHERE e.evidence_type = 'PREDICTIVE') AS predictive, count(*) FILTER (WHERE e.evidence_type = 'PROGNOSTIC') AS prognostic, count(*) FILTER (WHERE e.evidence_type = 'DIAGNOSTIC') AS diagnostic, count(*) FILTER (WHERE e.evidence_level = 'A') AS level_a, count(*) FILTER (WHERE e.evidence_level = 'B') AS level_b, count(*) FILTER (WHERE e.evidence_level = 'C') AS level_c, count(*) FILTER (WHERE e.evidence_level = 'D') AS level_d, count(*) FILTER (WHERE e.evidence_level = 'E') AS level_e, min(p.source_id) AS source_id FROM b JOIN civic_evidence_items e ON ${EV_SCOPE} JOIN cancers c ON c.id = e.cancer_id LEFT JOIN provenance p ON p.id = e.provenance_id GROUP BY c.id, c.slug, c.canonical_name ORDER BY n DESC, c.canonical_name LIMIT 200`), db.execute>(sql` ${byId}, ${DSET}, ev AS (SELECT t AS drug_id, e.significance, e.evidence_level, e.cancer_id, e.civic_id FROM b JOIN civic_evidence_items e ON ${EV_SCOPE} AND e.evidence_type = 'PREDICTIVE' CROSS JOIN LATERAL unnest(e.therapy_ids) t), ke AS (SELECT k.target_entity_id AS drug_id, k.direction, k.evidence_level, k.cancer_context_ids FROM b JOIN variants v ON ${VAR_SCOPE} JOIN knowledge_edges k ON k.source_entity_type = 'variant' AND k.source_entity_id = v.id AND k.target_entity_type = 'drug' AND k.relationship_type = 'PREDICTS_RESPONSE_TO' AND k.status = 'active') SELECT d.id, d.slug, d.name, d.kind, (SELECT count(*) FROM ev WHERE ev.drug_id = d.id) AS evidence_n, (SELECT count(*) FROM ev WHERE ev.drug_id = d.id AND ev.significance = 'SENSITIVITYRESPONSE') AS sensitivity, (SELECT count(*) FROM ev WHERE ev.drug_id = d.id AND ev.significance = 'RESISTANCE') AS resistance, (SELECT min(ev.evidence_level) FROM ev WHERE ev.drug_id = d.id) AS best_level, (SELECT array_agg(DISTINCT ev.civic_id) FROM ev WHERE ev.drug_id = d.id) AS civic_ids, (SELECT count(*) FROM ke WHERE ke.drug_id = d.id) AS edge_n, (SELECT count(*) FROM ke WHERE ke.drug_id = d.id AND ke.direction = 'sensitivity') AS edge_sensitivity, (SELECT count(*) FROM ke WHERE ke.drug_id = d.id AND ke.direction = 'resistance') AS edge_resistance, coalesce((SELECT array_agg(DISTINCT ev.cancer_id) FILTER (WHERE ev.cancer_id IS NOT NULL) FROM ev WHERE ev.drug_id = d.id), '{}') AS cancer_ids FROM dset JOIN drugs d ON d.id = dset.drug_id ORDER BY evidence_n DESC, edge_n DESC, d.name`), db.execute>(sql` ${byId}, ${DSET} SELECT a.id, a.drug_id, d.slug AS drug_slug, d.name AS drug_name, a.cancer_id, c.slug AS cancer_slug, c.canonical_name AS cancer_name, a.tumor_agnostic, a.jurisdiction, a.authority, a.indication, a.line_of_therapy, a.disease_stage, a.approval_type, a.accelerated, a.conditional, a.approval_date, a.withdrawal_date, a.status, a.application_number, a.source_id, p.source_url, p.retrieved_at, p.dataset_version, CASE WHEN a.drug_id IN (SELECT d2.drug_id FROM dset d2) AND ${TERM_MATCH} THEN 'both' WHEN a.drug_id IN (SELECT d2.drug_id FROM dset d2) THEN 'drug' ELSE 'indication' END AS matched_by FROM b, drug_approvals a JOIN drugs d ON d.id = a.drug_id LEFT JOIN cancers c ON c.id = a.cancer_id LEFT JOIN provenance p ON p.id = a.provenance_id WHERE a.drug_id IN (SELECT d2.drug_id FROM dset d2) OR ${TERM_MATCH} ORDER BY a.tumor_agnostic DESC, d.name, a.jurisdiction, a.approval_date DESC NULLS LAST LIMIT 500`), db.execute<{ total: string; active: string; recruiting: string; phase3: string }>(sql` ${byId}, ${DSET}, ${CSET}, ${TSET} SELECT count(*) AS total, count(*) FILTER (WHERE ${ACTIVE}) AS active, count(*) FILTER (WHERE t.overall_status = 'RECRUITING') AS recruiting, count(*) FILTER (WHERE ${ACTIVE} AND 'PHASE3' = ANY(t.phases)) AS phase3 FROM tset x JOIN clinical_trials t ON t.id = x.trial_id`), req.query.trialLimit ? db.execute>(sql` ${byId}, ${DSET}, ${CSET}, ${TSET} SELECT t.id, t.nct_id, t.brief_title, t.phases, t.overall_status, t.study_type, t.start_date, t.enrollment_count, t.lead_sponsor, t.countries, t.last_update_posted_date FROM tset x JOIN clinical_trials t ON t.id = x.trial_id WHERE ${ACTIVE} ORDER BY t.last_update_posted_date DESC NULLS LAST, t.nct_id LIMIT ${req.query.trialLimit}`) : Promise.resolve([]), req.query.publicationLimit && lit.ids.length ? db.execute>(sql` SELECT p.id, p.pmid, p.doi, p.title, p.journal, p.pub_year, p.retracted, x.method, x.confidence, x.edge_status, x.source_id FROM (SELECT DISTINCT ON (e.publication_id) e.publication_id, e.method, e.confidence, e.status AS edge_status, e.source_id FROM publication_entity_edges e WHERE e.entity_type = ${lit.type} AND e.entity_id = ANY(${sql.param(lit.ids)}::text[]) AND e.status <> 'rejected' ORDER BY e.publication_id, e.status) x JOIN publications p ON p.id = x.publication_id ORDER BY p.pub_year DESC NULLS LAST, p.id LIMIT ${req.query.publicationLimit}`) : Promise.resolve([]), ]); const tc = trialCounts[0] ?? { total: '0', active: '0', recruiting: '0', phase3: '0' }; const tumorAgnosticRows = approvals.filter((a) => a.tumor_agnostic === true); const { gene_symbols, ...restRow } = row; const data = { ...shapeBiomarker(restRow), geneSymbols: gene_symbols ?? [], genes: camelRows(genes), variants: camelRows(variants), noScopeReason: geneIds.length || variantIds.length ? null : (((row.measurement as Record | null)?.notes as string | undefined) ?? 'Not anchored to a gene or variant entity; gene-derived links cannot be computed.'), tumorAgnostic: { curated: ((row.measurement as Record | null)?.tumorAgnostic ?? false) === true, approvalRows: tumorAgnosticRows.map((a) => ({ id: a.id, drug: { id: a.drug_id, slug: a.drug_slug, name: a.drug_name }, jurisdiction: a.jurisdiction, authority: a.authority, approvalDate: a.approval_date, status: a.status, matchedBy: a.matched_by, category: 'regulatory_status' })), rule: 'Rows of drug_approvals with tumor_agnostic = true (source flag) reached through the derived drug set or an indication-text match; nothing is asserted without a row.', }, cancers: cancers.map((r) => ({ cancer: { id: r.id, slug: r.slug, name: r.name }, evidenceItems: num(r.n), byType: { predictive: num(r.predictive), prognostic: num(r.prognostic), diagnostic: num(r.diagnostic) }, byLevel: { A: num(r.level_a), B: num(r.level_b), C: num(r.level_c), D: num(r.level_d), E: num(r.level_e) }, sourceId: r.source_id, category: 'curated_evidence', })), drugs: drugs.map((r) => ({ drug: { id: r.id, slug: r.slug, name: r.name, kind: r.kind }, civic: { evidenceItems: num(r.evidence_n), sensitivity: num(r.sensitivity), resistance: num(r.resistance), bestLevel: r.best_level ?? null, civicEvidenceIds: r.civic_ids ?? [] }, knowledgeEdges: { edges: num(r.edge_n), sensitivity: num(r.edge_sensitivity), resistance: num(r.edge_resistance) }, cancerIds: r.cancer_ids ?? [], category: 'curated_evidence', })), approvals: approvals.map((r) => { const { drug_slug, drug_name, cancer_slug, cancer_name, source_url, retrieved_at, dataset_version, source_id, matched_by, ...rest } = r; return { ...camel(rest), drug: { id: r.drug_id, slug: drug_slug, name: drug_name }, cancer: r.cancer_id ? { id: r.cancer_id, slug: cancer_slug, name: cancer_name } : null, matchedBy: matched_by, provenance: { sourceId: source_id, url: source_url, retrievedAt: retrieved_at, datasetVersion: dataset_version, category: 'regulatory_status' } }; }), trials: { counts: { total: num(tc.total), active: num(tc.active), recruiting: num(tc.recruiting), phase3Active: num(tc.phase3) }, activeStatuses: ACTIVE_STATUSES, rows: camelRows(trials), category: 'published_evidence', }, publications: { entityType: lit.type, rows: pubs.map((r) => { const { method, confidence, edge_status, source_id, ...rest } = r; return { ...camel(rest), edge: { method, confidence, status: edge_status, sourceId: source_id } }; }) }, derived: { cancers: cancers.length, drugs: drugs.length, approvals: approvals.length, tumorAgnosticApprovals: tumorAgnosticRows.length, activeTrials: num(tc.active), ...derived }, }; return respond(app, data, ['ncit-evs', 'hgnc', 'civic', 'clinicaltrials', 'pubmed', ...pluck(cancers, 'source_id'), ...pluck(approvals, 'source_id'), ...pluck(pubs, 'source_id')]); }, ); };