//! ingest_semisym_scar — map §0 covariant semi-symmetry receipts → ENE rows.
//!
//! SKELETON (dry-run first). Reads the per-point receipt JSONs from
//! `semisym survey` (schemas `semisym_receipt_modp_v1`, `semisym_receipt_v1`,
//! `semisym_receipt_edge_v1`) and maps each lattice point to an
//! `ene.rrc_classifications` row for later data mining of the scar map.
//!
//! NO FLOATS (OTOM doctrine — the whole stack is integer / exact / Q16_16):
//! every quantity is stored EXACTLY. Rationals (β-norm, obstruction) are kept
//! as exact strings ("−628/9"); counts/signature are exact integers; the whole
//! bundle lives in a JSONB `metrics` column. Ordering, when needed, is exact
//! integer cross-multiplication (a/b < c/d ⇔ a·d < c·b), never IEEE float.
//! FUTURE revision (per "everything is logarithms", the prime-basis log): add a
//! prime-exponent vector νₚ per rational — exact integer log-coordinates,
//! multiplicatively additive, using the precomputed prime LUT as the basis.
//!
//! SAFETY (repo dry-run-by-default rule):
//!   * default            — parse + map, PRINT a summary + WRITE SQL to a file.
//!                          NO database connection.
//!   * --apply --dsn <..> — execute the SQL via `psql`. Requires an EXPLICIT
//!                          dsn; a bare --apply refuses (no baked endpoint).
//!                          RDS / rds_connect.py are never used; ENE is the
//!                          neon `research_stack`/`ene` schema.
//!
//! Ids are sha256(equation_id) so re-ingest is idempotent (ON CONFLICT UPDATE).
//! The legacy FLOAT columns (spectral_radius, score) are deliberately NOT
//! populated — they predate the float ban and should be migrated out.
//!
//! Usage:
//!   ingest_semisym_scar <ckpt_dir> [--out scar.sql] [--limit N]          # dry-run
//!   ingest_semisym_scar <ckpt_dir> --apply --dsn "postgresql://…/research_stack"

use std::collections::BTreeMap;
use std::fs;
use std::path::PathBuf;
use std::process::Command;

use serde::Deserialize;
use serde_json::json;
use sha2::{Digest, Sha256};

/// Union of the three receipt schemas; every field optional so one struct
/// deserializes modp / exact-escalated / edge receipts alike.
#[derive(Deserialize, Default)]
struct Receipt {
    schema: Option<String>,
    m: Option<u32>,
    prime: Option<u64>,
    geometrization: Option<String>,
    point: Option<Vec<String>>,
    gprime_signature: Option<(u32, u32)>, // JSON [pos, neg]
    gprime_beta_norm: Option<String>,     // exact rational as string
    verdict: Option<String>,
    obstruction_along_drift: Option<String>, // exact rational (v1 only)
    nabla_r_certified_nonzero: Option<u64>,
    nabla_r_nonzero: Option<u64>,
    nabla_r_total: Option<u64>,
    rr_nonzero_modp: Option<u64>,
    rr_nonzero: Option<u64>,
    baseline_constant_curvature_quarter: Option<bool>,
    reason: Option<String>, // edge receipts
    sound: Option<bool>,
    modular_pole: Option<bool>,
}

/// One `ene.rrc_classifications` row. Typed columns are exact-integer/text only;
/// all numeric detail (exact) is carried in `metrics` (JSONB).
struct RrcRow {
    id: String,
    equation_id: String,
    shape: String,          // verdict
    pist_label: String,     // geometrization / edge reason
    weak_axes: Option<u32>, // negative signature count (exact integer)
    metrics: String,        // exact JSONB bundle (no floats)
}

/// Deterministic UUID-shaped id from a key (sha256 → 8-4-4-4-12 hex).
fn stable_uuid(key: &str) -> String {
    let d = Sha256::digest(key.as_bytes());
    let h: String = d[..16].iter().map(|b| format!("{b:02x}")).collect();
    format!("{}-{}-{}-{}-{}", &h[0..8], &h[8..12], &h[12..16], &h[16..20], &h[20..32])
}

fn sql_str(s: &str) -> String {
    format!("'{}'", s.replace('\'', "''"))
}
fn sql_u32(x: Option<u32>) -> String {
    x.map(|v| v.to_string()).unwrap_or_else(|| "NULL".into())
}

fn map_receipt(r: &Receipt) -> RrcRow {
    let point = r.point.clone().unwrap_or_default().join("_");
    let equation_id = format!("semisym:m{}:{}", r.m.unwrap_or(0), point);
    let id = stable_uuid(&equation_id);
    let shape = r.verdict.clone().unwrap_or_else(|| "UNKNOWN".into());
    let pist_label = r
        .geometrization
        .clone()
        .or_else(|| r.reason.clone())
        .unwrap_or_default();
    let weak_axes = r.gprime_signature.map(|(_, neg)| neg);

    // exact JSONB bundle — rationals as strings, counts/signature as integers,
    // NO floats anywhere. This is the "combine both" richer field.
    let signature = r.gprime_signature.map(|(p, n)| json!([p, n]));
    let metrics = json!({
        "schema": r.schema,
        "verdict": r.verdict,
        "geometrization": r.geometrization,
        "prime": r.prime,
        "signature": signature,
        "beta_norm": r.gprime_beta_norm,                 // EXACT rational string
        "obstruction_exact": r.obstruction_along_drift,  // exact string or null (modp: null)
        "nabla_certified": r.nabla_r_certified_nonzero.or(r.nabla_r_nonzero),
        "nabla_total": r.nabla_r_total,
        "rr_nonzero": r.rr_nonzero_modp.or(r.rr_nonzero),
        "baseline_constant_curvature_quarter": r.baseline_constant_curvature_quarter,
        "sound": r.sound,
        "point": r.point,
        "reason": r.reason,
    });
    RrcRow {
        id,
        equation_id,
        shape,
        pist_label,
        weak_axes,
        metrics: serde_json::to_string(&metrics).unwrap(),
    }
}

fn row_sql(row: &RrcRow) -> String {
    format!(
        "INSERT INTO ene.rrc_classifications \
         (id, equation_id, shape, pist_label, weak_axes, metrics) VALUES \
         ({}, {}, {}, {}, {}, {}::jsonb) \
         ON CONFLICT (id) DO UPDATE SET shape=EXCLUDED.shape, pist_label=EXCLUDED.pist_label, \
         weak_axes=EXCLUDED.weak_axes, metrics=EXCLUDED.metrics;",
        sql_str(&row.id),
        sql_str(&row.equation_id),
        sql_str(&row.shape),
        sql_str(&row.pist_label),
        sql_u32(row.weak_axes),
        sql_str(&row.metrics),
    )
}

fn main() {
    let args: Vec<String> = std::env::args().collect();
    if args.len() < 2 {
        eprintln!("usage: ingest_semisym_scar <ckpt_dir> [--out f.sql] [--limit N] [--apply --dsn <conn>]");
        std::process::exit(2);
    }
    let dir = PathBuf::from(&args[1]);
    let mut out = PathBuf::from("semisym_scar.sql");
    let mut limit: Option<usize> = None;
    let mut apply = false;
    let mut dsn: Option<String> = None;
    let mut i = 2;
    while i < args.len() {
        match args[i].as_str() {
            "--out" => { out = PathBuf::from(&args[i + 1]); i += 2; }
            "--limit" => { limit = args[i + 1].parse().ok(); i += 2; }
            "--apply" => { apply = true; i += 1; }
            "--dsn" => { dsn = Some(args[i + 1].clone()); i += 2; }
            other => { eprintln!("unknown arg {other}"); std::process::exit(2); }
        }
    }

    let mut files: Vec<PathBuf> = fs::read_dir(&dir)
        .unwrap_or_else(|_| panic!("cannot read checkpoint dir {}", dir.display()))
        .filter_map(|e| e.ok().map(|e| e.path()))
        .filter(|p| p.extension().is_some_and(|x| x == "json"))
        .collect();
    files.sort();
    if let Some(n) = limit {
        files.truncate(n);
    }

    let mut rows: Vec<RrcRow> = Vec::with_capacity(files.len());
    let mut by_verdict: BTreeMap<String, usize> = BTreeMap::new();
    let mut parse_errors = 0usize;
    let mut edges = 0usize;
    for f in &files {
        let txt = match fs::read_to_string(f) { Ok(t) => t, Err(_) => { parse_errors += 1; continue; } };
        let r: Receipt = match serde_json::from_str(&txt) { Ok(r) => r, Err(_) => { parse_errors += 1; continue; } };
        if r.schema.as_deref() == Some("semisym_receipt_edge_v1") || r.modular_pole == Some(true) {
            edges += 1;
        }
        let row = map_receipt(&r);
        *by_verdict.entry(row.shape.clone()).or_insert(0) += 1;
        rows.push(row);
    }

    // generation only — no DB touch. Migration adds the JSONB metrics column.
    let mut sql = String::new();
    sql.push_str("BEGIN;\n");
    sql.push_str("ALTER TABLE ene.rrc_classifications ADD COLUMN IF NOT EXISTS metrics JSONB DEFAULT '{}'::jsonb;\n");
    for row in &rows {
        sql.push_str(&row_sql(row));
        sql.push('\n');
    }
    let audit_id = stable_uuid(&format!("semisym_scar_ingest:{}", dir.display()));
    sql.push_str(&format!(
        "INSERT INTO ene.ingest_events (id, session_id, event_type, payload) VALUES \
         ({}, NULL, 'semisym_scar_ingest', {}) ON CONFLICT (id) DO NOTHING;\n",
        sql_str(&audit_id),
        sql_str(&format!("{{\"rows\":{},\"source\":\"{}\"}}", rows.len(), dir.display())),
    ));
    sql.push_str("COMMIT;\n");
    fs::write(&out, &sql).expect("write sql");

    eprintln!("── semisym scar ingest (DRY-RUN, float-free) ──");
    eprintln!("checkpoint : {}", dir.display());
    eprintln!("receipts   : {} ({} parse errors, {} edge/pole)", files.len(), parse_errors, edges);
    eprintln!("rows        : {} → ene.rrc_classifications  (+ metrics JSONB, +1 ingest_events)", rows.len());
    eprintln!("verdicts    : {by_verdict:?}");
    eprintln!("sql written : {}", out.display());
    eprintln!("preview (first 2):");
    for row in rows.iter().take(2) {
        eprintln!("  {} shape={} weak_axes={:?}", row.equation_id, row.shape, row.weak_axes);
        eprintln!("    metrics={}", row.metrics);
    }

    if !apply {
        eprintln!("\n[dry-run] no database touched. To apply: --apply --dsn <postgres-conn>");
        return;
    }
    let Some(dsn) = dsn else {
        eprintln!("\nERROR: --apply requires --dsn <postgres-conn> (no default endpoint; confirm live ENE first)");
        std::process::exit(2);
    };
    eprintln!("\n[apply] executing {} via psql …", out.display());
    let status = Command::new("psql")
        .arg(&dsn)
        .arg("-v").arg("ON_ERROR_STOP=1")
        .arg("-f").arg(&out)
        .status()
        .expect("failed to spawn psql");
    if status.success() {
        eprintln!("[apply] OK — {} rows upserted into ene.rrc_classifications", rows.len());
    } else {
        eprintln!("[apply] psql failed ({status})");
        std::process::exit(1);
    }
}
