//! A swarm whose policy has every node judge every change (`access::policy`,
//! `access::screen`), with signatures required: honest writes go everywhere, a joining
//! node registers, a write a node lets through for a user the registry doesn't have
//! stays where it was made, and a remake carries the swarm over without refusing
//! anything.

use std::time::Duration;

use serde_json::json;

use super::{Sim, path};
use crate::document::signatures::Sign;
use crate::document::store;
use crate::state::AppState;
use crate::swarm::bootstrap::Start;

const SOON: Duration = Duration::from_secs(30);

/// Writes `value` at `at` on `node` for `user`, as the API does after its checks, or
/// without them.
async fn write_as(node: &AppState, user: &str, at: &str, value: serde_json::Value) {
    let at = path(at);
    let sign = Sign::Write {
        path: at.clone(),
        asked: false,
    };
    node.mutate_doc_for(Some(user.to_owned()), sign, move |doc| {
        store::write_path(doc, &at, &value)
    })
    .await
    .expect("write");
}

async fn holds(node: &AppState, at: &str) -> bool {
    let at = path(at);
    let replica = node.doc.read().await;
    store::read_path(&replica.doc, &at).ok().flatten().is_some()
}

async fn refused(node: &AppState) -> usize {
    use crate::locks::MutexExt;
    node.screening.lock_or_recover().refused().count()
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_swarm_judging_every_change() {
    let sim = Sim::with_identity(71);
    let a = sim.start_with("a", &[], Start::Genesis, |config| {
        config.first_users = vec!["alice".to_owned()];
    });
    let b = sim.start("b", &["a"], Start::Join);
    let c = sim.start("c", &["a"], Start::Join);
    sim.wait_ready(&[&a, &b, &c], SOON).await;

    write_as(
        &a,
        "alice",
        "_continuum/access/policy",
        json!({"version": 1, "signatures": "required"}),
    )
    .await;
    sim.wait_converged(&[&a, &b, &c], SOON).await;

    // Honest writes, on any node: everywhere.
    write_as(&b, "alice", "notes/x", json!(1)).await;
    write_as(&c, "alice", "notes/y", json!(2)).await;
    sim.wait_converged(&[&a, &b, &c], SOON).await;
    for node in [&a, &b, &c] {
        assert!(holds(node, "notes/x").await && holds(node, "notes/y").await);
    }

    // A node joining under the policy: it registers, and its writes go everywhere.
    let d = sim.start("d", &["a"], Start::Join);
    sim.wait_ready(&[&d], SOON).await;
    write_as(&d, "alice", "notes/z", json!(3)).await;
    sim.wait_converged(&[&a, &b, &c, &d], SOON).await;
    for node in [&a, &b, &c, &d] {
        assert_eq!(refused(node).await, 0, "an honest change was refused");
        assert!(holds(node, "notes/z").await);
    }

    // A write c lets through for a user the registry doesn't have: refused by the
    // others, and so is what c builds on it.
    write_as(&c, "mallory", "notes/forged", json!(4)).await;
    for node in [&a, &b, &d] {
        sim.wait_until("the forged write is refused", SOON, move || async move {
            refused(node).await > 0
        })
        .await;
        assert!(!holds(node, "notes/forged").await);
    }
}

/// A compaction under the policy: every node switches, the data and what was written
/// while it ran come along (replayed, signed at once after the switch), and nothing an
/// honest node wrote is refused, before or after.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_remake_under_the_policy() {
    let sim = Sim::with_identity(72);
    let a = sim.start_with("a", &[], Start::Genesis, |config| {
        config.first_users = vec!["alice".to_owned()];
    });
    let b = sim.start("b", &["a"], Start::Join);
    let c = sim.start("c", &["a"], Start::Join);
    sim.wait_ready(&[&a, &b, &c], SOON).await;
    let nodes = [a.clone(), b.clone(), c.clone()];
    sim.wait_until("every node is registered", SOON, || {
        let nodes = nodes.clone();
        async move {
            for node in &nodes {
                if crate::swarm::members::members(&node.doc.read().await.doc).len() < 3 {
                    return false;
                }
            }
            true
        }
    })
    .await;
    write_as(
        &a,
        "alice",
        "_continuum/access/policy",
        json!({"version": 1, "signatures": "required"}),
    )
    .await;
    write_as(&b, "alice", "notes/before", json!(1)).await;
    sim.wait_converged(&[&a, &b, &c], SOON).await;

    let (status, report) =
        crate::remake::testing::start(a.clone(), json!({ "kind": "compact" }), None).await;
    assert!(status.is_success(), "proposal refused: {status} {report}");
    // While it runs: replayed into the new epoch.
    write_as(&c, "alice", "notes/during", json!(2)).await;
    for node in [&a, &b, &c] {
        assert_eq!(refused(node).await, 0, "an honest change was refused");
    }
    sim.wait_until(
        "every node is on generation 1",
        Duration::from_secs(60),
        || {
            let nodes = nodes.clone();
            async move {
                for node in &nodes {
                    if node.local_epoch().generation != 1 || !node.is_ready().await {
                        return false;
                    }
                }
                true
            }
        },
    )
    .await;

    write_as(&b, "alice", "notes/after", json!(3)).await;
    for node in [&a, &b, &c] {
        sim.wait_until("every write is everywhere", SOON, move || async move {
            holds(node, "notes/before").await
                && holds(node, "notes/during").await
                && holds(node, "notes/after").await
        })
        .await;
        assert_eq!(refused(node).await, 0, "an honest change was refused");
    }
}