peat-tak 0.0.1

//! `PeatTakTransport` — [`MeshTransport`] adapter wiring [`TakServerTransport`]
//! and [`CotTranslator`] against a [`peat_mesh::Node`].
//!
//! Mirrors `peat-mesh-sapient`'s `PeatSapientTransport` shape: a codec-only
//! translator registered with [`TransportManager`] for fan-out, plus a
//! background recv loop that decodes inbound CoT and publishes to the mesh.
//!
//! [`MeshTransport`]: peat_mesh::transport::MeshTransport
//! [`TransportManager`]: peat_mesh::transport::TransportManager

use std::collections::HashMap;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, RwLock};
use std::time::Instant;

use async_trait::async_trait;
use futures::StreamExt;
use tokio::sync::mpsc;
use tokio::task::JoinHandle;
use tracing::{debug, info, warn};

use peat_mesh::transport::{
    ConnectionHealth, ConnectionState, DisconnectReason, MeshConnection, MeshTransport, NodeId,
    OutboundSink, PeerEvent, PeerEventReceiver, Result, TranslationContext, Translator, Transport,
    TransportCapabilities, TransportError, TransportType, PEER_EVENT_CHANNEL_CAPACITY,
};
use peat_mesh::Node as MeshNode;

use crate::cot_translator::CotTranslator;
use crate::error::TakError;
use crate::traits::{CotFilter, TakTransport};
use crate::TakServerTransport;

const TAK_ORIGIN: &str = "tak";
const TAK_TRANSPORT_TYPE_TAG: u32 = 0x544B; // "TK" ASCII-packed

const OUTBOUND_CHANNEL_DEPTH: usize = 64;

/// Configuration for the TAK mesh transport.
#[derive(Debug, Clone)]
pub struct TakMeshConfig {
    /// TAK Server address (host:port).
    pub server_addr: std::net::SocketAddr,
    /// Mesh-side node ID assigned to the TAK Server peer.
    pub peer_node_id: NodeId,
    /// Use TLS for the TAK Server connection.
    pub use_tls: bool,
    /// Client identity for mTLS — required when `use_tls` is true.
    pub identity: Option<crate::config::TakIdentity>,
}

struct PeerRecord {
    connected_at: Instant,
    alive: Arc<AtomicBool>,
    recv_task: JoinHandle<()>,
}

struct PeerRecordHandle {
    peer_id: NodeId,
    connected_at: Instant,
    alive: Arc<AtomicBool>,
}

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

    fn is_alive(&self) -> bool {
        self.alive.load(Ordering::Relaxed)
    }

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

type PeerMap = Arc<RwLock<HashMap<NodeId, PeerRecord>>>;
type EventSenders = Arc<RwLock<Vec<mpsc::Sender<PeerEvent>>>>;

/// [`OutboundSink`] for TAK — receives CoT XML bytes from the
/// TransportManager fan-out and forwards them to the TAK Server.
pub struct TakOutboundSink {
    sender: mpsc::Sender<Vec<u8>>,
}

#[async_trait]
impl OutboundSink for TakOutboundSink {
    async fn send_outbound(&self, bytes: Vec<u8>, _ctx: &TranslationContext) -> anyhow::Result<()> {
        self.sender
            .send(bytes)
            .await
            .map_err(|_| anyhow::anyhow!("tak outbound channel closed"))?;
        Ok(())
    }
}

/// `MeshTransport`/`Transport` impl for TAK (Cursor-on-Target).
///
/// Connects to a TAK Server, subscribes to inbound CoT events, decodes
/// them via [`CotTranslator`] into mesh documents, and publishes to the
/// [`Node`] with origin `"tak"`.
///
/// Outbound: callers register the translator with a [`TransportManager`]
/// using [`outbound_sink()`](PeatTakTransport::outbound_sink) — fan-out
/// encodes mesh documents via `CotTranslator::encode_outbound` into CoT
/// XML, which this transport sends to the TAK Server.
///
/// [`Node`]: peat_mesh::Node
/// [`TransportManager`]: peat_mesh::transport::TransportManager
pub struct PeatTakTransport {
    config: TakMeshConfig,
    translator: Arc<CotTranslator>,
    node: Arc<MeshNode>,
    peers: PeerMap,
    event_senders: EventSenders,
    started: RwLock<Option<Instant>>,
    worker_task: RwLock<Option<JoinHandle<()>>>,
    capabilities: TransportCapabilities,
    outbound_tx: mpsc::Sender<Vec<u8>>,
    outbound_rx: Arc<tokio::sync::Mutex<Option<mpsc::Receiver<Vec<u8>>>>>,
}

impl PeatTakTransport {
    pub fn new(config: TakMeshConfig, node: Arc<MeshNode>, translator: Arc<CotTranslator>) -> Self {
        let (outbound_tx, outbound_rx) = mpsc::channel(OUTBOUND_CHANNEL_DEPTH);
        Self {
            config,
            translator,
            node,
            peers: Arc::new(RwLock::new(HashMap::new())),
            event_senders: Arc::new(RwLock::new(Vec::new())),
            started: RwLock::new(None),
            worker_task: RwLock::new(None),
            capabilities: TransportCapabilities {
                transport_type: TransportType::Custom(TAK_TRANSPORT_TYPE_TAG),
                max_bandwidth_bps: 0,
                typical_latency_ms: 50,
                bidirectional: true,
                reliable: true,
                battery_impact: 0,
                ..Default::default()
            },
            outbound_tx,
            outbound_rx: Arc::new(tokio::sync::Mutex::new(Some(outbound_rx))),
        }
    }

    /// Returns an [`OutboundSink`] for use with [`TransportManager::register_translator`].
    ///
    /// [`TransportManager::register_translator`]: peat_mesh::transport::TransportManager::register_translator
    pub fn outbound_sink(&self) -> Arc<dyn OutboundSink> {
        Arc::new(TakOutboundSink {
            sender: self.outbound_tx.clone(),
        })
    }

    fn emit_event(&self, event: PeerEvent) {
        let senders = self.event_senders.read().unwrap_or_else(|e| e.into_inner());
        for sender in senders.iter() {
            let _ = sender.try_send(event.clone());
        }
    }

    fn register_peer(
        peers: &PeerMap,
        event_senders: &EventSenders,
        peer_id: NodeId,
        connected_at: Instant,
        alive: Arc<AtomicBool>,
        recv_task: JoinHandle<()>,
    ) {
        peers.write().unwrap_or_else(|e| e.into_inner()).insert(
            peer_id.clone(),
            PeerRecord {
                connected_at,
                alive,
                recv_task,
            },
        );
        let senders = event_senders.read().unwrap_or_else(|e| e.into_inner());
        for sender in senders.iter() {
            let _ = sender.try_send(PeerEvent::Connected {
                peer_id: peer_id.clone(),
                connected_at,
            });
        }
    }

    /// Main worker loop: connects to TAK Server, then runs inbound recv
    /// and outbound send concurrently via `select!`.
    async fn run_worker(
        config: TakMeshConfig,
        translator: Arc<CotTranslator>,
        node: Arc<MeshNode>,
        peers: PeerMap,
        event_senders: EventSenders,
        mut outbound_rx: mpsc::Receiver<Vec<u8>>,
    ) {
        let tak_config = crate::config::TakTransportConfig {
            mode: crate::config::TakTransportMode::TakServer {
                address: config.server_addr,
                use_tls: config.use_tls,
            },
            identity: config.identity.clone(),
            ..Default::default()
        };

        let mut transport = match TakServerTransport::new(tak_config) {
            Ok(t) => t,
            Err(err) => {
                warn!(%err, "tak: failed to create TakServerTransport");
                return;
            }
        };

        if let Err(err) = transport.connect().await {
            warn!(%err, addr = %config.server_addr, "tak: connect to TAK Server failed");
            return;
        }

        info!(addr = %config.server_addr, "tak: connected to TAK Server");

        let alive = Arc::new(AtomicBool::new(true));
        let connected_at = Instant::now();

        let recv_task = tokio::spawn(async {});
        Self::register_peer(
            &peers,
            &event_senders,
            config.peer_node_id.clone(),
            connected_at,
            alive.clone(),
            recv_task,
        );

        let mut inbound_stream = match transport.subscribe(CotFilter::all()).await {
            Ok(s) => s,
            Err(err) => {
                warn!(%err, "tak: subscribe failed");
                alive.store(false, Ordering::Relaxed);
                return;
            }
        };

        loop {
            tokio::select! {
                inbound = inbound_stream.next() => {
                    let Some(result) = inbound else { break; };
                    match result {
                        Ok(event) => {
                            let xml = match event.to_xml() {
                                Ok(xml) => xml,
                                Err(err) => {
                                    warn!(%err, uid = %event.uid, "tak: to_xml failed");
                                    continue;
                                }
                            };
                            let ctx = TranslationContext::inbound(
                                config.peer_node_id.as_str().to_string(),
                            );
                            match translator.decode_inbound(xml.as_bytes(), &ctx).await {
                                Ok(Some(doc)) => {
                                    if let Err(err) = node
                                        .publish_with_origin(
                                            translator.tracks_collection(),
                                            doc,
                                            Some(TAK_ORIGIN.to_string()),
                                        )
                                        .await
                                    {
                                        warn!(%err, "tak: publish_with_origin failed");
                                    }
                                }
                                Ok(None) => {}
                                Err(err) => {
                                    warn!(%err, "tak: decode_inbound failed");
                                }
                            }
                        }
                        Err(err) => {
                            warn!(%err, "tak: inbound stream error");
                        }
                    }
                }
                outbound = outbound_rx.recv() => {
                    let Some(xml_bytes) = outbound else { break; };
                    match send_xml_to_tak(&transport, &xml_bytes).await {
                        Ok(()) => {
                            debug!("tak: sent outbound CoT event");
                        }
                        Err(err) => {
                            warn!(%err, "tak: outbound send failed");
                        }
                    }
                }
            }
        }

        alive.store(false, Ordering::Relaxed);
        info!("tak: worker loop exited");
    }
}

/// Parse XML bytes back to a CotEvent and send via the TakServerTransport.
async fn send_xml_to_tak(
    transport: &TakServerTransport,
    xml_bytes: &[u8],
) -> std::result::Result<(), TakError> {
    let xml = std::str::from_utf8(xml_bytes)
        .map_err(|e| TakError::EncodingError(format!("outbound bytes not UTF-8: {e}")))?;
    let event = peat_protocol::cot::CotEvent::from_xml(xml)
        .map_err(|e| TakError::EncodingError(format!("outbound XML parse failed: {e}")))?;
    transport.send_cot(&event, 3).await
}

#[async_trait]
impl MeshTransport for PeatTakTransport {
    async fn start(&self) -> Result<()> {
        let outbound_rx = self
            .outbound_rx
            .lock()
            .await
            .take()
            .expect("start() called more than once");
        let task = tokio::spawn(Self::run_worker(
            self.config.clone(),
            self.translator.clone(),
            self.node.clone(),
            self.peers.clone(),
            self.event_senders.clone(),
            outbound_rx,
        ));
        *self.worker_task.write().unwrap_or_else(|e| e.into_inner()) = Some(task);
        *self.started.write().unwrap_or_else(|e| e.into_inner()) = Some(Instant::now());
        Ok(())
    }

    async fn stop(&self) -> Result<()> {
        if let Some(task) = self
            .worker_task
            .write()
            .unwrap_or_else(|e| e.into_inner())
            .take()
        {
            task.abort();
        }
        let mut peers = self.peers.write().unwrap_or_else(|e| e.into_inner());
        for (_, record) in peers.drain() {
            record.recv_task.abort();
        }
        *self.started.write().unwrap_or_else(|e| e.into_inner()) = None;
        Ok(())
    }

    async fn connect(&self, peer_id: &NodeId) -> Result<Box<dyn MeshConnection>> {
        self.get_connection(peer_id)
            .ok_or_else(|| TransportError::PeerNotFound(peer_id.to_string()))
    }

    async fn disconnect(&self, peer_id: &NodeId) -> Result<()> {
        let record = {
            let mut peers = self.peers.write().unwrap_or_else(|e| e.into_inner());
            peers
                .remove(peer_id)
                .ok_or_else(|| TransportError::PeerNotFound(peer_id.to_string()))?
        };
        let connection_duration = record.connected_at.elapsed();
        record.recv_task.abort();
        self.emit_event(PeerEvent::Disconnected {
            peer_id: peer_id.clone(),
            reason: DisconnectReason::LocalClosed,
            connection_duration,
        });
        Ok(())
    }

    fn get_connection(&self, peer_id: &NodeId) -> Option<Box<dyn MeshConnection>> {
        let peers = self.peers.read().unwrap_or_else(|e| e.into_inner());
        let record = peers.get(peer_id)?;
        Some(Box::new(PeerRecordHandle {
            peer_id: peer_id.clone(),
            connected_at: record.connected_at,
            alive: record.alive.clone(),
        }))
    }

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

    fn connected_peers(&self) -> Vec<NodeId> {
        self.peers
            .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> {
        let conn = self.get_connection(peer_id)?;
        Some(ConnectionHealth {
            state: if conn.is_alive() {
                ConnectionState::Healthy
            } else {
                ConnectionState::Dead
            },
            ..Default::default()
        })
    }
}

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

    fn is_available(&self) -> bool {
        self.started
            .read()
            .unwrap_or_else(|e| e.into_inner())
            .is_some()
    }

    fn can_reach(&self, peer_id: &NodeId) -> bool {
        self.is_connected(peer_id)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use peat_mesh::sync::DataSyncBackend;
    use peat_mesh::sync::InMemoryBackend;

    fn make_transport() -> PeatTakTransport {
        let backend: Arc<dyn DataSyncBackend> = Arc::new(InMemoryBackend::new_initialized());
        let node = Arc::new(MeshNode::new(backend));
        let translator = Arc::new(CotTranslator::new());
        PeatTakTransport::new(
            TakMeshConfig {
                server_addr: "127.0.0.1:8087".parse().unwrap(),
                peer_node_id: NodeId::from("tak-server-0"),
                use_tls: false,
                identity: None,
            },
            node,
            translator,
        )
    }

    fn register_fake_peer(transport: &PeatTakTransport, id: &str, alive: bool) {
        PeatTakTransport::register_peer(
            &transport.peers,
            &transport.event_senders,
            NodeId::from(id),
            Instant::now(),
            Arc::new(AtomicBool::new(alive)),
            tokio::spawn(async {}),
        );
    }

    #[test]
    fn capabilities_returns_tak_transport_type() {
        let transport = make_transport();
        assert!(matches!(
            transport.capabilities().transport_type,
            TransportType::Custom(TAK_TRANSPORT_TYPE_TAG)
        ));
        assert!(transport.capabilities().reliable);
        assert!(transport.capabilities().bidirectional);
    }

    #[test]
    fn is_available_false_before_start() {
        let transport = make_transport();
        assert!(!transport.is_available());
    }

    #[tokio::test]
    async fn connect_unknown_peer_errors() {
        let transport = make_transport();
        let result = transport.connect(&NodeId::from("ghost")).await;
        assert!(matches!(result, Err(TransportError::PeerNotFound(_))));
    }

    #[tokio::test]
    async fn connect_returns_connection_for_registered_peer() {
        let transport = make_transport();
        register_fake_peer(&transport, "peer-tak", true);

        let conn = transport
            .connect(&NodeId::from("peer-tak"))
            .await
            .expect("connect");
        assert_eq!(conn.peer_id(), &NodeId::from("peer-tak"));
        assert!(conn.is_alive());
    }

    #[tokio::test]
    async fn disconnect_emits_disconnected_event() {
        let transport = make_transport();
        let mut events = transport.subscribe_peer_events();

        register_fake_peer(&transport, "peer-dc", true);
        assert!(matches!(
            events.recv().await,
            Some(PeerEvent::Connected { .. })
        ));

        transport
            .disconnect(&NodeId::from("peer-dc"))
            .await
            .expect("disconnect");

        match events.recv().await {
            Some(PeerEvent::Disconnected {
                peer_id, reason, ..
            }) => {
                assert_eq!(peer_id, NodeId::from("peer-dc"));
                assert!(matches!(reason, DisconnectReason::LocalClosed));
            }
            other => panic!("expected Disconnected event, got {other:?}"),
        }
        assert_eq!(transport.peer_count(), 0);
    }

    #[tokio::test]
    async fn disconnect_unknown_peer_errors() {
        let transport = make_transport();
        let result = transport.disconnect(&NodeId::from("nope")).await;
        assert!(matches!(result, Err(TransportError::PeerNotFound(_))));
    }

    #[tokio::test]
    async fn get_peer_health_alive_reports_healthy() {
        let transport = make_transport();
        register_fake_peer(&transport, "h-peer", true);
        let health = transport
            .get_peer_health(&NodeId::from("h-peer"))
            .expect("health");
        assert!(matches!(health.state, ConnectionState::Healthy));
    }

    #[tokio::test]
    async fn get_peer_health_dead_reports_dead() {
        let transport = make_transport();
        register_fake_peer(&transport, "d-peer", false);
        let health = transport
            .get_peer_health(&NodeId::from("d-peer"))
            .expect("health");
        assert!(matches!(health.state, ConnectionState::Dead));
    }

    #[tokio::test]
    async fn can_reach_reflects_connected_peers() {
        let transport = make_transport();
        assert!(!transport.can_reach(&NodeId::from("peer-r")));
        register_fake_peer(&transport, "peer-r", true);
        assert!(transport.can_reach(&NodeId::from("peer-r")));
    }

    #[tokio::test]
    async fn stop_clears_peers_and_marks_unavailable() {
        let transport = make_transport();
        *transport.started.write().unwrap() = Some(Instant::now());
        register_fake_peer(&transport, "p1", true);
        register_fake_peer(&transport, "p2", true);
        assert_eq!(transport.peer_count(), 2);
        assert!(transport.is_available());

        transport.stop().await.expect("stop");

        assert_eq!(transport.peer_count(), 0);
        assert!(!transport.is_available());
    }

    #[tokio::test]
    async fn outbound_sink_accepts_bytes() {
        let transport = make_transport();
        let sink = transport.outbound_sink();
        let ctx = TranslationContext::outbound().with_collection("tracks");
        sink.send_outbound(b"<event/>".to_vec(), &ctx)
            .await
            .expect("sink should accept bytes");
    }

    #[tokio::test]
    async fn subscribe_peer_events_fans_out() {
        let transport = make_transport();
        let mut first = transport.subscribe_peer_events();
        let mut second = transport.subscribe_peer_events();

        register_fake_peer(&transport, "peer-fan", true);

        for events in [&mut first, &mut second] {
            assert!(matches!(
                events.recv().await,
                Some(PeerEvent::Connected { .. })
            ));
        }
    }
}