linkrs-storage 0.1.0

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

use crate::edge::{BatchInsertEntry, EdgeRecord, HotNbr};
use crate::engine::data_store::EdgeTableKey;
use crate::engine::{
    DeleteEdgesBatchParams, EdgeOperationParams, InsertEdgeParams, InsertEdgesBatchParams,
};
use crate::mvcc_visibility::PendingGate;
use crate::vertex::ShardedVertexTable;
use linkrs_core::types::{LabelId, Timestamp, VertexId};
use linkrs_core::{StorageError, StorageResult};

use super::helpers;
use super::GraphStorageContext;

type LabelPair = (LabelId, LabelId);
type EdgeEndpoints = Vec<(u32, u32, i64)>;
type PartitionedEdges = Vec<(LabelPair, EdgeEndpoints)>;

/// Project a stored internal endpoint id to its external `VertexId` using the
/// edge table's owning label. Unresolved ids stay unchanged (dangling edge).
fn endpoint_to_external(
    ctx: &GraphStorageContext,
    label: LabelId,
    vid: VertexId,
    ts: Timestamp,
) -> VertexId {
    if label == 0 {
        return vid;
    }
    match vid.as_internal_u32() {
        Some(internal) => ctx
            .get_external_id_by_internal_id(label, internal)
            .or_else(|| ctx.get_external_vertex_id(label, internal, ts))
            .unwrap_or(vid),
        None => vid,
    }
}

struct EdgeLabelLookupCtx<'a> {
    graph: &'a GraphStorageContext,
    vertex_tables: &'a HashMap<LabelId, Arc<ShardedVertexTable>>,
    src_id: &'a VertexId,
    src_label: LabelId,
    dst_id: &'a VertexId,
    dst_label: LabelId,
    edge_label: LabelId,
    ts: Timestamp,
}

impl GraphStorageContext {
    pub fn insert_edge(&self, params: InsertEdgeParams) -> StorageResult<()> {
        if !self.persistent.is_open.load(Ordering::Acquire) {
            return Err(StorageError::storage_not_open());
        }

        let (src_internal, dst_internal, actual_src_label, actual_dst_label) =
            self.persistent.data_store.with_vertex_tables(
                |vertex_tables| -> StorageResult<(u32, u32, LabelId, LabelId)> {
                    let src_internal = helpers::resolve_internal_id(
                        self,
                        vertex_tables,
                        params.src_label,
                        params.src_id,
                        params.ts,
                    )
                    .ok_or(StorageError::vertex_not_found())?;
                    let dst_internal = helpers::resolve_internal_id(
                        self,
                        vertex_tables,
                        params.dst_label,
                        params.dst_id,
                        params.ts,
                    )
                    .ok_or(StorageError::vertex_not_found())?;
                    let actual_src_label = if params.src_label == 0 {
                        helpers::resolve_internal_id_label(
                            self,
                            vertex_tables,
                            &params.src_id,
                            params.ts,
                        )
                        .ok_or(StorageError::vertex_not_found())?
                    } else {
                        params.src_label
                    };
                    let actual_dst_label = if params.dst_label == 0 {
                        helpers::resolve_internal_id_label(
                            self,
                            vertex_tables,
                            &params.dst_id,
                            params.ts,
                        )
                        .ok_or(StorageError::vertex_not_found())?
                    } else {
                        params.dst_label
                    };
                    Ok((
                        src_internal,
                        dst_internal,
                        actual_src_label,
                        actual_dst_label,
                    ))
                },
            )?;

        let key = EdgeTableKey::new(actual_src_label, actual_dst_label, params.edge_label);
        let template_key = EdgeTableKey::new(0, 0, params.edge_label);
        let stats_manager = self.persistent.stats_manager.clone();
        let maintenance_requested = self.persistent.data_store.with_edge_partition_mut(
            key,
            template_key,
            |template| {
                let mut s = template.schema().clone();
                s.src_label = actual_src_label;
                s.dst_label = actual_dst_label;
                let mut table = crate::edge::EdgeStore::new(s)?;
                if let Some(stats) = stats_manager {
                    table.set_stats_manager(stats);
                }
                Ok(table)
            },
            |edge_table| {
                edge_table.insert_edge(
                    src_internal,
                    dst_internal,
                    params.rank,
                    params.properties,
                    params.ts,
                )?;
                Ok(edge_table.needs_background_maintenance())
            },
        )?;
        if maintenance_requested {
            self.schedule_background_maintenance();
        }
        self.mark_edge_modified(params.edge_label);
        Ok(())
    }

    /// Insert many edges of one edge type with one staging commit per owner
    /// partition.
    ///
    /// Every endpoint resolves once under a single vertex-table read, then
    /// edges group by owner partition so each edge lands exactly where
    /// repeated single inserts put it. One partition lock and one staging
    /// commit serve each group instead of one per edge.
    pub fn insert_edges_batch(&self, params: InsertEdgesBatchParams) -> StorageResult<()> {
        if !self.persistent.is_open.load(Ordering::Acquire) {
            return Err(StorageError::storage_not_open());
        }
        if params.edges.is_empty() {
            return Ok(());
        }
        let resolved: Vec<(u32, u32, LabelId, LabelId)> =
            self.persistent.data_store.with_vertex_tables(
                |vertex_tables| -> StorageResult<Vec<(u32, u32, LabelId, LabelId)>> {
                    let mut resolved = Vec::with_capacity(params.edges.len());
                    for edge in params.edges {
                        let src_internal = helpers::resolve_internal_id(
                            self,
                            vertex_tables,
                            params.src_label,
                            edge.src_id,
                            params.ts,
                        )
                        .ok_or(StorageError::vertex_not_found())?;
                        let dst_internal = helpers::resolve_internal_id(
                            self,
                            vertex_tables,
                            params.dst_label,
                            edge.dst_id,
                            params.ts,
                        )
                        .ok_or(StorageError::vertex_not_found())?;
                        let actual_src_label = if params.src_label == 0 {
                            helpers::resolve_internal_id_label(
                                self,
                                vertex_tables,
                                &edge.src_id,
                                params.ts,
                            )
                            .ok_or(StorageError::vertex_not_found())?
                        } else {
                            params.src_label
                        };
                        let actual_dst_label = if params.dst_label == 0 {
                            helpers::resolve_internal_id_label(
                                self,
                                vertex_tables,
                                &edge.dst_id,
                                params.ts,
                            )
                            .ok_or(StorageError::vertex_not_found())?
                        } else {
                            params.dst_label
                        };
                        resolved.push((
                            src_internal,
                            dst_internal,
                            actual_src_label,
                            actual_dst_label,
                        ));
                    }
                    Ok(resolved)
                },
            )?;

        let mut by_partition: HashMap<(LabelId, LabelId), Vec<usize>> = HashMap::new();
        for (index, (_, _, actual_src, actual_dst)) in resolved.iter().enumerate() {
            by_partition
                .entry((*actual_src, *actual_dst))
                .or_default()
                .push(index);
        }
        let mut partitions: Vec<((LabelId, LabelId), Vec<usize>)> =
            by_partition.into_iter().collect();
        partitions.sort_unstable_by_key(|(key, _)| *key);

        let stats_manager = self.persistent.stats_manager.clone();
        let mut maintenance_requested = false;
        for ((actual_src, actual_dst), indices) in partitions {
            let key = EdgeTableKey::new(actual_src, actual_dst, params.edge_label);
            let template_key = EdgeTableKey::new(0, 0, params.edge_label);
            let entries: Vec<BatchInsertEntry> = indices
                .iter()
                .map(|&index| {
                    let (src_internal, dst_internal, _, _) = resolved[index];
                    let edge = &params.edges[index];
                    (
                        src_internal,
                        dst_internal,
                        edge.rank,
                        edge.properties,
                        params.ts,
                    )
                })
                .collect();
            let stats_manager = stats_manager.clone();
            let requested = self.persistent.data_store.with_edge_partition_mut(
                key,
                template_key,
                |template| {
                    let mut s = template.schema().clone();
                    s.src_label = actual_src;
                    s.dst_label = actual_dst;
                    let mut table = crate::edge::EdgeStore::new(s)?;
                    if let Some(stats) = stats_manager {
                        table.set_stats_manager(stats);
                    }
                    Ok(table)
                },
                |edge_table| {
                    edge_table.insert_edges_batch(&entries)?;
                    Ok(edge_table.needs_background_maintenance())
                },
            )?;
            maintenance_requested |= requested;
        }
        if maintenance_requested {
            self.schedule_background_maintenance();
        }
        self.mark_edge_modified(params.edge_label);
        Ok(())
    }

    /// Delete many edges of one edge type with one staging commit per owner
    /// partition.
    ///
    /// Every endpoint resolves once under a single vertex-table read, then
    /// keys group by owner partition so each delete lands exactly where
    /// repeated single deletes put it. One partition lock and one staging
    /// commit serve each group instead of one per edge. Keys whose endpoints
    /// do not resolve delete nothing and report no error, mirroring the
    /// single-delete no-op; keys resolving to tombstoned edges fail through
    /// the table batch like the single path. A missing partition table
    /// errors like the single path. Returns the number of net applied
    /// deletes.
    pub fn delete_edges_batch(&self, params: DeleteEdgesBatchParams) -> StorageResult<usize> {
        if !self.persistent.is_open.load(Ordering::Acquire) {
            return Err(StorageError::storage_not_open());
        }
        if params.edges.is_empty() {
            return Ok(0);
        }
        let resolved: Vec<Option<(u32, u32, LabelId, LabelId)>> = self
            .persistent
            .data_store
            .with_vertex_tables(|vertex_tables| {
                let mut resolved = Vec::with_capacity(params.edges.len());
                for edge in params.edges {
                    let src_internal = helpers::resolve_internal_id(
                        self,
                        vertex_tables,
                        params.src_label,
                        edge.src_id,
                        params.ts,
                    )
                    .or_else(|| {
                        helpers::resolve_internal_id_any(
                            self,
                            vertex_tables,
                            params.src_label,
                            edge.src_id,
                        )
                    });
                    let dst_internal = helpers::resolve_internal_id(
                        self,
                        vertex_tables,
                        params.dst_label,
                        edge.dst_id,
                        params.ts,
                    )
                    .or_else(|| {
                        helpers::resolve_internal_id_any(
                            self,
                            vertex_tables,
                            params.dst_label,
                            edge.dst_id,
                        )
                    });
                    let (Some(src_internal), Some(dst_internal)) = (src_internal, dst_internal)
                    else {
                        resolved.push(None);
                        continue;
                    };
                    let actual_src_label = if params.src_label == 0 {
                        helpers::resolve_internal_id_label(
                            self,
                            vertex_tables,
                            &edge.src_id,
                            params.ts,
                        )
                        .unwrap_or(params.src_label)
                    } else {
                        params.src_label
                    };
                    let actual_dst_label = if params.dst_label == 0 {
                        helpers::resolve_internal_id_label(
                            self,
                            vertex_tables,
                            &edge.dst_id,
                            params.ts,
                        )
                        .unwrap_or(params.dst_label)
                    } else {
                        params.dst_label
                    };
                    resolved.push(Some((
                        src_internal,
                        dst_internal,
                        actual_src_label,
                        actual_dst_label,
                    )));
                }
                resolved
            });

        let mut by_partition: HashMap<LabelPair, EdgeEndpoints> = HashMap::new();
        for (edge, slot) in params.edges.iter().zip(resolved.iter()) {
            if let Some((src_internal, dst_internal, actual_src, actual_dst)) = slot {
                by_partition
                    .entry((*actual_src, *actual_dst))
                    .or_default()
                    .push((*src_internal, *dst_internal, edge.rank));
            }
        }
        let mut partitions: PartitionedEdges = by_partition.into_iter().collect();
        partitions.sort_unstable_by_key(|(key, _)| *key);

        let mut applied = 0usize;
        let mut maintenance_requested = false;
        for ((actual_src, actual_dst), keys) in partitions {
            let key = EdgeTableKey::new(actual_src, actual_dst, params.edge_label);
            let (count, requested) =
                self.persistent
                    .data_store
                    .with_single_edge_table_mut(&key, |edge_table| {
                        let count = edge_table.delete_edges_batch(&keys, params.ts)?;
                        Ok((count, edge_table.needs_background_maintenance()))
                    })?;
            applied += count;
            maintenance_requested |= requested;
        }
        if maintenance_requested {
            self.schedule_background_maintenance();
        }
        self.mark_edge_modified(params.edge_label);
        Ok(applied)
    }

    fn resolve_edge_table_key(ctx: EdgeLabelLookupCtx) -> EdgeTableKey {
        let actual_src_label = if ctx.src_label == 0 {
            helpers::resolve_internal_id_label(ctx.graph, ctx.vertex_tables, ctx.src_id, ctx.ts)
                .unwrap_or(ctx.src_label)
        } else {
            ctx.src_label
        };
        let actual_dst_label = if ctx.dst_label == 0 {
            helpers::resolve_internal_id_label(ctx.graph, ctx.vertex_tables, ctx.dst_id, ctx.ts)
                .unwrap_or(ctx.dst_label)
        } else {
            ctx.dst_label
        };
        EdgeTableKey::new(actual_src_label, actual_dst_label, ctx.edge_label)
    }

    /// Resolve one endpoint's internal id, optionally falling back to an
    /// any-label scan (delete/erase paths, where data may be partially gone).
    fn resolve_endpoint_id(
        &self,
        vertex_tables: &HashMap<LabelId, Arc<ShardedVertexTable>>,
        label: LabelId,
        id: VertexId,
        ts: Timestamp,
        allow_any_fallback: bool,
    ) -> Option<u32> {
        helpers::resolve_internal_id(self, vertex_tables, label, id, ts).or_else(|| {
            allow_any_fallback
                .then(|| helpers::resolve_internal_id_any(self, vertex_tables, label, id))
                .flatten()
        })
    }

    /// Resolve both endpoints and the edge table key for point operations.
    /// `allow_any_fallback` enables the any-label id fallback used by
    /// delete/erase paths (data may already be partially gone).
    fn resolve_edge_endpoints(
        &self,
        params: &EdgeOperationParams,
        ts: Timestamp,
        allow_any_fallback: bool,
    ) -> Option<(u32, u32, EdgeTableKey)> {
        self.persistent.data_store.with_vertex_tables(
            |vertex_tables| -> Option<(u32, u32, EdgeTableKey)> {
                let src_internal = self.resolve_endpoint_id(
                    vertex_tables,
                    params.src_label,
                    params.src_id,
                    ts,
                    allow_any_fallback,
                )?;
                let dst_internal = self.resolve_endpoint_id(
                    vertex_tables,
                    params.dst_label,
                    params.dst_id,
                    ts,
                    allow_any_fallback,
                )?;
                let key = Self::resolve_edge_table_key(EdgeLabelLookupCtx {
                    graph: self,
                    vertex_tables,
                    src_id: &params.src_id,
                    src_label: params.src_label,
                    dst_id: &params.dst_id,
                    dst_label: params.dst_label,
                    edge_label: params.edge_label,
                    ts,
                });
                Some((src_internal, dst_internal, key))
            },
        )
    }

    /// Resolve one endpoint's internal id and actual owning label (label 0
    /// means resolve the owning label from the id). Shared by scan paths.
    fn resolve_node_endpoint(
        &self,
        label: LabelId,
        id: VertexId,
        ts: Timestamp,
    ) -> Option<(u32, LabelId)> {
        self.persistent
            .data_store
            .with_vertex_tables(|vertex_tables| {
                let internal = self.resolve_endpoint_id(vertex_tables, label, id, ts, false)?;
                let actual = if label == 0 {
                    helpers::resolve_internal_id_label(self, vertex_tables, &id, ts)
                        .unwrap_or(label)
                } else {
                    label
                };
                Some((internal, actual))
            })
    }

    pub fn get_edge(&self, params: &EdgeOperationParams, ts: Timestamp) -> Option<EdgeRecord> {
        if !self.persistent.is_open.load(Ordering::Acquire) {
            return None;
        }

        let (src_internal, dst_internal, key) = self.resolve_edge_endpoints(params, ts, false)?;

        // Pending-aware recheck of the point lookup: the plain timestamp
        // predicate inside `get_edge` cannot see slot states, so a foreign
        // uncommitted creation would leak (dirty read) and a foreign
        // uncommitted deletion would hide a live edge. Recheck through the
        // gate; a creation owned by a foreign pending transaction hides the
        // edge, a foreign pending deletion is ignored by re-reading below it.
        // Scan-class funnels share the same gate through the table
        // `*_with_gate` methods.
        let own_write = self
            .operation_context
            .as_ref()
            .and_then(|context| context.write_timestamp);
        let gate = PendingGate::new(&self.persistent.version_manager, own_write);
        // Single table lock: the fused lookup resolves the record and its
        // edge id in one row scan, and the gate verdict reuses that id with
        // no second scan and no second lock acquisition. One snapshot closes
        // the lock-to-lock race window the old triple acquisition retried
        // against, so no retry loop remains: no match reads empty, and a
        // gate-hidden match is a filtered dirty read (a foreign uncommitted
        // creation leaking through the plain predicate).
        // Scatter-gather point lookup: the catalog lock covers only the
        // handle collection inside `with_single_edge_table`, and the fused
        // row scan plus gate recheck run under the single table lock alone.
        self.persistent
            .data_store
            .with_single_edge_table(&key, |table| {
                Ok(
                    match table.get_edge_with_id(src_internal, dst_internal, params.rank, ts) {
                        Some((record, edge_id))
                            if table.mvcc.is_edge_visible_with_gate(edge_id, ts, &gate) =>
                        {
                            Some(record)
                        }
                        _ => None,
                    },
                )
            })
            .ok()
            .flatten()
    }

    /// Projected point lookup: same pending-aware recheck as `get_edge`
    /// but decodes only `projection` (`None` = all columns, `Some(&[])` =
    /// topology only). Empty query projection means all columns; topology-only
    /// is requested explicitly with an empty slice via batch/cursor paths.
    pub fn get_edge_projected(
        &self,
        params: &EdgeOperationParams,
        ts: Timestamp,
        projection: Option<&[Arc<str>]>,
    ) -> Option<EdgeRecord> {
        if !self
            .persistent
            .is_open
            .load(std::sync::atomic::Ordering::Acquire)
        {
            return None;
        }
        let (src_internal, dst_internal, key) = self.resolve_edge_endpoints(params, ts, false)?;
        let own_write = self
            .operation_context
            .as_ref()
            .and_then(|context| context.write_timestamp);
        let gate = PendingGate::new(&self.persistent.version_manager, own_write);
        // Single table lock mirroring `get_edge`: the fused lookup resolves
        // the projected record and its edge id in one gate-aware row scan,
        // and the recheck below reuses that id with no second scan and no
        // second lock acquisition. One snapshot closes the lock-to-lock race
        // window, so the old retry loop collapses to a single pass.
        self.persistent
            .data_store
            .with_single_edge_table(&key, |table| {
                Ok(
                    match table.get_edge_projected_with_id(
                        src_internal,
                        dst_internal,
                        params.rank,
                        ts,
                        &gate,
                        projection,
                    ) {
                        Some((record, edge_id))
                            if table.mvcc.is_edge_visible_with_gate(edge_id, ts, &gate) =>
                        {
                            Some(record)
                        }
                        _ => None,
                    },
                )
            })
            .ok()
            .flatten()
    }

    pub fn delete_edge(&self, params: &EdgeOperationParams, ts: Timestamp) -> StorageResult<bool> {
        if !self.persistent.is_open.load(Ordering::Acquire) {
            return Err(StorageError::storage_not_open());
        }

        let Some((src_internal, dst_internal, key)) = self.resolve_edge_endpoints(params, ts, true)
        else {
            return Ok(false);
        };

        let deleted = self
            .persistent
            .data_store
            .with_single_edge_table_mut(&key, |edge_table| {
                edge_table.delete_edge(src_internal, dst_internal, params.rank, ts)
            })?;
        if deleted {
            self.mark_edge_modified(params.edge_label);
        }

        Ok(deleted)
    }

    /// Physically erase an edge created by an uncommitted insert.
    ///
    /// Insert-undo entry point (see `UndoTarget::delete_edge`): aborts must
    /// leave no trace in any of the three copies, unlike user deletes which
    /// go through logical deletion. Missing endpoints, partitions or edges
    /// report `Ok(false)` so undo replay stays idempotent.
    pub fn erase_inserted_edge(
        &self,
        params: &EdgeOperationParams,
        ts: Timestamp,
    ) -> StorageResult<bool> {
        if !self.persistent.is_open.load(Ordering::Acquire) {
            return Err(StorageError::storage_not_open());
        }
        let resolved = self.resolve_edge_endpoints(params, ts, true);
        let Some((src_internal, dst_internal, key)) = resolved else {
            return Ok(false);
        };
        if self
            .persistent
            .data_store
            .try_get_edge_table_mut(&key)
            .is_none()
        {
            return Ok(false);
        }
        let erased = self
            .persistent
            .data_store
            .with_single_edge_table_mut(&key, |edge_table| {
                Ok(edge_table.erase_edge(src_internal, dst_internal, params.rank, ts))
            })?;
        if erased {
            self.mark_edge_modified(params.edge_label);
        }
        Ok(erased)
    }

    /// Visit raw out-edge neighbors of `src` without building a vector.
    ///
    /// Zero-copy fan-out entry point: resolves the internal id once, then
    /// streams every visible neighbor of every matching table into the
    /// visitor as `(src_internal, far_label, nbr)` pairs — `far_label` is the
    /// table's owning dst label so callers can project internal endpoints to
    /// external ids even when the edge type's endpoint tags are unconstrained.
    /// Returns the resolved internal src id.
    ///
    /// Lock contract: the visitor runs under one table read lock at a time
    /// and must never acquire a vertex, catalog or second table lock
    /// (no vertex-id resolution inside the callback). Stage neighbor ids in
    /// the callback and resolve them after the visit returns; reversing the
    /// order inverts the vertex-before-edge discipline and can deadlock
    /// against writers. Handles are collected under a brief catalog lock and
    /// the catalog is released before any table is touched, so DDL is never
    /// blocked by a long fan-out.
    /// Resolve many endpoint ids under a single vertex-catalog lock.
    ///
    /// Batch counterpart of [`Self::resolve_node_endpoint`]: one catalog
    /// acquisition serves the whole seed list instead of one per seed.
    /// Positions of unresolvable ids stay `None` so callers keep input order.
    fn resolve_node_endpoints(
        &self,
        label: LabelId,
        ids: &[VertexId],
        ts: Timestamp,
    ) -> Vec<Option<(u32, LabelId)>> {
        self.persistent
            .data_store
            .with_vertex_tables(|vertex_tables| {
                ids.iter()
                    .map(|id| {
                        let internal =
                            self.resolve_endpoint_id(vertex_tables, label, *id, ts, false)?;
                        let actual = if label == 0 {
                            helpers::resolve_internal_id_label(self, vertex_tables, id, ts)
                                .unwrap_or(label)
                        } else {
                            label
                        };
                        Some((internal, actual))
                    })
                    .collect()
            })
    }

    /// Visit raw out-edge neighbors for a whole seed batch.
    ///
    /// Each edge table is locked once for all seeds instead of once per
    /// seed. The visitor reports the seed index with every neighbor so
    /// callers can accumulate per-seed state.
    ///
    /// Lock contract: the visitor runs under one table read lock at a time
    /// and must never acquire a vertex, catalog or second table lock
    /// (no vertex-id resolution inside the callback). Stage neighbor ids in
    /// the callback and resolve them after the visit returns; reversing the
    /// order inverts the vertex-before-edge discipline and can deadlock
    /// against writers. Handles are collected under a brief catalog lock and
    /// the catalog is released before any table is touched, so DDL is never
    /// blocked by a long fan-out.
    pub fn visit_out_nbrs_batched<F>(
        &self,
        edge_label: LabelId,
        src_label: LabelId,
        _dst_label: LabelId,
        seeds: &[VertexId],
        ts: Timestamp,
        mut f: F,
    ) where
        F: FnMut(usize, u32, LabelId, HotNbr),
    {
        if !self.persistent.is_open.load(Ordering::Acquire) {
            return;
        }
        let resolved = self.resolve_node_endpoints(src_label, seeds, ts);
        let arcs = self
            .persistent
            .data_store
            .matching_edge_partition_arcs(edge_label);
        let gate = self.pending_gate();
        for arc in &arcs {
            let table = self.persistent.data_store.read_edge_table(arc);
            if table.label() != edge_label {
                continue;
            }
            let far_label = table.dst_label();
            for (index, seed) in resolved.iter().enumerate() {
                let Some((src_internal, actual_src)) = seed else {
                    continue;
                };
                if table.src_label() != *actual_src {
                    continue;
                }
                table.visit_out_with_gate(*src_internal, ts, &gate, |nbr| {
                    f(index, *src_internal, far_label, nbr)
                });
            }
        }
    }

    /// Visit raw in-edge neighbors for a whole seed batch.
    ///
    /// In-direction counterpart of [`Self::visit_out_nbrs_batched`] with the
    /// same single-lock-per-table behavior and lock contract.
    pub fn visit_in_nbrs_batched<F>(
        &self,
        edge_label: LabelId,
        _src_label: LabelId,
        dst_label: LabelId,
        seeds: &[VertexId],
        ts: Timestamp,
        mut f: F,
    ) where
        F: FnMut(usize, u32, LabelId, HotNbr),
    {
        if !self.persistent.is_open.load(Ordering::Acquire) {
            return;
        }
        let resolved = self.resolve_node_endpoints(dst_label, seeds, ts);
        let arcs = self
            .persistent
            .data_store
            .matching_edge_partition_arcs(edge_label);
        let gate = self.pending_gate();
        for arc in &arcs {
            let table = self.persistent.data_store.read_edge_table(arc);
            if table.label() != edge_label {
                continue;
            }
            let far_label = table.src_label();
            for (index, seed) in resolved.iter().enumerate() {
                let Some((dst_internal, actual_dst)) = seed else {
                    continue;
                };
                if table.dst_label() != *actual_dst {
                    continue;
                }
                table.visit_in_with_gate(*dst_internal, ts, &gate, |nbr| {
                    f(index, *dst_internal, far_label, nbr)
                });
            }
        }
    }

    pub fn out_edges_projected(
        &self,
        edge_label: LabelId,
        src_label: LabelId,
        src_id: VertexId,
        ts: Timestamp,
        projection: Option<&[Arc<str>]>,
    ) -> Option<Vec<EdgeRecord>> {
        self.out_edges_projected_impl(edge_label, src_label, src_id, ts, projection, None)
    }

    pub fn in_edges_projected(
        &self,
        edge_label: LabelId,
        dst_label: LabelId,
        dst_id: VertexId,
        ts: Timestamp,
        projection: Option<&[Arc<str>]>,
    ) -> Option<Vec<EdgeRecord>> {
        self.in_edges_projected_impl(edge_label, dst_label, dst_id, ts, projection, None)
    }

    pub fn out_edges_projected_limit(
        &self,
        edge_label: LabelId,
        src_label: LabelId,
        src_id: VertexId,
        ts: Timestamp,
        projection: Option<&[Arc<str>]>,
        limit: usize,
    ) -> Option<Vec<EdgeRecord>> {
        self.out_edges_projected_impl(edge_label, src_label, src_id, ts, projection, Some(limit))
    }

    pub fn in_edges_projected_limit(
        &self,
        edge_label: LabelId,
        dst_label: LabelId,
        dst_id: VertexId,
        ts: Timestamp,
        projection: Option<&[Arc<str>]>,
        limit: usize,
    ) -> Option<Vec<EdgeRecord>> {
        self.in_edges_projected_impl(edge_label, dst_label, dst_id, ts, projection, Some(limit))
    }

    /// Out-direction projected scan shared by the unbounded and limit
    /// variants; `limit` of `None` reads every matching edge.
    fn out_edges_projected_impl(
        &self,
        edge_label: LabelId,
        src_label: LabelId,
        src_id: VertexId,
        ts: Timestamp,
        projection: Option<&[Arc<str>]>,
        limit: Option<usize>,
    ) -> Option<Vec<EdgeRecord>> {
        if !self.persistent.is_open.load(Ordering::Acquire) {
            return None;
        }
        let (src_internal, actual_src) = self.resolve_node_endpoint(src_label, src_id, ts)?;
        let arcs = self
            .persistent
            .data_store
            .matching_edge_partition_arcs(edge_label);
        let gate = self.pending_gate();
        let mut staged: Vec<(LabelId, EdgeRecord)> = Vec::new();
        for arc in &arcs {
            let table = self.persistent.data_store.read_edge_table(arc);
            if table.label() != edge_label || table.src_label() != actual_src {
                continue;
            }
            let tbl_dst = table.dst_label();
            match limit {
                Some(limit) => {
                    let remaining = limit.saturating_sub(staged.len());
                    if remaining == 0 {
                        break;
                    }
                    for record in table.out_edges_with_gate_projected_limit(
                        src_internal,
                        ts,
                        &gate,
                        projection,
                        remaining,
                    ) {
                        staged.push((tbl_dst, record));
                    }
                }
                None => {
                    for record in
                        table.out_edges_with_gate_projected(src_internal, ts, &gate, projection)
                    {
                        staged.push((tbl_dst, record));
                    }
                }
            }
        }
        // Vertex resolution runs after the edge-table guards are released,
        // so no edge lock is held while taking vertex-table locks.
        let mut records = Vec::with_capacity(staged.len());
        for (tbl_dst, mut record) in staged.drain(..) {
            record.dst_vid = endpoint_to_external(self, tbl_dst, record.dst_vid, ts);
            records.push(record);
        }
        Some(records)
    }

    /// In-direction counterpart of `out_edges_projected_impl`.
    fn in_edges_projected_impl(
        &self,
        edge_label: LabelId,
        dst_label: LabelId,
        dst_id: VertexId,
        ts: Timestamp,
        projection: Option<&[Arc<str>]>,
        limit: Option<usize>,
    ) -> Option<Vec<EdgeRecord>> {
        if !self.persistent.is_open.load(Ordering::Acquire) {
            return None;
        }
        let (dst_internal, actual_dst) = self.resolve_node_endpoint(dst_label, dst_id, ts)?;
        let arcs = self
            .persistent
            .data_store
            .matching_edge_partition_arcs(edge_label);
        let gate = self.pending_gate();
        let mut staged: Vec<(LabelId, EdgeRecord)> = Vec::new();
        for arc in &arcs {
            let table = self.persistent.data_store.read_edge_table(arc);
            if table.label() != edge_label || table.dst_label() != actual_dst {
                continue;
            }
            let tbl_src = table.src_label();
            match limit {
                Some(limit) => {
                    let remaining = limit.saturating_sub(staged.len());
                    if remaining == 0 {
                        break;
                    }
                    for record in table.in_edges_with_gate_projected_limit(
                        dst_internal,
                        ts,
                        &gate,
                        projection,
                        remaining,
                    ) {
                        staged.push((tbl_src, record));
                    }
                }
                None => {
                    for record in
                        table.in_edges_with_gate_projected(dst_internal, ts, &gate, projection)
                    {
                        staged.push((tbl_src, record));
                    }
                }
            }
        }
        let mut records = Vec::with_capacity(staged.len());
        for (tbl_src, mut record) in staged.drain(..) {
            record.src_vid = endpoint_to_external(self, tbl_src, record.src_vid, ts);
            records.push(record);
        }
        Some(records)
    }
}