Dynamically add new contact subscriptions (#584)

* Dynamically add new contact subscriptions

* Review fixes: only subscribe if contact added

* Fix comment

Author: tompro

crates/bcr-ebill-api/src/service/notification_service/default_service.rs

@@ -335,7 +335,6 @@ impl NotificationServiceApi for DefaultNotificationService {
     async fn add_company_transport(&self, company: &Company, keys: &BcrKeys) -> Result<()> {
         self.add_company_client(company, keys).await
     }
-
     /// Sent when an identity chain is created or updated
     async fn send_identity_chain_events(&self, events: IdentityChainEvent) -> Result<()> {
         debug!(
@@ -883,6 +882,7 @@ mod tests {
                 root_event: Option<nostr::event::Event>) -> bcr_ebill_transport::Result<nostr::event::Event>;
             async fn resolve_contact(&self, node_id: &NodeId) -> Result<Option<bcr_ebill_transport::transport::NostrContactData>>;
             async fn resolve_public_chain(&self, id: &str, chain_type: BlockchainType) -> Result<Vec<nostr::event::Event>>;
+            async fn add_contact_subscription(&self, contact: &NodeId) -> Result<()>;
         }
     }
 

crates/bcr-ebill-api/src/service/notification_service/nostr.rs

@@ -383,6 +383,12 @@ impl NotificationJsonTransportApi for NostrClient {
             .fetch_events(chain_filter(id, chain_type), Some(SortOrder::Asc), None)
             .await?)
     }
+
+    async fn add_contact_subscription(&self, node_id: &NodeId) -> Result<()> {
+        debug!("adding nostr subscription for contact {node_id}");
+        self.subscribe(Filter::new().author(node_id.npub())).await?;
+        Ok(())
+    }
 }
 
 #[derive(Clone)]

crates/bcr-ebill-transport/src/handler/bill_chain_event_processor.rs

@@ -121,24 +121,30 @@ impl BillChainEventProcessor {
                 .iter()
                 .map(|r| r.as_str().to_owned())
                 .collect();
-            if let Err(e) = self
-                .nostr_contact_store
-                .upsert(&NostrContact {
+            self.upsert_contact(
+                node_id,
+                &NostrContact {
                     npub: node_id.npub(),
                     name: contact.metadata.name,
                     relays,
                     trust_level: TrustLevel::Participant,
                     handshake_status: HandshakeStatus::None,
-                })
-                .await
-            {
-                error!("Failed to save nostr contact information for node_id {node_id}: {e}");
-            }
+                },
+            )
+            .await;
         } else {
             info!("Could not resolve nostr contact information for node_id {node_id}");
         }
     }
 
+    async fn upsert_contact(&self, node_id: &NodeId, contact: &NostrContact) {
+        if let Err(e) = self.nostr_contact_store.upsert(contact).await {
+            error!("Failed to save nostr contact information for node_id {node_id}: {e}");
+        } else if let Err(e) = self.transport.add_contact_subscription(node_id).await {
+            error!("Failed to add nostr contact subscription for contact node_id {node_id}: {e}");
+        }
+    }
+
     async fn add_bill_blocks(
         &self,
         bill_id: &BillId,
@@ -549,7 +555,7 @@ mod tests {
         // If we don't have the contact in the store, we will try to resolve it via Nostr
         contact_store.expect_by_node_id().returning(|_| Ok(None));
 
-        // If we get data it should be store to the store
+        // If we get data it should be stored to the store
         transport.expect_resolve_contact().returning(|_| {
             Ok(Some(NostrContactData {
                 metadata: Metadata {
@@ -560,8 +566,14 @@ mod tests {
             }))
         });
 
+        // Store new contact
         contact_store.expect_upsert().returning(|_| Ok(()));
 
+        // Subscribe to contact
+        transport
+            .expect_add_contact_subscription()
+            .returning(|_| Ok(()));
+
         let handler = BillChainEventProcessor::new(
             Arc::new(bill_chain_store),
             Arc::new(bill_store),

crates/bcr-ebill-transport/src/transport.rs

@@ -66,6 +66,8 @@ pub trait NotificationJsonTransportApi: ServiceTraitBounds {
         id: &str,
         chain_type: BlockchainType,
     ) -> Result<Vec<Event>>;
+    /// Adds a new Nostr subscription on the primary client for an added contact
+    async fn add_contact_subscription(&self, contact: &NodeId) -> Result<()>;
 }
 
 #[derive(Debug, Clone)]