reddb_server/

runtime.rs

//! Embedded runtime with connection pooling, scans and health.

use std::cmp::Ordering;
use std::collections::{BTreeMap, BTreeSet, BinaryHeap, HashMap, HashSet, VecDeque};
use std::sync::atomic::{AtomicU64, Ordering as AtomicOrdering};
use std::sync::{Arc, Mutex};
use std::time::{SystemTime, UNIX_EPOCH};

use crate::api::{RedDBError, RedDBOptions, RedDBResult};
use crate::catalog::{
    CatalogAnalyticsJobStatus, CatalogAttentionSummary, CatalogGraphProjectionStatus,
    CatalogIndexStatus, CatalogModelSnapshot, CollectionDescriptor,
};
use crate::health::{HealthProvider, HealthReport};
use crate::index::IndexCatalog;
use crate::physical::{
    ExportDescriptor, ManifestEvent, PhysicalAnalyticsJob, PhysicalGraphProjection, PhysicalLayout,
    SnapshotDescriptor,
};
use crate::serde_json::Value as JsonValue;
use crate::storage::engine::pathfinding::{AStar, BellmanFord, Dijkstra, BFS, DFS};
use crate::storage::engine::{
    BetweennessCentrality, ClosenessCentrality, ClusteringCoefficient, ConnectedComponents,
    CycleDetector, DegreeCentrality, EigenvectorCentrality, GraphStore, IvfConfig, IvfIndex,
    IvfStats, LabelPropagation, Louvain, MetadataEntry, MetadataFilter as VectorMetadataFilter,
    MetadataValue as VectorMetadataValue, PageRank, PersonalizedPageRank, PhysicalFileHeader,
    StoredNode, StronglyConnectedComponents, WeaklyConnectedComponents, HITS,
};
use crate::storage::query::is_universal_entity_source as is_universal_query_source;
use crate::storage::query::planner::{
    CanonicalLogicalPlan, CanonicalPlanner, CostEstimator, QueryPlanner,
};
use crate::storage::query::unified::{UnifiedRecord, UnifiedResult};
use crate::storage::unified::dsl::{
    apply_filters, cosine_similarity, Filter as DslFilter, FilterOp as DslFilterOp,
    FilterValue as DslFilterValue, GraphPatternDsl, HybridQueryBuilder, MatchComponents,
    QueryResult as DslQueryResult, ScoredMatch, TextSearchBuilder,
};
use crate::storage::unified::store::{
    NativeCatalogSummary, NativeManifestSummary, NativePhysicalState, NativeRecoverySummary,
    NativeRegistrySummary,
};
use crate::storage::unified::{
    Metadata, MetadataValue as UnifiedMetadataValue, RefTarget, UnifiedMetadataFilter,
};
use crate::storage::{
    EntityData, EntityId, EntityKind, RedDB, RefType, SimilarResult, StoreStats, UnifiedEntity,
    UnifiedStore,
};
use reddb_rql::ast::{
    AlterOperation, AlterQueueQuery, AlterTableQuery, CompareOp, CreateCollectionQuery,
    CreateIndexQuery, CreateQueueQuery, CreateTableQuery, CreateTimeSeriesQuery, CreateTreeQuery,
    CreateVectorQuery, DeleteQuery, DropCollectionQuery, DropDocumentQuery, DropForkQuery,
    DropGraphQuery, DropIndexQuery, DropKvQuery, DropQueueQuery, DropTableQuery,
    DropTimeSeriesQuery, DropTreeQuery, DropVectorQuery, EventsBackfillQuery, ExplainAlterQuery,
    ExplainFormat, FieldRef, Filter, ForkStoreQuery, FusionStrategy, GraphCommand, HybridQuery,
    IndexMethod, InsertEntityType, InsertQuery, JoinQuery, JoinType, OrderByClause,
    ProbabilisticCommand, Projection, PromoteForkQuery, QueryExpr, QueueCommand, QueueSelectQuery,
    QueueSide, SearchCommand, TableQuery, TreeCommand, TruncateQuery, UpdateQuery, VectorQuery,
    VectorSource,
};
use reddb_rql::modes::{detect_mode, parse_multi, QueryMode};
use reddb_types::Value;

static TIMESERIES_TAG_INDEX_OBSERVED_POINTS: AtomicU64 = AtomicU64::new(0);

pub fn reset_timeseries_tag_index_observability() {
    TIMESERIES_TAG_INDEX_OBSERVED_POINTS.store(0, AtomicOrdering::Relaxed);
}

pub fn timeseries_tag_index_observed_points() -> u64 {
    TIMESERIES_TAG_INDEX_OBSERVED_POINTS.load(AtomicOrdering::Relaxed)
}

pub(crate) fn observe_timeseries_tag_index_point() {
    TIMESERIES_TAG_INDEX_OBSERVED_POINTS.fetch_add(1, AtomicOrdering::Relaxed);
}

#[derive(Debug, Clone)]
pub struct ConnectionPoolConfig {
    pub max_connections: usize,
    pub max_idle: usize,
}

impl Default for ConnectionPoolConfig {
    fn default() -> Self {
        Self {
            max_connections: 64,
            max_idle: 16,
        }
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ScanCursor {
    pub offset: usize,
}

#[derive(Debug, Clone)]
pub struct ScanPage {
    pub collection: String,
    pub items: Vec<UnifiedEntity>,
    pub next: Option<ScanCursor>,
    pub total: usize,
}

#[derive(Debug, Clone)]
pub struct SystemInfo {
    pub pid: u32,
    pub cpu_cores: usize,
    pub total_memory_bytes: u64,
    pub available_memory_bytes: u64,
    pub os: String,
    pub arch: String,
    pub hostname: String,
}

impl SystemInfo {
    /// Whether the system has enough cores to benefit from parallelism.
    /// Returns false on single-core machines where thread overhead > gains.
    pub fn should_parallelize() -> bool {
        reddb_engine::execution::should_parallelize()
    }

    pub fn collect() -> Self {
        Self {
            pid: std::process::id(),
            cpu_cores: std::thread::available_parallelism()
                .map(|p| p.get())
                .unwrap_or(1),
            total_memory_bytes: Self::read_total_memory(),
            available_memory_bytes: Self::read_available_memory(),
            os: std::env::consts::OS.to_string(),
            arch: std::env::consts::ARCH.to_string(),
            hostname: std::env::var("HOSTNAME")
                .or_else(|_| std::env::var("COMPUTERNAME"))
                .unwrap_or_else(|_| "unknown".to_string()),
        }
    }

    #[cfg(target_os = "linux")]
    fn read_total_memory() -> u64 {
        std::fs::read_to_string("/proc/meminfo")
            .ok()
            .and_then(|s| {
                s.lines()
                    .find(|l| l.starts_with("MemTotal:"))
                    .and_then(|l| {
                        l.split_whitespace()
                            .nth(1)
                            .and_then(|v| v.parse::<u64>().ok())
                    })
                    .map(|kb| kb * 1024)
            })
            .unwrap_or(0)
    }

    #[cfg(target_os = "linux")]
    fn read_available_memory() -> u64 {
        std::fs::read_to_string("/proc/meminfo")
            .ok()
            .and_then(|s| {
                s.lines()
                    .find(|l| l.starts_with("MemAvailable:"))
                    .and_then(|l| {
                        l.split_whitespace()
                            .nth(1)
                            .and_then(|v| v.parse::<u64>().ok())
                    })
                    .map(|kb| kb * 1024)
            })
            .unwrap_or(0)
    }

    #[cfg(not(target_os = "linux"))]
    fn read_total_memory() -> u64 {
        0
    }

    #[cfg(not(target_os = "linux"))]
    fn read_available_memory() -> u64 {
        0
    }
}

#[derive(Debug, Clone)]
pub struct RuntimeStats {
    pub active_connections: usize,
    pub idle_connections: usize,
    pub total_checkouts: u64,
    pub paged_mode: bool,
    pub started_at_unix_ms: u128,
    pub store: StoreStats,
    pub system: SystemInfo,
    pub result_blob_cache: crate::storage::cache::BlobCacheStats,
    pub kv: KvStats,
    pub metrics_ingest: MetricsIngestStats,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MetricsTenantActivityStats {
    pub tenant: String,
    pub namespace: String,
    pub operation: String,
    pub count: u64,
}

#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct MetricsIngestStats {
    pub samples_accepted: u64,
    pub series_accepted: u64,
    pub samples_rejected: u64,
    pub series_rejected: u64,
    pub series_rejected_cardinality_budget: u64,
}

#[derive(Debug, Default)]
pub(crate) struct MetricsIngestCounters {
    samples_accepted: AtomicU64,
    series_accepted: AtomicU64,
    samples_rejected: AtomicU64,
    series_rejected: AtomicU64,
    series_rejected_cardinality_budget: AtomicU64,
}

impl MetricsIngestCounters {
    pub(crate) fn record(
        &self,
        accepted_samples: u64,
        accepted_series: u64,
        rejected_samples: u64,
        rejected_series: u64,
    ) {
        self.samples_accepted
            .fetch_add(accepted_samples, AtomicOrdering::Relaxed);
        self.series_accepted
            .fetch_add(accepted_series, AtomicOrdering::Relaxed);
        self.samples_rejected
            .fetch_add(rejected_samples, AtomicOrdering::Relaxed);
        self.series_rejected
            .fetch_add(rejected_series, AtomicOrdering::Relaxed);
    }

    pub(crate) fn record_cardinality_budget_rejections(&self, rejected_series: u64) {
        self.series_rejected_cardinality_budget
            .fetch_add(rejected_series, AtomicOrdering::Relaxed);
    }

    pub(crate) fn snapshot(&self) -> MetricsIngestStats {
        MetricsIngestStats {
            samples_accepted: self.samples_accepted.load(AtomicOrdering::Relaxed),
            series_accepted: self.series_accepted.load(AtomicOrdering::Relaxed),
            samples_rejected: self.samples_rejected.load(AtomicOrdering::Relaxed),
            series_rejected: self.series_rejected.load(AtomicOrdering::Relaxed),
            series_rejected_cardinality_budget: self
                .series_rejected_cardinality_budget
                .load(AtomicOrdering::Relaxed),
        }
    }
}

#[derive(Debug, Default)]
pub(crate) struct MetricsTenantActivityCounters {
    inner: Mutex<BTreeMap<(String, String, String), u64>>,
}

impl MetricsTenantActivityCounters {
    pub(crate) fn record(&self, tenant: &str, namespace: &str, operation: &str) {
        let mut inner = self
            .inner
            .lock()
            .unwrap_or_else(|poison| poison.into_inner());
        let key = (
            tenant.to_string(),
            namespace.to_string(),
            operation.to_string(),
        );
        *inner.entry(key).or_insert(0) += 1;
    }

    pub(crate) fn snapshot(&self) -> Vec<MetricsTenantActivityStats> {
        let inner = self
            .inner
            .lock()
            .unwrap_or_else(|poison| poison.into_inner());
        inner
            .iter()
            .map(
                |((tenant, namespace, operation), count)| MetricsTenantActivityStats {
                    tenant: tenant.clone(),
                    namespace: namespace.clone(),
                    operation: operation.clone(),
                    count: *count,
                },
            )
            .collect()
    }
}

#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct KvStats {
    pub puts: u64,
    pub gets: u64,
    pub deletes: u64,
    pub incrs: u64,
    pub cas_success: u64,
    pub cas_conflict: u64,
    pub watch_streams_active: u64,
    pub watch_events_emitted: u64,
    pub watch_drops: u64,
}

#[derive(Debug, Default)]
pub(crate) struct KvStatsCounters {
    puts: AtomicU64,
    gets: AtomicU64,
    deletes: AtomicU64,
    incrs: AtomicU64,
    cas_success: AtomicU64,
    cas_conflict: AtomicU64,
    watch_streams_active: AtomicU64,
    watch_events_emitted: AtomicU64,
    watch_drops: AtomicU64,
}

#[derive(Debug, Default)]
pub(crate) struct KvTagIndex {
    tag_to_entries: parking_lot::RwLock<HashMap<(String, String), HashMap<String, EntityId>>>,
    key_to_tags: parking_lot::RwLock<HashMap<(String, String), BTreeSet<String>>>,
}

impl KvTagIndex {
    pub(crate) fn replace(&self, collection: &str, key: &str, id: EntityId, tags: &[String]) {
        let entry_key = (collection.to_string(), key.to_string());
        let new_tags: BTreeSet<String> = tags
            .iter()
            .map(|tag| tag.trim())
            .filter(|tag| !tag.is_empty())
            .map(ToOwned::to_owned)
            .collect();

        let old_tags = {
            let mut key_to_tags = self.key_to_tags.write();
            if new_tags.is_empty() {
                key_to_tags.remove(&entry_key)
            } else {
                key_to_tags.insert(entry_key.clone(), new_tags.clone())
            }
        };

        let mut tag_to_entries = self.tag_to_entries.write();
        if let Some(old_tags) = old_tags {
            for tag in old_tags {
                let scoped = (collection.to_string(), tag);
                let remove_scoped = if let Some(entries) = tag_to_entries.get_mut(&scoped) {
                    entries.remove(key);
                    entries.is_empty()
                } else {
                    false
                };
                if remove_scoped {
                    tag_to_entries.remove(&scoped);
                }
            }
        }

        for tag in new_tags {
            tag_to_entries
                .entry((collection.to_string(), tag))
                .or_default()
                .insert(key.to_string(), id);
        }
    }

    pub(crate) fn remove(&self, collection: &str, key: &str) {
        let entry_key = (collection.to_string(), key.to_string());
        let old_tags = self.key_to_tags.write().remove(&entry_key);
        let Some(old_tags) = old_tags else {
            return;
        };

        let mut tag_to_entries = self.tag_to_entries.write();
        for tag in old_tags {
            let scoped = (collection.to_string(), tag);
            let remove_scoped = if let Some(entries) = tag_to_entries.get_mut(&scoped) {
                entries.remove(key);
                entries.is_empty()
            } else {
                false
            };
            if remove_scoped {
                tag_to_entries.remove(&scoped);
            }
        }
    }

    pub(crate) fn entries_for_tags(
        &self,
        collection: &str,
        tags: &[String],
    ) -> Vec<(String, EntityId)> {
        if tags.is_empty() {
            return Vec::new();
        }

        let tag_to_entries = self.tag_to_entries.read();
        let mut out: HashMap<String, EntityId> = HashMap::new();
        for tag in tags {
            let scoped = (collection.to_string(), tag.trim().to_string());
            if let Some(entries) = tag_to_entries.get(&scoped) {
                for (key, id) in entries {
                    out.entry(key.clone()).or_insert(*id);
                }
            }
        }
        out.into_iter().collect()
    }

    pub(crate) fn tags_for_key(&self, collection: &str, key: &str) -> Vec<String> {
        self.key_to_tags
            .read()
            .get(&(collection.to_string(), key.to_string()))
            .map(|tags| tags.iter().cloned().collect())
            .unwrap_or_default()
    }
}

impl KvStatsCounters {
    pub(crate) fn snapshot(&self) -> KvStats {
        KvStats {
            puts: self.puts.load(AtomicOrdering::Relaxed),
            gets: self.gets.load(AtomicOrdering::Relaxed),
            deletes: self.deletes.load(AtomicOrdering::Relaxed),
            incrs: self.incrs.load(AtomicOrdering::Relaxed),
            cas_success: self.cas_success.load(AtomicOrdering::Relaxed),
            cas_conflict: self.cas_conflict.load(AtomicOrdering::Relaxed),
            watch_streams_active: self.watch_streams_active.load(AtomicOrdering::Relaxed),
            watch_events_emitted: self.watch_events_emitted.load(AtomicOrdering::Relaxed),
            watch_drops: self.watch_drops.load(AtomicOrdering::Relaxed),
        }
    }

    pub(crate) fn incr_puts(&self) {
        self.puts.fetch_add(1, AtomicOrdering::Relaxed);
    }

    pub(crate) fn incr_gets(&self) {
        self.gets.fetch_add(1, AtomicOrdering::Relaxed);
    }

    pub(crate) fn incr_deletes(&self) {
        self.deletes.fetch_add(1, AtomicOrdering::Relaxed);
    }

    pub(crate) fn incr_incrs(&self) {
        self.incrs.fetch_add(1, AtomicOrdering::Relaxed);
    }

    pub(crate) fn incr_cas_success(&self) {
        self.cas_success.fetch_add(1, AtomicOrdering::Relaxed);
    }

    pub(crate) fn incr_cas_conflict(&self) {
        self.cas_conflict.fetch_add(1, AtomicOrdering::Relaxed);
    }

    pub(crate) fn incr_watch_streams_active(&self) {
        self.watch_streams_active
            .fetch_add(1, AtomicOrdering::Relaxed);
    }

    pub(crate) fn decr_watch_streams_active(&self) {
        self.watch_streams_active
            .fetch_sub(1, AtomicOrdering::Relaxed);
    }

    pub(crate) fn incr_watch_events_emitted(&self) {
        self.watch_events_emitted
            .fetch_add(1, AtomicOrdering::Relaxed);
    }

    pub(crate) fn add_watch_drops(&self, count: u64) {
        self.watch_drops.fetch_add(count, AtomicOrdering::Relaxed);
    }
}

#[derive(Debug, Clone)]
pub struct RuntimeQueryResult {
    pub query: String,
    pub mode: QueryMode,
    pub statement: &'static str,
    pub engine: &'static str,
    pub result: UnifiedResult,
    pub affected_rows: u64,
    /// High-level statement type: "select", "insert", "update", "delete", "create", "drop", "alter"
    pub statement_type: &'static str,
    pub bookmark: Option<String>,
    pub notice: Option<String>,
}

impl RuntimeQueryResult {
    /// Construct a result representing a DML operation (insert/update/delete).
    pub fn dml_result(
        query: String,
        affected: u64,
        statement_type: &'static str,
        engine: &'static str,
    ) -> Self {
        Self {
            query,
            mode: QueryMode::Sql,
            statement: statement_type,
            engine,
            result: UnifiedResult::empty(),
            affected_rows: affected,
            statement_type,
            bookmark: None,
            notice: None,
        }
    }

    /// Construct a result representing a DDL message (create/drop/alter).
    pub fn ok_message(query: String, message: &str, statement_type: &'static str) -> Self {
        let mut result = UnifiedResult::empty();
        let mut record = UnifiedRecord::new();
        record.set("message", Value::text(message.to_string()));
        result.push(record);
        result.columns = vec!["message".to_string()];

        Self {
            query,
            mode: QueryMode::Sql,
            statement: statement_type,
            engine: "runtime-ddl",
            result,
            affected_rows: 0,
            statement_type,
            bookmark: None,
            notice: None,
        }
    }

    /// Construct a multi-column record result for read-only meta commands
    /// (EXPLAIN ALTER, schema introspection, etc.). Each row is a Vec of
    /// (column_name, value) pairs in column order.
    pub fn ok_records(
        query: String,
        columns: Vec<String>,
        rows: Vec<Vec<(String, Value)>>,
        statement_type: &'static str,
    ) -> Self {
        let mut result = UnifiedResult::empty();
        for row in rows {
            let mut record = UnifiedRecord::new();
            for (k, v) in row {
                record.set(&k, v);
            }
            result.push(record);
        }
        result.columns = columns;

        Self {
            query,
            mode: QueryMode::Sql,
            statement: statement_type,
            engine: "runtime-meta",
            result,
            affected_rows: 0,
            statement_type,
            bookmark: None,
            notice: None,
        }
    }

    pub fn bookmark_token(&self) -> Option<&str> {
        self.bookmark.as_deref()
    }
}

#[derive(Debug, Clone)]
pub struct RuntimeQueryExplain {
    pub query: String,
    pub mode: QueryMode,
    pub statement: &'static str,
    pub is_universal: bool,
    pub plan_cost: crate::storage::query::planner::PlanCost,
    pub estimated_rows: f64,
    pub estimated_selectivity: f64,
    pub estimated_confidence: f64,
    pub passes_applied: Vec<String>,
    pub logical_plan: CanonicalLogicalPlan,
    /// Names of any CTEs declared by a leading `WITH` clause. Empty
    /// for non-CTE queries. The plan tree is built against the
    /// post-inlining body, so each CTE's body is reachable inside
    /// `logical_plan` as a regular `Subquery` (or, for bare refs, the
    /// inlined Table node verbatim). This list lets renderers prepend
    /// `CteScan` markers so operators see which CTEs were resolved.
    pub cte_materializations: Vec<String>,
}

#[derive(Debug, Clone)]
pub struct RuntimeIvfMatch {
    pub entity_id: u64,
    pub distance: f32,
    pub entity: Option<UnifiedEntity>,
}

#[derive(Debug, Clone)]
pub struct RuntimeIvfSearchResult {
    pub collection: String,
    pub k: usize,
    pub n_lists: usize,
    pub n_probes: usize,
    pub stats: IvfStats,
    pub matches: Vec<RuntimeIvfMatch>,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RuntimeGraphDirection {
    Outgoing,
    Incoming,
    Both,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RuntimeGraphTraversalStrategy {
    Bfs,
    Dfs,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RuntimeGraphPathAlgorithm {
    Bfs,
    Dijkstra,
    AStar,
    BellmanFord,
}

#[derive(Debug, Clone)]
pub struct RuntimeGraphNode {
    pub id: String,
    pub label: String,
    pub node_type: String,
    pub out_edge_count: u32,
    pub in_edge_count: u32,
}

#[derive(Debug, Clone)]
pub struct RuntimeGraphEdge {
    pub source: String,
    pub target: String,
    pub edge_type: String,
    pub weight: f32,
}

#[derive(Debug, Clone)]
pub struct RuntimeGraphVisit {
    pub depth: usize,
    pub node: RuntimeGraphNode,
}

#[derive(Debug, Clone)]
pub struct RuntimeGraphNeighborhoodResult {
    pub source: String,
    pub direction: RuntimeGraphDirection,
    pub max_depth: usize,
    pub nodes: Vec<RuntimeGraphVisit>,
    pub edges: Vec<RuntimeGraphEdge>,
}

#[derive(Debug, Clone)]
pub struct RuntimeGraphTraversalResult {
    pub source: String,
    pub direction: RuntimeGraphDirection,
    pub strategy: RuntimeGraphTraversalStrategy,
    pub max_depth: usize,
    pub visits: Vec<RuntimeGraphVisit>,
    pub edges: Vec<RuntimeGraphEdge>,
}

#[derive(Debug, Clone)]
pub struct RuntimeGraphPath {
    pub hop_count: usize,
    pub total_weight: f64,
    pub nodes: Vec<RuntimeGraphNode>,
    pub edges: Vec<RuntimeGraphEdge>,
}

#[derive(Debug, Clone)]
pub struct RuntimeGraphPathResult {
    pub source: String,
    pub target: String,
    pub direction: RuntimeGraphDirection,
    pub algorithm: RuntimeGraphPathAlgorithm,
    pub nodes_visited: usize,
    pub negative_cycle_detected: Option<bool>,
    pub path: Option<RuntimeGraphPath>,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RuntimeGraphComponentsMode {
    Connected,
    Weak,
    Strong,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RuntimeGraphCentralityAlgorithm {
    Degree,
    Closeness,
    Betweenness,
    Eigenvector,
    PageRank,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RuntimeGraphCommunityAlgorithm {
    LabelPropagation,
    Louvain,
}

#[derive(Debug, Clone)]
pub struct RuntimeGraphComponent {
    pub id: String,
    pub size: usize,
    pub nodes: Vec<String>,
}

#[derive(Debug, Clone)]
pub struct RuntimeGraphComponentsResult {
    pub mode: RuntimeGraphComponentsMode,
    pub count: usize,
    pub components: Vec<RuntimeGraphComponent>,
}

#[derive(Debug, Clone)]
pub struct RuntimeGraphCentralityScore {
    pub node: RuntimeGraphNode,
    pub score: f64,
}

#[derive(Debug, Clone)]
pub struct RuntimeGraphDegreeScore {
    pub node: RuntimeGraphNode,
    pub in_degree: usize,
    pub out_degree: usize,
    pub total_degree: usize,
}

#[derive(Debug, Clone)]
pub struct RuntimeGraphCentralityResult {
    pub algorithm: RuntimeGraphCentralityAlgorithm,
    pub normalized: Option<bool>,
    pub iterations: Option<usize>,
    pub converged: Option<bool>,
    pub scores: Vec<RuntimeGraphCentralityScore>,
    pub degree_scores: Vec<RuntimeGraphDegreeScore>,
}

#[derive(Debug, Clone)]
pub struct RuntimeGraphCommunity {
    pub id: String,
    pub size: usize,
    pub nodes: Vec<String>,
}

#[derive(Debug, Clone)]
pub struct RuntimeGraphCommunityResult {
    pub algorithm: RuntimeGraphCommunityAlgorithm,
    pub count: usize,
    pub iterations: Option<usize>,
    pub converged: Option<bool>,
    pub modularity: Option<f64>,
    pub passes: Option<usize>,
    pub communities: Vec<RuntimeGraphCommunity>,
}

#[derive(Debug, Clone)]
pub struct RuntimeGraphClusteringResult {
    pub global: f64,
    pub local: Vec<RuntimeGraphCentralityScore>,
    pub triangle_count: Option<usize>,
}

#[derive(Debug, Clone)]
pub struct RuntimeGraphHitsResult {
    pub iterations: usize,
    pub converged: bool,
    pub hubs: Vec<RuntimeGraphCentralityScore>,
    pub authorities: Vec<RuntimeGraphCentralityScore>,
}

#[derive(Debug, Clone)]
pub struct RuntimeGraphCyclesResult {
    pub limit_reached: bool,
    pub cycles: Vec<RuntimeGraphPath>,
}

#[derive(Debug, Clone)]
pub struct RuntimeGraphTopologicalSortResult {
    pub acyclic: bool,
    pub ordered_nodes: Vec<RuntimeGraphNode>,
}

#[derive(Debug, Clone)]
pub struct RuntimeGraphPropertiesResult {
    pub node_count: usize,
    pub edge_count: usize,
    pub self_loop_count: usize,
    pub negative_edge_count: usize,
    pub connected_component_count: usize,
    pub weak_component_count: usize,
    pub strong_component_count: usize,
    pub is_empty: bool,
    pub is_connected: bool,
    pub is_weakly_connected: bool,
    pub is_strongly_connected: bool,
    pub is_complete: bool,
    pub is_complete_directed: bool,
    pub is_cyclic: bool,
    pub is_circular: bool,
    pub is_acyclic: bool,
    pub is_tree: bool,
    pub density: f64,
    pub density_directed: f64,
}

// ============================================================================
// Context Search types
// ============================================================================

#[derive(Debug, Clone)]
pub struct ContextSearchResult {
    pub query: String,
    pub tables: Vec<ContextEntity>,
    pub graph: ContextGraphResult,
    pub vectors: Vec<ContextEntity>,
    pub documents: Vec<ContextEntity>,
    pub key_values: Vec<ContextEntity>,
    pub connections: Vec<ContextConnection>,
    pub summary: ContextSummary,
}

#[derive(Debug, Clone)]
pub struct ContextEntity {
    pub entity: UnifiedEntity,
    pub score: f32,
    pub discovery: DiscoveryMethod,
    pub collection: String,
}

#[derive(Debug, Clone)]
pub enum DiscoveryMethod {
    Indexed {
        field: String,
    },
    GlobalScan,
    CrossReference {
        source_id: u64,
        ref_type: String,
    },
    GraphTraversal {
        source_id: u64,
        edge_type: String,
        depth: usize,
    },
    VectorQuery {
        similarity: f32,
    },
}

#[derive(Debug, Clone)]
pub struct ContextGraphResult {
    pub nodes: Vec<ContextEntity>,
    pub edges: Vec<ContextEntity>,
}

#[derive(Debug, Clone)]
pub struct ContextConnection {
    pub from_id: u64,
    pub to_id: u64,
    pub connection_type: ContextConnectionType,
    pub weight: f32,
}

#[derive(Debug, Clone)]
pub enum ContextConnectionType {
    CrossRef(String),
    GraphEdge(String),
    VectorSimilarity(f32),
}

#[derive(Debug, Clone)]
pub struct ContextSummary {
    pub total_entities: usize,
    pub direct_matches: usize,
    pub expanded_via_graph: usize,
    pub expanded_via_cross_refs: usize,
    pub expanded_via_vector_query: usize,
    pub collections_searched: usize,
    pub execution_time_us: u64,
    pub tiers_used: Vec<String>,
    pub entities_reindexed: usize,
}

struct PoolState {
    next_id: u64,
    active: usize,
    pooled_active: usize,
    idle: Vec<u64>,
    total_checkouts: u64,
}

impl Default for PoolState {
    fn default() -> Self {
        Self {
            next_id: 1,
            active: 0,
            pooled_active: 0,
            idle: Vec::new(),
            total_checkouts: 0,
        }
    }
}

#[derive(Debug, Clone)]
struct RuntimeResultCacheEntry {
    result: RuntimeQueryResult,
    cached_at: std::time::Instant,
    scopes: HashSet<String>,
}

pub const METRIC_CACHE_SHADOW_DIVERGENCE_TOTAL: &str = "cache_shadow_divergence_total";
/// Stable metric names for the query/graph-analytics result cache
/// (issue #802). Exposed under the `red.metrics` surface as
/// `reddb_result_cache_{hit,miss,evict}_total` and readable in-process
/// via `RedDBRuntime::result_cache_metrics`.
pub const METRIC_RESULT_CACHE_HIT_TOTAL: &str = "result_cache_hit_total";
pub const METRIC_RESULT_CACHE_MISS_TOTAL: &str = "result_cache_miss_total";
pub const METRIC_RESULT_CACHE_EVICT_TOTAL: &str = "result_cache_evict_total";
pub(crate) const ASK_ANSWER_CACHE_NAMESPACE: &str = "runtime.ask_answer_cache";
const RMW_LOCK_SHARDS: usize = 64;

struct RmwLockTable {
    shards: Vec<parking_lot::Mutex<HashMap<String, Arc<parking_lot::Mutex<()>>>>>,
}

impl RmwLockTable {
    fn new() -> Self {
        let shards = (0..RMW_LOCK_SHARDS)
            .map(|_| parking_lot::Mutex::new(HashMap::new()))
            .collect();
        Self { shards }
    }

    fn lock_for(&self, collection: &str, key: &str) -> Arc<parking_lot::Mutex<()>> {
        use std::hash::{Hash, Hasher};

        let mut hasher = std::collections::hash_map::DefaultHasher::new();
        collection.hash(&mut hasher);
        key.hash(&mut hasher);
        let shard_idx = (hasher.finish() as usize) % self.shards.len();
        let map_key = format!("{collection}\u{1f}{key}");
        let mut shard = self.shards[shard_idx].lock();
        shard
            .entry(map_key)
            .or_insert_with(|| Arc::new(parking_lot::Mutex::new(())))
            .clone()
    }
}

#[derive(Debug, Default)]
pub(crate) struct CheckpointProjectionStats {
    last_materialized_lsn: std::sync::atomic::AtomicU64,
    checkpoints_completed: std::sync::atomic::AtomicU64,
    last_checkpoint_duration_ms: std::sync::atomic::AtomicU64,
}

impl CheckpointProjectionStats {
    pub(crate) fn record_checkpoint(&self, lsn: u64, duration_ms: u64) {
        use std::sync::atomic::Ordering;

        self.last_materialized_lsn.store(lsn, Ordering::Relaxed);
        self.last_checkpoint_duration_ms
            .store(duration_ms, Ordering::Relaxed);
        self.checkpoints_completed.fetch_add(1, Ordering::Relaxed);
    }

    pub(crate) fn snapshot(&self, current_lsn: u64) -> CheckpointProjectionStatsSnapshot {
        use std::sync::atomic::Ordering;

        let last_materialized_lsn = self.last_materialized_lsn.load(Ordering::Relaxed);
        CheckpointProjectionStatsSnapshot {
            current_lsn,
            last_materialized_lsn,
            projection_lag: current_lsn.saturating_sub(last_materialized_lsn),
            checkpoints_completed: self.checkpoints_completed.load(Ordering::Relaxed),
            last_checkpoint_duration_ms: self.last_checkpoint_duration_ms.load(Ordering::Relaxed),
        }
    }
}

#[derive(Debug, Clone, Copy)]
pub(crate) struct CheckpointProjectionStatsSnapshot {
    pub(crate) current_lsn: u64,
    pub(crate) last_materialized_lsn: u64,
    pub(crate) projection_lag: u64,
    pub(crate) checkpoints_completed: u64,
    pub(crate) last_checkpoint_duration_ms: u64,
}

#[derive(Debug, Clone)]
pub(crate) struct ScrubStatsSnapshot {
    pub(crate) last_run_unix_ms: u64,
    pub(crate) last_findings_count: u64,
    pub(crate) background_status: String,
    pub(crate) background_verified_objects: u64,
    pub(crate) background_total_objects: u64,
    pub(crate) verified: reddb_file::StorageScrubVerifiedCounters,
}

#[derive(Debug, Clone)]
pub(crate) struct ScrubRuntimeState {
    pub(crate) last_run_unix_ms: u64,
    pub(crate) last_findings_count: u64,
    pub(crate) background_cursor: usize,
    pub(crate) background_status: String,
    pub(crate) background_verified_objects: u64,
    pub(crate) background_total_objects: u64,
    pub(crate) verified: reddb_file::StorageScrubVerifiedCounters,
}

impl Default for ScrubRuntimeState {
    fn default() -> Self {
        Self {
            last_run_unix_ms: 0,
            last_findings_count: 0,
            background_cursor: 0,
            background_status: "idle".to_string(),
            background_verified_objects: 0,
            background_total_objects: 0,
            verified: reddb_file::StorageScrubVerifiedCounters::default(),
        }
    }
}

impl ScrubRuntimeState {
    pub(crate) fn snapshot(&self) -> ScrubStatsSnapshot {
        ScrubStatsSnapshot {
            last_run_unix_ms: self.last_run_unix_ms,
            last_findings_count: self.last_findings_count,
            background_status: self.background_status.clone(),
            background_verified_objects: self.background_verified_objects,
            background_total_objects: self.background_total_objects,
            verified: self.verified.clone(),
        }
    }
}

struct RuntimeInner {
    functions: parking_lot::RwLock<function_catalog::FunctionCatalog>,
    db: Arc<RedDB>,
    layout: PhysicalLayout,
    embedded_single_file: bool,
    /// The one memory budget this process runs under (ADR 0073 §1),
    /// resolved once at boot and immutable for the process lifetime.
    pub(crate) memory_budget: crate::storage::memory_budget::MemoryBudget,
    /// The one shared accounting pool (ADR 0073 §2): each big consumer's share
    /// of the budget above, and the live usage they report into it.
    pub(crate) memory_accounting: Arc<crate::storage::memory_pools::MemoryAccounting>,
    memory_reservations: parking_lot::Mutex<memory_admission::MemoryReservations>,
    indices: IndexCatalog,
    pool_config: ConnectionPoolConfig,
    pool: Mutex<PoolState>,
    started_at_unix_ms: u128,
    probabilistic: probabilistic_store::ProbabilisticStore,
    index_store: index_store::IndexStore,
    cdc: crate::replication::cdc::CdcBuffer,
    pub(crate) checkpoint_projection_stats: CheckpointProjectionStats,
    pub(crate) scrub_state: parking_lot::Mutex<ScrubRuntimeState>,
    pub(crate) checkpoint_columnar_emission_budget_chunks: usize,
    pub(crate) columnar_projection_size_floor_rows: usize,
    backup_scheduler: crate::replication::scheduler::BackupScheduler,
    query_cache: parking_lot::RwLock<crate::storage::query::planner::cache::PlanCache>,
    result_cache: parking_lot::RwLock<(
        HashMap<String, RuntimeResultCacheEntry>,
        std::collections::VecDeque<String>,
    )>,
    result_blob_cache: crate::storage::cache::BlobCache,
    result_blob_entries: parking_lot::RwLock<(
        HashMap<String, RuntimeResultCacheEntry>,
        std::collections::VecDeque<String>,
    )>,
    ask_answer_cache_entries:
        parking_lot::RwLock<(HashSet<String>, std::collections::VecDeque<String>)>,
    result_cache_shadow_divergences: std::sync::atomic::AtomicU64,
    /// Result-cache observability counters (issue #802). Process-cumulative
    /// hit/miss/eviction tallies surfaced under `red.metrics`.
    result_cache_hits: std::sync::atomic::AtomicU64,
    result_cache_misses: std::sync::atomic::AtomicU64,
    result_cache_evictions: std::sync::atomic::AtomicU64,
    ask_daily_spend:
        parking_lot::RwLock<HashMap<String, crate::runtime::ai::cost_guard::DailyState>>,
    /// Process-local queue message locks used to emulate `SKIP LOCKED`-style
    /// claim semantics for concurrent queue consumers inside this runtime.
    queue_message_locks: parking_lot::RwLock<HashMap<String, Arc<parking_lot::Mutex<()>>>>,
    /// Process-local read-modify-write locks. The table is sharded by
    /// `(collection, key)` and each entry has its own mutex, so unrelated keys
    /// in the same collection do not serialize behind one global lock.
    rmw_locks: RmwLockTable,
    planner_dirty_tables: parking_lot::RwLock<HashSet<String>>,
    ec_registry: Arc<crate::ec::config::EcRegistry>,
    config_registry: Arc<crate::auth::registry::ConfigRegistry>,
    ec_worker: crate::ec::worker::EcWorker,
    /// Optional AuthStore — injected by server boot when auth is
    /// enabled. Required for `Value::Secret` auto-encrypt/decrypt
    /// because the AES key lives in the vault KV under the
    /// `red.secret.aes_key` entry.
    auth_store: parking_lot::RwLock<Option<Arc<crate::auth::store::AuthStore>>>,
    /// Optional OAuth/OIDC JWT validator. Wired by server boot when
    /// the operator configures issuer + JWKS via env / CLI. HTTP and
    /// wire transports read this on every bearer-token request and,
    /// when the token decodes as a JWT, validate it against the
    /// configured issuer + audience + signature before falling back to
    /// the local AuthStore lookup.
    oauth_validator: parking_lot::RwLock<Option<Arc<crate::auth::oauth::OAuthValidator>>>,
    /// Browser credential layer — the hybrid-token authority (issue
    /// #936, PRD #930). When wired (an operator opts the browser
    /// endpoint in), the RedWire WS handshake accepts the short-lived
    /// access JWT it mints, and the `/auth/browser/*` HTTP endpoints
    /// issue/rotate the access+refresh pair. `None` leaves the browser
    /// flow inert — default-deny, matching the WS endpoint itself.
    browser_token_authority:
        parking_lot::RwLock<Option<Arc<crate::auth::browser_token::BrowserTokenAuthority>>>,
    /// View registry (Phase 2.1 PG parity).
    ///
    /// Holds the parsed `SELECT` body for every view created via
    /// `CREATE [MATERIALIZED] VIEW`. Queries that reference a view name
    /// substitute the stored `QueryExpr` at execution time. Materialized
    /// views additionally back onto the shared `MaterializedViewCache`
    /// (see `RuntimeInner::materialized_views`).
    ///
    /// This is in-memory only in Phase 2.1 — view definitions do not
    /// survive a restart. Persistence is a Phase 3 follow-up.
    views: parking_lot::RwLock<HashMap<String, Arc<reddb_rql::ast::CreateViewQuery>>>,
    materialized_views: parking_lot::RwLock<crate::storage::cache::result::MaterializedViewCache>,
    /// Per-collection retention sweeper state (issue #584 slice 12).
    /// Tracks `last_sweep_at_ms`, row/segment retirement totals, and
    /// the latest pending-rows estimate that feeds `red.retention`.
    /// In-memory only; resets across restart.
    pub(crate) retention_sweeper:
        parking_lot::RwLock<crate::runtime::retention_sweeper::RetentionSweeperState>,
    /// Per-connection transaction contexts, local tenant overrides, and the
    /// shared MVCC snapshot manager behind the Transaction State interface.
    transaction_state: crate::runtime::transaction_state::TransactionState,
    /// Intent-lock hierarchy (IS/IX/S/X) used to break the implicit
    /// global-write serialisation in write paths. Populated at boot
    /// with `concurrency.locking.deadlock_timeout_ms` from the matrix
    /// and wired through DML/DDL dispatch in later P1 tasks.
    /// Dormant until P1.T3 flips the read path to `(Global,IS) →
    /// (Collection,IS)` and P1.T4/T5 pick up writes/DDL.
    lock_manager: Arc<crate::runtime::lock_manager::LockManager>,
    /// Perf-parity env-var overrides (`REDDB_<UP_DOTTED_KEY>`).
    /// Populated once at boot, read by every config getter; takes
    /// precedence over the persisted red_config value so operators
    /// can hot-fix a bad config by restarting with a different env
    /// var set. Keys are restricted to those declared in the matrix.
    env_config_overrides: HashMap<String, String>,
    /// Row-level security policies (Phase 2.5 PG parity).
    ///
    /// Keyed by `(table_name, policy_name)`; the set of tables with RLS
    /// enforcement toggled on lives in `rls_enabled_tables`. Filter
    /// enforcement hooks into the read path via `collect_rls_filters()`
    /// — see `runtime::impl_core`.
    rls_policies:
        parking_lot::RwLock<HashMap<(String, String), Arc<reddb_rql::ast::CreatePolicyQuery>>>,
    rls_enabled_tables: parking_lot::RwLock<HashSet<String>>,
    /// Foreign Data Wrapper registry (Phase 3.2 PG parity).
    ///
    /// Maps server names → wrapper instances and foreign-table names →
    /// definitions. Queries referencing a registered foreign table are
    /// re-routed to `ForeignTableRegistry::scan` by the read-path
    /// rewriter; reads against unknown names fall through to the native
    /// collection lookup.
    foreign_tables: Arc<crate::storage::fdw::ForeignTableRegistry>,
    /// Per-connection list of tuples marked for deletion by the current
    /// transaction (Phase 2.3.2b MVCC tombstones + 2.3.2e savepoints).
    /// Each entry is `(collection, entity_id, stamper_xid, previous_xmax)`
    /// — the xid that stamped xmax on the tuple plus the value it
    /// replaced. For a plain transaction the
    /// stamper equals `ctx.xid`; with savepoints the stamper equals
    /// the innermost open sub-xid so ROLLBACK TO SAVEPOINT can revive
    /// only the matching subset. COMMIT drains the whole conn list
    /// and keeps the committed tombstones; ROLLBACK (whole-tx) revives them all;
    /// ROLLBACK TO SAVEPOINT revives those with `stamper_xid >=
    /// savepoint_xid`. Autocommit DELETE bypasses this map.
    pending_tombstones: parking_lot::RwLock<
        HashMap<
            u64,
            Vec<(
                String,
                crate::storage::unified::entity::EntityId,
                crate::storage::transaction::snapshot::Xid,
                crate::storage::transaction::snapshot::Xid,
            )>,
        >,
    >,
    /// Per-connection table-row UPDATE versions created by an open
    /// transaction. Each entry is `(collection, old_entity_id,
    /// new_entity_id, stamper_xid, previous_xmax)`. COMMIT keeps both physical
    /// versions and drops the pending marker; ROLLBACK revives the old
    /// version and removes the new uncommitted version.
    pending_versioned_updates: parking_lot::RwLock<
        HashMap<
            u64,
            Vec<(
                String,
                crate::storage::unified::entity::EntityId,
                crate::storage::unified::entity::EntityId,
                crate::storage::transaction::snapshot::Xid,
                crate::storage::transaction::snapshot::Xid,
            )>,
        >,
    >,
    /// Per-connection push idempotency keys written by an open transaction.
    /// Each entry is `(queue, dedup_key, metadata_entity_id)`. COMMIT checks
    /// the queue/key subject for first-committer-wins conflicts; ROLLBACK
    /// removes the uncommitted metadata row so later producer retries enqueue
    /// normally.
    pending_queue_dedup: parking_lot::RwLock<HashMap<u64, Vec<(String, String, EntityId)>>>,
    pending_kv_watch_events:
        parking_lot::RwLock<HashMap<u64, Vec<(u64, crate::replication::cdc::KvWatchEvent)>>>,
    pending_vault_writes: parking_lot::RwLock<HashMap<u64, Vec<impl_kv::PendingVaultWrite>>>,
    pending_store_wal_actions:
        parking_lot::RwLock<HashMap<u64, crate::storage::unified::DeferredStoreWalActions>>,
    /// Transaction-scoped table UPDATE CLAIM ownership.
    ///
    /// Keyed by `(collection, logical_row_id)` and valued by connection id.
    /// Only claim selection consults this map: ordinary DML still follows
    /// MVCC first-committer-wins instead of waiting on claim locks.
    pending_claim_locks: parking_lot::RwLock<HashMap<(String, EntityId), u64>>,
    /// Table-scoped tenancy registry (Phase 2.5.4).
    ///
    /// Maps `table_name → tenant_column`. DML auto-fill looks here to
    /// inject `CURRENT_TENANT()` on INSERTs that omit the column, and
    /// DDL keeps the in-memory registry in sync with the
    /// `tenant_tables.*` keys in red_config. Read-side enforcement
    /// piggy-backs on the existing RLS infrastructure: every entry
    /// installs an implicit `col = CURRENT_TENANT()` policy.
    tenant_tables: parking_lot::RwLock<HashMap<String, String>>,
    /// Monotonic epoch bumped on every DDL / schema-mutating operation
    /// that calls `invalidate_plan_cache`. Prepared statements capture
    /// this at PREPARE and re-check at EXECUTE — a mismatch means the
    /// cached shape may reference dropped or renamed columns and the
    /// client must re-PREPARE.
    ddl_epoch: std::sync::atomic::AtomicU64,
    /// Public-mutation gate (PLAN.md W1).
    ///
    /// Built once at construction from the immutable subset of
    /// `RedDBOptions` (read_only flag + replication role). Every public
    /// mutation surface — SQL DML/DDL, gRPC mutating RPCs, HTTP/native
    /// wire mutations, admin maintenance endpoints, serverless
    /// lifecycle — consults `write_gate.check(WriteKind::*)` before
    /// dispatching to storage. The replica internal apply path
    /// (`LogicalChangeApplier`) reaches into the store directly and
    /// bypasses the gate by construction.
    write_gate: Arc<crate::runtime::write_gate::WriteGate>,
    /// Process lifecycle state machine (PLAN.md Phase 1 — Lifecycle
    /// Contract). Drives `/health/live`, `/health/ready`,
    /// `/health/startup`, and `POST /admin/shutdown` so any
    /// orchestrator (K8s preStop, Fly autostop, ECS task drain,
    /// systemd) can coordinate without losing data.
    lifecycle: crate::runtime::lifecycle::Lifecycle,
    /// Operator-imposed resource limits (PLAN.md Phase 4.1).
    /// Read once at boot from `RED_MAX_*` env vars; consulted by
    /// observability and (in follow-up commits) the per-write
    /// enforcement points.
    resource_limits: crate::runtime::resource_limits::ResourceLimits,
    query_budget_slots: AtomicU64,
    /// Append-only audit log for admin mutations (PLAN.md Phase
    /// 6.5). Lives next to the primary `.rdb` file so backup +
    /// restore flows ship it alongside the data.
    audit_log: Arc<crate::runtime::audit_log::AuditLogger>,
    /// Durable control-plane evidence ledger. Producer slices emit
    /// through this sink and decide fail-open/fail-closed using
    /// `control_event_config`.
    control_event_ledger:
        parking_lot::RwLock<Arc<dyn crate::runtime::control_events::ControlEventLedger>>,
    control_event_config: crate::runtime::control_events::ControlEventConfig,
    /// Data-plane query audit stream. Kept separate from the Control
    /// Event Ledger so scoped query metadata cannot be confused with
    /// governance evidence.
    query_audit: Arc<crate::runtime::query_audit::QueryAuditStream>,
    /// Serverless writer-lease state machine. `None` when the operator
    /// did not opt into lease fencing (`RED_LEASE_REQUIRED` unset/false).
    /// When set, owns the {acquire/refresh/release/lost} transitions and
    /// is the single place that mutates `write_gate.set_lease_state` and
    /// records lease/* audit entries — keeping those two side-effects
    /// from drifting.
    lease_lifecycle: std::sync::OnceLock<Arc<crate::runtime::lease_lifecycle::LeaseLifecycle>>,
    /// PLAN.md Phase 11.5 — counters bumped by the replica apply
    /// loop on `Gap` / `Divergence` / `Apply` errors so /metrics
    /// surfaces them as `reddb_replica_apply_errors_total{kind}`.
    replica_apply_metrics: Arc<crate::replication::logical::ReplicaApplyMetrics>,
    /// Issue #1243 (PRD #1237 Phase B) — primary↔replica reconnect counter
    /// bumped by the replica loop's health-persist chokepoint when the link
    /// drops and restores. Surfaced as `reddb_replication_reconnects_total`
    /// and in the red-ui replica status read model.
    replica_link_metrics: Arc<crate::replication::reconnect::ReplicaLinkMetrics>,
    /// PLAN.md Phase 4.4 — per-caller QPS quotas. Disabled (no-op)
    /// when `RED_MAX_QPS_PER_CALLER` is unset.
    quota_bucket: crate::runtime::quota_bucket::QuotaBucket,
    /// Issue #120 — token → schema entity reverse index, kept current
    /// incrementally on DDL events. Consumed by AskPipeline (issue
    /// #121) Stage 2 to narrow vector-search candidates before any
    /// embedding compute. Mutated only from DDL execution paths.
    schema_vocabulary: parking_lot::RwLock<crate::runtime::schema_vocabulary::SchemaVocabulary>,
    /// Issue #205 — dedicated slow-query sink (`red-slow.log`).
    /// Built once at runtime startup; below-threshold calls pay only a
    /// single relaxed atomic load. Threshold + sample-pct come from
    /// `runtime.slow_query.threshold_ms` / `.sample_pct` (config matrix)
    /// at construction; live tuning via the config tree is a follow-up.
    slow_query_logger: Arc<crate::telemetry::slow_query_logger::SlowQueryLogger>,
    /// Issue #1238 — operational telemetry substrate for slow-query events
    /// (ADR 0060, §2). Bounded ring buffer that the slow-query logger dual-
    /// writes into alongside `red-slow.log`. Exposed to read-model consumers
    /// via `slow_query_store()`.
    slow_query_store: Arc<crate::telemetry::slow_query_store::SlowQueryStore>,
    /// Process-local normal-KV operation counters. These are intentionally
    /// runtime-local; persistent accounting belongs in catalog stats.
    kv_stats: KvStatsCounters,
    metrics_ingest_stats: MetricsIngestCounters,
    metrics_tenant_activity_stats: MetricsTenantActivityCounters,
    /// Issue #1457 — Concurrent claim counters rendered onto `/metrics`.
    /// Process-local; counters reset on restart by design.
    claim_telemetry: std::sync::Arc<claim_telemetry::ClaimTelemetryCounters>,
    /// Slice 10 of issue #527 — Prometheus counters for the queue
    /// lifecycle (delivered/acked/nacked) rendered onto `/metrics`.
    /// Process-local; counters reset on restart by design.
    queue_telemetry: std::sync::Arc<queue_telemetry::QueueTelemetryCounters>,
    /// Issue #1241 — query latency histogram substrate (per-`kind`
    /// `reddb_query_duration_seconds`). Process-local measurement + read
    /// model that `/metrics`, `/cluster/status`, and the red-ui percentile
    /// panels all consume; counters reset on restart by design.
    query_latency_telemetry: std::sync::Arc<query_latency_telemetry::QueryLatencyTelemetry>,
    /// Issue #1244 — node CPU/RAM occupancy sampler (ADR 0060 §2 "node
    /// samples"). Process-local measurement + current-snapshot read model
    /// `/cluster/status` consumes; sampled on-demand from the status
    /// handler. See `occupancy_sampler` for the overhead/interval contract.
    occupancy_sampler: std::sync::Arc<occupancy_sampler::OccupancySampler>,
    /// Issue #1245 — per-node load telemetry (active query gauge +
    /// connect/disconnect churn counters). Three atomics; no per-client
    /// labels; feeds `/metrics`, `/cluster/status`, and the red-ui load
    /// panels (ADR 0060 §2 "node samples" data class).
    node_load_telemetry: std::sync::Arc<node_load_telemetry::NodeLoadTelemetry>,
    /// Issue #742 — consumer presence (heartbeat / lease / lifecycle
    /// state per (queue, group, consumer)). Process-local in this
    /// slice; durability across restart lands in the follow-up that
    /// mirrors the registry into `red_queue_meta` rows. Independent
    /// of pending-delivery state by design — see
    /// `storage::queue::presence` for the contract.
    queue_presence: std::sync::Arc<crate::storage::queue::presence::ConsumerPresenceRegistry>,
    /// Issue #743 — vector + TurboQuant introspection registry. Red UI
    /// vector toolbars (and `red.*` vector virtual tables) read
    /// per-collection metadata + artifact state from here so they
    /// never have to reach into `engine::vector_store` or
    /// `engine::turboquant::*` internals. Engine publish points (build
    /// start / finish, fallback toggle, drop) call into this in
    /// follow-up slices of PRD #735; the public Rust surface lives on
    /// the runtime and does not change when those land.
    vector_introspection:
        std::sync::Arc<crate::storage::vector::introspection::VectorIntrospectionRegistry>,
    /// Slice C of PRD #718 — local wait registry for `QUEUE READ … WAIT`.
    /// Producer commits notify; readers park here until wake-or-timeout
    /// or until `cancel_all` is invoked at shutdown.
    queue_wait_registry: std::sync::Arc<queue_wait_registry::QueueWaitRegistry>,
    /// Per-connection list of `(scope, queue)` pairs that should be
    /// notified through `queue_wait_registry` on COMMIT (slice C of
    /// PRD #718). Push paths buffer here while inside a txn; the
    /// COMMIT path drains and notifies. ROLLBACK discards — by design
    /// rolled-back enqueues do not deliver and do not wake waiters.
    pending_queue_wakes: parking_lot::RwLock<HashMap<u64, Vec<(String, String)>>>,
    /// Process-local normal-KV tag index used by `INVALIDATE TAGS`.
    kv_tag_index: KvTagIndex,
    /// Issue #524 — in-memory chain-tip cache per collection. Populated lazily
    /// by the first INSERT or `GET /chain-tip` call after restart and updated
    /// atomically with each chain INSERT. Backed by a single mutex so a chain
    /// INSERT serialises against concurrent submitters — the loser observes
    /// the advanced tip and surfaces `BlockchainConflict` to its caller.
    chain_tip_cache:
        parking_lot::Mutex<HashMap<String, crate::runtime::blockchain_kind::ChainTipFull>>,
    /// Issue #525 — in-memory mirror of the persisted `integrity` flag per
    /// chain collection.  `true` means INSERTs must be rejected with
    /// `ChainIntegrityBroken`.  Loaded lazily from `red_config` on first
    /// access so the flag survives restart.
    chain_integrity_broken: parking_lot::Mutex<HashMap<String, bool>>,
    /// Issue #765 / S6 — in-memory cache of input-stream integrity tombstone
    /// RID ranges. Loaded lazily from `red_config` on first read so the set
    /// survives restart (the durable list lives under
    /// `stream.integrity.tombstones`). `integrity_tombstones_state` is the
    /// hot-path gate: `0` = unloaded, `1` = loaded-empty (reads skip
    /// filtering after a single relaxed load), `2` = loaded-with-tombstones.
    integrity_tombstones:
        parking_lot::Mutex<Vec<crate::runtime::integrity_tombstone::TombstoneRange>>,
    integrity_tombstones_state: std::sync::atomic::AtomicU8,
}

#[derive(Clone)]
pub struct RedDBRuntime {
    // Field order matters: drain workers before dropping the foreground inner.
    // Temporary handles used by workers do not own their own shutdown guard.
    background_tasks: Option<Arc<background_tasks::RuntimeBackgroundTasks>>,
    inner: Arc<RuntimeInner>,
}

pub struct RuntimeConnection {
    id: u64,
    uses_pool_slot: bool,
    background_tasks: Option<Arc<background_tasks::RuntimeBackgroundTasks>>,
    inner: Arc<RuntimeInner>,
}

pub struct CausalSession {
    runtime: RedDBRuntime,
    bookmark: Option<crate::replication::CausalBookmark>,
    wait_timeout: std::time::Duration,
}

impl CausalSession {
    pub fn bookmark_token(&self) -> Option<String> {
        self.bookmark.map(|bookmark| bookmark.encode())
    }

    pub fn inject_bookmark(&mut self, token: &str) -> RedDBResult<()> {
        let bookmark = crate::replication::CausalBookmark::decode(token)
            .map_err(|err| RedDBError::InvalidOperation(err.to_string()))?;
        self.bookmark = Some(bookmark);
        Ok(())
    }

    pub fn execute_query(&mut self, query: &str) -> RedDBResult<RuntimeQueryResult> {
        if is_select_query_text(query) {
            if let Some(bookmark) = self.bookmark {
                self.runtime
                    .wait_for_bookmark(&bookmark, self.wait_timeout)?;
            }
        }
        let result = self.runtime.execute_query(query)?;
        if let Some(token) = result.bookmark.as_deref() {
            self.inject_bookmark(token)?;
        }
        Ok(result)
    }
}

fn is_select_query_text(query: &str) -> bool {
    query
        .trim_start()
        .get(..6)
        .is_some_and(|prefix| prefix.eq_ignore_ascii_case("select"))
}

pub mod ai;
mod allocator_pressure;
pub mod analytics_schema_registry;
pub(crate) mod analytics_source_catalog;
pub mod ask_pipeline;
pub mod audit_log;
pub mod audit_query;
pub mod authorized_search;
pub(crate) mod authz;
mod background_tasks;
pub mod batch_insert;
pub mod blockchain_kind;
mod collection_contract;
pub mod config_matrix;
pub mod config_overlay;
pub mod config_watcher;
pub mod continuous_materialized_view;
pub mod control_events;
pub(crate) mod ddl;
pub mod disk_space_monitor;
mod dml_target_scan;
pub mod evidence_export;
pub(crate) mod execution_context;
mod expr_eval;
mod graph_dsl;
mod graph_tvf;
mod health_connection;
mod impl_audit_control;
mod impl_backup;
mod impl_catalog_accessors;
mod impl_cdc;
mod impl_config;
mod impl_config_secret;
pub(crate) mod impl_core;
mod impl_core_outer_scope;
mod impl_core_result_cache_scopes;
mod impl_ddl;
mod impl_dml;
mod impl_dml_batch;
mod impl_dml_chain;
mod impl_dml_claim;
mod impl_dml_crypto;
mod impl_dml_returning;
mod impl_dml_support;
mod impl_dml_tenant;
mod impl_dml_ttl;
mod impl_dml_update_analysis;
mod impl_dml_update_target;
mod impl_ec;
pub mod impl_ephemeral;
mod impl_events;
mod impl_fast_path;
mod impl_graph;
mod impl_graph_commands;
pub mod impl_kv;
mod impl_lifecycle;
mod impl_materialized_view;
mod impl_migrations;
mod impl_native;
mod impl_physical;
mod impl_primary_replica_file;
mod impl_probabilistic;
pub mod impl_queue;
mod impl_ranking;
mod impl_replica_loop;
mod impl_replication_commit;
mod impl_runtime_index_registry;
mod impl_telemetry_accessors;
mod impl_tenant_registry;
pub(crate) use impl_queue::RedwireWaitOutcome;
pub(crate) mod claim_telemetry;
#[cfg(test)]
mod evaluator_differential;
mod function_budget;
pub(crate) mod function_catalog;
mod function_execution;
mod function_validation;
mod impl_scrub;
mod impl_search;
mod impl_serverless;
mod impl_timeseries;
mod impl_tree;
mod impl_vcs;
mod index_store;
#[cfg(test)]
pub(crate) use index_store::MutationTestPhase;
pub mod integrity_tombstone;
mod join_filter;
mod keyed_spine;
pub(crate) mod kv_read;
pub mod kv_watch;
pub(crate) mod kv_write;
pub mod lease_lifecycle;
pub mod lease_loop;
pub mod lease_timer_wheel;
pub mod lifecycle;
pub mod lock_manager;
pub mod locking;
pub(crate) mod materialization_limit;
/// Samples every governed pool's live usage into the shared accounting
/// (ADR 0073 §2).
pub mod memory_accounting;
pub(crate) mod memory_admission;
pub(crate) mod metric_descriptor_catalog;
pub(crate) mod mutation;
#[cfg(test)]
mod mutation_index_topology_tests;
#[cfg(test)]
mod mutation_memory_tests;
pub(crate) mod mvcc_lifecycle;
pub(crate) mod mvcc_maintenance;
pub(crate) mod node_load_telemetry;
pub(crate) mod occupancy_sampler;
pub(crate) mod primary_queue_store;
mod probabilistic_store;
pub mod query_audit;
pub(crate) mod query_budget;
pub(crate) mod query_exec;
pub(crate) mod query_latency_telemetry;
pub mod query_request;
pub(crate) mod queue_lifecycle;
pub(crate) mod queue_telemetry;
pub(crate) mod queue_wait_registry;
pub mod quota_bucket;
pub(crate) mod ranking_descriptor_catalog;
mod record_search;
mod red_schema;
pub(crate) mod replica_queue_store;
pub mod resource_limits;
mod result_cache_runtime;
pub(crate) mod retention_filter;
pub(crate) mod retention_sweeper;
mod rls_injection;
pub mod schema_diff;
pub mod schema_vocabulary;
pub(crate) mod score_sketch;
mod search_graph;
pub(crate) mod sessionize;
pub mod signed_chain;
pub mod signed_writes_kind;
pub(crate) mod slo_descriptor_catalog;
pub mod snapshot_reuse;
mod statement_frame;
mod table_row_mvcc_resolver;
pub(crate) mod transaction_state;
pub mod turbo_crash_inject;
mod unique_index_admission;
mod vcs_command;
mod vector_index;
pub mod vector_turbo_kind;
pub(crate) mod window_phase;
pub mod within_clause;
pub mod write_gate;

pub use self::claim_telemetry::ClaimTelemetrySnapshot;
pub use self::graph_dsl::*;
use self::join_filter::*;
use self::query_exec::*;
use self::record_search::*;
pub use self::statement_frame::EffectiveScope;

/// Re-exports for transports + tests that need per-connection
/// isolation, tenant / auth thread-locals, and MVCC snapshot
/// utilities. Mirrors what PG-wire / gRPC / HTTP middleware already
/// call, and is enough to emulate independent connections in
/// integration tests.
pub mod mvcc {
    pub use super::impl_core::{
        capture_current_snapshot, clear_current_auth_identity, clear_current_connection_id,
        clear_current_snapshot, clear_current_tenant, current_connection_id, current_tenant,
        entity_visible_under_current_snapshot, entity_visible_with_context,
        set_current_auth_identity, set_current_connection_id, set_current_snapshot,
        set_current_tenant, snapshot_bundle, with_snapshot_bundle, SnapshotBundle, SnapshotContext,
    };
}

/// Public helpers re-exported for use by the presentation layer.
pub mod record_search_helpers {
    use crate::storage::query::UnifiedRecord;
    use crate::storage::UnifiedEntity;
    use std::collections::BTreeSet;

    pub fn entity_type_and_capabilities(
        entity: &UnifiedEntity,
    ) -> (&'static str, BTreeSet<String>) {
        super::record_search::runtime_entity_type_and_capabilities(entity)
    }

    /// Materialise any entity kind (TableRow, Node, Edge, Vector,
    /// TimeSeriesPoint, QueueMessage) into a `UnifiedRecord` whose
    /// `values` carry the native fields. Used by the RLS evaluator
    /// when a non-table collection matches a `CompareExpr` policy.
    pub fn any_record_from_entity(entity: UnifiedEntity) -> Option<UnifiedRecord> {
        super::record_search::runtime_any_record_from_entity(entity)
    }
}