use crate::data::embed_data;
use chrono::NaiveDate;
use lazy_static::lazy_static;
use serde::Deserialize;
use std::collections::HashMap;
use std::error::Error;
use std::path::Path;
use std::time::Duration;

#[derive(Debug, Clone)]
/// Information about a specific distribution series
pub struct SeriesInformation {
    /// Distribution series
    pub series: String,
    /// Codename, i.e. full name of series
    pub codename: String,
    /// Series version as numbers
    pub version: Option<String>,
    /// Series creation date (absent if missing or invalid in the CSV data)
    pub created: Option<NaiveDate>,
    /// Series release date
    pub release: Option<NaiveDate>,
    /// Series end-of-life date
    pub eol: Option<NaiveDate>,
}

#[derive(Debug, Deserialize)]
struct SeriesInfo {
    local: String,
    network: String,
}

/// Architectures an archive mirror serves: an explicit list, or the `all`
/// sentinel meaning one mirror serves every architecture (Debian's mirror
/// setup — an exhaustive list would rot each time an arch is added)
#[derive(Debug, Deserialize)]
#[serde(untagged)]
enum MirrorArchs {
    /// The `all` sentinel
    All(String),
    /// An explicit list of dpkg architecture names
    List(Vec<String>),
}

impl MirrorArchs {
    /// Whether the mirror serves `arch`. A scalar other than the `all`
    /// sentinel serves nothing (a test locks that reading of the data).
    fn serves(&self, arch: &str) -> bool {
        match self {
            MirrorArchs::All(sentinel) => sentinel == "all",
            MirrorArchs::List(archs) => archs.iter().any(|a| a == arch),
        }
    }
}

/// One archive mirror of a distribution: a URL serving a set of
/// architectures, plus the sibling host serving its `-security` pocket
/// for the same architectures (ports mirrors serve their own security)
#[derive(Debug, Deserialize)]
pub struct Mirror {
    /// Base URL of the mirror (the primary mirror's URL doubles as the
    /// dist's base URL, see [`get_base_url`])
    pub url: String,
    /// Sibling host serving the `-security` pocket for these
    /// architectures; `None` when the mirror serves its own security
    #[serde(default)]
    security_url: Option<String>,
    /// Architectures the mirror serves (see [`MirrorArchs`])
    archs: MirrorArchs,
}

impl Mirror {
    /// Whether the mirror serves `arch`
    pub fn serves(&self, arch: &str) -> bool {
        self.archs.serves(arch)
    }
}

#[derive(Debug, Deserialize)]
struct DistData {
    mirrors: HashMap<String, Mirror>,
    archive_keyring: String,
    pockets: Vec<String>,
    #[serde(default)]
    sections: Vec<String>,
    components: Vec<String>,
    cross_pockets: Vec<String>,
    #[serde(default)]
    build_profiles: Vec<String>,
    /// Changelog suite names aliasing a distro-info series codename
    /// ('unstable' for Debian's 'sid'): the two names identify the same
    /// series ([`series_suite_alias`], [`resolve_suite_alias`])
    #[serde(default)]
    suite_aliases: HashMap<String, String>,
    series: SeriesInfo,
}

#[derive(Debug, Deserialize)]
struct Data {
    dist: HashMap<String, DistData>,
}

embed_data! {
    static ref DATA: Data = "../data/distro_info.yml"
}

lazy_static! {
    // Shared HTTP client used for all outgoing plain requests: timeouts keep
    // a hanging remote (connect or transfer) from stalling pkh indefinitely.
    // The short pool idle timeout and TCP keepalive avoid reusing keep-alive
    // connections that the remote closed in the meantime, which surfaces as
    // spurious 'error sending request' failures on busy mirrors/CDNs.
    static ref HTTP_CLIENT: reqwest::Client = reqwest::Client::builder()
        .connect_timeout(Duration::from_secs(10))
        .timeout(Duration::from_secs(30))
        .pool_idle_timeout(Duration::from_secs(10))
        .tcp_keepalive(Duration::from_secs(30))
        .build()
        .expect("building the shared HTTP client with static options cannot fail");
}

/// Shared HTTP client with a connect timeout (10s) and a total request
/// timeout (30s), to be used for all outgoing plain HTTP(S) requests
pub(crate) fn http_client() -> &'static reqwest::Client {
    &HTTP_CLIENT
}

/// GET `url` with bounded retries on transient transport errors (a pooled
/// keep-alive connection closed by the remote, a momentary network hiccup,
/// ...): these always succeed again on a fresh connection, and mirrors are
/// busy enough that unguarded single attempts make bulk operations flaky.
///
/// The response status is not inspected: 404s and the like are meaningful
/// answers, not transport failures.
pub(crate) async fn http_get_retried(url: &str) -> reqwest::Result<reqwest::Response> {
    http_get_retried_with_timeout(url, None).await
}

/// [`http_get_retried`] with a per-request timeout override, for large
/// streaming downloads that exceed the shared client's total timeout
pub(crate) async fn http_get_retried_with_timeout(
    url: &str,
    timeout: Option<Duration>,
) -> reqwest::Result<reqwest::Response> {
    const ATTEMPTS: u32 = 3;
    let mut last_error: Option<reqwest::Error> = None;
    for attempt in 0..ATTEMPTS {
        let mut request = http_client().get(url);
        if let Some(timeout) = timeout {
            request = request.timeout(timeout);
        }
        match request.send().await {
            Ok(response) => return Ok(response),
            Err(e) => {
                if attempt + 1 < ATTEMPTS {
                    log::debug!(
                        "GET '{url}' failed (attempt {}/{}, retrying): {}",
                        attempt + 1,
                        ATTEMPTS,
                        e
                    );
                    tokio::time::sleep(Duration::from_millis(300 * (u64::from(attempt) + 1))).await;
                }
                last_error = Some(e);
            }
        }
    }
    Err(last_error.expect("at least one attempt was made"))
}

/// Parse an optional '%Y-%m-%d' date from a CSV cell, warning instead of
/// panicking on invalid remote data
fn parse_optional_date(value: Option<&str>, series: &str, field: &str) -> Option<NaiveDate> {
    value.and_then(
        |date_str| match NaiveDate::parse_from_str(date_str, "%Y-%m-%d") {
            Ok(date) => Some(date),
            Err(e) => {
                log::warn!(
                    "Invalid '{}' date '{}' for series '{}': {}. Ignoring the date.",
                    field,
                    date_str,
                    series,
                    e
                );
                None
            }
        },
    )
}

fn parse_series_csv(content: &str) -> Result<Vec<SeriesInformation>, Box<dyn Error>> {
    let mut rdr = csv::ReaderBuilder::new()
        .flexible(true)
        .from_reader(content.as_bytes());

    let headers = rdr.headers()?.clone();
    let series_idx = headers
        .iter()
        .position(|h| h == "series")
        .ok_or("Column 'series' not found")?;
    let codename_idx = headers
        .iter()
        .position(|h| h == "codename")
        .ok_or("Column 'codename' not found")?;
    let version_idx = headers
        .iter()
        .position(|h| h == "version")
        .ok_or("Column 'version' not found")?;
    let created_idx = headers
        .iter()
        .position(|h| h == "created")
        .ok_or("Column 'created' not found")?;
    let release_idx = headers
        .iter()
        .position(|h| h == "release")
        .ok_or("Column 'release' not found")?;
    let eol_idx = headers
        .iter()
        .position(|h| h == "eol")
        .ok_or("Column 'eol' not found")?;

    let mut series_info_list = Vec::new();

    for result in rdr.records() {
        let record = match result {
            Ok(record) => record,
            Err(e) => {
                log::warn!("Skipping malformed series CSV row: {}", e);
                continue;
            }
        };

        // Rows missing essential identification fields are skipped: they
        // cannot be used nor reported meaningfully. Dates, on the other
        // hand, are all optional in the model, so a bad date keeps the row.
        let Some(series) = record.get(series_idx).filter(|s| !s.is_empty()) else {
            log::warn!(
                "Skipping series CSV row without a 'series' value: {:?}",
                record
            );
            continue;
        };
        let Some(codename) = record.get(codename_idx).filter(|s| !s.is_empty()) else {
            log::warn!(
                "Skipping series CSV row for series '{}' without a 'codename' value",
                series
            );
            continue;
        };
        let version = record.get(version_idx).map(|s| s.to_string());
        let created = parse_optional_date(record.get(created_idx), series, "created");
        let release = parse_optional_date(record.get(release_idx), series, "release");
        let eol = parse_optional_date(record.get(eol_idx), series, "eol");

        series_info_list.push(SeriesInformation {
            series: series.to_string(),
            codename: codename.to_string(),
            version,
            created,
            release,
            eol,
        });
    }

    // Revert to sort by most recent
    series_info_list.reverse();

    Ok(series_info_list)
}

/// List the distributions known to pkh (e.g. "debian", "ubuntu"), sorted so
/// that menus and error messages derived from it are deterministic
pub fn supported_dists() -> Vec<String> {
    let mut dists: Vec<String> = DATA.dist.keys().cloned().collect();
    dists.sort();
    dists
}

/// Name of a dist's primary mirror entry: its URL doubles as the dist's
/// base URL ([`get_base_url`]) and is the first candidate of
/// [`mirror_for_arch`]
const PRIMARY_MIRROR: &str = "primary";

/// The data of a known distribution: the shared "unknown distribution"
/// error of the per-dist accessors
fn dist_data(dist: &str) -> Result<&'static DistData, Box<dyn Error>> {
    DATA.dist.get(dist).ok_or_else(|| {
        format!(
            "Unknown distribution '{}'. Supported distributions are: {}.",
            dist,
            supported_dists().join(", ")
        )
        .into()
    })
}

/// Special changelog distribution marking an entry that has not been
/// released to any archive series yet
pub const UNRELEASED: &str = "UNRELEASED";

/// Whether `series` is the special [`UNRELEASED`] distribution rather than
/// a real archive series
pub fn is_unreleased(series: &str) -> bool {
    series == UNRELEASED
}

/// Get time-ordered list of series information for a distribution, development series first
pub async fn get_ordered_series(dist: &str) -> Result<Vec<SeriesInformation>, Box<dyn Error>> {
    let dist_data = DATA.dist.get(dist).ok_or_else(|| {
        format!(
            "Unknown distribution '{}'. Supported distributions are: {}.",
            dist,
            supported_dists().join(", ")
        )
    })?;
    let series_info = &dist_data.series;
    let content = if Path::new(series_info.local.as_str()).exists() {
        std::fs::read_to_string(series_info.local.as_str()).map_err(|e| {
            format!(
                "Failed to read distribution series data for '{dist}' \
                     from '{}': {}. The 'distro-info' package provides these CSV files.",
                series_info.local, e
            )
        })?
    } else {
        http_client()
            .get(series_info.network.as_str())
            .send()
            .await?
            .text()
            .await?
    };

    let series_info_list = parse_series_csv(&content)?;
    Ok(series_info_list)
}

/// Get time-ordered list of series names for a distribution, development series first
pub async fn get_ordered_series_name(dist: &str) -> Result<Vec<String>, Box<dyn Error>> {
    let series = get_ordered_series(dist).await?;
    Ok(series.iter().map(|info| info.series.clone()).collect())
}

/// The series to actually target when the changelog says [`UNRELEASED`]:
/// the development series of `dist`, i.e. the first entry of
/// [`get_ordered_series_name`] (which is documented "development series
/// first"). UNRELEASED work conventionally targets the next release, not
/// the last stable one. `dist` is matched case-insensitively, so vendor
/// names with original casing (dpkg's `Vendor:` field is e.g. "Ubuntu")
/// are accepted as-is. Any other `series` is returned unchanged. Errors
/// when `dist` is unknown or has no series list.
pub async fn effective_series(series: &str, dist: &str) -> Result<String, Box<dyn Error>> {
    if !is_unreleased(series) {
        return Ok(series.to_string());
    }

    // The series data keys are lowercase, unlike the vendor names that
    // callers typically resolve from dpkg
    let dist = dist.to_lowercase();

    get_ordered_series_name(&dist)
        .await?
        .into_iter()
        .next()
        .ok_or_else(|| format!("Distribution '{dist}' has no series to target").into())
}

/// Get the latest released series for a dist (excluding future releases and special cases like sid)
pub async fn get_latest_released_series(dist: &str) -> Result<String, Box<dyn Error>> {
    let latest = get_n_latest_released_series(dist, 1).await?;
    latest
        .first()
        .cloned()
        .ok_or("No released series found".into())
}

/// Get the N latest released series for a dist (excluding future releases and special cases like sid)
pub async fn get_n_latest_released_series(
    dist: &str,
    n: usize,
) -> Result<Vec<String>, Box<dyn Error>> {
    let series_info_list = get_ordered_series(dist).await?;

    let today = chrono::Local::now().date_naive();
    let mut released_series = Vec::new();

    for series_info in series_info_list {
        // Skip 'sid' and series without release dates or with future release dates
        if series_info.series != "sid"
            && series_info.series != "experimental"
            && series_info.release.is_some()
            && series_info.release.unwrap() <= today
        {
            released_series.push(series_info);
        }
    }

    // Sort by release date descending (newest first)
    released_series.sort_by_key(|b| std::cmp::Reverse(b.release));

    Ok(released_series
        .iter()
        .take(n)
        .map(|s| s.series.clone())
        .collect())
}

/// Obtain the distribution (eg. debian, ubuntu) from a distribution series (eg. noble, bookworm)
pub async fn get_dist_from_series(series: &str) -> Result<String, Box<dyn Error>> {
    for dist in DATA.dist.keys() {
        if get_ordered_series_name(dist)
            .await?
            .contains(&series.to_string())
        {
            return Ok(dist.to_string());
        }
    }
    Err(format!("Unknown series: {}", series).into())
}

/// The changelog suite name that aliases the series codename of `dist`
/// (Debian's 'unstable' for 'sid'): the two names identify the same
/// series. `None` when the series carries no suite alias.
pub fn series_suite_alias(dist: &str, series: &str) -> Option<String> {
    dist_data(dist)
        .ok()?
        .suite_aliases
        .iter()
        .find(|(_suite, codename)| codename.as_str() == series)
        .map(|(suite, _)| suite.clone())
}

/// Identify a changelog suite name with the distro-info series codename
/// it aliases (Debian's 'unstable' is 'sid'), and the dist that codename
/// belongs to. `None` when `suite` is not a known alias of any dist.
pub fn resolve_suite_alias(suite: &str) -> Option<(String, String)> {
    for (dist, data) in DATA.dist.iter() {
        if let Some(codename) = data.suite_aliases.get(suite) {
            return Some((dist.clone(), codename.clone()));
        }
    }
    None
}

/// Get the package pockets available for a given distribution, in search order
///
/// The main archive ('') comes first so that a search without an explicit
/// pocket prefers the released archive over its pockets; development pockets
/// (e.g. '-proposed') come last.
///
/// Example: get_dist_pockets(ubuntu) => ["", "updates", "security", "proposed"]
pub fn get_dist_pockets(dist: &str) -> Result<Vec<String>, Box<dyn Error>> {
    let mut pockets = dist_data(dist)?.pockets.clone();

    // Explicitely add 'main' pocket, which is just the empty string, first
    pockets.insert(0, "".to_string());

    Ok(pockets)
}

/// Get the archive components of a distribution (ubuntu's main,
/// restricted, universe, multiverse; Debian's main, contrib, non-free,
/// non-free-firmware): the default set a build environment enables on its
/// official sources. Live archive operations keep resolving components
/// from Release files ([`get_components`]); this is the offline default.
pub fn get_dist_components(dist: &str) -> Result<Vec<String>, Box<dyn Error>> {
    Ok(dist_data(dist)?.components.clone())
}

/// Get the pockets a cross-build environment enables for a series (the
/// `<series>-<pocket>` suite list is built from these): updates,
/// backports and security. Deliberately separate from
/// [`get_dist_pockets`], which is the *search order* of pull — folding
/// backports into it would change pull behavior.
pub fn get_cross_pockets(dist: &str) -> Result<Vec<String>, Box<dyn Error>> {
    Ok(dist_data(dist)?.cross_pockets.clone())
}

/// Get the default build profiles of a distribution's vendor (Ubuntu
/// activates `derivative.ubuntu noudeb`, Debian none), mirroring what
/// `Dpkg::BuildProfiles` resolves when `DEB_BUILD_PROFILES` is unset.
/// Vendors are matched case-insensitively by the caller (dpkg's `Vendor:`
/// field keeps its original casing).
pub fn get_build_profiles(dist: &str) -> Result<Vec<String>, Box<dyn Error>> {
    Ok(dist_data(dist)?.build_profiles.clone())
}

/// Get the valid `Section` values of a distribution's packages, as accepted
/// by its archives (a `section/subsection` in debian/control validates on
/// the part before the '/')
pub fn get_sections(dist: &str) -> Result<Vec<String>, Box<dyn Error>> {
    Ok(dist_data(dist)?.sections.clone())
}

/// Get the sources URL for a distribution, series, pocket, and component
pub fn get_sources_url(base_url: &str, series: &str, pocket: &str, component: &str) -> String {
    let pocket_full = if pocket.is_empty() {
        String::new()
    } else {
        format!("-{}", pocket)
    };
    format!("{base_url}/dists/{series}{pocket_full}/{component}/source/Sources.gz")
}

/// Get the archive base URL for a distribution: the URL of its primary
/// mirror (the former `base_url` key folded into `mirrors.primary.url`
/// when the mirrors were modeled — the signature is kept so the pull
/// paths do not churn)
///
/// Example: ubuntu => https://archive.ubuntu.com/ubuntu
pub fn get_base_url(dist: &str) -> Result<String, Box<dyn Error>> {
    let mirror = dist_data(dist)?
        .mirrors
        .get(PRIMARY_MIRROR)
        .ok_or_else(|| {
            format!(
                "Distribution '{dist}' has no '{PRIMARY_MIRROR}' mirror in the built-in \
             configuration. This is a bug; supported distributions are: {}.",
                supported_dists().join(", ")
            )
        })?;
    Ok(mirror.url.clone())
}

/// The mirror of `dist` serving `arch`: the primary mirror first, then
/// the other mirrors by name, so the answer is deterministic. Debian's
/// `all` sentinel makes its primary mirror serve every architecture.
/// Errors when the dist is unknown or no mirror serves the architecture
/// (an architecture the built-in data does not know about).
///
/// Example: mirror_for_arch(ubuntu, riscv64) => the ports mirror
pub fn mirror_for_arch(dist: &str, arch: &str) -> Result<&'static Mirror, Box<dyn Error>> {
    let data = dist_data(dist)?;
    if let Some(primary) = data.mirrors.get(PRIMARY_MIRROR)
        && primary.serves(arch)
    {
        return Ok(primary);
    }
    let mut others: Vec<&String> = data
        .mirrors
        .keys()
        .filter(|name| name.as_str() != PRIMARY_MIRROR)
        .collect();
    others.sort();
    others
        .into_iter()
        .filter_map(|name| data.mirrors.get(name))
        .find(|mirror| mirror.serves(arch))
        .ok_or_else(|| {
            format!(
                "No mirror of '{dist}' serves the '{arch}' architecture. Supported \
                 distributions are: {}.",
                supported_dists().join(", ")
            )
            .into()
        })
}

/// Host part of an apt-source URL: everything after the `://` scheme up
/// to the first `/` (a `:port` suffix stripped). URLs without a scheme
/// yield their leading segment.
fn url_host(url: &str) -> &str {
    let after_scheme = url.split_once("://").map_or(url, |(_, rest)| rest);
    let authority = after_scheme.split('/').next().unwrap_or(after_scheme);
    authority
        .split_once(':')
        .map_or(authority, |(host, _)| host)
}

/// Whether the host of `uri` is the host of `mirror_url` or a subdomain
/// of it: the old substring checks (`uri.contains("archive.ubuntu.com")`)
/// intentionally matched the country mirrors fronting each archive host
/// (`fr.archive.ubuntu.com`), and an exact host comparison would have
/// dropped them. The leading dot of the suffix keeps look-alike hosts
/// (`notarchive.ubuntu.com`) out.
fn uri_matches_url_host(uri: &str, mirror_url: &str) -> bool {
    let host = url_host(mirror_url);
    let uri_host = url_host(uri);
    uri_host == host || uri_host.ends_with(&format!(".{host}"))
}

/// Whether `uri` points at `mirror` or its security sibling: the URI's
/// host is the mirror's (or the sibling's) host or a subdomain of it (see
/// [`uri_matches_url_host`])
pub fn is_mirror_source(mirror: &Mirror, uri: &str) -> bool {
    uri_matches_url_host(uri, &mirror.url)
        || mirror
            .security_url
            .as_deref()
            .is_some_and(|security| uri_matches_url_host(uri, security))
}

/// Whether `uri` points at an official source of `dist` — one of its
/// archive mirrors or their security siblings — as opposed to a PPA or
/// another third-party repository. Unknown distributions match nothing.
pub fn is_official_source(dist: &str, uri: &str) -> bool {
    DATA.dist.get(dist).is_some_and(|data| {
        data.mirrors
            .values()
            .any(|mirror| is_mirror_source(mirror, uri))
    })
}

/// Obtain the URLs for the archive keyrings of a distribution series
///
/// For 'sid' and 'experimental', returns keyrings from the 3 latest releases
/// since sid needs keys from all recent releases.
pub async fn get_keyring_urls(series: &str) -> Result<Vec<String>, Box<dyn Error>> {
    let dist = get_dist_from_series(series).await?;
    let dist_data = DATA
        .dist
        .get(&dist)
        .ok_or(format!("Unsupported distribution: {}", dist))?;

    // For Debian, we need the series number to form the keyring URL
    if dist == "debian" {
        // Special case for 'sid' - use keyrings from the 3 latest released versions
        if series == "sid" || series == "experimental" {
            let latest_released = get_n_latest_released_series("debian", 3).await?;
            let mut urls = Vec::new();
            for released_series in latest_released {
                if let Some(series_num) = get_debian_series_number(&released_series).await? {
                    urls.push(
                        dist_data
                            .archive_keyring
                            .replace("{series_num}", &series_num),
                    );
                }
            }
            if urls.is_empty() {
                Err("No keyring URLs found for sid/experimental".into())
            } else {
                Ok(urls)
            }
        } else {
            let series_num = get_debian_series_number(series).await?.ok_or_else(|| {
                format!(
                    "Could not determine the version number for Debian series '{series}'. \
                         Make sure the 'distro-info' package is installed, which provides the \
                         series CSV data used to map series names to version numbers."
                )
            })?;
            // Replace {series_num} placeholder with the actual series number
            Ok(vec![
                dist_data
                    .archive_keyring
                    .replace("{series_num}", &series_num),
            ])
        }
    } else {
        // For other distributions like Ubuntu, use the keyring directly
        Ok(vec![dist_data.archive_keyring.clone()])
    }
}

/// Obtain the URL for the 'Release' file of a distribution series
fn get_release_url(base_url: &str, series: &str, pocket: &str) -> String {
    let pocket_full = if pocket.is_empty() {
        String::new()
    } else {
        format!("-{}", pocket)
    };
    format!("{base_url}/dists/{series}{pocket_full}/Release")
}

/// Obtain the components of a distribution series by parsing the 'Release' file
pub async fn get_components(
    base_url: &str,
    series: &str,
    pocket: &str,
) -> Result<Vec<String>, Box<dyn Error>> {
    let url = get_release_url(base_url, series, pocket);
    log::debug!("Fetching Release file from: {}", url);

    let content = http_client().get(&url).send().await?.text().await?;

    for line in content.lines() {
        if line.starts_with("Components:")
            && let Some((_, components)) = line.split_once(':')
        {
            return Ok(components
                .split_whitespace()
                .map(|s| s.to_string())
                .collect());
        }
    }

    Err("Components not found.".into())
}

/// Map a Debian series name to its version number
pub async fn get_debian_series_number(series: &str) -> Result<Option<String>, Box<dyn Error>> {
    let dist_data = DATA.dist.get("debian").ok_or_else(|| {
        format!(
            "Debian distribution data is missing from the built-in configuration. \
             This is a bug; supported distributions are: {}.",
            supported_dists().join(", ")
        )
    })?;
    let series_info = &dist_data.series;
    let content = if Path::new(series_info.local.as_str()).exists() {
        std::fs::read_to_string(series_info.local.as_str()).map_err(|e| {
            format!(
                "Failed to read Debian series data from '{}': {}. \
                     The 'distro-info' package provides this file.",
                series_info.local, e
            )
        })?
    } else {
        http_client()
            .get(series_info.network.as_str())
            .send()
            .await?
            .text()
            .await?
    };

    let mut rdr = csv::ReaderBuilder::new()
        .flexible(true)
        .from_reader(content.as_bytes());

    let headers = rdr.headers()?.clone();
    let series_idx = headers
        .iter()
        .position(|h| h == "series")
        .ok_or("Column 'series' not found")?;
    let version_idx = headers
        .iter()
        .position(|h| h == "version")
        .ok_or("Column 'version' not found")?;

    for result in rdr.records() {
        let record = result?;
        if let (Some(s), Some(v)) = (record.get(series_idx), record.get(version_idx))
            && s.to_lowercase() == series.to_lowercase()
        {
            return Ok(Some(v.to_string()));
        }
    }

    Ok(None)
}

/// The release number of a distribution series, paired with the dist it
/// belongs to: the version column of the series data, stripped to its
/// leading token ("12" for Debian bookworm, "26.04" out of Ubuntu
/// resolute's "26.04 LTS"). `None` when the series carries no version at
/// all (Debian's rolling sid/experimental have an empty column;
/// pseudo-versions like "unstable" pass through, callers validate per
/// vendor). Errors when no known distribution carries the series.
pub async fn get_series_release_number(
    series: &str,
) -> Result<Option<(String, String)>, Box<dyn Error>> {
    let dist = get_dist_from_series(series).await?;
    for info in get_ordered_series(&dist).await? {
        if info.series == series {
            let number = info
                .version
                .as_deref()
                .and_then(|version| version.split_whitespace().next())
                .map(str::to_string);
            return Ok(number.map(|number| (dist, number)));
        }
    }
    Ok(None)
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_get_sections() {
        // Both distributions bundle the policy section list
        for dist in ["debian", "ubuntu"] {
            let sections = get_sections(dist).unwrap();
            assert!(sections.contains(&"utils".to_string()));
            assert!(sections.contains(&"devel".to_string()));
            // 'unknown' is exactly what archives reject
            assert!(!sections.contains(&"unknown".to_string()));
        }
        assert!(get_sections("not-a-distro").is_err());
    }

    /// The primary mirror's URL is the former `base_url`, byte for byte:
    /// the pull paths build their archive URLs from it.
    #[test]
    fn test_primary_mirror_url_is_the_base_url() {
        assert_eq!(
            get_base_url("ubuntu").unwrap(),
            "https://archive.ubuntu.com/ubuntu"
        );
        assert_eq!(
            get_base_url("debian").unwrap(),
            "https://deb.debian.org/debian"
        );
        assert!(get_base_url("not-a-distro").is_err());
    }

    /// Mirror-per-architecture resolution: the local architectures come
    /// from Ubuntu's primary mirror, the others from ports; Debian's `all`
    /// sentinel makes its one mirror serve everything, including
    /// architectures the data never lists.
    #[test]
    fn test_mirror_for_arch() {
        assert_eq!(
            mirror_for_arch("ubuntu", "amd64").unwrap().url,
            "https://archive.ubuntu.com/ubuntu"
        );
        assert_eq!(
            mirror_for_arch("ubuntu", "riscv64").unwrap().url,
            "http://ports.ubuntu.com/ubuntu-ports"
        );
        for arch in ["amd64", "riscv64", "brand-new"] {
            assert_eq!(
                mirror_for_arch("debian", arch).unwrap().url,
                "https://deb.debian.org/debian",
                "the `all` sentinel serves every architecture, including {arch}"
            );
        }
        // An architecture no Ubuntu mirror serves, and an unknown dist.
        assert!(mirror_for_arch("ubuntu", "mips64el").is_err());
        assert!(mirror_for_arch("not-a-distro", "amd64").is_err());
    }

    /// The `archs` forms and their reading: the `all` sentinel serves
    /// everything, an explicit list serves exactly its members, and any
    /// other scalar serves nothing (a data bug a validation would have to
    /// catch, hence the documented reading).
    #[test]
    fn test_mirror_archs_forms() {
        let all: MirrorArchs = serde_yaml::from_str("all").unwrap();
        assert!(all.serves("anything"));
        let list: MirrorArchs = serde_yaml::from_str("[amd64, i386]").unwrap();
        assert!(list.serves("amd64"));
        assert!(!list.serves("arm64"));
        let typo: MirrorArchs = serde_yaml::from_str("every").unwrap();
        assert!(!typo.serves("amd64"));
    }

    /// Official-source matching is host-based but keeps matching the
    /// country mirrors the old substring checks matched (`fr.archive.
    /// ubuntu.com`): equality or a `.{host}` suffix, never a bare
    /// substring — `notarchive.ubuntu.com` must not match.
    #[test]
    fn test_is_official_source_matches_country_mirrors_only() {
        for uri in [
            "https://archive.ubuntu.com/ubuntu",
            "http://security.ubuntu.com/ubuntu",
            "http://ports.ubuntu.com/ubuntu-ports",
            // Country mirrors front the same archives.
            "http://fr.archive.ubuntu.com/ubuntu",
            "https://de.security.ubuntu.com/ubuntu",
        ] {
            assert!(is_official_source("ubuntu", uri), "{uri}");
        }
        for uri in [
            "https://deb.debian.org/debian",
            "https://ppa.launchpadcontent.net/user/ppa/ubuntu",
            "http://notarchive.ubuntu.com/ubuntu",
            "http://archive.ubuntu.com.evil.example/ubuntu",
        ] {
            assert!(!is_official_source("ubuntu", uri), "{uri}");
        }
        // Debian: its own mirror matches, Ubuntu's mirrors do not, and an
        // unknown dist matches nothing.
        assert!(is_official_source(
            "debian",
            "https://deb.debian.org/debian"
        ));
        assert!(!is_official_source(
            "debian",
            "http://security.ubuntu.com/ubuntu"
        ));
        assert!(!is_official_source(
            "not-a-distro",
            "https://deb.debian.org/debian"
        ));
    }

    /// The dist-level defaults the build paths read: components,
    /// cross-build pockets (deliberately not the pull search order) and
    /// vendor build profiles.
    #[test]
    fn test_dist_components_cross_pockets_and_build_profiles() {
        assert_eq!(
            get_dist_components("ubuntu").unwrap(),
            vec!["main", "restricted", "universe", "multiverse"]
        );
        assert!(
            get_dist_components("debian")
                .unwrap()
                .contains(&"non-free-firmware".to_string())
        );

        for dist in ["debian", "ubuntu"] {
            assert_eq!(
                get_cross_pockets(dist).unwrap(),
                vec!["updates", "backports", "security"]
            );
        }
        // Not the pull search order: no 'proposed', no empty main pocket.
        let cross = get_cross_pockets("ubuntu").unwrap();
        assert!(!cross.contains(&"proposed".to_string()));
        assert!(!cross.contains(&"".to_string()));

        assert_eq!(
            get_build_profiles("ubuntu").unwrap(),
            vec!["derivative.ubuntu", "noudeb"]
        );
        assert!(get_build_profiles("debian").unwrap().is_empty());

        for getter in [
            get_dist_components as fn(&str) -> Result<Vec<String>, Box<dyn Error>>,
            get_cross_pockets as fn(&str) -> Result<Vec<String>, Box<dyn Error>>,
            get_build_profiles as fn(&str) -> Result<Vec<String>, Box<dyn Error>>,
        ] {
            assert!(getter("not-a-distro").is_err());
        }
    }

    #[test]
    fn test_parse_series_csv_malformed_rows() {
        // A short row (missing 'codename') is skipped, a row with an invalid
        // 'created' date is kept without a date, and invalid 'release'/'eol'
        // dates become None: none of this may panic on remote data
        let csv_data = "series,codename,version,created,release,eol\n\
                        noble,Noble N,24.04,2023-10-26,2024-04-25,2029-04-25\n\
                        lonely\n\
                        badbad,Bad B,1.0,not-a-date,2020-01-01,also-bad\n\
                        sid,sid,unstable,1999-01-01,,\n";

        let series = parse_series_csv(csv_data).unwrap();

        // Rows are returned most recent first (the parser reverses the list),
        // with the malformed 'lonely' row skipped entirely
        let names: Vec<&str> = series.iter().map(|s| s.series.as_str()).collect();
        assert_eq!(names, vec!["sid", "badbad", "noble"]);

        let noble = &series[2];
        assert_eq!(noble.codename, "Noble N");
        assert_eq!(noble.version.as_deref(), Some("24.04"));
        assert_eq!(
            noble.created,
            Some(NaiveDate::from_ymd_opt(2023, 10, 26).unwrap())
        );
        assert_eq!(
            noble.release,
            Some(NaiveDate::from_ymd_opt(2024, 4, 25).unwrap())
        );
        assert_eq!(
            noble.eol,
            Some(NaiveDate::from_ymd_opt(2029, 4, 25).unwrap())
        );

        let badbad = &series[1];
        assert_eq!(badbad.created, None);
        assert_eq!(
            badbad.release,
            Some(NaiveDate::from_ymd_opt(2020, 1, 1).unwrap())
        );
        assert_eq!(badbad.eol, None);
    }

    #[test]
    fn test_get_dist_pockets_order() {
        // Without an explicit pocket, packages are searched in this order:
        // main archive first, then updates, security, and proposed last
        let pockets = get_dist_pockets("ubuntu").unwrap();
        assert_eq!(
            pockets,
            vec![
                "".to_string(),
                "updates".to_string(),
                "security".to_string(),
                "proposed".to_string()
            ]
        );

        let pockets = get_dist_pockets("debian").unwrap();
        assert_eq!(
            pockets,
            vec![
                "".to_string(),
                "updates".to_string(),
                "security".to_string(),
                "proposed-updates".to_string()
            ]
        );
    }

    #[test]
    fn test_is_unreleased() {
        // Matching is exact: UNRELEASED is uppercase by Debian convention
        assert!(is_unreleased("UNRELEASED"));
        assert!(!is_unreleased("unreleased"));
        assert!(!is_unreleased("noble"));
        assert!(!is_unreleased(""));
    }

    #[tokio::test]
    async fn test_effective_series_passthrough() {
        // A real series is returned unchanged, and the dist is not even
        // looked up (an unknown dist only matters for UNRELEASED)
        assert_eq!(effective_series("noble", "ubuntu").await.unwrap(), "noble");
        assert_eq!(effective_series("sid", "debian").await.unwrap(), "sid");
        assert_eq!(
            effective_series("noble", "unknown-distro").await.unwrap(),
            "noble"
        );
    }

    #[tokio::test]
    async fn test_effective_series_unreleased() {
        // UNRELEASED resolves to the development series of the dist, i.e.
        // the first entry of the time-ordered list. On current distro-info
        // data this is the next Ubuntu release, while Debian's list starts
        // with 'experimental' (sid comes second), so assert against the
        // data itself rather than a hardcoded name.
        for dist in ["ubuntu", "debian"] {
            let ordered = get_ordered_series_name(dist).await.unwrap();
            let resolved = effective_series(UNRELEASED, dist).await.unwrap();
            assert_eq!(resolved, ordered[0]);
            assert_ne!(resolved, UNRELEASED);
        }
    }

    #[tokio::test]
    async fn test_effective_series_unreleased_dist_case_insensitive() {
        // Distro data keys are lowercase but dpkg vendors keep original
        // casing ("Ubuntu"): the UNRELEASED lookup must resolve both
        let expected = effective_series(UNRELEASED, "ubuntu").await.unwrap();
        assert_eq!(
            effective_series(UNRELEASED, "Ubuntu").await.unwrap(),
            expected
        );
        assert_eq!(
            effective_series(UNRELEASED, "UBUNTU").await.unwrap(),
            expected
        );
    }

    #[tokio::test]
    async fn test_effective_series_unknown_dist() {
        // UNRELEASED on an unknown distribution cannot be resolved
        assert!(
            effective_series(UNRELEASED, "unknown-distro")
                .await
                .is_err()
        );
    }

    #[tokio::test]
    async fn test_get_debian_series() {
        let series = get_ordered_series_name("debian").await.unwrap();
        assert!(series.contains(&"sid".to_string()));
        assert!(series.contains(&"bookworm".to_string()));
    }

    #[tokio::test]
    async fn test_get_ubuntu_series() {
        let series = get_ordered_series_name("ubuntu").await.unwrap();
        assert!(series.contains(&"noble".to_string()));
        assert!(series.contains(&"jammy".to_string()));
    }

    /// Suite aliases identify a changelog suite name with the series
    /// codename of the same series: Debian's 'unstable' is 'sid'
    #[test]
    fn test_suite_aliases() {
        assert_eq!(
            resolve_suite_alias("unstable"),
            Some(("debian".to_string(), "sid".to_string()))
        );
        // A series codename or unknown suite is not an alias
        assert_eq!(resolve_suite_alias("sid"), None);
        assert_eq!(resolve_suite_alias("noble"), None);
        assert_eq!(
            series_suite_alias("debian", "sid"),
            Some("unstable".to_string())
        );
        assert_eq!(series_suite_alias("debian", "trixie"), None);
        assert_eq!(series_suite_alias("ubuntu", "noble"), None);
    }

    /// Every suite alias must map to a real series of its dist, or the
    /// selector would offer a phantom entry
    #[tokio::test]
    async fn test_suite_aliases_target_real_series() {
        for (dist, data) in DATA.dist.iter() {
            for (suite, codename) in &data.suite_aliases {
                let series = get_ordered_series_name(dist).await.unwrap_or_default();
                assert!(
                    series.contains(codename),
                    "suite alias '{suite}' of {dist} maps to '{codename}', \
                     which is not a known series"
                );
            }
        }
    }

    #[tokio::test]
    async fn test_get_dist_from_series() {
        assert_eq!(get_dist_from_series("sid").await.unwrap(), "debian");
        assert_eq!(get_dist_from_series("noble").await.unwrap(), "ubuntu");
    }

    #[tokio::test]
    async fn test_get_debian_series_number() {
        // Test with known Debian series
        let bookworm_number = get_debian_series_number("bookworm").await.unwrap();
        assert!(bookworm_number.is_some());
        assert_eq!(bookworm_number.unwrap(), "12");

        let trixie_number = get_debian_series_number("trixie").await.unwrap();
        assert!(trixie_number.is_some());
        assert_eq!(trixie_number.unwrap(), "13");

        // Test with unknown series
        let unknown_number = get_debian_series_number("unknown").await.unwrap();
        assert!(unknown_number.is_none());
    }

    #[tokio::test]
    async fn test_get_series_release_number() {
        let (dist, bookworm) = get_series_release_number("bookworm")
            .await
            .unwrap()
            .unwrap();
        assert_eq!(dist, "debian");
        assert_eq!(bookworm, "12");

        // Ubuntu LTS rows carry a " LTS" decoration: only the leading
        // YY.MM token is the release number
        let (dist, noble) = get_series_release_number("noble").await.unwrap().unwrap();
        assert_eq!(dist, "ubuntu");
        assert_eq!(noble, "24.04");

        // No known dist carries the series
        assert!(get_series_release_number("not-a-series").await.is_err());
    }

    #[tokio::test]
    async fn test_get_keyring_urls_sid() {
        // Test that 'sid' returns keyrings from the 3 latest released versions
        let sid_keyrings = get_keyring_urls("sid").await.unwrap();

        // Should have keyring URLs for sid
        assert!(!sid_keyrings.is_empty());
        assert!(sid_keyrings.len() <= 3);

        // Each URL should be a valid Debian keyring URL
        for url in &sid_keyrings {
            assert!(
                url.contains("ftp-master.debian.org/keys"),
                "URL '{}' does not contain expected pattern",
                url
            );
        }
    }

    #[tokio::test]
    async fn test_get_keyring_url_regular_series() {
        // Test that regular series (like bookworm) returns a single keyring URL
        let bookworm_keyring = &get_keyring_urls("bookworm").await.unwrap()[0];
        assert!(
            bookworm_keyring.contains("ftp-master.debian.org/keys"),
            "URL '{}' does not contain expected pattern",
            bookworm_keyring
        );
    }

    #[tokio::test]
    async fn test_get_n_latest_released_series() {
        // Test getting 3 latest released series
        let latest_3 = get_n_latest_released_series("debian", 3).await.unwrap();

        // Should have at most 3 series
        assert!(!latest_3.is_empty());
        assert!(latest_3.len() <= 3);

        // Should not contain 'sid' or 'experimental'
        assert!(!latest_3.contains(&"sid".to_string()));
        assert!(!latest_3.contains(&"experimental".to_string()));
    }

    #[tokio::test]
    async fn test_get_latest_released_debian_series() {
        // Test that we get a valid released series
        let latest_released = get_latest_released_series("debian").await.unwrap();

        // Should not be 'sid' or 'experimental'
        assert_ne!(latest_released, "sid");
        assert_ne!(latest_released, "experimental");

        // Should have a version number
        let version = get_debian_series_number(&latest_released).await.unwrap();
        assert!(version.is_some());
    }
}