fix: deliver completed Paykit proofs promptly

app/build.gradle.kts

@@ -589,6 +589,7 @@ dependencies {
     debugImplementation(libs.compose.ui.tooling)
     debugImplementation(libs.compose.ui.test.manifest)
     androidTestImplementation(libs.compose.ui.test.junit4)
+    testImplementation(libs.compose.ui.test.junit4)
     implementation(libs.accompanist.pager.indicators)
     implementation(libs.accompanist.permissions)
     implementation(libs.constraintlayout.compose)

app/src/androidTest/java/to/bitkit/ui/screens/wallets/send/SendPendingObservationDeviceTest.kt

@@ -151,6 +151,7 @@ class SendPendingObservationDeviceTest {
                                 onRecovered = { txid, amount -> recovered = txid to amount },
                                 savedStateHandle = SavedStateHandle(),
                                 onNavigateToPin = { error("Observation must not request spend authentication") },
+                                beginPaymentReview = { error("Observation must not hold payment review") },
                             )
                         } else {
                             Text("Original payment recovered")

app/src/main/java/to/bitkit/ext/Lifecycle.kt

@@ -0,0 +1,16 @@
+package to.bitkit.ext
+
+import androidx.lifecycle.Lifecycle
+import androidx.lifecycle.repeatOnLifecycle
+import kotlinx.coroutines.awaitCancellation
+
+internal suspend fun Lifecycle.withPaymentReview(beginReview: () -> (() -> Unit)) {
+    repeatOnLifecycle(Lifecycle.State.STARTED) {
+        val finishReview = beginReview()
+        try {
+            awaitCancellation()
+        } finally {
+            finishReview()
+        }
+    }
+}

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

@@ -17,6 +17,7 @@ import kotlinx.coroutines.flow.drop
 import kotlinx.coroutines.flow.flowOn
 import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.flow.update
+import kotlinx.coroutines.launch
 import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.sync.withLock
 import kotlinx.coroutines.withContext
@@ -32,6 +33,7 @@ import org.lightningdevkit.ldknode.PaymentDetails
 import org.lightningdevkit.ldknode.PaymentDirection
 import org.lightningdevkit.ldknode.PaymentKind
 import org.lightningdevkit.ldknode.PaymentStatus
+import to.bitkit.async.appScope
 import to.bitkit.di.IoDispatcher
 import to.bitkit.ext.fromHex
 import to.bitkit.ext.runSuspendCatching
@@ -41,6 +43,7 @@ import to.bitkit.models.PaykitPaymentStateBackup
 import to.bitkit.models.PaykitRequestPayment
 import to.bitkit.models.PubkyPublicKeyFormat
 import to.bitkit.models.WalletScope
+import to.bitkit.services.PaykitSdkOperationLock
 import to.bitkit.services.PaykitSdkService
 import to.bitkit.utils.Logger
 import to.bitkit.utils.ServiceError
@@ -127,6 +130,8 @@ class PaykitPaymentProofRepo @Inject constructor(
     }
 
     private val operationMutex = Mutex()
+    private val persistenceMutex = Mutex()
+    private val deliveryScope = appScope(ioDispatcher, TAG)
 
     // Process-local activity handoffs only; restored proofs must reconstruct their activity.
     private val observedHardwareProofs = mutableSetOf<PendingPaykitPaymentProof>()
@@ -158,7 +163,7 @@ class PaykitPaymentProofRepo @Inject constructor(
     )
 
     suspend fun backupSnapshot(): List<PaykitPaymentStateBackup.Proof> = withContext(ioDispatcher) {
-        operationMutex.withLock {
+        persistenceMutex.withLock {
             val proofs = store.load()
             check(proofs.none { it.hardwareDispatchDenied }) { "Hardware dispatch denial cleanup is pending" }
             proofs.map { PaykitPaymentStateBackup.Proof(it) }
@@ -187,12 +192,28 @@ class PaykitPaymentProofRepo @Inject constructor(
             val onchainAttempt = lightningRepo.currentOnchainSendAttempt()
             if (onchainAttempt?.requestId == request.id) throw PaykitPaymentRequestError.OperationInProgress
             operationMutex.withLock {
-                val proof = pendingProof(
-                    request,
-                    paymentEndpointIdentifier,
-                    paymentAppId,
-                    kind
-                ).copy(conversionQuoteId = paymentTerms?.quoteId)
+                if (
+                    paymentAppId.isBlank() ||
+                    paymentEndpointIdentifier !in request.acceptedPaymentEndpointIdentifiers ||
+                    !endpointSupports(paymentEndpointIdentifier, kind)
+                ) {
+                    throw PaykitPaymentRequestError.RequestUnavailable
+                }
+                val read = paykitSdkService.readPaymentRequests(request.counterparty)
+                val identityStatus = read.identityStatus
+                val identity = identityStatus?.publicKey?.let(PubkyPublicKeyFormat::normalized)
+                if (identityStatus?.capability != PubkyIdentityCapability.PRIVATE_LINK_CAPABLE || identity == null) {
+                    throw PaykitPaymentRequestError.RequestUnavailable
+                }
+                val proof = PendingPaykitPaymentProof(
+                    identity = identity,
+                    requestId = request.id,
+                    paymentEndpointIdentifier = paymentEndpointIdentifier,
+                    paymentAppId = paymentAppId,
+                    kind = kind,
+                    billingPeriod = request.billingPeriod,
+                    conversionQuoteId = paymentTerms?.quoteId,
+                )
                 val currentProofs = loadProofs()
                 if (currentProofs.any {
                         it.isStartedFor(proof.identity, request.id) ||
@@ -205,7 +226,7 @@ class PaykitPaymentProofRepo @Inject constructor(
                 ) {
                     throw PaykitPaymentRequestError.OperationInProgress
                 }
-                val alreadyPaid = paykitSdkService.paymentRequests().any { record ->
+                val alreadyPaid = read.paymentRequests.any { record ->
                     record.paymentRequestId == request.paymentRequestId &&
                         PubkyPublicKeyFormat.matches(record.counterparty, request.counterparty) &&
                         record.paymentProofs.any { it.billingPeriod.matches(request.billingPeriod) }
@@ -851,6 +872,9 @@ class PaykitPaymentProofRepo @Inject constructor(
     suspend fun retainedHardwareOnchainRequest(request: PaykitPaymentRequest): PendingPaykitPaymentProof? =
         withContext(ioDispatcher) {
             operationMutex.withLock {
+                if (loadProofs().none { it.requestId == request.id && it.hardwareSignedTransaction != null }) {
+                    return@withLock null
+                }
                 val identity = currentIdentity() ?: return@withLock null
                 loadProofs().singleOrNull {
                     it.isReopenableHardwareReceipt(identity) && it.matchesHardwareRetryRequest(request)
@@ -1381,20 +1405,24 @@ class PaykitPaymentProofRepo @Inject constructor(
         operationMutex.withLock { runSuspendCatching { submitReady(proof) } }
     }
 
-    private suspend fun submitReady(proof: PendingPaykitPaymentProof): Boolean {
+    private suspend fun submitReady(
+        proof: PendingPaykitPaymentProof,
+        priority: PaykitSdkOperationLock.Priority = PaykitSdkOperationLock.Priority.Ordered,
+    ): Boolean {
         // Missing legacy app provenance must remain readable, never guessed for delivery.
         if (proof.paymentAppId.isBlank()) return false
         if (proof.kind == PaykitPaymentProofKind.Onchain && proof.onchainAcceptanceVerified != true) return false
         val proofData = proof.proofData ?: return false
-        val identityStatus = paykitSdkService.identityStatus()
+        val read = paykitSdkService.readPaymentRequests(proof.requestId.counterparty, expectedIdentity = proof.identity)
+        val identityStatus = read.identityStatus
         if (
             identityStatus?.capability != PubkyIdentityCapability.PRIVATE_LINK_CAPABLE ||
             !PubkyPublicKeyFormat.matches(identityStatus.publicKey, proof.identity)
         ) {
             return false
         }
 
-        val record = paykitSdkService.paymentRequests().firstOrNull {
+        val record = read.paymentRequests.firstOrNull {
             it.paymentRequestId == proof.requestId.paymentRequestId &&
                 PubkyPublicKeyFormat.matches(it.counterparty, proof.requestId.counterparty)
         } ?: return false
@@ -1409,16 +1437,21 @@ class PaykitPaymentProofRepo @Inject constructor(
                 proofJson = proofJson,
                 conversionQuoteId = proof.conversionQuoteId,
                 billingPeriod = proof.billingPeriod,
+                priority = priority,
             )
             Logger.info("Queued a Paykit payment proof for private delivery", context = TAG)
-            runSuspendCatching { paykitSdkService.processPendingPrivateMessages() }
-                .onFailure {
-                    Logger.warn(
-                        "Paykit payment proof remains queued for private delivery",
-                        it,
-                        context = TAG,
-                    )
-                }
+        }
+        // Submission has durably queued the proof; transport does not need the local proof-state lock.
+        deliveryScope.launch {
+            runSuspendCatching {
+                paykitSdkService.processOutboundPrivateMessages(
+                    proof.requestId.counterparty,
+                    PaykitSdkOperationLock.Priority.Background,
+                    expectedIdentity = proof.identity,
+                )
+            }.onFailure {
+                Logger.warn("Retained Paykit payment proof for private delivery", it, context = TAG)
+            }
         }
         if (proof.kind != PaykitPaymentProofKind.Usdt) {
             runSuspendCatching {
@@ -1485,7 +1518,31 @@ class PaykitPaymentProofRepo @Inject constructor(
                     context = TAG,
                 )
             }.isSuccess
-        if (didPersist && !awaitDelivery) return true
+        if (didPersist && !awaitDelivery) {
+            deliveryScope.launch {
+                operationMutex.withLock {
+                    runSuspendCatching {
+                        for (proof in completedProofs) {
+                            if (proof in loadProofs()) {
+                                val priority = if (
+                                    proof.kind == PaykitPaymentProofKind.Lightning ||
+                                    (
+                                        proof.kind == PaykitPaymentProofKind.Onchain &&
+                                            proof.onchainWalletId == WalletScope.default
+                                        )
+                                ) {
+                                    PaykitSdkOperationLock.Priority.Interactive
+                                } else {
+                                    PaykitSdkOperationLock.Priority.Ordered
+                                }
+                                submitReady(proof, priority)
+                            }
+                        }
+                    }.onFailure { Logger.warn("Failed to queue a Paykit payment proof", it, context = TAG) }
+                }
+            }
+            return true
+        }
         var hasUndeliveredProof = false
         completedProofs.forEach { proof ->
             val wasDelivered = runSuspendCatching { submitReady(proof) }
@@ -1550,7 +1607,8 @@ class PaykitPaymentProofRepo @Inject constructor(
     }
 
     private suspend fun persist(proofs: List<PendingPaykitPaymentProof>) {
-        store.save(proofs)
+        // Snapshots wait for durable local writes, not the operation's remote SDK calls.
+        persistenceMutex.withLock { store.save(proofs) }
         observedHardwareProofs.retainAll(proofs.toSet())
     }
 }

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

@@ -367,6 +367,8 @@ class PaykitPaymentRequestRepo @Inject constructor(
 
     fun setPaymentSubmissionActive(active: Boolean) = paykitSdkService.setPaymentSubmissionActive(active)
 
+    fun beginPaymentReview(): () -> Unit = paykitSdkService.beginPaymentReview()
+
     @Volatile
     private var activeIdentity: String? = null
 
@@ -565,7 +567,7 @@ class PaykitPaymentRequestRepo @Inject constructor(
     private suspend fun refresh(
         mode: PaykitPaymentRequestRefreshMode,
         forceFresh: Boolean,
-        messagePriority: Priority = Priority.Ordered,
+        messagePriority: Priority = Priority.Background,
     ): Result<Unit> {
         val generation = stateGeneration.get()
         val expectedIdentity = activeIdentity
@@ -589,7 +591,13 @@ class PaykitPaymentRequestRepo @Inject constructor(
                     } ?: return@refresh
                     val snapshot = runSuspendCatching {
                         val priority = if (forceFresh) Priority.Ordered else Priority.Background
-                        fetchRequestSnapshot(expectedIdentity, mode, priority, messagePriority, observationOnly = true)
+                        fetchRequestSnapshot(
+                            expectedIdentity,
+                            mode,
+                            priority,
+                            if (forceFresh) Priority.Ordered else messagePriority,
+                            observationOnly = true,
+                        )
                     }
                     operationMutex.withLock {
                         if (!isCurrentState(generation, expectedIdentity)) return@withLock
@@ -612,7 +620,9 @@ class PaykitPaymentRequestRepo @Inject constructor(
                     }
                 }
             }.onFailure {
-                Logger.warn("Failed to refresh incoming Paykit payment requests", it, context = TAG)
+                if (it != PaykitPaymentRequestError.OperationInProgress) {
+                    Logger.warn("Failed to refresh incoming Paykit payment requests", it, context = TAG)
+                }
             }
         }
     }
@@ -1065,7 +1075,9 @@ class PaykitPaymentRequestRepo @Inject constructor(
                     }
                     val identity = activeIdentity ?: throw PaykitPaymentRequestError.RequestUnavailable
                     val generation = stateGeneration.get()
-                    ensurePaymentAllowed(it, forExecution = false).getOrThrow()
+                    if (it.isExpired(clock.now())) throw PaykitPaymentRequestError.RequestExpired
+                    if (!isLocallyPayable(it)) throw PaykitPaymentRequestError.RequestUnavailable
+                    // The SDK checks the current peer while atomically claiming and queueing acceptance.
                     val alreadySaved = it.id in presentationStore.loadAcceptedOneTimeIds(identity)
                     val acceptedIds = presentationStore.addAcceptedOneTimeId(identity, it.id)
                     if (!isCurrentState(generation, identity)) throw PaykitPaymentRequestError.RequestUnavailable
@@ -1701,7 +1713,10 @@ class PaykitPaymentRequestRepo @Inject constructor(
             }
         }
             .onSuccess(::logOutboundFailures)
-            .onFailure { Logger.warn("Failed to deliver pending Paykit private messages", it, context = TAG) }
+            .onFailure {
+                if (it == PaykitPaymentRequestError.OperationInProgress) throw it
+                Logger.warn("Failed to deliver pending Paykit private messages", it, context = TAG)
+            }
             .getOrDefault(emptyList())
 
     private fun logOutboundFailures(reports: List<OutboundPrivateCounterpartySendReport>) {

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

@@ -1645,26 +1645,21 @@ class PrivatePaykitRepo @Inject constructor(
 
     private suspend fun readPrivateMessageDrainState(priority: suspend () -> Priority): PrivateMessageDrainState? {
         val generation = preparationGeneration
-        val linkedPeers = runSuspendCatching { paykitSdkService.linkedPeers(priority()) }
+        val snapshot = runSuspendCatching { paykitSdkService.privateMessageSchedulingSnapshot(priority()) }
             .getOrElse {
-                Logger.warn("Failed to inspect private Paykit link state", it, context = TAG)
+                Logger.warn("Failed to inspect private Paykit scheduling state", it, context = TAG)
                 return null
             }
+        priority()
+        if (generation != preparationGeneration) return null
+        val linkedPeers = snapshot.linkedPeers
             .mapNotNull { peer ->
                 normalizedPublicKey(peer.counterparty)?.let { publicKey ->
                     publicKey to peer.state
                 }
             }
             .toMap()
-        val pendingPriority = priority()
-        if (generation != preparationGeneration) return null
-        val pendingOutbound = runSuspendCatching {
-            paykitSdkService.pendingOutboundPrivateCounterparties(pendingPriority)
-        }
-            .getOrElse {
-                Logger.warn("Failed to inspect pending private Paykit messages", it, context = TAG)
-                return null
-            }
+        val pendingOutbound = snapshot.pendingOutboundCounterparties
             .mapNotNull(::normalizedPublicKey)
             .toSet()
 

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

@@ -101,7 +101,7 @@ sealed class PubkyContactError(message: String) : AppError(message) {
  */
 class PubkySignIn internal constructor(val publicKey: String, internal val generation: Long)
 
-private fun Throwable.hasCause(error: PubkyContactError): Boolean =
+private fun Throwable.hasCause(error: AppError): Boolean =
     generateSequence(this) { it.cause }.any { it == error }
 
 private fun Throwable.isPaykitReadTimeout(): Boolean =
@@ -151,6 +151,9 @@ class PubkyRepo @Inject constructor(
         /** Shortest time between two contact list updates of a background profile refresh. */
         internal val CONTACT_REFRESH_BATCH_WINDOW = 300.milliseconds
 
+        /** Delay before retrying a contact observation deferred by payment review or submission. */
+        private val CONTACT_LOAD_RETRY_INTERVAL = 500.milliseconds
+
         /** Maximum automatic restore attempts during one foreground recovery window. */
         private const val DEFERRED_RESTORE_ATTEMPTS = 8
 
@@ -1240,8 +1243,13 @@ class PubkyRepo @Inject constructor(
                     loaded = true
                     shouldMarkLoadCompleted = true
                 }.onFailure {
-                    shouldMarkLoadCompleted = isCurrent(signIn)
-                    Logger.error("Failed to load contacts", it, context = TAG)
+                    if (it.hasCause(PaykitPaymentRequestError.OperationInProgress)) {
+                        reload = isCurrent(signIn)
+                        if (reload) delay(CONTACT_LOAD_RETRY_INTERVAL)
+                    } else {
+                        shouldMarkLoadCompleted = isCurrent(signIn)
+                        Logger.error("Failed to load contacts", it, context = TAG)
+                    }
                 }
             } while (reload && isCurrent(signIn))
         } finally {