//! 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. A strict one (`"values": "signed"`) takes the values its users sign, and
//! nothing else there.

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 = [("alice".to_owned(), Default::default())].into();
    });
    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 = [("alice".to_owned(), Default::default())].into();
    });
    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");
    }
}

/// Writes `value` at `at` on `node`, signed by `user` with `key` (`access::envelope`),
/// replacing the envelope there on `node`, as the store API writes one.
async fn write_signed(
    node: &AppState,
    user: &str,
    key: &ed25519_dalek::SigningKey,
    at: &str,
    value: serde_json::Value,
) {
    use crate::access::{envelope, sshsig};
    use crate::document::ScalarValue;
    use crate::document::typed::{TimestampFormat, Val};
    let at = path(at);
    let previous = match store::read_path(&node.doc.read().await.doc, &at) {
        Ok(Some(serde_json::Value::String(text))) if text.starts_with("cv1.") => Some(text),
        _ => None,
    };
    let payload = envelope::Payload {
        path: at.clone(),
        value,
        user: user.to_owned(),
        at: 1,
        prev: envelope::digest(previous.as_deref()),
    };
    let bytes = serde_json::to_vec(&payload).unwrap();
    let sealed = envelope::seal(&bytes, &sshsig::sign(key, envelope::NAMESPACE, &bytes)).unwrap();
    let sign = Sign::Write {
        path: at.clone(),
        asked: false,
    };
    node.mutate_doc_for(Some(user.to_owned()), sign, move |doc| {
        let writing = store::Writing::new(store::Sticky::No, TimestampFormat::default(), true);
        store::write_val_path(
            doc,
            &at,
            &Val::Scalar(ScalarValue::Str(sealed.into())),
            writing,
        )
        .map(|_| ())
    })
    .await
    .expect("signed write");
}

/// What `node` reads at `at`, envelopes opened (`store::read_plain`).
async fn plain(node: &AppState, at: &str) -> Option<serde_json::Value> {
    store::read_plain(&node.doc.read().await.doc, &path(at))
        .ok()
        .flatten()
}

/// A strict swarm (`"values": "signed"`), its first user's key given at genesis: values
/// signed by their user go everywhere, before and after a remake, which carries them
/// over; a node writing the registry plain, for a user that may, gets nothing past the
/// others.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_strict_swarm() {
    let key = ed25519_dalek::SigningKey::from_bytes(&[11; 32]);
    let line = crate::access::sshsig::PublicKey(key.verifying_key().to_bytes()).line();
    let sim = Sim::with_identity(73);
    let a = sim.start_with("a", &[], Start::Genesis, move |config| {
        config.first_users = [(
            "alice".to_owned(),
            [("laptop".to_owned(), line.clone())].into(),
        )]
        .into();
    });
    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/userSigned",
        json!([["config"]]),
    )
    .await;
    write_as(
        &a,
        "alice",
        "_continuum/access/policy",
        json!({"version": 1, "values": "signed"}),
    )
    .await;
    sim.wait_converged(&[&a, &b, &c], SOON).await;

    // Signed, on any node: everywhere.
    write_signed(&b, "alice", &key, "config", json!({"mode": "fast"})).await;
    for node in [&a, &b, &c] {
        sim.wait_until("the signed value is everywhere", SOON, move || async move {
            plain(node, "config").await == Some(json!({"mode": "fast"}))
        })
        .await;
    }

    // A remake carries it over; replacing it afterwards names it.
    let (status, report) =
        crate::remake::testing::start(a.clone(), json!({ "kind": "compact" }), None).await;
    assert!(status.is_success(), "proposal refused: {status} {report}");
    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_signed(&c, "alice", &key, "config", json!({"mode": "slow"})).await;
    for node in [&a, &b, &c] {
        sim.wait_until("the replacement is everywhere", SOON, move || async move {
            plain(node, "config").await == Some(json!({"mode": "slow"}))
        })
        .await;
        assert_eq!(refused(node).await, 0, "an honest change was refused");
    }

    // c writes the registry plain, as a user who may manage users: refused by the
    // others, and so is a plain write over the signed value.
    let mallory = crate::access::registry::entry("alice", 1, &crate::access::registry::all_caps());
    write_as(&c, "alice", "_continuum/users/mallory", mallory).await;
    write_as(&c, "alice", "config", json!({"mode": "forged"})).await;
    for node in [&a, &b] {
        sim.wait_until("the plain writes are refused", SOON, move || async move {
            refused(node).await > 0
        })
        .await;
        assert!(plain(node, "_continuum/users/mallory").await.is_none());
        assert_eq!(plain(node, "config").await, Some(json!({"mode": "slow"})));
    }
}