Implement LIVE and KILL statements

lib/src/dbs/options.rs

@@ -21,6 +21,8 @@ pub struct Options {
 	pub auth: Arc<Auth>,
 	// How many subqueries have we gone into?
 	pub dive: usize,
+	// Whether live queries are allowed?
+	pub live: bool,
 	// Should we debug query response SQL?
 	pub debug: bool,
 	// Should we force tables/events to re-run?
@@ -50,6 +52,7 @@ impl Options {
 			ns: None,
 			db: None,
 			dive: 0,
+			live: false,
 			perms: true,
 			debug: false,
 			force: false,
@@ -176,6 +179,13 @@ impl Options {
 		}
 	}
 
+	pub fn realtime(&self) -> Result<(), Error> {
+		if !self.live {
+			return Err(Error::RealtimeDisabled);
+		}
+		Ok(())
+	}
+
 	// Check whether the authentication permissions are ok
 	pub fn check(&self, level: Level) -> Result<(), Error> {
 		if !self.auth.check(level) {

lib/src/dbs/session.rs

@@ -8,15 +8,17 @@ use std::sync::Arc;
 pub struct Session {
 	/// The current [`Auth`] information
 	pub au: Arc<Auth>,
+	/// Whether realtime queries are supported
+	pub rt: bool,
 	/// The current connection IP address
 	pub ip: Option<String>,
 	/// The current connection origin
 	pub or: Option<String>,
 	/// The current connection ID
 	pub id: Option<String>,
-	/// THe currently selected namespace
+	/// The currently selected namespace
 	pub ns: Option<String>,
-	/// THe currently selected database
+	/// The currently selected database
 	pub db: Option<String>,
 	/// The currently selected authentication scope
 	pub sc: Option<String>,

lib/src/err/mod.rs

@@ -136,6 +136,10 @@ pub enum Error {
 	#[error("The table does not exist")]
 	TbNotFound,
 
+	/// Unable to perform the realtime query
+	#[error("Unable to perform the realtime query")]
+	RealtimeDisabled,
+
 	/// Too many recursive subqueries have been processed
 	#[error("Too many recursive subqueries have been processed")]
 	TooManySubqueries,
@@ -170,6 +174,18 @@ pub enum Error {
 		value: String,
 	},
 
+	/// Can not execute LIVE query using the specified value
+	#[error("Can not execute LIVE query using value '{value}'")]
+	LiveStatement {
+		value: String,
+	},
+
+	/// Can not execute KILL query using the specified id
+	#[error("Can not execute KILL query using id '{value}'")]
+	KillStatement {
+		value: String,
+	},
+
 	/// The permissions do not allow this query to be run on this table
 	#[error("You don't have permission to run this query on the `{table}` table")]
 	TablePermissions {

lib/src/key/lq.rs

@@ -0,0 +1,53 @@
+use crate::sql::uuid::Uuid;
+use derive::Key;
+use serde::{Deserialize, Serialize};
+
+#[derive(Clone, Debug, Eq, PartialEq, PartialOrd, Serialize, Deserialize, Key)]
+pub struct Lq {
+	__: u8,
+	_a: u8,
+	pub ns: String,
+	_b: u8,
+	pub db: String,
+	_c: u8,
+	_d: u8,
+	_e: u8,
+	pub lq: Uuid,
+}
+
+pub fn new(ns: &str, db: &str, lq: &Uuid) -> Lq {
+	Lq::new(ns.to_string(), db.to_string(), lq.to_owned())
+}
+
+impl Lq {
+	pub fn new(ns: String, db: String, lq: Uuid) -> Lq {
+		Lq {
+			__: 0x2f, // /
+			_a: 0x2a, // *
+			ns,
+			_b: 0x2a, // *
+			db,
+			_c: 0x21, // !
+			_d: 0x6c, // l
+			_e: 0x71, // v
+			lq,
+		}
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	#[test]
+	fn key() {
+		use super::*;
+		#[rustfmt::skip]
+		let val = Lq::new(
+			"test".to_string(),
+			"test".to_string(),
+			"00000000-0000-0000-0000-000000000000".into(),
+		);
+		let enc = Lq::encode(&val).unwrap();
+		let dec = Lq::decode(&enc).unwrap();
+		assert_eq!(val, dec);
+	}
+}

lib/src/key/lv.rs

@@ -1,3 +1,4 @@
+use crate::sql::uuid::Uuid;
 use derive::Key;
 use serde::{Deserialize, Serialize};
 
@@ -13,11 +14,11 @@ pub struct Lv {
 	_d: u8,
 	_e: u8,
 	_f: u8,
-	pub lv: String,
+	pub lv: Uuid,
 }
 
-pub fn new(ns: &str, db: &str, tb: &str, lv: &str) -> Lv {
+pub fn new(ns: &str, db: &str, tb: &str, lv: &Uuid) -> Lv {
-	Lv::new(ns.to_string(), db.to_string(), tb.to_string(), lv.to_string())
+	Lv::new(ns.to_string(), db.to_string(), tb.to_string(), lv.to_owned())
 }
 
 pub fn prefix(ns: &str, db: &str, tb: &str) -> Vec<u8> {
@@ -33,7 +34,7 @@ pub fn suffix(ns: &str, db: &str, tb: &str) -> Vec<u8> {
 }
 
 impl Lv {
-	pub fn new(ns: String, db: String, tb: String, lv: String) -> Lv {
+	pub fn new(ns: String, db: String, tb: String, lv: Uuid) -> Lv {
 		Lv {
 			__: 0x2f, // /
 			_a: 0x2a, // *
@@ -60,7 +61,7 @@ mod tests {
 			"test".to_string(),
 			"test".to_string(),
 			"test".to_string(),
-			"test".to_string(),
+			"00000000-0000-0000-0000-000000000000".into(),
 		);
 		let enc = Lv::encode(&val).unwrap();
 		let dec = Lv::decode(&enc).unwrap();

lib/src/key/mod.rs

@@ -12,6 +12,7 @@
 /// SC              /*{ns}*{db}!sc{sc}
 /// ST              /*{ns}*{db}!st{sc}!tk{tk}
 /// TB              /*{ns}*{db}!tb{tb}
+/// LQ              /*{ns}*{db}!lq{lq}
 ///
 /// Table           /*{ns}*{db}*{tb}
 /// FT              /*{ns}*{db}*{tb}!ft{ft}
@@ -37,6 +38,7 @@ pub mod graph;
 pub mod index;
 pub mod ix;
 pub mod kv;
+pub mod lq;
 pub mod lv;
 pub mod namespace;
 pub mod nl;

lib/src/kvs/ds.rs

@@ -163,10 +163,14 @@ impl Datastore {
 		let ctx = vars.attach(ctx);
 		// Parse the SQL query text
 		let ast = sql::parse(txt)?;
-		// Process all statements
+		// Setup the auth options
 		opt.auth = sess.au.clone();
+		// Setup the live options
+		opt.live = sess.rt;
+		// Set current NS and DB
 		opt.ns = sess.ns();
 		opt.db = sess.db();
+		// Process all statements
 		exe.execute(ctx, opt, ast).await
 	}
 
@@ -187,10 +191,14 @@ impl Datastore {
 		let ctx = sess.context(ctx);
 		// Store the query variables
 		let ctx = vars.attach(ctx);
-		// Process all statements
+		// Setup the auth options
 		opt.auth = sess.au.clone();
+		// Setup the live options
+		opt.live = sess.rt;
+		// Set current NS and DB
 		opt.ns = sess.ns();
 		opt.db = sess.db();
+		// Process all statements
 		exe.execute(ctx, opt, ast).await
 	}
 
@@ -213,8 +221,9 @@ impl Datastore {
 		let ctx = sess.context(ctx);
 		// Store the query variables
 		let ctx = vars.attach(ctx);
-		// Setup the query options
+		// Setup the auth options
 		opt.auth = sess.au.clone();
+		// Set current NS and DB
 		opt.ns = sess.ns();
 		opt.db = sess.db();
 		// Compute the value

lib/src/sql/statements/kill.rs

@@ -1,10 +1,11 @@
 use crate::ctx::Context;
+use crate::dbs::Level;
 use crate::dbs::Options;
 use crate::dbs::Transaction;
 use crate::err::Error;
 use crate::sql::comment::shouldbespace;
 use crate::sql::error::IResult;
-use crate::sql::ident::{ident, Ident};
+use crate::sql::uuid::{uuid, Uuid};
 use crate::sql::value::Value;
 use derive::Store;
 use nom::bytes::complete::tag_no_case;
@@ -13,18 +14,52 @@ use std::fmt;
 
 #[derive(Clone, Debug, Default, Eq, PartialEq, Serialize, Deserialize, Store)]
 pub struct KillStatement {
-	pub id: Ident,
+	pub id: Uuid,
 }
 
 impl KillStatement {
 	pub(crate) async fn compute(
 		&self,
 		_ctx: &Context<'_>,
-		_opt: &Options,
+		opt: &Options,
-		_txn: &Transaction,
+		txn: &Transaction,
 		_doc: Option<&Value>,
 	) -> Result<Value, Error> {
-		todo!()
+		// Allowed to run?
+		opt.realtime()?;
+		// Allowed to run?
+		opt.check(Level::No)?;
+		// Clone transaction
+		let run = txn.clone();
+		// Claim transaction
+		let mut run = run.lock().await;
+		// Create the live query key
+		let key = crate::key::lq::new(opt.ns(), opt.db(), &self.id);
+		// Fetch the live query key if it exists
+		match run.get(key).await? {
+			Some(val) => match std::str::from_utf8(&val) {
+				Ok(tb) => {
+					// Delete the live query
+					let key = crate::key::lq::new(opt.ns(), opt.db(), &self.id);
+					run.del(key).await?;
+					// Delete the table live query
+					let key = crate::key::lv::new(opt.ns(), opt.db(), tb, &self.id);
+					run.del(key).await?;
+				}
+				_ => {
+					return Err(Error::KillStatement {
+						value: self.id.to_string(),
+					})
+				}
+			},
+			None => {
+				return Err(Error::KillStatement {
+					value: self.id.to_string(),
+				})
+			}
+		}
+		// Return the query id
+		Ok(Value::None)
 	}
 }
 
@@ -37,7 +72,7 @@ impl fmt::Display for KillStatement {
 pub fn kill(i: &str) -> IResult<&str, KillStatement> {
 	let (i, _) = tag_no_case("KILL")(i)?;
 	let (i, _) = shouldbespace(i)?;
-	let (i, v) = ident(i)?;
+	let (i, v) = uuid(i)?;
 	Ok((
 		i,
 		KillStatement {

lib/src/sql/statements/live.rs

@@ -1,4 +1,5 @@
 use crate::ctx::Context;
+use crate::dbs::Level;
 use crate::dbs::Options;
 use crate::dbs::Transaction;
 use crate::err::Error;
@@ -7,10 +8,14 @@ use crate::sql::cond::{cond, Cond};
 use crate::sql::error::IResult;
 use crate::sql::fetch::{fetch, Fetchs};
 use crate::sql::field::{fields, Fields};
+use crate::sql::param::param;
+use crate::sql::table::table;
+use crate::sql::uuid::Uuid;
 use crate::sql::value::Value;
-use crate::sql::value::{whats, Values};
 use derive::Store;
+use nom::branch::alt;
 use nom::bytes::complete::tag_no_case;
+use nom::combinator::map;
 use nom::combinator::opt;
 use nom::sequence::preceded;
 use serde::{Deserialize, Serialize};
@@ -18,8 +23,9 @@ use std::fmt;
 
 #[derive(Clone, Debug, Default, Eq, PartialEq, Serialize, Deserialize, Store)]
 pub struct LiveStatement {
+	pub id: Uuid,
 	pub expr: Fields,
-	pub what: Values,
+	pub what: Value,
 	pub cond: Option<Cond>,
 	pub fetch: Option<Fetchs>,
 }
@@ -27,12 +33,37 @@ pub struct LiveStatement {
 impl LiveStatement {
 	pub(crate) async fn compute(
 		&self,
-		_ctx: &Context<'_>,
+		ctx: &Context<'_>,
-		_opt: &Options,
+		opt: &Options,
-		_txn: &Transaction,
+		txn: &Transaction,
-		_doc: Option<&Value>,
+		doc: Option<&Value>,
 	) -> Result<Value, Error> {
-		todo!()
+		// Allowed to run?
+		opt.realtime()?;
+		// Allowed to run?
+		opt.check(Level::No)?;
+		// Clone transaction
+		let run = txn.clone();
+		// Claim transaction
+		let mut run = run.lock().await;
+		// Process the live query table
+		match self.what.compute(ctx, opt, txn, doc).await? {
+			Value::Table(tb) => {
+				// Insert the live query
+				let key = crate::key::lq::new(opt.ns(), opt.db(), &self.id);
+				run.putc(key, tb.as_str(), None).await?;
+				// Insert the table live query
+				let key = crate::key::lv::new(opt.ns(), opt.db(), &tb, &self.id);
+				run.putc(key, self.clone(), None).await?;
+			}
+			v => {
+				return Err(Error::LiveStatement {
+					value: v.to_string(),
+				})
+			}
+		};
+		// Return the query id
+		Ok(self.id.clone().into())
 	}
 }
 
@@ -56,12 +87,13 @@ pub fn live(i: &str) -> IResult<&str, LiveStatement> {
 	let (i, _) = shouldbespace(i)?;
 	let (i, _) = tag_no_case("FROM")(i)?;
 	let (i, _) = shouldbespace(i)?;
-	let (i, what) = whats(i)?;
+	let (i, what) = alt((map(param, Value::from), map(table, Value::from)))(i)?;
 	let (i, cond) = opt(preceded(shouldbespace, cond))(i)?;
 	let (i, fetch) = opt(preceded(shouldbespace, fetch))(i)?;
 	Ok((
 		i,
 		LiveStatement {
+			id: Uuid::new(),
 			expr,
 			what,
 			cond,