//! 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;

/// Chunk names per `missing` question.

const ASK: usize = 20_000;

/// 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();

        // 20,000 names is about 1.3 MB of JSON: under a hub's 2 MB default

        // body limit, which refused 50,000 at a time (oxygenhub#147).

        for ask in m.needed().chunks(ASK) {

            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)

    }

}