//! The timing rules of `peers.md` on whole nodes: joining and converging, dormancy and
//! return, and late messages that must not undo what newer ones said.

use std::time::Duration;

use serde_json::json;

use super::{Sim, has_neighbour, heads, peer, read, write};
use crate::document::replica::Epoch;
use crate::state::{AppState, Liveness};
use crate::swarm::bootstrap::Start;
use crate::transport::sim::LinkParams;
use crate::transport::{HeadsSummary, HelloRequest, Keepalive, ParsedMessage, RunId};

/// Long enough for anything that should happen within a few keepalives.
const SOON: Duration = Duration::from_secs(10);

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn three_nodes_join_and_converge() {
    let sim = Sim::new(1);
    let a = sim.start("a", &[], Start::Genesis);
    let b = sim.start("b", &["a"], Start::Join);
    let c = sim.start("c", &["a"], Start::Join);
    let (a, b, c) = (&a, &b, &c);
    sim.wait_ready(&[a, b, c], SOON).await;

    let swarm = a.swarm_id().await;
    assert!(swarm.is_some());
    assert_eq!(b.swarm_id().await, swarm);
    assert_eq!(c.swarm_id().await, swarm);

    // A write anywhere reaches everyone.
    write(b, "users/bob", json!({"name": "Bob"})).await;
    sim.wait_until("c has b's write", SOON, move || async move {
        read(c, "users/bob/name").await == Some(json!("Bob"))
    })
    .await;

    // Concurrent writes merge to one document.
    write(a, "users/alice", json!(1)).await;
    write(c, "users/carol", json!(3)).await;
    sim.wait_converged(&[a, b, c], SOON).await;
    for node in [a, b, c] {
        assert_eq!(read(node, "users/alice").await, Some(json!(1)));
        assert_eq!(read(node, "users/carol").await, Some(json!(3)));
    }
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn converges_with_duplicated_and_reordered_messages() {
    let sim = Sim::new(2);
    sim.net.set_default_link(LinkParams {
        duplicate: 0.3,
        delay: (Duration::from_millis(1), Duration::from_millis(40)),
        ..LinkParams::default()
    });
    let a = sim.start("a", &[], Start::Genesis);
    let b = sim.start("b", &["a"], Start::Join);
    let c = sim.start("c", &["a"], Start::Join);
    let (a, b, c) = (&a, &b, &c);
    sim.wait_ready(&[a, b, c], SOON).await;

    for round in 0..10 {
        write(a, &format!("a/{round}"), json!(round)).await;
        write(b, &format!("b/{round}"), json!(round)).await;
        write(c, &format!("c/{round}"), json!(round)).await;
    }
    sim.wait_converged(&[a, b, c], SOON).await;
    for node in [a, b, c] {
        for prefix in ["a", "b", "c"] {
            assert_eq!(read(node, &format!("{prefix}/9")).await, Some(json!(9)));
        }
    }
    assert!(sim.net.counts().duplicated > 0);
}

/// A peer that hears nothing for `dormant_after` is dormant; once messages flow again
/// it is active, re-handshakes, and syncs what it missed.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn peer_goes_dormant_and_returns() {
    let sim = Sim::new(3);
    let a = sim.start("a", &[], Start::Genesis);
    let b = sim.start("b", &["a"], Start::Join);
    let (a, b) = (&a, &b);
    sim.wait_ready(&[a, b], SOON).await;
    // Only neighbours are judged, and a node joins the member list (which neighbours
    // are computed from) within 5 s of becoming ready: until then the other doesn't
    // count it as a neighbour.
    sim.wait_until(
        "a and b are each other's neighbours",
        SOON,
        move || async move { has_neighbour(a, "b") && has_neighbour(b, "a") },
    )
    .await;

    sim.net.take_down("b");
    sim.wait_until("a finds b dormant", SOON, move || is_dormant(a, "b"))
        .await;
    sim.wait_until("b finds a dormant", SOON, move || is_dormant(b, "a"))
        .await;
    let dormancies = peer(a, "b", |p| p.incidents.dormancies).await.unwrap();
    assert!(dormancies >= 1);

    // Written while apart.
    write(a, "while/apart", json!(true)).await;
    assert_ne!(heads(a).await, heads(b).await);

    sim.net.bring_up("b");
    sim.wait_until("a finds b active", SOON, move || async move {
        !is_dormant(a, "b").await
    })
    .await;
    sim.wait_until("b finds a active", SOON, move || async move {
        !is_dormant(b, "a").await
    })
    .await;
    sim.wait_until("b has what a wrote meanwhile", SOON, move || async move {
        read(b, "while/apart").await == Some(json!(true))
    })
    .await;
    sim.wait_converged(&[a, b], SOON).await;
}

async fn is_dormant(node: &AppState, other: &str) -> bool {
    peer(node, other, |p| {
        matches!(p.liveness, Liveness::Dormant { .. })
    })
    .await
        == Some(true)
}

/// R17: a message sent before the peer's last epoch change and delivered late doesn't
/// flip the epoch we recorded for it back; a new run of the peer starts its count over.
/// The peer is a "ghost" whose messages are injected, so their order is the test's.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn late_epoch_does_not_flip_a_peers_epoch() {
    let sim = Sim::new(4);
    let a = sim.start("a", &[], Start::Genesis);
    sim.wait_ready(&[&a], SOON).await;
    let swarm_id = a.swarm_id().await.unwrap();

    let epoch = |generation: u64| Epoch {
        generation,
        id: [generation as u8; 32],
    };
    let hello = |run: RunId, epoch: Epoch, epoch_seq: u32| {
        ParsedMessage::Hello(HelloRequest {
            swarm_id: Some(swarm_id.clone()),
            epoch,
            epoch_seq,
            auth: Vec::new(),
            ready: true,
            run_id: run,
        })
    };
    let ping = |epoch: Epoch, epoch_seq: u32| {
        ParsedMessage::Ping(Keepalive {
            heads: HeadsSummary::of(Vec::new()),
            epoch,
            epoch_seq,
            ready: true,
            syncs_with_you: false,
        })
    };
    let a = &a;
    let recorded = move || async move {
        peer(a, "ghost", |p| (p.epoch, p.epoch_seq))
            .await
            .expect("a knows the ghost")
    };

    let first_run: RunId = [1; 16];
    sim.net
        .inject("ghost", "a", &hello(first_run, Epoch::GENESIS, 0))
        .await
        .unwrap();
    assert_eq!(recorded().await, (Some(Epoch::GENESIS), Some(0)));

    // The ghost switched twice; we hear of the second switch first.
    sim.net
        .inject("ghost", "a", &ping(epoch(2), 2))
        .await
        .unwrap();
    assert_eq!(recorded().await, (Some(epoch(2)), Some(2)));

    // The PING from between the switches arrives late: ignored.
    sim.net
        .inject("ghost", "a", &ping(epoch(1), 1))
        .await
        .unwrap();
    assert_eq!(recorded().await, (Some(epoch(2)), Some(2)));
    // So is one from before both.
    sim.net
        .inject("ghost", "a", &ping(Epoch::GENESIS, 0))
        .await
        .unwrap();
    assert_eq!(recorded().await, (Some(epoch(2)), Some(2)));

    // The ghost restarted: its new run counts from 0 again.
    let second_run: RunId = [2; 16];
    sim.net
        .inject("ghost", "a", &hello(second_run, epoch(2), 0))
        .await
        .unwrap();
    assert_eq!(recorded().await, (Some(epoch(2)), Some(0)));
    assert_eq!(
        peer(a, "ghost", |p| p.run_id).await.unwrap(),
        Some(second_run)
    );
    sim.net
        .inject("ghost", "a", &ping(epoch(3), 1))
        .await
        .unwrap();
    assert_eq!(recorded().await, (Some(epoch(3)), Some(1)));

    // One ahead peer isn't enough for a ready node to rejoin.
    assert_eq!(a.local_epoch(), Epoch::GENESIS);
    assert!(a.is_ready().await);
}