//! 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, or that was given
//!   another (`--genesis-root`), 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`); and none
//!   before each of its seeds named one, or `ROOT_WAIT` passed since the first did, so
//!   the seed that answers first doesn't decide alone. Seeds are the nodes proven at
//!   the seeds' addresses (`seed_found`): a peer answering under identities of its own
//!   neither ends the wait nor starts its clock, only disputes. One seed hands a joiner
//!   another root only if it is all the joiner hears of its seeds within the wait.

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

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;

/// How long a joining node waits, after the first peer named the swarm's root, for its
/// other seeds to name theirs, before it takes the one named.
pub(crate) const ROOT_WAIT: Duration = Duration::from_secs(10);

/// 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>,
    /// The nodes proven at the seeds' addresses (`seed_found`): only theirs count
    /// toward the wait (`repin`).
    seeds: HashSet<String>,
    /// Joining with an empty document: how many seeds it waits for, and until when once
    /// the first named a root.
    awaiting: Option<(usize, Option<tokio::time::Instant>)>,
    /// 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 {
        // The root given: a peer naming another is another history, whatever this
        // node's document holds yet.
        if let Some(given) = state.genesis_root
            && given != root
        {
            if state
                .roots
                .lock_or_recover()
                .differing
                .insert(peer.to_owned())
            {
                error!(peer = %peer, given = %given, theirs = %root, "A peer's document starts from another root than the one given (--genesis-root): another history under this swarm's ID; not synced with");
            }
            bail!("{peer} starts the first epoch from root {root}, not the one given ({given})");
        }
        state
            .roots
            .lock_or_recover()
            .named
            .insert(peer.to_owned(), root);
        repin(state);
    }
    Ok(())
}

/// A node joining with an empty document through `seeds` seeds: it takes no root until
/// each named one, or `ROOT_WAIT` passed since the first did (`repin`). Not with a root
/// given, which it takes alone.
pub(crate) fn await_seeds(state: &AppState, seeds: usize) {
    if state.genesis_root.is_some() || seeds == 0 {
        return;
    }
    state.roots.lock_or_recover().awaiting = Some((seeds, None));
    repin(state);
}

/// `node` is the one at a seed's address (`bootstrap::resolve_seeds`): its root counts
/// toward the wait for the seeds (`repin`).
pub(crate) fn seed_found(state: &AppState, node: &str) {
    if state.roots.lock_or_recover().seeds.insert(node.to_owned()) {
        repin(state);
    }
}

/// Until when a joining node still waits for its seeds to name the swarm's root, if it
/// waits on a clock (`await_seeds`): then `repin`.
pub(crate) fn waiting_until(state: &AppState) -> Option<tokio::time::Instant> {
    state
        .roots
        .lock_or_recover()
        .awaiting
        .and_then(|(_, until)| until)
}

/// What an empty document of the first epoch takes as its root, from what the node was
/// given and heard, into `Screening`: logged when it changes.
pub(crate) fn repin(state: &AppState) {
    let now = tokio::time::Instant::now();
    let pin = {
        let mut roots = state.roots.lock_or_recover();
        // Seeds only: others' roots dispute, but neither end the wait nor start it.
        let named = roots
            .named
            .keys()
            .filter(|peer| roots.seeds.contains(*peer))
            .count();
        let awaiting = match &mut roots.awaiting {
            Some((seeds, until)) => {
                if until.is_none() && named > 0 {
                    *until = Some(now + ROOT_WAIT);
                }
                named < *seeds && until.is_none_or(|until| now < until)
            }
            None => false,
        };
        if !awaiting {
            roots.awaiting = None;
        }
        pin(state.genesis_root, &roots.named, awaiting)
    };
    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::Awaiting => {
                info!(wait = ?ROOT_WAIT, "Taking the swarm's root once every seed named it, or a while after the first did")
            }
            RootPin::Only(root) => {
                info!(root = %root, "The swarm's root, as given or as its peers name it")
            }
            RootPin::Any => {}
        }
        screening.root_pin = pin;
    }
}

/// What an empty document of the first epoch takes as its root: the one `given`, else,
/// while `awaiting` its seeds, none; 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>,
    awaiting: bool,
) -> RootPin {
    if let Some(given) = given {
        return RootPin::Only(given);
    }
    if awaiting {
        return RootPin::Awaiting;
    }
    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 none while waiting for the seeds; 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(&[]), false), RootPin::Any);
        assert_eq!(pin(None, &named(&[("p", a)]), true), RootPin::Awaiting);
        assert_eq!(
            pin(None, &named(&[("p", a), ("q", a)]), false),
            RootPin::Only(a)
        );
        assert_eq!(
            pin(None, &named(&[("p", a), ("q", b)]), false),
            RootPin::Disputed
        );
        assert_eq!(pin(Some(b), &named(&[("p", a)]), true), RootPin::Only(b));
    }
}