guardian-db 0.19.0

use crate::log::entry::Entry;
use crate::log::entry::EntryOrHash;
use crate::log::identity::Identity;
use crate::log::lamport_clock::LamportClock;
use crate::p2p::network::client::IrohClient;
use iroh_blobs::Hash;
use serde_json::json;
use std::cmp::Ordering;
use std::cmp::max;
use std::collections::BTreeMap; // For Log.entries (determinism in snapshots).
use std::collections::HashMap; // For internal helper methods (not serialized).
use std::collections::HashSet;
use std::fmt::{Display, Formatter, Result};
use std::sync::Arc;
use std::time::SystemTime;

pub mod access_control;
pub mod entry;
pub mod identity;
pub mod identity_provider;
pub mod lamport_clock;
pub mod traits;

// Type aliases to reduce type complexity.
type EntrySortFn = Box<dyn Fn(&Entry, &Entry) -> Ordering + Send + Sync>;

/// An immutable, operation-based conflict-free replicated data type ([CRDT]).
///
/// **IMPORTANT:** Uses `BTreeMap` instead of `HashMap` to guarantee a
/// deterministic order when serializing snapshots. This is critical so the
/// iroh-blobs BLAKE3 hash is consistent across runs.
///
/// [CRDT]: https://en.wikipedia.org/wiki/Conflict-free_replicated_data_type
pub struct Log {
    client: Arc<IrohClient>,
    id: String,
    identity: Identity,
    access: AdHocAccess,
    entries: BTreeMap<Hash, Arc<Entry>>, // BTreeMap for determinism.
    length: usize,
    heads: Vec<Arc<Entry>>,
    nexts: HashSet<Hash>,
    sort_fn: EntrySortFn,
    clock: LamportClock,
}

/// Options for constructing [`Log`].
///
/// Constructing log options using `LogOptions::new()` creates default log options:
/// * no identifier,
/// * no entries (and no heads among those non-existent entries),
/// * no Lamport clock,
/// * no sorting algorithm.
///
/// Use method chaining to set additional parameters:
///
/// ```ignore
/// let opts = LogOptions::new().id("some_id").clock(LamportClock::new().set_time(128));
/// let log = Log::new(/* identity */,opts);
/// ```
///
/// [`Log`]: ./struct.Log.html
pub struct LogOptions<'a> {
    pub id: Option<&'a str>,
    pub access: AdHocAccess,
    pub entries: &'a [Arc<Entry>],
    pub heads: &'a [Arc<Entry>],
    pub clock: Option<LamportClock>,
    pub sort_fn: Option<EntrySortFn>,
}

impl<'a> LogOptions<'a> {
    /// Constructs default log options.
    pub fn new() -> LogOptions<'a> {
        LogOptions::default()
    }

    /// Sets the identifier for the constructed log options.
    ///
    /// Allows method chaining.
    pub fn id(mut self, id: &'a str) -> LogOptions<'a> {
        self.id = Some(id);
        self
    }

    /// Sets the entries for the constructed log options.
    ///
    /// Allows method chaining.
    pub fn entries(mut self, es: &'a [Arc<Entry>]) -> LogOptions<'a> {
        self.entries = es;
        self
    }

    /// Sets the heads for the constructed log options.
    ///
    /// Allows method chaining.
    pub fn heads(mut self, hs: &'a [Arc<Entry>]) -> LogOptions<'a> {
        self.heads = hs;
        self
    }

    /// Sets the Lamport clock for the constructed log options.
    ///
    /// Allows method chaining.
    pub fn clock(mut self, clock: LamportClock) -> LogOptions<'a> {
        self.clock = Some(clock);
        self
    }

    /// Sets the sorting algorithm for the constructed log options.
    ///
    /// Allows method chaining.
    pub fn sort_fn<F>(mut self, sort_fn: F) -> LogOptions<'a>
    where
        F: 'static + Fn(&Entry, &Entry) -> Ordering + Send + Sync,
    {
        self.sort_fn = Some(Box::new(sort_fn));
        self
    }
}

impl<'a> Default for LogOptions<'a> {
    fn default() -> Self {
        LogOptions {
            id: None,
            access: AdHocAccess,
            entries: &[],
            heads: &[],
            clock: None,
            sort_fn: None,
        }
    }
}

impl Log {
    /// Constructs a new log owned by `identity`, using `opts` for constructor options.
    ///
    /// Use [`LogOptions::new()`] as `opts` for default constructor options.
    ///
    /// [`LogOptions::new()`]: ./struct.LogOptions.html#method.new
    pub fn new(client: Arc<IrohClient>, identity: Identity, opts: LogOptions) -> Log {
        let (id, access, entries, heads, clock, sort_fn) = (
            opts.id,
            opts.access,
            opts.entries,
            opts.heads,
            opts.clock,
            opts.sort_fn,
        );
        let sort_fn = Box::new(Entry::no_zeroes(
            sort_fn.unwrap_or(Box::new(Entry::last_write_wins)),
        ));
        let id = if let Some(s) = id {
            s.to_owned()
        } else {
            SystemTime::now()
                .duration_since(SystemTime::UNIX_EPOCH)
                .unwrap()
                .as_millis()
                .to_string()
        };
        let length = entries.len();

        let heads = Log::dedup(&if heads.is_empty() {
            Log::find_heads(entries)
        } else {
            heads.to_owned()
        });

        let mut nexts = HashSet::new();
        for e in entries {
            for n in e.next() {
                nexts.insert(n.to_owned());
            }
        }

        let mut entry_set = BTreeMap::new(); // BTreeMap for determinism.
        for e in entries {
            entry_set.insert(*e.hash(), e.clone());
        }

        let mut t_max = 0;
        if let Some(c) = clock {
            t_max = c.time();
        }
        for h in &heads {
            t_max = max(t_max, h.clock().time());
        }
        let clock = LamportClock::new(identity.pub_key()).set_time(t_max);

        Log {
            client: client.clone(),
            id,
            identity,
            access,
            entries: entry_set,
            length,
            heads,
            nexts,
            sort_fn,
            clock,
        }
    }

    /// Constructs a new log with the identity `identity` from an entry with the hash `hash`,
    /// using `opts` for constructor options.
    ///
    /// Use [`LogOptions::new()`] as `opts` for default constructor options.
    ///
    /// **N.B.** [`opts.entries(/* entries */)`] *and* [`opts.heads(/* heads */)`] *have no effect in the log created.*
    ///
    /// [`LogOptions::new()`]: ./struct.LogOptions.html#method.new
    /// [`opts.entries(/* entries */)`]: ./struct.LogOptions.html#method.entries
    /// [`opts.heads(/* heads */)`]: ./struct.LogOptions.html#method.heads
    pub fn from_hash(
        client: Arc<IrohClient>,
        identity: Identity,
        opts: LogOptions,
        hash: &Hash,
    ) -> Log {
        let es = Entry::fetch_entries(&client, &[*hash])
            .into_iter()
            .map(Arc::new)
            .collect::<Vec<Arc<Entry>>>();
        Log::new(client, identity, opts.entries(&es).heads(&[]))
    }

    /// Deprecated: use from_hash instead
    pub fn from_multihash(
        client: Arc<IrohClient>,
        identity: Identity,
        opts: LogOptions,
        hash: &str,
    ) -> Log {
        // Parse hex string to Hash
        let bytes = hex::decode(hash).expect("Invalid hex hash");
        let mut array = [0u8; 32];
        array.copy_from_slice(&bytes);
        let hash_obj = Hash::from(array);
        Self::from_hash(client, identity, opts, &hash_obj)
    }

    /// Appends `data` into the log as a new entry.
    ///
    /// Returns a reference to the newly created, appended entry.
    pub fn append(&mut self, data: &str, n_ptr: Option<usize>) -> &Entry {
        let mut t_new = self.clock.time();
        for h in &self.heads {
            t_new = max(t_new, h.clock().time());
        }
        t_new += 1;
        self.clock = LamportClock::new(self.clock.id()).set_time(t_new);

        let mut heads = Vec::new();
        for h in &self.heads {
            heads.push(h.clone());
        }
        let mut refs = self.traverse(
            &heads[..],
            Some(max(n_ptr.unwrap_or(1), self.heads.len())),
            None,
        );
        self.heads.reverse();
        self.heads = Log::dedup(&self.heads);
        self.heads.append(&mut refs);

        //should be created asynchronically in Client
        let mut entry = Entry::new(
            self.identity.clone(),
            &self.id,
            data.as_bytes(),
            &self
                .heads
                .iter()
                .map(|x| EntryOrHash::Hash(*x.hash()))
                .collect::<Vec<_>>()[..],
            Some(self.clock.clone()),
        );

        // Use a separate thread with a new Runtime to avoid "cannot start runtime from within a runtime".
        let client_clone = self.client.clone();
        let entry_clone = entry.clone();
        let hash_result = std::thread::spawn(move || {
            tokio::runtime::Runtime::new()
                .unwrap()
                .block_on(async move { Entry::hash_entry(&client_clone, &entry_clone).await })
        })
        .join()
        .unwrap()
        .unwrap();

        entry.set_hash(&hash_result);
        //should be queried asynchronically
        if !self.access.can_access(&entry) {
            panic!(
                "Could not append entry, key \"{}\" is not allowed to write in the log",
                self.identity.id()
            );
        }

        let eh = *entry.hash();
        let arc_entry = Arc::new(entry);
        self.entries.insert(eh, arc_entry.clone());
        for h in &self.heads {
            self.nexts.insert(*h.hash());
        }
        self.heads.clear();
        self.heads.push(arc_entry);
        self.length += 1;

        &self.entries[&eh]
    }

    /// Adds an existing entry to the log, preserving its original hash.
    /// This is used during sync/replication to add entries received from other nodes.
    /// Returns true if the entry was added, false if it already exists.
    pub fn add_entry(&mut self, entry: Entry) -> bool {
        let hash = *entry.hash();

        // Check if entry already exists
        if self.entries.contains_key(&hash) {
            return false;
        }

        // Update nexts from entry's next pointers
        for n in entry.next() {
            self.nexts.insert(*n);
        }

        // Add entry to entries map
        let arc_entry = Arc::new(entry);
        self.entries.insert(hash, arc_entry.clone());

        // Update heads: remove any heads that are in this entry's next list
        self.heads.retain(|h| !arc_entry.next().contains(h.hash()));

        // If this entry is not in any other entry's next list, it's a head
        if !self.nexts.contains(&hash) {
            self.heads.push(arc_entry);
        }

        self.length += 1;
        true
    }

    /// Joins the log `other` into this log. `other` is kept intact through and after the process.
    ///
    /// Optionally truncates the log into `size` after joining.
    ///
    /// Returns a reference to this log.
    pub fn join(&mut self, other: &Log, size: Option<usize>) -> Option<&Log> {
        if self.id != other.id {
            return None;
        }
        let new_items = other.diff(self);

        //something about identify provider and verification,
        //implement later
        //...
        //...

        for e in &new_items {
            if self.get(e.0).is_none() {
                self.length += 1;
            }
            for n in e.1.next() {
                self.nexts.insert(n.to_owned());
            }
        }

        for e in &new_items {
            self.entries.insert(e.0.to_owned(), e.1.clone());
        }

        let mut nexts_from_new_items = HashSet::new();
        new_items
            .into_iter()
            .map(|x| x.1.next().to_owned())
            .for_each(|n| {
                n.iter().for_each(|n| {
                    nexts_from_new_items.insert(n.to_owned());
                })
            });
        let all_heads = Log::find_heads(
            &self
                .heads
                .iter()
                .chain(other.heads.iter())
                .cloned()
                .collect::<Vec<_>>()[..],
        );
        let merged_heads: Vec<Arc<Entry>> = all_heads
            .into_iter()
            .filter(|x| !nexts_from_new_items.contains(x.hash()))
            .filter(|x| !self.nexts.contains(x.hash()))
            .collect();
        self.heads = Log::dedup(&merged_heads[..]);

        if let Some(n) = size {
            let mut vs = self.values();
            vs.reverse();
            vs = vs.into_iter().take(n).collect();

            self.entries.clear();
            for v in &vs {
                self.entries.insert(*v.hash(), v.clone());
            }

            self.heads = Log::find_heads(&Log::dedup(&vs));
            self.length = self.entries.len();
        }

        let mut t_max = 0;
        for h in &self.heads {
            t_max = max(t_max, h.clock().time());
        }
        self.clock = LamportClock::new(self.clock.id()).set_time(t_max);

        Some(self)
    }

    /// Returns a map of all the entries contained in this log but not in `other`.
    pub fn diff(&self, other: &Log) -> HashMap<Hash, Arc<Entry>> {
        let mut stack: Vec<Hash> = self.heads.iter().map(|x| *x.hash()).collect();
        let mut traversed = HashSet::<Hash>::new();
        let mut diff = HashMap::new();
        while !stack.is_empty() {
            let hash = stack.remove(0);
            let a = self.get(&hash);
            let b = other.get(&hash);
            if let Some(entry_a) = a
                && b.is_none()
                && entry_a.id() == other.id
            {
                for n in entry_a.next() {
                    if !traversed.contains(n) && other.get(n).is_none() {
                        stack.push(*n);
                        traversed.insert(*n);
                    }
                }
                traversed.insert(*entry_a.hash());
                diff.insert(*entry_a.hash(), entry_a.clone());
            }
        }
        diff
    }

    /// Returns the identifier of the log.
    pub fn id(&self) -> &str {
        &self.id
    }

    /// Returns `true` if the log contains an entry with the hash `hash`.
    /// Otherwise returns `false`.
    pub fn has(&self, hash: &Hash) -> bool {
        self.entries.contains_key(hash)
    }

    /// Returns a pointer to the entry with the hash `hash`.
    pub fn get(&self, hash: &Hash) -> Option<&Arc<Entry>> {
        self.entries.get(hash)
    }

    /// Returns the number of entries in the log.
    pub fn len(&self) -> usize {
        self.length
    }

    /// Returns whether the log is empty.
    pub fn is_empty(&self) -> bool {
        self.length == 0
    }

    pub fn find_heads(entries: &[Arc<Entry>]) -> Vec<Arc<Entry>> {
        let mut parents = HashMap::<Hash, Hash>::new();
        for e in entries {
            for n in e.next() {
                parents.insert(*n, *e.hash());
            }
        }
        let mut heads = Vec::new();
        for e in entries {
            if !parents.contains_key(e.hash()) {
                heads.push(e.clone());
            }
        }
        heads.sort_by(|a, b| {
            let diff = a.clock().id().cmp(b.clock().id());
            if diff == Ordering::Equal {
                Ordering::Less
            } else {
                diff
            }
        });
        heads
    }

    pub fn find_tails(entries: &[Arc<Entry>]) -> Vec<Arc<Entry>> {
        let mut no_nexts = Vec::new();
        let mut reverses = HashMap::new();
        let mut nexts = HashSet::new();
        let mut hashes: HashSet<Hash> = HashSet::new();
        for e in entries {
            if e.next().is_empty() {
                no_nexts.push(e.clone());
            }
            for n in e.next() {
                reverses.insert(n, e.clone());
                nexts.insert(*n);
            }
            hashes.insert(*e.hash());
        }
        //correct order?
        let mut tails = Log::dedup(
            &nexts
                .iter()
                .filter(|&&x| !hashes.contains(&x))
                .map(|x| reverses[x].clone())
                .chain(no_nexts)
                .collect::<Vec<_>>()[..],
        );
        tails.sort();
        tails
    }

    pub fn find_tail_hashes(entries: &[Arc<Entry>]) -> Vec<Hash> {
        let mut hashes: HashSet<Hash> = HashSet::new();
        for e in entries {
            hashes.insert(*e.hash());
        }
        let mut ths = Vec::new();
        for e in entries {
            for i in (0..e.next().len()).rev() {
                let n = &e.next()[i];
                if !hashes.contains(n) {
                    ths.push(*n);
                }
            }
        }
        ths.reverse();
        ths
    }

    fn dedup(v: &[Arc<Entry>]) -> Vec<Arc<Entry>> {
        let mut s = HashSet::new();
        v.iter().filter(|x| s.insert(x.hash())).cloned().collect()
    }

    pub fn set_identity(&mut self, identity: Identity) {
        let mut t_max = 0;
        for h in &self.heads {
            t_max = max(t_max, h.clock().time());
        }
        self.clock = LamportClock::new(identity.pub_key()).set_time(t_max);
        self.identity = identity;
    }

    pub fn clock(&self) -> &LamportClock {
        &self.clock
    }

    /// Updates the Lamport clock time to the maximum time found across all heads.
    /// This should be called after loading entries via `add_entry()` to ensure the
    /// clock is properly synchronized for future appends.
    pub fn sync_clock_from_heads(&mut self) {
        let mut t_max = self.clock.time();
        for h in &self.heads {
            t_max = max(t_max, h.clock().time());
        }
        self.clock = LamportClock::new(self.clock.id()).set_time(t_max);
    }

    pub fn values(&self) -> Vec<Arc<Entry>> {
        let mut es = self.traverse(&self.heads, None, None);
        es.reverse();
        es
    }

    pub fn heads(&self) -> Vec<Arc<Entry>> {
        let mut hs = Log::dedup(&self.heads);
        hs.sort_by(|a, b| (self.sort_fn)(a, b));
        hs.reverse();
        hs
    }

    pub fn tails(&self) -> Vec<Arc<Entry>> {
        Log::find_tails(&self.values())
    }

    pub fn tail_hashes(&self) -> Vec<Hash> {
        Log::find_tail_hashes(&self.values())
    }

    pub fn all(&self) -> String {
        let mut s = String::from("[ ");
        for e in &self.entries {
            if self.heads.iter().any(|x| x.hash() == e.1.hash()) {
                s.push('^');
            }
            s.push_str(&hex::encode(e.0.as_bytes()));
            s.push_str(", ");
        }
        s = String::from(&s[..s.len() - 2]);
        s.push_str(" ]");
        s
    }

    pub fn entries(&self) -> String {
        let mut s = String::new();
        for e in &self.entries {
            s.push_str(&hex::encode(e.0.as_bytes()));
            if !e.1.next().is_empty() {
                s.push_str("\t\t>");
                s.push_str(&hex::encode(e.1.next()[0].as_bytes()));
                s.push_str(", >");
                s.push_str(&hex::encode(e.1.next()[1].as_bytes()));
            } else {
                s.push_str("\t\t.,.");
            }
            s.push('\n');
        }
        s
    }

    pub fn traverse(
        &self,
        roots: &[Arc<Entry>],
        amount: Option<usize>,
        end_hash: Option<Hash>,
    ) -> Vec<Arc<Entry>> {
        let mut stack = Log::dedup(roots);
        stack.sort_by(|a, b| (self.sort_fn)(a, b));
        stack.reverse();
        let mut traversed = HashSet::<Hash>::new();
        let mut result = Vec::new();
        let mut count = 0;

        while !stack.is_empty() && (amount.is_none() || count < amount.unwrap()) {
            let e = stack.remove(0);
            let hash = *e.hash();
            count += 1;
            for h in e.next() {
                if let Some(e) = self.get(h)
                    && !traversed.contains(e.hash())
                {
                    stack.insert(0, e.clone());
                    stack.sort_by(|a, b| (self.sort_fn)(a, b));
                    stack.reverse();
                    traversed.insert(*e.hash());
                }
            }
            result.push(e);

            if let Some(ref eh) = end_hash
                && eh == &hash
            {
                break;
            }
        }

        result
    }

    pub fn json(&self) -> String {
        let mut hs = self.heads.to_owned();
        hs.sort_by(|a, b| (self.sort_fn)(a, b));
        hs.reverse();
        json!({
            "id": self.id,
            "heads": hs.into_iter().map(|x| hex::encode(x.hash().as_bytes())).collect::<Vec<_>>(),
        })
        .to_string()
    }

    /// Returns a snapshot representation of the log using postcard for consistent serialization.
    /// Used mainly for testing and state comparison.
    pub fn snapshot(&self) -> String {
        let hs = self.heads.to_owned();
        let vs = self.values().to_owned();

        // Use postcard for consistent serialization of the Entry values.
        let heads_serialized: Vec<String> = hs
            .into_iter()
            .map(|x| {
                crate::guardian::serializer::serialize(&*x)
                    .map(hex::encode)
                    .unwrap_or_default()
            })
            .collect();

        let values_serialized: Vec<String> = vs
            .into_iter()
            .map(|x| {
                crate::guardian::serializer::serialize(&*x)
                    .map(hex::encode)
                    .unwrap_or_default()
            })
            .collect();

        json!({
            "id": self.id,
            "heads": heads_serialized,
            "values": values_serialized,
        })
        .to_string()
    }

    pub fn buffer(&self) -> Vec<u8> {
        self.json().into_bytes()
    }
}

impl Display for Log {
    fn fmt(&self, f: &mut Formatter) -> Result {
        let mut es = self.values();
        es.reverse();
        let mut s = String::new();
        for e in es {
            let parents = Entry::find_children(&e, &self.values());
            if !parents.is_empty() {
                if parents.len() >= 2 {
                    for _ in 0..parents.len() - 1 {
                        s.push_str("  ");
                    }
                }
                s.push_str("└─");
            }
            s.push_str(&e.payload_str());
            s.push('\n');
        }
        write!(f, "{}", s)
    }
}

#[doc(hidden)]
#[derive(Copy, Clone)]
pub struct AdHocAccess;

impl AdHocAccess {
    fn can_access(&self, _entry: &Entry) -> bool {
        true
    }
}