//! Images by hash with a hub (oxygenflow#39, docs/jetstream.md §8):
//! `flow rootfs push` sends an image's missing chunks and its manifest;
//! `flow rootfs pull`, and a runner that lacks a job's image, fetch the
//! manifest and the chunks they lack and write the image back, checked
//! against its hash. Fetched chunks stay in the home's chunk cache
//! (`chunks/`), so a rebuilt image costs only the chunks that changed.

use crate::home::Home;
use anyhow::{bail, Context, Result};
use jet::images::{self, Chunk, Manifest};
use std::collections::HashSet;
use std::path::{Path, PathBuf};
use std::sync::Mutex;
use std::time::{Duration, Instant};

/// Requests in flight at once.
const PARALLEL: usize = 8;

/// A hub's image endpoints, with a token.
pub struct Client {
    root: String,
    base: String,
    token: String,
    agent: ureq::Agent,
}

/// What a push or a fetch moved.
pub struct Moved {
    pub chunks: usize,
    pub bytes: u64,
    /// Of `bytes`, those that came from peers rather than the hub.
    pub peer_bytes: u64,
    pub elapsed: Duration,
}

impl Client {
    pub fn new(hub: &str, token: &str, ca: Option<&Path>) -> Result<Client> {
        let tls = oxy_core::protocol::tls_client_config(ca).map_err(|e| anyhow::anyhow!("{e}"))?;
        let agent = ureq::AgentBuilder::new().tls_config(tls).timeout(Duration::from_secs(120)).build();
        let root = hub.trim_end_matches('/').to_string();
        Ok(Client { base: format!("{root}/api/v1"), root, token: token.into(), agent })
    }

    fn auth(&self) -> String {
        format!("Bearer {}", self.token)
    }

    fn fail(url: &str, e: ureq::Error) -> anyhow::Error {
        match e {
            ureq::Error::Status(code, r) => {
                let text = r.into_string().unwrap_or_default();
                let detail = serde_json::from_str::<serde_json::Value>(&text)
                    .ok()
                    .and_then(|v| v["detail"].as_str().map(String::from))
                    .unwrap_or(text);
                anyhow::anyhow!("{url}: HTTP {code}: {detail}")
            }
            e => anyhow::anyhow!("{url}: {e}"),
        }
    }

    /// The hub's manifest of image `hash`, `None` when it does not hold it.
    pub fn manifest(&self, hash: &str) -> Result<Option<Manifest>> {
        let url = format!("{}/flow/images/{hash}", self.base);
        match self.agent.get(&url).set("Authorization", &self.auth()).call() {
            Ok(r) => Ok(Some(r.into_json()?)),
            Err(ureq::Error::Status(404, _)) => Ok(None),
            Err(e) => Err(Self::fail(&url, e)),
        }
    }

    /// The hub's identity (`/.well-known/oxyhub-key`): the key its swarm
    /// tickets are signed with.
    pub fn hub_identity(&self) -> Result<oxy_core::identity::Identity> {
        let url = format!("{}/.well-known/oxyhub-key", self.root);
        let r = self.agent.get(&url).call().map_err(|e| Self::fail(&url, e))?;
        let mut out = Vec::new();
        std::io::Read::read_to_end(&mut r.into_reader(), &mut out)?;
        oxy_core::identity::Identity::decode(&out).map_err(|e| anyhow::anyhow!("{url}: not a hub identity: {e}"))
    }

    /// Tell the hub's tracker this runner is fetching image `hash`, or has
    /// stopped: the peers to ask and the ticket they admit.
    pub fn announce(&self, hash: &str, state: &str) -> Result<flow_core::api::Announced> {
        let url = format!("{}/flow/swarm/announce", self.base);
        let body = flow_core::api::Announce { image: hash.into(), state: state.into() };
        Ok(self.agent.post(&url).set("Authorization", &self.auth()).send_json(&body).map_err(|e| Self::fail(&url, e))?.into_json()?)
    }

    pub(crate) fn chunk(&self, hash: &str) -> Result<Vec<u8>> {
        let url = format!("{}/flow/images/chunks/{hash}", self.base);
        let r = self.agent.get(&url).set("Authorization", &self.auth()).call().map_err(|e| Self::fail(&url, e))?;
        let mut out = Vec::new();
        std::io::Read::read_to_end(&mut r.into_reader(), &mut out)?;
        Ok(out)
    }

    /// Send image `hash` from `home`: only the chunks the hub lacks, then
    /// the manifest, which the hub checks against the hash.
    pub fn push(&self, home: &Home, hash: &str) -> Result<Moved> {
        let start = Instant::now();
        let path = home.rootfs(hash);
        if !path.is_file() {
            bail!("{} is not in this Flow home ({}); import it first", &hash[..12.min(hash.len())], path.display());
        }
        let m = images::scan_file(&path, home.image_arch(hash), |_, _| Ok(()))?;
        if m.hash != hash {
            bail!("{} holds an image named {}, not {}: the file was changed", path.display(), &m.hash[..12], &hash[..12]);
        }
        let mut missing: HashSet<String> = HashSet::new();
        for ask in m.needed().chunks(50_000) {
            let url = format!("{}/flow/images/missing", self.base);
            let r: serde_json::Value = self
                .agent
                .post(&url)
                .set("Authorization", &self.auth())
                .send_json(serde_json::json!({ "chunks": ask }))
                .map_err(|e| Self::fail(&url, e))?
                .into_json()?;
            missing.extend(r["missing"].as_array().into_iter().flatten().filter_map(|h| h.as_str().map(String::from)));
        }
        // Read the image again, sending the missing chunks as they come.
        let (count, bytes) = (missing.len(), Mutex::new(0u64));
        let queue: Mutex<Vec<(String, Vec<u8>)>> = Mutex::new(Vec::new());
        let failed: Mutex<Option<anyhow::Error>> = Mutex::new(None);
        let sent = Mutex::new(HashSet::new());
        let send = |c: &str, b: &[u8]| -> Result<()> {
            let url = format!("{}/flow/images/chunks/{c}", self.base);
            self.agent.put(&url).set("Authorization", &self.auth()).send_bytes(b).map_err(|e| Self::fail(&url, e))?;
            *bytes.lock().unwrap() += b.len() as u64;
            Ok(())
        };
        images::scan_file(&path, None, |c: &Chunk, b: &[u8]| {
            if missing.contains(c.hash()) && sent.lock().unwrap().insert(c.hash().to_string()) {
                queue.lock().unwrap().push((c.hash().to_string(), b.to_vec()));
            }
            if queue.lock().unwrap().len() >= PARALLEL * 4 {
                let batch: Vec<_> = std::mem::take(&mut *queue.lock().unwrap());
                send_batch(&batch, &send, &failed);
            }
            match failed.lock().unwrap().take() {
                Some(e) => Err(images::Error::Bad(format!("{e:#}"))),
                None => Ok(()),
            }
        })?;
        let rest: Vec<_> = std::mem::take(&mut *queue.lock().unwrap());
        send_batch(&rest, &send, &failed);
        if let Some(e) = failed.lock().unwrap().take() {
            return Err(e);
        }
        let url = format!("{}/flow/images/{hash}", self.base);
        // The hub reads the whole image before it answers, which on a busy
        // disk takes far longer than the agent's 120 s (oxygenhub#341).
        self.agent
            .put(&url)
            .timeout(Duration::from_secs(3600))
            .set("Authorization", &self.auth())
            .send_json(&m)
            .map_err(|e| Self::fail(&url, e))?;
        let bytes = *bytes.lock().unwrap();
        Ok(Moved { chunks: count, bytes, peer_bytes: 0, elapsed: start.elapsed() })
    }

    /// Fetch image `hash` into `home`: the chunks its cache lacks, the
    /// image written and checked, its architecture recorded. `Ok(None)`
    /// when the hub does not hold it.
    pub fn fetch(&self, home: &Home, hash: &str) -> Result<Option<Moved>> {
        let start = Instant::now();
        let Some(m) = self.manifest(hash)? else { return Ok(None) };
        if m.hash != hash {
            bail!("the hub's manifest for {} names {}", &hash[..12], &m.hash[..12]);
        }
        let cache = chunk_cache(home);
        save_manifest(&home.root, &m)?;
        let wanted: Vec<String> = m.needed().into_iter().filter(|h| !cache_path(&cache, h).is_file()).map(String::from).collect();
        let (bytes, failed) = (Mutex::new(0u64), Mutex::new(None::<anyhow::Error>));
        let next = Mutex::new(wanted.iter());
        std::thread::scope(|s| {
            for _ in 0..PARALLEL {
                s.spawn(|| loop {
                    let Some(h) = next.lock().unwrap().next() else { break };
                    if failed.lock().unwrap().is_some() {
                        break;
                    }
                    let got = self.chunk(h).and_then(|b| {
                        if blake3::hash(&b).to_hex().as_str() != h {
                            bail!("chunk {} came back with other bytes", &h[..12]);
                        }
                        let p = cache_path(&cache, h);
                        std::fs::create_dir_all(p.parent().expect("a chunk directory"))?;
                        let tmp = p.with_extension("part");
                        std::fs::write(&tmp, &b)?;
                        std::fs::rename(&tmp, &p)?;
                        *bytes.lock().unwrap() += b.len() as u64;
                        Ok(())
                    });
                    if let Err(e) = got {
                        *failed.lock().unwrap() = Some(e);
                        break;
                    }
                });
            }
        });
        if let Some(e) = failed.into_inner().unwrap() {
            return Err(e);
        }
        write_image(home, &m)?;
        Ok(Some(Moved { chunks: wanted.len(), bytes: bytes.into_inner().unwrap(), peer_bytes: 0, elapsed: start.elapsed() }))
    }
}

/// Write image `m` into `home` from the chunk cache, checked against its
/// hash, and record its architecture.
pub(crate) fn write_image(home: &Home, m: &Manifest) -> Result<()> {
    let cache = chunk_cache(home);
    let out = home.rootfs(&m.hash);
    std::fs::create_dir_all(out.parent().expect("the cas directory"))?;
    images::assemble(m, &out, |c| Ok(std::fs::read(cache_path(&cache, c.hash()))?))
        .with_context(|| format!("write image {}", &m.hash[..12]))?;
    if let Some(arch) = &m.arch {
        let manifest = serde_json::json!({ "arch": arch, "imported_ms": std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_millis() as u64).unwrap_or(0), "from": "hub" });
        std::fs::write(home.image_manifest(&m.hash), serde_json::to_vec(&manifest)?)?;
    }
    Ok(())
}

impl Client {
    /// Tell the hub which images this home holds, and send the manifest of
    /// each it keeps none for (oxygenflow#94, jetstream.md §8.1.1): the
    /// swarm can then carry an image nobody pushed. A manifest not kept here
    /// yet is made by reading the image once, and kept. How many were sent;
    /// none from a hub that predates it.
    pub fn offer_held(&self, home: &Home) -> Result<usize> {
        let images = home.images();
        if images.is_empty() {
            return Ok(0);
        }
        let url = format!("{}/flow/images/held", self.base);
        let wanted: flow_core::api::WantedManifests =
            match self.agent.post(&url).set("Authorization", &self.auth()).send_json(flow_core::api::HeldImages { images }) {
                Ok(r) => r.into_json()?,
                // A hub that predates it: no such route (404), or a route
                // that takes other methods only (405).
                Err(ureq::Error::Status(404 | 405, _)) => return Ok(0),
                Err(e) => return Err(Self::fail(&url, e)),
            };
        let mut sent = 0;
        for hash in wanted.wanted.iter().filter(|h| images::is_hash(h)) {
            let path = home.rootfs(hash);
            if !path.is_file() {
                continue;
            }
            let kept = std::fs::read(manifest_path(&home.root, hash)).ok().and_then(|b| serde_json::from_slice::<Manifest>(&b).ok());
            let m = match kept.filter(|m| m.hash == *hash) {
                Some(m) => m,
                None => {
                    let m = images::scan_file(&path, home.image_arch(hash), |_, _| Ok(()))?;
                    if m.hash != *hash {
                        continue;
                    }
                    save_manifest(&home.root, &m)?;
                    m
                }
            };
            let url = format!("{}/flow/images/{hash}/held", self.base);
            self.agent.put(&url).set("Authorization", &self.auth()).send_json(&m).map_err(|e| Self::fail(&url, e))?;
            sent += 1;
        }
        Ok(sent)
    }
}

/// Where an image's chunk manifest is kept (`manifests/<hash>.json`), so
/// the runner can serve its chunks to peers out of the image itself.
pub(crate) fn manifest_path(root: &Path, hash: &str) -> PathBuf {
    root.join("manifests").join(format!("{hash}.json"))
}

pub(crate) fn save_manifest(root: &Path, m: &Manifest) -> Result<()> {
    let p = manifest_path(root, &m.hash);
    std::fs::create_dir_all(p.parent().expect("the manifests directory"))?;
    let tmp = p.with_extension("part");
    std::fs::write(&tmp, serde_json::to_vec(m)?)?;
    std::fs::rename(&tmp, &p)?;
    Ok(())
}

fn send_batch(batch: &[(String, Vec<u8>)], send: &(dyn Fn(&str, &[u8]) -> Result<()> + Sync), failed: &Mutex<Option<anyhow::Error>>) {
    std::thread::scope(|s| {
        for part in batch.chunks(batch.len().div_ceil(PARALLEL).max(1)) {
            s.spawn(move || {
                for (h, b) in part {
                    if let Err(e) = send(h, b) {
                        *failed.lock().unwrap() = Some(e);
                        return;
                    }
                }
            });
        }
    });
}

/// The home's chunk cache.
pub fn chunk_cache(home: &Home) -> PathBuf {
    home.root.join("chunks")
}

pub(crate) fn cache_path(cache: &Path, hash: &str) -> PathBuf {
    cache.join(&hash[..2]).join(hash)
}

/// A size for people: 991 MiB, 1.2 GiB.
pub fn human(bytes: u64) -> String {
    if bytes >= 1 << 30 {
        format!("{:.1} GiB", bytes as f64 / (1u64 << 30) as f64)
    } else {
        format!("{} MiB", bytes >> 20)
    }
}