spb/cancerindex

1import 'server-only';2import { run, sql, safe } from '@/lib/db';34export interface SourceRow {5  id: string;6  slug: string;7  name: string;8  organization: string | null;9  category: string;10  description: string | null;11  homepage: string | null;12  docs_url: string | null;13  terms_url: string | null;14  access_type: string;15  access_auth: string;16  license: string | null;17  license_status: string;18  commercial_use: string;19  redistribution: string;20  attribution: string | null;21  license_reviewed_at: Date | null;22  approved_for_production: boolean;23  update_frequency: string | null;24  supports_incremental: boolean;25  entities: string[];26  metrics: string[];27  rate_limit: string | null;28  status: string;29  tier: number;30  manifest: Record<string, unknown>;31  updated_at: Date;32  // joined33  health: string | null;34  health_detail: string | null;35  last_success_at: Date | null;36  last_attempt_at: Date | null;37  paused: boolean | null;38  record_count: string | number | null;39  last_run_id: string | null;40  last_run_status: string | null;41  last_run_finished_at: Date | null;42  last_run_fetched: number | null;43  last_run_created: number | null;44  last_run_updated: number | null;45  last_run_rejected: number | null;46  last_run_drift: unknown[] | null;47  last_dataset_version: string | null;48}4950const SOURCE_SELECT = sql`51  SELECT s.*, cc.health, cc.health_detail, cc.last_success_at, cc.last_attempt_at, cc.paused,52         (SELECT count(*) FROM source_records r WHERE r.source_id = s.id) AS record_count,53         lr.id AS last_run_id, lr.status AS last_run_status, lr.finished_at AS last_run_finished_at,54         lr.records_fetched AS last_run_fetched, lr.records_created AS last_run_created, lr.records_updated AS last_run_updated,55         lr.records_rejected AS last_run_rejected, lr.schema_drift AS last_run_drift, lr.dataset_version AS last_dataset_version56  FROM sources s57  LEFT JOIN connector_cursors cc ON cc.connector_id = s.slug58  LEFT JOIN LATERAL (SELECT * FROM ingest_runs ir WHERE ir.source_id = s.id ORDER BY ir.started_at DESC LIMIT 1) lr ON true`;5960export async function listSources(): Promise<SourceRow[]> {61  return safe(() => run<SourceRow>(sql`${SOURCE_SELECT} ORDER BY s.tier, s.category, s.name`), [] as SourceRow[]);62}6364export async function getSourceBySlug(slug: string): Promise<SourceRow | null> {65  const rows = await safe(() => run<SourceRow>(sql`${SOURCE_SELECT} WHERE s.slug = ${slug} LIMIT 1`), [] as SourceRow[]);66  return rows[0] ?? null;67}6869export interface RunRow {70  id: string;71  connector_id: string;72  source_id: string;73  mode: string;74  status: string;75  started_at: Date;76  finished_at: Date | null;77  duration_ms: number | null;78  records_fetched: number;79  records_created: number;80  records_updated: number;81  records_unchanged: number;82  records_rejected: number;83  http_requests: number;84  http_failures: number;85  rate_limit_events: number;86  validation_failures: number;87  schema_drift: unknown[];88  cursor_before: unknown;89  cursor_after: unknown;90  error: string | null;91  log: Array<{ t: string; level: string; msg: string }>;92  dataset_version: string | null;93  anomaly: string | null;94  source_slug?: string;95  source_name?: string;96}9798export async function listRuns(opts: { sourceId?: string; connectorId?: string; limit?: number } = {}): Promise<RunRow[]> {99  const limit = opts.limit ?? 25;100  return safe(101    () =>102      run<RunRow>(sql`103        SELECT r.*, s.slug AS source_slug, s.name AS source_name FROM ingest_runs r LEFT JOIN sources s ON s.id = r.source_id104        WHERE ${opts.sourceId ? sql`r.source_id = ${opts.sourceId}` : sql`true`} AND ${opts.connectorId ? sql`r.connector_id = ${opts.connectorId}` : sql`true`}105        ORDER BY r.started_at DESC LIMIT ${limit}`),106    [] as RunRow[],107  );108}109110export async function getRun(id: string): Promise<RunRow | null> {111  const rows = await safe(() => run<RunRow>(sql`SELECT r.*, s.slug AS source_slug, s.name AS source_name FROM ingest_runs r LEFT JOIN sources s ON s.id = r.source_id WHERE r.id = ${id}`), [] as RunRow[]);112  return rows[0] ?? null;113}114115export async function recordCountsByKind(sourceId: string): Promise<Array<{ entity_kind: string; n: number; status: string; last_retrieved: Date | null }>> {116  const rows = await safe(117    () => run<{ entity_kind: string; n: string; status: string; last_retrieved: Date | null }>(sql`SELECT entity_kind, status, count(*) AS n, max(retrieved_at) AS last_retrieved FROM source_records WHERE source_id = ${sourceId} GROUP BY entity_kind, status ORDER BY entity_kind, status`),118    [] as Array<{ entity_kind: string; n: string; status: string; last_retrieved: Date | null }>,119  );120  return rows.map((r) => ({ ...r, n: Number(r.n) }));121}122123/** Domains for the public coverage matrix (§307-308). */124export const COVERAGE_DOMAINS = ['Taxonomy', 'Epidemiology', 'Genomics', 'Variants', 'Trials', 'Drugs', 'Literature'] as const;125export type CoverageDomain = (typeof COVERAGE_DOMAINS)[number];126127/** Which domains a source contributes to, from its declared category/entities/metrics (manifest). */128export function sourceDomains(s: Pick<SourceRow, 'category' | 'entities' | 'metrics'>): Set<CoverageDomain> {129  const out = new Set<CoverageDomain>();130  const ents = new Set([...(s.entities ?? []), ...(s.metrics ?? [])].map((e) => e.toLowerCase()));131  const has = (...keys: string[]) => keys.some((k) => [...ents].some((e) => e.includes(k)));132  if (s.category === 'terminology' || has('cancer', 'tumor_type', 'concept', 'disease')) out.add('Taxonomy');133  if (s.category === 'epidemiology' || has('incidence', 'mortality', 'survival', 'prevalence')) out.add('Epidemiology');134  if (s.category === 'genomics' || has('gene', 'cohort', 'frequency', 'mutation')) out.add('Genomics');135  if (s.category === 'variants' || has('variant', 'evidence', 'clinical_significance')) out.add('Variants');136  if (s.category === 'trials' || has('trial', 'study')) out.add('Trials');137  if (s.category === 'drugs' || s.category === 'regulatory' || has('drug', 'therapy', 'approval')) out.add('Drugs');138  if (s.category === 'literature' || has('publication', 'literature', 'pubmed')) out.add('Literature');139  return out;140}141142/** Plain-language meaning of the license status (§308). */143export function licenseMeaning(status: string, redistribution: string, commercial: string): string {144  const base: Record<string, string> = {145    approved: 'Reviewed: CancerIndex may ingest and display this data with attribution.',146    review: 'Under review: terms are being assessed. No records are shown publicly until the review completes.',147    restricted: 'Restricted: data may be displayed with limits (e.g. no bulk redistribution).',148    blocked: 'Blocked: the license does not permit use by CancerIndex.',149  };150  const red: Record<string, string> = {151    allowed: 'Redistribution allowed.',152    attribution: 'Redistribution allowed with attribution.',153    restricted: 'Redistribution restricted — not included in downloads.',154    prohibited: 'Redistribution prohibited — not included in downloads.',155    unknown: 'Redistribution terms not yet determined.',156  };157  const com: Record<string, string> = { allowed: 'Commercial use allowed.', restricted: 'Commercial use restricted.', prohibited: 'Commercial use prohibited.', unknown: 'Commercial-use terms not yet determined.' };158  return [base[status] ?? status, red[redistribution] ?? '', com[commercial] ?? ''].filter(Boolean).join(' ');159}160