//! 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, MessageKind, ParsedMessage, RunId, Status};

/// Long enough for anything that should happen within a few keepalives.
const SOON: Duration = Duration::from_secs(10);
/// With lost messages: several reply timeouts (1 s, doubling) and heals in a row.
const LOSSY: Duration = Duration::from_secs(30);

#[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);
}

/// Lost messages are made up for: sync messages by the reply timeout, sessions out of
/// step by the keepalive heal. Joining under loss too: HELLOs are retried.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn converges_over_lossy_links() {
    let sim = Sim::new(5);
    sim.net.set_default_link(LinkParams {
        loss: 0.15,
        duplicate: 0.05,
        delay: (Duration::from_millis(1), Duration::from_millis(30)),
        ..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], LOSSY).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;
        tokio::time::sleep(Duration::from_millis(50)).await;
    }
    sim.wait_converged(&[a, b, c], LOSSY).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().lost > 0);
}

/// R24: a catch-up source that is alive (it answers handshakes and keepalives) but whose
/// sync messages never arrive is given up after three unanswered sync messages and
/// `delivery_limit` without one, and another peer takes over.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn stalled_catch_up_source_is_replaced() {
    let sim = Sim::new(6);
    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;
    write(a, "data", json!("x")).await;
    sim.wait_converged(&[a, b], SOON).await;
    // a's HELLO_REPLY lists only its neighbours, and b is one only once its
    // registration reached a: otherwise c would never hear of b.
    sim.wait_until("a counts b as a neighbour", SOON, move || async move {
        has_neighbour(a, "b")
    })
    .await;

    // c joins through a, so a becomes its source (the first ready peer it hears
    // from); b it learns of from a.
    sim.net.set_link(
        "a",
        "c",
        LinkParams {
            drop: vec![MessageKind::Sync],
            ..LinkParams::default()
        },
    );
    let c = sim.start("c", &["a"], Start::Join);
    let c = &c;
    sim.wait_ready(&[c], LOSSY).await;

    assert_eq!(read(c, "data").await, Some(json!("x")));
    let sources = c.stats.snapshot().catch_up_sources;
    assert!(sources >= 2, "{sources} catch-up sources");
}

/// A joining node whose only known peer stalls as its catch-up source asks it again
/// (a fresh HELLO): the reply lists the peers that joined meanwhile, and one of them
/// takes over.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn stalled_source_refreshes_the_peer_list() {
    let sim = Sim::new(7);
    let a = sim.start("a", &[], Start::Genesis);
    let a = &a;
    sim.wait_ready(&[a], SOON).await;
    write(a, "data", json!("x")).await;

    // c knows only a, which lists nobody: there is nobody else yet.
    sim.net.set_link(
        "a",
        "c",
        LinkParams {
            drop: vec![MessageKind::Sync],
            ..LinkParams::default()
        },
    );
    let c = sim.start("c", &["a"], Start::Join);
    let c = &c;
    sim.wait_until("c knows a", SOON, move || async move {
        c.get_peer_state("a").await.is_some()
    })
    .await;

    // b joins meanwhile; c hears of it only by asking a again (after giving it up, or
    // earlier if the keepalive heal re-handshakes first).
    let b = sim.start("b", &["a"], Start::Join);
    let b = &b;
    sim.wait_ready(&[b], SOON).await;

    sim.wait_ready(&[c], LOSSY).await;
    assert_eq!(read(c, "data").await, Some(json!("x")));
    assert!(c.get_peer_state("b").await.is_some());
}

/// 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)
}

/// R27: a resent HELLO (its reply was lost) or a late copy of one doesn't reset the
/// session it opened; only a new handshake does.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn resent_hello_keeps_the_session() {
    let sim = Sim::new(8);
    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;
    write(a, "data", json!(1)).await;
    sim.wait_converged(&[a, b], SOON).await;
    let opened = move || async move { peer(a, "b", |p| p.sync.has_opened()).await == Some(true) };
    sim.wait_until("a has heard b's heads", SOON, opened).await;

    let (run, handshake) = peer(a, "b", |p| p.last_hello).await.unwrap().unwrap();
    assert_eq!(run, b.run_id);
    let (epoch, epoch_seq) = b.epoch_stamp();
    let swarm_id = b.swarm_id().await;
    let hello = |handshake| {
        ParsedMessage::Hello(HelloRequest {
            swarm_id: swarm_id.clone(),
            epoch,
            epoch_seq,
            auth: Vec::new(),
            ready: true,
            run_id: run,
            handshake,
        })
    };
    for repeated in [handshake, handshake.saturating_sub(1)] {
        sim.net.inject("b", "a", &hello(repeated)).await.unwrap();
        assert!(opened().await, "HELLO {repeated} reset the session");
        assert_eq!(
            peer(a, "b", |p| p.last_hello).await.unwrap(),
            Some((run, handshake))
        );
    }
    sim.net
        .inject("b", "a", &hello(handshake + 1))
        .await
        .unwrap();
    assert_eq!(
        peer(a, "b", |p| p.last_hello).await.unwrap(),
        Some((run, handshake + 1))
    );
}

/// 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,
            handshake: 1,
        })
    };
    let status = |epoch: Epoch, epoch_seq: u32| {
        ParsedMessage::Status(Status {
            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", &status(epoch(2), 2))
        .await
        .unwrap();
    assert_eq!(recorded().await, (Some(epoch(2)), Some(2)));

    // The STATUS from between the switches arrives late: ignored.
    sim.net
        .inject("ghost", "a", &status(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", &status(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", &status(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);
}

/// A swarm-wide forget of a node that is alive: nobody forgets it, and it cancels the
/// record itself.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn forgotten_node_that_is_alive_cancels_its_record() {
    use crate::document::store;
    use crate::swarm::peer::forgotten_path;

    let sim = Sim::new(9);
    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 record = forgotten_path(b.local_id());
    let written = record.clone();
    a.mutate_doc(move |doc| store::write_path(doc, &written, &json!({"at": 0, "by": "a"})))
        .await
        .expect("write forget record");

    // The forget check runs every 5 s.
    for node in [a, b, c] {
        let record = record.clone();
        sim.wait_until("the record is cancelled", SOON + SOON, move || {
            let record = record.clone();
            async move {
                store::read_path(&node.doc.read().await.doc, &record)
                    .expect("read")
                    .is_none()
            }
        })
        .await;
    }
    for node in [a, c] {
        assert!(node.get_peer_state(b.local_id()).await.is_some());
    }
    sim.wait_converged(&[a, b, c], SOON).await;
}