mod common;

use std::collections::BTreeSet;
use std::fs;
use std::path::{Path, PathBuf};
use std::time::{Duration, Instant};

use common::{
    ask, challenge, client, collection, config_for, messages_of_type, poll, poll_messages, publish,
    query_envelope, register, spawn_relay, spawn_relay_with_capacity, test_envelope, unicast,
};
use frxd::crypto::Keypair;
use frxd::index::LocalIndex;
use frxd::message::{
    EXPOSURE_FULL, EXPOSURE_METADATA, Envelope, QueryBody, TYPE_AGGREGATE, TYPE_QUERY,
    TYPE_RESPONSE,
};
use frxd::node::{Node, NodeHandle};
use serde_json::{Value, json};

fn keys(value: &Value) -> BTreeSet<String> {
    value.as_object().expect("object").keys().cloned().collect()
}

async fn status(http: &reqwest::Client, addr: &str) -> Value {
    http.get(format!("http://{addr}/v1/local/status"))
        .send()
        .await
        .unwrap()
        .json()
        .await
        .unwrap()
}

fn corpus_dir(root: &Path, name: &str, files: &[(&str, &str)]) -> PathBuf {
    let dir = root.join(format!("{name}-docs"));
    fs::create_dir_all(&dir).unwrap();
    for (file, text) in files {
        fs::write(dir.join(file), text).unwrap();
    }
    dir
}

async fn start_node_with_corpus(
    root: &Path,
    name: &str,
    relay_url: &str,
    shared: bool,
    exposure: &str,
    files: &[(&str, &str)],
) -> NodeHandle {
    let docs = corpus_dir(root, name, files);
    let config = config_for(&root.join(name), name, relay_url);
    {
        let index = LocalIndex::open(&config.index_dir()).unwrap();
        index
            .add_collection(&collection("docs", &docs, shared, exposure))
            .unwrap();
    }
    Node::start(config).await.unwrap()
}

async fn collect_responses(
    http: &reqwest::Client,
    relay_url: &str,
    key: &Keypair,
    timeout_ms: u64,
) -> Vec<Value> {
    let start = Instant::now();
    let deadline = Duration::from_millis(timeout_ms);
    let mut responses = Vec::new();
    while start.elapsed() < deadline {
        let messages = poll_messages(http, relay_url, key, 200).await;
        responses.extend(messages_of_type(&messages, TYPE_RESPONSE));
        if !responses.is_empty() {
            break;
        }
    }
    responses
}

async fn raw_query(
    http: &reqwest::Client,
    relay_url: &str,
    asker: &Keypair,
    text: &str,
    timeout_ms: u64,
) -> Vec<Value> {
    register(http, relay_url, asker).await;
    let envelope = query_envelope(asker, &asker.public_hex(), text, 5);
    assert!(
        publish(http, relay_url, &envelope)
            .await
            .status()
            .is_success()
    );
    collect_responses(http, relay_url, asker, timeout_ms).await
}

#[test]
fn envelope_field_set_is_fixed() {
    let key = Keypair::generate();
    let envelope = test_envelope(&key, TYPE_QUERY, json!({"qid": "q", "text": "t"}));
    let value = serde_json::to_value(&envelope).unwrap();
    assert_eq!(
        keys(&value),
        BTreeSet::from([
            "body".to_string(),
            "from".to_string(),
            "key".to_string(),
            "nonce".to_string(),
            "sig".to_string(),
            "ts".to_string(),
            "type".to_string(),
        ])
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn scores_never_travel_and_selection_is_local() {
    let root = tempfile::tempdir().unwrap();
    let relay_url = spawn_relay().await;
    let _bob = start_node_with_corpus(
        root.path(),
        "bob",
        &relay_url,
        true,
        EXPOSURE_FULL,
        &[
            ("alpha.txt", "rust search engine document"),
            ("beta.txt", "rust notes"),
        ],
    )
    .await;
    let alice = Keypair::generate();
    let http = client();
    let responses = raw_query(&http, &relay_url, &alice, "rust", 700).await;
    assert_eq!(responses.len(), 1);
    let body = responses[0].pointer("/body").unwrap();
    let results = body.pointer("/results").and_then(Value::as_array).unwrap();
    assert_eq!(results.len(), 2);
    let urls: BTreeSet<&str> = results
        .iter()
        .map(|result| result.get("url").and_then(Value::as_str).unwrap())
        .collect();
    assert!(urls.iter().any(|url| url.contains("alpha")));
    assert!(urls.iter().any(|url| url.contains("beta")));
    for result in results {
        assert!(!keys(result).contains("score"));
        assert!(!keys(result).contains("rank"));
    }
    assert_eq!(
        body.pointer("/truncated").and_then(Value::as_bool),
        Some(false)
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn metadata_exposure_withholds_content() {
    let root = tempfile::tempdir().unwrap();
    let relay_url = spawn_relay().await;
    let _bob = start_node_with_corpus(
        root.path(),
        "bob",
        &relay_url,
        true,
        EXPOSURE_METADATA,
        &[("corpus.txt", "enrichment corpus about rust")],
    )
    .await;
    let alice = Keypair::generate();
    let http = client();
    let responses = raw_query(&http, &relay_url, &alice, "rust", 700).await;
    assert_eq!(responses.len(), 1);
    let result = &responses[0]
        .pointer("/body/results")
        .and_then(Value::as_array)
        .unwrap()[0];
    assert_eq!(
        result.get("exposure").and_then(Value::as_str),
        Some(EXPOSURE_METADATA)
    );
    assert!(result.get("content").unwrap().is_null());
    assert!(
        !result
            .get("summary")
            .and_then(Value::as_str)
            .unwrap()
            .is_empty()
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn full_exposure_carries_content() {
    let root = tempfile::tempdir().unwrap();
    let relay_url = spawn_relay().await;
    let _bob = start_node_with_corpus(
        root.path(),
        "bob",
        &relay_url,
        true,
        EXPOSURE_FULL,
        &[("corpus.txt", "full exposure rust content body")],
    )
    .await;
    let alice = Keypair::generate();
    let http = client();
    let responses = raw_query(&http, &relay_url, &alice, "rust", 700).await;
    let result = &responses[0]
        .pointer("/body/results")
        .and_then(Value::as_array)
        .unwrap()[0];
    assert_eq!(
        result.get("exposure").and_then(Value::as_str),
        Some(EXPOSURE_FULL)
    );
    assert!(
        result
            .get("content")
            .and_then(Value::as_str)
            .unwrap()
            .contains("full exposure rust content body")
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn private_collections_are_silent_egress() {
    let root = tempfile::tempdir().unwrap();
    let relay_url = spawn_relay().await;
    let _bob = start_node_with_corpus(
        root.path(),
        "bob",
        &relay_url,
        false,
        EXPOSURE_FULL,
        &[("private.txt", "rust private data")],
    )
    .await;
    let alice = Keypair::generate();
    let http = client();
    let responses = raw_query(&http, &relay_url, &alice, "rust", 500).await;
    assert!(responses.is_empty(), "private collection answered");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn one_result_is_a_conformant_response() {
    let root = tempfile::tempdir().unwrap();
    let relay_url = spawn_relay().await;
    let _bob = start_node_with_corpus(
        root.path(),
        "bob",
        &relay_url,
        true,
        EXPOSURE_FULL,
        &[("only.txt", "solitary rust document")],
    )
    .await;
    let alice = Keypair::generate();
    let http = client();
    let responses = raw_query(&http, &relay_url, &alice, "solitary", 700).await;
    assert_eq!(responses.len(), 1);
    assert_eq!(
        responses[0]
            .pointer("/body/results")
            .and_then(Value::as_array)
            .unwrap()
            .len(),
        1
    );
    assert_eq!(
        responses[0]
            .pointer("/body/truncated")
            .and_then(Value::as_bool),
        Some(false)
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn empty_query_earns_silence() {
    let root = tempfile::tempdir().unwrap();
    let relay_url = spawn_relay().await;
    let bob = start_node_with_corpus(
        root.path(),
        "bob",
        &relay_url,
        true,
        EXPOSURE_FULL,
        &[("doc.txt", "rust content")],
    )
    .await;
    let alice = start_node_with_corpus(
        root.path(),
        "alice",
        &relay_url,
        false,
        EXPOSURE_FULL,
        &[("mine.txt", "alice rust")],
    )
    .await;
    let response = client()
        .post(format!("http://{}/v1/local/query", alice.addr))
        .json(&json!({"text": "", "max_results": 5, "timeout_ms": 500}))
        .send()
        .await
        .unwrap()
        .json::<Value>()
        .await
        .unwrap();
    assert!(
        response
            .get("responses")
            .and_then(Value::as_array)
            .unwrap()
            .is_empty()
    );
    assert!(bob.node.doc_count() > 0);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn entities_are_optional_hints() {
    let root = tempfile::tempdir().unwrap();
    let relay_url = spawn_relay().await;
    let _bob = start_node_with_corpus(
        root.path(),
        "bob",
        &relay_url,
        true,
        EXPOSURE_FULL,
        &[("doc.txt", "entity tagged rust document")],
    )
    .await;
    let alice = Keypair::generate();
    let http = client();
    register(&http, &relay_url, &alice).await;
    let body = json!({
        "qid": "entities-test",
        "text": "rust",
        "entities": ["Q999999", "not-a-real-entity"],
        "budget": {"max_results": 5}
    });
    let envelope = test_envelope(&alice, TYPE_QUERY, body);
    assert!(
        publish(&http, &relay_url, &envelope)
            .await
            .status()
            .is_success()
    );
    let responses = collect_responses(&http, &relay_url, &alice, 700).await;
    assert_eq!(responses.len(), 1);
    assert_eq!(
        responses[0].pointer("/body/qid").and_then(Value::as_str),
        Some("entities-test")
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn member_directory_filters_senders() {
    let root = tempfile::tempdir().unwrap();
    let relay_url = spawn_relay().await;
    let alice = Keypair::generate();
    let untrusted = Keypair::generate();

    let docs = corpus_dir(root.path(), "bob", &[("doc.txt", "rust document")]);
    let config = config_for(&root.path().join("bob"), "bob", &relay_url);
    frxd::config::save_members(
        &config.members_path(),
        &[frxd::config::Member {
            name: "alice".to_string(),
            pubkey: alice.public_hex(),
            previous: Vec::new(),
        }],
    )
    .unwrap();
    {
        let index = LocalIndex::open(&config.index_dir()).unwrap();
        index
            .add_collection(&collection("docs", &docs, true, EXPOSURE_FULL))
            .unwrap();
    }
    let _bob = Node::start(config).await.unwrap();

    let http = client();
    let trusted = raw_query(&http, &relay_url, &alice, "rust", 700).await;
    assert_eq!(trusted.len(), 1, "trusted member got no response");

    register(&http, &relay_url, &untrusted).await;
    let envelope = query_envelope(&untrusted, &untrusted.public_hex(), "rust", 5);
    assert!(
        publish(&http, &relay_url, &envelope)
            .await
            .status()
            .is_success()
    );
    let denied = collect_responses(&http, &relay_url, &untrusted, 400).await;
    assert!(denied.is_empty(), "untrusted member received a response");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn rotated_keys_are_accepted_through_previous_listing() {
    let root = tempfile::tempdir().unwrap();
    let relay_url = spawn_relay().await;
    let old_key = Keypair::generate();
    let new_key = Keypair::generate();

    let docs = corpus_dir(root.path(), "bob", &[("doc.txt", "rotation rust document")]);
    let config = config_for(&root.path().join("bob"), "bob", &relay_url);
    let listing = |previous: Vec<String>| {
        vec![frxd::config::Member {
            name: "carol".to_string(),
            pubkey: new_key.public_hex(),
            previous,
        }]
    };
    frxd::config::save_members(&config.members_path(), &listing(vec![old_key.public_hex()]))
        .unwrap();
    {
        let index = LocalIndex::open(&config.index_dir()).unwrap();
        index
            .add_collection(&collection("docs", &docs, true, EXPOSURE_FULL))
            .unwrap();
    }
    let _bob = Node::start(config.clone()).await.unwrap();

    let http = client();
    register(&http, &relay_url, &old_key).await;
    let envelope = query_envelope(&old_key, &old_key.public_hex(), "rust", 5);
    assert!(
        publish(&http, &relay_url, &envelope)
            .await
            .status()
            .is_success()
    );
    let accepted = collect_responses(&http, &relay_url, &old_key, 700).await;
    assert_eq!(accepted.len(), 1, "listed previous key was rejected");

    tokio::time::sleep(Duration::from_millis(20)).await;
    frxd::config::save_members(&config.members_path(), &listing(Vec::new())).unwrap();
    let envelope = query_envelope(&old_key, &old_key.public_hex(), "rust", 5);
    assert!(
        publish(&http, &relay_url, &envelope)
            .await
            .status()
            .is_success()
    );
    let revoked = collect_responses(&http, &relay_url, &old_key, 500).await;
    assert!(revoked.is_empty(), "revoked key was still accepted");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn plain_http_transport_is_refused_off_loopback() {
    let root = tempfile::tempdir().unwrap();
    let mut config = config_for(&root.path().join("alice"), "alice", "http://10.0.0.1:1");
    let refused = Node::start(config.clone()).await;
    assert!(
        refused.is_err(),
        "plain http to a non-loopback relay must be refused"
    );

    config.node.allow_insecure = true;
    let node = Node::start(config).await.unwrap();
    let outcome = frxd::node::control_query(
        &format!("http://{}", node.addr),
        "rust",
        Some(5),
        Some(100),
        false,
    )
    .await
    .unwrap();
    assert_eq!(
        outcome.pointer("/local/total").and_then(Value::as_u64),
        Some(0)
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn stale_envelopes_are_rejected() {
    let relay_url = spawn_relay().await;
    let http = client();
    let key = Keypair::generate();

    let mut envelope = query_envelope(&key, &key.public_hex(), "stale", 5);
    envelope.ts = frxd::crypto::now_ts().saturating_sub(3600);
    envelope.sig = key.sign(&frxd::crypto::signing_bytes(&envelope));
    assert_eq!(
        publish(&http, &relay_url, &envelope).await.status(),
        reqwest::StatusCode::BAD_REQUEST
    );

    let mut response = test_envelope(
        &key,
        TYPE_RESPONSE,
        json!({"qid": "q1", "results": [], "truncated": false, "more_available": 0, "cursor": null}),
    );
    response.ts = frxd::crypto::now_ts().saturating_sub(3600);
    response.sig = key.sign(&frxd::crypto::signing_bytes(&response));
    assert_eq!(
        unicast(&http, &relay_url, &key.public_hex(), &response)
            .await
            .status(),
        reqwest::StatusCode::BAD_REQUEST
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn lagging_member_is_isolated_and_reported() {
    let relay_url = spawn_relay_with_capacity(1).await;
    let http = client();
    let active = Keypair::generate();
    let slow = Keypair::generate();
    let publisher = Keypair::generate();
    register(&http, &relay_url, &active).await;
    register(&http, &relay_url, &slow).await;

    let first = query_envelope(&publisher, &publisher.public_hex(), "one", 5);
    assert_eq!(
        publish(&http, &relay_url, &first).await.status(),
        reqwest::StatusCode::ACCEPTED
    );
    assert_eq!(
        poll_messages(&http, &relay_url, &active, 300).await.len(),
        1
    );

    let second = query_envelope(&publisher, &publisher.public_hex(), "two", 5);
    assert_eq!(
        publish(&http, &relay_url, &second).await.status(),
        reqwest::StatusCode::ACCEPTED,
        "a lagging subscriber must not stall publishers"
    );
    assert_eq!(
        poll_messages(&http, &relay_url, &active, 300).await.len(),
        1
    );

    let response = poll(&http, &relay_url, &slow, 300).await;
    assert_eq!(response.status(), reqwest::StatusCode::TOO_MANY_REQUESTS);
    assert_eq!(
        response
            .headers()
            .get("retry-after")
            .and_then(|value| value.to_str().ok()),
        Some("1")
    );
    let payload: Value = response.json().await.unwrap();
    assert!(
        payload.get("missed").and_then(Value::as_u64).unwrap_or(0) >= 1,
        "lagging member must be told how much it missed: {payload}"
    );
    assert_eq!(
        poll(&http, &relay_url, &slow, 50).await.status(),
        reqwest::StatusCode::NO_CONTENT
    );

    let third = query_envelope(&publisher, &publisher.public_hex(), "three", 5);
    assert_eq!(
        publish(&http, &relay_url, &third).await.status(),
        reqwest::StatusCode::ACCEPTED
    );
    assert_eq!(poll_messages(&http, &relay_url, &slow, 300).await.len(), 1);

    let fourth = query_envelope(&publisher, &publisher.public_hex(), "four", 5);
    assert_eq!(
        publish(&http, &relay_url, &fourth).await.status(),
        reqwest::StatusCode::ACCEPTED
    );
    let fifth = query_envelope(&publisher, &publisher.public_hex(), "five", 5);
    assert_eq!(
        publish(&http, &relay_url, &fifth).await.status(),
        reqwest::StatusCode::ACCEPTED
    );
    let direct = test_envelope(
        &publisher,
        TYPE_RESPONSE,
        json!({"qid": "q", "results": [], "truncated": false, "more_available": 0, "cursor": null}),
    );
    assert_eq!(
        unicast(&http, &relay_url, &slow.public_hex(), &direct)
            .await
            .status(),
        reqwest::StatusCode::TOO_MANY_REQUESTS,
        "unicast to a lagging member must be visibly rejected"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn unicast_is_need_to_know() {
    let relay_url = spawn_relay().await;
    let http = client();
    let alice = Keypair::generate();
    let bob = Keypair::generate();
    register(&http, &relay_url, &alice).await;
    register(&http, &relay_url, &bob).await;

    let query = query_envelope(&alice, &alice.public_hex(), "rust", 5);
    assert!(
        publish(&http, &relay_url, &query)
            .await
            .status()
            .is_success()
    );

    let response = test_envelope(
        &bob,
        TYPE_RESPONSE,
        json!({"qid": "q1", "results": [], "truncated": false, "more_available": 0, "cursor": null}),
    );
    assert_eq!(
        unicast(&http, &relay_url, &alice.public_hex(), &response)
            .await
            .status(),
        reqwest::StatusCode::OK
    );

    let alice_messages = poll_messages(&http, &relay_url, &alice, 300).await;
    assert_eq!(messages_of_type(&alice_messages, TYPE_RESPONSE).len(), 1);
    let bob_messages = poll_messages(&http, &relay_url, &bob, 300).await;
    assert!(
        messages_of_type(&bob_messages, TYPE_RESPONSE).is_empty(),
        "unicast response leaked to another member"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn unicast_to_unknown_member_is_visible() {
    let relay_url = spawn_relay().await;
    let http = client();
    let sender = Keypair::generate();
    let response = test_envelope(
        &sender,
        TYPE_RESPONSE,
        json!({"qid": "q1", "results": [], "truncated": false, "more_available": 0, "cursor": null}),
    );
    let stranger = Keypair::generate().public_hex();
    assert_eq!(
        unicast(&http, &relay_url, &stranger, &response)
            .await
            .status(),
        reqwest::StatusCode::NOT_FOUND
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn tampered_envelopes_are_rejected() {
    let relay_url = spawn_relay().await;
    let http = client();
    let key = Keypair::generate();
    let mut envelope = query_envelope(&key, &key.public_hex(), "rust", 5);
    envelope.body = json!({"qid": "q1", "text": "forged", "budget": {"max_results": 5}});
    let publish_response = publish(&http, &relay_url, &envelope).await;
    assert_eq!(publish_response.status(), reqwest::StatusCode::BAD_REQUEST);

    let mut response = test_envelope(
        &key,
        TYPE_RESPONSE,
        json!({"qid": "q1", "results": [], "truncated": false, "more_available": 0, "cursor": null}),
    );
    response.sig = "00".repeat(64);
    let unicast_response = unicast(&http, &relay_url, &key.public_hex(), &response).await;
    assert_eq!(unicast_response.status(), reqwest::StatusCode::BAD_REQUEST);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn poll_requires_proof_of_key() {
    let relay_url = spawn_relay().await;
    let http = client();
    let member = Keypair::generate();
    let imposter = Keypair::generate();

    assert!(
        challenge(&http, &relay_url, "not-hex-at-all")
            .await
            .is_none()
    );

    let nonce = challenge(&http, &relay_url, &member.public_hex())
        .await
        .unwrap();
    let forged = imposter.sign(&frxd::crypto::poll_signing_bytes(
        &member.public_hex(),
        &nonce,
    ));
    let response = http
        .get(format!(
            "{relay_url}/v1/poll?member={}&nonce={nonce}&sig={forged}&timeout_ms=30",
            member.public_hex()
        ))
        .send()
        .await
        .unwrap();
    assert_eq!(response.status(), reqwest::StatusCode::UNAUTHORIZED);

    let response = poll(&http, &relay_url, &member, 30).await;
    assert_eq!(response.status(), reqwest::StatusCode::NO_CONTENT);

    let nonce = challenge(&http, &relay_url, &member.public_hex())
        .await
        .unwrap();
    let sig = member.sign(&frxd::crypto::poll_signing_bytes(
        &member.public_hex(),
        &nonce,
    ));
    let url = format!(
        "{relay_url}/v1/poll?member={}&nonce={nonce}&sig={sig}&timeout_ms=30",
        member.public_hex()
    );
    assert_eq!(
        http.get(&url).send().await.unwrap().status(),
        reqwest::StatusCode::NO_CONTENT
    );
    assert_eq!(
        http.get(&url).send().await.unwrap().status(),
        reqwest::StatusCode::UNAUTHORIZED,
        "challenge nonce must be single-use"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn malformed_json_is_rejected() {
    let relay_url = spawn_relay().await;
    let http = client();
    let response = http
        .post(format!("{relay_url}/v1/publish"))
        .header("content-type", "application/json")
        .body("{not json")
        .send()
        .await
        .unwrap();
    assert_eq!(response.status(), reqwest::StatusCode::BAD_REQUEST);

    let root = tempfile::tempdir().unwrap();
    let config = config_for(&root.path().join("alice"), "alice", &relay_url);
    let node = Node::start(config).await.unwrap();
    let response = http
        .post(format!("http://{}/v1/local/query", node.addr))
        .header("content-type", "application/json")
        .body("{not json")
        .send()
        .await
        .unwrap();
    assert_eq!(response.status(), reqwest::StatusCode::BAD_REQUEST);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn node_does_not_answer_its_own_queries() {
    let root = tempfile::tempdir().unwrap();
    let relay_url = spawn_relay().await;
    let bob = start_node_with_corpus(
        root.path(),
        "bob",
        &relay_url,
        true,
        EXPOSURE_FULL,
        &[("doc.txt", "self query rust")],
    )
    .await;
    let (_status, _raw, value) = ask(&client(), &bob.addr.to_string(), "rust", 5).await;
    assert!(
        value
            .get("responses")
            .and_then(Value::as_array)
            .unwrap()
            .is_empty(),
        "node responded to itself: {value}"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn ingress_happens_only_locally() {
    let root = tempfile::tempdir().unwrap();
    let relay_url = spawn_relay().await;
    let bob = start_node_with_corpus(
        root.path(),
        "bob",
        &relay_url,
        true,
        EXPOSURE_FULL,
        &[("doc.txt", "rust served by bob")],
    )
    .await;
    let alice = start_node_with_corpus(
        root.path(),
        "alice",
        &relay_url,
        false,
        EXPOSURE_FULL,
        &[("mine.txt", "alice rust notes")],
    )
    .await;
    let http = client();
    let bob_before = status(&http, &bob.addr.to_string()).await;
    let alice_before = status(&http, &alice.addr.to_string()).await;

    let (_s, _r, value) = ask(&http, &alice.addr.to_string(), "rust", 5).await;
    assert_eq!(
        value
            .get("responses")
            .and_then(Value::as_array)
            .unwrap()
            .len(),
        1
    );

    let bob_after = status(&http, &bob.addr.to_string()).await;
    let alice_after = status(&http, &alice.addr.to_string()).await;
    assert_eq!(bob_before.get("doc_count"), bob_after.get("doc_count"));
    assert_eq!(alice_before.get("doc_count"), alice_after.get("doc_count"));
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn querier_and_responder_roles_coexist() {
    let root = tempfile::tempdir().unwrap();
    let relay_url = spawn_relay().await;
    let alice = start_node_with_corpus(
        root.path(),
        "alice",
        &relay_url,
        true,
        EXPOSURE_FULL,
        &[("a.txt", "alice rust document")],
    )
    .await;
    let bob = start_node_with_corpus(
        root.path(),
        "bob",
        &relay_url,
        true,
        EXPOSURE_FULL,
        &[("b.txt", "bob rust document")],
    )
    .await;

    let http = client();
    let alice_addr = alice.addr.to_string();
    let bob_addr = bob.addr.to_string();
    let alice_ask = async { common::ask(&http, &alice_addr, "rust", 5).await };
    let bob_ask = async { common::ask(&http, &bob_addr, "rust", 5).await };
    let (alice_result, bob_result) = tokio::join!(alice_ask, bob_ask);

    for (name, (_status, _raw, value)) in [("alice", alice_result), ("bob", bob_result)] {
        let responses = value.get("responses").and_then(Value::as_array).unwrap();
        assert_eq!(responses.len(), 1, "{name} got {value}");
        let member = responses[0].get("member").and_then(Value::as_str).unwrap();
        assert!(!member.is_empty());
    }
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn budget_is_clamped_and_never_exceeded() {
    let root = tempfile::tempdir().unwrap();
    let relay_url = spawn_relay().await;
    let _bob = start_node_with_corpus(
        root.path(),
        "bob",
        &relay_url,
        true,
        EXPOSURE_FULL,
        &[
            ("one.txt", "rust one"),
            ("two.txt", "rust two"),
            ("three.txt", "rust three"),
        ],
    )
    .await;
    let alice = start_node_with_corpus(
        root.path(),
        "alice",
        &relay_url,
        false,
        EXPOSURE_FULL,
        &[("mine.txt", "alice doc")],
    )
    .await;
    let http = client();

    let (_s, _r, value) = ask(&http, &alice.addr.to_string(), "rust", 0).await;
    let responses = value.get("responses").and_then(Value::as_array).unwrap();
    assert_eq!(responses.len(), 1);
    assert_eq!(
        responses[0]
            .get("results")
            .and_then(Value::as_array)
            .unwrap()
            .len(),
        1,
        "zero budget must clamp to one, never send nothing or more"
    );

    let (_s, _r, value) = ask(&http, &alice.addr.to_string(), "rust", 1000).await;
    let responses = value.get("responses").and_then(Value::as_array).unwrap();
    assert_eq!(
        responses[0]
            .get("results")
            .and_then(Value::as_array)
            .unwrap()
            .len(),
        3
    );
    assert_eq!(
        responses[0].get("truncated").and_then(Value::as_bool),
        Some(false)
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn aggregates_are_unicast_only_and_ignored_by_nodes() {
    let root = tempfile::tempdir().unwrap();
    let relay_url = spawn_relay().await;
    let bob = start_node_with_corpus(
        root.path(),
        "bob",
        &relay_url,
        true,
        EXPOSURE_FULL,
        &[("doc.txt", "aggregate test rust")],
    )
    .await;
    let http = client();
    let sender = Keypair::generate();
    register(&http, &relay_url, &bob.node.key).await;

    let broadcast = test_envelope(
        &sender,
        TYPE_AGGREGATE,
        json!({"period": "2026-03", "sent": 1, "passed": 0}),
    );
    assert_eq!(
        publish(&http, &relay_url, &broadcast).await.status(),
        reqwest::StatusCode::BAD_REQUEST,
        "aggregates must not travel the broadcast channel"
    );

    let unicast_envelope = test_envelope(
        &sender,
        TYPE_AGGREGATE,
        json!({"period": "2026-03", "sent": 1, "passed": 0}),
    );
    assert_eq!(
        unicast(&http, &relay_url, &bob.pubkey, &unicast_envelope)
            .await
            .status(),
        reqwest::StatusCode::OK
    );
    tokio::time::sleep(Duration::from_millis(200)).await;
    let (_s, _r, value) = ask(&http, &bob.addr.to_string(), "ignored", 5).await;
    assert_eq!(
        value.pointer("/local/total").and_then(Value::as_u64),
        Some(0)
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn node_survives_unreachable_relay() {
    let root = tempfile::tempdir().unwrap();
    let alice = start_node_with_corpus(
        root.path(),
        "alice",
        "http://127.0.0.1:1",
        false,
        EXPOSURE_FULL,
        &[("mine.txt", "offline rust note")],
    )
    .await;
    let (status_code, _raw, value) = ask(&client(), &alice.addr.to_string(), "rust", 5).await;
    assert!(status_code.is_success());
    assert_eq!(
        value.pointer("/local/total").and_then(Value::as_u64),
        Some(1)
    );
    assert!(
        value
            .get("responses")
            .and_then(Value::as_array)
            .unwrap()
            .is_empty()
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn duplicate_delivery_is_answered_once() {
    let root = tempfile::tempdir().unwrap();
    let relay_url = spawn_relay().await;
    let _bob = start_node_with_corpus(
        root.path(),
        "bob",
        &relay_url,
        true,
        EXPOSURE_FULL,
        &[("doc.txt", "duplicate rust doc")],
    )
    .await;
    let alice = Keypair::generate();
    let http = client();
    register(&http, &relay_url, &alice).await;
    let envelope = query_envelope(&alice, &alice.public_hex(), "rust", 5);
    assert!(
        publish(&http, &relay_url, &envelope)
            .await
            .status()
            .is_success()
    );
    assert!(
        publish(&http, &relay_url, &envelope)
            .await
            .status()
            .is_success()
    );
    let responses = collect_responses(&http, &relay_url, &alice, 700).await;
    assert_eq!(
        responses.len(),
        1,
        "duplicate query produced duplicate responses"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn multiple_responders_merge_with_provenance() {
    let root = tempfile::tempdir().unwrap();
    let relay_url = spawn_relay().await;
    let _bob = start_node_with_corpus(
        root.path(),
        "bob",
        &relay_url,
        true,
        EXPOSURE_FULL,
        &[("bob.txt", "bob rust contribution")],
    )
    .await;
    let _carol = start_node_with_corpus(
        root.path(),
        "carol",
        &relay_url,
        true,
        EXPOSURE_FULL,
        &[("carol.txt", "carol rust contribution")],
    )
    .await;
    let alice = start_node_with_corpus(
        root.path(),
        "alice",
        &relay_url,
        false,
        EXPOSURE_FULL,
        &[("mine.txt", "alice rust local")],
    )
    .await;
    let (_s, _r, value) = ask(&client(), &alice.addr.to_string(), "rust", 5).await;
    let responses = value.get("responses").and_then(Value::as_array).unwrap();
    assert_eq!(responses.len(), 2, "expected both responders: {value}");
    let members: BTreeSet<&str> = responses
        .iter()
        .map(|r| r.get("member").and_then(Value::as_str).unwrap())
        .collect();
    assert_eq!(members.len(), 2);
    let merged = value.get("merged").and_then(Value::as_array).unwrap();
    let provenances: BTreeSet<&str> = merged
        .iter()
        .map(|m| m.get("provenance").and_then(Value::as_str).unwrap())
        .collect();
    assert!(provenances.contains("local"));
    assert_eq!(provenances.len(), 3);
}

#[test]
fn canonical_signing_bytes_are_stable() {
    let mut envelope = Envelope {
        msg_type: TYPE_QUERY.to_string(),
        from: "alice.frx.example".to_string(),
        key: "aa".repeat(32),
        ts: 1_700_000_000,
        nonce: "00112233445566778899aabbccddeeff".to_string(),
        body: json!({"text": "rust", "qid": "q1", "budget": {"max_results": 5}, "entities": []}),
        sig: String::new(),
    };
    let first = frxd::crypto::signing_bytes(&envelope);
    envelope.body =
        json!({"entities": [], "budget": {"max_results": 5}, "qid": "q1", "text": "rust"});
    let second = frxd::crypto::signing_bytes(&envelope);
    assert_eq!(first, second);
    let key = frxd::crypto::Keypair::from_hex(&"00".repeat(32)).unwrap();
    let signature = key.sign(&first);
    assert_eq!(
        String::from_utf8(first).unwrap(),
        "FRX/0.5\n{\"body\":{\"budget\":{\"max_results\":5},\"entities\":[],\"qid\":\"q1\",\"text\":\"rust\"},\"from\":\"alice.frx.example\",\"key\":\"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa\",\"nonce\":\"00112233445566778899aabbccddeeff\",\"ts\":1700000000,\"type\":\"query\"}"
    );
    assert_eq!(
        signature,
        "2d5df0f85b5ea6b97d5dbc53d316b410c848d52e407583f9740d10889ba5a40fbcf03a930fa7d351689aa657884e3a32e0c6941ff11653a71d75dea58b3e9b09"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn query_body_without_entities_parses() {
    let root = tempfile::tempdir().unwrap();
    let relay_url = spawn_relay().await;
    let _bob = start_node_with_corpus(
        root.path(),
        "bob",
        &relay_url,
        true,
        EXPOSURE_FULL,
        &[("doc.txt", "minimal body rust")],
    )
    .await;
    let alice = Keypair::generate();
    let http = client();
    register(&http, &relay_url, &alice).await;
    let envelope = test_envelope(
        &alice,
        TYPE_QUERY,
        json!({"qid": "min", "text": "rust", "budget": {"max_results": 5}}),
    );
    assert!(
        publish(&http, &relay_url, &envelope)
            .await
            .status()
            .is_success()
    );
    let responses = collect_responses(&http, &relay_url, &alice, 700).await;
    assert_eq!(responses.len(), 1);
}

#[test]
fn query_body_budget_is_the_only_quantity_knob() {
    let value = serde_json::to_value(QueryBody::new("rust", 5)).unwrap();
    assert_eq!(
        keys(&value),
        BTreeSet::from([
            "qid".to_string(),
            "text".to_string(),
            "entities".to_string(),
            "budget".to_string()
        ])
    );
}