earthsciio 0.1.4

//! The `transport` registry (`spec/registries.md` §1).
//!
//! Keyed by URL scheme. A transport fetches a resolved URL's bytes into a
//! `tmp/<uuid>.part` staging path. Transports are **bypassed entirely in
//! offline mode** — the cache never constructs or calls one when `offline`.

mod cds;
mod file;
mod http;
mod s3;

pub use cds::{
    build_cds_url, cds_api_key, cds_auth, parse_cds_url, CdsTransport, CDS_API_URL, CDS_REALM,
};
pub use file::{file_url_to_path, FileTransport};
pub use http::HttpTransport;
pub use s3::{resolve_region, s3_https_url, S3Transport, DEFAULT_S3_REGION};

use std::collections::HashMap;
use std::path::Path;
use std::sync::Arc;

use crate::auth::AuthResolver;
use crate::error::Result;

/// `$var` as whole seconds, or `default_secs` when unset or unparsable — the
/// shared knob-reading for the transports' request timeouts.
pub(crate) fn env_secs(var: &str, default_secs: u64) -> std::time::Duration {
    std::env::var(var)
        .ok()
        .and_then(|v| v.parse::<u64>().ok())
        .map_or(
            std::time::Duration::from_secs(default_secs),
            std::time::Duration::from_secs,
        )
}

/// Conditional-GET validators carried from a prior manifest into a fetch. Empty
/// when there is nothing cached to revalidate against.
#[derive(Debug, Default, Clone)]
pub struct Conditional {
    /// Stored ETag, sent as `If-None-Match`.
    pub etag: Option<String>,
    /// Stored Last-Modified, sent as `If-Modified-Since`.
    pub last_modified: Option<String>,
}

impl Conditional {
    /// True when neither validator is present.
    pub fn is_empty(&self) -> bool {
        self.etag.is_none() && self.last_modified.is_none()
    }
}

/// Whether a fetch produced new bytes or revalidated an existing blob.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum FetchStatus {
    /// New bytes were written to the staging path.
    Downloaded,
    /// `304 Not Modified` — the cached blob is still valid; staging is untouched.
    NotModified,
}

/// The outcome of a transport fetch.
#[derive(Debug, Clone)]
pub struct FetchResult {
    /// Whether new bytes were downloaded or the cached blob was revalidated.
    pub status: FetchStatus,
    /// ETag from the response, to persist into the manifest.
    pub etag: Option<String>,
    /// Last-Modified from the response, to persist into the manifest.
    pub last_modified: Option<String>,
    /// Bytes written to the staging path (`0` for `NotModified`).
    pub bytes_written: u64,
}

/// Fetches a resolved URL's bytes into the cache. Keyed by URL scheme; never
/// constructed in offline mode.
pub trait Transport: Send + Sync {
    /// URL schemes this transport handles (e.g. `["http", "https"]`).
    fn schemes(&self) -> &'static [&'static str];

    /// Download `resolved_url` into `dest` (a staging path on the cache
    /// filesystem), honoring conditional validators and optional auth.
    fn fetch(
        &self,
        resolved_url: &str,
        dest: &Path,
        conditional: &Conditional,
        auth: Option<&dyn AuthResolver>,
    ) -> Result<FetchResult>;

    /// [`fetch`](Transport::fetch), plus the sha256 (lowercase hex) of the bytes
    /// written when the transport hashed them in transit. `None` means the
    /// caller hashes the staged file itself — the default, so an existing
    /// transport stays valid unchanged.
    fn fetch_hashed(
        &self,
        resolved_url: &str,
        dest: &Path,
        conditional: &Conditional,
        auth: Option<&dyn AuthResolver>,
    ) -> Result<(FetchResult, Option<String>)> {
        self.fetch(resolved_url, dest, conditional, auth)
            .map(|result| (result, None))
    }
}

/// Scheme → transport lookup. Adding a scheme is a registration, never a
/// Provider edit (`spec/registries.md` §1).
#[derive(Default, Clone)]
pub struct TransportRegistry {
    by_scheme: HashMap<String, Arc<dyn Transport>>,
}

impl TransportRegistry {
    /// An empty registry.
    pub fn new() -> Self {
        Self::default()
    }

    /// Registry with the built-in **active** transports: `http`/`https`, `file`,
    /// and `cds` (the Copernicus CDS API).
    pub fn with_builtins() -> Self {
        let mut r = Self::new();
        r.register(Arc::new(HttpTransport::new()));
        r.register(Arc::new(FileTransport::new()));
        r.register(Arc::new(CdsTransport::new()));
        r.register(Arc::new(S3Transport::new()));
        r
    }

    /// Register a transport under each of its schemes.
    pub fn register(&mut self, transport: Arc<dyn Transport>) -> &mut Self {
        for scheme in transport.schemes() {
            self.by_scheme
                .insert((*scheme).to_string(), transport.clone());
        }
        self
    }

    /// Look up the transport for a URL scheme.
    pub fn get(&self, scheme: &str) -> Option<Arc<dyn Transport>> {
        self.by_scheme.get(scheme).cloned()
    }

    /// All registered schemes.
    pub fn schemes(&self) -> Vec<String> {
        self.by_scheme.keys().cloned().collect()
    }
}

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

    #[test]
    fn builtins_cover_http_https_file_cds() {
        let r = TransportRegistry::with_builtins();
        assert!(r.get("http").is_some());
        assert!(r.get("https").is_some());
        assert!(r.get("file").is_some());
        assert!(r.get("cds").is_some()); // Copernicus CDS API
        assert!(r.get("s3").is_some()); // anonymous s3:// -> regional HTTPS rewrite
    }
}