Notification filter (#386)

* Add active filter for notification endpoint

* Add identity context to Nostr event handling

* Full filters

* Merged and docs

* Review fixes

Author: tompro

src/persistence/db/notification.rs

@@ -5,7 +5,7 @@ use serde_json::Value;
 use surrealdb::{engine::any::Any, sql::Thing, Surreal};
 
 use crate::{
-    persistence::notification::NotificationStoreApi,
+    persistence::notification::{NotificationFilter, NotificationStoreApi},
     service::notification_service::{ActionType, Notification, NotificationType},
     util::date::{now, DateTimeUtc},
 };
@@ -46,14 +46,28 @@ impl NotificationStoreApi for SurrealNotificationStore {
         }
     }
     /// Returns all currently active notifications from the database
-    async fn list(&self) -> Result<Vec<Notification>> {
-        let result: Vec<NotificationDb> = self
+    async fn list(&self, filter: NotificationFilter) -> Result<Vec<Notification>> {
+        let filters = filter.filters();
+        let mut query = self
             .db
-            .query("SELECT * FROM type::table($table) WHERE active = true ORDER BY datetime DESC")
+            .query(format!(
+                "SELECT * FROM type::table($table) {} ORDER BY datetime DESC LIMIT $limit START $offset",
+                filters
+            ))
             .bind(("table", Self::TABLE))
-            .await?
-            .take(0)?;
+            .bind(("limit", filter.get_limit()))
+            .bind(("offset", filter.get_offset()));
 
+        if let Some(active) = filter.get_active() {
+            query = query.bind(active.to_owned());
+        }
+        if let Some(reference_id) = filter.get_reference_id() {
+            query = query.bind(reference_id.to_owned());
+        }
+        if let Some(notification_type) = filter.get_notification_type() {
+            query = query.bind(notification_type.to_owned());
+        }
+        let result: Vec<NotificationDb> = query.await?.take(0)?;
         Ok(result.into_iter().map(|n| n.into()).collect())
     }
     /// Returns the latest active notification for the given reference and notification type
@@ -62,24 +76,27 @@ impl NotificationStoreApi for SurrealNotificationStore {
         reference: &str,
         notification_type: NotificationType,
     ) -> Result<Option<Notification>> {
-        let result: Vec<NotificationDb> = self.db.query("SELECT * FROM type::table($table) WHERE active = true AND reference_id = $reference_id AND notification_type = $notification_type ORDER BY datetime desc")
-            .bind(("table", Self::TABLE))
-            .bind(("reference_id", reference.to_owned()))
-            .bind(("notification_type", notification_type))
-            .await?
-            .take(0)?;
-
-        Ok(result.first().map(|n| n.clone().into()))
+        let result = self
+            .list(NotificationFilter {
+                active: Some(true),
+                reference_id: Some(reference.to_owned()),
+                notification_type: Some(notification_type.to_string()),
+                limit: Some(1),
+                ..Default::default()
+            })
+            .await?;
+        Ok(result.first().cloned())
     }
     /// Returns all notifications for the given reference and notification type that are active
     async fn list_by_type(&self, notification_type: NotificationType) -> Result<Vec<Notification>> {
-        let result: Vec<NotificationDb> = self.db.query("SELECT * FROM type::table($table) WHERE active = true AND notification_type = $notification_type ORDER BY datetime desc")
-            .bind(("table", Self::TABLE))
-            .bind(("notification_type", notification_type))
-            .await?
-        .take(0)?;
-
-        Ok(result.into_iter().map(|n| n.into()).collect())
+        let result = self
+            .list(NotificationFilter {
+                active: Some(true),
+                notification_type: Some(notification_type.to_string()),
+                ..Default::default()
+            })
+            .await?;
+        Ok(result)
     }
     /// Marks an active notification as done
     async fn mark_as_done(&self, notification_id: &str) -> Result<()> {
@@ -135,6 +152,7 @@ impl NotificationStoreApi for SurrealNotificationStore {
 #[derive(Debug, Clone, Serialize, Deserialize)]
 struct NotificationDb {
     pub id: Thing,
+    pub node_id: Option<String>,
     pub notification_type: NotificationType,
     pub reference_id: Option<String>,
     pub description: String,
@@ -147,6 +165,7 @@ impl From<NotificationDb> for Notification {
     fn from(value: NotificationDb) -> Self {
         Self {
             id: value.id.id.to_raw(),
+            node_id: value.node_id,
             notification_type: value.notification_type,
             reference_id: value.reference_id,
             description: value.description,
@@ -165,6 +184,7 @@ impl From<Notification> for NotificationDb {
                 value.id.to_owned(),
             )
                 .into(),
+            node_id: value.node_id,
             notification_type: value.notification_type,
             reference_id: value.reference_id,
             description: value.description,
@@ -243,36 +263,45 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn test_inserts_and_queries_notifiction() {
+    async fn test_inserts_and_queries_notification() {
         let store = get_store().await;
         let notification = test_notification("bill_id", Some(test_payload()));
         let r = store
             .add(notification.clone())
             .await
             .expect("could not create notification");
 
-        let all = store.list().await.expect("could not list notifications");
+        let all = store
+            .list(NotificationFilter::default())
+            .await
+            .expect("could not list notifications");
         assert!(!all.is_empty());
         assert_eq!(notification.id, r.id);
     }
 
     #[tokio::test]
-    async fn test_deletes_existing_notifiction() {
+    async fn test_deletes_existing_notification() {
         let store = get_store().await;
         let notification = test_notification("bill_id", Some(test_payload()));
         let r = store
             .add(notification.clone())
             .await
             .expect("could not create notification");
 
-        let all = store.list().await.expect("could not list notifications");
+        let all = store
+            .list(NotificationFilter::default())
+            .await
+            .expect("could not list notifications");
         assert!(!all.is_empty());
 
         store
             .delete(&r.id)
             .await
             .expect("could not delete notification");
-        let all = store.list().await.expect("could not list notifications");
+        let all = store
+            .list(NotificationFilter::default())
+            .await
+            .expect("could not list notifications");
         assert!(all.is_empty());
     }
 
@@ -285,15 +314,23 @@ mod tests {
             .await
             .expect("could not create notification");
 
-        let all = store.list().await.expect("could not list notifications");
+        let mut filter = NotificationFilter::default();
+        filter.active = Some(true);
+        let all = store
+            .list(filter.clone())
+            .await
+            .expect("could not list notifications");
         assert!(!all.is_empty());
 
         store
             .mark_as_done(&r.id)
             .await
             .expect("could not mark notification as done");
 
-        let all = store.list().await.expect("could not list notifications");
+        let all = store
+            .list(filter)
+            .await
+            .expect("could not list notifications");
         assert!(all.is_empty());
     }
 
@@ -369,7 +406,7 @@ mod tests {
     }
 
     fn test_notification(bill_id: &str, payload: Option<Value>) -> Notification {
-        Notification::new_bill_notification(bill_id, "test_notification", payload)
+        Notification::new_bill_notification(bill_id, "node_id", "test_notification", payload)
     }
 
     fn test_payload() -> Value {
@@ -379,6 +416,7 @@ mod tests {
     fn test_general_notification() -> Notification {
         Notification {
             id: Uuid::new_v4().to_string(),
+            node_id: Some("node_id".to_string()),
             notification_type: NotificationType::General,
             reference_id: Some("general".to_string()),
             description: "general desc".to_string(),

src/persistence/notification.rs

@@ -11,7 +11,7 @@ pub trait NotificationStoreApi: Send + Sync {
     /// Stores a new notification into the database
     async fn add(&self, notification: Notification) -> Result<Notification>;
     /// Returns all currently active notifications from the database
-    async fn list(&self) -> Result<Vec<Notification>>;
+    async fn list(&self, filter: NotificationFilter) -> Result<Vec<Notification>>;
     /// Returns the latest active notification for the given reference and notification type
     async fn get_latest_by_reference(
         &self,
@@ -41,3 +41,88 @@ pub trait NotificationStoreApi: Send + Sync {
         action_type: ActionType,
     ) -> Result<bool>;
 }
+
+#[derive(Default, Clone, PartialEq, Debug)]
+pub struct NotificationFilter {
+    pub active: Option<bool>,
+    pub reference_id: Option<String>,
+    pub notification_type: Option<String>,
+    pub limit: Option<i64>,
+    pub offset: Option<i64>,
+}
+
+impl NotificationFilter {
+    pub fn filters(&self) -> String {
+        let mut parts = vec![];
+        if self.active.is_some() {
+            parts.push("active = $active");
+        }
+        if self.reference_id.is_some() {
+            parts.push("reference_id = $reference_id");
+        }
+        if self.notification_type.is_some() {
+            parts.push("notification_type = $notification_type");
+        }
+
+        let filters = parts.join(" AND ");
+        if filters.is_empty() {
+            filters
+        } else {
+            format!("WHERE {}", filters)
+        }
+    }
+
+    pub fn get_limit(&self) -> i64 {
+        self.limit.unwrap_or(200)
+    }
+
+    pub fn get_offset(&self) -> i64 {
+        self.offset.unwrap_or(0)
+    }
+
+    pub fn get_active(&self) -> Option<(String, bool)> {
+        self.active.map(|active| ("active".to_string(), active))
+    }
+
+    pub fn get_reference_id(&self) -> Option<(String, String)> {
+        self.reference_id
+            .as_ref()
+            .map(|reference_id| ("reference_id".to_string(), reference_id.to_string()))
+    }
+
+    pub fn get_notification_type(&self) -> Option<(String, String)> {
+        self.notification_type.as_ref().map(|notification_type| {
+            (
+                "notification_type".to_string(),
+                notification_type.to_string(),
+            )
+        })
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    #[test]
+    fn test_query_filters() {
+        let empty = super::NotificationFilter::default();
+        assert_eq!(empty.filters(), "");
+
+        let active = super::NotificationFilter {
+            active: Some(true),
+            ..Default::default()
+        };
+        assert_eq!(active.filters(), "WHERE active = $active");
+
+        let all = super::NotificationFilter {
+            active: Some(true),
+            reference_id: Some("123".to_string()),
+            notification_type: Some("Bill".to_string()),
+            ..Default::default()
+        };
+
+        assert_eq!(
+            all.filters(),
+            "WHERE active = $active AND reference_id = $reference_id AND notification_type = $notification_type"
+        );
+    }
+}

src/service/bill_service.rs

@@ -42,6 +42,8 @@ use async_trait::async_trait;
 use borsh::to_vec;
 use borsh_derive::{BorshDeserialize, BorshSerialize};
 use log::info;
+#[cfg(test)]
+use mockall::automock;
 use rocket::http::ContentType;
 use rocket::Response;
 use rocket::{http::Status, response::Responder};
@@ -264,6 +266,7 @@ impl<'r, 'o: 'r> Responder<'r, 'o> for Error {
     }
 }
 
+#[cfg_attr(test, automock)]
 #[async_trait]
 pub trait BillServiceApi: Send + Sync {
     /// Get bill balances

src/service/mod.rs

@@ -240,7 +240,7 @@ pub async fn create_service_context(
     let notification_service =
         create_notification_service(nostr_client.clone(), db.notification_store.clone()).await?;
 
-    let bill_service = BillService::new(
+    let bill_service = Arc::new(BillService::new(
         client.clone(),
         db.bill_store,
         db.bill_blockchain_store.clone(),
@@ -252,7 +252,7 @@ pub async fn create_service_context(
         db.company_chain_store.clone(),
         db.contact_store.clone(),
         db.company_store.clone(),
-    );
+    ));
     let identity_service = IdentityService::new(
         db.identity_store.clone(),
         db.file_upload_store.clone(),
@@ -281,7 +281,7 @@ pub async fn create_service_context(
     .await?;
 
     let search_service = SearchService::new(
-        Arc::new(bill_service.clone()),
+        bill_service.clone(),
         contact_service.clone(),
         Arc::new(company_service.clone()),
     );
@@ -298,7 +298,7 @@ pub async fn create_service_context(
         dht_client: client,
         contact_service,
         search_service: Arc::new(search_service),
-        bill_service: Arc::new(bill_service),
+        bill_service,
         identity_service: Arc::new(identity_service),
         company_service: Arc::new(company_service),
         file_upload_service: Arc::new(file_upload_service),

src/service/notification_service/bill_action_event_handler.rs

@@ -62,12 +62,13 @@ impl NotificationHandlerApi for BillActionEventHandler {
         true
     }
 
-    async fn handle_event(&self, event: EventEnvelope) -> Result<()> {
+    async fn handle_event(&self, event: EventEnvelope, node_id: &str) -> Result<()> {
         let event: Option<Event<BillActionEventPayload>> = event.try_into().ok();
         if let Some(event) = event {
             // create notification
             let notification = Notification::new_bill_notification(
                 &event.data.bill_id,
+                node_id,
                 &self.event_description(&event.event_type),
                 Some(serde_json::to_value(&event.data)?),
             );