Switch to new chain transport (#565)

* Remove all DM transport payload

* Validate sender for action events

* Tests for action event validation

* Valid flag on chain events

* Respect valid flag on queries

* Chain invite tests

* Update crates/bcr-ebill-transport/src/handler/bill_action_event_handler.rs

Co-authored-by: Copilot <[email protected]>

* Reorder event sending, fix some log warinings

* Fix all cargo fmt errors

---------

Co-authored-by: Copilot <[email protected]>

Author: tompro

crates/bcr-ebill-api/src/external/bitcoin.rs

@@ -210,7 +210,7 @@ impl BitcoinClientApi for BitcoinClient {
 
     fn generate_link_to_pay(&self, address: &str, sum: u64, message: &str) -> String {
         let btc_sum = util::currency::sat_to_btc(sum);
-        let link = format!("bitcoin:{}?amount={}&message={}", address, btc_sum, message);
+        let link = format!("bitcoin:{address}?amount={btc_sum}&message={message}");
         link
     }
 

crates/bcr-ebill-api/src/external/mint.rs

@@ -238,7 +238,7 @@ impl MintClientApi for MintClient {
     async fn get_keyset_info(&self, mint_url: &str, keyset_id: &str) -> Result<cdk02::KeySet> {
         let base = reqwest::Url::parse(mint_url).map_err(|_| Error::InvalidMintUrl)?;
         let url = base
-            .join(&format!("/v1/keys/{}", keyset_id))
+            .join(&format!("/v1/keys/{keyset_id}"))
             .expect("keys relative path");
         let res = reqwest::Client::new().get(url).send().await.map_err(|e| {
             log::error!("Error getting keyset info from mint {mint_url}: {e}");

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

@@ -120,8 +120,7 @@ impl DefaultNotificationService {
                         .await
                     {
                         error!(
-                            "Failed to send block notification, will add it to retry queue: {}",
-                            e
+                            "Failed to send block notification, will add it to retry queue: {e}"
                         );
                         let queue_message = NostrQueuedMessage {
                             id: uuid::Uuid::new_v4().to_string(),
@@ -134,18 +133,15 @@ impl DefaultNotificationService {
                             .add_message(queue_message, Self::NOSTR_MAX_RETRIES)
                             .await
                         {
-                            error!("Failed to add block notification to retry queue: {}", e);
+                            error!("Failed to add block notification to retry queue: {e}");
                         }
                     }
                 } else {
-                    warn!(
-                        "Failed to find recipient in contacts for node_id: {}",
-                        node_id
-                    );
+                    warn!("Failed to find recipient in contacts for node_id: {node_id}");
                 }
             }
         } else {
-            warn!("No transport node found for sender node_id: {}", sender);
+            warn!("No transport node found for sender node_id: {sender}");
         }
         Ok(())
     }
@@ -206,6 +202,7 @@ impl DefaultNotificationService {
                         received: block_event.data.block.timestamp,
                         time: event.created_at.as_u64(),
                         payload: event,
+                        valid: true,
                     })
                     .await
                     .map_err(|_| {
@@ -263,8 +260,8 @@ impl NotificationServiceApi for DefaultNotificationService {
             None,
         );
 
-        self.send_all_events(&event.sender(), all_events).await?;
         self.send_bill_chain_events(event).await?;
+        self.send_all_events(&event.sender(), all_events).await?;
         Ok(())
     }
 
@@ -277,8 +274,8 @@ impl NotificationServiceApi for DefaultNotificationService {
             None,
             None,
         );
-        self.send_all_events(&event.sender(), all_events).await?;
         self.send_bill_chain_events(event).await?;
+        self.send_all_events(&event.sender(), all_events).await?;
         Ok(())
     }
 
@@ -294,8 +291,8 @@ impl NotificationServiceApi for DefaultNotificationService {
             None,
             None,
         );
-        self.send_all_events(&event.sender(), all_events).await?;
         self.send_bill_chain_events(event).await?;
+        self.send_all_events(&event.sender(), all_events).await?;
         Ok(())
     }
 
@@ -308,8 +305,8 @@ impl NotificationServiceApi for DefaultNotificationService {
             None,
             None,
         );
-        self.send_all_events(&event.sender(), all_events).await?;
         self.send_bill_chain_events(event).await?;
+        self.send_all_events(&event.sender(), all_events).await?;
         Ok(())
     }
 
@@ -322,8 +319,8 @@ impl NotificationServiceApi for DefaultNotificationService {
             None,
             None,
         );
-        self.send_all_events(&event.sender(), all_events).await?;
         self.send_bill_chain_events(event).await?;
+        self.send_all_events(&event.sender(), all_events).await?;
         Ok(())
     }
 
@@ -336,8 +333,8 @@ impl NotificationServiceApi for DefaultNotificationService {
             None,
             None,
         );
-        self.send_all_events(&bill.sender(), all_events).await?;
         self.send_bill_chain_events(bill).await?;
+        self.send_all_events(&bill.sender(), all_events).await?;
         Ok(())
     }
 
@@ -354,8 +351,8 @@ impl NotificationServiceApi for DefaultNotificationService {
             None,
             None,
         );
-        self.send_all_events(&event.sender(), all_events).await?;
         self.send_bill_chain_events(event).await?;
+        self.send_all_events(&event.sender(), all_events).await?;
         Ok(())
     }
 
@@ -372,8 +369,8 @@ impl NotificationServiceApi for DefaultNotificationService {
             None,
             None,
         );
-        self.send_all_events(&event.sender(), all_events).await?;
         self.send_bill_chain_events(event).await?;
+        self.send_all_events(&event.sender(), all_events).await?;
         Ok(())
     }
 
@@ -390,8 +387,8 @@ impl NotificationServiceApi for DefaultNotificationService {
             None,
             None,
         );
-        self.send_all_events(&event.sender(), all_events).await?;
         self.send_bill_chain_events(event).await?;
+        self.send_all_events(&event.sender(), all_events).await?;
         Ok(())
     }
 
@@ -406,8 +403,6 @@ impl NotificationServiceApi for DefaultNotificationService {
             bill_id: bill.id.clone(),
             action_type: Some(ActionType::CheckBill),
             sum: Some(bill.sum),
-            keys: None,
-            blocks: vec![],
         });
         if let Some(node) = self.notification_transport.get(sender_node_id) {
             node.send_private_event(mint, event.try_into()?).await?;
@@ -451,8 +446,6 @@ impl NotificationServiceApi for DefaultNotificationService {
                     bill_id: bill_id.to_owned(),
                     action_type: Some(ActionType::CheckBill),
                     sum,
-                    keys: None,
-                    blocks: vec![],
                 };
                 for (_, recipient) in unique {
                     let event = Event::new_bill(payload.clone());
@@ -479,8 +472,8 @@ impl NotificationServiceApi for DefaultNotificationService {
                 Some(BillEventType::BillBlock),
                 None,
             );
-            self.send_all_events(&event.sender(), all_events).await?;
             self.send_bill_chain_events(event).await?;
+            self.send_all_events(&event.sender(), all_events).await?;
         }
         Ok(())
     }
@@ -503,7 +496,7 @@ impl NotificationServiceApi for DefaultNotificationService {
             validate_node_id_network(node_id)?;
         }
         let result = self.notification_store.list(filter).await.map_err(|e| {
-            error!("Failed to get client notifications: {}", e);
+            error!("Failed to get client notifications: {e}");
             Error::Persistence("Failed to get client notifications".to_string())
         })?;
         Ok(result)
@@ -515,7 +508,7 @@ impl NotificationServiceApi for DefaultNotificationService {
             .mark_as_done(notification_id)
             .await
             .map_err(|e| {
-                error!("Failed to mark notification as done: {}", e);
+                error!("Failed to mark notification as done: {e}");
                 Error::Persistence("Failed to mark notification as done".to_string())
             })?;
         Ok(())
@@ -559,10 +552,7 @@ impl NotificationServiceApi for DefaultNotificationService {
             .bill_notification_sent(bill_id, block_height, action)
             .await
             .map_err(|e| {
-                error!(
-                    "Failed to check if bill notification was already sent: {}",
-                    e
-                );
+                error!("Failed to check if bill notification was already sent: {e}");
                 Error::Persistence(
                     "Failed to check if bill notification was already sent".to_string(),
                 )
@@ -581,7 +571,7 @@ impl NotificationServiceApi for DefaultNotificationService {
             .set_bill_notification_sent(bill_id, block_height, action)
             .await
             .map_err(|e| {
-                error!("Failed to mark bill notification as sent: {}", e);
+                error!("Failed to mark bill notification as sent: {e}");
                 Error::Persistence("Failed to mark bill notification as sent".to_string())
             })?;
         Ok(())
@@ -604,21 +594,21 @@ impl NotificationServiceApi for DefaultNotificationService {
                     )
                     .await
                 {
-                    error!("Failed to send retry message: {}", e);
+                    error!("Failed to send retry message: {e}");
                     failed_ids.push(queued_message.id.clone());
                 } else if let Err(e) = self
                     .queued_message_store
                     .succeed_retry(&queued_message.id)
                     .await
                 {
-                    error!("Failed to mark retry message as sent: {}", e);
+                    error!("Failed to mark retry message as sent: {e}");
                 }
             }
         }
 
         for failed in failed_ids {
             if let Err(e) = self.queued_message_store.fail_retry(&failed).await {
-                error!("Failed to store failed retry attemt: {}", e);
+                error!("Failed to store failed retry attemt: {e}");
             }
         }
         Ok(())
@@ -1248,32 +1238,32 @@ mod tests {
         let mut mock = MockNotificationJsonTransport::new();
         mock.expect_get_sender_node_id().returning(node_id_test);
 
-        mock.expect_send_private_event()
-            .returning(|_, _| Ok(()))
-            .once();
-
         mock.expect_send_private_event()
             .withf(move |_, e| {
                 let r: Result<Event<ChainInvite>> = e.clone().try_into();
-                r.is_err()
+                r.is_ok()
             })
-            .returning(|_, _| Err(Error::Network("Failed to send".to_string())));
+            .returning(|_, _| Ok(()));
 
         mock.expect_send_public_chain_event()
             .returning(|_, _, _, _, _, _, _| Ok(get_test_nostr_event()));
 
-        mock.expect_send_private_event()
-            .withf(move |_, e| {
-                let r: Result<Event<ChainInvite>> = e.clone().try_into();
-                r.is_ok()
-            })
-            .returning(|_, _| Ok(()));
-
         let mock_event_store = setup_event_store_expectations(
             chain.get_latest_block().previous_hash.to_owned().as_str(),
             &bill.id,
         );
 
+        mock.expect_send_private_event()
+            .returning(|_, _| Ok(()))
+            .once();
+
+        mock.expect_send_private_event()
+            .withf(move |_, e| {
+                let r: Result<Event<ChainInvite>> = e.clone().try_into();
+                r.is_err()
+            })
+            .returning(|_, _| Err(Error::Network("Failed to send".to_string())));
+
         let mut queue_mock = MockNostrQueuedMessageStore::new();
         queue_mock
             .expect_add_message()

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

@@ -40,10 +40,7 @@ pub async fn create_nostr_clients(
 ) -> Result<Vec<Arc<NostrClient>>> {
     // primary identity is required to launch
     let keys = identity_store.get_or_create_key_pair().await.map_err(|e| {
-        error!(
-            "Failed to get or create nostr key pair for nostr client: {}",
-            e
-        );
+        error!("Failed to get or create nostr key pair for nostr client: {e}");
         Error::Crypto("Failed to get or create nostr key pair".to_string())
     })?;
     let nostr_name = match identity_store.get().await {
@@ -62,7 +59,7 @@ pub async fn create_nostr_clients(
     let companies = match company_store.get_all().await {
         Ok(companies) => companies,
         Err(e) => {
-            error!("Failed to get companies for nostr client: {}", e);
+            error!("Failed to get companies for nostr client: {e}");
             HashMap::new()
         }
     };
@@ -137,7 +134,7 @@ pub async fn create_nostr_consumer(
     };
 
     let processor = Arc::new(BillChainEventProcessor::new(
-        bill_blockchain_store,
+        bill_blockchain_store.clone(),
         bill_store.clone(),
         transport.clone(),
         nostr_contact_store,

crates/bcr-ebill-core/src/lib.rs

@@ -616,6 +616,6 @@ pub enum ValidationError {
 
 impl From<crate::blockchain::Error> for ValidationError {
     fn from(e: crate::blockchain::Error) -> Self {
-        ValidationError::Blockchain(format!("Blockchain error: {}", e))
+        ValidationError::Blockchain(format!("Blockchain error: {e}"))
     }
 }

crates/bcr-ebill-core/src/notification.rs

@@ -64,7 +64,7 @@ pub enum NotificationType {
 
 impl Display for NotificationType {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        f.write_str(format!("{:?}", self).as_str())
+        f.write_str(format!("{self:?}").as_str())
     }
 }
 

crates/bcr-ebill-persistence/src/db/backup.rs

@@ -71,7 +71,7 @@ impl BackupStoreApi for SurrealBackupStore {
         let _ = self
             .db()
             .await?
-            .query(format!("REMOVE DATABASE {}", name))
+            .query(format!("REMOVE DATABASE {name}"))
             .await?;
         Ok(())
     }

crates/bcr-ebill-persistence/src/db/bill.rs

@@ -51,7 +51,7 @@ impl BillStoreApi for SurrealBillStore {
             .map(|id| {
                 (
                     SurrealBillStore::CACHE_TABLE.to_owned(),
-                    format!("{}{}", id, identity_node_id),
+                    format!("{id}{identity_node_id}"),
                 )
                     .into()
             })
@@ -76,7 +76,7 @@ impl BillStoreApi for SurrealBillStore {
     ) -> Result<Option<BitcreditBillResult>> {
         let result: Option<BitcreditBillResultDb> = self
             .db
-            .select_one(Self::CACHE_TABLE, format!("{}{}", id, identity_node_id))
+            .select_one(Self::CACHE_TABLE, format!("{id}{identity_node_id}"))
             .await?;
         match result {
             None => Ok(None),
@@ -96,11 +96,7 @@ impl BillStoreApi for SurrealBillStore {
         let entity: BitcreditBillResultDb = (bill, identity_node_id).into();
         let _: Option<BitcreditBillResultDb> = self
             .db
-            .upsert(
-                Self::CACHE_TABLE,
-                format!("{}{}", id, identity_node_id),
-                entity,
-            )
+            .upsert(Self::CACHE_TABLE, format!("{id}{identity_node_id}"), entity)
             .await?;
         Ok(())
     }