//! Changes passed on in a full mesh (`swarm::relay`) to the members the writer can't
//! reach: those on none of its transports.

use std::time::Duration;

use serde_json::json;

use super::{SYNC_REPLY_TIMEOUT, Sim, read, write};
use crate::state::Timing;
use crate::swarm::bootstrap::Start;
use crate::swarm::topology;
use crate::transport::sim::DELIVERY_LIMIT;

const SOON: Duration = Duration::from_secs(30);

/// a on `sim` alone, c on `sim2` alone, b on both; a full mesh at each (every member
/// it reaches is its neighbour), so a pushes its write to b alone, and b, which only
/// received it, must announce it to c. With a long keepalive, the regular STATUS (every
/// third keepalive, 15 s) can't be what brings it: c has it within a second (an
/// announcement waits 200 ms, then c pulls). Without the announcement, a STATUS that
/// happens to fall into that second would pass it too, once in about 15 runs.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn changes_reach_members_the_writer_cant_reach_at_once() {
    let keepalive = Duration::from_secs(5);
    let slow = move |config: &mut crate::node::NodeConfig| {
        config.timing = Timing::new(keepalive, SYNC_REPLY_TIMEOUT, DELIVERY_LIMIT);
    };
    let sim = Sim::with_identity(62);
    let a = sim.start_on_with("a", &["sim"], &[], Start::Genesis, slow);
    let b = sim.start_on_with("b", &["sim", "sim2"], &["a"], Start::Join, slow);
    let c = sim.start_on_with("c", &["sim2"], &["b"], Start::Join, slow);
    sim.wait_ready(&[&a, &b, &c], SOON).await;
    let c_id = c.local_id().to_owned();
    {
        let (a, b, c_id) = (&a, &b, c_id.clone());
        sim.wait_until(
            "a and b know their neighbours, a full mesh, c beyond a's reach",
            SOON,
            move || {
                let c_id = c_id.clone();
                async move {
                    topology::view_known(a)
                        && topology::view_known(b)
                        && topology::is_full_mesh(a)
                        && topology::is_full_mesh(b)
                        && topology::is_neighbour(b, &c_id)
                        && topology::transports_of(a, &c_id).contains("sim2")
                        && !a.transport.can_reach(&c_id)
                }
            },
        )
        .await;
    }
    sim.wait_converged(&[&a, &b, &c], SOON).await;

    write(&a, "x", json!(1)).await;
    {
        let c = &c;
        sim.wait_until(
            "c has a's write within a second",
            Duration::from_secs(1),
            move || async move { read(c, "x").await == Some(json!(1)) },
        )
        .await;
    }
}