//! Behavioral test for the peat-controlled `_peat._udp` browse (peat-mesh#265).
//!
//! Before this change, [`IrohTransport`] advertised on `_peat._udp` but had no
//! peat-controlled browse — discovery→connection relied on iroh's external
//! `MdnsAddressLookup` browse, which never fires on Android from a wildcard
//! bind. This pins the additive browse side: two transports built via
//! `from_formation_with_discovery_at_addr` advertise on **loopback** and each
//! transport's `peat_mdns_events()` stream must yield a `PeerFound` for the
//! *other* node — carrying the remote's `node_id` and a concrete `SocketAddr` —
//! while never emitting an event for its own advertisement (self-filter).
//!
//! `#[ignore]` for the same multicast reasons as the sibling mDNS tests
//! (`mdns_advertise_with_addr`, `mdns_iroh_address_lookup`): most containerized
//! CI blocks multicast joins, so this needs a multicast-capable runner or bare
//! metal. Locally:
//!
//! ```sh
//! cargo test --features automerge-backend --test peat_mdns_browse -- --ignored
//! ```

#![cfg(feature = "automerge-backend")]

use std::net::SocketAddr;
use std::time::{Duration, Instant};

use peat_mesh::discovery::DiscoveryEvent;
use peat_mesh::network::IrohTransport;

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
#[ignore = "requires multicast in CI; run with --ignored on bare metal"]
async fn browse_stream_yields_remote_peer_and_filters_self() {
    // Run-scoped suffix so two CI jobs sharing the multicast group (a
    // merge-push run racing a tag/release run on the same self-hosted runner)
    // don't collide on service-instance names. Each run only discovers its own
    // pair; any other `_peat._udp` advertiser on the segment is ignored.
    let uniq = format!(
        "{}-{}",
        std::process::id(),
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos()
    );
    let id_a = format!("node-A-{uniq}");
    let id_b = format!("node-B-{uniq}");

    // Same shared formation secret for both — discovery is formation-agnostic
    // at the mDNS layer, but using one keeps the test faithful to real use.
    let formation_secret = b"peat-mesh-265-shared-formation-secret";
    let app_id = format!("formation-{uniq}");

    let loopback: SocketAddr = "127.0.0.1:0".parse().unwrap();

    let transport_a = IrohTransport::from_formation_with_discovery_at_addr(
        formation_secret,
        &id_a,
        &app_id,
        loopback,
        false,
    )
    .await
    .expect("build transport A");
    let transport_b = IrohTransport::from_formation_with_discovery_at_addr(
        formation_secret,
        &id_b,
        &app_id,
        loopback,
        false,
    )
    .await
    .expect("build transport B");

    // Each node advertises its DIALABLE hex EndpointId as the TXT node_id (not
    // the formation node_id), so a browsing peer can dial it back. The browse
    // stream therefore carries the remote's hex endpoint id — match on that.
    let want_a = hex::encode(transport_a.endpoint_id().as_bytes());
    let want_b = hex::encode(transport_b.endpoint_id().as_bytes());

    let mut events_a = transport_a
        .peat_mdns_events()
        .expect("transport A exposes a peat_mdns browse stream");
    let mut events_b = transport_b
        .peat_mdns_events()
        .expect("transport B exposes a peat_mdns browse stream");

    // A must observe B (and never A), B must observe A (and never B). mDNS
    // announce + browse round-trip is typically 1-3s; allow 20s of headroom
    // for a cold-start announce on a contended runner.
    let start = Instant::now();
    let deadline = Duration::from_secs(20);

    let a_sees_b = await_remote(&mut events_a, &want_b, &want_a, deadline);
    let b_sees_a = await_remote(&mut events_b, &want_a, &want_b, deadline);
    let (a_sees_b, b_sees_a) = tokio::join!(a_sees_b, b_sees_a);

    let addr_b = a_sees_b.unwrap_or_else(|| {
        panic!(
            "transport A's peat_mdns browse did not yield a PeerFound for B \
             (endpoint {want_b}, formation {id_b}) within {:?}",
            start.elapsed()
        )
    });
    let addr_a = b_sees_a.unwrap_or_else(|| {
        panic!(
            "transport B's peat_mdns browse did not yield a PeerFound for A \
             (endpoint {want_a}, formation {id_a}) within {:?}",
            start.elapsed()
        )
    });

    assert!(
        addr_b.ip().is_loopback() && !addr_b.ip().is_unspecified(),
        "A resolved B at a non-concrete/non-loopback address: {addr_b}"
    );
    assert!(
        addr_a.ip().is_loopback() && !addr_a.ip().is_unspecified(),
        "B resolved A at a non-concrete/non-loopback address: {addr_a}"
    );

    let _ = transport_a.close().await;
    let _ = transport_b.close().await;
}

/// Drain `events` until a `PeerFound`/`PeerUpdated` for `want` arrives,
/// returning its first resolved `SocketAddr`. Fails the test immediately if an
/// event for `forbidden` (this node's own advertisement) is observed — the
/// self-filter must drop those before they reach the stream. Returns `None` on
/// timeout.
async fn await_remote(
    events: &mut tokio::sync::mpsc::Receiver<DiscoveryEvent>,
    want: &str,
    forbidden: &str,
    timeout: Duration,
) -> Option<SocketAddr> {
    let deadline = Instant::now() + timeout;
    loop {
        let remaining = deadline.checked_duration_since(Instant::now())?;
        let event = tokio::time::timeout(remaining, events.recv())
            .await
            .ok()??;
        match event {
            DiscoveryEvent::PeerFound(peer) | DiscoveryEvent::PeerUpdated(peer) => {
                assert_ne!(
                    peer.node_id, forbidden,
                    "self-filter leaked: browse stream emitted our own advertisement ({forbidden})"
                );
                if peer.node_id == want {
                    return peer.addresses.first().copied();
                }
            }
            DiscoveryEvent::PeerLost(id) => {
                assert_ne!(
                    id, forbidden,
                    "self-filter leaked: browse stream emitted a PeerLost for self ({forbidden})"
                );
            }
        }
    }
}
