reddb-io-server 1.27.4

//! Unified Store
//!
//! High-level API for the unified storage layer that combines tables, graphs,
//! and vectors into a single coherent interface.
//!
//! # Features
//!
//! - Multi-collection management
//! - Cross-collection queries
//! - Unified entity access
//! - Automatic ID generation
//! - Cross-reference management
//! - **Binary persistence** with pages, indices, and efficient encoding
//! - **Page-based storage** via Pager for ACID durability
//!
//! # Persistence Modes
//!
//! 1. **File Mode** (`save_to_file`/`load_from_file`): Simple binary dump
//! 2. **Paged Mode** (`open`/`persist`): Full page-based storage with B-tree indices

use std::collections::{BTreeMap, HashMap};
use std::fs::File;
use std::io::{BufReader, Read};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU32, AtomicU64, Ordering};
use std::sync::Arc;

use parking_lot::RwLock;

use super::context_index::ContextIndex;
use super::entity::{
    CrossRef, EmbeddingSlot, EntityData, EntityId, EntityKind, RefType, RowData,
    TimeSeriesPointKind, UnifiedEntity,
};
use super::manager::{ManagerConfig, ManagerStats, SegmentManager};
use super::metadata::{Metadata, MetadataFilter, MetadataValue};
use crate::api::{DurabilityMode, GroupCommitOptions};
use crate::physical::{ManifestEvent, ManifestEventKind};
use crate::storage::engine::pager::PagerError;
use crate::storage::engine::{BTree, BTreeError, Pager, PagerConfig, PhysicalFileHeader};
use crate::storage::primitives::encoding::read_varu64;
use reddb_types::types::Value;

pub use reddb_file::{
    is_supported_store_version, NativeCatalogCollectionSummary, NativeCatalogSummary,
    NativeExportSummary, NativeManifestEntrySummary, NativeManifestSummary,
    NativeMetadataStateSummary, NativeRecoverySummary, NativeRegistryIndexSummary,
    NativeRegistryJobSummary, NativeRegistryProjectionSummary, NativeRegistrySummary,
    NativeSnapshotSummary, NativeVectorArtifactPageSummary, NativeVectorArtifactSummary,
    ENTITY_RECORD_MAGIC, METADATA_MAGIC, METADATA_OVERFLOW_MAGIC, NATIVE_BLOB_MAGIC,
    NATIVE_CATALOG_MAGIC, NATIVE_COLLECTION_ROOTS_MAGIC, NATIVE_MANIFEST_MAGIC,
    NATIVE_MANIFEST_SAMPLE_LIMIT, NATIVE_METADATA_STATE_MAGIC, NATIVE_RECOVERY_MAGIC,
    NATIVE_REGISTRY_MAGIC, NATIVE_VECTOR_ARTIFACT_MAGIC, STORE_MAGIC, STORE_VERSION_CURRENT,
    STORE_VERSION_V1, STORE_VERSION_V11, STORE_VERSION_V12, STORE_VERSION_V2, STORE_VERSION_V3,
    STORE_VERSION_V4, STORE_VERSION_V5, STORE_VERSION_V6, STORE_VERSION_V7, STORE_VERSION_V8,
    STORE_VERSION_V9,
};

#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct MvccVacuumStats {
    pub scanned_versions: u64,
    pub retained_versions: u64,
    pub reclaimed_versions: u64,
    pub retained_history_versions: u64,
    pub reclaimed_history_versions: u64,
    pub retained_tombstones: u64,
    pub reclaimed_tombstones: u64,
}

impl MvccVacuumStats {
    pub fn add(&mut self, other: &Self) {
        self.scanned_versions += other.scanned_versions;
        self.retained_versions += other.retained_versions;
        self.reclaimed_versions += other.reclaimed_versions;
        self.retained_history_versions += other.retained_history_versions;
        self.reclaimed_history_versions += other.reclaimed_history_versions;
        self.retained_tombstones += other.retained_tombstones;
        self.reclaimed_tombstones += other.reclaimed_tombstones;
    }
}

#[derive(Debug, Clone)]
pub struct NativePhysicalState {
    pub header: PhysicalFileHeader,
    pub collection_roots: BTreeMap<String, u64>,
    pub manifest: Option<NativeManifestSummary>,
    pub registry: Option<NativeRegistrySummary>,
    pub recovery: Option<NativeRecoverySummary>,
    pub catalog: Option<NativeCatalogSummary>,
    pub metadata_state: Option<NativeMetadataStateSummary>,
    pub vector_artifact_pages: Option<Vec<NativeVectorArtifactPageSummary>>,
}

// ============================================================================
// Configuration
// ============================================================================

/// Configuration for UnifiedStore
#[derive(Debug, Clone)]
pub struct UnifiedStoreConfig {
    /// Configuration for segment managers
    pub manager_config: ManagerConfig,
    /// Automatically index cross-references on insert
    pub auto_index_refs: bool,
    /// Automatically build a HASH index on a user `id` column the first
    /// time a row carrying that column is inserted into a collection.
    /// Mirrors PostgreSQL's implicit primary-key index and Mongo's `_id`
    /// default index — without it, `WHERE id = N` falls through to a
    /// full segment scan because RedDB has no concept of an automatic
    /// primary-key index on user-declared columns. See `docs/perf/
    /// delete-sequential-2026-05-06.md` for the perf rationale.
    /// Defaults to `true`; set to `false` to opt out per workload.
    pub auto_index_id: bool,
    /// Maximum cross-references per entity
    pub max_cross_refs: usize,
    /// Enable write-ahead logging
    pub enable_wal: bool,
    /// Durability profile for paged writes.
    pub durability_mode: DurabilityMode,
    /// Group-commit batching knobs when using grouped durability.
    pub group_commit: GroupCommitOptions,
    /// Data directory path
    pub data_dir: Option<std::path::PathBuf>,
    /// Embedded single-file artifact used for the internal WAL stream.
    pub embedded_wal_path: Option<std::path::PathBuf>,
    /// Preallocated page-cache slot count (ADR 0073 §2). `None` leaves the
    /// pager on its own structural default; the promoted boot path always
    /// supplies a count derived from the page-cache budget share, so the
    /// default is reached only by direct library callers.
    pub page_cache_slots: Option<usize>,
}

impl Default for UnifiedStoreConfig {
    fn default() -> Self {
        Self {
            manager_config: ManagerConfig::default(),
            auto_index_refs: true,
            auto_index_id: true,
            max_cross_refs: 1000,
            enable_wal: false,
            // Mirrors `RedDBOptions::default().durability_mode` — see
            // `src/api.rs` for the rationale.
            durability_mode: DurabilityMode::WalDurableGrouped,
            group_commit: GroupCommitOptions::default(),
            data_dir: None,
            embedded_wal_path: None,
            page_cache_slots: None,
        }
    }
}

impl UnifiedStoreConfig {
    /// Create config with data directory
    pub fn with_data_dir(mut self, path: impl Into<std::path::PathBuf>) -> Self {
        self.data_dir = Some(path.into());
        self
    }

    /// Enable WAL
    pub fn with_wal(mut self) -> Self {
        self.enable_wal = true;
        self
    }

    pub fn with_durability_mode(mut self, mode: DurabilityMode) -> Self {
        self.durability_mode = mode;
        self
    }

    pub fn with_group_commit(mut self, options: GroupCommitOptions) -> Self {
        self.group_commit = options;
        self
    }

    pub fn with_embedded_wal_path(mut self, path: impl Into<std::path::PathBuf>) -> Self {
        self.embedded_wal_path = Some(path.into());
        self
    }

    /// Pre-size the page cache from its budget share (ADR 0073 §2).
    pub fn with_page_cache_slots(mut self, slots: usize) -> Self {
        self.page_cache_slots = Some(slots);
        self
    }

    /// Set max cross-references
    pub fn with_max_refs(mut self, max: usize) -> Self {
        self.max_cross_refs = max;
        self
    }
}

// ============================================================================
// Error Types
// ============================================================================

pub use reddb_engine::storage::error::StoreError;

// ============================================================================
// Statistics
// ============================================================================

/// Statistics for UnifiedStore
#[derive(Debug, Clone, Default)]
pub struct StoreStats {
    /// Number of collections
    pub collection_count: usize,
    /// Total entities across all collections
    pub total_entities: usize,
    /// Total memory usage in bytes
    pub total_memory_bytes: usize,
    /// Per-collection statistics
    pub collections: HashMap<String, ManagerStats>,
    /// Total cross-references
    pub cross_ref_count: usize,
}

impl StoreStats {
    /// Get average entities per collection
    pub fn avg_entities_per_collection(&self) -> f64 {
        if self.collection_count == 0 {
            0.0
        } else {
            self.total_entities as f64 / self.collection_count as f64
        }
    }

    /// Get memory in MB
    pub fn memory_mb(&self) -> f64 {
        self.total_memory_bytes as f64 / (1024.0 * 1024.0)
    }
}

// ============================================================================
// UnifiedStore - The Main API
// ============================================================================

struct CheckpointVersion {
    manager: std::sync::Weak<SegmentManager>,
    segments: Vec<(u64, u64)>,
    visibility_revision: u64,
    untracked_visibility_revision: u64,
    excluded_xids: Arc<std::collections::HashSet<u64>>,
    pending_vault: Arc<std::collections::HashSet<EntityId>>,
    format_version: u32,
    /// Raw rows omitted or normalized in the durable image. None retains the
    /// full validation path when more than 1,024 such rows are outstanding.
    /// Also carries bounded unprojected IDs across completed inline GC batches.
    deferred_rows: Option<Arc<Vec<EntityId>>>,
    /// Inline GC advanced the segment proof past rows still awaiting projection.
    has_unprojected_rows: bool,
}

impl CheckpointVersion {
    fn remember_deferred_row(&mut self, id: EntityId) {
        let Some(rows) = self.deferred_rows.as_mut() else {
            return;
        };
        if rows.len() == 1024 {
            self.deferred_rows = None;
        } else {
            Arc::make_mut(rows).push(id);
        }
    }
}

/// Unified storage for tables, graphs, and vectors
///
/// UnifiedStore provides a single coherent interface for all data types:
/// - **Tables**: Row-based data with columns
/// - **Graphs**: Nodes and edges with labels
/// - **Vectors**: Embeddings for similarity search
///
/// # Features
///
/// - Multi-collection management
/// - Cross-collection queries
/// - Cross-reference tracking between entities
/// - Automatic ID generation
/// - Segment-based storage with growing/sealed lifecycle
///
/// # Example
///
/// ```ignore
/// use reddb::storage::{Entity, Store};
///
/// let store = Store::new();
///
/// // Create a collection
/// store.create_collection("hosts")?;
///
/// // Insert an entity
/// let entity = Entity::table_row(1, "hosts", 1, vec![]);
/// let id = store.insert("hosts", entity)?;
///
/// // Query
/// let found = store.get("hosts", id);
/// ```
pub struct UnifiedStore {
    /// Canonical resident state; native fields below retain durability policy.
    resident: reddb_engine::storage::resident_store::ResidentStore,
    /// Store configuration
    config: UnifiedStoreConfig,
    /// File format version for serialization
    format_version: AtomicU32,
    /// Optional page-based storage via Pager
    pager: Option<Arc<Pager>>,
    /// Database file path (for paged mode)
    db_path: Option<PathBuf>,
    /// B-tree indices for O(log n) entity lookups by ID (per collection).
    /// Stored as `Arc<BTree>` so hot-path callers can clone the handle out
    /// under a read lock and release the map-level lock before doing the
    /// actual insert — previously the outer RwLock was held for the whole
    /// btree mutation, serialising every concurrent insert across every
    /// collection into one global write lock.
    btree_indices: RwLock<HashMap<String, Arc<BTree>>>,
    /// Reuse only a successfully published generation with the same live rows.
    checkpoint_versions: RwLock<HashMap<String, CheckpointVersion>>,
    checkpoint_durable_rows: RwLock<checkpoint_delta::DurableRows>,
    /// The immediately previous published trees and any older reader-pinned
    /// trees. Root identities are persisted with the paged catalog.
    retired_btrees: parking_lot::Mutex<Vec<RetiredBTree>>,
    /// Epoch in the durable catalog and the epoch being published, if any.
    ownership_catalog_epoch: AtomicU64,
    ownership_active_epoch: AtomicU64,
    /// Detached artifact pages stay unavailable until native metadata is durable.
    detached_native_blob_pages: parking_lot::Mutex<std::collections::BTreeSet<u32>>,
    /// Logical store WAL / grouped durability coordinator for paged mode.
    commit: Option<Arc<StoreCommitCoordinator>>,
    /// WAL-replayed `VectorInsert` records, captured at open time and
    /// drained per-collection on first `vector.turbo` access (issue
    /// #694). Boot-time recovery: the in-memory TurboQuant index is
    /// rebuilt by replaying these FP32 vectors in WAL order, so the
    /// rebuilt state is byte-deterministic against the pre-restart
    /// state under a fixed codec seed.
    pub(crate) replayed_turbo_inserts: parking_lot::Mutex<HashMap<String, Vec<(u64, Vec<f32>)>>>,
    /// WAL-replayed probabilistic mutation deltas captured at open time.
    /// Runtime recovery drains these after loading the latest full
    /// probabilistic snapshot from store state.
    pub(crate) replayed_probabilistic_deltas:
        parking_lot::Mutex<Vec<(u8, u8, String, Vec<Vec<u8>>)>>,
    /// Serialize vault publication with snapshot capture and WAL truncation.
    /// Physical metadata and native header publication take it too (#2376).
    /// Reentrant because embedded checkpoint calls the snapshot encoder while
    /// holding the barrier, and storage maintenance may nest those calls.
    pub(crate) vault_publication_lock: parking_lot::ReentrantMutex<()>,
    /// Live-only vault rows excluded from a reentrant WAL-full compaction.
    pub(crate) pending_vault_versions: RwLock<std::collections::HashSet<EntityId>>,
    /// Retain transaction outcomes through embedded snapshot-on-drop. Without
    /// this, aborted physical versions are reopened as committed rows.
    pub(crate) snapshot_manager:
        std::sync::OnceLock<Arc<crate::storage::transaction::snapshot::SnapshotManager>>,
    embedded_commit_session: std::sync::OnceLock<crate::storage::embedded::EmbeddedCommitSession>,
}

struct RetiredBTree {
    tree: Arc<BTree>,
    published: bool,
}

mod builder;
mod checkpoint_delta;
mod checkpoint_wal;
mod commit;
mod impl_entities;
mod impl_file;
mod impl_native_a;
mod impl_native_b;
mod impl_native_c;
mod impl_pages;
mod impl_vault_import;
mod keyed_write;
mod mvcc_reclaim;
pub(crate) use mvcc_reclaim::MvccReclaimCursor;
mod native_helpers;

pub use self::builder::EntityBuilder;
pub(crate) use self::commit::DeferredStoreWalActions;
use self::commit::{StoreCommitCoordinator, StoreWalAction};
use self::native_helpers::*;

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

    #[test]
    fn native_mutations_and_shared_reads_use_one_resident_state() {
        let store = UnifiedStore::new();
        let resident: &reddb_engine::storage::resident_store::ResidentStore = &store.resident;
        store.create_collection("nodes").expect("collection");
        let native = store.get_collection("nodes").expect("native manager");
        let shared = resident.get_collection("nodes").expect("shared manager");
        assert!(
            Arc::ptr_eq(&native, &shared),
            "one canonical collection owner"
        );
        let id = store
            .insert_auto(
                "nodes",
                UnifiedEntity::graph_node(EntityId::new(0), "Person", "person", HashMap::new()),
            )
            .expect("native mutation");
        assert_eq!(id.raw(), 1);
        assert_eq!(
            resident.get_any(id).expect("shared read").1.kind,
            store.get("nodes", id).expect("native read").kind
        );
        assert_eq!(
            resident.lookup_graph_nodes_by_label_in("nodes", "Person"),
            [id]
        );
        assert_eq!(resident.entity_cache_stats().entries, 1);
        assert!(store.delete("nodes", id).expect("native delete"));
        assert!(resident.get_any(id).is_none());
        assert!(resident.lookup_graph_nodes_by_label("Person").is_empty());
        assert_eq!(resident.entity_cache_stats().entries, 0);
        store.set_aux_metadata(b"catalog".to_vec());
        assert_eq!(resident.aux_metadata(), b"catalog");
        assert_eq!(
            store.next_entity_id().raw(),
            2,
            "native allocator is the shared allocator"
        );
    }
}