//! Nodes under the identity layer (`transport::identity`): named by node IDs, found
//! through addresses (seeds, HELLO_REPLYs, the member list), and believed only once a
//! node proved itself at an address.

use std::sync::Arc;
use std::time::Duration;

use serde_json::json;

use super::{Sim, read, sim_key, write};
use crate::identity::{self, NodeKey};
use crate::state::AppState;
use crate::swarm::bootstrap::Start;
use crate::swarm::members;
use crate::transport::identity::{WHO_LABEL, signed};
use crate::transport::sim::SimNet;
use crate::transport::{HeadsSummary, IdentityProof, Keepalive, ParsedMessage, Transport};

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

async fn peers_are_node_ids(node: &AppState) {
    for (peer, _) in node.peers_snapshot().await {
        assert!(identity::is_node_id(&peer), "peer {peer}");
    }
}

/// Seeds by name (resolved by the net) and by address; the third node finds the second
/// only through what the first lists in its HELLO_REPLY.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn nodes_are_their_node_ids() {
    let sim = Sim::with_identity(51);
    let a = sim.start("a", &[], Start::Genesis);
    let b = sim.start("b", &["a"], Start::Join);
    let c = sim.start("c", &["sim:a"], Start::Join);
    sim.wait_ready(&[&a, &b, &c], SOON).await;
    assert_eq!(a.local_id(), sim_key("a").id());

    write(&b, "x", json!(1)).await;
    sim.wait_converged(&[&a, &b, &c], SOON).await;
    assert_eq!(read(&c, "x").await, Some(json!(1)));
    for node in [&a, &b, &c] {
        peers_are_node_ids(node).await;
    }
    let (b_id, c_id) = (b.local_id().to_owned(), c.local_id().to_owned());
    let a = &a;
    sim.wait_until(
        "b and c are registered with their addresses",
        SOON,
        move || {
            let (b_id, c_id) = (b_id.clone(), c_id.clone());
            async move {
                let replica = a.doc.read().await;
                members::addresses(&replica.doc, &b_id) == ["sim:b"]
                    && members::addresses(&replica.doc, &c_id) == ["sim:c"]
            }
        },
    )
    .await;
}

/// With two neighbours each, most pairs of six nodes never meet through seeds: they
/// find each other's addresses in the member list.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn members_find_each_other_through_the_member_list() {
    let sim = Sim::with_identity(52);
    let two = |config: &mut crate::node::NodeConfig| config.neighbours_k = 2;
    let a = sim.start_with("a", &[], Start::Genesis, two);
    let others: Vec<AppState> = ["b", "c", "d", "e", "f"]
        .into_iter()
        .map(|id| sim.start_with(id, &["a"], Start::Join, two))
        .collect();
    let mut all: Vec<&AppState> = vec![&a];
    all.extend(others.iter());
    sim.wait_ready(&all, SOON).await;

    write(&others[4], "from/f", json!("f")).await;
    write(&others[0], "from/b", json!("b")).await;
    sim.wait_converged(&all, SOON).await;
    for node in &all {
        assert_eq!(read(node, "from/f").await, Some(json!("f")));
        assert_eq!(read(node, "from/b").await, Some(json!("b")));
        peers_are_node_ids(node).await;
    }
}

fn ping(node: &AppState) -> ParsedMessage {
    ParsedMessage::Ping(Keepalive {
        heads: HeadsSummary::of(Vec::new()),
        epoch: node.local_epoch(),
        epoch_seq: 0,
        ready: false,
        syncs_with_you: false,
    })
}

fn who(key: &NodeKey, from: &str, to: &str) -> ParsedMessage {
    ParsedMessage::Who(IdentityProof {
        key: key.public(),
        signature: key.sign(&signed(WHO_LABEL, from, to)),
    })
}

/// A proof counts only for the address it was made at: messages from an address with
/// a proof made elsewhere are dropped, and the node never hears of them.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_proof_holds_only_at_its_address() {
    let sim = Sim::with_identity(53);
    let a = sim.start("a", &[], Start::Genesis);
    sim.wait_ready(&[&a], SOON).await;
    let ghost = NodeKey::from_seed([9; 32]);

    // Made for another sender address, or for another recipient.
    for forged in [
        who(&ghost, "sim:elsewhere", "sim:a"),
        who(&ghost, "sim:ghost", "sim:b"),
    ] {
        sim.net.inject("ghost", "a", &forged).await.unwrap();
        sim.net.inject("ghost", "a", &ping(&a)).await.unwrap();
        assert!(a.get_peer_state(&ghost.id()).await.is_none());
    }

    sim.net
        .inject("ghost", "a", &who(&ghost, "sim:ghost", "sim:a"))
        .await
        .unwrap();
    sim.net.inject("ghost", "a", &ping(&a)).await.unwrap();
    assert!(a.get_peer_state(&ghost.id()).await.is_some());
    // Known by its node ID, not by its address.
    assert!(a.get_peer_state("ghost").await.is_none());
}

/// A swarm across two transports: a only on the first, c only on the second, b on both.
/// a and c never reach each other, and still share everything through b.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn nodes_on_both_transports_carry_the_others() {
    let sim = Sim::with_identity(54);
    let second = SimNet::new(55);
    let a = sim.start("a", &[], Start::Genesis);
    let b = sim.start_with("b", &["a"], Start::Join, |config| {
        let identity = config.identity.as_mut().unwrap();
        let transport: Arc<dyn Transport> = second.transport("b");
        identity.more.push(("other", transport));
    });
    let c = sim.start_with("c", &["other:b"], Start::Join, |config| {
        config.transport = second.transport("c");
        config.identity.as_mut().unwrap().scheme = "other";
    });
    sim.wait_ready(&[&a, &b, &c], SOON).await;
    assert_eq!(b.transport.local_addresses(), ["sim:b", "other:b"]);

    write(&c, "from/c", json!("c")).await;
    write(&a, "from/a", json!("a")).await;
    sim.wait_converged(&[&a, &b, &c], SOON).await;
    assert_eq!(read(&a, "from/c").await, Some(json!("c")));
    assert_eq!(read(&c, "from/a").await, Some(json!("a")));

    // The member list says where c is, on a transport a doesn't have.
    let c_id = c.local_id().to_owned();
    let a_ref = &a;
    sim.wait_until("a knows c's entry", SOON, move || {
        let c_id = c_id.clone();
        async move {
            let replica = a_ref.doc.read().await;
            members::addresses(&replica.doc, &c_id) == ["other:c"]
        }
    })
    .await;
    assert!(!a.transport.can_reach(c.local_id()));
    assert!(!c.transport.can_reach(a.local_id()));
    assert!(a.transport.can_reach(b.local_id()));
    assert!(c.transport.can_reach(b.local_id()));
}

/// Whole nodes over libp2p's in-memory transport, under the identity layer: seeds are
/// multiaddrs, and members find each other's listen addresses in the member list.
#[cfg(feature = "libp2p")]
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn nodes_over_libp2p() {
    use crate::transport::p2p::{Libp2pConfig, Libp2pTransport};
    use ::libp2p::identity::Keypair;
    use sha2::Digest;

    fn key(id: &str) -> Keypair {
        let mut seed: [u8; 32] = sha2::Sha256::digest(format!("libp2p {id}").as_bytes()).into();
        Keypair::ed25519_from_bytes(&mut seed).unwrap()
    }
    // Memory ports are process-wide; these are this test's.
    let address = |id: &str| -> ::libp2p::Multiaddr {
        let port = 50_000 + u64::from(id.as_bytes()[0]);
        format!("/memory/{port}").parse().unwrap()
    };
    let on_libp2p = |id: &'static str| {
        move |config: &mut crate::node::NodeConfig| {
            let transport = Libp2pTransport::start(Libp2pConfig {
                key: key(id),
                listen: vec![address(id)],
                announce: Vec::new(),
                relays: Vec::new(),
                relay: false,
                memory: true,
            })
            .unwrap();
            config.transport = Arc::new(transport);
            config.identity.as_mut().unwrap().scheme = "libp2p";
        }
    };
    let a_peer = key("a").public().to_peer_id();
    // With the peer ID, with the scheme but without the peer ID, and bare.
    let seeds = [
        format!("libp2p:{}", address("a").with_p2p(a_peer).unwrap()),
        format!("libp2p:{}", address("a")),
        address("a").to_string(),
    ];

    let sim = Sim::with_identity(56);
    let a = sim.start_with("a", &[], Start::Genesis, on_libp2p("a"));
    let others: Vec<AppState> = ["b", "c", "d"]
        .into_iter()
        .zip(&seeds)
        .map(|(id, seed)| sim.start_with(id, &[seed.as_str()], Start::Join, on_libp2p(id)))
        .collect();
    let mut all: Vec<&AppState> = vec![&a];
    all.extend(others.iter());
    sim.wait_ready(&all, SOON).await;

    write(&others[2], "from/d", json!("d")).await;
    sim.wait_converged(&all, SOON).await;
    for node in &all {
        assert_eq!(read(node, "from/d").await, Some(json!("d")));
        peers_are_node_ids(node).await;
    }
    let d_id = others[2].local_id().to_owned();
    let expected = format!(
        "libp2p:{}",
        address("d")
            .with_p2p(key("d").public().to_peer_id())
            .unwrap()
    );
    let a = &a;
    sim.wait_until(
        "d's listen address is in the member list",
        SOON,
        move || {
            let (d_id, expected) = (d_id.clone(), expected.clone());
            async move {
                let replica = a.doc.read().await;
                members::addresses(&replica.doc, &d_id) == [expected]
            }
        },
    )
    .await;
}