meshdb-storage 0.2.0

use meshdb_core::{Edge, Node, NodeId, Point, Property};
use meshdb_storage::{
    ConstraintScope, EdgePointIndexSpec, EdgePropertyIndexSpec, GraphMutation, PointIndexSpec,
    PropertyConstraintKind, PropertyIndexSpec, RocksDbStorageEngine as Store,
};
use tempfile::TempDir;

fn tmp_store() -> (Store, TempDir) {
    let dir = TempDir::new().unwrap();
    let store = Store::open(dir.path()).unwrap();
    (store, dir)
}

#[test]
fn put_get_node_roundtrips() {
    let (store, _dir) = tmp_store();
    let node = Node::new()
        .with_label("Person")
        .with_property("name", "Ada")
        .with_property("age", 37_i64);

    store.put_node(&node).unwrap();
    let fetched = store.get_node(node.id).unwrap().unwrap();
    assert_eq!(fetched, node);
}

#[test]
fn get_missing_node_returns_none() {
    let (store, _dir) = tmp_store();
    assert!(store.get_node(NodeId::new()).unwrap().is_none());
}

#[test]
fn edge_adjacency_is_bidirectional() {
    let (store, _dir) = tmp_store();
    let a = Node::new();
    let b = Node::new();
    store.put_node(&a).unwrap();
    store.put_node(&b).unwrap();

    let e = Edge::new("KNOWS", a.id, b.id).with_property("since", 2020_i64);
    store.put_edge(&e).unwrap();

    assert_eq!(store.outgoing(a.id).unwrap(), vec![(e.id, b.id)]);
    assert_eq!(store.incoming(b.id).unwrap(), vec![(e.id, a.id)]);
    assert!(store.outgoing(b.id).unwrap().is_empty());
    assert!(store.incoming(a.id).unwrap().is_empty());
}

#[test]
fn outgoing_isolates_per_node() {
    let (store, _dir) = tmp_store();
    let a = Node::new();
    let b = Node::new();
    let c = Node::new();
    for n in [&a, &b, &c] {
        store.put_node(n).unwrap();
    }
    let e1 = Edge::new("T", a.id, b.id);
    let e2 = Edge::new("T", a.id, c.id);
    let e3 = Edge::new("T", b.id, c.id);
    store.put_edge(&e1).unwrap();
    store.put_edge(&e2).unwrap();
    store.put_edge(&e3).unwrap();

    let mut out_a: Vec<_> = store
        .outgoing(a.id)
        .unwrap()
        .into_iter()
        .map(|(id, _)| id)
        .collect();
    out_a.sort();
    let mut expected = vec![e1.id, e2.id];
    expected.sort();
    assert_eq!(out_a, expected);

    assert_eq!(store.outgoing(b.id).unwrap(), vec![(e3.id, c.id)]);
}

#[test]
fn delete_edge_removes_both_adjacency_sides() {
    let (store, _dir) = tmp_store();
    let a = Node::new();
    let b = Node::new();
    store.put_node(&a).unwrap();
    store.put_node(&b).unwrap();
    let e = Edge::new("T", a.id, b.id);
    store.put_edge(&e).unwrap();

    store.delete_edge(e.id).unwrap();

    assert!(store.get_edge(e.id).unwrap().is_none());
    assert!(store.outgoing(a.id).unwrap().is_empty());
    assert!(store.incoming(b.id).unwrap().is_empty());
}

#[test]
fn detach_delete_removes_incident_edges_on_both_ends() {
    let (store, _dir) = tmp_store();
    let a = Node::new();
    let b = Node::new();
    let c = Node::new();
    for n in [&a, &b, &c] {
        store.put_node(n).unwrap();
    }
    let e_ab = Edge::new("T", a.id, b.id);
    let e_cb = Edge::new("T", c.id, b.id);
    let e_ba = Edge::new("T", b.id, a.id);
    store.put_edge(&e_ab).unwrap();
    store.put_edge(&e_cb).unwrap();
    store.put_edge(&e_ba).unwrap();

    store.detach_delete_node(b.id).unwrap();

    assert!(store.get_node(b.id).unwrap().is_none());
    assert!(store.get_edge(e_ab.id).unwrap().is_none());
    assert!(store.get_edge(e_cb.id).unwrap().is_none());
    assert!(store.get_edge(e_ba.id).unwrap().is_none());
    assert!(store.outgoing(a.id).unwrap().is_empty());
    assert!(store.incoming(a.id).unwrap().is_empty());
    assert!(store.outgoing(c.id).unwrap().is_empty());
}

#[test]
fn nodes_by_label_finds_all_matching_nodes() {
    let (store, _dir) = tmp_store();
    let ada = Node::new()
        .with_label("Person")
        .with_property("name", "Ada");
    let alan = Node::new()
        .with_label("Person")
        .with_property("name", "Alan");
    let py = Node::new()
        .with_label("Language")
        .with_property("name", "Python");
    store.put_node(&ada).unwrap();
    store.put_node(&alan).unwrap();
    store.put_node(&py).unwrap();

    let mut people = store.nodes_by_label("Person").unwrap();
    people.sort();
    let mut expected = vec![ada.id, alan.id];
    expected.sort();
    assert_eq!(people, expected);

    assert_eq!(store.nodes_by_label("Language").unwrap(), vec![py.id]);
    assert!(store.nodes_by_label("Nonexistent").unwrap().is_empty());
}

#[test]
fn multi_label_node_appears_under_each_label() {
    let (store, _dir) = tmp_store();
    let n = Node::new().with_label("Person").with_label("Employee");
    store.put_node(&n).unwrap();

    assert_eq!(store.nodes_by_label("Person").unwrap(), vec![n.id]);
    assert_eq!(store.nodes_by_label("Employee").unwrap(), vec![n.id]);
}

#[test]
fn put_node_diffs_labels_on_overwrite() {
    let (store, _dir) = tmp_store();
    let mut n = Node::new().with_label("Draft");
    store.put_node(&n).unwrap();
    assert_eq!(store.nodes_by_label("Draft").unwrap(), vec![n.id]);

    n.labels = vec!["Published".to_string()];
    store.put_node(&n).unwrap();

    assert!(store.nodes_by_label("Draft").unwrap().is_empty());
    assert_eq!(store.nodes_by_label("Published").unwrap(), vec![n.id]);
}

#[test]
fn detach_delete_removes_label_entries() {
    let (store, _dir) = tmp_store();
    let n = Node::new().with_label("Doomed");
    store.put_node(&n).unwrap();
    store.detach_delete_node(n.id).unwrap();
    assert!(store.nodes_by_label("Doomed").unwrap().is_empty());
}

#[test]
fn edges_by_type_finds_all_matching_edges() {
    let (store, _dir) = tmp_store();
    let a = Node::new();
    let b = Node::new();
    let c = Node::new();
    for n in [&a, &b, &c] {
        store.put_node(n).unwrap();
    }
    let knows_ab = Edge::new("KNOWS", a.id, b.id);
    let knows_ac = Edge::new("KNOWS", a.id, c.id);
    let likes_bc = Edge::new("LIKES", b.id, c.id);
    store.put_edge(&knows_ab).unwrap();
    store.put_edge(&knows_ac).unwrap();
    store.put_edge(&likes_bc).unwrap();

    let mut knows = store.edges_by_type("KNOWS").unwrap();
    knows.sort();
    let mut expected = vec![knows_ab.id, knows_ac.id];
    expected.sort();
    assert_eq!(knows, expected);

    assert_eq!(store.edges_by_type("LIKES").unwrap(), vec![likes_bc.id]);
    assert!(store.edges_by_type("HATES").unwrap().is_empty());
}

#[test]
fn delete_edge_removes_type_index_entry() {
    let (store, _dir) = tmp_store();
    let a = Node::new();
    let b = Node::new();
    store.put_node(&a).unwrap();
    store.put_node(&b).unwrap();
    let e = Edge::new("KNOWS", a.id, b.id);
    store.put_edge(&e).unwrap();

    store.delete_edge(e.id).unwrap();
    assert!(store.edges_by_type("KNOWS").unwrap().is_empty());
}

#[test]
fn detach_delete_removes_type_index_for_incident_edges() {
    let (store, _dir) = tmp_store();
    let a = Node::new();
    let b = Node::new();
    store.put_node(&a).unwrap();
    store.put_node(&b).unwrap();
    let e = Edge::new("KNOWS", a.id, b.id);
    store.put_edge(&e).unwrap();

    store.detach_delete_node(b.id).unwrap();
    assert!(store.edges_by_type("KNOWS").unwrap().is_empty());
}

#[test]
fn indexes_share_no_state_across_similar_prefixes() {
    let (store, _dir) = tmp_store();
    let p = Node::new().with_label("Pers");
    let person = Node::new().with_label("Person");
    store.put_node(&p).unwrap();
    store.put_node(&person).unwrap();

    assert_eq!(store.nodes_by_label("Pers").unwrap(), vec![p.id]);
    assert_eq!(store.nodes_by_label("Person").unwrap(), vec![person.id]);
}

#[test]
fn apply_batch_commits_multi_op_atomically() {
    let (store, _dir) = tmp_store();
    let a = Node::new()
        .with_label("Person")
        .with_property("name", "Ada");
    let b = Node::new()
        .with_label("Person")
        .with_property("name", "Grace");
    let edge = Edge::new("KNOWS", a.id, b.id);

    store
        .apply_batch(&[
            GraphMutation::PutNode(a.clone()),
            GraphMutation::PutNode(b.clone()),
            GraphMutation::PutEdge(edge.clone()),
        ])
        .unwrap();

    assert_eq!(store.get_node(a.id).unwrap().unwrap(), a);
    assert_eq!(store.get_node(b.id).unwrap().unwrap(), b);
    assert_eq!(store.get_edge(edge.id).unwrap().unwrap(), edge);

    // Both indexes should reflect the post-batch state.
    let person_ids = store.nodes_by_label("Person").unwrap();
    assert!(person_ids.contains(&a.id) && person_ids.contains(&b.id));
    assert_eq!(store.edges_by_type("KNOWS").unwrap(), vec![edge.id]);
}

#[test]
fn apply_batch_delete_edge_is_idempotent_for_missing_target() {
    let (store, _dir) = tmp_store();
    let a = Node::new();
    let b = Node::new();
    let edge = Edge::new("KNOWS", a.id, b.id);
    store.put_node(&a).unwrap();
    store.put_node(&b).unwrap();
    store.put_edge(&edge).unwrap();

    // Deleting twice in one batch must not error — the second sees the
    // edge already gone (live read) and skips.
    store
        .apply_batch(&[
            GraphMutation::DeleteEdge(edge.id),
            GraphMutation::DeleteEdge(edge.id),
        ])
        .unwrap();
    assert!(store.get_edge(edge.id).unwrap().is_none());
}

#[test]
fn apply_batch_detach_delete_then_reinsert_in_one_commit() {
    let (store, _dir) = tmp_store();
    let original = Node::new().with_label("Person").with_property("v", 1_i64);
    let id = original.id;
    store.put_node(&original).unwrap();

    // Replace an existing node atomically: delete then reinsert with a
    // different label set. The post-state must reflect the second put,
    // and label indexes must not leak the old label.
    let replacement = Node {
        id,
        labels: vec!["Hero".into()],
        properties: Default::default(),
    };
    store
        .apply_batch(&[
            GraphMutation::DetachDeleteNode(id),
            GraphMutation::PutNode(replacement.clone()),
        ])
        .unwrap();

    let fetched = store.get_node(id).unwrap().unwrap();
    assert_eq!(fetched.labels, vec!["Hero"]);
    assert!(store.nodes_by_label("Person").unwrap().is_empty());
    assert_eq!(store.nodes_by_label("Hero").unwrap(), vec![id]);
}

#[test]
fn apply_batch_empty_is_a_noop() {
    let (store, _dir) = tmp_store();
    store.apply_batch(&[]).unwrap();
    assert!(store.all_node_ids().unwrap().is_empty());
}

#[test]
fn store_reopens_and_data_persists() {
    let dir = TempDir::new().unwrap();
    let node = Node::new().with_label("Persistent");
    let node_id = node.id;
    {
        let store = Store::open(dir.path()).unwrap();
        store.put_node(&node).unwrap();
    }
    let store = Store::open(dir.path()).unwrap();
    let fetched = store.get_node(node_id).unwrap().unwrap();
    assert_eq!(fetched.labels, vec!["Persistent"]);
}

#[test]
fn create_checkpoint_produces_a_consistent_clone() {
    // Populate a source store, checkpoint it into a sibling dir,
    // then open that dir as a fresh Store and confirm the data is
    // visible. The checkpoint must reflect the source's state at
    // `create_checkpoint` call time, independent of any subsequent
    // mutations to the source.
    let src_dir = TempDir::new().unwrap();
    let cp_parent = TempDir::new().unwrap();
    let cp_path = cp_parent.path().join("snap");

    let src = Store::open(src_dir.path()).unwrap();
    let node = Node::new().with_label("Snapshotted");
    let node_id = node.id;
    src.put_node(&node).unwrap();

    src.create_checkpoint(&cp_path).unwrap();

    // After the checkpoint, mutate the source to prove the clone is
    // decoupled from subsequent writes.
    src.put_node(&Node::new().with_label("AfterSnapshot"))
        .unwrap();

    // Drop the source handle so the checkpoint dir isn't sharing any
    // open FDs with the live DB when we open it.
    drop(src);

    let cp = Store::open(&cp_path).unwrap();
    // The snapshotted node is visible via the checkpoint.
    let fetched = cp.get_node(node_id).unwrap().unwrap();
    assert_eq!(fetched.labels, vec!["Snapshotted"]);
    // The post-checkpoint write is NOT visible.
    let after: Vec<Vec<String>> = cp
        .all_nodes()
        .unwrap()
        .into_iter()
        .map(|n| n.labels)
        .collect();
    assert_eq!(after.len(), 1);
    assert_eq!(after[0], vec!["Snapshotted"]);
}

#[test]
fn property_index_create_backfills_existing_nodes() {
    let (store, _dir) = tmp_store();
    let ada = Node::new()
        .with_label("Person")
        .with_property("name", "Ada");
    let bob = Node::new()
        .with_label("Person")
        .with_property("name", "Bob");
    let cid = Node::new().with_label("Robot").with_property("name", "C3");
    store.put_node(&ada).unwrap();
    store.put_node(&bob).unwrap();
    store.put_node(&cid).unwrap();

    store.create_property_index("Person", "name").unwrap();

    let ada_hits = store
        .nodes_by_property("Person", "name", &Property::String("Ada".into()))
        .unwrap();
    assert_eq!(ada_hits, vec![ada.id]);
    // Different label: unindexed, backfill excluded it.
    let c3_hits = store
        .nodes_by_property("Person", "name", &Property::String("C3".into()))
        .unwrap();
    assert!(c3_hits.is_empty());
}

#[test]
fn property_index_tracks_subsequent_puts_and_overwrites() {
    let (store, _dir) = tmp_store();
    store.create_property_index("Person", "name").unwrap();

    let mut ada = Node::new()
        .with_label("Person")
        .with_property("name", "Ada");
    store.put_node(&ada).unwrap();
    assert_eq!(
        store
            .nodes_by_property("Person", "name", &Property::String("Ada".into()))
            .unwrap(),
        vec![ada.id]
    );

    ada.properties
        .insert("name".into(), Property::String("Ada Lovelace".into()));
    store.put_node(&ada).unwrap();
    // Old value no longer resolves.
    assert!(store
        .nodes_by_property("Person", "name", &Property::String("Ada".into()))
        .unwrap()
        .is_empty());
    assert_eq!(
        store
            .nodes_by_property("Person", "name", &Property::String("Ada Lovelace".into()))
            .unwrap(),
        vec![ada.id]
    );
}

#[test]
fn property_index_detach_delete_removes_entry() {
    let (store, _dir) = tmp_store();
    store.create_property_index("Person", "name").unwrap();
    let ada = Node::new()
        .with_label("Person")
        .with_property("name", "Ada");
    store.put_node(&ada).unwrap();
    store.detach_delete_node(ada.id).unwrap();
    assert!(store
        .nodes_by_property("Person", "name", &Property::String("Ada".into()))
        .unwrap()
        .is_empty());
}

#[test]
fn property_index_drop_clears_entries_and_registry() {
    let (store, _dir) = tmp_store();
    store.create_property_index("Person", "name").unwrap();
    let ada = Node::new()
        .with_label("Person")
        .with_property("name", "Ada");
    store.put_node(&ada).unwrap();

    store.drop_property_index("Person", "name").unwrap();
    assert!(store.list_property_indexes().is_empty());
    // Re-creating with no backfill source still works; a subsequent
    // put_node also won't resurrect the stale entry because drop
    // swept the CF.
    store.create_property_index("Person", "name").unwrap();
    assert!(store
        .nodes_by_property("Person", "name", &Property::String("Ada".into()))
        .unwrap()
        .contains(&ada.id));
}

#[test]
fn property_index_registry_survives_reopen() {
    let dir = TempDir::new().unwrap();
    {
        let store = Store::open(dir.path()).unwrap();
        store.create_property_index("Person", "name").unwrap();
        store
            .put_node(
                &Node::new()
                    .with_label("Person")
                    .with_property("name", "Ada"),
            )
            .unwrap();
    }
    let store = Store::open(dir.path()).unwrap();
    assert_eq!(
        store.list_property_indexes(),
        vec![PropertyIndexSpec {
            label: "Person".into(),
            properties: vec!["name".into()],
        }]
    );
    let hits = store
        .nodes_by_property("Person", "name", &Property::String("Ada".into()))
        .unwrap();
    assert_eq!(hits.len(), 1);
}

#[test]
fn property_index_rejects_float_values_at_query_time() {
    let (store, _dir) = tmp_store();
    store.create_property_index("M", "score").unwrap();
    let err = store
        .nodes_by_property("M", "score", &Property::Float64(1.5))
        .unwrap_err();
    assert!(err.to_string().contains("not indexable"));
}

#[test]
fn property_index_tracks_label_change() {
    // Adding an indexed label to an existing node should start
    // indexing it; removing the label should drop the entry.
    let (store, _dir) = tmp_store();
    store.create_property_index("Person", "name").unwrap();

    let mut n = Node::new().with_label("Thing").with_property("name", "Ada");
    store.put_node(&n).unwrap();
    assert!(store
        .nodes_by_property("Person", "name", &Property::String("Ada".into()))
        .unwrap()
        .is_empty());

    n.labels.push("Person".into());
    store.put_node(&n).unwrap();
    assert_eq!(
        store
            .nodes_by_property("Person", "name", &Property::String("Ada".into()))
            .unwrap(),
        vec![n.id]
    );

    n.labels.retain(|l| l != "Person");
    store.put_node(&n).unwrap();
    assert!(store
        .nodes_by_property("Person", "name", &Property::String("Ada".into()))
        .unwrap()
        .is_empty());
}

#[test]
fn property_index_int64_and_bool_values_round_trip() {
    let (store, _dir) = tmp_store();
    store.create_property_index("P", "age").unwrap();
    store.create_property_index("P", "active").unwrap();
    let n = Node::new()
        .with_label("P")
        .with_property("age", 37_i64)
        .with_property("active", true);
    store.put_node(&n).unwrap();

    assert_eq!(
        store
            .nodes_by_property("P", "age", &Property::Int64(37))
            .unwrap(),
        vec![n.id]
    );
    assert_eq!(
        store
            .nodes_by_property("P", "active", &Property::Bool(true))
            .unwrap(),
        vec![n.id]
    );
    assert!(store
        .nodes_by_property("P", "age", &Property::Int64(38))
        .unwrap()
        .is_empty());
}

#[test]
fn property_index_string_int_same_value_do_not_alias() {
    // Ensures the type-tag byte prevents a String "42" from aliasing
    // an Int64 42 under the same (label, prop) prefix.
    let (store, _dir) = tmp_store();
    store.create_property_index("M", "x").unwrap();
    let a = Node::new().with_label("M").with_property("x", "42");
    let b = Node::new().with_label("M").with_property("x", 42_i64);
    store.put_node(&a).unwrap();
    store.put_node(&b).unwrap();
    assert_eq!(
        store
            .nodes_by_property("M", "x", &Property::String("42".into()))
            .unwrap(),
        vec![a.id]
    );
    assert_eq!(
        store
            .nodes_by_property("M", "x", &Property::Int64(42))
            .unwrap(),
        vec![b.id]
    );
}

#[test]
fn edge_property_index_create_backfills_existing_edges() {
    let (store, _dir) = tmp_store();
    let a = Node::new();
    let b = Node::new();
    store.put_node(&a).unwrap();
    store.put_node(&b).unwrap();
    let e1 = Edge::new("KNOWS", a.id, b.id).with_property("since", 2020_i64);
    let e2 = Edge::new("KNOWS", a.id, b.id).with_property("since", 2024_i64);
    let e3 = Edge::new("WORKS_AT", a.id, b.id).with_property("since", 2020_i64);
    store.put_edge(&e1).unwrap();
    store.put_edge(&e2).unwrap();
    store.put_edge(&e3).unwrap();

    store.create_edge_property_index("KNOWS", "since").unwrap();

    let hits = store
        .edges_by_property("KNOWS", "since", &Property::Int64(2020))
        .unwrap();
    assert_eq!(hits, vec![e1.id]);
    // Different type — backfill left it alone.
    let ignored = store
        .edges_by_property("KNOWS", "since", &Property::Int64(9999))
        .unwrap();
    assert!(ignored.is_empty());
    // Different edge type entirely, no index entry exists.
    let wrong_type = store
        .edges_by_property("WORKS_AT", "since", &Property::Int64(2020))
        .unwrap();
    assert!(wrong_type.is_empty());
}

#[test]
fn edge_property_index_tracks_puts_overwrites_and_deletes() {
    let (store, _dir) = tmp_store();
    store.create_edge_property_index("KNOWS", "since").unwrap();
    let a = Node::new();
    let b = Node::new();
    store.put_node(&a).unwrap();
    store.put_node(&b).unwrap();

    let mut e = Edge::new("KNOWS", a.id, b.id).with_property("since", 2020_i64);
    store.put_edge(&e).unwrap();
    assert_eq!(
        store
            .edges_by_property("KNOWS", "since", &Property::Int64(2020))
            .unwrap(),
        vec![e.id]
    );

    // Overwrite with a new value — old entry should be swept.
    e.properties.insert("since".into(), Property::Int64(2024));
    store.put_edge(&e).unwrap();
    assert!(store
        .edges_by_property("KNOWS", "since", &Property::Int64(2020))
        .unwrap()
        .is_empty());
    assert_eq!(
        store
            .edges_by_property("KNOWS", "since", &Property::Int64(2024))
            .unwrap(),
        vec![e.id]
    );

    // Delete drops the index entry.
    store.delete_edge(e.id).unwrap();
    assert!(store
        .edges_by_property("KNOWS", "since", &Property::Int64(2024))
        .unwrap()
        .is_empty());
}

#[test]
fn edge_property_index_detach_delete_sweeps_incident_edges() {
    // A DETACH DELETE on a node removes incident edges — the edge
    // property index needs to drop the corresponding entries along
    // with the edge row and its adjacency / type-index keys.
    let (store, _dir) = tmp_store();
    store.create_edge_property_index("KNOWS", "since").unwrap();
    let a = Node::new();
    let b = Node::new();
    let c = Node::new();
    for n in [&a, &b, &c] {
        store.put_node(n).unwrap();
    }
    let e1 = Edge::new("KNOWS", a.id, b.id).with_property("since", 2020_i64);
    let e2 = Edge::new("KNOWS", c.id, a.id).with_property("since", 2021_i64);
    store.put_edge(&e1).unwrap();
    store.put_edge(&e2).unwrap();

    store.detach_delete_node(a.id).unwrap();

    assert!(store
        .edges_by_property("KNOWS", "since", &Property::Int64(2020))
        .unwrap()
        .is_empty());
    assert!(store
        .edges_by_property("KNOWS", "since", &Property::Int64(2021))
        .unwrap()
        .is_empty());
}

#[test]
fn edge_property_index_drop_clears_entries_and_registry() {
    let (store, _dir) = tmp_store();
    store.create_edge_property_index("KNOWS", "since").unwrap();
    let a = Node::new();
    let b = Node::new();
    store.put_node(&a).unwrap();
    store.put_node(&b).unwrap();
    let e = Edge::new("KNOWS", a.id, b.id).with_property("since", 2020_i64);
    store.put_edge(&e).unwrap();

    store.drop_edge_property_index("KNOWS", "since").unwrap();
    assert!(store.list_edge_property_indexes().is_empty());

    // A fresh create rebuilds via backfill.
    store.create_edge_property_index("KNOWS", "since").unwrap();
    assert_eq!(
        store
            .edges_by_property("KNOWS", "since", &Property::Int64(2020))
            .unwrap(),
        vec![e.id]
    );
}

#[test]
fn edge_property_index_registry_survives_reopen() {
    let dir = TempDir::new().unwrap();
    {
        let store = Store::open(dir.path()).unwrap();
        store.create_edge_property_index("KNOWS", "since").unwrap();
        let a = Node::new();
        let b = Node::new();
        store.put_node(&a).unwrap();
        store.put_node(&b).unwrap();
        store
            .put_edge(&Edge::new("KNOWS", a.id, b.id).with_property("since", 2020_i64))
            .unwrap();
    }
    let store = Store::open(dir.path()).unwrap();
    assert_eq!(
        store.list_edge_property_indexes(),
        vec![EdgePropertyIndexSpec {
            edge_type: "KNOWS".into(),
            properties: vec!["since".into()],
        }]
    );
    let hits = store
        .edges_by_property("KNOWS", "since", &Property::Int64(2020))
        .unwrap();
    assert_eq!(hits.len(), 1);
}

#[test]
fn edge_property_index_rejects_float_values_at_query_time() {
    let (store, _dir) = tmp_store();
    store.create_edge_property_index("R", "weight").unwrap();
    let err = store
        .edges_by_property("R", "weight", &Property::Float64(1.5))
        .unwrap_err();
    assert!(err.to_string().contains("not indexable"));
}

#[test]
fn relationship_unique_constraint_provisions_backing_edge_index() {
    // Creating a UNIQUE constraint on a relationship type should
    // implicitly provision a (edge_type, property) index so every
    // subsequent put_edge routes through O(log N) enforcement
    // instead of the O(E_type) scan that preceded edge indexes.
    let (store, _dir) = tmp_store();
    store
        .create_property_constraint(
            Some("u_knows_since"),
            &ConstraintScope::Relationship("KNOWS".into()),
            &["since".to_string()],
            PropertyConstraintKind::Unique,
            false,
        )
        .unwrap();
    assert_eq!(
        store.list_edge_property_indexes(),
        vec![EdgePropertyIndexSpec {
            edge_type: "KNOWS".into(),
            properties: vec!["since".into()],
        }]
    );
}

#[test]
fn relationship_unique_detects_duplicate_on_insert() {
    let (store, _dir) = tmp_store();
    let a = Node::new();
    let b = Node::new();
    store.put_node(&a).unwrap();
    store.put_node(&b).unwrap();
    store
        .create_property_constraint(
            None,
            &ConstraintScope::Relationship("KNOWS".into()),
            &["since".to_string()],
            PropertyConstraintKind::Unique,
            false,
        )
        .unwrap();
    store
        .put_edge(&Edge::new("KNOWS", a.id, b.id).with_property("since", 2020_i64))
        .unwrap();
    let err = store
        .put_edge(&Edge::new("KNOWS", a.id, b.id).with_property("since", 2020_i64))
        .unwrap_err();
    assert!(err.to_string().contains("value already held"), "{err}");
}

#[test]
fn relationship_unique_allows_self_update_on_same_edge() {
    // A second put_edge that carries the same id should be treated
    // as an update — the existing holder is self, so no conflict.
    let (store, _dir) = tmp_store();
    let a = Node::new();
    let b = Node::new();
    store.put_node(&a).unwrap();
    store.put_node(&b).unwrap();
    store
        .create_property_constraint(
            None,
            &ConstraintScope::Relationship("KNOWS".into()),
            &["since".to_string()],
            PropertyConstraintKind::Unique,
            false,
        )
        .unwrap();
    let e = Edge::new("KNOWS", a.id, b.id).with_property("since", 2020_i64);
    store.put_edge(&e).unwrap();
    // Same id + same value should be a no-op update.
    let mut updated = e.clone();
    updated
        .properties
        .insert("since".into(), Property::Int64(2020));
    store.put_edge(&updated).unwrap();
}

// ---------------------------------------------------------------
// Composite property index tests (slice 3).
// ---------------------------------------------------------------

#[test]
fn composite_property_index_backfills_matching_nodes() {
    let (store, _dir) = tmp_store();
    // Three nodes: two match both props, one only has the first.
    store
        .put_node(
            &Node::new()
                .with_label("P")
                .with_property("first", "Ada")
                .with_property("last", "Lovelace"),
        )
        .unwrap();
    store
        .put_node(
            &Node::new()
                .with_label("P")
                .with_property("first", "Ada")
                .with_property("last", "Smith"),
        )
        .unwrap();
    store
        .put_node(&Node::new().with_label("P").with_property("first", "Ada"))
        .unwrap();

    store
        .create_property_index_composite("P", &["first".to_string(), "last".to_string()])
        .unwrap();
    let hits = store
        .nodes_by_properties(
            "P",
            &["first".to_string(), "last".to_string()],
            &[
                Property::String("Ada".into()),
                Property::String("Lovelace".into()),
            ],
        )
        .unwrap();
    assert_eq!(hits.len(), 1);
}

#[test]
fn composite_property_index_skips_partial_tuples() {
    // A node missing any of the composite properties contributes no
    // entry — composite indexes are all-or-nothing on the tuple.
    let (store, _dir) = tmp_store();
    store
        .create_property_index_composite("P", &["first".to_string(), "last".to_string()])
        .unwrap();
    // Partial: only `first` is set.
    store
        .put_node(&Node::new().with_label("P").with_property("first", "Ada"))
        .unwrap();
    let hits = store
        .nodes_by_properties(
            "P",
            &["first".to_string(), "last".to_string()],
            &[
                Property::String("Ada".into()),
                Property::String("Lovelace".into()),
            ],
        )
        .unwrap();
    assert!(hits.is_empty());
}

#[test]
fn composite_property_index_tracks_put_updates() {
    let (store, _dir) = tmp_store();
    store
        .create_property_index_composite("P", &["a".to_string(), "b".to_string()])
        .unwrap();
    let mut n = Node::new()
        .with_label("P")
        .with_property("a", 1_i64)
        .with_property("b", 2_i64);
    let id = n.id;
    store.put_node(&n).unwrap();
    assert_eq!(
        store
            .nodes_by_properties(
                "P",
                &["a".to_string(), "b".to_string()],
                &[Property::Int64(1), Property::Int64(2)]
            )
            .unwrap(),
        vec![id]
    );

    // Update the tuple — old entry must drop, new entry appears.
    n.properties.insert("b".into(), Property::Int64(99));
    store.put_node(&n).unwrap();
    assert!(store
        .nodes_by_properties(
            "P",
            &["a".to_string(), "b".to_string()],
            &[Property::Int64(1), Property::Int64(2)]
        )
        .unwrap()
        .is_empty());
    assert_eq!(
        store
            .nodes_by_properties(
                "P",
                &["a".to_string(), "b".to_string()],
                &[Property::Int64(1), Property::Int64(99)]
            )
            .unwrap(),
        vec![id]
    );
}

#[test]
fn composite_property_index_detach_delete_sweeps_entries() {
    let (store, _dir) = tmp_store();
    store
        .create_property_index_composite("P", &["a".to_string(), "b".to_string()])
        .unwrap();
    let n = Node::new()
        .with_label("P")
        .with_property("a", 1_i64)
        .with_property("b", 2_i64);
    let id = n.id;
    store.put_node(&n).unwrap();
    store.detach_delete_node(id).unwrap();

    let hits = store
        .nodes_by_properties(
            "P",
            &["a".to_string(), "b".to_string()],
            &[Property::Int64(1), Property::Int64(2)],
        )
        .unwrap();
    assert!(hits.is_empty());
}

#[test]
fn composite_property_index_drop_clears_entries() {
    let (store, _dir) = tmp_store();
    store
        .create_property_index_composite("P", &["a".to_string(), "b".to_string()])
        .unwrap();
    store
        .put_node(
            &Node::new()
                .with_label("P")
                .with_property("a", 1_i64)
                .with_property("b", 2_i64),
        )
        .unwrap();
    store
        .drop_property_index_composite("P", &["a".to_string(), "b".to_string()])
        .unwrap();
    assert!(store.list_property_indexes().is_empty());
}

#[test]
fn composite_edge_index_backfills_and_seeks() {
    let (store, _dir) = tmp_store();
    let a = Node::new();
    let b = Node::new();
    store.put_node(&a).unwrap();
    store.put_node(&b).unwrap();
    store
        .put_edge(
            &Edge::new("KNOWS", a.id, b.id)
                .with_property("since", 2020_i64)
                .with_property("weight", 5_i64),
        )
        .unwrap();
    store
        .create_edge_property_index_composite("KNOWS", &["since".to_string(), "weight".to_string()])
        .unwrap();
    let hits = store
        .edges_by_properties(
            "KNOWS",
            &["since".to_string(), "weight".to_string()],
            &[Property::Int64(2020), Property::Int64(5)],
        )
        .unwrap();
    assert_eq!(hits.len(), 1);
}

#[test]
fn node_key_constraint_auto_provisions_composite_backing_index() {
    // NODE KEY needs an O(log N) enforcement path, so creating the
    // constraint auto-provisions a matching composite index on the
    // same `(label, properties)` shape. Mirrors how UNIQUE provisions
    // a single-property backing index.
    let (store, _dir) = tmp_store();
    store
        .create_property_constraint(
            Some("pk_person"),
            &ConstraintScope::Node("Person".into()),
            &["first".to_string(), "last".to_string()],
            PropertyConstraintKind::NodeKey,
            false,
        )
        .unwrap();

    // Backing index must exist on the same property tuple, in order.
    let indexes = store.list_property_indexes();
    assert!(
        indexes.iter().any(|s| s.label == "Person"
            && s.properties == vec!["first".to_string(), "last".to_string()]),
        "expected backing composite index on Person(first, last), got {indexes:?}",
    );
}

#[test]
fn node_key_constraint_backfilled_index_drives_duplicate_detection() {
    // Insert data first, then declare the constraint. The backing
    // index is populated via backfill during `create_property_index_composite`,
    // and a subsequent duplicate insert hits the constraint through
    // the seek rather than an O(N) scan.
    let (store, _dir) = tmp_store();
    let ada = Node::new()
        .with_label("Person")
        .with_property("first", "Ada")
        .with_property("last", "Lovelace");
    store.put_node(&ada).unwrap();

    store
        .create_property_constraint(
            None,
            &ConstraintScope::Node("Person".into()),
            &["first".to_string(), "last".to_string()],
            PropertyConstraintKind::NodeKey,
            false,
        )
        .unwrap();

    // Duplicate tuple — must be rejected by the constraint check.
    let dup = Node::new()
        .with_label("Person")
        .with_property("first", "Ada")
        .with_property("last", "Lovelace");
    let err = store.put_node(&dup).unwrap_err();
    assert!(err.to_string().contains("tuple already held"), "{err}");

    // Self-update (same id, same tuple) must not false-positive on
    // the seek: the seek returns the existing id and the enforcement
    // loop skips it with `id != node.id`.
    store.put_node(&ada).unwrap();

    // A different tuple — allowed.
    let grace = Node::new()
        .with_label("Person")
        .with_property("first", "Grace")
        .with_property("last", "Hopper");
    store.put_node(&grace).unwrap();
}

#[test]
fn single_property_index_via_composite_api_matches_legacy_api() {
    // Length-1 call through the composite API must produce the same
    // seek results as the legacy single-property API. Pins the
    // backward-compatibility of the on-disk key format.
    let (store, _dir) = tmp_store();
    store
        .put_node(&Node::new().with_label("P").with_property("name", "Ada"))
        .unwrap();
    store
        .create_property_index_composite("P", &["name".to_string()])
        .unwrap();
    let via_composite = store
        .nodes_by_properties(
            "P",
            &["name".to_string()],
            &[Property::String("Ada".into())],
        )
        .unwrap();
    let via_legacy = store
        .nodes_by_property("P", "name", &Property::String("Ada".into()))
        .unwrap();
    assert_eq!(via_composite, via_legacy);
    assert_eq!(via_composite.len(), 1);
}

// ---------------------------------------------------------------
// Drop-index-through-backing-constraint guard.
//
// UNIQUE and NODE KEY both auto-provision a property index and
// enforce through it. If `DROP INDEX` removed that index while the
// constraint was still live, subsequent duplicate inserts would pass
// silently — the seek that drives enforcement would return "nothing"
// for every lookup. The guard below forces `DROP CONSTRAINT` first.
// ---------------------------------------------------------------

#[test]
fn drop_index_rejected_when_backing_node_key_constraint() {
    let (store, _dir) = tmp_store();
    store
        .create_property_constraint(
            Some("pk_person"),
            &ConstraintScope::Node("Person".into()),
            &["first".to_string(), "last".to_string()],
            PropertyConstraintKind::NodeKey,
            false,
        )
        .unwrap();
    let err = store
        .drop_property_index_composite("Person", &["first".to_string(), "last".to_string()])
        .unwrap_err();
    assert!(
        err.to_string().contains("pk_person"),
        "expected constraint name in error, got: {err}",
    );
}

#[test]
fn drop_index_rejected_when_backing_unique_constraint() {
    let (store, _dir) = tmp_store();
    store
        .create_property_constraint(
            Some("u_person_email"),
            &ConstraintScope::Node("Person".into()),
            &["email".to_string()],
            PropertyConstraintKind::Unique,
            false,
        )
        .unwrap();
    let err = store.drop_property_index("Person", "email").unwrap_err();
    assert!(
        err.to_string().contains("u_person_email"),
        "expected constraint name in error, got: {err}",
    );
}

#[test]
fn drop_edge_index_rejected_when_backing_unique_constraint() {
    let (store, _dir) = tmp_store();
    store
        .create_property_constraint(
            Some("u_knows_since"),
            &ConstraintScope::Relationship("KNOWS".into()),
            &["since".to_string()],
            PropertyConstraintKind::Unique,
            false,
        )
        .unwrap();
    let err = store
        .drop_edge_property_index("KNOWS", "since")
        .unwrap_err();
    assert!(
        err.to_string().contains("u_knows_since"),
        "expected constraint name in error, got: {err}",
    );
}

#[test]
fn drop_index_succeeds_after_drop_constraint() {
    // The prescribed escape hatch: `DROP CONSTRAINT` first, then the
    // backing index is free to drop.
    let (store, _dir) = tmp_store();
    store
        .create_property_constraint(
            Some("pk_person"),
            &ConstraintScope::Node("Person".into()),
            &["first".to_string(), "last".to_string()],
            PropertyConstraintKind::NodeKey,
            false,
        )
        .unwrap();
    store.drop_property_constraint("pk_person", false).unwrap();
    store
        .drop_property_index_composite("Person", &["first".to_string(), "last".to_string()])
        .unwrap();
    assert!(store.list_property_indexes().is_empty());
}

#[test]
fn drop_index_succeeds_when_no_backing_constraint() {
    // A plain user-created index with no constraint over it drops
    // freely — the guard only fires when the exact (label,
    // properties) tuple carries a UNIQUE/NODE KEY constraint.
    let (store, _dir) = tmp_store();
    store.create_property_index("Person", "nickname").unwrap();
    store.drop_property_index("Person", "nickname").unwrap();
    assert!(store.list_property_indexes().is_empty());
}

// ---------------------------------------------------------------
// Point / spatial index.
//
// Z-order cell range scans, with exact bbox filter on stored
// coords. Covers: backfill, incremental maintenance across
// put_node / detach_delete_node, bbox containment (hits /
// misses / edge cases), cross-SRID isolation, drop/list.
// ---------------------------------------------------------------

fn wgs84(lon: f64, lat: f64) -> Property {
    Property::Point(Point {
        srid: meshdb_core::SRID_WGS84_2D,
        x: lon,
        y: lat,
        z: None,
    })
}

#[test]
fn point_index_backfills_existing_nodes() {
    let (store, _dir) = tmp_store();
    let p1 = Node::new()
        .with_label("City")
        .with_property("loc", wgs84(13.4, 52.5)); // Berlin
    let p2 = Node::new()
        .with_label("City")
        .with_property("loc", wgs84(-73.9, 40.7)); // NYC
    store.put_node(&p1).unwrap();
    store.put_node(&p2).unwrap();

    store.create_point_index("City", "loc").unwrap();
    assert_eq!(
        store.list_point_indexes(),
        vec![PointIndexSpec {
            label: "City".into(),
            property: "loc".into(),
        }]
    );

    // Bbox enclosing Berlin only.
    let hits = store
        .nodes_in_bbox("City", "loc", 4326, 10.0, 50.0, 20.0, 55.0)
        .unwrap();
    assert_eq!(hits, vec![p1.id]);
}

#[test]
fn point_index_maintains_on_put_and_delete() {
    let (store, _dir) = tmp_store();
    store.create_point_index("City", "loc").unwrap();

    let berlin = Node::new()
        .with_label("City")
        .with_property("loc", wgs84(13.4, 52.5));
    store.put_node(&berlin).unwrap();

    let hits = store
        .nodes_in_bbox("City", "loc", 4326, 10.0, 50.0, 20.0, 55.0)
        .unwrap();
    assert_eq!(hits, vec![berlin.id]);

    // Move the point — replacing the node's `loc` with a point in
    // South America should kick the old cell entry out.
    let mut moved = berlin.clone();
    moved.properties.insert("loc".into(), wgs84(-58.4, -34.6)); // Buenos Aires
    store.put_node(&moved).unwrap();

    let hits = store
        .nodes_in_bbox("City", "loc", 4326, 10.0, 50.0, 20.0, 55.0)
        .unwrap();
    assert!(hits.is_empty(), "old cell entry wasn't swept: {hits:?}");

    let hits = store
        .nodes_in_bbox("City", "loc", 4326, -60.0, -40.0, -50.0, -30.0)
        .unwrap();
    assert_eq!(hits, vec![berlin.id]);

    // Detach-delete removes the index entry too.
    store.detach_delete_node(berlin.id).unwrap();
    let hits = store
        .nodes_in_bbox("City", "loc", 4326, -60.0, -40.0, -50.0, -30.0)
        .unwrap();
    assert!(hits.is_empty());
}

#[test]
fn point_index_skips_nodes_without_label() {
    // A node carrying the indexed property but not the label must
    // not appear in the index — the (label, property) tuple is the
    // scope, so label-filtering is mandatory.
    let (store, _dir) = tmp_store();
    store.create_point_index("City", "loc").unwrap();
    let not_a_city = Node::new().with_property("loc", wgs84(13.4, 52.5));
    store.put_node(&not_a_city).unwrap();
    let hits = store
        .nodes_in_bbox("City", "loc", 4326, 0.0, 0.0, 180.0, 90.0)
        .unwrap();
    assert!(hits.is_empty());
}

#[test]
fn point_index_skips_nodes_with_non_point_value() {
    // A node with the label but a String at `loc` must not land in
    // the index. Mirrors how `encode_index_tuple` ignores
    // unindexable values at maintenance time.
    let (store, _dir) = tmp_store();
    store.create_point_index("City", "loc").unwrap();
    let weird = Node::new().with_label("City").with_property("loc", "here");
    store.put_node(&weird).unwrap();
    let hits = store
        .nodes_in_bbox("City", "loc", 4326, -180.0, -90.0, 180.0, 90.0)
        .unwrap();
    assert!(hits.is_empty());
}

#[test]
fn point_index_isolates_srids() {
    // Two points at identical (x, y) but different SRIDs must not
    // alias — the SRID is part of the key prefix, so a bbox query
    // under one SRID never touches entries tagged with another.
    let (store, _dir) = tmp_store();
    store.create_point_index("Spot", "loc").unwrap();
    let geo = Node::new()
        .with_label("Spot")
        .with_property("loc", wgs84(5.0, 5.0));
    let cart = Node::new().with_label("Spot").with_property(
        "loc",
        Property::Point(Point {
            srid: meshdb_core::SRID_CARTESIAN_2D,
            x: 5.0,
            y: 5.0,
            z: None,
        }),
    );
    store.put_node(&geo).unwrap();
    store.put_node(&cart).unwrap();

    let wgs = store
        .nodes_in_bbox("Spot", "loc", 4326, 0.0, 0.0, 10.0, 10.0)
        .unwrap();
    assert_eq!(wgs, vec![geo.id]);

    let cartesian = store
        .nodes_in_bbox("Spot", "loc", 7203, 0.0, 0.0, 10.0, 10.0)
        .unwrap();
    assert_eq!(cartesian, vec![cart.id]);
}

#[test]
fn point_index_returns_empty_when_index_missing() {
    // A query against a non-existent point index returns an empty
    // vector rather than erroring — matches the property-seek
    // convention so a DROP / query race stays soft-failing.
    let (store, _dir) = tmp_store();
    let hits = store
        .nodes_in_bbox("City", "loc", 4326, 0.0, 0.0, 10.0, 10.0)
        .unwrap();
    assert!(hits.is_empty());
}

#[test]
fn point_index_drop_sweeps_all_srid_entries() {
    let (store, _dir) = tmp_store();
    store.create_point_index("Spot", "loc").unwrap();
    let geo = Node::new()
        .with_label("Spot")
        .with_property("loc", wgs84(5.0, 5.0));
    let cart = Node::new().with_label("Spot").with_property(
        "loc",
        Property::Point(Point {
            srid: meshdb_core::SRID_CARTESIAN_2D,
            x: 5.0,
            y: 5.0,
            z: None,
        }),
    );
    store.put_node(&geo).unwrap();
    store.put_node(&cart).unwrap();

    store.drop_point_index("Spot", "loc").unwrap();
    assert!(store.list_point_indexes().is_empty());

    // After drop, queries return empty under every SRID — the
    // label-prop prefix sweep covered all variants.
    assert!(store
        .nodes_in_bbox("Spot", "loc", 4326, 0.0, 0.0, 10.0, 10.0)
        .unwrap()
        .is_empty());
    assert!(store
        .nodes_in_bbox("Spot", "loc", 7203, 0.0, 0.0, 10.0, 10.0)
        .unwrap()
        .is_empty());
}

#[test]
fn point_index_create_is_idempotent() {
    let (store, _dir) = tmp_store();
    store.create_point_index("City", "loc").unwrap();
    store.create_point_index("City", "loc").unwrap();
    assert_eq!(store.list_point_indexes().len(), 1);
}

#[test]
fn point_index_bbox_filters_morton_overshoot() {
    // Morton order zigzags, so the cell range covering a bbox
    // includes cells outside the bbox. The per-row precision filter
    // must discard those. Plant one point inside the bbox and one
    // adjacent-but-outside, at coordinates chosen so they share a
    // Z-order neighbourhood; the result must contain only the
    // inside one.
    let (store, _dir) = tmp_store();
    store.create_point_index("P", "loc").unwrap();
    let inside = Node::new()
        .with_label("P")
        .with_property("loc", wgs84(5.0, 5.0));
    let outside = Node::new()
        .with_label("P")
        .with_property("loc", wgs84(5.5, 20.0)); // outside the bbox below
    store.put_node(&inside).unwrap();
    store.put_node(&outside).unwrap();

    let hits = store
        .nodes_in_bbox("P", "loc", 4326, 0.0, 0.0, 10.0, 10.0)
        .unwrap();
    assert_eq!(hits, vec![inside.id]);
}

#[test]
fn point_index_survives_reopen() {
    // Meta is loaded from the CF at open(), so the registry must
    // come back with the spec list intact and the index continues
    // to answer queries after a restart.
    let dir = tempfile::TempDir::new().unwrap();
    {
        let store = Store::open(dir.path()).unwrap();
        store.create_point_index("City", "loc").unwrap();
        let berlin = Node::new()
            .with_label("City")
            .with_property("loc", wgs84(13.4, 52.5));
        store.put_node(&berlin).unwrap();
    }
    let store = Store::open(dir.path()).unwrap();
    assert_eq!(
        store.list_point_indexes(),
        vec![PointIndexSpec {
            label: "City".into(),
            property: "loc".into(),
        }]
    );
    let hits = store
        .nodes_in_bbox("City", "loc", 4326, 10.0, 50.0, 20.0, 55.0)
        .unwrap();
    assert_eq!(hits.len(), 1);
}

// ---------------------------------------------------------------
// Edge point index — relationship-scope spatial tests.
//
// Mirror of the node-scope suite (backfill, maintenance, drop,
// bbox containment, label isolation, SRID isolation).
// ---------------------------------------------------------------

#[test]
fn edge_point_index_backfills_existing_edges() {
    let (store, _dir) = tmp_store();
    let a = Node::new();
    let b = Node::new();
    store.put_node(&a).unwrap();
    store.put_node(&b).unwrap();
    let e1 = Edge::new("ROUTE", a.id, b.id).with_property("waypoint", wgs84(13.4, 52.5));
    let e2 = Edge::new("ROUTE", a.id, b.id).with_property("waypoint", wgs84(-73.9, 40.7));
    store.put_edge(&e1).unwrap();
    store.put_edge(&e2).unwrap();

    store.create_edge_point_index("ROUTE", "waypoint").unwrap();
    assert_eq!(
        store.list_edge_point_indexes(),
        vec![EdgePointIndexSpec {
            edge_type: "ROUTE".into(),
            property: "waypoint".into(),
        }]
    );

    // Bbox around Berlin only.
    let hits = store
        .edges_in_bbox("ROUTE", "waypoint", 4326, 10.0, 50.0, 20.0, 55.0)
        .unwrap();
    assert_eq!(hits, vec![e1.id]);
}

#[test]
fn edge_point_index_maintains_on_put_and_delete() {
    let (store, _dir) = tmp_store();
    let a = Node::new();
    let b = Node::new();
    store.put_node(&a).unwrap();
    store.put_node(&b).unwrap();
    store.create_edge_point_index("ROUTE", "waypoint").unwrap();

    let berlin_edge = Edge::new("ROUTE", a.id, b.id).with_property("waypoint", wgs84(13.4, 52.5));
    store.put_edge(&berlin_edge).unwrap();

    let hits = store
        .edges_in_bbox("ROUTE", "waypoint", 4326, 10.0, 50.0, 20.0, 55.0)
        .unwrap();
    assert_eq!(hits, vec![berlin_edge.id]);

    // Move the waypoint — update replaces the index entry.
    let mut moved = berlin_edge.clone();
    moved
        .properties
        .insert("waypoint".into(), wgs84(-58.4, -34.6));
    store.put_edge(&moved).unwrap();
    assert!(store
        .edges_in_bbox("ROUTE", "waypoint", 4326, 10.0, 50.0, 20.0, 55.0)
        .unwrap()
        .is_empty());
    assert_eq!(
        store
            .edges_in_bbox("ROUTE", "waypoint", 4326, -60.0, -40.0, -50.0, -30.0)
            .unwrap(),
        vec![berlin_edge.id]
    );

    // Delete the edge → index entry swept.
    store.delete_edge(berlin_edge.id).unwrap();
    assert!(store
        .edges_in_bbox("ROUTE", "waypoint", 4326, -60.0, -40.0, -50.0, -30.0)
        .unwrap()
        .is_empty());
}

#[test]
fn edge_point_index_skips_edges_of_other_types() {
    // The (edge_type, property) tuple is the scope — edges of a
    // different type carrying the same property name must NOT land
    // in the index.
    let (store, _dir) = tmp_store();
    let a = Node::new();
    let b = Node::new();
    store.put_node(&a).unwrap();
    store.put_node(&b).unwrap();
    store.create_edge_point_index("ROUTE", "waypoint").unwrap();
    let right_type = Edge::new("ROUTE", a.id, b.id).with_property("waypoint", wgs84(5.0, 5.0));
    let wrong_type = Edge::new("FLIGHT", a.id, b.id).with_property("waypoint", wgs84(5.0, 5.0));
    store.put_edge(&right_type).unwrap();
    store.put_edge(&wrong_type).unwrap();

    let hits = store
        .edges_in_bbox("ROUTE", "waypoint", 4326, 0.0, 0.0, 10.0, 10.0)
        .unwrap();
    assert_eq!(hits, vec![right_type.id]);
}

#[test]
fn edge_point_index_isolates_srids() {
    let (store, _dir) = tmp_store();
    let a = Node::new();
    let b = Node::new();
    store.put_node(&a).unwrap();
    store.put_node(&b).unwrap();
    store.create_edge_point_index("ROUTE", "waypoint").unwrap();
    let geo = Edge::new("ROUTE", a.id, b.id).with_property("waypoint", wgs84(5.0, 5.0));
    let cart = Edge::new("ROUTE", a.id, b.id).with_property(
        "waypoint",
        Property::Point(Point {
            srid: meshdb_core::SRID_CARTESIAN_2D,
            x: 5.0,
            y: 5.0,
            z: None,
        }),
    );
    store.put_edge(&geo).unwrap();
    store.put_edge(&cart).unwrap();

    assert_eq!(
        store
            .edges_in_bbox("ROUTE", "waypoint", 4326, 0.0, 0.0, 10.0, 10.0)
            .unwrap(),
        vec![geo.id]
    );
    assert_eq!(
        store
            .edges_in_bbox("ROUTE", "waypoint", 7203, 0.0, 0.0, 10.0, 10.0)
            .unwrap(),
        vec![cart.id]
    );
}

#[test]
fn edge_point_index_drop_sweeps_all_srid_entries() {
    let (store, _dir) = tmp_store();
    let a = Node::new();
    let b = Node::new();
    store.put_node(&a).unwrap();
    store.put_node(&b).unwrap();
    store.create_edge_point_index("ROUTE", "waypoint").unwrap();
    store
        .put_edge(&Edge::new("ROUTE", a.id, b.id).with_property("waypoint", wgs84(5.0, 5.0)))
        .unwrap();

    store.drop_edge_point_index("ROUTE", "waypoint").unwrap();
    assert!(store.list_edge_point_indexes().is_empty());
    assert!(store
        .edges_in_bbox("ROUTE", "waypoint", 4326, 0.0, 0.0, 10.0, 10.0)
        .unwrap()
        .is_empty());
}

#[test]
fn edge_point_index_detach_delete_node_sweeps_incident_edges() {
    // When a node is detach-deleted, every edge incident to it goes
    // away — including its edge-point-index entries. Mirror of the
    // detach_delete_node coverage for property indexes.
    let (store, _dir) = tmp_store();
    let a = Node::new();
    let b = Node::new();
    store.put_node(&a).unwrap();
    store.put_node(&b).unwrap();
    store.create_edge_point_index("ROUTE", "waypoint").unwrap();
    let e = Edge::new("ROUTE", a.id, b.id).with_property("waypoint", wgs84(5.0, 5.0));
    store.put_edge(&e).unwrap();

    store.detach_delete_node(a.id).unwrap();
    assert!(store
        .edges_in_bbox("ROUTE", "waypoint", 4326, 0.0, 0.0, 10.0, 10.0)
        .unwrap()
        .is_empty());
}

#[test]
fn edge_point_index_survives_reopen() {
    let dir = tempfile::TempDir::new().unwrap();
    {
        let store = Store::open(dir.path()).unwrap();
        let a = Node::new();
        let b = Node::new();
        store.put_node(&a).unwrap();
        store.put_node(&b).unwrap();
        store.create_edge_point_index("ROUTE", "waypoint").unwrap();
        store
            .put_edge(&Edge::new("ROUTE", a.id, b.id).with_property("waypoint", wgs84(13.4, 52.5)))
            .unwrap();
    }
    let store = Store::open(dir.path()).unwrap();
    assert_eq!(
        store.list_edge_point_indexes(),
        vec![EdgePointIndexSpec {
            edge_type: "ROUTE".into(),
            property: "waypoint".into(),
        }]
    );
    assert_eq!(
        store
            .edges_in_bbox("ROUTE", "waypoint", 4326, 10.0, 50.0, 20.0, 55.0)
            .unwrap()
            .len(),
        1
    );
}