use crate::acp::{
    deserialize_message, AcpClient, AcpClientError, AcpClientOptions, AcpClientProcessState,
    InboundRequestHandler, InboundRequestOutcome, JsonRpcError, JsonRpcId, JsonRpcMessage,
    JsonRpcNotification, JsonRpcRequest, JsonRpcResponse,
};
use serde_json::{json, Value};
use std::collections::BTreeMap;
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::io::{split, AsyncBufReadExt, AsyncWriteExt, BufReader, DuplexStream};

fn new_client(
    options: AcpClientOptions,
) -> (
    AcpClient,
    tokio::io::Lines<BufReader<tokio::io::ReadHalf<DuplexStream>>>,
    tokio::io::WriteHalf<DuplexStream>,
) {
    let (client_stream, server_stream) = tokio::io::duplex(8 * 1024);
    let (client_reader, client_writer) = split(client_stream);
    let (server_reader, server_writer) = split(server_stream);
    let client = AcpClient::new(client_reader, client_writer, options);
    (client, BufReader::new(server_reader).lines(), server_writer)
}

async fn read_message(
    reader: &mut tokio::io::Lines<BufReader<tokio::io::ReadHalf<DuplexStream>>>,
) -> JsonRpcMessage {
    let line = reader
        .next_line()
        .await
        .expect("read line")
        .expect("line should exist");
    deserialize_message(&line).expect("decode json-rpc line")
}

async fn write_raw(writer: &mut tokio::io::WriteHalf<DuplexStream>, line: &str) {
    writer
        .write_all(line.as_bytes())
        .await
        .expect("write raw line");
    writer.flush().await.expect("flush raw line");
}

async fn write_message(writer: &mut tokio::io::WriteHalf<DuplexStream>, message: &JsonRpcMessage) {
    let encoded = crate::acp::serialize_message(message).expect("encode json-rpc");
    write_raw(writer, &encoded).await;
}

async fn recv_notification(
    receiver: &mut tokio::sync::broadcast::Receiver<JsonRpcNotification>,
) -> JsonRpcNotification {
    tokio::time::timeout(Duration::from_secs(1), receiver.recv())
        .await
        .expect("notification timeout")
        .expect("receive notification")
}

#[tokio::test(flavor = "current_thread")]
async fn client_correlates_responses_and_forwards_notifications() {
    let (client, mut reader, mut writer) = new_client(AcpClientOptions::default());
    let mut notifications = client.subscribe_notifications();

    let request_task = tokio::spawn({
        let client = client.clone();
        async move {
            client
                .request(
                    "session/prompt",
                    Some(json!({ "sessionId": "session-1", "prompt": [{ "type": "text", "text": "hi" }] })),
                )
                .await
        }
    });

    let request = read_message(&mut reader).await;
    match request {
        JsonRpcMessage::Request(message) => {
            assert_eq!(message.method, "session/prompt");
            write_message(
                &mut writer,
                &JsonRpcMessage::Notification(JsonRpcNotification {
                    jsonrpc: String::from("2.0"),
                    method: String::from("session/update"),
                    params: Some(json!({ "status": "thinking" })),
                }),
            )
            .await;
            write_message(
                &mut writer,
                &JsonRpcMessage::Response(JsonRpcResponse::success(
                    message.id,
                    json!({ "status": "complete" }),
                )),
            )
            .await;
        }
        other => panic!("unexpected outbound frame: {other:?}"),
    }

    let notification = recv_notification(&mut notifications).await;
    assert_eq!(notification.method, "session/update");
    assert_eq!(notification.params, Some(json!({ "status": "thinking" })));

    let response = request_task.await.expect("request task").expect("request");
    assert_eq!(response.result(), Some(&json!({ "status": "complete" })));
}

#[tokio::test(flavor = "current_thread")]
async fn client_shims_modern_permission_requests_to_legacy_notifications() {
    let (client, mut reader, mut writer) = new_client(AcpClientOptions::default());
    let mut notifications = client.subscribe_notifications();

    let prompt_task = tokio::spawn({
        let client = client.clone();
        async move {
            client
                .request("session/prompt", Some(json!({ "sessionId": "session-1" })))
                .await
        }
    });

    let prompt_request = read_message(&mut reader).await;
    let prompt_id = match prompt_request {
        JsonRpcMessage::Request(request) => {
            assert_eq!(request.method, "session/prompt");
            request.id
        }
        other => panic!("unexpected prompt frame: {other:?}"),
    };

    write_message(
        &mut writer,
        &JsonRpcMessage::Request(JsonRpcRequest {
            jsonrpc: String::from("2.0"),
            id: JsonRpcId::String(String::from("perm-modern-1")),
            method: String::from("session/request_permission"),
            params: Some(json!({
                "sessionId": "session-1",
                "options": [
                    { "optionId": "allow_once", "kind": "allow_once" },
                    { "optionId": "allow_always", "kind": "allow_always" },
                    { "optionId": "reject_once", "kind": "reject_once" }
                ]
            })),
        }),
    )
    .await;

    let notification = recv_notification(&mut notifications).await;
    assert_eq!(notification.method, "request/permission");
    let params = notification.params.expect("permission params");
    assert_eq!(params["permissionId"], json!("perm-modern-1"));
    assert_eq!(params["_acpMethod"], json!("session/request_permission"));

    let permission_response = client
        .request(
            "request/permission",
            Some(json!({
                "permissionId": "perm-modern-1",
                "reply": "always"
            })),
        )
        .await
        .expect("permission response");
    assert_eq!(
        permission_response.result(),
        Some(&json!({
            "outcome": {
                "outcome": "selected",
                "optionId": "allow_always"
            }
        }))
    );

    let outbound_permission = read_message(&mut reader).await;
    match outbound_permission {
        JsonRpcMessage::Response(response) => {
            assert_eq!(
                response.id,
                JsonRpcId::String(String::from("perm-modern-1"))
            );
            assert_eq!(
                response.result(),
                Some(&json!({
                    "outcome": {
                        "outcome": "selected",
                        "optionId": "allow_always"
                    }
                }))
            );
        }
        other => panic!("unexpected permission response frame: {other:?}"),
    }

    write_message(
        &mut writer,
        &JsonRpcMessage::Response(JsonRpcResponse::success(
            prompt_id,
            json!({ "status": "complete" }),
        )),
    )
    .await;

    let prompt_response = prompt_task
        .await
        .expect("prompt task")
        .expect("prompt response");
    assert_eq!(
        prompt_response.result(),
        Some(&json!({ "status": "complete" }))
    );
}

#[tokio::test(flavor = "current_thread")]
async fn client_normalizes_opencode_style_permission_option_ids() {
    let (client, mut reader, mut writer) = new_client(AcpClientOptions::default());
    let mut notifications = client.subscribe_notifications();

    let prompt_task = tokio::spawn({
        let client = client.clone();
        async move {
            client
                .request("session/prompt", Some(json!({ "sessionId": "session-oc" })))
                .await
        }
    });

    let prompt_request = read_message(&mut reader).await;
    let prompt_id = match prompt_request {
        JsonRpcMessage::Request(request) => request.id,
        other => panic!("unexpected prompt frame: {other:?}"),
    };

    write_message(
        &mut writer,
        &JsonRpcMessage::Request(JsonRpcRequest {
            jsonrpc: String::from("2.0"),
            id: JsonRpcId::String(String::from("perm-opencode-1")),
            method: String::from("session/request_permission"),
            params: Some(json!({
                "sessionId": "session-oc",
                "options": [
                    { "optionId": "once", "kind": "allow_once" },
                    { "optionId": "always", "kind": "allow_always" },
                    { "optionId": "reject", "kind": "reject_once" }
                ]
            })),
        }),
    )
    .await;

    let _ = recv_notification(&mut notifications).await;

    client
        .request(
            "request/permission",
            Some(json!({
                "permissionId": "perm-opencode-1",
                "reply": "always"
            })),
        )
        .await
        .expect("permission response");

    let outbound_permission = read_message(&mut reader).await;
    match outbound_permission {
        JsonRpcMessage::Response(response) => {
            assert_eq!(
                response.result(),
                Some(&json!({
                    "outcome": {
                        "outcome": "selected",
                        "optionId": "always"
                    }
                }))
            );
        }
        other => panic!("unexpected permission response frame: {other:?}"),
    }

    write_message(
        &mut writer,
        &JsonRpcMessage::Response(JsonRpcResponse::success(prompt_id, json!({ "done": true }))),
    )
    .await;

    let prompt_response = prompt_task
        .await
        .expect("prompt task")
        .expect("prompt response");
    assert_eq!(prompt_response.result(), Some(&json!({ "done": true })));
}

#[tokio::test(flavor = "current_thread")]
async fn client_deduplicates_repeated_permission_request_ids() {
    let (client, mut reader, mut writer) = new_client(AcpClientOptions::default());
    let mut notifications = client.subscribe_notifications();

    let prompt_task = tokio::spawn({
        let client = client.clone();
        async move { client.request("session/prompt", Some(json!({}))).await }
    });

    let prompt_request = read_message(&mut reader).await;
    let prompt_id = match prompt_request {
        JsonRpcMessage::Request(request) => request.id,
        other => panic!("unexpected prompt frame: {other:?}"),
    };

    let permission_request = JsonRpcMessage::Request(JsonRpcRequest {
        jsonrpc: String::from("2.0"),
        id: JsonRpcId::String(String::from("perm-dup-1")),
        method: String::from("session/request_permission"),
        params: Some(json!({
            "options": [
                { "optionId": "allow_once", "kind": "allow_once" },
                { "optionId": "reject_once", "kind": "reject_once" }
            ]
        })),
    });
    write_message(&mut writer, &permission_request).await;
    write_message(&mut writer, &permission_request).await;

    let notification = recv_notification(&mut notifications).await;
    assert_eq!(
        notification.params.expect("permission params")["permissionId"],
        json!("perm-dup-1")
    );
    assert!(
        tokio::time::timeout(Duration::from_millis(50), notifications.recv())
            .await
            .is_err()
    );

    client
        .request(
            "request/permission",
            Some(json!({
                "permissionId": "perm-dup-1",
                "reply": "once"
            })),
        )
        .await
        .expect("permission response");

    let outbound_permission = read_message(&mut reader).await;
    match outbound_permission {
        JsonRpcMessage::Response(response) => {
            assert_eq!(response.id, JsonRpcId::String(String::from("perm-dup-1")));
        }
        other => panic!("unexpected permission response frame: {other:?}"),
    }

    write_message(
        &mut writer,
        &JsonRpcMessage::Response(JsonRpcResponse::success(prompt_id, json!({ "done": true }))),
    )
    .await;

    let _ = prompt_task
        .await
        .expect("prompt task")
        .expect("prompt response");
}

#[tokio::test(flavor = "current_thread")]
async fn client_falls_back_to_cancel_notification_when_request_form_is_unsupported() {
    let (client, mut reader, mut writer) = new_client(AcpClientOptions::default());

    let cancel_task = tokio::spawn({
        let client = client.clone();
        async move {
            client
                .request("session/cancel", Some(json!({ "sessionId": "session-1" })))
                .await
        }
    });

    let outbound_request = read_message(&mut reader).await;
    let request_id = match outbound_request {
        JsonRpcMessage::Request(request) => {
            assert_eq!(request.method, "session/cancel");
            request.id
        }
        other => panic!("unexpected cancel request: {other:?}"),
    };

    write_message(
        &mut writer,
        &JsonRpcMessage::Response(JsonRpcResponse::error_response(
            request_id,
            JsonRpcError {
                code: -32601,
                message: String::from("Method not found: session/cancel"),
                data: Some(json!({ "method": "session/cancel" })),
            },
        )),
    )
    .await;

    let fallback = read_message(&mut reader).await;
    match fallback {
        JsonRpcMessage::Notification(notification) => {
            assert_eq!(notification.method, "session/cancel");
            assert_eq!(
                notification.params,
                Some(json!({ "sessionId": "session-1" }))
            );
        }
        other => panic!("unexpected fallback frame: {other:?}"),
    }

    let response = cancel_task
        .await
        .expect("cancel task")
        .expect("cancel response");
    assert_eq!(
        response.result(),
        Some(&json!({
            "cancelled": false,
            "requested": true,
            "via": "notification-fallback"
        }))
    );
}

#[tokio::test(flavor = "current_thread")]
async fn client_timeout_errors_include_recent_activity() {
    let (client, mut reader, mut writer) = new_client(AcpClientOptions {
        timeout: Duration::from_millis(50),
        method_timeouts: BTreeMap::new(),
        request_handler: None,
        process_state_provider: Some(Arc::new(|| AcpClientProcessState {
            exit_code: Some(137),
            killed: Some(true),
        })),
        max_read_line_bytes: 16 * 1024 * 1024,
    });

    let request_task = tokio::spawn({
        let client = client.clone();
        async move {
            client
                .request("session/prompt", Some(json!({ "sessionId": "hang" })))
                .await
        }
    });

    let outbound_request = read_message(&mut reader).await;
    match outbound_request {
        JsonRpcMessage::Request(request) => {
            assert_eq!(request.method, "session/prompt");
        }
        other => panic!("unexpected request frame: {other:?}"),
    }

    write_raw(&mut writer, "[sandbox.require] start node:url /\n").await;
    write_message(
        &mut writer,
        &JsonRpcMessage::Notification(JsonRpcNotification {
            jsonrpc: String::from("2.0"),
            method: String::from("session/update"),
            params: Some(json!({ "status": "thinking" })),
        }),
    )
    .await;

    let error = request_task
        .await
        .expect("request task")
        .expect_err("request should time out");
    let message = error.to_string();
    assert!(message.contains("Recent ACP activity"));
    assert!(message.contains("invalid_json_rpc code=-32700 Parse error"));
    assert!(message.contains("received notification session/update"));
    assert!(message.contains("process exitCode=137"));
    assert!(message.contains("killed=true"));
}

#[tokio::test(flavor = "current_thread")]
async fn client_rejects_adapter_lines_over_configured_limit() {
    let (client, mut reader, mut writer) = new_client(AcpClientOptions {
        timeout: Duration::from_secs(1),
        method_timeouts: BTreeMap::new(),
        request_handler: None,
        process_state_provider: None,
        max_read_line_bytes: 32,
    });

    let request_task = tokio::spawn({
        let client = client.clone();
        async move {
            client
                .request(
                    "session/prompt",
                    Some(json!({ "sessionId": "oversized-line" })),
                )
                .await
        }
    });

    let outbound_request = read_message(&mut reader).await;
    match outbound_request {
        JsonRpcMessage::Request(request) => {
            assert_eq!(request.method, "session/prompt");
        }
        other => panic!("unexpected request frame: {other:?}"),
    }

    write_raw(&mut writer, "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx\n").await;

    let error = request_task
        .await
        .expect("request task")
        .expect_err("oversized line should fail the request");
    assert!(
        matches!(error, AcpClientError::Io(_)),
        "unexpected error: {error:?}"
    );
    assert!(
        error
            .to_string()
            .contains("ACP adapter emitted a line longer than 32 bytes"),
        "unexpected oversized-line error: {error}"
    );
}

#[tokio::test(flavor = "current_thread")]
async fn client_waits_for_exit_drain_before_rejecting_pending_requests() {
    let (client, mut reader, mut writer) = new_client(AcpClientOptions {
        timeout: Duration::from_secs(1),
        method_timeouts: BTreeMap::new(),
        request_handler: None,
        process_state_provider: None,
        max_read_line_bytes: 16 * 1024 * 1024,
    });

    let started_at = Instant::now();
    let request_task = tokio::spawn({
        let client = client.clone();
        async move {
            client
                .request("session/prompt", Some(json!({ "sessionId": "exit" })))
                .await
        }
    });

    let outbound_request = read_message(&mut reader).await;
    match outbound_request {
        JsonRpcMessage::Request(request) => {
            assert_eq!(request.method, "session/prompt");
        }
        other => panic!("unexpected request frame: {other:?}"),
    }

    writer.shutdown().await.expect("shutdown server writer");
    drop(writer);
    drop(reader);

    let error = request_task
        .await
        .expect("request task")
        .expect_err("request should fail after exit");
    assert!(
        matches!(error, AcpClientError::Closed(_)),
        "unexpected error: {error:?}"
    );
    assert!(started_at.elapsed() >= Duration::from_millis(45));
}

#[tokio::test(flavor = "current_thread")]
async fn client_handles_inbound_requests_with_registered_handler() {
    let handler: InboundRequestHandler = Arc::new(|request| {
        Box::pin(async move {
            Ok(Some(InboundRequestOutcome {
                result: Some(json!({
                    "echo": request.params.unwrap_or(Value::Null)
                })),
                error: None,
            }))
        })
    });

    let (client, mut reader, mut writer) = new_client(AcpClientOptions {
        timeout: Duration::from_secs(1),
        method_timeouts: BTreeMap::new(),
        request_handler: Some(handler),
        process_state_provider: None,
        max_read_line_bytes: 16 * 1024 * 1024,
    });
    let mut notifications = client.subscribe_notifications();

    write_message(
        &mut writer,
        &JsonRpcMessage::Request(JsonRpcRequest {
            jsonrpc: String::from("2.0"),
            id: JsonRpcId::Number(41),
            method: String::from("fs/read_text_file"),
            params: Some(json!({ "path": "/workspace/notes.txt" })),
        }),
    )
    .await;

    let notification = recv_notification(&mut notifications).await;
    assert_eq!(notification.method, "fs/read_text_file");
    assert_eq!(
        notification.params,
        Some(json!({
            "path": "/workspace/notes.txt",
            "requestId": 41
        }))
    );

    let response = read_message(&mut reader).await;
    match response {
        JsonRpcMessage::Response(response) => {
            assert_eq!(response.id, JsonRpcId::Number(41));
            assert_eq!(
                response.result(),
                Some(&json!({
                    "echo": {
                        "path": "/workspace/notes.txt"
                    }
                }))
            );
        }
        other => panic!("unexpected inbound response frame: {other:?}"),
    }
}
