reddb_engine/storage/

entity.rs

//! Unified Entity Model
//!
//! Provides a single entity type that can represent table rows, graph nodes,
//! graph edges, or vectors with seamless interoperability.

use std::collections::HashMap;
use std::sync::Arc;

use reddb_types::Value;

/// The first entity-id handed to user-inserted data. Ids `1..FIRST_USER_ENTITY_ID`
/// are reserved for the internal collection-descriptor and config-default entities
/// the engine seeds at boot, so the first user-inserted `rid` is a STABLE,
/// documented value regardless of how many config defaults a build ships.
///
/// Before this floor existed the offset drifted upward by one for every config
/// default added (101 → 114 over time), silently breaking the documented
/// file-format invariant (#1369). The boot sequence bumps the allocator up to
/// this floor after seeding internals; it only ever raises the counter, so a
/// database that already holds user data is untouched. Mirrors
/// `FIRST_USER_LABEL_ID` in the graph label registry.
pub const FIRST_USER_ENTITY_ID: u64 = 1024;

pub use super::entity_id::EntityId;

/// The kind of entity (what storage type it belongs to)
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub enum EntityKind {
    /// A row in a structured table (hot path — kept inline for cache performance)
    TableRow { table: Arc<str>, row_id: u64 },
    /// A node in the graph (boxed — saves ~56 bytes per entity for table rows)
    GraphNode(Box<GraphNodeKind>),
    /// An edge in the graph (boxed)
    GraphEdge(Box<GraphEdgeKind>),
    /// A vector in a collection
    Vector { collection: String },
    /// A time-series data point (boxed)
    TimeSeriesPoint(Box<TimeSeriesPointKind>),
    /// A queue message
    QueueMessage { queue: String, position: u64 },
}

#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct GraphNodeKind {
    pub label: String,
    pub node_type: String,
}

#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct GraphEdgeKind {
    pub label: String,
    pub from_node: String,
    pub to_node: String,
    pub weight: u32,
}

#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct TimeSeriesPointKind {
    pub series: String,
    pub metric: String,
}

impl EntityKind {
    /// Get the storage type as a string
    pub fn storage_type(&self) -> &'static str {
        match self {
            Self::TableRow { .. } => "table",
            Self::GraphNode(_) => "graph_node",
            Self::GraphEdge(_) => "graph_edge",
            Self::Vector { .. } => "vector",
            Self::TimeSeriesPoint(_) => "timeseries",
            Self::QueueMessage { .. } => "queue",
        }
    }

    /// Get the collection/table name
    pub fn collection(&self) -> &str {
        match self {
            Self::TableRow { table, .. } => table,
            Self::GraphNode(n) => &n.label,
            Self::GraphEdge(e) => &e.label,
            Self::Vector { collection } => collection,
            Self::TimeSeriesPoint(ts) => &ts.series,
            Self::QueueMessage { queue, .. } => queue,
        }
    }
}

/// The actual data content of an entity
#[derive(Debug, Clone, PartialEq)]
pub enum EntityData {
    /// Table row data
    Row(RowData),
    /// Graph node data
    Node(NodeData),
    /// Graph edge data
    Edge(EdgeData),
    /// Vector data
    Vector(VectorData),
    /// Time-series data point
    TimeSeries(TimeSeriesData),
    /// Queue message data
    QueueMessage(QueueMessageData),
}

impl EntityData {
    /// Check if this is row data
    pub fn is_row(&self) -> bool {
        matches!(self, Self::Row(_))
    }

    /// Check if this is node data
    pub fn is_node(&self) -> bool {
        matches!(self, Self::Node(_))
    }

    /// Check if this is edge data
    pub fn is_edge(&self) -> bool {
        matches!(self, Self::Edge(_))
    }

    /// Check if this is vector data
    pub fn is_vector(&self) -> bool {
        matches!(self, Self::Vector(_))
    }

    /// Get as row data
    pub fn as_row(&self) -> Option<&RowData> {
        match self {
            Self::Row(r) => Some(r),
            _ => None,
        }
    }

    /// Get as node data
    pub fn as_node(&self) -> Option<&NodeData> {
        match self {
            Self::Node(n) => Some(n),
            _ => None,
        }
    }

    /// Get as edge data
    pub fn as_edge(&self) -> Option<&EdgeData> {
        match self {
            Self::Edge(e) => Some(e),
            _ => None,
        }
    }

    /// Get as vector data
    pub fn as_vector(&self) -> Option<&VectorData> {
        match self {
            Self::Vector(v) => Some(v),
            _ => None,
        }
    }
}

/// Data for a table row
#[derive(Debug, Clone, PartialEq)]
pub struct RowData {
    /// Column values in schema order
    pub columns: Vec<Value>,
    /// Named column access (optional, for convenience)
    pub named: Option<HashMap<String, Value>>,
    /// Shared column schema: column names in order (maps index → name).
    /// When set, `columns` holds the values and `named` is None.
    /// This saves ~60% memory vs per-row HashMap.
    pub schema: Option<std::sync::Arc<Vec<String>>>,
}

impl RowData {
    /// Create new row data from column values
    pub fn new(columns: Vec<Value>) -> Self {
        Self {
            columns,
            named: None,
            schema: None,
        }
    }

    /// Create row data with named columns
    pub fn with_names(columns: Vec<Value>, names: Vec<String>) -> Self {
        let named: HashMap<String, Value> =
            names.into_iter().zip(columns.iter().cloned()).collect();
        Self {
            columns,
            named: Some(named),
            schema: None,
        }
    }

    /// Replace a decoded named row with values in the authoritative schema's
    /// order. Different dynamic shapes retain their named representation.
    #[cfg(feature = "full")]
    pub(crate) fn compact_with_schema(&mut self, schema: &std::sync::Arc<Vec<String>>) -> bool {
        let Some(named) = &self.named else {
            return false;
        };
        if named.len() != schema.len() || schema.iter().any(|name| !named.contains_key(name)) {
            return false;
        }
        // Existing positional values are independently observable. Only drop
        // their duplicate copy when the declared order agrees with them.
        if !self.columns.is_empty()
            && (self.columns.len() != schema.len()
                || schema
                    .iter()
                    .zip(&self.columns)
                    .any(|(name, value)| named.get(name) != Some(value)))
        {
            return false;
        }
        let mut named = self.named.take().expect("validated named row");
        self.columns = schema
            .iter()
            .map(|name| named.remove(name).expect("validated schema field"))
            .collect();
        self.schema = Some(std::sync::Arc::clone(schema));
        true
    }

    /// Get a named field value — checks named HashMap first, then schema+columns.
    pub fn get_field(&self, name: &str) -> Option<&Value> {
        // Fast path: named HashMap
        if let Some(ref named) = self.named {
            return named.get(name);
        }
        // Columnar path: use schema to find index
        if let Some(ref schema) = self.schema {
            if let Some(idx) = schema.iter().position(|s| s == name) {
                return self.columns.get(idx);
            }
        }
        None
    }

    /// Iterate over all (name, value) pairs — works for both named and columnar.
    pub fn iter_fields(&self) -> Box<dyn Iterator<Item = (&str, &Value)> + '_> {
        if let Some(ref named) = self.named {
            Box::new(named.iter().map(|(k, v)| (k.as_str(), v)))
        } else if let Some(ref schema) = self.schema {
            Box::new(
                schema
                    .iter()
                    .zip(self.columns.iter())
                    .map(|(k, v)| (k.as_str(), v)),
            )
        } else {
            Box::new(std::iter::empty())
        }
    }

    /// Get column by index
    pub fn get(&self, index: usize) -> Option<&Value> {
        self.columns.get(index)
    }

    /// Get column by name
    pub fn get_by_name(&self, name: &str) -> Option<&Value> {
        self.named.as_ref()?.get(name)
    }

    /// Number of columns
    pub fn len(&self) -> usize {
        self.columns.len()
    }

    /// Check if empty
    pub fn is_empty(&self) -> bool {
        self.columns.is_empty()
    }
}

/// Data for a graph node
#[derive(Debug, Clone, PartialEq)]
pub struct NodeData {
    /// Node properties
    pub properties: HashMap<String, Value>,
}

impl NodeData {
    /// Create new node data
    pub fn new() -> Self {
        Self {
            properties: HashMap::new(),
        }
    }

    /// Create with properties
    pub fn with_properties(properties: HashMap<String, Value>) -> Self {
        Self { properties }
    }

    /// Set a property
    pub fn set(&mut self, key: impl Into<String>, value: Value) {
        self.properties.insert(key.into(), value);
    }

    /// Get a property
    pub fn get(&self, key: &str) -> Option<&Value> {
        self.properties.get(key)
    }

    /// Check if property exists
    pub fn has(&self, key: &str) -> bool {
        self.properties.contains_key(key)
    }
}

impl Default for NodeData {
    fn default() -> Self {
        Self::new()
    }
}

/// Data for a graph edge
#[derive(Debug, Clone, PartialEq)]
pub struct EdgeData {
    /// Edge properties
    pub properties: HashMap<String, Value>,
    /// Edge weight (for weighted graphs)
    pub weight: f32,
}

impl EdgeData {
    /// Create new edge data
    pub fn new(weight: f32) -> Self {
        Self {
            properties: HashMap::new(),
            weight,
        }
    }

    /// Create with properties
    pub fn with_properties(weight: f32, properties: HashMap<String, Value>) -> Self {
        Self { properties, weight }
    }

    /// Set a property
    pub fn set(&mut self, key: impl Into<String>, value: Value) {
        self.properties.insert(key.into(), value);
    }

    /// Get a property
    pub fn get(&self, key: &str) -> Option<&Value> {
        self.properties.get(key)
    }
}

impl Default for EdgeData {
    fn default() -> Self {
        Self::new(1.0)
    }
}

/// Data for a vector
#[derive(Debug, Clone, PartialEq)]
pub struct VectorData {
    /// Dense vector (primary embedding)
    pub dense: Vec<f32>,
    /// Optional sparse vector
    pub sparse: Option<SparseVector>,
    /// Original content (if applicable)
    pub content: Option<String>,
}

impl VectorData {
    /// Create new vector data from dense vector
    pub fn new(dense: Vec<f32>) -> Self {
        Self {
            dense,
            sparse: None,
            content: None,
        }
    }

    /// Create with sparse vector
    pub fn with_sparse(dense: Vec<f32>, sparse: SparseVector) -> Self {
        Self {
            dense,
            sparse: Some(sparse),
            content: None,
        }
    }

    /// Set content
    pub fn with_content(mut self, content: impl Into<String>) -> Self {
        self.content = Some(content.into());
        self
    }

    /// Get dimension
    pub fn dimension(&self) -> usize {
        self.dense.len()
    }

    /// Check if has sparse component
    pub fn is_hybrid(&self) -> bool {
        self.sparse.is_some()
    }
}

/// Time-series data point
#[derive(Debug, Clone, PartialEq)]
pub struct TimeSeriesData {
    /// Metric name (e.g., "cpu.idle")
    pub metric: String,
    /// Interned `(metric, canonical tags)` id for native time-series points.
    pub series_id: Option<u64>,
    /// Timestamp in nanoseconds since epoch
    pub timestamp_ns: u64,
    /// Metric value
    pub value: f64,
    /// Legacy/read-through dimensional tags (e.g., {"host": "srv1"}).
    pub tags: std::collections::HashMap<String, String>,
    /// Per-point fields that are not part of the immutable series identity.
    pub fields: std::collections::HashMap<String, Value>,
}

/// Queue message data
#[derive(Debug, Clone, PartialEq)]
pub struct QueueMessageData {
    /// Message payload
    pub payload: Value,
    /// Optional priority (higher = more urgent)
    pub priority: Option<i32>,
    /// Enqueue timestamp (nanoseconds)
    pub enqueued_at_ns: u64,
    /// Number of delivery attempts
    pub attempts: u32,
    /// Maximum delivery attempts before DLQ
    pub max_attempts: u32,
    /// Whether the message has been acknowledged
    pub acked: bool,
}

/// Sparse vector representation
#[derive(Debug, Clone, PartialEq)]
pub struct SparseVector {
    /// Indices of non-zero elements
    pub indices: Vec<u32>,
    /// Values at those indices
    pub values: Vec<f32>,
    /// Total dimension (may be larger than indices.len())
    pub dimension: usize,
}

impl SparseVector {
    /// Create new sparse vector
    pub fn new(indices: Vec<u32>, values: Vec<f32>, dimension: usize) -> Self {
        debug_assert_eq!(indices.len(), values.len());
        Self {
            indices,
            values,
            dimension,
        }
    }

    /// Number of non-zero elements
    pub fn nnz(&self) -> usize {
        self.indices.len()
    }

    /// Sparsity ratio
    pub fn sparsity(&self) -> f32 {
        if self.dimension == 0 {
            1.0
        } else {
            1.0 - (self.nnz() as f32 / self.dimension as f32)
        }
    }

    /// Get value at index (0 if not present)
    pub fn get(&self, index: u32) -> f32 {
        self.indices
            .iter()
            .position(|&i| i == index)
            .map(|pos| self.values[pos])
            .unwrap_or(0.0)
    }
}

/// A slot for embedding a specific aspect of an entity
#[derive(Debug, Clone, PartialEq)]
pub struct EmbeddingSlot {
    /// Slot name (e.g., "content", "summary", "title", "code")
    pub name: String,
    /// The embedding vector
    pub vector: Vec<f32>,
    /// Model used to generate embedding
    pub model: String,
    /// Vector dimension
    pub dimension: usize,
    /// Generation timestamp
    pub generated_at: u64,
}

use crate::clock::current_unix_secs;

impl EmbeddingSlot {
    /// Create a new embedding slot
    pub fn new(name: impl Into<String>, vector: Vec<f32>, model: impl Into<String>) -> Self {
        let dimension = vector.len();
        Self {
            name: name.into(),
            vector,
            model: model.into(),
            dimension,
            generated_at: current_unix_secs(),
        }
    }
}

/// A unified entity that can represent any storage type
#[derive(Debug, Clone)]
pub struct UnifiedEntity {
    /// Unique entity identifier
    pub id: EntityId,
    /// Stable user-visible identity shared by all physical versions.
    ///
    /// `None` is the legacy encoding and resolves to `id`.
    logical_id: Option<EntityId>,
    /// What kind of entity this is
    pub kind: EntityKind,
    /// Creation timestamp
    pub created_at: u64,
    /// Last update timestamp
    pub updated_at: u64,
    /// The actual data content
    pub data: EntityData,
    /// Sequence ID for ordering/versioning
    pub sequence_id: u64,
    /// Field-name bloom filter (u64, zero-allocation).
    ///
    /// Each bit encodes one possible mid-character value: for field name `n`
    /// the bit position is `n.as_bytes()[n.len()/2] & 63`. OR of all user
    /// field names present in this entity. Cleared for schema-based bulk rows
    /// (all rows share the same schema so bloom is segment-level).
    ///
    /// The compiled filter computes `required_bloom` from predicate field names
    /// at compile time. If `entity.field_bloom & required_bloom != required_bloom`,
    /// the entity cannot match and is skipped before any HashMap probe.
    pub field_bloom: u64,
    /// MVCC creation transaction ID (Phase 2.3 PG parity).
    ///
    /// `0` means "pre-MVCC" / auto-commit — visible to every snapshot. When
    /// a BEGIN-wrapped INSERT runs, it stamps `xmin` with the transaction's
    /// snapshot id so other concurrent transactions only see the row after
    /// the writer commits (snapshot isolation semantics).
    ///
    /// Visibility rule: `xmin <= snapshot.xid && (xmax == 0 || xmax > snapshot.xid)`.
    pub xmin: u64,
    /// MVCC deletion transaction ID (Phase 2.3 PG parity).
    ///
    /// `0` means "live". Set to the deleting transaction's snapshot id on
    /// DELETE/UPDATE (row is kept until VACUUM reclaims it). Snapshots with
    /// `xid < xmax` still see the row; newer snapshots skip it.
    pub xmax: u64,
    /// Optional auxiliary data (embeddings, cross-refs).
    /// None for most table rows — saves 40 bytes/entity.
    aux: Option<Box<EntityAux>>,
}

/// Auxiliary entity data — only allocated when needed.
#[derive(Debug, Clone, Default)]
pub struct EntityAux {
    /// Embedding slots (for multi-vector support)
    pub embeddings: Vec<EmbeddingSlot>,
    /// Cross-references to other entities
    pub cross_refs: Vec<CrossRef>,
    /// VCS provenance is allocated only for opted-in entities. Epochs are
    /// immutable branch incarnations; recreating a branch uses a new epoch.
    pub branch: Option<BranchVersion>,
}

#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct BranchVersion {
    pub origin: u64,
    /// Deletions by other epochs must not overwrite the creator's xmax.
    pub deletions: Vec<(u64, u64)>,
}

impl UnifiedEntity {
    /// Normalize persisted runtime-owned collection identities while retaining
    /// exact user collection names. This is idempotent and intentionally does
    /// not rewrite graph labels, which are data rather than collection keys.
    pub fn canonicalize_system_collection_names(&mut self) {
        match &mut self.kind {
            EntityKind::TableRow { table, .. } => {
                if let Some(canonical) = super::system_collection_names::canonical_name(table) {
                    *table = Arc::from(canonical);
                }
            }
            EntityKind::Vector { collection } => {
                if let Some(canonical) = super::system_collection_names::canonical_name(collection)
                {
                    *collection = canonical.to_owned();
                }
            }
            EntityKind::TimeSeriesPoint(point) => {
                if let Some(canonical) =
                    super::system_collection_names::canonical_name(&point.series)
                {
                    point.series = canonical.to_owned();
                }
            }
            EntityKind::QueueMessage { queue, .. } => {
                if let Some(canonical) = super::system_collection_names::canonical_name(queue) {
                    *queue = canonical.to_owned();
                }
            }
            EntityKind::GraphNode(_) | EntityKind::GraphEdge(_) => {}
        }
        if let Some(aux) = &mut self.aux {
            for reference in &mut aux.cross_refs {
                if let Some(canonical) =
                    super::system_collection_names::canonical_name(&reference.target_collection)
                {
                    reference.target_collection = canonical.to_owned();
                }
            }
        }
    }

    /// Access embeddings (returns empty slice if no aux data).
    pub fn embeddings(&self) -> &[EmbeddingSlot] {
        self.aux
            .as_ref()
            .map(|a| a.embeddings.as_slice())
            .unwrap_or(&[])
    }

    /// Access cross-references (returns empty slice if no aux data).
    pub fn cross_refs(&self) -> &[CrossRef] {
        self.aux
            .as_ref()
            .map(|a| a.cross_refs.as_slice())
            .unwrap_or(&[])
    }

    /// Get mutable embeddings (allocates aux if needed).
    pub fn embeddings_mut(&mut self) -> &mut Vec<EmbeddingSlot> {
        &mut self.aux.get_or_insert_with(Default::default).embeddings
    }

    /// Get mutable cross-refs (allocates aux if needed).
    pub fn cross_refs_mut(&mut self) -> &mut Vec<CrossRef> {
        &mut self.aux.get_or_insert_with(Default::default).cross_refs
    }

    /// Check if entity has any auxiliary data.
    pub fn has_aux(&self) -> bool {
        self.aux.is_some()
    }
}

/// Compute one bit of a field-name bloom filter.
///
/// Uses the mid-character trick from MongoDB's field-name bloom:
/// the bit position is the mid-byte value clamped to 0..63. Zero-allocation,
/// ~1.5% false-positive rate for ≤5 distinct field names.
#[inline]
pub fn field_name_bloom(name: &str) -> u64 {
    let b = name.as_bytes();
    if b.is_empty() {
        return 0;
    }
    1u64 << (b[b.len() / 2] & 63)
}

/// Compute the combined field-name bloom for all user-level fields in `data`.
/// Returns 0 for schema-based rows (all rows share the same schema, so the
/// per-entity bloom would be identical — caller uses a segment-level bloom).
pub fn compute_entity_field_bloom(data: &EntityData) -> u64 {
    match data {
        EntityData::Row(row) => {
            if row.schema.is_some() {
                // Schema path: bloom is identical for every row in this table.
                // Don't store per-entity — use segment-level bloom instead.
                return 0;
            }
            if let Some(named) = &row.named {
                let mut bloom = named.keys().fold(0u64, |acc, k| acc | field_name_bloom(k));
                // Single-source documents no longer materialise promoted
                // columns, so the body's top-level keys are the only place a
                // `WHERE`/projection field lives. Fold them into the bloom so
                // the field-bloom gate doesn't reject a row before the body
                // read fallback ever runs. Inert for non-document rows (no
                // binary-container `body` field) — `container_field_names`
                // self-gates on the RDOC magic.
                if let Some(Value::Json(bytes)) = named.get("body") {
                    if let Ok(names) = reddb_types::document_body_codec::field_names(bytes) {
                        for name in names {
                            bloom |= field_name_bloom(&name);
                        }
                    }
                }
                bloom
            } else {
                0
            }
        }
        EntityData::Node(node) => node
            .properties
            .keys()
            .fold(0u64, |acc, k| acc | field_name_bloom(k)),
        EntityData::Edge(edge) => edge
            .properties
            .keys()
            .fold(0u64, |acc, k| acc | field_name_bloom(k)),
        // Vectors, time-series, queue: no user-named fields worth blooming.
        _ => 0,
    }
}

impl UnifiedEntity {
    /// Create a new unified entity
    pub fn new(id: EntityId, kind: EntityKind, data: EntityData) -> Self {
        let now = current_unix_secs();
        let field_bloom = compute_entity_field_bloom(&data);

        Self {
            id,
            logical_id: None,
            kind,
            created_at: now,
            updated_at: now,
            data,
            sequence_id: 0,
            field_bloom,
            // Pre-MVCC default: xmin/xmax = 0 means visible to every snapshot.
            // Transactional writers stamp real snapshot IDs after allocation.
            xmin: 0,
            xmax: 0,
            aux: None,
        }
    }

    /// MVCC visibility check (Phase 2.3 PG parity).
    ///
    /// Returns `true` when this tuple is visible under the provided
    /// snapshot xid. Pre-MVCC rows (`xmin == 0`, `xmax == 0`) are visible
    /// to every snapshot — preserves full compatibility with existing
    /// data inserted before the MVCC headers existed.
    ///
    /// Snapshot isolation rule:
    ///   - `xmin == 0 || xmin <= snapshot_xid`  (creator committed before snapshot)
    ///   - `xmax == 0 || xmax > snapshot_xid`   (deleter committed after snapshot)
    #[inline]
    pub fn is_visible(&self, snapshot_xid: u64) -> bool {
        if self.xmin != 0 && self.xmin > snapshot_xid {
            return false;
        }
        if self.xmax != 0 && self.xmax <= snapshot_xid {
            return false;
        }
        true
    }

    /// Stamp `xmin` (creation transaction ID). Called by the runtime on
    /// INSERT inside an active transaction.
    #[inline]
    pub fn set_xmin(&mut self, xid: u64) {
        self.xmin = xid;
    }

    /// Stamp `xmax` (deletion transaction ID). Called by the runtime on
    /// DELETE/UPDATE inside an active transaction — the tuple survives
    /// until VACUUM reclaims it.
    #[inline]
    pub fn set_xmax(&mut self, xid: u64) {
        self.xmax = xid;
    }

    pub fn branch_version(&self) -> Option<&BranchVersion> {
        self.aux.as_deref()?.branch.as_ref()
    }

    pub fn set_branch_origin(&mut self, epoch: u64) {
        self.aux.get_or_insert_with(Default::default).branch = Some(BranchVersion {
            origin: epoch,
            deletions: Vec::new(),
        });
    }

    pub fn clear_branch_version(&mut self) {
        if let Some(aux) = self.aux.as_deref_mut() {
            aux.branch = None;
        }
    }

    pub fn branch_xmax(&self, epoch: u64) -> u64 {
        match self.branch_version() {
            Some(branch) if branch.origin != epoch => branch
                .deletions
                .binary_search_by_key(&epoch, |(owner, _)| *owner)
                .map_or(0, |index| branch.deletions[index].1),
            _ => self.xmax,
        }
    }

    /// The caller holds the publication fence and starts from the current
    /// stored entity, so another branch's deletion cannot be lost on replace.
    pub fn set_branch_xmax(&mut self, epoch: u64, xid: u64) {
        let Some(branch) = self.aux.as_deref_mut().and_then(|aux| aux.branch.as_mut()) else {
            self.xmax = xid;
            return;
        };
        if branch.origin == epoch {
            self.xmax = xid;
            return;
        }
        match branch
            .deletions
            .binary_search_by_key(&epoch, |(owner, _)| *owner)
        {
            Ok(index) if xid == 0 => {
                branch.deletions.remove(index);
            }
            Ok(index) => branch.deletions[index].1 = xid,
            Err(index) if xid != 0 => branch.deletions.insert(index, (epoch, xid)),
            Err(_) => {}
        }
    }

    pub fn retain_branch_deletions(&mut self, mut keep: impl FnMut(u64) -> bool) {
        if let Some(branch) = self.aux.as_deref_mut().and_then(|aux| aux.branch.as_mut()) {
            branch.deletions.retain(|(_, xid)| keep(*xid));
        }
    }

    pub fn max_transaction_id(&self) -> u64 {
        self.branch_version()
            .map_or(self.xmin.max(self.xmax), |branch| {
                branch
                    .deletions
                    .iter()
                    .fold(self.xmin.max(self.xmax), |high, (_, xid)| high.max(*xid))
            })
    }

    /// Stable user-visible identity. Legacy rows without an explicit
    /// logical id map to their physical entity id.
    #[inline]
    pub fn logical_id(&self) -> EntityId {
        self.logical_id.unwrap_or(self.id)
    }

    /// Returns true when the entity carries an explicit logical id on disk.
    #[inline]
    pub fn has_explicit_logical_id(&self) -> bool {
        self.logical_id.is_some()
    }

    /// Set the stable user-visible identity for this physical version.
    #[inline]
    pub fn set_logical_id(&mut self, logical_id: EntityId) {
        self.logical_id = Some(logical_id);
    }

    /// Ensure entities written by the current engine carry explicit
    /// logical identity. Table rows + documents (Phase 1/2) and graph
    /// nodes/edges (Phase 3) participate in the multi-model MVCC
    /// versioning rollout and so need a stable logical id for
    /// version-chain selection. Stamping the logical id is inert for
    /// non-versioned collections: history only accrues through
    /// `install_versioned_table_row_update`, which is gated on the
    /// collection `versioned` flag, so a stamped-but-never-superseded
    /// entity keeps `logical_id == id` and behaves exactly as before.
    /// Vectors are intentionally excluded pending their read-path follow
    /// up (no snapshot-honoring `VECTOR SEARCH`).
    #[inline]
    pub fn ensure_table_logical_id(&mut self) {
        if self.logical_id.is_none()
            && matches!(
                self.kind,
                EntityKind::TableRow { .. } | EntityKind::GraphNode(_) | EntityKind::GraphEdge(_)
            )
        {
            self.logical_id = Some(self.id);
        }
    }

    /// Create a table row entity
    pub fn table_row(
        id: EntityId,
        table: impl Into<Arc<str>>,
        row_id: u64,
        columns: Vec<Value>,
    ) -> Self {
        Self::new(
            id,
            EntityKind::TableRow {
                table: table.into(),
                row_id,
            },
            EntityData::Row(RowData::new(columns)),
        )
    }

    /// Create a graph node entity
    pub fn graph_node(
        id: EntityId,
        label: impl Into<String>,
        node_type: impl Into<String>,
        properties: HashMap<String, Value>,
    ) -> Self {
        Self::new(
            id,
            EntityKind::GraphNode(Box::new(GraphNodeKind {
                label: label.into(),
                node_type: node_type.into(),
            })),
            EntityData::Node(NodeData::with_properties(properties)),
        )
    }

    /// Create a graph edge entity
    pub fn graph_edge(
        id: EntityId,
        label: impl Into<String>,
        from: impl Into<String>,
        to: impl Into<String>,
        weight: f32,
        properties: HashMap<String, Value>,
    ) -> Self {
        Self::new(
            id,
            EntityKind::GraphEdge(Box::new(GraphEdgeKind {
                label: label.into(),
                from_node: from.into(),
                to_node: to.into(),
                weight: (weight * 1000.0) as u32,
            })),
            EntityData::Edge(EdgeData::with_properties(weight, properties)),
        )
    }

    /// Create a vector entity
    pub fn vector(id: EntityId, collection: impl Into<String>, vector: Vec<f32>) -> Self {
        Self::new(
            id,
            EntityKind::Vector {
                collection: collection.into(),
            },
            EntityData::Vector(VectorData::new(vector)),
        )
    }

    /// Add an embedding to this entity
    pub fn add_embedding(&mut self, slot: EmbeddingSlot) {
        self.embeddings_mut().push(slot);
        self.touch();
    }

    /// Add a cross-reference
    pub fn add_cross_ref(&mut self, cross_ref: CrossRef) {
        self.cross_refs_mut().push(cross_ref);
        self.touch();
    }

    /// Get embedding by slot name
    pub fn get_embedding(&self, name: &str) -> Option<&EmbeddingSlot> {
        self.embeddings().iter().find(|e| e.name == name)
    }

    /// Update timestamp
    fn touch(&mut self) {
        self.updated_at = current_unix_secs();
    }

    /// Check if entity is stale (not updated in given seconds)
    pub fn is_stale(&self, max_age_secs: u64) -> bool {
        let now = current_unix_secs();
        now.saturating_sub(self.updated_at) > max_age_secs
    }
}

/// A cross-reference between entities
#[derive(Debug, Clone, PartialEq)]
pub struct CrossRef {
    /// Source entity ID (the entity that holds this reference)
    pub source: EntityId,
    /// Target entity ID
    pub target: EntityId,
    /// Target collection name
    pub target_collection: String,
    /// Type of reference
    pub ref_type: RefType,
    /// Reference weight/strength (0.0-1.0)
    pub weight: f32,
    /// When this reference was created
    pub created_at: u64,
}

impl CrossRef {
    /// Create a new cross-reference
    pub fn new(
        source: EntityId,
        target: EntityId,
        target_collection: impl Into<String>,
        ref_type: RefType,
    ) -> Self {
        Self {
            source,
            target,
            target_collection: target_collection.into(),
            ref_type,
            weight: 1.0,
            created_at: current_unix_secs(),
        }
    }

    /// Create with weight
    pub fn with_weight(
        source: EntityId,
        target: EntityId,
        target_collection: impl Into<String>,
        ref_type: RefType,
        weight: f32,
    ) -> Self {
        let mut cr = Self::new(source, target, target_collection, ref_type);
        cr.weight = weight;
        cr
    }
}

/// Types of cross-references between entities
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum RefType {
    // Table ↔ Graph
    RowToNode, // Table row represents a graph node
    RowToEdge, // Table row represents a graph edge
    NodeToRow, // Node links back to source row

    // Table ↔ Vector
    RowToVector, // Table row has embeddings
    VectorToRow, // Vector search → source row

    // Graph ↔ Vector
    NodeToVector, // Node has embeddings
    EdgeToVector, // Edge has embeddings
    VectorToNode, // Vector search → source node

    // Semantic links (discovered)
    SimilarTo,   // Discovered by vector similarity
    RelatedTo,   // Domain-specific relationship
    DerivesFrom, // Data lineage
    Mentions,    // Text mentions another entity
    Contains,    // Structural containment
    DependsOn,   // Dependency relationship
}

impl RefType {
    /// Get the inverse reference type (for bidirectional tracking)
    pub fn inverse(&self) -> Option<Self> {
        match self {
            Self::RowToNode => Some(Self::NodeToRow),
            Self::NodeToRow => Some(Self::RowToNode),
            Self::RowToVector => Some(Self::VectorToRow),
            Self::VectorToRow => Some(Self::RowToVector),
            Self::NodeToVector => Some(Self::VectorToNode),
            Self::VectorToNode => Some(Self::NodeToVector),
            Self::SimilarTo => Some(Self::SimilarTo), // Symmetric
            Self::RelatedTo => Some(Self::RelatedTo), // Symmetric
            _ => None,                                // One-directional references
        }
    }

    /// Check if this is a symmetric reference type
    pub fn is_symmetric(&self) -> bool {
        matches!(self, Self::SimilarTo | Self::RelatedTo)
    }

    /// Convert RefType to byte for binary serialization
    pub fn to_byte(&self) -> u8 {
        match self {
            Self::RowToNode => 0,
            Self::RowToEdge => 1,
            Self::NodeToRow => 2,
            Self::RowToVector => 3,
            Self::VectorToRow => 4,
            Self::NodeToVector => 5,
            Self::EdgeToVector => 6,
            Self::VectorToNode => 7,
            Self::SimilarTo => 8,
            Self::RelatedTo => 9,
            Self::DerivesFrom => 10,
            Self::Mentions => 11,
            Self::Contains => 12,
            Self::DependsOn => 13,
        }
    }

    /// Create RefType from byte (binary deserialization)
    pub fn from_byte(byte: u8) -> Self {
        match byte {
            0 => Self::RowToNode,
            1 => Self::RowToEdge,
            2 => Self::NodeToRow,
            3 => Self::RowToVector,
            4 => Self::VectorToRow,
            5 => Self::NodeToVector,
            6 => Self::EdgeToVector,
            7 => Self::VectorToNode,
            8 => Self::SimilarTo,
            9 => Self::RelatedTo,
            10 => Self::DerivesFrom,
            11 => Self::Mentions,
            12 => Self::Contains,
            13 => Self::DependsOn,
            _ => Self::RelatedTo, // Default fallback
        }
    }
}

/// Convert Vec<Value> to RowData
impl From<Vec<Value>> for RowData {
    fn from(columns: Vec<Value>) -> Self {
        RowData::new(columns)
    }
}

/// Convert HashMap to NodeData
impl From<HashMap<String, Value>> for NodeData {
    fn from(properties: HashMap<String, Value>) -> Self {
        NodeData::with_properties(properties)
    }
}

/// Convert dense vector to VectorData
impl From<Vec<f32>> for VectorData {
    fn from(dense: Vec<f32>) -> Self {
        VectorData::new(dense)
    }
}

/// Convert tuple (dense, sparse) to VectorData
impl From<(Vec<f32>, SparseVector)> for VectorData {
    fn from((dense, sparse): (Vec<f32>, SparseVector)) -> Self {
        VectorData::with_sparse(dense, sparse)
    }
}

// Helper trait for uniform entity creation
impl UnifiedEntity {
    /// Create a graph node entity from properties map
    pub fn from_properties(
        id: EntityId,
        label: impl Into<String>,
        node_type: impl Into<String>,
        properties: impl IntoIterator<Item = (impl Into<String>, Value)>,
    ) -> Self {
        let props: HashMap<String, Value> =
            properties.into_iter().map(|(k, v)| (k.into(), v)).collect();
        Self::graph_node(id, label, node_type, props)
    }

    /// Convert entity to row data if applicable
    pub fn into_row(self) -> Option<RowData> {
        match self.data {
            EntityData::Row(r) => Some(r),
            _ => None,
        }
    }

    /// Convert entity to node data if applicable
    pub fn into_node(self) -> Option<NodeData> {
        match self.data {
            EntityData::Node(n) => Some(n),
            _ => None,
        }
    }

    /// Convert entity to edge data if applicable
    pub fn into_edge(self) -> Option<EdgeData> {
        match self.data {
            EntityData::Edge(e) => Some(e),
            _ => None,
        }
    }

    /// Convert entity to vector data if applicable
    pub fn into_vector(self) -> Option<VectorData> {
        match self.data {
            EntityData::Vector(v) => Some(v),
            _ => None,
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_entity_creation() {
        let id = EntityId::new(1);
        let entity = UnifiedEntity::table_row(
            id,
            "users",
            100,
            vec![Value::text("alice".to_string()), Value::Integer(25)],
        );

        assert!(entity.data.is_row());
        assert_eq!(entity.kind.storage_type(), "table");
        assert_eq!(entity.kind.collection(), "users");
    }

    #[test]
    fn test_cross_refs() {
        let id1 = EntityId::new(1);
        let id2 = EntityId::new(2);

        let cross_ref = CrossRef::new(id1, id2, "nodes", RefType::RowToNode);
        assert_eq!(cross_ref.source, id1);
        assert_eq!(cross_ref.target, id2);
        assert_eq!(cross_ref.ref_type.inverse(), Some(RefType::NodeToRow));
    }

    #[test]
    fn test_sparse_vector() {
        let sparse = SparseVector::new(vec![0, 5, 10], vec![1.0, 2.0, 3.0], 100);

        assert_eq!(sparse.nnz(), 3);
        assert_eq!(sparse.get(5), 2.0);
        assert_eq!(sparse.get(3), 0.0);
        assert!(sparse.sparsity() > 0.9);
    }

    #[test]
    fn test_embedding_slots() {
        let mut entity = UnifiedEntity::table_row(
            EntityId::new(1),
            "documents",
            1,
            vec![Value::text("Hello world".to_string())],
        );

        entity.add_embedding(EmbeddingSlot::new(
            "content",
            vec![0.1, 0.2, 0.3],
            "text-embedding-3-small",
        ));

        assert_eq!(entity.embeddings().len(), 1);
        assert!(entity.get_embedding("content").is_some());
        assert!(entity.get_embedding("summary").is_none());
    }
}