Cancel play subscriptions when replacing them

service/src/main/java/org/whispersystems/textsecuregcm/controllers/SubscriptionController.java

@@ -420,6 +420,10 @@ public class SubscriptionController {
 
   After calling this method, the payment is confirmed. Callers must durably store their subscriberId before calling
   this method to ensure their payment is tracked.
+
+  Once a purchaseToken to is posted to a subscriberId, the same subscriberId must not be used with another payment
+  method. A different playbilling purchaseToken can be posted to the same subscriberId, in this case the subscription
+  associated with the old purchaseToken will be cancelled.
   """)
   @ApiResponse(responseCode = "200", description = "The purchaseToken was validated and acknowledged")
   @ApiResponse(responseCode = "402", description = "The purchaseToken payment is incomplete or invalid")

service/src/main/java/org/whispersystems/textsecuregcm/storage/SubscriptionManager.java

@@ -349,29 +349,42 @@ public class SubscriptionManager {
       final GooglePlayBillingManager googlePlayBillingManager,
       final String purchaseToken) {
 
-    return getSubscriber(subscriberCredentials).thenCompose(record -> {
+    // For IAP providers, the subscriptionId and the customerId are both just the purchaseToken. Changes to the
-      if (record.processorCustomer != null
+    // subscription always just result in a new purchaseToken
-          && record.processorCustomer.processor() != PaymentProvider.GOOGLE_PLAY_BILLING) {
+    final ProcessorCustomer pc = new ProcessorCustomer(purchaseToken, PaymentProvider.GOOGLE_PLAY_BILLING);
-        return CompletableFuture.failedFuture(
-            new SubscriptionException.ProcessorConflict("existing processor does not match"));
-      }
 
-      // For IAP providers, the subscriptionId and the customerId are both just the purchaseToken. Changes to the
+    return getSubscriber(subscriberCredentials)
-      // subscription always just result in a new purchaseToken
-      final ProcessorCustomer pc = new ProcessorCustomer(purchaseToken, PaymentProvider.GOOGLE_PLAY_BILLING);
 
-      return googlePlayBillingManager
+        // Check the record for an existing subscription
-          // Validating ensures we don't allow a user-determined token that's totally bunk into the subscription manager,
+        .thenCompose(record -> {
-          // but we don't want to acknowledge it until it's successfully persisted.
+          if (record.processorCustomer != null
-          .validateToken(purchaseToken)
+              && record.processorCustomer.processor() != PaymentProvider.GOOGLE_PLAY_BILLING) {
-          // Store the purchaseToken with the subscriber
+            return CompletableFuture.failedFuture(
-          .thenCompose(validatedToken -> subscriptions.setIapPurchase(
+                new SubscriptionException.ProcessorConflict("existing processor does not match"));
-                  record, pc, purchaseToken, validatedToken.getLevel(), subscriberCredentials.now())
+          }
-              // Now that the purchaseToken is durable, we can acknowledge it
-              .thenCompose(ignore -> validatedToken.acknowledgePurchase())
-              .thenApply(ignore -> validatedToken.getLevel()));
-    });
 
+          // If we're replacing an existing purchaseToken, cancel it first
+          return Optional.ofNullable(record.processorCustomer)
+              .map(ProcessorCustomer::customerId)
+              .filter(existingToken -> !purchaseToken.equals(existingToken))
+              .map(googlePlayBillingManager::cancelAllActiveSubscriptions)
+              .orElseGet(() -> CompletableFuture.completedFuture(null))
+              .thenApply(ignored -> record);
+        })
+
+        // Validate and set the purchaseToken
+        .thenCompose(record -> googlePlayBillingManager
+
+            // Validating ensures we don't allow a user-determined token that's totally bunk into the subscription manager,
+            // but we don't want to acknowledge it until it's successfully persisted.
+            .validateToken(purchaseToken)
+
+            // Store the purchaseToken with the subscriber
+            .thenCompose(validatedToken -> subscriptions.setIapPurchase(
+                    record, pc, purchaseToken, validatedToken.getLevel(), subscriberCredentials.now())
+                // Now that the purchaseToken is durable, we can acknowledge it
+                .thenCompose(ignore -> validatedToken.acknowledgePurchase())
+                .thenApply(ignore -> validatedToken.getLevel())));
   }
 
   private SubscriptionPaymentProcessor getProcessor(PaymentProvider provider) {

service/src/test/java/org/whispersystems/textsecuregcm/controllers/SubscriptionControllerTest.java

@@ -902,6 +902,56 @@ class SubscriptionControllerTest {
         eq(now));
   }
 
+  @Test
+  public void replacePlayPurchaseToken() {
+    final String oldPurchaseToken = "oldPurchaseToken";
+    final String newPurchaseToken = "newPurchaseToken";
+    final byte[] subscriberUserAndKey = new byte[32];
+    Arrays.fill(subscriberUserAndKey, (byte) 1);
+    final byte[] user = Arrays.copyOfRange(subscriberUserAndKey, 0, 16);
+    final String subscriberId = Base64.getEncoder().encodeToString(subscriberUserAndKey);
+
+    final Instant now = Instant.now();
+    when(CLOCK.instant()).thenReturn(now);
+
+    final ProcessorCustomer oldPc = new ProcessorCustomer(oldPurchaseToken, PaymentProvider.GOOGLE_PLAY_BILLING);
+    final Map<String, AttributeValue> dynamoItem = Map.of(Subscriptions.KEY_PASSWORD, b(new byte[16]),
+        Subscriptions.KEY_CREATED_AT, n(Instant.now().getEpochSecond()),
+        Subscriptions.KEY_ACCESSED_AT, n(Instant.now().getEpochSecond()),
+        Subscriptions.KEY_PROCESSOR_ID_CUSTOMER_ID, b(oldPc.toDynamoBytes()));
+    final Subscriptions.Record record = Subscriptions.Record.from(user, dynamoItem);
+    when(SUBSCRIPTIONS.get(any(), any()))
+        .thenReturn(CompletableFuture.completedFuture(Subscriptions.GetResult.found(record)));
+
+    final GooglePlayBillingManager.ValidatedToken validatedToken = mock(GooglePlayBillingManager.ValidatedToken.class);
+    when(validatedToken.getLevel()).thenReturn(99L);
+    when(validatedToken.acknowledgePurchase()).thenReturn(CompletableFuture.completedFuture(null));
+
+    when(PLAY_MANAGER.validateToken(eq(newPurchaseToken))).thenReturn(CompletableFuture.completedFuture(validatedToken));
+    when(PLAY_MANAGER.cancelAllActiveSubscriptions(eq(oldPurchaseToken)))
+        .thenReturn(CompletableFuture.completedFuture(null));
+
+    when(SUBSCRIPTIONS.setIapPurchase(any(), any(), anyString(), anyLong(), any()))
+        .thenReturn(CompletableFuture.completedFuture(null));
+
+    final Response response = RESOURCE_EXTENSION
+        .target(String.format("/v1/subscription/%s/playbilling/%s", subscriberId, newPurchaseToken))
+        .request()
+        .post(Entity.json(""));
+    assertThat(response.getStatus()).isEqualTo(200);
+    assertThat(response.readEntity(SubscriptionController.SetSubscriptionLevelSuccessResponse.class).level())
+        .isEqualTo(99L);
+
+    verify(SUBSCRIPTIONS, times(1)).setIapPurchase(
+        any(),
+        eq(new ProcessorCustomer(newPurchaseToken, PaymentProvider.GOOGLE_PLAY_BILLING)),
+        eq(newPurchaseToken),
+        eq(99L),
+        eq(now));
+
+    verify(PLAY_MANAGER, times(1)).cancelAllActiveSubscriptions(oldPurchaseToken);
+  }
+
   @ParameterizedTest
   @CsvSource({"5, P45D", "201, P13D"})
   public void createReceiptCredential(long level, Duration expectedExpirationWindow)