//! bl-018 — multi-actor convergence over the zero-infra FilesystemSync.
//!
//! Two actors each apply a concurrent write to the SAME entity in their own
//! substrate, exchange operations through a shared directory (ADR-027's
//! dependency-free default transport), ingest each other's ops, and
//! converge to one identical, verifiable state. Demonstrates criterion 6
//! ("concurrent-safe") with no broker, network, or extra process.

use ed25519_dalek::SigningKey;
use ontoref_core::Substrate;
use ontoref_sync::{FilesystemSync, SyncBackend};
use ontoref_types::{
    canonical,
    operation::{OpBody, OpPayload, Operation},
    AttrId, EntityId, Hlc, PublicKey, Value,
};

fn assert_op(actor: &SigningKey, entity: &str, value: i64, hlc: Hlc) -> Operation {
    Operation::sign(
        OpBody {
            actor: PublicKey::from_signing_key(actor),
            parents: vec![],
            payload: OpPayload::EntityAssert {
                attrs: vec![(AttrId::new("v"), Value::Int(value))],
                entity: EntityId::new(entity),
                valid_from: None,
            },
            timestamp: hlc,
        },
        actor,
    )
}

/// Publish an op's body under `sync`'s local peer (announce is a separate
/// `&mut self` call the caller makes).
fn share(sync: &FilesystemSync, op: &Operation) {
    sync.publish_body(&hex(&op.id().0), &canonical::encode(op).expect("encode"))
        .expect("publish");
}

fn hex(bytes: &[u8]) -> String {
    let mut s = String::with_capacity(bytes.len() * 2);
    for b in bytes {
        use std::fmt::Write as _;
        let _ = write!(s, "{b:02x}");
    }
    s
}

/// Pull every op a peer announced, decode it, and ingest into `sub`.
fn pull_and_ingest(sync: &FilesystemSync, peer: &str, sub: &mut Substrate) -> usize {
    let mut ops = Vec::new();
    for id in sync.announced_ops(peer).expect("announced") {
        if let Some(bytes) = sync.fetch_op(&id).expect("fetch") {
            ops.push(canonical::decode::<Operation>(&bytes).expect("decode"));
        }
    }
    sub.ingest(&ops).expect("ingest")
}

#[test]
fn two_actors_converge_on_the_same_cell() {
    let dir = tempfile::tempdir().expect("tempdir");
    let sync_dir = dir.path().join("sync");

    let alice_key = SigningKey::from_bytes(&[1u8; 32]);
    let bob_key = SigningKey::from_bytes(&[2u8; 32]);

    let mut alice = Substrate::open(dir.path().join("alice")).expect("alice substrate");
    let mut bob = Substrate::open(dir.path().join("bob")).expect("bob substrate");

    let mut alice_sync = FilesystemSync::new(&sync_dir, "alice").expect("alice sync");
    let mut bob_sync = FilesystemSync::new(&sync_dir, "bob").expect("bob sync");

    // Concurrent writes to the SAME entity "shared" — bob's HLC is higher,
    // so HlcLastWriterWins must pick bob's value on both sides.
    let alice_op = assert_op(&alice_key, "shared", 1, Hlc::new(10, 0));
    let bob_op = assert_op(&bob_key, "shared", 2, Hlc::new(20, 0));
    alice.apply(&alice_op).expect("alice apply");
    bob.apply(&bob_op).expect("bob apply");

    // Diverged before sync.
    assert_ne!(
        alice.state_root(),
        bob.state_root(),
        "actors diverge pre-sync"
    );

    // Each shares its op, then pulls + ingests the peer's.
    share(&alice_sync, &alice_op);
    alice_sync
        .announce(&hex(&alice_op.id().0))
        .expect("alice announce");
    share(&bob_sync, &bob_op);
    bob_sync
        .announce(&hex(&bob_op.id().0))
        .expect("bob announce");

    let into_bob = pull_and_ingest(&bob_sync, "alice", &mut bob);
    let into_alice = pull_and_ingest(&alice_sync, "bob", &mut alice);
    assert_eq!(into_bob, 1, "bob ingests alice's one op");
    assert_eq!(into_alice, 1, "alice ingests bob's one op");

    // Converged: identical state root on both sides.
    assert_eq!(
        alice.state_root(),
        bob.state_root(),
        "both actors converge to one state after exchanging ops"
    );

    // The convergent state is verifiable and ingest is idempotent: a second
    // pull adds no ops and leaves the root unchanged.
    let root_before = alice.state_root();
    let again = pull_and_ingest(&alice_sync, "bob", &mut alice);
    assert_eq!(again, 0, "re-ingest is idempotent");
    assert_eq!(
        alice.state_root(),
        root_before,
        "idempotent ingest leaves the root"
    );

    for op in alice.oplog().all_ops_in_hlc_order().expect("ops") {
        op.verify_signature()
            .expect("every converged witness verifies");
    }
}

fn retract_op(actor: &SigningKey, entity: &str, attr_names: &[&str], hlc: Hlc) -> Operation {
    Operation::sign(
        OpBody {
            actor: PublicKey::from_signing_key(actor),
            parents: vec![],
            payload: OpPayload::Retract {
                attrs: attr_names.iter().map(|n| AttrId::new(*n)).collect(),
                entity: EntityId::new(entity),
            },
            timestamp: hlc,
        },
        actor,
    )
}

#[test]
fn two_actors_converge_with_a_removal() {
    // bl-029 (29a): actor A asserts a cell, actor B retracts it at a higher
    // HLC. After exchanging ops both converge with the cell removed — the
    // retract wins by HLC, independent of arrival order.
    let dir = tempfile::tempdir().expect("tempdir");
    let a_key = SigningKey::from_bytes(&[1u8; 32]);
    let b_key = SigningKey::from_bytes(&[2u8; 32]);

    let mut alice = Substrate::open(dir.path().join("a")).expect("a");
    let mut bob = Substrate::open(dir.path().join("b")).expect("b");

    let assertion = assert_op(&a_key, "shared", 1, Hlc::new(10, 0));
    let removal = retract_op(&b_key, "shared", &["v"], Hlc::new(20, 0));
    alice.apply(&assertion).expect("alice asserts");
    bob.apply(&removal).expect("bob retracts");

    // Exchange: each ingests the other's op.
    alice
        .ingest(std::slice::from_ref(&removal))
        .expect("alice ingests retract");
    bob.ingest(std::slice::from_ref(&assertion))
        .expect("bob ingests assert");

    assert_eq!(
        alice.state_root(),
        bob.state_root(),
        "converge after exchanging assert + retract"
    );
    // The convergent state has the cell removed: it equals a substrate that
    // only ever saw the (higher-HLC) retract.
    let mut reference = Substrate::open(dir.path().join("ref")).expect("ref");
    reference
        .apply(&retract_op(&b_key, "shared", &["v"], Hlc::new(20, 0)))
        .expect("ref");
    assert_eq!(
        alice.state_root(),
        reference.state_root(),
        "retract wins by HLC — the cell is absent on both actors"
    );
}

fn assert_attr(actor: &SigningKey, entity: &str, attr: &str, value: Value, hlc: Hlc) -> Operation {
    Operation::sign(
        OpBody {
            actor: PublicKey::from_signing_key(actor),
            parents: vec![],
            payload: OpPayload::EntityAssert {
                attrs: vec![(AttrId::new(attr), value)],
                entity: EntityId::new(entity),
                valid_from: None,
            },
            timestamp: hlc,
        },
        actor,
    )
}

#[test]
fn union_domain_converges_across_actors_via_substrate() {
    // bl-029 (29b end-to-end): a Union ("tags") domain wired through
    // Substrate::open_with_policy converges by SET UNION across actors —
    // both elements are retained, unlike last-writer-wins.
    use ontoref_commit::{CellMergeKind, MergePolicy};

    let dir = tempfile::tempdir().expect("tempdir");
    let policy = MergePolicy::new().with("tags", CellMergeKind::Union);
    let a_key = SigningKey::from_bytes(&[1u8; 32]);
    let b_key = SigningKey::from_bytes(&[2u8; 32]);

    let mut alice = Substrate::open_with_policy(dir.path().join("a"), policy.clone()).expect("a");
    let mut bob = Substrate::open_with_policy(dir.path().join("b"), policy.clone()).expect("b");

    let oa = assert_attr(
        &a_key,
        "doc",
        "tags",
        Value::Str("red".into()),
        Hlc::new(10, 0),
    );
    let ob = assert_attr(
        &b_key,
        "doc",
        "tags",
        Value::Str("blue".into()),
        Hlc::new(20, 0),
    );
    alice.apply(&oa).expect("alice");
    bob.apply(&ob).expect("bob");

    alice
        .ingest(std::slice::from_ref(&ob))
        .expect("alice ingest");
    bob.ingest(std::slice::from_ref(&oa)).expect("bob ingest");

    assert_eq!(
        alice.state_root(),
        bob.state_root(),
        "union domain converges across actors"
    );

    // The converged root retains BOTH tags: it differs from a substrate that
    // only ever saw one element (proving union, not last-writer-wins).
    let mut only_one = Substrate::open_with_policy(dir.path().join("c"), policy).expect("c");
    only_one
        .apply(&assert_attr(
            &a_key,
            "doc",
            "tags",
            Value::Str("red".into()),
            Hlc::new(10, 0),
        ))
        .expect("c apply");
    assert_ne!(
        alice.state_root(),
        only_one.state_root(),
        "both actors' elements are retained — union, not overwrite"
    );
}

#[test]
fn semantic_validator_quarantines_a_peer_op_post_merge() {
    // bl-030: a peer op that passes structural checks (good signature,
    // valid HLC) but violates a semantic rule must be rejected by the
    // injected validator — never admitted into the merged state.
    let dir = tempfile::tempdir().expect("tempdir");
    let peer_key = SigningKey::from_bytes(&[3u8; 32]);

    let mut local = Substrate::open(dir.path().join("local")).expect("substrate");

    let good = assert_op(&peer_key, "ok", 1, Hlc::new(5, 0));
    let bad = assert_op(&peer_key, "forbidden", 999, Hlc::new(6, 0));

    // Domain rule: reject assertions to the "forbidden" entity.
    let report = local
        .ingest_with(&[good.clone(), bad.clone()], |op| match &op.body.payload {
            OpPayload::EntityAssert { entity, .. } if entity.0 == "forbidden" => {
                Err("entity 'forbidden' is not admissible".to_owned())
            }
            _ => Ok(()),
        })
        .expect("ingest_with");

    assert_eq!(report.ingested, 1, "only the admissible op is merged");
    assert_eq!(report.rejected.len(), 1, "the forbidden op is quarantined");
    assert_eq!(report.rejected[0].0, bad.id());

    // The rejected op is absent from the durable log — not just unapplied.
    let ids: Vec<_> = local
        .oplog()
        .all_ops_in_hlc_order()
        .expect("ops")
        .into_iter()
        .map(|o| o.id())
        .collect();
    assert!(ids.contains(&good.id()));
    assert!(
        !ids.contains(&bad.id()),
        "quarantined op never reaches the oplog"
    );
}
