//! IPC method dispatcher. One envelope in, one envelope out.
//!
//! Why a separate module from [`crate::exec::session`]: the dispatcher
//! is pure logic over an [`McpBackend`]. The session loop owns the
//! stream I/O. Splitting them lets tests exercise the routing without
//! plumbing pipes, and lets the CLI swap the backend (a real MCP
//! connection in production, a stub in integration tests) without
//! touching the dispatcher.

use async_trait::async_trait;
use serde_json::{json, Value};

#[cfg(test)]
use super::protocol::IpcResponseBody;
use super::protocol::{
    error_codes, HandshakeParams, HandshakeReply, IpcError, IpcMethod, IpcRequest, IpcResponse,
    COPROCESS_PROTOCOL_VERSION,
};

/// Contract the dispatcher requires from whatever owns the upstream MCP
/// connection.
///
/// Implementors run in the CLI binary (real `Client` over stdio or
/// streamable HTTP) and in this crate's tests (a stub that replays
/// canned responses). All methods are async because the production
/// implementation drives a tokio-based protocol client.
#[async_trait]
pub trait McpBackend: Send + Sync {
    /// Issue the MCP `initialize` handshake against the upstream server.
    /// Returns the negotiated server info that gets folded into the
    /// IPC `mcp.initialize` response.
    async fn initialize(&self) -> Result<Value, BackendError>;

    /// Forward a `tools/call`. Params shape: `{"tool": "<name>",
    /// "arguments": {...}}`.
    async fn call_tool(&self, tool: &str, arguments: Value) -> Result<Value, BackendError>;

    /// Forward a `tools/list`.
    async fn list_tools(&self) -> Result<Value, BackendError>;

    /// Forward a `resources/list`.
    async fn list_resources(&self) -> Result<Value, BackendError>;

    /// Forward a `resources/read`. Params shape: `{"uri": "<uri>"}`.
    async fn read_resource(&self, uri: &str) -> Result<Value, BackendError>;

    /// Forward a `prompts/get`. Params shape: `{"name": "<name>",
    /// "arguments": {...}}`.
    async fn call_prompt(&self, name: &str, arguments: Value) -> Result<Value, BackendError>;

    /// Configure the upload reporter destination. `url` is validated by
    /// the implementation; the dispatcher only forwards.
    async fn reporter_set_upload(&self, url: &str, token: Option<&str>)
        -> Result<(), BackendError>;

    /// Flush the upload reporter. Called by the SDK at session end.
    async fn reporter_flush(&self) -> Result<(), BackendError>;

    /// Switch the cache mode. Accepted values: `read`, `write`,
    /// `readwrite`, `off`.
    async fn cache_set_mode(&self, mode: &str) -> Result<(), BackendError>;

    /// Switch the cassette mode. Accepted values: `record`, `replay`,
    /// `off`.
    async fn cassette_set_mode(&self, mode: &str) -> Result<(), BackendError>;

    /// Graceful shutdown. Drains in-flight requests and flushes any
    /// pending reporter buffers.
    async fn shutdown(&self) -> Result<(), BackendError>;
}

/// Typed error returned by [`McpBackend`] implementations.
///
/// The dispatcher maps each variant onto a JSON-RPC error code so the
/// SDK sees a stable code set even when the upstream MCP server changes
/// its own error encoding.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum BackendError {
    /// Upstream MCP server returned an error response.
    Server {
        /// Server-supplied error message.
        message: String,
        /// Optional structured payload from the server.
        data: Option<Value>,
    },
    /// The MCP connection has dropped. Subsequent calls also fail
    /// until the SDK reconnects (out of scope for v1.1).
    Disconnected {
        /// Reason captured at the time the connection dropped.
        reason: String,
    },
    /// Caller passed params that did not satisfy the method contract.
    InvalidParams(String),
    /// Anything else (lock poisoning, JSON shape mismatch from the
    /// upstream server). Mapped to `-32603` internal error.
    Internal(String),
}

impl BackendError {
    fn into_ipc(self) -> IpcError {
        match self {
            BackendError::Server { message, data } => IpcError {
                code: error_codes::MCP_SERVER_ERROR,
                message,
                data,
            },
            BackendError::Disconnected { reason } => IpcError {
                code: error_codes::MCP_DISCONNECTED,
                message: reason,
                data: None,
            },
            BackendError::InvalidParams(message) => IpcError {
                code: error_codes::INVALID_PARAMS,
                message,
                data: None,
            },
            BackendError::Internal(message) => IpcError {
                code: error_codes::INTERNAL_ERROR,
                message,
                data: None,
            },
        }
    }
}

/// Dispatch one IPC request against a backend and produce one response.
///
/// Held as a struct (rather than free functions) so a future evolution
/// can carry per-session state (request counter, instrumentation hooks)
/// without changing the call sites.
pub struct Dispatcher<B: McpBackend> {
    backend: B,
}

impl<B: McpBackend> Dispatcher<B> {
    /// Wire a dispatcher around a backend.
    pub fn new(backend: B) -> Self {
        Self { backend }
    }

    /// Borrow the underlying backend. Used by the session loop to call
    /// methods (for example `shutdown`) that are not request-driven.
    pub fn backend(&self) -> &B {
        &self.backend
    }

    /// Route one envelope. Always returns an envelope; protocol errors
    /// fold into a `failure` body so the SDK never has to special-case
    /// "no response at all".
    pub async fn dispatch(&self, request: IpcRequest) -> IpcResponse {
        let id = request.id.clone().unwrap_or(Value::Null);
        if request.jsonrpc != "2.0" {
            return IpcResponse::failure(
                id,
                IpcError {
                    code: error_codes::INVALID_REQUEST,
                    message: format!("jsonrpc must be \"2.0\", got {:?}", request.jsonrpc),
                    data: None,
                },
            );
        }
        let Some(method) = IpcMethod::from_wire(&request.method) else {
            return IpcResponse::failure(
                id,
                IpcError {
                    code: error_codes::METHOD_NOT_FOUND,
                    message: format!("method not found: {}", request.method),
                    data: None,
                },
            );
        };
        let outcome = self
            .run(method, request.params.unwrap_or(Value::Null))
            .await;
        match outcome {
            Ok(result) => IpcResponse::success(id, result),
            Err(err) => IpcResponse::failure(id, err.into_ipc()),
        }
    }

    async fn run(&self, method: IpcMethod, params: Value) -> Result<Value, BackendError> {
        match method {
            IpcMethod::Handshake => handshake(params),
            IpcMethod::Initialize => {
                let server_info = self.backend.initialize().await?;
                Ok(json!({
                    "ipc_version": super::protocol::IPC_VERSION,
                    "server": server_info,
                    "ready": true,
                }))
            }
            IpcMethod::Call => {
                let tool = require_string(&params, "tool")?;
                let arguments = params
                    .get("arguments")
                    .cloned()
                    .unwrap_or_else(|| json!({}));
                self.backend.call_tool(&tool, arguments).await
            }
            IpcMethod::ListTools => self.backend.list_tools().await,
            IpcMethod::ListResources => self.backend.list_resources().await,
            IpcMethod::ReadResource => {
                let uri = require_string(&params, "uri")?;
                self.backend.read_resource(&uri).await
            }
            IpcMethod::CallPrompt => {
                let name = require_string(&params, "name")?;
                let arguments = params
                    .get("arguments")
                    .cloned()
                    .unwrap_or_else(|| json!({}));
                self.backend.call_prompt(&name, arguments).await
            }
            IpcMethod::ReporterSetUpload => {
                let url = require_string(&params, "url")?;
                let token = params
                    .get("token")
                    .and_then(Value::as_str)
                    .map(str::to_owned);
                self.backend
                    .reporter_set_upload(&url, token.as_deref())
                    .await?;
                Ok(json!({"ok": true}))
            }
            IpcMethod::ReporterFlush => {
                self.backend.reporter_flush().await?;
                Ok(json!({"ok": true}))
            }
            IpcMethod::CacheSetMode => {
                let mode = require_string(&params, "mode")?;
                self.backend.cache_set_mode(&mode).await?;
                Ok(json!({"mode": mode}))
            }
            IpcMethod::CassetteSetMode => {
                let mode = require_string(&params, "mode")?;
                self.backend.cassette_set_mode(&mode).await?;
                Ok(json!({"mode": mode}))
            }
            IpcMethod::Shutdown => {
                self.backend.shutdown().await?;
                Ok(json!({"ok": true}))
            }
        }
    }
}

/// Answer `coprocess/handshake` (WOR-1318).
///
/// A pure protocol concern, so it never touches the backend. Legacy
/// (v1) SDKs send no handshake at all and the session keeps serving
/// them; the version is enforced only when the request arrives. The
/// params parse strictly, so a handshake from a newer contract fails
/// here with both versions named instead of misparsing.
fn handshake(params: Value) -> Result<Value, BackendError> {
    let parsed: HandshakeParams = serde_json::from_value(params)
        .map_err(|e| BackendError::InvalidParams(format!("coprocess/handshake params: {e}")))?;
    let ours = COPROCESS_PROTOCOL_VERSION;
    if parsed.protocol_version != ours {
        let sdk = parsed.sdk.as_deref().unwrap_or("unknown");
        let theirs = parsed.protocol_version;
        let direction = if theirs > ours {
            "Upgrade the mcptest binary (or pin the SDK to the matching release)."
        } else {
            "Upgrade the SDK (or pin the mcptest binary to the matching release)."
        };
        return Err(BackendError::InvalidParams(format!(
            "coprocess protocol version mismatch: this mcptest binary speaks v{ours}, \
             the {sdk} SDK sent v{theirs}. {direction}"
        )));
    }
    serde_json::to_value(HandshakeReply {
        protocol_version: ours,
        binary_version: crate::version::semver().to_owned(),
    })
    .map_err(|e| BackendError::Internal(e.to_string()))
}

fn require_string(params: &Value, field: &str) -> Result<String, BackendError> {
    params
        .get(field)
        .and_then(Value::as_str)
        .map(str::to_owned)
        .ok_or_else(|| BackendError::InvalidParams(format!("missing string field `{field}`")))
}

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

    struct StubBackend {
        calls: Mutex<Vec<String>>,
        fail_initialize: bool,
    }

    impl StubBackend {
        fn new() -> Self {
            Self {
                calls: Mutex::new(Vec::new()),
                fail_initialize: false,
            }
        }

        fn record(&self, name: &str) {
            self.calls.lock().unwrap().push(name.to_owned());
        }
    }

    #[async_trait]
    impl McpBackend for StubBackend {
        async fn initialize(&self) -> Result<Value, BackendError> {
            self.record("initialize");
            if self.fail_initialize {
                return Err(BackendError::Disconnected {
                    reason: "no server".into(),
                });
            }
            Ok(json!({"name": "stub", "version": "0.0.1"}))
        }

        async fn call_tool(&self, tool: &str, _arguments: Value) -> Result<Value, BackendError> {
            self.record(&format!("call:{tool}"));
            Ok(json!({"content": [{"type": "text", "text": "hi"}], "isError": false}))
        }

        async fn list_tools(&self) -> Result<Value, BackendError> {
            self.record("list_tools");
            Ok(json!({"tools": []}))
        }

        async fn list_resources(&self) -> Result<Value, BackendError> {
            self.record("list_resources");
            Ok(json!({"resources": []}))
        }

        async fn read_resource(&self, uri: &str) -> Result<Value, BackendError> {
            self.record(&format!("read:{uri}"));
            Ok(json!({"contents": []}))
        }

        async fn call_prompt(&self, name: &str, _arguments: Value) -> Result<Value, BackendError> {
            self.record(&format!("prompt:{name}"));
            Ok(json!({"messages": []}))
        }

        async fn reporter_set_upload(
            &self,
            _url: &str,
            _token: Option<&str>,
        ) -> Result<(), BackendError> {
            self.record("reporter_set_upload");
            Ok(())
        }

        async fn reporter_flush(&self) -> Result<(), BackendError> {
            self.record("reporter_flush");
            Ok(())
        }

        async fn cache_set_mode(&self, _mode: &str) -> Result<(), BackendError> {
            self.record("cache_set_mode");
            Ok(())
        }

        async fn cassette_set_mode(&self, _mode: &str) -> Result<(), BackendError> {
            self.record("cassette_set_mode");
            Ok(())
        }

        async fn shutdown(&self) -> Result<(), BackendError> {
            self.record("shutdown");
            Ok(())
        }
    }

    fn make_request(method: &str, params: Value) -> IpcRequest {
        IpcRequest {
            jsonrpc: "2.0".into(),
            id: Some(json!(1)),
            method: method.into(),
            params: Some(params),
        }
    }

    #[tokio::test]
    async fn routes_initialize_through_backend() {
        let dispatcher = Dispatcher::new(StubBackend::new());
        let response = dispatcher
            .dispatch(make_request("mcp.initialize", json!({})))
            .await;
        match response.body {
            IpcResponseBody::Success { result } => {
                assert_eq!(result["ipc_version"], "v1");
                assert_eq!(result["server"]["name"], "stub");
                assert_eq!(result["ready"], true);
            }
            IpcResponseBody::Failure { error } => panic!("expected success, got {error:?}"),
        }
    }

    #[tokio::test]
    async fn missing_required_param_fails_with_invalid_params() {
        let dispatcher = Dispatcher::new(StubBackend::new());
        let response = dispatcher
            .dispatch(make_request("mcp.call", json!({})))
            .await;
        match response.body {
            IpcResponseBody::Failure { error } => {
                assert_eq!(error.code, error_codes::INVALID_PARAMS);
            }
            IpcResponseBody::Success { result } => panic!("expected failure, got {result:?}"),
        }
    }

    #[tokio::test]
    async fn unknown_method_returns_method_not_found() {
        let dispatcher = Dispatcher::new(StubBackend::new());
        let response = dispatcher
            .dispatch(make_request("mcp.unknown", json!({})))
            .await;
        match response.body {
            IpcResponseBody::Failure { error } => {
                assert_eq!(error.code, error_codes::METHOD_NOT_FOUND);
            }
            IpcResponseBody::Success { result } => panic!("expected failure, got {result:?}"),
        }
    }
}
