Keep Google from refunding purchases when the app is closed too soon

app/build.gradle.kts

@@ -204,6 +204,7 @@ tasks.withType<KotlinCompile>().configureEach {
 dependencies {
     addBaseKotlin()
     addBaseAndroid()
+    addWorkManager()
 
     addDagger()
 

app/src/gplay/java/eu/darken/amply/upgrade/core/UpgradeRepoGplay.kt

@@ -19,6 +19,7 @@ import eu.darken.amply.upgrade.core.billing.Sku
 import eu.darken.amply.upgrade.core.billing.SkuDetails
 import eu.darken.amply.upgrade.core.billing.UserCanceledBillingException
 import eu.darken.amply.upgrade.core.billing.client.redacted
+import eu.darken.amply.upgrade.core.billing.work.PurchaseAckScheduler
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Deferred
@@ -53,6 +54,7 @@ import javax.inject.Singleton
 class UpgradeRepoGplay @Inject constructor(
     private val billingManager: BillingManager,
     private val billingCache: BillingCache,
+    private val ackScheduler: PurchaseAckScheduler,
 ) : UpgradeRepo {
 
     private val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
@@ -244,6 +246,18 @@ class UpgradeRepoGplay @Inject constructor(
             return
         }
         try {
+            // Persistent ack safety net, launch trigger: armed and AWAITED before the Play sheet can
+            // open, so the WorkManager DB transaction lands even if the process dies around the
+            // sheet — the exact window behind Play's unacknowledged-purchase auto-refunds. Failure
+            // to arm never blocks the purchase; the foreground ack path still exists.
+            try {
+                ackScheduler.armForBillingFlowLaunch()
+            } catch (e: CancellationException) {
+                throw e
+            } catch (e: Exception) {
+                log(TAG, WARN) { "Failed to arm ack safety net for launch: ${e.asLog()}" }
+            }
+
             // Bounded, like every other Play path. useConnection waits for a healthy connection
             // indefinitely, so a Play outage between rendering the offers and this tap would park the
             // launch forever — with launchBusySku still held, which leaves every purchase button busy

app/src/gplay/java/eu/darken/amply/upgrade/core/billing/BillingManager.kt

@@ -15,6 +15,7 @@ import eu.darken.amply.upgrade.core.billing.client.BillingClientException
 import eu.darken.amply.upgrade.core.billing.client.BillingConnection
 import eu.darken.amply.upgrade.core.billing.client.BillingConnectionProvider
 import eu.darken.amply.upgrade.core.billing.client.redacted
+import eu.darken.amply.upgrade.core.billing.work.PurchaseAckScheduler
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
@@ -43,6 +44,8 @@ import kotlinx.coroutines.flow.shareIn
 import kotlinx.coroutines.flow.update
 import kotlinx.coroutines.flow.updateAndGet
 import kotlinx.coroutines.launch
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
 import kotlinx.coroutines.withTimeoutOrNull
 import javax.inject.Inject
 import javax.inject.Singleton
@@ -55,6 +58,7 @@ import javax.inject.Singleton
 @Singleton
 open class BillingManager @Inject constructor(
     connectionProvider: BillingConnectionProvider,
+    private val ackScheduler: PurchaseAckScheduler,
 ) {
 
     private val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
@@ -284,13 +288,18 @@ open class BillingManager @Inject constructor(
     // below. The immutable Purchase snapshot keeps reporting isAcknowledged=false until a fresh Play
     // query supersedes it, so the ack re-fires every emission until then; re-acking is a documented
     // no-op on Play's side, whereas skipping a needed ack gets the purchase auto-refunded after 3
-    // days. Single sequential collector (the ack pass below), no locking needed.
+    // days. Confined by [ackMutex] (the collector's pass and explicit sweeps both run under it).
     private val loggedAckTokens = mutableSetOf<String>()
 
     // Tokens whose PERMANENT ack failure was already reported. Play will keep rejecting these, so the
-    // error fires once per token instead of once per pass.
+    // error fires once per token instead of once per pass. Same [ackMutex] confinement.
     private val reportedAckFailures = mutableSetOf<String>()
 
+    // Serializes acknowledgement work between the reactive ack collector and explicit
+    // [ensureAllAcknowledged] sweeps (PurchaseAckWorker): both paths mutate the token bookkeeping
+    // sets above and both must never double-drive the same purchase's inline retry sequence.
+    private val ackMutex = Mutex()
+
     // At most one reschedule timer in flight: repeated failures must not stack timers.
     private val ackRetryPending = MutableStateFlow(false)
 
@@ -323,17 +332,36 @@ open class BillingManager @Inject constructor(
     // whose immutable snapshot stays false until a fresh Play query.
     private enum class AckOutcome { SUCCESS, TRANSIENT, PERMANENT }
 
+    /** Aggregate outcome of one ack pass; [ensureAllAcknowledged] maps it to a sweep result. */
+    data class AckPassOutcome(val transient: Int, val permanent: Int)
+
     // One acknowledgement pass over the canonical purchase list. Never throws except cancellation:
     // transient failures schedule a re-drive, permanent ones are reported and left to organic
     // fresh-data signals.
-    private suspend fun runAckPass(purchases: Collection<Purchase>) {
+    private suspend fun runAckPass(purchases: Collection<Purchase>): AckPassOutcome = ackMutex.withLock {
         val needAck = purchases.filter {
             val needsAck = !it.isAcknowledged
             if (needsAck) log(TAG) { "Needs ACK: ${it.redacted()}" } else log(TAG) { "Already ACK'ed: ${it.redacted()}" }
             needsAck
         }
 
+        if (needAck.isNotEmpty()) {
+            // Arm the persistent safety net BEFORE attempting anything, and AWAIT the enqueue (the
+            // scheduler bounds it): the inline retries below can span minutes, and a process death
+            // inside them must not strand the purchase until Play's 3-day auto-refund. A deferred
+            // signal (channel + collector) would reintroduce exactly that window. Fail-open: the net
+            // is an extra layer, never a reason to skip the acks themselves.
+            try {
+                ackScheduler.armForUnackedPurchases(needAck.maxOf { it.purchaseTime } + ACK_SAFETY_NET_DEADLINE_MS)
+            } catch (e: CancellationException) {
+                throw e
+            } catch (e: Exception) {
+                log(TAG, WARN) { "Failed to arm ack safety net: ${e.asLog()}" }
+            }
+        }
+
         var transientFailures = 0
+        var permanentFailures = 0
 
         for (purchase in needAck) {
             // First ack of a token is INFO; idempotent repeats drop to DEBUG. This never gates the
@@ -396,6 +424,7 @@ open class BillingManager @Inject constructor(
             }
 
             if (outcome == AckOutcome.TRANSIENT) transientFailures++
+            if (outcome == AckOutcome.PERMANENT) permanentFailures++
             if (abortPass) break
         }
 
@@ -406,6 +435,43 @@ open class BillingManager @Inject constructor(
             }
             scheduleAckRetry()
         }
+
+        AckPassOutcome(transient = transientFailures, permanent = permanentFailures)
+    }
+
+    /** Outcome of an explicit safety-net sweep, see [ensureAllAcknowledged]. */
+    enum class AckSweepResult { COMPLETE, RETRY, PERMANENT_FAILURE }
+
+    /**
+     * One self-contained acknowledgement sweep for the persistent safety net (PurchaseAckWorker):
+     * refresh from Play, then acknowledge everything unacknowledged IN THIS COROUTINE. The reactive
+     * ack collector consumes purchase state asynchronously, so a caller that needs proof the acks
+     * actually happened before it reports success (a worker deciding success vs retry) cannot rely
+     * on it. Never throws except cancellation.
+     */
+    open suspend fun ensureAllAcknowledged(): AckSweepResult {
+        log(TAG) { "ensureAllAcknowledged()" }
+        val fresh = try {
+            useConnection { refreshPurchases() }
+        } catch (e: CancellationException) {
+            throw e
+        } catch (e: Exception) {
+            log(TAG, WARN) { "ensureAllAcknowledged(): refresh failed: ${e.asLog()}" }
+            return AckSweepResult.RETRY
+        }
+        // Same bookkeeping every other refresh exit owes: grace episode clock + dead-binder teardown.
+        processReconciliation(fresh)
+        // Pending payments are filtered here, exactly like the reactive collector does: acknowledging
+        // one is a protocol error Play rejects permanently.
+        val outcome = runAckPass(fresh.purchases.purchased())
+        return when {
+            // An incomplete refresh may be hiding an unacknowledged purchase of the failed type, and
+            // a transient ack failure is retriable by definition.
+            outcome.transient > 0 || !fresh.isComplete -> AckSweepResult.RETRY
+            // Play will keep rejecting these no matter how often the worker comes back.
+            outcome.permanent > 0 -> AckSweepResult.PERMANENT_FAILURE
+            else -> AckSweepResult.COMPLETE
+        }
     }
 
     // A purchase Play will keep rejecting: report it once per token, then stay quiet. The pass still
@@ -581,6 +647,10 @@ open class BillingManager @Inject constructor(
             BillingResponseCode.ITEM_NOT_OWNED,
         )
 
+        // Play auto-refunds purchases not acknowledged within 3 days; every safety-net deadline
+        // derives from this.
+        const val ACK_SAFETY_NET_DEADLINE_MS = 3 * 24 * 60 * 60 * 1000L
+
         private const val INITIAL_REFRESH_TIMEOUT_MS = 30_000L
         private const val MAX_BACKOFF_MS = 300_000L
 

app/src/gplay/java/eu/darken/amply/upgrade/core/billing/work/PurchaseAckScheduler.kt

@@ -0,0 +1,118 @@
+package eu.darken.amply.upgrade.core.billing.work
+
+import androidx.work.BackoffPolicy
+import androidx.work.Constraints
+import androidx.work.ExistingWorkPolicy
+import androidx.work.NetworkType
+import androidx.work.OneTimeWorkRequestBuilder
+import androidx.work.WorkManager
+import androidx.work.await
+import androidx.work.workDataOf
+import eu.darken.amply.BuildConfig
+import eu.darken.amply.common.debug.logging.Logging.Priority.WARN
+import eu.darken.amply.common.debug.logging.log
+import eu.darken.amply.common.debug.logging.logTag
+import eu.darken.amply.upgrade.core.billing.BillingManager
+import kotlinx.coroutines.withTimeoutOrNull
+import java.util.concurrent.TimeUnit
+import javax.inject.Inject
+import javax.inject.Provider
+import javax.inject.Singleton
+
+/**
+ * Arms the [PurchaseAckWorker] safety net. Two triggers:
+ * - a billing flow is about to launch (armed and awaited BEFORE the Play sheet, so the WorkManager
+ *   DB transaction lands even if the process dies around the sheet),
+ * - an ack pass discovered unacknowledged purchases (called directly, pre-attempt, from
+ *   [BillingManager]'s ack pass).
+ *
+ * `open` for the same reason [BillingManager] is: the billing tests substitute it to assert the
+ * arming contract without standing up WorkManager.
+ */
+@Singleton
+open class PurchaseAckScheduler @Inject constructor(
+    // Resolved on the first arm, not at construction: AmplyApp eagerly injects UpgradeSurfaceSync
+    // (and with it the whole billing stack) during Application field injection, and resolving
+    // WorkManager there triggers its on-demand initialization before the Application's worker
+    // factory field is set.
+    private val workManager: Provider<WorkManager>,
+) {
+
+    // A genuinely new flow refreshes the watch window: REPLACE the previous LAUNCH watch. The
+    // worker sweeps ALL unacknowledged purchases, so replacing an older watch loses nothing — and a
+    // pending rescue for an already-discovered purchase has its own identity, so starting another
+    // purchase can never displace it. The long delay keeps the worker out of the window where the
+    // user may still be in the Play sheet.
+    open suspend fun armForBillingFlowLaunch() = arm(
+        name = WORK_NAME_LAUNCH,
+        policy = ExistingWorkPolicy.REPLACE,
+        expiresAt = System.currentTimeMillis() + BillingManager.ACK_SAFETY_NET_DEADLINE_MS,
+        initialDelayMs = LAUNCH_DELAY_MS,
+    )
+
+    // Any pending rescue already covers every unacknowledged purchase: KEEP it. Once completed work
+    // exists, KEEP inserts a fresh request. Short delay — the purchase already EXISTS (unlike the
+    // launch trigger), possibly for days, so waiting 30min could waste real deadline time.
+    // Accepted edge of KEEP, within the rescue lane only: a pending request keeps its original
+    // (possibly earlier) expiry; a newer purchase with a later deadline is only re-covered once a
+    // later pass re-arms after the old work completed. Bounded residual, only reachable via
+    // out-of-band purchases.
+    open suspend fun armForUnackedPurchases(expiresAt: Long) = arm(
+        name = WORK_NAME_RESCUE,
+        policy = ExistingWorkPolicy.KEEP,
+        expiresAt = expiresAt,
+        initialDelayMs = DISCOVERY_DELAY_MS,
+    )
+
+    private suspend fun arm(
+        name: String,
+        policy: ExistingWorkPolicy,
+        expiresAt: Long,
+        initialDelayMs: Long,
+    ) {
+        if (expiresAt <= System.currentTimeMillis()) {
+            // Play has already voided (or is about to void) such a purchase; a sweep can't help.
+            log(TAG, WARN) { "arm($policy): deadline $expiresAt already passed, not scheduling" }
+            return
+        }
+        val request = OneTimeWorkRequestBuilder<PurchaseAckWorker>().apply {
+            setConstraints(
+                Constraints.Builder().apply {
+                    setRequiredNetworkType(NetworkType.CONNECTED)
+                }.build(),
+            )
+            // Launch trigger: the worker must not run while the user may still be in the Play sheet
+            // — an immediate sweep would find nothing unacknowledged, report success, and complete
+            // the net before the purchase it exists for even happened.
+            setInitialDelay(initialDelayMs, TimeUnit.MILLISECONDS)
+            setBackoffCriteria(BackoffPolicy.EXPONENTIAL, BACKOFF_DELAY_MS, TimeUnit.MILLISECONDS)
+            setInputData(workDataOf(PurchaseAckWorker.KEY_EXPIRES_AT to expiresAt))
+        }.build()
+
+        // Await the enqueue: the caller arms this because the process may die at any moment — a
+        // fire-and-forget enqueue could be lost with it. Cancellable and BOUNDED: every caller needs
+        // a durable enqueue without an unbounded stall — a WorkManager that never settles must
+        // become an exception (handled fail-open by every caller) instead of a hang (which on the
+        // launch lane would park the purchase and its busy guard forever).
+        val operation = workManager.get().enqueueUniqueWork(name, policy, request)
+        withTimeoutOrNull(ENQUEUE_TIMEOUT_MS) { operation.await() }
+            ?: throw IllegalStateException("WorkManager enqueue did not settle within ${ENQUEUE_TIMEOUT_MS}ms")
+        log(TAG) { "arm($policy): safety net armed, expiresAt=$expiresAt" }
+    }
+
+    companion object {
+        // WorkManager persists these names AND the worker's class name in its DB across app updates:
+        // keep all of them stable while old work may exist (hence the version suffix for future
+        // changes). Separate identities per trigger: the launch watch's REPLACE must not be able to
+        // displace a pending rescue for a purchase that already exists.
+        private val WORK_NAME_LAUNCH = "${BuildConfig.APPLICATION_ID}.gplay.purchase-ack.launch.v1"
+        private val WORK_NAME_RESCUE = "${BuildConfig.APPLICATION_ID}.gplay.purchase-ack.rescue.v1"
+
+        private const val LAUNCH_DELAY_MS = 30 * 60 * 1000L
+        private const val DISCOVERY_DELAY_MS = 60 * 1000L
+        private const val BACKOFF_DELAY_MS = 30 * 60 * 1000L
+        private const val ENQUEUE_TIMEOUT_MS = 10 * 1000L
+
+        val TAG: String = logTag("Upgrade", "Gplay", "Billing", "AckScheduler")
+    }
+}

app/src/gplay/java/eu/darken/amply/upgrade/core/billing/work/PurchaseAckWorker.kt

@@ -0,0 +1,85 @@
+package eu.darken.amply.upgrade.core.billing.work
+
+import android.content.Context
+import androidx.hilt.work.HiltWorker
+import androidx.work.CoroutineWorker
+import androidx.work.ListenableWorker.Result
+import androidx.work.WorkerParameters
+import dagger.assisted.Assisted
+import dagger.assisted.AssistedInject
+import eu.darken.amply.common.debug.logging.Logging.Priority.INFO
+import eu.darken.amply.common.debug.logging.Logging.Priority.WARN
+import eu.darken.amply.common.debug.logging.log
+import eu.darken.amply.common.debug.logging.logTag
+import eu.darken.amply.upgrade.core.billing.BillingManager
+import kotlinx.coroutines.withTimeoutOrNull
+
+/**
+ * Persistent acknowledgement safety net, armed by [PurchaseAckScheduler].
+ *
+ * Play auto-refunds (and revokes) any purchase not acknowledged within 3 days. The in-process ack
+ * machinery in [BillingManager] handles every case where the process lives long enough — this
+ * worker covers the case it can't: the process dies around the Play purchase sheet (OEM task
+ * killers) and the user doesn't reopen the app before the deadline. Play voids such purchases and
+ * revokes the entitlement, so the user loses what they paid for.
+ *
+ * Self-completing by design: nothing cancels this work from the foreground ack path (an ack pass can
+ * legitimately see zero unacknowledged purchases while the Play sheet is still open, which must not
+ * tear down the net). The redundant sweep after a successful foreground ack is one purchase query.
+ */
+@HiltWorker
+class PurchaseAckWorker @AssistedInject constructor(
+    @Assisted private val context: Context,
+    @Assisted private val params: WorkerParameters,
+    private val billingManager: BillingManager,
+) : CoroutineWorker(context, params) {
+
+    override suspend fun doWork(): Result {
+        val expiresAt = inputData.getLong(KEY_EXPIRES_AT, 0L)
+        log(TAG) { "doWork(): attempt=$runAttemptCount, expiresAt=$expiresAt" }
+
+        if (!isWorthSweeping(System.currentTimeMillis(), expiresAt)) {
+            // Past Play's refund deadline (or malformed input): retrying can't achieve anything.
+            // failure() is deliberate over success() — it is visible in WorkManager diagnostics, and
+            // a completed state lets a later KEEP enqueue insert fresh work.
+            log(TAG, WARN) { "doWork(): deadline passed, giving up" }
+            return Result.failure()
+        }
+
+        // Bounded well below WorkManager's 10-minute execution limit, but generous enough for the
+        // connection wait plus the per-purchase inline retries. A sweep that ran out of time is a
+        // transient outcome, not a verdict. External cancellation propagates out of doWork — it must
+        // never be converted into success.
+        val sweep = withTimeoutOrNull(SWEEP_TIMEOUT_MS) {
+            billingManager.ensureAllAcknowledged()
+        }
+        log(TAG, INFO) { "doWork(): sweep=$sweep" }
+
+        return mapSweep(sweep, System.currentTimeMillis(), expiresAt)
+    }
+
+    companion object {
+        // Persisted in WorkManager's request data — keep the key stable while old work may exist.
+        const val KEY_EXPIRES_AT = "purchase.ack.expiresAt"
+
+        private const val SWEEP_TIMEOUT_MS = 4 * 60 * 1000L
+
+        // Pure so the retry/expiry decision is unit-testable without a WorkManager test harness.
+        internal fun isWorthSweeping(now: Long, expiresAt: Long): Boolean =
+            expiresAt > 0L && now < expiresAt
+
+        internal fun mapSweep(
+            sweep: BillingManager.AckSweepResult?,
+            now: Long,
+            expiresAt: Long,
+        ): Result = when (sweep) {
+            BillingManager.AckSweepResult.COMPLETE -> Result.success()
+            BillingManager.AckSweepResult.PERMANENT_FAILURE -> Result.failure()
+            // RETRY or timeout (null): keep trying until the deadline. WorkManager's exponential
+            // backoff caps at 5h, so the 3-day window still yields many attempts.
+            else -> if (now < expiresAt) Result.retry() else Result.failure()
+        }
+
+        val TAG: String = logTag("Upgrade", "Gplay", "Billing", "AckWorker")
+    }
+}