//! In-process channel transport for MCP.
//!
//! This transport is intended for embedded deployments where both endpoints
//! live in the same process or runtime. It uses typed Tokio channels instead of
//! stdio framing, so it avoids newline parsing and JSON text serialization on
//! the transport boundary.
//!
//! `InProcessTransport` is useful when:
//! - Running an MCP server on a background task or thread
//! - Embedding an MCP server inside another Rust application
//! - Connecting two local MCP endpoints with minimal transport overhead

use tokio::sync::{Mutex, mpsc};

use super::traits::{IncomingMessage, JsonRpcNotification, Transport, TransportError, sealed};
use crate::protocol::types::{JsonRpcRequest, JsonRpcResponse};
use crate::server::multiplexer::JsonRpcClientRequest;

/// A bidirectional in-process transport backed by Tokio channels.
///
/// Use [`InProcessTransport::pair`] to create two connected endpoints.
pub struct InProcessTransport {
    inbox_rx: Mutex<mpsc::Receiver<IncomingMessage>>,
    peer_tx: Option<mpsc::Sender<IncomingMessage>>,
    closed: bool,
}

impl InProcessTransport {
    /// Default channel capacity used by [`Self::pair`].
    pub const DEFAULT_CHANNEL_CAPACITY: usize = 256;

    /// Create two connected in-process transport endpoints.
    pub fn pair() -> (Self, Self) {
        Self::with_capacity(Self::DEFAULT_CHANNEL_CAPACITY)
    }

    /// Create two connected in-process transport endpoints with a custom
    /// bounded channel capacity.
    pub fn with_capacity(capacity: usize) -> (Self, Self) {
        assert!(
            capacity > 0,
            "in-process transport capacity must be greater than zero"
        );

        let (a_tx, a_rx) = mpsc::channel(capacity);
        let (b_tx, b_rx) = mpsc::channel(capacity);

        (
            Self {
                inbox_rx: Mutex::new(a_rx),
                peer_tx: Some(b_tx),
                closed: false,
            },
            Self {
                inbox_rx: Mutex::new(b_rx),
                peer_tx: Some(a_tx),
                closed: false,
            },
        )
    }

    fn peer_tx(&self) -> Result<mpsc::Sender<IncomingMessage>, TransportError> {
        if self.closed {
            return Err(TransportError::Closed);
        }

        self.peer_tx.clone().ok_or(TransportError::Closed)
    }
}

impl sealed::Sealed for InProcessTransport {}

impl Transport for InProcessTransport {
    fn read_message<'a>(&'a mut self) -> super::traits::TransportFuture<'a, JsonRpcRequest> {
        Box::pin(async move {
            match self.read_incoming().await? {
                IncomingMessage::Request(request) => Ok(request),
                IncomingMessage::Response(_) => {
                    tracing::warn!("Received response when expecting request");
                    self.read_message().await
                }
            }
        })
    }

    fn read_incoming<'a>(&'a mut self) -> super::traits::TransportFuture<'a, IncomingMessage> {
        if self.closed {
            return Box::pin(async { Err(TransportError::Closed) });
        }

        let rx = &self.inbox_rx;
        Box::pin(async move {
            let mut rx = rx.lock().await;
            rx.recv().await.ok_or(TransportError::Closed)
        })
    }

    fn write_message<'a>(
        &'a mut self,
        response: JsonRpcResponse,
    ) -> super::traits::TransportFuture<'a, ()> {
        let tx = self.peer_tx();
        Box::pin(async move {
            tx?.send(IncomingMessage::Response(response))
                .await
                .map_err(|_| TransportError::Closed)
        })
    }

    fn send_notification<'a>(
        &'a mut self,
        notification: JsonRpcNotification,
    ) -> super::traits::TransportFuture<'a, ()> {
        let tx = self.peer_tx();
        let request = JsonRpcRequest::notification(notification.method, notification.params);

        Box::pin(async move {
            tx?.send(IncomingMessage::Request(request))
                .await
                .map_err(|_| TransportError::Closed)
        })
    }

    fn send_request<'a>(
        &'a mut self,
        request: JsonRpcClientRequest,
    ) -> super::traits::TransportFuture<'a, ()> {
        let tx = self.peer_tx();
        let request = JsonRpcRequest::new(
            serde_json::Value::String(request.id),
            request.method,
            request.params,
        );

        Box::pin(async move {
            tx?.send(IncomingMessage::Request(request))
                .await
                .map_err(|_| TransportError::Closed)
        })
    }

    fn supports_bidirectional(&self) -> bool {
        true
    }

    fn shutdown<'a>(&'a mut self) -> super::traits::TransportFuture<'a, ()> {
        self.closed = true;
        self.peer_tx = None;
        let rx = &self.inbox_rx;

        Box::pin(async move {
            let mut rx = rx.lock().await;
            rx.close();
            Ok(())
        })
    }
}

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

    #[tokio::test]
    async fn pair_routes_requests_and_responses() {
        let (mut left, mut right) = InProcessTransport::pair();

        left.send_request(JsonRpcClientRequest::new(
            "req-1",
            "ping",
            Some(json!({"x": 1})),
        ))
        .await
        .unwrap();

        let request = match right.read_incoming().await.unwrap() {
            IncomingMessage::Request(request) => request,
            IncomingMessage::Response(_) => panic!("expected request"),
        };

        assert_eq!(request.method, "ping");
        assert_eq!(request.id, Some(json!("req-1")));
        assert_eq!(request.params, Some(json!({"x": 1})));

        right
            .write_message(JsonRpcResponse::success(
                json!("req-1"),
                json!({"ok": true}),
            ))
            .await
            .unwrap();

        let response = match left.read_incoming().await.unwrap() {
            IncomingMessage::Response(response) => response,
            IncomingMessage::Request(_) => panic!("expected response"),
        };

        assert_eq!(response.id, json!("req-1"));
        assert_eq!(response.result, Some(json!({"ok": true})));
    }

    #[tokio::test]
    async fn notifications_arrive_as_idless_requests() {
        let (mut left, mut right) = InProcessTransport::pair();

        left.send_notification(JsonRpcNotification::new(
            "notifications/message",
            Some(json!({"level": "info"})),
        ))
        .await
        .unwrap();

        let notification = match right.read_incoming().await.unwrap() {
            IncomingMessage::Request(request) => request,
            IncomingMessage::Response(_) => panic!("expected notification"),
        };

        assert_eq!(notification.method, "notifications/message");
        assert_eq!(notification.id, None);
        assert_eq!(notification.params, Some(json!({"level": "info"})));
    }
}
