dragoman 0.4.57

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())))?;
    // WAL admits readers alongside one writer, but a read still takes a read mark
    // in the wal-index, and that collides with WAL recovery and with a checkpoint
    // — over a large WAL left behind by an interrupted writer, a long window. The
    // default timeout is 0, so such a read failed outright with "database is
    // locked" rather than waiting a moment, surfacing as `Internal error: read
    // container` for a plain SELECT. commonmeta 0.9.86 gives its own readers the
    // 30 s its writers use; these connections are opened here, so they need it too.
    conn.busy_timeout(std::time::Duration::from_secs(30))
        .map_err(|e| {
            AppError::Internal(format!("sqlite busy_timeout '{}': {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)
}

/// Strip the resolver prefix from a pid. `pid_relations` keys `source_id` and
/// `target_id` on **bare** pids (`10.x/y`, a bare ORCID, a bare ROR), never on
/// resolver URLs, so every query against that table must bare its argument first.
/// Mirrors commonmeta's private `bare_pid`.
fn bare_pid(id: &str) -> &str {
    id.trim_start_matches("https://doi.org/")
        .trim_start_matches("http://doi.org/")
        .trim_start_matches("https://dx.doi.org/")
        .trim_start_matches("http://dx.doi.org/")
        .trim_start_matches("https://orcid.org/")
        .trim_start_matches("http://orcid.org/")
        .trim_start_matches("https://ror.org/")
        .trim_start_matches("http://ror.org/")
}

/// Re-expand a bare DOI read back out of `pid_relations` into the resolver URL
/// form `Relation.id` is expected to carry. Stored targets are already lowercased
/// and suffix-encoded by `normalize_doi`, so this only re-adds the prefix.
fn doi_url(bare: &str) -> String {
    format!("https://doi.org/{bare}")
}

/// 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()
}

/// One page of a container's locally stored works, with its total work count.
/// Only the page is hydrated, so the cost does not scale with the container's
/// size. `(vec![], 0)` when the identifier resolves to nothing.
pub fn container_works_local_page(
    path: &PathBuf,
    identifier: &str,
    limit: usize,
    offset: usize,
) -> (Vec<Work>, usize) {
    commonmeta::read_sqlite_by_container_page(identifier, path, Some(limit), offset)
        .unwrap_or_default()
}

/// One page of the works linked to an ORCID person, with their total count.
pub fn orcid_works_page(
    path: &PathBuf,
    orcid_url: &str,
    limit: usize,
    offset: usize,
) -> (Vec<Work>, usize) {
    commonmeta::read_sqlite_by_orcid_page(orcid_url, path, Some(limit), offset).unwrap_or_default()
}

/// One page of the works linked to a ROR organization, with their total count.
pub fn ror_works_page(
    path: &PathBuf,
    ror_url: &str,
    limit: usize,
    offset: usize,
) -> (Vec<Work>, usize) {
    commonmeta::read_sqlite_by_ror_page(ror_url, path, Some(limit), offset).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}")))
}

/// One search suggestion: a navigable identifier (the record's bare pid — a
/// DOI, ORCID or ROR id, which the entity router resolves directly), its
/// display name, and the entity kind used to group results in the UI.
#[derive(serde::Serialize)]
pub struct SearchResult {
    pub id: String,
    pub title: String,
    pub entity_type: String,
}

/// The entity kind a search is restricted to.
///
/// One kind per query rather than all four at once: mixing them meant a work
/// always consumed most of the result list, and a store with 850k works and 30
/// people could never show a person above the works that outranked it.
#[derive(Clone, Copy, PartialEq, Eq, Debug, Default)]
pub enum SearchKind {
    /// Crossref and DataCite records (index sources 1 and 2). The default: the
    /// bulk of any store, and the one kind every store has.
    #[default]
    Works,
    /// ORCID persons (index source 4).
    People,
    /// ROR organizations (index source 3).
    Organizations,
    /// Journals, repositories and other containers (index source 5).
    Containers,
}

impl SearchKind {
    /// Parse the `kind` query parameter. Absent, empty or unrecognised means
    /// [`SearchKind::Works`], so a malformed request still searches something
    /// rather than erroring or returning nothing.
    pub fn parse(value: Option<&str>) -> Self {
        match value.unwrap_or("").trim().to_ascii_lowercase().as_str() {
            "people" | "person" | "persons" => Self::People,
            "organizations" | "organization" | "orgs" | "org" => Self::Organizations,
            "containers" | "container" => Self::Containers,
            _ => Self::Works,
        }
    }

    /// The kind name the tantivy index stores documents under.
    fn index_kind(self) -> &'static str {
        match self {
            Self::Works => "work",
            Self::People => "person",
            Self::Organizations => "organization",
            Self::Containers => "container",
        }
    }

    /// The `entity_type` the UI groups and labels results by.
    fn entity_type(self) -> &'static str {
        match self {
            Self::Works => "Work",
            Self::People => "Person",
            Self::Organizations => "Organization",
            Self::Containers => "Container",
        }
    }
}

/// Suggestions returned for one kind. Uniform across kinds now that a query
/// covers only one: people and organizations previously shared the list with
/// works and were capped at 5.
const SEARCH_LIMIT: usize = 10;

/// How many extra works to ask the index for before versions are dropped.
/// A versioned work is typically one record plus one version, so asking for
/// twice the page keeps a full page after the duplicates go.
const SEARCH_VERSION_OVERFETCH: usize = 2;

/// Full-text search over the tantivy index commonmeta builds beside the store
/// (a sibling `<db>.search` directory; see `commonmeta search --index`),
/// restricted to one [`SearchKind`] and ranked by relevance.
/// Returns an empty list when no index has been built yet.
pub fn search(path: &PathBuf, q: &str, kind: SearchKind) -> Result<Vec<SearchResult>, AppError> {
    let q = q.trim();
    if q.is_empty() || !commonmeta::search::index_dir(path).exists() {
        return Ok(vec![]);
    }
    let mut results: Vec<SearchResult> = Vec::new();

    if kind != SearchKind::Containers {
        // Works are asked for in surplus because versions are dropped below; a
        // page of ten would otherwise come back half empty.
        let want = if kind == SearchKind::Works {
            SEARCH_LIMIT * SEARCH_VERSION_OVERFETCH
        } else {
            SEARCH_LIMIT
        };
        // A query the parser rejects (or a kind absent from the index) yields no
        // suggestions rather than failing the request.
        let Ok(hits) = commonmeta::search::search(path, q, None, Some(kind.index_kind()), want)
        else {
            return Ok(results);
        };
        results.extend(hits.into_iter().map(|h| SearchResult {
            id: h.pid,
            title: h.name,
            entity_type: kind.entity_type().to_string(),
        }));
        if kind == SearchKind::Works {
            drop_superseded_versions(path, &mut results);
            results.truncate(SEARCH_LIMIT);
        }
        return Ok(results);
    }

    // Containers are indexed under a surrogate UUID pid the entity router cannot
    // resolve, so map each hit to a navigable identifier — ISSN, then DOI, then
    // OpenAlex source id, then DataCite client id — from pid_annotations. A
    // container with none of these is dropped: there is nothing to link to.
    if let Ok(hits) = commonmeta::search::search(
        path,
        q,
        None,
        Some(SearchKind::Containers.index_kind()),
        SEARCH_LIMIT,
    ) && !hits.is_empty()
    {
        let pids: Vec<String> = hits.iter().map(|h| h.pid.clone()).collect();
        let ids = container_navigable_ids(path, &pids);
        for h in hits {
            if let Some(id) = ids.get(&h.pid).filter(|s| !s.is_empty()) {
                results.push(SearchResult {
                    id: id.clone(),
                    title: h.name,
                    entity_type: "Container".to_string(),
                });
            }
        }
    }
    Ok(results)
}

/// Map container surrogate pids to a navigable identifier from `pid_annotations`,
/// preferring ISSN, then DOI, then OpenAlex source id (bare `S…`), then DataCite
/// client id — the forms the container router accepts. Pids with no resolvable
/// identifier are omitted. Best-effort: a query error yields an empty map.
/// Drop a result that is a version of a higher-ranked result in the same list.
///
/// A versioned work is stored as two records — the version and what it is a
/// version of — and both match the same query, so a search for one returned the
/// same title twice. `pid_relations` records the link in both directions
/// (`IsVersionOf` and its `HasVersion` inverse); only `IsVersionOf` is consulted
/// here, since the direction is what says which record is the version.
///
/// Rank decides the survivor, not the direction of the link: the first of a
/// versioned set to appear is kept and the rest drop out. Direction is no guide
/// because the index ranks a version above its original as readily as below it
/// — in the pair that prompted this, the `IsVersionOf` source ranks first — and
/// rank is also what keeps the 44 pairs in this store that are `IsVersionOf`
/// each other from cancelling out and losing the work from the results.
fn drop_superseded_versions(path: &PathBuf, results: &mut Vec<SearchResult>) {
    if results.len() < 2 {
        return;
    }
    // `pid_relations` keys on bare pids, so bare the arguments before matching.
    let pids: Vec<String> = results
        .iter()
        .map(|r| bare_pid(&r.id).to_string())
        .collect();
    let Ok(conn) = connect(path) else {
        return;
    };
    let placeholders = std::iter::repeat_n("?", pids.len())
        .collect::<Vec<_>>()
        .join(",");
    // Only links where both ends are on this page can hide a duplicate from it.
    let sql = format!(
        "SELECT source_id, target_id FROM pid_relations \
         WHERE type = 'IsVersionOf' \
           AND source_id IN ({placeholders}) AND target_id IN ({placeholders})"
    );
    let Ok(mut stmt) = conn.prepare(&sql) else {
        return;
    };
    let params = pids.iter().chain(pids.iter());
    let Ok(rows) = stmt.query_map(rusqlite::params_from_iter(params), |row| {
        Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
    }) else {
        return;
    };
    // Linked in both directions: the index ranks a version above the record it
    // is a version of as readily as below it, so which side of the relation a
    // hit sits on says nothing about which one the reader should keep.
    let mut linked: std::collections::HashMap<String, Vec<String>> =
        std::collections::HashMap::new();
    for (source, target) in rows.flatten() {
        linked
            .entry(source.clone())
            .or_default()
            .push(target.clone());
        linked.entry(target).or_default().push(source);
    }
    if linked.is_empty() {
        return;
    }
    // `retain` visits in rank order, so "already kept" means "ranked above this
    // one" — the first of a versioned set survives and the rest drop out.
    let mut kept: std::collections::HashSet<String> = std::collections::HashSet::new();
    results.retain(|r| {
        let pid = bare_pid(&r.id).to_string();
        let duplicates_kept = linked
            .get(&pid)
            .is_some_and(|others| others.iter().any(|o| kept.contains(o)));
        if !duplicates_kept {
            kept.insert(pid);
        }
        !duplicates_kept
    });
}

fn container_navigable_ids(
    path: &PathBuf,
    pids: &[String],
) -> std::collections::HashMap<String, String> {
    let mut out = std::collections::HashMap::new();
    if pids.is_empty() {
        return out;
    }
    let Ok(conn) = connect(path) else {
        return out;
    };
    let placeholders = std::iter::repeat_n("?", pids.len())
        .collect::<Vec<_>>()
        .join(",");
    let sql = format!(
        "SELECT pid, issn, doi, openalex, datacite FROM pid_annotations WHERE pid IN ({placeholders})"
    );
    let Ok(mut stmt) = conn.prepare(&sql) else {
        return out;
    };
    let rows = stmt.query_map(rusqlite::params_from_iter(pids.iter()), |row| {
        let col = |i: usize| -> String {
            row.get::<_, Option<String>>(i)
                .ok()
                .flatten()
                .unwrap_or_default()
        };
        let pid: String = row.get(0)?;
        let (issn, doi, openalex, datacite) = (col(1), col(2), col(3), col(4));
        // The OpenAlex column stores a resolver URL; the router wants the bare id.
        let id = if !issn.is_empty() {
            issn
        } else if !doi.is_empty() {
            doi
        } else if !openalex.is_empty() {
            openalex
                .trim_start_matches("https://openalex.org/")
                .to_string()
        } else {
            datacite
        };
        Ok((pid, id))
    });
    if let Ok(rows) = rows {
        for (pid, id) in rows.flatten() {
            out.insert(pid, id);
        }
    }
    out
}

/// 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> {
    // pid_relations keys on BARE pids (`10.x/y`, `0000-0002-…`, `05gq02987`), so
    // every lookup must strip the resolver prefix before matching — the URL form
    // never matches the stored bare target.
    if id.starts_with("https://doi.org/") {
        // Per DataCite, a work is cited by any work pointing to it with a
        // reference-type relation (References / Cites / IsSupplementedBy),
        // counting distinct sources. commonmeta::count_sqlite_citations bares the
        // DOI (matching the same encoding the store uses) and applies exactly
        // this definition.
        // https://support.datacite.org/docs/contributing-citations-and-references
        let n = commonmeta::count_sqlite_citations(id, path)
            .map_err(|e| AppError::Internal(format!("count citations: {e}")))?
            as i64;
        Ok(EntityCount {
            citations: Some(n),
            works: None,
        })
    } else if let Some(orcid) = id.strip_prefix("https://orcid.org/") {
        // Work→person links are `hasPerson` rows in pid_relations (target = bare ORCID).
        let conn = connect(path)?;
        // unwrap_or(0) silently handles "no such table" for un-enriched databases.
        let n: i64 = conn.query_row(
            r#"SELECT COUNT(*) FROM pid_relations WHERE "target_id" = ?1 AND "type" = 'hasPerson'"#,
            params![orcid], |row| row.get(0),
        ).unwrap_or(0);
        Ok(EntityCount {
            citations: None,
            works: Some(n),
        })
    } else if let Some(ror) = id.strip_prefix("https://ror.org/") {
        // Work→organization links are `hasOrganization` rows (target = bare ROR).
        let conn = connect(path)?;
        let n: i64 = conn.query_row(
            r#"SELECT COUNT(*) FROM pid_relations WHERE "target_id" = ?1 AND "type" = 'hasOrganization'"#,
            params![ror], |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 = bare_pid(&commonmeta::doi_utils::normalize_doi(doi)).to_string();
    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
             AND "type" NOT IN ('hasPerson', 'hasOrganization', 'hasContainer')
           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((doi_url(&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 = bare_pid(&commonmeta::doi_utils::normalize_doi(doi)).to_string();
    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
             AND "type" NOT IN ('hasPerson', 'hasOrganization', 'hasContainer')
           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((
                doi_url(&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(),
    };
    // Callers pass resolver-URL ids, but pid_relations keys on bare pids.
    let bare: Vec<String> = ids.iter().map(|id| bare_pid(id).to_string()).collect();
    let placeholders = bare
        .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> =
        bare.iter().map(|s| s as &dyn rusqlite::ToSql).collect();
    // Re-expand to the resolver-URL form the caller compares against.
    match stmt.query_map(params.as_slice(), |row| row.get::<_, String>(0)) {
        Ok(rows) => rows.filter_map(|r| r.ok()).map(|id| doi_url(&id)).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()
}

/// 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 connect_sets_busy_timeout() {
        let dir = tempfile::tempdir().unwrap();
        let path = make_test_db(&dir.path().join("test.sqlite3"));
        let conn = connect(&path).expect("connect");
        // A reader left at the default 0 fails a contended SELECT outright
        // instead of waiting; 30 s matches what commonmeta gives its own readers.
        let timeout: i64 = conn
            .query_row("PRAGMA busy_timeout", [], |row| row.get(0))
            .expect("read busy_timeout");
        assert_eq!(
            timeout, 30_000,
            "expected a 30 s busy_timeout, got {timeout}"
        );
    }

    #[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 search_kind_parses_and_defaults_to_works() {
        // The UI sends these four values; everything else must still search
        // something rather than erroring or coming back empty.
        assert_eq!(SearchKind::parse(Some("works")), SearchKind::Works);
        assert_eq!(SearchKind::parse(Some("people")), SearchKind::People);
        assert_eq!(
            SearchKind::parse(Some("organizations")),
            SearchKind::Organizations
        );
        assert_eq!(
            SearchKind::parse(Some("containers")),
            SearchKind::Containers
        );
        // Case and surrounding space are not the caller's problem.
        assert_eq!(SearchKind::parse(Some("  People ")), SearchKind::People);
        // Singular forms, since the index kinds themselves are singular.
        assert_eq!(SearchKind::parse(Some("person")), SearchKind::People);
        assert_eq!(SearchKind::parse(Some("container")), SearchKind::Containers);
        // Absent, empty and unknown all mean works.
        assert_eq!(SearchKind::parse(None), SearchKind::Works);
        assert_eq!(SearchKind::parse(Some("")), SearchKind::Works);
        assert_eq!(SearchKind::parse(Some("nonsense")), SearchKind::Works);
        assert_eq!(SearchKind::default(), SearchKind::Works);
    }

    #[test]
    fn search_empty_query_returns_no_results() {
        let dir = tempfile::tempdir().unwrap();
        let path = make_test_db(&dir.path().join("test.sqlite3"));
        assert!(search(&path, "   ", SearchKind::Works).unwrap().is_empty());
    }

    #[test]
    fn search_without_index_returns_empty() {
        // No sibling `<db>.search` directory exists, so search yields nothing
        // rather than erroring — the server falls back to plain PID resolution.
        let dir = tempfile::tempdir().unwrap();
        let path = make_test_db(&dir.path().join("test.sqlite3"));
        assert!(!commonmeta::search::index_dir(&path).exists());
        assert!(
            search(&path, "carberry", SearchKind::Works)
                .unwrap()
                .is_empty()
        );
    }

    #[test]
    fn container_navigable_ids_prefers_issn_then_doi_then_openalex_then_datacite() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("c.sqlite3");
        let conn = rusqlite::Connection::open(&path).unwrap();
        conn.execute_batch(
            "CREATE TABLE pid_annotations (pid TEXT, issn TEXT DEFAULT '', doi TEXT DEFAULT '', \
             openalex TEXT DEFAULT '', datacite TEXT DEFAULT '');
             INSERT INTO pid_annotations VALUES
               ('u-issn','2749-9952','https://doi.org/10.1/x','https://openalex.org/S1','a.b'),
               ('u-doi','','https://doi.org/10.2/y','https://openalex.org/S2','c.d'),
               ('u-oa','','','https://openalex.org/S3','e.f'),
               ('u-dc','','','','g.h'),
               ('u-none','','','','');",
        )
        .unwrap();
        let pids: Vec<String> = ["u-issn", "u-doi", "u-oa", "u-dc", "u-none"]
            .iter()
            .map(|s| s.to_string())
            .collect();
        let ids = container_navigable_ids(&path, &pids);
        assert_eq!(ids.get("u-issn").map(String::as_str), Some("2749-9952"));
        assert_eq!(
            ids.get("u-doi").map(String::as_str),
            Some("https://doi.org/10.2/y")
        );
        // The OpenAlex resolver URL is reduced to the bare source id.
        assert_eq!(ids.get("u-oa").map(String::as_str), Some("S3"));
        assert_eq!(ids.get("u-dc").map(String::as_str), Some("g.h"));
        // No navigable identifier → empty, which search() then drops.
        assert_eq!(ids.get("u-none").map(String::as_str), Some(""));
    }

    #[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");
        // Bare pids, exactly as the real store keys them.
        for (src, tgt, rel_type) in &[
            (
                "10.5281/zenodo.1000000",
                "10.5281/zenodo.1000001",
                "HasVersion",
            ),
            (
                "10.5281/zenodo.1000000",
                "10.5281/zenodo.1000002",
                "HasVersion",
            ),
            (
                "10.5281/zenodo.9999999",
                "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()
    }

    // ── drop_superseded_versions ──────────────────────────────────────────────

    /// Build a relations store holding only the version links a test names.
    fn versions_db(path: &std::path::Path, links: &[(&str, &str, &str)]) -> PathBuf {
        let conn = Connection::open(path).expect("open versions 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) in links {
            conn.execute(
                r#"INSERT INTO pid_relations (source_id, target_id, "type") VALUES (?1, ?2, ?3)"#,
                params![src, tgt, rel],
            )
            .expect("insert relation");
        }
        path.to_path_buf()
    }

    fn work_hits(ids: &[&str]) -> Vec<SearchResult> {
        ids.iter()
            .map(|id| SearchResult {
                id: (*id).to_string(),
                title: "same title".to_string(),
                entity_type: "Work".to_string(),
            })
            .collect()
    }

    fn ids_of(results: &[SearchResult]) -> Vec<&str> {
        results.iter().map(|r| r.id.as_str()).collect()
    }

    #[test]
    fn drop_superseded_versions_keeps_the_higher_ranked_record() {
        // The real pair behind the report: both records carry the same title, and
        // pid_relations stores the link in both directions.
        let dir = tempfile::tempdir().unwrap();
        let path = versions_db(
            &dir.path().join("v.sqlite3"),
            &[
                (
                    "10.59350/27wky-wpg41",
                    "10.59350/4w0sm-qww29",
                    "IsVersionOf",
                ),
                ("10.59350/4w0sm-qww29", "10.59350/27wky-wpg41", "HasVersion"),
            ],
        );
        // Ranked with the canonical record first: the version below it goes.
        let mut results = work_hits(&["10.59350/4w0sm-qww29", "10.59350/27wky-wpg41"]);
        drop_superseded_versions(&path, &mut results);
        assert_eq!(ids_of(&results), vec!["10.59350/4w0sm-qww29"]);

        // Ranked the other way, the survivor is whichever ranked higher — never
        // nothing, and never both.
        let mut results = work_hits(&["10.59350/27wky-wpg41", "10.59350/4w0sm-qww29"]);
        drop_superseded_versions(&path, &mut results);
        assert_eq!(ids_of(&results), vec!["10.59350/27wky-wpg41"]);
    }

    #[test]
    fn drop_superseded_versions_survives_a_mutual_pair() {
        // This store holds 44 pairs that are IsVersionOf each other. Dropping on
        // the relation alone would take both and lose the work entirely.
        let dir = tempfile::tempdir().unwrap();
        let path = versions_db(
            &dir.path().join("v.sqlite3"),
            &[
                ("10.1/a", "10.1/b", "IsVersionOf"),
                ("10.1/b", "10.1/a", "IsVersionOf"),
            ],
        );
        let mut results = work_hits(&["10.1/a", "10.1/b"]);
        drop_superseded_versions(&path, &mut results);
        assert_eq!(ids_of(&results), vec!["10.1/a"], "exactly one must survive");
    }

    #[test]
    fn drop_superseded_versions_leaves_unrelated_results_alone() {
        let dir = tempfile::tempdir().unwrap();
        let path = versions_db(
            &dir.path().join("v.sqlite3"),
            &[("10.1/a", "10.1/elsewhere", "IsVersionOf")],
        );
        // 10.1/a is a version of something that is not on this page, so it stays:
        // only a duplicate visible in the same list is worth hiding.
        let mut results = work_hits(&["10.1/a", "10.1/b", "10.1/c"]);
        drop_superseded_versions(&path, &mut results);
        assert_eq!(ids_of(&results), vec!["10.1/a", "10.1/b", "10.1/c"]);
    }

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

    /// Citation/credit fixture keyed on BARE pids — the form the real enriched store
    /// uses (`enrich` writes `10.x/y`, `0000-…`, `05gq…`, never the resolver URL).
    /// Every reader strips the resolver prefix before matching, so all of their tests
    /// must run against this shape or they will pass against data production never has.
    pub(crate) fn make_test_pid_relations_db(path: &std::path::Path) -> PathBuf {
        let conn = Connection::open(path).expect("open pid_relations 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");
        // Z is cited by A and B (References) and by S (IsSupplementedBy) → 3 citations.
        // A and B each credit ORCID 0000-0002-1825-0097 → 2 works; A credits ROR
        // 04aj4c181 → 1 work. All pids bare, exactly as the store keys them.
        let rows: &[(&str, &str, &str, i64)] = &[
            ("10.1234/citing-a", "10.5678/cited-z", "References", 2023),
            ("10.1234/citing-b", "10.5678/cited-z", "References", 2024),
            (
                "10.1234/supplement-s",
                "10.5678/cited-z",
                "IsSupplementedBy",
                2025,
            ),
            ("10.1234/citing-c", "10.5678/other", "References", 2024),
            ("10.1234/citing-a", "0000-0002-1825-0097", "hasPerson", 0),
            ("10.1234/citing-b", "0000-0002-1825-0097", "hasPerson", 0),
            ("10.1234/citing-a", "04aj4c181", "hasOrganization", 0),
        ];
        for (source_id, target_id, type_, source_year) in rows {
            conn.execute(
                r#"INSERT INTO pid_relations (source_id, target_id, type, source_year) VALUES (?1, ?2, ?3, ?4)"#,
                params![source_id, target_id, type_, source_year],
            )
            .expect("insert pid_relations row");
        }
        path.to_path_buf()
    }

    #[test]
    fn lookup_target_relations_empty_doi_returns_empty() {
        let dir = tempfile::tempdir().unwrap();
        let path = make_test_pid_relations_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_pid_relations_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_pid_relations_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_pid_relations_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_pid_relations_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_pid_relations_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_pid_relations_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_pid_relations_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 relation_readers_exclude_structural_entity_edges() {
        // hasPerson/hasOrganization rows carry bare ORCID/ROR targets and exist only to
        // power the /api/count tallies. Surfacing them as work relations would emit a
        // Relation.id that is not a resolvable work, so both readers must drop them.
        let dir = tempfile::tempdir().unwrap();
        let path = make_test_pid_relations_db(&dir.path().join("refs.sqlite3"));

        // citing-a has hasPerson + hasOrganization edges alongside its References row.
        let types: Vec<String> = lookup_source_relations(&path, "https://doi.org/10.1234/citing-a")
            .unwrap()
            .into_iter()
            .map(|(_id, type_)| type_)
            .collect();
        assert!(
            !types
                .iter()
                .any(|t| t == "hasPerson" || t == "hasOrganization"),
            "structural edges leaked into source relations: {types:?}"
        );

        // Querying the person as a target must not surface the works crediting them.
        let inbound =
            lookup_target_relations(&path, "https://orcid.org/0000-0002-1825-0097").unwrap();
        assert!(
            inbound.is_empty(),
            "structural edges leaked into target relations: {inbound:?}"
        );
    }

    #[test]
    fn lookup_target_relations_returns_resolver_urls_not_bare_pids() {
        // The store keys bare, but Relation.id must carry the resolver URL form.
        let dir = tempfile::tempdir().unwrap();
        let path = make_test_pid_relations_db(&dir.path().join("refs.sqlite3"));
        let rels = lookup_target_relations(&path, "https://doi.org/10.5678/cited-z").unwrap();
        assert!(!rels.is_empty(), "fixture should yield inbound citations");
        assert!(
            rels.iter()
                .all(|(id, _, _)| id.starts_with("https://doi.org/")),
            "ids must be re-expanded to resolver URLs: {rels:?}"
        );
    }

    #[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, total) = container_works_local_page(&path, "10.53731/front_matter", 50, 0);
        assert_eq!(total, 1);
        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"));
        let (works, total) = container_works_local_page(&path, "0000-0000", 50, 0);
        assert!(works.is_empty());
        assert_eq!(total, 0, "an unknown container has no works to page over");
    }

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