fix(upgrade): Recover billing from stale purchase data and mid-flow errors

This commit is contained in:
Matthias Urhahn
2026-07-11 12:15:17 +02:00
committed by GitHub
parent 778d45170f
commit bd36c5e5d6
5 changed files with 308 additions and 24 deletions
@@ -1,6 +1,7 @@
package eu.darken.capod.common.upgrade.core
import android.app.Activity
import com.android.billingclient.api.BillingClient
import eu.darken.capod.common.coroutine.AppScope
import eu.darken.capod.common.debug.logging.Logging.Priority.INFO
import eu.darken.capod.common.debug.logging.Logging.Priority.VERBOSE
@@ -8,6 +9,7 @@ import eu.darken.capod.common.debug.logging.Logging.Priority.WARN
import eu.darken.capod.common.debug.logging.asLog
import eu.darken.capod.common.debug.logging.log
import eu.darken.capod.common.debug.logging.logTag
import eu.darken.capod.common.flow.setupCommonEventHandlers
import eu.darken.capod.common.upgrade.UpgradeRepo
import eu.darken.capod.common.upgrade.core.client.ItemAlreadyOwnedBillingException
import eu.darken.capod.common.upgrade.core.data.BillingData
@@ -20,7 +22,11 @@ import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.SharingStarted
import kotlinx.coroutines.flow.catch
import kotlinx.coroutines.flow.distinctUntilChanged
import kotlinx.coroutines.flow.filter
import kotlinx.coroutines.flow.launchIn
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.flow.onEach
import kotlinx.coroutines.flow.onStart
import kotlinx.coroutines.flow.shareIn
import kotlinx.coroutines.withTimeoutOrNull
@@ -47,6 +53,44 @@ class UpgradeRepoGplay @Inject constructor(
private val anchorLock = Any()
init {
// Fresh-provenance grace stamping: freshBillingData carries every successful query result
// and push payload as an event — unlike the equality-deduped billingData state, an
// unchanged steady-owner query still stamps, and stale listener data can't sneak in.
// The reactive upgradeInfo mapping deliberately writes nothing anymore.
billingDataRepo.freshBillingData
.onEach { data ->
try {
recordProState(data)
} catch (e: CancellationException) {
throw e
} catch (e: Exception) {
// A failed DataStore write must not kill this process-lifetime collector.
log(TAG, WARN) { "Failed to record pro state: ${e.asLog()}" }
}
}
.setupCommonEventHandlers(TAG) { "proStateRecorder" }
.launchIn(scope)
// Async variant of the launch-result ITEM_ALREADY_OWNED case: Play told us mid-flow that
// the user already owns it. Reconcile silently — Play shows its own UI for purchase-sheet
// failures, so no app-side dialog here.
billingDataRepo.purchaseFailures
.filter { it.responseCode == BillingClient.BillingResponseCode.ITEM_ALREADY_OWNED }
.onEach {
log(TAG, INFO) { "Async already-owned event -> restoring purchase" }
try {
withTimeoutOrNull(RESTORE_ON_OWNED_TIMEOUT_MS) { restorePurchaseNow() }
} catch (e: CancellationException) {
throw e
} catch (e: Exception) {
log(TAG, WARN) { "Async already-owned restore failed: ${e.asLog()}" }
}
}
.setupCommonEventHandlers(TAG) { "asyncAlreadyOwned" }
.launchIn(scope)
}
// Grace window depends on what was last owned: a permanent one-time purchase should almost
// never be dropped on a Play hiccup, so it gets a long window; a subscription legitimately
// lapses, so it keeps the short one (also used for unknown/legacy last SKUs).
@@ -72,7 +116,9 @@ class UpgradeRepoGplay @Inject constructor(
// True once we've ever confirmed a known Pro purchase on this install; drives the proactive
// restore banner. Local signal only — a fresh install or switched Google account starts false.
val wasEverPro: Flow<Boolean> = billingCache.lastProStateAt.flow.map { it > 0 }
val wasEverPro: Flow<Boolean> = billingCache.lastProStateAt.flow
.map { it > 0 }
.distinctUntilChanged()
// Explicit "Restore purchase": query Play now and evaluate Pro from the returned data in the
// same coroutine (real happens-before), so we never read a stale upgradeInfo replay. Billing
@@ -80,7 +126,11 @@ class UpgradeRepoGplay @Inject constructor(
suspend fun restorePurchaseNow(): Info {
log(TAG) { "restorePurchaseNow()" }
return try {
billingDataRepo.refresh().toUpgradeInfo()
val data = billingDataRepo.refresh()
// Returned data is fresh by definition — stamp it even if the flows dedupe the
// unchanged result and the init collector never sees a new emission.
recordProState(data)
data.toUpgradeInfo()
} catch (e: CancellationException) {
throw e
} catch (e: Exception) {
@@ -97,30 +147,15 @@ class UpgradeRepoGplay @Inject constructor(
}
// Shared Pro/grace mapping used by both the reactive upgradeInfo flow and restorePurchaseNow().
// Only relinquishes Pro if we haven't had it for a while (grace period).
// Only relinquishes Pro if we haven't had it for a while (grace period). READ-ONLY: this also
// runs on replayed shared-flow data, so it must never stamp the grace cache — a refunded
// purchase could otherwise keep re-stamping its own grace window. See recordProState().
private fun BillingData?.toUpgradeInfo(): Info {
val now = System.currentTimeMillis()
val proSku = this?.getProSku()
log(TAG) { "toUpgradeInfo(): now=$now, lastProStateAt=$lastProStateAt, data=$this" }
return when {
proSku != null -> {
val upgrades = this!!.getProSkus()
// Record which Pro SKU anchors the grace window; the permanent IAP wins when both
// are owned. An IAP anchor is sticky: purchase data may lack the IAP because that
// query failed on a fresh connection, and a subscription seen in the meantime must
// not shrink the 30d window of an owner whose IAP was never disproven. Trade-off:
// a refunded IAP keeps the long window — consistent with the fail-open cache.
// SKU first, timestamp last — the timestamp is the gate, so a crash between the
// two writes stays conservative. Locked: mappings run concurrently from the hot
// reactive flow and direct restores, and the sticky check-then-write must not race.
synchronized(anchorLock) {
preferredProSku(upgrades)
?.takeIf { it.type == Sku.Type.IAP || !anchorIsIap() }
?.let { lastProStateSku = it.id }
lastProStateAt = now
}
Info(billingData = this, upgrades = upgrades)
}
proSku != null -> Info(billingData = this, upgrades = this!!.getProSkus())
(now - lastProStateAt) < graceWindowMs() -> {
log(TAG, VERBOSE) { "We are not pro, but were recently, did GPlay try annoy us again?" }
@@ -131,6 +166,26 @@ class UpgradeRepoGplay @Inject constructor(
}
}
// Persists "we saw a known Pro purchase" for the grace machinery. Callers must only pass FRESH
// data (returned query results, or new emissions seen by the init collector) — never replayed
// flow data. The permanent IAP wins as anchor when both are owned, and an IAP anchor is sticky:
// purchase data may lack the IAP because that query failed on a fresh connection, and a
// subscription seen in the meantime must not shrink the 30d window of an owner whose IAP was
// never disproven (trade-off: a refunded IAP keeps the long window — consistent with the
// fail-open cache). SKU before timestamp — the timestamp is the gate, so a crash between the
// two writes stays conservative. Locked: runs concurrently from the init collector and direct
// restores, and the sticky check-then-write must not race.
private fun recordProState(data: BillingData) {
val upgrades = data.getProSkus()
val preferred = preferredProSku(upgrades) ?: return
synchronized(anchorLock) {
preferred
.takeIf { it.type == Sku.Type.IAP || !anchorIsIap() }
?.let { lastProStateSku = it.id }
lastProStateAt = System.currentTimeMillis()
}
}
data class Info(
private val gracePeriod: Boolean = false,
private val billingData: BillingData?,
@@ -26,6 +26,7 @@ import kotlinx.coroutines.async
import kotlinx.coroutines.coroutineScope
import kotlinx.coroutines.withContext
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.combine
import kotlinx.coroutines.flow.update
@@ -35,7 +36,18 @@ import kotlinx.coroutines.sync.withLock
data class BillingClientConnection(
private val client: BillingClient,
private val purchasesGlobal: Flow<Collection<Purchase>>,
private val freshObservations: MutableSharedFlow<Collection<Purchase>>,
private val purchaseFailuresGlobal: Flow<BillingResult>,
) {
// Non-OK results from onPurchasesUpdated (e.g. async ITEM_ALREADY_OWNED after the Play sheet
// opened). Consumed by a single persistent collector in UpgradeRepoGplay — not an event bus.
val purchaseFailures: Flow<BillingResult> = purchaseFailuresGlobal
// Every conclusive fresh look at PURCHASED purchases (successful queries and push payloads),
// regardless of whether it differs from the previous one — the combined `purchases` state is
// equality-deduped and can mix in stale listener data, so grace stamping must not use it.
val freshPurchases: Flow<Collection<Purchase>> = freshObservations
private data class QueryCaches(
val iaps: Collection<Purchase>? = null,
val subs: Collection<Purchase>? = null,
@@ -96,6 +108,10 @@ data class BillingClientConnection(
)
}
// A conclusive refresh is a fresh observation for the grace stamping, even when the result
// equals the previous one and the state flows dedupe it away.
freshObservations.tryEmit(combined)
combined
}
@@ -14,14 +14,17 @@ import eu.darken.capod.common.debug.logging.log
import eu.darken.capod.common.debug.logging.logTag
import eu.darken.capod.common.flow.setupCommonEventHandlers
import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.channels.BufferOverflow
import kotlinx.coroutines.channels.awaitClose
import kotlinx.coroutines.channels.trySendBlocking
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.callbackFlow
import kotlinx.coroutines.flow.retryWhen
import kotlinx.coroutines.launch
import kotlinx.coroutines.withTimeoutOrNull
import javax.inject.Inject
import javax.inject.Singleton
@@ -32,6 +35,23 @@ class BillingClientConnectionProvider @Inject constructor(
private val connectionProvider: Flow<BillingClientConnection> = callbackFlow {
val purchasePublisher = MutableStateFlow<Collection<Purchase>>(emptySet())
// Events, not state: fresh observations feed the grace stamping (every successful query or
// push payload counts, even if equal to the previous one — Purchase.equals would dedupe a
// StateFlow), and failures must not be conflated away (Play reuses BillingResult instances,
// so a repeated ITEM_ALREADY_OWNED could be a same-instance emission).
// replay=1 on observations: the connect-time query can complete before the grace recorder
// subscribes (construction order race) — the latest fresh observation must not be lost.
// Failures stay replay=0: they can only originate from a purchase flow, which requires the
// consumer to already exist, and a consumed event must not be re-delivered.
val freshPurchaseObservations = MutableSharedFlow<Collection<Purchase>>(
replay = 1,
extraBufferCapacity = 16,
onBufferOverflow = BufferOverflow.DROP_OLDEST,
)
val purchaseFailureEvents = MutableSharedFlow<BillingResult>(
extraBufferCapacity = 8,
onBufferOverflow = BufferOverflow.DROP_OLDEST,
)
val client = newBuilder(context).apply {
enablePendingPurchases(
@@ -46,10 +66,16 @@ class BillingClientConnectionProvider @Inject constructor(
"onPurchasesUpdated(code=${result.responseCode}, message=${result.debugMessage}, purchases=$purchases)"
}
purchasePublisher.value = purchases.orEmpty()
freshPurchaseObservations.tryEmit(
purchases.orEmpty().filter { it.purchaseState == Purchase.PurchaseState.PURCHASED }
)
} else {
log(TAG, WARN) {
"error: onPurchasesUpdated(code=${result.responseCode}, message=${result.debugMessage}, purchases=$purchases)"
}
// Failures are published too: async ITEM_ALREADY_OWNED (Play telling us mid-flow
// that the user already owns it) drives the auto-restore in UpgradeRepoGplay.
purchaseFailureEvents.tryEmit(result)
}
}
}.build()
@@ -64,14 +90,27 @@ class BillingClientConnectionProvider @Inject constructor(
when (result.responseCode) {
BillingResponseCode.OK -> {
val connection = BillingClientConnection(client, purchasePublisher)
val connection = BillingClientConnection(
client = client,
purchasesGlobal = purchasePublisher,
freshObservations = freshPurchaseObservations,
purchaseFailuresGlobal = purchaseFailureEvents,
)
trySendBlocking(connection)
launch {
try {
connection.refreshPurchases()
log(TAG) { "Initial purchase query successful." }
// Bounded: a hung Play callback would otherwise hold the refresh
// lock indefinitely and starve every later refresh on this
// connection (foreground, manual restore, already-owned recovery).
val initial = withTimeoutOrNull(INITIAL_QUERY_TIMEOUT_MS) {
connection.refreshPurchases()
}
if (initial != null) log(TAG) { "Initial purchase query successful." }
else log(TAG, WARN) { "Initial purchase query timed out." }
} catch (e: CancellationException) {
throw e
} catch (e: Exception) {
log(TAG, ERROR) { "Initial purchase query failed:\n${e.asLog()}" }
}
@@ -127,6 +166,7 @@ class BillingClientConnectionProvider @Inject constructor(
}
companion object {
private const val INITIAL_QUERY_TIMEOUT_MS = 30_000L
val TAG: String = logTag("Upgrade", "Gplay", "Billing", "Client", "ConnectionProvider")
}
}
@@ -2,6 +2,7 @@ package eu.darken.capod.common.upgrade.core.data
import android.app.Activity
import com.android.billingclient.api.BillingClient
import com.android.billingclient.api.BillingResult
import com.android.billingclient.api.Purchase
import eu.darken.capod.common.AppForegroundState
import eu.darken.capod.common.TimeSource
@@ -49,6 +50,20 @@ class BillingDataRepo @Inject constructor(
.setupCommonEventHandlers(TAG) { "billingData" }
.replayingShare(scope)
// Async purchase failures from onPurchasesUpdated; UpgradeRepoGplay reconciles
// ITEM_ALREADY_OWNED silently.
val purchaseFailures: Flow<BillingResult> = connectionProvider
.flatMapLatest { it.purchaseFailures }
.setupCommonEventHandlers(TAG) { "purchaseFailures" }
// Every fresh observation of PURCHASED purchases (successful queries and push payloads) —
// unlike billingData this is not equality-deduped state and never mixes in stale listener
// data, so it is the only valid source for grace stamping.
val freshBillingData: Flow<BillingData> = connectionProvider
.flatMapLatest { it.freshPurchases }
.map { BillingData(purchases = it) }
.setupCommonEventHandlers(TAG) { "freshBillingData" }
init {
connectionProvider
.flatMapLatest { client ->
@@ -14,12 +14,18 @@ import io.kotest.matchers.longs.shouldBeGreaterThan
import io.kotest.matchers.nulls.shouldBeNull
import io.kotest.matchers.nulls.shouldNotBeNull
import io.kotest.matchers.shouldBe
import com.android.billingclient.api.BillingClient.BillingResponseCode
import com.android.billingclient.api.BillingResult
import eu.darken.capod.common.datastore.DataStoreValue
import io.mockk.coEvery
import io.mockk.coVerify
import io.mockk.every
import io.mockk.mockk
import kotlinx.coroutines.cancel
import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.emptyFlow
import kotlinx.coroutines.flow.first
import kotlinx.coroutines.flow.flowOf
import kotlinx.coroutines.flow.toList
import kotlinx.coroutines.launch
import kotlinx.coroutines.test.TestScope
@@ -38,6 +44,7 @@ class UpgradeRepoGplayTest : BaseTest() {
lateinit var tempDir: File
private lateinit var billingDataFlow: MutableSharedFlow<BillingData>
private lateinit var freshDataFlow: MutableSharedFlow<BillingData>
private lateinit var billingDataRepo: BillingDataRepo
private lateinit var billingCache: BillingCache
@@ -46,8 +53,11 @@ class UpgradeRepoGplayTest : BaseTest() {
@BeforeEach
fun setup() {
billingDataFlow = MutableSharedFlow()
freshDataFlow = MutableSharedFlow()
billingDataRepo = mockk {
every { billingData } returns billingDataFlow
every { freshBillingData } returns freshDataFlow
every { purchaseFailures } returns emptyFlow()
}
val dataStore = PreferenceDataStoreFactory.create(
produceFile = { File(tempDir, "test_billing_cache_${dsCounter++}.preferences_pb") }
@@ -430,4 +440,152 @@ class UpgradeRepoGplayTest : BaseTest() {
testScope.cancel()
}
private fun mockBillingResult(code: Int): BillingResult = mockk {
every { responseCode } returns code
every { debugMessage } returns "mock"
}
@Test
fun `fresh observations stamp the grace cache via the persistent collector`() = runTest2 {
val testScope = TestScope(UnconfinedTestDispatcher(testScheduler))
createRepo(testScope)
// No upgradeInfo collection at all — the init collector alone must stamp.
freshDataFlow.emit(BillingData(purchases = listOf(mockPurchase(CapodSku.Iap.PRO_UPGRADE.id))))
billingCache.lastProStateAt.valueBlocking shouldBeGreaterThan 0L
billingCache.lastProStateSku.valueBlocking shouldBe CapodSku.Iap.PRO_UPGRADE.id
testScope.cancel()
}
@Test
fun `the reactive mapping is read-only, only the collector stamps`() = runTest2 {
val testScope = TestScope(UnconfinedTestDispatcher(testScheduler))
// Count writes via mocked DataStore values: the cold freshBillingData flow delivers one
// pro observation to the init collector (1 stamp), and collecting upgradeInfo runs the
// mapping on top (onStart-null + pro data) — if the mapping still stamped, the count
// would exceed 1.
val proData = BillingData(purchases = listOf(mockPurchase(CapodSku.Iap.PRO_UPGRADE.id)))
val lastProAtMock = mockk<DataStoreValue<Long>>(relaxed = true) {
every { flow } returns flowOf(0L)
}
val lastProSkuMock = mockk<DataStoreValue<String>>(relaxed = true) {
every { flow } returns flowOf("")
}
val cache = mockk<BillingCache> {
every { lastProStateAt } returns lastProAtMock
every { lastProStateSku } returns lastProSkuMock
}
val repo = UpgradeRepoGplay(
scope = testScope,
billingDataRepo = mockk {
every { billingData } returns flowOf(proData)
every { freshBillingData } returns flowOf(proData)
every { purchaseFailures } returns emptyFlow()
},
billingCache = cache,
)
repo.upgradeInfo.first { it.isPro }.isPro shouldBe true
coVerify(exactly = 1) { lastProAtMock.update(any()) }
testScope.cancel()
}
@Test
fun `an observation emitted before the recorder subscribes still stamps via replay`() = runTest2 {
val testScope = TestScope(UnconfinedTestDispatcher(testScheduler))
// The connect-time query can complete before UpgradeRepoGplay is constructed — the
// observation stream carries replay=1 so that first Pro observation isn't lost.
val replayingFresh = MutableSharedFlow<BillingData>(replay = 1)
replayingFresh.tryEmit(BillingData(purchases = listOf(mockPurchase(CapodSku.Iap.PRO_UPGRADE.id))))
every { billingDataRepo.freshBillingData } returns replayingFresh
createRepo(testScope)
billingCache.lastProStateAt.valueBlocking shouldBeGreaterThan 0L
testScope.cancel()
}
@Test
fun `equal consecutive fresh observations both stamp`() = runTest2 {
val testScope = TestScope(UnconfinedTestDispatcher(testScheduler))
// Purchase equality dedupes the state flows for a steady owner — the observation stream
// must not dedupe, or a long-lived process would stop refreshing the grace timestamp.
val proData = BillingData(purchases = listOf(mockPurchase(CapodSku.Iap.PRO_UPGRADE.id)))
val lastProAtMock = mockk<DataStoreValue<Long>>(relaxed = true) {
every { flow } returns flowOf(0L)
}
val lastProSkuMock = mockk<DataStoreValue<String>>(relaxed = true) {
every { flow } returns flowOf("")
}
val cache = mockk<BillingCache> {
every { lastProStateAt } returns lastProAtMock
every { lastProStateSku } returns lastProSkuMock
}
UpgradeRepoGplay(
scope = testScope,
billingDataRepo = mockk {
every { billingData } returns emptyFlow()
every { freshBillingData } returns flowOf(proData, proData)
every { purchaseFailures } returns emptyFlow()
},
billingCache = cache,
)
coVerify(exactly = 2) { lastProAtMock.update(any()) }
testScope.cancel()
}
@Test
fun `the same failure instance delivered twice triggers two restores`() = runTest2 {
val testScope = TestScope(UnconfinedTestDispatcher(testScheduler))
coEvery { billingDataRepo.refresh() } returns BillingData(
purchases = listOf(mockPurchase(CapodSku.Iap.PRO_UPGRADE.id))
)
// Play reuses static BillingResult instances — a repeat of the same object must still
// trigger (a conflating/deduping state flow would drop it).
val sameInstance = mockBillingResult(BillingResponseCode.ITEM_ALREADY_OWNED)
every { billingDataRepo.purchaseFailures } returns flowOf(sameInstance, sameInstance)
createRepo(testScope)
coVerify(exactly = 2) { billingDataRepo.refresh() }
testScope.cancel()
}
@Test
fun `async already-owned purchase event triggers a silent restore`() = runTest2 {
val testScope = TestScope(UnconfinedTestDispatcher(testScheduler))
coEvery { billingDataRepo.refresh() } returns BillingData(
purchases = listOf(mockPurchase(CapodSku.Iap.PRO_UPGRADE.id))
)
every { billingDataRepo.purchaseFailures } returns
flowOf(mockBillingResult(BillingResponseCode.ITEM_ALREADY_OWNED))
createRepo(testScope)
coVerify(exactly = 1) { billingDataRepo.refresh() }
testScope.cancel()
}
@Test
fun `other async purchase failures do not trigger a restore`() = runTest2 {
val testScope = TestScope(UnconfinedTestDispatcher(testScheduler))
every { billingDataRepo.purchaseFailures } returns
flowOf(mockBillingResult(BillingResponseCode.DEVELOPER_ERROR))
createRepo(testScope)
coVerify(exactly = 0) { billingDataRepo.refresh() }
testScope.cancel()
}
}