dragoman 0.4.20

use std::path::{Path, PathBuf};

use commonmeta::Data;
use rusqlite::{Connection, OpenFlags, params};

use crate::error::AppError;

const CACHE_DDL: &str = "
    CREATE TABLE IF NOT EXISTS pid_records (
        pid               TEXT PRIMARY KEY NOT NULL,
        source_id         INTEGER NOT NULL,
        resource_url      TEXT NOT NULL DEFAULT '',
        last_modified     TEXT,
        last_fetched      TEXT NOT NULL DEFAULT (datetime('now')),
        raw_metadata      TEXT NOT NULL,
        raw_metadata_type TEXT NOT NULL
    );
    CREATE INDEX IF NOT EXISTS idx_pid_records_fetched  ON pid_records(last_fetched);
    CREATE INDEX IF NOT EXISTS idx_pid_records_modified ON pid_records(last_modified);
";

fn source_to_id(source: &str) -> Result<i64, AppError> {
    match source {
        "crossref"  => Ok(1),
        "datacite"  => Ok(2),
        "openalex"  => Ok(3),
        "orcid"     => Ok(4),
        other => Err(AppError::Internal(format!("unknown source: {other}"))),
    }
}

#[cfg(test)]
pub(crate) const TEST_DDL: &str = r#"
CREATE TABLE works (
    "id"             TEXT PRIMARY KEY NOT NULL,
    "type"           TEXT NOT NULL DEFAULT '',
    "url"            TEXT NOT NULL DEFAULT '',
    "title"          TEXT NOT NULL DEFAULT '',
    "subjects"       TEXT NOT NULL DEFAULT '[]',
    "language"       TEXT NOT NULL DEFAULT '',
    "date_published" TEXT NOT NULL DEFAULT '',
    "date_updated"   TEXT NOT NULL DEFAULT '',
    "provider"       TEXT NOT NULL DEFAULT '',
    "pmid"           TEXT NOT NULL DEFAULT '',
    "pmcid"          TEXT NOT NULL DEFAULT '',
    "openalex"       TEXT NOT NULL DEFAULT '',
    "valid"          INTEGER NOT NULL DEFAULT 0,
    "metadata"       BLOB NOT NULL DEFAULT x''
);
CREATE INDEX works_pmid ON works("pmid") WHERE "pmid" != '';
CREATE INDEX works_pmcid ON works("pmcid") WHERE "pmcid" != '';
CREATE INDEX works_openalex ON works("openalex") WHERE "openalex" != ''
"#;

fn connect(path: &PathBuf) -> Result<Connection, AppError> {
    let conn = Connection::open_with_flags(path, OpenFlags::SQLITE_OPEN_READ_ONLY)
        .map_err(|e| AppError::Internal(format!("sqlite open '{}': {e}", path.display())))?;
    // Map the full database into virtual address space so the kernel's page
    // cache (not SQLite's tiny internal cache) serves all index lookups via
    // page-faults. On a server with 128 GB RAM the hot indexes for
    // works_references, works_orcid and works_ror will stay resident.
    // Virtual address space on 64-bit Linux is ~128 TB so a large mmap_size
    // costs nothing until pages are actually touched.
    // SQLite page cache is kept small because mmap bypasses it for mapped pages.
    conn.execute_batch(
        "PRAGMA mmap_size = 549755813888; PRAGMA cache_size = -8192; PRAGMA temp_store = MEMORY;"
    ).ok();
    Ok(conn)
}

fn connect_rw(path: &PathBuf) -> Result<Connection, AppError> {
    Connection::open(path)
        .map_err(|e| AppError::Internal(format!("sqlite open rw '{}': {e}", path.display())))
}

/// Read-write cache connection that ensures the `pid_records` table exists.
/// `commonmeta import --from cache` drops the table after a successful import,
/// so cache writes recreate it (idempotent DDL) rather than failing on a
/// long-running server.
fn connect_cache_rw(path: &PathBuf) -> Result<Connection, AppError> {
    let conn = connect_rw(path)?;
    conn.execute_batch(CACHE_DDL)
        .map_err(|e| AppError::Internal(format!("cache init: {e}")))?;
    Ok(conn)
}

/// True when a query failed because the referenced table doesn't exist — the
/// cache's `pid_records` table was dropped by an import flush. Treated as a
/// cache miss; the next cache write recreates the table via `connect_cache_rw`.
fn is_missing_table(e: &rusqlite::Error) -> bool {
    e.to_string().contains("no such table")
}

/// Validate that `path` can be opened as a SQLite database with a `works` table.
///
/// Returns `Ok(None)` when the file does not exist or contains no `works` table
/// (e.g. a commonmeta database used only for organisations or settings).
/// Returns `Err` only when the file exists but cannot be opened at all.
pub fn open(path: &std::path::Path) -> Result<Option<PathBuf>, AppError> {
    if !path.exists() {
        tracing::warn!(path = %path.display(), "sqlite file not found, running without local database");
        return Ok(None);
    }
    let path = path.to_path_buf();
    let conn = connect(&path)?;
    let has_works: bool = conn
        .query_row(
            "SELECT 1 FROM sqlite_master WHERE type='table' AND name='works'",
            [],
            |_| Ok(true),
        )
        .unwrap_or(false);
    if !has_works {
        tracing::info!(path = %path.display(), "sqlite database has no works table, running without local database");
        return Ok(None);
    }
    Ok(Some(path))
}

/// Return a random `id` chosen uniformly across the works, organizations, and people
/// tables (only tables that exist and are non-empty are included).
pub fn random_id(path: &PathBuf) -> Result<Option<String>, AppError> {
    let conn = connect(path)?;

    // Build the list of non-empty tables available.
    let mut tables: Vec<&str> = Vec::new();

    let works_ok: bool = conn
        .query_row("SELECT 1 FROM works LIMIT 1", [], |_| Ok(true))
        .unwrap_or(false);
    if works_ok {
        tables.push("works");
    }

    for tbl in &["organizations", "people"] {
        let exists: bool = conn
            .query_row(
                "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?1",
                params![tbl],
                |_| Ok(true),
            )
            .unwrap_or(false);
        if exists {
            let has_rows: bool = conn
                .query_row(&format!("SELECT 1 FROM {tbl} LIMIT 1"), [], |_| Ok(true))
                .unwrap_or(false);
            if has_rows {
                tables.push(tbl);
            }
        }
    }

    if tables.is_empty() {
        return Ok(None);
    }

    // Pick a random table, then a random row from it.
    // O(log n) per query: max(rowid) + WHERE rowid >= avoids ORDER BY RANDOM() O(n log n).
    let n = tables.len() as i64;
    let idx = conn
        .query_row("SELECT abs(random()) % ?1", params![n], |r| r.get::<_, i64>(0))
        .unwrap_or(0) as usize;
    let table = tables[idx];

    let query = format!(
        "SELECT id FROM {table} \
         WHERE rowid >= abs(random()) % (SELECT max(rowid) FROM {table}) + 1 \
         LIMIT 1"
    );
    match conn.query_row(&query, [], |row| row.get(0)) {
        Ok(id) => Ok(Some(id)),
        Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
        Err(e) => Err(AppError::Internal(format!("sqlite random: {e}"))),
    }
}

#[derive(serde::Serialize)]
pub struct SearchResult {
    pub id: String,
    pub title: String,
    pub entity_type: String,
}

/// Full-text search across works_fts, people_fts, and organizations_fts (whichever exist).
pub fn search(path: &PathBuf, q: &str) -> Result<Vec<SearchResult>, AppError> {
    let q = q.trim();
    if q.is_empty() {
        return Ok(vec![]);
    }
    let conn = connect(path)?;
    let mut results: Vec<SearchResult> = Vec::new();

    let fts_tables: &[(&str, &str, &str, &str, usize)] = &[
        ("works_fts",         "works",         "title", "Work",         10),
        ("people_fts",        "people",        "name",  "Person",       10),
        ("organizations_fts", "organizations", "name",  "Organization", 10),
    ];

    for (fts, base, col, entity_type, limit) in fts_tables {
        let exists: bool = conn
            .query_row(
                "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?1",
                params![fts],
                |_| Ok(true),
            )
            .unwrap_or(false);
        if !exists {
            continue;
        }
        let sql = format!(
            "SELECT {base}.id, {fts}.{col} FROM {fts} \
             JOIN {base} ON {base}.rowid = {fts}.rowid \
             WHERE {fts} MATCH ?1||'*' ORDER BY {fts}.rank LIMIT {limit}"
        );
        let mut stmt = match conn.prepare(&sql) {
            Ok(s) => s,
            Err(_) => continue,
        };
        let rows = match stmt.query_map(params![q], |row| {
            Ok(SearchResult {
                id:          row.get(0)?,
                title:       row.get::<_, Option<String>>(1)?.unwrap_or_default(),
                entity_type: entity_type.to_string(),
            })
        }) {
            Ok(r) => r,
            Err(_) => continue,
        };
        results.extend(rows.flatten());
    }

    results.truncate(15);
    Ok(results)
}

/// Look up a single DOI in a commonmeta SQLite database.
pub fn lookup(path: &PathBuf, doi: &str) -> Result<Option<Data>, AppError> {
    let id = commonmeta::doi_utils::normalize_doi(doi);
    if id.is_empty() {
        return Ok(None);
    }
    let conn = connect(path)?;
    fetch_metadata(&conn, "SELECT metadata FROM works WHERE id = ?1", &id)
}

/// Return a live COUNT for an entity from the appropriate junction table.
/// Returns `citations` for DOIs, `works` for ORCID/ROR URLs.
/// Tables that don't exist yet return 0 rather than an error.
#[derive(serde::Serialize)]
pub struct EntityCount {
    #[serde(skip_serializing_if = "Option::is_none")]
    pub citations: Option<i64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub works: Option<i64>,
}

pub fn count_for_id(path: &PathBuf, id: &str) -> Result<EntityCount, AppError> {
    let conn = connect(path)?;
    // unwrap_or(0) silently handles "no such table" for databases that
    // haven't been indexed yet — no need for a separate table_exists check.
    if id.starts_with("https://doi.org/") {
        let normalized = commonmeta::doi_utils::normalize_doi(id);
        let n: i64 = conn.query_row(
            r#"SELECT COUNT(*) FROM works_relations WHERE "target_id" = ?1 AND "type" = 'References'"#,
            params![normalized], |row| row.get(0),
        ).unwrap_or(0);
        Ok(EntityCount { citations: Some(n), works: None })
    } else if id.starts_with("https://orcid.org/") {
        let n: i64 = conn.query_row(
            r#"SELECT COUNT(*) FROM works_people WHERE "person_id" = ?1"#,
            params![id], |row| row.get(0),
        ).unwrap_or(0);
        Ok(EntityCount { citations: None, works: Some(n) })
    } else if id.starts_with("https://ror.org/") {
        let n: i64 = conn.query_row(
            r#"SELECT COUNT(*) FROM works_organizations WHERE "organization_id" = ?1"#,
            params![id], |row| row.get(0),
        ).unwrap_or(0);
        Ok(EntityCount { citations: None, works: Some(n) })
    } else {
        Ok(EntityCount { citations: None, works: None })
    }
}

/// Return all (target_id, type) pairs from `works_relations` where source_id matches.
/// Single query covering HasVersion, IsReviewOf, and any future source-keyed relation types.
/// Uses the PRIMARY KEY index — O(log n) seek + O(k) scan. Caller filters by type.
pub fn lookup_source_relations(path: &PathBuf, doi: &str) -> Result<Vec<(String, String)>, AppError> {
    let id = commonmeta::doi_utils::normalize_doi(doi);
    if id.is_empty() {
        return Ok(vec![]);
    }
    let conn = connect(path)?;
    let mut stmt = match conn.prepare(
        r#"SELECT "target_id", "type" FROM works_relations WHERE "source_id" = ?1 LIMIT 200"#,
    ) {
        Ok(s) => s,
        Err(_) => return Ok(vec![]),  // table not yet created on this database
    };
    let rels = stmt
        .query_map(params![id], |row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)))
        .map_err(|e| AppError::Internal(format!("query source_relations: {e}")))?
        .filter_map(|r| r.ok())
        .collect();
    Ok(rels)
}

/// Return all (source_id, type, source_year) rows from `works_relations` where target_id matches.
/// Single query; caller inverts each type to get the target work's perspective.
/// source_year is 0 when the column is absent or the value is NULL.
pub fn lookup_target_relations(path: &PathBuf, doi: &str) -> Result<Vec<(String, String, i64)>, AppError> {
    let id = commonmeta::doi_utils::normalize_doi(doi);
    if id.is_empty() {
        return Ok(vec![]);
    }
    let conn = connect(path)?;
    let mut stmt = match conn.prepare(
        r#"SELECT "source_id", "type", COALESCE("source_year", 0) FROM works_relations WHERE "target_id" = ?1 ORDER BY "source_year" DESC LIMIT 200"#,
    ) {
        Ok(s) => s,
        Err(_) => return Ok(vec![]),  // table not yet created on this database
    };
    let rels = stmt
        .query_map(params![id], |row| {
            Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?, row.get::<_, i64>(2).unwrap_or(0)))
        })
        .map_err(|e| AppError::Internal(format!("query target_relations: {e}")))?
        .filter_map(|r| r.ok())
        .collect();
    Ok(rels)
}

/// Return the subset of `ids` that appear as `source_id` in `works_relations` with type `IsVersionOf`.
/// Used to filter out non-canonical versioned works from relation lists.
/// Returns an empty set when the table doesn't exist or on any error.
pub fn filter_versioned_ids(path: &PathBuf, ids: &[String]) -> std::collections::HashSet<String> {
    if ids.is_empty() {
        return std::collections::HashSet::new();
    }
    let conn = match connect(path) {
        Ok(c) => c,
        Err(_) => return std::collections::HashSet::new(),
    };
    let placeholders = ids.iter().enumerate().map(|(i, _)| format!("?{}", i + 1)).collect::<Vec<_>>().join(", ");
    let sql = format!(
        r#"SELECT DISTINCT "source_id" FROM works_relations WHERE "type" = 'IsVersionOf' AND "source_id" IN ({placeholders})"#
    );
    let mut stmt = match conn.prepare(&sql) {
        Ok(s) => s,
        Err(_) => return std::collections::HashSet::new(),
    };
    let params: Vec<&dyn rusqlite::ToSql> = ids.iter().map(|s| s as &dyn rusqlite::ToSql).collect();
    match stmt.query_map(params.as_slice(), |row| row.get::<_, String>(0)) {
        Ok(rows) => rows.filter_map(|r| r.ok()).collect(),
        Err(_) => std::collections::HashSet::new(),
    }
}

/// Look up a single record by PMID in a commonmeta SQLite database.
pub fn lookup_by_pmid(path: &PathBuf, pmid: &str) -> Result<Option<Data>, AppError> {
    if pmid.is_empty() {
        return Ok(None);
    }
    let conn = connect(path)?;
    fetch_metadata(&conn, "SELECT metadata FROM works WHERE pmid = ?1", pmid)
}

/// Look up a single record by arXiv ID in a commonmeta SQLite database.
pub fn lookup_by_arxiv(path: &PathBuf, id: &str) -> Result<Option<Data>, AppError> {
    if id.is_empty() {
        return Ok(None);
    }
    let conn = connect(path)?;
    fetch_metadata(&conn, "SELECT metadata FROM works WHERE arxiv = ?1", id)
}

/// Look up a single record by OpenAlex ID in a commonmeta SQLite database.
/// The column always stores the URL form (`https://openalex.org/W…`);
/// normalise any input (bare W ID, or URL with optional /works/ segment) before querying.
pub fn lookup_by_openalex(path: &PathBuf, id: &str) -> Result<Option<Data>, AppError> {
    if id.is_empty() {
        return Ok(None);
    }
    // Strip any URL prefix and /works/ segment to get the bare W-prefixed ID.
    let bare = id
        .trim_start_matches("https://openalex.org/works/")
        .trim_start_matches("https://openalex.org/")
        .trim_start_matches("http://openalex.org/works/")
        .trim_start_matches("http://openalex.org/");
    let canonical = format!("https://openalex.org/{bare}");
    let conn = connect(path)?;
    fetch_metadata(&conn, "SELECT metadata FROM works WHERE openalex = ?1", &canonical)
}

/// Look up a single record by PMCID in a commonmeta SQLite database.
pub fn lookup_by_pmcid(path: &PathBuf, pmcid: &str) -> Result<Option<Data>, AppError> {
    if pmcid.is_empty() {
        return Ok(None);
    }
    let conn = connect(path)?;
    fetch_metadata(&conn, "SELECT metadata FROM works WHERE pmcid = ?1", pmcid)
}

/// Return the ROR `types` array (e.g. `["education","funder"]`) from the `organizations` table.
/// Returns an empty vec when the table doesn't exist or the record isn't found.
pub fn lookup_ror_types(path: &PathBuf, ror_url: &str) -> Result<Vec<String>, AppError> {
    if ror_url.is_empty() {
        return Ok(Vec::new());
    }
    let conn = connect(path)?;
    let has_orgs: bool = conn
        .query_row(
            "SELECT 1 FROM sqlite_master WHERE type='table' AND name='organizations'",
            [],
            |_| Ok(true),
        )
        .unwrap_or(false);
    if !has_orgs {
        return Ok(Vec::new());
    }
    match conn.query_row(
        "SELECT types FROM organizations WHERE id = ?1",
        params![ror_url],
        |row| row.get::<_, String>(0),
    ) {
        Ok(json) => Ok(serde_json::from_str::<Vec<String>>(&json).unwrap_or_default()),
        Err(rusqlite::Error::QueryReturnedNoRows) => Ok(Vec::new()),
        Err(e) => Err(AppError::Internal(format!("sqlite ror types: {e}"))),
    }
}

/// Look up a ROR organization by its full URL from the `organizations` table.
/// The blob stores zstd-compressed ROR JSON (parsed with `commonmeta::read("ror", ...)`).
/// Returns `Ok(None)` if the table doesn't exist or the ID isn't found.
pub fn lookup_by_ror(
    path: &PathBuf,
    ror_url: &str,
) -> Result<Option<Data>, AppError> {
    if ror_url.is_empty() {
        return Ok(None);
    }
    let conn = connect(path)?;
    let has_orgs: bool = conn
        .query_row(
            "SELECT 1 FROM sqlite_master WHERE type='table' AND name='organizations'",
            [],
            |_| Ok(true),
        )
        .unwrap_or(false);
    if !has_orgs {
        return Ok(None);
    }
    drop(conn);

    // Fetch the raw Ror struct, enrich location from the geonames table (same
    // DB), then convert to Data via a JSON round-trip (from_ror is private).
    let mut ror = match commonmeta::fetch_ror_raw_sqlite(ror_url, path) {
        Ok(r) => r,
        Err(_) => return Ok(None),
    };
    commonmeta::enrich_ror_locations(&mut ror, path);
    let json = commonmeta::write_ror_v2_json(&ror)
        .map_err(|e| AppError::Internal(format!("ror serialize: {e}")))?;
    let json_str = String::from_utf8(json)
        .map_err(|e| AppError::Internal(format!("ror utf8: {e}")))?;
    commonmeta::read("ror", &json_str)
        .map(|data| Some(data))
        .map_err(|e| AppError::Internal(format!("ror parse: {e}")))
}

fn fetch_metadata(conn: &Connection, sql: &str, param: &str) -> Result<Option<Data>, AppError> {
    let result = conn.query_row(sql, params![param], |row| row.get::<_, Vec<u8>>(0));
    let blob = match result {
        Ok(b) => b,
        Err(rusqlite::Error::QueryReturnedNoRows) => return Ok(None),
        Err(e) => return Err(AppError::Internal(format!("sqlite query: {e}"))),
    };
    let json = zstd::decode_all(blob.as_slice())
        .map_err(|e| AppError::Internal(format!("zstd decompress: {e}")))?;
    let json_str = String::from_utf8(json)
        .map_err(|e| AppError::Internal(format!("metadata utf8: {e}")))?;
    commonmeta::read("commonmeta", &json_str)
        .map(Some)
        .map_err(|e| AppError::Internal(format!("metadata parse: {e}")))
}

/// Open (or create) the dragoman-managed cache database with WAL mode and the
/// `pid_records` schema (same structure as vraix).
pub fn cache_open(path: &Path) -> Result<PathBuf, AppError> {
    let path = path.to_path_buf();
    let conn = connect_rw(&path)?;
    let _: String = conn
        .query_row("PRAGMA journal_mode=WAL", [], |row| row.get(0))
        .map_err(|e| AppError::Internal(format!("cache wal: {e}")))?;
    conn.execute_batch(CACHE_DDL)
        .map_err(|e| AppError::Internal(format!("cache init: {e}")))?;
    Ok(path)
}

/// Search pid_records for any cached work that has `doi` as a relation target,
/// returning `(source_pid, relation_type)` pairs. The caller inverts the type.
/// Return the subset of `ids` that have `IsVersionOf` in their own cached relations.
/// Used together with `filter_versioned_ids` to remove non-canonical versioned works.
pub fn filter_versioned_ids_from_cache(path: &PathBuf, ids: &[String]) -> std::collections::HashSet<String> {
    if ids.is_empty() {
        return std::collections::HashSet::new();
    }
    let conn = match connect(path) {
        Ok(c) => c,
        Err(_) => return std::collections::HashSet::new(),
    };
    let placeholders = ids.iter().enumerate().map(|(i, _)| format!("?{}", i + 1)).collect::<Vec<_>>().join(", ");
    let sql = format!(
        r#"SELECT DISTINCT p.pid
           FROM pid_records p, json_each(p.raw_metadata, '$.relations') r
           WHERE p.raw_metadata_type = 'commonmeta'
             AND r.value->>'$.type' = 'IsVersionOf'
             AND p.pid IN ({placeholders})"#
    );
    let mut stmt = match conn.prepare(&sql) {
        Ok(s) => s,
        Err(_) => return std::collections::HashSet::new(),
    };
    let params: Vec<&dyn rusqlite::ToSql> = ids.iter().map(|s| s as &dyn rusqlite::ToSql).collect();
    match stmt.query_map(params.as_slice(), |row| row.get::<_, String>(0)) {
        Ok(rows) => rows.filter_map(|r| r.ok()).collect(),
        Err(_) => std::collections::HashSet::new(),
    }
}

pub fn lookup_cache_target_relations(path: &PathBuf, doi: &str) -> Result<Vec<(String, String)>, AppError> {
    let id = commonmeta::doi_utils::normalize_doi(doi);
    if id.is_empty() {
        return Ok(vec![]);
    }
    let conn = connect(path)?;
    let mut stmt = match conn.prepare(
        r#"SELECT p.pid, r.value->>'$.type'
           FROM pid_records p, json_each(p.raw_metadata, '$.relations') r
           WHERE p.raw_metadata_type = 'commonmeta'
             AND r.value->>'$.id' = ?1"#,
    ) {
        Ok(s) => s,
        Err(_) => return Ok(vec![]),
    };
    let rels = stmt
        .query_map(params![id], |row| {
            Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
        })
        .map_err(|e| AppError::Internal(format!("lookup_cache_target_relations: {e}")))?
        .filter_map(|r| r.ok())
        .filter(|(_, t)| !t.is_empty())
        .collect();
    Ok(rels)
}

/// Look up a DOI in the cache. Records are stored in commonmeta format; older entries
/// stored in source format are treated as cache misses (triggering a fresh fetch).
pub fn cache_lookup(path: &PathBuf, doi: &str) -> Result<Option<Data>, AppError> {
    let id = commonmeta::doi_utils::normalize_doi(doi);
    if id.is_empty() {
        return Ok(None);
    }
    let conn = connect(path)?;
    let result = conn.query_row(
        "SELECT raw_metadata FROM pid_records WHERE pid = ?1",
        params![id],
        |row| row.get::<_, String>(0),
    );
    match result {
        Ok(raw) => match commonmeta::read("commonmeta", &raw) {
            Ok(data) => Ok(Some(data)),
            Err(_) => Ok(None), // stale entry in old source format — treat as cache miss
        },
        Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
        Err(ref e) if is_missing_table(e) => Ok(None),
        Err(e) => Err(AppError::Internal(format!("cache lookup: {e}"))),
    }
}

/// Return only the `resource_url` for a DOI from the cache (no JSON parsing).
#[allow(dead_code)]
pub fn cache_lookup_url(path: &PathBuf, doi: &str) -> Result<Option<String>, AppError> {
    let id = commonmeta::doi_utils::normalize_doi(doi);
    if id.is_empty() {
        return Ok(None);
    }
    let conn = connect(path)?;
    let result = conn.query_row(
        "SELECT resource_url FROM pid_records WHERE pid = ?1",
        params![id],
        |row| row.get::<_, String>(0),
    );
    match result {
        Ok(url) if !url.is_empty() => Ok(Some(url)),
        Ok(_) => Ok(None),
        Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
        Err(ref e) if is_missing_table(e) => Ok(None),
        Err(e) => Err(AppError::Internal(format!("cache url lookup: {e}"))),
    }
}

/// Insert or replace a person record fetched from the ORCID API into the cache.
/// The raw ORCID 3.0 person JSON is stored so the `people` table can be
/// populated with full fidelity (including keywords) via `commonmeta import --from cache`.
pub fn cache_insert_person(path: &PathBuf, orcid_url: &str, json: &serde_json::Value) -> Result<(), AppError> {
    let raw_metadata = serde_json::to_string(json)
        .map_err(|e| AppError::Internal(format!("serialize orcid json: {e}")))?;
    let conn = connect_cache_rw(path)?;
    conn.execute(
        "INSERT OR REPLACE INTO pid_records
         (pid, source_id, resource_url, raw_metadata, raw_metadata_type)
         VALUES (?1, ?2, ?3, ?4, ?5)",
        params![orcid_url, 4i64, "", raw_metadata, "orcid"],
    )
    .map_err(|e| AppError::Internal(format!("cache person insert: {e}")))?;
    Ok(())
}

/// Store the rendered entity-page JSON (person + works, commonmeta array) for
/// an ORCID URL so subsequent requests can be served from cache without API calls.
pub fn cache_insert_orcid_page(path: &PathBuf, orcid_url: &str, bytes: &[u8]) -> Result<(), AppError> {
    let json_str = std::str::from_utf8(bytes)
        .map_err(|e| AppError::Internal(format!("orcid page utf8: {e}")))?;
    let conn = connect_cache_rw(path)?;
    conn.execute(
        "INSERT OR REPLACE INTO pid_records
         (pid, source_id, resource_url, raw_metadata, raw_metadata_type)
         VALUES (?1, ?2, ?3, ?4, ?5)",
        params![orcid_url, 4i64, "", json_str, "orcid_page"],
    )
    .map_err(|e| AppError::Internal(format!("orcid page cache insert: {e}")))?;
    Ok(())
}

/// Return the cached entity-page JSON bytes for an ORCID URL, or `None` on miss.
pub fn cache_lookup_orcid_page(path: &PathBuf, orcid_url: &str) -> Result<Option<Vec<u8>>, AppError> {
    let conn = connect_rw(path)?;
    let result = conn.query_row(
        "SELECT raw_metadata FROM pid_records WHERE pid = ?1 AND raw_metadata_type = 'orcid_page'",
        params![orcid_url],
        |row| row.get::<_, String>(0),
    );
    match result {
        Ok(s) => Ok(Some(s.into_bytes())),
        Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
        Err(ref e) if is_missing_table(e) => Ok(None),
        Err(e) => Err(AppError::Internal(format!("orcid page cache lookup: {e}"))),
    }
}

/// Store the rendered entity-page JSON (org-only) for a ROR URL so subsequent
/// requests can be served from cache without an API call.
pub fn cache_insert_ror_page(path: &PathBuf, ror_url: &str, bytes: &[u8]) -> Result<(), AppError> {
    let json_str = std::str::from_utf8(bytes)
        .map_err(|e| AppError::Internal(format!("ror page utf8: {e}")))?;
    let conn = connect_cache_rw(path)?;
    conn.execute(
        "INSERT OR REPLACE INTO pid_records
         (pid, source_id, resource_url, raw_metadata, raw_metadata_type)
         VALUES (?1, ?2, ?3, ?4, ?5)",
        params![ror_url, 3i64, "", json_str, "ror_page"],
    )
    .map_err(|e| AppError::Internal(format!("ror page cache insert: {e}")))?;
    Ok(())
}

/// Return the cached entity-page JSON bytes for a ROR URL, or `None` on miss.
pub fn cache_lookup_ror_page(path: &PathBuf, ror_url: &str) -> Result<Option<Vec<u8>>, AppError> {
    let conn = connect_rw(path)?;
    let result = conn.query_row(
        "SELECT raw_metadata FROM pid_records WHERE pid = ?1 AND raw_metadata_type = 'ror_page'",
        params![ror_url],
        |row| row.get::<_, String>(0),
    );
    match result {
        Ok(s) => Ok(Some(s.into_bytes())),
        Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
        Err(ref e) if is_missing_table(e) => Ok(None),
        Err(e) => Err(AppError::Internal(format!("ror page cache lookup: {e}"))),
    }
}

/// Store the prepared works JSON array for a ROR URL.
pub fn cache_insert_ror_works(path: &PathBuf, ror_url: &str, bytes: &[u8]) -> Result<(), AppError> {
    let json_str = std::str::from_utf8(bytes)
        .map_err(|e| AppError::Internal(format!("ror works utf8: {e}")))?;
    let conn = connect_cache_rw(path)?;
    conn.execute(
        "INSERT OR REPLACE INTO pid_records
         (pid, source_id, resource_url, raw_metadata, raw_metadata_type)
         VALUES (?1, ?2, ?3, ?4, ?5)",
        params![format!("{ror_url}#works"), 3i64, "", json_str, "ror_works"],
    )
    .map_err(|e| AppError::Internal(format!("ror works cache insert: {e}")))?;
    Ok(())
}

/// Return the cached works JSON bytes for a ROR URL, or `None` on miss.
pub fn cache_lookup_ror_works(path: &PathBuf, ror_url: &str) -> Result<Option<Vec<u8>>, AppError> {
    let conn = connect_rw(path)?;
    let result = conn.query_row(
        "SELECT raw_metadata FROM pid_records WHERE pid = ?1 AND raw_metadata_type = 'ror_works'",
        params![format!("{ror_url}#works")],
        |row| row.get::<_, String>(0),
    );
    match result {
        Ok(s) => Ok(Some(s.into_bytes())),
        Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
        Err(ref e) if is_missing_table(e) => Ok(None),
        Err(e) => Err(AppError::Internal(format!("ror works cache lookup: {e}"))),
    }
}

/// Store the prepared works JSON array for an ORCID URL so subsequent
/// `/api/person/{id}/works` calls are served from cache without API calls.
pub fn cache_insert_orcid_works(path: &PathBuf, orcid_url: &str, bytes: &[u8]) -> Result<(), AppError> {
    let json_str = std::str::from_utf8(bytes)
        .map_err(|e| AppError::Internal(format!("orcid works utf8: {e}")))?;
    let conn = connect_cache_rw(path)?;
    conn.execute(
        "INSERT OR REPLACE INTO pid_records
         (pid, source_id, resource_url, raw_metadata, raw_metadata_type)
         VALUES (?1, ?2, ?3, ?4, ?5)",
        params![format!("{orcid_url}#works"), 4i64, "", json_str, "orcid_works"],
    )
    .map_err(|e| AppError::Internal(format!("orcid works cache insert: {e}")))?;
    Ok(())
}

/// Return the cached works JSON bytes for an ORCID URL, or `None` on miss.
pub fn cache_lookup_orcid_works(path: &PathBuf, orcid_url: &str) -> Result<Option<Vec<u8>>, AppError> {
    let conn = connect_rw(path)?;
    let result = conn.query_row(
        "SELECT raw_metadata FROM pid_records WHERE pid = ?1 AND raw_metadata_type = 'orcid_works'",
        params![format!("{orcid_url}#works")],
        |row| row.get::<_, String>(0),
    );
    match result {
        Ok(s) => Ok(Some(s.into_bytes())),
        Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
        Err(ref e) if is_missing_table(e) => Ok(None),
        Err(e) => Err(AppError::Internal(format!("orcid works cache lookup: {e}"))),
    }
}

/// Insert or replace a record in the cache. The metadata is stored in the
/// original source format (crossref/datacite) so it can be re-converted using
/// the latest commonmeta-rs logic on each read.
pub fn cache_insert(path: &PathBuf, data: &Data, source: &str) -> Result<(), AppError> {
    let raw_bytes = commonmeta::write("commonmeta", data)
        .map_err(|e| AppError::Internal(format!("serialize to commonmeta: {e}")))?;
    let raw_metadata = String::from_utf8(raw_bytes)
        .map_err(|e| AppError::Internal(format!("utf8: {e}")))?;

    let sid = source_to_id(source)?;
    let conn = connect_cache_rw(path)?;
    conn.execute(
        "INSERT OR REPLACE INTO pid_records
         (pid, source_id, resource_url, raw_metadata, raw_metadata_type)
         VALUES (?1, ?2, ?3, ?4, ?5)",
        params![data.id, sid, data.url, raw_metadata, "commonmeta"],
    )
    .map_err(|e| AppError::Internal(format!("cache insert: {e}")))?;
    Ok(())
}

#[cfg(test)]
pub(crate) mod tests {
    use super::*;
    use std::path::Path;

    pub(crate) fn make_test_cache_db(path: &Path) -> PathBuf {
        let db_path = cache_open(path).expect("cache_open");
        let data = commonmeta::Data {
            id: "https://doi.org/10.5678/cached".to_string(),
            type_: "JournalArticle".to_string(),
            url: "https://example.com/cached-article".to_string(),
            title: "Cached Article".to_string(),
            date_published: "2024-06-01".to_string(),
            provider: "Crossref".to_string(),
            ..commonmeta::Data::default()
        };
        cache_insert(&db_path, &data, "crossref").expect("cache_insert");
        db_path
    }

    pub(crate) fn make_test_db(path: &Path) -> PathBuf {
        let conn = Connection::open(path).expect("open test db");
        conn.execute_batch(TEST_DDL).expect("create schema");

        // Record without PMID/PMCID (DOI-only lookup)
        let data = commonmeta::Data {
            id: "https://doi.org/10.1234/test".to_string(),
            type_: "JournalArticle".to_string(),
            url: "https://example.com/test-article".to_string(),
            title: "Test Article on Content Negotiation".to_string(),
            date_published: "2024-01-15".to_string(),
            provider: "Crossref".to_string(),
            ..commonmeta::Data::default()
        };
        let compressed = compress_data(&data);
        conn.execute(
            r#"INSERT INTO works ("id","type","url","title","date_published","provider","metadata")
               VALUES (?1,?2,?3,?4,?5,?6,?7)"#,
            params![&data.id, &data.type_, &data.url, &data.title,
                    &data.date_published, &data.provider, compressed],
        ).expect("insert doi-only record");

        // Record with PMID and PMCID
        let data_pmid = commonmeta::Data {
            id: "https://doi.org/10.1234/pmid-test".to_string(),
            type_: "JournalArticle".to_string(),
            url: "https://example.com/pmid-article".to_string(),
            title: "Article with PubMed IDs".to_string(),
            date_published: "2024-02-01".to_string(),
            provider: "PubMed".to_string(),
            identifiers: vec![
                commonmeta::data::Identifier {
                    identifier: "12345678".to_string(),
                    identifier_type: "PMID".to_string(),
                    ..Default::default()
                },
                commonmeta::data::Identifier {
                    identifier: "PMC9876543".to_string(),
                    identifier_type: "PMCID".to_string(),
                    ..Default::default()
                },
            ],
            ..commonmeta::Data::default()
        };
        let compressed_pmid = compress_data(&data_pmid);
        conn.execute(
            r#"INSERT INTO works ("id","type","url","title","date_published","provider",
                                  "pmid","pmcid","metadata")
               VALUES (?1,?2,?3,?4,?5,?6,?7,?8,?9)"#,
            params![&data_pmid.id, &data_pmid.type_, &data_pmid.url, &data_pmid.title,
                    &data_pmid.date_published, &data_pmid.provider,
                    "12345678", "PMC9876543", compressed_pmid],
        ).expect("insert pmid record");

        path.to_path_buf()
    }

    fn compress_data(data: &commonmeta::Data) -> Vec<u8> {
        let json_bytes = commonmeta::write("commonmeta", data).unwrap();
        zstd::encode_all(json_bytes.as_slice(), 0).unwrap()
    }

    #[test]
    fn open_returns_none_for_missing_file() {
        let result = open(Path::new("/nonexistent/path/db.sqlite3"));
        assert!(matches!(result, Ok(None)), "expected Ok(None), got {result:?}");
    }

    #[test]
    fn open_returns_none_when_no_works_table() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("no_works.sqlite3");
        let conn = Connection::open(&path).unwrap();
        conn.execute_batch("CREATE TABLE organizations (id TEXT PRIMARY KEY, name TEXT);")
            .unwrap();
        drop(conn);
        let result = open(&path);
        assert!(matches!(result, Ok(None)), "expected Ok(None), got {result:?}");
    }

    #[test]
    fn lookup_returns_none_for_unknown_doi() {
        let dir = tempfile::tempdir().unwrap();
        let path = make_test_db(&dir.path().join("test.sqlite3"));
        let result = lookup(&path, "10.9999/does-not-exist").unwrap();
        assert!(result.is_none());
    }

    #[test]
    fn lookup_finds_existing_doi() {
        let dir = tempfile::tempdir().unwrap();
        let path = make_test_db(&dir.path().join("test.sqlite3"));
        let data = lookup(&path, "10.1234/test").unwrap().expect("should find DOI");
        assert_eq!(data.id, "https://doi.org/10.1234/test");
        assert_eq!(data.title, "Test Article on Content Negotiation");
        assert_eq!(data.url, "https://example.com/test-article");
        assert_eq!(data.type_, "JournalArticle");
    }

    #[test]
    fn lookup_normalises_doi_prefix_form() {
        let dir = tempfile::tempdir().unwrap();
        let path = make_test_db(&dir.path().join("test.sqlite3"));
        for doi in &[
            "10.1234/test",
            "https://doi.org/10.1234/test",
            "http://dx.doi.org/10.1234/test",
        ] {
            assert!(
                lookup(&path, doi).unwrap().is_some(),
                "should find DOI in form '{doi}'"
            );
        }
    }

    #[test]
    fn lookup_empty_string_returns_none() {
        let dir = tempfile::tempdir().unwrap();
        let path = make_test_db(&dir.path().join("test.sqlite3"));
        assert!(lookup(&path, "").unwrap().is_none());
    }

    #[test]
    fn lookup_by_pmid_finds_record() {
        let dir = tempfile::tempdir().unwrap();
        let path = make_test_db(&dir.path().join("test.sqlite3"));
        let data = lookup_by_pmid(&path, "12345678").unwrap().expect("should find by PMID");
        assert_eq!(data.id, "https://doi.org/10.1234/pmid-test");
    }

    #[test]
    fn lookup_by_pmid_returns_none_for_unknown() {
        let dir = tempfile::tempdir().unwrap();
        let path = make_test_db(&dir.path().join("test.sqlite3"));
        assert!(lookup_by_pmid(&path, "99999999").unwrap().is_none());
    }

    #[test]
    fn lookup_by_pmid_empty_returns_none() {
        let dir = tempfile::tempdir().unwrap();
        let path = make_test_db(&dir.path().join("test.sqlite3"));
        assert!(lookup_by_pmid(&path, "").unwrap().is_none());
    }

    #[test]
    fn lookup_by_pmcid_finds_record() {
        let dir = tempfile::tempdir().unwrap();
        let path = make_test_db(&dir.path().join("test.sqlite3"));
        let data = lookup_by_pmcid(&path, "PMC9876543").unwrap().expect("should find by PMCID");
        assert_eq!(data.id, "https://doi.org/10.1234/pmid-test");
    }

    #[test]
    fn lookup_by_pmcid_returns_none_for_unknown() {
        let dir = tempfile::tempdir().unwrap();
        let path = make_test_db(&dir.path().join("test.sqlite3"));
        assert!(lookup_by_pmcid(&path, "PMC0000000").unwrap().is_none());
    }

    // ── cache ─────────────────────────────────────────────────────────────────

    #[test]
    fn cache_open_creates_pid_records_table() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("cache.sqlite3");
        cache_open(&path).unwrap();

        let conn = Connection::open(&path).unwrap();
        let cols: Vec<String> = conn
            .prepare("PRAGMA table_info(pid_records)").unwrap()
            .query_map([], |row| row.get::<_, String>(1)).unwrap()
            .filter_map(|r| r.ok())
            .collect();

        for col in &["pid", "source_id", "resource_url", "last_fetched", "raw_metadata", "raw_metadata_type"] {
            assert!(cols.iter().any(|c| c == col), "missing column: {col}");
        }
    }

    #[test]
    fn cache_schema_compatible_with_vraix_transport_table() {
        // pid + source_id + raw_metadata are required by find_transport_table
        // in commonmeta-rs so that stream_cache_to_sqlite can consume cache.sqlite3.
        // (raw_metadata holds commonmeta JSON, which stream_cache_to_sqlite parses directly.)
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("cache.sqlite3");
        cache_open(&path).unwrap();

        let conn = Connection::open(&path).unwrap();
        let cols: Vec<String> = conn
            .prepare("PRAGMA table_info(pid_records)").unwrap()
            .query_map([], |row| row.get::<_, String>(1)).unwrap()
            .filter_map(|r| r.ok())
            .collect();

        for required in &["pid", "source_id", "raw_metadata"] {
            assert!(cols.iter().any(|c| c == required), "missing vraix-required column: {required}");
        }
    }

    #[test]
    fn cache_lookup_returns_none_for_unknown_doi() {
        let dir = tempfile::tempdir().unwrap();
        let path = make_test_cache_db(&dir.path().join("cache.sqlite3"));
        assert!(cache_lookup(&path, "10.9999/nope").unwrap().is_none());
    }

    #[test]
    fn cache_lookup_finds_inserted_record() {
        let dir = tempfile::tempdir().unwrap();
        let path = make_test_cache_db(&dir.path().join("cache.sqlite3"));
        let data = cache_lookup(&path, "10.5678/cached").unwrap().expect("should find cached DOI");
        assert_eq!(data.id, "https://doi.org/10.5678/cached");
        assert_eq!(data.title, "Cached Article");
        assert_eq!(data.url, "https://example.com/cached-article");
        assert_eq!(data.type_, "JournalArticle");
    }

    #[test]
    fn cache_lookup_empty_string_returns_none() {
        let dir = tempfile::tempdir().unwrap();
        let path = make_test_cache_db(&dir.path().join("cache.sqlite3"));
        assert!(cache_lookup(&path, "").unwrap().is_none());
    }

    #[test]
    fn cache_lookup_url_returns_url_without_json_parsing() {
        let dir = tempfile::tempdir().unwrap();
        let path = make_test_cache_db(&dir.path().join("cache.sqlite3"));
        let url = cache_lookup_url(&path, "10.5678/cached").unwrap();
        assert_eq!(url.as_deref(), Some("https://example.com/cached-article"));
    }

    #[test]
    fn cache_lookup_url_returns_none_for_unknown_doi() {
        let dir = tempfile::tempdir().unwrap();
        let path = make_test_cache_db(&dir.path().join("cache.sqlite3"));
        assert!(cache_lookup_url(&path, "10.9999/nope").unwrap().is_none());
    }

    #[test]
    fn cache_insert_unknown_source_returns_error() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("cache.sqlite3");
        cache_open(&path).unwrap();
        let data = commonmeta::Data {
            id: "https://doi.org/10.1/x".to_string(),
            ..commonmeta::Data::default()
        };
        assert!(cache_insert(&path, &data, "unknown_source").is_err());
    }

    #[test]
    fn cache_insert_replaces_existing_record() {
        let dir = tempfile::tempdir().unwrap();
        let path = make_test_cache_db(&dir.path().join("cache.sqlite3"));
        let updated = commonmeta::Data {
            id: "https://doi.org/10.5678/cached".to_string(),
            type_: "JournalArticle".to_string(),
            url: "https://example.com/cached-article".to_string(),
            title: "Updated Title".to_string(),
            ..commonmeta::Data::default()
        };
        cache_insert(&path, &updated, "crossref").unwrap();
        let data = cache_lookup(&path, "10.5678/cached").unwrap().expect("should still exist");
        assert_eq!(data.title, "Updated Title");
    }

    // ── lookup_source_relations ───────────────────────────────────────────────

    pub(crate) fn make_test_relations_db(path: &std::path::Path) -> PathBuf {
        let conn = Connection::open(path).expect("open relations db");
        conn.execute_batch(
            r#"CREATE TABLE works_relations (
                   source_id   TEXT    NOT NULL,
                   target_id   TEXT    NOT NULL,
                   "type"      TEXT    NOT NULL,
                   source_year INTEGER NOT NULL DEFAULT 0,
                   PRIMARY KEY (source_id, target_id, "type")
               )"#,
        )
        .expect("create works_relations");
        for (src, tgt, rel_type) in &[
            (
                "https://doi.org/10.5281/zenodo.1000000",
                "https://doi.org/10.5281/zenodo.1000001",
                "HasVersion",
            ),
            (
                "https://doi.org/10.5281/zenodo.1000000",
                "https://doi.org/10.5281/zenodo.1000002",
                "HasVersion",
            ),
            (
                "https://doi.org/10.5281/zenodo.9999999",
                "https://doi.org/10.5281/zenodo.1000000",
                "IsVersionOf",
            ),
        ] {
            conn.execute(
                r#"INSERT INTO works_relations (source_id, target_id, "type") VALUES (?1, ?2, ?3)"#,
                params![src, tgt, rel_type],
            )
            .expect("insert relation");
        }
        path.to_path_buf()
    }

    // ── lookup_target_relations ───────────────────────────────────────────────

    pub(crate) fn make_test_references_db(path: &std::path::Path) -> PathBuf {
        let conn = Connection::open(path).expect("open references db");
        conn.execute_batch(
            r#"CREATE TABLE works_relations (
                   source_id        TEXT NOT NULL,
                   target_id        TEXT NOT NULL,
                   type             TEXT NOT NULL DEFAULT 'References',
                   source_year      INTEGER NOT NULL DEFAULT 0,
                   PRIMARY KEY (source_id, target_id, type)
               );
               CREATE TABLE works_people (
                   work_id   TEXT NOT NULL,
                   person_id TEXT NOT NULL,
                   PRIMARY KEY (work_id, person_id)
               );
               CREATE TABLE works_organizations (
                   work_id         TEXT NOT NULL,
                   organization_id TEXT NOT NULL,
                   PRIMARY KEY (work_id, organization_id)
               )"#,
        )
        .expect("create reference tables");

        // Work A and B both cite Work Z; Work C cites a different work
        for (source_id, target_id, source_year) in &[
            ("https://doi.org/10.1234/citing-a", "https://doi.org/10.5678/cited-z", 2023i64),
            ("https://doi.org/10.1234/citing-b", "https://doi.org/10.5678/cited-z", 2024i64),
            ("https://doi.org/10.1234/citing-c", "https://doi.org/10.5678/other",   2024i64),
        ] {
            conn.execute(
                r#"INSERT INTO works_relations (source_id, target_id, type, source_year) VALUES (?1, ?2, 'References', ?3)"#,
                params![source_id, target_id, source_year],
            )
            .expect("insert reference row");
        }

        for (work_id, person_id) in &[
            ("https://doi.org/10.1234/citing-a", "https://orcid.org/0000-0002-1825-0097"),
            ("https://doi.org/10.1234/citing-b", "https://orcid.org/0000-0002-1825-0097"),
        ] {
            conn.execute(
                "INSERT INTO works_people (work_id, person_id) VALUES (?1, ?2)",
                params![work_id, person_id],
            )
            .expect("insert person row");
        }

        for (work_id, organization_id) in &[
            ("https://doi.org/10.1234/citing-a", "https://ror.org/04aj4c181"),
        ] {
            conn.execute(
                "INSERT INTO works_organizations (work_id, organization_id) VALUES (?1, ?2)",
                params![work_id, organization_id],
            )
            .expect("insert organization row");
        }

        path.to_path_buf()
    }

    #[test]
    fn lookup_target_relations_empty_doi_returns_empty() {
        let dir = tempfile::tempdir().unwrap();
        let path = make_test_references_db(&dir.path().join("refs.sqlite3"));
        let result = lookup_target_relations(&path, "").unwrap();
        assert!(result.is_empty());
    }

    #[test]
    fn lookup_target_relations_table_absent_returns_empty() {
        let dir = tempfile::tempdir().unwrap();
        let path = make_test_db(&dir.path().join("test.sqlite3"));
        let result = lookup_target_relations(&path, "https://doi.org/10.5678/cited-z").unwrap();
        assert!(result.is_empty());
    }

    #[test]
    fn lookup_target_relations_finds_citing_works() {
        let dir = tempfile::tempdir().unwrap();
        let path = make_test_references_db(&dir.path().join("refs.sqlite3"));
        let mut ids: Vec<_> = lookup_target_relations(&path, "https://doi.org/10.5678/cited-z")
            .unwrap()
            .into_iter()
            .map(|(id, _type_, _year)| id)
            .collect();
        ids.sort();
        assert_eq!(
            ids,
            vec![
                "https://doi.org/10.1234/citing-a",
                "https://doi.org/10.1234/citing-b",
            ]
        );
    }

    #[test]
    fn lookup_target_relations_returns_empty_for_uncited_doi() {
        let dir = tempfile::tempdir().unwrap();
        let path = make_test_references_db(&dir.path().join("refs.sqlite3"));
        let result = lookup_target_relations(&path, "https://doi.org/10.9999/no-such-doi").unwrap();
        assert!(result.is_empty());
    }

    // ── count_for_id ──────────────────────────────────────────────────────────

    #[test]
    fn count_for_id_returns_citation_count_for_doi() {
        let dir = tempfile::tempdir().unwrap();
        let path = make_test_references_db(&dir.path().join("refs.sqlite3"));
        let count = count_for_id(&path, "https://doi.org/10.5678/cited-z").unwrap();
        assert_eq!(count.citations, Some(2));
        assert_eq!(count.works, None);
    }

    #[test]
    fn count_for_id_returns_zero_for_uncited_doi() {
        let dir = tempfile::tempdir().unwrap();
        let path = make_test_references_db(&dir.path().join("refs.sqlite3"));
        let count = count_for_id(&path, "https://doi.org/10.9999/uncited").unwrap();
        assert_eq!(count.citations, Some(0));
        assert_eq!(count.works, None);
    }

    #[test]
    fn count_for_id_returns_works_count_for_orcid() {
        let dir = tempfile::tempdir().unwrap();
        let path = make_test_references_db(&dir.path().join("refs.sqlite3"));
        let count = count_for_id(&path, "https://orcid.org/0000-0002-1825-0097").unwrap();
        assert_eq!(count.citations, None);
        assert_eq!(count.works, Some(2));
    }

    #[test]
    fn count_for_id_returns_works_count_for_ror() {
        let dir = tempfile::tempdir().unwrap();
        let path = make_test_references_db(&dir.path().join("refs.sqlite3"));
        let count = count_for_id(&path, "https://ror.org/04aj4c181").unwrap();
        assert_eq!(count.citations, None);
        assert_eq!(count.works, Some(1));
    }

    #[test]
    fn count_for_id_returns_none_for_unrecognised_id() {
        let dir = tempfile::tempdir().unwrap();
        let path = make_test_references_db(&dir.path().join("refs.sqlite3"));
        let count = count_for_id(&path, "https://example.com/not-a-pid").unwrap();
        assert_eq!(count.citations, None);
        assert_eq!(count.works, None);
    }

    #[test]
    fn lookup_source_relations_empty_doi_returns_empty() {
        let dir = tempfile::tempdir().unwrap();
        let path = make_test_relations_db(&dir.path().join("r.sqlite3"));
        let result = lookup_source_relations(&path, "").unwrap();
        assert!(result.is_empty());
    }

    #[test]
    fn lookup_source_relations_table_absent_returns_empty() {
        // make_test_db creates a works table but not works_relations
        let dir = tempfile::tempdir().unwrap();
        let path = make_test_db(&dir.path().join("test.sqlite3"));
        let result = lookup_source_relations(&path, "https://doi.org/10.5281/zenodo.1000000").unwrap();
        assert!(result.is_empty());
    }

    #[test]
    fn lookup_source_relations_finds_has_version_rows() {
        let dir = tempfile::tempdir().unwrap();
        let path = make_test_relations_db(&dir.path().join("r.sqlite3"));
        let mut rels = lookup_source_relations(&path, "https://doi.org/10.5281/zenodo.1000000").unwrap();
        rels.sort();
        assert_eq!(
            rels,
            vec![
                ("https://doi.org/10.5281/zenodo.1000001".to_string(), "HasVersion".to_string()),
                ("https://doi.org/10.5281/zenodo.1000002".to_string(), "HasVersion".to_string()),
            ]
        );
    }

    #[test]
    fn lookup_source_relations_accepts_doi_suffix_form() {
        // validate_doi returns the suffix (e.g. "10.5281/zenodo.1000000"), not the full URL.
        // lookup_source_relations must normalize internally so the query matches DB rows.
        let dir = tempfile::tempdir().unwrap();
        let path = make_test_relations_db(&dir.path().join("r.sqlite3"));
        let mut rels = lookup_source_relations(&path, "10.5281/zenodo.1000000").unwrap();
        rels.sort();
        assert_eq!(rels.len(), 2);
        assert!(rels.iter().any(|(id, t)| id == "https://doi.org/10.5281/zenodo.1000001" && t == "HasVersion"));
    }

    #[test]
    fn lookup_source_relations_returns_empty_for_unknown_doi() {
        let dir = tempfile::tempdir().unwrap();
        let path = make_test_relations_db(&dir.path().join("r.sqlite3"));
        let result = lookup_source_relations(&path, "https://doi.org/10.9999/no-such-doi").unwrap();
        assert!(result.is_empty());
    }

    #[test]
    fn lookup_source_relations_returns_all_types() {
        // zenodo.9999999 has an IsVersionOf row — returned as-is, caller filters
        let dir = tempfile::tempdir().unwrap();
        let path = make_test_relations_db(&dir.path().join("r.sqlite3"));
        let result = lookup_source_relations(&path, "https://doi.org/10.5281/zenodo.9999999").unwrap();
        assert_eq!(result.len(), 1);
        assert_eq!(result[0], ("https://doi.org/10.5281/zenodo.1000000".to_string(), "IsVersionOf".to_string()));
    }
}