Skip to main content

cloudillo_types/
rtdb_adapter.rs

1// SPDX-FileCopyrightText: Szilárd Hajba
2// SPDX-License-Identifier: LGPL-3.0-or-later
3
4//! Real-Time Database Adapter
5//!
6//! Trait and types for pluggable real-time database backends that store JSON documents
7//! using hierarchical path-based access (e.g., `posts/abc123/comments/xyz789`).
8//!
9//! Read operations (query, get, subscribe) work directly on the adapter.
10//! Write operations (create, update, delete) require a transaction for atomicity.
11//!
12//! Each adapter implementation provides its own constructor handling backend-specific
13//! initialization (database path, connection settings, etc.).
14
15use async_trait::async_trait;
16use futures_core::Stream;
17use serde::{Deserialize, Serialize};
18use serde_json::Value;
19use std::collections::HashMap;
20use std::fmt::Debug;
21use std::pin::Pin;
22
23use crate::prelude::*;
24use crate::types::CompactReport;
25
26/// Lock mode for document locking.
27#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
28#[serde(rename_all = "camelCase")]
29pub enum LockMode {
30	Soft,
31	Hard,
32}
33
34/// Information about an active lock on a document path.
35#[derive(Debug, Clone)]
36pub struct LockInfo {
37	pub user_id: Box<str>,
38	pub mode: LockMode,
39	pub acquired_at: u64,
40	pub ttl_secs: u64,
41}
42
43/// An aggregation operation to compute per group.
44#[derive(Debug, Clone, Serialize, Deserialize)]
45#[serde(tag = "op", rename_all = "camelCase")]
46pub enum AggregateOp {
47	Sum { field: String },
48	Avg { field: String },
49	Min { field: String },
50	Max { field: String },
51}
52
53/// Aggregation options: group by a field and compute statistics.
54#[derive(Debug, Clone, Serialize, Deserialize)]
55#[serde(rename_all = "camelCase")]
56pub struct AggregateOptions {
57	/// Field to group by. For array fields, each element becomes a separate group.
58	pub group_by: String,
59
60	/// Additional operations per group (count is always included implicitly).
61	#[serde(default, skip_serializing_if = "Vec::is_empty")]
62	pub ops: Vec<AggregateOp>,
63}
64
65/// Query filter for selecting documents.
66///
67/// Supports multiple filter operations on JSON document fields.
68/// A document matches if ALL specified conditions are satisfied (AND logic).
69#[derive(Debug, Clone, Default, Serialize, Deserialize)]
70pub struct QueryFilter {
71	/// Field equality constraints: field_name -> expected_value
72	#[serde(default, skip_serializing_if = "HashMap::is_empty")]
73	pub equals: HashMap<String, Value>,
74
75	/// Field not-equal constraints: field_name -> expected_value
76	#[serde(default, skip_serializing_if = "HashMap::is_empty", rename = "notEquals")]
77	pub not_equals: HashMap<String, Value>,
78
79	/// Field greater-than constraints: field_name -> threshold_value
80	#[serde(default, skip_serializing_if = "HashMap::is_empty", rename = "greaterThan")]
81	pub greater_than: HashMap<String, Value>,
82
83	/// Field greater-than-or-equal constraints: field_name -> threshold_value
84	#[serde(default, skip_serializing_if = "HashMap::is_empty", rename = "greaterThanOrEqual")]
85	pub greater_than_or_equal: HashMap<String, Value>,
86
87	/// Field less-than constraints: field_name -> threshold_value
88	#[serde(default, skip_serializing_if = "HashMap::is_empty", rename = "lessThan")]
89	pub less_than: HashMap<String, Value>,
90
91	/// Field less-than-or-equal constraints: field_name -> threshold_value
92	#[serde(default, skip_serializing_if = "HashMap::is_empty", rename = "lessThanOrEqual")]
93	pub less_than_or_equal: HashMap<String, Value>,
94
95	/// Field in-array constraints: field_name -> array of allowed values
96	#[serde(default, skip_serializing_if = "HashMap::is_empty", rename = "inArray")]
97	pub in_array: HashMap<String, Vec<Value>>,
98
99	/// Array-contains constraints: field_name -> value that must be in the array field
100	#[serde(default, skip_serializing_if = "HashMap::is_empty", rename = "arrayContains")]
101	pub array_contains: HashMap<String, Value>,
102
103	/// Not-in-array constraints: field_name -> array of excluded values
104	#[serde(default, skip_serializing_if = "HashMap::is_empty", rename = "notInArray")]
105	pub not_in_array: HashMap<String, Vec<Value>>,
106
107	/// Array-contains-any constraints: field_name -> array of values (at least one must be in the array field)
108	#[serde(default, skip_serializing_if = "HashMap::is_empty", rename = "arrayContainsAny")]
109	pub array_contains_any: HashMap<String, Vec<Value>>,
110
111	/// Array-contains-all constraints: field_name -> array of values (all must be in the array field)
112	#[serde(default, skip_serializing_if = "HashMap::is_empty", rename = "arrayContainsAll")]
113	pub array_contains_all: HashMap<String, Vec<Value>>,
114}
115
116impl QueryFilter {
117	/// Create a new empty filter (matches all documents).
118	pub fn new() -> Self {
119		Self::default()
120	}
121
122	/// Create a filter with a single equality constraint.
123	pub fn equals_one(field: impl Into<String>, value: Value) -> Self {
124		let mut equals = HashMap::new();
125		equals.insert(field.into(), value);
126		Self { equals, ..Default::default() }
127	}
128
129	/// Add an equality constraint to this filter (builder pattern).
130	pub fn with_equals(mut self, field: impl Into<String>, value: Value) -> Self {
131		self.equals.insert(field.into(), value);
132		self
133	}
134
135	/// Add a not-equal constraint to this filter (builder pattern).
136	pub fn with_not_equals(mut self, field: impl Into<String>, value: Value) -> Self {
137		self.not_equals.insert(field.into(), value);
138		self
139	}
140
141	/// Add a greater-than constraint to this filter (builder pattern).
142	pub fn with_greater_than(mut self, field: impl Into<String>, value: Value) -> Self {
143		self.greater_than.insert(field.into(), value);
144		self
145	}
146
147	/// Add a greater-than-or-equal constraint to this filter (builder pattern).
148	pub fn with_greater_than_or_equal(mut self, field: impl Into<String>, value: Value) -> Self {
149		self.greater_than_or_equal.insert(field.into(), value);
150		self
151	}
152
153	/// Add a less-than constraint to this filter (builder pattern).
154	pub fn with_less_than(mut self, field: impl Into<String>, value: Value) -> Self {
155		self.less_than.insert(field.into(), value);
156		self
157	}
158
159	/// Add a less-than-or-equal constraint to this filter (builder pattern).
160	pub fn with_less_than_or_equal(mut self, field: impl Into<String>, value: Value) -> Self {
161		self.less_than_or_equal.insert(field.into(), value);
162		self
163	}
164
165	/// Add an in-array constraint to this filter (builder pattern).
166	pub fn with_in_array(mut self, field: impl Into<String>, values: Vec<Value>) -> Self {
167		self.in_array.insert(field.into(), values);
168		self
169	}
170
171	/// Add an array-contains constraint to this filter (builder pattern).
172	pub fn with_array_contains(mut self, field: impl Into<String>, value: Value) -> Self {
173		self.array_contains.insert(field.into(), value);
174		self
175	}
176
177	/// Add a not-in-array constraint to this filter (builder pattern).
178	pub fn with_not_in_array(mut self, field: impl Into<String>, values: Vec<Value>) -> Self {
179		self.not_in_array.insert(field.into(), values);
180		self
181	}
182
183	/// Add an array-contains-any constraint to this filter (builder pattern).
184	pub fn with_array_contains_any(mut self, field: impl Into<String>, values: Vec<Value>) -> Self {
185		self.array_contains_any.insert(field.into(), values);
186		self
187	}
188
189	/// Add an array-contains-all constraint to this filter (builder pattern).
190	pub fn with_array_contains_all(mut self, field: impl Into<String>, values: Vec<Value>) -> Self {
191		self.array_contains_all.insert(field.into(), values);
192		self
193	}
194
195	/// Check if a document matches this filter (all conditions must be satisfied).
196	pub fn matches(&self, doc: &Value) -> bool {
197		// Equality checks
198		for (field, expected) in &self.equals {
199			if doc.get(field) != Some(expected) {
200				return false;
201			}
202		}
203
204		// Not-equal checks (missing fields are inherently "not equal")
205		for (field, expected) in &self.not_equals {
206			if doc.get(field) == Some(expected) {
207				return false;
208			}
209		}
210
211		// Greater-than checks
212		for (field, threshold) in &self.greater_than {
213			match doc.get(field) {
214				Some(actual)
215					if compare_json_values(Some(actual), Some(threshold))
216						== std::cmp::Ordering::Greater => {}
217				_ => return false,
218			}
219		}
220
221		// Greater-than-or-equal checks
222		for (field, threshold) in &self.greater_than_or_equal {
223			match doc.get(field) {
224				Some(actual) => {
225					let ord = compare_json_values(Some(actual), Some(threshold));
226					if ord != std::cmp::Ordering::Greater && ord != std::cmp::Ordering::Equal {
227						return false;
228					}
229				}
230				_ => return false,
231			}
232		}
233
234		// Less-than checks
235		for (field, threshold) in &self.less_than {
236			match doc.get(field) {
237				Some(actual)
238					if compare_json_values(Some(actual), Some(threshold))
239						== std::cmp::Ordering::Less => {}
240				_ => return false,
241			}
242		}
243
244		// Less-than-or-equal checks
245		for (field, threshold) in &self.less_than_or_equal {
246			match doc.get(field) {
247				Some(actual) => {
248					let ord = compare_json_values(Some(actual), Some(threshold));
249					if ord != std::cmp::Ordering::Less && ord != std::cmp::Ordering::Equal {
250						return false;
251					}
252				}
253				_ => return false,
254			}
255		}
256
257		// In-array checks (field value must be in the provided array)
258		for (field, allowed_values) in &self.in_array {
259			match doc.get(field) {
260				Some(actual) if allowed_values.contains(actual) => {}
261				_ => return false,
262			}
263		}
264
265		// Array-contains checks (field must be an array containing the value)
266		for (field, required_value) in &self.array_contains {
267			match doc.get(field) {
268				Some(Value::Array(arr)) if arr.contains(required_value) => {}
269				_ => return false,
270			}
271		}
272
273		// Not-in-array checks (field value must NOT be in the provided array; missing fields pass)
274		for (field, excluded_values) in &self.not_in_array {
275			if let Some(actual) = doc.get(field)
276				&& excluded_values.contains(actual)
277			{
278				return false;
279			}
280		}
281
282		// Array-contains-any checks
283		for (field, candidate_values) in &self.array_contains_any {
284			match doc.get(field) {
285				Some(Value::Array(arr)) if candidate_values.iter().any(|v| arr.contains(v)) => {}
286				_ => return false,
287			}
288		}
289
290		// Array-contains-all checks
291		for (field, required_values) in &self.array_contains_all {
292			match doc.get(field) {
293				Some(Value::Array(arr)) if required_values.iter().all(|v| arr.contains(v)) => {}
294				_ => return false,
295			}
296		}
297
298		true
299	}
300
301	/// Check if this filter is empty (matches all documents).
302	pub fn is_empty(&self) -> bool {
303		self.equals.is_empty()
304			&& self.not_equals.is_empty()
305			&& self.greater_than.is_empty()
306			&& self.greater_than_or_equal.is_empty()
307			&& self.less_than.is_empty()
308			&& self.less_than_or_equal.is_empty()
309			&& self.in_array.is_empty()
310			&& self.array_contains.is_empty()
311			&& self.not_in_array.is_empty()
312			&& self.array_contains_any.is_empty()
313			&& self.array_contains_all.is_empty()
314	}
315}
316
317/// Sort order for a field.
318#[derive(Debug, Clone, Serialize, Deserialize)]
319pub struct SortField {
320	/// Field name to sort by
321	pub field: String,
322
323	/// Sort direction: true for ascending, false for descending
324	pub ascending: bool,
325}
326
327impl SortField {
328	/// Create ascending sort order.
329	pub fn asc(field: impl Into<String>) -> Self {
330		Self { field: field.into(), ascending: true }
331	}
332
333	/// Create descending sort order.
334	pub fn desc(field: impl Into<String>) -> Self {
335		Self { field: field.into(), ascending: false }
336	}
337}
338
339/// Options for querying documents (filter, sort, limit, offset).
340#[derive(Debug, Clone, Default)]
341pub struct QueryOptions {
342	/// Optional filter to select documents
343	pub filter: Option<QueryFilter>,
344
345	/// Optional sort order (multiple fields supported)
346	pub sort: Option<Vec<SortField>>,
347
348	/// Optional limit on number of results
349	pub limit: Option<u32>,
350
351	/// Optional offset for pagination
352	pub offset: Option<u32>,
353
354	/// When set, returns aggregated groups instead of documents.
355	pub aggregate: Option<AggregateOptions>,
356
357	/// Optional field projection. When set, only these top-level fields (plus
358	/// `id`) are returned. `None` returns whole documents.
359	pub select: Option<Vec<String>>,
360}
361
362impl QueryOptions {
363	/// Create new empty query options (no filter, sort, or limit).
364	pub fn new() -> Self {
365		Self::default()
366	}
367
368	/// Set the filter.
369	pub fn with_filter(mut self, filter: QueryFilter) -> Self {
370		self.filter = Some(filter);
371		self
372	}
373
374	/// Set the sort order.
375	pub fn with_sort(mut self, sort: Vec<SortField>) -> Self {
376		self.sort = Some(sort);
377		self
378	}
379
380	/// Set the limit.
381	pub fn with_limit(mut self, limit: u32) -> Self {
382		self.limit = Some(limit);
383		self
384	}
385
386	/// Set the offset.
387	pub fn with_offset(mut self, offset: u32) -> Self {
388		self.offset = Some(offset);
389		self
390	}
391
392	/// Set the aggregation options.
393	pub fn with_aggregate(mut self, aggregate: AggregateOptions) -> Self {
394		self.aggregate = Some(aggregate);
395		self
396	}
397
398	/// Set the field projection.
399	pub fn with_select(mut self, select: Vec<String>) -> Self {
400		self.select = Some(select);
401		self
402	}
403}
404
405/// Restrict a document to the selected top-level fields.
406///
407/// `id` is always retained: it is injected by the adapter rather than stored, and
408/// every caller keys results by it, so a projection that dropped it would return
409/// documents nothing could address.
410///
411/// Only top-level fields are addressable, matching the same restriction sorting
412/// and filtering already carry. A non-object value is returned untouched.
413pub fn project_doc(doc: &Value, select: &[String]) -> Value {
414	let Some(obj) = doc.as_object() else { return doc.clone() };
415
416	let mut out = serde_json::Map::with_capacity(select.len() + 1);
417	if let Some(id) = obj.get("id") {
418		out.insert("id".to_string(), id.clone());
419	}
420	for field in select {
421		if let Some(value) = obj.get(field) {
422			out.insert(field.clone(), value.clone());
423		}
424	}
425
426	Value::Object(out)
427}
428
429/// True when any selected field differs between two versions of a document.
430///
431/// Drives event suppression on projected subscriptions: a subscriber that asked
432/// for four fields has no way to observe a write that touched none of them, so
433/// waking it costs a full client-side rebuild for nothing.
434pub fn selection_changed(old: Option<&Value>, new: &Value, select: &[String]) -> bool {
435	let Some(old) = old else { return true };
436	select.iter().any(|field| old.get(field) != new.get(field))
437}
438
439/// Options for subscribing to real-time changes.
440#[derive(Debug, Clone)]
441pub struct SubscriptionOptions {
442	/// Path to subscribe to (e.g., "posts", "posts/abc123/comments")
443	pub path: Box<str>,
444
445	/// Optional filter (only matching changes are sent)
446	pub filter: Option<QueryFilter>,
447
448	/// Optional field projection, applied to `Create`/`Update` payloads. When
449	/// set, an event touching none of these fields is not delivered at all.
450	pub select: Option<Vec<String>>,
451}
452
453impl SubscriptionOptions {
454	/// Create a subscription to all changes at a path.
455	pub fn all(path: impl Into<Box<str>>) -> Self {
456		Self { path: path.into(), filter: None, select: None }
457	}
458
459	/// Create a subscription with a filter.
460	pub fn filtered(path: impl Into<Box<str>>, filter: QueryFilter) -> Self {
461		Self { path: path.into(), filter: Some(filter), select: None }
462	}
463
464	/// Set the field projection.
465	pub fn with_select(mut self, select: Option<Vec<String>>) -> Self {
466		self.select = select;
467		self
468	}
469}
470
471/// Real-time change event emitted when a document is created, updated, or deleted.
472#[derive(Debug, Clone, Serialize, Deserialize)]
473#[serde(tag = "action", rename_all = "camelCase")]
474pub enum ChangeEvent {
475	/// A new document was created
476	Create {
477		/// Full path to the document (e.g., "posts/abc123" or "posts/abc123/comments/xyz789")
478		path: Box<str>,
479		/// Full document data
480		data: Value,
481	},
482
483	/// An existing document was updated
484	Update {
485		/// Full path to the document
486		path: Box<str>,
487		/// Full updated document data
488		data: Value,
489		/// Previous document data (for incremental aggregate computation)
490		#[serde(default, skip_serializing_if = "Option::is_none")]
491		old_data: Option<Value>,
492	},
493
494	/// A document was deleted
495	Delete {
496		/// Full path to the document
497		path: Box<str>,
498		/// Document data before deletion (for incremental aggregate computation)
499		#[serde(default, skip_serializing_if = "Option::is_none")]
500		old_data: Option<Value>,
501	},
502
503	/// A lock was acquired on a document path
504	Lock {
505		/// Full path to the locked document
506		path: Box<str>,
507		/// Lock metadata (userId, mode)
508		data: Value,
509	},
510
511	/// A lock was released on a document path
512	Unlock {
513		/// Full path to the unlocked document
514		path: Box<str>,
515		/// Unlock metadata (userId)
516		data: Value,
517	},
518
519	/// Signals that all initial documents have been yielded for a subscription
520	Ready {
521		/// Subscription path
522		path: Box<str>,
523		/// Optional initial dataset
524		#[serde(default, skip_serializing_if = "Option::is_none")]
525		data: Option<Value>,
526	},
527
528	/// A complete result set replacing everything the subscriber holds.
529	///
530	/// Distinct from [`ChangeEvent::Ready`], which is the *one-shot*
531	/// initial-snapshot signal a client resolves its loading state on, and from
532	/// [`ChangeEvent::Update`], which a client merges as a delta. A min/max
533	/// aggregate cannot express its recompute as either: the recompute yields the
534	/// whole group set, so a group that emptied is simply absent rather than
535	/// zeroed, and merging would keep it forever.
536	Replace {
537		/// Subscription path
538		path: Box<str>,
539		/// The complete new dataset
540		#[serde(default, skip_serializing_if = "Option::is_none")]
541		data: Option<Value>,
542	},
543}
544
545impl ChangeEvent {
546	/// Get the full path from this event.
547	pub fn path(&self) -> &str {
548		match self {
549			ChangeEvent::Create { path, .. }
550			| ChangeEvent::Update { path, .. }
551			| ChangeEvent::Delete { path, .. }
552			| ChangeEvent::Lock { path, .. }
553			| ChangeEvent::Unlock { path, .. }
554			| ChangeEvent::Ready { path, .. }
555			| ChangeEvent::Replace { path, .. } => path,
556		}
557	}
558
559	/// Get the document ID (last segment of the path).
560	pub fn id(&self) -> Option<&str> {
561		self.path().split('/').next_back()
562	}
563
564	/// Get the parent path (all segments except the last).
565	pub fn parent_path(&self) -> Option<&str> {
566		let path = self.path();
567		path.rfind('/').map(|pos| &path[..pos])
568	}
569
570	/// Get the document data if this is a Create or Update event.
571	pub fn data(&self) -> Option<&Value> {
572		match self {
573			ChangeEvent::Create { data, .. }
574			| ChangeEvent::Update { data, .. }
575			| ChangeEvent::Lock { data, .. }
576			| ChangeEvent::Unlock { data, .. } => Some(data),
577			ChangeEvent::Delete { .. } => None,
578			ChangeEvent::Ready { data, .. } | ChangeEvent::Replace { data, .. } => data.as_ref(),
579		}
580	}
581
582	/// Check if this is a Create event.
583	pub fn is_create(&self) -> bool {
584		matches!(self, ChangeEvent::Create { .. })
585	}
586
587	/// Check if this is an Update event.
588	pub fn is_update(&self) -> bool {
589		matches!(self, ChangeEvent::Update { .. })
590	}
591
592	/// Check if this is a Delete event.
593	pub fn is_delete(&self) -> bool {
594		matches!(self, ChangeEvent::Delete { .. })
595	}
596}
597
598/// Compare two JSON values for ordering (used by filter range operators).
599fn compare_json_values(a: Option<&Value>, b: Option<&Value>) -> std::cmp::Ordering {
600	match (a, b) {
601		(None, None) => std::cmp::Ordering::Equal,
602		(None, Some(_)) => std::cmp::Ordering::Less,
603		(Some(_), None) => std::cmp::Ordering::Greater,
604		(Some(Value::Number(a)), Some(Value::Number(b))) => {
605			a.as_f64().partial_cmp(&b.as_f64()).unwrap_or(std::cmp::Ordering::Equal)
606		}
607		(Some(Value::String(a)), Some(Value::String(b))) => a.cmp(b),
608		(Some(Value::Bool(a)), Some(Value::Bool(b))) => a.cmp(b),
609		(Some(a), Some(b)) => a.to_string().cmp(&b.to_string()),
610	}
611}
612
613/// Convert a JSON value to a string key for aggregate group indexing.
614pub fn value_to_group_string(value: &Value) -> String {
615	match value {
616		Value::String(s) => s.clone(),
617		Value::Number(n) => n.to_string(),
618		Value::Bool(b) => b.to_string(),
619		Value::Null => "null".to_string(),
620		_ => serde_json::to_string(value).unwrap_or_default(),
621	}
622}
623
624/// Database statistics.
625#[derive(Debug, Clone, Serialize, Deserialize)]
626pub struct DbStats {
627	/// Total size of database files in bytes
628	pub size_bytes: u64,
629
630	/// Total number of documents across all tables
631	pub record_count: u64,
632
633	/// Number of tables in the database
634	pub table_count: u32,
635}
636
637/// Transaction for atomic write operations.
638///
639/// All write operations must be performed within a transaction to ensure atomicity.
640#[async_trait]
641pub trait Transaction: Send + Sync {
642	/// Create a new document with auto-generated ID. Returns the generated ID.
643	async fn create(&mut self, path: &str, data: Value) -> ClResult<Box<str>>;
644
645	/// Update an existing document (stores the provided data as-is).
646	///
647	/// Note: This method performs a full document replacement at the storage level.
648	/// Merge/PATCH semantics should be handled by the caller before invoking this method.
649	async fn update(&mut self, path: &str, data: Value) -> ClResult<()>;
650
651	/// Delete a document at a path.
652	async fn delete(&mut self, path: &str) -> ClResult<()>;
653
654	/// Read a document from the transaction's view.
655	///
656	/// This method provides transaction-local reads with "read-your-own-writes" semantics:
657	/// - Returns uncommitted changes made by this transaction
658	/// - Provides snapshot isolation from other concurrent transactions
659	/// - Essential for atomic operations like increment, append, etc.
660	///
661	/// # Returns
662	/// - `Ok(Some(value))` if document exists (either committed or written by this transaction)
663	/// - `Ok(None)` if document doesn't exist or was deleted by this transaction
664	/// - `Err` if read operation fails
665	async fn get(&self, path: &str) -> ClResult<Option<Value>>;
666
667	/// Query documents from the transaction's view, with the same
668	/// read-your-own-writes semantics as [`Transaction::get`].
669	///
670	/// Exists so a caller inside a transaction never has to reach back into
671	/// [`RtdbAdapter::query`] — see [`RtdbAdapter::transaction`] for why that
672	/// deadlocks.
673	async fn query(&self, path: &str, opts: &QueryOptions) -> ClResult<Vec<Value>>;
674
675	/// Read a hard/soft lock from the transaction's view.
676	///
677	/// Same reason as [`Transaction::query`]: the adapter-level
678	/// [`RtdbAdapter::check_lock`] must not be called while a transaction on the
679	/// same file is open.
680	async fn check_lock(&self, path: &str) -> ClResult<Option<LockInfo>>;
681
682	/// Commit the transaction, applying all changes atomically.
683	async fn commit(&mut self) -> ClResult<()>;
684
685	/// Rollback the transaction, discarding all changes.
686	async fn rollback(&mut self) -> ClResult<()>;
687}
688
689/// Real-Time Database Adapter trait.
690///
691/// Unified interface for database backends. Provides transaction-based writes,
692/// queries, and real-time subscriptions.
693#[async_trait]
694pub trait RtdbAdapter: Debug + Send + Sync {
695	/// Begin a new transaction for write operations.
696	///
697	/// **While a transaction is open, no code path may call back into this trait
698	/// for the same file.** A backend may hold the file's maintenance barrier for
699	/// the transaction's whole life, and re-entering through an adapter method
700	/// takes a second guard on it — which a queued `compact_storage` writer
701	/// deadlocks against permanently, hanging both the transaction and the
702	/// maintenance sweep. Every read a transaction needs is on [`Transaction`]
703	/// itself: [`Transaction::get`], [`Transaction::query`],
704	/// [`Transaction::check_lock`].
705	async fn transaction(&self, tn_id: TnId, db_id: &str) -> ClResult<Box<dyn Transaction>>;
706
707	/// Close a database instance, flushing pending changes to disk.
708	async fn close_db(&self, tn_id: TnId, db_id: &str) -> ClResult<()>;
709
710	/// Query documents at a path with optional filtering, sorting, and pagination.
711	async fn query(
712		&self,
713		tn_id: TnId,
714		db_id: &str,
715		path: &str,
716		opts: QueryOptions,
717	) -> ClResult<Vec<Value>>;
718
719	/// Get a document at a specific path. Returns None if not found.
720	async fn get(&self, tn_id: TnId, db_id: &str, path: &str) -> ClResult<Option<Value>>;
721
722	/// Subscribe to real-time changes at a path. Returns a stream of ChangeEvents.
723	async fn subscribe(
724		&self,
725		tn_id: TnId,
726		db_id: &str,
727		opts: SubscriptionOptions,
728	) -> ClResult<Pin<Box<dyn Stream<Item = ChangeEvent> + Send>>>;
729
730	/// Create an index on a field to improve query performance.
731	async fn create_index(&self, tn_id: TnId, db_id: &str, path: &str, field: &str)
732	-> ClResult<()>;
733
734	/// Get database statistics (size, record count, table count).
735	async fn stats(&self, tn_id: TnId, db_id: &str) -> ClResult<DbStats>;
736
737	/// Export all documents from a database.
738	///
739	/// Returns all `(path, document)` pairs. The path is relative to the db_id
740	/// (e.g., `posts/abc123`). Used for duplicating RTDB files.
741	async fn export_all(&self, tn_id: TnId, db_id: &str) -> ClResult<Vec<(Box<str>, Value)>>;
742
743	/// Acquire a lock on a document path.
744	///
745	/// Returns `Ok(None)` if the lock was acquired successfully.
746	/// Returns `Ok(Some(LockInfo))` if the path is already locked by another user (denied).
747	async fn acquire_lock(
748		&self,
749		tn_id: TnId,
750		db_id: &str,
751		path: &str,
752		user_id: &str,
753		mode: LockMode,
754		conn_id: &str,
755	) -> ClResult<Option<LockInfo>>;
756
757	/// Release a lock on a document path.
758	async fn release_lock(
759		&self,
760		tn_id: TnId,
761		db_id: &str,
762		path: &str,
763		user_id: &str,
764		conn_id: &str,
765	) -> ClResult<()>;
766
767	/// Check if a path has an active lock. Returns the lock info if locked.
768	async fn check_lock(&self, tn_id: TnId, db_id: &str, path: &str) -> ClResult<Option<LockInfo>>;
769
770	/// Release all locks held by a specific user (called on disconnect).
771	async fn release_all_locks(
772		&self,
773		tn_id: TnId,
774		db_id: &str,
775		user_id: &str,
776		conn_id: &str,
777	) -> ClResult<()>;
778
779	/// Delete every RTDB database owned by the tenant.
780	///
781	/// Used by tenant purge orchestration. Implementations should treat a
782	/// missing tenant store as success.
783	async fn delete_tenant_databases(&self, tn_id: TnId) -> ClResult<()>;
784
785	/// Rewrite every storage file, returning the space already freed inside them
786	/// to the filesystem.
787	///
788	/// Called from the nightly maintenance task, never from a request path: a
789	/// backend may have to close and reopen its files, which blocks every reader
790	/// and writer of the one being rewritten.
791	///
792	/// Defaults to a no-op report, the honest answer for a backend with nothing
793	/// to compact.
794	async fn compact_storage(&self) -> ClResult<CompactReport> {
795		Ok(CompactReport::default())
796	}
797}
798
799// vim: ts=4