open-standards/wai/wai-ledger/src/ledger.rs

//! The ledger's state: each tariff with its held sources and its chain of invoices
//! (extensions/receipt-ledger.md §6.9, §9). 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); of them, a source may not bring again
    /// those its tariff's version bills once ([`rl::billed_once`]).
    pub billed: BTreeSet<(u8, [u8; 32])>,
}

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

    /// Whether the next invoice bills `h`: a source the chain has not billed, or, under a tariff
    /// of version 4 or 5, 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.invoices.last().is_some_and(|(i, _, _)| h.facts.heads().iter().all(|r| i.records.contains(r)))
        };
        !self.billed(h) || held()
    }
}

pub struct Ledger {
    dir: Dir,
    key: SigningKey,
    pub chains: BTreeMap<[u8; 32], Chain>,
}

/// 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::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();
        for d in dirs {
            let at = d.display().to_string();
            let tb = std::fs::read(d.join("tariff.wlt")).map_err(|e| 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"));
            }
            let mut chain = Chain { tariff, sources: Vec::new(), invoices: Vec::new(), billed: BTreeSet::new() };
            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 = [0u8; 32];
            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 })
    }

    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.
    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) {
            write_whole(&self.dir.tariff_dir(&hex(&d)).join("tariff.wlt"), bytes).map_err(store)?;
            self.chains.insert(d, Chain { tariff: t, sources: Vec::new(), invoices: Vec::new(), billed: BTreeSet::new() });
        }
        Ok(d)
    }

    /// `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)?;
        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)?;
        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.invoices.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.invoices.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),
            "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(),
        }))
    }
}