use std::collections::{HashMap, HashSet};
use std::net::SocketAddr;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::{Duration, Instant};

use nostr::nips::nip17;
use nostr::nips::nip19::ToBech32;
use nostr::prelude::{
    Alphabet, Event, EventBuilder, EventId, Filter, Kind, PublicKey, RelayUrl, SingleLetterTag,
    Tag, TagKind, Timestamp,
};
use nostr_sdk::{Client, ClientOptions, prelude::RelayPoolNotification};
use serde::Serialize;
use tokio::sync::{Mutex, Notify, RwLock, Semaphore, broadcast, mpsc, oneshot};
use tokio::task::JoinHandle;
use tracing::{debug, info, trace, warn};

use super::failure_state::FailureState;
use super::signal::{
    FreshnessOutcome, SignalEnvelope, build_signal_event, create_traversal_answer,
    create_traversal_offer, estimate_clock_skew, unwrap_signal_event, validate_offer_freshness,
    validate_traversal_answer_for_offer,
};
use super::stun::observe_traversal_addresses;
use super::traversal::{nonce, now_ms, planned_remote_endpoints, run_punch_attempt};
use super::types::{
    ADVERT_IDENTIFIER, ADVERT_KIND, ADVERT_VERSION, BootstrapError, BootstrapEvent,
    CachedOverlayAdvert, MeshTraversalSignal, NostrFailureDecision, NostrPeerFailureView,
    NostrRefetchOutcome, NostrRelayStatus, OverlayAdvert, OverlayEndpointAdvert, PROTOCOL_VERSION,
    PunchHint, SIGNAL_KIND, TraversalAnswer, TraversalOffer, advert_d_tag,
};
use crate::config::{NostrDiscoveryConfig, PeerConfig};
use crate::discovery::EstablishedTraversal;

const ADVERT_CACHE_STALE_GRACE_MULTIPLIER: u64 = 2;

fn bind_traversal_udp_socket() -> std::io::Result<std::net::UdpSocket> {
    #[cfg(any(target_os = "linux", target_os = "macos"))]
    {
        use socket2::{Domain, Protocol, Socket, Type};

        let socket = Socket::new(Domain::IPV4, Type::DGRAM, Some(Protocol::UDP))?;
        let _ = socket.set_reuse_address(true);
        let _ = socket.set_reuse_port(true);
        socket.bind(&SocketAddr::from(([0, 0, 0, 0], 0)).into())?;
        let socket: std::net::UdpSocket = socket.into();
        socket.set_nonblocking(true)?;
        Ok(socket)
    }
    #[cfg(not(any(target_os = "linux", target_os = "macos")))]
    {
        let socket = std::net::UdpSocket::bind(("0.0.0.0", 0))?;
        socket.set_nonblocking(true)?;
        Ok(socket)
    }
}

fn short_npub(npub: &str) -> String {
    npub.strip_prefix("npub1")
        .filter(|s| s.len() >= 8)
        .map(|s| format!("npub1{}..{}", &s[..4], &s[s.len() - 4..]))
        .unwrap_or_else(|| npub.to_string())
}

fn short_id(id: &str) -> String {
    if id.len() > 8 {
        id[..8].to_string()
    } else {
        id.to_string()
    }
}

fn endpoint_summary(endpoints: &[OverlayEndpointAdvert]) -> String {
    endpoints
        .iter()
        .map(|e| format!("{:?}:{}", e.transport, e.addr).to_lowercase())
        .collect::<Vec<_>>()
        .join(",")
}

fn is_unroutable_direct_advert_ip(ip: std::net::IpAddr) -> bool {
    match ip {
        std::net::IpAddr::V4(v4) => {
            v4.is_private()
                || v4.is_loopback()
                || v4.is_link_local()
                || v4.is_unspecified()
                || v4.is_multicast()
                || v4.is_broadcast()
                || v4.is_documentation()
                || (v4.octets()[0] == 100 && (v4.octets()[1] & 0xc0) == 64)
        }
        std::net::IpAddr::V6(v6) => {
            v6.is_loopback()
                || v6.is_unspecified()
                || v6.is_unique_local()
                || v6.is_multicast()
                || (v6.segments()[0] & 0xffc0) == 0xfe80
        }
    }
}

fn endpoint_advert_is_publicly_usable(endpoint: &OverlayEndpointAdvert) -> bool {
    let addr = endpoint.addr.trim();
    if addr.is_empty() {
        return false;
    }

    if endpoint.transport == super::types::OverlayTransportKind::Udp
        && addr.eq_ignore_ascii_case("nat")
    {
        return true;
    }
    if addr.eq_ignore_ascii_case("nat") {
        return false;
    }

    match endpoint.transport {
        super::types::OverlayTransportKind::Udp | super::types::OverlayTransportKind::Tcp => {
            let Ok(socket_addr) = addr.parse::<SocketAddr>() else {
                let Some((host, port)) = addr.rsplit_once(':') else {
                    return false;
                };
                let host = host.trim().trim_start_matches('[').trim_end_matches(']');
                if host.is_empty() || port.trim().parse::<u16>().ok().is_none_or(|p| p == 0) {
                    return false;
                }
                if host.eq_ignore_ascii_case("localhost") {
                    return false;
                }
                return host
                    .parse::<std::net::IpAddr>()
                    .ok()
                    .is_none_or(|ip| !is_unroutable_direct_advert_ip(ip));
            };
            socket_addr.port() != 0 && !is_unroutable_direct_advert_ip(socket_addr.ip())
        }
        super::types::OverlayTransportKind::Tor => true,
        super::types::OverlayTransportKind::WebRtc => is_compressed_pubkey_hex(addr),
    }
}

fn is_compressed_pubkey_hex(addr: &str) -> bool {
    addr.len() == 66
        && (addr.starts_with("02") || addr.starts_with("03"))
        && addr.as_bytes().iter().all(u8::is_ascii_hexdigit)
}

/// Cached STUN-derived public address for an advert-eligible UDP transport
/// bound to a wildcard. Lives on `NostrDiscovery` so the freshness window
/// survives advert refresh cycles.
struct CachedPublicUdpAddr {
    /// Most recent STUN observation. `None` means the last attempt failed
    /// (recorded so we don't re-spam STUN every refresh tick on broken
    /// network conditions).
    addr: Option<SocketAddr>,
    fetched_at: Instant,
}

/// Cache lifetime for a *failed* STUN observation. Held briefly so that
/// transient flakes (slow startup network, momentary STUN-server
/// blip) get retried within ~a minute and the advert grows its UDP
/// endpoint as soon as STUN starts working — rather than waiting a
/// full `advert_refresh_secs` (30 min) for the success-path TTL to
/// expire. Successful results use the longer per-config TTL.
const PUBLIC_UDP_ADDR_FAILURE_TTL: Duration = Duration::from_secs(60);
const RELAY_STARTUP_OP_TIMEOUT: Duration = Duration::from_secs(5);
const ADVERT_PUBLISH_TIMEOUT: Duration = Duration::from_secs(10);

#[derive(Debug, Clone, PartialEq, Eq)]
struct NostrRelayConfig {
    advert_relays: Vec<String>,
    dm_relays: Vec<String>,
}

impl From<&NostrDiscoveryConfig> for NostrRelayConfig {
    fn from(config: &NostrDiscoveryConfig) -> Self {
        Self {
            advert_relays: config.advert_relays.clone(),
            dm_relays: config.dm_relays.clone(),
        }
    }
}

impl NostrRelayConfig {
    fn union(&self) -> HashSet<String> {
        self.advert_relays
            .iter()
            .chain(self.dm_relays.iter())
            .cloned()
            .collect()
    }
}

pub struct NostrDiscovery {
    client: Client,
    keys: nostr::Keys,
    pubkey: PublicKey,
    npub: String,
    config: NostrDiscoveryConfig,
    relay_config: RwLock<NostrRelayConfig>,
    advert_cache: RwLock<HashMap<String, CachedOverlayAdvert>>,
    local_advert: RwLock<Option<OverlayAdvert>>,
    current_advert_event_id: RwLock<Option<EventId>>,
    pending_answers: Mutex<HashMap<String, oneshot::Sender<SignalEnvelope<TraversalAnswer>>>>,
    active_initiators: Mutex<HashSet<String>>,
    active_refetches: Mutex<HashSet<String>>,
    seen_sessions: Mutex<HashMap<String, u64>>,
    offer_slots: Arc<Semaphore>,
    event_tx: mpsc::Sender<BootstrapEvent>,
    event_rx: Mutex<mpsc::Receiver<BootstrapEvent>>,
    mesh_signal_tx: mpsc::Sender<MeshTraversalSignal>,
    mesh_signal_rx: Mutex<mpsc::Receiver<MeshTraversalSignal>>,
    connect_task: Mutex<Option<JoinHandle<()>>>,
    relay_startup_task: Mutex<Option<JoinHandle<()>>>,
    publish_task: Mutex<Option<JoinHandle<()>>>,
    publish_notify: Notify,
    notify_task: Mutex<Option<JoinHandle<()>>>,
    advertise_task: Mutex<Option<JoinHandle<()>>>,
    failure_state: FailureState,
    /// STUN-derived public address per advert-eligible UDP transport
    /// (keyed by `TransportId.as_u32()`). Populated on demand by
    /// `learn_public_udp_addr()` and refreshed by TTL.
    public_udp_addr_cache: RwLock<HashMap<u32, CachedPublicUdpAddr>>,
    /// Capacity gate refreshed by `Node` once per tick. The inbound Msg1 gate
    /// remains authoritative; this only suppresses wasted traversal work.
    outbound_admission: AtomicBool,
    /// Capacity gate for racing a better path to an already-authenticated
    /// peer. This bypasses the peer-slot cap while still honoring
    /// connection/link caps.
    direct_refresh_admission: AtomicBool,
}

impl NostrDiscovery {
    pub async fn start(
        identity: &crate::Identity,
        config: NostrDiscoveryConfig,
    ) -> Result<Arc<Self>, BootstrapError> {
        if !config.enabled {
            return Err(BootstrapError::Disabled);
        }

        let keys = nostr::Keys::parse(&hex::encode(identity.keypair().secret_bytes()))
            .map_err(|e| BootstrapError::Nostr(e.to_string()))?;
        let client = Client::builder()
            .signer(keys.clone())
            .opts(ClientOptions::new().autoconnect(false))
            .build();

        let mut relay_union = HashSet::new();
        relay_union.extend(config.advert_relays.iter().cloned());
        relay_union.extend(config.dm_relays.iter().cloned());
        for relay in relay_union {
            client
                .add_relay(&relay)
                .await
                .map_err(|e| BootstrapError::Nostr(e.to_string()))?;
        }
        let pubkey = keys.public_key();
        let npub = crate::encode_npub(&identity.pubkey());
        let (event_tx, event_rx) = mpsc::channel(event_channel_capacity(&config));
        let (mesh_signal_tx, mesh_signal_rx) = mpsc::channel(event_channel_capacity(&config));
        let offer_slots = Arc::new(Semaphore::new(config.max_concurrent_incoming_offers));

        let failure_state = FailureState::new(
            config.failure_streak_threshold,
            config.extended_cooldown_secs,
            config.warn_log_interval_secs,
            config.failure_state_max_entries,
        );

        let runtime = Arc::new(Self {
            client,
            keys,
            pubkey,
            npub,
            relay_config: RwLock::new(NostrRelayConfig::from(&config)),
            config,
            advert_cache: RwLock::new(HashMap::new()),
            local_advert: RwLock::new(None),
            current_advert_event_id: RwLock::new(None),
            pending_answers: Mutex::new(HashMap::new()),
            active_initiators: Mutex::new(HashSet::new()),
            active_refetches: Mutex::new(HashSet::new()),
            seen_sessions: Mutex::new(HashMap::new()),
            offer_slots,
            event_tx,
            event_rx: Mutex::new(event_rx),
            mesh_signal_tx,
            mesh_signal_rx: Mutex::new(mesh_signal_rx),
            connect_task: Mutex::new(None),
            relay_startup_task: Mutex::new(None),
            publish_task: Mutex::new(None),
            publish_notify: Notify::new(),
            notify_task: Mutex::new(None),
            advertise_task: Mutex::new(None),
            failure_state,
            public_udp_addr_cache: RwLock::new(HashMap::new()),
            outbound_admission: AtomicBool::new(true),
            direct_refresh_admission: AtomicBool::new(true),
        });

        // Subscribe to the relay-pool broadcast channel BEFORE issuing the
        // Nostr REQs. tokio's broadcast channel only delivers messages sent
        // after the receiver is created — historical events that arrive in
        // response to subscribe() (REQ replays) would otherwise be dropped
        // by the pool's `external_notification_sender.send(...)` returning
        // `Err(SendError)` when no subscriber exists yet. Without this,
        // freshly-restarted nodes with `policy: open` waited up to one
        // `advert_refresh_secs` interval (default 30 min) for non-configured
        // peers to re-publish before discovering them.
        let notifications = runtime.client.notifications();
        *runtime.publish_task.lock().await = Some(runtime.clone().spawn_publish_loop());
        *runtime.connect_task.lock().await = Some(runtime.clone().spawn_connect_loop());
        *runtime.relay_startup_task.lock().await = Some(runtime.clone().spawn_relay_startup_loop());
        *runtime.advertise_task.lock().await = Some(runtime.clone().spawn_advertise_loop());
        *runtime.notify_task.lock().await = Some(runtime.clone().spawn_notify_loop(notifications));

        Ok(runtime)
    }

    pub fn set_outbound_admission(&self, allow: bool) {
        self.outbound_admission.store(allow, Ordering::Relaxed);
    }

    pub(crate) fn outbound_admission_allowed(&self) -> bool {
        self.outbound_admission.load(Ordering::Relaxed)
    }

    pub fn set_direct_refresh_admission(&self, allow: bool) {
        self.direct_refresh_admission
            .store(allow, Ordering::Relaxed);
    }

    pub(crate) fn direct_refresh_admission_allowed(&self) -> bool {
        self.direct_refresh_admission.load(Ordering::Relaxed)
    }

    fn traversal_initiator_admission_allowed(&self, mesh_signaling_allowed: bool) -> bool {
        if mesh_signaling_allowed {
            self.direct_refresh_admission_allowed()
        } else {
            self.outbound_admission_allowed()
        }
    }

    pub async fn relay_statuses(&self) -> Vec<NostrRelayStatus> {
        let relay_config = self.relay_config.read().await.clone();
        let mut statuses = relay_config
            .advert_relays
            .iter()
            .chain(relay_config.dm_relays.iter())
            .map(|url| {
                (
                    url.clone(),
                    NostrRelayStatus {
                        url: url.clone(),
                        status: "unknown".to_string(),
                    },
                )
            })
            .collect::<HashMap<_, _>>();

        for (relay_url, relay) in self.client.relays().await {
            let url = relay_url.to_string();
            statuses.insert(
                url.clone(),
                NostrRelayStatus {
                    url,
                    status: relay.status().to_string().to_ascii_lowercase(),
                },
            );
        }

        let mut statuses = statuses.into_values().collect::<Vec<_>>();
        statuses.sort_by(|lhs, rhs| lhs.url.cmp(&rhs.url));
        statuses
    }

    pub async fn update_relays(
        &self,
        advert_relays: Vec<String>,
        dm_relays: Vec<String>,
    ) -> Result<(), BootstrapError> {
        let next = NostrRelayConfig {
            advert_relays,
            dm_relays,
        };

        let previous = self.relay_config.read().await.clone();
        if previous == next {
            return Ok(());
        }

        let previous_union = previous.union();
        let next_union = next.union();

        for relay in next_union.difference(&previous_union) {
            self.client
                .add_relay(relay)
                .await
                .map_err(|e| BootstrapError::Nostr(e.to_string()))?;
        }

        for relay in previous_union.difference(&next_union) {
            self.client
                .force_remove_relay(relay)
                .await
                .map_err(|e| BootstrapError::Nostr(e.to_string()))?;
        }

        {
            let mut relay_config = self.relay_config.write().await;
            *relay_config = next;
        }

        for relay in next_union {
            if let Err(error) = self.client.connect_relay(relay.clone()).await {
                warn!(relay = %relay, error = %error, "failed to connect updated Nostr relay");
            }
        }

        if let Err(error) = self.subscribe().await {
            warn!(error = %error, "failed to subscribe updated Nostr relays");
        }
        if let Err(error) = self.publish_inbox_relays().await {
            warn!(error = %error, "failed to publish updated Nostr inbox relay list");
        }
        self.request_publish_advert();
        Ok(())
    }

    pub async fn request_connect(self: &Arc<Self>, peer_config: PeerConfig) {
        let _ = self
            .request_connect_with_mesh_signaling(peer_config, false)
            .await;
    }

    pub(crate) async fn request_connect_with_mesh_signaling(
        self: &Arc<Self>,
        peer_config: PeerConfig,
        mesh_signaling_allowed: bool,
    ) -> bool {
        let peer_npub = peer_config.npub.clone();
        {
            let mut active = self.active_initiators.lock().await;
            if !active.insert(peer_npub.clone()) {
                return false;
            }
        }

        let runtime = Arc::clone(self);
        tokio::spawn(async move {
            let event = match runtime
                .connect_peer(peer_config.clone(), mesh_signaling_allowed)
                .await
            {
                Ok(traversal) => BootstrapEvent::Established { traversal },
                Err(err) => BootstrapEvent::Failed {
                    peer_config,
                    reason: err.to_string(),
                },
            };
            runtime.emit_event(event).await;
            runtime.active_initiators.lock().await.remove(&peer_npub);
        });
        true
    }

    /// Record a NAT-traversal failure for `npub`, returning the
    /// resulting decision (WARN suppression + extended cooldown +
    /// threshold-crossing flag for the B6 re-fetch).
    pub fn record_traversal_failure(&self, npub: &str, now_ms: u64) -> NostrFailureDecision {
        let d = self.failure_state.record_failure(npub, now_ms);
        NostrFailureDecision {
            consecutive_failures: d.consecutive_failures,
            should_warn: d.should_warn,
            cooldown_until_ms: d.cooldown_until_ms,
            crossed_threshold: d.crossed_threshold,
        }
    }

    /// Mark a traversal-backed path as unstable after it completed but later
    /// died under link liveness. This records diagnostics and threshold
    /// crossing without creating a new peer-wide cooldown; direct probing is
    /// paced by the node retry loop.
    pub fn record_unstable_path(&self, npub: &str, now_ms: u64) -> NostrFailureDecision {
        let d = self.failure_state.record_unstable_path(npub, now_ms);
        NostrFailureDecision {
            consecutive_failures: d.consecutive_failures,
            should_warn: d.should_warn,
            cooldown_until_ms: d.cooldown_until_ms,
            crossed_threshold: d.crossed_threshold,
        }
    }

    /// Record a successful traversal — clears the streak/cooldown.
    pub fn record_traversal_success(&self, npub: &str, now_ms: u64) {
        self.failure_state.record_success(npub, now_ms);
    }

    /// Cooldown wall-clock ms if the peer is currently suppressed,
    /// else None. Used by the open-discovery sweep to skip enqueue.
    pub fn cooldown_until(&self, npub: &str, now_ms: u64) -> Option<u64> {
        self.failure_state.cooldown_until(npub, now_ms)
    }

    /// Record a fatal protocol mismatch (e.g. `Unknown FMP version` on a
    /// Nostr-adopted bootstrap transport). Returns `true` if this is a
    /// fresh observation worth a WARN log; `false` if the peer is already
    /// inside a comparable mismatch cooldown.
    ///
    /// The cooldown is `protocol_mismatch_cooldown_secs` from config —
    /// much longer than `extended_cooldown_secs` because mismatches are
    /// structural (only resolves when one side upgrades) rather than
    /// transient.
    pub fn record_protocol_mismatch(&self, npub: &str, now_ms: u64) -> bool {
        let cooldown_ms = self
            .config
            .protocol_mismatch_cooldown_secs
            .saturating_mul(1000);
        self.failure_state
            .record_protocol_mismatch(npub, now_ms, cooldown_ms)
    }

    /// Configured protocol-mismatch cooldown in seconds. Exposed so log
    /// emitters can include the duration without re-reading config.
    pub fn protocol_mismatch_cooldown_secs(&self) -> u64 {
        self.config.protocol_mismatch_cooldown_secs
    }

    /// Snapshot of per-npub failure state for `show_peers` rendering.
    pub fn failure_state_snapshot(&self) -> Vec<NostrPeerFailureView> {
        self.failure_state
            .snapshot()
            .into_iter()
            .map(|(npub, rec)| NostrPeerFailureView {
                npub,
                consecutive_failures: rec.consecutive_failures,
                cooldown_until_ms: rec.cooldown_until_ms,
                last_observed_skew_ms: rec.last_observed_skew_ms,
            })
            .collect()
    }

    /// Discover (or return cached) the public-Internet address for an
    /// advert-eligible UDP transport bound to a wildcard. Used by
    /// `build_overlay_advert` to avoid emitting `udp:0.0.0.0:port`,
    /// which is invalid as an advertised endpoint. Result is the
    /// reflexive IP (from STUN against the daemon's first
    /// `stun_servers` reachable) combined with the configured
    /// `advertise_port`.
    ///
    /// Asymmetric cache TTL: a successful observation is cached for
    /// `advert_refresh_secs` (default 1800 = same as advert refresh)
    /// so we don't re-STUN every refresh tick. A failed observation
    /// is cached for `PUBLIC_UDP_ADDR_FAILURE_TTL` (60s) so we retry
    /// soon after a transient STUN flake at startup, instead of
    /// blocking advertise-as-public for half an hour. Once a success
    /// is cached, subsequent ticks are zero-overhead.
    pub async fn learn_public_udp_addr(
        &self,
        transport_id_key: u32,
        advertise_port: u16,
    ) -> Option<SocketAddr> {
        if let Some(entry) = self
            .public_udp_addr_cache
            .read()
            .await
            .get(&transport_id_key)
        {
            let ttl = if entry.addr.is_some() {
                Duration::from_secs(self.config.advert_refresh_secs.max(60))
            } else {
                PUBLIC_UDP_ADDR_FAILURE_TTL
            };
            if entry.fetched_at.elapsed() < ttl {
                return entry.addr;
            }
        }
        let resolved = self.stun_observe_public_ip(advertise_port).await;
        let mut cache = self.public_udp_addr_cache.write().await;
        cache.insert(
            transport_id_key,
            CachedPublicUdpAddr {
                addr: resolved,
                fetched_at: Instant::now(),
            },
        );
        resolved
    }

    /// Run a one-shot STUN observation against an ephemeral UDP socket
    /// to learn this host's public IPv4 (or IPv6, if the local STUN
    /// server returns one). Returns `<reflexive_ip>:<advertise_port>`,
    /// or `None` if STUN failed or no `stun_servers` are configured.
    ///
    /// The STUN-reported port is the ephemeral source port and is
    /// discarded — what we want to advertise is the bound listener
    /// port, which the kernel preserves through 1:1 NAT (AWS EIP,
    /// GCP/Azure external IPs) and which the operator has explicitly
    /// chosen via `bind_addr`.
    async fn stun_observe_public_ip(&self, advertise_port: u16) -> Option<SocketAddr> {
        if self.config.stun_servers.is_empty() {
            return None;
        }
        let socket = match bind_traversal_udp_socket() {
            Ok(s) => s,
            Err(err) => {
                debug!(error = %err, "public-udp-addr: ephemeral bind failed");
                return None;
            }
        };
        let observed = match super::stun::observe_traversal_addresses(
            &socket,
            &self.config.stun_servers,
            false,
            super::stun::ADVERT_STUN_TIMEOUT,
        )
        .await
        {
            Ok((reflexive, _local, stun_server)) => {
                debug!(
                    stun = %stun_server.as_deref().unwrap_or("-"),
                    reflexive = %reflexive
                        .as_ref()
                        .map(|a| format!("{}:{}", a.ip, a.port))
                        .unwrap_or_else(|| "-".into()),
                    "public-udp-addr: STUN observation"
                );
                reflexive
            }
            Err(err) => {
                debug!(error = %err, "public-udp-addr: STUN failed");
                return None;
            }
        };
        observed.and_then(|addr| {
            let parsed_ip: std::net::IpAddr = addr.ip.parse().ok()?;
            Some(SocketAddr::new(parsed_ip, advertise_port))
        })
    }

    /// Stale-advert re-check (B6). Called by lifecycle on the
    /// streak-threshold transition. Actively re-queries the peer's
    /// Kind 37195 advert from `advert_relays`; evicts the cache entry
    /// if absent, refreshes if newer than the cached `created_at`,
    /// otherwise leaves the cache untouched.
    pub async fn refetch_advert_for_stale_check(&self, peer_npub: &str) -> NostrRefetchOutcome {
        let target_pubkey = match PublicKey::parse(peer_npub) {
            Ok(p) => p,
            Err(_) => return NostrRefetchOutcome::Skipped,
        };
        let relay_config = self.relay_config.read().await.clone();
        if relay_config.advert_relays.is_empty() {
            return NostrRefetchOutcome::Skipped;
        }
        let cached_created_at = self
            .advert_cache
            .read()
            .await
            .get(peer_npub)
            .map(|c| c.created_at);

        let events = match self
            .client
            .fetch_events_from(
                relay_config.advert_relays.clone(),
                Filter::new()
                    .author(target_pubkey)
                    .kind(Kind::Custom(ADVERT_KIND))
                    .identifier(advert_d_tag(&self.config.app)),
                Duration::from_secs(2),
            )
            .await
        {
            Ok(e) => e,
            Err(_) => return NostrRefetchOutcome::Skipped,
        };

        let mut newest: Option<(u64, &Event)> = None;
        for ev in events.iter() {
            let ts = ev.created_at.as_secs();
            match newest {
                Some((cur, _)) if ts <= cur => {}
                _ => newest = Some((ts, ev)),
            }
        }

        let Some((relay_created_at, ev)) = newest else {
            // Absent on relays. Evict any stale cache entry.
            self.advert_cache.write().await.remove(peer_npub);
            self.failure_state.reset_streak_after_refresh(peer_npub);
            return NostrRefetchOutcome::Evicted;
        };

        match cached_created_at {
            Some(cached) if relay_created_at <= cached => NostrRefetchOutcome::SameAdvert,
            _ => {
                let Some(valid_until_ms) = self.event_valid_until_ms(ev) else {
                    return NostrRefetchOutcome::Skipped;
                };
                let Ok(verified_event) = VerifiedEvent::try_from(ev) else {
                    return NostrRefetchOutcome::Skipped;
                };
                let Ok(advert) = Self::parse_overlay_advert_event(verified_event, &self.config.app)
                else {
                    return NostrRefetchOutcome::Skipped;
                };
                let updated = CachedOverlayAdvert {
                    author_npub: peer_npub.to_string(),
                    advert,
                    created_at: relay_created_at,
                    valid_until_ms,
                };
                self.advert_cache
                    .write()
                    .await
                    .insert(peer_npub.to_string(), updated);
                self.failure_state.reset_streak_after_refresh(peer_npub);
                NostrRefetchOutcome::Refreshed
            }
        }
    }

    pub async fn request_advert_stale_check(self: &Arc<Self>, peer_npub: String) -> bool {
        let relay_config = self.relay_config.read().await.clone();
        if relay_config.advert_relays.is_empty() {
            return false;
        }
        {
            let mut active = self.active_refetches.lock().await;
            if !active.insert(peer_npub.clone()) {
                return false;
            }
        }

        let runtime = Arc::clone(self);
        tokio::spawn(async move {
            let outcome = runtime.refetch_advert_for_stale_check(&peer_npub).await;
            match outcome {
                NostrRefetchOutcome::Evicted => info!(
                    npub = %peer_npub,
                    "stale-advert sweep: peer evicted from advert cache"
                ),
                NostrRefetchOutcome::Refreshed => info!(
                    npub = %peer_npub,
                    "stale-advert sweep: peer republished, cache refreshed and streak reset"
                ),
                NostrRefetchOutcome::SameAdvert => debug!(
                    npub = %peer_npub,
                    "stale-advert sweep: relay still has same advert"
                ),
                NostrRefetchOutcome::Skipped => debug!(
                    npub = %peer_npub,
                    "stale-advert sweep: skipped"
                ),
            }
            runtime.active_refetches.lock().await.remove(&peer_npub);
        });
        true
    }

    pub async fn drain_events(&self) -> Vec<BootstrapEvent> {
        let mut out = Vec::new();
        let mut rx = self.event_rx.lock().await;
        while let Ok(event) = rx.try_recv() {
            out.push(event);
        }
        out
    }

    pub(crate) async fn drain_mesh_signals(&self) -> Vec<MeshTraversalSignal> {
        let mut out = Vec::new();
        let mut rx = self.mesh_signal_rx.lock().await;
        while let Ok(signal) = rx.try_recv() {
            out.push(signal);
        }
        out
    }

    pub(crate) fn requeue_mesh_signal(&self, signal: MeshTraversalSignal) -> bool {
        let (kind, peer_npub) = match &signal {
            MeshTraversalSignal::Offer { peer_npub, .. } => ("offer", peer_npub.clone()),
            MeshTraversalSignal::Answer { peer_npub, .. } => ("answer", peer_npub.clone()),
        };
        if let Err(error) = self.mesh_signal_tx.try_send(signal) {
            debug!(
                kind,
                peer = %short_npub(&peer_npub),
                error = %error,
                "dropping deferred mesh traversal signal because node signal channel is full"
            );
            return false;
        }
        true
    }

    async fn emit_event(&self, event: BootstrapEvent) {
        let (kind, peer_npub) = match &event {
            BootstrapEvent::Established { traversal } => {
                ("established", traversal.peer_npub.clone())
            }
            BootstrapEvent::Failed { peer_config, .. } => ("failed", peer_config.npub.clone()),
        };
        if self.event_tx.send(event).await.is_err() {
            debug!(
                kind,
                peer = %short_npub(&peer_npub),
                "dropping Nostr traversal event because node event receiver is closed"
            );
        }
    }

    async fn emit_mesh_signal(&self, signal: MeshTraversalSignal) -> bool {
        let (kind, peer_npub) = match &signal {
            MeshTraversalSignal::Offer { peer_npub, .. } => ("offer", peer_npub.clone()),
            MeshTraversalSignal::Answer { peer_npub, .. } => ("answer", peer_npub.clone()),
        };
        if self.mesh_signal_tx.send(signal).await.is_err() {
            debug!(
                kind,
                peer = %short_npub(&peer_npub),
                "dropping mesh traversal signal because node signal channel is closed"
            );
            return false;
        }
        true
    }

    pub async fn update_local_advert(
        self: &Arc<Self>,
        advert: Option<OverlayAdvert>,
    ) -> Result<(), BootstrapError> {
        let changed = {
            let mut slot = self.local_advert.write().await;
            if *slot == advert {
                false
            } else {
                *slot = advert;
                true
            }
        };
        if !changed {
            return Ok(());
        }
        self.request_publish_advert();
        Ok(())
    }

    pub async fn advert_endpoints_for_peer(
        &self,
        peer_npub: &str,
    ) -> Result<Vec<OverlayEndpointAdvert>, BootstrapError> {
        let target_pubkey =
            PublicKey::parse(peer_npub).map_err(|e| BootstrapError::InvalidPeerNpub {
                npub: peer_npub.to_string(),
                reason: e.to_string(),
            })?;
        let advert = self.fetch_advert(peer_npub, target_pubkey).await?;
        Ok(advert.endpoints)
    }

    pub async fn cached_advert_endpoints_for_peer(
        &self,
        peer_npub: &str,
    ) -> Option<Vec<OverlayEndpointAdvert>> {
        self.prune_advert_cache().await;
        let now = now_ms();
        self.advert_cache
            .read()
            .await
            .get(peer_npub)
            .filter(|cached| cached.valid_until_ms > now)
            .map(|cached| cached.advert.endpoints.clone())
    }

    pub async fn cached_open_discovery_candidates(
        &self,
        max: usize,
    ) -> Vec<(String, Vec<OverlayEndpointAdvert>, u64)> {
        self.prune_advert_cache().await;
        let now = now_ms();
        let cache = self.advert_cache.read().await;
        cache
            .values()
            .filter(|entry| entry.author_npub != self.npub)
            .filter(|entry| entry.valid_until_ms > now)
            .map(|entry| {
                (
                    entry.author_npub.clone(),
                    entry.advert.endpoints.clone(),
                    entry.created_at,
                )
            })
            .take(max)
            .collect()
    }

    pub async fn shutdown(&self) -> Result<(), BootstrapError> {
        if let Some(handle) = self.advertise_task.lock().await.take() {
            handle.abort();
        }
        if let Some(handle) = self.connect_task.lock().await.take() {
            handle.abort();
        }
        if let Some(handle) = self.relay_startup_task.lock().await.take() {
            handle.abort();
        }
        if let Some(handle) = self.publish_task.lock().await.take() {
            handle.abort();
        }

        // Don't proactively retract the advert via NIP-09 on shutdown.
        // Parameterized-replaceable semantics handle restart supersedence,
        // and NIP-40 expiration (advert_ttl_secs) bounds staleness on
        // permanent shutdown. An explicit retraction races with the next
        // daemon's republish on strict relays (e.g. Damus rate-limits the
        // burst, leaving the advert deleted and never restored).
        let _ = self.current_advert_event_id.write().await.take();

        if let Some(handle) = self.notify_task.lock().await.take() {
            handle.abort();
        }

        Ok(())
    }

    fn spawn_notify_loop(
        self: Arc<Self>,
        mut notifications: broadcast::Receiver<RelayPoolNotification>,
    ) -> JoinHandle<()> {
        tokio::spawn(async move {
            let started_at = Instant::now();
            let mut first_event_seen = false;
            info!("nostr notify loop entered");
            loop {
                let notification = match notifications.recv().await {
                    Ok(notification) => notification,
                    Err(broadcast::error::RecvError::Lagged(skipped)) => {
                        warn!(
                            skipped,
                            "nostr notification channel lagged; advert/signal events dropped"
                        );
                        continue;
                    }
                    Err(broadcast::error::RecvError::Closed) => {
                        warn!("nostr notification channel closed; notify loop exiting");
                        break;
                    }
                };
                if !first_event_seen {
                    first_event_seen = true;
                    info!(
                        elapsed_ms = started_at.elapsed().as_millis() as u64,
                        "nostr notify loop received first event"
                    );
                }
                if let RelayPoolNotification::Event { event, .. } = notification {
                    if event.kind == Kind::Custom(ADVERT_KIND) {
                        let Ok(verified_event) = VerifiedEvent::try_from(event.as_ref()) else {
                            continue;
                        };
                        let author_npub = verified_event.pubkey().to_bech32().expect("infallible");
                        if let Some(valid_until_ms) = self.event_valid_until_ms(&event)
                            && let Ok(advert) =
                                Self::parse_overlay_advert_event(verified_event, &self.config.app)
                        {
                            let mut cache = self.advert_cache.write().await;
                            let should_replace = cache
                                .get(&author_npub)
                                .map(|existing| existing.created_at <= event.created_at.as_secs())
                                .unwrap_or(true);
                            if should_replace && author_npub != self.npub {
                                debug!(
                                    peer = %short_npub(&author_npub),
                                    endpoints = %endpoint_summary(&advert.endpoints),
                                    event = %short_id(&event.id.to_string()),
                                    "advert: peer cached"
                                );
                            }
                            if should_replace {
                                cache.insert(
                                    author_npub.clone(),
                                    CachedOverlayAdvert {
                                        author_npub,
                                        advert,
                                        created_at: event.created_at.as_secs(),
                                        valid_until_ms,
                                    },
                                );
                            }
                        }
                        self.prune_advert_cache().await;
                        continue;
                    }

                    if event.kind != Kind::Custom(SIGNAL_KIND) {
                        continue;
                    }

                    let unwrapped = match unwrap_signal_event(&self.keys, &event).await {
                        Ok(unwrapped) => unwrapped,
                        Err(err) => {
                            trace!(error = %err, "failed to unwrap traversal signal");
                            continue;
                        }
                    };
                    let sender_npub = match unwrapped.sender.to_bech32() {
                        Ok(npub) => npub,
                        Err(err) => {
                            debug!(error = %err, "failed to encode traversal sender npub");
                            continue;
                        }
                    };

                    if let Ok(answer) =
                        serde_json::from_str::<TraversalAnswer>(&unwrapped.rumor.content)
                        && answer.message_type == "answer"
                        && answer.recipient_npub == self.npub
                    {
                        if let Some(tx) = self
                            .pending_answers
                            .lock()
                            .await
                            .remove(&answer.in_reply_to)
                        {
                            let _ = tx.send(SignalEnvelope {
                                payload: answer,
                                event_id: Some(event.id),
                                sender_npub: sender_npub.clone(),
                            });
                        }
                        continue;
                    }

                    if let Ok(offer) =
                        serde_json::from_str::<TraversalOffer>(&unwrapped.rumor.content)
                        && offer.message_type == "offer"
                        && offer.recipient_npub == self.npub
                    {
                        let Ok(permit) = self.offer_slots.clone().try_acquire_owned() else {
                            warn!(
                                sender_npub = %sender_npub,
                                limit = self.config.max_concurrent_incoming_offers,
                                "rate-limited inbound traversal offer (max_concurrent_incoming_offers reached); offer dropped"
                            );
                            continue;
                        };
                        let runtime = Arc::clone(&self);
                        tokio::spawn(async move {
                            let _permit = permit;
                            if let Err(err) = runtime
                                .handle_incoming_offer(offer, unwrapped.sender, sender_npub)
                                .await
                            {
                                debug!(error = %err, "failed to handle traversal offer");
                            }
                        });
                    }
                }
            }
        })
    }

    fn spawn_advertise_loop(self: Arc<Self>) -> JoinHandle<()> {
        tokio::spawn(async move {
            let mut interval =
                tokio::time::interval(Duration::from_secs(self.config.advert_refresh_secs.max(1)));
            // Swallow the immediate first tick: Node::start() requests the
            // initial advert publish via update_local_advert().
            interval.tick().await;
            loop {
                interval.tick().await;
                self.request_publish_advert();
            }
        })
    }

    fn spawn_relay_startup_loop(self: Arc<Self>) -> JoinHandle<()> {
        tokio::spawn(async move {
            let mut retry_delay = Duration::from_secs(2);
            loop {
                let subscribed =
                    match tokio::time::timeout(RELAY_STARTUP_OP_TIMEOUT, self.subscribe()).await {
                        Ok(Ok(())) => true,
                        Ok(Err(err)) => {
                            warn!(error = %err, "failed to subscribe to Nostr discovery relays");
                            false
                        }
                        Err(_) => {
                            warn!(
                                timeout_ms = RELAY_STARTUP_OP_TIMEOUT.as_millis() as u64,
                                "Nostr discovery relay subscribe timed out"
                            );
                            false
                        }
                    };
                match tokio::time::timeout(RELAY_STARTUP_OP_TIMEOUT, self.publish_inbox_relays())
                    .await
                {
                    Ok(Ok(())) => {}
                    Ok(Err(err)) => {
                        warn!(error = %err, "failed to publish Nostr inbox relay list");
                    }
                    Err(_) => {
                        warn!(
                            timeout_ms = RELAY_STARTUP_OP_TIMEOUT.as_millis() as u64,
                            "Nostr inbox relay publish timed out"
                        );
                    }
                }

                self.request_publish_advert();

                if subscribed {
                    break;
                }

                tokio::time::sleep(retry_delay).await;
                retry_delay = retry_delay.saturating_mul(2).min(Duration::from_secs(60));
            }
        })
    }

    fn spawn_connect_loop(self: Arc<Self>) -> JoinHandle<()> {
        tokio::spawn(async move {
            self.client.connect().await;
        })
    }

    fn spawn_publish_loop(self: Arc<Self>) -> JoinHandle<()> {
        tokio::spawn(async move {
            loop {
                self.publish_notify.notified().await;
                match tokio::time::timeout(ADVERT_PUBLISH_TIMEOUT, self.publish_advert()).await {
                    Ok(Ok(())) => {}
                    Ok(Err(err)) => {
                        warn!(error = %err, "failed to publish traversal advert");
                    }
                    Err(_) => {
                        warn!(
                            timeout_ms = ADVERT_PUBLISH_TIMEOUT.as_millis() as u64,
                            "Nostr traversal advert publish timed out"
                        );
                    }
                }
            }
        })
    }

    fn request_publish_advert(&self) {
        self.publish_notify.notify_one();
    }

    fn punch_hint(&self) -> PunchHint {
        PunchHint {
            start_at_ms: now_ms() + self.config.punch_start_delay_ms,
            interval_ms: self.config.punch_interval_ms,
            duration_ms: self.config.punch_duration_ms,
        }
    }

    async fn subscribe(&self) -> Result<(), BootstrapError> {
        let relay_config = self.relay_config.read().await.clone();
        self.client
            .subscribe_to(
                relay_config.dm_relays.clone(),
                Filter::new()
                    .kind(Kind::Custom(SIGNAL_KIND))
                    .pubkey(self.pubkey)
                    .limit(0),
                None,
            )
            .await
            .map_err(|e| BootstrapError::Nostr(e.to_string()))?;

        self.client
            .subscribe_to(
                relay_config.advert_relays.clone(),
                Filter::new()
                    .kind(Kind::Custom(ADVERT_KIND))
                    .identifier(advert_d_tag(&self.config.app)),
                None,
            )
            .await
            .map_err(|e| BootstrapError::Nostr(e.to_string()))?;

        Ok(())
    }

    async fn publish_inbox_relays(&self) -> Result<(), BootstrapError> {
        let relay_config = self.relay_config.read().await.clone();
        let tags = relay_config
            .dm_relays
            .iter()
            .filter_map(|relay| RelayUrl::parse(relay).ok())
            .map(|relay| {
                Tag::custom(
                    TagKind::SingleLetter(SingleLetterTag::lowercase(Alphabet::R)),
                    [relay.to_string()],
                )
            })
            .collect::<Vec<_>>();

        let event = EventBuilder::new(Kind::InboxRelays, "")
            .tags(tags)
            .sign_with_keys(&self.keys)
            .map_err(|e| BootstrapError::Nostr(e.to_string()))?;
        self.client
            .send_event_to(relay_config.dm_relays.clone(), &event)
            .await
            .map_err(|e| BootstrapError::Nostr(e.to_string()))?;
        Ok(())
    }

    async fn publish_advert(&self) -> Result<(), BootstrapError> {
        let previous_event_id = self.current_advert_event_id.read().await.to_owned();
        if !self.config.advertise {
            if let Some(event_id) = previous_event_id {
                let relay_config = self.relay_config.read().await.clone();
                self.publish_delete(&relay_config.advert_relays, [event_id])
                    .await?;
                *self.current_advert_event_id.write().await = None;
            }
            return Ok(());
        }

        let mut advert = match self.local_advert.read().await.clone() {
            Some(advert) => advert,
            // Transient absence (e.g., a single tick during startup where
            // build_overlay_advert briefly returns None). Don't proactively
            // emit a NIP-09 delete: the next publish supersedes the old
            // event via parameterized-replaceable semantics, and the NIP-40
            // expiration tag bounds the worst case if we never re-publish.
            None => return Ok(()),
        };

        advert.identifier = ADVERT_IDENTIFIER.to_string();
        advert.version = ADVERT_VERSION;
        advert.endpoints.retain(endpoint_advert_is_publicly_usable);
        // Defensive: build_overlay_advert returns None on empty endpoints,
        // so this is only reachable from non-lifecycle callers.
        if advert.endpoints.is_empty() {
            return Ok(());
        }

        if advert.has_udp_nat_endpoint() {
            if advert
                .signal_relays
                .as_ref()
                .is_none_or(|relays| relays.is_empty())
            {
                return Err(BootstrapError::InvalidAdvert(
                    "udp:nat endpoint requires non-empty signalRelays".to_string(),
                ));
            }
            if advert
                .stun_servers
                .as_ref()
                .is_none_or(|servers| servers.is_empty())
            {
                return Err(BootstrapError::InvalidAdvert(
                    "udp:nat endpoint requires non-empty stunServers".to_string(),
                ));
            }
        } else {
            advert.signal_relays = None;
            advert.stun_servers = None;
        }

        let expires_at = now_ms() + self.config.advert_ttl_secs * 1000;
        let tags = vec![
            Tag::identifier(advert_d_tag(&self.config.app)),
            Tag::custom(TagKind::custom("protocol"), [self.config.app.clone()]),
            Tag::custom(TagKind::custom("version"), [PROTOCOL_VERSION.to_string()]),
            Tag::expiration(Timestamp::from((expires_at / 1000).max(1))),
        ];

        let event = EventBuilder::new(Kind::Custom(ADVERT_KIND), serde_json::to_string(&advert)?)
            .tags(tags)
            .sign_with_keys(&self.keys)
            .map_err(|e| BootstrapError::Nostr(e.to_string()))?;
        let relay_config = self.relay_config.read().await.clone();
        self.client
            .send_event_to(relay_config.advert_relays.clone(), &event)
            .await
            .map_err(|e| BootstrapError::Nostr(e.to_string()))?;
        debug!(
            event = %short_id(&event.id.to_string()),
            relays = relay_config.advert_relays.len(),
            endpoints = %endpoint_summary(&advert.endpoints),
            ttl_secs = self.config.advert_ttl_secs,
            "advert: published"
        );
        // Kind 37195 lives in NIP-01's parameterized replaceable range
        // (30000–39999). Relays supersede the previous event for the same
        // (pubkey, kind, d-tag) triple by created_at — emitting an explicit
        // NIP-09 delete here is redundant and races with the replacement
        // publish, which strict relays (e.g. Damus) honor by removing the
        // new advert too.
        *self.current_advert_event_id.write().await = Some(event.id);
        Ok(())
    }

    async fn connect_peer(
        &self,
        peer_config: PeerConfig,
        mesh_signaling_allowed: bool,
    ) -> Result<EstablishedTraversal, BootstrapError> {
        let peer_short = short_npub(&peer_config.npub);
        if !self.traversal_initiator_admission_allowed(mesh_signaling_allowed) {
            debug!(
                peer = %peer_short,
                mesh_signaling_allowed,
                "traversal: initiator suppressed, Node at capacity"
            );
            return Err(BootstrapError::Disabled);
        }
        debug!(
            peer = %peer_short,
            mesh_signaling_allowed,
            "traversal: initiator starting"
        );
        let target_pubkey =
            PublicKey::parse(&peer_config.npub).map_err(|e| BootstrapError::InvalidPeerNpub {
                npub: peer_config.npub.clone(),
                reason: e.to_string(),
            })?;

        let mut relays = Vec::new();
        let mut nostr_setup_error = None;
        match self.fetch_advert(&peer_config.npub, target_pubkey).await {
            Ok(advert) => {
                if advert.has_udp_nat_endpoint() {
                    match self
                        .preferred_signal_relays(target_pubkey, Some(&advert))
                        .await
                    {
                        Ok(preferred) => relays = preferred,
                        Err(err) if mesh_signaling_allowed => {
                            debug!(
                                peer = %peer_short,
                                error = %err,
                                "traversal: continuing with mesh signaling after relay lookup failure"
                            );
                            nostr_setup_error = Some(err);
                        }
                        Err(err) => return Err(err),
                    }
                } else if mesh_signaling_allowed {
                    debug!(
                        peer = %peer_short,
                        "traversal: peer advert has no udp:nat endpoint; trying mesh signaling"
                    );
                } else {
                    return Err(BootstrapError::MissingNatEndpoint(peer_config.npub.clone()));
                }
            }
            Err(err) if mesh_signaling_allowed => {
                debug!(
                    peer = %peer_short,
                    error = %err,
                    "traversal: continuing with mesh signaling after advert lookup failure"
                );
                nostr_setup_error = Some(err);
            }
            Err(err) => return Err(err),
        }

        let base_socket = bind_traversal_udp_socket()?;

        let (reflexive_address, local_addresses, stun_server) = observe_traversal_addresses(
            &base_socket,
            &self.config.stun_servers,
            self.config.share_local_candidates,
            super::stun::TRAVERSAL_STUN_TIMEOUT,
        )
        .await?;
        debug!(
            peer = %peer_short,
            reflexive = %reflexive_address.as_ref().map(|a| format!("{}:{}", a.ip, a.port)).unwrap_or_else(|| "-".into()),
            local = local_addresses.len(),
            stun = %stun_server.as_deref().unwrap_or("-"),
            "traversal: initiator STUN observed"
        );
        let session_id = nonce();
        let offer = create_traversal_offer(
            session_id.clone(),
            now_ms(),
            self.config.signal_ttl_secs * 1000,
            session_id.clone(),
            self.npub.clone(),
            peer_config.npub.clone(),
            reflexive_address,
            local_addresses,
            stun_server,
        );

        let (tx, rx) = oneshot::channel();
        self.pending_answers
            .lock()
            .await
            .insert(offer.nonce.clone(), tx);

        let mesh_offer_sent = mesh_signaling_allowed
            && self
                .emit_mesh_signal(MeshTraversalSignal::Offer {
                    peer_npub: peer_config.npub.clone(),
                    offer: offer.clone(),
                })
                .await;
        if mesh_offer_sent {
            debug!(
                peer = %peer_short,
                session = %short_id(&offer.session_id),
                "traversal: offer queued for FIPS mesh signaling"
            );
        }

        let offer_event = if relays.is_empty() {
            None
        } else {
            match self.send_signal(&relays, target_pubkey, &offer).await {
                Ok(event) => {
                    debug!(
                        peer = %peer_short,
                        session = %short_id(&offer.session_id),
                        relays = relays.len(),
                        event = %short_id(&event.id.to_string()),
                        "traversal: offer sent over Nostr"
                    );
                    Some(event)
                }
                Err(err) if mesh_offer_sent => {
                    debug!(
                        peer = %peer_short,
                        session = %short_id(&offer.session_id),
                        error = %err,
                        "traversal: Nostr offer failed; waiting for mesh-signaled answer"
                    );
                    nostr_setup_error = Some(err);
                    None
                }
                Err(err) => {
                    let _ = self.pending_answers.lock().await.remove(&offer.nonce);
                    return Err(err);
                }
            }
        };

        if !mesh_offer_sent && offer_event.is_none() {
            let _ = self.pending_answers.lock().await.remove(&offer.nonce);
            return Err(nostr_setup_error
                .unwrap_or_else(|| BootstrapError::MissingRelays(peer_config.npub.clone())));
        }

        let answer = match tokio::time::timeout(
            Duration::from_secs(self.config.signal_ttl_secs),
            rx,
        )
        .await
        {
            Ok(Ok(answer)) => answer,
            Ok(Err(_)) => {
                let _ = self.pending_answers.lock().await.remove(&offer.nonce);
                return Err(BootstrapError::Protocol(
                    "answer channel closed".to_string(),
                ));
            }
            Err(_) => {
                let _ = self.pending_answers.lock().await.remove(&offer.nonce);
                return Err(BootstrapError::SignalTimeout(peer_config.npub));
            }
        };

        let answer_received_at = now_ms();
        debug!(
            peer = %peer_short,
            session = %short_id(&offer.session_id),
            accepted = answer.payload.accepted,
            signal_path = if answer.event_id.is_some() { "nostr" } else { "fips-mesh" },
            reflexive = %answer.payload.reflexive_address.as_ref().map(|a| format!("{}:{}", a.ip, a.port)).unwrap_or_else(|| "-".into()),
            local = answer.payload.local_addresses.len(),
            "traversal: answer received"
        );
        if let Some(observed_skew_ms) =
            estimate_clock_skew(&offer, &answer.payload, answer_received_at)
        {
            self.failure_state.note_observed_skew(
                &peer_config.npub,
                observed_skew_ms,
                answer_received_at,
            );
            let abs_skew = observed_skew_ms.unsigned_abs();
            // 30s threshold: well below the 60s SKEW_TOLERANCE wall but loud
            // enough to surface a real clock problem on either side.
            if abs_skew >= 30_000 {
                debug!(
                    peer = %peer_short,
                    session = %short_id(&offer.session_id),
                    skew_ms = observed_skew_ms,
                    "traversal: significant peer clock skew observed"
                );
            } else {
                trace!(
                    peer = %peer_short,
                    skew_ms = observed_skew_ms,
                    "traversal: peer clock skew within nominal range"
                );
            }
        }
        let outcome = validate_traversal_answer_for_offer(
            &offer,
            &answer.payload,
            answer_received_at,
            self.config.signal_ttl_secs * 1000,
            &answer.sender_npub,
            &self.npub,
        )?;
        if outcome == FreshnessOutcome::FreshWithinSkewTolerance {
            debug!(
                peer = %peer_short,
                session = %short_id(&offer.session_id),
                "traversal: answer accepted within clock-skew tolerance"
            );
        }
        if !answer.payload.accepted {
            return Err(BootstrapError::Protocol(
                answer
                    .payload
                    .reason
                    .unwrap_or_else(|| "remote rejected traversal".to_string()),
            ));
        }

        let planned_remotes = planned_remote_endpoints(
            &offer.local_addresses,
            offer.reflexive_address.as_ref(),
            &answer.payload.local_addresses,
            answer.payload.reflexive_address.as_ref(),
            true,
        )?;

        let remote_addr = run_punch_attempt(
            &base_socket,
            &session_id,
            &planned_remotes.remotes,
            self.punch_hint(),
            Duration::from_secs(self.config.attempt_timeout_secs),
            planned_remotes.preferred_count,
        )
        .await
        .map_err(|_| BootstrapError::PunchTimeout(peer_config.npub.clone()))?;
        debug!(
            peer = %peer_short,
            session = %short_id(&session_id),
            remote = %remote_addr,
            "traversal: initiator punch succeeded"
        );

        let answer_via_nostr = answer.event_id.is_some();
        let delete_ids = offer_event
            .as_ref()
            .map(|event| event.id)
            .into_iter()
            .chain(answer.event_id)
            .collect::<Vec<_>>();
        if !delete_ids.is_empty() {
            let _ = self.publish_delete(&relays, delete_ids).await;
        }

        self.failure_state
            .record_success(&peer_config.npub, now_ms());

        let transport_name = if answer_via_nostr {
            "nostr-nat"
        } else {
            "fips-mesh-nat"
        };
        Ok(
            EstablishedTraversal::new(session_id, peer_config.npub, remote_addr, base_socket)
                .with_transport_name(transport_name),
        )
    }

    pub(crate) async fn receive_mesh_traversal_answer(
        &self,
        answer: TraversalAnswer,
        sender_npub: String,
    ) {
        if answer.message_type != "answer" || answer.recipient_npub != self.npub {
            debug!(
                peer = %short_npub(&sender_npub),
                session = %short_id(&answer.session_id),
                "traversal: ignoring mesh answer with mismatched type or recipient"
            );
            return;
        }

        if let Some(tx) = self
            .pending_answers
            .lock()
            .await
            .remove(&answer.in_reply_to)
        {
            let _ = tx.send(SignalEnvelope {
                payload: answer,
                event_id: None,
                sender_npub,
            });
        } else {
            debug!(
                peer = %short_npub(&sender_npub),
                session = %short_id(&answer.session_id),
                "traversal: ignoring mesh answer without pending offer"
            );
        }
    }

    pub(crate) async fn receive_mesh_traversal_offer(
        self: &Arc<Self>,
        offer: TraversalOffer,
        sender_npub: String,
    ) {
        if offer.message_type != "offer" || offer.recipient_npub != self.npub {
            debug!(
                peer = %short_npub(&sender_npub),
                session = %short_id(&offer.session_id),
                "traversal: ignoring mesh offer with mismatched type or recipient"
            );
            return;
        }

        let Ok(permit) = self.offer_slots.clone().try_acquire_owned() else {
            warn!(
                sender_npub = %sender_npub,
                limit = self.config.max_concurrent_incoming_offers,
                "rate-limited inbound mesh traversal offer (max_concurrent_incoming_offers reached); offer dropped"
            );
            return;
        };

        let runtime = Arc::clone(self);
        tokio::spawn(async move {
            let _permit = permit;
            if let Err(err) = runtime.handle_incoming_mesh_offer(offer, sender_npub).await {
                debug!(error = %err, "failed to handle mesh traversal offer");
            }
        });
    }

    async fn handle_incoming_mesh_offer(
        self: Arc<Self>,
        offer: TraversalOffer,
        sender_npub: String,
    ) -> Result<(), BootstrapError> {
        let peer_short = short_npub(&sender_npub);
        if !self.direct_refresh_admission_allowed() {
            debug!(
                peer = %peer_short,
                session = %short_id(&offer.session_id),
                "traversal: incoming mesh offer dropped, Node at connection/link capacity"
            );
            return Ok(());
        }
        let offer_received_at = now_ms();
        debug!(
            peer = %peer_short,
            session = %short_id(&offer.session_id),
            reflexive = %offer.reflexive_address.as_ref().map(|a| format!("{}:{}", a.ip, a.port)).unwrap_or_else(|| "-".into()),
            local = offer.local_addresses.len(),
            "traversal: mesh offer received"
        );
        let outcome = validate_offer_freshness(
            &offer,
            offer_received_at,
            self.config.signal_ttl_secs * 1000,
            &sender_npub,
            &self.npub,
        )?;
        if outcome == FreshnessOutcome::FreshWithinSkewTolerance {
            debug!(
                peer = %peer_short,
                session = %short_id(&offer.session_id),
                offer_issued_at = offer.issued_at,
                offer_received_at = offer_received_at,
                "traversal: mesh offer accepted within clock-skew tolerance"
            );
        }
        self.mark_session_seen(&offer.session_id).await?;

        let base_socket = bind_traversal_udp_socket()?;
        let (reflexive_address, local_addresses, stun_server) = observe_traversal_addresses(
            &base_socket,
            &self.config.stun_servers,
            self.config.share_local_candidates,
            super::stun::TRAVERSAL_STUN_TIMEOUT,
        )
        .await?;
        let accepted = reflexive_address.is_some() || !local_addresses.is_empty();
        debug!(
            peer = %peer_short,
            session = %short_id(&offer.session_id),
            accepted = accepted,
            reflexive = %reflexive_address.as_ref().map(|a| format!("{}:{}", a.ip, a.port)).unwrap_or_else(|| "-".into()),
            local = local_addresses.len(),
            "traversal: mesh responder STUN observed"
        );
        let answer = create_traversal_answer(
            offer.session_id.clone(),
            now_ms(),
            self.config.signal_ttl_secs * 1000,
            nonce(),
            self.npub.clone(),
            offer.sender_npub.clone(),
            offer.nonce.clone(),
            accepted,
            reflexive_address,
            local_addresses,
            stun_server,
            accepted.then(|| self.punch_hint()),
            (!accepted).then_some("no-usable-addresses".to_string()),
            Some(offer_received_at),
        );
        if !self
            .emit_mesh_signal(MeshTraversalSignal::Answer {
                peer_npub: sender_npub.clone(),
                answer: answer.clone(),
            })
            .await
        {
            return Err(BootstrapError::Protocol(
                "mesh traversal answer queue full".to_string(),
            ));
        }
        debug!(
            peer = %peer_short,
            session = %short_id(&offer.session_id),
            accepted = accepted,
            "traversal: answer queued for FIPS mesh signaling"
        );
        if !accepted {
            return Ok(());
        }

        let planned_remotes = planned_remote_endpoints(
            &answer.local_addresses,
            answer.reflexive_address.as_ref(),
            &offer.local_addresses,
            offer.reflexive_address.as_ref(),
            true,
        )?;

        if let Ok(remote_addr) = run_punch_attempt(
            &base_socket,
            &offer.session_id,
            &planned_remotes.remotes,
            answer
                .punch
                .clone()
                .expect("accepted answers always include a punch hint"),
            Duration::from_secs(self.config.attempt_timeout_secs),
            planned_remotes.preferred_count,
        )
        .await
        {
            debug!(
                peer = %peer_short,
                session = %short_id(&offer.session_id),
                remote = %remote_addr,
                "traversal: mesh responder punch succeeded"
            );
            self.emit_event(BootstrapEvent::Established {
                traversal: EstablishedTraversal::new(
                    offer.session_id,
                    offer.sender_npub,
                    remote_addr,
                    base_socket,
                )
                .with_transport_name("fips-mesh-nat"),
            })
            .await;
        }

        Ok(())
    }

    async fn handle_incoming_offer(
        self: Arc<Self>,
        offer: TraversalOffer,
        sender: PublicKey,
        sender_npub: String,
    ) -> Result<(), BootstrapError> {
        let peer_short = short_npub(&sender_npub);
        if !self.outbound_admission_allowed() {
            debug!(
                peer = %peer_short,
                session = %short_id(&offer.session_id),
                "traversal: incoming offer dropped, Node at capacity"
            );
            return Ok(());
        }
        let offer_received_at = now_ms();
        debug!(
            peer = %peer_short,
            session = %short_id(&offer.session_id),
            reflexive = %offer.reflexive_address.as_ref().map(|a| format!("{}:{}", a.ip, a.port)).unwrap_or_else(|| "-".into()),
            local = offer.local_addresses.len(),
            "traversal: offer received"
        );
        let outcome = validate_offer_freshness(
            &offer,
            offer_received_at,
            self.config.signal_ttl_secs * 1000,
            &sender_npub,
            &self.npub,
        )?;
        if outcome == FreshnessOutcome::FreshWithinSkewTolerance {
            debug!(
                peer = %peer_short,
                session = %short_id(&offer.session_id),
                offer_issued_at = offer.issued_at,
                offer_received_at = offer_received_at,
                "traversal: offer accepted within clock-skew tolerance"
            );
        }
        self.mark_session_seen(&offer.session_id).await?;

        let base_socket = bind_traversal_udp_socket()?;
        let (reflexive_address, local_addresses, stun_server) = observe_traversal_addresses(
            &base_socket,
            &self.config.stun_servers,
            self.config.share_local_candidates,
            super::stun::TRAVERSAL_STUN_TIMEOUT,
        )
        .await?;
        let accepted = reflexive_address.is_some() || !local_addresses.is_empty();
        debug!(
            peer = %peer_short,
            session = %short_id(&offer.session_id),
            accepted = accepted,
            reflexive = %reflexive_address.as_ref().map(|a| format!("{}:{}", a.ip, a.port)).unwrap_or_else(|| "-".into()),
            local = local_addresses.len(),
            "traversal: responder STUN observed"
        );
        let answer = create_traversal_answer(
            offer.session_id.clone(),
            now_ms(),
            self.config.signal_ttl_secs * 1000,
            nonce(),
            self.npub.clone(),
            offer.sender_npub.clone(),
            offer.nonce.clone(),
            accepted,
            reflexive_address,
            local_addresses,
            stun_server,
            accepted.then(|| self.punch_hint()),
            (!accepted).then_some("no-usable-addresses".to_string()),
            Some(offer_received_at),
        );
        let relays = self.preferred_signal_relays(sender, None).await?;
        let answer_event = self.send_signal(&relays, sender, &answer).await?;
        debug!(
            peer = %peer_short,
            session = %short_id(&offer.session_id),
            accepted = accepted,
            relays = relays.len(),
            event = %short_id(&answer_event.id.to_string()),
            "traversal: answer sent"
        );
        if !accepted {
            let _ = self.publish_delete(&relays, [answer_event.id]).await;
            return Ok(());
        }

        let planned_remotes = planned_remote_endpoints(
            &answer.local_addresses,
            answer.reflexive_address.as_ref(),
            &offer.local_addresses,
            offer.reflexive_address.as_ref(),
            true,
        )?;

        if let Ok(remote_addr) = run_punch_attempt(
            &base_socket,
            &offer.session_id,
            &planned_remotes.remotes,
            answer
                .punch
                .clone()
                .expect("accepted answers always include a punch hint"),
            Duration::from_secs(self.config.attempt_timeout_secs),
            planned_remotes.preferred_count,
        )
        .await
        {
            debug!(
                peer = %peer_short,
                session = %short_id(&offer.session_id),
                remote = %remote_addr,
                "traversal: responder punch succeeded"
            );
            self.emit_event(BootstrapEvent::Established {
                traversal: EstablishedTraversal::new(
                    offer.session_id,
                    offer.sender_npub,
                    remote_addr,
                    base_socket,
                )
                .with_transport_name("nostr-nat"),
            })
            .await;
        }

        let _ = self.publish_delete(&relays, [answer_event.id]).await;
        Ok(())
    }

    async fn fetch_advert(
        &self,
        peer_npub: &str,
        target_pubkey: PublicKey,
    ) -> Result<OverlayAdvert, BootstrapError> {
        self.prune_advert_cache().await;
        if let Some(cached) = self.advert_cache.read().await.get(peer_npub).cloned() {
            debug!(
                peer = %short_npub(peer_npub),
                source = "cache",
                endpoints = %endpoint_summary(&cached.advert.endpoints),
                "advert: resolved"
            );
            return Ok(cached.advert);
        }

        let relay_config = self.relay_config.read().await.clone();
        let events = self
            .client
            .fetch_events_from(
                relay_config.advert_relays.clone(),
                Filter::new()
                    .author(target_pubkey)
                    .kind(Kind::Custom(ADVERT_KIND))
                    .identifier(advert_d_tag(&self.config.app)),
                Duration::from_secs(2),
            )
            .await
            .map_err(|e| BootstrapError::Nostr(e.to_string()))?;

        let mut best: Option<CachedOverlayAdvert> = None;
        for event in events.iter() {
            let Some(valid_until_ms) = self.event_valid_until_ms(event) else {
                continue;
            };
            let Ok(verified_event) = VerifiedEvent::try_from(event) else {
                continue;
            };
            let author_npub = verified_event.pubkey().to_bech32().expect("infallible");
            let Ok(advert) = Self::parse_overlay_advert_event(verified_event, &self.config.app)
            else {
                continue;
            };
            if author_npub != peer_npub {
                continue;
            }
            let replace = best
                .as_ref()
                .map(|current| event.created_at.as_secs() >= current.created_at)
                .unwrap_or(true);
            if replace {
                best = Some(CachedOverlayAdvert {
                    author_npub,
                    advert,
                    created_at: event.created_at.as_secs(),
                    valid_until_ms,
                });
            }
        }

        let cached = best.ok_or_else(|| BootstrapError::MissingAdvert(peer_npub.to_string()))?;
        debug!(
            peer = %short_npub(peer_npub),
            source = "relay-fetch",
            endpoints = %endpoint_summary(&cached.advert.endpoints),
            "advert: resolved"
        );
        self.advert_cache
            .write()
            .await
            .insert(peer_npub.to_string(), cached.clone());
        self.prune_advert_cache().await;
        Ok(cached.advert)
    }

    async fn preferred_signal_relays(
        &self,
        target_pubkey: PublicKey,
        advert: Option<&OverlayAdvert>,
    ) -> Result<Vec<String>, BootstrapError> {
        let mut merged = self.find_recipient_inbox_relays(target_pubkey).await?;
        if let Some(advert) = advert
            && let Some(relays) = advert.signal_relays.as_ref()
        {
            for relay in relays {
                if !merged.contains(relay) {
                    merged.push(relay.clone());
                }
            }
        }
        let relay_config = self.relay_config.read().await.clone();
        for relay in &relay_config.dm_relays {
            if !merged.contains(relay) {
                merged.push(relay.clone());
            }
        }
        Ok(merged)
    }

    async fn find_recipient_inbox_relays(
        &self,
        target_pubkey: PublicKey,
    ) -> Result<Vec<String>, BootstrapError> {
        let relay_config = self.relay_config.read().await.clone();
        let mut lookup_relays = relay_config.dm_relays.clone();
        for relay in &relay_config.advert_relays {
            if !lookup_relays.contains(relay) {
                lookup_relays.push(relay.clone());
            }
        }
        let events = self
            .client
            .fetch_events_from(
                lookup_relays,
                Filter::new()
                    .author(target_pubkey)
                    .kind(Kind::InboxRelays)
                    .since(Timestamp::from(
                        Timestamp::now().as_secs().saturating_sub(30 * 24 * 60 * 60),
                    )),
                Duration::from_millis(1500),
            )
            .await;
        let events = match events {
            Ok(events) => events,
            Err(err) => {
                debug!(error = %err, "failed to fetch inbox relays, falling back to configured DM relays");
                return Ok(self.relay_config.read().await.dm_relays.clone());
            }
        };
        let newest = events.iter().max_by_key(|event| event.created_at.as_secs());
        if let Some(event) = newest {
            let relays = nip17::extract_relay_list(event)
                .map(|relay| relay.to_string())
                .collect::<Vec<_>>();
            if !relays.is_empty() {
                return Ok(relays);
            }
        }
        Ok(self.relay_config.read().await.dm_relays.clone())
    }

    pub(super) fn parse_overlay_advert_event(
        event: VerifiedEvent<'_>,
        expected_app: &str,
    ) -> Result<OverlayAdvert, BootstrapError> {
        let event = event.as_event();
        if event.kind != Kind::Custom(ADVERT_KIND) {
            return Err(BootstrapError::InvalidAdvert(
                "unexpected advert event kind".to_string(),
            ));
        }

        let advertised_app = event
            .tags
            .find(TagKind::custom("protocol"))
            .and_then(|tag| tag.content())
            .ok_or_else(|| {
                BootstrapError::InvalidAdvert("missing required protocol tag".to_string())
            })?;
        if advertised_app != expected_app {
            return Err(BootstrapError::InvalidAdvert(format!(
                "unsupported protocol '{}'",
                advertised_app
            )));
        }

        let advert: OverlayAdvert = serde_json::from_str(&event.content)?;
        Self::validate_overlay_advert(advert)
    }

    pub(super) fn validate_overlay_advert(
        mut advert: OverlayAdvert,
    ) -> Result<OverlayAdvert, BootstrapError> {
        if advert.identifier != ADVERT_IDENTIFIER {
            return Err(BootstrapError::InvalidAdvert(format!(
                "unsupported identifier '{}'",
                advert.identifier
            )));
        }
        if advert.version != ADVERT_VERSION {
            return Err(BootstrapError::InvalidAdvert(format!(
                "unsupported version '{}'",
                advert.version
            )));
        }
        if advert.endpoints.is_empty() {
            return Err(BootstrapError::InvalidAdvert(
                "missing required endpoints".to_string(),
            ));
        }
        advert.endpoints.retain(endpoint_advert_is_publicly_usable);
        if advert.endpoints.is_empty() {
            return Err(BootstrapError::InvalidAdvert(
                "missing publicly routable endpoints".to_string(),
            ));
        }

        let has_nat = advert.has_udp_nat_endpoint();
        let has_webrtc = advert
            .endpoints
            .iter()
            .any(|endpoint| endpoint.transport == super::types::OverlayTransportKind::WebRtc);
        if has_nat {
            if advert
                .signal_relays
                .as_ref()
                .is_none_or(|relays| relays.is_empty())
            {
                return Err(BootstrapError::InvalidAdvert(
                    "udp:nat endpoint requires signalRelays".to_string(),
                ));
            }
            if advert
                .stun_servers
                .as_ref()
                .is_none_or(|servers| servers.is_empty())
            {
                return Err(BootstrapError::InvalidAdvert(
                    "udp:nat endpoint requires stunServers".to_string(),
                ));
            }
        } else if has_webrtc {
            if advert
                .signal_relays
                .as_ref()
                .is_none_or(|relays| relays.is_empty())
            {
                return Err(BootstrapError::InvalidAdvert(
                    "webrtc endpoint requires signalRelays".to_string(),
                ));
            }
        } else {
            advert.signal_relays = None;
            advert.stun_servers = None;
        }

        Ok(advert)
    }

    async fn prune_advert_cache(&self) {
        let now = now_ms();
        let mut cache = self.advert_cache.write().await;
        cache.retain(|_, entry| entry.valid_until_ms > now);
        if cache.len() <= self.config.advert_cache_max_entries {
            return;
        }

        let mut oldest = cache
            .iter()
            .map(|(npub, entry)| (npub.clone(), entry.valid_until_ms))
            .collect::<Vec<_>>();
        oldest.sort_by_key(|(_, ts)| *ts);
        let overflow = cache
            .len()
            .saturating_sub(self.config.advert_cache_max_entries);
        for (npub, _) in oldest.into_iter().take(overflow) {
            cache.remove(&npub);
        }
        debug!(
            evicted = overflow,
            retained = cache.len(),
            cap = self.config.advert_cache_max_entries,
            "advert cache overflow; evicted oldest entries"
        );
    }

    fn advert_max_age_ms(&self) -> u64 {
        self.config.advert_ttl_secs * 1000 * ADVERT_CACHE_STALE_GRACE_MULTIPLIER
    }

    fn event_valid_until_ms(&self, event: &Event) -> Option<u64> {
        Self::compute_advert_valid_until_ms(event, self.advert_max_age_ms(), now_ms())
    }

    pub(super) fn compute_advert_valid_until_ms(
        event: &Event,
        advert_max_age_ms: u64,
        now_ms: u64,
    ) -> Option<u64> {
        if event.is_expired() {
            return None;
        }

        let created_ms = event.created_at.as_secs().saturating_mul(1000);
        let created_window_until = created_ms.saturating_add(advert_max_age_ms);
        if created_window_until <= now_ms {
            return None;
        }

        let expires_ms = event
            .tags
            .expiration()
            .map(|timestamp| timestamp.as_secs().saturating_mul(1000));
        let valid_until_ms = expires_ms
            .map(|expires| expires.min(created_window_until))
            .unwrap_or(created_window_until);

        (valid_until_ms > now_ms).then_some(valid_until_ms)
    }

    async fn send_signal<T: Serialize>(
        &self,
        relays: &[String],
        receiver: PublicKey,
        payload: &T,
    ) -> Result<Event, BootstrapError> {
        let rumor = EventBuilder::private_msg_rumor(receiver, serde_json::to_string(payload)?)
            .build(self.pubkey);
        let signal = build_signal_event(
            &self.keys,
            receiver,
            rumor,
            Timestamp::from((now_ms() + self.config.signal_ttl_secs * 1000) / 1000),
        )
        .await?;
        let relays = self.ensure_signal_relays(relays).await?;
        self.client
            .send_event_to(relays, &signal)
            .await
            .map_err(|e| BootstrapError::Nostr(e.to_string()))?;
        Ok(signal)
    }

    async fn publish_delete<I>(&self, relays: &[String], ids: I) -> Result<(), BootstrapError>
    where
        I: IntoIterator<Item = EventId>,
    {
        let event = EventBuilder::delete(nostr::nips::nip09::EventDeletionRequest::new().ids(ids))
            .sign_with_keys(&self.keys)
            .map_err(|e| BootstrapError::Nostr(e.to_string()))?;
        let relays = self.ensure_signal_relays(relays).await?;
        self.client
            .send_event_to(relays, &event)
            .await
            .map_err(|e| BootstrapError::Nostr(e.to_string()))?;
        Ok(())
    }

    async fn ensure_signal_relays(&self, relays: &[String]) -> Result<Vec<String>, BootstrapError> {
        let mut usable = Vec::new();
        for relay in relays {
            match self.client.add_relay(relay).await {
                Ok(_) => usable.push(relay.clone()),
                Err(error) => {
                    debug!(relay = %relay, error = %error, "failed to add signal relay");
                }
            }
        }
        if usable.is_empty() {
            return Err(BootstrapError::Nostr("no usable signal relays".to_string()));
        }
        self.client.connect().await;
        Ok(usable)
    }

    async fn mark_session_seen(&self, session_id: &str) -> Result<(), BootstrapError> {
        let now = now_ms();
        let expiry = now + self.config.replay_window_secs * 1000;
        let mut seen = self.seen_sessions.lock().await;
        seen.retain(|_, expires_at| *expires_at > now);
        if seen.contains_key(session_id) {
            return Err(BootstrapError::Replay(session_id.to_string()));
        }
        seen.insert(session_id.to_string(), expiry);
        if seen.len() > self.config.seen_sessions_max_entries {
            let mut oldest = seen
                .iter()
                .map(|(session, expires_at)| (session.clone(), *expires_at))
                .collect::<Vec<_>>();
            oldest.sort_by_key(|(_, expires_at)| *expires_at);
            let overflow = seen
                .len()
                .saturating_sub(self.config.seen_sessions_max_entries);
            for (session, _) in oldest.into_iter().take(overflow) {
                seen.remove(&session);
            }
            debug!(
                evicted = overflow,
                retained = seen.len(),
                cap = self.config.seen_sessions_max_entries,
                "seen-sessions cache overflow; evicted oldest entries"
            );
        }
        Ok(())
    }
}

#[derive(Debug)]
pub(super) struct VerifiedEvent<'a> {
    event: &'a Event,
}

impl<'a> VerifiedEvent<'a> {
    fn as_event(&self) -> &'a Event {
        self.event
    }

    fn pubkey(&self) -> &'a PublicKey {
        &self.event.pubkey
    }
}

impl<'a> TryFrom<&'a Event> for VerifiedEvent<'a> {
    type Error = BootstrapError;

    fn try_from(event: &'a Event) -> Result<Self, Self::Error> {
        event
            .verify()
            .map_err(|e| BootstrapError::InvalidAdvert(format!("invalid event signature: {e}")))?;
        Ok(Self { event })
    }
}

#[cfg(test)]
impl NostrDiscovery {
    /// Build a minimal `NostrDiscovery` for unit tests. No relay client is
    /// connected and no background tasks are spawned; only the in-memory
    /// `advert_cache` and `npub` are usable. Intended for cache-injection
    /// tests of consumers (e.g. `Node::run_open_discovery_sweep`).
    pub(crate) fn new_for_test() -> Self {
        let keys = nostr::Keys::generate();
        let pubkey = keys.public_key();
        let npub = pubkey.to_bech32().expect("bech32 encode");
        let client = Client::builder()
            .signer(keys.clone())
            .opts(ClientOptions::new().autoconnect(false))
            .build();
        let config = NostrDiscoveryConfig::default();
        let offer_slots = Arc::new(Semaphore::new(config.max_concurrent_incoming_offers));
        let (event_tx, event_rx) = mpsc::channel(event_channel_capacity(&config));
        let (mesh_signal_tx, mesh_signal_rx) = mpsc::channel(event_channel_capacity(&config));
        let failure_state = FailureState::new(
            config.failure_streak_threshold,
            config.extended_cooldown_secs,
            config.warn_log_interval_secs,
            config.failure_state_max_entries,
        );
        Self {
            client,
            keys,
            pubkey,
            npub,
            relay_config: RwLock::new(NostrRelayConfig::from(&config)),
            config,
            advert_cache: RwLock::new(HashMap::new()),
            local_advert: RwLock::new(None),
            current_advert_event_id: RwLock::new(None),
            pending_answers: Mutex::new(HashMap::new()),
            active_initiators: Mutex::new(HashSet::new()),
            active_refetches: Mutex::new(HashSet::new()),
            seen_sessions: Mutex::new(HashMap::new()),
            offer_slots,
            event_tx,
            event_rx: Mutex::new(event_rx),
            mesh_signal_tx,
            mesh_signal_rx: Mutex::new(mesh_signal_rx),
            connect_task: Mutex::new(None),
            relay_startup_task: Mutex::new(None),
            publish_task: Mutex::new(None),
            publish_notify: Notify::new(),
            notify_task: Mutex::new(None),
            advertise_task: Mutex::new(None),
            failure_state,
            public_udp_addr_cache: RwLock::new(HashMap::new()),
            outbound_admission: AtomicBool::new(true),
            direct_refresh_admission: AtomicBool::new(true),
        }
    }

    /// Build a `CachedOverlayAdvert` for tests with a single endpoint and
    /// a generous validity window (one hour from `now_ms()`).
    pub(crate) fn cached_advert_for_test(
        author_npub: String,
        endpoint: OverlayEndpointAdvert,
        created_at_secs: u64,
    ) -> CachedOverlayAdvert {
        CachedOverlayAdvert {
            author_npub: author_npub.clone(),
            advert: OverlayAdvert {
                identifier: ADVERT_IDENTIFIER.to_string(),
                version: ADVERT_VERSION,
                endpoints: vec![endpoint],
                signal_relays: None,
                stun_servers: None,
            },
            created_at: created_at_secs,
            valid_until_ms: now_ms().saturating_add(3_600_000),
        }
    }

    /// Insert a cached advert directly into the in-memory cache. Used by
    /// unit tests to set up consumer-side state without needing live relays.
    pub(crate) async fn insert_advert_for_test(&self, npub: String, advert: CachedOverlayAdvert) {
        let mut cache = self.advert_cache.write().await;
        cache.insert(npub, advert);
    }

    /// Queue a bootstrap event directly for lifecycle tests without live relays
    /// or a running traversal task.
    pub(crate) fn push_event_for_test(&self, event: BootstrapEvent) {
        let _ = self.event_tx.try_send(event);
    }

    pub(crate) fn push_mesh_signal_for_test(&self, signal: MeshTraversalSignal) {
        let _ = self.mesh_signal_tx.try_send(signal);
    }

    pub(crate) async fn active_initiator_count_for_test(&self) -> usize {
        self.active_initiators.lock().await.len()
    }
}

fn event_channel_capacity(config: &NostrDiscoveryConfig) -> usize {
    let work_limit = config
        .open_discovery_max_pending
        .max(config.max_concurrent_incoming_offers)
        .max(1);

    // This channel carries traversal outcomes/signals back to the node, not
    // just permission to start new open-discovery work. Give it enough burst
    // room for roster peers, startup cache sweeps, and inbound offers without
    // turning the open-discovery cap into an event-loss trigger.
    work_limit.saturating_mul(4).clamp(64, 4096)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::discovery::nostr::TraversalAddress;

    #[test]
    fn event_channel_capacity_tracks_open_and_inbound_limits() {
        let mut config = NostrDiscoveryConfig {
            open_discovery_max_pending: 8,
            max_concurrent_incoming_offers: 16,
            ..Default::default()
        };
        assert_eq!(event_channel_capacity(&config), 64);

        config.open_discovery_max_pending = 32;
        config.max_concurrent_incoming_offers = 4;
        assert_eq!(event_channel_capacity(&config), 128);

        config.open_discovery_max_pending = 0;
        config.max_concurrent_incoming_offers = 0;
        assert_eq!(event_channel_capacity(&config), 64);

        config.open_discovery_max_pending = 5000;
        config.max_concurrent_incoming_offers = 1;
        assert_eq!(event_channel_capacity(&config), 4096);
    }

    #[test]
    fn mesh_signaled_initiators_use_direct_refresh_admission() {
        let discovery = NostrDiscovery::new_for_test();

        discovery.set_outbound_admission(false);
        discovery.set_direct_refresh_admission(true);

        assert!(
            !discovery.traversal_initiator_admission_allowed(false),
            "ordinary Nostr traversal should still obey peer-slot capacity"
        );
        assert!(
            discovery.traversal_initiator_admission_allowed(true),
            "mesh-signaled direct refresh should bypass only the peer-slot cap"
        );

        discovery.set_direct_refresh_admission(false);
        assert!(
            !discovery.traversal_initiator_admission_allowed(true),
            "mesh-signaled direct refresh should still obey connection/link capacity"
        );
    }

    #[tokio::test]
    async fn duplicate_connect_request_reports_already_active() {
        let discovery = Arc::new(NostrDiscovery::new_for_test());
        let peer_npub = nostr::Keys::generate()
            .public_key()
            .to_bech32()
            .expect("peer npub");
        let peer_config = PeerConfig::new(peer_npub, "udp", "nat");

        assert!(
            discovery
                .request_connect_with_mesh_signaling(peer_config.clone(), true)
                .await,
            "first request should spawn an initiator"
        );
        assert!(
            !discovery
                .request_connect_with_mesh_signaling(peer_config, true)
                .await,
            "second request for the same peer should be deduped"
        );
        assert_eq!(discovery.active_initiator_count_for_test().await, 1);
    }

    #[tokio::test]
    async fn mesh_signal_channel_roundtrips_offer() {
        let discovery = NostrDiscovery::new_for_test();
        let offer = TraversalOffer {
            message_type: "offer".to_string(),
            session_id: "session".to_string(),
            issued_at: 1,
            expires_at: 2,
            nonce: "nonce".to_string(),
            sender_npub: discovery.npub.clone(),
            recipient_npub: "npub1peer".to_string(),
            reflexive_address: None,
            local_addresses: Vec::new(),
            stun_server: None,
        };

        assert!(
            discovery
                .emit_mesh_signal(MeshTraversalSignal::Offer {
                    peer_npub: "npub1peer".to_string(),
                    offer: offer.clone(),
                })
                .await
        );

        let signals = discovery.drain_mesh_signals().await;
        assert_eq!(signals.len(), 1);
        match &signals[0] {
            MeshTraversalSignal::Offer {
                peer_npub,
                offer: got,
            } => {
                assert_eq!(peer_npub, "npub1peer");
                assert_eq!(got, &offer);
            }
            MeshTraversalSignal::Answer { .. } => panic!("expected mesh offer"),
        }
    }

    #[tokio::test]
    async fn mesh_answer_resolves_pending_offer_without_nostr_event() {
        let discovery = NostrDiscovery::new_for_test();
        let (tx, rx) = oneshot::channel();
        discovery
            .pending_answers
            .lock()
            .await
            .insert("offer-nonce".to_string(), tx);
        let answer = TraversalAnswer {
            message_type: "answer".to_string(),
            session_id: "session".to_string(),
            issued_at: 1,
            expires_at: 2,
            nonce: "answer-nonce".to_string(),
            sender_npub: "npub1peer".to_string(),
            recipient_npub: discovery.npub.clone(),
            in_reply_to: "offer-nonce".to_string(),
            accepted: true,
            reflexive_address: None,
            local_addresses: vec![TraversalAddress {
                protocol: "udp".to_string(),
                ip: "127.0.0.1".to_string(),
                port: 51820,
            }],
            stun_server: None,
            punch: None,
            reason: None,
            offer_received_at: None,
        };

        discovery
            .receive_mesh_traversal_answer(answer.clone(), "npub1peer".to_string())
            .await;

        let envelope = rx.await.expect("pending answer should resolve");
        assert_eq!(envelope.payload, answer);
        assert!(envelope.event_id.is_none());
        assert_eq!(envelope.sender_npub, "npub1peer");
    }
}
