//! 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::identity::Identity;

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>,

    /// `append.lock`: held, exclusively, around each append, so several

    /// processes sharing the data directory append one at a time (#274).

    lock: File,

    /// The latest signed checkpoint: its size and its note.

    checkpoint: Mutex<Option<(u64, String)>>,

    /// Witnesses this log asks to cosign its checkpoints.

    witnesses: std::sync::RwLock<Vec<Peer>>,

    /// Logs this hub witnesses: their origins and keys.

    witness_for: std::sync::RwLock<Vec<(String, Identity)>>,

    /// Whether this process writes the log: the web role's (#272). A

    /// process without it only reads, catching up with the writers.

    writer: bool,

}

/// A witness of this hub's log.

#[derive(Clone, Debug)]

pub struct Peer {

    pub name: String,

    /// Its add-checkpoint address.

    pub url: String,

    pub identity: Identity,

}

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> {

        Self::open_as(dir, origin, true)

    }

    /// Open the log to read only, as a process without the web role does

    /// (#272): it neither appends nor signs, and leaves the files as they

    /// are, so it never cuts short an append a writer is making. Several

    /// web processes may share the data directory and write (#274).

    pub fn open_reader(dir: &Path, origin: &str) -> anyhow::Result<Log> {

        Self::open_as(dir, origin, false)

    }

    fn open_as(dir: &Path, origin: &str, writer: bool) -> 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 lock = OpenOptions::new().read(true).write(true).create(true).truncate(false).open(dir.join("append.lock"))?;

        let mut files = Files { entries, offsets, hashes, size: 0, entries_len: 0 };

        if writer {

            lock.lock()?;

            let done = Self::recover(&mut files, true);

            lock.unlock()?;

            done?;

        } else {

            Self::recover(&mut files, false)?;

        }

        let size = files.size;

        Ok(Log {

            dir: dir.to_path_buf(),

            origin: origin.to_string(),

            files: Mutex::new(files),

            lock,

            checkpoint: Mutex::new(Self::read_checkpoint(dir, size)),

            witnesses: Default::default(),

            witness_for: Default::default(),

            writer,

        })

    }

    /// Read the log's extent from its files, as another process may have

    /// appended since (#274). The whole entries count: those whose offset

    /// and bytes were written, and whose stored hashes all were. A writer,

    /// holding the append lock, also finishes what an append cut short left

    /// (it truncates partial bytes and computes missing hashes again); a

    /// reader changes nothing and takes the whole entries only.

    fn recover(files: &mut Files, writer: bool) -> anyhow::Result<()> {

        let entries_on_disk = files.entries.metadata()?.len();

        let mut size = files.offsets.metadata()?.len() / 8;

        let mut entries_len = 0;

        while size > 0 {

            let off = read_u64_at(&files.offsets, (size - 1) * 8)?;

            let mut len = [0u8; 2];

            if off + 2 <= entries_on_disk && files.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;

        }

        // The hashes of every leaf whose hashes were all written; the rest

        // are computed again from the entries.

        let have = files.hashes.metadata()?.len() / 32;

        let mut whole = size;

        while tlog::stored_hash_count(whole) > have {

            whole -= 1;

        }

        if !writer {

            files.size = whole;

            files.entries_len = entries_len;

            return Ok(());

        }

        files.offsets.set_len(size * 8)?;

        files.entries.set_len(entries_len)?;

        files.hashes.set_len(tlog::stored_hash_count(whole) * 32)?;

        files.entries_len = 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()?;

        Ok(())

    }

    /// Catch up with what other processes appended (#274).

    fn catch_up(&self, f: &mut Files) {

        if let Err(e) = Self::recover(f, false) {

            tracing::warn!("tlog: reading the log's extent: {e:#}");

        }

    }

    /// The signed checkpoint kept in `dir`, if it is of at most `size` entries.

    fn read_checkpoint(dir: &Path, size: u64) -> Option<(u64, String)> {

        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))

        })

    }

    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)

    }

    /// Entry `n`'s bytes.

    pub fn entry(&self, n: u64) -> anyhow::Result<Vec<u8>> {

        let mut f = self.files.lock().unwrap();

        if n >= f.size {

            self.catch_up(&mut f);

        }

        anyhow::ensure!(n < f.size, "entry {n} is not in the log");

        Self::entry_bytes(&f, n)

    }

    /// Entries in the log.

    pub fn size(&self) -> u64 {

        let mut f = self.files.lock().unwrap();

        self.catch_up(&mut f);

        f.size

    }

    /// Append entries, durably; their numbers.

    pub fn append(&self, entries: &[Entry]) -> anyhow::Result<Vec<u64>> {

        if !self.writer {

            anyhow::bail!("this process does not write the log: it has no web role");

        }

        let mut f = self.files.lock().unwrap();

        // One append at a time across the processes sharing the directory

        // (#274), each from the log as the last left it.

        self.lock.lock()?;

        let appended = Self::recover(&mut f, true).and_then(|()| Self::append_locked(&mut f, entries));

        self.lock.unlock()?;

        appended

    }

    fn append_locked(f: &mut Files, entries: &[Entry]) -> anyhow::Result<Vec<u64>> {

        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)?))

    }

    /// Take `[tlog]` from the configuration: witnesses and witnessed logs.

    pub fn configure(&self, c: &crate::config::Tlog) -> anyhow::Result<()> {

        let key = |b64: &str, what: &str| -> anyhow::Result<Identity> {

            let bytes = oxy_core::dsse::base64_decode(b64.trim()).ok_or_else(|| anyhow::anyhow!("[tlog] {what}: the key is not base64"))?;

            Identity::decode(&bytes).map_err(|e| anyhow::anyhow!("[tlog] {what}: {e}"))

        };

        let mut ws = Vec::new();

        for w in &c.witnesses {

            ws.push(Peer { name: w.name.clone(), url: w.url.clone(), identity: key(&w.key, &w.name)? });

        }

        let mut wf = Vec::new();

        for l in &c.witness_for {

            wf.push((l.origin.clone(), key(&l.key, &l.origin)?));

        }

        self.set_witnesses(ws);

        for (o, k) in wf {

            self.add_witness_for(&o, k);

        }

        Ok(())

    }

    pub fn set_witnesses(&self, peers: Vec<Peer>) {

        *self.witnesses.write().unwrap() = peers;

    }

    pub fn witnesses(&self) -> Vec<Peer> {

        self.witnesses.read().unwrap().clone()

    }

    /// Witness the log named `origin`, signed with `key`.

    pub fn add_witness_for(&self, origin: &str, key: Identity) {

        let mut w = self.witness_for.write().unwrap();

        w.retain(|(o, _)| o != origin);

        w.push((origin.to_string(), key));

    }

    fn witnessed_key(&self, origin: &str) -> Option<Identity> {

        self.witness_for.read().unwrap().iter().find(|(o, _)| o == origin).map(|(_, k)| k.clone())

    }

    /// This hub's name as a witness: its log's host.

    pub fn witness_name(&self) -> &str {

        self.origin.trim_end_matches("/log")

    }

    /// Replace the latest checkpoint's note with `note` (the same

    /// checkpoint, with cosignatures added), unless a newer one was signed.

    fn set_note(&self, size: u64, note: String) -> anyhow::Result<()> {

        let mut cur = self.checkpoint.lock().unwrap();

        if cur.as_ref().map(|(s, _)| *s) != Some(size) {

            return Ok(());

        }

        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"))?;

        *cur = Some((size, note));

        Ok(())

    }

    /// The latest signed checkpoint note: the newer of this process's and

    /// the one kept in the directory, which the process leading the

    /// checkpointer signs (#274).

    pub fn checkpoint(&self) -> Option<String> {

        let mut held = self.checkpoint.lock().unwrap();

        if let Some(kept) = Self::read_checkpoint(&self.dir, u64::MAX) {

            if held.as_ref().is_none_or(|(s, _)| kept.0 > *s) {

                *held = Some(kept);

            }

        }

        held.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> {

        if !self.writer {

            return Ok(false);

        }

        let size = self.size();

        // From the newest kept, whichever process signed it.

        let _ = self.checkpoint();

        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}");