1use 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#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
28#[serde(rename_all = "camelCase")]
29pub enum LockMode {
30 Soft,
31 Hard,
32}
33
34#[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#[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#[derive(Debug, Clone, Serialize, Deserialize)]
55#[serde(rename_all = "camelCase")]
56pub struct AggregateOptions {
57 pub group_by: String,
59
60 #[serde(default, skip_serializing_if = "Vec::is_empty")]
62 pub ops: Vec<AggregateOp>,
63}
64
65#[derive(Debug, Clone, Default, Serialize, Deserialize)]
70pub struct QueryFilter {
71 #[serde(default, skip_serializing_if = "HashMap::is_empty")]
73 pub equals: HashMap<String, Value>,
74
75 #[serde(default, skip_serializing_if = "HashMap::is_empty", rename = "notEquals")]
77 pub not_equals: HashMap<String, Value>,
78
79 #[serde(default, skip_serializing_if = "HashMap::is_empty", rename = "greaterThan")]
81 pub greater_than: HashMap<String, Value>,
82
83 #[serde(default, skip_serializing_if = "HashMap::is_empty", rename = "greaterThanOrEqual")]
85 pub greater_than_or_equal: HashMap<String, Value>,
86
87 #[serde(default, skip_serializing_if = "HashMap::is_empty", rename = "lessThan")]
89 pub less_than: HashMap<String, Value>,
90
91 #[serde(default, skip_serializing_if = "HashMap::is_empty", rename = "lessThanOrEqual")]
93 pub less_than_or_equal: HashMap<String, Value>,
94
95 #[serde(default, skip_serializing_if = "HashMap::is_empty", rename = "inArray")]
97 pub in_array: HashMap<String, Vec<Value>>,
98
99 #[serde(default, skip_serializing_if = "HashMap::is_empty", rename = "arrayContains")]
101 pub array_contains: HashMap<String, Value>,
102
103 #[serde(default, skip_serializing_if = "HashMap::is_empty", rename = "notInArray")]
105 pub not_in_array: HashMap<String, Vec<Value>>,
106
107 #[serde(default, skip_serializing_if = "HashMap::is_empty", rename = "arrayContainsAny")]
109 pub array_contains_any: HashMap<String, Vec<Value>>,
110
111 #[serde(default, skip_serializing_if = "HashMap::is_empty", rename = "arrayContainsAll")]
113 pub array_contains_all: HashMap<String, Vec<Value>>,
114}
115
116impl QueryFilter {
117 pub fn new() -> Self {
119 Self::default()
120 }
121
122 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 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 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 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 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 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 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 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 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 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 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 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 pub fn matches(&self, doc: &Value) -> bool {
197 for (field, expected) in &self.equals {
199 if doc.get(field) != Some(expected) {
200 return false;
201 }
202 }
203
204 for (field, expected) in &self.not_equals {
206 if doc.get(field) == Some(expected) {
207 return false;
208 }
209 }
210
211 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 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 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 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 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 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 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 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 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 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#[derive(Debug, Clone, Serialize, Deserialize)]
319pub struct SortField {
320 pub field: String,
322
323 pub ascending: bool,
325}
326
327impl SortField {
328 pub fn asc(field: impl Into<String>) -> Self {
330 Self { field: field.into(), ascending: true }
331 }
332
333 pub fn desc(field: impl Into<String>) -> Self {
335 Self { field: field.into(), ascending: false }
336 }
337}
338
339#[derive(Debug, Clone, Default)]
341pub struct QueryOptions {
342 pub filter: Option<QueryFilter>,
344
345 pub sort: Option<Vec<SortField>>,
347
348 pub limit: Option<u32>,
350
351 pub offset: Option<u32>,
353
354 pub aggregate: Option<AggregateOptions>,
356
357 pub select: Option<Vec<String>>,
360}
361
362impl QueryOptions {
363 pub fn new() -> Self {
365 Self::default()
366 }
367
368 pub fn with_filter(mut self, filter: QueryFilter) -> Self {
370 self.filter = Some(filter);
371 self
372 }
373
374 pub fn with_sort(mut self, sort: Vec<SortField>) -> Self {
376 self.sort = Some(sort);
377 self
378 }
379
380 pub fn with_limit(mut self, limit: u32) -> Self {
382 self.limit = Some(limit);
383 self
384 }
385
386 pub fn with_offset(mut self, offset: u32) -> Self {
388 self.offset = Some(offset);
389 self
390 }
391
392 pub fn with_aggregate(mut self, aggregate: AggregateOptions) -> Self {
394 self.aggregate = Some(aggregate);
395 self
396 }
397
398 pub fn with_select(mut self, select: Vec<String>) -> Self {
400 self.select = Some(select);
401 self
402 }
403}
404
405pub 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
429pub 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#[derive(Debug, Clone)]
441pub struct SubscriptionOptions {
442 pub path: Box<str>,
444
445 pub filter: Option<QueryFilter>,
447
448 pub select: Option<Vec<String>>,
451}
452
453impl SubscriptionOptions {
454 pub fn all(path: impl Into<Box<str>>) -> Self {
456 Self { path: path.into(), filter: None, select: None }
457 }
458
459 pub fn filtered(path: impl Into<Box<str>>, filter: QueryFilter) -> Self {
461 Self { path: path.into(), filter: Some(filter), select: None }
462 }
463
464 pub fn with_select(mut self, select: Option<Vec<String>>) -> Self {
466 self.select = select;
467 self
468 }
469}
470
471#[derive(Debug, Clone, Serialize, Deserialize)]
473#[serde(tag = "action", rename_all = "camelCase")]
474pub enum ChangeEvent {
475 Create {
477 path: Box<str>,
479 data: Value,
481 },
482
483 Update {
485 path: Box<str>,
487 data: Value,
489 #[serde(default, skip_serializing_if = "Option::is_none")]
491 old_data: Option<Value>,
492 },
493
494 Delete {
496 path: Box<str>,
498 #[serde(default, skip_serializing_if = "Option::is_none")]
500 old_data: Option<Value>,
501 },
502
503 Lock {
505 path: Box<str>,
507 data: Value,
509 },
510
511 Unlock {
513 path: Box<str>,
515 data: Value,
517 },
518
519 Ready {
521 path: Box<str>,
523 #[serde(default, skip_serializing_if = "Option::is_none")]
525 data: Option<Value>,
526 },
527
528 Replace {
537 path: Box<str>,
539 #[serde(default, skip_serializing_if = "Option::is_none")]
541 data: Option<Value>,
542 },
543}
544
545impl ChangeEvent {
546 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 pub fn id(&self) -> Option<&str> {
561 self.path().split('/').next_back()
562 }
563
564 pub fn parent_path(&self) -> Option<&str> {
566 let path = self.path();
567 path.rfind('/').map(|pos| &path[..pos])
568 }
569
570 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 pub fn is_create(&self) -> bool {
584 matches!(self, ChangeEvent::Create { .. })
585 }
586
587 pub fn is_update(&self) -> bool {
589 matches!(self, ChangeEvent::Update { .. })
590 }
591
592 pub fn is_delete(&self) -> bool {
594 matches!(self, ChangeEvent::Delete { .. })
595 }
596}
597
598fn 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
613pub 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#[derive(Debug, Clone, Serialize, Deserialize)]
626pub struct DbStats {
627 pub size_bytes: u64,
629
630 pub record_count: u64,
632
633 pub table_count: u32,
635}
636
637#[async_trait]
641pub trait Transaction: Send + Sync {
642 async fn create(&mut self, path: &str, data: Value) -> ClResult<Box<str>>;
644
645 async fn update(&mut self, path: &str, data: Value) -> ClResult<()>;
650
651 async fn delete(&mut self, path: &str) -> ClResult<()>;
653
654 async fn get(&self, path: &str) -> ClResult<Option<Value>>;
666
667 async fn query(&self, path: &str, opts: &QueryOptions) -> ClResult<Vec<Value>>;
674
675 async fn check_lock(&self, path: &str) -> ClResult<Option<LockInfo>>;
681
682 async fn commit(&mut self) -> ClResult<()>;
684
685 async fn rollback(&mut self) -> ClResult<()>;
687}
688
689#[async_trait]
694pub trait RtdbAdapter: Debug + Send + Sync {
695 async fn transaction(&self, tn_id: TnId, db_id: &str) -> ClResult<Box<dyn Transaction>>;
706
707 async fn close_db(&self, tn_id: TnId, db_id: &str) -> ClResult<()>;
709
710 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 async fn get(&self, tn_id: TnId, db_id: &str, path: &str) -> ClResult<Option<Value>>;
721
722 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 async fn create_index(&self, tn_id: TnId, db_id: &str, path: &str, field: &str)
732 -> ClResult<()>;
733
734 async fn stats(&self, tn_id: TnId, db_id: &str) -> ClResult<DbStats>;
736
737 async fn export_all(&self, tn_id: TnId, db_id: &str) -> ClResult<Vec<(Box<str>, Value)>>;
742
743 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 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 async fn check_lock(&self, tn_id: TnId, db_id: &str, path: &str) -> ClResult<Option<LockInfo>>;
769
770 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 async fn delete_tenant_databases(&self, tn_id: TnId) -> ClResult<()>;
784
785 async fn compact_storage(&self) -> ClResult<CompactReport> {
795 Ok(CompactReport::default())
796 }
797}
798
799