//! HyParView (Leitão, Pereira and Rodrigues, "HyParView: a membership protocol for
//! reliable gossip-based broadcast", 2007), slimmed down, as an experiment
//! (`--topology hyparview`, `swarm::topology::Topology`).
//!
//! Each node keeps an active view: its neighbours, about `k` of them (the swarm's
//! neighbours, as with the member list's graph), each of which keeps it in its own: a
//! link both ends agreed to (NEIGHBOUR, answered by NEIGHBOUR_REPLY), not one every
//! node computes alike. A node with fewer asks others from its passive view, one at a
//! time per missing neighbour; one that has none asks with high priority, and is taken
//! even where that means dropping another (told so: DISCONNECT). A neighbour that fails
//! (judged unreachable: unanswered HELLOs, dormant, not alive lately) leaves the active
//! view, and another is asked. Every neighbour is asked again now and then whether it
//! still is (`REAFFIRM_EVERY`): the paper's links are TCP connections, whose ends know
//! when the other closed them; ours, without, could be left in one end's view.
//!
//! The paper's passive view comes from its own messages (JOIN, FORWARD_JOIN, SHUFFLE),
//! as its nodes know no member list. Ours has one, in the document, so the passive view
//! is drawn from it (`passive_view`): every member we may reach that isn't in the
//! active view. Those messages could replace it later without changing the rest.
//!
//! There is no graph every node shares (`Topology::graph`): what needs one does nothing
//! (writers' trees, recorded links, indirect probes), so changes pass on by a
//! dissemination that needs none (`--dissemination flood`, its default here, or
//! `plumtree`), or by announcements.
//!
//! A member being asked counts as a neighbour already: it gets HELLOs, and NEIGHBOUR
//! goes once the handshake is done (a swarm's members only take messages from peers
//! that did one). Refused, or unanswered for `ASK_FOR`, it is not asked again for
//! `REFUSED_FOR`, and stops counting.

use std::collections::{BTreeMap, BTreeSet};
use std::sync::Mutex;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Duration;

use async_trait::async_trait;
use tokio::time::Instant;
use tracing::debug;

use crate::locks::MutexExt;
use crate::state::{AppState, HelloState};
use crate::swarm::topology::{self, Inputs, Topology};
use crate::transport::{NeighbourAnswer, NeighbourRequest, ParsedMessage};

/// How long a member asked to be a neighbour has to agree, the handshake before
/// included: over Tor, a few round trips of up to 3 s.
const ASK_FOR: Duration = Duration::from_secs(20);

/// A member that refused, didn't answer, or left isn't asked again for this long.
const REFUSED_FOR: Duration = Duration::from_secs(60);

/// A member one neighbour short asks this seldom: in a swarm whose members mostly have
/// `k`, those that refuse are full, and asking each in turn kept NEIGHBOURs going in a
/// quiet swarm (up to 300 in 20 s at 50 members). Two or more short, it asks at once.
const ONE_SHORT_EVERY: Duration = Duration::from_secs(300);

/// Each neighbour is asked again this often whether it still is (NEIGHBOUR, as a new
/// ask would be): a NEIGHBOUR_REPLY or DISCONNECT lost on a lossy link leaves a link in
/// one end's view only, and nothing else would tell that end: the other still answers
/// it. A neighbour that has us agrees at once; one that doesn't takes us if it has room,
/// else refuses, and we drop it.
const REAFFIRM_EVERY: Duration = Duration::from_secs(60);

/// A member asked to be a neighbour.
#[derive(Debug)]
struct Asking {
    since: Instant,
    /// With high priority: we had no neighbour when we asked.
    high: bool,
    /// NEIGHBOUR sent (after the handshake).
    sent: bool,
}

/// The views.
#[derive(Debug, Default)]
struct Views {
    /// Our neighbours: each agreed, or asked us and we agreed.
    active: BTreeSet<String>,
    /// Asked, not answered yet.
    asking: BTreeMap<String, Asking>,
    /// Refused, unanswered or left lately: since when.
    refused: BTreeMap<String, Instant>,
    /// When we last asked while one neighbour short (`ONE_SHORT_EVERY`).
    asked_one_short: Option<Instant>,
    /// When the neighbours were last asked whether they still are (`REAFFIRM_EVERY`).
    reaffirmed: Option<Instant>,
}

#[derive(Default)]
pub(crate) struct HyParView {
    views: Mutex<Views>,
    /// Bumped whenever the views change what `neighbours` gives (`Topology::version`).
    version: AtomicU64,
}

impl HyParView {
    /// The members we could ask to be neighbours: from the member list, every member on
    /// one of our transports that we don't judge unreachable, not us. The paper keeps
    /// these from its own messages instead (`passive_view`, and the module's docs).
    fn reachable_members(local: &str, inputs: &Inputs<'_>) -> BTreeSet<String> {
        inputs
            .transports
            .iter()
            .filter(|(member, on)| {
                member.as_str() != local
                    && !on.is_disjoint(inputs.local_transports)
                    && !(inputs.unreachable)(member)
            })
            .map(|(member, _)| member.clone())
            .collect()
    }

    /// The passive view: whom we may ask when we lack neighbours. Drawn from the member
    /// list (`reachable_members`), not from JOIN and SHUFFLE as in the paper.
    fn passive_view(views: &Views, reachable: &BTreeSet<String>) -> Vec<String> {
        reachable
            .iter()
            .filter(|member| {
                !views.active.contains(*member)
                    && !views.asking.contains_key(*member)
                    && !views.refused.contains_key(*member)
            })
            .cloned()
            .collect()
    }

    fn changed(&self, state: &AppState) {
        self.version.fetch_add(1, Ordering::Relaxed);
        state.topology_changed.notify_one();
    }

    async fn send(state: &AppState, peer: &str, message: ParsedMessage) {
        if let Err(error) = state.transport.send_message(peer, &message).await {
            debug!(peer = %peer, error = %format!("{error:#}"), "A HyParView message failed");
        }
    }

    fn answer(state: &AppState, agreed: bool) -> NeighbourAnswer {
        let (epoch, epoch_seq) = state.epoch_stamp();
        NeighbourAnswer {
            epoch,
            epoch_seq,
            agreed,
        }
    }
}

/// `count` of `members`, picked at random.
fn some_of(mut members: Vec<String>, count: usize) -> Vec<String> {
    let count = count.min(members.len());
    for picked in 0..count {
        let rest = (members.len() - picked) as u64;
        let other = picked + (crate::transport::paths::random_nonce() % rest) as usize;
        members.swap(picked, other);
    }
    members.truncate(count);
    members
}

#[async_trait]
impl Topology for HyParView {
    fn name(&self) -> &'static str {
        "hyparview"
    }

    fn graph(
        &self,
        members: &[String],
        k: usize,
        unreachable: &dyn Fn(&str) -> bool,
        stable: &dyn Fn(&str) -> bool,
    ) -> Option<BTreeMap<String, BTreeSet<String>>> {
        let _ = (members, k, unreachable, stable);
        None
    }

    /// The active view, and those being asked.
    fn neighbours(&self, local: &str, inputs: &Inputs<'_>) -> BTreeSet<String> {
        let _ = (local, inputs);
        let views = self.views.lock_or_recover();
        views
            .active
            .iter()
            .chain(views.asking.keys())
            .cloned()
            .collect()
    }

    /// Failed neighbours dropped, unanswered asks given up, NEIGHBOUR sent to those
    /// asked once the handshake is done, and new members asked while we lack
    /// neighbours.
    async fn maintain(&self, state: &AppState, inputs: &Inputs<'_>) {
        let local = state.local_id().to_owned();
        let reachable = Self::reachable_members(&local, inputs);
        let now = Instant::now();
        let mut to_leave = Vec::new();
        let (changed, to_ask) = {
            let mut views = self.views.lock_or_recover();
            let mut changed = false;
            // Told all the same: one we judge unreachable may only be silent towards
            // us (a link that works one way), and kept us in its view, it would be our
            // neighbour at its end alone.
            let failed: Vec<String> = views
                .active
                .iter()
                .filter(|member| !reachable.contains(*member))
                .cloned()
                .collect();
            if !failed.is_empty() {
                debug!(
                    dropped = failed.len(),
                    "Neighbours judged unreachable left the active view"
                );
                for member in failed {
                    views.active.remove(&member);
                    to_leave.push(member);
                }
                changed = true;
            }
            let given_up: Vec<String> = views
                .asking
                .iter()
                .filter(|(member, asking)| {
                    !reachable.contains(*member) || now.duration_since(asking.since) > ASK_FOR
                })
                .map(|(member, _)| member.clone())
                .collect();
            for member in given_up {
                // Asked: told we gave up, should it agree after all (its answer lost,
                // or still on its way).
                if views
                    .asking
                    .remove(&member)
                    .is_some_and(|asking| asking.sent)
                {
                    to_leave.push(member.clone());
                }
                views.refused.insert(member, now);
                changed = true;
            }
            views
                .refused
                .retain(|_, since| now.duration_since(*since) < REFUSED_FOR);
            // More than one beyond `k` (asks sent before others asked us came back
            // agreed): the excess dropped, at random, and told.
            if views.active.len() > inputs.k + 1 {
                let excess = views.active.len() - inputs.k;
                let all: Vec<String> = views.active.iter().cloned().collect();
                for member in some_of(all, excess) {
                    views.active.remove(&member);
                    views.refused.insert(member.clone(), now);
                    to_leave.push(member);
                }
                changed = true;
            }
            let missing = inputs
                .k
                .saturating_sub(views.active.len() + views.asking.len());
            // One short, with neighbours: not again before `ONE_SHORT_EVERY`.
            let one_short = missing == 1 && !views.active.is_empty();
            let waiting = one_short
                && views
                    .asked_one_short
                    .is_some_and(|at| now.duration_since(at) < ONE_SHORT_EVERY);
            if missing > 0 && !waiting {
                if one_short {
                    views.asked_one_short = Some(now);
                }
                // With high priority only the first, while we have no neighbour: the
                // others would make room for us where none is needed.
                let mut high = views.active.is_empty() && views.asking.is_empty();
                for member in some_of(Self::passive_view(&views, &reachable), missing) {
                    views.asking.insert(
                        member,
                        Asking {
                            since: now,
                            high,
                            sent: false,
                        },
                    );
                    high = false;
                    changed = true;
                }
            }
            let mut to_ask: Vec<(String, bool)> = views
                .asking
                .iter()
                .filter(|(_, asking)| !asking.sent)
                .map(|(member, asking)| (member.clone(), asking.high))
                .collect();
            if views
                .reaffirmed
                .is_none_or(|at| now.duration_since(at) >= REAFFIRM_EVERY)
            {
                views.reaffirmed = Some(now);
                to_ask.extend(views.active.iter().map(|member| (member.clone(), false)));
            }
            (changed, to_ask)
        };
        if changed {
            self.version.fetch_add(1, Ordering::Relaxed);
        }
        for member in to_leave {
            Self::send(
                state,
                &member,
                ParsedMessage::Disconnect(Self::answer(state, false)),
            )
            .await;
        }
        for (member, high) in to_ask {
            let ready = match state.get_peer_state(&member).await {
                Some(peer_state) => peer_state.lock().await.hello_state == HelloState::Established,
                None => false,
            };
            if !ready {
                continue;
            }
            // A neighbour asked again isn't asking: nothing to mark.
            if let Some(asking) = self.views.lock_or_recover().asking.get_mut(&member) {
                asking.sent = true;
            }
            let (epoch, epoch_seq) = state.epoch_stamp();
            Self::send(
                state,
                &member,
                ParsedMessage::Neighbour(NeighbourRequest {
                    epoch,
                    epoch_seq,
                    high,
                }),
            )
            .await;
        }
    }

    async fn received(&self, state: &AppState, peer: &str, message: &ParsedMessage) {
        match message {
            ParsedMessage::Neighbour(request) => {
                let k = topology::k(state);
                let (agreed, dropped) = {
                    let mut views = self.views.lock_or_recover();
                    if views.active.contains(peer) {
                        (true, None)
                    } else if request.high || views.active.len() < k {
                        // Full: room made by dropping another, at random.
                        let dropped = if views.active.len() >= k {
                            let others: Vec<String> = views.active.iter().cloned().collect();
                            let dropped = some_of(others, 1).pop();
                            if let Some(dropped) = &dropped {
                                views.active.remove(dropped);
                                views.refused.insert(dropped.clone(), Instant::now());
                            }
                            dropped
                        } else {
                            None
                        };
                        views.active.insert(peer.to_owned());
                        views.asking.remove(peer);
                        views.refused.remove(peer);
                        // Asks not sent yet no longer needed: given up.
                        let room = k.saturating_sub(views.active.len());
                        let unsent: Vec<String> = views
                            .asking
                            .iter()
                            .filter(|(_, asking)| !asking.sent)
                            .map(|(member, _)| member.clone())
                            .collect();
                        let sent = views.asking.len() - unsent.len();
                        for member in unsent.into_iter().skip(room.saturating_sub(sent)) {
                            views.asking.remove(&member);
                        }
                        (true, dropped)
                    } else {
                        (false, None)
                    }
                };
                if let Some(dropped) = dropped {
                    Self::send(
                        state,
                        &dropped,
                        ParsedMessage::Disconnect(Self::answer(state, false)),
                    )
                    .await;
                }
                Self::send(
                    state,
                    peer,
                    ParsedMessage::NeighbourReply(Self::answer(state, agreed)),
                )
                .await;
                if agreed {
                    self.changed(state);
                }
            }
            ParsedMessage::NeighbourReply(answer) => {
                {
                    let mut views = self.views.lock_or_recover();
                    // Not asked: a neighbour asked again whether it still is
                    // (`REAFFIRM_EVERY`), dropped if it refused; or one we gave up on,
                    // and told so (`maintain`), which undoes its agreeing.
                    if views.asking.remove(peer).is_none() {
                        if answer.agreed || !views.active.remove(peer) {
                            return;
                        }
                        views.refused.insert(peer.to_owned(), Instant::now());
                        drop(views);
                        self.changed(state);
                        return;
                    }
                    if answer.agreed {
                        views.active.insert(peer.to_owned());
                    } else {
                        views.refused.insert(peer.to_owned(), Instant::now());
                    }
                }
                self.changed(state);
            }
            ParsedMessage::Disconnect(_) => {
                {
                    let mut views = self.views.lock_or_recover();
                    views.active.remove(peer);
                    views.asking.remove(peer);
                    views.refused.insert(peer.to_owned(), Instant::now());
                }
                self.changed(state);
            }
            _ => {}
        }
    }

    fn version(&self) -> u64 {
        self.version.load(Ordering::Relaxed)
    }

    /// Those we had as neighbours asked first: the handshakes are done, so NEIGHBOUR
    /// goes with the next `maintain`. Up to `k` of them, at random, as `k` isn't known
    /// here: the most `maintain` asks itself is what we lack.
    fn seed(&self, neighbours: &BTreeSet<String>) {
        let now = Instant::now();
        let mut views = self.views.lock_or_recover();
        for member in neighbours {
            views.asking.insert(
                member.clone(),
                Asking {
                    since: now,
                    high: false,
                    sent: false,
                },
            );
        }
    }

    #[cfg(test)]
    fn active(&self) -> Option<BTreeSet<String>> {
        Some(self.views.lock_or_recover().active.clone())
    }
}