Add google play billing token conversion endpoint and job.

app/src/main/java/org/thoughtcrime/securesms/backup/v2/ui/subscription/MessageBackupsFlowState.kt

@@ -18,5 +18,6 @@ data class MessageBackupsFlowState(
   val inAppPayment: InAppPaymentTable.InAppPayment? = null,
   val startScreen: MessageBackupsStage,
   val stage: MessageBackupsStage = startScreen,
-  val backupKey: BackupKey = SignalStore.svr.getOrCreateMasterKey().deriveBackupKey()
+  val backupKey: BackupKey = SignalStore.svr.getOrCreateMasterKey().deriveBackupKey(),
+  val failure: Throwable? = null
 )

app/src/main/java/org/thoughtcrime/securesms/backup/v2/ui/subscription/MessageBackupsFlowViewModel.kt

@@ -8,10 +8,16 @@ package org.thoughtcrime.securesms.backup.v2.ui.subscription
 import androidx.lifecycle.ViewModel
 import androidx.lifecycle.viewModelScope
 import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.FlowPreview
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.StateFlow
+import kotlinx.coroutines.flow.filter
+import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.flow.take
+import kotlinx.coroutines.flow.timeout
 import kotlinx.coroutines.flow.update
 import kotlinx.coroutines.launch
+import kotlinx.coroutines.reactive.asFlow
 import kotlinx.coroutines.withContext
 import org.signal.core.util.billing.BillingPurchaseResult
 import org.signal.core.util.money.FiatMoney
@@ -19,15 +25,21 @@ import org.signal.donations.InAppPaymentType
 import org.thoughtcrime.securesms.backup.v2.BackupRepository
 import org.thoughtcrime.securesms.backup.v2.MessageBackupTier
 import org.thoughtcrime.securesms.components.settings.app.subscription.DonationSerializationHelper.toFiatValue
+import org.thoughtcrime.securesms.components.settings.app.subscription.InAppPaymentsRepository
+import org.thoughtcrime.securesms.components.settings.app.subscription.RecurringInAppPaymentRepository
+import org.thoughtcrime.securesms.components.settings.app.subscription.donate.InAppPaymentError
 import org.thoughtcrime.securesms.database.InAppPaymentTable
 import org.thoughtcrime.securesms.database.SignalDatabase
+import org.thoughtcrime.securesms.database.model.InAppPaymentSubscriberRecord
 import org.thoughtcrime.securesms.database.model.databaseprotos.InAppPaymentData
 import org.thoughtcrime.securesms.dependencies.AppDependencies
+import org.thoughtcrime.securesms.jobs.InAppPaymentPurchaseTokenJob
 import org.thoughtcrime.securesms.keyvalue.SignalStore
 import org.thoughtcrime.securesms.recipients.Recipient
 import org.thoughtcrime.securesms.util.RemoteConfig
 import org.whispersystems.signalservice.internal.push.SubscriptionsConfiguration
 import java.math.BigDecimal
+import kotlin.time.Duration.Companion.seconds
 
 class MessageBackupsFlowViewModel : ViewModel() {
 
@@ -45,6 +57,17 @@ class MessageBackupsFlowViewModel : ViewModel() {
     check(SignalStore.backup.backupTier != MessageBackupTier.PAID) { "This screen does not support cancellation or downgrades." }
 
     viewModelScope.launch {
+      try {
+        ensureSubscriberIdForBackups()
+      } catch (e: Exception) {
+        internalStateFlow.update {
+          it.copy(
+            stage = MessageBackupsStage.FAILURE,
+            failure = e
+          )
+        }
+      }
+
       internalStateFlow.update {
         it.copy(
           availableBackupTypes = BackupRepository.getAvailableBackupsTypes(
@@ -55,12 +78,30 @@ class MessageBackupsFlowViewModel : ViewModel() {
     }
 
     viewModelScope.launch {
-      AppDependencies.billingApi.getBillingPurchaseResults().collect {
-        when (it) {
+      AppDependencies.billingApi.getBillingPurchaseResults().collect { result ->
+        when (result) {
           is BillingPurchaseResult.Success -> {
-            // 1. Copy the purchaseToken into our inAppPaymentData
-            // 2. Enqueue the redemption chain
-            goToNextStage()
+            internalStateFlow.update { it.copy(stage = MessageBackupsStage.PROCESS_PAYMENT) }
+
+            try {
+              handleSuccess(
+                result,
+                internalStateFlow.value.inAppPayment!!.id
+              )
+
+              internalStateFlow.update {
+                it.copy(
+                  stage = MessageBackupsStage.COMPLETED
+                )
+              }
+            } catch (e: Exception) {
+              internalStateFlow.update {
+                it.copy(
+                  stage = MessageBackupsStage.FAILURE,
+                  failure = e
+                )
+              }
+            }
           }
 
           else -> goToPreviousStage()
@@ -81,9 +122,10 @@ class MessageBackupsFlowViewModel : ViewModel() {
         MessageBackupsStage.TYPE_SELECTION -> validateTypeAndUpdateState(it)
         MessageBackupsStage.CHECKOUT_SHEET -> validateGatewayAndUpdateState(it)
         MessageBackupsStage.CREATING_IN_APP_PAYMENT -> error("This is driven by an async coroutine.")
-        MessageBackupsStage.PROCESS_PAYMENT -> it.copy(stage = MessageBackupsStage.COMPLETED)
-        MessageBackupsStage.PROCESS_FREE -> it.copy(stage = MessageBackupsStage.COMPLETED)
+        MessageBackupsStage.PROCESS_PAYMENT -> error("This is driven by an async coroutine.")
+        MessageBackupsStage.PROCESS_FREE -> error("This is driven by an async coroutine.")
         MessageBackupsStage.COMPLETED -> error("Unsupported state transition from terminal state COMPLETED")
+        MessageBackupsStage.FAILURE -> error("Unsupported state transition from terminal state FAILURE")
       }
     }
   }
@@ -103,6 +145,7 @@ class MessageBackupsFlowViewModel : ViewModel() {
           MessageBackupsStage.PROCESS_PAYMENT -> MessageBackupsStage.PROCESS_PAYMENT
           MessageBackupsStage.PROCESS_FREE -> MessageBackupsStage.PROCESS_FREE
           MessageBackupsStage.COMPLETED -> error("Unsupported state transition from terminal state COMPLETED")
+          MessageBackupsStage.FAILURE -> error("Unsupported state transition from terminal state FAILURE")
         }
 
         it.copy(stage = previousScreen)
@@ -146,7 +189,7 @@ class MessageBackupsFlowViewModel : ViewModel() {
       val id = SignalDatabase.inAppPayments.insert(
         type = InAppPaymentType.RECURRING_BACKUP,
         state = InAppPaymentTable.State.CREATED,
-        subscriberId = null,
+        subscriberId = InAppPaymentsRepository.requireSubscriber(InAppPaymentSubscriberRecord.Type.BACKUP).subscriberId,
         endOfPeriod = null,
         inAppPaymentData = InAppPaymentData(
           badge = null,
@@ -170,4 +213,51 @@ class MessageBackupsFlowViewModel : ViewModel() {
 
     return state.copy(stage = MessageBackupsStage.CREATING_IN_APP_PAYMENT)
   }
+
+  /**
+   * Ensures we have a SubscriberId created and available for use. This is considered safe because
+   * the screen this is called in is assumed to only be accessible if the user does not currently have
+   * a subscription.
+   */
+  private suspend fun ensureSubscriberIdForBackups() {
+    val product = AppDependencies.billingApi.queryProduct() ?: error("No product available.")
+    SignalStore.inAppPayments.setSubscriberCurrency(product.price.currency, InAppPaymentSubscriberRecord.Type.BACKUP)
+    RecurringInAppPaymentRepository.ensureSubscriberId(InAppPaymentSubscriberRecord.Type.BACKUP).blockingAwait()
+  }
+
+  /**
+   * Handles a successful BillingPurchaseResult. Updates the in app payment, enqueues the appropriate job chain,
+   * and handles any resulting error. Like donations, we will wait up to 10s for the completion of the job chain.
+   */
+  @OptIn(FlowPreview::class)
+  private suspend fun handleSuccess(result: BillingPurchaseResult.Success, inAppPaymentId: InAppPaymentTable.InAppPaymentId) {
+    withContext(Dispatchers.IO) {
+      val inAppPayment = SignalDatabase.inAppPayments.getById(inAppPaymentId)!!
+      SignalDatabase.inAppPayments.update(
+        inAppPayment.copy(
+          data = inAppPayment.data.copy(
+            redemption = inAppPayment.data.redemption!!.copy(
+              googlePlayBillingPurchaseToken = result.purchaseToken
+            )
+          )
+        )
+      )
+
+      InAppPaymentPurchaseTokenJob.createJobChain(inAppPayment).enqueue()
+    }
+
+    val terminalInAppPayment = withContext(Dispatchers.IO) {
+      InAppPaymentsRepository.observeUpdates(inAppPaymentId).asFlow()
+        .filter { it.state == InAppPaymentTable.State.END }
+        .take(1)
+        .timeout(10.seconds)
+        .first()
+    }
+
+    if (terminalInAppPayment.data.error != null) {
+      throw InAppPaymentError(terminalInAppPayment.data.error)
+    } else {
+      return
+    }
+  }
 }

app/src/main/java/org/thoughtcrime/securesms/backup/v2/ui/subscription/MessageBackupsStage.kt

@@ -19,7 +19,8 @@ enum class MessageBackupsStage(
   CREATING_IN_APP_PAYMENT(route = Route.TYPE_SELECTION),
   PROCESS_PAYMENT(route = Route.TYPE_SELECTION),
   PROCESS_FREE(route = Route.TYPE_SELECTION),
-  COMPLETED(route = Route.TYPE_SELECTION);
+  COMPLETED(route = Route.TYPE_SELECTION),
+  FAILURE(route = Route.TYPE_SELECTION);
 
   /**
    * Compose navigation route to display while in a given stage.