ontoref-code/crates/ontoref-query/tests/basic.rs

use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};

use ed25519_dalek::SigningKey;
use ontoref_triples::TripleStore;
use ontoref_types::{
    operation::{OpBody, OpPayload, Operation},
    AttrId, EntityId, Hlc, PublicKey, Value,
};

use ontoref_query::QueryEngine;

fn signing_key() -> SigningKey {
    SigningKey::from_bytes(&[13u8; 32])
}

fn assert_op(entity: &str, attrs: Vec<(&str, Value)>, tx: Hlc) -> Operation {
    let key = signing_key();
    let body = OpBody {
        actor: PublicKey::from_signing_key(&key),
        parents: vec![],
        payload: OpPayload::EntityAssert {
            attrs: attrs
                .into_iter()
                .map(|(name, value)| (AttrId::new(name), value))
                .collect(),
            entity: EntityId::new(entity),
            valid_from: None,
        },
        timestamp: tx,
    };
    Operation::sign(body, &key)
}

fn fresh_engine() -> (tempfile::TempDir, QueryEngine) {
    let dir = tempfile::tempdir().unwrap();
    let triples = TripleStore::open(dir.path().join("triples.redb")).unwrap();
    (dir, QueryEngine::new(triples))
}

#[test]
fn transitive_closure_linear_chain() {
    let (_dir, mut engine) = fresh_engine();
    engine
        .apply(&assert_op(
            "a",
            vec![("next", Value::Ref(EntityId::new("b")))],
            Hlc::new(1, 0),
        ))
        .unwrap();
    engine
        .apply(&assert_op(
            "b",
            vec![("next", Value::Ref(EntityId::new("c")))],
            Hlc::new(2, 0),
        ))
        .unwrap();
    engine
        .apply(&assert_op(
            "c",
            vec![("name", Value::Str("Cee".into()))],
            Hlc::new(3, 0),
        ))
        .unwrap();

    let reached = engine
        .transitive_closure(
            &EntityId::new("a"),
            &AttrId::new("next"),
            Hlc::new(10, 0),
            Hlc::new(10, 0),
        )
        .unwrap();
    let expected: std::collections::HashSet<EntityId> =
        ["a", "b", "c"].iter().map(|n| EntityId::new(*n)).collect();
    assert_eq!(reached, expected);
}

#[test]
fn transitive_closure_terminates_on_cycle() {
    let (_dir, mut engine) = fresh_engine();
    engine
        .apply(&assert_op(
            "a",
            vec![("next", Value::Ref(EntityId::new("b")))],
            Hlc::new(1, 0),
        ))
        .unwrap();
    engine
        .apply(&assert_op(
            "b",
            vec![("next", Value::Ref(EntityId::new("a")))],
            Hlc::new(2, 0),
        ))
        .unwrap();

    let reached = engine
        .transitive_closure(
            &EntityId::new("a"),
            &AttrId::new("next"),
            Hlc::new(10, 0),
            Hlc::new(10, 0),
        )
        .unwrap();
    let expected: std::collections::HashSet<EntityId> =
        ["a", "b"].iter().map(|n| EntityId::new(*n)).collect();
    assert_eq!(reached, expected);
}

#[test]
fn transitive_closure_respects_as_of_horizon() {
    let (_dir, mut engine) = fresh_engine();
    engine
        .apply(&assert_op(
            "a",
            vec![("next", Value::Ref(EntityId::new("b")))],
            Hlc::new(10, 0),
        ))
        .unwrap();
    engine
        .apply(&assert_op(
            "b",
            vec![("next", Value::Ref(EntityId::new("c")))],
            Hlc::new(20, 0),
        ))
        .unwrap();

    // Looking from before tx2: b has no successor yet.
    let early = engine
        .transitive_closure(
            &EntityId::new("a"),
            &AttrId::new("next"),
            Hlc::new(15, 0),
            Hlc::new(15, 0),
        )
        .unwrap();
    let expected_early: std::collections::HashSet<EntityId> =
        ["a", "b"].iter().map(|n| EntityId::new(*n)).collect();
    assert_eq!(early, expected_early);

    // After both ops: c is reachable.
    let late = engine
        .transitive_closure(
            &EntityId::new("a"),
            &AttrId::new("next"),
            Hlc::new(30, 0),
            Hlc::new(30, 0),
        )
        .unwrap();
    let expected_late: std::collections::HashSet<EntityId> = ["a", "b", "c"]
        .iter()
        .map(|n| EntityId::new(*n))
        .collect();
    assert_eq!(late, expected_late);
}

#[test]
fn transitive_closure_only_follows_named_ref_attr() {
    let (_dir, mut engine) = fresh_engine();
    engine
        .apply(&assert_op(
            "a",
            vec![
                ("next", Value::Ref(EntityId::new("b"))),
                ("other", Value::Ref(EntityId::new("z"))),
            ],
            Hlc::new(1, 0),
        ))
        .unwrap();
    engine
        .apply(&assert_op(
            "b",
            vec![("name", Value::Str("Bee".into()))],
            Hlc::new(2, 0),
        ))
        .unwrap();

    let reached = engine
        .transitive_closure(
            &EntityId::new("a"),
            &AttrId::new("next"),
            Hlc::new(10, 0),
            Hlc::new(10, 0),
        )
        .unwrap();
    let expected: std::collections::HashSet<EntityId> =
        ["a", "b"].iter().map(|n| EntityId::new(*n)).collect();
    assert_eq!(reached, expected, "must not follow `other` ref");
}

#[test]
fn subscription_delivers_once_per_apply() {
    let (_dir, mut engine) = fresh_engine();
    let counter = Arc::new(AtomicUsize::new(0));
    {
        let counter = Arc::clone(&counter);
        engine.subscribe(move |_op| {
            counter.fetch_add(1, Ordering::SeqCst);
        });
    }

    for i in 0..5u64 {
        engine
            .apply(&assert_op(
                "e",
                vec![("v", Value::Int(i as i64))],
                Hlc::new(i + 1, 0),
            ))
            .unwrap();
    }
    assert_eq!(counter.load(Ordering::SeqCst), 5);
}

#[test]
fn subscription_delivers_to_every_subscriber() {
    let (_dir, mut engine) = fresh_engine();
    let a = Arc::new(AtomicUsize::new(0));
    let b = Arc::new(AtomicUsize::new(0));
    {
        let a = Arc::clone(&a);
        engine.subscribe(move |_op| {
            a.fetch_add(1, Ordering::SeqCst);
        });
        let b = Arc::clone(&b);
        engine.subscribe(move |_op| {
            b.fetch_add(1, Ordering::SeqCst);
        });
    }
    for i in 0..3u64 {
        engine
            .apply(&assert_op(
                "e",
                vec![("v", Value::Int(i as i64))],
                Hlc::new(i + 1, 0),
            ))
            .unwrap();
    }
    assert_eq!(a.load(Ordering::SeqCst), 3);
    assert_eq!(b.load(Ordering::SeqCst), 3);
}

#[test]
fn subscription_delivers_op_payload_to_handler() {
    let (_dir, mut engine) = fresh_engine();
    let captured: Arc<Mutex<Vec<EntityId>>> = Arc::new(Mutex::new(Vec::new()));
    {
        let captured = Arc::clone(&captured);
        engine.subscribe(move |op| {
            if let OpPayload::EntityAssert { entity, .. } = &op.body.payload {
                captured.lock().unwrap().push(entity.clone());
            }
        });
    }

    engine
        .apply(&assert_op(
            "alpha",
            vec![("v", Value::Int(1))],
            Hlc::new(1, 0),
        ))
        .unwrap();
    engine
        .apply(&assert_op(
            "beta",
            vec![("v", Value::Int(2))],
            Hlc::new(2, 0),
        ))
        .unwrap();

    let seen = captured.lock().unwrap();
    assert_eq!(
        *seen,
        vec![EntityId::new("alpha"), EntityId::new("beta")]
    );
}