fix: keep valid tags when a bulk upsert hits a bad record (#134)

CHANGELOG.md

@@ -2,6 +2,8 @@
 
 ## Unreleased
 
+- `upsert_tags` no longer discards a whole batch when one record cannot be written. Records with an empty activity ID, an empty wallet ID, an empty tag, or a missing parent activity are now skipped and reported, and everything else in the batch is persisted. The function returns `UpsertTagsResult { inserted, skipped }` instead of nothing, where each `SkippedTag` carries the wallet ID, activity ID, the offending tag (if any) and the reason. Only batch-level failures (transaction, statement or commit errors) still return an `ActivityError`. FFI signature change: callers that ignored the previous empty return keep working, and restore flows can now log which tags were dropped instead of losing the entire category.
+
 - The Android AAR now ships targeted R8 consumer keep rules for the UniFFI/JNA FFI surface, so consuming apps can enable R8 full mode without extra keep rules for this library.
 - Add a generic hardware-wallet catalog with Foundation Passport support, multipart UR QR encoding and decoding, Passport single-signature account export parsing, and signed PSBT finalization across the UniFFI bindings.
 - Swap status updates now reconcile against Boltz's REST status whenever a swap is (re)subscribed, both on `boltz_start_swap_updates` and when `boltz_create_reverse_swap` adds a swap to a running stream. A confirmed reverse-swap lockup is therefore caught up and auto-claimed even when its live WebSocket event was missed (for example because the updates stream was down while the lockup confirmed), instead of the swap silently stalling until a manual claim. No FFI signature change.

README.md

@@ -180,6 +180,10 @@
     ```rust
     fn get_all_unique_tags() -> Result<Vec<String>, ActivityError>
     ```
+  - [upsert_tags](src/modules/activity/README.md#usage-examples): Bulk insert tags, skipping and reporting records that cannot be written
+    ```rust
+    fn upsert_tags(activity_tags: Vec<ActivityTags>) -> Result<UpsertTagsResult, ActivityError>
+    ```
   - [get_activities_by_tag](src/modules/activity/README.md#usage-examples): Get activities with a specific tag and optional wallet scope
     ```rust
     fn get_activities_by_tag(

src/lib.rs

@@ -30,7 +30,7 @@ use crate::activity::{
     derive_wallet_id as derive_wallet_id_inner, Activity, ActivityDB, ActivityError,
     ActivityFilter, ActivityTags, ClosedChannelDetails, DbError, LightningActivity,
     OnchainActivity, PaymentType, PreActivityMetadata, SortDirection, TransactionDetails,
-    DEFAULT_WALLET_ID,
+    UpsertTagsResult, DEFAULT_WALLET_ID,
 };
 use crate::modules::blocktank::{
     BlocktankDB, BlocktankError, BtOrderState2, CJitStateEnum, ChannelLiquidityOptions,
@@ -753,7 +753,7 @@ pub fn get_all_activities_tags() -> Result<Vec<ActivityTags>, ActivityError> {
 }
 
 #[uniffi::export]
-pub fn upsert_tags(activity_tags: Vec<ActivityTags>) -> Result<(), ActivityError> {
+pub fn upsert_tags(activity_tags: Vec<ActivityTags>) -> Result<UpsertTagsResult, ActivityError> {
     let mut guard = get_activity_db()?;
     let db = guard
         .activity_db

src/modules/activity/README.md

@@ -72,6 +72,12 @@ fn remove_tags(wallet_id: String, activity_id: String, tags: Vec<String>) -> Res
 fn get_tags(wallet_id: String, activity_id: String) -> Result<Vec<String>, ActivityError>
 fn get_all_unique_tags() -> Result<Vec<String>, ActivityError>
 
+// Bulk insert tags for multiple activities. Records that cannot be written (empty activity ID,
+// empty wallet ID, empty tag, or a tag whose parent activity is missing) are skipped and
+// reported in `UpsertTagsResult.skipped` instead of failing the batch.
+fn upsert_tags(activity_tags: Vec<ActivityTags>) -> Result<UpsertTagsResult, ActivityError>
+fn get_all_activities_tags() -> Result<Vec<ActivityTags>, ActivityError>
+
 // Pre-activity metadata
 fn add_pre_activity_metadata(pre_activity_metadata: PreActivityMetadata) -> Result<(), ActivityError>
 fn add_pre_activity_metadata_tags(wallet_id: String, payment_id: String, tags: Vec<String>) -> Result<(), ActivityError>

src/modules/activity/implementation.rs

@@ -1,7 +1,7 @@
 use crate::activity::{
     Activity, ActivityError, ActivityFilter, ActivityTags, ClosedChannelDetails, LightningActivity,
-    OnchainActivity, PaymentState, PaymentType, PreActivityMetadata, SortDirection,
-    TransactionDetails, TxInput, TxOutput,
+    OnchainActivity, PaymentState, PaymentType, PreActivityMetadata, SkippedTag, SortDirection,
+    TransactionDetails, TxInput, TxOutput, UpsertTagsResult,
 };
 use rusqlite::{Connection, OptionalExtension};
 use serde_json;
@@ -2202,10 +2202,21 @@ impl ActivityDB {
         Ok(result)
     }
 
-    /// Bulk upsert tags for multiple activities
-    pub fn upsert_tags(&mut self, activity_tags: &[ActivityTags]) -> Result<(), ActivityError> {
+    /// Bulk upsert tags for multiple activities.
+    ///
+    /// Records that cannot be written (empty activity ID, empty wallet ID, empty tag, or a
+    /// tag whose parent activity is missing) are skipped and reported in the result instead
+    /// of failing the batch, so one unusable record cannot discard the rest. Only failures
+    /// affecting the batch as a whole (transaction, statement or commit failures) return an
+    /// error.
+    pub fn upsert_tags(
+        &mut self,
+        activity_tags: &[ActivityTags],
+    ) -> Result<UpsertTagsResult, ActivityError> {
+        let mut result = UpsertTagsResult::default();
+
         if activity_tags.is_empty() {
-            return Ok(());
+            return Ok(result);
         }
 
         let tx = self
@@ -2226,25 +2237,44 @@ impl ActivityDB {
                 })?;
 
             for activity_tag in activity_tags {
-                if activity_tag.activity_id.is_empty() {
-                    return Err(ActivityError::DataError {
-                        error_details: "Activity ID cannot be empty".to_string(),
+                let mut skip_record = |tag: Option<&String>, reason: String| {
+                    result.skipped.push(SkippedTag {
+                        wallet_id: activity_tag.wallet_id.clone(),
+                        activity_id: activity_tag.activity_id.clone(),
+                        tag: tag.cloned(),
+                        reason,
                     });
+                };
+
+                if activity_tag.activity_id.is_empty() {
+                    skip_record(None, "Activity ID cannot be empty".to_string());
+                    continue;
                 }
-                let wallet_id = Self::normalize_wallet_id(&activity_tag.wallet_id)?;
+
+                let wallet_id = match Self::normalize_wallet_id(&activity_tag.wallet_id) {
+                    Ok(wallet_id) => wallet_id,
+                    Err(e) => {
+                        skip_record(None, e.to_string());
+                        continue;
+                    }
+                };
 
                 for tag in &activity_tag.tags {
                     if tag.is_empty() {
-                        continue; // Skip empty tags
+                        skip_record(Some(tag), "Tag cannot be empty".to_string());
+                        continue;
                     }
-                    stmt.execute(rusqlite::params![
+
+                    // A failing statement rolls back only itself, so the rest of the batch
+                    // still commits.
+                    match stmt.execute(rusqlite::params![
                         &wallet_id,
                         &activity_tag.activity_id,
                         tag
-                    ])
-                    .map_err(|e| ActivityError::DataError {
-                        error_details: format!("Failed to insert tag: {}", e),
-                    })?;
+                    ]) {
+                        Ok(_) => result.inserted += 1,
+                        Err(e) => skip_record(Some(tag), format!("Failed to insert tag: {}", e)),
+                    }
                 }
             }
         }
@@ -2253,7 +2283,7 @@ impl ActivityDB {
             error_details: format!("Failed to commit transaction: {}", e),
         })?;
 
-        Ok(())
+        Ok(result)
     }
 
     /// Add pre-activity metadata for an onchain address or lightning invoice.

src/modules/activity/tests.rs

@@ -3118,10 +3118,14 @@ mod tests {
             activity_id: activity.id.clone(),
             tags: vec!["payment".to_string(), "coffee".to_string()],
         }];
-        assert!(db.upsert_tags(&activity_tags).is_ok());
+        let result = db.upsert_tags(&activity_tags).unwrap();
+        assert_eq!(result.inserted, 2);
+        assert!(result.skipped.is_empty());
 
         // Second upsert with same tags (should be idempotent)
-        assert!(db.upsert_tags(&activity_tags).is_ok());
+        let result = db.upsert_tags(&activity_tags).unwrap();
+        assert_eq!(result.inserted, 2);
+        assert!(result.skipped.is_empty());
 
         // Verify tags are still there and not duplicated
         let tags = db.get_tags(DEFAULT_WALLET_ID, &activity.id).unwrap();
@@ -3178,7 +3182,11 @@ mod tests {
             activity_id: activity.id.clone(),
             tags: vec!["payment".to_string(), "".to_string(), "coffee".to_string()],
         }];
-        assert!(db.upsert_tags(&activity_tags).is_ok());
+        let result = db.upsert_tags(&activity_tags).unwrap();
+        assert_eq!(result.inserted, 2);
+        assert_eq!(result.skipped.len(), 1);
+        assert_eq!(result.skipped[0].tag, Some("".to_string()));
+        assert_eq!(result.skipped[0].reason, "Tag cannot be empty");
 
         // Verify only non-empty tags were added
         let tags = db.get_tags(DEFAULT_WALLET_ID, &activity.id).unwrap();
@@ -3398,7 +3406,9 @@ mod tests {
         let (mut db, db_path) = setup();
 
         // Test with empty vector
-        assert!(db.upsert_tags(&[]).is_ok());
+        let result = db.upsert_tags(&[]).unwrap();
+        assert_eq!(result.inserted, 0);
+        assert!(result.skipped.is_empty());
 
         cleanup(&db_path);
     }
@@ -3414,7 +3424,101 @@ mod tests {
             tags: vec!["payment".to_string()],
         }];
 
-        assert!(db.upsert_tags(&activity_tags).is_err());
+        let result = db.upsert_tags(&activity_tags).unwrap();
+        assert_eq!(result.inserted, 0);
+        assert_eq!(result.skipped.len(), 1);
+        assert_eq!(result.skipped[0].activity_id, "");
+        assert_eq!(result.skipped[0].tag, None);
+        assert_eq!(result.skipped[0].reason, "Activity ID cannot be empty");
+
+        cleanup(&db_path);
+    }
+
+    #[test]
+    fn test_upsert_tags_keeps_valid_records_when_one_is_unusable() {
+        let (mut db, db_path) = setup();
+
+        let activity = create_test_onchain_activity();
+        let mut other_activity = create_test_onchain_activity();
+        other_activity.id = "test_onchain_2".to_string();
+        other_activity.tx_id = "test_onchain_2_txid".to_string();
+
+        db.insert_onchain_activity(&activity).unwrap();
+        db.insert_onchain_activity(&other_activity).unwrap();
+
+        // A batch where the second record has no parent activity, the third has an empty
+        // activity ID and the fourth has an empty wallet ID.
+        let activity_tags = vec![
+            ActivityTags {
+                wallet_id: DEFAULT_WALLET_ID.to_string(),
+                activity_id: activity.id.clone(),
+                tags: vec!["payment".to_string(), "coffee".to_string()],
+            },
+            ActivityTags {
+                wallet_id: DEFAULT_WALLET_ID.to_string(),
+                activity_id: "missing_activity".to_string(),
+                tags: vec!["orphan".to_string()],
+            },
+            ActivityTags {
+                wallet_id: DEFAULT_WALLET_ID.to_string(),
+                activity_id: "".to_string(),
+                tags: vec!["nameless".to_string()],
+            },
+            ActivityTags {
+                wallet_id: "   ".to_string(),
+                activity_id: activity.id.clone(),
+                tags: vec!["walletless".to_string()],
+            },
+            ActivityTags {
+                wallet_id: DEFAULT_WALLET_ID.to_string(),
+                activity_id: other_activity.id.clone(),
+                tags: vec!["food".to_string()],
+            },
+        ];
+
+        let result = db.upsert_tags(&activity_tags).unwrap();
+
+        // The valid records on both sides of the bad ones are persisted.
+        assert_eq!(result.inserted, 3);
+        let mut tags = db.get_tags(DEFAULT_WALLET_ID, &activity.id).unwrap();
+        tags.sort();
+        assert_eq!(tags, vec!["coffee".to_string(), "payment".to_string()]);
+        assert_eq!(
+            db.get_tags(DEFAULT_WALLET_ID, &other_activity.id).unwrap(),
+            vec!["food".to_string()]
+        );
+
+        // And the caller can tell which records were skipped and why.
+        assert_eq!(result.skipped.len(), 3);
+
+        let orphan = &result.skipped[0];
+        assert_eq!(orphan.activity_id, "missing_activity");
+        assert_eq!(orphan.tag, Some("orphan".to_string()));
+        assert!(
+            orphan.reason.contains("FOREIGN KEY constraint failed"),
+            "unexpected reason: {}",
+            orphan.reason
+        );
+
+        let nameless = &result.skipped[1];
+        assert_eq!(nameless.activity_id, "");
+        assert_eq!(nameless.tag, None);
+        assert_eq!(nameless.reason, "Activity ID cannot be empty");
+
+        let walletless = &result.skipped[2];
+        assert_eq!(walletless.wallet_id, "   ");
+        assert_eq!(walletless.tag, None);
+        assert!(
+            walletless.reason.contains("Wallet ID cannot be empty"),
+            "unexpected reason: {}",
+            walletless.reason
+        );
+
+        assert!(db
+            .get_all_unique_tags()
+            .unwrap()
+            .iter()
+            .all(|tag| tag != "orphan" && tag != "nameless" && tag != "walletless"));
 
         cleanup(&db_path);
     }

src/modules/activity/types.rs

@@ -222,6 +222,27 @@ pub struct ActivityTags {
     pub tags: Vec<String>,
 }
 
+/// A tag record that a bulk upsert could not write.
+#[derive(Debug, Clone, uniffi::Record, Serialize, Deserialize)]
+pub struct SkippedTag {
+    pub wallet_id: String,
+    pub activity_id: String,
+    /// The individual tag that failed, or `None` when the whole record was
+    /// rejected before any of its tags were attempted.
+    pub tag: Option<String>,
+    /// Why the record was skipped.
+    pub reason: String,
+}
+
+/// Outcome of a bulk tag upsert.
+#[derive(Debug, Clone, Default, uniffi::Record, Serialize, Deserialize)]
+pub struct UpsertTagsResult {
+    /// Number of tags persisted, including tags that were already present.
+    pub inserted: u32,
+    /// Records that could not be written, with the reason for each.
+    pub skipped: Vec<SkippedTag>,
+}
+
 #[derive(Debug, Clone, uniffi::Record, Serialize, Deserialize)]
 pub struct PreActivityMetadata {
     #[serde(default = "default_wallet_id")]