//! The ledger's state: each tariff with its held sources and its chain of invoices
//! (extensions/receipt-ledger.md §6.9, §9). A tariff that follows another (§4.14) continues that
//! tariff's chain, which it closes. Every change is written to the directory before it is
//! answered, and [`Ledger::open`] rebuilds the state from the directory, verifying again every
//! tariff, source and invoice it reads.

use std::collections::{BTreeMap, BTreeSet};
use std::path::Path;

use ed25519_dalek::SigningKey;
use serde_json::{json, Value};
use wai::receipt_ledger::{self as rl, Facts, Invoice, Reason, Refusal, Tariff};

use crate::store::{name, numbered, write_whole, Dir};

pub fn hex(b: &[u8]) -> String {
    b.iter().map(|x| format!("{x:02x}")).collect()
}

pub fn unhex(s: &str) -> Option<Vec<u8>> {
    if !s.len().is_multiple_of(2) || !s.bytes().all(|c| c.is_ascii_digit() || (b'a'..=b'f').contains(&c)) {
        return None;
    }
    (0..s.len()).step_by(2).map(|i| u8::from_str_radix(&s[i..i + 2], 16).ok()).collect()
}

/// Why a request was not done: a refusal with its reason and place, or a failure to write.
#[derive(Debug)]
pub enum Error {
    Refused(Refusal),
    Store(String),
}

impl From<Refusal> for Error {
    fn from(r: Refusal) -> Error {
        Error::Refused(r)
    }
}

fn refuse(code: &'static str, at: Value) -> Error {
    Error::Refused((code, at))
}

fn store(e: String) -> Error {
    Error::Store(e)
}

/// A source the ledger holds: its document and what its verification gave.
#[derive(Clone)]
pub struct Held {
    pub doc: Value,
    pub facts: Facts,
}

/// One tariff and what is held under it.
pub struct Chain {
    pub tariff: Tariff,
    pub sources: Vec<Held>,
    /// The chain, in order: each invoice's bytes and its verification document.
    pub invoices: Vec<(Invoice, Vec<u8>, Value)>,
    /// The records of every invoice of the chain (§6.9), the chain a following tariff continues
    /// among them (§4.14); of them, a source may not bring again those its tariff's version bills
    /// once ([`rl::billed_once`]).
    pub billed: BTreeSet<(u8, [u8; 32])>,
    /// Under a tariff that follows another (§4.14): the last invoice of the chain it continues,
    /// and its bytes, which its first invoice names.
    pub head: Option<(Invoice, Vec<u8>)>,
    /// The tariff that follows this one, which closes it (§9).
    pub followed_by: Option<[u8; 32]>,
}

impl Chain {
    fn of(tariff: Tariff) -> Chain {
        Chain { tariff, sources: Vec::new(), invoices: Vec::new(), billed: BTreeSet::new(), head: None, followed_by: None }
    }

    fn billed(&self, h: &Held) -> bool {
        h.facts.heads().iter().all(|r| self.billed.contains(r))
    }

    /// The invoice the next one names: the chain's last, or, before its first under a tariff
    /// that follows another, the last of the chain it continues (§4.14).
    fn last(&self) -> Option<(&Invoice, &Vec<u8>)> {
        self.invoices.last().map(|(i, b, _)| (i, b)).or(self.head.as_ref().map(|(i, b)| (i, b)))
    }

    /// Whether the next invoice bills `h`: a source the chain has not billed, or, under a tariff
    /// of version 4 or later, a package the last invoice holds, whose storage each period bills
    /// (§4.7).
    fn bills(&self, h: &Held) -> bool {
        let held = || matches!(h.facts, Facts::Package(_)) && rl::bills_spending(self.tariff.version) && self.last().is_some_and(|(i, _)| h.facts.heads().iter().all(|r| i.records.contains(r)));
        !self.billed(h) || held()
    }

    /// A closed tariff takes no source and no invoice: its chain continues under the tariff that
    /// follows it (§4.14, §9).
    fn open_for_more(&self) -> Result<(), Error> {
        match self.followed_by {
            Some(_) => Err(refuse(Reason::TariffMismatch.code(), json!(["tariff"]))),
            None => Ok(()),
        }
    }
}

/// What a following tariff takes over (§4.14): the last invoice of the chain it continues and its
/// bytes, the records that chain billed, and the sources it holds that are billed next.
type Continued = ((Invoice, Vec<u8>), BTreeSet<(u8, [u8; 32])>, Vec<Held>);

pub struct Ledger {
    dir: Dir,
    key: SigningKey,
    pub chains: BTreeMap<[u8; 32], Chain>,
    /// Tariff directories this open passed over because they hold no tariff
    /// and no invoice: an `add_tariff` that stopped after writing the sources
    /// a following tariff carries and before the tariff itself (§4.14). Named
    /// rather than passed over in silence, so a restart that found one says
    /// so.
    pub incomplete: Vec<String>,
}

/// An issued invoice: its bytes, its verification document and the outcome a verifier gives it.
pub struct Issued {
    pub invoice: Vec<u8>,
    pub document: Value,
    pub outcome: Value,
}

impl Ledger {
    /// Open the directory and rebuild the state from it.
    pub fn open(root: &Path, key: SigningKey) -> Result<Ledger, String> {
        let dir = Dir::open(root)?;
        let mut chains: BTreeMap<[u8; 32], Chain> = BTreeMap::new();
        let rd = std::fs::read_dir(dir.root.join("tariffs")).map_err(|e| e.to_string())?;
        let mut dirs: Vec<_> = rd.flatten().map(|e| e.path()).filter(|p| p.is_dir()).collect();
        dirs.sort();
        let mut tariffs = Vec::with_capacity(dirs.len());
        let mut incomplete = Vec::new();
        for d in dirs {
            let at = d.display().to_string();
            let tb = match std::fs::read(d.join("tariff.wlt")) {
                Ok(bytes) => bytes,
                Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
                    // A tariff that follows another carries the sources of the
                    // chain it continues, and `add_tariff` writes those before
                    // the `tariff.wlt` that makes this directory a tariff. A
                    // stop between the two leaves exactly this: a directory
                    // with sources and no tariff. Nothing is billed under a
                    // tariff that was never written, so nothing here is lost,
                    // and adding the tariff again completes it. Refusing would
                    // refuse the whole ledger -- every chain, every invoice --
                    // over a tariff it never had, and the service would not
                    // start again to be given it.
                    let invoices = numbered(&d.join("invoices"), "wli")?;
                    if !invoices.is_empty() {
                        return Err(format!(
                            "{at}: it holds {} invoice(s) and no tariff.wlt, which no stop leaves: \
                             the tariff is written before any invoice can be issued under it",
                            invoices.len()
                        ));
                    }
                    incomplete.push(at);
                    continue;
                }
                Err(e) => return Err(format!("{at}: {e}")),
            };
            let tariff = Tariff::parse(&tb).map_err(|(r, w)| format!("{at}: the tariff is refused {r} at {w}"))?;
            if tariff.seller != key.verifying_key().to_bytes() {
                return Err(format!("{at}: the tariff's seller is not this ledger's key"));
            }
            tariffs.push((d, tariff));
        }
        // Each chain after the chain it continues (§4.14): a tariff that follows another is
        // rebuilt once that one is, from the last invoice and the records of its chain.
        while !tariffs.is_empty() {
            let Some(k) = tariffs.iter().position(|(_, t)| t.followed().is_none_or(|f| chains.contains_key(&f))) else {
                return Err(format!("{}: the tariff it follows is not in the directory", tariffs[0].0.display()));
            };
            let (d, tariff) = tariffs.remove(k);
            let at = d.display().to_string();
            let mut chain = Chain::of(tariff);
            if let Some(f) = chain.tariff.followed() {
                let followed = chains.get_mut(&f).expect("rebuilt first");
                let Some((i, b, _)) = followed.invoices.last() else {
                    return Err(format!("{at}: the chain it follows has no invoice"));
                };
                if followed.followed_by.is_some() {
                    return Err(format!("{at}: the tariff it follows is followed twice"));
                }
                chain.head = Some((i.clone(), b.clone()));
                chain.billed = followed.billed.clone();
                followed.followed_by = Some(chain.tariff.digest());
            }
            for (_, p) in numbered(&d.join("sources"), "json")? {
                let text = std::fs::read_to_string(&p).map_err(|e| format!("{}: {e}", p.display()))?;
                let doc: Value = serde_json::from_str(&text).map_err(|e| format!("{}: {e}", p.display()))?;
                let facts = rl::verify_source_of(&doc, &rl::signers(&chain.tariff), chain.tariff.version).map_err(|(r, w)| format!("{}: refused {r} at {w}", p.display()))?;
                chain.sources.push(Held { doc, facts });
            }
            let mut prev = chain.head.as_ref().map_or([0u8; 32], |(i, _)| i.digest());
            for (n, p) in numbered(&d.join("invoices"), "wli")? {
                let b = std::fs::read(&p).map_err(|e| format!("{}: {e}", p.display()))?;
                let inv = Invoice::parse(&b).map_err(|r| format!("{}: refused {r}", p.display()))?;
                if inv.prev != prev || inv.tariff != chain.tariff.digest() {
                    return Err(format!("{}: the chain is broken", p.display()));
                }
                let dp = d.join("invoices").join(name(n, "json"));
                let doc: Value = std::fs::read_to_string(&dp).map_err(|e| format!("{}: {e}", dp.display())).and_then(|t| serde_json::from_str(&t).map_err(|e| format!("{}: {e}", dp.display())))?;
                rl::verify(&doc).map_err(|(r, w)| format!("{}: refused {r} at {w}", dp.display()))?;
                prev = inv.digest();
                chain.billed.extend(inv.records.iter().copied());
                chain.invoices.push((inv, b, doc));
            }
            chains.insert(chain.tariff.digest(), chain);
        }
        Ok(Ledger { dir, key, chains, incomplete })
    }

    pub fn seller(&self) -> [u8; 32] {
        self.key.verifying_key().to_bytes()
    }

    fn chain(&self, tariff: &[u8; 32]) -> Result<&Chain, Error> {
        self.chains.get(tariff).ok_or(refuse("record_missing", json!(["tariff"])))
    }

    /// `POST /v1/tariffs`: a tariff whose seller is this ledger, both signatures verified. A
    /// tariff that follows another (§4.14) continues that tariff's chain and closes it.
    pub fn add_tariff(&mut self, bytes: &[u8]) -> Result<[u8; 32], Error> {
        let t = Tariff::parse(bytes)?;
        if t.seller != self.seller() {
            return Err(refuse(Reason::TariffMismatch.code(), json!(["tariff", "seller"])));
        }
        let d = t.digest();
        if self.chains.contains_key(&d) {
            return Ok(d);
        }
        let mut chain = Chain::of(t);
        if let Some(f) = chain.tariff.followed() {
            let (head, billed, carried) = self.continued(&chain.tariff, &f)?;
            (chain.head, chain.billed, chain.sources) = (Some(head), billed, carried);
        }
        let td = self.dir.tariff_dir(&hex(&d));
        for (n, h) in chain.sources.iter().enumerate() {
            write_whole(&td.join("sources").join(name(n as u64, "json")), serde_json::to_string(&h.doc).expect("JSON").as_bytes()).map_err(store)?;
        }
        write_whole(&td.join("tariff.wlt"), bytes).map_err(store)?;
        if let Some(f) = chain.tariff.followed() {
            self.chains.get_mut(&f).expect("the chain it follows").followed_by = Some(d);
        }
        self.chains.insert(d, chain);
        Ok(d)
    }

    /// What a tariff `t` that follows the tariff `f` continues (§4.14, §9): the last invoice of
    /// `f`'s chain, the records its chain billed, and every source its next invoice would bill,
    /// each verified again under `t` (§7.2 steps 5 and 6). A package that chain holds whose
    /// packager `t` does not name is left behind, billed no more; any other source `t` refuses
    /// refuses `t`, at the source's place, so the seller bills it under `f` first.
    fn continued(&self, t: &Tariff, f: &[u8; 32]) -> Result<Continued, Error> {
        let at = || json!(["tariff", "follows"]);
        let c = self.chains.get(f).ok_or(refuse("record_missing", at()))?;
        if c.tariff.buyer != t.buyer || c.followed_by.is_some() {
            return Err(refuse(Reason::TariffMismatch.code(), at()));
        }
        let (last, bytes) = c.invoices.last().map(|(i, b, _)| (i.clone(), b.clone())).ok_or(refuse("record_missing", at()))?;
        let place = |k: usize, rest: Value| {
            let mut a = vec![json!("tariff"), json!("follows"), json!("source"), json!(k)];
            a.extend(rest.as_array().cloned().unwrap_or_default());
            Value::Array(a)
        };
        let mut carried: Vec<Held> = Vec::new();
        let mut from: Vec<usize> = Vec::new();
        for (k, h) in c.sources.iter().enumerate().filter(|(_, h)| c.bills(h)) {
            let held = matches!(h.facts, Facts::Package(_)) && c.billed(h);
            let verified = rl::verify_source_of(&h.doc, &rl::signers(t), t.version).and_then(|facts| rl::check_source(t, t.start, t.end, &facts).map(|_| facts).map_err(|r| (r.code(), json!([]))));
            // A source whose records the following tariff would no longer bill, a stitched record of
            // a retired key among them, is outside its scope (§4.4).
            let verified = verified.and_then(|facts| if facts.heads() == h.facts.heads() { Ok(facts) } else { Err((Reason::OutsideScope.code(), json!([]))) });
            match verified {
                Ok(facts) => {
                    carried.push(Held { doc: h.doc.clone(), facts });
                    from.push(k);
                }
                Err(_) if held => {}
                Err((r, rest)) => return Err(refuse(r, place(k, rest))),
            }
        }
        let facts: Vec<Facts> = carried.iter().map(|h| h.facts.clone()).collect();
        rl::check_across(t.version, &facts, None).map_err(|(k, r)| refuse(r, place(from[k], json!([]))))?;
        Ok(((last, bytes), c.billed.clone(), carried))
    }

    /// `POST /v1/sources`: the source verified (§7.2 step 5, against the tariff's own period)
    /// and checked against everything held under the tariff (step 6), then kept.
    pub fn add_source(&mut self, tariff: &[u8; 32], doc: Value) -> Result<Vec<(u8, [u8; 32])>, Error> {
        let c = self.chain(tariff)?;
        c.open_for_more()?;
        let facts = rl::verify_source_of(&doc, &rl::signers(&c.tariff), c.tariff.version).map_err(|(r, at)| {
            let mut a = vec![json!("source")];
            a.extend(at.as_array().cloned().unwrap_or_default());
            Error::Refused((r, Value::Array(a)))
        })?;
        rl::check_source(&c.tariff, c.tariff.start, c.tariff.end, &facts).map_err(|r| refuse(r.code(), json!(["source"])))?;
        let mut all: Vec<Facts> = c.sources.iter().map(|h| h.facts.clone()).collect();
        all.push(facts.clone());
        rl::check_across(c.tariff.version, &all, None).map_err(|(_, r)| refuse(r, json!(["source"])))?;
        // A record the chain billed is billed never again; under a tariff of versions 3 to 5, a
        // job's output receipt is cited again by a re-run, whose hop and route record are new
        // (§6.9). A package source is brought once: the ledger holds it, and under versions 4 and
        // 5 bills its storage each period from what it holds (§4.7).
        let package_again = matches!(facts, Facts::Package(_)) && facts.heads().iter().all(|h| c.billed.contains(h));
        if package_again || facts.records().iter().any(|r| rl::billed_once(c.tariff.version, r.0) && c.billed.contains(r)) {
            return Err(refuse(Reason::RecordRebilled.code(), json!(["source"])));
        }
        let n = c.sources.len() as u64;
        let path = self.dir.tariff_dir(&hex(tariff)).join("sources").join(name(n, "json"));
        let text = serde_json::to_string(&doc).expect("JSON");
        write_whole(&path, text.as_bytes()).map_err(store)?;
        let heads = facts.heads();
        self.chains.get_mut(tariff).expect("the chain").sources.push(Held { doc, facts });
        Ok(heads)
    }

    /// `POST /v1/invoices`: the invoice of the period over the sources held and not yet billed
    /// that lie in it (a package always does), chained, signed, verified and kept.
    pub fn invoice(&mut self, tariff: &[u8; 32], start: u64, end: u64) -> Result<Issued, Error> {
        let c = self.chain(tariff)?;
        c.open_for_more()?;
        let t = &c.tariff;
        let selected: Vec<Value> = c.sources.iter().filter(|h| c.bills(h) && rl::check_source(t, start, end, &h.facts).is_ok()).map(|h| h.doc.clone()).collect();
        let prev = c.last().map(|(i, _)| i);
        let inv = rl::make_invoice(&self.key, t, prev, start, end, &selected)?;
        let bytes = inv.to_bytes();
        let document = json!({
            "tariff": hex(&t.to_bytes()),
            "invoice": hex(&bytes),
            "previous": c.last().map(|(_, b)| hex(b)),
            "sources": selected,
        });
        // What a buyer will check: the document verifies, here, before it is kept.
        let outcome = rl::outcome_json(&rl::verify(&document));
        if outcome["outcome"] != "accept" {
            return Err(store(format!("the invoice computed does not verify: {outcome}")));
        }
        let n = c.invoices.len() as u64;
        let d = self.dir.tariff_dir(&hex(tariff)).join("invoices");
        write_whole(&d.join(name(n, "json")), serde_json::to_string(&document).expect("JSON").as_bytes()).map_err(store)?;
        write_whole(&d.join(name(n, "wli")), &bytes).map_err(store)?;
        let c = self.chains.get_mut(tariff).expect("the chain");
        c.billed.extend(inv.records.iter().copied());
        c.invoices.push((inv, bytes.clone(), document.clone()));
        Ok(Issued { invoice: bytes, document, outcome })
    }

    /// `GET /v1/invoices/{digest}`: the invoice and its verification document.
    pub fn find_invoice(&self, digest: &[u8; 32]) -> Option<(&Vec<u8>, &Value)> {
        self.chains.values().flat_map(|c| c.invoices.iter()).find(|(i, _, _)| i.digest() == *digest).map(|(_, b, d)| (b, d))
    }

    /// `GET /v1/tariffs/{digest}`: the tariff, its chain and what it holds.
    pub fn describe_tariff(&self, digest: &[u8; 32]) -> Option<Value> {
        let c = self.chains.get(digest)?;
        let t = &c.tariff;
        Some(json!({
            "tariff": hex(digest),
            "seller": hex(&t.seller),
            "buyer": hex(&t.buyer),
            "start": t.start,
            "end": t.end,
            "currency": String::from_utf8_lossy(&t.currency),
            "follows": t.followed().map(|f| hex(&f)),
            "followed_by": c.followed_by.map(|f| hex(&f)),
            "invoices": c.invoices.iter().map(|(i, _, _)| json!({"digest": hex(&i.digest()), "start": i.start, "end": i.end, "amount": i.amount})).collect::<Vec<_>>(),
            "sources_held": c.sources.len(),
            "sources_unbilled": c.sources.iter().filter(|h| !c.billed(h)).count(),
        }))
    }
}