//! Route harness requests into its V8 inbox and preserve incremental NDJSON replies.
use bytes::Bytes;
use http::{Request, Response};
use http_body_util::{BodyExt, Full};
use serde_json::{Value, json};
use std::{
    pin::Pin,
    task::{Context, Poll},
};
use tokio::sync::mpsc;

struct Body(mpsc::UnboundedReceiver<Value>);
impl http_body::Body for Body {
    type Data = Bytes;
    type Error = wasmtime_wasi_http::Error;
    fn poll_frame(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Result<http_body::Frame<Bytes>, Self::Error>>> {
        self.0.poll_recv(cx).map(|value| {
            value.map(|value| {
                let mut bytes = serde_json::to_vec(&value).expect("harness JSON");
                bytes.push(b'\n');
                Ok(http_body::Frame::data(Bytes::from(bytes)))
            })
        })
    }
}
fn json_response(status: u16, value: Value) -> Response<wasmtime_wasi_http::WasiBody> {
    Response::builder()
        .status(status)
        .header("content-type", "application/json")
        .body(
            Full::new(Bytes::from(value.to_string()))
                .map_err(|never| match never {})
                .boxed_unsync(),
        )
        .unwrap()
}
pub(super) async fn handle(
    api: &super::Api,
    caller: crate::cred::Caller,
    mut request: Request<Bytes>,
) -> Response<wasmtime_wasi_http::WasiBody> {
    if request.method() == http::Method::GET
        && request.uri().path() == "/harness/credential-requests"
    {
        *request.uri_mut() = "/cred/requests?state=pending".parse().unwrap();
        let response = super::handle(api, caller, request).await;
        let mut value: Value = serde_json::from_slice(response.body()).expect("credential JSON");
        if let Some(requests) = value.as_array_mut() {
            for request in requests {
                request["approvalUrl"] =
                    json!(format!("/#approve/{}", request["id"].as_str().unwrap()));
            }
            value = json!({"requests":value});
        }
        return json_response(response.status().as_u16(), value);
    }
    let sender = api.harness.lock().unwrap().clone();
    let Some(sender) = sender else {
        return json_response(503, json!({"error":"harness_unavailable"}));
    };
    let body = if request.body().is_empty() {
        Value::Null
    } else {
        match serde_json::from_slice(request.body()) {
            Ok(body) => body,
            Err(_) => return json_response(400, json!({"error":"invalid_json"})),
        }
    };
    let (reply, mut receiver) = mpsc::unbounded_channel();
    let message = crate::harness::input::Input::Api {
        id: uuid::Uuid::new_v4().to_string(),
        method: request.method().to_string(),
        path: request.uri().path_and_query().unwrap().to_string(),
        body,
        reply,
    };
    if sender.send(message).is_err() {
        return json_response(503, json!({"error":"harness_unavailable"}));
    }
    let Some(first) = receiver.recv().await else {
        return json_response(503, json!({"error":"harness_stopped"}));
    };
    let status = first["status"].as_u64().unwrap_or(500) as u16;
    if first["stream"] == true {
        Response::builder()
            .status(status)
            .header("content-type", "application/x-ndjson")
            .body(Body(receiver).boxed_unsync())
            .unwrap()
    } else {
        json_response(status, first["body"].clone())
    }
}