use super::{DiscoveryError, DiscoveryEvent, DiscoveryStrategy, PeerInfo, Result};
use async_trait::async_trait;
use mdns_sd::{IfKind, ResolvedService, ServiceDaemon, ServiceEvent, ServiceInfo};
use std::collections::HashMap;
use std::net::SocketAddr;
use std::sync::Arc;
use tokio::sync::{mpsc, RwLock};
use tracing::{debug, error, info, warn};

/// mDNS-based peer discovery for local network discovery
pub struct MdnsDiscovery {
    daemon: ServiceDaemon,
    service_type: String,
    discovered: Arc<RwLock<HashMap<String, PeerInfo>>>,
    events_tx: mpsc::Sender<DiscoveryEvent>,
    events_rx: Option<mpsc::Receiver<DiscoveryEvent>>,
    running: Arc<RwLock<bool>>,
}

impl MdnsDiscovery {
    /// Create a new mDNS discovery instance
    pub fn new() -> Result<Self> {
        Self::with_service_type("_peat._udp.local.")
    }

    /// Create a new mDNS discovery instance with a custom service type
    pub fn with_service_type(service_type: &str) -> Result<Self> {
        let daemon = ServiceDaemon::new().map_err(|e| DiscoveryError::MdnsError(e.to_string()))?;

        let (events_tx, events_rx) = mpsc::channel(100);

        Ok(Self {
            daemon,
            service_type: service_type.to_string(),
            discovered: Arc::new(RwLock::new(HashMap::new())),
            events_tx,
            events_rx: Some(events_rx),
            running: Arc::new(RwLock::new(false)),
        })
    }

    /// Advertise this node on the local network
    ///
    /// Uses `mdns_sd::ServiceInfo::enable_addr_auto` to publish every
    /// non-loopback local interface IP. Right for `--bind 0.0.0.0:PORT` /
    /// wildcard binds where the caller wants every reachable interface
    /// advertised. **Wrong** for `--bind 127.0.0.1:PORT` and other explicit
    /// non-default binds — `enable_addr_auto` skips loopback by design, so
    /// the published record won't match where the QUIC listener actually
    /// lives. For explicit binds, use [`Self::advertise_with_addr`].
    pub fn advertise(
        &self,
        node_id: &str,
        port: u16,
        metadata: Option<HashMap<String, String>>,
    ) -> Result<()> {
        let (instance_name, hostname, properties) = Self::build_advertise_parts(node_id, metadata);

        let service = ServiceInfo::new(
            &self.service_type,
            &instance_name,
            &hostname,
            "",
            port,
            Some(properties),
        )
        .map_err(|e| DiscoveryError::MdnsError(e.to_string()))?
        .enable_addr_auto();

        self.daemon
            .register(service)
            .map_err(|e| DiscoveryError::MdnsError(e.to_string()))?;

        info!("Advertising node {} on port {} via mDNS", node_id, port);

        Ok(())
    }

    /// Advertise this node on the local network at an exact `SocketAddr`.
    ///
    /// Unlike [`Self::advertise`], this skips `enable_addr_auto` and publishes
    /// only the IP+port passed in. Use this when the QUIC/TCP listener is
    /// bound to a specific address (loopback, a single explicit interface,
    /// etc.) and the caller doesn't want every interface auto-detected.
    ///
    /// Closes peat#897: `iroh::Endpoint::addr().addrs` does not include the
    /// caller's loopback bind, and `enable_addr_auto` skips loopback — so
    /// without this path, `--bind 127.0.0.1:PORT --mdns` advertises an IP
    /// that the listener isn't bound to and remote dials never land.
    ///
    /// # Errors
    ///
    /// Rejects unspecified addresses (`0.0.0.0` / `::`) at the boundary —
    /// publishing them produces an unroutable record. For wildcard binds,
    /// use [`Self::advertise`] instead.
    pub fn advertise_with_addr(
        &self,
        node_id: &str,
        addr: SocketAddr,
        metadata: Option<HashMap<String, String>>,
    ) -> Result<()> {
        if addr.ip().is_unspecified() {
            return Err(DiscoveryError::MdnsError(format!(
                "advertise_with_addr requires a concrete IP (got unspecified {}); \
                 use advertise() with enable_addr_auto for wildcard binds",
                addr.ip()
            )));
        }

        let (instance_name, hostname, properties) = Self::build_advertise_parts(node_id, metadata);
        let ip_str = addr.ip().to_string();

        let service = ServiceInfo::new(
            &self.service_type,
            &instance_name,
            &hostname,
            ip_str.as_str(),
            addr.port(),
            Some(properties),
        )
        .map_err(|e| DiscoveryError::MdnsError(e.to_string()))?;

        self.daemon
            .register(service)
            .map_err(|e| DiscoveryError::MdnsError(e.to_string()))?;

        info!(
            "Advertising node {} at {} via mDNS (explicit addr)",
            node_id, addr
        );

        Ok(())
    }

    /// Maximum size of the `node_id` slice we can embed in a DNS-SD
    /// instance label. DNS labels are limited to **63 octets** per
    /// RFC 1035 §2.3.4; our instance label is `peat-{node_id}`, so the
    /// node_id portion has at most `63 - "peat-".len() = 58` octets to
    /// work with. A 64-char hex `EndpointId` (the common case) exceeds
    /// this and causes the daemon to silently reject the registration.
    const MAX_NODE_ID_IN_LABEL: usize = 63 - "peat-".len();

    /// Build the DNS-SD instance label for a given `node_id`, truncating
    /// to at most [`Self::MAX_NODE_ID_IN_LABEL`] **octets** (not
    /// codepoints — RFC 1035 measures labels in octets).
    ///
    /// # Identity model
    ///
    /// After truncation, two node_ids that share their first
    /// 58 bytes will publish the **same** instance label and hostname.
    /// For 256-bit Iroh keys hex-encoded to 64 ASCII chars, a 58-byte
    /// prefix retains 232 bits of entropy — random collision is
    /// astronomically improbable. But the contract has shifted from
    /// "full id uniqueness on the wire" to "58-octet prefix uniqueness
    /// on the wire". The TXT-record `node_id` property remains the
    /// **authoritative** identifier; downstream dedup, routing, and
    /// connection management must key off the TXT value, not the DNS
    /// label. [`handle_removed`] looks up `discovered` entries by
    /// truncated instance label to bridge the two views.
    ///
    /// # Truncation policy
    ///
    /// The byte-cap is applied via `char_indices` so a multibyte UTF-8
    /// boundary is never split. Non-ASCII `node_id`s remain valid
    /// `String`s after truncation, though most peat consumers pass
    /// ASCII hex.
    fn instance_name(node_id: &str) -> String {
        // `str::len()` is already the octet count — measuring in bytes
        // is exactly what the RFC 1035 §2.3.4 limit cares about.
        let end = if node_id.len() <= Self::MAX_NODE_ID_IN_LABEL {
            node_id.len()
        } else {
            // Find the largest `i ≤ MAX_NODE_ID_IN_LABEL` such that `i`
            // is on a UTF-8 char boundary. `str::is_char_boundary` is
            // O(1) so this loop is at most `MAX_NODE_ID_IN_LABEL` cheap
            // checks.
            let mut i = Self::MAX_NODE_ID_IN_LABEL;
            while i > 0 && !node_id.is_char_boundary(i) {
                i -= 1;
            }
            i
        };
        format!("peat-{}", &node_id[..end])
    }

    /// Shared instance-name / hostname / properties construction for the
    /// two advertise paths. Pure; no I/O.
    fn build_advertise_parts(
        node_id: &str,
        metadata: Option<HashMap<String, String>>,
    ) -> (String, String, HashMap<String, String>) {
        let instance_name = Self::instance_name(node_id);

        let mut properties = HashMap::new();
        properties.insert("node_id".to_string(), node_id.to_string());

        if let Some(meta) = metadata {
            for (k, v) in meta {
                properties.insert(k, v);
            }
        }

        // Same 63-octet DNS-label cap applies to the hostname label.
        let host_label = node_id
            .chars()
            .take(Self::MAX_NODE_ID_IN_LABEL)
            .map(|c| {
                if c.is_ascii_alphanumeric() || c == '-' {
                    c
                } else {
                    '-'
                }
            })
            .collect::<String>();
        let hostname = format!("{}.local.", host_label);

        (instance_name, hostname, properties)
    }

    /// Unadvertise this node from the local network
    pub fn unadvertise(&self, node_id: &str) -> Result<()> {
        let instance_name = Self::instance_name(node_id);
        let fullname = format!("{}.{}", instance_name, self.service_type);

        self.daemon
            .unregister(&fullname)
            .map_err(|e| DiscoveryError::MdnsError(e.to_string()))?;

        info!("Unadvertised node {} from mDNS", node_id);

        Ok(())
    }

    /// Parse a `ResolvedService` (mdns_sd 0.19+) into a `PeerInfo`.
    ///
    /// mdns_sd's 0.12+ rewrite split `ServiceInfo` (publish side) from
    /// `ResolvedService` (resolve side). The resolved type's `addresses`
    /// is `HashSet<ScopedIp>` instead of `HashSet<IpAddr>`; we use
    /// `ScopedIp::to_ip_addr()` to recover the plain `IpAddr` peat-mesh
    /// `PeerInfo` expects.
    fn parse_service_info(resolved: &ResolvedService) -> Option<PeerInfo> {
        // Extract node_id from TXT records
        let properties = &resolved.txt_properties;
        let node_id = properties.get("node_id").map(|v| v.val_str().to_string())?;

        // Get all addresses for this service
        let addresses: Vec<SocketAddr> = resolved
            .addresses
            .iter()
            .map(|scoped| SocketAddr::new(scoped.to_ip_addr(), resolved.port))
            .collect();

        if addresses.is_empty() {
            warn!("Service {} has no addresses", node_id);
            return None;
        }

        let mut metadata = HashMap::new();
        for prop in properties.iter() {
            let key = prop.key();
            if key != "node_id" {
                metadata.insert(key.to_string(), prop.val_str().to_string());
            }
        }

        debug!(
            "Parsed service info: node_id={}, addresses={:?}",
            node_id, addresses
        );

        Some(PeerInfo {
            node_id,
            addresses,
            relay_url: None,
            last_seen: std::time::Instant::now(),
            metadata,
        })
    }

    /// Handle a resolved mDNS service event.
    async fn handle_resolved(
        discovered: &RwLock<HashMap<String, PeerInfo>>,
        events_tx: &mpsc::Sender<DiscoveryEvent>,
        resolved: &ResolvedService,
    ) {
        debug!("mDNS service resolved: {:?}", resolved.get_fullname());

        if let Some(peer_info) = Self::parse_service_info(resolved) {
            let node_id = peer_info.node_id.clone();

            // Update discovered peers
            let mut peers = discovered.write().await;
            let is_new = !peers.contains_key(&node_id);
            peers.insert(node_id.clone(), peer_info.clone());
            drop(peers);

            // Send event
            let event = if is_new {
                info!("Discovered new peer: {}", node_id);
                DiscoveryEvent::PeerFound(peer_info)
            } else {
                debug!("Updated peer: {}", node_id);
                DiscoveryEvent::PeerUpdated(peer_info)
            };

            if let Err(e) = events_tx.send(event).await {
                error!("Failed to send discovery event: {}", e);
            }
        }
    }

    /// Handle a removed mDNS service event.
    ///
    /// The DNS-SD fullname carries `peat-{label}` where `label` is the
    /// possibly-truncated instance form of the node id (see
    /// [`Self::instance_name`]). The `discovered` map, however, is keyed
    /// by the **full** node id pulled from the TXT record at resolve
    /// time. A direct `peers.remove(&label)` would miss every entry
    /// whose full id was longer than 58 chars — i.e., every Iroh
    /// `EndpointId` consumer. Look up by matching the stored full id's
    /// truncated instance form against the label from the fullname.
    async fn handle_removed(
        discovered: &RwLock<HashMap<String, PeerInfo>>,
        events_tx: &mpsc::Sender<DiscoveryEvent>,
        fullname: &str,
    ) {
        debug!("mDNS service removed: {}", fullname);

        let Some(removed_label) = Self::extract_node_id(fullname) else {
            return;
        };

        // Find the full id whose truncated form matches the label. n is
        // bounded by the number of peers on the local segment — small in
        // practice — so a linear scan is fine and avoids a parallel
        // truncated→full index that could drift from `discovered`.
        let mut peers = discovered.write().await;
        let matching_full_id = peers
            .keys()
            .find(|full| {
                Self::instance_name(full)
                    .strip_prefix("peat-")
                    .map(|t| t == removed_label)
                    .unwrap_or(false)
            })
            .cloned();
        let Some(full_id) = matching_full_id else {
            // Either we never saw a resolve for this label (race: remove
            // before resolve), or the publisher used a different prefix
            // scheme. Nothing to clean up; nothing to emit.
            return;
        };
        peers.remove(&full_id);
        drop(peers);

        info!("Lost peer: {}", full_id);

        if let Err(e) = events_tx.send(DiscoveryEvent::PeerLost(full_id)).await {
            error!("Failed to send discovery event: {}", e);
        }
    }

    /// Extract the instance label from a service fullname.
    ///
    /// Returns the `peat-` stripped label (which is the truncated form
    /// of the node id for long ids — see [`Self::instance_name`]). The
    /// label is **not** the full node_id; callers that need the full id
    /// must consult `discovered` (keyed by the TXT-record `node_id`).
    fn extract_node_id(fullname: &str) -> Option<String> {
        // Format: "peat-{label}._peat._udp.local."
        let parts: Vec<&str> = fullname.split('.').collect();
        if parts.is_empty() {
            return None;
        }

        let instance = parts[0];
        if !instance.starts_with("peat-") {
            return None;
        }

        Some(instance.strip_prefix("peat-")?.to_string())
    }
}

impl Default for MdnsDiscovery {
    fn default() -> Self {
        Self::new().expect("Failed to create default MdnsDiscovery")
    }
}

#[async_trait]
impl DiscoveryStrategy for MdnsDiscovery {
    async fn start(&mut self) -> Result<()> {
        let mut running = self.running.write().await;
        if *running {
            warn!("mDNS discovery already running");
            return Ok(());
        }

        *running = true;
        drop(running);

        // Enable loopback multicast on the ADVERTISE daemon too (peat#897), so
        // same-host instances advertising on 127.0.0.1 are reachable; the browse
        // daemon enables it separately inside the supervisor below.
        if let Err(e) = self.daemon.enable_interface(IfKind::LoopbackV4) {
            warn!("mDNS: failed to enable LoopbackV4 multicast (advertise): {e}");
        }
        if let Err(e) = self.daemon.enable_interface(IfKind::LoopbackV6) {
            warn!("mDNS: failed to enable LoopbackV6 multicast (advertise): {e}");
        }

        let discovered = self.discovered.clone();
        let events_tx = self.events_tx.clone();
        let running_clone = self.running.clone();
        let service_type = self.service_type.clone();

        info!("Starting mDNS discovery for {}", service_type);

        // Spawn a *supervised* browse task. The browse runs on its OWN
        // `ServiceDaemon` (separate from the advertise daemon on `self.daemon`)
        // so it can be torn down and recreated independently.
        //
        // Why: `mdns-sd` 0.19's wire encoder `assert!`s that every DNS label is
        // < 64 bytes and PANICS (rather than erroring) when it isn't. A single
        // malformed/oversized record on the segment kills the daemon thread,
        // which closes the browse channel — and the old code `break`'d out of
        // the loop, stopping discovery *permanently* (observed on Android: one
        // panic → peer count stuck at 0 until the node is recreated). The fix
        // is to treat a dead browse daemon as transient: drop it, recreate a
        // fresh `ServiceDaemon`, re-`browse`, and resume — with capped backoff
        // so a persistently-bad segment doesn't hot-loop. The advertise daemon
        // is independent and unaffected.
        tokio::spawn(async move {
            let mut backoff = std::time::Duration::from_millis(500);
            let max_backoff = std::time::Duration::from_secs(10);

            'supervise: while *running_clone.read().await {
                // (Re)create a dedicated browse daemon.
                let daemon = match ServiceDaemon::new() {
                    Ok(d) => d,
                    Err(e) => {
                        error!(
                            "mDNS: failed to create browse daemon: {e}; retrying in {backoff:?}"
                        );
                        tokio::time::sleep(backoff).await;
                        backoff = (backoff * 2).min(max_backoff);
                        continue 'supervise;
                    }
                };
                // Opt the daemon into loopback interfaces (defenseunicorns/peat#897)
                // so same-host instances advertising on 127.0.0.1 can find each
                // other; errors are non-fatal (hosts without loopback are rare).
                if let Err(e) = daemon.enable_interface(IfKind::LoopbackV4) {
                    warn!("mDNS: failed to enable LoopbackV4 multicast: {e}");
                }
                if let Err(e) = daemon.enable_interface(IfKind::LoopbackV6) {
                    warn!("mDNS: failed to enable LoopbackV6 multicast: {e}");
                }

                let receiver = match daemon.browse(&service_type) {
                    Ok(r) => r,
                    Err(e) => {
                        error!("mDNS: browse({service_type}) failed: {e}; retrying in {backoff:?}");
                        tokio::time::sleep(backoff).await;
                        backoff = (backoff * 2).min(max_backoff);
                        continue 'supervise;
                    }
                };
                // Healthy (re)start — reset the backoff.
                backoff = std::time::Duration::from_millis(500);

                while *running_clone.read().await {
                    match receiver.recv_async().await {
                        Ok(event) => match event {
                            ServiceEvent::ServiceResolved(info) => {
                                Self::handle_resolved(&discovered, &events_tx, &info).await;
                            }
                            ServiceEvent::ServiceRemoved(_, fullname) => {
                                Self::handle_removed(&discovered, &events_tx, &fullname).await;
                            }
                            ServiceEvent::SearchStarted(_) => {
                                debug!("mDNS search started");
                            }
                            ServiceEvent::SearchStopped(_) => {
                                debug!("mDNS search stopped");
                            }
                            _ => {}
                        },
                        Err(e) => {
                            // Browse daemon thread died (e.g. mdns-sd encoder
                            // panic on an oversized label). Treat as transient:
                            // drop this daemon and respawn a fresh one rather
                            // than stopping discovery forever.
                            warn!("mDNS browse daemon ended ({e}); respawning discovery");
                            break;
                        }
                    }
                }
                // `daemon` drops here, tearing down the dead browse daemon before
                // the supervisor loop recreates one.
            }

            info!("mDNS discovery task stopped");
        });

        Ok(())
    }

    async fn advertise(&self, node_id: &str, port: u16) -> Result<()> {
        self.advertise(node_id, port, None)
    }

    async fn stop(&mut self) -> Result<()> {
        let mut running = self.running.write().await;
        if !*running {
            return Ok(());
        }

        info!("Stopping mDNS discovery");
        *running = false;

        // Give the background task time to exit
        tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;

        Ok(())
    }

    async fn discovered_peers(&self) -> Vec<PeerInfo> {
        self.discovered.read().await.values().cloned().collect()
    }

    fn event_stream(&mut self) -> Result<mpsc::Receiver<DiscoveryEvent>> {
        self.events_rx
            .take()
            .ok_or(DiscoveryError::EventStreamConsumed)
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_extract_node_id() {
        let fullname = "peat-platform-1._peat._udp.local.";
        let node_id = MdnsDiscovery::extract_node_id(fullname);
        assert_eq!(node_id, Some("platform-1".to_string()));

        let invalid = "invalid._peat._udp.local.";
        let node_id = MdnsDiscovery::extract_node_id(invalid);
        assert_eq!(node_id, None);
    }

    #[test]
    fn test_extract_node_id_empty() {
        assert_eq!(MdnsDiscovery::extract_node_id(""), None);
    }

    #[test]
    fn test_extract_node_id_just_prefix() {
        // "peat-" prefix with empty node id
        let result = MdnsDiscovery::extract_node_id("peat-._peat._udp.local.");
        assert_eq!(result, Some("".to_string()));
    }

    #[test]
    fn test_extract_node_id_complex() {
        let fullname = "peat-cell-bravo-3._peat._udp.local.";
        let node_id = MdnsDiscovery::extract_node_id(fullname);
        assert_eq!(node_id, Some("cell-bravo-3".to_string()));
    }

    #[tokio::test]
    async fn test_mdns_discovery_lifecycle() {
        let mut discovery = MdnsDiscovery::new().unwrap();

        // Start discovery
        discovery.start().await.unwrap();

        // Check that it's running
        assert!(*discovery.running.read().await);

        // Stop discovery
        discovery.stop().await.unwrap();

        // Check that it's stopped
        assert!(!*discovery.running.read().await);
    }

    #[tokio::test]
    async fn test_mdns_discovery_start_twice() {
        let mut discovery = MdnsDiscovery::new().unwrap();
        discovery.start().await.unwrap();
        assert!(*discovery.running.read().await);

        // Starting again should be idempotent
        discovery.start().await.unwrap();
        assert!(*discovery.running.read().await);

        discovery.stop().await.unwrap();
    }

    #[tokio::test]
    async fn test_mdns_discovery_stop_when_not_running() {
        let mut discovery = MdnsDiscovery::new().unwrap();
        // Stop without start should be fine
        discovery.stop().await.unwrap();
        assert!(!*discovery.running.read().await);
    }

    #[tokio::test]
    async fn test_mdns_discovered_peers_initially_empty() {
        let discovery = MdnsDiscovery::new().unwrap();
        let peers = discovery.discovered_peers().await;
        assert!(peers.is_empty());
    }

    #[tokio::test]
    async fn test_mdns_event_stream_consumed() {
        let mut discovery = MdnsDiscovery::new().unwrap();
        let _stream = discovery.event_stream().unwrap();

        // Second call should fail
        let result = discovery.event_stream();
        assert!(result.is_err());
        assert!(matches!(
            result.unwrap_err(),
            DiscoveryError::EventStreamConsumed
        ));
    }

    #[test]
    fn test_mdns_with_custom_service_type() {
        let discovery = MdnsDiscovery::with_service_type("_custom._tcp.local.");
        assert!(discovery.is_ok());
        assert_eq!(discovery.unwrap().service_type, "_custom._tcp.local.");
    }

    #[test]
    fn test_parse_service_info_with_address() {
        let mut props = HashMap::new();
        props.insert("node_id".to_string(), "test-node".to_string());
        props.insert("role".to_string(), "leader".to_string());

        let info = ServiceInfo::new(
            "_peat._udp.local.",
            "peat-test-node",
            "localhost",
            "192.168.1.10",
            5000,
            Some(props),
        )
        .unwrap();

        let peer = MdnsDiscovery::parse_service_info(&info.clone().as_resolved_service()).unwrap();
        assert_eq!(peer.node_id, "test-node");
        assert!(!peer.addresses.is_empty());
        assert_eq!(peer.addresses[0].port(), 5000);
        assert!(peer.relay_url.is_none());
        // QA on peat-mesh#149: tighten the receive-side contract — the
        // resolved IP must exactly match the IP that was published.
        // `ScopedIp::to_ip_addr()` collapses scope info; this checks the
        // round-trip through that lossy-by-design accessor still
        // preserves the host bits. The publish→resolve silent rewrite
        // was the exact failure shape peat#897 was about.
        let expected_ip: std::net::IpAddr = "192.168.1.10".parse().unwrap();
        assert!(
            peer.addresses.iter().any(|s| s.ip() == expected_ip),
            "expected published IP {expected_ip} preserved through ScopedIp::to_ip_addr round-trip, got {:?}",
            peer.addresses
        );
        // "role" should be in metadata but "node_id" should not
        assert_eq!(peer.metadata.get("role").unwrap(), "leader");
        assert!(!peer.metadata.contains_key("node_id"));
    }

    #[test]
    fn test_parse_service_info_no_node_id() {
        // No "node_id" property → should return None
        let info = ServiceInfo::new(
            "_peat._udp.local.",
            "peat-anon",
            "localhost",
            "10.0.0.1",
            4000,
            None::<HashMap<String, String>>,
        )
        .unwrap();

        let result = MdnsDiscovery::parse_service_info(&info.clone().as_resolved_service());
        assert!(result.is_none());
    }

    #[test]
    fn test_parse_service_info_no_addresses() {
        let mut props = HashMap::new();
        props.insert("node_id".to_string(), "ghost".to_string());

        // Empty string for IP → no addresses
        let info = ServiceInfo::new(
            "_peat._udp.local.",
            "peat-ghost",
            "localhost",
            "",
            3000,
            Some(props),
        )
        .unwrap();

        let result = MdnsDiscovery::parse_service_info(&info.clone().as_resolved_service());
        assert!(result.is_none());
    }

    #[test]
    fn test_parse_service_info_multiple_metadata() {
        let mut props = HashMap::new();
        props.insert("node_id".to_string(), "multi".to_string());
        props.insert("version".to_string(), "2.0".to_string());
        props.insert("region".to_string(), "us-east".to_string());
        props.insert("priority".to_string(), "high".to_string());

        let info = ServiceInfo::new(
            "_peat._udp.local.",
            "peat-multi",
            "localhost",
            "10.0.0.5",
            6000,
            Some(props),
        )
        .unwrap();

        let peer = MdnsDiscovery::parse_service_info(&info.clone().as_resolved_service()).unwrap();
        assert_eq!(peer.node_id, "multi");
        assert_eq!(peer.metadata.len(), 3); // version, region, priority (not node_id)
        assert_eq!(peer.metadata["version"], "2.0");
        assert_eq!(peer.metadata["region"], "us-east");
        assert_eq!(peer.metadata["priority"], "high");
        // peat-mesh#149 QA: receive-side IP-value pin (see
        // test_parse_service_info_with_address for the rationale).
        let expected_ip: std::net::IpAddr = "10.0.0.5".parse().unwrap();
        assert!(
            peer.addresses.iter().any(|s| s.ip() == expected_ip),
            "expected published IP {expected_ip}, got {:?}",
            peer.addresses
        );
        assert_eq!(peer.addresses[0].port(), 6000);
    }

    #[test]
    fn test_advertise_error_maps_correctly() {
        // advertise with empty hostname (which ServiceInfo::new passes through)
        // exercises the error-mapping path: ServiceInfo::new → MdnsError
        let discovery = MdnsDiscovery::new().unwrap();
        // The current code passes "" as hostname to ServiceInfo::new, which
        // may fail depending on the mdns-sd version. Either outcome exercises code.
        let result = discovery.advertise("node-42", 8080, None);
        // We don't assert success — just that it doesn't panic.
        let _ = result;
    }

    #[test]
    fn test_advertise_with_metadata_path() {
        let discovery = MdnsDiscovery::new().unwrap();
        let mut meta = HashMap::new();
        meta.insert("role".to_string(), "leader".to_string());
        meta.insert("version".to_string(), "1.0".to_string());
        // Exercise the metadata-merging path in advertise()
        let result = discovery.advertise("node-meta", 9090, Some(meta));
        let _ = result;
    }

    // --- advertise_with_addr regression tests (peat#897) ---
    //
    // The boundary-check tests below pin the API contract that
    // unspecified IPs (0.0.0.0 / ::) are rejected, while loopback and
    // concrete IPs are accepted. They do not exercise multicast — the
    // mutual-discovery test that does is in the integration test file
    // `tests/mdns_advertise_with_addr.rs`, guarded by `#[ignore]` because
    // CI multicast is unreliable.

    #[test]
    fn test_advertise_with_addr_rejects_unspecified_v4() {
        let discovery = MdnsDiscovery::new().unwrap();
        let result = discovery.advertise_with_addr("n1", "0.0.0.0:9000".parse().unwrap(), None);
        match result {
            Err(DiscoveryError::MdnsError(msg)) => {
                assert!(
                    msg.contains("unspecified") && msg.contains("advertise()"),
                    "error must explain the unspecified-IP rejection and point at advertise(): {msg}"
                );
            }
            other => panic!("expected MdnsError for unspecified bind, got: {other:?}"),
        }
    }

    #[test]
    fn test_advertise_with_addr_rejects_unspecified_v6() {
        let discovery = MdnsDiscovery::new().unwrap();
        let result = discovery.advertise_with_addr("n2", "[::]:9000".parse().unwrap(), None);
        assert!(
            matches!(result, Err(DiscoveryError::MdnsError(_))),
            "unspecified v6 must be rejected; got {result:?}"
        );
    }

    /// Build a `ServiceInfo` using the same construction as
    /// `advertise_with_addr`, so the unit tests can inspect the resulting
    /// record's addresses without standing up a daemon. This pins the
    /// publish-side contract — exactly which IP ends up in the record —
    /// at unit tier, addressing the QA suggestion on peat-mesh#147 that
    /// `is_ok()`-only checks would not catch a silent loopback→LAN
    /// rewrite (the very failure mode peat#897 was about).
    fn build_explicit_addr_service_info(
        node_id: &str,
        addr: std::net::SocketAddr,
    ) -> mdns_sd::ServiceInfo {
        let (instance_name, hostname, properties) =
            MdnsDiscovery::build_advertise_parts(node_id, None);
        let ip_str = addr.ip().to_string();
        mdns_sd::ServiceInfo::new(
            "_peat._udp.local.",
            &instance_name,
            &hostname,
            ip_str.as_str(),
            addr.port(),
            Some(properties),
        )
        .expect("ServiceInfo::new must accept explicit addr")
    }

    #[test]
    fn test_advertise_with_addr_accepts_loopback_v4() {
        let discovery = MdnsDiscovery::new().unwrap();
        // This is the case peat#897 was about: --bind 127.0.0.1:PORT must
        // be advertisable as-is, not silently rewritten to a LAN IP.
        let addr: std::net::SocketAddr = "127.0.0.1:39001".parse().unwrap();
        let result = discovery.advertise_with_addr("n-loop", addr, None);
        assert!(
            result.is_ok(),
            "loopback bind must be acceptable: {result:?}"
        );

        // Inspect the published record (built with the same arguments
        // advertise_with_addr feeds to mdns_sd): the IP must be exactly
        // 127.0.0.1 and the port exactly 39001. Per the QA suggestion
        // on peat-mesh#147, an `is_ok()`-only check would not catch a
        // silent rewrite — which is precisely peat#897's failure mode.
        let info = build_explicit_addr_service_info("n-loop", addr);
        assert_eq!(info.get_port(), 39001);
        let addrs: Vec<std::net::IpAddr> = info.get_addresses().iter().copied().collect();
        assert!(
            addrs.contains(&"127.0.0.1".parse().unwrap()),
            "advertised record must contain 127.0.0.1; got {addrs:?}"
        );
        assert!(
            !info.is_addr_auto(),
            "explicit-addr path must not enable_addr_auto"
        );
    }

    #[test]
    fn test_advertise_with_addr_accepts_concrete_v4() {
        let discovery = MdnsDiscovery::new().unwrap();
        let addr: std::net::SocketAddr = "192.168.1.10:39001".parse().unwrap();
        let result = discovery.advertise_with_addr("n-lan", addr, None);
        assert!(
            result.is_ok(),
            "concrete LAN IP must be acceptable: {result:?}"
        );

        let info = build_explicit_addr_service_info("n-lan", addr);
        let addrs: Vec<std::net::IpAddr> = info.get_addresses().iter().copied().collect();
        assert!(
            addrs.contains(&"192.168.1.10".parse().unwrap()),
            "advertised record must contain 192.168.1.10; got {addrs:?}"
        );
    }

    #[test]
    fn test_advertise_with_addr_accepts_concrete_v6_loopback() {
        let discovery = MdnsDiscovery::new().unwrap();
        let addr: std::net::SocketAddr = "[::1]:39001".parse().unwrap();
        let result = discovery.advertise_with_addr("n-v6", addr, None);
        assert!(result.is_ok(), "v6 loopback must be acceptable: {result:?}");

        let info = build_explicit_addr_service_info("n-v6", addr);
        let addrs: Vec<std::net::IpAddr> = info.get_addresses().iter().copied().collect();
        assert!(
            addrs.contains(&"::1".parse().unwrap()),
            "advertised record must contain ::1; got {addrs:?}"
        );
    }

    #[test]
    fn test_build_advertise_parts_truncates_long_node_id() {
        // A 64-char hex Iroh EndpointId would yield `peat-<64chars>` =
        // 69 octets, blowing the 63-byte DNS label limit and causing
        // `mdns_sd` to silently reject the registration. The truncation
        // in `build_advertise_parts` keeps the label valid; the full
        // node_id stays in the TXT record so receivers can recover it.
        let long_id = "a".repeat(64);
        let (instance, hostname, props) = MdnsDiscovery::build_advertise_parts(&long_id, None);
        assert!(
            instance.len() <= 63,
            "instance name must be <= 63 octets (RFC 1035), got {}: {instance}",
            instance.len()
        );
        // hostname is `<label>.local.`; label portion alone must be ≤63.
        let host_label = hostname.trim_end_matches(".local.");
        assert!(
            host_label.len() <= 63,
            "hostname label must be <= 63 octets, got {}: {host_label}",
            host_label.len()
        );
        // Full node_id survives in the TXT record.
        assert_eq!(props.get("node_id"), Some(&long_id));
    }

    #[test]
    fn test_instance_name_short_node_id_unchanged() {
        // Existing short-name callers (peat-mesh-node, etc.) must keep
        // exactly the prior instance name shape, so on-the-wire identity
        // and `unadvertise` cleanup don't drift.
        assert_eq!(MdnsDiscovery::instance_name("node-A"), "peat-node-A");
        assert_eq!(MdnsDiscovery::instance_name("ad-node"), "peat-ad-node");
    }

    #[test]
    fn test_advertise_with_addr_metadata_passthrough() {
        let discovery = MdnsDiscovery::new().unwrap();
        let mut meta = HashMap::new();
        meta.insert("version".to_string(), "1".to_string());
        let result =
            discovery.advertise_with_addr("n-meta", "127.0.0.1:39001".parse().unwrap(), Some(meta));
        assert!(result.is_ok());
    }

    #[test]
    fn test_build_advertise_parts_node_id_in_properties() {
        let (instance, hostname, props) = MdnsDiscovery::build_advertise_parts("abc123", None);
        assert_eq!(instance, "peat-abc123");
        assert_eq!(hostname, "abc123.local.");
        assert_eq!(props.get("node_id"), Some(&"abc123".to_string()));
    }

    #[test]
    fn test_build_advertise_parts_sanitizes_hostname() {
        // Non-alphanumeric, non-dash chars get replaced with '-'.
        let (_, hostname, _) = MdnsDiscovery::build_advertise_parts("a.b/c d", None);
        assert_eq!(hostname, "a-b-c-d.local.");
    }

    #[test]
    fn test_build_advertise_parts_metadata_merged() {
        let mut meta = HashMap::new();
        meta.insert("version".to_string(), "2".to_string());
        meta.insert("role".to_string(), "edge".to_string());
        let (_, _, props) = MdnsDiscovery::build_advertise_parts("n1", Some(meta));
        assert_eq!(props.get("version"), Some(&"2".to_string()));
        assert_eq!(props.get("role"), Some(&"edge".to_string()));
        // node_id is always present
        assert_eq!(props.get("node_id"), Some(&"n1".to_string()));
    }

    #[test]
    fn test_unadvertise_formats_fullname() {
        let discovery = MdnsDiscovery::new().unwrap();
        // unadvertise just formats the fullname and calls daemon.unregister
        let result = discovery.unadvertise("node-42");
        // May fail if not registered, but exercises the code path
        let _ = result;
    }

    #[test]
    fn test_mdns_default() {
        let discovery = MdnsDiscovery::default();
        assert_eq!(discovery.service_type, "_peat._udp.local.");
    }

    #[test]
    fn test_extract_node_id_no_prefix_match() {
        // Various non-matching prefixes
        assert_eq!(
            MdnsDiscovery::extract_node_id("other-node._udp.local."),
            None
        );
        assert_eq!(
            MdnsDiscovery::extract_node_id("PEAT-upper._udp.local."),
            None
        );
    }

    #[tokio::test]
    async fn test_mdns_discovered_peers_after_advertise() {
        let discovery = MdnsDiscovery::new().unwrap();
        let _ = discovery.advertise("ad-node", 7777, None);
        // Peers list should still be empty since we haven't started discovery browsing
        let peers = discovery.discovered_peers().await;
        assert!(peers.is_empty());
    }

    /// Test handle_resolved with a new peer (PeerFound path).
    #[tokio::test]
    async fn test_handle_resolved_new_peer() {
        let discovered = Arc::new(RwLock::new(HashMap::new()));
        let (events_tx, mut events_rx) = mpsc::channel(10);

        let mut props = HashMap::new();
        props.insert("node_id".to_string(), "peer-A".to_string());
        let info = ServiceInfo::new(
            "_peat._udp.local.",
            "peat-peer-A",
            "host.local.",
            "10.0.0.1",
            5000,
            Some(props),
        )
        .unwrap();

        MdnsDiscovery::handle_resolved(
            &discovered,
            &events_tx,
            &info.clone().as_resolved_service(),
        )
        .await;

        // Peer should be in discovered map
        let peers = discovered.read().await;
        assert!(peers.contains_key("peer-A"));
        drop(peers);

        // Should receive PeerFound event
        let event = events_rx.try_recv().unwrap();
        assert!(matches!(event, DiscoveryEvent::PeerFound(ref p) if p.node_id == "peer-A"));
    }

    /// Test handle_resolved with an existing peer (PeerUpdated path).
    #[tokio::test]
    async fn test_handle_resolved_existing_peer() {
        let discovered = Arc::new(RwLock::new(HashMap::new()));
        let (events_tx, mut events_rx) = mpsc::channel(10);

        let mut props = HashMap::new();
        props.insert("node_id".to_string(), "peer-B".to_string());
        let info = ServiceInfo::new(
            "_peat._udp.local.",
            "peat-peer-B",
            "host.local.",
            "10.0.0.2",
            5001,
            Some(props),
        )
        .unwrap();

        // First resolve → PeerFound
        MdnsDiscovery::handle_resolved(
            &discovered,
            &events_tx,
            &info.clone().as_resolved_service(),
        )
        .await;
        let event = events_rx.try_recv().unwrap();
        assert!(matches!(event, DiscoveryEvent::PeerFound(_)));

        // Second resolve → PeerUpdated
        MdnsDiscovery::handle_resolved(
            &discovered,
            &events_tx,
            &info.clone().as_resolved_service(),
        )
        .await;
        let event = events_rx.try_recv().unwrap();
        assert!(matches!(event, DiscoveryEvent::PeerUpdated(_)));
    }

    /// Test handle_resolved with no node_id property (no event sent).
    #[tokio::test]
    async fn test_handle_resolved_no_node_id() {
        let discovered = Arc::new(RwLock::new(HashMap::new()));
        let (events_tx, mut events_rx) = mpsc::channel(10);

        let info = ServiceInfo::new(
            "_peat._udp.local.",
            "anon",
            "host.local.",
            "10.0.0.3",
            5002,
            None::<HashMap<String, String>>,
        )
        .unwrap();

        MdnsDiscovery::handle_resolved(
            &discovered,
            &events_tx,
            &info.clone().as_resolved_service(),
        )
        .await;

        // No event should be sent
        assert!(events_rx.try_recv().is_err());
    }

    /// Test handle_removed with a known peer.
    #[tokio::test]
    async fn test_handle_removed_known_peer() {
        let discovered = Arc::new(RwLock::new(HashMap::new()));
        let (events_tx, mut events_rx) = mpsc::channel(10);

        // Pre-populate discovered peers
        let peer = PeerInfo {
            node_id: "peer-C".to_string(),
            addresses: vec!["10.0.0.4:5003".parse().unwrap()],
            relay_url: None,
            last_seen: std::time::Instant::now(),
            metadata: HashMap::new(),
        };
        discovered.write().await.insert("peer-C".to_string(), peer);

        MdnsDiscovery::handle_removed(&discovered, &events_tx, "peat-peer-C._peat._udp.local.")
            .await;

        // Peer should be removed from map
        assert!(discovered.read().await.is_empty());

        // Should receive PeerLost event
        let event = events_rx.try_recv().unwrap();
        assert!(matches!(event, DiscoveryEvent::PeerLost(ref id) if id == "peer-C"));
    }

    /// Test handle_removed with an unknown peer (no event sent).
    #[tokio::test]
    async fn test_handle_removed_unknown_peer() {
        let discovered = Arc::new(RwLock::new(HashMap::new()));
        let (events_tx, mut events_rx) = mpsc::channel(10);

        MdnsDiscovery::handle_removed(&discovered, &events_tx, "peat-unknown._peat._udp.local.")
            .await;

        // No event should be sent
        assert!(events_rx.try_recv().is_err());
    }

    /// Test handle_removed with invalid fullname (no peat- prefix).
    #[tokio::test]
    async fn test_handle_removed_invalid_fullname() {
        let discovered = Arc::new(RwLock::new(HashMap::new()));
        let (events_tx, mut events_rx) = mpsc::channel(10);

        MdnsDiscovery::handle_removed(&discovered, &events_tx, "other._tcp.local.").await;

        // No event should be sent
        assert!(events_rx.try_recv().is_err());
    }

    /// Regression: a long-id peer (64-char Iroh `EndpointId`) is stored
    /// in `discovered` by its FULL id (from the TXT record), but the
    /// remove path only sees the TRUNCATED instance label in the
    /// fullname. Before the lookup-by-truncated-form fix, `PeerLost`
    /// never fired for long-id peers and stale entries accumulated.
    /// This is the BLOCKER finding on peat-mesh#147's QA review.
    #[tokio::test]
    async fn test_handle_removed_long_id_peer() {
        let discovered = Arc::new(RwLock::new(HashMap::new()));
        let (events_tx, mut events_rx) = mpsc::channel(10);

        let full_id = "a".repeat(64); // 64-char hex EndpointId
        let peer = PeerInfo {
            node_id: full_id.clone(),
            addresses: vec!["10.0.0.4:5003".parse().unwrap()],
            relay_url: None,
            last_seen: std::time::Instant::now(),
            metadata: HashMap::new(),
        };
        discovered.write().await.insert(full_id.clone(), peer);

        // The fullname carries the TRUNCATED form (58 chars after `peat-`),
        // matching what mdns_sd actually puts on the wire.
        let truncated_label = MdnsDiscovery::instance_name(&full_id);
        let fullname = format!("{truncated_label}._peat._udp.local.");

        MdnsDiscovery::handle_removed(&discovered, &events_tx, &fullname).await;

        // Entry must be gone and PeerLost must carry the FULL id, not
        // the truncated label.
        assert!(discovered.read().await.is_empty());
        let event = events_rx.try_recv().unwrap();
        assert!(
            matches!(event, DiscoveryEvent::PeerLost(ref id) if id == &full_id),
            "PeerLost must carry the full 64-char id, got {event:?}"
        );
    }

    /// Locks in the byte-count truncation invariant for `instance_name`:
    /// even with non-ASCII input, the resulting label must be ≤ 63 octets
    /// AND a valid `String` (no split multibyte sequence).
    #[test]
    fn test_instance_name_truncates_on_byte_boundary_for_non_ascii() {
        // 'ñ' is 2 bytes in UTF-8. Build a node_id that overflows the
        // 58-byte budget *and* whose 58th byte falls inside a
        // multibyte sequence.
        let id: String = "ñ".repeat(30); // 60 bytes, 30 chars
        let label = MdnsDiscovery::instance_name(&id);
        // Label is "peat-{prefix}"; full label ≤ 63 octets.
        // `str::len()` is the byte count; using it directly per clippy
        // (needless_as_bytes — `String::len` already returns octets).
        assert!(
            label.len() <= 63,
            "label must be <= 63 octets, got {} for {label:?}",
            label.len()
        );
        // The trailing `peat-` prefix is 5 bytes; the truncated id
        // portion is at most 58 bytes and must be a valid UTF-8 string.
        let id_part = label.strip_prefix("peat-").unwrap();
        assert!(id_part.len() <= 58);
        // Round-trips through str without panicking — implicit in the
        // operations above, but assert for clarity.
        let _: &str = id_part;
    }

    /// Test that the spawned task's SearchStarted/SearchStopped handling works.
    /// When we start and stop, the daemon fires these meta-events.
    #[tokio::test]
    async fn test_mdns_start_stop_search_events() {
        use std::time::Duration;

        let mut discovery = MdnsDiscovery::new().unwrap();
        discovery.start().await.unwrap();

        // Wait briefly for SearchStarted event
        tokio::time::sleep(Duration::from_millis(200)).await;

        // Stop discovery
        discovery.stop().await.unwrap();

        // Wait for background task to exit
        tokio::time::sleep(Duration::from_millis(200)).await;
    }
}
