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