1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
use crate::ctx::Context;
use crate::dbs::{Options, Transaction};
use crate::doc::CursorDoc;
use crate::err::Error;
use crate::iam::Auth;
use crate::{Cond, Fetchs, Fields, Uuid, Value};
use derive::Store;
use revision::revisioned;
use serde::{Deserialize, Serialize};
use std::fmt;
#[derive(Clone, Debug, Default, Eq, PartialEq, PartialOrd, Serialize, Deserialize, Store, Hash)]
#[revisioned(revision = 2)]
pub struct LiveStatement {
pub id: Uuid,
pub node: Uuid,
pub expr: Fields,
pub what: Value,
pub cond: Option<Cond>,
pub fetch: Option<Fetchs>,
// When a live query is marked for archiving, this will
// be set to the node ID that archived the query. This
// is an internal property, set by the database runtime.
// This is optional, and os only set when archived.
#[doc(hidden)]
pub archived: Option<Uuid>,
// When a live query is created, we must also store the
// authenticated session of the user who made the query,
// so we can chack it later when sending notifications.
// This is optional as it is only set by the database
// runtime when storing the live query to storage.
#[revision(start = 2)]
#[doc(hidden)]
pub session: Option<Value>,
// When a live query is created, we must also store the
// authenticated session of the user who made the query,
// so we can chack it later when sending notifications.
// This is optional as it is only set by the database
// runtime when storing the live query to storage.
#[doc(hidden)]
pub auth: Option<Auth>,
}
impl LiveStatement {
/// Creates a live statement from parts that can be set during a query.
pub(crate) fn from_source_parts(
expr: Fields,
what: Value,
cond: Option<Cond>,
fetch: Option<Fetchs>,
) -> Self {
LiveStatement {
id: Uuid::new_v4(),
node: Uuid::new_v4(),
expr,
what,
cond,
fetch,
..Default::default()
}
}
/// Process this type returning a computed simple Value
pub(crate) async fn compute(
&self,
ctx: &Context<'_>,
opt: &Options,
txn: &Transaction,
doc: Option<&CursorDoc<'_>>,
) -> Result<Value, Error> {
// Is realtime enabled?
opt.realtime()?;
// Valid options?
opt.valid_for_db()?;
// Get the Node ID
let nid = opt.id()?;
// Check that auth has been set
let mut stm = LiveStatement {
// Use the current session authentication
// for when we store the LIVE Statement
session: ctx.value("session").cloned(),
// Use the current session authentication
// for when we store the LIVE Statement
auth: Some(opt.auth.as_ref().clone()),
// Clone the rest of the original fields
// from the LIVE statement to the new one
..self.clone()
};
let id = stm.id.0;
// Claim transaction
let mut run = txn.lock().await;
// Process the live query table
match stm.what.compute(ctx, opt, txn, doc).await? {
Value::Table(tb) => {
// Store the current Node ID
stm.node = nid.into();
// Insert the node live query
run.putc_ndlq(nid, id, opt.ns(), opt.db(), tb.as_str(), None).await?;
// Insert the table live query
run.putc_tblq(opt.ns(), opt.db(), &tb, stm, None).await?;
}
v => {
return Err(Error::LiveStatement {
value: v.to_string(),
})
}
};
// Return the query id
Ok(id.into())
}
pub(crate) fn archive(mut self, node_id: Uuid) -> LiveStatement {
self.archived = Some(node_id);
self
}
}
impl fmt::Display for LiveStatement {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
write!(f, "LIVE SELECT {} FROM {}", self.expr, self.what)?;
if let Some(ref v) = self.cond {
write!(f, " {v}")?
}
if let Some(ref v) = self.fetch {
write!(f, " {v}")?
}
Ok(())
}
}