//! Does a command that outlives the answer already given for it report back?
//!
//! A `start_print` is acknowledged the moment it is accepted, because the
//! transfer takes minutes and the cloud gives up after ten seconds. So its
//! real outcome can only ever arrive late, and before the outbound queue
//! existed there was nowhere for it to arrive from: the uplink's `select!`
//! loop was the only writer and a spawned task could not reach the socket.
//! A failed start therefore reported success and the reason stayed in this
//! gateway's log, on a machine in someone else's workshop.
//!
//! This stands up a real websocket server and runs the real `uplink::run`
//! against it. **It touches no printer and makes no outbound request**: the
//! fleet is empty, and `dispatch` checks the fleet for the printer id
//! *before* downloading anything, so the command fails on a lookup.

use std::sync::Arc;
use std::time::Duration;

use futures_util::{SinkExt, StreamExt};
use serde_json::{json, Value};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpListener;
use tokio::sync::Mutex;
use tokio_tungstenite::WebSocketStream;
use tokio_tungstenite::tungstenite::Message;
use tokio_tungstenite::tungstenite::handshake::server::{Request, Response};

use continuum_proxy::config::Config;
use continuum_proxy::uplink;

/// Long enough for a loopback round trip, short enough that a regression
/// fails the test instead of hanging the suite.
const WAIT: Duration = Duration::from_secs(10);

fn config(port: u16) -> Config {
    Config {
        gateway_id: "test-gw".into(),
        farm_id: "test-farm".into(),
        gateway_token: "test-token".into(),
        uplink_url: format!("ws://127.0.0.1:{port}/ws/edge/v1"),
        // Nothing here starts go2rtc or reads the fleet file - `uplink::run`
        // is handed its fleet rather than loading one - but Config has no
        // Default, so these have to say something.
        go2rtc_bin: "/nonexistent-go2rtc".into(),
        go2rtc_config: "/nonexistent/go2rtc.yaml".into(),
        go2rtc_api_listen: "127.0.0.1:1984".into(),
        stream_base_url: None,
        printers_path: "/nonexistent/printers.toml".into(),
        store_path: "/nonexistent/continuum.db".into(),
        // Off, which is the default and also the only safe thing in a test:
        // an enabled one would sweep whatever network the test host is on.
        discovery: Default::default(),
    }
}

/// The next frame as JSON. Heartbeats and telemetry share this socket, so
/// callers filter; `interval` fires on its first tick, which means a
/// heartbeat arrives immediately after the hello.
async fn next_json<S>(ws: &mut WebSocketStream<S>) -> Value
where
    S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin,
{
    loop {
        let frame = tokio::time::timeout(WAIT, ws.next())
            .await
            .expect("timed out waiting for a frame from the gateway")
            .expect("the gateway closed the connection")
            .expect("websocket error");
        if let Message::Text(text) = frame {
            return serde_json::from_str(&text).expect("the gateway sent invalid JSON");
        }
    }
}

/// Plays the cloud's `/auth/edge-token`: reads one HTTP request, checks that
/// the gateway sent its farm and long-lived token, and answers with a JWT.
/// Returns the request text so the caller can assert on it.
async fn answer_token_exchange(listener: &TcpListener) -> String {
    let (mut stream, _) = listener.accept().await.unwrap();
    let mut request = Vec::new();
    let mut buf = [0u8; 4096];
    loop {
        let n = tokio::time::timeout(WAIT, stream.read(&mut buf)).await.unwrap().unwrap();
        assert!(n > 0, "the gateway hung up before finishing its token request");
        request.extend_from_slice(&buf[..n]);
        let text = String::from_utf8_lossy(&request).into_owned();
        if let Some((head, body)) = text.split_once("\r\n\r\n") {
            let length = head
                .lines()
                .find_map(|l| l.to_ascii_lowercase().strip_prefix("content-length:").map(|v| v.trim().parse::<usize>().unwrap()))
                .unwrap_or(0);
            if body.len() >= length {
                break;
            }
        }
    }
    let body = r#"{"token":"minted-edge-jwt","expiresInSeconds":3600}"#;
    stream
        .write_all(format!("HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", body.len()).as_bytes())
        .await
        .unwrap();
    String::from_utf8_lossy(&request).into_owned()
}

#[tokio::test]
async fn a_long_command_that_fails_reports_its_reason_after_the_fact() {
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let port = listener.local_addr().unwrap().port();

    let fleet: uplink::SharedFleet = Arc::new(Mutex::new(Vec::new()));
    let (outbound, inbox) = uplink::channel();
    tokio::spawn(uplink::run(config(port), fleet, None, outbound, inbox));

    // The gateway trades its long-lived token for an edge JWT first, and the
    // JWT - not the token it was configured with - is what opens the socket.
    let exchange = answer_token_exchange(&listener).await;
    assert!(exchange.starts_with("POST /auth/edge-token "), "{exchange}");
    assert!(exchange.contains(r#""farmId":"test-farm""#) && exchange.contains(r#""edgeProxyToken":"test-token""#), "{exchange}");

    let (stream, _) = listener.accept().await.unwrap();
    let mut upgrade_auth = String::new();
    let mut ws = tokio_tungstenite::accept_hdr_async(stream, |request: &Request, response: Response| {
        upgrade_auth = request.headers().get("authorization").unwrap().to_str().unwrap().to_string();
        Ok(response)
    })
    .await
    .unwrap();
    assert_eq!(upgrade_auth, "Bearer minted-edge-jwt", "the socket must be opened with the minted JWT");

    let hello = next_json(&mut ws).await;
    assert_eq!(hello["type"], "hello", "a gateway introduces itself first");

    ws.send(Message::Text(
        json!({
            "type": "command",
            "command_id": "cmd-1",
            "printer_id": "not-in-this-fleet",
            "command": {
                "action": "start_print",
                "job_id": "7c9e6679-0000-4000-8000-000000000000",
                "file_name": "cube.gcode.3mf",
                // Never fetched: the fleet lookup fails first. Port 1 so a
                // regression that *did* reach the network fails fast and
                // loudly rather than hanging on a real host.
                "url": "http://127.0.0.1:1/never-fetched",
            }
        })
        .to_string(),
    ))
    .await
    .unwrap();

    let mut results = Vec::new();
    while results.len() < 2 {
        let msg = next_json(&mut ws).await;
        if msg["type"] == "command_result" {
            results.push(msg);
        }
    }

    // Two results for one command, in this order, and that is the contract -
    // not a bug to be deduplicated away. The first is "accepted, working on
    // it", which is all the cloud ever trusted a start reply to mean; the
    // second is what actually happened.
    assert_eq!(results[0]["command_id"], "cmd-1");
    assert_eq!(results[0]["ok"], true, "a start is acknowledged on acceptance");
    assert!(results[0]["error"].is_null());

    assert_eq!(results[1]["command_id"], "cmd-1", "the late result reuses the id");
    assert_eq!(results[1]["ok"], false);

    // The reason is the entire point of the queue: without it the cloud sees
    // a job that never started printing and can say nothing about why.
    let reason = results[1]["error"].as_str().expect("a failure carries its reason");
    assert!(
        reason.contains("not-in-this-fleet"),
        "the reason should name the printer that could not be found, got: {reason}"
    );
}