#![allow(dead_code)]

use std::io::Read;
use std::path::Path;
use serde::{Deserialize, Serialize};

use crate::data::{Data, Identifier, PersonUrl};
use crate::error::{Error, Result};

// ── Figshare release ──────────────────────────────────────────────────────────

const ORCID_FIGSHARE_ARTICLE_ID: u64 = 30375589;

/// Metadata about an ORCID Public Data File release on figshare.
///
/// The release is identified by year and batch number (the second numeric
/// component in the filename, e.g. `ORCID_2025_10_summaries.tar.gz` → year=2025,
/// batch=10). Download the summaries archive (~46 GB) manually and pass its
/// local path to [`stream_summaries_to_sqlite`].
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct OrcidRelease {
    /// Snapshot year, e.g. 2025.
    pub year: u16,
    /// Batch number within the year, e.g. 10.
    pub batch: u16,
    /// Filename of the summaries archive, e.g. `ORCID_2025_10_summaries.tar.gz`.
    pub filename: String,
    /// Direct download URL for the summaries archive.
    pub download_url: String,
    /// Compressed size in bytes (~46 GB for the 2025 release).
    pub size_bytes: u64,
}

// ── XML deserialization structs ───────────────────────────────────────────────
// quick_xml 0.40 non-namespace-aware deserialization strips namespace prefixes
// from element names before matching struct fields. `<common:uri>` matches the
// field rename `"uri"`, not `"common:uri"`. All renames use local names only.

#[derive(Deserialize, Default)]
#[serde(default)]
struct OrcidXml {
    #[serde(rename = "orcid-identifier")]
    orcid_identifier: XmlOrcidId,
    history: XmlHistory,
    person: XmlPerson,
    #[serde(rename = "activities-summary")]
    activities_summary: XmlActivitiesSummary,
}

// ── Activities / affiliations XML structs ─────────────────────────────────────

#[derive(Deserialize, Default)]
#[serde(default)]
struct XmlActivitiesSummary {
    employments: XmlEmploymentsList,
    educations: XmlEducationsList,
}

#[derive(Deserialize, Default)]
#[serde(default)]
struct XmlEmploymentsList {
    #[serde(rename = "affiliation-group")]
    affiliation_groups: Vec<XmlEmploymentGroup>,
}

#[derive(Deserialize, Default)]
#[serde(default)]
struct XmlEmploymentGroup {
    #[serde(rename = "employment-summary")]
    employment_summary: Option<XmlAffSummary>,
}

#[derive(Deserialize, Default)]
#[serde(default)]
struct XmlEducationsList {
    #[serde(rename = "affiliation-group")]
    affiliation_groups: Vec<XmlEducationGroup>,
}

#[derive(Deserialize, Default)]
#[serde(default)]
struct XmlEducationGroup {
    #[serde(rename = "education-summary")]
    education_summary: Option<XmlAffSummary>,
}

/// Shared summary struct for both employment-summary and education-summary elements.
#[derive(Deserialize, Default)]
#[serde(default)]
struct XmlAffSummary {
    organization: XmlOrgSummary,
    #[serde(rename = "department-name")]
    department_name: String,
    #[serde(rename = "role-title")]
    role_title: String,
    #[serde(rename = "start-date")]
    start_date: Option<XmlAffDate>,
    #[serde(rename = "end-date")]
    end_date: Option<XmlAffDate>,
}

#[derive(Deserialize, Default)]
#[serde(default)]
struct XmlOrgSummary {
    name: String,
    #[serde(rename = "disambiguated-organization")]
    disambiguated: Option<XmlDisambiguatedOrg>,
}

#[derive(Deserialize, Default)]
#[serde(default)]
struct XmlDisambiguatedOrg {
    #[serde(rename = "disambiguated-organization-identifier")]
    identifier: String,
    #[serde(rename = "disambiguation-source")]
    source: String,
}

/// Date as it appears in activities XML: year/month/day as direct text children.
#[derive(Deserialize, Default)]
#[serde(default)]
struct XmlAffDate {
    year: String,
    month: String,
    day: String,
}


#[derive(Deserialize, Default)]
#[serde(default)]
struct XmlOrcidId {
    uri: String,
    path: String,
}

#[derive(Deserialize, Default)]
#[serde(default)]
struct XmlHistory {
    #[serde(rename = "last-modified-date")]
    last_modified_date: String,
}

#[derive(Deserialize, Default)]
#[serde(default)]
struct XmlPerson {
    name: XmlPersonName,
    #[serde(rename = "other-names")]
    other_names: XmlOtherNames,
    addresses: XmlAddresses,
    #[serde(rename = "external-identifiers")]
    external_identifiers: XmlExternalIdentifiers,
}

#[derive(Deserialize, Default)]
#[serde(default)]
struct XmlPersonName {
    #[serde(rename = "given-names")]
    given_names: String,
    #[serde(rename = "family-name")]
    family_name: String,
    #[serde(rename = "credit-name")]
    credit_name: String,
}

#[derive(Deserialize, Default)]
#[serde(default)]
struct XmlAddresses {
    #[serde(rename = "address")]
    addresses: Vec<XmlAddress>,
}

#[derive(Deserialize, Default)]
#[serde(default)]
struct XmlAddress {
    country: String,
}

#[derive(Deserialize, Default)]
#[serde(default)]
struct XmlExternalIdentifiers {
    #[serde(rename = "external-identifier")]
    identifiers: Vec<XmlExternalIdentifier>,
}

/// Person-level external identifier (ResearcherID, Scopus Author ID, etc.).
/// Distinct from work-level external IDs in `<activities:activities-summary>`.
#[derive(Deserialize, Default)]
#[serde(default)]
struct XmlExternalIdentifier {
    #[serde(rename = "external-id-type")]
    type_: String,
    #[serde(rename = "external-id-value")]
    value: String,
}

#[derive(Deserialize, Default)]
#[serde(default)]
struct XmlOtherNames {
    #[serde(rename = "other-name")]
    names: Vec<XmlOtherName>,
}

#[derive(Deserialize, Default)]
#[serde(default)]
struct XmlOtherName {
    content: String,
}

// ── Parser ────────────────────────────────────────────────────────────────────

/// Parse a single ORCID summary XML record from raw bytes.
///
/// Returns `None` for locked / deprecated records (`<error:error>` root) and
/// for records with no ORCID URI (malformed or structurally unexpected).
fn parse_orcid_xml(xml_bytes: &[u8]) -> Option<OrcidXml> {
    let xml = std::str::from_utf8(xml_bytes).ok()?;
    // Locked / deprecated records use <error:error> as the root element; skip them.
    if xml.contains("<error:error") {
        return None;
    }
    let record: OrcidXml = quick_xml::de::from_str(xml).ok()?;
    if record.orcid_identifier.uri.is_empty() {
        return None;
    }
    Some(record)
}

// ── Data conversion ───────────────────────────────────────────────────────────

fn xml_aff_date_to_iso(date: &XmlAffDate) -> Option<String> {
    if date.year.is_empty() {
        return None;
    }
    match (date.month.is_empty(), date.day.is_empty()) {
        (false, false) => Some(format!("{}-{:0>2}-{:0>2}", date.year, date.month, date.day)),
        (false, _) => Some(format!("{}-{:0>2}", date.year, date.month)),
        _ => Some(date.year.clone()),
    }
}

fn xml_aff_summary_to_affiliation(s: &XmlAffSummary) -> crate::data::Affiliation {
    let (identifier, identifier_type) = s
        .organization
        .disambiguated
        .as_ref()
        .and_then(|d| resolve_identifier_xml(d))
        .unwrap_or_default();
    let name = if s.department_name.is_empty() {
        s.organization.name.clone()
    } else {
        format!("{}, {}", s.organization.name, s.department_name)
    };
    crate::data::Affiliation {
        identifier,
        identifier_type,
        name,
        role: s.role_title.clone(),
        start_date: s.start_date.as_ref().and_then(xml_aff_date_to_iso).unwrap_or_default(),
        end_date: s.end_date.as_ref().and_then(xml_aff_date_to_iso).unwrap_or_default(),
        ..Default::default()
    }
}

fn resolve_identifier_xml(d: &XmlDisambiguatedOrg) -> Option<(String, String)> {
    if d.identifier.is_empty() {
        return None;
    }
    match d.source.to_uppercase().as_str() {
        "ROR" => {
            let bare = d.identifier
                .trim_start_matches("https://ror.org/")
                .trim_start_matches("http://ror.org/");
            Some((format!("https://ror.org/{bare}"), "ROR".to_string()))
        }
        "FUNDREF" | "OFR" => Some((d.identifier.clone(), "FundRef".to_string())),
        "GRID" => Some((d.identifier.clone(), "GRID".to_string())),
        "ISNI" => Some((d.identifier.clone(), "ISNI".to_string())),
        "RINGGOLD" => Some((d.identifier.clone(), "Ringgold".to_string())),
        "WIKIDATA" => Some((d.identifier.clone(), "Wikidata".to_string())),
        _ => None,
    }
}

fn display_name(record: &OrcidXml) -> String {
    let name = &record.person.name;
    if !name.credit_name.is_empty() {
        return name.credit_name.clone();
    }
    match (name.given_names.is_empty(), name.family_name.is_empty()) {
        (false, false) => format!("{} {}", name.given_names, name.family_name),
        (true, false) => name.family_name.clone(),
        (false, true) => name.given_names.clone(),
        (true, true) => String::new(),
    }
}

/// Convert a parsed ORCID record to a commonmeta `Data` struct.
fn from_orcid(record: OrcidXml) -> Data {
    let id = record.orcid_identifier.uri.clone();
    let name_obj = &record.person.name;
    let given_name = name_obj.given_names.clone();
    let family_name = name_obj.family_name.clone();
    let name = name_obj.credit_name.clone();
    let title = display_name(&record);

    let additional_names: Vec<String> = record.person.other_names.names
        .iter()
        .map(|n| n.content.clone())
        .filter(|s| !s.is_empty())
        .collect();

    let identifiers: Vec<Identifier> = record
        .person
        .external_identifiers
        .identifiers
        .iter()
        .filter(|e| !e.type_.is_empty() && !e.value.is_empty())
        .map(|e| {
            let (cm_type, scheme) = map_orcid_ext_id_type(&e.type_);
            Identifier {
                identifier: e.value.clone(),
                identifier_type: cm_type.to_string(),
                scheme: scheme.unwrap_or_default(),
                ..Default::default()
            }
        })
        .collect();

    let country = record.person.addresses.addresses.first()
        .map(|a| a.country.clone())
        .unwrap_or_default();

    let affiliations: Vec<crate::data::Affiliation> = record
        .activities_summary
        .employments
        .affiliation_groups
        .iter()
        .filter_map(|g| g.employment_summary.as_ref())
        .filter(|s| !s.organization.name.is_empty())
        .map(xml_aff_summary_to_affiliation)
        .chain(
            record
                .activities_summary
                .educations
                .affiliation_groups
                .iter()
                .filter_map(|g| g.education_summary.as_ref())
                .filter(|s| !s.organization.name.is_empty())
                .map(|s| {
                    let mut aff = xml_aff_summary_to_affiliation(s);
                    aff.role = if aff.role.is_empty() {
                        "Education".to_string()
                    } else {
                        format!("Education: {}", aff.role)
                    };
                    aff
                }),
        )
        .collect();

    Data {
        id,
        type_: "Person".to_string(),
        given_name,
        family_name,
        name,
        additional_names,
        title,
        identifiers,
        country,
        affiliations,
        asserted_by: "ORCID".to_string(),
        date_updated: record.history.last_modified_date.clone(),
        provider: "ORCID".to_string(),
        ..Data::default()
    }
}

// ── SQLite schema ─────────────────────────────────────────────────────────────

const PEOPLE_DDL: &str = r#"PRAGMA synchronous=NORMAL;
CREATE TABLE IF NOT EXISTS settings (
    "key"   TEXT PRIMARY KEY NOT NULL,
    "value" TEXT NOT NULL DEFAULT ''
);
CREATE TABLE IF NOT EXISTS people (
    "id"                   TEXT PRIMARY KEY NOT NULL,
    "orcid"                TEXT NOT NULL DEFAULT '',
    "given_name"           TEXT NOT NULL DEFAULT '',
    "family_name"          TEXT NOT NULL DEFAULT '',
    "credit_name"          TEXT NOT NULL DEFAULT '',
    "name"                 TEXT NOT NULL DEFAULT '',
    "country"              TEXT NOT NULL DEFAULT '',
    "other_names"          TEXT NOT NULL DEFAULT '',
    "external_identifiers" TEXT NOT NULL DEFAULT '',
    "affiliations"         TEXT NOT NULL DEFAULT '',
    "date_updated"         TEXT NOT NULL DEFAULT '',
    "valid"                INTEGER NOT NULL DEFAULT 0,
    "metadata"             BLOB NOT NULL DEFAULT x''
);
CREATE INDEX IF NOT EXISTS people_country ON people("country");
CREATE INDEX IF NOT EXISTS people_date_updated ON people("date_updated");"#;

// Migrate databases created before other_names / external_identifiers were added.
// SQLite does not support ADD COLUMN IF NOT EXISTS; ignoring errors is idiomatic.
const PEOPLE_MIGRATE_DDL: &[&str] = &[
    "ALTER TABLE people ADD COLUMN other_names TEXT NOT NULL DEFAULT ''",
    "ALTER TABLE people ADD COLUMN external_identifiers TEXT NOT NULL DEFAULT ''",
    "ALTER TABLE people ADD COLUMN affiliations TEXT NOT NULL DEFAULT ''",
    "ALTER TABLE people ADD COLUMN \"valid\" INTEGER NOT NULL DEFAULT 0",
];

const PEOPLE_FTS5_DDL: &str = "CREATE VIRTUAL TABLE people_fts USING fts5(\
    name, other_names, \
    content=\"people\", \
    content_rowid=\"rowid\", \
    tokenize=\"unicode61 remove_diacritics 1\"\
)";

const PEOPLE_INSERT: &str = r#"INSERT OR REPLACE INTO people (
    "id", "orcid", "given_name", "family_name", "credit_name", "name",
    "country", "other_names", "external_identifiers", "affiliations", "date_updated", "valid", "metadata"
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13)"#;

struct PeopleRow {
    id: String,
    orcid: String,
    given_name: String,
    family_name: String,
    credit_name: String,
    name: String,
    country: String,
    other_names: String,
    external_identifiers: String,
    affiliations: String,
    date_updated: String,
    valid: bool,
    metadata: Vec<u8>,
}

fn aff_to_person_affiliation(aff: crate::data::Affiliation) -> PersonAffiliation {
    PersonAffiliation {
        identifier: if aff.identifier.is_empty() { None } else { Some(aff.identifier) },
        identifier_type: if aff.identifier_type.is_empty() { None } else { Some(aff.identifier_type) },
        name: aff.name,
        role: if aff.role.is_empty() { None } else { Some(aff.role) },
        start_date: if aff.start_date.is_empty() { None } else { Some(aff.start_date) },
        end_date: if aff.end_date.is_empty() { None } else { Some(aff.end_date) },
    }
}

fn serialize_to_people_row(record: &OrcidXml, xml_bytes: &[u8]) -> PeopleRow {
    let name = display_name(record);
    let country = record
        .person
        .addresses
        .addresses
        .first()
        .map(|a| a.country.clone())
        .unwrap_or_default();
    let other_names = record
        .person
        .other_names
        .names
        .iter()
        .map(|n| n.content.as_str())
        .filter(|s| !s.is_empty())
        .collect::<Vec<_>>()
        .join(" ");
    let ext_ids: Vec<serde_json::Value> = record
        .person
        .external_identifiers
        .identifiers
        .iter()
        .filter(|e| !e.type_.is_empty() && !e.value.is_empty())
        .map(|e| serde_json::json!({"type": e.type_, "value": e.value}))
        .collect();
    let external_identifiers = serde_json::to_string(&ext_ids).unwrap_or_default();
    let affs: Vec<PersonAffiliation> = record
        .activities_summary
        .employments
        .affiliation_groups
        .iter()
        .filter_map(|g| g.employment_summary.as_ref())
        .filter(|s| !s.organization.name.is_empty())
        .map(|s| aff_to_person_affiliation(xml_aff_summary_to_affiliation(s)))
        .chain(
            record
                .activities_summary
                .educations
                .affiliation_groups
                .iter()
                .filter_map(|g| g.education_summary.as_ref())
                .filter(|s| !s.organization.name.is_empty())
                .map(|s| {
                    let mut aff = xml_aff_summary_to_affiliation(s);
                    aff.role = if aff.role.is_empty() {
                        "Education".to_string()
                    } else {
                        format!("Education: {}", aff.role)
                    };
                    aff_to_person_affiliation(aff)
                }),
        )
        .collect();
    let affiliations = affiliations_to_json(&affs);
    let metadata = zstd::encode_all(xml_bytes, 0).unwrap_or_else(|_| xml_bytes.to_vec());
    PeopleRow {
        id: record.orcid_identifier.uri.clone(),
        orcid: record.orcid_identifier.path.clone(),
        given_name: record.person.name.given_names.clone(),
        family_name: record.person.name.family_name.clone(),
        credit_name: record.person.name.credit_name.clone(),
        name,
        country,
        other_names,
        external_identifiers,
        affiliations,
        date_updated: record.history.last_modified_date.clone(),
        valid: false,
        metadata,
    }
}

// ── Bulk import ───────────────────────────────────────────────────────────────

const BATCH_SIZE: usize = 50_000;

/// Stream all records from an ORCID summaries tar.gz into the `people` table
/// of the SQLite database at `output_path`.
///
/// The tar.gz contains one XML file per ORCID record arranged in prefix
/// subdirectories (`000/`, `001/`, …). Records with `<error:error>` roots
/// (locked or deprecated ORCIDs) are skipped silently.
///
/// The FTS5 index is rebuilt once at the end of the import. Existing rows are
/// replaced if their ORCID ID already exists in the table.
///
/// Returns the total number of rows inserted or replaced.
/// Iterate XML entries from a tar archive (any `Read` source) and upsert them
/// into an already-open, already-schema'd SQLite connection. Does not touch FTS.
/// `limit = 0` means no limit; any other value stops after that many valid records.
fn drain_tar_into_conn<R: Read>(
    conn: &rusqlite::Connection,
    reader: R,
    limit: usize,
) -> Result<usize> {
    use flate2::read::GzDecoder;
    use tar::Archive;

    let gz = GzDecoder::new(reader);
    let mut archive = Archive::new(gz);

    let mut batch: Vec<PeopleRow> = Vec::with_capacity(BATCH_SIZE.min(if limit > 0 { limit } else { BATCH_SIZE }));
    let mut total = 0usize;
    let entries = archive
        .entries()
        .map_err(|e| Error::Parse(format!("read tar entries: {}", e)))?;

    'entries: for entry in entries {
        let mut entry = match entry {
            Ok(e) => e,
            Err(_) => continue,
        };

        let is_xml = entry
            .path()
            .ok()
            .and_then(|p| p.extension().map(|e| e == "xml"))
            .unwrap_or(false);
        if !is_xml {
            continue;
        }

        let mut xml_bytes = Vec::new();
        if entry.read_to_end(&mut xml_bytes).is_err() {
            continue;
        }

        if let Some(record) = parse_orcid_xml(&xml_bytes) {
            batch.push(serialize_to_people_row(&record, &xml_bytes));
        }

        if limit > 0 && total + batch.len() >= limit {
            batch.truncate(limit - total);
            break 'entries;
        }

        if batch.len() >= BATCH_SIZE {
            let count = write_people_batch(conn, &batch)?;
            total += count;
            eprintln!("orcid: {} inserted in batch ({} total)", count, total);
            batch.clear();
        }
    }

    if !batch.is_empty() {
        let count = write_people_batch(conn, &batch)?;
        total += count;
    }

    eprintln!("orcid: {} people imported; building FTS index…", total);
    Ok(total)
}

fn open_people_conn(output_path: &Path) -> Result<rusqlite::Connection> {
    use rusqlite::Connection;
    if let Some(parent) = output_path.parent() {
        if !parent.as_os_str().is_empty() && !parent.exists() {
            std::fs::create_dir_all(parent)
                .map_err(|e| Error::Parse(format!("create output dir: {}", e)))?;
        }
    }
    let conn = Connection::open(output_path)
        .map_err(|e| Error::Parse(format!("open sqlite '{}': {}", output_path.display(), e)))?;
    conn.query_row("PRAGMA journal_mode=WAL", [], |r: &rusqlite::Row<'_>| r.get::<_, String>(0))
        .map_err(|e| Error::Parse(format!("WAL mode: {}", e)))?;
    ensure_people_schema(&conn)?;
    // Drop FTS before bulk import; rebuild at the end for throughput.
    let _ = conn.execute("DROP TABLE IF EXISTS people_fts", []);
    Ok(conn)
}

fn finish_people_import(conn: &rusqlite::Connection, total: usize) -> Result<usize> {
    conn.execute_batch(PEOPLE_FTS5_DDL)
        .map_err(|e| Error::Parse(format!("FTS5 DDL: {}", e)))?;
    conn.execute("INSERT INTO people_fts(people_fts) VALUES('rebuild')", [])
        .map_err(|e| Error::Parse(format!("FTS5 rebuild: {}", e)))?;
    let _ = conn.execute("PRAGMA wal_checkpoint(PASSIVE)", []);
    Ok(total)
}

/// Import a local tar.gz summaries file into the `people` SQLite table.
/// `limit = 0` imports all records; any other value stops after that many valid records.
pub fn stream_summaries_to_sqlite(
    tar_gz_path: &Path,
    output_path: &Path,
    limit: usize,
) -> Result<usize> {
    let conn = open_people_conn(output_path)?;
    let file = std::fs::File::open(tar_gz_path)
        .map_err(|e| Error::Parse(format!("open '{}': {}", tar_gz_path.display(), e)))?;
    let total = drain_tar_into_conn(&conn, file, limit)?;
    finish_people_import(&conn, total)
}

/// Returns a descriptive error when figshare blocks an ORCID download with 403.
/// Cloud server IPs are sometimes blocked by figshare's ndownloader CDN.
fn figshare_blocked_err(url: &str) -> Error {
    Error::Http(format!(
        "figshare blocked {url} (403) — server IP may not be authorized for large-file downloads.\n\
        \n\
        To import on this host:\n  \
        1. On a machine with figshare access, find the file URL:\n     \
           commonmeta import --from orcid --list-releases\n  \
        2. Download the file on that machine:\n     \
           wget -O ORCID_summaries.tar.gz \"<URL from step 1>\"\n  \
        3. Transfer and import (rsync is resumable if interrupted):\n     \
           rsync -az --progress ORCID_summaries.tar.gz user@server:/data/\n     \
           commonmeta import --from orcid /data/ORCID_summaries.tar.gz\n  \
        For --sample only: run --sample locally, then rsync the resulting people.db to this server."
    ))
}

/// Download the ORCID Public Data File summaries from an HTTP URL to a local
/// cache file using a single sequential GET request (no parallel Range
/// requests), then return the path. Creates parent directories as needed.
/// Skips the download if a non-empty file already exists at `cache_path`.
fn download_summaries_to_cache(url: &str, cache_path: &std::path::Path) -> Result<()> {
    if cache_path.exists() && cache_path.metadata().map(|m| m.len()).unwrap_or(0) > 0 {
        eprintln!("orcid: using cached {}", cache_path.display());
        return Ok(());
    }
    if let Some(parent) = cache_path.parent() {
        std::fs::create_dir_all(parent)
            .map_err(|e| Error::Parse(format!("mkdir: {}", e)))?;
    }
    eprintln!("orcid: downloading to {} …", cache_path.display());
    let client = reqwest::blocking::Client::builder()
        // A 46 GB file at 10 MB/s takes ~78 min; allow 6 h for slow links.
        .timeout(std::time::Duration::from_secs(6 * 3600))
        .build()
        .map_err(|e| Error::Http(e.to_string()))?;
    let raw = client
        .get(url)
        .send()
        .map_err(|e| Error::Http(format!("GET {url}: {e}")))?;
    if raw.status().as_u16() == 403 && url.contains("figshare.com") {
        return Err(figshare_blocked_err(url));
    }
    let mut resp = raw
        .error_for_status()
        .map_err(|e| Error::Http(format!("HTTP error: {e}")))?;
    let mut file = std::fs::File::create(cache_path)
        .map_err(|e| Error::Parse(format!("create {}: {}", cache_path.display(), e)))?;
    let bytes = std::io::copy(&mut resp, &mut file)
        .map_err(|e| Error::Parse(format!("download write: {}", e)))?;
    eprintln!(
        "orcid: {:.1} GB saved to {}",
        bytes as f64 / 1_073_741_824.0,
        cache_path.display()
    );
    Ok(())
}

fn write_people_batch(conn: &rusqlite::Connection, batch: &[PeopleRow]) -> Result<usize> {
    use rusqlite::params;
    let tx = conn
        .unchecked_transaction()
        .map_err(|e| Error::Parse(format!("begin transaction: {}", e)))?;
    {
        let mut stmt = tx
            .prepare(PEOPLE_INSERT)
            .map_err(|e| Error::Parse(format!("prepare insert: {}", e)))?;
        for row in batch {
            stmt.execute(params![
                row.id,
                row.orcid,
                row.given_name,
                row.family_name,
                row.credit_name,
                row.name,
                row.country,
                row.other_names,
                row.external_identifiers,
                row.affiliations,
                row.date_updated,
                row.valid as i32,
                row.metadata,
            ])
            .map_err(|e| Error::Parse(format!("insert '{}': {}", row.id, e)))?;
        }
    }
    tx.commit()
        .map_err(|e| Error::Parse(format!("commit transaction: {}", e)))?;
    Ok(batch.len())
}

// ── ORCID public API (JSON) ───────────────────────────────────────────────────

// Serde structs for the ORCID member/public API `/v3.0/{orcid}/person` endpoint.
// The JSON uses a mix of plain strings and `{value: "..."}` wrappers.

#[derive(Deserialize, Default)]
#[serde(default)]
struct PersonJson {
    #[serde(rename = "last-modified-date")]
    last_modified_date: Option<EpochMs>,
    name: NameJson,
    biography: Option<BiographyJson>,
    #[serde(rename = "other-names")]
    other_names: OtherNamesJson,
    addresses: AddressesJson,
    #[serde(rename = "researcher-urls")]
    researcher_urls: ResearcherUrlsJson,
    #[serde(rename = "external-identifiers")]
    external_identifiers: ExtIdsJson,
    path: String, // "/0000-0003-1419-2405/person"
}

#[derive(Deserialize, Default)]
#[serde(default)]
struct BiographyJson {
    content: Option<String>,
}

#[derive(Deserialize, Default)]
#[serde(default)]
struct OtherNamesJson {
    #[serde(rename = "other-name")]
    other_names: Vec<OtherNameItemJson>,
}

#[derive(Deserialize, Default)]
#[serde(default)]
struct OtherNameItemJson {
    content: String,
}

#[derive(Deserialize, Default)]
#[serde(default)]
struct EpochMs {
    value: i64,
}

#[derive(Deserialize, Default)]
#[serde(default)]
struct NameJson {
    // These can be `null` (not just absent) when the person has not set them.
    #[serde(rename = "given-names")]
    given_names: Option<StringValue>,
    #[serde(rename = "family-name")]
    family_name: Option<StringValue>,
    #[serde(rename = "credit-name")]
    credit_name: Option<StringValue>,
}

#[derive(Deserialize, Default)]
#[serde(default)]
struct StringValue {
    value: String,
}

#[derive(Deserialize, Default)]
#[serde(default)]
struct AddressesJson {
    address: Vec<AddressItemJson>,
}

#[derive(Deserialize, Default)]
#[serde(default)]
struct AddressItemJson {
    country: StringValue,
}

#[derive(Deserialize, Default)]
#[serde(default)]
struct ResearcherUrlsJson {
    #[serde(rename = "researcher-url")]
    urls: Vec<ResearcherUrlItemJson>,
}

#[derive(Deserialize, Default)]
#[serde(default)]
struct ResearcherUrlItemJson {
    #[serde(rename = "url-name")]
    url_name: Option<String>,
    url: StringValue,
}

#[derive(Deserialize, Default)]
#[serde(default)]
struct ExtIdsJson {
    #[serde(rename = "external-identifier")]
    identifiers: Vec<ExtIdItemJson>,
}

#[derive(Deserialize, Default)]
#[serde(default)]
struct ExtIdItemJson {
    #[serde(rename = "external-id-type")]
    type_: String,
    #[serde(rename = "external-id-value")]
    value: String,
}

fn epoch_ms_to_iso(ms: i64) -> String {
    use chrono::{TimeZone, Utc};
    if ms == 0 {
        return String::new();
    }
    Utc.timestamp_millis_opt(ms)
        .single()
        .map(|dt| dt.to_rfc3339_opts(chrono::SecondsFormat::Secs, true))
        .unwrap_or_default()
}

/// Build a `Data` struct from a parsed ORCID person JSON response.
fn person_json_to_data(person: &PersonJson, orcid_url: &str) -> Data {
    let name_obj = &person.name;
    let given_name = name_obj.given_names.as_ref().map(|s| s.value.clone()).unwrap_or_default();
    let family_name = name_obj.family_name.as_ref().map(|s| s.value.clone()).unwrap_or_default();
    let name = name_obj.credit_name.as_ref().map(|s| s.value.clone()).unwrap_or_default();
    let title = if !name.is_empty() {
        name.clone()
    } else {
        match (given_name.is_empty(), family_name.is_empty()) {
            (false, false) => format!("{} {}", given_name, family_name),
            (true, false) => family_name.clone(),
            (false, true) => given_name.clone(),
            (true, true) => String::new(),
        }
    };
    let additional_names: Vec<String> = person.other_names.other_names
        .iter()
        .map(|n| n.content.clone())
        .filter(|s| !s.is_empty())
        .collect();
    let description = person
        .biography
        .as_ref()
        .and_then(|b| b.content.as_deref())
        .filter(|s| !s.is_empty())
        .unwrap_or_default()
        .to_string();
    let identifiers: Vec<Identifier> = person
        .external_identifiers
        .identifiers
        .iter()
        .filter(|e| !e.type_.is_empty() && !e.value.is_empty())
        .map(|e| {
            let (cm_type, scheme) = map_orcid_ext_id_type(&e.type_);
            Identifier {
                identifier: e.value.clone(),
                identifier_type: cm_type.to_string(),
                scheme: scheme.unwrap_or_default(),
                ..Default::default()
            }
        })
        .collect();
    let urls: Vec<PersonUrl> = person.researcher_urls.urls
        .iter()
        .filter(|r| !r.url.value.is_empty())
        .map(|r| PersonUrl {
            name: r.url_name.clone().unwrap_or_default(),
            url: r.url.value.clone(),
        })
        .collect();
    let country = person.addresses.address.first()
        .map(|a| a.country.value.clone())
        .unwrap_or_default();
    Data {
        id: orcid_url.to_string(),
        type_: "Person".to_string(),
        given_name,
        family_name,
        name,
        additional_names,
        title,
        description,
        identifiers,
        urls,
        country,
        asserted_by: "ORCID".to_string(),
        date_updated: epoch_ms_to_iso(person.last_modified_date.as_ref().map(|d| d.value).unwrap_or(0)),
        provider: "ORCID".to_string(),
        ..Data::default()
    }
}

/// Fetch a person record from the ORCID public API and return a `PeopleRow`
/// ready for SQLite insertion. `orcid_url` must be a canonical ORCID URL.
fn fetch_person_api(orcid_url: &str) -> Result<PeopleRow> {
    let (mut row, _) = fetch_person_api_with_json(orcid_url)?;
    // Best-effort: fetch employment + education affiliations (no ROR cross-ref without db_path).
    if let Ok(affs) = fetch_person_affiliations(orcid_url, None) {
        row.affiliations = affiliations_to_json(&affs);
    }
    Ok(row)
}

fn fetch_person_api_with_json(orcid_url: &str) -> Result<(PeopleRow, serde_json::Value)> {
    let orcid = orcid_url.trim_start_matches("https://orcid.org/");
    let api_url = format!("https://pub.orcid.org/v3.0/{}/person", orcid);
    let client = reqwest::blocking::Client::builder()
        .user_agent(crate::io_utils::commonmeta_user_agent())
        .timeout(std::time::Duration::from_secs(30))
        .build()
        .map_err(|e| Error::Http(e.to_string()))?;
    let bytes = client
        .get(&api_url)
        .header("Accept", "application/vnd.orcid+json")
        .send()
        .map_err(|e| Error::Http(format!("ORCID API: {}", e)))?
        .error_for_status()
        .map_err(|e| Error::Http(format!("ORCID API: {}", e)))?
        .bytes()
        .map_err(|e| Error::Http(e.to_string()))?;

    let json_value: serde_json::Value =
        serde_json::from_slice(&bytes).map_err(|e| Error::Parse(e.to_string()))?;
    let person: PersonJson =
        serde_json::from_slice(&bytes).map_err(|e| Error::Parse(e.to_string()))?;

    let country = person
        .addresses
        .address
        .first()
        .map(|a| a.country.value.clone())
        .unwrap_or_default();
    let other_names = person
        .other_names
        .other_names
        .iter()
        .map(|n| n.content.as_str())
        .filter(|s| !s.is_empty())
        .collect::<Vec<_>>()
        .join(" ");
    let ext_ids: Vec<serde_json::Value> = person
        .external_identifiers
        .identifiers
        .iter()
        .filter(|e| !e.type_.is_empty() && !e.value.is_empty())
        .map(|e| serde_json::json!({"type": e.type_, "value": e.value}))
        .collect();
    let external_identifiers = serde_json::to_string(&ext_ids).unwrap_or_default();
    let metadata = zstd::encode_all(bytes.as_ref(), 0).unwrap_or_else(|_| bytes.to_vec());
    let data = person_json_to_data(&person, orcid_url);

    let row = PeopleRow {
        id: data.id.clone(),
        orcid: orcid.to_string(),
        given_name: person.name.given_names.as_ref().map(|s| s.value.clone()).unwrap_or_default(),
        family_name: person.name.family_name.as_ref().map(|s| s.value.clone()).unwrap_or_default(),
        credit_name: person.name.credit_name.as_ref().map(|s| s.value.clone()).unwrap_or_default(),
        name: data.title,
        country,
        other_names,
        external_identifiers,
        affiliations: String::new(), // populated by fetch_person_api after fetch_person_affiliations
        date_updated: data.date_updated,
        valid: false,
        metadata,
    };
    Ok((row, json_value))
}

/// Fetch a person from the ORCID public API and return their record as `Data`.
/// Accepts a bare ORCID iD (`0000-0003-1419-2405`) or a full ORCID URL.
pub fn fetch_orcid(id: &str) -> Result<Data> {
    let url = crate::utils::normalize_orcid(id);
    if url.is_empty() {
        return Err(Error::Parse(format!("'{}' is not a valid ORCID identifier", id)));
    }
    let (_, json) = fetch_person_api_with_json(&url)?;
    let person: PersonJson =
        serde_json::from_value(json).map_err(|e| Error::Parse(e.to_string()))?;
    Ok(person_json_to_data(&person, &url))
}

/// Fetch a person from the ORCID public API and return both the parsed [`Data`]
/// and the raw ORCID 3.0 person JSON in a single HTTP request.
///
/// Prefer this over calling [`fetch_orcid`] and [`fetch_person_json`] separately
/// when you need both — e.g. when caching the raw JSON for later import into the
/// `people` table while also assembling the response.
pub fn fetch_orcid_with_json(id: &str) -> Result<(Data, serde_json::Value)> {
    let url = crate::utils::normalize_orcid(id);
    if url.is_empty() {
        return Err(Error::Parse(format!("'{}' is not a valid ORCID identifier", id)));
    }
    let (_, json) = fetch_person_api_with_json(&url)?;
    let person: PersonJson =
        serde_json::from_value(json.clone()).map_err(|e| Error::Parse(e.to_string()))?;
    Ok((person_json_to_data(&person, &url), json))
}

/// Fetch a person from the ORCID public API and return the raw JSON response
/// conforming to the ORCID 3.0 person schema (`orcid_schema_v3.0.json`).
pub fn fetch_person_json(id: &str) -> Result<serde_json::Value> {
    let url = crate::utils::normalize_orcid(id);
    if url.is_empty() {
        return Err(Error::Parse(format!("'{}' is not a valid ORCID identifier", id)));
    }
    let (_, json) = fetch_person_api_with_json(&url)?;
    Ok(json)
}

/// Build an ORCID 3.0 person JSON value from a parsed XML summary record.
///
/// The summaries archive contains a subset of the full person record (no
/// biography, no researcher-URLs, no emails). Fields not present in the XML
/// are rendered as empty containers matching the schema shape.
fn xml_record_to_person_json(record: &OrcidXml) -> serde_json::Value {
    let orcid = &record.orcid_identifier.path;

    // Convert ISO-8601 string (from XML) to epoch ms (JSON schema format).
    let last_modified_ms = chrono::DateTime::parse_from_rfc3339(&record.history.last_modified_date)
        .map(|dt| dt.timestamp_millis())
        .unwrap_or(0);

    let given = if record.person.name.given_names.is_empty() {
        serde_json::Value::Null
    } else {
        serde_json::json!({"value": record.person.name.given_names})
    };
    let family = if record.person.name.family_name.is_empty() {
        serde_json::Value::Null
    } else {
        serde_json::json!({"value": record.person.name.family_name})
    };
    let credit = if record.person.name.credit_name.is_empty() {
        serde_json::Value::Null
    } else {
        serde_json::json!({"value": record.person.name.credit_name})
    };

    let other_name_items: Vec<serde_json::Value> = record
        .person
        .other_names
        .names
        .iter()
        .map(|n| serde_json::json!({"content": n.content, "visibility": "public"}))
        .collect();

    let address_items: Vec<serde_json::Value> = record
        .person
        .addresses
        .addresses
        .iter()
        .map(|a| serde_json::json!({"country": {"value": a.country}, "visibility": "public"}))
        .collect();

    let keyword_items: Vec<serde_json::Value> = Vec::new();

    let ext_id_items: Vec<serde_json::Value> = record
        .person
        .external_identifiers
        .identifiers
        .iter()
        .map(|e| serde_json::json!({
            "external-id-type": e.type_,
            "external-id-value": e.value,
            "external-id-relationship": "self",
            "visibility": "public"
        }))
        .collect();

    serde_json::json!({
        "last-modified-date": {"value": last_modified_ms},
        "name": {
            "given-names": given,
            "family-name": family,
            "credit-name": credit,
            "visibility": "public",
            "path": orcid
        },
        "other-names": {
            "other-name": other_name_items,
            "path": format!("/{}/other-names", orcid)
        },
        "biography": null,
        "researcher-urls": {
            "researcher-url": [],
            "path": format!("/{}/researcher-urls", orcid)
        },
        "emails": {
            "email": [],
            "path": format!("/{}/email", orcid)
        },
        "addresses": {
            "address": address_items,
            "path": format!("/{}/address", orcid)
        },
        "keywords": {
            "keyword": keyword_items,
            "path": format!("/{}/keywords", orcid)
        },
        "external-identifiers": {
            "external-identifier": ext_id_items,
            "path": format!("/{}/external-identifiers", orcid)
        },
        "path": format!("/{}/person", orcid)
    })
}

/// Look up a person in the local SQLite database and return the raw ORCID 3.0
/// person JSON conforming to `orcid_schema_v3.0.json`.
///
/// JSON metadata blobs (from [`import_person`]) are decompressed and returned
/// as-is. XML blobs (from [`stream_summaries_to_sqlite`]) are reconstructed
/// into the equivalent JSON shape.
pub fn fetch_person_json_sqlite(id: &str, db_path: &Path) -> Result<serde_json::Value> {
    use rusqlite::{params, Connection};

    let url = crate::utils::normalize_orcid(id);
    if url.is_empty() {
        return Err(Error::Parse(format!("'{}' is not a valid ORCID identifier", id)));
    }

    let conn = Connection::open(db_path)
        .map_err(|e| Error::Parse(format!("open sqlite '{}': {}", db_path.display(), e)))?;
    let blob: Vec<u8> = conn
        .query_row(
            "SELECT metadata FROM people WHERE id = ?1 LIMIT 1",
            params![url],
            |row| row.get(0),
        )
        .map_err(|e| {
            if matches!(e, rusqlite::Error::QueryReturnedNoRows) {
                Error::Parse(format!("person '{}' not found in local database", id))
            } else {
                Error::Parse(format!("sqlite query failed: {}", e))
            }
        })?;
    let raw = zstd::decode_all(std::io::Cursor::new(&blob))
        .map_err(|e| Error::Parse(format!("decompress metadata for '{}': {}", id, e)))?;

    if raw.iter().find(|&&b| b != b' ' && b != b'\n' && b != b'\r' && b != b'\t') == Some(&b'<') {
        // XML blob from bulk import — reconstruct JSON shape.
        let record = parse_orcid_xml(&raw)
            .ok_or_else(|| Error::Parse(format!("re-parse XML for '{}'", id)))?;
        Ok(xml_record_to_person_json(&record))
    } else {
        // JSON blob from API import — return directly.
        serde_json::from_slice(&raw).map_err(|e| Error::Parse(format!("re-parse JSON for '{}': {}", id, e)))
    }
}

/// Ensure the `people` table and its FTS5 index exist and are current.
///
/// Safe to call on both new and existing databases:
/// - New databases: creates table and FTS5 with all columns.
/// - Existing databases: migrates missing columns and recreates the FTS5
///   index (with a content rebuild) when the schema has changed.
fn ensure_people_schema(conn: &rusqlite::Connection) -> Result<()> {
    conn.execute_batch(PEOPLE_DDL)
        .map_err(|e| Error::Parse(format!("people DDL: {}", e)))?;

    // Add new columns to existing tables (errors are silently ignored because
    // SQLite has no ADD COLUMN IF NOT EXISTS).
    for stmt in PEOPLE_MIGRATE_DDL {
        let _ = conn.execute_batch(stmt);
    }

    // Drop the keywords column (removed from v1.0 schema). The FTS must be
    // dropped first because it references the content table; SQLite ≥ 3.35 is
    // required for ALTER TABLE … DROP COLUMN (errors silently ignored).
    let fts_before: String = conn
        .query_row(
            "SELECT COALESCE(sql, '') FROM sqlite_master WHERE name='people_fts'",
            [],
            |r| r.get(0),
        )
        .unwrap_or_default();
    if fts_before.contains("keywords") {
        let _ = conn.execute_batch("DROP TABLE IF EXISTS people_fts");
        let _ = conn.execute_batch("ALTER TABLE people DROP COLUMN keywords");
    }

    // Recreate FTS5 if missing or predating other_names.
    let fts_sql: String = conn
        .query_row(
            "SELECT COALESCE(sql, '') FROM sqlite_master WHERE name='people_fts'",
            [],
            |r| r.get(0),
        )
        .unwrap_or_default();

    if !fts_sql.contains("other_names") {
        conn.execute_batch("DROP TABLE IF EXISTS people_fts")
            .map_err(|e| Error::Parse(format!("drop FTS5: {}", e)))?;
        conn.execute_batch(PEOPLE_FTS5_DDL)
            .map_err(|e| Error::Parse(format!("FTS5 DDL: {}", e)))?;
        let row_count: i64 = conn
            .query_row("SELECT COUNT(*) FROM people", [], |r| r.get(0))
            .unwrap_or(0);
        if row_count > 0 {
            conn.execute("INSERT INTO people_fts(people_fts) VALUES('rebuild')", [])
                .map_err(|e| Error::Parse(format!("FTS5 rebuild: {}", e)))?;
        }
    }

    Ok(())
}

/// Fetch a single person from the ORCID public API and upsert them into the
/// `people` table of the SQLite database at `db_path`.
///
/// Accepts a bare ORCID iD (`0000-0003-1419-2405`) or a full ORCID URL.
/// Creates the table and FTS5 index if they don't already exist.
/// Fetch an ORCID person from the public API and upsert into the `people`
/// table at `people_db`. Also fetches the person's works from Crossref and
/// DataCite and upserts them into the `works` table at `works_db` (may be the
/// same path as `people_db`). Returns the number of works written.
pub fn import_person(id: &str, people_db: &Path, works_db: &Path) -> Result<usize> {
    let orcid_url = crate::utils::normalize_orcid(id);
    if orcid_url.is_empty() {
        return Err(Error::Parse(format!("'{}' is not a valid ORCID identifier", id)));
    }

    use rusqlite::Connection;
    let conn = Connection::open(people_db)
        .map_err(|e| Error::Parse(format!("open sqlite '{}': {}", people_db.display(), e)))?;
    let _: String = conn
        .query_row("PRAGMA journal_mode=WAL", [], |r| r.get(0))
        .map_err(|e| Error::Parse(format!("WAL mode: {}", e)))?;
    ensure_people_schema(&conn)?;

    let row = fetch_person_api(&orcid_url)?;
    write_people_batch(&conn, std::slice::from_ref(&row))?;

    conn.execute("INSERT INTO people_fts(people_fts) VALUES('rebuild')", [])
        .map_err(|e| Error::Parse(format!("FTS5 rebuild: {}", e)))?;

    // Fetch works from Crossref and DataCite, deduplicate, and upsert.
    const MAX_WORKS: usize = 50;
    let mut works: Vec<crate::Data> = Vec::new();
    let mut cr = crate::formats::crossref::fetch_by_orcid(&orcid_url, MAX_WORKS, 1).unwrap_or_default();
    let mut dc = crate::formats::datacite::fetch_by_orcid(&orcid_url, MAX_WORKS, 1).unwrap_or_default();
    works.append(&mut cr);
    works.append(&mut dc);

    let n = if !works.is_empty() {
        let mut seen = std::collections::HashSet::new();
        works.retain(|d| seen.insert(d.id.clone()));
        works.sort_by(|a, b| b.date_published.cmp(&a.date_published));
        works.truncate(MAX_WORKS);
        let n = works.len();
        crate::formats::commonmeta::upsert_sqlite(&works, works_db)?;
        let work_ids: Vec<String> = works.iter().map(|d| d.id.clone()).collect();
        crate::formats::commonmeta::upsert_works_people_links(&work_ids, &orcid_url, works_db)?;
        n
    } else {
        0
    };

    Ok(n)
}

// ── Cache import ─────────────────────────────────────────────────────────────

/// Read ORCID person rows from a dragoman `cache.sqlite3` (`pid_records` table,
/// `raw_metadata_type = 'orcid'`) and upsert them into the `people` table at
/// `people_path` (which may be the same file as the main `commonmeta.sqlite3`).
///
/// Two storage formats are accepted transparently:
/// - **Commonmeta JSON** (written by dragoman before `fetch_orcid_with_json`
///   was available): top-level `"type": "Person"` field.
/// - **Raw ORCID 3.0 person JSON** (written by dragoman using
///   [`fetch_orcid_with_json`]).
///
/// Returns the number of rows upserted.  FTS5 is rebuilt once at the end.
pub fn stream_cache_orcid_to_people_sqlite(
    cache_path: &Path,
    people_path: &Path,
) -> Result<usize> {
    use rusqlite::Connection;

    let cache_conn = Connection::open(cache_path)
        .map_err(|e| Error::Parse(format!("open cache '{}': {}", cache_path.display(), e)))?;

    let mut stmt = cache_conn
        .prepare(
            "SELECT pid, raw_metadata FROM pid_records WHERE raw_metadata_type = 'orcid'",
        )
        .map_err(|e| Error::Parse(format!("prepare cache query: {}", e)))?;

    let rows: Vec<(String, String)> = stmt
        .query_map([], |row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)))
        .map_err(|e| Error::Parse(format!("query cache: {}", e)))?
        .filter_map(|r| r.ok())
        .collect();

    if rows.is_empty() {
        return Ok(0);
    }

    let people_conn = Connection::open(people_path)
        .map_err(|e| Error::Parse(format!("open people db '{}': {}", people_path.display(), e)))?;
    let _: String = people_conn
        .query_row("PRAGMA journal_mode=WAL", [], |r| r.get(0))
        .map_err(|e| Error::Parse(format!("WAL mode: {}", e)))?;
    ensure_people_schema(&people_conn)?;

    let mut batch: Vec<PeopleRow> = Vec::with_capacity(rows.len());
    for (pid, raw) in &rows {
        let raw_bytes = raw.as_bytes();
        let val: serde_json::Value = match serde_json::from_str(raw) {
            Ok(v) => v,
            Err(_) => {
                eprintln!("orcid cache: skipping '{}': JSON parse error", pid);
                continue;
            }
        };

        let row = if val.get("type").and_then(|v| v.as_str()) == Some("Person") {
            // Commonmeta JSON (dragoman < 0.3.16).
            let data: crate::Data = match serde_json::from_value(val) {
                Ok(d) => d,
                Err(e) => {
                    eprintln!("orcid cache: skipping '{}': {}", pid, e);
                    continue;
                }
            };
            let orcid = pid.trim_start_matches("https://orcid.org/").to_string();
            let other_names = data.additional_names.join(" ");
            let ext_ids: Vec<serde_json::Value> = data
                .identifiers
                .iter()
                .map(|i| serde_json::json!({"type": i.identifier_type, "value": i.identifier}))
                .collect();
            let external_identifiers = serde_json::to_string(&ext_ids).unwrap_or_default();
            let affiliations = {
                let aff_vals: Vec<serde_json::Value> = data.affiliations.iter().map(|a| {
                    let mut obj = serde_json::Map::new();
                    if !a.identifier.is_empty() { obj.insert("identifier".into(), serde_json::Value::String(a.identifier.clone())); }
                    if !a.identifier_type.is_empty() { obj.insert("identifier_type".into(), serde_json::Value::String(a.identifier_type.clone())); }
                    obj.insert("name".into(), serde_json::Value::String(a.name.clone()));
                    if !a.role.is_empty() { obj.insert("role".into(), serde_json::Value::String(a.role.clone())); }
                    if !a.start_date.is_empty() { obj.insert("start_date".into(), serde_json::Value::String(a.start_date.clone())); }
                    if !a.end_date.is_empty() { obj.insert("end_date".into(), serde_json::Value::String(a.end_date.clone())); }
                    serde_json::Value::Object(obj)
                }).collect();
                if aff_vals.is_empty() { String::new() } else { serde_json::to_string(&aff_vals).unwrap_or_default() }
            };
            let metadata = zstd::encode_all(raw_bytes, 0).unwrap_or_else(|_| raw_bytes.to_vec());
            PeopleRow {
                id: data.id,
                orcid,
                given_name: data.given_name,
                family_name: data.family_name,
                credit_name: data.name.clone(),
                name: if !data.title.is_empty() { data.title } else { data.name },
                country: data.country,
                other_names,
                external_identifiers,
                affiliations,
                date_updated: data.date_updated,
                valid: false,
                metadata,
            }
        } else {
            // Raw ORCID 3.0 person JSON (dragoman >= 0.3.16).
            let person: PersonJson = match serde_json::from_value(val) {
                Ok(p) => p,
                Err(e) => {
                    eprintln!("orcid cache: skipping '{}': {}", pid, e);
                    continue;
                }
            };
            let data = person_json_to_data(&person, pid);
            let other_names = person
                .other_names
                .other_names
                .iter()
                .map(|n| n.content.as_str())
                .filter(|s| !s.is_empty())
                .collect::<Vec<_>>()
                .join(" ");
            let ext_ids: Vec<serde_json::Value> = person
                .external_identifiers
                .identifiers
                .iter()
                .filter(|e| !e.type_.is_empty() && !e.value.is_empty())
                .map(|e| serde_json::json!({"type": e.type_, "value": e.value}))
                .collect();
            let external_identifiers = serde_json::to_string(&ext_ids).unwrap_or_default();
            let metadata = zstd::encode_all(raw_bytes, 0).unwrap_or_else(|_| raw_bytes.to_vec());
            let orcid = pid.trim_start_matches("https://orcid.org/").to_string();
            PeopleRow {
                id: data.id,
                orcid,
                given_name: data.given_name,
                family_name: data.family_name,
                credit_name: data.name.clone(),
                name: if !data.title.is_empty() { data.title } else { data.name },
                country: data.country,
                other_names,
                external_identifiers,
                affiliations: String::new(),
                date_updated: data.date_updated,
                valid: false,
                metadata,
            }
        };

        batch.push(row);
    }

    if batch.is_empty() {
        return Ok(0);
    }

    let n = write_people_batch(&people_conn, &batch)?;
    people_conn
        .execute("INSERT INTO people_fts(people_fts) VALUES('rebuild')", [])
        .map_err(|e| Error::Parse(format!("FTS5 rebuild: {}", e)))?;
    Ok(n)
}

// ── Lookup ────────────────────────────────────────────────────────────────────

/// Look up a person by their ORCID iD or URL from a local SQLite database.
/// Accepts a bare ORCID iD or a full ORCID URL.
/// Handles both XML blobs (from [`stream_summaries_to_sqlite`]) and JSON blobs
/// (from [`import_person`]).
pub fn fetch_sqlite(id: &str, db_path: &Path) -> Result<Data> {
    use rusqlite::{params, Connection};

    let url = crate::utils::normalize_orcid(id);
    if url.is_empty() {
        return Err(Error::Parse(format!("'{}' is not a valid ORCID identifier", id)));
    }

    let conn = Connection::open(db_path)
        .map_err(|e| Error::Parse(format!("open sqlite '{}': {}", db_path.display(), e)))?;
    let (blob, affiliations_json): (Vec<u8>, String) = conn
        .query_row(
            "SELECT metadata, COALESCE(affiliations, '') FROM people WHERE id = ?1 LIMIT 1",
            params![url],
            |row| Ok((row.get(0)?, row.get(1)?)),
        )
        .map_err(|e| {
            if matches!(e, rusqlite::Error::QueryReturnedNoRows) {
                Error::Parse(format!("person '{}' not found in local database", id))
            } else {
                Error::Parse(format!("sqlite query failed: {}", e))
            }
        })?;
    let raw = zstd::decode_all(std::io::Cursor::new(&blob))
        .map_err(|e| Error::Parse(format!("decompress metadata for '{}': {}", id, e)))?;

    // Records from stream_summaries_to_sqlite store XML; records from
    // import_person store JSON. Detect by the first non-whitespace byte.
    if raw.iter().find(|&&b| b != b' ' && b != b'\n' && b != b'\r' && b != b'\t') == Some(&b'<') {
        let record = parse_orcid_xml(&raw)
            .ok_or_else(|| Error::Parse(format!("re-parse XML for '{}'", id)))?;
        Ok(from_orcid(record))
    } else {
        let person: PersonJson = serde_json::from_slice(&raw)
            .map_err(|e| Error::Parse(format!("re-parse JSON for '{}': {}", id, e)))?;
        let mut data = person_json_to_data(&person, &url);
        // For JSON-stored records (API import), populate affiliations from the
        // dedicated column which was populated by fetch_person_affiliations.
        if data.affiliations.is_empty() && !affiliations_json.is_empty() {
            if let Ok(vals) = serde_json::from_str::<Vec<serde_json::Value>>(&affiliations_json) {
                data.affiliations = vals
                    .iter()
                    .filter_map(|v| {
                        let name = v.get("name")?.as_str()?;
                        {
                            let (identifier, identifier_type) = if let Some(id_str) = v.get("identifier").and_then(|s| s.as_str()) {
                                (id_str.to_string(), v.get("identifier_type").and_then(|s| s.as_str()).unwrap_or("").to_string())
                            } else if let Some(old_id) = v.get("id").and_then(|s| s.as_str()) {
                                (old_id.to_string(), "ROR".to_string())
                            } else {
                                (String::new(), String::new())
                            };
                            Some(crate::data::Affiliation {
                                identifier,
                                identifier_type,
                                name: name.to_string(),
                                role: v.get("role").and_then(|s| s.as_str()).unwrap_or("").to_string(),
                                start_date: v.get("start_date").and_then(|s| s.as_str()).unwrap_or("").to_string(),
                                end_date: v.get("end_date").and_then(|s| s.as_str()).unwrap_or("").to_string(),
                                ..Default::default()
                            })
                        }
                    })
                    .collect();
            }
        }
        Ok(data)
    }
}

// ── Figshare release helpers ──────────────────────────────────────────────────

#[derive(Deserialize)]
struct FigshareFile {
    name: String,
    size: u64,
    download_url: String,
}

#[derive(Deserialize)]
struct FigshareArticle {
    files: Vec<FigshareFile>,
}

/// Fetch metadata for the ORCID summaries file from the figshare article API
/// (no download). Pass the figshare article `version` (1-indexed; 1 for the
/// 2025 dataset).
///
/// The returned [`OrcidRelease`] gives you the filename, download URL, and
/// size so you can download the ~46 GB file yourself and pass its local path
/// to [`stream_summaries_to_sqlite`].
pub fn fetch_orcid_release(version: u32) -> Result<OrcidRelease> {
    let url = format!(
        "https://api.figshare.com/v2/articles/{}/versions/{}",
        ORCID_FIGSHARE_ARTICLE_ID, version
    );
    let client = reqwest::blocking::Client::builder()
        .user_agent(format!(
            "commonmeta-rs/{} (https://github.com/front-matter/commonmeta-rs; mailto:info@front-matter.de)",
            env!("CARGO_PKG_VERSION")
        ))
        .build()
        .map_err(|e| Error::Http(e.to_string()))?;
    let text = client
        .get(&url)
        .send()
        .map_err(|e| Error::Http(format!("figshare API: {}", e)))?
        .error_for_status()
        .map_err(|e| Error::Http(format!("figshare API status: {}", e)))?
        .text()
        .map_err(|e| Error::Http(e.to_string()))?;
    let article: FigshareArticle =
        serde_json::from_str(&text).map_err(|e| Error::Parse(e.to_string()))?;
    let summaries = article
        .files
        .into_iter()
        .find(|f| f.name.ends_with("_summaries.tar.gz"))
        .ok_or_else(|| Error::Parse("no summaries tar.gz found in figshare article".into()))?;
    let (year, batch) = parse_filename_date(&summaries.name);
    Ok(OrcidRelease {
        year,
        batch,
        filename: summaries.name,
        download_url: summaries.download_url,
        size_bytes: summaries.size,
    })
}

/// Extract year and batch number from `ORCID_{year}_{batch}_summaries.tar.gz`.
fn parse_filename_date(name: &str) -> (u16, u16) {
    let parts: Vec<&str> = name.split('_').collect();
    let year = parts.get(1).and_then(|s| s.parse().ok()).unwrap_or(0);
    let batch = parts.get(2).and_then(|s| s.parse().ok()).unwrap_or(0);
    (year, batch)
}

/// Fetch the latest ORCID Public Data File release from figshare.
///
/// Uses `GET /v2/articles/{id}/versions` (public, no auth) to find the highest
/// version number, then calls [`fetch_orcid_release`] for the file metadata.
pub fn fetch_latest_orcid_release() -> Result<OrcidRelease> {
    #[derive(Deserialize)]
    struct VersionEntry {
        version: u32,
    }
    let url = format!(
        "https://api.figshare.com/v2/articles/{}/versions",
        ORCID_FIGSHARE_ARTICLE_ID
    );
    let client = reqwest::blocking::Client::builder()
        .user_agent(format!(
            "commonmeta-rs/{} (https://github.com/front-matter/commonmeta-rs; mailto:info@front-matter.de)",
            env!("CARGO_PKG_VERSION")
        ))
        .build()
        .map_err(|e| Error::Http(e.to_string()))?;
    let text = client
        .get(&url)
        .send()
        .map_err(|e| Error::Http(format!("figshare versions API: {}", e)))?
        .error_for_status()
        .map_err(|e| Error::Http(format!("figshare versions API status: {}", e)))?
        .text()
        .map_err(|e| Error::Http(e.to_string()))?;
    let entries: Vec<VersionEntry> =
        serde_json::from_str(&text).map_err(|e| Error::Parse(e.to_string()))?;
    let latest = entries
        .into_iter()
        .map(|e| e.version)
        .max()
        .ok_or_else(|| Error::Parse("figshare returned empty versions list".into()))?;
    fetch_orcid_release(latest)
}

/// Drop and rebuild the `people_fts` FTS5 virtual table.
pub fn rebuild_people_fts(path: &Path) -> Result<()> {
    let conn = rusqlite::Connection::open(path)
        .map_err(|e| Error::Parse(format!("open sqlite '{}': {}", path.display(), e)))?;
    let _ = conn.execute("DROP TABLE IF EXISTS people_fts", []);
    conn.execute_batch(PEOPLE_FTS5_DDL)
        .map_err(|e| Error::Parse(format!("people_fts DDL: {}", e)))?;
    conn.execute("INSERT INTO people_fts(people_fts) VALUES('rebuild')", [])
        .map_err(|e| Error::Parse(format!("people_fts rebuild: {}", e)))?;
    Ok(())
}

/// Read the installed ORCID Public Data File version from the `settings` table.
/// Returns `None` when no version has been recorded yet.
pub fn fetch_installed_orcid_public_data_version(db_path: &Path) -> Result<Option<String>> {
    use rusqlite::Connection;
    use rusqlite::Error as SqliteError;
    let conn = Connection::open(db_path)
        .map_err(|e| Error::Parse(format!("failed to open sqlite: {}", e)))?;
    match conn.query_row(
        "SELECT value FROM settings WHERE key = 'orcid_public_data_version' LIMIT 1",
        [],
        |row| row.get::<_, String>(0),
    ) {
        Ok(v) => Ok(Some(v)),
        Err(SqliteError::QueryReturnedNoRows) => Ok(None),
        Err(_) => Ok(None),
    }
}

/// Probe `url` with a HEAD request (no redirect follow) to extract the filename
/// from `Content-Disposition` and the file size from `Content-Length`.
/// Returns `(filename, size_bytes)`.  Both fall back to safe defaults on failure.
fn probe_download_url(url: &str) -> (String, u64) {
    let client = match reqwest::blocking::Client::builder()
        .user_agent(format!(
            "commonmeta-rs/{} (https://github.com/front-matter/commonmeta-rs; mailto:info@front-matter.de)",
            env!("CARGO_PKG_VERSION")
        ))
        .redirect(reqwest::redirect::Policy::none())
        .timeout(std::time::Duration::from_secs(30))
        .build()
    {
        Ok(c) => c,
        Err(_) => return ("orcid_summaries.tar.gz".to_string(), 0),
    };
    let resp = match client.head(url).send() {
        Ok(r) => r,
        Err(_) => return ("orcid_summaries.tar.gz".to_string(), 0),
    };
    let size = resp.content_length().unwrap_or(0);
    let filename = resp
        .headers()
        .get("content-disposition")
        .and_then(|v| v.to_str().ok())
        .and_then(|s| {
            s.split(';')
                .find_map(|part| {
                    let part = part.trim();
                    part.strip_prefix("filename=").map(|f| f.trim_matches('"').to_string())
                })
        })
        .filter(|f| f.ends_with(".tar.gz"))
        .unwrap_or_else(|| "orcid_summaries.tar.gz".to_string());
    (filename, size)
}

/// Download the ORCID Public Data File (summaries tar.gz) if it is not already
/// cached, then stream it into the `people` table at `output_path`.
///
const ORCID_SAMPLE_SIZE: usize = 1_000;

/// Download or load the ORCID Public Data File summaries and stream them into
/// the `people` table at `output_path`.
///
/// `source` can be:
/// - `None` — auto-discover the latest release from the figshare API
/// - `Some(local_path)` — stream directly from a local `.tar.gz` file (no network)
/// - `Some(url)` starting with `http` — download from that URL, bypassing the
///   figshare API (get the URL via `--list-releases` on a machine that can reach
///   `api.figshare.com`)
///
/// When `sample` is true, stops after [`ORCID_SAMPLE_SIZE`] valid records.
/// For a URL source with no cache, samples without downloading the full file.
pub fn import_orcid_public_data(
    output_path: &Path,
    source: Option<&str>,
    no_network: bool,
    sample: bool,
) -> Result<usize> {
    use rusqlite::params;
    use rusqlite::Connection;

    let limit = if sample { ORCID_SAMPLE_SIZE } else { 0 };

    fn write_version(output_path: &Path, version_key: &str) -> Result<()> {
        if version_key.starts_with("0_") {
            return Ok(());
        }
        let conn = Connection::open(output_path)
            .map_err(|e| Error::Parse(format!("open sqlite: {}", e)))?;
        conn.execute(
            "INSERT OR REPLACE INTO settings (key, value) VALUES ('orcid_public_data_version', ?1)",
            params![version_key],
        )
        .map_err(|e| Error::Parse(format!("write settings: {}", e)))?;
        Ok(())
    }

    // ── Local file: stream directly, no network ───────────────────────────────
    if let Some(s) = source {
        if !s.starts_with("http") {
            let local = std::path::Path::new(s);
            if !local.exists() {
                return Err(Error::Parse(format!("file not found: {}", s)));
            }
            let filename = local
                .file_name()
                .and_then(|n| n.to_str())
                .unwrap_or("orcid_summaries.tar.gz")
                .to_string();
            let (year, batch) = parse_filename_date(&filename);
            let version_key = format!("{}_{}", year, batch);
            eprintln!("orcid: importing {} …", local.display());
            let count = stream_summaries_to_sqlite(local, output_path, limit)?;
            if !sample {
                write_version(output_path, &version_key)?;
            }
            eprintln!("orcid: import complete — {} people", count);
            return Ok(count);
        }
    }

    if no_network {
        return Err(Error::Parse(
            "--no-network: pass a local file path or remove --no-network".to_string(),
        ));
    }

    // ── HTTP URL provided directly → download to cache (or sample-stream) ────
    // Full import: single sequential GET → cache file → stream_summaries_to_sqlite.
    // Sample without cache: stream the first N records directly without writing
    // the full ~46 GB to disk (same pattern as DataCite --sample without cache).
    if let Some(url) = source.filter(|s| s.starts_with("http")) {
        if url.contains("/ndownloader/articles/") {
            return Err(Error::Parse(
                "that URL downloads the full article bundle (~221 GB, all files).\n\
                Pass the summaries-only file (~46 GB) instead.\n\
                On a machine where figshare is reachable:\n  \
                commonmeta import --from orcid --list-releases\n\
                Or download the file there and transfer it (rsync is resumable if interrupted):\n  \
                rsync -az --progress ORCID_2025_10_summaries.tar.gz user@server:/data/\n  \
                commonmeta import --from orcid /data/ORCID_2025_10_summaries.tar.gz"
                    .to_string(),
            ));
        }
        // Derive version from the filename embedded in the URL path
        // (e.g. ORCID_2025_10_summaries.tar.gz). No HEAD request needed.
        let url_filename = url
            .rsplit('/')
            .next()
            .filter(|f| f.ends_with(".tar.gz"))
            .unwrap_or("orcid_summaries.tar.gz")
            .to_string();
        let (year, batch) = parse_filename_date(&url_filename);
        let version_key = format!("{}_{}", year, batch);
        let installed = if output_path.exists() {
            fetch_installed_orcid_public_data_version(output_path)?
        } else {
            None
        };
        if !sample && !version_key.starts_with("0_") && installed.as_deref() == Some(&version_key) {
            eprintln!("orcid: {} already installed", version_key);
            return Ok(0);
        }
        let cache_path = crate::io_utils::cache_dir("orcid").join(&url_filename);
        let cached = cache_path.exists() && cache_path.metadata().map(|m| m.len()).unwrap_or(0) > 0;
        if sample && !cached {
            // Sample without cache: stream first N records directly, skip disk write.
            eprintln!("orcid: streaming first {} records from {} (no cache)", ORCID_SAMPLE_SIZE, url);
            let client = reqwest::blocking::Client::builder()
                .timeout(std::time::Duration::from_secs(6 * 3600))
                .build()
                .map_err(|e| Error::Http(e.to_string()))?;
            let raw = client
                .get(url)
                .send()
                .map_err(|e| Error::Http(format!("GET {url}: {e}")))?;
            if raw.status().as_u16() == 403 && url.contains("figshare.com") {
                return Err(figshare_blocked_err(url));
            }
            let resp = raw
                .error_for_status()
                .map_err(|e| Error::Http(format!("HTTP error: {e}")))?;
            let conn = open_people_conn(output_path)?;
            let buffered = std::io::BufReader::with_capacity(256 * 1024, resp);
            let total = drain_tar_into_conn(&conn, buffered, limit)?;
            return finish_people_import(&conn, total);
        }
        download_summaries_to_cache(url, &cache_path)?;
        eprintln!("orcid: importing {} into {} …", url_filename, output_path.display());
        let count = stream_summaries_to_sqlite(&cache_path, output_path, limit)?;
        if !sample {
            write_version(output_path, &version_key)?;
        }
        eprintln!("orcid: import complete — {} people", count);
        return Ok(count);
    }

    // ── Resolve download URL and filename (figshare auto-discover) ────────────
    let (download_url, filename, size_bytes) = match source {
        Some(_) => unreachable!("non-http source handled above as local file"),
        None => {
            eprintln!("orcid: checking latest release on figshare …");
            let release = fetch_latest_orcid_release().map_err(|e| {
                let msg = e.to_string();
                if msg.contains("403") {
                    Error::Http(format!(
                        "{msg}\n\
                        figshare is blocked on this host. Options:\n\
                        A) download the file on another machine and transfer it here:\n  \
                           commonmeta import --from orcid --list-releases   # get URL\n  \
                           wget <SUMMARIES_URL>   # ~46 GB\n  \
                           rsync -az --progress ORCID_*_summaries.tar.gz user@server:/data/\n  \
                           commonmeta import --from orcid /data/ORCID_*_summaries.tar.gz\n\
                        B) pass the URL directly (from --list-releases on another machine):\n  \
                           commonmeta import --from orcid \"<SUMMARIES_URL>\""
                    ))
                } else {
                    e
                }
            })?;
            eprintln!(
                "orcid: {} ({}_{}) {:.1} GB",
                release.filename, release.year, release.batch,
                release.size_bytes as f64 / 1_073_741_824.0,
            );
            (release.download_url, release.filename, release.size_bytes)
        }
    };

    let (year, batch) = parse_filename_date(&filename);
    let version_key = format!("{}_{}", year, batch);

    let installed = if output_path.exists() {
        fetch_installed_orcid_public_data_version(output_path)?
    } else {
        None
    };
    if !sample && !version_key.starts_with("0_") && installed.as_deref() == Some(&version_key) {
        eprintln!("orcid: {} already installed", version_key);
        return Ok(0);
    }

    let cache_path = crate::io_utils::cache_dir("orcid").join(&filename);
    let cached = cache_path.exists()
        && size_bytes > 0
        && cache_path.metadata().map(|m| m.len()).unwrap_or(0) == size_bytes;

    if sample && !cached {
        eprintln!(
            "orcid: streaming first {} records from {} (no cache) …",
            ORCID_SAMPLE_SIZE, download_url
        );
        let client = reqwest::blocking::Client::builder()
            .timeout(std::time::Duration::from_secs(6 * 3600))
            .build()
            .map_err(|e| Error::Http(e.to_string()))?;
        let raw = client
            .get(&download_url)
            .send()
            .map_err(|e| Error::Http(format!("GET: {e}")))?;
        if raw.status().as_u16() == 403 && download_url.contains("figshare.com") {
            return Err(figshare_blocked_err(&download_url));
        }
        let resp = raw
            .error_for_status()
            .map_err(|e| Error::Http(format!("HTTP error: {e}")))?;
        let conn = open_people_conn(output_path)?;
        let buffered = std::io::BufReader::with_capacity(256 * 1024, resp);
        let total = drain_tar_into_conn(&conn, buffered, limit)?;
        return finish_people_import(&conn, total);
    }

    if !cached {
        eprintln!("orcid: downloading to {} …", cache_path.display());
        if let Some(parent) = cache_path.parent() {
            std::fs::create_dir_all(parent)
                .map_err(|e| Error::Parse(format!("mkdir: {}", e)))?;
        }
        download_summaries_to_cache(&download_url, &cache_path)
            .map_err(|e| Error::Parse(format!("download: {}", e)))?;
    } else {
        eprintln!("orcid: using cached {}", cache_path.display());
    }

    eprintln!("orcid: importing {} into {} …", filename, output_path.display());
    let count = stream_summaries_to_sqlite(&cache_path, output_path, limit)?;
    if !sample {
        write_version(output_path, &version_key)?;
    }
    eprintln!("orcid: import complete — {} people", count);
    Ok(count)
}

// ── Employment / affiliations ─────────────────────────────────────────────────

#[derive(Deserialize, Default)]
#[serde(default)]
struct EmploymentsResponse {
    #[serde(rename = "affiliation-group")]
    affiliation_groups: Vec<AffiliationGroupJson>,
}

#[derive(Deserialize, Default)]
#[serde(default)]
struct AffiliationGroupJson {
    summaries: Vec<AffiliationSummaryWrapper>,
}

#[derive(Deserialize, Default)]
#[serde(default)]
struct AffiliationSummaryWrapper {
    #[serde(rename = "employment-summary")]
    employment_summary: Option<EmploymentSummaryJson>,
}

#[derive(Deserialize, Default)]
#[serde(default)]
struct EmploymentSummaryJson {
    organization: OrganizationJson,
    #[serde(rename = "department-name")]
    department_name: String,
    #[serde(rename = "role-title")]
    role_title: String,
    #[serde(rename = "start-date")]
    start_date: Option<OrcidDateJson>,
    #[serde(rename = "end-date")]
    end_date: Option<OrcidDateJson>,
}

#[derive(Deserialize, Default)]
#[serde(default)]
struct OrganizationJson {
    name: String,
    #[serde(rename = "disambiguated-organization")]
    disambiguated: Option<DisambiguatedOrgJson>,
}

#[derive(Deserialize, Default)]
#[serde(default)]
struct DisambiguatedOrgJson {
    #[serde(rename = "disambiguated-organization-identifier")]
    identifier: String,
    #[serde(rename = "disambiguation-source")]
    source: String,
}

#[derive(Deserialize, Default)]
#[serde(default)]
struct OrcidDateJson {
    year: Option<StringValue>,
    month: Option<StringValue>,
    day: Option<StringValue>,
}

pub struct PersonAffiliation {
    pub identifier: Option<String>,
    pub identifier_type: Option<String>,
    pub name: String,
    pub role: Option<String>,
    pub start_date: Option<String>,
    pub end_date: Option<String>,
}

/// Pre-loaded lookup index for bulk affiliation fill. Opens one connection and loads
/// the Ringgold→ROR map once so the hot batch loop has O(1) Ringgold lookups with
/// no per-record connection opens.
struct FillConn {
    conn: rusqlite::Connection,
    ringgold: std::collections::HashMap<String, (String, String)>,
}

impl FillConn {
    fn new(db_path: &std::path::Path) -> Result<Self> {
        let conn = rusqlite::Connection::open(db_path)
            .map_err(|e| Error::Parse(format!("open fill db: {}", e)))?;
        let ringgold = Self::load_ringgold_map(&conn)?;
        Ok(Self { conn, ringgold })
    }

    fn load_ringgold_map(
        conn: &rusqlite::Connection,
    ) -> Result<std::collections::HashMap<String, (String, String)>> {
        let mut stmt = conn
            .prepare(
                "SELECT o.id, o.name, ext.value \
                 FROM organizations o, json_each(o.external_ids) ext \
                 WHERE LOWER(json_extract(ext.value, '$.type')) = 'ringgold'",
            )
            .map_err(|e| Error::Parse(format!("prepare ringgold map: {}", e)))?;
        let mut map = std::collections::HashMap::new();
        let rows = stmt
            .query_map(rusqlite::params![], |row| {
                Ok((
                    row.get::<_, String>(0)?,
                    row.get::<_, String>(1)?,
                    row.get::<_, String>(2)?,
                ))
            })
            .map_err(|e| Error::Parse(format!("query ringgold map: {}", e)))?;
        for row in rows.flatten() {
            let (ror_url, name, ext_json) = row;
            if let Ok(val) = serde_json::from_str::<serde_json::Value>(&ext_json) {
                if let Some(preferred) = val.get("preferred").and_then(|v| v.as_str()) {
                    if !preferred.is_empty() {
                        map.entry(preferred.to_string())
                            .or_insert_with(|| (ror_url.clone(), name.clone()));
                    }
                }
                if let Some(all) = val.get("all").and_then(|v| v.as_array()) {
                    for item in all {
                        if let Some(id) = item.as_str() {
                            map.entry(id.to_string())
                                .or_insert_with(|| (ror_url.clone(), name.clone()));
                        }
                    }
                }
            }
        }
        Ok(map)
    }

    fn lookup_ringgold(&self, id: &str) -> Option<(String, String)> {
        self.ringgold.get(id).cloned()
    }

    fn lookup_ror(&self, id: &str) -> Option<(String, String)> {
        let (normalized, id_type) = crate::utils::validate_id(id);
        if id_type != "ROR" {
            return None;
        }
        let ror_url = format!("https://ror.org/{}", normalized);
        self.conn
            .query_row(
                "SELECT id, name FROM organizations WHERE id = ?1 LIMIT 1",
                rusqlite::params![ror_url],
                |row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)),
            )
            .ok()
    }

    fn lookup_by_external_id(&self, ext_type: &str, ext_value: &str) -> Option<(String, String)> {
        let ror_url = self.conn.query_row(
            "SELECT o.id FROM organizations o, json_each(o.external_ids) ext \
             WHERE LOWER(json_extract(ext.value, '$.type')) = ?1 \
               AND (json_extract(ext.value, '$.preferred') = ?2 \
                    OR EXISTS ( \
                        SELECT 1 FROM json_each(json_extract(ext.value, '$.all')) a \
                        WHERE a.value = ?2 \
                    )) \
             LIMIT 1",
            rusqlite::params![ext_type, ext_value],
            |row| row.get::<_, String>(0),
        ).ok()?;
        self.conn.query_row(
            "SELECT id, name FROM organizations WHERE id = ?1 LIMIT 1",
            rusqlite::params![ror_url],
            |row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)),
        ).ok()
    }
}

/// Rename the legacy `"id"` field to `"identifier"` in a stored affiliations JSON string.
/// Old records stored ROR URLs as `"id"` before the schema settled on `"identifier"`.
/// Returns `(normalized_json, changed)`.
fn normalize_affiliations_json(affs_json: &str) -> (String, bool) {
    if affs_json.is_empty() {
        return (affs_json.to_string(), false);
    }
    let mut affs: Vec<serde_json::Value> = match serde_json::from_str(affs_json) {
        Ok(v) => v,
        Err(_) => return (affs_json.to_string(), false),
    };
    let mut changed = false;
    for aff in affs.iter_mut() {
        if let Some(obj) = aff.as_object_mut() {
            if !obj.contains_key("identifier") {
                if let Some(id_val) = obj.remove("id") {
                    obj.insert("identifier".to_string(), id_val);
                    obj.entry("identifier_type")
                        .or_insert_with(|| serde_json::Value::String("ROR".to_string()));
                    changed = true;
                }
            }
        }
    }
    if changed {
        (serde_json::to_string(&affs).unwrap_or_else(|_| affs_json.to_string()), true)
    } else {
        (affs_json.to_string(), false)
    }
}

fn fill_affiliations_json(affs_json: &str, fill: &FillConn) -> String {
    // Normalize legacy "id" field before filling.
    let (affs_json, pre_changed) = normalize_affiliations_json(affs_json);
    let affs_json = affs_json.as_str();

    let mut affs: Vec<serde_json::Value> = match serde_json::from_str(affs_json) {
        Ok(v) => v,
        Err(_) => return affs_json.to_string(),
    };
    let mut changed = pre_changed;
    for aff in affs.iter_mut() {
        let id = match aff.get("identifier").and_then(|v| v.as_str()) {
            Some(s) if !s.is_empty() => s.to_string(),
            _ => continue,
        };
        let id_type = aff.get("identifier_type")
            .and_then(|v| v.as_str())
            .unwrap_or("")
            .to_uppercase();
        let resolved: Option<(String, String)> = match id_type.as_str() {
            "RINGGOLD" => fill.lookup_ringgold(&id),
            "ROR" => {
                let name_empty = aff.get("name").and_then(|v| v.as_str()).unwrap_or("").is_empty();
                if name_empty { fill.lookup_ror(&id) } else { None }
            }
            "GRID" | "ISNI" | "FUNDREF" | "WIKIDATA" => {
                let ext_type = match id_type.as_str() {
                    "FUNDREF" => "fundref",
                    "WIKIDATA" => "wikidata",
                    "ISNI" => "isni",
                    _ => "grid",
                };
                fill.lookup_by_external_id(ext_type, &id)
            }
            _ => None,
        };
        if let Some((ror_url, org_name)) = resolved {
            if let Some(obj) = aff.as_object_mut() {
                obj.insert("identifier".into(), serde_json::Value::String(ror_url));
                obj.insert("identifier_type".into(), serde_json::Value::String("ROR".into()));
                let name_empty = obj.get("name").and_then(|v| v.as_str()).unwrap_or("").is_empty();
                if name_empty && !org_name.is_empty() {
                    obj.insert("name".into(), serde_json::Value::String(org_name));
                }
                changed = true;
            }
        }
    }
    if changed {
        serde_json::to_string(&affs).unwrap_or_else(|_| affs_json.to_string())
    } else {
        affs_json.to_string()
    }
}

/// Resolve Ringgold, GRID, ISNI, FundRef, and Wikidata identifiers in the `affiliations`
/// column of every row in the `people` table. Runs as a post-import pass so the bulk
/// import itself stays fast. Opens one connection and pre-loads the Ringgold→ROR map.
pub fn fill_people_sqlite(people_db_path: &Path, org_db_path: &Path) -> Result<usize> {
    eprintln!("fill: loading organization index…");
    let fill = FillConn::new(org_db_path)?;
    eprintln!("fill: Ringgold index loaded ({} entries)", fill.ringgold.len());

    let conn = rusqlite::Connection::open(people_db_path)
        .map_err(|e| Error::Parse(format!("open people db: {}", e)))?;

    let mut updated = 0usize;
    let mut last_id = String::new();
    loop {
        let batch: Vec<(String, String)> = {
            let mut stmt = conn
                .prepare(
                    "SELECT id, affiliations FROM people \
                     WHERE affiliations != '' AND id > ?1 \
                     ORDER BY id LIMIT 1000",
                )
                .map_err(|e| Error::Parse(format!("prepare fill: {}", e)))?;
            stmt.query_map(rusqlite::params![last_id], |row| {
                Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
            })
            .map_err(|e| Error::Parse(format!("query fill: {}", e)))?
            .flatten()
            .collect()
        };
        if batch.is_empty() {
            break;
        }
        last_id = batch.last().unwrap().0.clone();

        let tx = conn
            .unchecked_transaction()
            .map_err(|e| Error::Parse(format!("begin fill tx: {}", e)))?;
        for (id, affs_json) in &batch {
            let new_json = fill_affiliations_json(affs_json, &fill);
            if new_json != *affs_json {
                tx.execute(
                    "UPDATE people SET affiliations = ?1 WHERE id = ?2",
                    rusqlite::params![new_json, id],
                )
                .map_err(|e| Error::Parse(format!("update '{}': {}", id, e)))?;
                updated += 1;
            }
        }
        tx.commit()
            .map_err(|e| Error::Parse(format!("commit fill tx: {}", e)))?;
    }
    eprintln!("fill: {} people affiliations updated", updated);
    Ok(updated)
}

/// Validate all records (or a filtered subset) in the `people` table against the
/// `$defs/person` sub-schema of the commonmeta v1.0rc1 JSON schema.
///
/// The person JSON is assembled from structured columns rather than the compressed
/// XML blob so that the same normalisation rules apply across all import paths.
///
/// Returns a [`ValidationReport`] with per-record error details.
pub fn validate_people_sqlite(
    path: &Path,
    id: Option<&str>,
    limit: usize,
    fix: bool,
    recheck: bool,
    sample: bool,
) -> Result<crate::formats::commonmeta::ValidationReport> {
    use serde_json::Value;
    use crate::schema_utils::SCHEMA_JSON;
    use crate::formats::commonmeta::{ValidationError, ValidationReport};
    use crate::schema_utils::collect_leaf_errors;

    let schema_json: Value = serde_json::from_str(SCHEMA_JSON)
        .map_err(|e| Error::Parse(format!("failed to parse commonmeta schema: {e}")))?;
    let defs = schema_json.get("$defs").cloned()
        .unwrap_or(Value::Object(serde_json::Map::new()));
    let schema_id = schema_json.get("$id").cloned()
        .unwrap_or(Value::String("https://commonmeta.org/commonmeta_v1.0rc1.json".to_string()));
    let validation_schema = serde_json::json!({
        "$schema": "https://json-schema.org/draft/2020-12/schema",
        "$id": schema_id,
        "$ref": "#/$defs/person",
        "$defs": defs,
    });
    let compiled = jsonschema::validator_for(&validation_schema)
        .map_err(|e| Error::Parse(format!("failed to compile person schema: {e}")))?;

    let conn = rusqlite::Connection::open(path)
        .map_err(|e| Error::Parse(format!("failed to open '{}': {}", path.display(), e)))?;
    let _ = conn.execute_batch("PRAGMA cache_size=-65536; PRAGMA mmap_size=4294967296;");
    let _ = conn.execute_batch(r#"ALTER TABLE people ADD COLUMN "valid" INTEGER NOT NULL DEFAULT 0;"#);

    const DEFAULT_SAMPLE: usize = 1_000;
    let effective_limit = if sample && limit == 0 { DEFAULT_SAMPLE } else { limit };

    let mut where_parts: Vec<&str> = Vec::new();
    if recheck { where_parts.push(r#"people."valid" = 0"#); }
    if id.is_some() { where_parts.push(r#"people."id" = ?1"#); }
    let where_sql = if where_parts.is_empty() {
        String::new()
    } else {
        format!("WHERE {}", where_parts.join(" AND "))
    };

    let count_sql = if where_sql.is_empty() {
        "SELECT COALESCE(MAX(rowid), 0) FROM people".to_string()
    } else {
        format!("SELECT COUNT(rowid) FROM people {where_sql}")
    };
    let (limit_ph, offset_ph) = if id.is_some() { ("?2", "?3") } else { ("?1", "?2") };
    let order_by = if sample { "RANDOM()" } else { "rowid" };
    let cursor_sql = format!(
        r#"SELECT id, given_name, family_name, name, affiliations FROM people {where_sql}
           ORDER BY {order_by} LIMIT {limit_ph} OFFSET {offset_ph}"#
    );

    let id_param = id.unwrap_or("");
    let row_count: u64 = if id.is_some() {
        conn.query_row(&count_sql, rusqlite::params![id_param], |r| r.get::<_, i64>(0))
    } else {
        conn.query_row(&count_sql, [], |r| r.get::<_, i64>(0))
    }.unwrap_or(0).max(0) as u64;
    let total_to_check = if effective_limit == 0 { row_count } else { row_count.min(effective_limit as u64) };

    let bar = crate::progress::count_bar("validating people", total_to_check);

    let mut stmt = conn.prepare(&cursor_sql)
        .map_err(|e| Error::Parse(e.to_string()))?;

    const BATCH: usize = 10_000;
    let mut valid = 0usize;
    let mut fixed = 0usize;
    let mut report_errors: Vec<ValidationError> = Vec::new();
    let mut offset = 0usize;
    let mut total = 0usize;

    loop {
        let remaining = if effective_limit == 0 { BATCH } else { effective_limit.saturating_sub(total) };
        if remaining == 0 { break; }
        let batch_size = if sample { remaining } else { BATCH.min(remaining) };
        let batch_offset = if recheck || sample { 0 } else { offset };

        let raw_batch: Vec<(String, String, String, String, String)> = {
            let mut rows = if id.is_some() {
                stmt.query(rusqlite::params![id_param, batch_size as i64, batch_offset as i64])
            } else {
                stmt.query(rusqlite::params![batch_size as i64, batch_offset as i64])
            }.map_err(|e| Error::Parse(e.to_string()))?;
            let mut v = Vec::with_capacity(batch_size);
            while let Some(row) = rows.next().map_err(|e| Error::Parse(e.to_string()))? {
                let rid: String         = row.get(0).map_err(|e| Error::Parse(e.to_string()))?;
                let given: String       = row.get(1).map_err(|e| Error::Parse(e.to_string()))?;
                let family: String      = row.get(2).map_err(|e| Error::Parse(e.to_string()))?;
                let name: String        = row.get(3).map_err(|e| Error::Parse(e.to_string()))?;
                let affiliations: String = row.get(4).map_err(|e| Error::Parse(e.to_string()))?;
                v.push((rid, given, family, name, affiliations));
            }
            v
        };

        if raw_batch.is_empty() { break; }
        let batch_count = raw_batch.len();

        let mut passing_ids: Vec<String> = Vec::new();
        // (original_id, new_affiliations?, new_canonical_id?) — written back when fix is enabled.
        let mut fix_updates: Vec<(String, Option<String>, Option<String>)> = Vec::new();
        let mut error_pairs: Vec<(String, Vec<String>)> = Vec::new();

        for (rid, given, family, name, affiliations) in &raw_batch {
            // Normalize legacy "id" → "identifier" in affiliations before validating.
            let (affiliations_norm, aff_changed) = normalize_affiliations_json(affiliations);

            // Normalize non-canonical ORCID URLs (e.g. prod.orcid.org → orcid.org).
            let canonical_id = crate::utils::normalize_orcid(rid);
            let id_changed = !canonical_id.is_empty() && &canonical_id != rid;
            let validation_id = if id_changed { canonical_id.as_str() } else { rid.as_str() };

            let mut obj = serde_json::Map::new();
            if !rid.is_empty()    { obj.insert("id".into(),          Value::String(validation_id.to_string())); }
            if !given.is_empty()  { obj.insert("given_name".into(),  Value::String(given.clone())); }
            if !family.is_empty() { obj.insert("family_name".into(), Value::String(family.clone())); }
            if !name.is_empty()   { obj.insert("name".into(),        Value::String(name.clone())); }
            if !affiliations_norm.is_empty() {
                if let Ok(aff_val) = serde_json::from_str::<Value>(&affiliations_norm) {
                    if aff_val.as_array().map(|a| !a.is_empty()).unwrap_or(false) {
                        obj.insert("affiliations".into(), aff_val);
                    }
                }
            }
            let doc = Value::Object(obj);
            let raw_errs: Vec<jsonschema::ValidationError<'_>> = compiled.iter_errors(&doc).collect();
            let errs = collect_leaf_errors(&raw_errs);
            if errs.is_empty() {
                if fix && (aff_changed || id_changed) {
                    fix_updates.push((
                        rid.clone(),
                        if aff_changed { Some(affiliations_norm) } else { None },
                        if id_changed { Some(canonical_id) } else { None },
                    ));
                } else {
                    passing_ids.push(rid.clone());
                }
                valid += 1;
            } else {
                error_pairs.push((rid.clone(), errs.clone()));
                report_errors.push(ValidationError { id: rid.clone(), errors: errs });
            }
            bar.inc(1);
        }

        let tx = conn.unchecked_transaction()
            .map_err(|e| Error::Parse(format!("begin transaction: {e}")))?;
        for pid in &passing_ids {
            tx.execute(r#"UPDATE people SET "valid" = 1 WHERE id = ?1"#, [pid]).ok();
        }
        for (original_id, new_affs, new_id) in &fix_updates {
            match (new_affs, new_id) {
                (Some(aff), Some(nid)) => {
                    let bare = nid.strip_prefix("https://orcid.org/").unwrap_or(nid.as_str());
                    tx.execute(
                        r#"UPDATE people SET "id" = ?1, "orcid" = ?2, "affiliations" = ?3, "valid" = 1 WHERE id = ?4"#,
                        rusqlite::params![nid, bare, aff, original_id],
                    ).ok();
                    tx.execute(
                        r#"UPDATE works_people SET "person_id" = ?1 WHERE "person_id" = ?2"#,
                        rusqlite::params![nid, original_id],
                    ).ok();
                }
                (Some(aff), None) => {
                    tx.execute(
                        r#"UPDATE people SET "affiliations" = ?1, "valid" = 1 WHERE id = ?2"#,
                        rusqlite::params![aff, original_id],
                    ).ok();
                }
                (None, Some(nid)) => {
                    let bare = nid.strip_prefix("https://orcid.org/").unwrap_or(nid.as_str());
                    tx.execute(
                        r#"UPDATE people SET "id" = ?1, "orcid" = ?2, "valid" = 1 WHERE id = ?3"#,
                        rusqlite::params![nid, bare, original_id],
                    ).ok();
                    tx.execute(
                        r#"UPDATE works_people SET "person_id" = ?1 WHERE "person_id" = ?2"#,
                        rusqlite::params![nid, original_id],
                    ).ok();
                }
                (None, None) => {}
            }
        }
        for (pid, errors) in &error_pairs {
            let _ = errors;
            tx.execute(r#"UPDATE people SET "valid" = 0 WHERE id = ?1"#, [pid]).ok();
        }
        tx.commit().map_err(|e| Error::Parse(format!("commit transaction: {e}")))?;

        total  += batch_count;
        offset += batch_count;
        fixed  += fix_updates.len();
        if batch_count < batch_size { break; }
        if (recheck || sample) && passing_ids.is_empty() && fix_updates.is_empty() { break; }
    }
    bar.finish_and_clear();

    Ok(ValidationReport {
        total,
        valid,
        invalid: report_errors.len(),
        fixed,
        errors: report_errors,
    })
}

/// Resolve Ringgold, GRID, ISNI, FundRef, and Wikidata identifiers in a list of
/// `PersonAffiliation`s to ROR URLs using the local organizations table at `db_path`.
/// Already-resolved ROR entries with an empty name are also filled.
pub fn fill_person_affiliations(affiliations: &mut [PersonAffiliation], db_path: &std::path::Path) {
    for aff in affiliations.iter_mut() {
        let id = match aff.identifier.as_deref() {
            Some(s) if !s.is_empty() => s,
            _ => continue,
        };
        let id_type = aff.identifier_type.as_deref().unwrap_or("").to_uppercase();

        let resolved: Option<(String, String)> = match id_type.as_str() {
            "RINGGOLD" => crate::formats::ror::lookup_ringgold_to_ror(id, db_path),
            "ROR" => {
                if aff.name.is_empty() {
                    crate::formats::ror::lookup_org_sqlite(id, db_path)
                } else {
                    None
                }
            }
            "GRID" | "ISNI" | "FUNDREF" | "WIKIDATA" => {
                let ext_type = match id_type.as_str() {
                    "FUNDREF" => "fundref",
                    "WIKIDATA" => "wikidata",
                    "ISNI" => "isni",
                    _ => "grid",
                };
                ror_id_from_sqlite(ext_type, id, db_path).and_then(|ror_url| {
                    crate::formats::ror::lookup_org_sqlite(&ror_url, db_path)
                })
            }
            _ => None,
        };

        if let Some((ror_url, org_name)) = resolved {
            aff.identifier = Some(ror_url);
            aff.identifier_type = Some("ROR".to_string());
            if aff.name.is_empty() {
                aff.name = org_name;
            }
        }
    }
}

fn orcid_date_to_iso(date: &OrcidDateJson) -> Option<String> {
    let year = date.year.as_ref()?.value.clone();
    if year.is_empty() {
        return None;
    }
    match (&date.month, &date.day) {
        (Some(m), Some(d)) if !m.value.is_empty() && !d.value.is_empty() => {
            Some(format!("{}-{:0>2}-{:0>2}", year, m.value, d.value))
        }
        (Some(m), _) if !m.value.is_empty() => Some(format!("{}-{:0>2}", year, m.value)),
        _ => Some(year),
    }
}

/// Look up a ROR ID in the `organizations` SQLite table by external identifier type and value.
/// `ext_type` must be lowercase (e.g. "grid", "isni", "fundref", "wikidata").
/// Returns the full ROR URL (e.g. `https://ror.org/04wxnsj81`) or `None`.
fn ror_id_from_sqlite(ext_type: &str, ext_value: &str, db_path: &Path) -> Option<String> {
    use rusqlite::{Connection, params};
    let conn = Connection::open(db_path).ok()?;
    conn.query_row(
        "SELECT o.id FROM organizations o, json_each(o.external_ids) ext \
         WHERE LOWER(json_extract(ext.value, '$.type')) = ?1 \
           AND (json_extract(ext.value, '$.preferred') = ?2 \
                OR EXISTS ( \
                    SELECT 1 FROM json_each(json_extract(ext.value, '$.all')) a \
                    WHERE a.value = ?2 \
                )) \
         LIMIT 1",
        params![ext_type, ext_value],
        |row| row.get::<_, String>(0),
    )
    .ok()
}

/// Resolve an identifier and its type from an ORCID disambiguation entry.
/// Tries to resolve to ROR via SQLite first (for GRID, ISNI, FundRef, Wikidata).
/// Falls back to the original identifier and its source type when no ROR match is found.
/// Returns `None` only when the identifier is empty.
fn resolve_identifier(d: &DisambiguatedOrgJson, db_path: Option<&Path>) -> Option<(String, String)> {
    if d.identifier.is_empty() {
        return None;
    }
    let source_upper = d.source.to_uppercase();
    match source_upper.as_str() {
        "ROR" => {
            let bare = d.identifier
                .trim_start_matches("https://ror.org/")
                .trim_start_matches("http://ror.org/");
            Some((format!("https://ror.org/{bare}"), "ROR".to_string()))
        }
        "FUNDREF" | "OFR" => {
            if let Some(db) = db_path {
                if let Some(ror_url) = ror_id_from_sqlite("fundref", &d.identifier, db) {
                    return Some((ror_url, "ROR".to_string()));
                }
            }
            Some((d.identifier.clone(), "FundRef".to_string()))
        }
        "GRID" => {
            if let Some(db) = db_path {
                if let Some(ror_url) = ror_id_from_sqlite("grid", &d.identifier, db) {
                    return Some((ror_url, "ROR".to_string()));
                }
            }
            Some((d.identifier.clone(), "GRID".to_string()))
        }
        "ISNI" => {
            if let Some(db) = db_path {
                if let Some(ror_url) = ror_id_from_sqlite("isni", &d.identifier, db) {
                    return Some((ror_url, "ROR".to_string()));
                }
            }
            Some((d.identifier.clone(), "ISNI".to_string()))
        }
        "RINGGOLD" => {
            if let Some(db) = db_path {
                if let Some(ror_url) = ror_id_from_sqlite("ringgold", &d.identifier, db) {
                    return Some((ror_url, "ROR".to_string()));
                }
            }
            Some((d.identifier.clone(), "Ringgold".to_string()))
        }
        "WIKIDATA" => {
            if let Some(db) = db_path {
                if let Some(ror_url) = ror_id_from_sqlite("wikidata", &d.identifier, db) {
                    return Some((ror_url, "ROR".to_string()));
                }
            }
            Some((d.identifier.clone(), "Wikidata".to_string()))
        }
        _ => None,
    }
}

/// Fetch employment records from the ORCID public API and return them as a list
/// of affiliations sorted by start date ascending. Accepts a bare ORCID iD or
/// full ORCID URL. When `db_path` points to the local `organizations` SQLite,
/// non-ROR identifiers (GRID, ISNI, FundRef, Wikidata) are resolved to ROR IDs.
pub fn fetch_person_employments(orcid_url: &str, db_path: Option<&Path>) -> Result<Vec<PersonAffiliation>> {
    let orcid = orcid_url.trim_start_matches("https://orcid.org/");
    let api_url = format!("https://pub.orcid.org/v3.0/{}/employments", orcid);
    let client = reqwest::blocking::Client::builder()
        .user_agent(crate::io_utils::commonmeta_user_agent())
        .timeout(std::time::Duration::from_secs(30))
        .build()
        .map_err(|e| Error::Http(e.to_string()))?;
    let resp: EmploymentsResponse = client
        .get(&api_url)
        .header("Accept", "application/vnd.orcid+json")
        .send()
        .map_err(|e| Error::Http(format!("ORCID API: {}", e)))?
        .error_for_status()
        .map_err(|e| Error::Http(format!("ORCID API: {}", e)))?
        .json()
        .map_err(|e| Error::Parse(e.to_string()))?;

    let mut affiliations: Vec<PersonAffiliation> = resp
        .affiliation_groups
        .iter()
        .flat_map(|g| g.summaries.iter())
        .filter_map(|w| w.employment_summary.as_ref())
        .filter(|emp| !emp.organization.name.is_empty())
        .map(|emp| {
            let (identifier, identifier_type) = emp
                .organization
                .disambiguated
                .as_ref()
                .and_then(|d| resolve_identifier(d, db_path))
                .map(|(id, typ)| (Some(id), Some(typ)))
                .unwrap_or((None, None));
            let name = if emp.department_name.is_empty() {
                emp.organization.name.clone()
            } else {
                format!("{}, {}", emp.organization.name, emp.department_name)
            };
            PersonAffiliation {
                identifier,
                identifier_type,
                name,
                role: if emp.role_title.is_empty() { None } else { Some(emp.role_title.clone()) },
                start_date: emp.start_date.as_ref().and_then(orcid_date_to_iso),
                end_date: emp.end_date.as_ref().and_then(orcid_date_to_iso),
            }
        })
        .collect();

    affiliations.sort_by(|a, b| match (&a.start_date, &b.start_date) {
        (Some(da), Some(db)) => da.cmp(db),
        (Some(_), None) => std::cmp::Ordering::Less,
        (None, Some(_)) => std::cmp::Ordering::Greater,
        (None, None) => std::cmp::Ordering::Equal,
    });
    Ok(affiliations)
}

// Education API uses the same affiliation-group shape but with education-summary.
#[derive(Deserialize, Default)]
#[serde(default)]
struct EducationsResponse {
    #[serde(rename = "affiliation-group")]
    affiliation_groups: Vec<EducationGroupJson>,
}

#[derive(Deserialize, Default)]
#[serde(default)]
struct EducationGroupJson {
    summaries: Vec<EducationSummaryWrapper>,
}

#[derive(Deserialize, Default)]
#[serde(default)]
struct EducationSummaryWrapper {
    #[serde(rename = "education-summary")]
    education_summary: Option<EmploymentSummaryJson>,
}

/// Read affiliations stored in the `affiliations` column of the `people` SQLite table.
///
/// Returns an empty vec when the record is absent, the column is empty, or the
/// JSON cannot be parsed — never returns an error.
pub fn fetch_person_affiliations_sqlite(
    orcid_url: &str,
    db_path: &Path,
) -> Vec<PersonAffiliation> {
    let url = crate::utils::normalize_orcid(orcid_url);
    if url.is_empty() {
        return Vec::new();
    }
    let conn = match rusqlite::Connection::open(db_path) {
        Ok(c) => c,
        Err(_) => return Vec::new(),
    };
    let json: String = match conn.query_row(
        "SELECT affiliations FROM people WHERE id = ?1 LIMIT 1",
        rusqlite::params![url],
        |row| row.get(0),
    ) {
        Ok(s) => s,
        Err(_) => return Vec::new(),
    };
    if json.is_empty() {
        return Vec::new();
    }
    let vals: Vec<serde_json::Value> = match serde_json::from_str(&json) {
        Ok(v) => v,
        Err(_) => return Vec::new(),
    };
    vals.iter()
        .filter_map(|v| {
            let name = v.get("name")?.as_str()?;
            // Support both new format (identifier + identifier_type) and legacy format (id as ROR URL).
            let (identifier, identifier_type) = if let Some(id_str) = v.get("identifier").and_then(|s| s.as_str()) {
                let id_type = v.get("identifier_type").and_then(|s| s.as_str()).unwrap_or("").to_string();
                (Some(id_str.to_string()), if id_type.is_empty() { None } else { Some(id_type) })
            } else if let Some(old_id) = v.get("id").and_then(|s| s.as_str()) {
                (Some(old_id.to_string()), Some("ROR".to_string()))
            } else {
                (None, None)
            };
            Some(PersonAffiliation {
                identifier,
                identifier_type,
                name: name.to_string(),
                role: v.get("role").and_then(|s| s.as_str()).map(|s| s.to_string()),
                start_date: v.get("start_date").and_then(|s| s.as_str()).map(|s| s.to_string()),
                end_date: v.get("end_date").and_then(|s| s.as_str()).map(|s| s.to_string()),
            })
        })
        .collect()
}

/// Fetch employment **and** education records from the ORCID public API, returning
/// them as a combined list sorted by start date ascending.
///
/// Accepts a bare ORCID iD or full ORCID URL. When `db_path` points to the local
/// `organizations` SQLite, non-ROR identifiers (GRID, ISNI, FundRef, Wikidata) are
/// resolved to ROR IDs.
pub fn fetch_person_affiliations(orcid_url: &str, db_path: Option<&Path>) -> Result<Vec<PersonAffiliation>> {
    let orcid = orcid_url.trim_start_matches("https://orcid.org/");
    let client = reqwest::blocking::Client::builder()
        .user_agent(crate::io_utils::commonmeta_user_agent())
        .timeout(std::time::Duration::from_secs(30))
        .build()
        .map_err(|e| Error::Http(e.to_string()))?;

    let make_emp_aff = |emp: &EmploymentSummaryJson| -> PersonAffiliation {
        let (identifier, identifier_type) = emp.organization.disambiguated.as_ref()
            .and_then(|d| resolve_identifier(d, db_path))
            .map(|(id, typ)| (Some(id), Some(typ)))
            .unwrap_or((None, None));
        let name = if emp.department_name.is_empty() {
            emp.organization.name.clone()
        } else {
            format!("{}, {}", emp.organization.name, emp.department_name)
        };
        PersonAffiliation {
            identifier,
            identifier_type,
            name,
            role: if emp.role_title.is_empty() { None } else { Some(emp.role_title.clone()) },
            start_date: emp.start_date.as_ref().and_then(orcid_date_to_iso),
            end_date: emp.end_date.as_ref().and_then(orcid_date_to_iso),
        }
    };
    let make_edu_aff = |edu: &EmploymentSummaryJson| -> PersonAffiliation {
        let (identifier, identifier_type) = edu.organization.disambiguated.as_ref()
            .and_then(|d| resolve_identifier(d, db_path))
            .map(|(id, typ)| (Some(id), Some(typ)))
            .unwrap_or((None, None));
        let name = if edu.department_name.is_empty() {
            edu.organization.name.clone()
        } else {
            format!("{}, {}", edu.organization.name, edu.department_name)
        };
        PersonAffiliation {
            identifier,
            identifier_type,
            name,
            role: Some(if edu.role_title.is_empty() {
                "Education".to_string()
            } else {
                format!("Education: {}", edu.role_title)
            }),
            start_date: edu.start_date.as_ref().and_then(orcid_date_to_iso),
            end_date: edu.end_date.as_ref().and_then(orcid_date_to_iso),
        }
    };

    let emp_url = format!("https://pub.orcid.org/v3.0/{orcid}/employments");
    let emp_resp: EmploymentsResponse = client
        .get(&emp_url)
        .header("Accept", "application/vnd.orcid+json")
        .send()
        .map_err(|e| Error::Http(format!("ORCID API: {}", e)))?
        .error_for_status()
        .map_err(|e| Error::Http(format!("ORCID API: {}", e)))?
        .json()
        .map_err(|e| Error::Parse(e.to_string()))?;

    let edu_url = format!("https://pub.orcid.org/v3.0/{orcid}/educations");
    let edu_resp: EducationsResponse = client
        .get(&edu_url)
        .header("Accept", "application/vnd.orcid+json")
        .send()
        .map_err(|e| Error::Http(format!("ORCID API: {}", e)))?
        .error_for_status()
        .map_err(|e| Error::Http(format!("ORCID API: {}", e)))?
        .json()
        .map_err(|e| Error::Parse(e.to_string()))?;

    let mut affiliations: Vec<PersonAffiliation> = emp_resp
        .affiliation_groups
        .iter()
        .flat_map(|g| g.summaries.iter())
        .filter_map(|w| w.employment_summary.as_ref())
        .filter(|e| !e.organization.name.is_empty())
        .map(&make_emp_aff)
        .chain(
            edu_resp
                .affiliation_groups
                .iter()
                .flat_map(|g| g.summaries.iter())
                .filter_map(|w| w.education_summary.as_ref())
                .filter(|e| !e.organization.name.is_empty())
                .map(&make_edu_aff),
        )
        .collect();

    affiliations.sort_by(|a, b| match (&a.start_date, &b.start_date) {
        (Some(da), Some(db)) => da.cmp(db),
        (Some(_), None) => std::cmp::Ordering::Less,
        (None, Some(_)) => std::cmp::Ordering::Greater,
        (None, None) => std::cmp::Ordering::Equal,
    });
    Ok(affiliations)
}

/// Serialize a `PersonAffiliation` slice to a compact JSON string for SQLite storage.
fn affiliations_to_json(affs: &[PersonAffiliation]) -> String {
    if affs.is_empty() {
        return String::new();
    }
    let vals: Vec<serde_json::Value> = affs
        .iter()
        .map(|a| {
            let mut obj = serde_json::Map::new();
            if let Some(ref id) = a.identifier {
                obj.insert("identifier".into(), serde_json::Value::String(id.clone()));
            }
            if let Some(ref id_type) = a.identifier_type {
                obj.insert("identifier_type".into(), serde_json::Value::String(id_type.clone()));
            }
            obj.insert("name".into(), serde_json::Value::String(a.name.clone()));
            if let Some(ref role) = a.role {
                obj.insert("role".into(), serde_json::Value::String(role.clone()));
            }
            if let Some(ref date) = a.start_date {
                obj.insert("start_date".into(), serde_json::Value::String(date.clone()));
            }
            if let Some(ref date) = a.end_date {
                obj.insert("end_date".into(), serde_json::Value::String(date.clone()));
            }
            serde_json::Value::Object(obj)
        })
        .collect();
    serde_json::to_string(&vals).unwrap_or_default()
}

// ── Works / publications ──────────────────────────────────────────────────────

#[derive(Deserialize, Default)]
#[serde(default)]
struct WorksResponse {
    group: Vec<WorkGroup>,
}

#[derive(Deserialize, Default)]
#[serde(default)]
struct WorkGroup {
    #[serde(rename = "work-summary")]
    work_summaries: Vec<WorkSummaryJson>,
}

#[derive(Deserialize, Default)]
#[serde(default)]
struct WorkSummaryJson {
    #[serde(rename = "external-ids")]
    external_ids: WorkExternalIds,
}

#[derive(Deserialize, Default)]
#[serde(default)]
struct WorkExternalIds {
    #[serde(rename = "external-id")]
    external_id: Vec<WorkExternalId>,
}

#[derive(Deserialize, Default)]
#[serde(default)]
struct WorkExternalId {
    #[serde(rename = "external-id-type")]
    id_type: String,
    #[serde(rename = "external-id-value")]
    id_value: String,
}

/// Fetch the DOIs of all works listed on an ORCID profile.
///
/// Calls `GET https://pub.orcid.org/v3.0/{orcid}/works`, extracts one DOI per
/// work group (preferring the first summary's identifiers), and returns
/// normalised `https://doi.org/…` URLs deduplicated and in response order.
pub fn fetch_orcid_work_dois(orcid_url: &str) -> Result<Vec<String>> {
    let orcid = orcid_url.trim_start_matches("https://orcid.org/");
    let api_url = format!("https://pub.orcid.org/v3.0/{}/works", orcid);
    let client = reqwest::blocking::Client::builder()
        .user_agent(crate::io_utils::commonmeta_user_agent())
        .timeout(std::time::Duration::from_secs(30))
        .build()
        .map_err(|e| Error::Http(e.to_string()))?;
    let resp: WorksResponse = client
        .get(&api_url)
        .header("Accept", "application/vnd.orcid+json")
        .send()
        .map_err(|e| Error::Http(format!("ORCID API: {}", e)))?
        .error_for_status()
        .map_err(|e| Error::Http(format!("ORCID API: {}", e)))?
        .json()
        .map_err(|e| Error::Parse(e.to_string()))?;

    let mut dois: Vec<String> = Vec::new();
    for group in &resp.group {
        // Each group is one logical work; take the first summary's identifiers.
        if let Some(summary) = group.work_summaries.first() {
            if let Some(doi) = summary
                .external_ids
                .external_id
                .iter()
                .find(|id| id.id_type == "doi")
                .map(|id| crate::doi_utils::normalize_doi(&id.id_value))
                .filter(|s| !s.is_empty())
            {
                if !dois.contains(&doi) {
                    dois.push(doi);
                }
            }
        }
    }
    Ok(dois)
}

// ── InvenioRDM YAML output ────────────────────────────────────────────────────

/// Double-quote a YAML scalar, escaping backslashes, double-quotes, and newlines.
fn yaml_dq(s: &str) -> String {
    format!(
        "\"{}\"",
        s.replace('\\', "\\\\")
            .replace('"', "\\\"")
            .replace('\n', "\\n")
            .replace('\r', "")
            .replace('\t', "\\t")
    )
}

/// Return `s` as a plain YAML scalar when safe; otherwise double-quote it.
fn yaml_scalar(s: &str) -> String {
    if s.is_empty() {
        return yaml_dq(s);
    }
    let first = s.chars().next().unwrap();
    let needs_q = ":-?|>!'\"#&*{}[],%@`".contains(first)
        || s.contains(": ")
        || s.contains(" #")
        || s.contains('\n')
        || matches!(s, "true" | "false" | "null" | "~");
    if needs_q { yaml_dq(s) } else { s.to_string() }
}

/// Serialize a person to InvenioRDM names YAML format (list form).
///
/// `person_json` must be the ORCID 3.0 `/person` endpoint response; `affiliations`
/// is the employment list from [`fetch_person_employments`] (may be empty).
pub fn write_inveniordm_person_yaml(
    person_json: &serde_json::Value,
    affiliations: &[PersonAffiliation],
) -> Result<Vec<u8>> {
    let name_obj = person_json.get("name");
    let given_name = name_obj
        .and_then(|n| n.get("given-names"))
        .and_then(|g| g.get("value"))
        .and_then(|v| v.as_str())
        .unwrap_or("")
        .to_string();
    let family_name = name_obj
        .and_then(|n| n.get("family-name"))
        .and_then(|f| f.get("value"))
        .and_then(|v| v.as_str())
        .unwrap_or("")
        .to_string();
    let credit_name = name_obj
        .and_then(|n| n.get("credit-name"))
        .and_then(|c| c.get("value"))
        .and_then(|v| v.as_str())
        .unwrap_or("")
        .to_string();
    let display_name = if !credit_name.is_empty() {
        credit_name
    } else {
        format!("{} {}", given_name, family_name).trim().to_string()
    };

    let path = person_json.get("path").and_then(|p| p.as_str()).unwrap_or("");
    let orcid_id = path.trim_start_matches('/').split('/').next().unwrap_or("").to_string();

    let biography = person_json
        .get("biography")
        .and_then(|b| b.get("content"))
        .and_then(|c| c.as_str())
        .unwrap_or("")
        .to_string();

    let mut out = String::new();
    out.push_str(&format!("- id: {}\n", orcid_id));
    out.push_str(&format!("  name: {}\n", yaml_scalar(&display_name)));
    out.push_str(&format!("  given_name: {}\n", yaml_scalar(&given_name)));
    out.push_str(&format!("  family_name: {}\n", yaml_scalar(&family_name)));
    out.push_str("  identifiers:\n");
    out.push_str(&format!("    - identifier: https://orcid.org/{}\n", orcid_id));

    if !affiliations.is_empty() {
        out.push_str("  affiliations:\n");
        for aff in affiliations {
            if let Some(ref id) = aff.identifier {
                out.push_str(&format!("    - identifier: {}\n", id));
                if let Some(ref id_type) = aff.identifier_type {
                    out.push_str(&format!("      identifier_type: {}\n", id_type));
                }
                out.push_str(&format!("      name: {}\n", yaml_scalar(&aff.name)));
            } else {
                out.push_str(&format!("    - name: {}\n", yaml_scalar(&aff.name)));
            }
            if let Some(date) = &aff.start_date {
                out.push_str(&format!("      start_date: '{}'\n", date));
            }
            if let Some(date) = &aff.end_date {
                out.push_str(&format!("      end_date: '{}'\n", date));
            }
        }
    }

    if !biography.is_empty() {
        out.push_str(&format!("  description: {}\n", yaml_dq(&biography)));
    }

    Ok(out.into_bytes())
}

// ── JSON output ───────────────────────────────────────────────────────────────

/// Serialize an ORCID 3.0 person JSON value to bytes.
///
/// The value should conform to `orcid_schema_v3.0.json` — i.e., the shape
/// returned by [`fetch_person_json`] or [`fetch_person_json_sqlite`].
pub fn write_orcid_json(value: &serde_json::Value) -> Result<Vec<u8>> {
    serde_json::to_vec(value).map_err(|e| Error::Serialize(e.to_string()))
}

/// Convert ORCID 3.0 person JSON + resolved affiliations to a commonmeta person entity.
pub fn orcid_person_to_commonmeta(
    person_json: &serde_json::Value,
    affiliations: &[PersonAffiliation],
) -> serde_json::Value {
    let mut obj = serde_json::Map::new();

    // id — path may be "/0000-0003-1419-2405/person" (API) or "0000-0003-1419-2405" (SQLite XML)
    if let Some(path) = person_json.get("path").and_then(|v| v.as_str()) {
        let orcid_id = path.trim_start_matches('/').split('/').next().unwrap_or("");
        if !orcid_id.is_empty() {
            obj.insert(
                "id".to_string(),
                serde_json::Value::String(format!("https://orcid.org/{orcid_id}")),
            );
        }
    }

    // given_name / family_name / name (display_name: credit-name, else given + family)
    if let Some(name_obj) = person_json.get("name") {
        let given = name_obj
            .get("given-names")
            .and_then(|n| n.get("value"))
            .and_then(|v| v.as_str())
            .unwrap_or("");
        let family = name_obj
            .get("family-name")
            .and_then(|n| n.get("value"))
            .and_then(|v| v.as_str())
            .unwrap_or("");
        let credit = name_obj
            .get("credit-name")
            .and_then(|n| n.get("value"))
            .and_then(|v| v.as_str())
            .unwrap_or("");
        if !given.is_empty() {
            obj.insert("given_name".to_string(), serde_json::Value::String(given.to_string()));
        }
        if !family.is_empty() {
            obj.insert("family_name".to_string(), serde_json::Value::String(family.to_string()));
        }
        let display_name = if !credit.is_empty() {
            credit.to_string()
        } else {
            match (given.is_empty(), family.is_empty()) {
                (false, false) => format!("{} {}", given, family),
                (true, false) => family.to_string(),
                (false, true) => given.to_string(),
                (true, true) => String::new(),
            }
        };
        if !display_name.is_empty() {
            obj.insert("name".to_string(), serde_json::Value::String(display_name));
        }
    }

    // description (biography)
    if let Some(bio) = person_json
        .get("biography")
        .and_then(|b| b.get("content"))
        .and_then(|v| v.as_str())
        .filter(|s| !s.is_empty())
    {
        obj.insert("description".to_string(), serde_json::Value::String(bio.to_string()));
    }

    // additional_names (other-names)
    if let Some(others) = person_json
        .get("other-names")
        .and_then(|o| o.get("other-name"))
        .and_then(|v| v.as_array())
    {
        let names: Vec<serde_json::Value> = others
            .iter()
            .filter_map(|n| n.get("content").and_then(|v| v.as_str()))
            .filter(|s| !s.is_empty())
            .map(|s| serde_json::Value::String(s.to_string()))
            .collect();
        if !names.is_empty() {
            obj.insert("additional_names".to_string(), serde_json::Value::Array(names));
        }
    }

    // affiliations (from resolved employments)
    if !affiliations.is_empty() {
        let affs: Vec<serde_json::Value> = affiliations
            .iter()
            .map(|a| {
                let mut aff = serde_json::Map::new();
                if let Some(ref id) = a.identifier {
                    aff.insert("identifier".to_string(), serde_json::Value::String(id.clone()));
                }
                if let Some(ref id_type) = a.identifier_type {
                    aff.insert("identifier_type".to_string(), serde_json::Value::String(id_type.clone()));
                }
                aff.insert("name".to_string(), serde_json::Value::String(a.name.clone()));
                if let Some(ref role) = a.role {
                    aff.insert("role".to_string(), serde_json::Value::String(role.clone()));
                }
                if let Some(ref date) = a.start_date {
                    aff.insert("start_date".to_string(), serde_json::Value::String(date.clone()));
                }
                if let Some(ref date) = a.end_date {
                    aff.insert("end_date".to_string(), serde_json::Value::String(date.clone()));
                }
                serde_json::Value::Object(aff)
            })
            .collect();
        obj.insert("affiliations".to_string(), serde_json::Value::Array(affs));
    }

    // identifiers (external-identifiers)
    if let Some(ext_ids) = person_json
        .get("external-identifiers")
        .and_then(|e| e.get("external-identifier"))
        .and_then(|v| v.as_array())
    {
        let ids: Vec<serde_json::Value> = ext_ids
            .iter()
            .filter_map(|e| {
                let id_type = e.get("external-id-type")?.as_str()?;
                let id_value = e.get("external-id-value")?.as_str()?;
                if id_value.is_empty() {
                    return None;
                }
                let (cm_type, scheme) = map_orcid_ext_id_type(id_type);
                let mut id_obj = serde_json::Map::new();
                id_obj.insert("identifier".to_string(), serde_json::Value::String(id_value.to_string()));
                id_obj.insert("identifier_type".to_string(), serde_json::Value::String(cm_type.to_string()));
                if let Some(s) = scheme {
                    id_obj.insert("scheme".to_string(), serde_json::Value::String(s.to_string()));
                }
                Some(serde_json::Value::Object(id_obj))
            })
            .collect();
        if !ids.is_empty() {
            obj.insert("identifiers".to_string(), serde_json::Value::Array(ids));
        }
    }

    // urls (researcher-urls)
    if let Some(ru) = person_json
        .get("researcher-urls")
        .and_then(|r| r.get("researcher-url"))
        .and_then(|v| v.as_array())
    {
        let urls: Vec<serde_json::Value> = ru
            .iter()
            .filter_map(|r| {
                let url = r.get("url")?.get("value")?.as_str()?;
                if url.is_empty() {
                    return None;
                }
                let mut url_obj = serde_json::Map::new();
                if let Some(name) = r.get("url-name").and_then(|v| v.as_str()).filter(|s| !s.is_empty()) {
                    url_obj.insert("name".to_string(), serde_json::Value::String(name.to_string()));
                }
                url_obj.insert("url".to_string(), serde_json::Value::String(url.to_string()));
                Some(serde_json::Value::Object(url_obj))
            })
            .collect();
        if !urls.is_empty() {
            obj.insert("urls".to_string(), serde_json::Value::Array(urls));
        }
    }

    // country (first address)
    if let Some(country) = person_json
        .get("addresses")
        .and_then(|a| a.get("address"))
        .and_then(|v| v.as_array())
        .and_then(|arr| arr.first())
        .and_then(|a| a.get("country"))
        .and_then(|c| c.get("value"))
        .and_then(|v| v.as_str())
        .filter(|s| !s.is_empty())
    {
        obj.insert("country".to_string(), serde_json::Value::String(country.to_string()));
    }

    obj.insert("asserted_by".to_string(), serde_json::Value::String("ORCID".to_string()));

    serde_json::Value::Object(obj)
}

fn map_orcid_ext_id_type(id_type: &str) -> (&'static str, Option<String>) {
    match id_type {
        "ResearcherID" | "rid"     => ("ResearcherID", None),
        "Scopus Author ID"         => ("ScopusID", None),
        "ISNI"                     => ("ISNI", None),
        "Wikidata"                 => ("Wikidata", None),
        _                          => ("Other", Some(id_type.to_string())),
    }
}

/// Serialize a commonmeta person entity optionally followed by works as a JSON
/// array validated against the commonmeta v1.0 schema.
pub fn write_commonmeta_person(
    person_json: &serde_json::Value,
    affiliations: &[PersonAffiliation],
    works: &[Data],
) -> Result<Vec<u8>> {
    let entity = orcid_person_to_commonmeta(person_json, affiliations);
    let mut items = vec![entity];
    for work in works {
        let prepared = crate::formats::commonmeta::prepare(work);
        let v = serde_json::to_value(&prepared).map_err(|e| Error::Serialize(e.to_string()))?;
        items.push(v);
    }
    let arr = serde_json::Value::Array(items);
    let bytes = serde_json::to_vec_pretty(&arr).map_err(|e| Error::Serialize(e.to_string()))?;
    crate::schema_utils::json_schema_errors(&bytes, Some("commonmeta"))?;
    Ok(bytes)
}

// ── Tests ─────────────────────────────────────────────────────────────────────

#[cfg(test)]
mod tests {
    use super::*;

    const MINIMAL_RECORD: &str = r#"<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
<record:record xmlns:common="http://www.orcid.org/ns/common"
               xmlns:history="http://www.orcid.org/ns/history"
               xmlns:person="http://www.orcid.org/ns/person"
               xmlns:personal-details="http://www.orcid.org/ns/personal-details"
               xmlns:address="http://www.orcid.org/ns/address"
               xmlns:keyword="http://www.orcid.org/ns/keyword"
               xmlns:external-identifier="http://www.orcid.org/ns/external-identifier"
               xmlns:record="http://www.orcid.org/ns/record"
               path="/0009-0007-0779-1000">
    <common:orcid-identifier>
        <common:uri>https://orcid.org/0009-0007-0779-1000</common:uri>
        <common:path>0009-0007-0779-1000</common:path>
        <common:host>orcid.org</common:host>
    </common:orcid-identifier>
    <history:history>
        <common:last-modified-date>2023-08-20T05:55:33.757Z</common:last-modified-date>
    </history:history>
    <person:person path="/0009-0007-0779-1000/person">
        <person:name visibility="public" path="0009-0007-0779-1000">
            <personal-details:given-names>Yumi</personal-details:given-names>
            <personal-details:family-name>Shin</personal-details:family-name>
        </person:name>
        <address:addresses path="/0009-0007-0779-1000/address"/>
        <keyword:keywords path="/0009-0007-0779-1000/keywords"/>
        <external-identifier:external-identifiers path="/0009-0007-0779-1000/external-identifiers"/>
    </person:person>
</record:record>"#;

    const RICH_RECORD: &str = r#"<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
<record:record xmlns:common="http://www.orcid.org/ns/common"
               xmlns:history="http://www.orcid.org/ns/history"
               xmlns:person="http://www.orcid.org/ns/person"
               xmlns:personal-details="http://www.orcid.org/ns/personal-details"
               xmlns:address="http://www.orcid.org/ns/address"
               xmlns:keyword="http://www.orcid.org/ns/keyword"
               xmlns:external-identifier="http://www.orcid.org/ns/external-identifier"
               xmlns:activities="http://www.orcid.org/ns/activities"
               xmlns:employment="http://www.orcid.org/ns/employment"
               xmlns:education="http://www.orcid.org/ns/education"
               xmlns:record="http://www.orcid.org/ns/record"
               path="/0000-0001-8188-0000">
    <common:orcid-identifier>
        <common:uri>https://orcid.org/0000-0001-8188-0000</common:uri>
        <common:path>0000-0001-8188-0000</common:path>
        <common:host>orcid.org</common:host>
    </common:orcid-identifier>
    <history:history>
        <common:last-modified-date>2022-05-26T07:06:31.543Z</common:last-modified-date>
    </history:history>
    <person:person path="/0000-0001-8188-0000/person">
        <person:name visibility="public" path="0000-0001-8188-0000">
            <personal-details:given-names>Ana</personal-details:given-names>
            <personal-details:family-name>Souza</personal-details:family-name>
        </person:name>
        <address:addresses path="/0000-0001-8188-0000/address">
            <address:address put-code="12345" visibility="public" path="/0000-0001-8188-0000/address/12345">
                <address:country>BR</address:country>
            </address:address>
        </address:addresses>
        <keyword:keywords path="/0000-0001-8188-0000/keywords">
            <keyword:keyword put-code="111" visibility="public">
                <keyword:content>bioinformatics</keyword:content>
            </keyword:keyword>
            <keyword:keyword put-code="222" visibility="public">
                <keyword:content>genomics</keyword:content>
            </keyword:keyword>
        </keyword:keywords>
        <external-identifier:external-identifiers path="/0000-0001-8188-0000/external-identifiers">
            <external-identifier:external-identifier put-code="848742" visibility="public">
                <common:external-id-type>ResearcherID</common:external-id-type>
                <common:external-id-value>D-1073-2012</common:external-id-value>
                <common:external-id-url>http://www.researcherid.com/rid/D-1073-2012</common:external-id-url>
            </external-identifier:external-identifier>
            <external-identifier:external-identifier put-code="848755" visibility="public">
                <common:external-id-type>Scopus Author ID</common:external-id-type>
                <common:external-id-value>36852681700</common:external-id-value>
            </external-identifier:external-identifier>
        </external-identifier:external-identifiers>
    </person:person>
    <activities:activities-summary path="/0000-0001-8188-0000/activities">
        <activities:employments path="/0000-0001-8188-0000/employments">
            <activities:affiliation-group>
                <employment:employment-summary put-code="11111" visibility="public">
                    <common:organization>
                        <common:name>University of São Paulo</common:name>
                        <common:address>
                            <common:city>São Paulo</common:city>
                            <common:country>BR</common:country>
                        </common:address>
                        <common:disambiguated-organization>
                            <common:disambiguated-organization-identifier>036rp1748</common:disambiguated-organization-identifier>
                            <common:disambiguation-source>ROR</common:disambiguation-source>
                        </common:disambiguated-organization>
                    </common:organization>
                    <common:role-title>Research Associate</common:role-title>
                    <common:start-date>
                        <common:year>2019</common:year>
                        <common:month>03</common:month>
                    </common:start-date>
                </employment:employment-summary>
            </activities:affiliation-group>
        </activities:employments>
        <activities:educations path="/0000-0001-8188-0000/educations">
            <activities:affiliation-group>
                <education:education-summary put-code="22222" visibility="public">
                    <common:organization>
                        <common:name>UNICAMP</common:name>
                        <common:address>
                            <common:city>Campinas</common:city>
                            <common:country>BR</common:country>
                        </common:address>
                    </common:organization>
                    <common:role-title>PhD</common:role-title>
                    <common:start-date>
                        <common:year>2013</common:year>
                    </common:start-date>
                    <common:end-date>
                        <common:year>2018</common:year>
                    </common:end-date>
                </education:education-summary>
            </activities:affiliation-group>
        </activities:educations>
    </activities:activities-summary>
</record:record>"#;

    const DEPT_RECORD: &str = r#"<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
<record:record xmlns:common="http://www.orcid.org/ns/common"
               xmlns:history="http://www.orcid.org/ns/history"
               xmlns:person="http://www.orcid.org/ns/person"
               xmlns:personal-details="http://www.orcid.org/ns/personal-details"
               xmlns:address="http://www.orcid.org/ns/address"
               xmlns:keyword="http://www.orcid.org/ns/keyword"
               xmlns:external-identifier="http://www.orcid.org/ns/external-identifier"
               xmlns:activities="http://www.orcid.org/ns/activities"
               xmlns:employment="http://www.orcid.org/ns/employment"
               xmlns:education="http://www.orcid.org/ns/education"
               xmlns:record="http://www.orcid.org/ns/record"
               path="/0000-0001-8177-3280">
    <common:orcid-identifier>
        <common:uri>https://orcid.org/0000-0001-8177-3280</common:uri>
        <common:path>0000-0001-8177-3280</common:path>
        <common:host>orcid.org</common:host>
    </common:orcid-identifier>
    <history:history>
        <common:last-modified-date>2024-01-01T00:00:00.000Z</common:last-modified-date>
    </history:history>
    <person:person path="/0000-0001-8177-3280/person">
        <person:name visibility="public" path="0000-0001-8177-3280">
            <personal-details:given-names>Petra</personal-details:given-names>
            <personal-details:family-name>Dersch</personal-details:family-name>
        </person:name>
        <address:addresses path="/0000-0001-8177-3280/address"/>
        <keyword:keywords path="/0000-0001-8177-3280/keywords"/>
        <external-identifier:external-identifiers path="/0000-0001-8177-3280/external-identifiers"/>
    </person:person>
    <activities:activities-summary path="/0000-0001-8177-3280/activities">
        <activities:employments path="/0000-0001-8177-3280/employments">
            <activities:affiliation-group>
                <employment:employment-summary put-code="99999" visibility="public">
                    <common:department-name>Institute of Infectiology</common:department-name>
                    <common:role-title>Head of the Institute</common:role-title>
                    <common:organization>
                        <common:name>University of Münster</common:name>
                        <common:disambiguated-organization>
                            <common:disambiguated-organization-identifier>00pd74e08</common:disambiguated-organization-identifier>
                            <common:disambiguation-source>ROR</common:disambiguation-source>
                        </common:disambiguated-organization>
                    </common:organization>
                    <common:start-date>
                        <common:year>2019</common:year>
                        <common:month>02</common:month>
                    </common:start-date>
                </employment:employment-summary>
            </activities:affiliation-group>
        </activities:employments>
        <activities:educations path="/0000-0001-8177-3280/educations"/>
    </activities:activities-summary>
</record:record>"#;

    const ERROR_RECORD: &str = r#"<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
<error:error xmlns:error="http://www.orcid.org/ns/error">
    <error:response-code>409</error:response-code>
    <error:developer-message>409 Conflict: The ORCID record is locked.</error:developer-message>
    <error:user-message>The ORCID record is locked.</error:user-message>
    <error:error-code>9018</error:error-code>
</error:error>"#;

    // ── helpers ───────────────────────────────────────────────────────────────

    /// Build an in-memory tar.gz containing a single XML entry at `path`.
    fn make_tar_gz(path: &str, xml: &str) -> Vec<u8> {
        use flate2::write::GzEncoder;
        use flate2::Compression;

        let mut gz_data: Vec<u8> = Vec::new();
        let enc = GzEncoder::new(&mut gz_data, Compression::default());
        let mut builder = tar::Builder::new(enc);
        let content = xml.as_bytes();
        let mut header = tar::Header::new_gnu();
        header.set_size(content.len() as u64);
        header.set_mode(0o644);
        header.set_mtime(0);
        header.set_cksum();
        builder.append_data(&mut header, path, content).unwrap();
        let enc = builder.into_inner().unwrap();
        enc.finish().unwrap();
        gz_data
    }

    /// Create a minimal cache SQLite (pid_records schema) with one ORCID row.
    fn make_cache_db(dir: &std::path::Path, pid: &str, json: &str) -> std::path::PathBuf {
        let path = dir.join("cache.sqlite3");
        let conn = rusqlite::Connection::open(&path).unwrap();
        conn.execute_batch(
            "CREATE TABLE pid_records (
                pid TEXT PRIMARY KEY, source_id INTEGER NOT NULL,
                resource_url TEXT NOT NULL DEFAULT '',
                raw_metadata TEXT NOT NULL, raw_metadata_type TEXT NOT NULL
            );",
        )
        .unwrap();
        conn.execute(
            "INSERT INTO pid_records (pid, source_id, raw_metadata, raw_metadata_type)
             VALUES (?1, 4, ?2, 'orcid')",
            rusqlite::params![pid, json],
        )
        .unwrap();
        path
    }

    // ── XML fixture data ──────────────────────────────────────────────────────

    #[test]
    fn test_parse_minimal_record() {
        let record = parse_orcid_xml(MINIMAL_RECORD.as_bytes()).unwrap();
        assert_eq!(record.orcid_identifier.uri, "https://orcid.org/0009-0007-0779-1000");
        assert_eq!(record.orcid_identifier.path, "0009-0007-0779-1000");
        assert_eq!(record.history.last_modified_date, "2023-08-20T05:55:33.757Z");
        assert_eq!(record.person.name.given_names, "Yumi");
        assert_eq!(record.person.name.family_name, "Shin");
        assert!(record.person.name.credit_name.is_empty());
        assert!(record.person.addresses.addresses.is_empty());
        assert!(record.person.external_identifiers.identifiers.is_empty());
    }

    #[test]
    fn test_parse_rich_record() {
        let record = parse_orcid_xml(RICH_RECORD.as_bytes()).unwrap();
        assert_eq!(record.orcid_identifier.uri, "https://orcid.org/0000-0001-8188-0000");
        assert_eq!(record.person.name.given_names, "Ana");
        assert_eq!(record.person.name.family_name, "Souza");
        assert_eq!(record.person.addresses.addresses.len(), 1);
        assert_eq!(record.person.addresses.addresses[0].country, "BR");
        assert_eq!(record.person.external_identifiers.identifiers.len(), 2);
        let ext = &record.person.external_identifiers.identifiers[0];
        assert_eq!(ext.type_, "ResearcherID");
        assert_eq!(ext.value, "D-1073-2012");
    }

    #[test]
    fn test_parse_employment_from_xml() {
        let record = parse_orcid_xml(RICH_RECORD.as_bytes()).unwrap();
        let emps = &record.activities_summary.employments.affiliation_groups;
        assert_eq!(emps.len(), 1);
        let emp = emps[0].employment_summary.as_ref().unwrap();
        assert_eq!(emp.organization.name, "University of São Paulo");
        assert_eq!(emp.role_title, "Research Associate");
        let start = emp.start_date.as_ref().unwrap();
        assert_eq!(start.year, "2019");
        assert_eq!(start.month, "03");
        let dis = emp.organization.disambiguated.as_ref().unwrap();
        assert_eq!(dis.source, "ROR");
        assert_eq!(dis.identifier, "036rp1748");
    }

    #[test]
    fn test_parse_education_from_xml() {
        let record = parse_orcid_xml(RICH_RECORD.as_bytes()).unwrap();
        let edus = &record.activities_summary.educations.affiliation_groups;
        assert_eq!(edus.len(), 1);
        let edu = edus[0].education_summary.as_ref().unwrap();
        assert_eq!(edu.organization.name, "UNICAMP");
        assert_eq!(edu.role_title, "PhD");
        assert_eq!(edu.start_date.as_ref().unwrap().year, "2013");
        assert_eq!(edu.end_date.as_ref().unwrap().year, "2018");
    }

    #[test]
    fn test_from_orcid_affiliations() {
        let record = parse_orcid_xml(RICH_RECORD.as_bytes()).unwrap();
        let data = from_orcid(record);
        assert_eq!(data.affiliations.len(), 2);
        // employment first
        let emp = &data.affiliations[0];
        assert_eq!(emp.name, "University of São Paulo");
        assert_eq!(emp.role, "Research Associate");
        assert_eq!(emp.identifier, "https://ror.org/036rp1748");
        assert_eq!(emp.identifier_type, "ROR");
        assert_eq!(emp.start_date, "2019-03");
        assert!(emp.end_date.is_empty());
        // education second — role gets "Education: " prefix
        let edu = &data.affiliations[1];
        assert_eq!(edu.name, "UNICAMP");
        assert_eq!(edu.role, "Education: PhD");
        assert!(edu.identifier.is_empty()); // no disambiguated org
        assert_eq!(edu.start_date, "2013");
        assert_eq!(edu.end_date, "2018");
    }

    #[test]
    fn test_serialize_row_affiliations() {
        let record = parse_orcid_xml(RICH_RECORD.as_bytes()).unwrap();
        let row = serialize_to_people_row(&record, RICH_RECORD.as_bytes());
        let affs: Vec<serde_json::Value> = serde_json::from_str(&row.affiliations).unwrap();
        assert_eq!(affs.len(), 2);
        assert_eq!(affs[0]["name"], "University of São Paulo");
        assert_eq!(affs[0]["identifier"], "https://ror.org/036rp1748");
        assert_eq!(affs[0]["identifier_type"], "ROR");
        assert_eq!(affs[0]["role"], "Research Associate");
        assert_eq!(affs[0]["start_date"], "2019-03");
        assert_eq!(affs[1]["name"], "UNICAMP");
        assert_eq!(affs[1]["role"], "Education: PhD");
        assert_eq!(affs[1]["start_date"], "2013");
        assert_eq!(affs[1]["end_date"], "2018");
    }

    #[test]
    fn test_department_name_included_in_affiliation_name() {
        let record = parse_orcid_xml(DEPT_RECORD.as_bytes()).unwrap();
        let data = from_orcid(record);
        assert_eq!(data.affiliations.len(), 1);
        let aff = &data.affiliations[0];
        assert_eq!(aff.name, "University of Münster, Institute of Infectiology");
        assert_eq!(aff.role, "Head of the Institute");
        assert_eq!(aff.identifier, "https://ror.org/00pd74e08");
        assert_eq!(aff.identifier_type, "ROR");
        assert_eq!(aff.start_date, "2019-02");

        // serialize_to_people_row also includes the combined name
        let record2 = parse_orcid_xml(DEPT_RECORD.as_bytes()).unwrap();
        let row = serialize_to_people_row(&record2, DEPT_RECORD.as_bytes());
        let affs: Vec<serde_json::Value> = serde_json::from_str(&row.affiliations).unwrap();
        assert_eq!(affs[0]["name"], "University of Münster, Institute of Infectiology");
        assert_eq!(affs[0]["role"], "Head of the Institute");
    }

    #[test]
    fn test_minimal_record_no_affiliations() {
        let record = parse_orcid_xml(MINIMAL_RECORD.as_bytes()).unwrap();
        let data = from_orcid(record);
        assert!(data.affiliations.is_empty());
        // Parse again to avoid use-after-move (from_orcid consumes the record).
        let record2 = parse_orcid_xml(MINIMAL_RECORD.as_bytes()).unwrap();
        let row = serialize_to_people_row(&record2, MINIMAL_RECORD.as_bytes());
        assert!(row.affiliations.is_empty());
    }

    #[test]
    fn test_error_record_returns_none() {
        assert!(parse_orcid_xml(ERROR_RECORD.as_bytes()).is_none());
    }

    #[test]
    fn test_display_name_given_family() {
        let record = parse_orcid_xml(MINIMAL_RECORD.as_bytes()).unwrap();
        assert_eq!(display_name(&record), "Yumi Shin");
    }

    #[test]
    fn test_display_name_family_only() {
        let xml = MINIMAL_RECORD.replace(
            "<personal-details:given-names>Yumi</personal-details:given-names>",
            "",
        );
        let record = parse_orcid_xml(xml.as_bytes()).unwrap();
        assert_eq!(display_name(&record), "Shin");
    }

    #[test]
    fn test_from_orcid_data() {
        let record = parse_orcid_xml(RICH_RECORD.as_bytes()).unwrap();
        let data = from_orcid(record);
        assert_eq!(data.id, "https://orcid.org/0000-0001-8188-0000");
        assert_eq!(data.type_, "Person");
        assert_eq!(data.title, "Ana Souza");
        assert_eq!(data.provider, "ORCID");
        assert_eq!(data.identifiers.len(), 2);
        assert!(data.identifiers.iter().any(|i| i.identifier_type == "ResearcherID"));
    }

    #[test]
    fn test_serialize_to_people_row() {
        let record = parse_orcid_xml(RICH_RECORD.as_bytes()).unwrap();
        let row = serialize_to_people_row(&record, RICH_RECORD.as_bytes());
        assert_eq!(row.id, "https://orcid.org/0000-0001-8188-0000");
        assert_eq!(row.orcid, "0000-0001-8188-0000");
        assert_eq!(row.given_name, "Ana");
        assert_eq!(row.family_name, "Souza");
        assert_eq!(row.name, "Ana Souza");
        assert_eq!(row.country, "BR");
        let ext: Vec<serde_json::Value> =
            serde_json::from_str(&row.external_identifiers).unwrap();
        assert_eq!(ext.len(), 2);
        assert!(!row.metadata.is_empty());
    }

    #[test]
    fn test_parse_filename_date() {
        assert_eq!(parse_filename_date("ORCID_2025_10_summaries.tar.gz"), (2025, 10));
        assert_eq!(parse_filename_date("bad_name.tar.gz"), (0, 0));
    }

    #[test]
    fn test_person_validates_against_orcid_schema() {
        let bytes = include_bytes!("../../tests/fixtures/orcid/person_0000-0003-1419-2405.json");
        let result = crate::schema_utils::json_schema_errors(bytes, Some("orcid"));
        assert!(result.is_ok(), "ORCID person fixture should validate: {result:?}");
    }

    // ── bulk import (tar.gz → SQLite) ─────────────────────────────────────────

    #[test]
    fn test_stream_summaries_to_sqlite_inserts_record() {
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("people.sqlite3");
        let tar_gz = make_tar_gz("000/0009-0007-0779-1000.xml", MINIMAL_RECORD);
        let tar_gz_path = dir.path().join("summaries.tar.gz");
        std::fs::write(&tar_gz_path, &tar_gz).unwrap();

        let count = stream_summaries_to_sqlite(&tar_gz_path, &db_path, 0).unwrap();
        assert_eq!(count, 1);

        let conn = rusqlite::Connection::open(&db_path).unwrap();
        let name: String = conn
            .query_row(
                "SELECT name FROM people WHERE id = ?1",
                rusqlite::params!["https://orcid.org/0009-0007-0779-1000"],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(name, "Yumi Shin");
    }

    #[test]
    fn test_stream_summaries_skips_error_records() {
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("people.sqlite3");
        let tar_gz = make_tar_gz("000/locked.xml", ERROR_RECORD);
        let tar_gz_path = dir.path().join("summaries.tar.gz");
        std::fs::write(&tar_gz_path, &tar_gz).unwrap();

        let count = stream_summaries_to_sqlite(&tar_gz_path, &db_path, 0).unwrap();
        assert_eq!(count, 0);
    }

    #[test]
    fn test_stream_summaries_people_table_has_no_keywords_column() {
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("people.sqlite3");
        let tar_gz = make_tar_gz("000/0009-0007-0779-1000.xml", MINIMAL_RECORD);
        std::fs::write(dir.path().join("summaries.tar.gz"), &tar_gz).unwrap();
        stream_summaries_to_sqlite(&dir.path().join("summaries.tar.gz"), &db_path, 0).unwrap();

        let conn = rusqlite::Connection::open(&db_path).unwrap();
        let cols: Vec<String> = conn
            .prepare("PRAGMA table_info(people)")
            .unwrap()
            .query_map([], |r| r.get::<_, String>(1))
            .unwrap()
            .filter_map(|r| r.ok())
            .collect();
        assert!(!cols.iter().any(|c| c == "keywords"), "keywords column must not exist");
    }

    // ── SQLite lookup ─────────────────────────────────────────────────────────

    #[test]
    fn test_fetch_sqlite_after_bulk_import() {
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("people.sqlite3");
        let tar_gz = make_tar_gz("000/0000-0001-8188-0000.xml", RICH_RECORD);
        std::fs::write(dir.path().join("summaries.tar.gz"), &tar_gz).unwrap();
        stream_summaries_to_sqlite(&dir.path().join("summaries.tar.gz"), &db_path, 0).unwrap();

        let data = fetch_sqlite("0000-0001-8188-0000", &db_path).unwrap();
        assert_eq!(data.id, "https://orcid.org/0000-0001-8188-0000");
        assert_eq!(data.type_, "Person");
        assert_eq!(data.given_name, "Ana");
        assert_eq!(data.family_name, "Souza");
        assert_eq!(data.country, "BR");
        assert_eq!(data.provider, "ORCID");
        assert_eq!(data.identifiers.len(), 2);
        assert!(data.identifiers.iter().any(|i| i.identifier_type == "ResearcherID"));
    }

    #[test]
    fn test_fetch_person_json_sqlite_reconstructs_json_from_xml_blob() {
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("people.sqlite3");
        let tar_gz = make_tar_gz("000/0000-0001-8188-0000.xml", RICH_RECORD);
        std::fs::write(dir.path().join("summaries.tar.gz"), &tar_gz).unwrap();
        stream_summaries_to_sqlite(&dir.path().join("summaries.tar.gz"), &db_path, 0).unwrap();

        let json = fetch_person_json_sqlite("0000-0001-8188-0000", &db_path).unwrap();
        assert_eq!(
            json.pointer("/name/given-names/value").and_then(|v| v.as_str()),
            Some("Ana")
        );
        assert_eq!(
            json.pointer("/name/family-name/value").and_then(|v| v.as_str()),
            Some("Souza")
        );
        // keywords not stored; reconstructed blob has empty array
        let kw = json.pointer("/keywords/keyword").and_then(|v| v.as_array());
        assert!(kw.map(|a| a.is_empty()).unwrap_or(true));
    }

    // ── cache integration ─────────────────────────────────────────────────────

    // Minimal ORCID 3.0 person JSON as stored by dragoman (fetch_orcid_with_json).
    const PERSON_JSON: &str = r#"{
        "last-modified-date": {"value": 1748000000000},
        "name": {
            "given-names": {"value": "Ana"},
            "family-name": {"value": "Souza"},
            "credit-name": null,
            "visibility": "public",
            "path": "0000-0001-8188-0000"
        },
        "other-names": {"other-name": [], "path": "/0000-0001-8188-0000/other-names"},
        "biography": null,
        "researcher-urls": {"researcher-url": [], "path": "/0000-0001-8188-0000/researcher-urls"},
        "emails": {"email": [], "path": "/0000-0001-8188-0000/email"},
        "addresses": {
            "address": [{"country": {"value": "BR"}, "visibility": "public"}],
            "path": "/0000-0001-8188-0000/address"
        },
        "external-identifiers": {
            "external-identifier": [
                {"external-id-type": "ResearcherID", "external-id-value": "D-1073-2012",
                 "external-id-relationship": "self", "visibility": "public"}
            ],
            "path": "/0000-0001-8188-0000/external-identifiers"
        },
        "path": "/0000-0001-8188-0000/person"
    }"#;

    #[test]
    fn test_stream_cache_empty_cache_returns_zero() {
        let dir = tempfile::tempdir().unwrap();
        let cache_path = dir.path().join("cache.sqlite3");
        let people_path = dir.path().join("people.sqlite3");
        let conn = rusqlite::Connection::open(&cache_path).unwrap();
        conn.execute_batch(
            "CREATE TABLE pid_records (pid TEXT PRIMARY KEY, source_id INTEGER,
             resource_url TEXT, raw_metadata TEXT, raw_metadata_type TEXT);",
        )
        .unwrap();
        drop(conn);

        let count = stream_cache_orcid_to_people_sqlite(&cache_path, &people_path).unwrap();
        assert_eq!(count, 0);
    }

    #[test]
    fn test_stream_cache_orcid_json_into_people() {
        let dir = tempfile::tempdir().unwrap();
        let cache_path = make_cache_db(
            dir.path(),
            "https://orcid.org/0000-0001-8188-0000",
            PERSON_JSON,
        );
        let people_path = dir.path().join("people.sqlite3");

        let count = stream_cache_orcid_to_people_sqlite(&cache_path, &people_path).unwrap();
        assert_eq!(count, 1);

        let data = fetch_sqlite("0000-0001-8188-0000", &people_path).unwrap();
        assert_eq!(data.id, "https://orcid.org/0000-0001-8188-0000");
        assert_eq!(data.given_name, "Ana");
        assert_eq!(data.family_name, "Souza");
        assert_eq!(data.country, "BR");
        assert_eq!(data.identifiers.len(), 1);
        assert_eq!(data.identifiers[0].identifier_type, "ResearcherID");
    }

    #[test]
    fn test_stream_cache_person_json_sqlite_roundtrip() {
        let dir = tempfile::tempdir().unwrap();
        let cache_path = make_cache_db(
            dir.path(),
            "https://orcid.org/0000-0001-8188-0000",
            PERSON_JSON,
        );
        let people_path = dir.path().join("people.sqlite3");
        stream_cache_orcid_to_people_sqlite(&cache_path, &people_path).unwrap();

        // fetch_person_json_sqlite returns the stored JSON blob without reparse
        let json = fetch_person_json_sqlite("0000-0001-8188-0000", &people_path).unwrap();
        assert_eq!(
            json.pointer("/name/given-names/value").and_then(|v| v.as_str()),
            Some("Ana")
        );
        assert_eq!(
            json.pointer("/name/family-name/value").and_then(|v| v.as_str()),
            Some("Souza")
        );
    }

    // ── schema migration ──────────────────────────────────────────────────────

    #[test]
    fn test_migration_drops_keywords_column_and_rebuilds_fts() {
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("people_old.sqlite3");

        // Simulate a database created with the old schema (keywords column + FTS with keywords).
        {
            let conn = rusqlite::Connection::open(&db_path).unwrap();
            conn.execute_batch(
                "CREATE TABLE settings (key TEXT PRIMARY KEY, value TEXT DEFAULT '');
                 CREATE TABLE people (
                     id TEXT PRIMARY KEY, orcid TEXT DEFAULT '',
                     given_name TEXT DEFAULT '', family_name TEXT DEFAULT '',
                     credit_name TEXT DEFAULT '', name TEXT DEFAULT '',
                     country TEXT DEFAULT '', keywords TEXT DEFAULT '',
                     other_names TEXT DEFAULT '', external_identifiers TEXT DEFAULT '',
                     date_updated TEXT DEFAULT '', metadata BLOB DEFAULT x''
                 );
                 CREATE VIRTUAL TABLE people_fts USING fts5(
                     name, keywords, other_names,
                     content='people', content_rowid='rowid',
                     tokenize='unicode61 remove_diacritics 1'
                 );",
            )
            .unwrap();
            conn.execute(
                "INSERT INTO people (id, orcid, name, keywords, other_names)
                 VALUES (?1, ?2, ?3, ?4, ?5)",
                rusqlite::params![
                    "https://orcid.org/0000-0001-8188-0000",
                    "0000-0001-8188-0000",
                    "Ana Souza",
                    "bioinformatics",
                    ""
                ],
            )
            .unwrap();
        }

        // Run migration.
        {
            let conn = rusqlite::Connection::open(&db_path).unwrap();
            ensure_people_schema(&conn).unwrap();
        }

        let conn = rusqlite::Connection::open(&db_path).unwrap();

        // keywords column must be gone.
        let cols: Vec<String> = conn
            .prepare("PRAGMA table_info(people)")
            .unwrap()
            .query_map([], |r| r.get::<_, String>(1))
            .unwrap()
            .filter_map(|r| r.ok())
            .collect();
        assert!(!cols.iter().any(|c| c == "keywords"), "keywords column should be dropped");

        // FTS must not reference keywords.
        let fts_sql: String = conn
            .query_row(
                "SELECT COALESCE(sql,'') FROM sqlite_master WHERE name='people_fts'",
                [],
                |r| r.get(0),
            )
            .unwrap_or_default();
        assert!(!fts_sql.contains("keywords"), "FTS should not reference keywords: {fts_sql}");
        assert!(fts_sql.contains("other_names"), "FTS should index other_names: {fts_sql}");

        // Data row is intact.
        let name: String = conn
            .query_row(
                "SELECT name FROM people WHERE id = ?1",
                rusqlite::params!["https://orcid.org/0000-0001-8188-0000"],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(name, "Ana Souza");
    }

    // ── commonmeta output ─────────────────────────────────────────────────────

    #[test]
    fn test_orcid_person_to_commonmeta_fields() {
        let json: serde_json::Value = serde_json::from_str(PERSON_JSON).unwrap();
        let out = orcid_person_to_commonmeta(&json, &[]);
        assert_eq!(out.get("id").and_then(|v| v.as_str()), Some("https://orcid.org/0000-0001-8188-0000"));
        assert_eq!(out.get("given_name").and_then(|v| v.as_str()), Some("Ana"));
        assert_eq!(out.get("family_name").and_then(|v| v.as_str()), Some("Souza"));
        assert_eq!(out.get("country").and_then(|v| v.as_str()), Some("BR"));
        assert_eq!(out.get("asserted_by").and_then(|v| v.as_str()), Some("ORCID"));
        assert!(out.get("keywords").is_none(), "keywords must not appear in commonmeta output");
    }

    #[test]
    #[ignore = "network"]
    fn test_fetch_orcid_release() {
        let release = fetch_orcid_release(1).unwrap();
        assert_eq!(release.year, 2025);
        assert!(release.batch > 0);
        assert!(release.filename.ends_with("_summaries.tar.gz"));
        assert!(release.download_url.contains("figshare.com"));
        assert!(release.size_bytes > 40_000_000_000);
    }
}
