General: Add a persistent acknowledgement safety net for Play purchases

Play auto-refunds (and revokes) purchases not acknowledged within 3 days.
The in-process ack machinery covers every case where the process lives
long enough; what it cannot cover is a process death around the Play
sheet (aggressive OEM task killers) followed by the user not reopening
the app before the deadline.

Add a gplay-only WorkManager safety net:
- PurchaseAckWorker: self-completing sweep via a new bounded
  BillingManager.ensureAllAcknowledged() that refreshes and acknowledges
  in the same coroutine (the reactive ack collector is async, so a worker
  cannot prove its acks happened through it). Retries with exponential
  backoff until the purchase's refund deadline, then gives up visibly.
- PurchaseAckScheduler: two unique work identities. A launch watch
  (REPLACE, armed and awaited before startIapFlow with a 30min delay so
  it cannot complete while the user is still in the sheet) and a
  discovered-purchase rescue (KEEP, 1min delay, armed directly from an
  ack pass that finds unacknowledged purchases, pre-attempt). Separate
  identities so a new purchase flow can never displace a pending rescue.
  Both triggers are fail-open: a broken WorkManager never blocks a
  purchase or an ack. WorkManager resolves via Provider at first arm,
  because AmplyApp eagerly injects UpgradeSurfaceSync (and with it the
  whole billing stack) during Application field injection, where
  resolving WorkManager would trigger its on-demand initialization before
  the worker factory field is set.
- Nothing cancels the work from the foreground path: an ack pass can see
  zero unacked purchases while the sheet is still open, so the worker
  completes itself after its own reconciliation instead.

The ack pass now runs under a mutex (the worker sweep and the reactive
collector would otherwise race the token bookkeeping) and reports
per-outcome counts for the sweep result mapping.

New wiring this needs: androidx.hilt:hilt-work plus its KSP compiler,
AmplyApp implements Configuration.Provider with an injected
HiltWorkerFactory, the androidx.startup WorkManagerInitializer is removed
from the manifest so that configuration is actually used, and WorkManager
becomes injectable (WorkManagerModule). SettleRefreshWorker is
deliberately left as-is: HiltWorkerFactory delegates workers it doesn't
know to WorkManager's default reflective factory.

This is a semantic port of d4rken-org/sdmaid-se#2685 — the behaviour is
the same, the shape follows amply's own billing architecture (a
BillingManager that owns its scope, pending purchases filtered at the
call site, hand-written fakes instead of a mocking framework).

FOSS stays untouched: all new billing types live in src/gplay, workers
need no manifest entry, and the HiltWorkerFactory resolves the worker
only in gplay variants.

Author: d4rken

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")
+    }
+}