frxd/tests/aggregates.rs

mod common;

use std::fs;
use std::path::Path;

use common::{ask, client, collection, config_for, spawn_relay};
use frxd::config::{CLASS_ENRICHMENT, CLASS_SOURCE, Member, save_members};
use frxd::index::LocalIndex;
use frxd::message::EXPOSURE_FULL;
use frxd::node::{self, Node, NodeHandle, current_period};
use serde_json::Value;

fn corpus_dir(root: &Path, name: &str, files: &[(&str, &str)]) -> std::path::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(
    root: &Path,
    name: &str,
    relay_url: &str,
    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, true, exposure))
            .unwrap();
    }
    Node::start(config).await.unwrap()
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn aggregates_count_sent_and_passed_per_member() {
    let root = tempfile::tempdir().unwrap();
    let relay_url = spawn_relay().await;
    let _bob = start_node(
        root.path(),
        "bob",
        &relay_url,
        EXPOSURE_FULL,
        &[("doc.txt", "aggregate rust document")],
    )
    .await;
    let alice = start_node(
        root.path(),
        "alice",
        &relay_url,
        EXPOSURE_FULL,
        &[("mine.txt", "alice rust note")],
    )
    .await;

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

    let period = current_period();
    let aggregate = node::control_aggregate_request(
        &format!("http://{}", _bob.addr),
        &alice.pubkey,
        &period,
        Some(700),
    )
    .await
    .unwrap();
    assert_eq!(aggregate.get("sent").and_then(Value::as_u64), Some(1));
    assert_eq!(aggregate.get("passed").and_then(Value::as_u64), Some(1));
    assert!(aggregate.get("cited").is_none());
    assert_eq!(
        aggregate.get("period").and_then(Value::as_str),
        Some(period.as_str())
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn aggregate_rollup_for_year_sums_months() {
    let root = tempfile::tempdir().unwrap();
    let relay_url = spawn_relay().await;
    let bob = start_node(
        root.path(),
        "bob",
        &relay_url,
        EXPOSURE_FULL,
        &[("doc.txt", "rollup rust document")],
    )
    .await;
    let alice = start_node(
        root.path(),
        "alice",
        &relay_url,
        EXPOSURE_FULL,
        &[("mine.txt", "alice rust note")],
    )
    .await;

    let (_status, _raw, _value) = ask(&client(), &alice.addr.to_string(), "rust", 5).await;
    let year = current_period()[0..4].to_string();
    let aggregate = node::control_aggregate_request(
        &format!("http://{}", bob.addr),
        &alice.pubkey,
        &year,
        Some(700),
    )
    .await
    .unwrap();
    assert_eq!(aggregate.get("sent").and_then(Value::as_u64), Some(1));
    assert_eq!(
        aggregate.get("period").and_then(Value::as_str),
        Some(year.as_str())
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn aggregate_floor_rejects_finer_than_month() {
    let root = tempfile::tempdir().unwrap();
    let alice = start_node(
        root.path(),
        "alice",
        &spawn_relay().await,
        EXPOSURE_FULL,
        &[("mine.txt", "alice rust note")],
    )
    .await;
    let http = client();

    let response = http
        .get(format!(
            "http://{}/v1/local/aggregates?period=2026-09-15",
            alice.addr
        ))
        .send()
        .await
        .unwrap();
    assert_eq!(response.status(), reqwest::StatusCode::BAD_REQUEST);

    let local = http
        .get(format!("http://{}/v1/local/aggregates", alice.addr))
        .send()
        .await
        .unwrap();
    assert!(local.status().is_success());

    let config = alice.node.config.clone();
    let direct = alice
        .node
        .request_aggregate(&config.node.relays[0], "2026-09-15", Some(50))
        .await;
    assert!(direct.is_err());
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn enrichment_members_cannot_send_content() {
    let root = tempfile::tempdir().unwrap();
    let relay_url = spawn_relay().await;
    let bob_full = start_node(
        root.path(),
        "bob",
        &relay_url,
        EXPOSURE_FULL,
        &[("doc.txt", "enrichment test rust content")],
    )
    .await;
    let carol_meta = start_node(
        root.path(),
        "carol",
        &relay_url,
        "metadata",
        &[("doc.txt", "enrichment test rust metadata")],
    )
    .await;

    let alice = start_node(
        root.path(),
        "alice",
        &relay_url,
        EXPOSURE_FULL,
        &[("mine.txt", "alice local")],
    )
    .await;
    save_members(
        &alice.node.config.members_path(),
        &[
            Member {
                name: "bob".to_string(),
                pubkey: bob_full.pubkey.clone(),
                class: CLASS_ENRICHMENT.to_string(),
                previous: Vec::new(),
            },
            Member {
                name: "carol".to_string(),
                pubkey: carol_meta.pubkey.clone(),
                class: CLASS_ENRICHMENT.to_string(),
                previous: Vec::new(),
            },
        ],
    )
    .unwrap();

    let (_status, _raw, 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(),
        1,
        "full-exposure enrichment reply must be dropped"
    );
    assert_eq!(
        responses[0].get("member").and_then(Value::as_str),
        Some(carol_meta.pubkey.as_str())
    );
    assert!(
        responses[0]
            .get("results")
            .and_then(Value::as_array)
            .unwrap()[0]
            .get("content")
            .unwrap()
            .is_null()
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn source_members_may_send_content() {
    let root = tempfile::tempdir().unwrap();
    let relay_url = spawn_relay().await;
    let bob = start_node(
        root.path(),
        "bob",
        &relay_url,
        EXPOSURE_FULL,
        &[("doc.txt", "source test rust content")],
    )
    .await;
    let alice = start_node(
        root.path(),
        "alice",
        &relay_url,
        EXPOSURE_FULL,
        &[("mine.txt", "alice local")],
    )
    .await;
    save_members(
        &alice.node.config.members_path(),
        &[Member {
            name: "bob".to_string(),
            pubkey: bob.pubkey.clone(),
            class: CLASS_SOURCE.to_string(),
            previous: Vec::new(),
        }],
    )
    .unwrap();

    let (_status, _raw, 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(), 1);
    assert!(
        responses[0]
            .get("results")
            .and_then(Value::as_array)
            .unwrap()[0]
            .get("content")
            .unwrap()
            .is_string()
    );
}