dragoman 0.4.21

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

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

use crate::error::AppError;

#[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)
}

/// 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 as a commonmeta database.
/// Initialises an empty commonmeta schema (`works`/`works_relations`) so reads
/// before the first write resolve as misses rather than errors; the `people`
/// and `organizations` tables are created lazily by their respective upserts.
/// `commonmeta import --from cache` copies these tables into the main database.
pub fn cache_open(path: &Path) -> Result<PathBuf, AppError> {
    let path = path.to_path_buf();
    commonmeta::upsert_sqlite(&[], &path)
        .map_err(|e| AppError::Internal(format!("cache init: {e}")))?;
    Ok(path)
}

/// Look up a DOI in the cache's `works` table. A cache with no `works` table yet
/// (nothing cached) is treated as a miss, not an error.
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);
    }
    Ok(commonmeta::read_sqlite_by_id(&id, path).unwrap_or(None))
}

/// Return the cached ORCID person entity-page JSON (person + works) from the
/// cache's `people`/`works` tables, or `None` when the person isn't cached.
pub fn cache_lookup_person_page(path: &PathBuf, orcid_url: &str) -> Option<Vec<u8>> {
    commonmeta::read_person_commonmeta_sqlite(orcid_url, path).ok()
}

/// Return the cached ROR org entity-page JSON (org + works) from the cache's
/// `organizations`/`works` tables, or `None` when the org isn't cached.
pub fn cache_lookup_org_page(path: &PathBuf, ror_url: &str) -> Option<Vec<u8>> {
    commonmeta::read_org_commonmeta_sqlite(ror_url, path).ok()
}

/// Return all cached works for an ORCID (empty when none cached).
pub fn cache_lookup_orcid_works(path: &PathBuf, orcid_url: &str) -> Vec<Data> {
    commonmeta::read_sqlite_by_orcid(orcid_url, path).unwrap_or_default()
}

/// Return all cached works for a ROR org (empty when none cached).
pub fn cache_lookup_ror_works(path: &PathBuf, ror_url: &str) -> Vec<Data> {
    commonmeta::read_sqlite_by_ror(ror_url, path).unwrap_or_default()
}

/// Upsert a single work record into the cache's `works`/`works_relations` tables.
pub fn cache_upsert_work(path: &PathBuf, data: &Data) -> Result<(), AppError> {
    commonmeta::upsert_sqlite(std::slice::from_ref(data), path)
        .map_err(|e| AppError::Internal(format!("cache upsert work: {e}")))
}

/// Upsert a batch of work records into the cache.
pub fn cache_upsert_works(path: &PathBuf, works: &[Data]) -> Result<(), AppError> {
    commonmeta::upsert_sqlite(works, path)
        .map_err(|e| AppError::Internal(format!("cache upsert works: {e}")))
}

/// Upsert an ORCID person (raw ORCID 3.0 JSON) into the cache's `people` table.
pub fn cache_upsert_person(path: &PathBuf, orcid_url: &str, json: &serde_json::Value) -> Result<(), AppError> {
    commonmeta::upsert_orcid_person_sqlite(json, orcid_url, path)
        .map_err(|e| AppError::Internal(format!("cache upsert person: {e}")))
}

/// Upsert a ROR organization into the cache's `organizations` table.
pub fn cache_upsert_org(path: &PathBuf, ror: &commonmeta::Ror) -> Result<(), AppError> {
    commonmeta::upsert_ror_org_sqlite(ror, path)
        .map_err(|e| AppError::Internal(format!("cache upsert org: {e}")))
}

#[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_upsert_work(&db_path, &data).expect("cache_upsert_work");
        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_commonmeta_works_schema() {
        // The cache is a commonmeta database; `commonmeta import --from cache`
        // copies its works/works_relations tables into the main database.
        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(works)").unwrap()
            .query_map([], |row| row.get::<_, String>(1)).unwrap()
            .filter_map(|r| r.ok())
            .collect();

        for col in &["id", "metadata"] {
            assert!(cols.iter().any(|c| c == col), "missing works column: {col}");
        }
        // No legacy transport table.
        let has_pid_records: bool = conn
            .query_row(
                "SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='pid_records'",
                [],
                |r| r.get::<_, i64>(0),
            )
            .unwrap() > 0;
        assert!(!has_pid_records, "pid_records table must not be created");
    }

    #[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_on_empty_cache_is_a_miss_not_an_error() {
        // A freshly opened cache (no works yet) must return None, not error.
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("cache.sqlite3");
        cache_open(&path).unwrap();
        assert!(cache_lookup(&path, "10.9999/nope").unwrap().is_none());
    }

    #[test]
    fn cache_upsert_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_upsert_work(&path, &updated).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()));
    }
}