1use std::collections::{BTreeMap, BTreeSet};
10use std::fmt;
11use std::io;
12use std::path::{Path, PathBuf};
13use std::sync::Arc;
14use std::time::{SystemTime, UNIX_EPOCH};
15
16use crate::auth::AuthConfig;
17use crate::replication::ReplicationConfig;
18
19pub const DEFAULT_SNAPSHOT_RETENTION: usize = 16;
20pub const DEFAULT_EXPORT_RETENTION: usize = 16;
21
22pub const REDDB_PROTOCOL_VERSION: &str = "reddb-v2";
23pub const REDDB_FORMAT_VERSION: u32 = 2;
24pub const DEFAULT_GROUP_COMMIT_WINDOW_MS: u64 = 0;
37pub const DEFAULT_GROUP_COMMIT_MAX_STATEMENTS: usize = 128;
38pub const DEFAULT_GROUP_COMMIT_MAX_WAL_BYTES: u64 = 1024 * 1024;
39
40pub type RedDBResult<T> = Result<T, RedDBError>;
41
42#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Default)]
43pub enum StorageMode {
44 #[default]
46 Persistent,
47}
48
49impl StorageMode {
50 pub const fn is_persistent(self) -> bool {
51 matches!(self, Self::Persistent)
52 }
53}
54
55#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Default)]
56pub enum DurabilityMode {
57 #[default]
58 Strict,
59 WalDurableGrouped,
60 Async,
68}
69
70impl DurabilityMode {
71 pub const fn as_str(self) -> &'static str {
72 match self {
73 Self::Strict => "strict",
74 Self::WalDurableGrouped => "wal_durable_grouped",
75 Self::Async => "async",
76 }
77 }
78
79 pub fn from_str(value: &str) -> Option<Self> {
80 let normalized = value.trim().to_ascii_lowercase();
81 match normalized.as_str() {
82 "strict" => Some(Self::Strict),
84 "sync"
90 | "wal_durable_grouped"
91 | "wal-durable-grouped"
92 | "grouped"
93 | "wal_grouped"
94 | "wal-grouped" => Some(Self::WalDurableGrouped),
95 "async" | "fire_and_forget" | "fire-and-forget" => Some(Self::Async),
99 _ => None,
100 }
101 }
102}
103
104#[derive(Debug, Clone, Copy, PartialEq, Eq)]
105pub struct GroupCommitOptions {
106 pub window_ms: u64,
107 pub max_statements: usize,
108 pub max_wal_bytes: u64,
109}
110
111impl Default for GroupCommitOptions {
112 fn default() -> Self {
113 Self {
114 window_ms: DEFAULT_GROUP_COMMIT_WINDOW_MS,
115 max_statements: DEFAULT_GROUP_COMMIT_MAX_STATEMENTS,
116 max_wal_bytes: DEFAULT_GROUP_COMMIT_MAX_WAL_BYTES,
117 }
118 }
119}
120
121#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
122pub enum Capability {
123 Table,
125 Graph,
127 Vector,
129 FullText,
131 Security,
133 Encryption,
135}
136
137impl Capability {
138 pub const fn as_str(self) -> &'static str {
139 match self {
140 Self::Table => "table",
141 Self::Graph => "graph",
142 Self::Vector => "vector",
143 Self::FullText => "fulltext",
144 Self::Security => "security",
145 Self::Encryption => "encryption",
146 }
147 }
148}
149
150#[derive(Debug, Clone, Default)]
151pub struct CapabilitySet {
152 items: BTreeSet<Capability>,
153}
154
155impl CapabilitySet {
156 pub fn new() -> Self {
157 Self::default()
158 }
159
160 pub fn with(mut self, capability: Capability) -> Self {
161 self.items.insert(capability);
162 self
163 }
164
165 pub fn with_all(mut self, capabilities: &[Capability]) -> Self {
166 capabilities.iter().copied().for_each(|capability| {
167 self.items.insert(capability);
168 });
169 self
170 }
171
172 pub fn has(&self, capability: Capability) -> bool {
173 self.items.contains(&capability)
174 }
175
176 pub fn as_slice(&self) -> Vec<Capability> {
177 self.items.iter().copied().collect()
178 }
179}
180
181pub struct RedDBOptions {
182 pub mode: StorageMode,
183 pub data_path: Option<PathBuf>,
184 pub read_only: bool,
185 pub create_if_missing: bool,
186 pub verify_checksums: bool,
187 pub durability_mode: DurabilityMode,
188 pub group_commit: GroupCommitOptions,
189 pub auto_checkpoint_pages: u32,
190 pub cache_pages: usize,
191 pub snapshot_retention: usize,
192 pub export_retention: usize,
193 pub feature_gates: CapabilitySet,
194 pub force_create: bool,
195 pub metadata: BTreeMap<String, String>,
196 pub remote_backend: Option<Arc<dyn crate::storage::backend::RemoteBackend>>,
198 pub remote_backend_atomic: Option<Arc<dyn crate::storage::backend::AtomicRemoteBackend>>,
205 pub remote_key: Option<String>,
207 pub replication: ReplicationConfig,
209 pub auth: AuthConfig,
211 pub auto_index_id: bool,
217 pub layout: crate::storage::layout::StorageLayout,
223 pub layout_overrides: crate::storage::layout::LayoutOverrides,
225 pub layout_explicit: bool,
231}
232
233impl fmt::Debug for RedDBOptions {
234 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
235 let backend_name = self.remote_backend.as_ref().map(|b| b.name().to_string());
236 f.debug_struct("RedDBOptions")
237 .field("mode", &self.mode)
238 .field("data_path", &self.data_path)
239 .field("read_only", &self.read_only)
240 .field("create_if_missing", &self.create_if_missing)
241 .field("verify_checksums", &self.verify_checksums)
242 .field("durability_mode", &self.durability_mode)
243 .field("group_commit", &self.group_commit)
244 .field("auto_checkpoint_pages", &self.auto_checkpoint_pages)
245 .field("cache_pages", &self.cache_pages)
246 .field("snapshot_retention", &self.snapshot_retention)
247 .field("export_retention", &self.export_retention)
248 .field("feature_gates", &self.feature_gates)
249 .field("force_create", &self.force_create)
250 .field("metadata", &self.metadata)
251 .field("remote_backend", &backend_name)
252 .field("remote_key", &self.remote_key)
253 .field("replication", &self.replication)
254 .field("auth", &self.auth)
255 .field("layout", &self.layout)
256 .field("layout_overrides", &self.layout_overrides)
257 .finish()
258 }
259}
260
261impl Clone for RedDBOptions {
262 fn clone(&self) -> Self {
263 Self {
264 mode: self.mode,
265 data_path: self.data_path.clone(),
266 read_only: self.read_only,
267 create_if_missing: self.create_if_missing,
268 verify_checksums: self.verify_checksums,
269 durability_mode: self.durability_mode,
270 group_commit: self.group_commit,
271 auto_checkpoint_pages: self.auto_checkpoint_pages,
272 cache_pages: self.cache_pages,
273 snapshot_retention: self.snapshot_retention,
274 export_retention: self.export_retention,
275 feature_gates: self.feature_gates.clone(),
276 force_create: self.force_create,
277 metadata: self.metadata.clone(),
278 remote_backend: self.remote_backend.clone(),
279 remote_backend_atomic: self.remote_backend_atomic.clone(),
280 remote_key: self.remote_key.clone(),
281 replication: self.replication.clone(),
282 auth: self.auth.clone(),
283 auto_index_id: self.auto_index_id,
284 layout: self.layout,
285 layout_overrides: self.layout_overrides.clone(),
286 layout_explicit: self.layout_explicit,
287 }
288 }
289}
290
291impl Default for RedDBOptions {
292 fn default() -> Self {
293 Self {
294 mode: StorageMode::Persistent,
295 data_path: None,
296 read_only: false,
297 create_if_missing: true,
298 verify_checksums: true,
299 durability_mode: DurabilityMode::WalDurableGrouped,
305 group_commit: GroupCommitOptions::default(),
306 auto_checkpoint_pages: 1000,
307 cache_pages: 10_000,
308 snapshot_retention: DEFAULT_SNAPSHOT_RETENTION,
309 export_retention: DEFAULT_EXPORT_RETENTION,
310 feature_gates: CapabilitySet::new()
311 .with(Capability::Table)
312 .with(Capability::Graph)
313 .with(Capability::Vector),
314 force_create: true,
315 metadata: BTreeMap::new(),
316 remote_backend: None,
317 remote_backend_atomic: None,
318 remote_key: None,
319 replication: ReplicationConfig::standalone(),
320 auth: AuthConfig::default(),
321 auto_index_id: true,
322 layout: crate::storage::layout::StorageLayout::default(),
323 layout_overrides: crate::storage::layout::LayoutOverrides::default(),
324 layout_explicit: false,
325 }
326 }
327}
328
329impl RedDBOptions {
330 pub fn persistent<P: Into<PathBuf>>(path: P) -> Self {
331 Self {
332 mode: StorageMode::Persistent,
333 data_path: Some(path.into()),
334 ..Default::default()
335 }
336 }
337
338 pub fn in_memory() -> Self {
345 static NEXT_EPHEMERAL_ID: std::sync::atomic::AtomicU64 =
346 std::sync::atomic::AtomicU64::new(0);
347
348 let now_nanos = std::time::SystemTime::now()
349 .duration_since(std::time::UNIX_EPOCH)
350 .map(|duration| duration.as_nanos())
351 .unwrap_or(0);
352 let unique = NEXT_EPHEMERAL_ID.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
353 let path = std::env::temp_dir().join(format!(
354 "reddb-ephemeral-{}-{}-{}.rdb",
355 std::process::id(),
356 now_nanos,
357 unique
358 ));
359 let _ = std::fs::remove_file(&path);
360 Self {
361 mode: StorageMode::Persistent,
362 data_path: Some(path),
363 auto_checkpoint_pages: 0,
364 cache_pages: 2_000,
365 snapshot_retention: DEFAULT_SNAPSHOT_RETENTION,
366 export_retention: DEFAULT_EXPORT_RETENTION,
367 read_only: false,
368 force_create: true,
369 ..Default::default()
370 }
371 }
372
373 pub fn with_mode(mut self, mode: StorageMode) -> Self {
374 self.mode = mode;
375 self
376 }
377
378 pub fn with_data_path<P: Into<PathBuf>>(mut self, path: P) -> Self {
379 self.data_path = Some(path.into());
380 self
381 }
382
383 pub fn with_read_only(mut self, read_only: bool) -> Self {
384 self.read_only = read_only;
385 self
386 }
387
388 pub fn with_auto_checkpoint(mut self, pages: u32) -> Self {
389 self.auto_checkpoint_pages = pages;
390 self
391 }
392
393 pub fn with_durability_mode(mut self, mode: DurabilityMode) -> Self {
394 self.durability_mode = mode;
395 self
396 }
397
398 pub fn with_group_commit_window_ms(mut self, window_ms: u64) -> Self {
399 self.group_commit.window_ms = window_ms;
402 self
403 }
404
405 pub fn with_group_commit_max_statements(mut self, max_statements: usize) -> Self {
406 self.group_commit.max_statements = max_statements.max(1);
407 self
408 }
409
410 pub fn with_group_commit_max_wal_bytes(mut self, max_wal_bytes: u64) -> Self {
411 self.group_commit.max_wal_bytes = max_wal_bytes.max(1);
412 self
413 }
414
415 pub fn with_cache_pages(mut self, pages: usize) -> Self {
416 self.cache_pages = pages.max(2);
417 self
418 }
419
420 pub fn with_snapshot_retention(mut self, limit: usize) -> Self {
421 self.snapshot_retention = limit.max(1);
422 self
423 }
424
425 pub fn with_export_retention(mut self, limit: usize) -> Self {
426 self.export_retention = limit.max(1);
427 self
428 }
429
430 pub fn with_metadata<K: Into<String>, V: Into<String>>(mut self, key: K, value: V) -> Self {
431 self.metadata.insert(key.into(), value.into());
432 self
433 }
434
435 pub fn with_auto_index_id(mut self, enabled: bool) -> Self {
439 self.auto_index_id = enabled;
440 self
441 }
442
443 pub fn with_capability(mut self, capability: Capability) -> Self {
444 self.feature_gates = self.feature_gates.with(capability);
445 self
446 }
447
448 pub fn with_remote_backend(
454 mut self,
455 backend: Arc<dyn crate::storage::backend::RemoteBackend>,
456 key: impl Into<String>,
457 ) -> Self {
458 self.remote_backend = Some(backend);
459 self.remote_key = Some(key.into());
460 self
461 }
462
463 pub fn with_atomic_remote_backend(
469 mut self,
470 backend: Arc<dyn crate::storage::backend::AtomicRemoteBackend>,
471 ) -> Self {
472 self.remote_backend_atomic = Some(backend);
473 self
474 }
475
476 pub fn with_replication(mut self, config: ReplicationConfig) -> Self {
477 self.replication = config;
478 self
479 }
480
481 pub fn with_auth(mut self, config: AuthConfig) -> Self {
482 self.auth = config;
483 self
484 }
485
486 pub fn resolved_path(&self, fallback: impl AsRef<Path>) -> PathBuf {
487 self.data_path
488 .clone()
489 .unwrap_or_else(|| fallback.as_ref().to_path_buf())
490 }
491
492 pub fn remote_namespace_prefix(&self) -> String {
493 let Some(remote_key) = &self.remote_key else {
494 return String::new();
495 };
496 let normalized = remote_key.trim_matches('/');
497 if normalized.is_empty() {
498 return String::new();
499 }
500 match normalized.rsplit_once('/') {
501 Some((parent, _)) if !parent.is_empty() => format!("{parent}/"),
502 _ => String::new(),
503 }
504 }
505
506 pub fn default_backup_head_key(&self) -> String {
507 if let Some(value) = self.metadata.get("red.config.backup.head_key") {
508 return value.clone();
509 }
510 format!("{}manifests/head.json", self.remote_namespace_prefix())
511 }
512
513 pub fn default_snapshot_prefix(&self) -> String {
514 if let Some(value) = self.metadata.get("red.config.backup.snapshot_prefix") {
515 return value.clone();
516 }
517 format!("{}snapshots/", self.remote_namespace_prefix())
518 }
519
520 pub fn default_wal_archive_prefix(&self) -> String {
521 if let Some(value) = self.metadata.get("red.config.wal.archive.prefix") {
522 return value.clone();
523 }
524 format!("{}wal/", self.remote_namespace_prefix())
525 }
526
527 pub fn has_capability(&self, capability: Capability) -> bool {
528 self.feature_gates.has(capability)
529 }
530
531 pub fn with_layout(mut self, layout: crate::storage::layout::StorageLayout) -> Self {
534 self.layout = layout;
535 self.layout_explicit = true;
536 self
537 }
538
539 pub fn with_layout_overrides(
542 mut self,
543 overrides: crate::storage::layout::LayoutOverrides,
544 ) -> Self {
545 self.layout_overrides = overrides;
546 self
547 }
548
549 pub fn resolve_tiered_layout(
553 &self,
554 ) -> Option<(PathBuf, crate::storage::layout::TieredLayoutPaths)> {
555 let data_path = self.data_path.clone()?;
556 let paths = crate::storage::layout::TieredLayoutPaths::new(
557 &data_path,
558 self.layout,
559 self.layout_overrides.clone(),
560 );
561 Some((data_path, paths))
562 }
563
564 pub fn apply_tier_defaults(&self) {
585 use crate::storage::layout::StorageLayout;
586
587 if !self.layout_explicit {
592 if let Some((_, paths)) = self.resolve_tiered_layout() {
593 tier_wiring::stash_layout_paths(paths);
594 }
595 return;
596 }
597
598 let layout = self.layout;
599 crate::physical::set_meta_json_sidecar_enabled(matches!(layout, StorageLayout::Max));
601
602 crate::physical::set_seqn_journal_enabled(matches!(layout, StorageLayout::Max));
606 crate::physical::set_seqn_journal_retention(match layout {
607 StorageLayout::Max => crate::physical::DEFAULT_METADATA_JOURNAL_RETENTION,
608 _ => crate::physical::OPT_IN_METADATA_JOURNAL_RETENTION,
609 });
610
611 crate::physical::set_shm_provisioning_enabled(matches!(
613 layout,
614 StorageLayout::Standard | StorageLayout::Performance | StorageLayout::Max
615 ));
616
617 crate::physical::set_fold_pager_meta_enabled(matches!(layout, StorageLayout::Max));
620 crate::physical::set_fold_dwb_into_wal_enabled(matches!(layout, StorageLayout::Max));
621
622 if let Some((_, paths)) = self.resolve_tiered_layout() {
624 tier_wiring::stash_layout_paths(paths);
625 }
626 }
627}
628
629pub mod tier_wiring {
633 use std::sync::Mutex;
634
635 use crate::storage::layout::{LogDestination, TieredLayoutPaths};
636
637 static CURRENT_LAYOUT_PATHS: Mutex<Option<TieredLayoutPaths>> = Mutex::new(None);
638
639 pub fn stash_layout_paths(paths: TieredLayoutPaths) {
640 if let Ok(mut slot) = CURRENT_LAYOUT_PATHS.lock() {
641 *slot = Some(paths);
642 }
643 }
644
645 pub fn current_layout_paths() -> Option<TieredLayoutPaths> {
646 CURRENT_LAYOUT_PATHS.lock().ok().and_then(|slot| slot.clone())
647 }
648
649 pub fn current_log_destinations() -> (LogDestination, LogDestination) {
653 match current_layout_paths() {
654 Some(p) => (p.audit_log_destination, p.slow_log_destination),
655 None => (LogDestination::Stderr, LogDestination::Stderr),
656 }
657 }
658}
659
660#[derive(Debug, Clone, Default)]
661pub struct CollectionStats {
662 pub entities: usize,
663 pub cross_refs: usize,
664 pub segments: usize,
665}
666
667#[derive(Debug, Clone)]
668pub struct CatalogSnapshot {
669 pub name: String,
670 pub total_entities: usize,
671 pub total_collections: usize,
672 pub stats_by_collection: BTreeMap<String, CollectionStats>,
673 pub updated_at: SystemTime,
674}
675
676impl Default for CatalogSnapshot {
677 fn default() -> Self {
678 Self {
679 name: String::new(),
680 total_entities: 0,
681 total_collections: 0,
682 stats_by_collection: BTreeMap::new(),
683 updated_at: UNIX_EPOCH,
684 }
685 }
686}
687
688#[derive(Debug, Clone)]
689pub struct SchemaManifest {
690 pub format_version: u32,
691 pub created_at_unix_ms: u128,
692 pub updated_at_unix_ms: u128,
693 pub options: RedDBOptions,
694 pub collection_count: usize,
695}
696
697impl SchemaManifest {
698 pub fn now(options: RedDBOptions, collection_count: usize) -> Self {
699 let now = SystemTime::now()
700 .duration_since(UNIX_EPOCH)
701 .unwrap_or_default()
702 .as_millis();
703 Self {
704 format_version: REDDB_FORMAT_VERSION,
705 created_at_unix_ms: now,
706 updated_at_unix_ms: now,
707 options,
708 collection_count,
709 }
710 }
711}
712
713#[derive(Debug)]
714pub enum RedDBError {
715 InvalidConfig(String),
716 SchemaVersionMismatch {
717 expected: u32,
718 found: u32,
719 },
720 FeatureNotEnabled(String),
721 NotFound(String),
722 ReadOnly(String),
723 InvalidOperation(String),
724 Engine(String),
725 Catalog(String),
726 Query(String),
727 Validation {
728 message: String,
729 validation: crate::json::Value,
730 },
731 Io(io::Error),
732 VersionUnavailable,
733 QuotaExceeded(String),
739 Internal(String),
740}
741
742impl fmt::Display for RedDBError {
743 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
744 match self {
745 Self::InvalidConfig(msg) => write!(f, "invalid config: {msg}"),
746 Self::SchemaVersionMismatch { expected, found } => {
747 write!(
748 f,
749 "schema version mismatch: expected {expected}, found {found}"
750 )
751 }
752 Self::FeatureNotEnabled(msg) => write!(f, "feature disabled: {msg}"),
753 Self::NotFound(msg) => write!(f, "not found: {msg}"),
754 Self::ReadOnly(msg) => write!(f, "read-only violation: {msg}"),
755 Self::InvalidOperation(msg) => write!(f, "INVALID_OPERATION: {msg}"),
756 Self::Engine(msg) => write!(f, "engine error: {msg}"),
757 Self::Catalog(msg) => write!(f, "catalog error: {msg}"),
758 Self::Query(msg) => write!(f, "query error: {msg}"),
759 Self::Validation { message, .. } => write!(f, "validation error: {message}"),
760 Self::Io(err) => write!(f, "io error: {err}"),
761 Self::VersionUnavailable => write!(f, "version information unavailable"),
762 Self::QuotaExceeded(msg) => write!(f, "quota exceeded: {msg}"),
763 Self::Internal(msg) => write!(f, "internal error: {msg}"),
764 }
765 }
766}
767
768impl std::error::Error for RedDBError {}
769
770impl From<io::Error> for RedDBError {
771 fn from(err: io::Error) -> Self {
772 Self::Io(err)
773 }
774}
775
776impl From<crate::storage::engine::DatabaseError> for RedDBError {
777 fn from(err: crate::storage::engine::DatabaseError) -> Self {
778 Self::Engine(err.to_string())
779 }
780}
781
782impl From<crate::storage::wal::TxError> for RedDBError {
783 fn from(err: crate::storage::wal::TxError) -> Self {
784 Self::Engine(err.to_string())
785 }
786}
787
788impl From<crate::storage::StoreError> for RedDBError {
789 fn from(err: crate::storage::StoreError) -> Self {
790 Self::Catalog(err.to_string())
791 }
792}
793
794impl From<crate::storage::unified::devx::DevXError> for RedDBError {
795 fn from(err: crate::storage::unified::devx::DevXError) -> Self {
796 match err {
797 crate::storage::unified::devx::DevXError::Validation(msg) => Self::InvalidConfig(msg),
798 crate::storage::unified::devx::DevXError::Storage(msg) => Self::Engine(msg),
799 crate::storage::unified::devx::DevXError::NotFound(msg) => Self::NotFound(msg),
800 }
801 }
802}
803
804pub trait CatalogService {
805 fn list_collections(&self) -> Vec<String>;
806 fn collection_stats(&self, collection: &str) -> Option<CollectionStats>;
807 fn catalog_snapshot(&self) -> CatalogSnapshot;
808}
809
810pub trait QueryPlanner {
811 fn plan_cost(&self, query: &str) -> Option<f64>;
812}
813
814pub trait DataOps {
815 fn execute_query(&self, query: &str) -> RedDBResult<()>;
816}
817
818pub mod prelude {
819 pub use super::{
820 Capability, CapabilitySet, CatalogService, CatalogSnapshot, CollectionStats, DataOps,
821 QueryPlanner, RedDBError, RedDBOptions, RedDBResult, SchemaManifest, StorageMode,
822 REDDB_FORMAT_VERSION, REDDB_PROTOCOL_VERSION,
823 };
824}