//! An epoch's root, pinned: the change without dependencies a document of that epoch
//! starts from. For a remake's epoch it is the epoch's ID (its snapshot change's hash),
//! which the rules hold every root to; for the first epoch, whose ID is all zeros, the
//! genesis change, which a node joining with an empty document would take from whoever
//! sent it first. So every node names its epoch's root in its handshakes
//! (`HelloRequest::root`, `HelloReply::root`), and:
//!
//! * a node whose document holds another root of the same epoch refuses the handshake:
//!   another history under this swarm's ID (a forged root, or a genesis started again);
//! * a joining node takes as its root only the one given (`--genesis-root`), or else the
//!   one its peers name, and none while they name different ones (`RootPin`), so one
//!   peer can't hand it another root unless it is all the joiner hears.

use std::collections::{HashMap, HashSet};

use anyhow::{Result, bail};
use tracing::{error, info, warn};

use crate::access::screen::RootPin;
use crate::document::replica::Epoch;
use crate::document::{ChangeHash, Document};
use crate::locks::MutexExt;
use crate::state::AppState;

/// What the node heard of its epoch's root, and knows of its own.
#[derive(Debug, Default)]
pub(crate) struct Roots {
    /// Of the first epoch, as each peer named it.
    named: HashMap<String, ChangeHash>,
    /// Its own, for the document (`Replica::epoch_seq`), once known.
    own: Option<(u32, ChangeHash)>,
    /// Peers whose root differs, logged once each.
    differing: HashSet<String>,
}

/// A change hash in hex (`--genesis-root`).
pub(crate) fn parse(hex: &str) -> Option<ChangeHash> {
    let hex = hex.trim();
    if hex.len() != 64 {
        return None;
    }
    let mut bytes = [0u8; 32];
    for (index, byte) in bytes.iter_mut().enumerate() {
        *byte = u8::from_str_radix(hex.get(index * 2..index * 2 + 2)?, 16).ok()?;
    }
    Some(ChangeHash(bytes))
}

/// `doc`'s root, of `epoch`: the epoch's ID for a remake's, the change without
/// dependencies of the first epoch's; None for an empty document.
pub(crate) fn of(doc: &Document, epoch: Epoch) -> Option<ChangeHash> {
    if epoch.generation > 0 {
        return Some(ChangeHash(epoch.id));
    }
    if doc.get_heads().is_empty() {
        return None;
    }
    doc.get_changes(&[])
        .iter()
        .find(|change| change.deps().is_empty())
        .map(|change| change.hash())
}

/// This node's epoch's root, if its document holds one; found once per document.
pub(crate) async fn own(state: &AppState) -> Option<ChangeHash> {
    let (epoch, epoch_seq) = state.epoch_stamp();
    if epoch.generation > 0 {
        return Some(ChangeHash(epoch.id));
    }
    if let Some((seq, root)) = state.roots.lock_or_recover().own
        && seq == epoch_seq
    {
        return Some(root);
    }
    let replica = state.doc.read().await;
    let root = of(&replica.doc, replica.epoch)?;
    state.roots.lock_or_recover().own = Some((replica.epoch_seq, root));
    Some(root)
}

/// `peer`, in a handshake of `epoch`, names `root` as its root. Refused (an error) if
/// this node's document holds another of the same epoch; of the first epoch, what an
/// empty document takes as its root follows (`pin`).
pub(crate) async fn heard(
    state: &AppState,
    peer: &str,
    epoch: Epoch,
    root: Option<&[u8]>,
) -> Result<()> {
    let Some(root) = root.and_then(|bytes| <[u8; 32]>::try_from(bytes).ok()) else {
        return Ok(());
    };
    let root = ChangeHash(root);
    if epoch != state.epoch_stamp().0 {
        // Another epoch: catching up is its business.
        return Ok(());
    }
    if let Some(ours) = own(state).await
        && ours != root
    {
        if state
            .roots
            .lock_or_recover()
            .differing
            .insert(peer.to_owned())
        {
            error!(peer = %peer, ours = %ours, theirs = %root, epoch = %epoch, "A peer's document starts from another root of this epoch: another history under this swarm's ID (a forged root, or a genesis started again); not synced with");
        }
        bail!("{peer} starts epoch {epoch} from root {root}, ours from {ours}");
    }
    if epoch.generation == 0 {
        let pin = {
            let mut roots = state.roots.lock_or_recover();
            roots.named.insert(peer.to_owned(), root);
            pin(state.genesis_root, &roots.named)
        };
        let mut screening = state.screening.lock_or_recover();
        if screening.root_pin != pin {
            match &pin {
                RootPin::Disputed => {
                    warn!(
                        "Peers name different roots of the swarm's first epoch: taking none until they agree, or one is given (--genesis-root)"
                    );
                }
                RootPin::Only(root) => {
                    info!(root = %root, "The swarm's root, as given or as its peers name it")
                }
                RootPin::Any => {}
            }
            screening.root_pin = pin;
        }
    }
    Ok(())
}

/// What an empty document of the first epoch takes as its root: the one `given`, else
/// the one all peers name (`named`), none while they differ, any while none is named.
pub(crate) fn pin(given: Option<ChangeHash>, named: &HashMap<String, ChangeHash>) -> RootPin {
    if let Some(given) = given {
        return RootPin::Only(given);
    }
    let mut roots: Vec<&ChangeHash> = named.values().collect();
    roots.sort_unstable();
    roots.dedup();
    match roots.as_slice() {
        [] => RootPin::Any,
        [root] => RootPin::Only(**root),
        _ => RootPin::Disputed,
    }
}

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

    /// The root given wins; else the one every peer names; none while they differ.
    #[test]
    fn what_a_joiner_takes_as_its_root() {
        let (a, b) = (ChangeHash([1; 32]), ChangeHash([2; 32]));
        let named = |roots: &[(&str, ChangeHash)]| -> HashMap<String, ChangeHash> {
            roots
                .iter()
                .map(|(peer, root)| ((*peer).to_owned(), *root))
                .collect()
        };
        assert_eq!(pin(None, &named(&[])), RootPin::Any);
        assert_eq!(pin(None, &named(&[("p", a), ("q", a)])), RootPin::Only(a));
        assert_eq!(pin(None, &named(&[("p", a), ("q", b)])), RootPin::Disputed);
        assert_eq!(pin(Some(b), &named(&[("p", a)])), RootPin::Only(b));
    }
}