diff --git a/core/src/doc/lives.rs b/core/src/doc/lives.rs index 14d9a280..1d548255 100644 --- a/core/src/doc/lives.rs +++ b/core/src/doc/lives.rs @@ -12,9 +12,12 @@ use crate::sql::paths::SC; use crate::sql::paths::SD; use crate::sql::paths::TK; use crate::sql::permission::Permission; +use crate::sql::statements::LiveStatement; use crate::sql::Value; +use channel::Sender; use std::ops::Deref; use std::sync::Arc; +use uuid::Uuid; impl<'a> Document<'a> { pub async fn lives( @@ -36,125 +39,14 @@ impl<'a> Document<'a> { // Check if we can send notifications if let Some(chn) = &opt.sender { // Loop through all index statements - for lv in self.lv(opt, txn).await?.iter() { - // Create a new statement - let lq = Statement::from(lv); - // Get the event action - let met = if stm.is_delete() { - Value::from("DELETE") - } else if self.is_new() { - Value::from("CREATE") - } else { - Value::from("UPDATE") - }; - // Check if this is a delete statement - let doc = match stm.is_delete() { - true => &self.initial, - false => &self.current, - }; - // Ensure that a session exists on the LIVE query - let sess = match lv.session.as_ref() { - Some(v) => v, - None => continue, - }; - // Ensure that auth info exists on the LIVE query - let auth = match lv.auth.clone() { - Some(v) => v, - None => continue, - }; - // We need to create a new context which we will - // use for processing this LIVE query statement. - // This ensures that we are using the session - // of the user who created the LIVE query. - let mut lqctx = Context::background(); - lqctx.add_value("auth", sess.pick(SD.as_ref())); - lqctx.add_value("scope", sess.pick(SC.as_ref())); - lqctx.add_value("token", sess.pick(TK.as_ref())); - lqctx.add_value("session", sess); - // We need to create a new options which we will - // use for processing this LIVE query statement. - // This ensures that we are using the auth data - // of the user who created the LIVE query. - let lqopt = opt.new_with_perms(true).with_auth(Arc::from(auth)); - // Add $before, $after, $value, and $event params - // to this LIVE query so that user can use these - // within field projections and WHERE clauses. - lqctx.add_value("event", met); - lqctx.add_value("value", self.current.doc.deref()); - lqctx.add_value("after", self.current.doc.deref()); - lqctx.add_value("before", self.initial.doc.deref()); - // First of all, let's check to see if the WHERE - // clause of the LIVE query is matched by this - // document. If it is then we can continue. - match self.lq_check(&lqctx, &lqopt, txn, &lq, doc).await { - Err(Error::Ignore) => continue, - Err(e) => return Err(e), - Ok(_) => (), - } - // Secondly, let's check to see if any PERMISSIONS - // clause for this table allows this document to - // be viewed by the user who created this LIVE - // query. If it does, then we can continue. - match self.lq_allow(&lqctx, &lqopt, txn, &lq, doc).await { - Err(Error::Ignore) => continue, - Err(e) => return Err(e), - Ok(_) => (), - } - // Finally, let's check what type of statement - // caused this LIVE query to run, and send the - // relevant notification based on the statement. - if stm.is_delete() { - // Send a DELETE notification - if opt.id()? == lv.node.0 { - chn.send(Notification { - id: lv.id, - action: Action::Delete, - result: { - // Ensure futures are run - let lqopt: &Options = &lqopt.new_with_futures(true); - // Output the full document before any changes were applied - let mut value = - doc.doc.compute(&lqctx, lqopt, txn, Some(doc)).await?; - // Remove metadata fields on output - value.del(&lqctx, lqopt, txn, &*META).await?; - // Output result - value - }, - }) - .await?; - } else { - // TODO: Send to storage - } - } else if self.is_new() { - // Send a CREATE notification - if opt.id()? == lv.node.0 { - chn.send(Notification { - id: lv.id, - action: Action::Create, - result: self.pluck(&lqctx, &lqopt, txn, &lq).await?, - }) - .await?; - } else { - // TODO: Send to storage - } - } else { - // Send a UPDATE notification - if opt.id()? == lv.node.0 { - chn.send(Notification { - id: lv.id, - action: Action::Update, - result: self.pluck(&lqctx, &lqopt, txn, &lq).await?, - }) - .await?; - } else { - // TODO: Send to storage - } - }; - } + let lq_stms = self.lv(opt, txn).await?; + let borrows = lq_stms.iter().collect::>(); + self.check_lqs_and_send_notifications(opt, stm, txn, borrows.as_slice(), chn).await?; } // Carry on Ok(()) } + /// Check the WHERE clause for a LIVE query async fn lq_check( &self, @@ -205,4 +97,161 @@ impl<'a> Document<'a> { // Carry on Ok(()) } + + /// Process live query for notifications + pub(crate) async fn check_lqs_and_send_notifications( + &self, + opt: &Options, + stm: &Statement<'_>, + txn: &Transaction, + live_statements: &[&LiveStatement], + sender: &Sender, + ) -> Result<(), Error> { + trace!( + "Called check_lqs_and_send_notifications with {} live statements", + live_statements.len() + ); + for lv in live_statements { + // Create a new statement + let lq = Statement::from(*lv); + // Get the event action + let met = if stm.is_delete() { + Value::from("DELETE") + } else if self.is_new() { + Value::from("CREATE") + } else { + Value::from("UPDATE") + }; + // Check if this is a delete statement + let doc = match stm.is_delete() { + true => &self.initial, + false => &self.current, + }; + // Ensure that a session exists on the LIVE query + let sess = match lv.session.as_ref() { + Some(v) => v, + None => { + trace!("live query did not have a session, skipping"); + continue; + } + }; + // Ensure that auth info exists on the LIVE query + let auth = match lv.auth.clone() { + Some(v) => v, + None => { + trace!("live query did not have auth info, skipping"); + continue; + } + }; + // We need to create a new context which we will + // use for processing this LIVE query statement. + // This ensures that we are using the session + // of the user who created the LIVE query. + let mut lqctx = Context::background(); + lqctx.add_value("auth", sess.pick(SD.as_ref())); + lqctx.add_value("scope", sess.pick(SC.as_ref())); + lqctx.add_value("token", sess.pick(TK.as_ref())); + lqctx.add_value("session", sess); + // We need to create a new options which we will + // use for processing this LIVE query statement. + // This ensures that we are using the auth data + // of the user who created the LIVE query. + let lqopt = opt.new_with_perms(true).with_auth(Arc::from(auth)); + // Add $before, $after, $value, and $event params + // to this LIVE query so that user can use these + // within field projections and WHERE clauses. + lqctx.add_value("event", met); + lqctx.add_value("value", self.current.doc.deref()); + lqctx.add_value("after", self.current.doc.deref()); + lqctx.add_value("before", self.initial.doc.deref()); + // First of all, let's check to see if the WHERE + // clause of the LIVE query is matched by this + // document. If it is then we can continue. + match self.lq_check(&lqctx, &lqopt, txn, &lq, doc).await { + Err(Error::Ignore) => { + trace!("live query did not match the where clause, skipping"); + continue; + } + Err(e) => return Err(e), + Ok(_) => (), + } + // Secondly, let's check to see if any PERMISSIONS + // clause for this table allows this document to + // be viewed by the user who created this LIVE + // query. If it does, then we can continue. + match self.lq_allow(&lqctx, &lqopt, txn, &lq, doc).await { + Err(Error::Ignore) => { + trace!("live query did not have permission to view this document, skipping"); + continue; + } + Err(e) => return Err(e), + Ok(_) => (), + } + // Finally, let's check what type of statement + // caused this LIVE query to run, and send the + // relevant notification based on the statement. + let default_node_id = Uuid::default(); + let node_id = opt.id().unwrap_or(default_node_id); + // This bool is deprecated since lq v2 on cf + // We check against defaults because clients register live queries with their local node id + // But the cf scanner uses the server node id, which is different from the client + let node_matches_live_query = + node_id == default_node_id || lv.node.0 == default_node_id || node_id == lv.node.0; + trace!( + "Notification node matches live query: {} ({} != {})", + node_matches_live_query, + node_id, + lv.node.0 + ); + if stm.is_delete() { + // Send a DELETE notification + if node_matches_live_query { + trace!("Sending lq delete notification"); + sender + .send(Notification { + id: lv.id, + action: Action::Delete, + result: { + // Ensure futures are run + let lqopt: &Options = &lqopt.new_with_futures(true); + // Output the full document before any changes were applied + let mut value = + doc.doc.compute(&lqctx, lqopt, txn, Some(doc)).await?; + // Remove metadata fields on output + value.del(&lqctx, lqopt, txn, &*META).await?; + // Output result + value + }, + }) + .await?; + } + } else if self.is_new() { + // Send a CREATE notification + if node_matches_live_query { + trace!("Sending lq create notification"); + sender + .send(Notification { + id: lv.id, + action: Action::Create, + result: self.pluck(&lqctx, &lqopt, txn, &lq).await?, + }) + .await?; + } + } else { + // Send a UPDATE notification + if node_matches_live_query { + trace!("Sending lq update notification"); + sender + .send(Notification { + id: lv.id, + action: Action::Update, + result: self.pluck(&lqctx, &lqopt, txn, &lq).await?, + }) + .await?; + } + }; + } + trace!("Ended check_lqs_and_send_notifications"); + Ok(()) + } }