hawdb-storage 0.14.0

// Copyright 2026 Nowledge
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

//! Ordered checkpoint/delta overlay used by the embedded graph store.
//!
//! The facade captures the delta and tombstone snapshot. This owner merges it
//! with a lazy canonical reader without collecting the checkpoint into memory.

use crate::cow::CowSegment;
use crate::{
    canonical::{CanonicalNodeIterator, CanonicalRelationshipIterator, CanonicalSegmentError},
    NodeId, NodeRecord, RelId, RelRecord,
};
use hawdb_core::{HawDBError, Result};
use std::collections::BTreeSet;
use std::iter::Peekable;

mod checkpoint;
mod cursor;
pub(crate) use cursor::Cursor as CheckpointOverlayCursor;
#[cfg(test)]
mod tests;
pub(crate) use checkpoint::{CheckpointOverlayIterator, CheckpointRecordRef};

pub struct GraphNodeIterator {
    inner: OverlayIterator<NodeRecord, CanonicalNodeIterator>,
}

impl GraphNodeIterator {
    /// Expose one physical overlay decision per step, including tombstones.
    /// Checkpoint builders can admit each decision instead of charging a whole
    /// tombstone run as one logical record operation.
    #[doc(hidden)]
    pub fn checkpoint_steps(mut self) -> impl Iterator<Item = Result<Option<NodeRecord>>> {
        std::iter::from_fn(move || self.inner.next_step())
    }
}

impl Iterator for GraphNodeIterator {
    type Item = Result<NodeRecord>;

    fn next(&mut self) -> Option<Self::Item> {
        self.inner.next()
    }
}

pub struct GraphRelationshipIterator {
    inner: OverlayIterator<RelRecord, CanonicalRelationshipIterator>,
}

impl GraphRelationshipIterator {
    #[doc(hidden)]
    pub fn checkpoint_steps(mut self) -> impl Iterator<Item = Result<Option<RelRecord>>> {
        std::iter::from_fn(move || self.inner.next_step())
    }
}

impl Iterator for GraphRelationshipIterator {
    type Item = Result<RelRecord>;

    fn next(&mut self) -> Option<Self::Item> {
        self.inner.next()
    }
}

/// Assemble an owned snapshot from ID-ordered inputs captured by the facade.
pub fn node_records(
    base: Option<CanonicalNodeIterator>,
    delta: Vec<NodeRecord>,
    tombstones: CowSegment<BTreeSet<NodeId>>,
) -> GraphNodeIterator {
    GraphNodeIterator {
        inner: OverlayIterator::new(base, delta, tombstones),
    }
}

/// Assemble an owned snapshot from ID-ordered inputs captured by the facade.
pub fn relationship_records(
    base: Option<CanonicalRelationshipIterator>,
    delta: Vec<RelRecord>,
    tombstones: CowSegment<BTreeSet<RelId>>,
) -> GraphRelationshipIterator {
    GraphRelationshipIterator {
        inner: OverlayIterator::new(base, delta, tombstones),
    }
}

pub(crate) trait OverlayRecord {
    type Id: Copy + Ord;

    fn id(&self) -> Self::Id;
}

impl OverlayRecord for NodeRecord {
    type Id = NodeId;

    fn id(&self) -> NodeId {
        self.id
    }
}

impl OverlayRecord for RelRecord {
    type Id = RelId;

    fn id(&self) -> RelId {
        self.id
    }
}

struct OverlayIterator<R: OverlayRecord, B: Iterator> {
    base: Option<Peekable<B>>,
    delta: Peekable<std::vec::IntoIter<R>>,
    tombstones: CowSegment<BTreeSet<R::Id>>,
}

impl<R: OverlayRecord, B: Iterator> OverlayIterator<R, B> {
    fn new(base: Option<B>, delta: Vec<R>, tombstones: CowSegment<BTreeSet<R::Id>>) -> Self {
        Self {
            base: base.map(Iterator::peekable),
            delta: delta.into_iter().peekable(),
            tombstones,
        }
    }
}

impl<R: OverlayRecord, B: Iterator<Item = std::result::Result<R, CanonicalSegmentError>>>
    OverlayIterator<R, B>
{
    fn next_base(&mut self) -> Result<R> {
        self.base
            .as_mut()
            .and_then(Iterator::next)
            .expect("peeked base record exists")
            .map_err(|error| HawDBError::StorageIntegrity(error.to_string()))
    }
}

impl<R: OverlayRecord, B: Iterator<Item = std::result::Result<R, CanonicalSegmentError>>>
    OverlayIterator<R, B>
{
    fn next_step(&mut self) -> Option<Result<Option<R>>> {
        // A tombstone or a newer delta must never hide a physical read error.
        let base_id = match self.base.as_mut().and_then(|base| base.peek()) {
            Some(Ok(record)) => Some(record.id()),
            Some(Err(_)) => return Some(self.next_base().map(Some)),
            None => None,
        };
        let delta_id = self.delta.peek().map(OverlayRecord::id);
        let record = match (base_id, delta_id) {
            (None, None) => return None,
            (Some(base_id), Some(delta_id)) if base_id == delta_id => {
                if let Err(error) = self.next_base() {
                    return Some(Err(error));
                }
                Ok(self.delta.next().expect("matching delta record exists"))
            }
            (None, Some(_)) => Ok(self.delta.next().expect("peeked delta record exists")),
            (Some(base_id), Some(delta_id)) if base_id > delta_id => {
                Ok(self.delta.next().expect("peeked delta record exists"))
            }
            (Some(_), _) => self.next_base(),
        };
        Some(record.map(|record| {
            if self.tombstones.contains(&record.id()) {
                None
            } else {
                Some(record)
            }
        }))
    }
}

impl<R: OverlayRecord, B: Iterator<Item = std::result::Result<R, CanonicalSegmentError>>> Iterator
    for OverlayIterator<R, B>
{
    type Item = Result<R>;

    fn next(&mut self) -> Option<Self::Item> {
        loop {
            match self.next_step()? {
                Ok(None) => continue,
                Ok(Some(record)) => return Some(Ok(record)),
                Err(error) => return Some(Err(error)),
            }
        }
    }
}