//! 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; one declaring values sealed, only those sealed for their
//! readers.

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");
}

/// Writes `value` at `at` on `node` as `user`, past the checks a node makes of its own
/// writes (the API's, and judging the change as every node will): as a member that
/// bypasses them would. Signed by the node, as its own.
async fn forge(node: &AppState, user: &str, at: &str, value: serde_json::Value) {
    let at = path(at);
    let user = user.to_owned();
    let state = node.clone();
    tokio::task::spawn_blocking(move || {
        let mut replica = state.doc.blocking_write();
        let writer = crate::document::replica::Writer {
            node: state.local_id(),
            user: Some(&user),
        };
        replica
            .transact(state.clock.now_millis(), &writer, |doc| {
                store::write_path(doc, &at, &value)
            })
            .expect("forged write");
        let heads = replica.doc.get_heads();
        if let (Some(key), [hash]) = (&state.node_key, &heads[..]) {
            let replica = &mut *replica;
            replica.signatures.sign(&replica.doc, *hash, key);
        }
        replica.refresh_heads();
    })
    .await
    .expect("forged write");
    node.publish_change();
}

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.
    forge(&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());
    forge(&c, "alice", "_continuum/users/mallory", mallory).await;
    forge(&c, "alice", "config", json!({"mode": "forged"})).await;
    // By whichever node judges them first: c, told, drops them before the other may.
    sim.wait_until("the plain writes are refused", SOON, || async {
        refused(&a).await + refused(&b).await > 0
    })
    .await;
    sim.wait_until("c drops them", SOON, || async {
        plain(&c, "_continuum/users/mallory").await.is_none()
    })
    .await;
    for node in [&a, &b] {
        assert!(plain(node, "_continuum/users/mallory").await.is_none());
        assert_eq!(plain(node, "config").await, Some(json!({"mode": "slow"})));
    }
}

/// A swarm declaring values sealed (`access::sealed`), with the policy: a value sealed
/// for its declared readers goes everywhere; one a node writes plain there, or sealed
/// for a stranger too, gets nothing past the others.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn sealed_values() {
    use crate::access::sealed::testing::sealed_for;
    use crate::access::sshsig::PublicKey;
    let alice_key = PublicKey([21; 32]);
    let line = alice_key.line();
    let sim = Sim::with_identity(74);
    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/sealed",
        json!([{"path": ["secrets", "*"], "readers": ["alice"]}]),
    )
    .await;
    write_as(
        &a,
        "alice",
        "_continuum/access/policy",
        json!({"version": 1}),
    )
    .await;
    sim.wait_converged(&[&a, &b, &c], SOON).await;

    // Sealed for alice, written on b: everywhere.
    let sealed = sealed_for(&[alice_key]);
    write_as(&b, "alice", "secrets/db", json!(sealed.clone())).await;
    for node in [&a, &b, &c] {
        let sealed = sealed.clone();
        sim.wait_until("the sealed value is everywhere", SOON, move || {
            let sealed = sealed.clone();
            async move { plain(node, "secrets/db").await == Some(json!(sealed)) }
        })
        .await;
        assert_eq!(refused(node).await, 0, "an honest change was refused");
    }

    // c writes one sealed for a stranger too: refused by the others, and so is what it
    // builds on it.
    let stray = sealed_for(&[alice_key, PublicKey([22; 32])]);
    forge(&c, "alice", "secrets/web", json!(stray)).await;
    forge(&c, "alice", "secrets/api", json!("hunter2")).await;
    // By whichever node judges them first: c, told, drops them before the other may.
    sim.wait_until("the badly sealed writes are refused", SOON, || async {
        refused(&a).await + refused(&b).await > 0
    })
    .await;
    sim.wait_until("c drops them", SOON, || async {
        !holds(&c, "secrets/web").await
    })
    .await;
    for node in [&a, &b] {
        assert!(!holds(node, "secrets/web").await);
        assert!(!holds(node, "secrets/api").await);
    }
}

/// A node that took a change the rules refuse, unjudged (as under a policy it didn't
/// know yet): the others refuse what it writes on it, tell it so (REFUSALS), and it
/// drops the change and what it built on it, its own write among them quarantined, and
/// what it writes from then on goes everywhere.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_change_taken_unjudged() {
    let sim = Sim::with_identity(75);
    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}),
    )
    .await;
    sim.wait_converged(&[&a, &b, &c], SOON).await;

    // A change of a node that isn't a member, into c's document as if received
    // unjudged.
    let forged = {
        let mut doc = c.doc.read().await.doc.fork();
        let writer = crate::document::replica::Writer {
            node: "continuum:nobody",
            user: Some("alice"),
        };
        crate::document::replica::transact(&mut doc, crate::clock::UnixMillis(1), &writer, |doc| {
            store::write_path(doc, &path("notes/forged"), &json!(1))
        })
        .unwrap();
        doc.get_last_local_change().unwrap().clone()
    };
    {
        let doc = std::sync::Arc::clone(&c.doc);
        tokio::task::spawn_blocking(move || {
            let mut replica = doc.blocking_write();
            replica.doc.apply_changes([forged]).unwrap();
            replica.refresh_heads();
        })
        .await
        .unwrap();
    }
    assert!(holds(&c, "notes/forged").await);
    // c's write builds on it: refused by the others, who tell c.
    write_as(&c, "alice", "notes/after", json!(2)).await;
    sim.wait_until("c drops the change taken unjudged", SOON, || async {
        !holds(&c, "notes/forged").await
    })
    .await;
    assert!(
        !holds(&c, "notes/after").await,
        "built on the change dropped"
    );
    assert_eq!(c.remake.list_quarantine(|entry| entry.id.clone()).len(), 1);

    write_as(&c, "alice", "notes/later", json!(3)).await;
    for node in [&a, &b] {
        sim.wait_until("c's later write is everywhere", SOON, move || async move {
            holds(node, "notes/later").await
        })
        .await;
        assert!(!holds(node, "notes/forged").await);
        assert!(!holds(node, "notes/after").await);
    }
}

/// A member writing against the rules (past its API's checks): refused by every other
/// node (as a whole, its own change being signed, for as long as it takes); told which
/// change broke the rules (REFUSALS), it judges it itself, drops it, its write
/// quarantined, and is taken again with what it writes next.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_member_writing_against_the_rules() {
    let sim = Sim::with_identity(76);
    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;
    sim.wait_converged(&[&a, &b, &c], SOON).await;

    forge(&c, "mallory", "notes/forged", json!(1)).await;
    let rogue = c.local_id().to_owned();

    // By whichever node judges it first: the other may still be waiting for its
    // signature when c drops it, and then never gets to judge it.
    sim.wait_until("the change is refused", SOON, || async {
        refused(&a).await + refused(&b).await > 0
    })
    .await;
    sim.wait_until("c drops its change", SOON, || async {
        !holds(&c, "notes/forged").await
    })
    .await;
    assert_eq!(c.remake.list_quarantine(|entry| entry.id.clone()).len(), 1);
    write_as(&c, "alice", "notes/honest", json!(2)).await;
    for node in [&a, &b] {
        let rogue = rogue.clone();
        sim.wait_until("c's next write is taken, and c with it", SOON, move || {
            let rogue = rogue.clone();
            async move {
                use crate::locks::MutexExt;
                holds(node, "notes/honest").await
                    && node.screening.lock_or_recover().refusing(&rogue).is_none()
            }
        })
        .await;
        assert!(!holds(node, "notes/forged").await);
    }
}

/// A policy this node doesn't know (another version): it takes no writes through the
/// API, as it can't tell what the others would refuse, and syncs on.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_policy_this_node_doesnt_know() {
    let sim = Sim::with_identity(77);
    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);
    sim.wait_ready(&[&a, &b], SOON).await;
    write_as(&a, "alice", "notes/before", json!(1)).await;
    write_as(
        &a,
        "alice",
        "_continuum/access/policy",
        json!({"version": 2}),
    )
    .await;
    sim.wait_converged(&[&a, &b], SOON).await;
    assert!(holds(&b, "notes/before").await);
    let at = path("notes/after");
    let refused = b
        .mutate_doc_for(
            Some("alice".to_owned()),
            Sign::Write {
                path: at.clone(),
                asked: false,
            },
            move |doc| store::write_path(doc, &at, &json!(2)),
        )
        .await;
    assert!(
        matches!(refused, Err(store::StoreError::UnknownPolicy(_))),
        "a write under a policy this node doesn't know"
    );
}

/// A REFUSALS naming the swarm's first change, as a member could send: the node
/// receiving it keeps its document, as no rule refuses an epoch's root.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_refusal_naming_the_root() {
    let sim = Sim::with_identity(78);
    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);
    sim.wait_ready(&[&a, &b], SOON).await;
    write_as(
        &a,
        "alice",
        "_continuum/access/policy",
        json!({"version": 1}),
    )
    .await;
    write_as(&a, "alice", "notes/x", json!(1)).await;
    sim.wait_converged(&[&a, &b], SOON).await;
    let (root, epoch, epoch_seq) = {
        let replica = b.doc.read().await;
        (
            replica.doc.get_changes(&[])[0].hash(),
            replica.epoch,
            replica.epoch_seq,
        )
    };
    crate::swarm::refusals::tell(&a, b.local_id(), epoch, epoch_seq, vec![root]).await;
    tokio::time::sleep(Duration::from_secs(3)).await;
    assert!(holds(&b, "notes/x").await, "b dropped its document");
    assert_eq!(b.remake.list_quarantine(|entry| entry.id.clone()).len(), 0);
}

/// A write the API would let through but every node's rules refuse (here: past the
/// API, to a path its user may not write): judged before it leaves, refused, rolled
/// back; what the node writes next goes everywhere.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_write_every_node_would_refuse() {
    let sim = Sim::with_identity(79);
    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);
    sim.wait_ready(&[&a, &b], SOON).await;
    let bob = crate::access::registry::entry(
        "alice",
        1,
        &[(path("notes"), &[crate::access::Right::Write][..])],
    );
    write_as(&a, "alice", "_continuum/users/bob", bob).await;
    write_as(
        &a,
        "alice",
        "_continuum/access/policy",
        json!({"version": 1}),
    )
    .await;
    sim.wait_converged(&[&a, &b], SOON).await;

    let heads = b.doc.read().await.doc.get_heads();
    let at = path("secrets/x");
    let refused_write = b
        .mutate_doc_for(
            Some("bob".to_owned()),
            Sign::Write {
                path: at.clone(),
                asked: false,
            },
            move |doc| store::write_path(doc, &at, &json!(1)),
        )
        .await;
    assert!(
        matches!(refused_write, Err(store::StoreError::Refused(_))),
        "a write every node would refuse"
    );
    assert_eq!(b.doc.read().await.doc.get_heads(), heads, "rolled back");
    write_as(&b, "bob", "notes/x", json!(2)).await;
    sim.wait_until("bob's next write reaches a", SOON, || async {
        holds(&a, "notes/x").await
    })
    .await;
    assert!(!holds(&a, "secrets/x").await);
    assert_eq!(refused(&a).await, 0);
}