znippy-common 0.10.1

//! Typed specialized package views — the READ side of the plugin contract.
//!
//! A znippy plugin writes per-ecosystem metadata into Arrow columns at compress
//! time (maven: `group_id`/`artifact_id`/`version`; python: `name`/`version`;
//! rust: `crate_name`/`version`), discriminated by `pkg_type`. This module is the
//! symmetric READ side: it turns those columns back into a typed, coord-addressed
//! view so consumers never re-derive coordinates by parsing file paths.
//!
//! ## Performance contract (HARD)
//!
//! 1. **The coord index is built ONCE**, at view construction, via
//!    [`read_znippy_index_filtered`](crate::read_znippy_index_filtered) with an
//!    [`IndexFilter`](crate::IndexFilter) pinned to the plugin's `pkg_type` — so
//!    only that one sub-index stream is read and parsed. The result is a
//!    `HashMap<Key, FileLoc>` interned into the view; the holder ([`ZnippyArchive`])
//!    caches it (one slot per format), so repeated `as_maven()` is free.
//! 2. **`get(coords)` is an O(1) map lookup** returning a lightweight handle
//!    ([`Package`]): coords + a cloned `FileLoc` + the shared fd. No scan, no
//!    Arrow re-parse, no decompression, no bytes copied.
//! 3. **Bytes are LAZY** — [`Package::bytes`] is the only thing that preads the
//!    blob(s) and decompresses (the same loop as [`crate::ZnippyArchive::extract_file`]),
//!    and only when called. A `fetch` that only needs existence/size never decompresses.

use std::collections::{HashMap, HashSet};
use std::fs::File;
use std::hash::Hash;
use std::os::unix::fs::FileExt;
use std::path::Path;
use std::sync::Arc;

use anyhow::{anyhow, Result};
use arrow::array::{Array, StringArray};
use arrow::record_batch::RecordBatch;

use crate::codec;
use crate::index::{read_znippy_index_filtered, IndexFilter};
use crate::plugin::file_name;

// ─── pkg_type discriminants (single source of truth: each plugin's `type_id()`) ──
//
// These mirror the discriminant each plugin returns from `ArchiveTypePlugin::type_id()`:
//   CargoPlugin::type_id()        == 1   (znippy-common/src/plugins/native/cargo_native.rs)
//   NativePythonPlugin::type_id() == 2   (znippy-plugin-python)
//   NativeMavenPlugin::type_id()  == 3   (znippy-plugin-maven)
// They live here because `ZnippyArchive` (in this crate) builds the filtered index
// keyed on them, and the maven/python plugin crates depend on this crate (not the
// other way round), so the constant cannot live in those crates without a cycle.

/// `pkg_type` discriminant written by the rust/cargo plugin.
pub const RUST_PKG_TYPE: i8 = 1;
/// `pkg_type` discriminant written by the python plugin.
pub const PYTHON_PKG_TYPE: i8 = 2;
/// `pkg_type` discriminant written by the maven plugin.
pub const MAVEN_PKG_TYPE: i8 = 3;
/// `pkg_type` discriminant written by the npm plugin (`plugins::npm_native`).
pub const NPM_PKG_TYPE: i8 = 6;
/// `pkg_type` discriminant written by the gem plugin (`plugins::gem_native`).
pub const GEM_PKG_TYPE: i8 = 11;
/// `pkg_type` discriminant written by the rpm plugin (`plugins::rpm_native`).
pub const RPM_PKG_TYPE: i8 = 8;
/// `pkg_type` discriminant written by the deb plugin (`plugins::deb_native`).
pub const DEB_PKG_TYPE: i8 = 9;
/// `pkg_type` discriminant written by the conda plugin (`plugins::conda_native`).
pub const CONDA_PKG_TYPE: i8 = 14;

/// One file's chunk locations within the archive blob region — everything needed
/// to pread + decompress its bytes, with **no** path involved. Built from the base
/// index columns on the same rows that carried the coord match.
#[derive(Debug, Clone)]
pub struct FileLoc {
    /// The row key this file was read under: its `relative_path` column.
    ///
    /// The typed views index BY COORDINATE (name+version, group+artifact+version,
    /// …), which is what a client asks with — and they threw the path away while
    /// building that index, so a caller holding a `RustPackage` had no key to ask
    /// the lookup sub-index with. It was never absent, only discarded:
    /// `group_rows_by_file` keys its map by exactly this string.
    ///
    /// It is what a CHANGE TOKEN needs: `sign::file_digest_from_parts` binds the
    /// relative path, so without it the six backends that list through the typed
    /// views (maven, npm, gem, conda, python, build-thing) have no key to fold a
    /// token from.
    pub relative_path: String,
    /// The file's chunks, ordered by `fdata_offset` (concatenation order).
    pub chunks: Vec<ChunkRef>,
    /// Total uncompressed size across all chunks.
    pub uncompressed_size: u64,
}

/// One chunk's blob location.
#[derive(Debug, Clone, Copy)]
pub struct ChunkRef {
    pub blob_offset: u64,
    pub blob_size: u64,
    pub fdata_offset: u64,
    pub compressed: bool,
}

impl FileLoc {
    /// Read + decompress this file's bytes — the lone I/O of the read API.
    /// Reuses the exact pread/decompress loop of [`ZnippyArchive::extract_file`].
    fn read_bytes(&self, archive: &File) -> Result<Vec<u8>> {
        let mut result = Vec::with_capacity(self.uncompressed_size as usize);
        let mut blob = Vec::new();
        let mut decomp = Vec::new();
        for chunk in &self.chunks {
            blob.resize(chunk.blob_size as usize, 0);
            archive.read_exact_at(&mut blob, chunk.blob_offset)?;
            if chunk.compressed {
                codec::decompress_into(&blob, &mut decomp)?;
                result.extend_from_slice(&decomp);
            } else {
                result.extend_from_slice(&blob);
            }
        }
        Ok(result)
    }
}

/// Project the base location columns of a row into a [`ChunkRef`], appending to a
/// per-file [`FileLoc`] keyed by `relative_path` (so chunked files group correctly).
fn group_rows_by_file(batch: &RecordBatch) -> Result<HashMap<String, FileLoc>> {
    use arrow::array::{BooleanArray, UInt64Array};

    let col = |n: &str| {
        batch
            .column_by_name(n)
            .ok_or_else(|| anyhow!("index missing column {n}"))
    };
    let paths = col("relative_path")?
        .as_any()
        .downcast_ref::<StringArray>()
        .ok_or_else(|| anyhow!("relative_path not StringArray"))?;
    let compressed = col("compressed")?
        .as_any()
        .downcast_ref::<BooleanArray>()
        .ok_or_else(|| anyhow!("compressed not BooleanArray"))?;
    let sizes = col("uncompressed_size")?
        .as_any()
        .downcast_ref::<UInt64Array>()
        .ok_or_else(|| anyhow!("uncompressed_size not UInt64Array"))?;
    let blob_offset = col("blob_offset")?
        .as_any()
        .downcast_ref::<UInt64Array>()
        .ok_or_else(|| anyhow!("blob_offset not UInt64Array"))?;
    let blob_size = col("blob_size")?
        .as_any()
        .downcast_ref::<UInt64Array>()
        .ok_or_else(|| anyhow!("blob_size not UInt64Array"))?;
    let fdata = col("fdata_offset")?
        .as_any()
        .downcast_ref::<UInt64Array>()
        .ok_or_else(|| anyhow!("fdata_offset not UInt64Array"))?;

    let mut by_path: HashMap<String, FileLoc> = HashMap::new();
    for i in 0..batch.num_rows() {
        let path = paths.value(i);
        let entry = by_path.entry(path.to_string()).or_insert_with(|| FileLoc {
            relative_path: path.to_string(),
            chunks: Vec::new(),
            uncompressed_size: 0,
        });
        entry.uncompressed_size += sizes.value(i);
        entry.chunks.push(ChunkRef {
            blob_offset: blob_offset.value(i),
            blob_size: blob_size.value(i),
            fdata_offset: fdata.value(i),
            compressed: compressed.value(i),
        });
    }
    for f in by_path.values_mut() {
        f.chunks.sort_by_key(|c| c.fdata_offset);
    }
    Ok(by_path)
}



// ════════════════════════════════════════════════════════════════════════════
// The generic coord view
// ════════════════════════════════════════════════════════════════════════════

mod sealed {
    pub trait Sealed {}
}

/// How one format's sub-index maps to coords: which `pkg_type` it reads, which
/// columns, and how a row becomes a `(Key, Extra)`. One small impl per format;
/// everything else (filtered read, grouping by path, the O(1) map, the lazy
/// bytes) is [`CoordView`]'s and written once.
pub trait CoordSpec: sealed::Sealed + Send + Sync + 'static {
    /// The `pkg_type` sub-index this view reads.
    const PKG_TYPE: i8;
    /// This format's slot in the archive's view cache.
    const SLOT: usize;
    /// The columns a row is read from. The first [`Self::REQUIRED`] must exist
    /// in the batch (else the batch is skipped) and be non-null (else the row is).
    const COLS: &'static [&'static str];
    const REQUIRED: usize;
    /// The lookup key (the coords).
    type Key: Eq + Hash + Clone + Send + Sync;
    /// Per-file metadata riding beside the location.
    type Extra: Clone + Send + Sync;
    /// One row → its key and extra; `path` is the row's `relative_path`.
    fn row(c: &Cells<'_>, path: &str) -> (Self::Key, Self::Extra);
    /// Whether a later row replaces an earlier one with the same key (default: yes).
    fn replaces(_new: &Self::Extra) -> bool {
        true
    }
}

/// One row's cells, in [`CoordSpec::COLS`] order.
pub struct Cells<'a> {
    cols: &'a [Option<&'a StringArray>],
    i: usize,
}

impl Cells<'_> {
    /// Column `j`'s value, `None` when the column is absent or the cell null.
    pub fn opt(&self, j: usize) -> Option<String> {
        self.cols[j].filter(|a| !a.is_null(self.i)).map(|a| a.value(self.i).to_string())
    }
    /// Column `j`'s value, `""` when absent/null.
    pub fn get(&self, j: usize) -> String {
        self.opt(j).unwrap_or_default()
    }
    /// Column `j`'s value when present and non-empty.
    pub fn non_empty(&self, j: usize) -> Option<String> {
        self.opt(j).filter(|s| !s.is_empty())
    }
}

/// A typed, coord-addressed view over one format's sub-index. Built once;
/// lookups are O(1).
pub struct CoordView<S: CoordSpec> {
    archive: Arc<File>,
    coords: HashMap<S::Key, (FileLoc, S::Extra)>,
}

/// A handle to one file in a [`CoordView`]: authoritative coords (read from the
/// columns) + its location. Bytes are lazy — [`Package::bytes`].
pub struct Package<S: CoordSpec> {
    archive: Arc<File>,
    loc: FileLoc,
    key: S::Key,
    extra: S::Extra,
}

impl<S: CoordSpec> CoordView<S> {
    pub(crate) fn build(path: &Path, archive: Arc<File>) -> Result<Self> {
        let (_schema, batches) =
            read_znippy_index_filtered(path, &IndexFilter { pkg_type: Some(S::PKG_TYPE), repo: None })?;
        let mut coords: HashMap<S::Key, (FileLoc, S::Extra)> = HashMap::new();
        for batch in &batches {
            let cols: Vec<Option<&StringArray>> = S::COLS
                .iter()
                .map(|n| batch.column_by_name(n).and_then(|c| c.as_any().downcast_ref::<StringArray>()))
                .collect();
            if cols[..S::REQUIRED].iter().any(Option::is_none) {
                continue;
            }
            let paths = batch
                .column_by_name("relative_path")
                .and_then(|c| c.as_any().downcast_ref::<StringArray>())
                .ok_or_else(|| anyhow!("missing relative_path"))?;
            let locs = group_rows_by_file(batch)?;
            let mut seen = HashSet::new();
            for i in 0..batch.num_rows() {
                let p = paths.value(i);
                // The first row of a path carries its coords; later rows are chunks.
                if !seen.insert(p) || cols[..S::REQUIRED].iter().any(|c| c.is_some_and(|a| a.is_null(i))) {
                    continue;
                }
                let Some(loc) = locs.get(p) else { continue };
                let (key, extra) = S::row(&Cells { cols: &cols, i }, p);
                if S::replaces(&extra) || !coords.contains_key(&key) {
                    coords.insert(key, (loc.clone(), extra));
                }
            }
        }
        Ok(Self { archive, coords })
    }

    fn package(&self, key: &S::Key) -> Option<Package<S>> {
        let (loc, extra) = self.coords.get(key)?;
        Some(Package { archive: Arc::clone(&self.archive), loc: loc.clone(), key: key.clone(), extra: extra.clone() })
    }

    /// Every entry's key and extra (unordered).
    fn entries(&self) -> impl Iterator<Item = (&S::Key, &S::Extra)> {
        self.coords.iter().map(|(k, (_, e))| (k, e))
    }

    /// Number of files indexed.
    pub fn len(&self) -> usize {
        self.coords.len()
    }
    pub fn is_empty(&self) -> bool {
        self.coords.is_empty()
    }
}

impl<S: CoordSpec> Package<S> {
    /// This package's row key in the archive — its `relative_path` column. What
    /// lets a caller holding a handle ask the lookup sub-index, or fold a
    /// path-bound per-file digest, without listing the whole archive.
    pub fn relative_path(&self) -> &str {
        &self.loc.relative_path
    }
    /// Uncompressed size in bytes (no decompression).
    pub fn size(&self) -> u64 {
        self.loc.uncompressed_size
    }
    /// LAZY: pread + decompress the file's bytes. The only I/O of the read API.
    pub fn bytes(&self) -> Result<Vec<u8>> {
        self.loc.read_bytes(&self.archive)
    }
    pub fn into_bytes(self) -> Result<Vec<u8>> {
        self.loc.read_bytes(&self.archive)
    }
}

/// Declare a spec type: `spec!(RustSpec)`.
macro_rules! spec {
    ($t:ident) => {
        #[doc = concat!("The coord spec behind [`", stringify!($t), "`]'s view.")]
        pub struct $t;
        impl sealed::Sealed for $t {}
    };
}

fn s2(k: &(String, String)) -> (String, String) {
    k.clone()
}

// ─── RUST: (crate_name, version) ────────────────────────────────────────────

spec!(RustSpec);
impl CoordSpec for RustSpec {
    const PKG_TYPE: i8 = RUST_PKG_TYPE;
    const SLOT: usize = 0;
    const COLS: &'static [&'static str] = &["crate_name", "version"];
    const REQUIRED: usize = 2;
    type Key = (String, String);
    type Extra = ();
    fn row(c: &Cells<'_>, _: &str) -> (Self::Key, ()) {
        ((c.get(0), c.get(1)), ())
    }
}
/// Typed view over the rust (cargo) sub-index.
pub type RustView = CoordView<RustSpec>;
/// A handle to one crate.
pub type RustPackage = Package<RustSpec>;

impl RustView {
    /// O(1) lookup → handle. `None` if the crate is not in the archive.
    pub fn get(&self, name: &str, version: &str) -> Option<RustPackage> {
        self.package(&(name.to_string(), version.to_string()))
    }
    /// Authoritative `(name, version)` coords of every crate in the view.
    pub fn list(&self) -> Vec<(String, String)> {
        self.entries().map(|(k, _)| s2(k)).collect()
    }
}
impl RustPackage {
    /// Authoritative crate name (from the `crate_name` column, not a path parse).
    pub fn name(&self) -> &str {
        &self.key.0
    }
    pub fn version(&self) -> &str {
        &self.key.1
    }
}

// ─── MAVEN: (group, artifact, version, classifier?) ─────────────────────────

spec!(MavenSpec);
impl CoordSpec for MavenSpec {
    const PKG_TYPE: i8 = MAVEN_PKG_TYPE;
    const SLOT: usize = 1;
    const COLS: &'static [&'static str] = &["group_id", "artifact_id", "version", "classifier"];
    const REQUIRED: usize = 3;
    /// The classifier is part of the key so `-sources`/`-javadoc` resolve distinctly.
    type Key = (String, String, String, Option<String>);
    type Extra = ();
    fn row(c: &Cells<'_>, path: &str) -> (Self::Key, ()) {
        let (g, a, v) = (c.get(0), c.get(1), c.get(2));
        // Prefer the column; derive from the filename when absent/null (older archives).
        let classifier = c.non_empty(3).or_else(|| derive_classifier(file_name(path), &a, &v));
        ((g, a, v, classifier), ())
    }
}
/// Typed view over the maven sub-index.
pub type MavenView = CoordView<MavenSpec>;
/// A handle to one maven artifact.
pub type MavenPackage = Package<MavenSpec>;

impl MavenView {
    /// O(1) lookup of the primary artifact (no classifier) for a GAV.
    pub fn get(&self, group: &str, artifact: &str, version: &str) -> Option<MavenPackage> {
        self.get_classified(group, artifact, version, None)
    }
    /// O(1) lookup of a specific classifier (`Some("sources")`) or the primary (`None`).
    pub fn get_classified(
        &self,
        group: &str,
        artifact: &str,
        version: &str,
        classifier: Option<&str>,
    ) -> Option<MavenPackage> {
        self.package(&(group.into(), artifact.into(), version.into(), classifier.map(str::to_string)))
    }
    /// Authoritative coords of every artifact: `(group, artifact, version, classifier?)`.
    pub fn list(&self) -> Vec<(String, String, String, Option<String>)> {
        self.entries().map(|(k, _)| k.clone()).collect()
    }
}

/// Best-effort classifier recovery from a filename when the column is absent.
/// Maven filename: `{artifact}-{version}[-{classifier}].{ext}`.
fn derive_classifier(filename: &str, artifact: &str, version: &str) -> Option<String> {
    let stem = filename.rsplit_once('.').map(|(s, _)| s).unwrap_or(filename);
    let rest = stem.strip_prefix(&format!("{artifact}-{version}"))?.strip_prefix('-')?;
    (!rest.is_empty()).then(|| rest.to_string())
}

impl MavenPackage {
    pub fn group(&self) -> &str {
        &self.key.0
    }
    pub fn artifact(&self) -> &str {
        &self.key.1
    }
    pub fn version(&self) -> &str {
        &self.key.2
    }
    /// The classifier (`sources`, `javadoc`, …) or `None` for the primary artifact.
    pub fn classifier(&self) -> Option<&str> {
        self.key.3.as_deref()
    }
    /// `(group, artifact, version, classifier?)` — authoritative coords.
    pub fn coords(&self) -> (&str, &str, &str, Option<&str>) {
        (&self.key.0, &self.key.1, &self.key.2, self.key.3.as_deref())
    }
}

// ─── PYTHON: (name, version), wheel preferred over sdist ────────────────────

/// Wheel vs sdist discriminant.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PythonKind {
    Wheel,
    Sdist,
}

spec!(PythonSpec);
impl CoordSpec for PythonSpec {
    const PKG_TYPE: i8 = PYTHON_PKG_TYPE;
    const SLOT: usize = 2;
    const COLS: &'static [&'static str] = &["name", "version"];
    const REQUIRED: usize = 2;
    type Key = (String, String);
    type Extra = PythonKind;
    fn row(c: &Cells<'_>, path: &str) -> (Self::Key, PythonKind) {
        let kind = if file_name(path).ends_with(".whl") { PythonKind::Wheel } else { PythonKind::Sdist };
        ((c.get(0), c.get(1)), kind)
    }
    /// Prefer a wheel over an sdist when both share a `(name, version)`.
    fn replaces(new: &PythonKind) -> bool {
        *new == PythonKind::Wheel
    }
}
/// Typed view over the python sub-index.
pub type PythonView = CoordView<PythonSpec>;
/// A handle to one python distribution.
pub type PythonPackage = Package<PythonSpec>;

impl PythonView {
    /// O(1) lookup → handle. `None` if the distribution is not in the archive.
    pub fn get(&self, name: &str, version: &str) -> Option<PythonPackage> {
        self.package(&(name.to_string(), version.to_string()))
    }
    /// Authoritative `(name, version)` coords of every distribution.
    pub fn list(&self) -> Vec<(String, String)> {
        self.entries().map(|(k, _)| s2(k)).collect()
    }
}
impl PythonPackage {
    pub fn name(&self) -> &str {
        &self.key.0
    }
    pub fn version(&self) -> &str {
        &self.key.1
    }
    /// Wheel or sdist (derived from the matched row's filename).
    pub fn kind(&self) -> PythonKind {
        self.extra
    }
}

// ─── NPM: (name incl. @scope, version) ──────────────────────────────────────

spec!(NpmSpec);
impl CoordSpec for NpmSpec {
    const PKG_TYPE: i8 = NPM_PKG_TYPE;
    const SLOT: usize = 3;
    const COLS: &'static [&'static str] = &["name", "version"];
    const REQUIRED: usize = 2;
    /// `name` is the authoritative `package.json` name, including the `@scope/`
    /// prefix the tarball filename drops.
    type Key = (String, String);
    type Extra = ();
    fn row(c: &Cells<'_>, _: &str) -> (Self::Key, ()) {
        ((c.get(0), c.get(1)), ())
    }
}
/// Typed view over the npm sub-index.
pub type NpmView = CoordView<NpmSpec>;
/// A handle to one npm package tarball.
pub type NpmPackage = Package<NpmSpec>;

impl NpmView {
    /// O(1) lookup → handle (pass `@scope/pkg` for scoped packages).
    pub fn get(&self, name: &str, version: &str) -> Option<NpmPackage> {
        self.package(&(name.to_string(), version.to_string()))
    }
    /// Authoritative `(name, version)` coords of every package in the view.
    pub fn list(&self) -> Vec<(String, String)> {
        self.entries().map(|(k, _)| s2(k)).collect()
    }
}
impl NpmPackage {
    /// Authoritative package name (from the `name` column — includes `@scope/`).
    pub fn name(&self) -> &str {
        &self.key.0
    }
    pub fn version(&self) -> &str {
        &self.key.1
    }
}

// ─── GEM: (name, version, platform) ─────────────────────────────────────────

spec!(GemSpec);
impl CoordSpec for GemSpec {
    const PKG_TYPE: i8 = GEM_PKG_TYPE;
    const SLOT: usize = 4;
    const COLS: &'static [&'static str] = &["name", "version", "platform"];
    const REQUIRED: usize = 2;
    /// `platform` is part of the key so `foo-1.2.3-java.gem` resolves apart from
    /// the pure-ruby gem of the same version. All three come from `metadata.gz`.
    type Key = (String, String, String);
    type Extra = ();
    fn row(c: &Cells<'_>, _: &str) -> (Self::Key, ()) {
        ((c.get(0), c.get(1), c.non_empty(2).unwrap_or_else(|| "ruby".into())), ())
    }
}
/// Typed view over the gem sub-index.
pub type GemView = CoordView<GemSpec>;
/// A handle to one gem.
pub type GemPackage = Package<GemSpec>;

impl GemView {
    /// O(1) lookup of the `ruby`-platform gem for a `(name, version)`.
    pub fn get(&self, name: &str, version: &str) -> Option<GemPackage> {
        self.get_platform(name, version, "ruby")
    }
    /// O(1) lookup of a specific platform (`java`, `x86_64-linux`, …).
    pub fn get_platform(&self, name: &str, version: &str, platform: &str) -> Option<GemPackage> {
        self.package(&(name.into(), version.into(), platform.into()))
    }
    /// Authoritative `(name, version, platform)` coords of every gem in the view.
    pub fn list(&self) -> Vec<(String, String, String)> {
        self.entries().map(|(k, _)| k.clone()).collect()
    }
}
impl GemPackage {
    pub fn name(&self) -> &str {
        &self.key.0
    }
    pub fn version(&self) -> &str {
        &self.key.1
    }
    /// The gem platform (`ruby` default, e.g. `java` for a native gem).
    pub fn platform(&self) -> &str {
        &self.key.2
    }
}

// ─── CONDA: (name, version, build, subdir) ──────────────────────────────────

spec!(CondaSpec);
impl CoordSpec for CondaSpec {
    const PKG_TYPE: i8 = CONDA_PKG_TYPE;
    const SLOT: usize = 5;
    const COLS: &'static [&'static str] = &["name", "version", "build", "subdir"];
    const REQUIRED: usize = 2;
    /// `build` + `subdir` are part of the key so one `(name, version)` resolves
    /// distinctly across builds and platforms (all from `info/index.json`).
    type Key = (String, String, String, String);
    type Extra = ();
    fn row(c: &Cells<'_>, _: &str) -> (Self::Key, ()) {
        ((c.get(0), c.get(1), c.get(2), c.get(3)), ())
    }
}
/// Typed view over the conda sub-index.
pub type CondaView = CoordView<CondaSpec>;
/// A handle to one conda package.
pub type CondaPackage = Package<CondaSpec>;

impl CondaView {
    /// The first `(name, version)` match across any build/subdir (a scan).
    /// Use [`get_exact`](CondaView::get_exact) to pin the build + subdir.
    pub fn get(&self, name: &str, version: &str) -> Option<CondaPackage> {
        let key = self.coords.keys().find(|k| k.0 == name && k.1 == version)?.clone();
        self.package(&key)
    }
    /// O(1) lookup of an exact `(name, version, build, subdir)` coord.
    pub fn get_exact(&self, name: &str, version: &str, build: &str, subdir: &str) -> Option<CondaPackage> {
        self.package(&(name.into(), version.into(), build.into(), subdir.into()))
    }
    /// Authoritative `(name, version, build, subdir)` coords of every package.
    pub fn list(&self) -> Vec<(String, String, String, String)> {
        self.entries().map(|(k, _)| k.clone()).collect()
    }
}
impl CondaPackage {
    pub fn name(&self) -> &str {
        &self.key.0
    }
    pub fn version(&self) -> &str {
        &self.key.1
    }
    /// The build string (e.g. `py311h1234567_0`), from `info/index.json`.
    pub fn build(&self) -> &str {
        &self.key.2
    }
    /// The subdir/platform (e.g. `linux-64`), from `info/index.json`.
    pub fn subdir(&self) -> &str {
        &self.key.3
    }
}

// ─── RPM: (name, version, release, arch) + header metadata ──────────────────

/// One rpm's authoritative header metadata — NEVRA plus the rpm-md `primary.xml`
/// fields — the shape holger's `primary.xml` synthesis feeds off.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RpmMetaRow {
    pub name: String,
    pub version: String,
    pub release: String,
    pub arch: String,
    pub epoch: Option<String>,
    pub summary: Option<String>,
    pub license: Option<String>,
    pub url: Option<String>,
    pub vendor: Option<String>,
    pub sourcerpm: Option<String>,
    pub provides: Vec<String>,
    pub requires: Vec<String>,
}

spec!(RpmSpec);
impl CoordSpec for RpmSpec {
    const PKG_TYPE: i8 = RPM_PKG_TYPE;
    const SLOT: usize = 6;
    const COLS: &'static [&'static str] = &[
        "name", "version", "release", "arch", "epoch", "summary", "license", "url", "vendor",
        "sourcerpm", "provides", "requires",
    ];
    const REQUIRED: usize = 2;
    /// Epoch is NOT a lookup key (an rpm filename / dnf request carries none); it
    /// rides in the extra so the read side can surface it.
    type Key = (String, String, String, String);
    /// The full row; its NEVRA duplicates the key.
    type Extra = RpmMetaRow;
    fn row(c: &Cells<'_>, _: &str) -> (Self::Key, RpmMetaRow) {
        let lines = |j| split_lines(c.opt(j));
        let m = RpmMetaRow {
            name: c.get(0),
            version: c.get(1),
            release: c.get(2),
            arch: c.get(3),
            epoch: c.opt(4),
            summary: c.opt(5),
            license: c.opt(6),
            url: c.opt(7),
            vendor: c.opt(8),
            sourcerpm: c.opt(9),
            provides: lines(10),
            requires: lines(11),
        };
        ((m.name.clone(), m.version.clone(), m.release.clone(), m.arch.clone()), m)
    }
}
/// Typed view over the rpm sub-index (coords incl. `epoch` from the RPM header).
pub type RpmView = CoordView<RpmSpec>;
/// A handle to one rpm.
pub type RpmPackage = Package<RpmSpec>;

impl RpmView {
    /// O(1) lookup of the rpm for `(name, version, release, arch)`.
    pub fn get(&self, name: &str, version: &str, release: &str, arch: &str) -> Option<RpmPackage> {
        self.package(&(name.into(), version.into(), release.into(), arch.into()))
    }
    /// `(name, version, release, arch, epoch)` of every rpm — the NEVRA projection.
    pub fn list(&self) -> Vec<(String, String, String, String, Option<String>)> {
        self.entries().map(|(k, e)| (k.0.clone(), k.1.clone(), k.2.clone(), k.3.clone(), e.epoch.clone())).collect()
    }
    /// Every rpm's full header metadata — what holger's `primary.xml` synthesis emits.
    pub fn list_meta(&self) -> Vec<RpmMetaRow> {
        self.entries().map(|(_, e)| e.clone()).collect()
    }
}
impl RpmPackage {
    pub fn name(&self) -> &str {
        &self.key.0
    }
    pub fn version(&self) -> &str {
        &self.key.1
    }
    pub fn release(&self) -> &str {
        &self.key.2
    }
    pub fn arch(&self) -> &str {
        &self.key.3
    }
    /// The authoritative `Epoch` (from the header), or `None` when unset.
    pub fn epoch(&self) -> Option<&str> {
        self.extra.epoch.as_deref()
    }
}

/// Split a newline-joined column value (how the rpm plugin stores dependency
/// lists) back into its parts, dropping empties. `None`/`""` ⇒ `[]`.
fn split_lines(v: Option<String>) -> Vec<String> {
    v.map(|s| s.lines().filter(|l| !l.is_empty()).map(str::to_string).collect()).unwrap_or_default()
}

// ─── DEB: (name, version, arch) + the raw control stanza ────────────────────

spec!(DebSpec);
impl CoordSpec for DebSpec {
    const PKG_TYPE: i8 = DEB_PKG_TYPE;
    const SLOT: usize = 7;
    const COLS: &'static [&'static str] = &["name", "version", "arch", "control"];
    const REQUIRED: usize = 1;
    type Key = (String, String, String);
    /// The raw `control` stanza; `None` when the `.deb` was not parseable at
    /// ingest (coords then came from the filename).
    type Extra = Option<String>;
    fn row(c: &Cells<'_>, _: &str) -> (Self::Key, Option<String>) {
        ((c.get(0), c.get(1), c.get(2)), c.opt(3))
    }
}
/// Typed view over the deb sub-index.
pub type DebView = CoordView<DebSpec>;
/// A handle to one deb.
pub type DebPackage = Package<DebSpec>;

impl DebView {
    /// O(1) lookup of the deb for `(name, version, arch)`.
    pub fn get(&self, name: &str, version: &str, arch: &str) -> Option<DebPackage> {
        self.package(&(name.into(), version.into(), arch.into()))
    }
    /// `(name, version, arch, control)` of every deb — the APT `Packages` synthesis feeds off this.
    pub fn list(&self) -> Vec<(String, String, String, Option<String>)> {
        self.entries().map(|(k, e)| (k.0.clone(), k.1.clone(), k.2.clone(), e.clone())).collect()
    }
}
impl DebPackage {
    pub fn name(&self) -> &str {
        &self.key.0
    }
    pub fn version(&self) -> &str {
        &self.key.1
    }
    pub fn arch(&self) -> &str {
        &self.key.2
    }
    /// The raw `control` stanza, or `None` when the `.deb` wasn't parseable at ingest.
    pub fn control(&self) -> Option<&str> {
        self.extra.as_deref()
    }
}

/// Number of view-cache slots ([`CoordSpec::SLOT`] is below it).
pub(crate) const VIEW_SLOTS: usize = 8;