//! Cassette replay transport.
//!
//! Serves a recorded [`Cassette`] to the protocol [`Client`] in place of a
//! live server. The per-exchange request matching and normalization already
//! live in [`crate::cassette`]; this module is the thin bridge that turns that
//! pure logic into the channel-pair [`Transport`] the client drives.
//!
//! ## How it works
//!
//! A background task drains the client's outbound queue. For each message:
//!
//! - A **request** (carries an id) is matched by content against the recorded
//!   exchanges via [`replay_with_strategy`]: the first not-yet-consumed
//!   exchange whose method and params match wins, and is consumed so each
//!   recording answers exactly once. On a match the recorded response is
//!   pushed back with its id rewritten to the live request's id (the client
//!   routes responses by id, and recorded ids do not survive across runs).
//! - A **notification** (no id, for example `notifications/initialized`) gets
//!   no response and consumes nothing, mirroring the wire.
//! - A request with **no matching exchange** gets a JSON-RPC error response,
//!   but replay keeps serving later requests, so one unrecorded call fails
//!   only its own test instead of every test sharing the transport.
//!
//! ## Why content matching instead of a cursor (WOR-1283)
//!
//! Replay used to walk a cursor in recorded order. Within one test, calls
//! arrive in order, so a cursor works; but tests sharing one cassette-backed
//! server run in parallel, and their interleaving across tests is
//! nondeterministic, so a multi-call cassette failed intermittently. Matching
//! by content (method plus params, `initialize` by method alone) makes replay
//! order independent. Scanning in recorded order and consuming each exchange
//! once means exact duplicate calls still replay their responses in recorded
//! order. The cassette file format is unchanged.

use tokio::sync::mpsc;
use tokio::task::JoinHandle;

use crate::cassette::{replay_with_strategy, Cassette, ReplayOutcome, ReplayStrategy};
use crate::protocol::{JsonRpcErrorBody, JsonRpcMessage, Transport};

/// JSON-RPC error code for a replay failure (mismatch or exhausted cassette).
/// Sits in the implementation-defined server-error range so it never collides
/// with a code a recorded server might have returned.
const REPLAY_ERROR_CODE: i64 = -32020;

/// Handle to the spawned replay task. Dropping it detaches the task; call
/// [`ReplayHandle::shutdown`] to await a clean stop.
#[derive(Debug)]
pub struct ReplayHandle {
    task: JoinHandle<()>,
}

impl ReplayHandle {
    /// Abort the replay task and wait for it to wind down. The task holds no
    /// OS resources, so an abort is clean; we await the join so shutdown is
    /// observable the way the stdio and HTTP handles are.
    pub async fn shutdown(self) {
        self.task.abort();
        let _ = self.task.await;
    }
}

/// Cassette-backed transport. Replays `cassette` to the client with the given
/// request-matching `strategy`.
#[derive(Debug)]
pub struct ReplayTransport;

impl ReplayTransport {
    /// Spawn the replay task and return the channel-pair transport plus its
    /// handle. The task runs until the client closes its outbound channel or
    /// the handle is shut down; an unmatched request only fails itself.
    pub fn spawn(cassette: Cassette, strategy: ReplayStrategy) -> (Transport, ReplayHandle) {
        let (outbound_tx, mut outbound_rx) = mpsc::unbounded_channel::<JsonRpcMessage>();
        let (inbound_tx, inbound_rx) = mpsc::unbounded_channel::<JsonRpcMessage>();

        let task = tokio::spawn(async move {
            // One slot per recorded exchange. Consuming a slot on match keeps
            // every recording single-use, which is what lets exact duplicate
            // calls fall back to recorded order.
            let mut consumed = vec![false; cassette.exchanges.len()];
            while let Some(msg) = outbound_rx.recv().await {
                // Notifications carry no id and expect no response; swallow
                // them without consuming an exchange, the same as the wire.
                let Some(id) = msg.id.clone() else {
                    continue;
                };
                match find_unconsumed_match(&cassette, &consumed, &msg, strategy) {
                    Some(idx) => {
                        consumed[idx] = true;
                        // The client routes responses by the id it assigned,
                        // not the recorded one. Rewrite before handing back.
                        let mut response = cassette.exchanges[idx].response.clone();
                        response.id = Some(id);
                        if inbound_tx.send(response).is_err() {
                            break;
                        }
                    }
                    None => {
                        // Answer the offending request with an error but keep
                        // replaying: under parallel tests, a sibling's calls
                        // may still match the remaining exchanges.
                        let detail = no_match_detail(&cassette, &consumed, &msg);
                        if inbound_tx.send(error_for(id, detail)).is_err() {
                            break;
                        }
                    }
                }
            }
        });

        (
            Transport::new(outbound_tx, inbound_rx),
            ReplayHandle { task },
        )
    }
}

/// Find the first not-yet-consumed exchange that matches `msg` by content,
/// scanning in recorded order so exact duplicates replay their responses in
/// the order they were recorded. The handshake's params carry the client's
/// version and capabilities, which drift across builds and are not a
/// meaningful replay key, so `initialize` matches on method alone and a
/// cassette survives an mcptest upgrade.
fn find_unconsumed_match(
    cassette: &Cassette,
    consumed: &[bool],
    msg: &JsonRpcMessage,
    strategy: ReplayStrategy,
) -> Option<usize> {
    let live_is_initialize = msg.method.as_deref() == Some("initialize");
    for (idx, exchange) in cassette.exchanges.iter().enumerate() {
        if consumed[idx] {
            continue;
        }
        if live_is_initialize {
            if exchange.request.method.as_deref() == Some("initialize") {
                return Some(idx);
            }
            continue;
        }
        if matches!(
            replay_with_strategy(cassette, idx, msg, strategy),
            ReplayOutcome::Match(_)
        ) {
            return Some(idx);
        }
    }
    None
}

/// Explain why no recorded exchange answered `msg`: either every exchange was
/// already consumed (exhausted) or the remaining ones do not match the live
/// request (mismatch). The wording keeps the `exhausted` / `mismatch` markers
/// operators and tests already key on.
fn no_match_detail(cassette: &Cassette, consumed: &[bool], msg: &JsonRpcMessage) -> String {
    let method = msg.method.as_deref().unwrap_or("(none)");
    let remaining = consumed.iter().filter(|c| !**c).count();
    if remaining == 0 {
        format!(
            "cassette exhausted: all {} recorded exchange(s) already replayed; \
             no recording left for method {method:?}",
            cassette.exchanges.len(),
        )
    } else {
        format!(
            "cassette replay mismatch: none of the {remaining} remaining recorded \
             exchange(s) match method {method:?} with these params\nsee: {}",
            crate::error_anchors::CASSETTE_REPLAY_MISMATCH
        )
    }
}

/// Build a JSON-RPC error response for `id` carrying `message`.
fn error_for(id: crate::protocol::JsonRpcId, message: String) -> JsonRpcMessage {
    JsonRpcMessage::error_response(
        id,
        JsonRpcErrorBody {
            code: REPLAY_ERROR_CODE,
            message,
            data: None,
        },
    )
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::cassette::Exchange;
    use crate::protocol::JsonRpcId;
    use serde_json::json;

    fn cassette_with_handshake_and_tool() -> Cassette {
        let mut c = Cassette::new();
        c.push(Exchange {
            request: JsonRpcMessage::request(
                JsonRpcId::Number(1),
                "initialize",
                json!({"protocolVersion": "2025-06-18"}),
            ),
            response: JsonRpcMessage::response(
                JsonRpcId::Number(1),
                json!({"protocolVersion": "2025-06-18", "capabilities": {}}),
            ),
        });
        c.push(Exchange {
            request: JsonRpcMessage::request(JsonRpcId::Number(2), "tools/list", json!({})),
            response: JsonRpcMessage::response(
                JsonRpcId::Number(2),
                json!({"tools": [{"name": "search"}]}),
            ),
        });
        c
    }

    #[tokio::test]
    async fn replays_recorded_responses_with_live_ids() {
        let (transport, handle) =
            ReplayTransport::spawn(cassette_with_handshake_and_tool(), ReplayStrategy::Semantic);
        let Transport {
            outbound_tx,
            mut inbound_rx,
        } = transport;

        // initialize with a client-assigned id that differs from the recording.
        outbound_tx
            .send(JsonRpcMessage::request(
                JsonRpcId::Number(100),
                "initialize",
                json!({"protocolVersion": "2025-06-18"}),
            ))
            .unwrap();
        let resp = inbound_rx.recv().await.expect("init response");
        assert_eq!(
            resp.id,
            Some(JsonRpcId::Number(100)),
            "id rewritten to live"
        );
        assert!(resp.result.is_some());

        // a notification gets no response and does not consume an exchange.
        outbound_tx
            .send(JsonRpcMessage::notification(
                "notifications/initialized",
                json!({}),
            ))
            .unwrap();

        outbound_tx
            .send(JsonRpcMessage::request(
                JsonRpcId::Number(101),
                "tools/list",
                json!({}),
            ))
            .unwrap();
        let resp = inbound_rx.recv().await.expect("tools/list response");
        assert_eq!(resp.id, Some(JsonRpcId::Number(101)));
        assert_eq!(resp.result.as_ref().unwrap()["tools"][0]["name"], "search");

        handle.shutdown().await;
    }

    #[tokio::test]
    async fn initialize_matches_on_method_alone_despite_param_drift() {
        // Recorded with one client version; replayed by another. The handshake
        // params differ but the cassette must still match so an mcptest upgrade
        // does not invalidate every recording.
        let mut c = Cassette::new();
        c.push(Exchange {
            request: JsonRpcMessage::request(
                JsonRpcId::Number(1),
                "initialize",
                json!({"clientInfo": {"name": "mcptest", "version": "1.0.0"}}),
            ),
            response: JsonRpcMessage::response(
                JsonRpcId::Number(1),
                json!({"protocolVersion": "2025-06-18", "capabilities": {}}),
            ),
        });
        let (transport, handle) = ReplayTransport::spawn(c, ReplayStrategy::Semantic);
        let Transport {
            outbound_tx,
            mut inbound_rx,
        } = transport;
        outbound_tx
            .send(JsonRpcMessage::request(
                JsonRpcId::Number(7),
                "initialize",
                json!({"clientInfo": {"name": "mcptest", "version": "9.9.9"}}),
            ))
            .unwrap();
        let resp = inbound_rx.recv().await.expect("init response");
        assert_eq!(resp.id, Some(JsonRpcId::Number(7)));
        assert!(resp.result.is_some(), "initialize should match on method");
        handle.shutdown().await;
    }

    #[tokio::test]
    async fn exhausted_cassette_returns_an_error_response() {
        let (transport, handle) = ReplayTransport::spawn(Cassette::new(), ReplayStrategy::Semantic);
        let Transport {
            outbound_tx,
            mut inbound_rx,
        } = transport;
        outbound_tx
            .send(JsonRpcMessage::request(
                JsonRpcId::Number(1),
                "initialize",
                json!({}),
            ))
            .unwrap();
        let resp = inbound_rx.recv().await.expect("error response");
        let err = resp.error.expect("error body");
        assert_eq!(err.code, REPLAY_ERROR_CODE);
        assert!(err.message.contains("exhausted"), "got: {}", err.message);
        handle.shutdown().await;
    }

    #[tokio::test]
    async fn method_mismatch_returns_an_error_response() {
        let (transport, handle) =
            ReplayTransport::spawn(cassette_with_handshake_and_tool(), ReplayStrategy::Semantic);
        let Transport {
            outbound_tx,
            mut inbound_rx,
        } = transport;
        // The cassette recorded `initialize` and `tools/list`; send a call
        // it never saw.
        outbound_tx
            .send(JsonRpcMessage::request(
                JsonRpcId::Number(1),
                "tools/call",
                json!({"name": "x"}),
            ))
            .unwrap();
        let resp = inbound_rx.recv().await.expect("error response");
        let err = resp.error.expect("error body");
        assert_eq!(err.code, REPLAY_ERROR_CODE);
        assert!(err.message.contains("mismatch"), "got: {}", err.message);
        assert!(
            err.message
                .contains(crate::error_anchors::CASSETTE_REPLAY_MISMATCH),
            "mismatch error should link the troubleshooting entry, got: {}",
            err.message
        );
        handle.shutdown().await;
    }

    fn tool_exchange(id: i64, args: serde_json::Value, text: &str) -> Exchange {
        Exchange {
            request: JsonRpcMessage::request(
                JsonRpcId::Number(id),
                "tools/call",
                json!({"name": "echo", "arguments": args}),
            ),
            response: JsonRpcMessage::response(
                JsonRpcId::Number(id),
                json!({"content": [{"type": "text", "text": text}]}),
            ),
        }
    }

    #[tokio::test]
    async fn distinct_calls_replay_out_of_recorded_order() {
        // WOR-1283: parallel tests interleave their calls arbitrarily, so
        // the transport must answer by content, not by recorded position.
        let mut c = Cassette::new();
        c.push(tool_exchange(1, json!({"msg": "one"}), "first"));
        c.push(tool_exchange(2, json!({"msg": "two"}), "second"));
        let (transport, handle) = ReplayTransport::spawn(c, ReplayStrategy::Semantic);
        let Transport {
            outbound_tx,
            mut inbound_rx,
        } = transport;

        // Ask for the second recording before the first.
        outbound_tx
            .send(JsonRpcMessage::request(
                JsonRpcId::Number(10),
                "tools/call",
                json!({"name": "echo", "arguments": {"msg": "two"}}),
            ))
            .unwrap();
        let resp = inbound_rx.recv().await.expect("second recording");
        assert_eq!(resp.result.unwrap()["content"][0]["text"], "second");

        outbound_tx
            .send(JsonRpcMessage::request(
                JsonRpcId::Number(11),
                "tools/call",
                json!({"name": "echo", "arguments": {"msg": "one"}}),
            ))
            .unwrap();
        let resp = inbound_rx.recv().await.expect("first recording");
        assert_eq!(resp.result.unwrap()["content"][0]["text"], "first");
        handle.shutdown().await;
    }

    #[tokio::test]
    async fn exact_duplicate_calls_consume_recordings_in_recorded_order() {
        // Two identical requests with different recorded responses: the
        // first live call takes the first recording, the second takes the
        // second, and a third errors because both are consumed.
        let mut c = Cassette::new();
        c.push(tool_exchange(1, json!({"msg": "hi"}), "first"));
        c.push(tool_exchange(2, json!({"msg": "hi"}), "second"));
        let (transport, handle) = ReplayTransport::spawn(c, ReplayStrategy::Semantic);
        let Transport {
            outbound_tx,
            mut inbound_rx,
        } = transport;

        for expected in ["first", "second"] {
            outbound_tx
                .send(JsonRpcMessage::request(
                    JsonRpcId::Number(20),
                    "tools/call",
                    json!({"name": "echo", "arguments": {"msg": "hi"}}),
                ))
                .unwrap();
            let resp = inbound_rx.recv().await.expect("recorded response");
            assert_eq!(resp.result.unwrap()["content"][0]["text"], expected);
        }

        outbound_tx
            .send(JsonRpcMessage::request(
                JsonRpcId::Number(21),
                "tools/call",
                json!({"name": "echo", "arguments": {"msg": "hi"}}),
            ))
            .unwrap();
        let resp = inbound_rx.recv().await.expect("error response");
        let err = resp.error.expect("error body");
        assert!(err.message.contains("exhausted"), "got: {}", err.message);
        handle.shutdown().await;
    }

    #[tokio::test]
    async fn unrecorded_call_fails_alone_and_replay_keeps_serving() {
        // One bad request must not kill the transport for the tests that
        // still hold matching recordings (they share one replay task).
        let (transport, handle) =
            ReplayTransport::spawn(cassette_with_handshake_and_tool(), ReplayStrategy::Semantic);
        let Transport {
            outbound_tx,
            mut inbound_rx,
        } = transport;

        outbound_tx
            .send(JsonRpcMessage::request(
                JsonRpcId::Number(30),
                "tools/call",
                json!({"name": "never-recorded"}),
            ))
            .unwrap();
        let resp = inbound_rx.recv().await.expect("error response");
        assert!(resp.error.is_some(), "unrecorded call errors");

        outbound_tx
            .send(JsonRpcMessage::request(
                JsonRpcId::Number(31),
                "initialize",
                json!({}),
            ))
            .unwrap();
        let resp = inbound_rx.recv().await.expect("init still replays");
        assert!(resp.result.is_some(), "later matching calls still answered");
        handle.shutdown().await;
    }
}
