import { sql } from 'drizzle-orm';
import { type Database, metricDefinitions } from '@cancerindex/database';
import { persistSnapshot, type RankingResult, type Scope } from './engine.js';
import type { RankInput } from './rank.js';
import { STOP_REASON_RULES_VERSION, classifyStopReason } from './trial-stop-reasons.js';

// The stop-reason classifier is part of this work package's public surface (API + web reuse it).
export * from './trial-stop-reasons.js';

export const TRIAL_INTELLIGENCE_FORMULA_VERSION = 'ci-trial-intel-v1';

/** Registry statuses counted as "active" (same list as counters.ts so numbers match entity_counters). */
export const TRIAL_INTEL_ACTIVE_STATUSES = ['RECRUITING', 'NOT_YET_RECRUITING', 'ENROLLING_BY_INVITATION', 'ACTIVE_NOT_RECRUITING'] as const;
// Inline literal array: drizzle expands a JS array parameter into a ($1,$2,…) tuple, which breaks ANY().
const ACTIVE_SQL = `ARRAY[${TRIAL_INTEL_ACTIVE_STATUSES.map((s) => `'${s}'`).join(',')}]::text[]`;

/** Every threshold used by the formulas — persisted in `inputs.thresholds` of each row. */
export const TRIAL_INTEL_THRESHOLDS = {
  /** trial_growth_yoy is null when fewer studies were first posted in the prior 12-month window. */
  growthMinPriorTrials: 20,
  /** sponsor_hhi is null when fewer active interventional studies. */
  hhiMinActiveTrials: 10,
  /** termination_share is null when fewer terminal studies (completed + terminated + withdrawn). */
  terminationMinTerminalTrials: 30,
  /** termination_share only considers studies first posted on/after this date. */
  terminationSince: '2010-01-01',
  /** burden-normalized ratios require at least this many annual deaths. */
  burdenMinDeaths: 100,
  /** NCIt-hierarchy traversal depth for descendants (same as counters.ts). */
  maxHierarchyDepth: 12,
  /** Burden scope for trials_per_1000_deaths / trials_per_100k_cases. */
  burdenGeography: 'USA',
} as const;

export interface TrialIntelligenceResult {
  rows: number;
  topRows: number;
  allRows: number;
  stopReasonsClassified: number;
  rankings: RankingResult[];
  ms: number;
}

/**
 * Growth windows as ISO dates for a reference day (pure, unit-tested):
 *   new_trials_12m       ← first_posted_date ∈ [asOf − 12 months, asOf)
 *   new_trials_prior_12m ← first_posted_date ∈ [asOf − 24 months, asOf − 12 months)
 */
export function growthWindows(asOf: string): { asOf: string; new12m: { from: string; to: string }; prior12m: { from: string; to: string } } {
  const d = new Date(`${asOf}T00:00:00Z`);
  if (!/^\d{4}-\d{2}-\d{2}$/.test(asOf) || Number.isNaN(d.getTime())) throw new Error(`growthWindows: invalid asOf date "${asOf}"`);
  const minus = (months: number) => {
    const x = new Date(d);
    const day = x.getUTCDate();
    x.setUTCDate(1);
    x.setUTCMonth(x.getUTCMonth() - months);
    const last = new Date(Date.UTC(x.getUTCFullYear(), x.getUTCMonth() + 1, 0)).getUTCDate();
    x.setUTCDate(Math.min(day, last));
    return x.toISOString().slice(0, 10);
  };
  const m12 = minus(12);
  const m24 = minus(24);
  return { asOf, new12m: { from: m12, to: asOf }, prior12m: { from: m24, to: m12 } };
}

/** (new − prior) / prior, null under the eligibility threshold (pure, unit-tested). */
export function growthYoy(new12: number, prior12: number, minPrior: number = TRIAL_INTEL_THRESHOLDS.growthMinPriorTrials): number | null {
  if (!Number.isFinite(new12) || !Number.isFinite(prior12) || prior12 < minPrior || prior12 <= 0) return null;
  return (new12 - prior12) / prior12;
}

/** Herfindahl–Hirschman index of a count distribution: Σ (nᵢ / Σn)² ∈ (0, 1]; null when empty (pure, unit-tested). */
export function hhi(counts: Iterable<number>): number | null {
  let total = 0;
  let sq = 0;
  for (const n of counts) {
    if (!Number.isFinite(n) || n < 0) continue;
    total += n;
    sq += n * n;
  }
  return total > 0 ? sq / (total * total) : null;
}

/** (terminated + withdrawn) / (completed + terminated + withdrawn); null under the threshold (pure, unit-tested). */
export function terminationShare(completed: number, terminated: number, withdrawn: number, minTerminal: number = TRIAL_INTEL_THRESHOLDS.terminationMinTerminalTrials): number | null {
  const denom = completed + terminated + withdrawn;
  if (denom < minTerminal || denom <= 0) return null;
  return (terminated + withdrawn) / denom;
}

/** active / (deaths / 1000) and active / (incidence / 100 000); null when deaths are under the threshold (pure, unit-tested). */
export function burdenNormalized(active: number, deaths: number | null, incidence: number | null, minDeaths: number = TRIAL_INTEL_THRESHOLDS.burdenMinDeaths): { per1000Deaths: number | null; per100kCases: number | null } {
  if (deaths == null || !Number.isFinite(deaths) || deaths < minDeaths) return { per1000Deaths: null, per100kCases: null };
  return {
    per1000Deaths: active / (deaths / 1000),
    per100kCases: incidence != null && Number.isFinite(incidence) && incidence > 0 ? active / (incidence / 100_000) : null,
  };
}

/**
 * Recompute `trial_intelligence` for every top-level cancer (entity_level = 'top') and every active
 * malignant entity with ≥ 1 mapped trial (entity_level = 'all'). Trials attach to a cancer through
 * `trial_conditions.cancer_id` over its NCIt-hierarchy descendants (depth ≤ 12, DISTINCT trials) —
 * exactly like entity_counters. Counts are over interventional studies. Set-based SQL over temp
 * tables; one transaction; deterministic for a given `asOf` day. Rankings for the four
 * trial-intelligence metrics are refreshed at the end.
 */
export async function computeTrialIntelligence(db: Database, opts: { asOf?: string } = {}): Promise<TrialIntelligenceResult> {
  const t0 = Date.now();
  const asOf = opts.asOf ?? new Date().toISOString().slice(0, 10);
  const windows = growthWindows(asOf);
  const th = TRIAL_INTEL_THRESHOLDS;

  // 1. Classify registrant-reported stop reasons in TypeScript (pure rules), to be joined in SQL.
  const stopped = await db.execute<{ id: string; why_stopped: string | null }>(sql`
    SELECT id, why_stopped FROM clinical_trials WHERE overall_status IN ('TERMINATED','WITHDRAWN','SUSPENDED')`);
  const stopIds: string[] = [];
  const stopCats: string[] = [];
  for (const r of stopped) {
    stopIds.push(r.id);
    stopCats.push(classifyStopReason(r.why_stopped).category);
  }

  const { top, all } = await db.transaction(async (tx) => {
    await tx.execute(sql`CREATE TEMP TABLE _ti_desc (ancestor varchar(32), descendant varchar(32)) ON COMMIT DROP`);
    await tx.execute(sql`
      INSERT INTO _ti_desc
      WITH RECURSIVE d AS (
        SELECT id AS ancestor, id AS descendant, 0 AS depth FROM cancers WHERE status = 'active'
        UNION
        SELECT d.ancestor, h.child_id, d.depth + 1 FROM d JOIN cancer_hierarchy h ON h.parent_id = d.descendant
        WHERE d.depth < ${sql.raw(String(th.maxHierarchyDepth))}
      )
      SELECT DISTINCT ancestor, descendant FROM d`);
    await tx.execute(sql`CREATE INDEX ON _ti_desc (descendant)`);

    await tx.execute(sql`CREATE TEMP TABLE _ti_stop (trial_id varchar(32) PRIMARY KEY, category text NOT NULL) ON COMMIT DROP`);
    const chunk = 20_000;
    for (let i = 0; i < stopIds.length; i += chunk) {
      await tx.execute(sql`INSERT INTO _ti_stop (trial_id, category) SELECT * FROM unnest(${sql.param(stopIds.slice(i, i + chunk))}::text[], ${sql.param(stopCats.slice(i, i + chunk))}::text[])`);
    }

    // 2. cancer × trial map (DISTINCT over descendants) with the trial fields the formulas need.
    await tx.execute(sql`
      CREATE TEMP TABLE _ti_map ON COMMIT DROP AS
      SELECT m.cancer_id, t.id AS trial_id,
        t.study_type = 'INTERVENTIONAL' AS interventional,
        t.overall_status AS status,
        t.overall_status = ANY(${sql.raw(ACTIVE_SQL)}) AS active,
        t.phases, t.first_posted_date::date AS first_posted, t.enrollment_count, t.lead_sponsor, t.lead_sponsor_class, t.countries, t.has_results
      FROM (SELECT DISTINCT d.ancestor AS cancer_id, tc.trial_id FROM trial_conditions tc JOIN _ti_desc d ON d.descendant = tc.cancer_id WHERE tc.cancer_id IS NOT NULL) m
      JOIN clinical_trials t ON t.id = m.trial_id`);
    await tx.execute(sql`CREATE INDEX ON _ti_map (cancer_id)`);

    // 3. Per-cancer aggregates (computed once, inserted for each entity level the cancer belongs to).
    await tx.execute(sql`
      CREATE TEMP TABLE _ti_agg ON COMMIT DROP AS
      WITH base AS (
        SELECT cancer_id,
          count(*) AS mapped_any_type,
          count(*) FILTER (WHERE interventional) AS total_trials,
          count(*) FILTER (WHERE interventional AND active) AS active_trials,
          count(*) FILTER (WHERE interventional AND status = 'RECRUITING') AS recruiting_trials,
          count(*) FILTER (WHERE interventional AND active AND 'PHASE1' = ANY(phases)) AS phase1_active,
          count(*) FILTER (WHERE interventional AND active AND 'PHASE2' = ANY(phases)) AS phase2_active,
          count(*) FILTER (WHERE interventional AND active AND 'PHASE3' = ANY(phases)) AS phase3_active,
          count(*) FILTER (WHERE interventional AND status = 'RECRUITING' AND 'PHASE3' = ANY(phases)) AS phase3_recruiting,
          count(*) FILTER (WHERE interventional AND active AND 'PHASE4' = ANY(phases)) AS phase4_active,
          count(*) FILTER (WHERE interventional AND status = 'COMPLETED') AS completed_trials,
          count(*) FILTER (WHERE interventional AND status = 'TERMINATED') AS terminated_trials,
          count(*) FILTER (WHERE interventional AND status = 'WITHDRAWN') AS withdrawn_trials,
          count(*) FILTER (WHERE interventional AND status = 'SUSPENDED') AS suspended_trials,
          count(*) FILTER (WHERE interventional AND has_results) AS with_results,
          count(*) FILTER (WHERE interventional AND first_posted >= ${windows.new12m.from}::date AND first_posted < ${windows.new12m.to}::date) AS new_trials_12m,
          count(*) FILTER (WHERE interventional AND first_posted >= ${windows.prior12m.from}::date AND first_posted < ${windows.prior12m.to}::date) AS new_trials_prior_12m,
          avg(enrollment_count) FILTER (WHERE interventional AND active) AS avg_enrollment,
          percentile_cont(0.5) WITHIN GROUP (ORDER BY enrollment_count) FILTER (WHERE interventional AND active AND enrollment_count IS NOT NULL) AS median_enrollment,
          sum(enrollment_count) FILTER (WHERE interventional AND active) AS total_enrollment_active,
          count(*) FILTER (WHERE interventional AND active AND lead_sponsor_class = 'INDUSTRY') AS industry_active,
          count(*) FILTER (WHERE interventional AND active AND 'United States' = ANY(countries)) AS us_active,
          count(*) FILTER (WHERE interventional AND first_posted >= ${th.terminationSince}::date AND status = 'COMPLETED') AS term_completed,
          count(*) FILTER (WHERE interventional AND first_posted >= ${th.terminationSince}::date AND status = 'TERMINATED') AS term_terminated,
          count(*) FILTER (WHERE interventional AND first_posted >= ${th.terminationSince}::date AND status = 'WITHDRAWN') AS term_withdrawn
        FROM _ti_map GROUP BY cancer_id
      ),
      sp AS (
        SELECT cancer_id, lead_sponsor, count(*) AS n FROM _ti_map WHERE interventional AND active AND lead_sponsor IS NOT NULL GROUP BY 1, 2
      ),
      spa AS (
        SELECT cancer_id, count(*) AS distinct_sponsors, sum(n) AS sponsor_denominator, sum(n * n)::double precision / (sum(n) * sum(n)) AS sponsor_hhi,
          (array_agg(lead_sponsor ORDER BY n DESC, lead_sponsor))[1] AS top_sponsor, max(n) AS top_sponsor_n
        FROM sp GROUP BY cancer_id
      ),
      co AS (
        SELECT cancer_id, c AS country, count(*) AS n FROM _ti_map, unnest(countries) c WHERE interventional AND active GROUP BY 1, 2
      ),
      coa AS (
        SELECT cancer_id, count(*) AS distinct_countries, sum(n) AS country_pairs, sum(n * n)::double precision / (sum(n) * sum(n)) AS country_hhi,
          (array_agg(country ORDER BY n DESC, country))[1] AS top_country, max(n) AS top_country_n
        FROM co GROUP BY cancer_id
      ),
      ws AS (
        SELECT m.cancer_id, s.category, count(*) AS n FROM _ti_map m JOIN _ti_stop s ON s.trial_id = m.trial_id
        WHERE m.interventional AND m.status IN ('TERMINATED','WITHDRAWN','SUSPENDED') GROUP BY 1, 2
      ),
      wsa AS (
        SELECT cancer_id, jsonb_object_agg(category, n ORDER BY category) AS why_stopped_breakdown FROM ws GROUP BY cancer_id
      ),
      bur AS (
        SELECT DISTINCT ON (x.cancer_id) x.* FROM (
          SELECT m.cancer_id, m.year, m.source_id, m.value AS deaths, i.value AS incidence, m.id AS mortality_obs_id, i.id AS incidence_obs_id
          FROM epidemiology_observations m
          JOIN geographies g ON g.id = m.geography_id AND g.iso3 = ${th.burdenGeography}
          JOIN epidemiology_observations i ON i.cancer_id = m.cancer_id AND i.geography_id = m.geography_id AND i.year = m.year AND i.source_id = m.source_id
            AND i.sex = m.sex AND i.age_group = m.age_group AND i.metric = 'incidence_count'
          WHERE m.metric = 'mortality_count' AND m.sex = 'all' AND m.age_group = 'all' AND m.value >= ${sql.raw(String(th.burdenMinDeaths))}
        ) x ORDER BY x.cancer_id, x.year DESC, x.source_id, x.mortality_obs_id, x.incidence_obs_id
      )
      SELECT c.id AS cancer_id, c.top_level, c.malignant,
        b.mapped_any_type, b.total_trials, b.active_trials, b.recruiting_trials, b.phase1_active, b.phase2_active, b.phase3_active, b.phase3_recruiting, b.phase4_active,
        b.completed_trials, b.terminated_trials, b.withdrawn_trials, b.suspended_trials, b.with_results,
        b.new_trials_12m, b.new_trials_prior_12m,
        CASE WHEN b.new_trials_prior_12m >= ${sql.raw(String(th.growthMinPriorTrials))} THEN (b.new_trials_12m - b.new_trials_prior_12m)::double precision / b.new_trials_prior_12m END AS trial_growth_yoy,
        b.avg_enrollment, b.median_enrollment, b.total_enrollment_active,
        COALESCE(s.distinct_sponsors, 0) AS distinct_sponsors,
        CASE WHEN b.active_trials > 0 THEN b.industry_active::double precision / b.active_trials END AS industry_share,
        CASE WHEN b.active_trials >= ${sql.raw(String(th.hhiMinActiveTrials))} THEN s.sponsor_hhi END AS sponsor_hhi,
        s.top_sponsor,
        CASE WHEN s.sponsor_denominator > 0 THEN s.top_sponsor_n::double precision / s.sponsor_denominator END AS top_sponsor_share,
        COALESCE(s.sponsor_denominator, 0) AS sponsor_denominator,
        COALESCE(g.distinct_countries, 0) AS distinct_countries,
        CASE WHEN b.active_trials > 0 THEN b.us_active::double precision / b.active_trials END AS us_share,
        g.top_country,
        CASE WHEN b.active_trials > 0 THEN g.top_country_n::double precision / b.active_trials END AS top_country_share,
        g.country_hhi,
        COALESCE(g.country_pairs, 0) AS country_pairs,
        b.term_completed, b.term_terminated, b.term_withdrawn,
        CASE WHEN (b.term_completed + b.term_terminated + b.term_withdrawn) >= ${sql.raw(String(th.terminationMinTerminalTrials))}
          THEN (b.term_terminated + b.term_withdrawn)::double precision / (b.term_completed + b.term_terminated + b.term_withdrawn) END AS termination_share,
        COALESCE(w.why_stopped_breakdown, '{}'::jsonb) AS why_stopped_breakdown,
        CASE WHEN c.top_level THEN b.active_trials::double precision / (r.deaths / 1000) END AS trials_per_1000_deaths,
        CASE WHEN c.top_level AND r.incidence > 0 THEN b.active_trials::double precision / (r.incidence / 100000) END AS trials_per_100k_cases,
        CASE WHEN c.top_level AND r.cancer_id IS NOT NULL THEN ${th.burdenGeography}::text END AS burden_geography,
        CASE WHEN c.top_level THEN r.year END AS burden_year,
        CASE WHEN c.top_level THEN r.source_id END AS burden_source_id,
        CASE WHEN c.top_level THEN r.deaths END AS burden_deaths,
        CASE WHEN c.top_level THEN r.incidence END AS burden_incidence,
        CASE WHEN c.top_level THEN r.mortality_obs_id END AS mortality_obs_id,
        CASE WHEN c.top_level THEN r.incidence_obs_id END AS incidence_obs_id
      FROM cancers c
      JOIN base b ON b.cancer_id = c.id
      LEFT JOIN spa s ON s.cancer_id = c.id
      LEFT JOIN coa g ON g.cancer_id = c.id
      LEFT JOIN wsa w ON w.cancer_id = c.id
      LEFT JOIN bur r ON r.cancer_id = c.id
      WHERE c.status = 'active'`);

    const inputsJson = sql`jsonb_strip_nulls(jsonb_build_object(
        'asOf', ${asOf}::text,
        'windows', ${JSON.stringify({ new12m: windows.new12m, prior12m: windows.prior12m })}::jsonb,
        'thresholds', ${JSON.stringify(th)}::jsonb,
        'activeStatuses', ${JSON.stringify(TRIAL_INTEL_ACTIVE_STATUSES)}::jsonb,
        'stopReasonRulesVersion', ${STOP_REASON_RULES_VERSION}::text,
        'aggregation', 'descendants',
        'studyType', 'INTERVENTIONAL',
        'mappedTrialsAnyType', a.mapped_any_type,
        'denominators', jsonb_build_object('sponsor', a.sponsor_denominator, 'countryPairs', a.country_pairs, 'terminal', a.term_completed + a.term_terminated + a.term_withdrawn, 'terminalCompleted', a.term_completed, 'terminalTerminated', a.term_terminated, 'terminalWithdrawn', a.term_withdrawn),
        'burden', CASE WHEN a.burden_source_id IS NOT NULL THEN jsonb_build_object('geography', a.burden_geography, 'year', a.burden_year, 'sourceId', a.burden_source_id, 'deaths', a.burden_deaths, 'incidence', a.burden_incidence, 'mortalityObservationId', a.mortality_obs_id, 'incidenceObservationId', a.incidence_obs_id, 'sex', 'all', 'ageGroup', 'all') END
      ))`;

    const insertFor = (level: 'top' | 'all') => sql`
      INSERT INTO trial_intelligence (cancer_id, entity_level, total_trials, active_trials, recruiting_trials, phase1_active, phase2_active, phase3_active, phase3_recruiting, phase4_active,
        completed_trials, terminated_trials, withdrawn_trials, suspended_trials, with_results, new_trials_12m, new_trials_prior_12m, trial_growth_yoy,
        avg_enrollment, median_enrollment, total_enrollment_active, distinct_sponsors, industry_share, sponsor_hhi, top_sponsor, top_sponsor_share,
        distinct_countries, us_share, top_country, top_country_share, country_hhi, termination_share, why_stopped_breakdown,
        trials_per_1000_deaths, trials_per_100k_cases, burden_geography, burden_year, burden_source_id, formula_version, inputs, updated_at)
      SELECT a.cancer_id, ${level}, a.total_trials, a.active_trials, a.recruiting_trials, a.phase1_active, a.phase2_active, a.phase3_active, a.phase3_recruiting, a.phase4_active,
        a.completed_trials, a.terminated_trials, a.withdrawn_trials, a.suspended_trials, a.with_results, a.new_trials_12m, a.new_trials_prior_12m, a.trial_growth_yoy,
        a.avg_enrollment, a.median_enrollment, a.total_enrollment_active, a.distinct_sponsors, a.industry_share, a.sponsor_hhi, a.top_sponsor, a.top_sponsor_share,
        a.distinct_countries, a.us_share, a.top_country, a.top_country_share, a.country_hhi, a.termination_share, a.why_stopped_breakdown,
        a.trials_per_1000_deaths, a.trials_per_100k_cases, a.burden_geography, a.burden_year, a.burden_source_id, ${TRIAL_INTELLIGENCE_FORMULA_VERSION}, ${inputsJson}, now()
      FROM _ti_agg a
      ${level === 'top' ? sql`WHERE a.top_level` : sql`WHERE a.malignant AND a.mapped_any_type > 0`}`;

    await tx.execute(sql`DELETE FROM trial_intelligence`);
    // Top-level cancers without any mapped trial still get a row (true zeros) so the top-level table is complete.
    await tx.execute(sql`
      INSERT INTO _ti_agg (cancer_id, top_level, malignant, mapped_any_type, total_trials, active_trials, recruiting_trials, phase1_active, phase2_active, phase3_active, phase3_recruiting, phase4_active,
        completed_trials, terminated_trials, withdrawn_trials, suspended_trials, with_results, new_trials_12m, new_trials_prior_12m, distinct_sponsors, sponsor_denominator, distinct_countries, country_pairs,
        term_completed, term_terminated, term_withdrawn, why_stopped_breakdown)
      SELECT c.id, true, c.malignant, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, '{}'::jsonb
      FROM cancers c WHERE c.status = 'active' AND c.top_level AND NOT EXISTS (SELECT 1 FROM _ti_agg a WHERE a.cancer_id = c.id)`);
    const topRows = await tx.execute<{ n: string }>(sql`WITH ins AS (${insertFor('top')} RETURNING 1) SELECT count(*)::text AS n FROM ins`);
    const allRows = await tx.execute<{ n: string }>(sql`WITH ins AS (${insertFor('all')} RETURNING 1) SELECT count(*)::text AS n FROM ins`);
    return { top: Number(topRows[0]?.n ?? 0), all: Number(allRows[0]?.n ?? 0) };
  });

  const rankings = await rankTrialIntelligence(db);
  return { rows: top + all, topRows: top, allRows: all, stopReasonsClassified: stopIds.length, rankings, ms: Date.now() - t0 };
}

interface IntelRow extends Record<string, unknown> {
  cancer_id: string;
  entity_level: 'top' | 'all';
  phase3_recruiting: number;
  active_trials: number;
  new_trials_12m: number;
  new_trials_prior_12m: number;
  trial_growth_yoy: number | null;
  termination_share: number | null;
  sponsor_hhi: number | null;
  distinct_sponsors: number;
  top_sponsor: string | null;
  top_sponsor_share: number | null;
  inputs: Record<string, unknown>;
}

/**
 * Ranking snapshots for the four trial-intelligence metrics (WORLD, all sexes/ages, latest, per entity
 * level). Only eligible entities are ranked: non-null value, count metrics > 0. `trial_termination_share`
 * ranks descending too (rank 1 = highest share; `higher_is_worse` is display information).
 */
export async function rankTrialIntelligence(db: Database): Promise<RankingResult[]> {
  const slugs = ['phase3_recruiting_trials', 'trial_growth_yoy', 'trial_termination_share', 'sponsor_concentration'];
  const defs = await db.select().from(metricDefinitions);
  const byslug = new Map(defs.filter((d) => slugs.includes(d.slug)).map((d) => [d.slug, d]));
  const rows = await db.execute<IntelRow>(sql`
    SELECT ti.cancer_id, ti.entity_level, ti.phase3_recruiting, ti.active_trials, ti.new_trials_12m, ti.new_trials_prior_12m, ti.trial_growth_yoy, ti.termination_share, ti.sponsor_hhi,
      ti.distinct_sponsors, ti.top_sponsor, ti.top_sponsor_share, ti.inputs
    FROM trial_intelligence ti JOIN cancers c ON c.id = ti.cancer_id WHERE c.status = 'active' AND ti.formula_version = ${TRIAL_INTELLIGENCE_FORMULA_VERSION}`);
  const out: RankingResult[] = [];
  for (const level of ['top', 'all'] as const) {
    const scope: Scope = { geography: 'WORLD', sex: 'all', ageGroup: 'all', year: null, entityLevel: level };
    const lv = rows.filter((r) => r.entity_level === level);
    const common = (r: IntelRow) => ({ formulaVersion: TRIAL_INTELLIGENCE_FORMULA_VERSION, asOf: r.inputs.asOf, aggregation: 'descendants', studyType: 'INTERVENTIONAL', activeStatuses: r.inputs.activeStatuses });
    const metrics: Array<{ slug: string; items: RankInput[] }> = [
      {
        slug: 'phase3_recruiting_trials',
        items: lv.filter((r) => Number(r.phase3_recruiting) > 0).map((r) => ({ id: r.cancer_id, value: Number(r.phase3_recruiting), confidence: 'HIGH' as const, inputs: { ...common(r), phase3Recruiting: Number(r.phase3_recruiting), activeTrials: Number(r.active_trials) } })),
      },
      {
        slug: 'trial_growth_yoy',
        items: lv
          .filter((r) => r.trial_growth_yoy != null && Number.isFinite(Number(r.trial_growth_yoy)))
          .map((r) => ({ id: r.cancer_id, value: Number(r.trial_growth_yoy), confidence: (Number(r.new_trials_prior_12m) >= 100 ? 'HIGH' : 'MEDIUM') as 'HIGH' | 'MEDIUM', inputs: { ...common(r), newTrials12m: Number(r.new_trials_12m), newTrialsPrior12m: Number(r.new_trials_prior_12m), windows: r.inputs.windows, minPriorTrials: TRIAL_INTEL_THRESHOLDS.growthMinPriorTrials } })),
      },
      {
        slug: 'trial_termination_share',
        items: lv
          .filter((r) => r.termination_share != null && Number.isFinite(Number(r.termination_share)))
          .map((r) => {
            const d = (r.inputs.denominators ?? {}) as Record<string, number>;
            return { id: r.cancer_id, value: Number(r.termination_share), confidence: ((d.terminal ?? 0) >= 100 ? 'HIGH' : 'MEDIUM') as 'HIGH' | 'MEDIUM', inputs: { ...common(r), completed: d.terminalCompleted, terminated: d.terminalTerminated, withdrawn: d.terminalWithdrawn, terminal: d.terminal, since: TRIAL_INTEL_THRESHOLDS.terminationSince, minTerminalTrials: TRIAL_INTEL_THRESHOLDS.terminationMinTerminalTrials } };
          }),
      },
      {
        slug: 'sponsor_concentration',
        items: lv
          .filter((r) => r.sponsor_hhi != null && Number.isFinite(Number(r.sponsor_hhi)))
          .map((r) => {
            const d = (r.inputs.denominators ?? {}) as Record<string, number>;
            return { id: r.cancer_id, value: Number(r.sponsor_hhi), confidence: (Number(r.active_trials) >= 50 ? 'HIGH' : 'MEDIUM') as 'HIGH' | 'MEDIUM', inputs: { ...common(r), activeTrials: Number(r.active_trials), activeWithSponsor: d.sponsor, distinctSponsors: Number(r.distinct_sponsors), topSponsor: r.top_sponsor, topSponsorShare: r.top_sponsor_share == null ? null : Number(r.top_sponsor_share), minActiveTrials: TRIAL_INTEL_THRESHOLDS.hhiMinActiveTrials } };
          }),
      },
    ];
    for (const m of metrics) {
      const def = byslug.get(m.slug);
      if (!def || m.items.length < 3) continue;
      out.push(await persistSnapshot(db, def, scope, m.items, { descending: true, sourceIds: def.sourceSlugs.length ? def.sourceSlugs : ['clinicaltrials'] }));
    }
  }
  return out;
}