//! The hub's transparency log (oxygenhub#238, docs/transparency.md, Oxygen
//! spec §14.8): one entry per signed object the hub accepts, in the order
//! it accepts them, kept as append-only files and served as C2SP
//! tlog-tiles with a signed checkpoint, so anyone can check the hub shows
//! everyone the same history.
//!
//! Files in `<data>/tlog/`:
//! - `entries`: each entry as a big-endian u16 length and its bytes;
//! - `offsets`: each entry's offset in `entries`, a u64 (little-endian);
//! - `hashes`: every hash of the tree in stored order (Oxygen spec §14.8);
//! - `checkpoint`: the latest signed checkpoint note (replaced, not
//!   appended, as it is only a signature over the others).
//!
//! Nothing is rewritten: a crash mid-append leaves at most a tail that is
//! cut back to the last whole entry on open, and the hashes are rebuilt
//! from the entries. The database's `tlog_entries` points from objects to
//! entry numbers.

use crate::{db, Shared};
use axum::extract::{Path as UrlPath, State};
use axum::http::{header, StatusCode};
use axum::response::{IntoResponse, Response};
use oxy_core::hash::{KeyId, ObjectRef};
use oxy_core::repo::{PushReport, Repo};
use oxy_core::tlog::{self, Checkpoint, Entry, Hash, HashReader, Tile, TlogError, TILE_WIDTH};
use std::fs::{File, OpenOptions};
use std::io::Write;
use std::os::unix::fs::FileExt;
use std::path::{Path, PathBuf};
use std::sync::Mutex;

/// How often a new checkpoint is signed while entries arrive.
pub const CHECKPOINT_EVERY: std::time::Duration = std::time::Duration::from_secs(5);

struct Files {
    entries: File,
    offsets: File,
    hashes: File,
    /// Entries in the log.
    size: u64,
    /// Bytes in `entries`.
    entries_len: u64,
}

/// The stored hashes in a file, by stored index.
struct FileHashes<'a>(&'a File);

impl HashReader for FileHashes<'_> {
    fn read_hashes(&self, indexes: &[u64]) -> Result<Vec<Hash>, TlogError> {
        indexes
            .iter()
            .map(|&i| {
                let mut h = [0u8; 32];
                self.0.read_exact_at(&mut h, i * 32).map_err(|e| TlogError::Read(format!("stored hash {i}: {e}")))?;
                Ok(h)
            })
            .collect()
    }
}

/// The log on disk.
pub struct Log {
    dir: PathBuf,
    /// The log's name in its checkpoints: the hub's host and `/log`.
    pub origin: String,
    files: Mutex<Files>,
    /// The latest signed checkpoint: its size and its note.
    checkpoint: Mutex<Option<(u64, String)>>,
}

fn read_u64_at(f: &File, at: u64) -> std::io::Result<u64> {
    let mut b = [0u8; 8];
    f.read_exact_at(&mut b, at)?;
    Ok(u64::from_le_bytes(b))
}

/// The origin of a hub's log: its host (and port) and `/log`.
pub fn origin_of(base_url: &str) -> String {
    let host = base_url.split_once("://").map(|(_, h)| h).unwrap_or(base_url);
    format!("{}/log", host.trim_end_matches('/'))
}

impl Log {
    /// Open (or start) the log in `dir`, recovering from an interrupted
    /// append.
    pub fn open(dir: &Path, origin: &str) -> anyhow::Result<Log> {
        std::fs::create_dir_all(dir)?;
        let open = |name: &str| OpenOptions::new().read(true).append(true).create(true).open(dir.join(name));
        let (entries, offsets, hashes) = (open("entries")?, open("offsets")?, open("hashes")?);
        let entries_on_disk = entries.metadata()?.len();
        // The whole entries: those whose offset was written and whose
        // bytes all were.
        let mut size = offsets.metadata()?.len() / 8;
        let mut entries_len = 0;
        while size > 0 {
            let off = read_u64_at(&offsets, (size - 1) * 8)?;
            let mut len = [0u8; 2];
            if off + 2 <= entries_on_disk && entries.read_exact_at(&mut len, off).is_ok() {
                let end = off + 2 + u16::from_be_bytes(len) as u64;
                if end <= entries_on_disk {
                    entries_len = end;
                    break;
                }
            }
            size -= 1;
        }
        offsets.set_len(size * 8)?;
        entries.set_len(entries_len)?;
        // The hashes of every leaf whose hashes were all written; the rest
        // are computed again from the entries.
        let have = hashes.metadata()?.len() / 32;
        let mut whole = size;
        while tlog::stored_hash_count(whole) > have {
            whole -= 1;
        }
        hashes.set_len(tlog::stored_hash_count(whole) * 32)?;
        let mut files = Files { entries, offsets, hashes, size: whole, entries_len };
        for n in whole..size {
            let bytes = Self::entry_bytes(&files, n)?;
            let add = tlog::stored_hashes_for_leaf(n, tlog::leaf_hash(&bytes), &FileHashes(&files.hashes))?;
            for h in add {
                files.hashes.write_all(&h)?;
            }
            files.size = n + 1;
        }
        files.size = size;
        files.hashes.sync_data()?;
        let checkpoint = std::fs::read_to_string(dir.join("checkpoint")).ok().and_then(|note| {
            let n = tlog::Note::parse(&note).ok()?;
            let c = Checkpoint::parse(&n.text).ok()?;
            (c.size <= size).then_some((c.size, note))
        });
        Ok(Log { dir: dir.to_path_buf(), origin: origin.to_string(), files: Mutex::new(files), checkpoint: Mutex::new(checkpoint) })
    }

    fn entry_bytes(f: &Files, n: u64) -> anyhow::Result<Vec<u8>> {
        let off = read_u64_at(&f.offsets, n * 8)?;
        let mut len = [0u8; 2];
        f.entries.read_exact_at(&mut len, off)?;
        let mut b = vec![0u8; u16::from_be_bytes(len) as usize];
        f.entries.read_exact_at(&mut b, off + 2)?;
        Ok(b)
    }

    /// Entries in the log.
    pub fn size(&self) -> u64 {
        self.files.lock().unwrap().size
    }

    /// Append entries, durably; their numbers.
    pub fn append(&self, entries: &[Entry]) -> anyhow::Result<Vec<u64>> {
        let mut f = self.files.lock().unwrap();
        let mut out = Vec::with_capacity(entries.len());
        for e in entries {
            let bytes = e.encode();
            let n = f.size;
            let add = tlog::stored_hashes_for_leaf(n, tlog::leaf_hash(&bytes), &FileHashes(&f.hashes))?;
            let off = f.entries_len;
            let mut rec = (bytes.len() as u16).to_be_bytes().to_vec();
            rec.extend_from_slice(&bytes);
            f.entries.write_all(&rec)?;
            f.entries_len += rec.len() as u64;
            f.offsets.write_all(&off.to_le_bytes())?;
            let mut hs = Vec::with_capacity(add.len() * 32);
            for h in add {
                hs.extend_from_slice(&h);
            }
            f.hashes.write_all(&hs)?;
            f.size = n + 1;
            out.push(n);
        }
        f.entries.sync_data()?;
        f.offsets.sync_data()?;
        f.hashes.sync_data()?;
        Ok(out)
    }

    /// The root of the first `n` entries.
    pub fn tree_hash(&self, n: u64) -> anyhow::Result<Hash> {
        let f = self.files.lock().unwrap();
        Ok(tlog::tree_hash(n, &FileHashes(&f.hashes))?)
    }

    /// Entry `n`'s inclusion proof in the tree of `t` entries.
    pub fn prove_inclusion(&self, t: u64, n: u64) -> anyhow::Result<Vec<Hash>> {
        let f = self.files.lock().unwrap();
        Ok(tlog::prove_inclusion(t, n, &FileHashes(&f.hashes))?)
    }

    /// The consistency proof from `n` entries to `t`.
    pub fn prove_consistency(&self, t: u64, n: u64) -> anyhow::Result<Vec<Hash>> {
        let f = self.files.lock().unwrap();
        Ok(tlog::prove_consistency(t, n, &FileHashes(&f.hashes))?)
    }

    /// A tile's hashes, when the log holds all of them.
    pub fn tile(&self, t: &Tile) -> anyhow::Result<Option<Vec<u8>>> {
        let f = self.files.lock().unwrap();
        let at_level = f.size >> (tlog::TILE_HEIGHT * t.level);
        if at_level < (t.n << tlog::TILE_HEIGHT) + t.width as u64 {
            return Ok(None);
        }
        Ok(Some(t.read_data(&FileHashes(&f.hashes))?))
    }

    /// The entry bundle beside level-0 tile `t`, when the log holds it.
    pub fn bundle(&self, t: &Tile) -> anyhow::Result<Option<Vec<u8>>> {
        let f = self.files.lock().unwrap();
        let first = t.n << tlog::TILE_HEIGHT;
        if f.size < first + t.width as u64 {
            return Ok(None);
        }
        let entries: Vec<Vec<u8>> = (first..first + t.width as u64).map(|n| Self::entry_bytes(&f, n)).collect::<anyhow::Result<_>>()?;
        Ok(Some(tlog::encode_bundle(&entries)?))
    }

    /// The latest signed checkpoint note.
    pub fn checkpoint(&self) -> Option<String> {
        self.checkpoint.lock().unwrap().as_ref().map(|(_, n)| n.clone())
    }

    /// Sign a checkpoint of the log as it is, unless the last one already
    /// covers it. True when a new one was signed.
    pub fn refresh_checkpoint(&self, identity: &oxy_core::identity::Identity, sk: &oxy_core::identity::SecretKey) -> anyhow::Result<bool> {
        let size = self.size();
        if self.checkpoint.lock().unwrap().as_ref().is_some_and(|(s, _)| *s >= size) {
            return Ok(false);
        }
        let c = Checkpoint { origin: self.origin.clone(), size, root: self.tree_hash(size)?, extensions: vec![] };
        let note = c.sign(identity, sk)?.encode();
        let tmp = self.dir.join("checkpoint.new");
        std::fs::write(&tmp, &note)?;
        File::open(&tmp)?.sync_all()?;
        std::fs::rename(&tmp, self.dir.join("checkpoint"))?;
        *self.checkpoint.lock().unwrap() = Some((size, note));
        Ok(true)
    }
}

/// An object to log.
#[derive(Clone, Debug)]
pub struct Pending {
    pub kind: &'static str,
    /// The repository's id (zero for none) and its row on this hub.
    pub repo: [u8; 16],
    pub repo_row: Option<i64>,
    pub object: ObjectRef,
    pub signers: Vec<KeyId>,
}

/// Log what is not logged yet. A failure is reported, never fatal: the
/// object stays accepted (enforcing logging before acceptance comes with
/// inclusion proofs in push responses).
pub async fn log(app: &Shared, items: Vec<Pending>) -> Vec<u64> {
    let mut fresh = Vec::new();
    for p in items {
        let object = p.object.to_hex();
        match db::tlog::entry_of(&app.db, &object, p.kind).await {
            Ok(Some(_)) => continue,
            Ok(None) => {}
            Err(e) => {
                eprintln!("tlog: looking up {} {}: {e}", p.kind, &object[..12]);
                continue;
            }
        }
        if !fresh.iter().any(|q: &Pending| q.kind == p.kind && q.object == p.object) {
            fresh.push(p);
        }
    }
    if fresh.is_empty() {
        return vec![];
    }
    let now = db::now_ms();
    let entries: Vec<Entry> = fresh
        .iter()
        .map(|p| Entry { kind: p.kind.to_string(), repo: p.repo, object: p.object, signers: p.signers.clone(), received: now as u64 })
        .collect();
    let tl = app.tlog.clone();
    let numbers = match tokio::task::spawn_blocking(move || tl.append(&entries)).await {
        Ok(Ok(n)) => n,
        Ok(Err(e)) => {
            eprintln!("tlog: appending {} entries: {e}", fresh.len());
            return vec![];
        }
        Err(e) => {
            eprintln!("tlog: appending: {e}");
            return vec![];
        }
    };
    for (p, n) in fresh.iter().zip(&numbers) {
        if let Err(e) = db::tlog::record(&app.db, *n as i64, &p.object.to_hex(), p.kind, p.repo_row, now).await {
            eprintln!("tlog: indexing entry {n}: {e}");
        }
    }
    numbers
}

fn sig_keys(sigs: &[oxy_core::identity::Signature]) -> Vec<KeyId> {
    let mut keys: Vec<KeyId> = sigs.iter().map(|s| s.key).collect();
    keys.sort();
    keys.dedup();
    keys
}

/// The certificates of short-lived keys among `signers`, to log beside
/// what they signed.
fn certificates(repo: &Repo, rid: [u8; 16], row: Option<i64>, signers: &[KeyId], out: &mut Vec<Pending>) {
    for k in signers {
        if let Ok(Some(c)) = repo.store().certificate_of(k) {
            let issuer = repo.store().get_certificate(&c).map(|sc| vec![sc.cert.issuer]).unwrap_or_default();
            out.push(Pending { kind: "certificate", repo: rid, repo_row: row, object: c, signers: issuer });
        }
    }
}

/// What a push brought that the log keeps: the attestations of the
/// references it moved (and the hub's acceptance of them), tags, seals,
/// the repository's policy, and the certificates of short-lived signers.
pub fn push_entries(repo: &Repo, row: Option<i64>, report: &PushReport) -> Vec<Pending> {
    let Ok(rid) = repo.id() else { return vec![] };
    let mut out = Vec::new();
    let add = |out: &mut Vec<Pending>, kind: &'static str, object: ObjectRef, signers: Vec<KeyId>| {
        certificates(repo, rid, row, &signers, out);
        out.push(Pending { kind, repo: rid, repo_row: row, object, signers });
    };
    for u in &report.updated {
        if u.new_state.is_zero() || !report.results.iter().any(|r| r.ok && r.name == u.name) {
            continue;
        }
        if let Some(tag) = u.name.strip_prefix("tags/") {
            if let Ok(Some((r, sa))) = repo.tag_attest(tag) {
                add(&mut out, "tag", r, sig_keys(&sa.sigs));
            }
        } else if let Ok(Some((r, sa))) = repo.latest_ref_attest(&u.name) {
            add(&mut out, "ref", r, sig_keys(&sa.sigs));
            if let Ok(Some(acc)) = repo.acceptance_of(&r) {
                add(&mut out, "accepted", acc.attest.hash(), sig_keys(&acc.sigs));
            }
        }
    }
    let mut seals = std::collections::BTreeSet::new();
    for h in &report.imported {
        for s in repo.store().seals_for(h).unwrap_or_default() {
            seals.insert(s);
        }
    }
    for s in seals {
        let signers = sig_keys(&repo.store().signatures(&s).unwrap_or_default());
        add(&mut out, "seal", s, signers);
    }
    if let Ok(Some((r, sp))) = repo.current_policy() {
        add(&mut out, "policy", r, sig_keys(&sp.sigs));
    }
    out
}

/// Sign a new checkpoint every few seconds while entries arrive.
pub async fn checkpointer(app: Shared) {
    let mut tick = tokio::time::interval(CHECKPOINT_EVERY);
    loop {
        tick.tick().await;
        let (identity, keys) = (app.hub_key.identity.clone(), app.config.keys_dir());
        let tl = app.tlog.clone();
        let done = tokio::task::spawn_blocking(move || -> anyhow::Result<bool> {
            let (_, sk) = crate::keys::load_secret(&keys)?;
            tl.refresh_checkpoint(&identity, &sk)
        })
        .await;
        if let Ok(Err(e)) = done {
            eprintln!("tlog: signing a checkpoint: {e}");
        }
    }
}

fn plain(status: StatusCode, why: &str) -> Response {
    (status, [(header::CONTENT_TYPE, "text/plain; charset=utf-8")], format!("{why}\n")).into_response()
}

/// `GET /log/checkpoint`: the latest signed checkpoint (C2SP
/// tlog-checkpoint, a signed note).
pub async fn checkpoint(State(app): State<Shared>) -> Response {
    match app.tlog.checkpoint() {
        Some(note) => ([(header::CONTENT_TYPE, "text/plain; charset=utf-8"), (header::CACHE_CONTROL, "no-cache")], note).into_response(),
        None => plain(StatusCode::SERVICE_UNAVAILABLE, "no checkpoint yet: the log signs one within seconds"),
    }
}

/// `GET /log/tile/{L}/{N}[.p/{W}]` and `/log/tile/entries/{N}[.p/{W}]`:
/// a tile of hashes or an entry bundle (C2SP tlog-tiles). Full tiles never
/// change and are cached for good.
pub async fn tile(State(app): State<Shared>, UrlPath(path): UrlPath<String>) -> Response {
    let (entries, rest) = match path.strip_prefix("entries/") {
        Some(r) => (true, format!("0/{r}")),
        None => (false, path.clone()),
    };
    let Some(t) = Tile::parse(&rest) else {
        return plain(StatusCode::NOT_FOUND, "no such tile");
    };
    let tl = app.tlog.clone();
    let got = tokio::task::spawn_blocking(move || if entries { tl.bundle(&t) } else { tl.tile(&t) }).await;
    match got {
        Ok(Ok(Some(data))) => {
            let cache = if t.width as u64 == TILE_WIDTH { "public, max-age=31536000, immutable" } else { "public, max-age=5" };
            ([(header::CONTENT_TYPE, "application/octet-stream"), (header::CACHE_CONTROL, cache)], data).into_response()
        }
        Ok(Ok(None)) => plain(StatusCode::NOT_FOUND, "the log does not hold that tile yet"),
        Ok(Err(e)) => plain(StatusCode::INTERNAL_SERVER_ERROR, &e.to_string()),
        Err(e) => plain(StatusCode::INTERNAL_SERVER_ERROR, &e.to_string()),
    }
}

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

    fn entry(i: u64) -> Entry {
        Entry { kind: "ref".into(), repo: [1; 16], object: ObjectRef([i as u8; 32]), signers: vec![], received: i }
    }

    #[test]
    fn the_log_survives_a_torn_append_and_keeps_its_root() {
        let dir = tempfile::tempdir().unwrap();
        let log = Log::open(dir.path(), "hub.example/log").unwrap();
        let first: Vec<Entry> = (0..300).map(entry).collect();
        assert_eq!(log.append(&first).unwrap(), (0..300).collect::<Vec<_>>());
        let root = log.tree_hash(300).unwrap();
        // Leaves alone, as RFC 6962 roots them.
        let mut stored: Vec<Hash> = Vec::new();
        for (n, e) in first.iter().enumerate() {
            stored.extend(tlog::stored_hashes_for_leaf(n as u64, e.leaf_hash(), &stored).unwrap());
        }
        assert_eq!(root, tlog::tree_hash(300, &stored).unwrap());
        drop(log);
        // A crash after the entry and part of the hashes were written.
        let torn = dir.path().join("hashes");
        let len = std::fs::metadata(&torn).unwrap().len();
        OpenOptions::new().write(true).open(&torn).unwrap().set_len(len - 40).unwrap();
        let mut f = OpenOptions::new().append(true).open(dir.path().join("entries")).unwrap();
        f.write_all(&[0, 200, 1, 2]).unwrap(); // a partial entry
        let log = Log::open(dir.path(), "hub.example/log").unwrap();
        assert_eq!(log.size(), 300);
        assert_eq!(log.tree_hash(300).unwrap(), root);
        log.append(&[entry(300)]).unwrap();
        assert_eq!(log.size(), 301);
        let p = log.prove_inclusion(301, 7).unwrap();
        tlog::check_inclusion(&p, 301, &log.tree_hash(301).unwrap(), 7, &first[7].leaf_hash()).unwrap();
        let c = log.prove_consistency(301, 300).unwrap();
        tlog::check_consistency(&c, 301, &log.tree_hash(301).unwrap(), 300, &root).unwrap();
        // Tiles and bundles: the full level-0 tile and the partial one.
        let full = log.tile(&Tile { level: 0, n: 0, width: 256 }).unwrap().unwrap();
        assert_eq!(full.len(), 256 * 32);
        assert!(log.tile(&Tile { level: 0, n: 1, width: 45 }).unwrap().is_some());
        assert!(log.tile(&Tile { level: 0, n: 1, width: 46 }).unwrap().is_none(), "not yet");
        let bundle = log.bundle(&Tile { level: 0, n: 1, width: 45 }).unwrap().unwrap();
        assert_eq!(tlog::decode_bundle(&bundle).unwrap()[0], first[256].encode());
    }

    #[test]
    fn checkpoints_are_signed_when_the_log_grows() {
        let dir = tempfile::tempdir().unwrap();
        let sk = oxy_core::identity::SecretKey::generate();
        let hub = sk.identity("hub", "hub@example.org", 1);
        let log = Log::open(dir.path(), &origin_of("https://hub.example:7343/")).unwrap();
        assert_eq!(log.origin, "hub.example:7343/log");
        assert!(log.refresh_checkpoint(&hub, &sk).unwrap(), "the empty log's");
        assert!(!log.refresh_checkpoint(&hub, &sk).unwrap(), "nothing new");
        log.append(&[entry(1), entry(2)]).unwrap();
        assert!(log.refresh_checkpoint(&hub, &sk).unwrap());
        let note = tlog::Note::parse(&log.checkpoint().unwrap()).unwrap();
        let c = Checkpoint::verify(&note, "hub.example:7343/log", &hub).unwrap();
        assert_eq!((c.size, c.root), (2, log.tree_hash(2).unwrap()));
        // Kept across a restart.
        drop(log);
        let log = Log::open(dir.path(), "hub.example:7343/log").unwrap();
        assert_eq!(log.checkpoint().map(|n| n.contains("\n2\n")), Some(true));
    }
}