//! 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, and a skewed clock.
//!
//! 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::{Sim, read, write};
use crate::document::replica::Epoch;
use crate::remake::testing;
use crate::state::AppState;
use crate::swarm::bootstrap::Start;
use crate::transport::sim::LinkParams;

/// Joining, registering and converging.
const SOON: Duration = Duration::from_secs(10);
/// 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)));
}