//! The `jcp-gateway` binary over HTTP (`--listen`), in front of a real server
//! process, reached over a real socket the way a remote caller reaches it:
//! a signed MCP 2026-07-28 request in, a sealed decision out, and a retry
//! after a restart answered from the record without the call running again.
//!
//! The server is awk, as in `stdio_binary.rs`: a request (a line with an
//! `id`) gets one response carrying that id.

use base64::Engine as _;
use ed25519_dalek::SigningKey;
use jcp_bridge::mcp_resource;
use jcp_gateway::{present_grant, JCP_META_NS};
use jcp_schema::{Capability, Did, Grant, SignedGrant};
use serde_json::{json, Value};
use std::io::{BufRead, BufReader, Read, Write};
use std::path::PathBuf;
use std::process::{Child, Command, Stdio};
use std::sync::mpsc::{self, Receiver};
use std::time::{Duration, SystemTime, UNIX_EPOCH};

const ENDPOINT: &str = "https://gw.example/mcp";
const AGENT_URL: &str = "https://agent.example/.well-known/http-message-signatures-directory";
const STEWARD_SEED: [u8; 32] = [0x33; 32];
const DEADLINE: Duration = Duration::from_secs(120);

const ANSWERS_REQUESTS: &str = r#"/"id"/ {
  id = "null"
  if (match($0, /"id":[0-9]+/)) id = substr($0, RSTART + 5, RLENGTH - 5)
  printf "{\"jsonrpc\":\"2.0\",\"id\":%s,\"result\":{\"resultType\":\"complete\",\"content\":[{\"type\":\"text\",\"text\":\"done\"}]}}\n", id
  fflush()
}"#;

/// The awk command line, with mawk told to read lines as they come
/// (`stdio_binary.rs` says why).
fn awk_server(script: &str) -> Vec<String> {
    const AWK: &str = "/usr/bin/awk";
    let is_mawk = Command::new(AWK)
        .args(["-W", "version"])
        .stdin(Stdio::null())
        .output()
        .map(|o| String::from_utf8_lossy(&o.stdout).contains("mawk") || String::from_utf8_lossy(&o.stderr).contains("mawk"))
        .unwrap_or(false);
    let mut argv = vec![AWK.to_string()];
    if is_mawk {
        argv.extend(["-W".to_string(), "interactive".to_string()]);
    }
    argv.push(script.to_string());
    argv
}

fn key(n: u8) -> SigningKey {
    SigningKey::from_bytes(&[n; 32])
}
fn did(k: &SigningKey) -> Did {
    Did::from_verifying_key(&k.verifying_key())
}
fn now() -> u64 {
    SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs()
}

fn path(name: &str, what: &str) -> PathBuf {
    std::path::Path::new(env!("CARGO_TARGET_TMPDIR")).join(format!("jcp-gateway-http-{name}-{}.{what}", std::process::id()))
}

/// The gateway binary serving HTTP on a port the system chose.
struct Running {
    child: Child,
    addr: String,
    log: std::sync::Mutex<Receiver<String>>,
}

impl Running {
    fn start(name: &str) -> Running {
        Running::start_with(name, &[], &awk_server(ANSWERS_REQUESTS))
    }

    fn start_with(name: &str, extra: &[&str], server: &[String]) -> Running {
        Running::start_env(name, extra, server, &[])
    }

    fn start_env(name: &str, extra: &[&str], server: &[String], env: &[(&str, &str)]) -> Running {
        let key_file = path(name, "key");
        let hex: String = STEWARD_SEED.iter().map(|b| format!("{b:02x}")).collect();
        std::fs::write(&key_file, hex).unwrap();
        let mut child = Command::new(env!("CARGO_BIN_EXE_jcp-gateway"))
            .args(["--server", "srv", "--listen", "127.0.0.1:0", "--endpoint", ENDPOINT])
            .arg("--steward-key")
            .arg(&key_file)
            .args(["--trust", &did(&key(1)).to_did_key()])
            .args(extra)
            .arg("--")
            .args(server)
            .env("JCP_GATEWAY_RECEIPTS", path(name, "cose"))
            .env("JCP_GATEWAY_CALLS", path(name, "jsonl"))
            .envs(env.iter().copied())
            .stdin(Stdio::null())
            .stdout(Stdio::null())
            .stderr(Stdio::piped())
            .spawn()
            .expect("spawn the jcp-gateway binary");
        let (tx, log) = mpsc::channel();
        let err = child.stderr.take().unwrap();
        std::thread::spawn(move || {
            for line in BufReader::new(err).lines() {
                let Ok(line) = line else { break };
                if tx.send(line).is_err() {
                    break;
                }
            }
        });
        let mut seen = Vec::new();
        loop {
            match log.recv_timeout(DEADLINE) {
                Ok(line) => {
                    if let Some(addr) = line.split(" over HTTP on ").nth(1) {
                        let addr = addr.split([',', ' ']).next().unwrap_or("").to_string();
                        return Running { child, addr, log: std::sync::Mutex::new(log) };
                    }
                    seen.push(line);
                }
                Err(_) => {
                    let _ = child.kill();
                    let _ = child.wait();
                    panic!("the gateway did not start serving; it said:\n{}", seen.join("\n"));
                }
            }
        }
    }

    /// Send `raw` on a connection of its own and read the whole response.
    fn exchange(&self, raw: &[u8]) -> (u16, String, Value) {
        let mut c = std::net::TcpStream::connect(&self.addr).unwrap();
        c.set_read_timeout(Some(DEADLINE)).unwrap();
        c.write_all(raw).unwrap();
        let mut out = String::new();
        if let Err(e) = c.read_to_string(&mut out) {
            let said: Vec<String> = self.log.lock().unwrap().try_iter().collect();
            panic!("no response ({e}); the gateway said:\n{}", said.join("\n"));
        }
        let (head, body) = out.split_once("\r\n\r\n").unwrap_or((&out, ""));
        let status = head.split(' ').nth(1).and_then(|s| s.parse().ok()).unwrap_or(0);
        (status, head.to_ascii_lowercase(), serde_json::from_str(body).unwrap_or(Value::Null))
    }
}

impl Drop for Running {
    fn drop(&mut self) {
        let _ = self.child.kill();
        let _ = self.child.wait();
    }
}

fn root() -> SignedGrant {
    root_for("echo")
}

fn root_for(tool: &str) -> SignedGrant {
    let g = Grant {
        issuer: did(&key(1)),
        subject: did(&key(2)),
        capability: Capability::new(mcp_resource("srv", tool), [tool]),
        budget_uj: 1_000_000_000_000,
        budget_funds: None,
        caveats: Vec::new(),
        expiry: 0,
        parent: None,
        delegate_depth: None,
        nonce: [1; 16],
    };
    jcp_grant::issue(&key(1), g).unwrap()
}

/// A `tools/call` of `echo`, presenting `chain`, as raw HTTP/1.1, signed by
/// `signer` under `nonce` with `call_id` when given.
fn request(id: u64, chain: &[SignedGrant], signer: Option<&SigningKey>, nonce: &str, call_id: Option<[u8; 32]>) -> Vec<u8> {
    request_tool("echo", id, chain, signer, nonce, call_id)
}

fn request_tool(tool: &str, id: u64, chain: &[SignedGrant], signer: Option<&SigningKey>, nonce: &str, call_id: Option<[u8; 32]>) -> Vec<u8> {
    let body = json!({"jsonrpc": "2.0", "id": id, "method": "tools/call", "params": {
        "name": tool, "arguments": {"text": "hi"},
        "_meta": {
            "io.modelcontextprotocol/protocolVersion": "2026-07-28",
            JCP_META_NS: {"grants": chain.iter().map(present_grant).collect::<Vec<_>>()},
        }}})
    .to_string()
    .into_bytes();
    let mut r = jcp_httpsig::Request::new("POST", ENDPOINT).unwrap();
    for (n, v) in [
        ("Content-Type", "application/json"),
        ("Accept", "application/json, text/event-stream"),
        ("MCP-Protocol-Version", "2026-07-28"),
        ("Mcp-Method", "tools/call"),
        ("Mcp-Name", tool),
    ] {
        r.headers.append(n, v);
    }
    if let Some(k) = signer {
        let created = i64::try_from(now()).unwrap();
        jcp_httpsig::jcp::sign_call(&mut r, &body, "sig1", AGENT_URL, k, created, nonce, call_id.as_ref()).unwrap();
    }
    let mut raw = String::from("POST /mcp HTTP/1.1\r\nHost: gw.example\r\nConnection: close\r\n");
    for (n, v) in r.headers.iter() {
        raw.push_str(&format!("{n}: {v}\r\n"));
    }
    raw.push_str(&format!("Content-Length: {}\r\n\r\n", body.len()));
    let mut out = raw.into_bytes();
    out.extend_from_slice(&body);
    out
}

/// The receipt a response carries, verified under the steward key the
/// gateway was given.
fn receipt(body: &Value) -> jcp_receipt::ReceiptCore {
    let ns = JCP_META_NS.replace('/', "~1");
    let b64 = body
        .pointer(&format!("/result/_meta/{ns}/receipt_b64"))
        .or_else(|| body.pointer(&format!("/error/data/_meta/{ns}/receipt_b64")))
        .and_then(Value::as_str)
        .unwrap_or_else(|| panic!("no receipt in {body}"));
    let env = base64::engine::general_purpose::STANDARD.decode(b64).unwrap();
    let pubkey = SigningKey::from_bytes(&STEWARD_SEED).verifying_key().to_bytes();
    jcp_receipt::extract(&env, "did:jcp:gateway:srv", &pubkey).expect("a receipt the gateway's steward key signed")
}

#[test]
fn a_signed_call_is_decided_sealed_and_answered_and_its_retry_survives_a_restart() {
    let name = "restart";
    for what in ["cose", "jsonl"] {
        let _ = std::fs::remove_file(path(name, what));
    }
    let gw = Running::start(name);
    let chain = [root()];

    let (status, _, body) = gw.exchange(&request(1, &chain, Some(&key(2)), "n1", Some([7; 32])));
    assert_eq!(status, 200, "{body}");
    assert_eq!(body["result"]["content"][0]["text"], "done");
    assert_eq!(receipt(&body).verdict, "allow");

    let (status, head, body) = gw.exchange(&request(2, &chain, None, "", None));
    assert_eq!(status, 403, "{body}");
    assert!(head.contains("accept-signature: sig1=("), "{head}");
    assert_eq!(receipt(&body).deny_reason.as_deref(), Some("bad-signature"));

    drop(gw);
    let sealed = std::fs::metadata(path(name, "cose")).unwrap().len();
    let gw = Running::start(name);

    // The retry: a new request id, a new signature, the same call id.
    let (status, _, retry) = gw.exchange(&request(3, &chain, Some(&key(2)), "n3", Some([7; 32])));
    assert_eq!(status, 200, "{retry}");
    assert_eq!(retry["id"], 3);
    assert_eq!(retry["result"]["content"][0]["text"], "done");
    assert_eq!(std::fs::metadata(path(name, "cose")).unwrap().len(), sealed, "answered from the record: nothing decided, nothing sealed");

    let (status, _, fresh) = gw.exchange(&request(4, &chain, Some(&key(2)), "n4", Some([8; 32])));
    assert_eq!(status, 200, "{fresh}");
    assert!(std::fs::metadata(path(name, "cose")).unwrap().len() > sealed, "a new call is decided and sealed");
}

#[test]
fn a_gateway_that_answers_the_network_needs_a_steward_key_of_its_own() {
    let out = Command::new(env!("CARGO_BIN_EXE_jcp-gateway"))
        .args(["--server", "srv", "--listen", "127.0.0.1:0", "--endpoint", ENDPOINT, "--", "/usr/bin/true"])
        .stdin(Stdio::null())
        .output()
        .unwrap();
    assert_eq!(out.status.code(), Some(2));
    assert!(String::from_utf8_lossy(&out.stderr).contains("--steward-key"));
}

/// A server that answers each request on a process of its own, so answers
/// come back in the order the work finishes: `slow` after two seconds, any
/// other at once.
const ANSWERS_OUT_OF_ORDER: &str = r#"while IFS= read -r l; do
  ( id=$(printf '%s\n' "$l" | sed -n 's/.*"id":\([0-9][0-9]*\).*/\1/p')
    case "$l" in *'"name":"slow"'*) sleep 2 ;; esac
    [ -n "$id" ] && printf '{"jsonrpc":"2.0","id":%s,"result":{"resultType":"complete","content":[{"type":"text","text":"done"}]}}\n' "$id" ) &
done"#;

/// With `--concurrent`, a call does not wait behind another: a fast call
/// sent while a slow one runs is answered first, each is decided and
/// sealed, and the slow one, having run beside the fast, is charged its share
/// of what was read while both ran.
#[test]
fn calls_run_at_once_and_a_call_that_ran_beside_another_is_charged_its_share() {
    let name = "concurrent";
    for what in ["cose", "jsonl"] {
        let _ = std::fs::remove_file(path(name, what));
    }
    let server = vec!["/bin/sh".to_string(), "-c".to_string(), ANSWERS_OUT_OF_ORDER.to_string()];
    let gw = std::sync::Arc::new(Running::start_with(name, &["--concurrent", "4"], &server));
    let slow = {
        let gw = std::sync::Arc::clone(&gw);
        std::thread::spawn(move || {
            let started = std::time::Instant::now();
            let out = gw.exchange(&request_tool("slow", 1, &[root_for("slow")], Some(&key(2)), "s1", Some([1; 32])));
            (out, started.elapsed())
        })
    };
    std::thread::sleep(Duration::from_millis(400));
    let started = std::time::Instant::now();
    let (status, _, fast) = gw.exchange(&request_tool("echo", 2, &[root_for("echo")], Some(&key(2)), "s2", Some([2; 32])));
    let fast_took = started.elapsed();
    assert_eq!(status, 200, "{fast}");
    assert!(fast_took < Duration::from_millis(1500), "the fast call waited behind the slow one: {fast_took:?}");
    let ((status, _, slow_body), slow_took) = slow.join().unwrap();
    assert_eq!(status, 200, "{slow_body}");
    assert!(slow_took >= Duration::from_secs(2));
    assert_eq!((receipt(&fast).verdict, receipt(&slow_body).verdict), ("allow".to_string(), "allow".to_string()));
    let ns = JCP_META_NS.replace('/', "~1");
    assert_eq!(slow_body.pointer(&format!("/result/_meta/{ns}/energy_domain")).and_then(Value::as_str), Some("machine-shared"), "{slow_body}");
}

/// A node of a fleet on `dir`, with a short lease and a follower directory
/// of its own.
fn fleet_node(name: &str, node: &str, dir: &std::path::Path) -> Running {
    fleet_node_env(name, node, dir, &[])
}

fn fleet_node_env(name: &str, node: &str, dir: &std::path::Path, env: &[(&str, &str)]) -> Running {
    let dir = dir.to_str().unwrap().to_string();
    let data = path(name, "follow");
    let _ = std::fs::remove_dir_all(&data);
    let data = data.to_str().unwrap().to_string();
    Running::start_env(
        name,
        &["--fleet", &dir, "--node", node, "--fleet-ttl-ms", "1500", "--fleet-data", &data],
        &awk_server(ANSWERS_REQUESTS),
        env,
    )
}

/// What a call's receipt left the grant, and what it debited.
fn left_and_debited(body: &Value) -> (u64, u64) {
    let r = receipt(body);
    assert_eq!(r.verdict, "allow", "{body}");
    (r.budget_remaining_uj, r.debited_uj.unwrap_or(r.cost_uj))
}

/// Two gateway processes on one store (`--fleet`): the first call takes the
/// record; a call sent to the other node is forwarded to the owner, whole, and
/// charged against the same budget; a retry sent to either is answered from
/// the owner's call record; and once the owner dies and its lease lapses the
/// other node takes the record, resuming its spend and its calls.
///
/// The owner acknowledges through its follower, the other node, and uploads
/// nothing in the test's time: when it is killed its receipts are on the
/// follower's disk and not in the store, so the takeover has them only by
/// sealing the follower and uploading what it holds.
#[test]
fn a_fleet_serves_one_record_from_either_node_and_a_takeover_carries_it() {
    let dir = path("fleet", "store");
    let _ = std::fs::remove_dir_all(&dir);
    let a = fleet_node_env("fleet-a", "a", &dir, &[("JCP_FLEET_UPLOAD_MS", "600000")]);
    let b = fleet_node("fleet-b", "b", &dir);
    let chain = [root()];
    let record = root().grant.id().to_hex();

    let (status, _, first) = a.exchange(&request(1, &chain, Some(&key(2)), "f1", Some([1; 32])));
    assert_eq!(status, 200, "{first}");
    let (left1, debited1) = left_and_debited(&first);
    assert_eq!(left1 + debited1, 1_000_000_000_000);

    let (status, _, second) = b.exchange(&request(2, &chain, Some(&key(2)), "f2", Some([2; 32])));
    assert_eq!(status, 200, "{second}");
    let (left2, debited2) = left_and_debited(&second);
    assert_eq!(left2 + debited2, left1, "the call sent to b was charged against the budget a holds");
    let said: Vec<String> = b.log.lock().unwrap().try_iter().collect();
    assert!(said.iter().any(|l| l.contains("forwarded to the record's owner")), "{}", said.join("\n"));

    let (status, _, retry) = b.exchange(&request(3, &chain, Some(&key(2)), "f3", Some([2; 32])));
    assert_eq!(status, 200, "{retry}");
    assert_eq!(retry["id"], 3);
    assert_eq!(receipt(&retry).budget_remaining_uj, left2, "the retry got the call's own answer, from the owner's call record");

    let in_store = |epoch: u64| {
        std::fs::read_dir(dir.join("rec").join(&record).join(format!("{epoch:020}"))).map_or(0, |d| d.count()) - 1
    };
    assert_eq!(in_store(1), 0, "a's entries are on its follower, not in the store");
    let said: Vec<String> = a.log.lock().unwrap().try_iter().collect();
    assert!(said.iter().any(|l| l.contains("are acknowledged once")), "{}", said.join("\n"));
    drop(a);
    std::thread::sleep(Duration::from_millis(1800));
    let (status, _, third) = b.exchange(&request(4, &chain, Some(&key(2)), "f4", Some([4; 32])));
    assert_eq!(status, 200, "{third}");
    let (left3, debited3) = left_and_debited(&third);
    assert_eq!(left3 + debited3, left2, "b took the record and charged against what a left: nothing handed back");
    let said: Vec<String> = b.log.lock().unwrap().try_iter().collect();
    assert!(said.iter().any(|l| l.contains("taken at epoch 2")), "{}", said.join("\n"));

    let (status, _, again) = b.exchange(&request(5, &chain, Some(&key(2)), "f5", Some([2; 32])));
    assert_eq!(status, 200, "{again}");
    assert_eq!(receipt(&again).budget_remaining_uj, left2, "a's call record came with the record: the retry runs nothing");

    let store = jcp_fleet::DirStore::open(&dir).unwrap();
    let restored = jcp_fleet::restore(&store, &root().grant.id().to_hex()).unwrap();
    let receipts: Vec<Vec<u8>> = restored.stream.iter().filter(|e| e.first() == Some(&0xd2)).cloned().collect();
    let pubkey = SigningKey::from_bytes(&STEWARD_SEED).verifying_key().to_bytes();
    jcp_receipt::verify_log(&receipts, "did:jcp:gateway:srv", &pubkey).expect("the record's receipts are one chain, across both owners");
    assert!(restored.stream.iter().any(|e| e.first() == Some(&b'{')), "the call record is in the store beside the receipts");
    assert!(restored.fenced.is_empty());
}

/// A node that cannot renew its lease stops before it lapses past, writing
/// `SELF-FENCE:` and exiting 3, rather than serve records another node may
/// already have taken.
#[cfg(unix)]
#[test]
fn a_node_that_cannot_renew_its_lease_fences_itself() {
    use std::os::unix::fs::PermissionsExt as _;
    let dir = path("fence", "store");
    let _ = std::fs::remove_dir_all(&dir);
    let mut c = fleet_node("fence-c", "c", &dir);
    let nodes = dir.join("nodes");
    std::fs::set_permissions(&nodes, std::fs::Permissions::from_mode(0o555)).unwrap();
    if std::fs::write(nodes.join("probe-write"), b"x").is_ok() {
        // Permissions do not bind this user (root): the renewal cannot be made to fail this way.
        let _ = std::fs::remove_file(nodes.join("probe-write"));
        std::fs::set_permissions(&nodes, std::fs::Permissions::from_mode(0o755)).unwrap();
        eprintln!("skipped: directory permissions do not bind this user");
        return;
    }
    let status = {
        let deadline = std::time::Instant::now() + Duration::from_secs(20);
        loop {
            if let Some(status) = c.child.try_wait().unwrap() {
                break Some(status);
            }
            if std::time::Instant::now() > deadline {
                break None;
            }
            std::thread::sleep(Duration::from_millis(100));
        }
    };
    std::fs::set_permissions(&nodes, std::fs::Permissions::from_mode(0o755)).unwrap();
    let status = status.expect("the node stopped once its lease lapsed");
    assert_eq!(status.code(), Some(3));
    let said: Vec<String> = c.log.lock().unwrap().try_iter().collect();
    assert!(said.iter().any(|l| l.starts_with("SELF-FENCE:")), "{}", said.join("\n"));
}