meshdb-executor 0.2.0

use crate::error::Result;
use meshdb_core::{Edge, EdgeId, Node, NodeId, Property};
use meshdb_storage::{PropertyConstraintSpec, StorageEngine};

/// Read-side counterpart to [`crate::GraphWriter`]. Gives the executor a
/// uniform view of the graph regardless of whether the data behind it lives
/// entirely in a local storage engine (single-node or full-replica Raft
/// mode) or is sharded across cluster peers (routing mode, where a
/// partitioned reader fans out point reads to owners and scatter-gathers
/// bulk scans).
///
/// Methods are sync because the executor's iterator model is sync. Async-
/// backed implementations (e.g. a remote reader that talks gRPC) bridge via
/// `Handle::block_on`; callers must run the executor inside `spawn_blocking`
/// so they don't stall the tokio runtime.
pub trait GraphReader: Send + Sync {
    fn get_node(&self, id: NodeId) -> Result<Option<Node>>;
    fn get_edge(&self, id: EdgeId) -> Result<Option<Edge>>;
    fn all_node_ids(&self) -> Result<Vec<NodeId>>;
    fn nodes_by_label(&self, label: &str) -> Result<Vec<NodeId>>;
    /// Equality lookup via a property index. Callers (planner) must
    /// have verified the `(label, property)` index exists before
    /// emitting this call — fallback implementations are free to do a
    /// label-scan-and-filter for impls that don't maintain their own
    /// property index, but the storage-backed reader treats a call on
    /// a non-existent index as an empty result since no entries are
    /// maintained.
    fn nodes_by_property(
        &self,
        label: &str,
        property: &str,
        value: &Property,
    ) -> Result<Vec<NodeId>>;
    /// Composite form of [`Self::nodes_by_property`]. `properties`
    /// and `values` are parallel slices of equal length — every
    /// slot must be present for the call to land a match. The
    /// default impl delegates to `nodes_by_property` for length-1
    /// slices and returns empty otherwise, so readers that haven't
    /// wired a native composite seek degrade to no-results rather
    /// than mis-answering. The storage-backed blanket overrides
    /// with a real composite lookup.
    fn nodes_by_properties(
        &self,
        label: &str,
        properties: &[String],
        values: &[Property],
    ) -> Result<Vec<NodeId>> {
        if properties.len() == 1 && values.len() == 1 {
            return self.nodes_by_property(label, &properties[0], &values[0]);
        }
        Ok(Vec::new())
    }
    /// Relationship-scope analogue of [`Self::nodes_by_property`].
    /// The planner only emits an `EdgeSeek` after confirming a
    /// `(edge_type, property)` index is registered via
    /// [`Self::list_edge_property_indexes`]. Default impl returns
    /// empty so readers that haven't wired a native seek path
    /// degrade to no-results rather than mis-answering; the
    /// storage-backed blanket overrides with a real lookup.
    fn edges_by_property(
        &self,
        _edge_type: &str,
        _property: &str,
        _value: &Property,
    ) -> Result<Vec<EdgeId>> {
        Ok(Vec::new())
    }
    /// Snapshot the `(label, property)` pairs of every property
    /// index visible through this reader. Used by `SHOW INDEXES`.
    /// Default impl returns empty — the storage-backed reader
    /// overrides via the blanket impl, and partitioned/overlay
    /// readers delegate to their bases.
    fn list_property_indexes(&self) -> Result<Vec<(String, Vec<String>)>> {
        Ok(Vec::new())
    }
    /// Relationship-scope analogue of [`Self::list_property_indexes`].
    /// Returns `(edge_type, property)` pairs for every registered
    /// edge property index. Default impl returns empty; overlay
    /// and partitioned readers delegate to their bases.
    fn list_edge_property_indexes(&self) -> Result<Vec<(String, Vec<String>)>> {
        Ok(Vec::new())
    }
    /// Snapshot the `(label, property)` pairs of every point /
    /// spatial index visible through this reader. Used by `SHOW
    /// POINT INDEXES`. Default impl returns empty; storage-backed
    /// readers override via the blanket impl.
    fn list_point_indexes(&self) -> Result<Vec<(String, String)>> {
        Ok(Vec::new())
    }
    /// Relationship-scope analogue of [`Self::list_point_indexes`].
    /// `(edge_type, property)` pairs. Default impl returns empty.
    fn list_edge_point_indexes(&self) -> Result<Vec<(String, String)>> {
        Ok(Vec::new())
    }
    /// Axis-aligned bounding-box range query over the point index
    /// `(label, property)`. Returns every node carrying `label`
    /// whose `property` is a `Property::Point` under `srid` that
    /// falls inside `[xlo..xhi] × [ylo..yhi]`.
    ///
    /// Default impl does the naive scan (iterate `nodes_by_label`,
    /// filter in memory) so that readers that don't maintain a
    /// spatial index — or that haven't wired a native range query —
    /// stay correct, just slow. Partitioned readers inherit this
    /// default and get cluster-wide correctness through the
    /// scatter-gather `nodes_by_label` underneath. The
    /// storage-backed blanket overrides with the Z-order seek.
    fn nodes_in_bbox(
        &self,
        label: &str,
        property: &str,
        srid: i32,
        xlo: f64,
        ylo: f64,
        xhi: f64,
        yhi: f64,
    ) -> Result<Vec<NodeId>> {
        let (lox, hix) = if xlo <= xhi { (xlo, xhi) } else { (xhi, xlo) };
        let (loy, hiy) = if ylo <= yhi { (ylo, yhi) } else { (yhi, ylo) };
        let mut result: Vec<NodeId> = Vec::new();
        for id in self.nodes_by_label(label)? {
            let Some(node) = self.get_node(id)? else {
                continue;
            };
            if let Some(Property::Point(p)) = node.properties.get(property) {
                if p.srid == srid && p.x >= lox && p.x <= hix && p.y >= loy && p.y <= hiy {
                    result.push(id);
                }
            }
        }
        Ok(result)
    }
    /// Relationship-scope analogue of [`Self::nodes_in_bbox`].
    /// Default impl returns empty — edge-scoped bbox queries aren't
    /// yet part of the planner's rewrite surface, so no read path
    /// exercises this on anything but the storage-backed blanket.
    /// When the edge point-seek lowering lands this default should
    /// grow a naive scan via `edges_by_type` (needs to be added to
    /// the trait first).
    fn edges_in_bbox(
        &self,
        _edge_type: &str,
        _property: &str,
        _srid: i32,
        _xlo: f64,
        _ylo: f64,
        _xhi: f64,
        _yhi: f64,
    ) -> Result<Vec<EdgeId>> {
        Ok(Vec::new())
    }
    /// Snapshot every registered constraint visible through this
    /// reader, for `SHOW CONSTRAINTS` and `db.constraints()`. Default
    /// impl returns empty; storage-backed readers override.
    fn list_property_constraints(&self) -> Result<Vec<PropertyConstraintSpec>> {
        Ok(Vec::new())
    }
    fn outgoing(&self, id: NodeId) -> Result<Vec<(EdgeId, NodeId)>>;
    fn incoming(&self, id: NodeId) -> Result<Vec<(EdgeId, NodeId)>>;
}

/// Blanket impl: any **sized** type that implements [`StorageEngine`]
/// is automatically a [`GraphReader`]. Covers the concrete
/// [`meshdb_storage::RocksDbStorageEngine`] — a `&RocksDbStorageEngine`
/// coerces to `&dyn GraphReader` via this blanket.
///
/// Not covered: `dyn StorageEngine` itself. Rust does not transitively
/// coerce `&dyn StorageEngine` to `&dyn GraphReader` because trait
/// objects of unrelated traits carry different vtables and there's no
/// supertrait relationship connecting them. Call sites that hold a
/// `&dyn StorageEngine` should use [`StorageReaderAdapter`] to wrap it
/// as a `GraphReader`.
impl<T: StorageEngine> GraphReader for T {
    fn get_node(&self, id: NodeId) -> Result<Option<Node>> {
        Ok(StorageEngine::get_node(self, id)?)
    }

    fn get_edge(&self, id: EdgeId) -> Result<Option<Edge>> {
        Ok(StorageEngine::get_edge(self, id)?)
    }

    fn all_node_ids(&self) -> Result<Vec<NodeId>> {
        Ok(StorageEngine::all_node_ids(self)?)
    }

    fn nodes_by_label(&self, label: &str) -> Result<Vec<NodeId>> {
        Ok(StorageEngine::nodes_by_label(self, label)?)
    }

    fn nodes_by_property(
        &self,
        label: &str,
        property: &str,
        value: &Property,
    ) -> Result<Vec<NodeId>> {
        Ok(StorageEngine::nodes_by_property(
            self, label, property, value,
        )?)
    }

    fn nodes_by_properties(
        &self,
        label: &str,
        properties: &[String],
        values: &[Property],
    ) -> Result<Vec<NodeId>> {
        Ok(StorageEngine::nodes_by_properties(
            self, label, properties, values,
        )?)
    }

    fn edges_by_property(
        &self,
        edge_type: &str,
        property: &str,
        value: &Property,
    ) -> Result<Vec<EdgeId>> {
        Ok(StorageEngine::edges_by_property(
            self, edge_type, property, value,
        )?)
    }

    fn list_property_indexes(&self) -> Result<Vec<(String, Vec<String>)>> {
        Ok(StorageEngine::list_property_indexes(self)
            .into_iter()
            .map(|s| (s.label, s.properties))
            .collect())
    }

    fn list_edge_property_indexes(&self) -> Result<Vec<(String, Vec<String>)>> {
        Ok(StorageEngine::list_edge_property_indexes(self)
            .into_iter()
            .map(|s| (s.edge_type, s.properties))
            .collect())
    }

    fn list_point_indexes(&self) -> Result<Vec<(String, String)>> {
        Ok(StorageEngine::list_point_indexes(self)
            .into_iter()
            .map(|s| (s.label, s.property))
            .collect())
    }

    fn list_edge_point_indexes(&self) -> Result<Vec<(String, String)>> {
        Ok(StorageEngine::list_edge_point_indexes(self)
            .into_iter()
            .map(|s| (s.edge_type, s.property))
            .collect())
    }

    fn nodes_in_bbox(
        &self,
        label: &str,
        property: &str,
        srid: i32,
        xlo: f64,
        ylo: f64,
        xhi: f64,
        yhi: f64,
    ) -> Result<Vec<NodeId>> {
        // Storage-backed readers go through the Z-order seek path;
        // skip the default naive scan.
        Ok(StorageEngine::nodes_in_bbox(
            self, label, property, srid, xlo, ylo, xhi, yhi,
        )?)
    }

    fn edges_in_bbox(
        &self,
        edge_type: &str,
        property: &str,
        srid: i32,
        xlo: f64,
        ylo: f64,
        xhi: f64,
        yhi: f64,
    ) -> Result<Vec<EdgeId>> {
        Ok(StorageEngine::edges_in_bbox(
            self, edge_type, property, srid, xlo, ylo, xhi, yhi,
        )?)
    }

    fn list_property_constraints(&self) -> Result<Vec<PropertyConstraintSpec>> {
        Ok(StorageEngine::list_property_constraints(self))
    }

    fn outgoing(&self, id: NodeId) -> Result<Vec<(EdgeId, NodeId)>> {
        Ok(StorageEngine::outgoing(self, id)?)
    }

    fn incoming(&self, id: NodeId) -> Result<Vec<(EdgeId, NodeId)>> {
        Ok(StorageEngine::incoming(self, id)?)
    }
}

/// Adapter that lets a `&dyn StorageEngine` act as a `GraphReader`.
/// Needed because trait objects of unrelated traits don't coerce —
/// see the note on the blanket `impl<T: StorageEngine> GraphReader for T`.
/// Wraps a trait-object reference; no heap allocation. Works for both
/// reads and writes via the sibling [`crate::writer::StorageWriterAdapter`].
pub struct StorageReaderAdapter<'a>(pub &'a dyn StorageEngine);

impl GraphReader for StorageReaderAdapter<'_> {
    fn get_node(&self, id: NodeId) -> Result<Option<Node>> {
        Ok(self.0.get_node(id)?)
    }

    fn get_edge(&self, id: EdgeId) -> Result<Option<Edge>> {
        Ok(self.0.get_edge(id)?)
    }

    fn all_node_ids(&self) -> Result<Vec<NodeId>> {
        Ok(self.0.all_node_ids()?)
    }

    fn nodes_by_label(&self, label: &str) -> Result<Vec<NodeId>> {
        Ok(self.0.nodes_by_label(label)?)
    }

    fn nodes_by_property(
        &self,
        label: &str,
        property: &str,
        value: &Property,
    ) -> Result<Vec<NodeId>> {
        Ok(self.0.nodes_by_property(label, property, value)?)
    }

    fn edges_by_property(
        &self,
        edge_type: &str,
        property: &str,
        value: &Property,
    ) -> Result<Vec<EdgeId>> {
        Ok(self.0.edges_by_property(edge_type, property, value)?)
    }

    fn list_property_indexes(&self) -> Result<Vec<(String, Vec<String>)>> {
        Ok(self
            .0
            .list_property_indexes()
            .into_iter()
            .map(|s| (s.label, s.properties))
            .collect())
    }

    fn list_edge_property_indexes(&self) -> Result<Vec<(String, Vec<String>)>> {
        Ok(self
            .0
            .list_edge_property_indexes()
            .into_iter()
            .map(|s| (s.edge_type, s.properties))
            .collect())
    }

    fn list_point_indexes(&self) -> Result<Vec<(String, String)>> {
        Ok(self
            .0
            .list_point_indexes()
            .into_iter()
            .map(|s| (s.label, s.property))
            .collect())
    }

    fn list_edge_point_indexes(&self) -> Result<Vec<(String, String)>> {
        Ok(self
            .0
            .list_edge_point_indexes()
            .into_iter()
            .map(|s| (s.edge_type, s.property))
            .collect())
    }

    fn nodes_in_bbox(
        &self,
        label: &str,
        property: &str,
        srid: i32,
        xlo: f64,
        ylo: f64,
        xhi: f64,
        yhi: f64,
    ) -> Result<Vec<NodeId>> {
        Ok(self
            .0
            .nodes_in_bbox(label, property, srid, xlo, ylo, xhi, yhi)?)
    }

    fn edges_in_bbox(
        &self,
        edge_type: &str,
        property: &str,
        srid: i32,
        xlo: f64,
        ylo: f64,
        xhi: f64,
        yhi: f64,
    ) -> Result<Vec<EdgeId>> {
        Ok(self
            .0
            .edges_in_bbox(edge_type, property, srid, xlo, ylo, xhi, yhi)?)
    }

    fn list_property_constraints(&self) -> Result<Vec<PropertyConstraintSpec>> {
        Ok(self.0.list_property_constraints())
    }

    fn outgoing(&self, id: NodeId) -> Result<Vec<(EdgeId, NodeId)>> {
        Ok(self.0.outgoing(id)?)
    }

    fn incoming(&self, id: NodeId) -> Result<Vec<(EdgeId, NodeId)>> {
        Ok(self.0.incoming(id)?)
    }
}