//! Links relayed (`swarm::links`): a member the asker can't reach, on a transport the
//! asker isn't on, is asked through a node on both, for as long as the asker renews.

use std::time::Duration;

use serde_json::json;

use super::{Sim, read, write};
use crate::state::AppState;
use crate::swarm::bootstrap::Start;
use crate::swarm::{links, members, topology};

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

/// Whether `node` has `member`'s links, listing `peer`.
fn has_links(node: &AppState, member: &str, peer: &str) -> bool {
    node.links
        .replies()
        .iter()
        .any(|(from, _, links)| from == member && links.iter().any(|link| link.peer == peer))
}

/// a on `sim` alone, c on `sim2` alone, b on both: a and c never reach each other, yet
/// a's writes reach c, and a gets c's links through b, while it renews; once it stops,
/// the subscription runs out at b and at c.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn links_are_relayed_to_members_out_of_reach() {
    let sim = Sim::with_identity(61);
    let a = sim.start_on("a", &["sim"], &[], Start::Genesis);
    let b = sim.start_on("b", &["sim", "sim2"], &["a"], Start::Join);
    let c = sim.start_on("c", &["sim2"], &["b"], Start::Join);
    sim.wait_ready(&[&a, &b, &c], SOON).await;
    let (a_id, b_id, c_id) = (
        a.local_id().to_owned(),
        b.local_id().to_owned(),
        c.local_id().to_owned(),
    );

    {
        let (a, c_id) = (&a, c_id.clone());
        sim.wait_until("a knows c, out of its reach", SOON, move || {
            let c_id = c_id.clone();
            async move {
                let listed = members::members(&a.doc.read().await.doc).contains(&c_id);
                // On `sim2` as a's view of the member list has it, which picks the relay.
                let on = topology::transports_of(a, &c_id);
                listed && on.contains("sim2") && !a.transport.can_reach(&c_id)
            }
        })
        .await;
    }

    // Changes pass through b.
    write(&a, "x", json!(1)).await;
    {
        let c = &c;
        sim.wait_until("c has a's write", SOON, move || async move {
            read(c, "x").await == Some(json!(1))
        })
        .await;
    }

    // c's links reach a through b; b asks c for a.
    let lease = Duration::from_secs(3);
    links::ask_all(&a, vec![b_id.clone(), c_id.clone()], lease).await;
    {
        let (a, b_id, c_id) = (&a, b_id.clone(), c_id.clone());
        sim.wait_until("a has c's links, relayed by b", SOON, move || {
            let (b_id, c_id) = (b_id.clone(), c_id.clone());
            async move { has_links(a, &c_id, &b_id) }
        })
        .await;
    }
    assert!(
        b.links
            .relaying()
            .contains(&(c_id.clone(), vec![a_id.clone()])),
        "{:?}",
        b.links.relaying()
    );
    assert!(c.links.subscribers().contains(&b_id));
    assert!(!c.links.subscribers().contains(&a_id));

    // Renewed: it lasts beyond its lease.
    for _ in 0..3 {
        tokio::time::sleep(lease / 2).await;
        links::ask_all(&a, vec![c_id.clone()], lease).await;
    }
    assert!(!b.links.relaying().is_empty());
    assert!(c.links.subscribers().contains(&b_id));

    // Not renewed: it runs out at every hop, without a message.
    {
        let (b, c, b_id) = (&b, &c, b_id.clone());
        sim.wait_until(
            "the relayed subscription ran out at b and at c",
            lease * 2,
            move || {
                let b_id = b_id.clone();
                async move { b.links.relaying().is_empty() && !c.links.subscribers().contains(&b_id) }
            },
        )
        .await;
    }
}