fix: publish rbf replacements before success

app/src/main/java/to/bitkit/repositories/ActivityRepo.kt

@@ -48,6 +48,7 @@ import to.bitkit.ext.nowMillis
 import to.bitkit.ext.nowTimestamp
 import to.bitkit.ext.rawId
 import to.bitkit.ext.runSuspendCatching
+import to.bitkit.ext.scopedActivityId
 import to.bitkit.ext.walletId
 import to.bitkit.models.ActivityBackupV1
 import to.bitkit.models.PubkyPublicKeyFormat
@@ -80,6 +81,8 @@ class ActivityRepo @Inject constructor(
     private val syncMutex = Mutex()
     private var activeSync: CompletableDeferred<Result<Unit>>? = null
 
+    private val publishedRbfReplacements = MutableStateFlow(RbfPublications())
+
     private val _state = MutableStateFlow(ActivityState())
     val state: StateFlow<ActivityState> = _state
 
@@ -105,6 +108,7 @@ class ActivityRepo @Inject constructor(
     }
 
     suspend fun resetState() = withContext(bgDispatcher) {
+        publishedRbfReplacements.update { RbfPublications(generation = it.generation + 1) }
         _state.update { ActivityState() }
         notifyActivitiesChanged()
         Logger.debug("Activity state reset", context = TAG)
@@ -320,9 +324,42 @@ class ActivityRepo @Inject constructor(
         }
     }
 
+    suspend fun publishRbfBoost(originalTxId: String, replacementTxId: String): Result<Unit> {
+        val generation = publishedRbfReplacements.value.generation
+        return withContext(bgDispatcher) {
+            runSuspendCatching {
+                val replacement = coreService.activity.publishRbfBoost(originalTxId, replacementTxId)
+                publishedRbfReplacements.update {
+                    check(it.generation == generation) { "Wallet activity reset during RBF publication" }
+                    it.copy(
+                        replacements = it.replacements +
+                            (scopedActivityId(replacement.v1.walletId, originalTxId) to replacement)
+                    )
+                }
+                notifyActivitiesChanged(tagsChanged = true)
+            }
+        }
+    }
+
+    fun isLocalRbfBoost(txid: String): Boolean =
+        scopedActivityId(WalletScope.default, txid) in publishedRbfReplacements.value.replacements
+
+    fun resolvePublishedRbfActivity(activity: Activity): Activity {
+        val replacements = publishedRbfReplacements.value.replacements
+        val visited = mutableSetOf<String>()
+        var current = activity
+        while (current is Activity.Onchain) {
+            if (current.v1.confirmed) return current
+            val key = scopedActivityId(current.v1.walletId, current.v1.txId)
+            if (!visited.add(key)) return current
+            current = replacements[key] ?: return current
+        }
+        return current
+    }
+
     suspend fun handleOnchainTransactionReplaced(txid: String, conflicts: List<String>) {
         coreService.activity.handleOnchainTransactionReplaced(txid, conflicts)
-        notifyActivitiesChanged()
+        notifyActivitiesChanged(tagsChanged = true)
     }
 
     suspend fun handleOnchainTransactionReorged(txid: String) {
@@ -1221,3 +1258,8 @@ private fun OnchainActivity.toPreActivityMetadata(tags: List<String>): PreActivi
 }
 
 private data class SyncedDuplicate(val synced: OnchainActivity, val restoredId: String)
+
+private data class RbfPublications(
+    val generation: Long = 0,
+    val replacements: Map<String, Activity.Onchain> = emptyMap(),
+)

app/src/main/java/to/bitkit/services/CoreService.kt

@@ -1664,20 +1664,58 @@ class ActivityService(
         }
     }
 
-    suspend fun handleOnchainTransactionReplaced(txid: String, conflicts: List<String>) {
+    suspend fun publishRbfBoost(originalTxId: String, replacementTxId: String): Activity.Onchain =
         ServiceQueue.CORE.background {
-            runCatching {
-                val replacedActivity = getOnchainActivityByTxId(txid)
-                markReplacedActivity(txid, replacedActivity, conflicts)
+            require(originalTxId.isNotBlank() && replacementTxId.isNotBlank() && originalTxId != replacementTxId) {
+                "RBF must replace the original transaction"
+            }
+            checkNotNull(getActivityByTxId(defaultWalletId, originalTxId)) { "RBF original is not available" }
+            if (getOrCreateReplacementActivity(replacementTxId) == null) {
+                // RBF returns after broadcast, before wallet sync discovers the new payment.
+                lightningService.sync()
+                checkNotNull(getOrCreateReplacementActivity(replacementTxId)) { "RBF replacement is not available" }
+            }
+            val cache = cacheStore.data.first()
+            val original = checkNotNull(getActivityByTxId(defaultWalletId, originalTxId))
+            val replacement = checkNotNull(getActivityByTxId(defaultWalletId, replacementTxId))
+            check(original.txType == PaymentType.SENT && replacement.txType == PaymentType.SENT)
+            check(!cache.isActivityDeleted(original.id, defaultWalletId))
+            check(!cache.isActivityDeleted(replacement.id, defaultWalletId))
+            val updated = updateReplacementActivity(
+                originalTxId,
+                replacementTxId,
+                replacement,
+                original,
+                cache.isContactDetached(replacement.id, defaultWalletId),
+            )
+            markReplacedActivity(originalTxId, original, listOf(replacementTxId))
+            updated
+        }
 
+    suspend fun handleOnchainTransactionReplaced(txid: String, conflicts: List<String>) {
+        ServiceQueue.CORE.background {
+            runSuspendCatching {
+                markReplacedActivity(txid, getActivityByTxId(defaultWalletId, txid), conflicts)
                 for (conflictTxid in conflicts) {
-                    val replacementActivity = getOrCreateReplacementActivity(conflictTxid)
-                    if (replacementActivity != null && !replacementActivity.boostTxIds.contains(txid)) {
-                        updateReplacementActivity(txid, conflictTxid, replacementActivity, replacedActivity)
+                    runSuspendCatching {
+                        getOrCreateReplacementActivity(conflictTxid) ?: return@runSuspendCatching
+                        val cache = cacheStore.data.first()
+                        val replacedActivity = getActivityByTxId(defaultWalletId, txid)
+                        val replacement = getActivityByTxId(defaultWalletId, conflictTxid)
+                            ?: return@runSuspendCatching
+                        updateReplacementActivity(
+                            txid,
+                            conflictTxid,
+                            replacement,
+                            replacedActivity,
+                            cache.isContactDetached(replacement.id, defaultWalletId),
+                        )
+                    }.onFailure {
+                        Logger.error("Failed to handle replacement '$conflictTxid' for '$txid'", it, context = TAG)
                     }
                 }
-            }.onFailure { e ->
-                Logger.error("Error handling onchain transaction replaced for $txid", e, context = TAG)
+            }.onFailure {
+                Logger.error("Failed to handle replacement for '$txid'", it, context = TAG)
             }
         }
     }
@@ -1697,7 +1735,7 @@ class ActivityService(
             val updatedActivity = replacedActivity.copy(
                 doesExist = false,
                 isBoosted = false,
-                updatedAt = System.currentTimeMillis().toULong() / 1000u
+                updatedAt = nowTimestamp().epochSecond.toULong()
             )
             paykitActivityRevision++
             updateActivity(activityId = replacedActivity.id, activity = Activity.Onchain(updatedActivity))
@@ -1712,82 +1750,43 @@ class ActivityService(
     }
 
     private suspend fun getOrCreateReplacementActivity(conflictTxid: String): OnchainActivity? {
-        var replacementActivity = getOnchainActivityByTxId(conflictTxid)
-
-        if (replacementActivity == null) {
-            val payments = lightningService.listPayments()
-            val replacementPayment = payments?.firstOrNull { payment ->
-                (payment.kind as? PaymentKind.Onchain)?.txid == conflictTxid
-            }
-
-            if (replacementPayment != null) {
-                Logger.info(
-                    "Processing replacement transaction $conflictTxid that was already in payments list",
-                    context = TAG
-                )
-                val processResult = runCatching {
-                    processOnchainPayment(
-                        kind = replacementPayment.kind as PaymentKind.Onchain,
-                        payment = replacementPayment,
-                        forceUpdate = false,
-                        channelId = null,
-                    )
-                    getOnchainActivityByTxId(conflictTxid)
-                }
-                processResult.onFailure { e ->
-                    Logger.error(
-                        "Failed to process replacement transaction $conflictTxid",
-                        e,
-                        context = TAG
-                    )
-                }
-                replacementActivity = processResult.getOrNull()
-            }
-        }
-
-        return replacementActivity
+        getActivityByTxId(defaultWalletId, conflictTxid)?.let { return it }
+        val payment = lightningService.listPayments()?.firstOrNull {
+            (it.kind as? PaymentKind.Onchain)?.txid == conflictTxid
+        } ?: return null
+        processOnchainPayment(
+            kind = payment.kind as PaymentKind.Onchain,
+            payment = payment,
+            forceUpdate = false,
+            channelId = null,
+        )
+        return getActivityByTxId(defaultWalletId, conflictTxid)
     }
 
-    private suspend fun updateReplacementActivity(
+    private fun updateReplacementActivity(
         txid: String,
         conflictTxid: String,
         replacementActivity: OnchainActivity,
         replacedActivity: OnchainActivity?,
-    ) {
-        val updatedActivity = replacementActivity.copy(
-            boostTxIds = replacementActivity.boostTxIds + txid,
-            isBoosted = true,
-            contact = replacementActivity.contact ?: replacedActivity?.contact,
-            updatedAt = System.currentTimeMillis().toULong() / 1000u,
+        contactDetached: Boolean,
+    ): Activity.Onchain {
+        if (txid in replacementActivity.boostTxIds) return Activity.Onchain(replacementActivity)
+        val updated = Activity.Onchain(
+            replacementActivity.copy(
+                boostTxIds = (replacementActivity.boostTxIds + txid).distinct(),
+                isBoosted = true,
+                contact = replacementActivity.contact ?: replacedActivity?.contact.takeUnless { contactDetached },
+                updatedAt = nowTimestamp().epochSecond.toULong(),
+            )
         )
-        update(replacementActivity.id, Activity.Onchain(updatedActivity))
-
         if (replacedActivity != null) {
-            copyTagsFromReplacedActivity(txid, conflictTxid, replacedActivity.id, replacementActivity.id)
-        }
-
-        Logger.info("Updated replacement transaction $conflictTxid with boostTxId $txid", context = TAG)
-    }
-
-    private suspend fun copyTagsFromReplacedActivity(
-        txid: String,
-        conflictTxid: String,
-        replacedActivityId: String,
-        replacementActivityId: String,
-    ) {
-        runCatching {
-            val replacedTags = tags(replacedActivityId)
-            if (replacedTags.isNotEmpty()) {
-                appendTags(replacementActivityId, replacedTags)
-            }
-        }.onFailure { e ->
-            Logger.error(
-                "Failed to copy tags from replaced transaction $txid " +
-                    "to replacement transaction $conflictTxid",
-                e,
-                context = TAG
-            )
+            val tags = getTags(defaultWalletId, replacedActivity.id)
+            if (tags.isNotEmpty()) addTags(defaultWalletId, replacementActivity.id, tags)
         }
+        paykitActivityRevision++
+        updateActivity(replacementActivity.id, updated)
+        Logger.info("Updated replacement '$conflictTxid' with boostTxId '$txid'", context = TAG)
+        return updated
     }
 
     suspend fun handleOnchainTransactionReorged(txid: String) {

app/src/main/java/to/bitkit/ui/ContentView.kt

@@ -1208,7 +1208,11 @@ private fun NavGraphBuilder.home(
         SavingsWalletScreen(
             onchainActivities = onchainActivities ?: persistentListOf(),
             onAllActivityButtonClick = { navController.navigateToAllActivity(activityListViewModel::clearFilters) },
-            onActivityItemClick = { navController.navToActivityDetail(it) },
+            onActivityItemClick = {
+                navController.navToActivityDetail(
+                    activityListViewModel.resolvePublishedRbfActivity(it)
+                )
+            },
             onEmptyActivityRowClick = { appViewModel.showSheet(Sheet.Receive()) },
             onTransferToSpendingClick = { appViewModel.showSheet(Sheet.TransferUnavailable) },
             onBackClick = { navController.popBackStack() },
@@ -1224,7 +1228,11 @@ private fun NavGraphBuilder.home(
             channels = lightningState.channels,
             lightningActivities = lightningActivities ?: persistentListOf(),
             onAllActivityButtonClick = { navController.navigateToAllActivity(activityListViewModel::clearFilters) },
-            onActivityItemClick = { navController.navToActivityDetail(it) },
+            onActivityItemClick = {
+                navController.navToActivityDetail(
+                    activityListViewModel.resolvePublishedRbfActivity(it)
+                )
+            },
             onEmptyActivityRowClick = { appViewModel.showSheet(Sheet.Receive()) },
             onTransferToSavingsClick = {
                 if (!hasSeenSavingsIntro) {
@@ -1241,7 +1249,11 @@ private fun NavGraphBuilder.home(
         val walletId = it.toRoute<Routes.HardwareWallet>().walletId
         HardwareWalletScreen(
             walletId = walletId,
-            onActivityItemClick = { navController.navToActivityDetail(it) },
+            onActivityItemClick = {
+                navController.navToActivityDetail(
+                    activityListViewModel.resolvePublishedRbfActivity(it)
+                )
+            },
             onTransferToSpendingClick = { appViewModel.showSheet(Sheet.TransferUnavailable) },
             onBackClick = { navController.popBackStack() },
         )
@@ -1256,7 +1268,11 @@ private fun NavGraphBuilder.allActivity(
         AllActivityScreen(
             viewModel = activityListViewModel,
             onBack = { navController.popBackStack() },
-            onActivityItemClick = { navController.navToActivityDetail(it) },
+            onActivityItemClick = {
+                navController.navToActivityDetail(
+                    activityListViewModel.resolvePublishedRbfActivity(it)
+                )
+            },
         )
     }
 }
@@ -1432,12 +1448,17 @@ private fun NavGraphBuilder.contacts(
         }
     }
     deepLinkableComposable<Routes.ContactActivity> {
+        val listViewModel = activityListViewModel
         PaykitRouteGuard(settingsViewModel, navController) {
             val viewModel: ContactActivityViewModel = hiltViewModel()
             ContactActivityScreen(
                 viewModel = viewModel,
                 onBackClick = { navController.popBackStack() },
-                onActivityItemClick = { navController.navToActivityDetail(it) },
+                onActivityItemClick = {
+                    navController.navToActivityDetail(
+                        listViewModel?.resolvePublishedRbfActivity(it) ?: it
+                    )
+                },
             )
         }
     }

app/src/main/java/to/bitkit/ui/screens/wallets/HomeScreen.kt

@@ -391,7 +391,11 @@ fun HomeScreen(
         onDismissWidgetsOnboardingHint = homeViewModel::dismissWidgetsOnboardingHint,
         onNavigateToAppStatus = { rootNavController.navigate(Routes.AppStatus) },
         onNavigateToAllActivity = { rootNavController.navigateToAllActivity(activityListViewModel::clearFilters) },
-        onNavigateToActivityItem = { rootNavController.navToActivityDetail(it) },
+        onNavigateToActivityItem = {
+            rootNavController.navToActivityDetail(
+                activityListViewModel.resolvePublishedRbfActivity(it)
+            )
+        },
         onNavigateToSavings = { walletNavController.navigate(Routes.Savings) },
         onNavigateToSpending = { walletNavController.navigate(Routes.Spending) },
         onClickHardwareWallet = { walletNavController.navigateTo(Routes.HardwareWallet(it)) },

app/src/main/java/to/bitkit/ui/screens/wallets/activity/ActivityDetailScreen.kt

@@ -292,6 +292,15 @@ fun ActivityDetailScreen(
                                 )
                                 detailViewModel.onDismissBoostSheet()
                             },
+                            onPending = {
+                                app.toast(
+                                    type = Toast.ToastType.WARNING,
+                                    title = resources.getString(R.string.wallet__boost_pending_title),
+                                    description = resources.getString(R.string.wallet__boost_pending_msg),
+                                    testTag = "BoostPendingToast"
+                                )
+                                detailViewModel.onDismissBoostSheet()
+                            },
                             onMaxFee = {
                                 app.toast(
                                     type = Toast.ToastType.ERROR,