//! 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;
    }
}

/// a, b and c all on `sim`, but nothing passes between a and c (as behind NAT): once a
/// judges c unreachable, it asks c's links through b, though it has c's address.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn links_are_relayed_to_members_that_dont_answer() {
    let sim = Sim::with_identity(64);
    let cut = crate::transport::sim::LinkParams {
        loss: 1.0,
        ..Default::default()
    };
    sim.net.set_link("a", "c", cut.clone());
    sim.net.set_link("c", "a", cut);
    let b = sim.start_on("b", &["sim"], &[], Start::Genesis);
    let a = sim.start_on("a", &["sim"], &["b"], Start::Join);
    let c = sim.start_on("c", &["sim"], &["b"], Start::Join);
    sim.wait_ready(&[&a, &b, &c], SOON).await;
    let (b_id, c_id) = (b.local_id().to_owned(), c.local_id().to_owned());
    {
        let (a, c_id) = (&a, c_id.clone());
        sim.wait_until(
            "a knows c, and that it doesn't answer",
            SOON * 3,
            move || {
                let c_id = c_id.clone();
                async move {
                    let listed = members::members(&a.doc.read().await.doc).contains(&c_id);
                    listed && a.transport.can_reach(&c_id) && !topology::reaches(a, &c_id).await
                }
            },
        )
        .await;
    }
    links::ask_all(
        &a,
        vec![b_id.clone(), c_id.clone()],
        Duration::from_secs(10),
    )
    .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!(!c.links.subscribers().contains(&a.local_id().to_owned()));
}

/// a and c can't talk; b and d both could relay. Once the relay a asks through is cut
/// off from c (and from the other relay, so it can't pass the ask on), it brings
/// nothing: after `FAILED_AFTER` a's renewal goes through the other one, and the
/// subscription at the first runs out by itself.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn links_fail_over_to_another_relay() {
    let sim = Sim::with_identity(67);
    let cut = crate::transport::sim::LinkParams {
        loss: 1.0,
        ..Default::default()
    };
    let cut_between = |x: &str, y: &str| {
        sim.net.set_link(x, y, cut.clone());
        sim.net.set_link(y, x, cut.clone());
    };
    cut_between("a", "c");
    let b = sim.start_on("b", &["sim"], &[], Start::Genesis);
    let d = sim.start_on("d", &["sim"], &["b"], Start::Join);
    let a = sim.start_on("a", &["sim"], &["b"], Start::Join);
    let c = sim.start_on("c", &["sim"], &["b"], Start::Join);
    sim.wait_ready(&[&a, &b, &c, &d], SOON).await;
    let (a_id, c_id) = (a.local_id().to_owned(), c.local_id().to_owned());
    {
        let (a, c_id) = (&a, c_id.clone());
        sim.wait_until(
            "a knows c, and that it doesn't answer",
            SOON * 3,
            move || {
                let c_id = c_id.clone();
                async move {
                    let listed = members::members(&a.doc.read().await.doc).contains(&c_id);
                    listed && !topology::reaches(a, &c_id).await
                }
            },
        )
        .await;
    }

    let lease = Duration::from_secs(3);
    links::ask_all(&a, vec![c_id.clone()], lease).await;
    {
        let (b, d, c_id, a_id) = (&b, &d, c_id.clone(), a_id.clone());
        sim.wait_until("one of b and d relays c's links to a", SOON, move || {
            let relayed = (c_id.clone(), vec![a_id.clone()]);
            async move {
                b.links.relaying().contains(&relayed) || d.links.relaying().contains(&relayed)
            }
        })
        .await;
    }
    let relayed = (c_id.clone(), vec![a_id.clone()]);
    // With their names on the simulated net, which links are cut by.
    let ((first, first_name), (other, other_name)) = if b.links.relaying().contains(&relayed) {
        ((&b, "b"), (&d, "d"))
    } else {
        ((&d, "d"), (&b, "b"))
    };

    // The relay in use reaches c no more, nor the other relay; a still reaches it.
    cut_between(first_name, "c");
    cut_between(first_name, other_name);
    let cut_at = tokio::time::Instant::now();

    // a renews as a dashboard does, and gets c's links through the other relay.
    let renewing = {
        let (a, c_id) = (a.clone(), c_id.clone());
        tokio::spawn(async move {
            loop {
                tokio::time::sleep(lease / 2).await;
                links::ask_all(&a, vec![c_id.clone()], lease).await;
            }
        })
    };
    {
        let (a, other, relayed, c_id) = (&a, other, relayed.clone(), c_id.clone());
        sim.wait_until(
            "the other relay relays c's links to a, and they arrive",
            SOON * 3,
            move || {
                let (relayed, c_id) = (relayed.clone(), c_id.clone());
                async move {
                    other.links.relaying().contains(&relayed)
                        && a.links
                            .replies()
                            .iter()
                            .any(|(from, age, _)| from == &c_id && *age < cut_at.elapsed())
                }
            },
        )
        .await;
    }
    {
        let (first, relayed) = (first, relayed.clone());
        sim.wait_until(
            "the first relay's subscription ran out",
            lease * 2,
            move || {
                let relayed = relayed.clone();
                async move { !first.links.relaying().contains(&relayed) }
            },
        )
        .await;
    }
    renewing.abort();
}