//! How a change reaches every member: whom a node sends a writer's changes to at once,
//! and whom it leaves to the announcement rounds (`swarm::relay`, which repair what it
//! misses: announcements, pulls, the slow round). Writers' trees (`swarm::tree`) are
//! one way; others can be tried against them in the benchmark (`docs/future-work.md`,
//! Swappable topology and dissemination). One per swarm: a node's part in passing a
//! change on is worked out by itself and by its neighbours, and must agree.
//!
//! What every way keeps: the base underneath, pairs of neighbours syncing with
//! Automerge, which delivers everything in the end; a way only decides whom to push to
//! first, and so how soon.

use async_trait::async_trait;
use tracing::trace;

use crate::document::ChangeHash;
use crate::state::AppState;
use crate::swarm::tree::Plan;

#[async_trait]
pub(crate) trait Dissemination: Send + Sync + 'static {
    /// Its name (in logs, and once chosen by the swarm, in the document).
    fn name(&self) -> &'static str;

    /// Our part in passing on `writer`'s changes (by its pair key, `topology::pair_key`):
    /// whom we send them to at once, and how the announcement rounds treat our other
    /// neighbours (`Plan`). None: we have no part, and announcements do it all.
    async fn plan(&self, state: &AppState, writer: u64) -> Option<Plan>;

    /// `writer`'s `changes` came from `from` (none: we had them already, and they came
    /// as the writer's now): passed on at once to those we send them to, the others
    /// counted as having them where they get them elsewhere. Returns our part, as
    /// `plan`.
    async fn forward(
        &self,
        state: &AppState,
        writer: u64,
        from: &str,
        changes: &[ChangeHash],
    ) -> Option<Plan>;

    /// What a sync message from `peer` says about how well this works (`/admin/stats`):
    /// whether it came as `writer`'s and `carried` changes, and the new changes it
    /// brought, by author. Not while catching up.
    async fn note_arrival(
        &self,
        state: &AppState,
        peer: &str,
        writer: Option<u64>,
        carried: bool,
        authors: &[(String, Vec<ChangeHash>)],
    );

    /// `peer` asks us to send it `writer`'s changes at once (`eager`, a GRAFT) or to
    /// leave them to announcements (a PRUNE), for a way that keeps such links
    /// (PlumTree). By default nothing.
    async fn linked(&self, state: &AppState, peer: &str, writer: u64, eager: bool) {
        let _ = (state, peer, writer, eager);
    }

    /// `peer` told us of `heads` we lack (a STATUS: an announcement, or its keepalive's),
    /// for a way that minds which neighbour has changes first (PlumTree). By default
    /// nothing.
    fn announced(&self, peer: &str, heads: &[ChangeHash]) {
        let _ = (peer, heads);
    }

    /// The neighbours whose links are lazy, for a way that keeps such links (PlumTree),
    /// for the benchmark to report. By default none.
    #[cfg(test)]
    fn lazy(&self) -> Option<std::collections::HashSet<String>> {
        None
    }
}

/// What every way does with its `plan` for `writer`'s `changes` from `from` (none: we
/// had them already): those it leaves to others (`Plan::covered`) count as having them
/// (`SyncSession::known_to_peer`): else, with writers in turn, each change coming along
/// another way, the next message we pass on to one would carry all we got since our
/// last to it, mostly what its other parents brought it; should one lack them after
/// all, it asks, and gets them (`resend_if_needed`). Those it sends to
/// (`Plan::children`, but `from`) get them at once, with those of them that `changes`
/// build on, if they were counted so lately (`SyncSession::with_predecessors`).
pub(crate) async fn pass_on(
    state: &AppState,
    writer: u64,
    from: &str,
    changes: &[ChangeHash],
    mut plan: Plan,
) -> Plan {
    if !changes.is_empty() {
        for neighbour in &plan.covered {
            if let Some(peer_state) = state.get_peer_state(neighbour).await {
                peer_state
                    .lock()
                    .await
                    .sync
                    .known_to_peer(changes, tokio::time::Instant::now());
            }
        }
    }
    plan.children.retain(|child| child != from);
    let children = &plan.children;
    let replica = if changes.is_empty() {
        None
    } else {
        Some(state.doc.read().await)
    };
    for child in children {
        if let Some(peer_state) = state.get_peer_state(child).await {
            let mut peer_state = peer_state.lock().await;
            peer_state.tree_writer = Some(writer);
            if let Some(replica) = &replica {
                peer_state.sync.with_predecessors(
                    &replica.doc,
                    changes,
                    tokio::time::Instant::now(),
                );
            }
        }
        state.stats.tree_forward();
        state.request_sync(child);
    }
    if !children.is_empty() {
        trace!(
            writer,
            children = children.len(),
            "Passing a change on along the writer's tree"
        );
    }
    plan
}

/// The ways to choose from (`--dissemination`); one per swarm.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "kebab-case")]
pub(crate) enum Kind {
    /// Writers' trees (`swarm::tree`).
    #[default]
    WritersTrees,
    /// PlumTree (`swarm::plumtree`).
    Plumtree,
    /// Flooding (`swarm::flood`).
    Flood,
}

impl Kind {
    pub(crate) fn make(self) -> std::sync::Arc<dyn Dissemination> {
        match self {
            Kind::WritersTrees => std::sync::Arc::new(crate::swarm::tree::WritersTrees),
            Kind::Plumtree => std::sync::Arc::new(crate::swarm::plumtree::PlumTree::default()),
            Kind::Flood => std::sync::Arc::new(crate::swarm::flood::Flood),
        }
    }
}