//! Remakes on whole nodes (`docs/remakes.md`): a compaction everywhere, competing
//! proposals across a partition, a node that misses the switch, a coordinator that is
//! gone, a skewed clock, writes while a node starts to rejoin, and two remakes committed
//! for one epoch (a coordinator cut off after it committed; electorates of a partition
//! older than the electorate window).
//!
//! The simulated nodes' timings make a remake quick: settle 2 s (two `dormant_after`),
//! switch delay a keepalive, votes within half a keepalive.

use std::time::Duration;

use axum::http::StatusCode;
use serde_json::{Value as JsonValue, json};

use super::{SOON, Sim, read, write};
use crate::document::ChangeHash;
use crate::document::replica::Epoch;
use crate::document::store;
use crate::remake::testing;
use crate::state::AppState;
use crate::swarm::bootstrap::Start;
use crate::transport::sim::LinkParams;

/// A remake from proposal to every node switched, rejoins included.
const REMAKE: Duration = Duration::from_secs(30);

/// Three nodes, ready, registered in the member list (so in the electorate) and in step.
async fn three(sim: &Sim) -> (AppState, AppState, AppState) {
    let a = sim.start("a", &[], Start::Genesis);
    let b = sim.start("b", &["a"], Start::Join);
    let c = sim.start("c", &["a"], Start::Join);
    sim.wait_ready(&[&a, &b, &c], SOON).await;
    let nodes = [a.clone(), b.clone(), c.clone()];
    sim.wait_until("every node is registered", SOON, || {
        let nodes = nodes.clone();
        async move {
            for node in &nodes {
                let members = crate::swarm::members::members(&node.doc.read().await.doc);
                if members.len() < 3 {
                    return false;
                }
            }
            true
        }
    })
    .await;
    (a, b, c)
}

async fn propose(node: &AppState, body: JsonValue) -> String {
    let (status, report) = testing::start(node.clone(), body, None).await;
    assert!(status.is_success(), "proposal refused: {status} {report}");
    report["id"].as_str().expect("remake id").to_owned()
}

async fn on_generation(sim: &Sim, nodes: &[&AppState], generation: u64) {
    sim.wait_until(
        &format!("every node is on generation {generation}"),
        REMAKE,
        move || async move {
            for node in nodes {
                if node.local_epoch().generation != generation || !node.is_ready().await {
                    return false;
                }
            }
            true
        },
    )
    .await;
}

fn same_epoch(nodes: &[&AppState]) -> Epoch {
    let epoch = nodes[0].local_epoch();
    for node in nodes {
        assert_eq!(node.local_epoch(), epoch);
    }
    epoch
}

/// A compaction moves every node to the next epoch with the data as it was, and the
/// swarm works on in it.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn compaction_switches_every_node() {
    let sim = Sim::new(31);
    let (a, b, c) = three(&sim).await;
    write(&b, "before", json!(1)).await;
    sim.wait_converged(&[&a, &b, &c], SOON).await;

    propose(&a, json!({ "kind": "compact" })).await;
    on_generation(&sim, &[&a, &b, &c], 1).await;
    same_epoch(&[&a, &b, &c]);
    for node in [&a, &b, &c] {
        assert_eq!(read(node, "before").await, Some(json!(1)));
    }

    write(&c, "after", json!(2)).await;
    let a = &a;
    sim.wait_until(
        "a has what c wrote in the new epoch",
        SOON,
        move || async move { read(a, "after").await == Some(json!(2)) },
    )
    .await;
}

/// Two proposals on two sides of a partition: the side with the majority commits, the
/// other's can't, and once the partition heals every node is on the one new epoch.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn competing_proposals_commit_at_most_once() {
    let sim = Sim::new(32);
    let (a, b, c) = three(&sim).await;
    write(&a, "data", json!("x")).await;
    sim.wait_converged(&[&a, &b, &c], SOON).await;

    let cut = LinkParams {
        loss: 1.0,
        ..LinkParams::default()
    };
    for other in ["b", "c"] {
        sim.net.set_link("a", other, cut.clone());
        sim.net.set_link(other, "a", cut.clone());
    }
    let lonely = propose(&a, json!({ "kind": "compact" })).await;
    propose(&b, json!({ "kind": "compact" })).await;
    on_generation(&sim, &[&b, &c], 1).await;
    assert_eq!(a.local_epoch().generation, 0, "a's remake had no majority");

    for other in ["b", "c"] {
        sim.net.set_link("a", other, LinkParams::default());
        sim.net.set_link(other, "a", LinkParams::default());
    }
    on_generation(&sim, &[&a, &b, &c], 1).await;
    same_epoch(&[&a, &b, &c]);
    assert_eq!(read(&a, "data").await, Some(json!("x")));
    // a's own remake never committed (its record stayed in the old epoch).
    assert!(
        testing::report(&a, &lonely)
            .await
            .is_none_or(|report| report["phase"] != "committed")
    );
}

/// A node that is away during the remake catches up with the new epoch when it
/// returns, and what it wrote meanwhile isn't lost.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn node_away_during_the_switch_rejoins() {
    let sim = Sim::new(33);
    let (a, b, c) = three(&sim).await;
    write(&a, "data", json!(1)).await;
    sim.wait_converged(&[&a, &b, &c], SOON).await;

    sim.net.take_down("c");
    write(&c, "while/away", json!(true)).await;
    // c counts in the electorate (its alive stamp is recent); a and b are a majority.
    propose(&a, json!({ "kind": "compact" })).await;
    on_generation(&sim, &[&a, &b], 1).await;
    assert_eq!(c.local_epoch().generation, 0);

    sim.net.bring_up("c");
    on_generation(&sim, &[&a, &b, &c], 1).await;
    same_epoch(&[&a, &b, &c]);
    assert_eq!(read(&c, "data").await, Some(json!(1)));
    let a = &a;
    sim.wait_until(
        "a has what c wrote while away",
        REMAKE,
        move || async move { read(a, "while/away").await == Some(json!(true)) },
    )
    .await;
}

/// A coordinator that is gone before it decides: its remake expires, another node
/// aborts it (which frees the votes), and a new proposal commits without it.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn remake_of_a_gone_coordinator_expires_and_is_aborted() {
    let sim = Sim::new(34);
    let (a, b, c) = three(&sim).await;

    // The proposal reaches b and c, but their votes never reach a.
    let cut = LinkParams {
        loss: 1.0,
        ..LinkParams::default()
    };
    sim.net.set_link("b", "a", cut.clone());
    sim.net.set_link("c", "a", cut);
    let orphan = propose(&a, json!({ "kind": "compact" })).await;
    let (b_ref, c_ref, orphan_ref) = (&b, &c, orphan.as_str());
    sim.wait_until("b has the proposal", SOON, move || async move {
        testing::report(b_ref, orphan_ref).await.is_some()
    })
    .await;
    sim.net.take_down("a");

    // Expired after its settle period plus a grace period, abortable a grace period
    // later (until then the answer is 409).
    sim.wait_until(
        "the orphaned remake can be aborted",
        REMAKE,
        move || async move {
            let (status, _) = testing::abort(b_ref.clone(), orphan_ref, None).await;
            status == StatusCode::OK
        },
    )
    .await;
    sim.wait_until("c sees it aborted", SOON, move || async move {
        testing::report(c_ref, orphan_ref)
            .await
            .is_some_and(|report| report["phase"] == "aborted")
    })
    .await;

    propose(&b, json!({ "kind": "compact" })).await;
    on_generation(&sim, &[&b, &c], 1).await;
    same_epoch(&[&b, &c]);
}

/// A node whose clock is ahead (within the grace period) takes part as any other.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn compaction_with_a_skewed_clock() {
    let sim = Sim::new(35);
    let (a, b, c) = three(&sim).await;
    // dormant_after (the grace period) is 1 s here.
    c.clock.set_offset_ms(700);
    write(&a, "data", json!(1)).await;
    sim.wait_converged(&[&a, &b, &c], SOON).await;

    propose(&c, json!({ "kind": "compact" })).await;
    on_generation(&sim, &[&a, &b, &c], 1).await;
    same_epoch(&[&a, &b, &c]);
    assert_eq!(read(&b, "data").await, Some(json!(1)));
}

/// Writes on a node right as it starts to rejoin (`Replica::writable`): each is either
/// answered and kept, or refused; none lands in the fresh document before the epoch's
/// snapshot, which every change of the new epoch descends from.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn writes_while_a_rejoin_starts_are_kept_or_refused() {
    use std::sync::atomic::{AtomicBool, Ordering};
    use std::sync::{Arc, Mutex};

    let sim = Sim::new(36);
    let (a, b, c) = three(&sim).await;
    sim.net.take_down("c");
    propose(&a, json!({ "kind": "compact" })).await;
    on_generation(&sim, &[&a, &b], 1).await;

    // c writes on and on, through its return and rejoin.
    let answered = Arc::new(Mutex::new(Vec::<u64>::new()));
    let stop = Arc::new(AtomicBool::new(false));
    let writer = {
        let (c, answered, stop) = (c.clone(), Arc::clone(&answered), Arc::clone(&stop));
        tokio::spawn(async move {
            let mut n = 0_u64;
            while !stop.load(Ordering::Relaxed) {
                let at = vec!["w".to_owned(), n.to_string()];
                let written = c
                    .mutate_doc(move |doc| store::write_path(doc, &at, &json!(n)))
                    .await;
                if written.is_ok() {
                    answered.lock().unwrap().push(n);
                }
                n += 1;
                tokio::time::sleep(Duration::from_millis(2)).await;
            }
        })
    };
    tokio::time::sleep(Duration::from_millis(200)).await;
    sim.net.bring_up("c");
    on_generation(&sim, &[&a, &b, &c], 1).await;
    tokio::time::sleep(Duration::from_millis(200)).await;
    stop.store(true, Ordering::Relaxed);
    writer.await.unwrap();

    // Every change of c's document descends from the epoch's snapshot.
    let epoch = same_epoch(&[&a, &b, &c]);
    {
        let replica = c.doc.read().await;
        let snapshot = ChangeHash(epoch.id);
        let deps: std::collections::HashMap<ChangeHash, Vec<ChangeHash>> = replica
            .doc
            .get_changes(&[])
            .iter()
            .map(|change| (change.hash(), change.deps().to_vec()))
            .collect();
        for start in deps.keys() {
            let (mut todo, mut seen) = (vec![*start], std::collections::HashSet::new());
            let mut reached = false;
            while let Some(hash) = todo.pop() {
                if hash == snapshot {
                    reached = true;
                    break;
                }
                if seen.insert(hash) {
                    todo.extend(deps.get(&hash).into_iter().flatten().copied());
                }
            }
            assert!(reached, "change {start} doesn't descend from the snapshot");
        }
    }
    // Every write answered is there (those before the rejoin replayed into it).
    let answered = answered.lock().unwrap().clone();
    assert!(!answered.is_empty());
    let c = &c;
    sim.wait_until("c has every write it answered", SOON, move || {
        let answered = answered.clone();
        async move {
            for n in &answered {
                if read(c, &format!("w/{n}")).await != Some(json!(n)) {
                    return false;
                }
            }
            true
        }
    })
    .await;
}

/// A coordinator cut off right after it committed: the others never hear of the commit,
/// abort the remake once it expired and commit another. The coordinator switches to its
/// own; once back, it sees more peers on the other epoch of that generation, rejoins
/// that one sideways (`catchup::sideways`), and the swarm is on one epoch again.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn coordinator_cut_off_after_committing_rejoins_the_others() {
    let sim = Sim::new(37);
    let (a, b, c) = three(&sim).await;
    write(&b, "data", json!(1)).await;
    sim.wait_converged(&[&a, &b, &c], SOON).await;

    let cut = LinkParams {
        loss: 1.0,
        ..LinkParams::default()
    };
    let lost = propose(&a, json!({ "kind": "compact" })).await;
    let (a_ref, b_ref, c_ref, lost_ref) = (&a, &b, &c, lost.as_str());
    sim.wait_until("b and c have the proposal", SOON, move || async move {
        testing::report(b_ref, lost_ref).await.is_some()
            && testing::report(c_ref, lost_ref).await.is_some()
    })
    .await;
    // Their votes still reach a; nothing from a reaches them.
    for other in ["b", "c"] {
        sim.net.set_link("a", other, cut.clone());
    }
    sim.wait_until("a committed", SOON, move || async move {
        testing::report(a_ref, lost_ref)
            .await
            .is_some_and(|report| report["phase"] == "committed")
    })
    .await;
    sim.net.take_down("a");
    on_generation(&sim, &[&a], 1).await;

    // For b and c, the remake expired undecided.
    sim.wait_until("b can abort the lost remake", REMAKE, move || async move {
        let (status, _) = testing::abort(b_ref.clone(), lost_ref, None).await;
        status == StatusCode::OK
    })
    .await;
    propose(&b, json!({ "kind": "compact" })).await;
    on_generation(&sim, &[&b, &c], 1).await;
    let theirs = same_epoch(&[&b, &c]);
    assert_ne!(a.local_epoch(), theirs, "two epochs of generation 1");

    for other in ["b", "c"] {
        sim.net.set_link("a", other, LinkParams::default());
    }
    sim.net.bring_up("a");
    sim.wait_until("a is on b's and c's epoch", REMAKE, move || async move {
        a_ref.local_epoch() == theirs && a_ref.is_ready().await
    })
    .await;
    assert_eq!(read(&a, "data").await, Some(json!(1)));
}

/// Electorates that don't overlap: on one side of a partition the other side's alive
/// stamps look older than `--electorate-window` (here by the clocks of d and e, a month
/// ahead, as a partition that long would make them), so each side's remake gets a
/// majority of its own electorate and both commit. Once the partition heals, the two
/// nodes of the smaller side rejoin the larger side's epoch.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn both_sides_of_an_old_partition_commit_and_converge() {
    use crate::remake::Host;

    let sim = Sim::new(38);
    let a = sim.start("a", &[], Start::Genesis);
    let [b, c, d, e] = ["b", "c", "d", "e"].map(|id| sim.start(id, &["a"], Start::Join));
    let all = [&a, &b, &c, &d, &e];
    sim.wait_ready(&all, SOON).await;
    let nodes = [a.clone(), b.clone(), c.clone(), d.clone(), e.clone()];
    sim.wait_until("every node is registered", SOON, || {
        let nodes = nodes.clone();
        async move {
            for node in &nodes {
                if crate::swarm::members::members(&node.doc.read().await.doc).len() < 5 {
                    return false;
                }
            }
            true
        }
    })
    .await;
    write(&a, "data", json!(1)).await;
    sim.wait_converged(&all, SOON).await;

    let cut = LinkParams {
        loss: 1.0,
        ..LinkParams::default()
    };
    let link = |params: &LinkParams| {
        for one in ["a", "b", "c"] {
            for other in ["d", "e"] {
                sim.net.set_link(one, other, params.clone());
                sim.net.set_link(other, one, params.clone());
            }
        }
    };
    link(&cut);
    let month = 31 * 24 * 3600 * 1000;
    d.clock.set_offset_ms(month);
    e.clock.set_offset_ms(month);
    // d and e renew their stamps (by their clocks, theirs were a month old), and count
    // only themselves.
    let d_ref = &d;
    sim.wait_until("d's electorate is d and e", SOON, move || async move {
        let doc = d_ref.doc.read().await;
        d_ref.electorate(&doc.doc, d_ref.now_millis()).len() == 2
    })
    .await;

    propose(&a, json!({ "kind": "compact" })).await;
    propose(&d, json!({ "kind": "compact" })).await;
    on_generation(&sim, &[&a, &b, &c], 1).await;
    on_generation(&sim, &[&d, &e], 1).await;
    let larger = same_epoch(&[&a, &b, &c]);
    assert_ne!(same_epoch(&[&d, &e]), larger, "two epochs of generation 1");

    link(&LinkParams::default());
    let (d_ref, e_ref) = (&d, &e);
    sim.wait_until("d and e are on a's epoch", REMAKE, move || async move {
        d_ref.local_epoch() == larger
            && e_ref.local_epoch() == larger
            && d_ref.is_ready().await
            && e_ref.is_ready().await
    })
    .await;
    assert_eq!(read(&e, "data").await, Some(json!(1)));
}