feltdb 0.3.0

/// In-process transport using Tokio channels
///
/// Provides fast, deterministic delivery of ReplicationMessage for testing.
/// No network delay, no packet loss by default.
///
/// Transport Guarantees:
/// - Ordered delivery per connection
/// - Serialization integrity (bincode framing)
/// - Bounded channel semantics (messages queued until received)
/// - Connection lifecycle tracking
///
/// Deduplication/idempotency is the protocol's responsibility via DistributedTransactionExecutor.
use crate::distributed_transactions::ReplicationMessage;
use crate::replication_protocol::{ProtocolError, ProtocolMetrics, ProtocolTransport};
use std::sync::Arc;
use tokio::sync::mpsc;

/// In-process message delivery via Tokio channel
pub struct InProcessTransport {
    local_node_id: String,
    peer_node_id: String,
    tx: mpsc::UnboundedSender<ReplicationMessage>,
    rx: Arc<tokio::sync::Mutex<mpsc::UnboundedReceiver<ReplicationMessage>>>,
    metrics: Arc<ProtocolMetrics>,
    is_connected: bool,
}

impl InProcessTransport {
    /// Create a new in-process transport pair
    pub fn create_pair(
        local_id: String,
        peer_id: String,
    ) -> (Self, mpsc::UnboundedReceiver<ReplicationMessage>) {
        let (tx, rx) = mpsc::unbounded_channel();

        let transport = Self {
            local_node_id: local_id,
            peer_node_id: peer_id,
            tx,
            rx: Arc::new(tokio::sync::Mutex::new(rx)),
            metrics: Arc::new(ProtocolMetrics::new()),
            is_connected: true,
        };

        // For the pair variant, return both the transport and a separate receiver
        // This isn't ideal - use create_bidirectional_pair for proper two-way communication
        let (tx2, rx2) = mpsc::unbounded_channel();
        let transport_with_output = Self {
            local_node_id: transport.local_node_id.clone(),
            peer_node_id: transport.peer_node_id.clone(),
            tx: tx2,
            rx: transport.rx.clone(),
            metrics: transport.metrics.clone(),
            is_connected: transport.is_connected,
        };

        (transport_with_output, rx2)
    }

    /// Create a bidirectional pair of transports
    pub fn create_bidirectional_pair(local_id: String, peer_id: String) -> (Self, Self) {
        let (local_tx, local_rx) = mpsc::unbounded_channel();
        let (peer_tx, peer_rx) = mpsc::unbounded_channel();

        let local = Self {
            local_node_id: local_id.clone(),
            peer_node_id: peer_id.clone(),
            tx: peer_tx,
            rx: Arc::new(tokio::sync::Mutex::new(local_rx)),
            metrics: Arc::new(ProtocolMetrics::new()),
            is_connected: true,
        };

        let peer = Self {
            local_node_id: peer_id,
            peer_node_id: local_id,
            tx: local_tx,
            rx: Arc::new(tokio::sync::Mutex::new(peer_rx)),
            metrics: Arc::new(ProtocolMetrics::new()),
            is_connected: true,
        };

        (local, peer)
    }

    pub fn metrics(&self) -> &ProtocolMetrics {
        &self.metrics
    }
}

#[async_trait::async_trait]
impl ProtocolTransport for InProcessTransport {
    async fn send(&self, message: ReplicationMessage) -> Result<(), ProtocolError> {
        // Attempt serialization for metrics, but don't fail if it doesn't work
        if let Ok(serialized) = bincode::serialize(&message) {
            self.metrics.record_send(serialized.len());
        } else {
            // Estimate size as fallback
            self.metrics.record_send(1024);
        }

        self.tx
            .send(message)
            .map_err(|_| ProtocolError::TransportError("channel closed".to_string()))?;

        Ok(())
    }

    async fn receive(&self) -> Result<Vec<ReplicationMessage>, ProtocolError> {
        // This is a non-blocking receive - collect all available messages
        let mut messages = Vec::new();
        let mut rx = self.rx.lock().await;

        loop {
            match rx.try_recv() {
                Ok(msg) => {
                    // Attempt serialization for metrics, but don't fail if it doesn't work
                    if let Ok(serialized) = bincode::serialize(&msg) {
                        self.metrics.record_receive(serialized.len());
                    } else {
                        self.metrics.record_receive(1024);
                    }
                    messages.push(msg);
                }
                Err(mpsc::error::TryRecvError::Empty) => break,
                Err(mpsc::error::TryRecvError::Disconnected) => {
                    return Err(ProtocolError::TransportError(
                        "channel disconnected".to_string(),
                    ))
                }
            }
        }

        Ok(messages)
    }

    async fn connect(&mut self, _peer_addr: &str) -> Result<(), ProtocolError> {
        // In-process: connection is already established
        self.is_connected = true;
        self.metrics.record_connection_state_change();
        Ok(())
    }

    async fn disconnect(&mut self) -> Result<(), ProtocolError> {
        self.is_connected = false;
        self.metrics.record_connection_state_change();
        Ok(())
    }

    fn is_connected(&self) -> bool {
        self.is_connected
    }
}

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

    #[tokio::test]
    async fn in_process_bidirectional_sends_and_receives() {
        let (local, peer) =
            InProcessTransport::create_bidirectional_pair("local".to_string(), "peer".to_string());

        let msg = ReplicationMessage::new(
            crate::distributed_transactions::TransactionEnvelope::new(
                "local".to_string(),
                1,
                "tx-1".to_string(),
                vec![],
                crate::transactions::ConsistencyContract::local(),
            ),
            "local".to_string(),
            "peer".to_string(),
            1,
        );

        local.send(msg).await.expect("send");

        let received = peer.receive().await.expect("receive");
        assert_eq!(received.len(), 1);
        assert_eq!(peer.metrics().get_messages_received(), 1);
        assert_eq!(local.metrics().get_messages_sent(), 1);
    }

    #[tokio::test]
    async fn metrics_track_connection_state() {
        let (mut transport, _) =
            InProcessTransport::create_bidirectional_pair("local".to_string(), "peer".to_string());

        let initial_state = transport.is_connected();
        assert!(initial_state);

        // Disconnect and check metrics
        transport.disconnect().await.expect("disconnect");
        let metrics = transport.metrics();
        assert_eq!(metrics.get_connection_state_changes(), 1);
    }

    #[tokio::test]
    async fn connection_status_tracked() {
        let (mut transport, _) =
            InProcessTransport::create_bidirectional_pair("local".to_string(), "peer".to_string());

        assert!(transport.is_connected());

        transport.disconnect().await.expect("disconnect");
        assert!(!transport.is_connected());

        transport.connect("fake://addr").await.expect("connect");
        assert!(transport.is_connected());

        assert_eq!(transport.metrics().get_connection_state_changes(), 2);
    }
}