factbook 0.1.3

// Project:   factbook
// File:      src/table/mod.rs
// Purpose:   Arbitrary tabular sources: fetch, index, look up by key
// Language:  Rust
//
// License:   Apache-2.0
// Copyright: (c) 2026 HYPERI PTY LIMITED

//! A table fetched from wherever the operator names, keyed by whatever reaches
//! its rows.
//!
//! The geo half of this crate is opinionated on purpose: it knows what a city
//! database is and what an ASN database is, and it can only fetch the providers
//! that publish them. Plenty of useful reference data is neither -- an ASN-keyed
//! side table, a JSON list of relays, an internal CSV of customer prefixes --
//! and the only thing those have in common with a GeoLite2 build is how they
//! are acquired.
//!
//! So acquisition is shared and interpretation is stated. A [`TableSource`] is
//! config: a URL, a file name, an encoding, where the column names come from,
//! and which key reaches a row. Everything the geo download path does applies
//! to it unchanged -- the published digest is verified, a body that turns out
//! to be a login page is refused, a replacement a fraction of the size of the
//! copy on disk is refused, and a refusal leaves that copy exactly where it is.
//!
//! # Two surfaces, not one
//!
//! [`GeoIp`](crate::geoip::GeoIp) answers `lookup(ip)`, which a table keyed by
//! an autonomous system number cannot. [`Table`] is the generic surface: it
//! knows about columns and keys and nothing about geography.
//!
//! # A table is held in memory, or it is refused
//!
//! [`Table::ensure`] reads the source into memory and serves it from there. A
//! source whose rows and index outgrow `resident_max_bytes` while they are built
//! is abandoned -- what it had taken is released, and the call fails with
//! [`TableError::OverResidentCeiling`] rather than holding a table the
//! deployment said it did not want held.
//!
//! The ceiling is hit, never predicted. Whether a source fits is not knowable
//! from its size on disk: an encoding, a quoting style and a column count all
//! move the answer.
//!
//! Refusing is the whole of it. Writing an oversized source out as a MaxMind DB
//! and serving it from there is future work, tracked as issue #2.
//!
//! # One key, or several conditions
//!
//! [`Table::get`] and [`Table::all`] answer the enrichment question -- what is
//! filed under this key -- in a hash lookup and a hand-back. That is the common
//! case and the fast one.
//!
//! A host whose lookups arrive as condition objects asks something else: the
//! rows where country is `AU` and status is `active`. [`Table::find`] takes a
//! [`Query`] of those conditions, and an equality on the indexed column still
//! reaches the index rather than walking the table.
//!
//! # Example
//!
//! ```yaml
//! # Fetch a CSV and reach its rows by autonomous system number.
//! url: https://example.net/networks.csv
//! checksum_url: https://example.net/networks.csv.sha256
//! file: networks.csv
//! format: csv
//! index:
//!   column: asn
//! ```
//!
//! ```rust,no_run
//! use factbook::geoip::AutoDownloadConfig;
//! use factbook::table::{Index, Table, TableFormat, TableSource};
//!
//! # async fn run() -> Result<(), Box<dyn std::error::Error>> {
//! let source = TableSource::new(
//!     "https://example.net/networks.csv",
//!     "networks.csv",
//!     TableFormat::Csv { header: true },
//!     Index::Column("asn".to_string()),
//! );
//!
//! let table = Table::ensure(&source, &AutoDownloadConfig::default()).await?;
//! if let Some(row) = table.get("13335") {
//!     println!("{:?}", row.get("name"));
//! }
//! # Ok(())
//! # }
//! ```

mod config;
mod parse;
mod query;

use std::collections::HashMap;
use std::collections::hash_map::Entry;
use std::fmt;
use std::fs;
use std::hash::Hash;
use std::io::{self, BufRead, BufReader};
use std::net::IpAddr;
use std::ops::Range;
use std::path::{Path, PathBuf};
use std::slice;
use std::time::Duration;

use ipnet::IpNet;
use prefix_trie::joint::JointPrefixMap;
use tracing::{debug, warn};

use crate::geoip::AutoDownloadConfig;
use crate::geoip::DatabaseFormat;
use crate::geoip::download::fetch::{self, Archive, Credential, Transfer};
use crate::geoip::download::verify::Guard;
use crate::geoip::download::{self, Attempt, GeoIpDownloadError, SECS_PER_DAY, is_fresh};
use parse::Parsed;

pub use config::{Index, Schema, SourceArchive, TableFormat, TableSource};
pub(crate) use parse::probe;
pub use query::{Condition, Fields, Matched, Matches, Query};

/// Column names an [`Index::Ip`] source is likely to have written, lowercase.
const ADDRESS_COLUMNS: [&str; 5] = ["ip", "ip_address", "ipaddress", "address", "addr"];

/// Column names a CIDR-keyed source is likely to have written, lowercase.
const PREFIX_COLUMNS: [&str; 5] = ["prefix", "network", "cidr", "range", "subnet"];

/// Rows sampled when working out which column holds addresses.
const ADDRESS_SAMPLE: usize = 64;

/// No row is filed under this key.
const NO_ROWS: &[usize] = &[];

/// Bytes one new range costs the prefix index: its trie node, and the branch
/// node a binary trie adds at most once per leaf.
const PREFIX_ENTRY_BYTES: usize =
    2 * size_of::<(IpNet, Option<Vec<usize>>, Option<usize>, Option<usize>)>();

/// One cell of a row. `None` is a value the source did not supply.
type Cell = Option<Box<str>>;

/// Errors raised while provisioning or reading a table.
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum TableError {
    /// The file could not be fetched.
    #[error("fetching the table failed: {0}")]
    Fetch(#[from] GeoIpDownloadError),

    /// A filesystem operation failed.
    #[error("IO error: {0}")]
    Io(#[from] io::Error),

    /// The source names somewhere the transfer cannot go.
    #[error("{url} is not a URL that can be fetched: {detail}")]
    BadUrl {
        /// URL as the operator wrote it, minus its userinfo and query values.
        url: String,
        /// What the URL parser refused it for.
        detail: String,
    },

    /// The file name would write outside the data directory.
    #[error("file {file} is not a plain file name")]
    UnsafeFileName {
        /// Name as the operator wrote it.
        file: String,
    },

    /// The file is not the shape its format states.
    #[error("the table is malformed at line {line}: {detail}")]
    Malformed {
        /// Line the record that was refused started on.
        line: usize,
        /// What the reader refused it for.
        detail: String,
    },

    /// The document is not a JSON array of objects.
    #[error("the source is not a JSON array of objects: {detail}")]
    NotAnArrayOfObjects {
        /// What the reader refused it for.
        detail: String,
    },

    /// A CSV with no header row supplies no column names.
    #[error("a CSV with no header row needs schema.named to supply the column names")]
    NamesRequired,

    /// The stated schema names no columns.
    #[error("schema.named names no columns")]
    NoNames,

    /// The file holds no rows.
    #[error("the source holds no rows")]
    Empty,

    /// The index names a column the source does not have.
    #[error("the index names column {column}, which is not one of: {columns}")]
    UnknownColumn {
        /// Column the index asked for.
        column: String,
        /// Columns the source actually has.
        columns: String,
    },

    /// No column holds IP addresses.
    #[error("no column holds IP addresses; name one with index.column instead. columns: {columns}")]
    NoAddressColumn {
        /// Columns the source actually has.
        columns: String,
    },

    /// No column holds CIDR ranges.
    #[error("no column holds CIDR ranges; name one with index.column instead. columns: {columns}")]
    NoPrefixColumn {
        /// Columns the source actually has.
        columns: String,
    },

    /// The rows, or the rows and their index, outgrew the memory the source
    /// was given.
    ///
    /// The read stops here rather than finishing: a source too large for
    /// `resident_max_bytes` is refused, and there is nowhere else for it to go.
    /// Writing one out as a database and serving it from there is future work.
    #[error(
        "the table outgrew the {ceiling}-byte resident ceiling; raise \
         auto_download.resident_max_bytes or point the source at less data"
    )]
    OverResidentCeiling {
        /// Ceiling the read was given, in bytes.
        ceiling: u64,
    },
}

/// Where a table's rows are held.
///
/// One backing today. A source that will not fit in memory is refused rather
/// than moved anywhere else, so this reports rather than chooses -- and it is
/// what a metrics recorder reads to see that a table loaded at all.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
#[non_exhaustive]
pub enum TableBacking {
    /// Read into memory when the table was loaded.
    Resident,
}

impl TableBacking {
    /// Label used in metric series and log fields.
    #[must_use]
    pub const fn label(self) -> &'static str {
        match self {
            Self::Resident => "resident",
        }
    }
}

/// How the rows of a table are reached.
///
/// The address and prefix kinds carry the rows their index could not file --
/// a key cell that is there and does not parse. Those rows are unreachable by
/// address and still reachable by the text they hold, so a
/// [`Query`] narrowed by the index walks them as well.
#[derive(Debug)]
enum Keys {
    /// Filed under the text of a column.
    Text(HashMap<Box<str>, Vec<usize>>),

    /// Filed under a parsed address.
    Address {
        /// Rows, by the address their key cell parsed to.
        filed: HashMap<IpAddr, Vec<usize>>,

        /// Rows whose key cell is there and is not an address.
        unindexed: Vec<usize>,
    },

    /// Filed under a CIDR range, reached by the longest one containing an
    /// address.
    Prefix {
        /// Rows, by the range their key cell parsed to.
        filed: Box<JointPrefixMap<IpNet, Vec<usize>>>,

        /// Rows whose key cell is there and is not a range.
        unindexed: Vec<usize>,
    },
}

/// A tabular source, in memory and indexed.
///
/// Rows are kept whole and in file order; the index is a second view of them,
/// so a key that appears twice keeps both rows rather than one silently
/// replacing the other.
///
/// ```
/// use factbook::table::{Index, Schema, Table, TableFormat};
///
/// let csv = "asn,name\n13335,CLOUDFLARENET\n15169,GOOGLE\n";
/// let table = Table::from_reader(
///     csv.as_bytes(),
///     TableFormat::Csv { header: true },
///     &Schema::Auto,
///     &Index::Column("asn".to_string()),
/// )?;
///
/// assert_eq!(table.len(), 2);
/// assert_eq!(table.get("13335").unwrap().get("name"), Some("CLOUDFLARENET"));
/// assert!(table.get("64496").is_none());
/// # Ok::<(), factbook::table::TableError>(())
/// ```
#[derive(Debug)]
pub struct Table {
    /// Column names, in the order a row's cells are stored.
    columns: Vec<String>,

    /// Column the index is built on.
    key_column: usize,

    /// Every row read, in file order.
    rows: Vec<Vec<Cell>>,

    /// Where a key leads.
    keys: Keys,
}

impl Table {
    /// Ensure the source is on disk, downloading when the local copy is missing
    /// or stale, then read it.
    ///
    /// Downloading is non-fatal by the same contract as the geo path: a
    /// transfer that fails over a copy already on disk logs a warning and reads
    /// that copy, because a stale table answers most lookups and no table
    /// answers none. A refusal or a rate limit starts the same cool-down the geo
    /// path keeps, and the copy on disk serves until it passes.
    ///
    /// Reading is not. The rows and their index are held to
    /// `auto.resident_max_bytes`, and a source that outgrows it is abandoned
    /// mid-read or mid-index and refused -- there is
    /// nowhere else for a table to be served from, so a ceiling it crosses is
    /// the end of the load rather than a switch to something slower.
    ///
    /// The read and the index build run on the blocking pool, so a large
    /// source holds no runtime worker while it loads.
    ///
    /// # Errors
    ///
    /// [`TableError::Fetch`] when the download fails or is cooling down and
    /// there is no local copy to fall back to,
    /// [`TableError::OverResidentCeiling`] when the source will not fit, or any
    /// of the read errors of [`Table::load`].
    pub async fn ensure(
        source: &TableSource,
        auto: &AutoDownloadConfig,
    ) -> Result<Self, TableError> {
        source.validate()?;

        let dest = auto.data_dir.join(&source.file);

        if !auto.enabled {
            debug!(file = %dest.display(), "auto-download is off, reading what is on disk");
        } else if is_fresh(&dest, u64::from(auto.max_age_days) * SECS_PER_DAY) {
            debug!(file = %dest.display(), "table is fresh");
        } else {
            download::check_settings(auto, source.http_client.is_some())?;
            let client = fetch::client(
                source.http_client.as_ref(),
                Duration::from_secs(auto.connect_timeout_secs),
                Duration::from_secs(auto.read_timeout_secs),
            )?;

            let transfer = transfer(source, dest.clone());
            let guard = Guard::for_table(source.format, &source.schema, &source.index, auto);

            let asked = download::fingerprint(std::slice::from_ref(&transfer));
            let fetched = download::attempt("table", Some(&dest), asked, || {
                transfer.run_guarded(&client, guard)
            })
            .await;
            match fetched {
                Attempt::Ran(Ok(_)) => {}
                Attempt::Ran(Err(e)) if dest.exists() => warn!(
                    file = %dest.display(),
                    error = %e,
                    "table download failed, reading the copy already on disk"
                ),
                Attempt::Ran(Err(e)) => return Err(e.into()),
                Attempt::Waiting(_) if dest.exists() => {}
                Attempt::Waiting(wait) => {
                    return Err(TableError::Fetch(GeoIpDownloadError::CoolingDown {
                        path: dest.display().to_string(),
                        wait_secs: wait.as_secs(),
                    }));
                }
            }
        }

        let (format, schema, index) = (source.format, source.schema.clone(), source.index.clone());
        let ceiling = auto.resident_max_bytes;
        let table = tokio::task::spawn_blocking(move || {
            Self::read_file(&dest, format, &schema, &index, ceiling)
        })
        .await
        .map_err(io::Error::from)??;

        telemetry::backing(&source.file, table.backing());
        Ok(table)
    }

    /// Read and index the file at `path`, refusing past `ceiling` bytes.
    /// Blocking: it reads, parses and indexes the whole file.
    pub(crate) fn read_file(
        path: &Path,
        format: TableFormat,
        schema: &Schema,
        index: &Index,
        ceiling: u64,
    ) -> Result<Self, TableError> {
        #[cfg(test)]
        crate::geoip::download::testkit::checkpoint(path);

        let parsed = parse::read(
            BufReader::new(fs::File::open(path)?),
            format,
            schema,
            ceiling,
        )?;
        Self::build(parsed, index, ceiling)
    }

    /// Read a table off disk.
    ///
    /// Held in memory whatever its size. The ceiling belongs to
    /// [`Table::ensure`], which is the call that reads the setting.
    ///
    /// # Errors
    ///
    /// [`TableError::Io`] when the file cannot be read, and the reader's own
    /// errors -- [`TableError::Malformed`], [`TableError::NotAnArrayOfObjects`],
    /// [`TableError::NamesRequired`], [`TableError::Empty`] -- when its contents
    /// are not the table the source states.
    pub fn load(
        path: &Path,
        format: TableFormat,
        schema: &Schema,
        index: &Index,
    ) -> Result<Self, TableError> {
        Self::read_file(path, format, schema, index, u64::MAX)
    }

    /// Read a table from bytes already in hand.
    ///
    /// # Errors
    ///
    /// The errors of [`Table::load`], other than the ones that come from
    /// opening a file.
    pub fn from_reader(
        reader: impl BufRead,
        format: TableFormat,
        schema: &Schema,
        index: &Index,
    ) -> Result<Self, TableError> {
        Self::build(
            parse::read(reader, format, schema, u64::MAX)?,
            index,
            u64::MAX,
        )
    }

    /// Column names, in the order a row's cells are stored.
    #[must_use]
    pub fn columns(&self) -> &[String] {
        &self.columns
    }

    /// Column the rows are indexed by.
    ///
    /// Worth reporting because [`Index::Ip`] locates its own column, so this is
    /// how a deployment sees which one it settled on.
    #[must_use]
    pub fn key_column(&self) -> &str {
        &self.columns[self.key_column]
    }

    /// Whether rows are reached by an address or by the text of a column.
    ///
    /// A lookup that does not match answers nothing rather than failing, so a
    /// source whose `index` was edited from a column to a prefix builds, reports
    /// its full [`len`](Self::len), and returns `None` from every
    /// [`get`](Self::get). This is how a caller tells the two apart.
    #[must_use]
    pub const fn keyed_by_address(&self) -> bool {
        matches!(self.keys, Keys::Address { .. } | Keys::Prefix { .. })
    }

    /// Where the rows are held.
    ///
    /// One answer today, and reported rather than configured. The same answer
    /// reaches a metrics recorder as `enrichment_table_backing`.
    #[must_use]
    pub const fn backing(&self) -> TableBacking {
        TableBacking::Resident
    }

    /// How many rows the table holds.
    #[must_use]
    pub fn len(&self) -> usize {
        self.rows.len()
    }

    /// Whether the table holds no rows.
    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.rows.is_empty()
    }

    /// Every row, in file order.
    #[must_use]
    pub fn rows(&self) -> Rows<'_> {
        Rows {
            table: self,
            positions: Positions::All(0..self.rows.len()),
        }
    }

    /// The first row filed under a text key.
    ///
    /// Answers nothing on a table keyed by address, which
    /// [`keyed_by_address`](Self::keyed_by_address) reports.
    #[must_use]
    pub fn get(&self, key: &str) -> Option<Row<'_>> {
        self.positions_of(key).first().map(|&at| self.row(at))
    }

    /// Every row filed under a text key.
    #[must_use]
    pub fn all(&self, key: &str) -> Rows<'_> {
        Rows {
            table: self,
            positions: Positions::Listed(self.positions_of(key).iter()),
        }
    }

    /// The first row filed under an address.
    ///
    /// Answers nothing on a table keyed by a column, which
    /// [`keyed_by_address`](Self::keyed_by_address) reports.
    #[must_use]
    pub fn get_by_address(&self, address: IpAddr) -> Option<Row<'_>> {
        self.positions_at(address).first().map(|&at| self.row(at))
    }

    /// Every row filed under an address.
    #[must_use]
    pub fn all_by_address(&self, address: IpAddr) -> Rows<'_> {
        Rows {
            table: self,
            positions: Positions::Listed(self.positions_at(address).iter()),
        }
    }

    /// Index parsed rows by the key the source states, refusing once the rows,
    /// the column names and the index together cost more than `ceiling` bytes.
    fn build(parsed: Parsed, index: &Index, ceiling: u64) -> Result<Self, TableError> {
        let Parsed {
            columns,
            mut rows,
            held,
        } = parsed;

        // The parse charged one slot per row, not the capacity the vector
        // doubled into.
        rows.shrink_to_fit();
        let mut budget = Budget { held, ceiling };
        for column in &columns {
            budget.charge(column.len() + size_of::<String>())?;
        }

        let key_column = match index {
            Index::Column(name) => columns
                .iter()
                .position(|column| column == name)
                .ok_or_else(|| TableError::UnknownColumn {
                    column: name.clone(),
                    columns: columns.join(", "),
                })?,
            Index::Ip => keyed_column(&columns, &rows, &ADDRESS_COLUMNS, |text| {
                as_address(text).is_some()
            })
            .ok_or_else(|| TableError::NoAddressColumn {
                columns: columns.join(", "),
            })?,
            Index::Prefix => keyed_column(&columns, &rows, &PREFIX_COLUMNS, |text| {
                as_prefix(text).is_some()
            })
            .ok_or_else(|| TableError::NoPrefixColumn {
                columns: columns.join(", "),
            })?,
        };

        let mut reachable = 0;
        let keys = match index {
            Index::Column(_) => {
                let mut filed: HashMap<Box<str>, Vec<usize>> = HashMap::new();
                for (at, row) in rows.iter().enumerate() {
                    if let Some(key) = cell(row, key_column) {
                        budget.file(&mut filed, Box::from(key), key.len(), at)?;
                        reachable += 1;
                    }
                }
                Keys::Text(filed)
            }
            Index::Ip => {
                let mut filed: HashMap<IpAddr, Vec<usize>> = HashMap::new();
                let mut unindexed = Vec::new();
                for (at, row) in rows.iter().enumerate() {
                    let Some(text) = cell(row, key_column) else {
                        continue;
                    };
                    if let Some(address) = as_address(text) {
                        budget.file(&mut filed, address, 0, at)?;
                        reachable += 1;
                    } else {
                        budget.push(&mut unindexed, at)?;
                    }
                }
                Keys::Address { filed, unindexed }
            }
            Index::Prefix => {
                let mut filed: JointPrefixMap<IpNet, Vec<usize>> = JointPrefixMap::default();
                let mut unindexed = Vec::new();
                for (at, row) in rows.iter().enumerate() {
                    let Some(text) = cell(row, key_column) else {
                        continue;
                    };
                    if let Some(prefix) = as_prefix(text) {
                        // A repeated range keeps every row, the same as a
                        // repeated text key does.
                        if let Some(list) = filed.get_mut(&prefix) {
                            budget.push(list, at)?;
                        } else {
                            budget.charge(PREFIX_ENTRY_BYTES)?;
                            let mut list = Vec::new();
                            budget.push(&mut list, at)?;
                            filed.insert(prefix, list);
                        }
                        reachable += 1;
                    } else {
                        budget.push(&mut unindexed, at)?;
                    }
                }
                Keys::Prefix {
                    filed: Box::new(filed),
                    unindexed,
                }
            }
        };

        // Rows whose key is missing or unparseable are kept and unreachable,
        // which is worth saying once rather than losing silently.
        if reachable < rows.len() {
            debug!(
                column = columns[key_column].as_str(),
                rows = rows.len(),
                unreachable = rows.len() - reachable,
                "some table rows carry no key"
            );
        }

        Ok(Self {
            columns,
            key_column,
            rows,
            keys,
        })
    }

    /// Where a text key leads.
    fn positions_of(&self, key: &str) -> &[usize] {
        match &self.keys {
            Keys::Text(filed) => filed.get(key).map_or(NO_ROWS, Vec::as_slice),
            Keys::Address { .. } | Keys::Prefix { .. } => NO_ROWS,
        }
    }

    /// Where an address leads.
    ///
    /// An exact index answers the address itself; a prefix index answers the
    /// most specific range that contains it, so a `/24` wins over the `/8`
    /// around it.
    fn positions_at(&self, address: IpAddr) -> &[usize] {
        match &self.keys {
            Keys::Address { filed, .. } => filed.get(&address).map_or(NO_ROWS, Vec::as_slice),
            Keys::Prefix { filed, .. } => filed
                .get_lpm(&IpNet::from(address))
                .map_or(NO_ROWS, |(_, rows)| rows.as_slice()),
            Keys::Text(_) => NO_ROWS,
        }
    }

    /// The row at a position.
    fn row(&self, at: usize) -> Row<'_> {
        Row {
            table: self,
            cells: &self.rows[at],
        }
    }
}

impl TableSource {
    /// Check the source can be acted on, before anything is downloaded.
    ///
    /// This is the config-load check: a URL nothing can fetch, a file name that
    /// would write outside the data directory, or a headerless CSV that names
    /// no columns is reported here rather than part way through a transfer.
    ///
    /// # Errors
    ///
    /// [`TableError::BadUrl`], [`TableError::UnsafeFileName`],
    /// [`TableError::NamesRequired`] or [`TableError::NoNames`], naming the
    /// field to fix.
    pub fn validate(&self) -> Result<(), TableError> {
        reqwest::Url::parse(&self.url).map_err(|e| TableError::BadUrl {
            url: fetch::loggable(&self.url),
            detail: e.to_string(),
        })?;

        // A source states a file name, never a path: it is written into a
        // directory the operator chose, and it may not climb out of it.
        if Path::new(&self.file).file_name() != Some(self.file.as_ref()) {
            return Err(TableError::UnsafeFileName {
                file: self.file.clone(),
            });
        }

        match (&self.format, &self.schema) {
            (TableFormat::Csv { header: false }, Schema::Auto) => Err(TableError::NamesRequired),
            (_, Schema::Named(named)) if named.is_empty() => Err(TableError::NoNames),
            _ => Ok(()),
        }
    }
}

/// One row of a table, borrowed from it.
#[derive(Clone, Copy)]
pub struct Row<'a> {
    /// Table the row belongs to, which is what names its cells.
    table: &'a Table,

    /// Cells of this row, as wide as the table's columns.
    cells: &'a [Cell],
}

impl<'a> Row<'a> {
    /// The value under a column name.
    #[must_use]
    pub fn get(&self, column: &str) -> Option<&'a str> {
        self.at(self.table.position_of(column)?)
    }

    /// The value at a column position.
    #[must_use]
    pub fn at(&self, column: usize) -> Option<&'a str> {
        cell(self.cells, column)
    }

    /// The value this row is indexed by.
    #[must_use]
    pub fn key(&self) -> Option<&'a str> {
        self.at(self.table.key_column)
    }

    /// Column names, in the order the cells are stored.
    #[must_use]
    pub fn columns(&self) -> &'a [String] {
        &self.table.columns
    }
}

impl fmt::Debug for Row<'_> {
    /// Prints the row's own cells rather than the table behind it, which every
    /// row would otherwise carry a copy of.
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        let mut map = f.debug_map();
        for (column, value) in self.table.columns.iter().zip(self.cells) {
            map.entry(column, &value.as_deref());
        }
        map.finish()
    }
}

/// Which rows a [`Rows`] walks.
#[derive(Debug)]
enum Positions<'a> {
    /// The rows filed under one key.
    Listed(slice::Iter<'a, usize>),

    /// Every row, in file order.
    All(Range<usize>),
}

/// Rows of a table, in file order.
#[derive(Debug)]
pub struct Rows<'a> {
    /// Table the rows belong to.
    table: &'a Table,

    /// Positions still to walk.
    positions: Positions<'a>,
}

impl<'a> Iterator for Rows<'a> {
    type Item = Row<'a>;

    fn next(&mut self) -> Option<Self::Item> {
        let at = match &mut self.positions {
            Positions::Listed(listed) => *listed.next()?,
            Positions::All(all) => all.next()?,
        };
        Some(self.table.row(at))
    }

    fn size_hint(&self) -> (usize, Option<usize>) {
        match &self.positions {
            Positions::Listed(listed) => listed.size_hint(),
            Positions::All(all) => all.size_hint(),
        }
    }
}

impl ExactSizeIterator for Rows<'_> {}

/// The transfer that fetches one table source.
fn transfer(source: &TableSource, dest: PathBuf) -> Transfer {
    Transfer {
        url: source.url.clone(),
        // A user-supplied source states one URL: there is no publication month
        // to fall back behind.
        fallback_url: None,
        checksum_url: source.checksum_url.clone(),
        dest,
        archive: match source.archive {
            SourceArchive::Raw => Archive::Raw,
            SourceArchive::Gzip => Archive::Gzip,
        },
        format: match source.format {
            TableFormat::Csv { .. } => DatabaseFormat::Csv,
            TableFormat::Json => DatabaseFormat::Json,
        },
        credential: Credential::None,
    }
}

/// The value at a position, absent when the source did not supply one.
fn cell(cells: &[Cell], at: usize) -> Option<&str> {
    cells.get(at).and_then(Option::as_deref)
}

/// A cell as an address, when it is one.
fn as_address(text: &str) -> Option<IpAddr> {
    text.parse().ok()
}

/// A cell as a CIDR range, when it is one.
///
/// A bare address is accepted as the range holding only itself, because
/// publishers mix single hosts into a prefix list.
fn as_prefix(text: &str) -> Option<IpNet> {
    text.parse()
        .ok()
        .or_else(|| as_address(text).map(IpNet::from))
}

/// Which column a source is keyed by.
///
/// A conventional name is preferred, and a source that uses none is sampled
/// instead, because the field is named differently by every publisher. The
/// sampling policy is shared by both key kinds and changes for both at once.
fn keyed_column(
    columns: &[String],
    rows: &[Vec<Cell>],
    names: &[&str],
    parses: fn(&str) -> bool,
) -> Option<usize> {
    let holds = |at: usize| {
        let mut seen = 0;

        for row in rows.iter().take(ADDRESS_SAMPLE) {
            let Some(text) = cell(row, at) else {
                continue;
            };
            if !parses(text) {
                return false;
            }
            seen += 1;
        }

        seen > 0
    };

    let conventional = columns
        .iter()
        .position(|column| names.contains(&column.to_ascii_lowercase().as_str()));

    if let Some(at) = conventional
        && holds(at)
    {
        return Some(at);
    }

    (0..columns.len()).find(|&at| holds(at))
}

/// Bytes a table may hold while its index is built, and what it holds so far.
///
/// An estimate on the same footing as the parse's: what each structure asks
/// the allocator for, with no account of how the allocator rounds it.
struct Budget {
    /// Bytes charged so far.
    held: u64,

    /// Bytes the table may cost before the build is abandoned.
    ceiling: u64,
}

impl Budget {
    /// Add `bytes`, refusing once the total passes the ceiling.
    fn charge(&mut self, bytes: usize) -> Result<(), TableError> {
        let bytes = u64::try_from(bytes).unwrap_or(u64::MAX);
        self.held = self.held.saturating_add(bytes);
        if self.held > self.ceiling {
            return Err(TableError::OverResidentCeiling {
                ceiling: self.ceiling,
            });
        }
        Ok(())
    }

    /// Append a row position, charging whatever capacity the list grows by.
    fn push(&mut self, list: &mut Vec<usize>, at: usize) -> Result<(), TableError> {
        let before = list.capacity();
        list.push(at);
        self.charge((list.capacity() - before) * size_of::<usize>())
    }

    /// File a row position under `key`, charging the `key_bytes` it holds on
    /// the heap, the list it lands in and the buckets the map grows by.
    fn file<K: Hash + Eq>(
        &mut self,
        filed: &mut HashMap<K, Vec<usize>>,
        key: K,
        key_bytes: usize,
        at: usize,
    ) -> Result<(), TableError> {
        let slots = filed.capacity();
        let list = match filed.entry(key) {
            Entry::Occupied(entry) => entry.into_mut(),
            Entry::Vacant(entry) => {
                self.charge(key_bytes)?;
                entry.insert(Vec::new())
            }
        };
        self.push(list, at)?;

        // A bucket holds one entry and one control byte, and the map keeps an
        // eighth of its buckets empty.
        let bucket = size_of::<(K, Vec<usize>)>() + 1;
        self.charge((filed.capacity() - slots) * bucket * 8 / 7)
    }
}

/// The backing each table got, through the facade.
///
/// On the same footing as the enricher's own backing gauge: it is emitted once
/// per table per load, so it costs nothing a lookup can feel.
#[cfg(feature = "metrics")]
mod telemetry {
    use super::TableBacking;

    /// Value of the `type` label on every series.
    const TYPE: &str = "table";

    /// Report where a loaded table's rows are.
    pub(super) fn backing(file: &str, chosen: TableBacking) {
        metrics::gauge!(
            "enrichment_table_backing",
            "type" => TYPE,
            "file" => file.to_string(),
            "backing" => chosen.label(),
        )
        .set(1.0);
    }
}

/// Backing emission, compiled out.
#[cfg(not(feature = "metrics"))]
mod telemetry {
    use super::TableBacking;

    /// Report where a loaded table's rows are.
    pub(super) const fn backing(_file: &str, _chosen: TableBacking) {}
}

#[cfg(test)]
mod tests;