dragoman 0.5.17

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

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

use crate::error::AppError;
use crate::pool;

/// 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 commonmeta store: a SQLite database
/// with a `pid_records` 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 = pool::get(&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 from the `pid_records` table, which holds works,
/// people, organizations and containers alike, keyed by `pid`. An older store
/// may also have the legacy `organizations` table (keyed by `id`), which is drawn
/// from too when it is there and not empty.
pub fn random_id(path: &PathBuf) -> Result<Option<String>, AppError> {
    let conn = pool::get(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]` as a commonmeta JSON array. 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.
///
/// The record only, never its works: `/api/container/works` serves those paged.
/// Embedding them sent every one at once — 1292 works, 3.7 MB, for a blog — to
/// readers that wanted the container itself: the entity page, a resolve of its
/// url, and a bookmark lookup that kept only the first element.
pub fn lookup_container(path: &PathBuf, identifier: &str) -> Result<Option<Vec<u8>>, AppError> {
    let conn = pool::get(path)?;
    let Some(container) = commonmeta::read_container_sqlite_conn(identifier, &conn)
        .map_err(|e| AppError::Internal(format!("read container: {e}")))?
    else {
        return Ok(None);
    };
    let bytes = serde_json::to_vec_pretty(&[container])
        .map_err(|e| AppError::Internal(format!("container serialize: {e}")))?;
    Ok(Some(bytes))
}

/// 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_conn(identifier, &*pool::get(path).ok()?)
        .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,
    page: commonmeta::WorksPage<'_>,
) -> (Vec<Work>, usize) {
    let Ok(conn) = pool::get(path) else {
        return Default::default();
    };
    commonmeta::read_sqlite_by_container_page_conn(identifier, &conn, page).unwrap_or_default()
}

/// Counts of a container's works by type and year, or `None` when the list is
/// too long to count (see [`FACETS_MAX_WORKS`]) or the store cannot answer.
pub fn container_work_facets(
    path: &PathBuf,
    identifier: &str,
    max_works: usize,
) -> Option<commonmeta::WorkFacets> {
    commonmeta::read_sqlite_container_facets_conn(
        identifier,
        &*pool::get(path).ok()?,
        max_works,
        true,
    )
    .ok()
    .flatten()
}

/// One page of the works linked to an ORCID person, with their total count.
pub fn orcid_works_page(
    path: &PathBuf,
    orcid_url: &str,
    page: commonmeta::WorksPage<'_>,
) -> (Vec<Work>, usize) {
    let Ok(conn) = pool::get(path) else {
        return Default::default();
    };
    commonmeta::read_sqlite_by_orcid_page_conn(orcid_url, &conn, page).unwrap_or_default()
}

/// Counts of a person's works by type and year. See [`container_work_facets`].
pub fn orcid_work_facets(
    path: &PathBuf,
    orcid_url: &str,
    max_works: usize,
) -> Option<commonmeta::WorkFacets> {
    commonmeta::read_sqlite_orcid_facets_conn(orcid_url, &*pool::get(path).ok()?, max_works, true)
        .ok()
        .flatten()
}

/// One page of the works linked to a ROR organization, with their total count.
pub fn ror_works_page(
    path: &PathBuf,
    ror_url: &str,
    page: commonmeta::WorksPage<'_>,
) -> (Vec<Work>, usize) {
    let Ok(conn) = pool::get(path) else {
        return Default::default();
    };
    commonmeta::read_sqlite_by_ror_page_conn(ror_url, &conn, page).unwrap_or_default()
}

/// Counts of an organization's works by type and year. See
/// [`container_work_facets`].
pub fn ror_work_facets(
    path: &PathBuf,
    ror_url: &str,
    max_works: usize,
) -> Option<commonmeta::WorkFacets> {
    commonmeta::read_sqlite_ror_facets_conn(ror_url, &*pool::get(path).ok()?, max_works, true)
        .ok()
        .flatten()
}

/// 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_conn(&id, &*pool::get(path)?)
        .map_err(|e| AppError::Internal(format!("sqlite lookup: {e}")))
}

/// The URL a DOI resolves to, from a store's `pid_records`, without building the
/// work: all a redirect needs. `None` when the DOI is absent or names no URL. A
/// store that cannot answer is a miss, as for [`cache_lookup`].
pub fn lookup_url(path: &PathBuf, doi: &str) -> Option<String> {
    let id = commonmeta::doi_utils::normalize_doi(doi);
    if id.is_empty() {
        return None;
    }
    commonmeta::read_sqlite_url_by_id_conn(&id, &*pool::get(path).ok()?)
        .ok()
        .flatten()
}

/// 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 other names the entity is known by — an organization's acronym and
    /// its alternates. Absent for a kind that has none.
    #[serde(skip_serializing_if = "Vec::is_empty", default)]
    pub also_known_as: Vec<String>,
    /// Where it is, as a place name, else the country.
    #[serde(skip_serializing_if = "String::is_empty", default)]
    pub location: String,
    /// An organization's ROR types, lowercased — `education`, `funder`.
    #[serde(skip_serializing_if = "Vec::is_empty", default)]
    pub types: Vec<String>,
    /// A person's given name on its own.
    #[serde(skip_serializing_if = "String::is_empty", default)]
    pub given_name: String,
    /// A person's family name on its own.
    #[serde(skip_serializing_if = "String::is_empty", default)]
    pub family_name: String,
    /// A person's country, as its two-letter code.
    #[serde(skip_serializing_if = "String::is_empty", default)]
    pub country: String,
    /// A person's biography, as plain text.
    #[serde(skip_serializing_if = "String::is_empty", default)]
    pub description: String,
    /// A container's publisher, by name.
    #[serde(skip_serializing_if = "String::is_empty", default)]
    pub publisher: String,
    /// The publisher's own id. A ROR links to the organization's page here, as
    /// it does on the container card; anything else is shown as text.
    #[serde(skip_serializing_if = "String::is_empty", default)]
    pub publisher_id: String,
    /// A container's other identifiers — the ISSN, DOI, OpenAlex source id and
    /// DataCite client id the card lists under "also known as".
    #[serde(skip_serializing_if = "Vec::is_empty", default)]
    pub identifiers: Vec<SearchIdentifier>,
    /// A container's commonmeta type, which decides whether its works are
    /// called works or blog posts.
    #[serde(skip_serializing_if = "String::is_empty", default)]
    pub container_type: String,
    /// How many works the container holds, versions left out — the number its
    /// own page shows.
    #[serde(skip_serializing_if = "is_zero", default)]
    pub works: usize,
}

fn is_zero(n: &usize) -> bool {
    *n == 0
}

/// One identifier of a search result, as the card labels and links it.
#[derive(serde::Serialize)]
pub struct SearchIdentifier {
    pub identifier: String,
    pub identifier_type: String,
}

impl SearchResult {
    /// A hit with nothing beyond its name — what the index alone can say.
    fn named(id: String, title: String, entity_type: &str) -> Self {
        SearchResult {
            id,
            title,
            entity_type: entity_type.to_string(),
            also_known_as: Vec::new(),
            location: String::new(),
            types: Vec::new(),
            given_name: String::new(),
            family_name: String::new(),
            country: String::new(),
            description: String::new(),
            publisher: String::new(),
            publisher_id: String::new(),
            identifiers: Vec::new(),
            container_type: String::new(),
            works: 0,
        }
    }
}

/// 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",
        }
    }
}

/// Results per page of the result list. The UI asks for a page size; this is
/// what a request that names none gets.
pub const SEARCH_LIMIT: usize = 10;

/// The most results a page may ask for, so a hand-written request cannot ask
/// the index for an unbounded page.
pub const SEARCH_LIMIT_MAX: usize = 100;

/// How many records of each kind a search can find.
#[derive(serde::Serialize, Default, Clone, Copy)]
pub struct StoreCounts {
    pub works: usize,
    pub people: usize,
    pub organizations: usize,
    pub containers: usize,
}

/// What the store holds of each kind, as the number a search for it returns.
///
/// Read from the search index, where a kind's total is its term's document
/// frequency: constant time, whatever the store holds. `commonmeta stats`
/// counts the same four from `pid_annotations` and agrees to the record —
/// 993,755 works, 117,299 people, 144,717 organizations, 318,671 containers on
/// one store — but by scanning the source index, which is 60 ms for 1.5 M rows
/// and so about eleven seconds for 300 M.
///
/// These are the records a search can *reach*, which is what the sentence they
/// appear in promises. A store enriched beyond its last reindex therefore reads
/// low, which is the honest number for "search 993,755 works": the rest cannot
/// be found yet.
pub fn store_counts(path: &PathBuf) -> StoreCounts {
    let of = |kind: SearchKind| {
        search_page(path, SEARCH_ALL, kind, &SearchFilters::default(), 1, 0)
            .map(|(_, total)| total)
            .unwrap_or(0)
    };
    StoreCounts {
        works: of(SearchKind::Works),
        people: of(SearchKind::People),
        organizations: of(SearchKind::Organizations),
        containers: of(SearchKind::Containers),
    }
}

/// The query that asks for everything of a kind.
pub const SEARCH_ALL: &str = "*";

/// The most matches worth faceting.
///
/// A facet walks every match: measured at 0.38 s per million on a 994 k-work
/// index, and the cost follows the matches rather than the store, so the same
/// rate holds for a larger one. A million keeps the sidebar inside about four
/// tenths of a second; past that the reader gets the list and no sidebar,
/// which is how `/api/facets` already treats a works list too long to count.
///
/// commonmeta enforces it from the term dictionary, before walking anything,
/// so a query arriving with a facet value already picked is bounded by what the
/// query itself matches rather than by what the filter left of it.
pub const SEARCH_FACETS_MAX_MATCHES: usize = 1_000_000;

/// How many values each facet offers.
///
/// Every facet runs well past ten in a large store — "university" matches
/// organizations in forty countries — and a sidebar is read, not scrolled. What
/// is cut is the long tail: the values are ranked by count, except the years,
/// which are ranked by date because a year is read as a date and the ten
/// *largest* would scatter decades up the list in an order nothing explains.
pub const SEARCH_FACET_VALUES: usize = 10;

/// The most recent years, newest first — but a year still to come sorts after
/// every year that has, as a work dated past today does in a works list. Such
/// a date is a deposit error, and one work dated 2029 would otherwise take the
/// first row ahead of the 165 of this year, pushing a real year off the end of
/// the list.
///
/// Indexing only ever writes a four-digit year, so every value parses; one that
/// somehow does not sorts last rather than being dropped.
fn search_facets_years(
    counts: Vec<commonmeta::search::FacetCount>,
    keep: usize,
    this_year: i64,
) -> Vec<SearchFacetCount> {
    let mut years: Vec<SearchFacetCount> = counts
        .into_iter()
        .map(|c| SearchFacetCount {
            value: c.value,
            count: c.count,
        })
        .collect();
    years.sort_by_key(|c| {
        let year = c.value.parse::<i64>().unwrap_or(i64::MIN);
        (year > this_year, std::cmp::Reverse(year))
    });
    years.truncate(keep);
    years
}

/// The current year, for telling a published year from one still to come.
fn this_year() -> i64 {
    let secs = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs() as i64)
        .unwrap_or(0);
    // Days since the epoch over the mean Gregorian year, which is exact enough
    // to name a year and needs no date library.
    1970 + (secs / 86_400) * 400 / 146_097
}

/// 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;

/// What a result list is narrowed by.
///
/// Each field holds the values picked for one facet, and each value is one the
/// index counted, so a filter the UI offers is one that has results behind it.
/// Within a facet the values are ORed and across facets ANDed — see
/// `commonmeta::search::SearchFilters` — so a reader can ask for two countries
/// without giving up the type they also picked.
#[derive(Clone, Debug, Default)]
pub struct SearchFilters {
    pub year: Vec<String>,
    pub language: Vec<String>,
    pub work_type: Vec<String>,
    pub subject: Vec<String>,
    /// The countries of a person or an organization, by two-letter code.
    pub country: Vec<String>,
}

impl SearchFilters {
    /// The same values as `&str`, which is what commonmeta takes.
    ///
    /// It cannot be built in one step: the slices commonmeta borrows have to
    /// outlive the call, so they are held here and handed over from this.
    fn borrowed(&self) -> BorrowedFilters<'_> {
        fn of(values: &[String]) -> Vec<&str> {
            values.iter().map(String::as_str).collect()
        }
        BorrowedFilters {
            year: of(&self.year),
            language: of(&self.language),
            work_type: of(&self.work_type),
            subject: of(&self.subject),
            country: of(&self.country),
        }
    }
}

/// [`SearchFilters`] borrowed, so commonmeta can be handed slices of `&str`.
struct BorrowedFilters<'a> {
    year: Vec<&'a str>,
    language: Vec<&'a str>,
    work_type: Vec<&'a str>,
    subject: Vec<&'a str>,
    country: Vec<&'a str>,
}

impl BorrowedFilters<'_> {
    fn as_commonmeta(&self) -> commonmeta::search::SearchFilters<'_> {
        commonmeta::search::SearchFilters {
            year: &self.year,
            language: &self.language,
            work_type: &self.work_type,
            subject: &self.subject,
            country: &self.country,
        }
    }
}

/// One facet value and how many results carry it.
#[derive(serde::Serialize)]
pub struct SearchFacetCount {
    pub value: String,
    pub count: u64,
}

/// The counts behind a result list, each most common first.
///
/// `counted` is false when the index was built before facets existed, which
/// lets the UI leave the sidebar out rather than draw four empty facets. See
/// `commonmeta::search::SearchFacets`.
#[derive(serde::Serialize, Default)]
pub struct SearchFacets {
    pub years: Vec<SearchFacetCount>,
    pub languages: Vec<SearchFacetCount>,
    pub types: Vec<SearchFacetCount>,
    pub subjects: Vec<SearchFacetCount>,
    pub countries: Vec<SearchFacetCount>,
    pub counted: bool,
}

/// Log an index that could not answer, and treat it as a miss.
///
/// A query the parser rejects — or a kind absent from the index — yields no
/// results rather than failing the request, but it is logged: an index that
/// cannot be opened looks exactly like one that matched nothing, and twice that
/// has read as "search is broken" with nothing in the log to say why.
fn warn_index_unavailable(path: &Path, kind: &str, error: &dyn std::fmt::Display) {
    tracing::warn!(
        kind,
        index = %commonmeta::search::index_dir(path).display(),
        error = %error,
        "search index unavailable, returning no results"
    );
}

/// One page of the result list, and the size of the list it came from.
///
/// 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`], narrowed by `filters` and ranked by
/// relevance. An empty list when no index has been built yet.
///
/// The total is the index's count of the same query, so pages can be counted
/// before they are fetched. Two things can still make a page shorter than
/// `limit`: a work that is a superseded version of another on the same page,
/// and a container with no identifier to link to. Both are rare and neither
/// empties a page — unlike dropping versions from a client-side slice, which
/// emptied the later pages of a blog.
pub fn search_page(
    path: &PathBuf,
    q: &str,
    kind: SearchKind,
    filters: &SearchFilters,
    limit: usize,
    offset: usize,
) -> Result<(Vec<SearchResult>, usize), AppError> {
    let q = q.trim();
    if q.is_empty() || !commonmeta::search::index_dir(path).exists() {
        return Ok((vec![], 0));
    }
    let limit = limit.clamp(1, SEARCH_LIMIT_MAX);
    // Works are asked for in surplus because versions are dropped below; a page
    // would otherwise come back short.
    let want = if kind == SearchKind::Works {
        limit * SEARCH_VERSION_OVERFETCH
    } else {
        limit
    };
    let borrowed = filters.borrowed();
    let (hits, total) = match commonmeta::search::search_page(
        path,
        q,
        None,
        Some(kind.index_kind()),
        borrowed.as_commonmeta(),
        want,
        offset,
    ) {
        Ok(found) => found,
        Err(e) => {
            warn_index_unavailable(path, kind.index_kind(), &e);
            return Ok((vec![], 0));
        }
    };

    // 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 kind == SearchKind::Containers {
        return Ok((container_results(path, &hits), total));
    }

    // An organization is shown by name, the names it is also known by, where it
    // is and its ROR id, which needs the record: the index holds its names
    // concatenated into one searchable field, so the hit's `name` reads
    // "Freie Universität Berlin FU Berlin …" rather than a name.
    if kind == SearchKind::Organizations {
        return Ok((organization_results(path, &hits), total));
    }
    if kind == SearchKind::People {
        return Ok((person_results(path, &hits), total));
    }

    let mut results: Vec<SearchResult> = hits
        .into_iter()
        .map(|h| SearchResult::named(h.pid, h.name, kind.entity_type()))
        .collect();
    if kind == SearchKind::Works {
        drop_superseded_versions(path, &mut results);
    }
    results.truncate(limit);
    Ok((results, total))
}

/// Organization hits as rows: the name, the names it is also known by, where it
/// is, and the ROR id — read from the store in one batch.
///
/// A record the store cannot produce keeps the hit rather than dropping it: the
/// ROR id alone is still worth linking, and the index found it for a reason.
/// The indexed name is the fallback, which is why it is only ever a fallback —
/// it holds every name the organization has, run together.
fn organization_results(
    path: &PathBuf,
    hits: &[commonmeta::search::SearchHit],
) -> Vec<SearchResult> {
    let pids: Vec<String> = hits.iter().map(|h| h.pid.clone()).collect();
    let records: HashMap<String, Organization> =
        match commonmeta::pidbox::read_sqlite_orgs_by_pid(path, &pids) {
            Ok(found) => found.into_iter().collect(),
            Err(e) => {
                tracing::warn!(error = %e, "reading organizations for search results failed");
                HashMap::new()
            }
        };
    // The works the store links to each, counted for the page in one query.
    // Only 13,307 of this store's 144,717 organizations have any such edge, so
    // most rows carry no count — and say nothing rather than claiming none. An
    // organization's own page falls back to a Crossref/DataCite fetch when the
    // store has no edges, which is not a thing to do ten times for a page of
    // results.
    let works = works_counts(path, &pids, "hasOrganization");
    hits.iter()
        .map(|hit| {
            let counted = works.get(&hit.pid).copied().unwrap_or(0);
            let Some(org) = records.get(&hit.pid) else {
                return SearchResult {
                    works: counted,
                    ..SearchResult::named(hit.pid.clone(), hit.name.clone(), "Organization")
                };
            };
            // The acronym first, then the alternates, each once and none of them
            // the name again — ROR records repeat the name among its aliases.
            let mut also_known_as: Vec<String> = Vec::new();
            for name in std::iter::once(&org.acronym).chain(org.additional_names.iter()) {
                let name = name.trim();
                if !name.is_empty()
                    && !name.eq_ignore_ascii_case(org.name.trim())
                    && !also_known_as
                        .iter()
                        .any(|seen| seen.eq_ignore_ascii_case(name))
                {
                    also_known_as.push(name.to_string());
                }
            }
            // The place if the record names one, else the country it is in.
            let location = org
                .geo_locations
                .iter()
                .map(|geo| geo.place.trim())
                .find(|place| !place.is_empty())
                .unwrap_or(org.country.trim())
                .to_string();
            SearchResult {
                id: hit.pid.clone(),
                title: if org.name.trim().is_empty() {
                    hit.name.clone()
                } else {
                    org.name.clone()
                },
                entity_type: "Organization".to_string(),
                also_known_as,
                location,
                types: org.types.clone(),
                works: counted,
                ..SearchResult::named(String::new(), String::new(), "Organization")
            }
        })
        .collect()
}

/// How many works each of `pids` holds under `relation`, versions left out —
/// one query for the page rather than one per row.
///
/// The same list the entity's own page counts, so a result and the page it
/// links to agree. Versions have to be excluded for that: Zenodo holds 88,639
/// `hasContainer` edges and 54,755 works, and its page says 54,755; Humboldt
/// has 84 `hasOrganization` edges and 78 works, and its page says 78.
///
/// The cost is proportional to the edges counted, not to the store: the
/// `(target_id, type)` index covers the outer scan and the subquery hits the
/// primary key. Measured on a 33.7 M-row `pid_relations`: a page of ten real
/// journals (5,496 works between them) takes 30 ms and one holding Zenodo
/// 150 ms warm, while the ten largest organizations — 655 works at the top —
/// take 10 ms, `hasOrganization` being 62 k rows against `hasContainer`'s
/// millions.
fn works_counts(path: &PathBuf, pids: &[String], relation: &str) -> HashMap<String, usize> {
    let mut counts = HashMap::new();
    if pids.is_empty() {
        return counts;
    }
    let Ok(conn) = pool::get(path) else {
        return counts;
    };
    let placeholders = std::iter::repeat_n("?", pids.len())
        .collect::<Vec<_>>()
        .join(",");
    let sql = format!(
        "SELECT r.target_id, COUNT(*) FROM pid_relations r \
          WHERE r.type = ?1 AND r.target_id IN ({placeholders}) \
            AND NOT EXISTS (SELECT 1 FROM pid_relations v \
                             WHERE v.source_id = r.source_id AND v.type = 'IsVersionOf') \
          GROUP BY r.target_id"
    );
    let Ok(mut stmt) = conn.prepare(&sql) else {
        return counts;
    };
    let params = std::iter::once(relation).chain(pids.iter().map(String::as_str));
    let rows = stmt.query_map(rusqlite::params_from_iter(params), |row| {
        Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)? as usize))
    });
    if let Ok(rows) = rows {
        counts.extend(rows.flatten());
    }
    counts
}

/// Container hits as rows: the title, its other titles, the publisher, the
/// country and the other identifiers it is known by — as the container card
/// shows them.
///
/// A container is indexed under a surrogate pid the entity router cannot
/// resolve, so each hit is mapped to a navigable identifier — ISSN, then DOI,
/// then OpenAlex source id, then DataCite client id. One with none of those is
/// dropped: there is nothing to link to. The record supplies the rest, and its
/// title in particular: the index holds the title run together with every
/// alternate, so the hit's name read "Zenodo (CERN European Organization for
/// Nuclear Research) Research. Shared".
fn container_results(path: &PathBuf, hits: &[commonmeta::search::SearchHit]) -> Vec<SearchResult> {
    let pids: Vec<String> = hits.iter().map(|h| h.pid.clone()).collect();
    let navigable = container_navigable_ids(path, &pids);
    // One read per row on one pooled connection: commonmeta batches people and
    // organizations by pid but has no such reader for containers, and a page is
    // ten rows. Keyed by the navigable identifier, which is what the container
    // reader takes and what the row links by.
    let records: HashMap<String, commonmeta::metadata::ContainerRecord> = match pool::get(path) {
        Ok(conn) => navigable
            .values()
            .filter(|id| !id.is_empty())
            .filter_map(|id| {
                commonmeta::read_container_sqlite_conn(id, &conn)
                    .ok()
                    .flatten()
                    .map(|container| (id.clone(), container))
            })
            .collect(),
        Err(e) => {
            tracing::warn!(error = %e, "reading containers for search results failed");
            HashMap::new()
        }
    };
    let works = works_counts(path, &pids, "hasContainer");
    hits.iter()
        .filter_map(|hit| {
            let id = navigable.get(&hit.pid).filter(|s| !s.is_empty())?;
            let counted = works.get(&hit.pid).copied().unwrap_or(0);
            let Some(container) = records.get(id) else {
                return Some(SearchResult {
                    works: counted,
                    ..SearchResult::named(id.clone(), hit.name.clone(), "Container")
                });
            };
            let title = if container.title.trim().is_empty() {
                hit.name.clone()
            } else {
                container.title.trim().to_string()
            };
            let also_known_as: Vec<String> = container
                .additional_titles
                .iter()
                .map(|t| t.trim().to_string())
                .filter(|t| !t.is_empty() && !t.eq_ignore_ascii_case(&title))
                .collect();
            // Every identifier but the one the row already links by, so the
            // list says what else this container is known as.
            let identifiers: Vec<SearchIdentifier> = container
                .identifiers
                .iter()
                .filter(|i| !i.identifier.trim().is_empty())
                .filter(|i| bare_pid(&i.identifier) != bare_pid(id))
                .map(|i| SearchIdentifier {
                    identifier: i.identifier.trim().to_string(),
                    identifier_type: i.identifier_type.clone(),
                })
                .collect();
            let publisher = container.publisher.as_ref();
            Some(SearchResult {
                id: id.clone(),
                title,
                also_known_as,
                country: container.country.trim().to_string(),
                publisher: publisher
                    .map(|p| p.name.trim().to_string())
                    .unwrap_or_default(),
                publisher_id: publisher
                    .map(|p| p.id.trim().to_string())
                    .unwrap_or_default(),
                identifiers,
                container_type: container.type_.trim().to_string(),
                works: counted,
                ..SearchResult::named(String::new(), String::new(), "Container")
            })
        })
        .collect()
}

/// Person hits as rows: the display name, the given and family names on their
/// own, the names they are also known by, the country and the ORCID iD — read
/// from the store in one batch, as for organizations.
///
/// The index holds a person's names concatenated into one searchable field, so
/// the hit's name is the display name run together with the name parts and
/// every alternate. The record is what a row is built from; the hit's name
/// stands in only when the store cannot produce one.
fn person_results(path: &PathBuf, hits: &[commonmeta::search::SearchHit]) -> Vec<SearchResult> {
    let pids: Vec<String> = hits.iter().map(|h| h.pid.clone()).collect();
    let records: HashMap<String, commonmeta::Person> =
        match commonmeta::pidbox::read_sqlite_persons_by_pid(path, &pids) {
            Ok(found) => found.into_iter().collect(),
            Err(e) => {
                tracing::warn!(error = %e, "reading people for search results failed");
                HashMap::new()
            }
        };
    hits.iter()
        .map(|hit| {
            let Some(person) = records.get(&hit.pid) else {
                return SearchResult::named(hit.pid.clone(), hit.name.clone(), "Person");
            };
            // The name as the record gives it, else the parts joined — a record
            // may carry either.
            let joined = [person.given_name.trim(), person.family_name.trim()]
                .into_iter()
                .filter(|part| !part.is_empty())
                .collect::<Vec<_>>()
                .join(" ");
            let title = [person.name.trim(), joined.as_str(), hit.name.as_str()]
                .into_iter()
                .find(|name| !name.is_empty())
                .unwrap_or_default()
                .to_string();
            // An alternate that is only the name again says nothing.
            let also_known_as: Vec<String> = person
                .additional_names
                .iter()
                .map(|name| name.trim().to_string())
                .filter(|name| !name.is_empty() && !name.eq_ignore_ascii_case(&title))
                .collect();
            SearchResult {
                id: hit.pid.clone(),
                title,
                also_known_as,
                given_name: person.given_name.trim().to_string(),
                family_name: person.family_name.trim().to_string(),
                country: person.country.trim().to_string(),
                description: person.description.trim().to_string(),
                ..SearchResult::named(String::new(), String::new(), "Person")
            }
        })
        .collect()
}

/// The facet counts of everything a query matched, unnarrowed by the filters
/// in force — so a sidebar's numbers stay put as choices are picked.
///
/// Only works carry facets: a person has no publication year and an
/// organization no language, so another kind counts nothing and says so.
pub fn search_facets(path: &PathBuf, q: &str, kind: SearchKind) -> SearchFacets {
    let q = q.trim();
    // A container has nothing to narrow by: its works are its own page's
    // business, and its title and publisher are what the query already matched.
    //
    if q.is_empty()
        || kind == SearchKind::Containers
        || !commonmeta::search::index_dir(path).exists()
    {
        return SearchFacets::default();
    }
    // Every value, because which ones to keep is decided here: the index ranks
    // them by count, and the years are wanted by date. A query matches at most
    // a few hundred distinct values across the four facets, so asking for all
    // of them costs nothing.
    let counts = match commonmeta::search::search_facets(
        path,
        q,
        None,
        Some(kind.index_kind()),
        0,
        SEARCH_FACETS_MAX_MATCHES,
    ) {
        Ok(counts) => counts,
        Err(e) => {
            warn_index_unavailable(path, kind.index_kind(), &e);
            return SearchFacets::default();
        }
    };
    /// The commonest values, in the order the index ranked them.
    fn commonest(
        counts: Vec<commonmeta::search::FacetCount>,
        keep: usize,
    ) -> Vec<SearchFacetCount> {
        counts
            .into_iter()
            .take(keep)
            .map(|c| SearchFacetCount {
                value: c.value,
                count: c.count,
            })
            .collect()
    }
    SearchFacets {
        years: search_facets_years(counts.years, SEARCH_FACET_VALUES, this_year()),
        languages: commonest(counts.languages, SEARCH_FACET_VALUES),
        types: commonest(counts.types, SEARCH_FACET_VALUES),
        subjects: commonest(counts.subjects, SEARCH_FACET_VALUES),
        countries: commonest(counts.countries, SEARCH_FACET_VALUES),
        counted: counts.counted,
    }
}

/// 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) = pool::get(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) = pool::get(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_conn(id, &*pool::get(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 = pool::get(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 = pool::get(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 = pool::get(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 = pool::get(path)?;
    // A publication year past this one is a data error rather than news, and
    // with only 200 rows taken those records would crowd out real ones at the
    // head of the list. They sort last, as they do in the works lists. SQLite
    // supplies the year — in UTC, as `Utc::now()` does on the commonmeta side —
    // rather than this reaching for a date library to name it.
    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 (COALESCE("source_year", 0) > CAST(strftime('%Y', 'now') AS INTEGER)) ASC,
                    "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 pool::get(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 legacy
/// `organizations` table. Current stores keep organizations in `pid_records` and
/// no longer have it, so this returns an empty vec there, as it does when 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 = pool::get(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 in the legacy `organizations` table,
/// whose blob holds zstd-compressed ROR JSON (parsed with `commonmeta::read("ror",
/// ...)`). Current stores keep organizations in `pid_records` and no longer have
/// the table, so this returns `Ok(None)` there, as it does when 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 = pool::get(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 `pid_records`. A cache with nothing cached yet 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);
    }
    let Ok(conn) = pool::get(path) else {
        return Ok(None);
    };
    Ok(commonmeta::read_sqlite_by_id_conn(&id, &conn).unwrap_or(None))
}

/// A person's entity-page JSON from a store — the person alone, its works are
/// paged in separately — or `None` when the store does not hold them.
pub fn person_page(path: &Path, orcid: &str) -> Option<Vec<u8>> {
    commonmeta::read_person_commonmeta_sqlite_conn(orcid, &*pool::get(path).ok()?).ok()
}

/// An organization's entity-page JSON from a store, the record alone, without
/// the works it links; `None` when the store does not hold it.
pub fn org_record_page(path: &Path, ror: &str) -> Option<Vec<u8>> {
    commonmeta::read_org_record_commonmeta_sqlite_conn(ror, &*pool::get(path).ok()?).ok()
}

/// Every work linked to an ORCID person in a store, newest first.
pub fn orcid_works(path: &Path, orcid_url: &str) -> Vec<Work> {
    let Ok(conn) = pool::get(path) else {
        return Vec::new();
    };
    commonmeta::read_sqlite_by_orcid_conn(orcid_url, &conn).unwrap_or_default()
}

/// Every work linked to a ROR organization in a store, newest first.
pub fn ror_works(path: &Path, ror_url: &str) -> Vec<Work> {
    let Ok(conn) = pool::get(path) else {
        return Vec::new();
    };
    commonmeta::read_sqlite_by_ror_conn(ror_url, &conn).unwrap_or_default()
}

/// One page of the works citing a DOI, with their total. `(vec![], 0)` when the
/// store holds no citation of it.
pub fn citations_page(
    path: &PathBuf,
    doi: &str,
    page: commonmeta::WorksPage<'_>,
) -> (Vec<Work>, usize) {
    let Ok(conn) = pool::get(path) else {
        return Default::default();
    };
    commonmeta::read_sqlite_by_citation_page_conn(doi, &conn, page).unwrap_or_default()
}

/// Counts of the works citing a DOI by type and year. See
/// [`container_work_facets`].
pub fn citation_facets(
    path: &PathBuf,
    doi: &str,
    max_works: usize,
) -> Option<commonmeta::WorkFacets> {
    commonmeta::read_sqlite_citation_facets_conn(doi, &*pool::get(path).ok()?, max_works, true)
        .ok()
        .flatten()
}

/// Return the cached ORCID person's entity-page JSON — the person alone, its works
/// are paged in separately — from the raw ORCID record in the cache's
/// `pid_records` (source id 4), or `None` when the person isn't cached.
pub fn cache_lookup_person_page(path: &PathBuf, orcid_url: &str) -> Option<Vec<u8>> {
    person_page(path, orcid_url)
}

/// Return the cached ROR organization's entity-page JSON — the organization and
/// the works its `hasOrganization` relations link — from the cache's
/// `pid_records` (source id 3), or `None` when the organization isn't cached.
pub fn cache_lookup_org_page(path: &PathBuf, ror_url: &str) -> Option<Vec<u8>> {
    commonmeta::read_org_commonmeta_sqlite_conn(ror_url, &*pool::get(path).ok()?).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), which is what `build --from cache` copies into the
/// main store's `pid_records` and enriches.
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 rusqlite::Connection;
    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()
    }

    /// A store holding two ROR organizations, as source id 3 raw payloads.
    fn make_test_orgs_db(path: &Path) -> PathBuf {
        // ROR repeats the name among the aliases, and some records carry no
        // place at all — both of which the result row has to cope with.
        let humboldt = serde_json::json!({
            "id": "https://ror.org/01hcx6992",
            "names": [
                {"value": "Humboldt-Universität zu Berlin", "types": ["ror_display", "label"]},
                {"value": "HU", "types": ["acronym"]},
                {"value": "HU Berlin", "types": ["alias"]},
                {"value": "Humboldt University Berlin", "types": ["alias"]},
                // ROR repeats the display name among the aliases.
                {"value": "Humboldt-Universität zu Berlin", "types": ["alias"]},
            ],
            "locations": [{
                "geonames_id": 2950159,
                "geonames_details": {
                    "name": "Berlin", "country_name": "Germany", "country_code": "DE",
                    "country_subdivision_name": "State of Berlin",
                    "lat": 52.52437, "lng": 13.41053,
                },
            }],
            "types": ["education"],
        })
        .to_string();
        let placeless = serde_json::json!({
            "id": "https://ror.org/02nr0ka47",
            "names": [{"value": "Universität Münster", "types": ["ror_display"]}],
            // No coordinates, so no place: the country has to stand in.
            "locations": [{"geonames_details": {"country_code": "DE", "country_name": "Germany"}}],
            "types": ["education"],
        })
        .to_string();
        commonmeta::upsert_pid_records(
            &[
                ("01hcx6992".to_string(), 3, humboldt),
                ("02nr0ka47".to_string(), 3, placeless),
            ],
            path,
        )
        .expect("upsert organizations into pid_records");

        // Two works affiliated with the first organization and none with the
        // second — one of them a version, which the count leaves out as the
        // organization's own page does. `pid_relations` keys on bare pids.
        // A writable connection: the pool hands out read-only ones.
        let conn = Connection::open(path).expect("open store for writing");
        // The table as the real store keys it: bare pids, the three-column
        // primary key. `upsert_pid_records` does not create it.
        conn.execute_batch(
            r#"CREATE TABLE IF NOT EXISTS 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 (source, target, rel) in [
            ("10.1234/aff-1", "01hcx6992", "hasOrganization"),
            ("10.1234/aff-2", "01hcx6992", "hasOrganization"),
            ("10.1234/aff-2", "10.1234/aff-1", "IsVersionOf"),
        ] {
            conn.execute(
                r#"INSERT OR IGNORE INTO pid_relations (source_id, target_id, "type") VALUES (?1, ?2, ?3)"#,
                params![source, target, rel],
            )
            .expect("insert relation");
        }
        path.to_path_buf()
    }

    #[test]
    fn an_organization_row_names_it_properly_and_once() {
        // The index holds every name an organization has run together into one
        // searchable field, so the hit's name is no name at all. The row comes
        // from the record instead.
        let dir = tempfile::tempdir().unwrap();
        let path = make_test_orgs_db(&dir.path().join("orgs.sqlite3"));
        let hit = |pid: &str| commonmeta::search::SearchHit {
            pid: pid.to_string(),
            kind: "organization".to_string(),
            name: "Humboldt-Universität zu Berlin HU HU Berlin".to_string(),
            score: 1.0,
        };
        let rows = organization_results(&path, &[hit("01hcx6992"), hit("02nr0ka47")]);
        assert_eq!(rows.len(), 2);

        let humboldt = &rows[0];
        assert_eq!(humboldt.title, "Humboldt-Universität zu Berlin");
        assert_eq!(humboldt.entity_type, "Organization");
        // The acronym first, then the alternates — and the name is not repeated
        // among them, though the record repeats it.
        assert_eq!(
            humboldt.also_known_as,
            vec!["HU", "HU Berlin", "Humboldt University Berlin"]
        );
        assert!(
            humboldt.location.contains("Berlin"),
            "{}",
            humboldt.location
        );
        // The ROR types, as the card shows them: lowercased keys the UI looks
        // up a label for.
        assert_eq!(humboldt.types, vec!["education"]);

        // The works the store links to it, the version among them left out —
        // the number its own page shows.
        assert_eq!(humboldt.works, 1);

        // No place on the record, so the country stands in rather than nothing.
        let münster = &rows[1];
        assert_eq!(münster.title, "Universität Münster");
        assert!(münster.also_known_as.is_empty());
        assert_eq!(münster.location, "DE");
        // Nothing links to it, so the row says nothing rather than "0 works":
        // only 9% of a real store's organizations have any works edge, and
        // their pages fall back to a Crossref fetch.
        assert_eq!(münster.works, 0);

        // A pid the store does not hold keeps its hit: the ROR id is still
        // worth linking, and the indexed name is all there is to show.
        let rows = organization_results(&path, &[hit("09zzzzzz9")]);
        assert_eq!(rows.len(), 1);
        assert_eq!(rows[0].id, "09zzzzzz9");
        assert_eq!(rows[0].title, "Humboldt-Universität zu Berlin HU HU Berlin");
        assert!(rows[0].also_known_as.is_empty());
    }

    /// A store holding two ORCID people. Source id 4 holds the ORCID *XML*
    /// record, which is what the transport store carries and what
    /// `read_sqlite_persons_by_pid` converts.
    fn make_test_people_db(path: &Path) -> PathBuf {
        let record = |orcid: &str, names: &str, extras: &str| {
            format!(
                r#"<?xml version="1.0" encoding="UTF-8"?>
<record:record xmlns:record="http://www.orcid.org/ns/record"
               xmlns:common="http://www.orcid.org/ns/common"
               xmlns:person="http://www.orcid.org/ns/person"
               xmlns:other-name="http://www.orcid.org/ns/other-name"
               xmlns:address="http://www.orcid.org/ns/address"
               xmlns:personal-details="http://www.orcid.org/ns/personal-details">
  <common:orcid-identifier>
    <common:uri>https://orcid.org/{orcid}</common:uri>
    <common:path>{orcid}</common:path>
  </common:orcid-identifier>
  <person:person>
    <person:name>{names}</person:name>
    {extras}
  </person:person>
</record:record>"#
            )
        };
        let carberry = record(
            "0000-0002-1825-0097",
            "<personal-details:given-names>Josiah</personal-details:given-names>\
             <personal-details:family-name>Carberry</personal-details:family-name>\
             <personal-details:credit-name>Josiah S. Carberry</personal-details:credit-name>",
            // An alternate that is only the name again says nothing, so it is
            // left out of the row.
            r#"<other-name:other-names>
                 <other-name:other-name><other-name:content>J. S. Carberry</other-name:content></other-name:other-name>
                 <other-name:other-name><other-name:content>Josiah S. Carberry</other-name:content></other-name:other-name>
               </other-name:other-names>
               <address:addresses>
                 <address:address><address:country>US</address:country></address:address>
               </address:addresses>
               <person:biography visibility="public">
                 <personal-details:content>  Professor of Psychoceramics.  </personal-details:content>
               </person:biography>"#,
        );
        // No name parts, no country, no alternates: the row has the name and
        // the iD, and nothing else.
        let sparse = record(
            "0000-0003-0920-415X",
            "<personal-details:credit-name>A Sparse Record</personal-details:credit-name>",
            "",
        );
        commonmeta::upsert_pid_records(
            &[
                ("0000-0002-1825-0097".to_string(), 4, carberry),
                ("0000-0003-0920-415X".to_string(), 4, sparse),
            ],
            path,
        )
        .expect("upsert people into pid_records");
        path.to_path_buf()
    }

    #[test]
    fn a_container_row_carries_its_publisher_country_and_other_identifiers() {
        let dir = tempfile::tempdir().unwrap();
        let path = make_test_containers_db(&dir.path().join("c.sqlite3"));
        // A container is indexed under a surrogate pid; the row has to map it to
        // something the router resolves, and read the record for the rest.
        let surrogate = {
            let conn = pool::get(&path).unwrap();
            conn.query_row(
                "SELECT pid FROM pid_records WHERE source_id = 5 LIMIT 1",
                [],
                |row| row.get::<_, String>(0),
            )
            .expect("the fixture registered a container")
        };
        let rows = container_results(
            &path,
            &[commonmeta::search::SearchHit {
                pid: surrogate,
                kind: "container".to_string(),
                // What the index holds: the title and every alternate, run together.
                name: "Front Matter Front Matter Blog".to_string(),
                score: 1.0,
            }],
        );
        assert_eq!(rows.len(), 1);
        let row = &rows[0];
        // Linked by the ISSN, which the router resolves — not the surrogate.
        assert_eq!(row.id, "2749-9952");
        // The title is the record's, not the index's concatenation.
        assert_eq!(row.title, "Front Matter");
        assert_eq!(row.entity_type, "Container");
        assert_eq!(row.publisher, "Front Matter");
        assert_eq!(row.country, "DE");
        // Every identifier but the one the row links by, so the list says what
        // else the container is known as.
        let listed: Vec<(&str, &str)> = row
            .identifiers
            .iter()
            .map(|i| (i.identifier_type.as_str(), i.identifier.as_str()))
            .collect();
        assert!(
            listed.contains(&("OpenAlex", "https://openalex.org/S7407056926")),
            "{listed:?}"
        );
        assert!(
            listed.contains(&("DOI", "https://doi.org/10.53731/front_matter")),
            "{listed:?}"
        );
        assert!(
            !listed.iter().any(|(_, id)| *id == "2749-9952"),
            "the ISSN the row links by is not also listed: {listed:?}"
        );
        // Nothing a person's or an organization's row carries.
        assert!(row.given_name.is_empty() && row.types.is_empty());
        // The works it holds, which the fixture gives it one of, and the type
        // that decides whether they are called works or blog posts. The
        // fixture's publisher-type override makes this a Blog.
        assert_eq!(row.works, 1);
        assert_eq!(row.container_type, "Blog");
    }

    #[test]
    fn a_person_row_carries_the_name_its_parts_and_the_country() {
        let dir = tempfile::tempdir().unwrap();
        let path = make_test_people_db(&dir.path().join("people.sqlite3"));
        let hit = |pid: &str| commonmeta::search::SearchHit {
            pid: pid.to_string(),
            kind: "person".to_string(),
            // What the index holds: every name run together.
            name: "Josiah S. Carberry Josiah Carberry J. S. Carberry".to_string(),
            score: 1.0,
        };
        let rows = person_results(
            &path,
            &[hit("0000-0002-1825-0097"), hit("0000-0003-0920-415X")],
        );
        assert_eq!(rows.len(), 2);

        let carberry = &rows[0];
        // The header is the name, not the index's concatenation of them all.
        assert_eq!(carberry.title, "Josiah S. Carberry");
        assert_eq!(carberry.entity_type, "Person");
        assert_eq!(carberry.given_name, "Josiah");
        assert_eq!(carberry.family_name, "Carberry");
        assert_eq!(carberry.country, "US");
        assert_eq!(carberry.description, "Professor of Psychoceramics.");
        // The alternate that repeats the name is left out.
        assert_eq!(carberry.also_known_as, vec!["J. S. Carberry"]);
        // Nothing an organization's row carries.
        assert!(carberry.types.is_empty() && carberry.location.is_empty());

        let sparse = &rows[1];
        assert_eq!(sparse.title, "A Sparse Record");
        assert!(sparse.given_name.is_empty() && sparse.family_name.is_empty());
        assert!(sparse.country.is_empty() && sparse.also_known_as.is_empty());
        assert!(sparse.description.is_empty());

        // A pid the store does not hold keeps its hit, as for organizations.
        let rows = person_results(&path, &[hit("0000-0000-0000-0000")]);
        assert_eq!(rows.len(), 1);
        assert_eq!(rows[0].id, "0000-0000-0000-0000");
        assert!(rows[0].given_name.is_empty());
    }

    #[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 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"));
        let (hits, total) = search_page(
            &path,
            "   ",
            SearchKind::Works,
            &SearchFilters::default(),
            10,
            0,
        )
        .unwrap();
        assert!(hits.is_empty());
        assert_eq!(total, 0);
        // And nothing to narrow it by, rather than four empty facets.
        assert!(!search_facets(&path, "   ", SearchKind::Works).counted);
    }

    #[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());
        let (hits, total) = search_page(
            &path,
            "carberry",
            SearchKind::Works,
            &SearchFilters::default(),
            10,
            0,
        )
        .unwrap();
        assert!(hits.is_empty());
        assert_eq!(total, 0);
        assert!(!search_facets(&path, "carberry", SearchKind::Works).counted);
    }

    #[test]
    fn this_year_is_the_current_year() {
        // The arithmetic replaces a date library, so it is pinned: it must
        // agree with the clock, not merely look plausible.
        let year = this_year();
        assert!(
            (2026..=2100).contains(&year),
            "{year} is not a plausible current year"
        );
    }

    #[test]
    fn the_year_facet_reads_newest_first_with_future_years_last() {
        let facet = |value: &str, count: u64| commonmeta::search::FacetCount {
            value: value.to_string(),
            count,
        };
        // As the index ranks them: by count, so in no useful date order.
        let counts = vec![
            facet("2023", 303),
            facet("2026", 165),
            facet("2024", 71),
            facet("2025", 70),
            facet("2021", 45),
            facet("2022", 43),
            facet("2018", 32),
            facet("2019", 26),
            facet("2020", 36),
            facet("2017", 12),
            facet("2016", 8),
            // Deposit errors: dates that have not happened.
            facet("2029", 1),
            facet("2031", 2),
        ];
        let years = search_facets_years(counts, 10, 2026);
        let shown: Vec<&str> = years.iter().map(|c| c.value.as_str()).collect();
        // Newest first, and the two future years are not among the ten at all
        // — they sort behind every year that has happened.
        assert_eq!(
            shown,
            vec![
                "2026", "2025", "2024", "2023", "2022", "2021", "2020", "2019", "2018", "2017"
            ]
        );

        // With room to spare they appear, but still behind — and descending
        // among themselves, so the whole list reads newest first within each
        // group. That is the works list's rule: `(year > this year) ASC, year
        // DESC`.
        let years = search_facets_years(
            vec![facet("2031", 2), facet("2026", 1), facet("2029", 1)],
            10,
            2026,
        );
        let shown: Vec<&str> = years.iter().map(|c| c.value.as_str()).collect();
        assert_eq!(shown, vec!["2026", "2031", "2029"]);
    }

    #[test]
    fn only_works_carry_facets() {
        // A person has no publication year and an organization no language, so
        // another kind counts nothing and says so — the UI leaves the sidebar
        // out rather than drawing four empty facets.
        let dir = tempfile::tempdir().unwrap();
        let path = make_test_db(&dir.path().join("test.sqlite3"));
        for kind in [
            SearchKind::People,
            SearchKind::Organizations,
            SearchKind::Containers,
        ] {
            assert!(
                !search_facets(&path, "carberry", kind).counted,
                "{kind:?} should carry no facets"
            );
        }
    }

    #[test]
    fn a_search_page_is_clamped_to_a_sane_size() {
        // A hand-written request cannot ask the index for an unbounded page.
        let dir = tempfile::tempdir().unwrap();
        let path = make_test_db(&dir.path().join("test.sqlite3"));
        // No index, so this only exercises the clamp arithmetic, not the index.
        for limit in [0, 1, SEARCH_LIMIT, SEARCH_LIMIT_MAX, usize::MAX] {
            let page = search_page(
                &path,
                "carberry",
                SearchKind::Works,
                &SearchFilters::default(),
                limit,
                0,
            );
            assert!(page.is_ok(), "limit {limit} should not fail");
        }
    }

    #[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::named((*id).to_string(), "same title".to_string(), "Work"))
            .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 ──────────────────────────────────────────────────────

    /// Add `n` more `hasContainer` edges to the container carrying `issn`, their
    /// years cycling over 25. Facet counting reads only the relation rows, so the
    /// works behind these edges need not exist.
    pub(crate) fn add_container_edges(path: &std::path::Path, issn: &str, n: usize) -> String {
        let conn = rusqlite::Connection::open(path).expect("open");
        let cpid: String = conn
            .query_row(
                r#"SELECT pid FROM pid_annotations WHERE "issn" = ?1"#,
                rusqlite::params![issn],
                |r| r.get(0),
            )
            .expect("container pid");
        let tx = conn.unchecked_transaction().expect("tx");
        for i in 0..n {
            tx.execute(
                r#"INSERT OR IGNORE INTO pid_relations
                   ("source_id","target_id","type","source_year","source_type")
                   VALUES (?1, ?2, 'hasContainer', ?3, 'JournalArticle')"#,
                rusqlite::params![format!("10.5555/w{i}"), &cpid, 2000 + (i % 25) as i64],
            )
            .expect("edge");
        }
        tx.commit().expect("commit");
        cpid
    }

    /// Link works to a person with `hasPerson` rows, each with the year and type
    /// the works filters read. An empty type and a year of 0 are what a row
    /// carries when it was written without them.
    pub(crate) fn add_person_edges(
        path: &std::path::Path,
        orcid: &str,
        rows: &[(&str, i64, &str)],
    ) {
        let conn = rusqlite::Connection::open(path).expect("open");
        conn.execute_batch(
            r#"CREATE TABLE IF NOT EXISTS pid_relations (
                   source_id   TEXT    NOT NULL,
                   target_id   TEXT    NOT NULL,
                   "type"      TEXT    NOT NULL,
                   source_year INTEGER NOT NULL DEFAULT 0,
                   source_type TEXT    NOT NULL DEFAULT '',
                   PRIMARY KEY (source_id, target_id, "type")
               );"#,
        )
        .expect("pid_relations");
        for (doi, year, type_) in rows {
            conn.execute(
                r#"INSERT INTO pid_relations
                   ("source_id","target_id","type","source_year","source_type")
                   VALUES (?1, ?2, 'hasPerson', ?3, ?4)"#,
                rusqlite::params![doi, orcid, year, type_],
            )
            .expect("edge");
        }
    }

    /// Give `pid` a `pid_annotations` row carrying a PMID and an OpenAlex id.
    pub(crate) fn add_work_identifiers(
        path: &std::path::Path,
        pid: &str,
        pmid: &str,
        openalex: &str,
        source_id: i64,
    ) {
        let conn = rusqlite::Connection::open(path).expect("open");
        conn.execute_batch(
            r#"CREATE TABLE IF NOT EXISTS pid_annotations (
                   "pid" TEXT PRIMARY KEY NOT NULL, "valid" INTEGER NOT NULL DEFAULT 1,
                   "pmid" TEXT NOT NULL DEFAULT '', "openalex" TEXT NOT NULL DEFAULT '',
                   "source_id" INTEGER NOT NULL DEFAULT 0);"#,
        )
        .expect("pid_annotations");
        conn.execute(
            r#"INSERT INTO pid_annotations ("pid","pmid","openalex","source_id") VALUES (?1,?2,?3,?4)"#,
            rusqlite::params![pid, pmid, openalex, source_id],
        )
        .expect("annotation");
    }

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

    /// The containers fixture with a second container row carrying the same
    /// OpenAlex source id, and a work of its own — the shape the production
    /// store is in, where one journal was registered under several surrogate
    /// pids (from its ISSN, its DOI, an OpenAlex source id, a DataCite client
    /// id) and each work's `hasContainer` edge points at whichever pid existed
    /// when it was enriched.
    ///
    /// Written directly, because `write_container_sqlite` merges rows that share
    /// an identifier onto one survivor: the duplicates it leaves behind are the
    /// ones registered before that merge existed.
    pub(crate) fn make_test_duplicate_containers_db(path: &std::path::Path) -> PathBuf {
        let path = make_test_containers_db(path);

        // A second work, in pid_records so the junction can hydrate it.
        let raw = serde_json::json!({
            "DOI": "10.53731/article-2",
            "type": "journal-article",
            "title": ["A second blog post"],
            "container-title": ["Front Matter"],
        })
        .to_string();
        commonmeta::upsert_pid_records(&[("10.53731/article-2".to_string(), 1, raw)], &path)
            .expect("upsert second work");

        let conn = rusqlite::Connection::open(&path).expect("open fixture");
        conn.execute(
            r#"INSERT INTO pid_annotations ("pid","openalex","source_id")
               VALUES ('duplicate-container-pid', 'https://openalex.org/S7407056926', 5)"#,
            [],
        )
        .expect("insert duplicate container row");
        conn.execute(
            r#"INSERT INTO pid_relations ("source_id","target_id","type","source_year")
               VALUES ('10.53731/article-2', 'duplicate-container-pid', 'hasContainer', 2024)"#,
            [],
        )
        .expect("link the second work to the duplicate");
        path
    }

    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_carries_no_works() {
        // The fixture links one work to the container; it is served by
        // /api/container/works, not inline with the record.
        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();
        assert_eq!(arr.as_array().map(Vec::len), Some(1), "the container alone");
    }

    #[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 page = commonmeta::WorksPage::new(Some(50), 0);
        let (works, total) = container_works_local_page(&path, "10.53731/front_matter", page);
        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 page = commonmeta::WorksPage::new(Some(50), 0);
        let (works, total) = container_works_local_page(&path, "0000-0000", page);
        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()
            )
        );
    }
}