//! Iroh-backed [`MeshTransport`] implementation.
//!
//! This module provides a [`MeshTransport`] implementation backed by
//! [`IrohTransport`](crate::network::iroh_transport::IrohTransport)'s QUIC
//! endpoint. It manages `NodeId ↔ EndpointId` mapping for discovery,
//! emits connection lifecycle events, and runs a background cleanup +
//! reconnection task for static peers.
//!
//! Per ADR-062 Phase 2 (peat#926), this type moved from
//! `peat_protocol::transport::iroh::IrohMeshTransport` to peat-mesh.
//! peat-mesh is the canonical home for backend-specific implementations
//! of the [`MeshTransport`] trait — peat-protocol's transport-agnosticism
//! rule prohibits backend-specific impls in the protocol crate.
//!
//! ## API change from peat-protocol's IrohMeshTransport
//!
//! peat-protocol's version took `Arc<RwLock<PeerConfig>>` — coupling the
//! mesh transport to peat-protocol's formation-key model. The
//! `formation` and `local` fields of `PeerConfig` were never used inside
//! the transport; only `config.peers: Vec<PeerInfo>` was consulted, for
//! address resolution and reconnection. The peat-mesh version takes
//! `Vec<PeerInfo>` directly, leaving the formation-key model where it
//! belongs (peat-protocol's security layer).
//!
//! ## Features
//!
//! - **Connection Liveness Detection (Issue #251)**: Uses QUIC
//!   `close_reason()` to detect disconnected peers.
//! - **Peer Events (Issue #252)**: Emits [`PeerEvent`] notifications on
//!   connect/disconnect.
//! - **Connection Health (Issue #254)**: Tracks connection establishment
//!   time and disconnect reasons via [`HealthMonitor`].
//! - **Automatic Reconnection (Issue #253)**: Static peers are
//!   reconnected with exponential backoff when their connections drop.

use super::capabilities::{Transport, TransportCapabilities};
use super::health::{HealthMonitor, HeartbeatConfig};
use super::reconnection::{ReconnectionManager, ReconnectionPolicy};
use super::{
    ConnectionHealth, DisconnectReason, MeshConnection, MeshTransport, NodeId, PeerEvent,
    PeerEventReceiver, PeerEventSender, Result, TransportError, PEER_EVENT_CHANNEL_CAPACITY,
};
use crate::network::iroh_transport::IrohTransport;
use crate::network::peer_info::PeerInfo;
use async_trait::async_trait;
use iroh::endpoint::Connection;
use iroh::EndpointId;
use std::collections::HashMap;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, RwLock};
use std::time::{Duration, Instant};
use tokio::sync::mpsc;
use tracing::{debug, info, warn};

/// Default interval for stale peer cleanup (Issue #244).
pub const DEFAULT_CLEANUP_INTERVAL: Duration = Duration::from_secs(5);

/// Iroh-based mesh transport implementation.
///
/// Wraps [`IrohTransport`] to provide the [`MeshTransport`] interface
/// with:
/// - `NodeId ↔ EndpointId` mapping for discovery
/// - Static peer registration (a `Vec<PeerInfo>` provided at construction)
/// - Connection lifecycle management
///
/// # Example
///
/// ```ignore
/// use peat_mesh::transport::iroh_mesh::IrohMeshTransport;
/// use peat_mesh::network::iroh_transport::IrohTransport;
/// use peat_mesh::network::peer_info::PeerInfo;
/// use std::sync::Arc;
///
/// let iroh_transport = Arc::new(IrohTransport::new().await?);
/// let static_peers: Vec<PeerInfo> = vec![/* ... */];
/// let mesh_transport = IrohMeshTransport::new(iroh_transport, static_peers);
///
/// mesh_transport.start().await?;
/// ```
pub struct IrohMeshTransport {
    transport: Arc<IrohTransport>,
    static_peers_list: Arc<RwLock<Vec<PeerInfo>>>,
    node_to_endpoint: Arc<RwLock<HashMap<NodeId, EndpointId>>>,
    endpoint_to_node: Arc<RwLock<HashMap<EndpointId, NodeId>>>,
    connections: Arc<RwLock<HashMap<NodeId, IrohMeshConnection>>>,
    event_senders: Arc<RwLock<Vec<PeerEventSender>>>,
    cleanup_running: Arc<AtomicBool>,
    cleanup_interval: Duration,
    reconnection: Arc<RwLock<ReconnectionManager>>,
    static_peers: Arc<RwLock<std::collections::HashSet<NodeId>>>,
    health_monitor: Arc<HealthMonitor>,
    capabilities: TransportCapabilities,
}

impl IrohMeshTransport {
    /// Create a new Iroh mesh transport with default settings.
    pub fn new(transport: Arc<IrohTransport>, static_peers_list: Vec<PeerInfo>) -> Self {
        Self::with_cleanup_interval(transport, static_peers_list, DEFAULT_CLEANUP_INTERVAL)
    }

    /// Create a new Iroh mesh transport with a custom cleanup interval.
    pub fn with_cleanup_interval(
        transport: Arc<IrohTransport>,
        static_peers_list: Vec<PeerInfo>,
        cleanup_interval: Duration,
    ) -> Self {
        Self::with_reconnection_policy(
            transport,
            static_peers_list,
            cleanup_interval,
            ReconnectionPolicy::default(),
        )
    }

    /// Create a new Iroh mesh transport with a custom reconnection policy.
    pub fn with_reconnection_policy(
        transport: Arc<IrohTransport>,
        static_peers_list: Vec<PeerInfo>,
        cleanup_interval: Duration,
        reconnection_policy: ReconnectionPolicy,
    ) -> Self {
        Self::with_full_config(
            transport,
            static_peers_list,
            cleanup_interval,
            reconnection_policy,
            HeartbeatConfig::default(),
        )
    }

    /// Create a new Iroh mesh transport with full configuration including
    /// health monitoring.
    pub fn with_full_config(
        transport: Arc<IrohTransport>,
        static_peers_list: Vec<PeerInfo>,
        cleanup_interval: Duration,
        reconnection_policy: ReconnectionPolicy,
        heartbeat_config: HeartbeatConfig,
    ) -> Self {
        Self {
            transport,
            static_peers_list: Arc::new(RwLock::new(static_peers_list)),
            node_to_endpoint: Arc::new(RwLock::new(HashMap::new())),
            endpoint_to_node: Arc::new(RwLock::new(HashMap::new())),
            connections: Arc::new(RwLock::new(HashMap::new())),
            event_senders: Arc::new(RwLock::new(Vec::new())),
            cleanup_running: Arc::new(AtomicBool::new(false)),
            cleanup_interval,
            reconnection: Arc::new(RwLock::new(ReconnectionManager::new(reconnection_policy))),
            static_peers: Arc::new(RwLock::new(std::collections::HashSet::new())),
            health_monitor: Arc::new(HealthMonitor::new(heartbeat_config)),
            capabilities: TransportCapabilities::quic(),
        }
    }

    /// Replace the static peers list at runtime.
    ///
    /// Every entry whose `endpoint_id()` parses successfully is registered
    /// into the `NodeId ↔ EndpointId` map and inserted into the static-peer
    /// set, so after this call:
    ///
    /// - [`Self::connect`] for any new peer succeeds (the [`PeerNotFound`]
    ///   path on a missing `NodeId → EndpointId` entry is closed).
    /// - The cleanup/reconnection task treats every entry as a static peer
    ///   eligible for reconnection.
    ///
    /// Peers with malformed `node_id` hex are skipped with a `warn!` —
    /// same surfacing as [`Self::start`].
    ///
    /// This method is **additive** with respect to the existing
    /// registration maps: entries previously registered are not removed
    /// from `node_to_endpoint` / `endpoint_to_node` / `static_peers` when
    /// they fall out of the new list. Callers that need that semantic
    /// should `disconnect()` explicitly. The original peat-protocol
    /// `Arc<RwLock<PeerConfig>>` path had the same property.
    ///
    /// [`PeerNotFound`]: TransportError::PeerNotFound
    pub fn set_static_peers(&self, peers: Vec<PeerInfo>) {
        // Pre-parse so we can release `static_peers_list`'s write lock
        // before taking the registry locks (avoids any cross-lock order
        // surprises with the cleanup task).
        let parsed: Vec<(NodeId, iroh::EndpointId)> = peers
            .iter()
            .filter_map(|p| match p.endpoint_id() {
                Ok(eid) => Some((NodeId::new(p.name.clone()), eid)),
                Err(e) => {
                    warn!(
                        peer = %p.name,
                        "Skipping static peer with invalid endpoint_id: {:#}",
                        e
                    );
                    None
                }
            })
            .collect();

        *self
            .static_peers_list
            .write()
            .unwrap_or_else(|e| e.into_inner()) = peers;

        let mut static_peers = self.static_peers.write().unwrap_or_else(|e| e.into_inner());
        for (node_id, endpoint_id) in parsed {
            self.register_peer(node_id.clone(), endpoint_id);
            static_peers.insert(node_id);
        }
    }

    /// Get the health monitor for this transport.
    pub fn health_monitor(&self) -> &Arc<HealthMonitor> {
        &self.health_monitor
    }

    fn emit_event(&self, event: PeerEvent) {
        let mut senders = self
            .event_senders
            .write()
            .unwrap_or_else(|e| e.into_inner());

        senders.retain(|sender| match sender.try_send(event.clone()) {
            Ok(()) => true,
            Err(mpsc::error::TrySendError::Full(_)) => {
                warn!(
                    "Peer event channel full, dropping event for one subscriber: {:?}",
                    event
                );
                true
            }
            Err(mpsc::error::TrySendError::Closed(_)) => {
                debug!("Peer event subscriber disconnected, removing channel");
                false
            }
        });
    }

    /// Clean up dead connections and emit disconnect events
    /// (Issue #251 + #252).
    pub fn cleanup_dead_connections(&self) {
        let mut connections = self.connections.write().unwrap_or_else(|e| e.into_inner());
        let dead_peers: Vec<_> = connections
            .iter()
            .filter(|(_, conn)| !conn.is_alive())
            .map(|(id, conn)| (id.clone(), conn.disconnect_reason(), conn.connected_at()))
            .collect();

        for (peer_id, reason, connected_at) in dead_peers {
            connections.remove(&peer_id);

            let event = PeerEvent::Disconnected {
                peer_id: peer_id.clone(),
                reason: reason.unwrap_or(DisconnectReason::Unknown),
                connection_duration: connected_at.elapsed(),
            };

            debug!("Peer {} disconnected: {:?}", peer_id, event);
            self.emit_event(event);
        }
    }

    /// Register a peer (NodeId → EndpointId mapping).
    pub fn register_peer(&self, node_id: NodeId, endpoint_id: EndpointId) {
        self.node_to_endpoint
            .write()
            .unwrap_or_else(|e| e.into_inner())
            .insert(node_id.clone(), endpoint_id);
        self.endpoint_to_node
            .write()
            .unwrap_or_else(|e| e.into_inner())
            .insert(endpoint_id, node_id);
    }

    /// Get NodeId from EndpointId (for incoming connections).
    pub fn get_node_id(&self, endpoint_id: &EndpointId) -> Option<NodeId> {
        self.endpoint_to_node
            .read()
            .unwrap_or_else(|e| e.into_inner())
            .get(endpoint_id)
            .cloned()
    }

    /// Get EndpointId from NodeId (for outgoing connections).
    pub fn get_endpoint_id(&self, node_id: &NodeId) -> Option<EndpointId> {
        self.node_to_endpoint
            .read()
            .unwrap_or_else(|e| e.into_inner())
            .get(node_id)
            .copied()
    }

    /// Get the underlying [`IrohTransport`].
    pub fn transport(&self) -> &Arc<IrohTransport> {
        &self.transport
    }

    /// Start the background cleanup and reconnection task
    /// (Issue #244 + #253).
    fn start_cleanup_task(&self) {
        if self
            .cleanup_running
            .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
            .is_err()
        {
            return;
        }

        let connections = Arc::clone(&self.connections);
        let event_senders = Arc::clone(&self.event_senders);
        let cleanup_running = Arc::clone(&self.cleanup_running);
        let interval = self.cleanup_interval;
        let reconnection = Arc::clone(&self.reconnection);
        let static_peers = Arc::clone(&self.static_peers);
        let transport = Arc::clone(&self.transport);
        let static_peers_list = Arc::clone(&self.static_peers_list);
        let health_monitor = Arc::clone(&self.health_monitor);

        tokio::spawn(async move {
            info!(
                "Started peer cleanup/reconnection task with interval {:?}",
                interval
            );

            let emit_event = |event: PeerEvent, senders: &Arc<RwLock<Vec<PeerEventSender>>>| {
                let mut senders = senders.write().unwrap_or_else(|e| e.into_inner());
                senders.retain(|sender| match sender.try_send(event.clone()) {
                    Ok(()) => true,
                    Err(mpsc::error::TrySendError::Full(_)) => true,
                    Err(mpsc::error::TrySendError::Closed(_)) => false,
                });
            };

            while cleanup_running.load(Ordering::SeqCst) {
                tokio::time::sleep(interval).await;

                if !cleanup_running.load(Ordering::SeqCst) {
                    break;
                }

                // Phase 1: Check for dead connections
                let dead_peers: Vec<_> = {
                    let conns = connections.read().unwrap_or_else(|e| e.into_inner());
                    conns
                        .iter()
                        .filter(|(_, conn)| !conn.is_alive())
                        .map(|(id, conn)| {
                            (id.clone(), conn.disconnect_reason(), conn.connected_at())
                        })
                        .collect()
                };

                if !dead_peers.is_empty() {
                    let mut conns = connections.write().unwrap_or_else(|e| e.into_inner());
                    let static_set = static_peers.read().unwrap_or_else(|e| e.into_inner());
                    let mut recon = reconnection.write().unwrap_or_else(|e| e.into_inner());

                    for (peer_id, reason, connected_at) in dead_peers {
                        conns.remove(&peer_id);

                        health_monitor.stop_monitoring(&peer_id);

                        let event = PeerEvent::Disconnected {
                            peer_id: peer_id.clone(),
                            reason: reason.unwrap_or(DisconnectReason::Unknown),
                            connection_duration: connected_at.elapsed(),
                        };

                        debug!("Cleanup: Peer {} disconnected: {:?}", peer_id, event);
                        emit_event(event, &event_senders);

                        let is_static = static_set.contains(&peer_id);
                        if is_static {
                            recon.schedule_reconnect(peer_id.clone(), true);
                            debug!("Scheduled reconnection for static peer: {}", peer_id);
                        }
                    }
                }

                // Phase 2: Check health timeouts (Issue #254)
                let newly_dead_from_health: Vec<NodeId> = health_monitor.check_timeouts();
                for peer_id in newly_dead_from_health {
                    let is_static = static_peers
                        .read()
                        .unwrap_or_else(|e| e.into_inner())
                        .contains(&peer_id);
                    if is_static {
                        let mut recon = reconnection.write().unwrap_or_else(|e| e.into_inner());
                        recon.schedule_reconnect(peer_id.clone(), true);
                        debug!(
                            "Health monitor detected dead peer, scheduling reconnection: {}",
                            peer_id
                        );
                    }
                }

                // Phase 3: Attempt due reconnections (Issue #253)
                let due_peers: Vec<NodeId> = {
                    let recon = reconnection.read().unwrap_or_else(|e| e.into_inner());
                    if !recon.is_enabled() {
                        continue;
                    }
                    recon.due_reconnections()
                };

                for peer_id in due_peers {
                    let (attempt, max_attempts) = {
                        let recon = reconnection.read().unwrap_or_else(|e| e.into_inner());
                        let state = recon.get_state(&peer_id);
                        let attempt = state.map(|s| s.attempts + 1).unwrap_or(1);
                        let max = recon.policy().max_retries;
                        (attempt, max)
                    };

                    emit_event(
                        PeerEvent::Reconnecting {
                            peer_id: peer_id.clone(),
                            attempt,
                            max_attempts,
                        },
                        &event_senders,
                    );

                    let peer_info_opt = {
                        let peers = static_peers_list.read().unwrap_or_else(|e| e.into_inner());
                        peers.iter().find(|p| p.name == peer_id.as_str()).cloned()
                    };

                    let result = if let Some(peer_info) = peer_info_opt {
                        transport.connect_peer(&peer_info).await
                    } else {
                        Err(anyhow::anyhow!(
                            "Peer not found in static list: {}",
                            peer_id
                        ))
                    };

                    match result {
                        Ok(Some(conn)) => {
                            let connected_at = Instant::now();
                            let mesh_conn =
                                IrohMeshConnection::new(peer_id.clone(), conn, connected_at);
                            connections
                                .write()
                                .unwrap_or_else(|e| e.into_inner())
                                .insert(peer_id.clone(), mesh_conn);
                            reconnection
                                .write()
                                .unwrap_or_else(|e| e.into_inner())
                                .reconnected(&peer_id);

                            health_monitor.start_monitoring(peer_id.clone());

                            info!("Reconnected to peer: {} (attempt {})", peer_id, attempt);
                            emit_event(
                                PeerEvent::Connected {
                                    peer_id: peer_id.clone(),
                                    connected_at,
                                },
                                &event_senders,
                            );
                        }
                        Ok(None) => {
                            reconnection
                                .write()
                                .unwrap_or_else(|e| e.into_inner())
                                .reconnected(&peer_id);
                            debug!("Reconnection to {} handled by accept path", peer_id);
                        }
                        Err(e) => {
                            let error_msg = e.to_string();
                            let will_retry = {
                                let mut recon =
                                    reconnection.write().unwrap_or_else(|e| e.into_inner());
                                let will_retry = recon.failed(&peer_id, error_msg.clone());
                                if !will_retry {
                                    recon.remove(&peer_id);
                                }
                                will_retry
                            };

                            warn!(
                                "Reconnection to {} failed (attempt {}): {} - will_retry={}",
                                peer_id, attempt, error_msg, will_retry
                            );

                            emit_event(
                                PeerEvent::ReconnectFailed {
                                    peer_id: peer_id.clone(),
                                    attempt,
                                    error: error_msg,
                                    will_retry,
                                },
                                &event_senders,
                            );
                        }
                    }
                }
            }

            info!("Stopped peer cleanup/reconnection task");
        });
    }

    fn stop_cleanup_task(&self) {
        self.cleanup_running.store(false, Ordering::SeqCst);
    }
}

#[async_trait]
impl MeshTransport for IrohMeshTransport {
    async fn start(&self) -> Result<()> {
        self.transport
            .start_accept_loop()
            .map_err(|e| TransportError::ConnectionFailed(e.to_string()))?;

        let peers = self
            .static_peers_list
            .read()
            .unwrap_or_else(|e| e.into_inner());
        let mut static_peers = self.static_peers.write().unwrap_or_else(|e| e.into_inner());
        for peer_info in peers.iter() {
            let node_id = NodeId::new(peer_info.name.clone());
            match peer_info.endpoint_id() {
                Ok(endpoint_id) => {
                    self.register_peer(node_id.clone(), endpoint_id);
                    static_peers.insert(node_id);
                }
                Err(e) => warn!(
                    peer = %peer_info.name,
                    "Skipping static peer with invalid endpoint_id: {:#}",
                    e
                ),
            }
        }
        drop(static_peers);
        drop(peers);

        self.start_cleanup_task();

        Ok(())
    }

    async fn stop(&self) -> Result<()> {
        self.stop_cleanup_task();

        self.transport
            .stop_accept_loop()
            .map_err(|e| TransportError::ConnectionFailed(e.to_string()))?;

        self.health_monitor.clear();

        let connections = self
            .connections
            .write()
            .unwrap_or_else(|e| e.into_inner())
            .drain()
            .collect::<Vec<_>>();
        for (_node_id, _conn) in connections {
            // Connections close when dropped
        }

        Ok(())
    }

    async fn connect(&self, peer_id: &NodeId) -> Result<Box<dyn MeshConnection>> {
        if let Some(conn) = self.get_connection(peer_id) {
            if conn.is_alive() {
                return Ok(conn);
            }
            debug!("Existing connection to {} is dead, reconnecting", peer_id);
            self.cleanup_dead_connections();
        }

        let _endpoint_id = self
            .node_to_endpoint
            .read()
            .unwrap_or_else(|e| e.into_inner())
            .get(peer_id)
            .copied()
            .ok_or_else(|| TransportError::PeerNotFound(peer_id.as_str().to_string()))?;

        let peer_info = {
            let peers = self
                .static_peers_list
                .read()
                .unwrap_or_else(|e| e.into_inner());
            peers
                .iter()
                .find(|p| p.name == peer_id.as_str())
                .cloned()
                .ok_or_else(|| TransportError::PeerNotFound(peer_id.as_str().to_string()))?
        };

        let conn_opt = self
            .transport
            .connect_peer(&peer_info)
            .await
            .map_err(|e| TransportError::ConnectionFailed(e.to_string()))?;

        match conn_opt {
            Some(conn) => {
                let connected_at = Instant::now();
                let mesh_conn = IrohMeshConnection::new(peer_id.clone(), conn, connected_at);
                self.connections
                    .write()
                    .unwrap_or_else(|e| e.into_inner())
                    .insert(peer_id.clone(), mesh_conn.clone());

                self.health_monitor.start_monitoring(peer_id.clone());

                self.emit_event(PeerEvent::Connected {
                    peer_id: peer_id.clone(),
                    connected_at,
                });

                debug!("Connected to peer: {}", peer_id);
                Ok(Box::new(mesh_conn))
            }
            None => self
                .connections
                .read()
                .unwrap_or_else(|e| e.into_inner())
                .get(peer_id)
                .cloned()
                .map(|c| Box::new(c) as Box<dyn MeshConnection>)
                .ok_or_else(|| {
                    TransportError::ConnectionFailed(
                        "Connection being handled by accept path".to_string(),
                    )
                }),
        }
    }

    async fn disconnect(&self, peer_id: &NodeId) -> Result<()> {
        if let Some(conn) = self
            .connections
            .write()
            .unwrap_or_else(|e| e.into_inner())
            .remove(peer_id)
        {
            self.health_monitor.stop_monitoring(peer_id);

            let event = PeerEvent::Disconnected {
                peer_id: peer_id.clone(),
                reason: DisconnectReason::LocalClosed,
                connection_duration: conn.connected_at().elapsed(),
            };
            debug!("Disconnected from peer: {}", peer_id);
            self.emit_event(event);
        }
        Ok(())
    }

    fn get_connection(&self, peer_id: &NodeId) -> Option<Box<dyn MeshConnection>> {
        self.connections
            .read()
            .unwrap_or_else(|e| e.into_inner())
            .get(peer_id)
            .cloned()
            .map(|c| Box::new(c) as Box<dyn MeshConnection>)
    }

    fn peer_count(&self) -> usize {
        self.connections
            .read()
            .unwrap_or_else(|e| e.into_inner())
            .len()
    }

    fn connected_peers(&self) -> Vec<NodeId> {
        self.connections
            .read()
            .unwrap_or_else(|e| e.into_inner())
            .keys()
            .cloned()
            .collect()
    }

    fn subscribe_peer_events(&self) -> PeerEventReceiver {
        let (tx, rx) = mpsc::channel(PEER_EVENT_CHANNEL_CAPACITY);
        self.event_senders
            .write()
            .unwrap_or_else(|e| e.into_inner())
            .push(tx);
        rx
    }

    fn get_peer_health(&self, peer_id: &NodeId) -> Option<ConnectionHealth> {
        self.health_monitor.get_health(peer_id)
    }
}

impl Transport for IrohMeshTransport {
    fn capabilities(&self) -> &TransportCapabilities {
        &self.capabilities
    }

    fn is_available(&self) -> bool {
        self.cleanup_running.load(Ordering::SeqCst)
    }

    fn signal_quality(&self) -> Option<u8> {
        None
    }

    fn can_reach(&self, peer_id: &NodeId) -> bool {
        self.node_to_endpoint
            .read()
            .unwrap_or_else(|e| e.into_inner())
            .contains_key(peer_id)
    }
}

/// Iroh mesh connection implementation.
///
/// Wraps an Iroh QUIC connection with the [`MeshConnection`] interface.
#[derive(Clone)]
pub struct IrohMeshConnection {
    peer_id: NodeId,
    connection: Connection,
    connected_at: Instant,
}

impl IrohMeshConnection {
    pub fn new(peer_id: NodeId, connection: Connection, connected_at: Instant) -> Self {
        Self {
            peer_id,
            connection,
            connected_at,
        }
    }

    pub fn connection(&self) -> &Connection {
        &self.connection
    }

    fn parse_close_reason(&self) -> Option<DisconnectReason> {
        self.connection.close_reason().map(map_connection_error)
    }
}

/// Map a structured iroh [`ConnectionError`](iroh::endpoint::ConnectionError)
/// to peat-mesh's transport-tier [`DisconnectReason`].
///
/// Replaces the prior substring-on-`Display` matching (peat-mesh#164) so
/// upstream wording tweaks to `iroh::endpoint::ConnectionError`'s
/// `thiserror`-derived `Display` impl no longer silently shift every
/// disconnect into the `NetworkError(_)` fallback arm. The mapping
/// preserves the load-bearing behavior of the legacy code:
///
/// - timeouts → [`DisconnectReason::IdleTimeout`]
/// - peer-initiated close / reset → [`DisconnectReason::RemoteClosed`]
/// - locally-initiated close → [`DisconnectReason::LocalClosed`]
///   (legacy code lumped this with `RemoteClosed` because the Display
///   text for both contains "closed"; structured match disambiguates)
/// - application-level close → [`DisconnectReason::ApplicationError`]
///   with the close frame's reason bytes formatted into the payload
/// - everything else (`VersionMismatch`, `TransportError`,
///   `CidsExhausted`) → [`DisconnectReason::NetworkError`] with a
///   stable per-variant tag in the payload rather than the raw
///   `Display`, so audit log lines stay parseable across iroh bumps.
///
/// Exhaustive match — `iroh::endpoint::ConnectionError` is not marked
/// `#[non_exhaustive]` upstream as of iroh 1.0.0-rc.0 / noq-proto
/// 1.0.0-rc.0, so a future variant addition fails compilation here at
/// peat-mesh CI before silently falling through to `NetworkError`.
fn map_connection_error(err: iroh::endpoint::ConnectionError) -> DisconnectReason {
    use iroh::endpoint::ConnectionError;
    match err {
        ConnectionError::TimedOut => DisconnectReason::IdleTimeout,
        ConnectionError::Reset => DisconnectReason::RemoteClosed,
        ConnectionError::ConnectionClosed(_) => DisconnectReason::RemoteClosed,
        ConnectionError::LocallyClosed => DisconnectReason::LocalClosed,
        // `close.error_code.into_inner()` produces a byte-identical
        // format vs. the pre-extraction `format!("...{}...", close.error_code)`
        // **only because** `quinn_proto::VarInt`'s `Display` impl is a
        // transparent passthrough to its inner `u64` (asserted by
        // quinn-proto 0.11 + noq-proto 1.0.0-rc.0). If a future quinn
        // bump silently changes `VarInt::Display` to add a prefix /
        // hex-encode / etc., this conversion stops being equivalent —
        // catch via the `format_application_close_payload_pin` test
        // which asserts the helper's contract directly.
        ConnectionError::ApplicationClosed(close) => DisconnectReason::ApplicationError(
            format_application_close(close.error_code.into_inner(), &close.reason),
        ),
        ConnectionError::VersionMismatch => {
            DisconnectReason::NetworkError("version-mismatch".to_string())
        }
        ConnectionError::TransportError(e) => {
            DisconnectReason::NetworkError(format!("transport-error: {}", e))
        }
        ConnectionError::CidsExhausted => {
            DisconnectReason::NetworkError("cids-exhausted".to_string())
        }
    }
}

/// Format an `ApplicationClose` frame's contents into the stable
/// `ApplicationError` payload format documented on
/// [`DisconnectReason`]: `"code={error_code} reason={utf8-lossy(reason)}"`.
///
/// Extracted so the payload contract can be pinned in isolation —
/// `iroh::endpoint::ApplicationClose` carries private / non-exhaustive
/// fields that prevent test code from constructing it directly, but
/// the format itself is the stable downstream contract operators
/// parse, so it needs a regression guard independent of variant
/// construction.
fn format_application_close(error_code: u64, reason: &[u8]) -> String {
    format!(
        "code={} reason={}",
        error_code,
        String::from_utf8_lossy(reason)
    )
}

impl MeshConnection for IrohMeshConnection {
    fn peer_id(&self) -> &NodeId {
        &self.peer_id
    }

    fn is_alive(&self) -> bool {
        self.connection.close_reason().is_none()
    }

    fn connected_at(&self) -> Instant {
        self.connected_at
    }

    fn disconnect_reason(&self) -> Option<DisconnectReason> {
        self.parse_close_reason()
    }
}

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

    #[tokio::test]
    async fn test_iroh_mesh_transport_creation() {
        let transport = Arc::new(IrohTransport::new().await.unwrap());
        let mesh_transport = IrohMeshTransport::new(transport, Vec::new());

        assert_eq!(mesh_transport.peer_count(), 0);
    }

    #[tokio::test]
    async fn test_peer_registration() {
        let transport = Arc::new(IrohTransport::new().await.unwrap());
        let mesh_transport = IrohMeshTransport::new(transport.clone(), Vec::new());

        let node_id = NodeId::new("test-node".to_string());
        let endpoint_id = transport.endpoint_id();
        mesh_transport.register_peer(node_id.clone(), endpoint_id);

        assert_eq!(mesh_transport.get_endpoint_id(&node_id), Some(endpoint_id));
        assert_eq!(mesh_transport.get_node_id(&endpoint_id), Some(node_id));
    }

    #[tokio::test]
    async fn test_start_stop_lifecycle() {
        let transport = Arc::new(IrohTransport::new().await.unwrap());
        let mesh_transport = IrohMeshTransport::new(transport.clone(), Vec::new());

        mesh_transport.start().await.unwrap();
        assert!(transport.is_accept_loop_running());

        mesh_transport.stop().await.unwrap();
        assert!(!transport.is_accept_loop_running());
    }

    #[tokio::test]
    async fn test_connect_to_unknown_peer() {
        let transport = Arc::new(IrohTransport::new().await.unwrap());
        let mesh_transport = IrohMeshTransport::new(transport, Vec::new());

        mesh_transport.start().await.unwrap();

        let unknown_peer = NodeId::new("unknown".to_string());
        let result = mesh_transport.connect(&unknown_peer).await;

        assert!(result.is_err());
        match result {
            Err(TransportError::PeerNotFound(_)) => {}
            _ => panic!("Expected PeerNotFound error"),
        }
    }

    #[tokio::test]
    async fn test_disconnect() {
        let transport = Arc::new(IrohTransport::new().await.unwrap());
        let mesh_transport = IrohMeshTransport::new(transport, Vec::new());

        mesh_transport.start().await.unwrap();

        let peer_id = NodeId::new("test".to_string());
        let result = mesh_transport.disconnect(&peer_id).await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_static_peers_registered_on_start() {
        use std::net::SocketAddr;

        let bind_addr: SocketAddr = "127.0.0.1:0".parse().unwrap();
        let transport = Arc::new(IrohTransport::bind(bind_addr).await.unwrap());
        let endpoint_id = transport.endpoint_id();

        let static_peers = vec![PeerInfo {
            name: "test-peer".to_string(),
            node_id: hex::encode(endpoint_id.as_bytes()),
            addresses: vec!["127.0.0.1:9999".to_string()],
            relay_url: None,
        }];

        let mesh_transport = IrohMeshTransport::new(transport, static_peers);

        mesh_transport.start().await.unwrap();

        let node_id = NodeId::new("test-peer".to_string());
        assert_eq!(mesh_transport.get_endpoint_id(&node_id), Some(endpoint_id));
    }

    #[tokio::test]
    async fn test_set_static_peers_runtime() {
        // Pre-start path: set, then start, peer is registered.
        let transport = Arc::new(IrohTransport::new().await.unwrap());
        let mesh_transport = IrohMeshTransport::new(transport.clone(), Vec::new());

        let endpoint_id = transport.endpoint_id();
        let new_peers = vec![PeerInfo {
            name: "runtime-peer".to_string(),
            node_id: hex::encode(endpoint_id.as_bytes()),
            addresses: vec!["127.0.0.1:9999".to_string()],
            relay_url: None,
        }];

        mesh_transport.set_static_peers(new_peers);

        mesh_transport.start().await.unwrap();
        let node_id = NodeId::new("runtime-peer".to_string());
        assert_eq!(mesh_transport.get_endpoint_id(&node_id), Some(endpoint_id));
    }

    #[tokio::test]
    async fn test_set_static_peers_post_start_registers_inline() {
        // Post-start path (the runtime-update case the method's docstring
        // claims): start with an empty list, then add a peer at runtime,
        // verify it's registered without needing another start() call.
        // Regression guard against the partial-semantics case the QA
        // reviewer flagged on peat-mesh#162.
        let transport = Arc::new(IrohTransport::new().await.unwrap());
        let mesh_transport = IrohMeshTransport::new(transport.clone(), Vec::new());

        mesh_transport.start().await.unwrap();

        let node_id = NodeId::new("late-peer".to_string());
        assert!(
            mesh_transport.get_endpoint_id(&node_id).is_none(),
            "Peer should not yet be registered pre-set_static_peers"
        );

        let endpoint_id = transport.endpoint_id();
        mesh_transport.set_static_peers(vec![PeerInfo {
            name: "late-peer".to_string(),
            node_id: hex::encode(endpoint_id.as_bytes()),
            addresses: vec!["127.0.0.1:9999".to_string()],
            relay_url: None,
        }]);

        assert_eq!(
            mesh_transport.get_endpoint_id(&node_id),
            Some(endpoint_id),
            "set_static_peers must register entries inline so connect() \
             can find them without another start() call"
        );
    }

    #[tokio::test]
    async fn test_set_static_peers_skips_malformed() {
        // Surface-tier proof for the [SUGGESTION] warn!-on-malformed-peer
        // policy that start() and set_static_peers() share. Malformed
        // entries must not poison the registration of well-formed ones.
        let transport = Arc::new(IrohTransport::new().await.unwrap());
        let mesh_transport = IrohMeshTransport::new(transport.clone(), Vec::new());

        let endpoint_id = transport.endpoint_id();
        mesh_transport.set_static_peers(vec![
            PeerInfo {
                name: "malformed".to_string(),
                node_id: "not-valid-hex".to_string(),
                addresses: vec!["127.0.0.1:9999".to_string()],
                relay_url: None,
            },
            PeerInfo {
                name: "good".to_string(),
                node_id: hex::encode(endpoint_id.as_bytes()),
                addresses: vec!["127.0.0.1:9999".to_string()],
                relay_url: None,
            },
        ]);

        assert!(
            mesh_transport
                .get_endpoint_id(&NodeId::new("malformed".to_string()))
                .is_none(),
            "Malformed peer must be skipped"
        );
        assert_eq!(
            mesh_transport.get_endpoint_id(&NodeId::new("good".to_string())),
            Some(endpoint_id),
            "Well-formed peer must still register despite preceding malformed entry"
        );
    }

    /// Bidirectional surface-tier proof: two real `IrohMeshTransport`
    /// instances establish a QUIC connection, with verifiable evidence
    /// on both sides — `PeerEvent::Connected` on the dialer's mesh-tier
    /// subscription (outbound path), and `peer_count() > 0` on the
    /// receiver's underlying `IrohTransport` (inbound accept-loop path).
    ///
    /// Pinned shape per QA review of peat#926: surface-tier round-trip
    /// against the new canonical `peat_mesh::transport::iroh_mesh`
    /// surface, not just the legacy peat-protocol byte-for-byte move.
    #[tokio::test]
    async fn test_two_iroh_mesh_transports_real_connect() {
        use std::net::SocketAddr;

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

        // A: accept side
        let transport_a = Arc::new(IrohTransport::bind(bind_addr).await.unwrap());
        let mesh_a = Arc::new(IrohMeshTransport::new(Arc::clone(&transport_a), Vec::new()));
        mesh_a.start().await.unwrap();

        // B: dial side
        let transport_b = Arc::new(IrohTransport::bind(bind_addr).await.unwrap());
        let mesh_b = Arc::new(IrohMeshTransport::new(Arc::clone(&transport_b), Vec::new()));
        mesh_b.start().await.unwrap();

        // Subscribe to B's mesh-tier events BEFORE dialing.
        let mut b_events = mesh_b.subscribe_peer_events();

        // B learns about A via static-peer registration.
        let addr_a = transport_a
            .bound_socket_addr_string()
            .expect("A must have a bound socket addr in bind() mode");
        let endpoint_id_a = transport_a.endpoint_id();
        mesh_b.set_static_peers(vec![PeerInfo {
            name: "node-a".to_string(),
            node_id: hex::encode(endpoint_id_a.as_bytes()),
            addresses: vec![addr_a],
            relay_url: None,
        }]);

        // Dial: B → A.
        let node_id_a = NodeId::new("node-a".to_string());
        let conn = mesh_b
            .connect(&node_id_a)
            .await
            .expect("B.connect(A) should succeed against A's accept loop");
        assert!(conn.is_alive(), "B's freshly-opened connection is alive");

        // B's mesh-tier subscription observes Connected for A.
        let event = tokio::time::timeout(Duration::from_secs(2), b_events.recv())
            .await
            .expect("B should observe a peer event within 2s")
            .expect("event channel must not close");
        match event {
            PeerEvent::Connected { peer_id, .. } => {
                assert_eq!(peer_id, node_id_a, "Connected event must name A");
            }
            other => panic!("expected PeerEvent::Connected, got {:?}", other),
        }

        // A's underlying IrohTransport saw the accept side land.
        // The accept loop is async — give it a brief window to register
        // the inbound connection before we assert.
        for _ in 0..20 {
            if transport_a.peer_count() > 0 {
                break;
            }
            tokio::time::sleep(Duration::from_millis(50)).await;
        }
        assert!(
            transport_a.peer_count() > 0,
            "A's IrohTransport must have accepted the inbound connection"
        );

        // Cleanup.
        let _ = mesh_b.stop().await;
        let _ = mesh_a.stop().await;
    }

    /// Per-variant pin for the iroh `ConnectionError` → peat-mesh
    /// `DisconnectReason` mapping (peat-mesh#164). Covers the five
    /// unit-shape variants directly; the three data-bearing variants
    /// (`ConnectionClosed`, `ApplicationClosed`, `TransportError`)
    /// can't be constructed from outside noq-proto (their inner types
    /// are `#[non_exhaustive]` or carry private fields), but the
    /// **exhaustive `match` in `map_connection_error` itself** is the
    /// load-bearing regression guard for any future iroh variant
    /// addition — a new variant would refuse to compile until added
    /// to the match, surfacing at peat-mesh CI rather than after a
    /// downstream binary swallows it as `NetworkError`.
    ///
    /// The unit-variant cases catch the original substring-on-`Display`
    /// regression class:
    ///
    /// - `TimedOut` and `LocallyClosed` no longer depend on the words
    ///   "timeout"/"idle"/"closed" appearing in iroh's `thiserror`
    ///   wording.
    /// - `LocallyClosed` ↔ `RemoteClosed` are now disambiguated; the
    ///   legacy code collapsed them by matching "closed" in both.
    /// - `VersionMismatch` and `CidsExhausted` get stable
    ///   per-variant tags (`version-mismatch`, `cids-exhausted`) in
    ///   the `NetworkError` payload instead of the raw `Display`,
    ///   so audit-log parsers don't break on upstream wording tweaks.
    #[test]
    fn parse_close_reason_per_variant_pin() {
        use iroh::endpoint::ConnectionError;

        assert!(matches!(
            map_connection_error(ConnectionError::TimedOut),
            DisconnectReason::IdleTimeout
        ));

        assert!(matches!(
            map_connection_error(ConnectionError::Reset),
            DisconnectReason::RemoteClosed
        ));

        // Disambiguation gain over the legacy substring code which
        // collapsed `LocallyClosed` into `RemoteClosed` via "closed".
        assert!(matches!(
            map_connection_error(ConnectionError::LocallyClosed),
            DisconnectReason::LocalClosed
        ));

        match map_connection_error(ConnectionError::VersionMismatch) {
            DisconnectReason::NetworkError(payload) => {
                assert_eq!(payload, "version-mismatch");
            }
            other => panic!("expected NetworkError, got {:?}", other),
        }

        match map_connection_error(ConnectionError::CidsExhausted) {
            DisconnectReason::NetworkError(payload) => {
                assert_eq!(payload, "cids-exhausted");
            }
            other => panic!("expected NetworkError, got {:?}", other),
        }
    }

    /// Format pin for the `ApplicationError(_)` payload contract
    /// documented on [`DisconnectReason`]. The contract is
    /// `"code={error_code} reason={utf8-lossy(reason)}"`; downstream
    /// operators / audit-log parsers rely on the `code=` and `reason=`
    /// keys staying stable. Tests the extracted
    /// [`format_application_close`] helper directly since
    /// `iroh::endpoint::ApplicationClose` can't be constructed from
    /// outside `noq-proto` (private fields).
    #[test]
    fn format_application_close_payload_pin() {
        // Plain ASCII reason
        assert_eq!(
            format_application_close(42, b"goodbye"),
            "code=42 reason=goodbye"
        );

        // Zero error code, empty reason — operationally common
        assert_eq!(format_application_close(0, b""), "code=0 reason=");

        // Non-UTF-8 reason bytes — must not panic; uses lossy decode
        let non_utf8: &[u8] = &[0xff, 0xfe, b'a'];
        let formatted = format_application_close(7, non_utf8);
        assert!(
            formatted.starts_with("code=7 reason="),
            "lossy decode broke the prefix contract: {}",
            formatted
        );

        // Large error_code (QUIC VarInts go up to 2^62 - 1) — format
        // must not lose precision AND the surrounding `code=` /
        // ` reason=` framing must stay byte-exact (full-string
        // equality, not `.contains()`, catches a regression that
        // smuggled extra characters into the prefix while still
        // containing the big number).
        let big = (1u64 << 50) + 17;
        assert_eq!(
            format_application_close(big, b"ok"),
            format!("code={} reason=ok", big),
        );
    }
}
