fix: keep incoming request refreshes together

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

@@ -612,6 +612,7 @@ class PaykitPaymentRequestRepo @Inject constructor(
                             priority,
                             if (forceFresh) priority else messagePriority,
                             observationOnly = true,
+                            combineReceiveObservation = !forceFresh,
                         )
                     }
                     operationMutex.withLock {
@@ -1285,8 +1286,21 @@ class PaykitPaymentRequestRepo @Inject constructor(
         priority: Priority = Priority.Ordered,
         messagePriority: Priority = Priority.Ordered,
         observationOnly: Boolean = false,
+        combineReceiveObservation: Boolean = false,
     ): RequestSnapshot {
         if (mode == PaykitPaymentRequestRefreshMode.FULL) processPendingMessages(priority = messagePriority)
+        if (combineReceiveObservation && observationOnly && mode != PaykitPaymentRequestRefreshMode.STORED &&
+            priority == Priority.Background && messagePriority == Priority.Background
+        ) {
+            val observation = paykitSdkService.receiveAndObserveState(
+                requireNotNull(expectedIdentity),
+                ::logIntakeFailures,
+            )
+            return RequestSnapshot(
+                observation.paymentRequests,
+                observation.linkedPeers.filter { it.state == LinkedPeerState.BLOCKED },
+            )
+        }
         if (mode != PaykitPaymentRequestRefreshMode.STORED) {
             paykitSdkService.receivePrivateMessagesFromLinkedPeers(messagePriority).also(::logIntakeFailures)
         }

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

@@ -983,6 +983,50 @@ class PaykitSdkService @Inject constructor(
         }
     }
 
+    internal suspend fun receiveAndObserveState(
+        expectedIdentity: String,
+        onReceive: (List<PrivateStreamCounterpartyIntakeReport>) -> Unit,
+    ): PaykitStateObservation {
+        isSetup.await()
+        val generation = runtimeGeneration
+        return operationLock.withLock(Priority.Background) {
+            check(runtimeGeneration == generation) { "Paykit runtime changed before inbox refresh" }
+            checkRefreshAdmission(Priority.Background)
+            withPaykitKey { handle ->
+                val key = cachedPaykitKey
+                check(key == null || PubkyPublicKeyFormat.matches(key.publicKey, expectedIdentity)) {
+                    "Paykit inbox identity changed"
+                }
+                val reports = withStateRevisionTracking(
+                    handle,
+                    beforeOperation = {
+                        operationLock.withoutLock {
+                            check(runtimeGeneration == generation && sdk === handle && cachedPaykitKey == key) {
+                                "Paykit runtime changed before inbox receive"
+                            }
+                            checkRefreshAdmission(Priority.Background)
+                        }
+                    },
+                ) { completeSdkCall { handle.receivePrivateMessagesFromLinkedPeers() } }
+                onReceive(reports)
+                operationLock.withoutLock {
+                    check(runtimeGeneration == generation && sdk === handle && cachedPaykitKey == key) {
+                        "Paykit runtime changed during inbox refresh"
+                    }
+                    checkRefreshAdmission(Priority.Background)
+                    val observation = completeSdkCall { handle.observeState() }
+                    check(PubkyPublicKeyFormat.matches(observation.publicKey, expectedIdentity)) {
+                        "Paykit observation identity changed"
+                    }
+                    key?.let {
+                        check(observation.keyGeneration == it.generation) { "Paykit observation key changed" }
+                    }
+                    observation
+                }
+            }
+        }
+    }
+
     suspend fun receivePrivateMessages(counterparty: String) = receivePrivateMessages(counterparty, Priority.Ordered)
 
     internal suspend fun receivePrivateMessages(counterparty: String, priority: Priority) = run {
@@ -1771,32 +1815,38 @@ class PaykitSdkService @Inject constructor(
         beforeOperation: suspend () -> Unit = {},
         block: suspend (PaykitSdk) -> T,
     ): T = withPaykitKey { handle ->
-        withPaykitBackupStateTracking(
-            readRevision = {
-                val started = TimeSource.Monotonic.markNow()
-                try {
-                    runSuspendCatching { completeSdkCall { handle.backupStateRevision() } }
-                        .onFailure(::invalidatePaykitKeyIfNeeded)
-                        .getOrThrow()
-                } finally {
-                    if (BuildConfig.DEBUG) {
-                        Logger.debug("Read Paykit backup revision duration='${started.elapsedNow()}'", context = TAG)
-                    }
-                }
-            },
-            readStateRevision = { handle.stateRevision() },
-            readObservedSnapshot = {
-                handle.observedBackupStateRevision()?.let {
-                    PaykitBackupStateSnapshot(it.stateRevision, it.backupRevision)
+        withStateRevisionTracking(handle, beforeOperation) { block(handle) }
+    }
+
+    private suspend fun <T> withStateRevisionTracking(
+        handle: PaykitSdk,
+        beforeOperation: suspend () -> Unit,
+        block: suspend () -> T,
+    ): T = withPaykitBackupStateTracking(
+        readRevision = {
+            val started = TimeSource.Monotonic.markNow()
+            try {
+                runSuspendCatching { completeSdkCall { handle.backupStateRevision() } }
+                    .onFailure(::invalidatePaykitKeyIfNeeded)
+                    .getOrThrow()
+            } finally {
+                if (BuildConfig.DEBUG) {
+                    Logger.debug("Read Paykit backup revision duration='${started.elapsedNow()}'", context = TAG)
                 }
-            },
-            cachedSnapshot = cachedBackupState,
-            onSnapshot = { cachedBackupState = it },
-            onChange = ::notifyBackupStateChanged,
-            beforeOperation = beforeOperation,
-        ) {
-            block(handle)
-        }
+            }
+        },
+        readStateRevision = { handle.stateRevision() },
+        readObservedSnapshot = {
+            handle.observedBackupStateRevision()?.let {
+                PaykitBackupStateSnapshot(it.stateRevision, it.backupRevision)
+            }
+        },
+        cachedSnapshot = cachedBackupState,
+        onSnapshot = { cachedBackupState = it },
+        onChange = ::notifyBackupStateChanged,
+        beforeOperation = beforeOperation,
+    ) {
+        block()
     }
 
     private suspend fun handle(): PaykitSdk = handleMutex.withLock {

app/src/main/java/to/bitkit/viewmodels/AppViewModel.kt

@@ -964,7 +964,11 @@ class AppViewModel @Inject constructor(
             refreshPrivateOnlyPaykitApp("contact sync", onlyIfNeeded = !identityChanged)
             if (!state.contactsLoaded || !synchronizeSavedPaykitContacts(state.contactKeys, isStillCurrent)) return
             if (isPaymentRequestPollingStopped.value) return
-            val result = refreshIncomingPaykitPaymentRequests(forceFresh = true, onlyWhileActive = true)
+            val result = refreshIncomingPaykitPaymentRequests(
+                forceFresh = true,
+                refreshPriority = Priority.Background,
+                onlyWhileActive = true,
+            )
             if (!isPaymentRequestPollingStopped.value) refreshPaymentRequestTargets(force = true)
             result
         } finally {
@@ -1045,7 +1049,11 @@ class AppViewModel @Inject constructor(
             isStillCurrent = isStillCurrent,
         )
         if (isPaymentRequestPollingStopped.value) return
-        refreshIncomingPaykitPaymentRequests(forceFresh = true, onlyWhileActive = true)
+        refreshIncomingPaykitPaymentRequests(
+            forceFresh = true,
+            refreshPriority = Priority.Background,
+            onlyWhileActive = true,
+        )
         if (!isPaymentRequestPollingStopped.value) refreshPaymentRequestTargets(force = true)
     }
 

app/src/test/java/to/bitkit/repositories/PaykitPaymentRequestRepoSubscriptionTest.kt

@@ -21,6 +21,7 @@ import com.synonym.paykit.PaymentRequestRecord
 import com.synonym.paykit.PaymentRequestRecurrence
 import com.synonym.paykit.PaymentRequestTerms
 import com.synonym.paykit.PrivateJsonObject
+import com.synonym.paykit.PrivateStreamCounterpartyIntakeReport
 import com.synonym.paykit.PubkyIdentityCapability
 import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.ExperimentalCoroutinesApi
@@ -53,6 +54,7 @@ import to.bitkit.models.PaykitAmount
 import to.bitkit.models.PaykitAsset
 import to.bitkit.models.PaykitRequestPricing
 import to.bitkit.services.PaykitPaymentRequestProposalTerms
+import to.bitkit.services.PaykitSdkOperationLock.Priority
 import to.bitkit.services.PaykitSdkService
 import to.bitkit.test.BaseUnitTest
 import kotlin.test.assertEquals
@@ -116,6 +118,11 @@ class PaykitPaymentRequestRepoSubscriptionTest : BaseUnitTest(StandardTestDispat
         whenever(paykitSdkService.receivePrivateMessagesFromLinkedPeers(any())).doSuspendableAnswer {
             paykitSdkService.receivePrivateMessagesFromLinkedPeers()
         }
+        whenever(paykitSdkService.receiveAndObserveState(any(), any())).doSuspendableAnswer {
+            val reports = paykitSdkService.receivePrivateMessagesFromLinkedPeers(Priority.Background)
+            it.getArgument<(List<PrivateStreamCounterpartyIntakeReport>) -> Unit>(1)(reports)
+            paykitSdkService.observeState(it.getArgument(0), Priority.Background)
+        }
         stubRequestReads()
         whenever(paykitSdkService.processOutboundPrivateMessages(any(), any())).doSuspendableAnswer {
             paykitSdkService.processOutboundPrivateMessages(it.getArgument(0))

app/src/test/java/to/bitkit/repositories/PaykitPaymentRequestRepoTest.kt

@@ -316,6 +316,11 @@ class PaykitPaymentRequestRepoTest : BaseUnitTest(StandardTestDispatcher()) {
         whenever(paykitSdkService.receivePrivateMessagesFromLinkedPeers(any())).doSuspendableAnswer {
             paykitSdkService.receivePrivateMessagesFromLinkedPeers()
         }
+        whenever(paykitSdkService.receiveAndObserveState(any(), any())).doSuspendableAnswer {
+            val reports = paykitSdkService.receivePrivateMessagesFromLinkedPeers(Priority.Background)
+            it.getArgument<(List<PrivateStreamCounterpartyIntakeReport>) -> Unit>(1)(reports)
+            paykitSdkService.observeState(it.getArgument(0), Priority.Background)
+        }
         stubRequestReads()
         whenever(paykitSdkService.processOutboundPrivateMessages(any(), any())).doSuspendableAnswer {
             paykitSdkService.processOutboundPrivateMessages(it.getArgument(0))
@@ -497,6 +502,58 @@ class PaykitPaymentRequestRepoTest : BaseUnitTest(StandardTestDispatcher()) {
         verify(paykitSdkService, never()).linkedPeers(any())
         verify(paykitSdkService).processPendingPrivateMessages(Priority.Background)
         verify(paykitSdkService, times(2)).receivePrivateMessagesFromLinkedPeers(Priority.Background)
+        verify(paykitSdkService, times(2)).receiveAndObserveState(eq(LOCAL_IDENTITY), any())
+    }
+
+    @Test
+    fun `forced and non-background refresh retain separate receive and observation`() = test {
+        for (mode in listOf(PaykitPaymentRequestRefreshMode.INBOX, PaykitPaymentRequestRefreshMode.FULL)) {
+            for (priority in listOf(Priority.Background, Priority.Ordered, Priority.Interactive)) {
+                clearInvocations(paykitSdkService)
+                if (priority == Priority.Background) {
+                    sut.refreshAfterStateChange(mode, priority).getOrThrow()
+                } else {
+                    sut.refresh(mode, priority).getOrThrow()
+                }
+
+                verify(paykitSdkService).receivePrivateMessagesFromLinkedPeers(priority)
+                verify(paykitSdkService).observeState(LOCAL_IDENTITY, Priority.Background)
+                verify(paykitSdkService, never()).receiveAndObserveState(any(), any())
+            }
+        }
+    }
+
+    @Test
+    fun `automatic full refresh finishes pending drain before receive observation and stops on deferral`() = test {
+        for (deferred in listOf(false, true)) {
+            clearInvocations(paykitSdkService)
+            val release = CompletableDeferred<Unit>()
+            doSuspendableAnswer {
+                release.await()
+                if (deferred) throw PaykitPaymentRequestError.OperationInProgress
+                emptyList<OutboundPrivateCounterpartySendReport>()
+            }.whenever(paykitSdkService).processPendingPrivateMessages(Priority.Background)
+            val refresh = async { sut.refresh() }
+            runCurrent()
+            verify(paykitSdkService).processPendingPrivateMessages(Priority.Background)
+            verify(paykitSdkService, never()).receiveAndObserveState(any(), any())
+            verify(paykitSdkService, never()).receivePrivateMessagesFromLinkedPeers(any())
+            verify(paykitSdkService, never()).observeState(any(), any())
+            release.complete(Unit)
+            val result = refresh.await()
+
+            if (deferred) {
+                assertSame(PaykitPaymentRequestError.OperationInProgress, result.exceptionOrNull())
+                verify(paykitSdkService, never()).receiveAndObserveState(any(), any())
+                verify(paykitSdkService, never()).receivePrivateMessagesFromLinkedPeers(any())
+                verify(paykitSdkService, never()).observeState(any(), any())
+            } else {
+                result.getOrThrow()
+                verify(paykitSdkService).receiveAndObserveState(eq(LOCAL_IDENTITY), any())
+                verify(paykitSdkService).receivePrivateMessagesFromLinkedPeers(Priority.Background)
+                verify(paykitSdkService).observeState(LOCAL_IDENTITY, Priority.Background)
+            }
+        }
     }
 
     @Test
@@ -1134,6 +1191,36 @@ class PaykitPaymentRequestRepoTest : BaseUnitTest(StandardTestDispatcher()) {
         }
     }
 
+    @Test
+    fun `inbox reports intake failures once before observation fails and keeps warm requests`() = test {
+        for (failure in listOf(
+            PaykitPaymentRequestError.OperationInProgress,
+            PaykitException.SharedStateBusy("shared_state_busy", "Locked"),
+        )) {
+            stubRequestReads()
+            val record = paymentRequestRecord()
+            whenever(paykitSdkService.allPaymentRequests(anyOrNull())).thenReturn(listOf(record))
+            sut.refresh(PaykitPaymentRequestRefreshMode.STORED).getOrThrow()
+            val pending = sut.pendingRequests.value
+            val intakeError = mock<PrivateOperationError> {
+                on { redactedContext() } doReturn "transport failure"
+            }
+            whenever(paykitSdkService.receivePrivateMessagesFromLinkedPeers()).thenReturn(
+                listOf(PrivateStreamCounterpartyIntakeReport(COUNTERPARTY, null, intakeError)),
+            )
+            whenever(paykitSdkService.observeState(any(), any())).doSuspendableAnswer { throw failure }
+            clearInvocations(paykitSdkService)
+
+            val result = sut.refresh(PaykitPaymentRequestRefreshMode.INBOX)
+
+            assertSame(failure, result.exceptionOrNull())
+            assertEquals(pending, sut.pendingRequests.value)
+            verify(intakeError).redactedContext()
+            verify(paykitSdkService).receiveAndObserveState(eq(LOCAL_IDENTITY), any())
+            verify(paykitSdkService).receivePrivateMessagesFromLinkedPeers()
+        }
+    }
+
     @Test
     fun `peer intake failure does not drop received requests`() = test {
         val record = paymentRequestRecord()