//! Stdio man-in-the-middle recorder behind `mcptest record` (WOR-1264).
//!
//! A developer points their MCP client (Claude Code, Cursor) at
//! `mcptest record --target <real server>` instead of the real server. The
//! proxy speaks NDJSON JSON-RPC on its own stdin/stdout, forwards every
//! envelope verbatim to the real target (ids untouched, notifications both
//! ways, no second handshake), and tees request/response pairs through the
//! same [`RecordingTransport`] the cassette refresher uses. After each
//! completed exchange the cassette file is rewritten crash-safe (write to
//! `<output>.tmp`, then rename), so an abrupt client exit never loses
//! finished work. The recording then feeds `mcptest distill`.

use std::path::{Path, PathBuf};

use tokio::io::{AsyncBufRead, AsyncBufReadExt, AsyncWrite, AsyncWriteExt};

pub use crate::cassette::LiveTarget;
use crate::cassette::{
    finalize_recording, live::connect_live, Cassette, META_SOURCE_COMMAND, META_SOURCE_URL,
};
use crate::network::ProxyConfig;
use crate::protocol::{JsonRpcMessage, Transport};
use crate::transport::{RecordingHandle, RecordingTransport};

/// What to record and where to put it.
///
/// A plain-data options struct so the CLI handler stays a thin translation
/// layer: it parses flags into this and hands off to [`run_stdio_front`].
#[derive(Debug, Clone)]
pub struct ProxyOptions {
    /// Real MCP server the proxy forwards to: a stdio command line or a
    /// streamable-http URL.
    pub target: LiveTarget,
    /// Cassette file to write. An existing parseable cassette here is a
    /// seed: its exchanges are kept and new ones appended, because clients
    /// like Claude Code restart servers between sessions.
    pub output: PathBuf,
    /// Egress proxy configuration for HTTP targets, from the global
    /// `--proxy` flags.
    pub proxy: ProxyConfig,
}

/// Errors raised while proxying. Typed so the CLI can attach the right
/// operator hint per failure (start the target, check the path, free disk).
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum ProxyError {
    /// The record target could not be spawned or connected.
    #[error("could not connect to the record target: {detail}")]
    Connect {
        /// Transport-level failure detail.
        detail: String,
    },
    /// Reading from or writing to the client side (the proxy's own
    /// stdin/stdout) failed.
    #[error("client side of the record proxy failed: {source}")]
    FrontIo {
        /// Underlying IO error.
        #[source]
        source: std::io::Error,
    },
    /// The target server closed its transport while the proxy still had
    /// traffic to forward.
    #[error("the record target closed its transport while a request was in flight")]
    BackClosed,
    /// The cassette file could not be written.
    #[error("could not write cassette {path}: {source}")]
    Write {
        /// Cassette (or its `.tmp` sibling) that failed to write.
        path: PathBuf,
        /// Underlying IO error.
        #[source]
        source: std::io::Error,
    },
}

/// What a finished recording session produced, for the CLI's summary line.
#[derive(Debug, Clone)]
pub struct ProxySummary {
    /// Total exchanges on disk, seed included.
    pub exchanges: usize,
    /// Cassette file the session wrote.
    pub path: PathBuf,
}

/// Run one stdio-front recording session until the client or the target
/// hangs up, then return what landed on disk.
///
/// Generic over the front streams so tests can drive a whole session over
/// [`tokio::io::duplex`] instead of real process pipes. The CLI passes
/// `BufReader::new(tokio::io::stdin())` and `tokio::io::stdout()`.
pub async fn run_stdio_front<R, W>(
    opts: &ProxyOptions,
    front_in: R,
    front_out: W,
) -> Result<ProxySummary, ProxyError>
where
    R: AsyncBufRead + Unpin,
    W: AsyncWrite + Unpin,
{
    let mut seed = load_or_new_cassette(&opts.output);
    stamp_source_meta(&mut seed, &opts.target);
    let (transport, guard) = connect_live(opts.target.clone(), &opts.proxy)
        .map_err(|detail| ProxyError::Connect { detail })?;
    let result = drive(transport, seed, &opts.output, front_in, front_out).await;
    guard.shutdown().await;
    result
}

/// Everything between connect and shutdown: wrap the back transport in the
/// recording tap, pump until either side closes, then write the final tape.
///
/// Split from [`run_stdio_front`] so unit tests can inject a hand-built
/// channel-pair back transport instead of spawning a real target.
async fn drive<R, W>(
    back: Transport,
    seed: Cassette,
    output: &Path,
    front_in: R,
    mut front_out: W,
) -> Result<ProxySummary, ProxyError>
where
    R: AsyncBufRead + Unpin,
    W: AsyncWrite + Unpin,
{
    let (tap, handle) = RecordingTransport::wrap(back);
    let pumped = pump(tap, &handle, &seed, output, front_in, &mut front_out).await;
    let recorded = handle.finish().await;
    // Surface a pump failure only after the tap is stopped, so the final
    // persist below never races the forwarding task.
    pumped?;
    let exchanges = persist(recorded, &seed, output)?;
    Ok(ProxySummary {
        exchanges,
        path: output.to_path_buf(),
    })
}

/// The forwarding loop: front line in, envelope to the back; back envelope
/// in, line to the front, with a cassette rewrite after each response.
async fn pump<R, W>(
    tap: Transport,
    handle: &RecordingHandle,
    seed: &Cassette,
    output: &Path,
    front_in: R,
    front_out: &mut W,
) -> Result<(), ProxyError>
where
    R: AsyncBufRead + Unpin,
    W: AsyncWrite + Unpin,
{
    let Transport {
        outbound_tx: to_back,
        inbound_rx: mut from_back,
    } = tap;
    let mut lines = front_in.lines();
    loop {
        tokio::select! {
            line = lines.next_line() => {
                let raw = line.map_err(|source| ProxyError::FrontIo { source })?;
                let Some(raw) = raw else {
                    break; // client closed its end: the session is over
                };
                if raw.trim().is_empty() {
                    continue;
                }
                match serde_json::from_str::<JsonRpcMessage>(&raw) {
                    Ok(message) => {
                        if to_back.send(message).is_err() {
                            return Err(ProxyError::BackClosed);
                        }
                    }
                    // Mirror the mock server's stdio loop: a malformed line
                    // earns a local -32700 reply and never reaches the
                    // target or the tape.
                    Err(err) => write_line(front_out, &parse_error_line(&err)).await?,
                }
            }
            inbound = from_back.recv() => {
                let Some(message) = inbound else {
                    break; // target hung up; everything completed is taped
                };
                // Persist before forwarding so any response the client saw
                // is already on disk, whatever happens next.
                if message.id.is_some() && message.method.is_none() {
                    persist(handle.snapshot(), seed, output)?;
                }
                // A parsed envelope always re-serializes; fold the
                // impossible failure into the front-IO error.
                let line = serde_json::to_string(&message).map_err(|e| ProxyError::FrontIo {
                    source: std::io::Error::other(e),
                })?;
                write_line(front_out, &line).await?;
            }
        }
    }
    Ok(())
}

/// Seed the session from an existing cassette so recordings accumulate
/// across client restarts (Claude Code respawns its MCP servers per
/// session); a missing file starts a fresh tape.
fn load_or_new_cassette(path: &Path) -> Cassette {
    let Ok(raw) = std::fs::read_to_string(path) else {
        return Cassette::new();
    };
    match serde_json::from_str(&raw) {
        Ok(cassette) => cassette,
        Err(err) => {
            // A corrupt file must not block a recording session; the
            // crash-safe rewrite replaces it wholesale. Warn so the
            // operator knows the old contents were not appended to.
            tracing::warn!(
                path = %path.display(),
                %err,
                "existing cassette did not parse; starting a fresh recording"
            );
            Cassette::new()
        }
    }
}

/// Stamp the live target into the cassette's `meta` block so the recording
/// remembers where it came from: that is what `mcptest run
/// --update-cassettes` re-records from later.
fn stamp_source_meta(cassette: &mut Cassette, target: &LiveTarget) {
    match target {
        LiveTarget::Command(argv) => {
            cassette
                .meta
                .insert(META_SOURCE_COMMAND.to_string(), shell_words::join(argv));
        }
        LiveTarget::Url(url) => {
            cassette
                .meta
                .insert(META_SOURCE_URL.to_string(), url.clone());
        }
    }
}

/// Write seed plus freshly recorded exchanges to disk, crash-safe.
///
/// The whole file is rewritten through `<output>.tmp` plus rename so a kill
/// mid-write leaves the previous complete tape, never a truncated one.
/// Returns the total exchange count on disk.
fn persist(fresh: Cassette, seed: &Cassette, output: &Path) -> Result<usize, ProxyError> {
    let write_err = |path: &Path, source: std::io::Error| ProxyError::Write {
        path: path.to_path_buf(),
        source,
    };
    let mut combined = Cassette::new();
    combined.exchanges = seed.exchanges.clone();
    combined.exchanges.extend(fresh.exchanges);
    let finished = finalize_recording(combined, &seed.meta);
    // A cassette always serializes (plain data, string keys); fold the
    // impossible failure into the write error rather than mint a variant.
    let mut raw = serde_json::to_string_pretty(&finished)
        .map_err(|e| write_err(output, std::io::Error::other(e)))?;
    raw.push('\n');
    if let Some(parent) = output.parent() {
        if !parent.as_os_str().is_empty() {
            std::fs::create_dir_all(parent).map_err(|e| write_err(output, e))?;
        }
    }
    let mut tmp = output.as_os_str().to_owned();
    tmp.push(".tmp");
    let tmp = PathBuf::from(tmp);
    std::fs::write(&tmp, &raw).map_err(|e| write_err(&tmp, e))?;
    std::fs::rename(&tmp, output).map_err(|e| write_err(output, e))?;
    Ok(finished.exchanges.len())
}

/// Local JSON-RPC parse-error reply (`-32700`, null id), matching what the
/// mock server emits for an unparseable line.
fn parse_error_line(err: &serde_json::Error) -> String {
    serde_json::json!({
        "jsonrpc": "2.0",
        "id": serde_json::Value::Null,
        "error": { "code": -32700, "message": format!("parse error: {err}") },
    })
    .to_string()
}

/// One NDJSON line to the client, flushed so the client never waits on a
/// buffered response.
async fn write_line<W>(out: &mut W, line: &str) -> Result<(), ProxyError>
where
    W: AsyncWrite + Unpin,
{
    let front_io = |source| ProxyError::FrontIo { source };
    out.write_all(line.as_bytes()).await.map_err(front_io)?;
    out.write_all(b"\n").await.map_err(front_io)?;
    out.flush().await.map_err(front_io)
}

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

    /// Hand-built back transport, mirroring the record.rs test style: the
    /// test holds the "real server" ends of the channel pair.
    fn back_pair() -> (
        Transport,
        mpsc::UnboundedReceiver<JsonRpcMessage>,
        mpsc::UnboundedSender<JsonRpcMessage>,
    ) {
        let (real_out_tx, server_rx) = mpsc::unbounded_channel::<JsonRpcMessage>();
        let (server_tx, real_in_rx) = mpsc::unbounded_channel::<JsonRpcMessage>();
        (
            Transport::new(real_out_tx, real_in_rx),
            server_rx,
            server_tx,
        )
    }

    /// Spawn `drive` over a duplex front. Returns the session task plus the
    /// test-side writer (client -> proxy) and line reader (proxy -> client).
    fn spawn_session(
        back: Transport,
        seed: Cassette,
        output: PathBuf,
    ) -> (
        tokio::task::JoinHandle<Result<ProxySummary, ProxyError>>,
        tokio::io::WriteHalf<tokio::io::DuplexStream>,
        tokio::io::Lines<BufReader<tokio::io::ReadHalf<tokio::io::DuplexStream>>>,
    ) {
        let (test_side, proxy_side) = tokio::io::duplex(4096);
        let (proxy_read, proxy_write) = tokio::io::split(proxy_side);
        let (test_read, test_write) = tokio::io::split(test_side);
        let task = tokio::spawn(async move {
            drive(back, seed, &output, BufReader::new(proxy_read), proxy_write).await
        });
        (task, test_write, BufReader::new(test_read).lines())
    }

    async fn send_line<W: AsyncWrite + Unpin>(w: &mut W, line: &str) {
        w.write_all(line.as_bytes()).await.unwrap();
        w.write_all(b"\n").await.unwrap();
        w.flush().await.unwrap();
    }

    fn read_cassette(path: &Path) -> Cassette {
        serde_json::from_str(&std::fs::read_to_string(path).unwrap()).unwrap()
    }

    #[tokio::test]
    async fn forwards_requests_and_responses_preserving_string_ids() {
        let dir = tempfile::tempdir().unwrap();
        let output = dir.path().join("tape.json");
        let (back, mut server_rx, server_tx) = back_pair();
        let (task, mut front, mut lines) = spawn_session(back, Cassette::new(), output.clone());

        send_line(
            &mut front,
            r#"{"jsonrpc":"2.0","id":"abc","method":"tools/list","params":{}}"#,
        )
        .await;
        let forwarded = server_rx.recv().await.expect("request reaches the target");
        assert_eq!(forwarded.id, Some(JsonRpcId::String("abc".into())));
        assert_eq!(forwarded.method.as_deref(), Some("tools/list"));

        server_tx
            .send(JsonRpcMessage::response(
                JsonRpcId::String("abc".into()),
                json!({"tools": []}),
            ))
            .unwrap();
        let line = lines.next_line().await.unwrap().expect("response line");
        let value: serde_json::Value = serde_json::from_str(&line).unwrap();
        assert_eq!(value["id"], "abc", "id must come back untouched");

        front.shutdown().await.unwrap();
        let summary = task.await.unwrap().unwrap();
        assert_eq!(summary.exchanges, 1);
        assert_eq!(summary.path, output);
    }

    #[tokio::test]
    async fn notifications_flow_both_directions_unrecorded() {
        let dir = tempfile::tempdir().unwrap();
        let output = dir.path().join("tape.json");
        let (back, mut server_rx, server_tx) = back_pair();
        let (task, mut front, mut lines) = spawn_session(back, Cassette::new(), output.clone());

        send_line(
            &mut front,
            r#"{"jsonrpc":"2.0","method":"notifications/initialized","params":{}}"#,
        )
        .await;
        let forwarded = server_rx.recv().await.expect("notification forwarded");
        assert_eq!(forwarded.id, None);

        server_tx
            .send(JsonRpcMessage::notification(
                "notifications/message",
                json!({"level": "info"}),
            ))
            .unwrap();
        let line = lines
            .next_line()
            .await
            .unwrap()
            .expect("server notification");
        let value: serde_json::Value = serde_json::from_str(&line).unwrap();
        assert_eq!(value["method"], "notifications/message");

        front.shutdown().await.unwrap();
        let summary = task.await.unwrap().unwrap();
        assert_eq!(summary.exchanges, 0, "notifications have no exchange");
        assert!(read_cassette(&output).exchanges.is_empty());
    }

    #[tokio::test]
    async fn unparseable_line_gets_parse_error_and_is_not_recorded() {
        let dir = tempfile::tempdir().unwrap();
        let output = dir.path().join("tape.json");
        let (back, mut server_rx, _server_tx) = back_pair();
        let (task, mut front, mut lines) = spawn_session(back, Cassette::new(), output.clone());

        send_line(&mut front, "this is not json").await;
        let line = lines.next_line().await.unwrap().expect("local error reply");
        let value: serde_json::Value = serde_json::from_str(&line).unwrap();
        assert_eq!(value["error"]["code"], -32700);
        assert!(value["id"].is_null());

        front.shutdown().await.unwrap();
        let summary = task.await.unwrap().unwrap();
        assert_eq!(summary.exchanges, 0);
        assert!(
            server_rx.try_recv().is_err(),
            "a malformed line must never reach the target"
        );
    }

    #[tokio::test]
    async fn cassette_file_is_rewritten_after_each_completed_exchange() {
        let dir = tempfile::tempdir().unwrap();
        let output = dir.path().join("tape.json");
        let (back, mut server_rx, server_tx) = back_pair();
        let (task, mut front, mut lines) = spawn_session(back, Cassette::new(), output.clone());

        for (id, expected_on_disk) in [(1, 1), (2, 2)] {
            send_line(
                &mut front,
                &format!(
                    r#"{{"jsonrpc":"2.0","id":{id},"method":"tools/call","params":{{"name":"x"}}}}"#
                ),
            )
            .await;
            let _ = server_rx.recv().await.unwrap();
            server_tx
                .send(JsonRpcMessage::response(JsonRpcId::Number(id), json!({})))
                .unwrap();
            let _ = lines.next_line().await.unwrap().unwrap();
            // The response was forwarded, so the exchange is already on
            // disk: that is the crash-safety contract.
            assert_eq!(read_cassette(&output).exchanges.len(), expected_on_disk);
        }

        front.shutdown().await.unwrap();
        let summary = task.await.unwrap().unwrap();
        assert_eq!(summary.exchanges, 2);
    }

    #[tokio::test]
    async fn existing_cassette_is_appended_not_overwritten() {
        let dir = tempfile::tempdir().unwrap();
        let output = dir.path().join("tape.json");
        let mut existing = Cassette::new();
        existing.push(Exchange {
            request: JsonRpcMessage::request(JsonRpcId::Number(1), "initialize", json!({})),
            response: JsonRpcMessage::response(JsonRpcId::Number(1), json!({"capabilities": {}})),
        });
        std::fs::write(&output, serde_json::to_string_pretty(&existing).unwrap()).unwrap();

        let seed = load_or_new_cassette(&output);
        let (back, mut server_rx, server_tx) = back_pair();
        let (task, mut front, mut lines) = spawn_session(back, seed, output.clone());

        send_line(
            &mut front,
            r#"{"jsonrpc":"2.0","id":7,"method":"tools/call","params":{"name":"echo"}}"#,
        )
        .await;
        let _ = server_rx.recv().await.unwrap();
        server_tx
            .send(JsonRpcMessage::response(
                JsonRpcId::Number(7),
                json!({"ok": true}),
            ))
            .unwrap();
        let _ = lines.next_line().await.unwrap().unwrap();
        front.shutdown().await.unwrap();

        let summary = task.await.unwrap().unwrap();
        assert_eq!(summary.exchanges, 2, "seed exchange plus the new one");
        let taped = read_cassette(&output);
        assert_eq!(
            taped.exchanges[0].request.method.as_deref(),
            Some("initialize")
        );
        assert_eq!(
            taped.exchanges[1].request.method.as_deref(),
            Some("tools/call")
        );
    }

    #[tokio::test]
    async fn target_url_stamps_source_url_meta() {
        let dir = tempfile::tempdir().unwrap();
        let output = dir.path().join("tape.json");
        let mut seed = Cassette::new();
        stamp_source_meta(
            &mut seed,
            &LiveTarget::Url("http://localhost:7777/mcp".into()),
        );

        let (back, _server_rx, _server_tx) = back_pair();
        let (task, mut front, _lines) = spawn_session(back, seed, output.clone());
        front.shutdown().await.unwrap();
        let _ = task.await.unwrap().unwrap();

        let taped = read_cassette(&output);
        assert_eq!(
            taped.meta.get(META_SOURCE_URL).map(String::as_str),
            Some("http://localhost:7777/mcp")
        );
    }
}
