dragoman 0.4.43

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

use commonmeta::{Organization, Work};
use rusqlite::{Connection, OpenFlags, params};

use crate::error::AppError;

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 `pid_records`
/// 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)?;
    // The 0.9.53+ store keeps works in the `pid_records` transport table (the
    // legacy `works` table was dropped); require it to serve from the DB.
    let has_pid_records: bool = conn
        .query_row(
            "SELECT 1 FROM sqlite_master WHERE type='table' AND name='pid_records'",
            [],
            |_| Ok(true),
        )
        .unwrap_or(false);
    if !has_pid_records {
        tracing::info!(path = %path.display(), "sqlite database has no pid_records table, running without local database");
        return Ok(None);
    }
    Ok(Some(path))
}

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

    // (table, id column) pairs that exist and are non-empty.
    let mut tables: Vec<(&str, &str)> = Vec::new();
    for (tbl, col) in &[("pid_records", "pid"), ("organizations", "id")] {
        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, col));
        }
    }

    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, col) = tables[idx];

    let query = format!(
        "SELECT {col} 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}"))),
    }
}

// Full-text search was removed in commonmeta 0.9.53 (FTS5 dropped); the search
// endpoint and its `works_fts`/`people_fts`/`organizations_fts`/`containers_fts`
// queries were removed with it.

/// Look up a container by identifier (ISSN, OpenAlex source id, or DOI) and return
/// `[container, …works]` as a commonmeta JSON array — the container entity followed
/// by its works resolved from the `pid_records` transport store via `hasContainer`
/// links. The container is read via [`commonmeta::read_container_sqlite`] (from
/// `pid_records` source id 5 + the annotation overlay, with the Front Matter → Blog
/// override applied). Returns `Ok(None)` when no container matches.
pub fn lookup_container(path: &PathBuf, identifier: &str) -> Result<Option<Vec<u8>>, AppError> {
    let Some(container) = commonmeta::read_container_sqlite(identifier, path)
        .map_err(|e| AppError::Internal(format!("read container: {e}")))?
    else {
        return Ok(None);
    };

    // The container being served — used to fill a title-less work container so the
    // works card can show and link the container name (repository works from the
    // Crossref/DataCite readers often arrive with only an identifier).
    let container_title = container.title.clone();
    let container_type = container.type_.clone();
    let container_value = serde_json::to_value(&container)
        .map_err(|e| AppError::Internal(format!("container serialize: {e}")))?;

    // Append the container's cached works (pid_records `hasContainer` links). The
    // main store carries none (they are cached per-visit); a repeat visit to a
    // resolved container serves them from the cache. Empty is fine — the UI also
    // loads works lazily via /api/container/works.
    let mut items = vec![container_value];
    for w in commonmeta::read_sqlite_by_container(identifier, path).unwrap_or_default() {
        if let Ok(mut v) = serde_json::to_value(commonmeta::prepare_commonmeta(&w)) {
            fill_work_container_title(&mut v, &container_title, &container_type);
            items.push(v);
        }
    }

    let bytes = serde_json::to_vec_pretty(&serde_json::Value::Array(items))
        .map_err(|e| AppError::Internal(format!("container serialize: {e}")))?;
    Ok(Some(bytes))
}

/// Fill a work's title-less container (that has an identifier) from the container
/// being served — its `title` and, if absent, `type`. No-op when the container
/// already has a title, has no identifiers, or the served container has no title.
fn fill_work_container_title(work: &mut serde_json::Value, title: &str, type_: &str) {
    if title.is_empty() {
        return;
    }
    let Some(c) = work.get_mut("container").and_then(|c| c.as_object_mut()) else {
        return;
    };
    let has_title = c
        .get("title")
        .and_then(|t| t.as_str())
        .is_some_and(|s| !s.is_empty());
    let has_ident = c
        .get("identifiers")
        .and_then(|i| i.as_array())
        .is_some_and(|a| !a.is_empty());
    if has_title || !has_ident {
        return;
    }
    c.insert(
        "title".to_string(),
        serde_json::Value::String(title.to_string()),
    );
    let has_type = c
        .get("type")
        .and_then(|t| t.as_str())
        .is_some_and(|s| !s.is_empty());
    if !has_type && !type_.is_empty() {
        c.insert(
            "type".to_string(),
            serde_json::Value::String(type_.to_string()),
        );
    }
}

/// The container's stored commonmeta `ContainerRecord` for `identifier`, read from
/// the `pid_records` transport store (source id 5 + annotation overlay, Front
/// Matter → Blog override applied). `None` when not present. Used to drive the
/// Crossref/DataCite works fetch.
pub fn container_record(
    path: &PathBuf,
    identifier: &str,
) -> Option<commonmeta::metadata::ContainerRecord> {
    commonmeta::read_container_sqlite(identifier, path)
        .ok()
        .flatten()
}

/// A container's works already cached locally, resolved from the `pid_records`
/// transport store via its `hasContainer` links (any container identifier).
/// Empty when nothing is linked yet.
pub fn container_works_local(path: &PathBuf, identifier: &str) -> Vec<Work> {
    commonmeta::read_sqlite_by_container(identifier, path).unwrap_or_default()
}

/// Resolve a container from the OpenAlex Sources API by identifier (ISSN, OpenAlex
/// source id, or `openalex.org` source URL). Network call; `None` when it does not
/// resolve to a container. Used as the fallback when a container isn't stored locally.
pub fn resolve_container_remote(identifier: &str) -> Option<commonmeta::metadata::ContainerRecord> {
    commonmeta::read("openalex_source", identifier)
        .ok()?
        .into_container()
}

/// Cache a container resolved from a remote source into the cache's `pid_records`
/// transport store (source id 5), so a repeat visit serves it via
/// [`lookup_container`] / [`container_record`] without re-resolving it remotely.
pub fn cache_upsert_container(
    cache: &PathBuf,
    container: &commonmeta::metadata::ContainerRecord,
) -> Result<(), AppError> {
    commonmeta::write_container_sqlite(std::slice::from_ref(container), cache)
        .map(|_| ())
        .map_err(|e| AppError::Internal(format!("cache container: {e}")))
}

/// Look up a single DOI in a commonmeta SQLite database. Resolves the DOI from
/// the `pid_records` transport store (raw provider record converted on demand).
pub fn lookup(path: &PathBuf, doi: &str) -> Result<Option<Work>, AppError> {
    let id = commonmeta::doi_utils::normalize_doi(doi);
    if id.is_empty() {
        return Ok(None);
    }
    commonmeta::read_sqlite_by_id(&id, path)
        .map_err(|e| AppError::Internal(format!("sqlite lookup: {e}")))
}

/// 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);
        // Per DataCite, a work is cited by any work that points to it with a reference-type
        // relation (References / Cites / IsSupplementedBy). COUNT(DISTINCT source) so a work
        // that points with more than one such type is counted once.
        // https://support.datacite.org/docs/contributing-citations-and-references
        let n: i64 = conn.query_row(
            r#"SELECT COUNT(DISTINCT "source_id") FROM pid_relations WHERE "target_id" = ?1 AND "type" IN ('References', 'Cites', 'IsSupplementedBy')"#,
            params![normalized], |row| row.get(0),
        ).unwrap_or(0);
        Ok(EntityCount {
            citations: Some(n),
            works: None,
        })
    } else if id.starts_with("https://orcid.org/") {
        // Work→person links are `hasPerson` rows in pid_relations (target = ORCID).
        let n: i64 = conn.query_row(
            r#"SELECT COUNT(*) FROM pid_relations WHERE "target_id" = ?1 AND "type" = 'hasPerson'"#,
            params![id], |row| row.get(0),
        ).unwrap_or(0);
        Ok(EntityCount {
            citations: None,
            works: Some(n),
        })
    } else if id.starts_with("https://ror.org/") {
        // Work→organization links are `hasOrganization` rows (target = ROR).
        let n: i64 = conn.query_row(
            r#"SELECT COUNT(*) FROM pid_relations WHERE "target_id" = ?1 AND "type" = 'hasOrganization'"#,
            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 pid_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 pid_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 pid_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(),
    }
}

/// 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<Organization>, 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 an Organization 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}")))?;
    let entity = commonmeta::read("ror", &json_str)
        .map_err(|e| AppError::Internal(format!("ror parse: {e}")))?;
    entity
        .into_organization()
        .map(Some)
        .ok_or_else(|| AppError::Internal(format!("'{ror_url}' did not parse as an organization")))
}

/// Open (or create) the dragoman-managed cache as a commonmeta `pid_records`
/// transport store. Initialises the transport schema so reads before the first
/// write resolve as misses rather than errors. `commonmeta import --from cache`
/// copies the transport rows into the main database.
pub fn cache_open(path: &Path) -> Result<PathBuf, AppError> {
    // The cache is a `pid_records` transport store. Create it with the transport
    // schema up front so reads before the first write resolve as misses (not
    // errors) and `commonmeta import --from cache` finds the expected layout.
    let path = path.to_path_buf();
    commonmeta::open_pid_records_writer(&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<Work>, 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<Work> {
    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<Work> {
    commonmeta::read_sqlite_by_ror(ror_url, path).unwrap_or_default()
}

/// Cache raw provider records (works) into cache.sqlite3's `pid_records`. Each
/// entry is `(bare_pid, source_id, raw)` — source id 1 = Crossref, 2 = DataCite —
/// and the raw payload is stored verbatim, converted to commonmeta lazily on read.
pub fn cache_upsert_raw(path: &PathBuf, entries: &[(String, i64, String)]) -> Result<(), AppError> {
    commonmeta::upsert_pid_records(entries, path)
        .map_err(|e| AppError::Internal(format!("cache upsert raw: {e}")))
}

/// Upsert an ORCID person (raw ORCID `/record` XML) into the cache as a
/// `pid_records` transport row (source id 4).
pub fn cache_upsert_person(path: &PathBuf, orcid_url: &str, xml: &str) -> Result<(), AppError> {
    commonmeta::upsert_orcid_record(orcid_url, xml, path)
        .map_err(|e| AppError::Internal(format!("cache upsert person: {e}")))
}

/// Upsert a ROR organization into the cache's `pid_records` transport table as a
/// raw ROR row (source id 3). `import --from cache` materializes source-id-3 rows
/// into the main store's `organizations` table, so the org must live in the
/// transport table to transport at all (a separate `organizations` table in the
/// cache is not read by import).
pub fn cache_upsert_org(path: &PathBuf, ror: &commonmeta::Ror) -> Result<(), AppError> {
    let value = serde_json::to_value(ror)
        .map_err(|e| AppError::Internal(format!("cache upsert org: serialize: {e}")))?;
    commonmeta::upsert_ror_record(&ror.id, &value, 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");
        // The cache stores RAW provider payloads; a Crossref message (source id 1)
        // converts to a Work on read.
        let raw = serde_json::json!({
            "DOI": "10.5678/cached",
            "type": "journal-article",
            "title": ["Cached Article"],
            "resource": {"primary": {"URL": "https://example.com/cached-article"}},
        })
        .to_string();
        cache_upsert_raw(&db_path, &[("10.5678/cached".to_string(), 1, raw)])
            .expect("cache_upsert_raw");
        db_path
    }

    pub(crate) fn make_test_db(path: &Path) -> PathBuf {
        // Works live in the commonmeta `pid_records` transport store as RAW provider
        // payloads keyed by bare DOI. A Crossref message (source id 1) converts to a
        // Work on read.
        let article = serde_json::json!({
            "DOI": "10.1234/test",
            "type": "journal-article",
            "title": ["Test Article on Content Negotiation"],
            "resource": {"primary": {"URL": "https://example.com/test-article"}},
        })
        .to_string();
        let article2 = serde_json::json!({
            "DOI": "10.1234/second",
            "type": "journal-article",
            "title": ["Second Test Article"],
            "resource": {"primary": {"URL": "https://example.com/second-article"}},
        })
        .to_string();

        commonmeta::upsert_pid_records(
            &[
                ("10.1234/test".to_string(), 1, article),
                ("10.1234/second".to_string(), 1, article2),
            ],
            path,
        )
        .expect("upsert works into pid_records");

        path.to_path_buf()
    }

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

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

    #[test]
    fn cache_open_creates_pid_records_schema() {
        // The cache is a commonmeta `pid_records` transport store; `commonmeta
        // import --from cache` copies its transport rows 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(pid_records)")
            .unwrap()
            .query_map([], |row| row.get::<_, String>(1))
            .unwrap()
            .filter_map(|r| r.ok())
            .collect();

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

    #[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 raw = serde_json::json!({
            "DOI": "10.5678/cached",
            "type": "journal-article",
            "title": ["Updated Title"],
        })
        .to_string();
        cache_upsert_raw(&path, &[("10.5678/cached".to_string(), 1, raw)]).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 pid_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 pid_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 pid_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 pid_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)
               )"#,
        )
        .expect("create pid_relations");

        // 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 pid_relations (source_id, target_id, type, source_year) VALUES (?1, ?2, 'References', ?3)"#,
                params![source_id, target_id, source_year],
            )
            .expect("insert reference row");
        }

        // Work S points to Work Z with IsSupplementedBy (a DataCite reference-type), so
        // Z is cited by S — it counts as a citation of Z.
        conn.execute(
            r#"INSERT INTO pid_relations (source_id, target_id, type, source_year) VALUES (?1, ?2, 'IsSupplementedBy', ?3)"#,
            params!["https://doi.org/10.1234/supplement-s", "https://doi.org/10.5678/cited-z", 2025i64],
        )
        .expect("insert supplement row");

        // Work→person links are `hasPerson` rows (target = ORCID url), replacing
        // the legacy `works_people` junction.
        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(
                r#"INSERT INTO pid_relations (source_id, target_id, type, source_year) VALUES (?1, ?2, 'hasPerson', 0)"#,
                params![work_id, person_id],
            )
            .expect("insert person row");
        }

        // Work→organization links are `hasOrganization` rows (target = ROR url),
        // replacing the legacy `works_organizations` junction.
        for (work_id, organization_id) in &[(
            "https://doi.org/10.1234/citing-a",
            "https://ror.org/04aj4c181",
        )] {
            conn.execute(
                r#"INSERT INTO pid_relations (source_id, target_id, type, source_year) VALUES (?1, ?2, 'hasOrganization', 0)"#,
                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",
                "https://doi.org/10.1234/supplement-s",
            ]
        );
    }

    #[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"));
        // Two inbound References + one inbound IsSupplementedBy (both DataCite
        // reference-types) → three citations of Z.
        let count = count_for_id(&path, "https://doi.org/10.5678/cited-z").unwrap();
        assert_eq!(count.citations, Some(3));
        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());
    }

    // ── lookup_container ──────────────────────────────────────────────────────

    pub(crate) fn make_test_containers_db(path: &std::path::Path) -> PathBuf {
        use commonmeta::metadata::{ContainerRecord, Identifier, Publisher};

        // Register the container in the pid_records transport store (source id 5) +
        // the annotation overlay. Stored type "Journal" with publisher "Front
        // Matter" is written back as "Blog" by the publisher-type override.
        let container = ContainerRecord {
            identifiers: vec![
                Identifier {
                    identifier: "https://openalex.org/S7407056926".to_string(),
                    identifier_type: "OpenAlex".to_string(),
                    ..Default::default()
                },
                Identifier {
                    identifier: "2749-9952".to_string(),
                    identifier_type: "ISSN".to_string(),
                    ..Default::default()
                },
                Identifier {
                    // Stored verbatim in the `doi` annotation column; the read side
                    // normalizes lookups to this full-URL form, so it must match.
                    identifier: "https://doi.org/10.53731/front_matter".to_string(),
                    identifier_type: "DOI".to_string(),
                    ..Default::default()
                },
            ],
            type_: "Journal".to_string(),
            title: "Front Matter".to_string(),
            publisher: Some(Publisher {
                name: "Front Matter".to_string(),
                ..Default::default()
            }),
            country: "DE".to_string(),
            url: "https://blog.front-matter.io".to_string(),
            ..Default::default()
        };
        commonmeta::write_container_sqlite(std::slice::from_ref(&container), path)
            .expect("register container");

        // One work in the container: a raw Crossref record (source id 1) whose ISSN
        // matches the container. `backfill_pid_relations` resolves the work's embedded
        // container identifier to the registered container's surrogate pid, writing a
        // `hasContainer` edge so a lookup by ISSN or DOI resolves it.
        let raw = serde_json::json!({
            "DOI": "10.53731/article-1",
            "type": "journal-article",
            "title": ["A blog post"],
            "container-title": ["Front Matter"],
            "ISSN": ["2749-9952"],
        })
        .to_string();
        commonmeta::upsert_pid_records(&[("10.53731/article-1".to_string(), 1, raw)], path)
            .expect("upsert container work");
        commonmeta::backfill_pid_relations(path, &[1]).expect("backfill hasContainer");

        path.to_path_buf()
    }

    fn container_from_lookup(bytes: &[u8]) -> serde_json::Value {
        let arr: serde_json::Value = serde_json::from_slice(bytes).unwrap();
        arr.as_array().unwrap()[0].clone()
    }

    #[test]
    fn lookup_container_by_issn_openalex_and_doi() {
        let dir = tempfile::tempdir().unwrap();
        let path = make_test_containers_db(&dir.path().join("c.sqlite3"));
        for id in &[
            "2749-9952",
            "S7407056926",
            "https://openalex.org/S7407056926",
            "10.53731/front_matter",
            "https://doi.org/10.53731/front_matter",
        ] {
            let bytes = lookup_container(&path, id)
                .unwrap()
                .unwrap_or_else(|| panic!("container should resolve for '{id}'"));
            let c = container_from_lookup(&bytes);
            assert_eq!(c["title"], "Front Matter", "id '{id}'");
        }
    }

    #[test]
    fn lookup_container_applies_front_matter_blog_override() {
        let dir = tempfile::tempdir().unwrap();
        let path = make_test_containers_db(&dir.path().join("c.sqlite3"));
        let bytes = lookup_container(&path, "2749-9952").unwrap().unwrap();
        let c = container_from_lookup(&bytes);
        // Stored as "Journal"; Front Matter publisher forces "Blog".
        assert_eq!(c["type"], "Blog");
    }

    #[test]
    fn lookup_container_includes_linked_works() {
        let dir = tempfile::tempdir().unwrap();
        let path = make_test_containers_db(&dir.path().join("c.sqlite3"));
        let bytes = lookup_container(&path, "2749-9952").unwrap().unwrap();
        let arr: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
        let arr = arr.as_array().unwrap();
        assert_eq!(arr.len(), 2, "container + one linked work");
        assert_eq!(arr[1]["id"], "https://doi.org/10.53731/article-1");
    }

    #[test]
    fn lookup_container_returns_none_for_unknown() {
        let dir = tempfile::tempdir().unwrap();
        let path = make_test_containers_db(&dir.path().join("c.sqlite3"));
        assert!(lookup_container(&path, "0000-0000").unwrap().is_none());
        assert!(lookup_container(&path, "S9999999999").unwrap().is_none());
    }

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

    #[test]
    fn container_record_reads_and_applies_override() {
        let dir = tempfile::tempdir().unwrap();
        let path = make_test_containers_db(&dir.path().join("c.sqlite3"));
        let rec = container_record(&path, "2749-9952").expect("record present");
        assert_eq!(rec.title, "Front Matter");
        assert_eq!(rec.type_, "Blog"); // stored as Journal, overridden
        assert!(
            rec.identifiers
                .iter()
                .any(|i| i.identifier_type == "ISSN" && i.identifier == "2749-9952")
        );
    }

    #[test]
    fn container_works_local_reads_linked_works() {
        let dir = tempfile::tempdir().unwrap();
        let path = make_test_containers_db(&dir.path().join("c.sqlite3"));
        let works = container_works_local(&path, "10.53731/front_matter");
        assert_eq!(works.len(), 1);
        assert_eq!(works[0].id, "https://doi.org/10.53731/article-1");
    }

    #[test]
    fn container_works_local_empty_for_unknown() {
        let dir = tempfile::tempdir().unwrap();
        let path = make_test_containers_db(&dir.path().join("c.sqlite3"));
        assert!(container_works_local(&path, "0000-0000").is_empty());
    }

    #[test]
    fn cache_upsert_raw_round_trips() {
        let dir = tempfile::tempdir().unwrap();
        let cache = cache_open(&dir.path().join("cache.sqlite3")).expect("cache_open");
        // Simulates the API-fetch path: cache a raw Crossref record (source id 1).
        // It lands in the transport table keyed by bare DOI and converts to a Work
        // on read.
        let raw = serde_json::json!({
            "DOI": "10.53731/fetched-1",
            "type": "journal-article",
            "title": ["Fetched work"],
        })
        .to_string();
        cache_upsert_raw(&cache, &[("10.53731/fetched-1".to_string(), 1, raw)])
            .expect("cache upsert raw");
        let got = cache_lookup(&cache, "10.53731/fetched-1")
            .unwrap()
            .expect("cached work");
        assert_eq!(got.id, "https://doi.org/10.53731/fetched-1");
        assert_eq!(got.title, "Fetched work");
    }

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