Notification filter with node_ids (#489)

tompro · web-flow · commit 213276ae4913 · 2025-04-29T15:50:39.000Z
* Notification filter with node_ids

* Review fixes
diff --git a/crates/bcr-ebill-persistence/src/db/notification.rs b/crates/bcr-ebill-persistence/src/db/notification.rs
@@ -84,6 +84,9 @@ impl NotificationStoreApi for SurrealNotificationStore {
         if let Some(notification_type) = filter.get_notification_type() {
             query = query.bind(notification_type.to_owned());
         }
+        if let Some(node_ids) = filter.get_node_ids() {
+            query = query.bind(node_ids.to_owned());
+        }
         let result: Vec<NotificationDb> = query.await?.take(0)?;
         Ok(result.into_iter().map(|n| n.into()).collect())
     }
diff --git a/crates/bcr-ebill-persistence/src/notification.rs b/crates/bcr-ebill-persistence/src/notification.rs
@@ -52,6 +52,7 @@ pub struct NotificationFilter {
     pub active: Option<bool>,
     pub reference_id: Option<String>,
     pub notification_type: Option<String>,
+    pub node_ids: Vec<String>,
     pub limit: Option<i64>,
     pub offset: Option<i64>,
 }
@@ -69,6 +70,10 @@ impl NotificationFilter {
             parts.push("notification_type = $notification_type");
         }
 
+        if !self.node_ids.is_empty() {
+            parts.push("node_id IN $node_ids");
+        }
+
         let filters = parts.join(" AND ");
         if filters.is_empty() {
             filters
@@ -103,6 +108,14 @@ impl NotificationFilter {
             )
         })
     }
+
+    pub fn get_node_ids(&self) -> Option<(String, Vec<String>)> {
+        if !self.node_ids.is_empty() {
+            Some(("node_ids".to_string(), self.node_ids.clone()))
+        } else {
+            None
+        }
+    }
 }
 
 #[cfg(test)]
@@ -118,16 +131,32 @@ mod tests {
         };
         assert_eq!(active.filters(), "WHERE active = $active");
 
+        let node_ids = super::NotificationFilter {
+            node_ids: vec!["123".to_string(), "456".to_string()],
+            ..Default::default()
+        };
+
+        assert_eq!(node_ids.filters(), "WHERE node_id IN $node_ids");
+
+        assert_eq!(
+            node_ids.get_node_ids(),
+            Some((
+                "node_ids".to_string(),
+                vec!["123".to_string(), "456".to_string()]
+            ))
+        );
+
         let all = super::NotificationFilter {
             active: Some(true),
             reference_id: Some("123".to_string()),
             notification_type: Some("Bill".to_string()),
+            node_ids: vec!["123".to_string()],
             ..Default::default()
         };
 
         assert_eq!(
             all.filters(),
-            "WHERE active = $active AND reference_id = $reference_id AND notification_type = $notification_type"
+            "WHERE active = $active AND reference_id = $reference_id AND notification_type = $notification_type AND node_id IN $node_ids"
         );
     }
 }
diff --git a/crates/bcr-ebill-transport/src/handler/bill_chain_event_handler.rs b/crates/bcr-ebill-transport/src/handler/bill_chain_event_handler.rs
@@ -16,10 +16,7 @@ use bcr_ebill_core::notification::{Notification, NotificationType};
 use bcr_ebill_persistence::NotificationStoreApi;
 use bcr_ebill_persistence::bill::BillChainStoreApi;
 use bcr_ebill_persistence::bill::BillStoreApi;
-use log::debug;
-use log::error;
-use log::info;
-use log::warn;
+use log::{debug, error, info, trace, warn};
 use std::sync::Arc;
 
 #[derive(Clone)]
@@ -50,7 +47,7 @@ impl BillChainEventHandler {
         event: &BillChainEventPayload,
         node_id: &str,
     ) -> Result<()> {
-        debug!("creating notification {event:?} for {node_id}");
+        trace!("creating notification {event:?} for {node_id}");
         // no action no notification required
         if event.action_type.is_none() {
             return Ok(());
@@ -95,7 +92,7 @@ impl BillChainEventHandler {
         // send push notification to connected clients
         match serde_json::to_value(notification) {
             Ok(notification) => {
-                debug!("sending notification {notification:?} for {node_id}");
+                trace!("sending notification {notification:?} for {node_id}");
                 self.push_service.send(notification).await;
             }
             Err(e) => {
@@ -384,7 +381,7 @@ impl NotificationHandlerApi for BillChainEventHandler {
     }
 
     async fn handle_event(&self, event: EventEnvelope, node_id: &str) -> Result<()> {
-        debug!("incoming bill chain event {event:?} for {node_id}");
+        debug!("incoming bill chain event for {node_id}");
         if let Ok(decoded) = Event::<BillChainEventPayload>::try_from(event.clone()) {
             if !decoded.data.blocks.is_empty() {
                 if let Err(e) = self
diff --git a/crates/bcr-ebill-transport/src/handler/mod.rs b/crates/bcr-ebill-transport/src/handler/mod.rs
@@ -1,7 +1,7 @@
 use crate::Result;
 use async_trait::async_trait;
 use bcr_ebill_core::ServiceTraitBounds;
-use log::info;
+use log::trace;
 #[cfg(test)]
 use mockall::automock;
 
@@ -47,7 +47,7 @@ impl NotificationHandlerApi for LoggingEventHandler {
     }
 
     async fn handle_event(&self, event: EventEnvelope, identity: &str) -> Result<()> {
-        info!("Received event: {event:?} for identity: {identity}");
+        trace!("Received event: {event:?} for identity: {identity}");
         Ok(())
     }
 }
diff --git a/crates/bcr-ebill-transport/src/push_notification.rs b/crates/bcr-ebill-transport/src/push_notification.rs
@@ -1,5 +1,5 @@
 use async_trait::async_trait;
-use log::error;
+use log::trace;
 use std::sync::Arc;
 
 use async_broadcast::{InactiveReceiver, Receiver, Sender};
@@ -23,7 +23,9 @@ pub struct PushService {
 
 impl PushService {
     pub fn new() -> Self {
-        let (tx, rx) = async_broadcast::broadcast::<Value>(5);
+        let (mut tx, rx) = async_broadcast::broadcast::<Value>(5);
+        tx.set_overflow(true);
+        tx.set_await_active(false);
         let inactive = rx.deactivate();
         Self {
             sender: Arc::new(tx),
@@ -44,7 +46,7 @@ impl PushApi for PushService {
         match self.sender.broadcast(value).await {
             Ok(_) => {}
             Err(err) => {
-                error!("Error sending push message: {}", err);
+                trace!("Error sending push message: {}", err);
             }
         }
     }
diff --git a/crates/bcr-ebill-wasm/main.js b/crates/bcr-ebill-wasm/main.js
@@ -133,7 +133,8 @@ async function start() {
   console.log("status: ", status);
 
   // Notifications
-  let notifications = await notificationApi.list();
+  let filter = current_identity ? { node_ids: [current_identity.node_id] } : null;
+  let notifications = await notificationApi.list(filter);
   console.log("notifications: ", notifications);
   return { companyApi, generalApi, identityApi, billApi, contactApi, notificationApi };
 }
diff --git a/crates/bcr-ebill-wasm/src/api/notification.rs b/crates/bcr-ebill-wasm/src/api/notification.rs
@@ -1,7 +1,7 @@
 use super::Result;
 use crate::{
     context::get_ctx,
-    data::{IntoWeb, notification::NotificationWeb},
+    data::{FromWeb, IntoWeb, NotificationFilters, notification::NotificationWeb},
 };
 use bcr_ebill_api::NotificationFilter;
 use log::{error, info};
@@ -40,21 +40,17 @@ impl Notification {
     #[wasm_bindgen(unchecked_return_type = "NotificationWeb[]")]
     pub async fn list(
         &self,
-        active: Option<bool>,
-        reference_id: Option<String>,
-        notification_type: Option<String>,
-        limit: Option<i64>,
-        offset: Option<i64>,
+        #[wasm_bindgen(unchecked_param_type = "NotificationFilters")] filters: JsValue,
     ) -> Result<JsValue> {
+        let filter = NotificationFilter::from_web(
+            serde_wasm_bindgen::from_value::<NotificationFilters>(filters)
+                .ok()
+                .unwrap_or_default(),
+        );
+
         let notifications = get_ctx()
             .notification_service
-            .get_client_notifications(NotificationFilter {
-                active,
-                reference_id,
-                notification_type,
-                limit,
-                offset,
-            })
+            .get_client_notifications(filter)
             .await?;
 
         let web: Vec<NotificationWeb> = notifications.into_iter().map(|n| n.into_web()).collect();
diff --git a/crates/bcr-ebill-wasm/src/data/mod.rs b/crates/bcr-ebill-wasm/src/data/mod.rs
@@ -1,5 +1,6 @@
 use async_trait::async_trait;
 use bcr_ebill_api::{
+    NotificationFilter,
     data::{
         File, GeneralSearchFilterItemType, GeneralSearchResult, OptionalPostalAddress,
         PostalAddress, UploadFileResult,
@@ -188,6 +189,30 @@ impl IntoWeb<PostalAddressWeb> for PostalAddress {
     }
 }
 
+#[derive(Tsify, Debug, Clone, Serialize, Deserialize, Default)]
+#[tsify(into_wasm_abi, from_wasm_abi)]
+pub struct NotificationFilters {
+    pub active: Option<bool>,
+    pub reference_id: Option<String>,
+    pub notification_type: Option<String>,
+    pub node_ids: Option<Vec<String>>,
+    pub limit: Option<i64>,
+    pub offset: Option<i64>,
+}
+
+impl FromWeb<NotificationFilters> for NotificationFilter {
+    fn from_web(value: NotificationFilters) -> Self {
+        Self {
+            active: value.active,
+            reference_id: value.reference_id,
+            notification_type: value.notification_type,
+            node_ids: value.node_ids.unwrap_or_default(),
+            limit: value.limit,
+            offset: value.offset,
+        }
+    }
+}
+
 #[derive(Tsify, Debug, Clone, Serialize, Deserialize)]
 #[tsify(into_wasm_abi, from_wasm_abi)]
 pub struct FileWeb {
diff --git a/crates/bcr-ebill-web/src/handlers/notifications.rs b/crates/bcr-ebill-web/src/handlers/notifications.rs
@@ -19,16 +19,18 @@ use serde_json::Value;
         ("active" = Option<bool>, Query, description = "Returns only active notifications when true, inactive when false and all when left out"),
         ("reference_id" = Option<String>, Query, description = "The id of the entity to filter by (eg. a bill id)"),
         ("notification_type" = Option<String>, Query, description = "The type of notifications to return (eg. Bill)"),
+        ("node_ids" = Option<Vec<String>>, Query, description = "The node_ids of the identity for which to query notifications"),
         ("limit" = Option<i64>, Query, description = "The max number of notifications to return"),
         ("offset" = Option<i64>, Query, description = "The number of notifications to skip at the start of the result")
     )
 )]
-#[get("/notifications?<active>&<reference_id>&<notification_type>&<limit>&<offset>")]
+#[get("/notifications?<active>&<reference_id>&<notification_type>&<limit>&<offset>&<node_ids>")]
 pub async fn list_notifications(
     state: &State<ServiceContext>,
     active: Option<bool>,
     reference_id: Option<String>,
     notification_type: Option<String>,
+    node_ids: Option<Vec<String>>,
     limit: Option<i64>,
     offset: Option<i64>,
 ) -> Result<Json<Vec<NotificationWeb>>> {
@@ -38,6 +40,7 @@ pub async fn list_notifications(
             active,
             reference_id,
             notification_type,
+            node_ids: node_ids.unwrap_or_default(),
             limit,
             offset,
         })

Original file line number	Diff line number	Diff line change
`@@ -84,6 +84,9 @@ impl NotificationStoreApi for SurrealNotificationStore {`
`84`	`84`	`if let Some(notification_type) = filter.get_notification_type() {`
`85`	`85`	`query = query.bind(notification_type.to_owned());`
`86`	`86`	`}`
	`87`	`+ if let Some(node_ids) = filter.get_node_ids() {`
	`88`	`+ query = query.bind(node_ids.to_owned());`
	`89`	`+ }`
`87`	`90`	`let result: Vec<NotificationDb> = query.await?.take(0)?;`
`88`	`91`	`Ok(result.into_iter().map(\|n\| n.into()).collect())`
`89`	`92`	`}`
Original file line number	Diff line number	Diff line change
`@@ -1,7 +1,7 @@`
`1`	`1`	`use crate::Result;`
`2`	`2`	`use async_trait::async_trait;`
`3`	`3`	`use bcr_ebill_core::ServiceTraitBounds;`
`4`		`-use log::info;`
	`4`	`+use log::trace;`
`5`	`5`	`#[cfg(test)]`
`6`	`6`	`use mockall::automock;`
`7`	`7`
`@@ -47,7 +47,7 @@ impl NotificationHandlerApi for LoggingEventHandler {`
`47`	`47`	`}`
`48`	`48`
`49`	`49`	`async fn handle_event(&self, event: EventEnvelope, identity: &str) -> Result<()> {`
`50`		`- info!("Received event: {event:?} for identity: {identity}");`
	`50`	`+ trace!("Received event: {event:?} for identity: {identity}");`
`51`	`51`	`Ok(())`
`52`	`52`	`}`
`53`	`53`	`}`
Original file line number	Diff line number	Diff line change
`@@ -133,7 +133,8 @@ async function start() {`
`133`	`133`	`console.log("status: ", status);`
`134`	`134`
`135`	`135`	`// Notifications`
`136`		`- let notifications = await notificationApi.list();`
	`136`	`+ let filter = current_identity ? { node_ids: [current_identity.node_id] } : null;`
	`137`	`+ let notifications = await notificationApi.list(filter);`
`137`	`138`	`console.log("notifications: ", notifications);`
`138`	`139`	`return { companyApi, generalApi, identityApi, billApi, contactApi, notificationApi };`
`139`	`140`	`}`