From 90b280d6a0b662072164ecfb86adcfd2273dcd4a Mon Sep 17 00:00:00 2001 From: Chris Eager Date: Wed, 20 May 2026 12:47:51 -0500 Subject: [PATCH] Convert Subscriptions to sync DynamoDB client --- .../textsecuregcm/WhisperServerService.java | 2 +- .../storage/SubscriptionManager.java | 21 +- .../textsecuregcm/storage/Subscriptions.java | 146 +++++++------- .../workers/CommandDependencies.java | 2 +- .../SubscriptionControllerTest.java | 82 +++----- .../storage/SubscriptionsTest.java | 185 ++++++++---------- 6 files changed, 188 insertions(+), 250 deletions(-) diff --git a/service/src/main/java/org/whispersystems/textsecuregcm/WhisperServerService.java b/service/src/main/java/org/whispersystems/textsecuregcm/WhisperServerService.java index fe0bdeb77..986c1a3a9 100644 --- a/service/src/main/java/org/whispersystems/textsecuregcm/WhisperServerService.java +++ b/service/src/main/java/org/whispersystems/textsecuregcm/WhisperServerService.java @@ -740,7 +740,7 @@ public class WhisperServerService extends Application new UncheckedIOException(new IOException("processor must now exist"))); @@ -330,7 +329,7 @@ public class SubscriptionManager { subscriptions.subscriptionLevelChanged(subscriberCredentials.subscriberUser(), subscriberCredentials.now(), level, - updatedSubscriptionId.id()).join(); + updatedSubscriptionId.id()); } else { // Otherwise, we don't have a subscription yet so create it and then record the subscription id long lastSubscriptionCreatedAt = record.subscriptionCreatedAt != null @@ -437,7 +436,7 @@ public class SubscriptionManager { final ProcessorCustomer pc = new ProcessorCustomer(originalTransactionId, PaymentProvider.APPLE_APP_STORE); final Long level = appleAppStoreManager.validateTransaction(originalTransactionId); - subscriptions.setIapPurchase(record, pc, originalTransactionId, level, subscriberCredentials.now()).join(); + subscriptions.setIapPurchase(record, pc, originalTransactionId, level, subscriberCredentials.now()); return level; } diff --git a/service/src/main/java/org/whispersystems/textsecuregcm/storage/Subscriptions.java b/service/src/main/java/org/whispersystems/textsecuregcm/storage/Subscriptions.java index e5a851b5a..47cd50d7d 100644 --- a/service/src/main/java/org/whispersystems/textsecuregcm/storage/Subscriptions.java +++ b/service/src/main/java/org/whispersystems/textsecuregcm/storage/Subscriptions.java @@ -10,7 +10,6 @@ import static org.whispersystems.textsecuregcm.util.AttributeValues.n; import static org.whispersystems.textsecuregcm.util.AttributeValues.s; import com.google.common.annotations.VisibleForTesting; -import com.google.common.base.Throwables; import jakarta.ws.rs.ClientErrorException; import jakarta.ws.rs.core.Response; import java.nio.charset.StandardCharsets; @@ -19,8 +18,6 @@ import java.time.Instant; import java.util.Map; import java.util.Objects; import java.util.Optional; -import java.util.concurrent.CompletableFuture; -import java.util.concurrent.CompletionException; import javax.annotation.Nonnull; import javax.annotation.Nullable; import org.slf4j.Logger; @@ -28,15 +25,16 @@ import org.slf4j.LoggerFactory; import org.whispersystems.textsecuregcm.subscriptions.PaymentProvider; import org.whispersystems.textsecuregcm.subscriptions.ProcessorCustomer; import org.whispersystems.textsecuregcm.util.Pair; -import org.whispersystems.textsecuregcm.util.Util; -import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient; +import software.amazon.awssdk.services.dynamodb.DynamoDbClient; import software.amazon.awssdk.services.dynamodb.model.AttributeValue; import software.amazon.awssdk.services.dynamodb.model.ConditionalCheckFailedException; import software.amazon.awssdk.services.dynamodb.model.GetItemRequest; import software.amazon.awssdk.services.dynamodb.model.GetItemResponse; import software.amazon.awssdk.services.dynamodb.model.QueryRequest; +import software.amazon.awssdk.services.dynamodb.model.QueryResponse; import software.amazon.awssdk.services.dynamodb.model.ReturnValue; import software.amazon.awssdk.services.dynamodb.model.UpdateItemRequest; +import software.amazon.awssdk.services.dynamodb.model.UpdateItemResponse; public class Subscriptions { @@ -161,11 +159,11 @@ public class Subscriptions { } private final String table; - private final DynamoDbAsyncClient client; + private final DynamoDbClient client; public Subscriptions( @Nonnull String table, - @Nonnull DynamoDbAsyncClient client) { + @Nonnull DynamoDbClient client) { this.table = Objects.requireNonNull(table); this.client = Objects.requireNonNull(client); } @@ -173,7 +171,7 @@ public class Subscriptions { /** * Looks in the GSI for a record with the given customer id and returns the user id. */ - public CompletableFuture getSubscriberUserByProcessorCustomer(ProcessorCustomer processorCustomer) { + public byte[] getSubscriberUserByProcessorCustomer(ProcessorCustomer processorCustomer) { QueryRequest query = QueryRequest.builder() .tableName(table) .indexName(INDEX_NAME) @@ -185,20 +183,20 @@ public class Subscriptions { .expressionAttributeValues(Map.of( ":processor_customer_id", b(processorCustomer.toDynamoBytes()))) .build(); - return client.query(query).thenApply(queryResponse -> { - int count = queryResponse.count(); - if (count == 0) { - return null; - } else if (count > 1) { - logger.error("expected invariant of 1-1 subscriber-customer violated for customer {} ({})", - processorCustomer.customerId(), processorCustomer.processor()); - throw new IllegalStateException( - "expected invariant of 1-1 subscriber-customer violated for customer " + processorCustomer); - } else { - Map result = queryResponse.items().getFirst(); - return result.get(KEY_USER).b().asByteArray(); - } - }); + final QueryResponse queryResponse = client.query(query); + + int count = queryResponse.count(); + if (count == 0) { + return null; + } else if (count > 1) { + logger.error("expected invariant of 1-1 subscriber-customer violated for customer {} ({})", + processorCustomer.customerId(), processorCustomer.processor()); + throw new IllegalStateException( + "expected invariant of 1-1 subscriber-customer violated for customer " + processorCustomer); + } else { + Map result = queryResponse.items().getFirst(); + return result.get(KEY_USER).b().asByteArray(); + } } public static class GetResult { @@ -228,21 +226,22 @@ public class Subscriptions { /** * Looks up a record with the given {@code user} and validates the {@code hmac} before returning it. */ - public CompletableFuture get(byte[] user, byte[] hmac) { - return getUser(user).thenApply(getItemResponse -> { - if (!getItemResponse.hasItem()) { - return GetResult.NOT_STORED; - } + public GetResult get(byte[] user, byte[] hmac) { + final GetItemResponse getItemResponse = getUser(user); - Record record = Record.from(user, getItemResponse.item()); - if (!MessageDigest.isEqual(hmac, record.password)) { - return GetResult.PASSWORD_MISMATCH; - } - return GetResult.found(record); - }); + if (!getItemResponse.hasItem()) { + + return GetResult.NOT_STORED; + } + + Record record = Record.from(user, getItemResponse.item()); + if (!MessageDigest.isEqual(hmac, record.password)) { + return GetResult.PASSWORD_MISMATCH; + } + return GetResult.found(record); } - private CompletableFuture getUser(byte[] user) { + private GetItemResponse getUser(byte[] user) { checkUserLength(user); GetItemRequest request = GetItemRequest.builder() @@ -254,10 +253,11 @@ public class Subscriptions { return client.getItem(request); } - public CompletableFuture create(byte[] user, byte[] password, Instant createdAt) { + @Nullable + public Record create(byte[] user, byte[] password, Instant createdAt) { checkUserLength(user); - UpdateItemRequest request = UpdateItemRequest.builder() + final UpdateItemRequest request = UpdateItemRequest.builder() .tableName(table) .key(Map.of(KEY_USER, b(user))) .returnValues(ReturnValue.ALL_NEW) @@ -279,17 +279,13 @@ public class Subscriptions { ":accessed_at", n(createdAt.getEpochSecond())) ) .build(); - return client.updateItem(request).handle((updateItemResponse, throwable) -> { - if (throwable != null) { - if (Throwables.getRootCause(throwable) instanceof ConditionalCheckFailedException) { - return null; - } - Throwables.throwIfUnchecked(throwable); - throw new CompletionException(throwable); - } + try { + final UpdateItemResponse updateItemResponse = client.updateItem(request); return Record.from(user, updateItemResponse.attributes()); - }); + } catch (ConditionalCheckFailedException _) { + return null; + } } /** @@ -297,10 +293,10 @@ public class Subscriptions { * * @return the user record. */ - public CompletableFuture setProcessorAndCustomerId(Record userRecord, + public Record setProcessorAndCustomerId(Record userRecord, ProcessorCustomer activeProcessorCustomer, Instant updatedAt) { - UpdateItemRequest request = UpdateItemRequest.builder() + final UpdateItemRequest request = UpdateItemRequest.builder() .tableName(table) .key(Map.of(KEY_USER, b(userRecord.user))) .returnValues(ReturnValue.ALL_NEW) @@ -318,15 +314,13 @@ public class Subscriptions { ":processor_customer_id", b(activeProcessorCustomer.toDynamoBytes()) )).build(); - return client.updateItem(request) - .thenApply(updateItemResponse -> Record.from(userRecord.user, updateItemResponse.attributes())) - .exceptionallyCompose(throwable -> { - if (Throwables.getRootCause(throwable) instanceof ConditionalCheckFailedException) { - throw new ClientErrorException(Response.Status.CONFLICT); - } - Throwables.throwIfUnchecked(throwable); - throw new CompletionException(throwable); - }); + try { + final UpdateItemResponse updateItemResponse = client.updateItem(request); + + return Record.from(userRecord.user, updateItemResponse.attributes()); + } catch (ConditionalCheckFailedException _) { + throw new ClientErrorException(Response.Status.CONFLICT); + } } /** @@ -345,7 +339,7 @@ public class Subscriptions { * @param updatedAt The time of this update * @return A stage that completes once the record has been updated */ - public CompletableFuture setIapPurchase( + public void setIapPurchase( final Record record, final ProcessorCustomer processorCustomer, final String subscriptionId, @@ -390,21 +384,17 @@ public class Subscriptions { ":subscription_level_changed_at", n(updatedAt.getEpochSecond()))) .build(); - return client.updateItem(request) - .exceptionallyCompose(throwable -> { - if (Throwables.getRootCause(throwable) instanceof ConditionalCheckFailedException) { - throw new ClientErrorException(Response.Status.CONFLICT); - } - Throwables.throwIfUnchecked(throwable); - throw new CompletionException(throwable); - }) - .thenRun(Util.NOOP); + try { + client.updateItem(request); + } catch (ConditionalCheckFailedException _) { + throw new ClientErrorException(Response.Status.CONFLICT); + } } - public CompletableFuture accessedAt(byte[] user, Instant accessedAt) { + public void accessedAt(byte[] user, Instant accessedAt) { checkUserLength(user); - UpdateItemRequest request = UpdateItemRequest.builder() + final UpdateItemRequest request = UpdateItemRequest.builder() .tableName(table) .key(Map.of(KEY_USER, b(user))) .returnValues(ReturnValue.NONE) @@ -412,13 +402,13 @@ public class Subscriptions { .expressionAttributeNames(Map.of("#accessed_at", KEY_ACCESSED_AT)) .expressionAttributeValues(Map.of(":accessed_at", n(accessedAt.getEpochSecond()))) .build(); - return client.updateItem(request).thenApply(updateItemResponse -> null); + client.updateItem(request); } - public CompletableFuture setCanceledAt(byte[] user, Instant canceledAt) { + public void setCanceledAt(byte[] user, Instant canceledAt) { checkUserLength(user); - UpdateItemRequest request = UpdateItemRequest.builder() + final UpdateItemRequest request = UpdateItemRequest.builder() .tableName(table) .key(Map.of(KEY_USER, b(user))) .returnValues(ReturnValue.NONE) @@ -434,14 +424,14 @@ public class Subscriptions { ":accessed_at", n(canceledAt.getEpochSecond()), ":canceled_at", n(canceledAt.getEpochSecond()))) .build(); - return client.updateItem(request).thenApply(updateItemResponse -> null); + client.updateItem(request); } - public CompletableFuture subscriptionCreated( + public void subscriptionCreated( byte[] user, String subscriptionId, Instant subscriptionCreatedAt, long level) { checkUserLength(user); - UpdateItemRequest request = UpdateItemRequest.builder() + final UpdateItemRequest request = UpdateItemRequest.builder() .tableName(table) .key(Map.of(KEY_USER, b(user))) .returnValues(ReturnValue.NONE) @@ -466,14 +456,14 @@ public class Subscriptions { ":subscription_level", n(level), ":subscription_level_changed_at", n(subscriptionCreatedAt.getEpochSecond()))) .build(); - return client.updateItem(request).thenApply(updateItemResponse -> null); + client.updateItem(request); } - public CompletableFuture subscriptionLevelChanged( + public void subscriptionLevelChanged( byte[] user, Instant subscriptionLevelChangedAt, long level, String subscriptionId) { checkUserLength(user); - UpdateItemRequest request = UpdateItemRequest.builder() + final UpdateItemRequest request = UpdateItemRequest.builder() .tableName(table) .key(Map.of(KEY_USER, b(user))) .returnValues(ReturnValue.NONE) @@ -495,7 +485,7 @@ public class Subscriptions { ":subscription_level", n(level), ":subscription_level_changed_at", n(subscriptionLevelChangedAt.getEpochSecond()))) .build(); - return client.updateItem(request).thenApply(updateItemResponse -> null); + client.updateItem(request); } private static byte[] checkUserLength(final byte[] user) { diff --git a/service/src/main/java/org/whispersystems/textsecuregcm/workers/CommandDependencies.java b/service/src/main/java/org/whispersystems/textsecuregcm/workers/CommandDependencies.java index 5af83a5ad..2a3610433 100644 --- a/service/src/main/java/org/whispersystems/textsecuregcm/workers/CommandDependencies.java +++ b/service/src/main/java/org/whispersystems/textsecuregcm/workers/CommandDependencies.java @@ -344,7 +344,7 @@ public record CommandDependencies( configuration.getAppleAppStore().subscriptionGroupId(), configuration.getAppleAppStore().productIdToLevel()); final SubscriptionManager subscriptionManager = new SubscriptionManager( - new Subscriptions(configuration.getDynamoDbTables().getSubscriptions().getTableName(), dynamoDbAsyncClient), + new Subscriptions(configuration.getDynamoDbTables().getSubscriptions().getTableName(), dynamoDbClient), List.of(googlePlayBillingManager, appleAppStoreManager), zkReceiptOperations, issuedReceiptsManager); diff --git a/service/src/test/java/org/whispersystems/textsecuregcm/controllers/SubscriptionControllerTest.java b/service/src/test/java/org/whispersystems/textsecuregcm/controllers/SubscriptionControllerTest.java index 89e7411b7..1f49194ab 100644 --- a/service/src/test/java/org/whispersystems/textsecuregcm/controllers/SubscriptionControllerTest.java +++ b/service/src/test/java/org/whispersystems/textsecuregcm/controllers/SubscriptionControllerTest.java @@ -175,10 +175,7 @@ class SubscriptionControllerTest extends AbstractV1SubscriptionControllerTest { final Subscriptions.Record record = Subscriptions.Record.from( Arrays.copyOfRange(subscriberUserAndKey, 0, 16), dynamoItem); when(SUBSCRIPTIONS.get(eq(Arrays.copyOfRange(subscriberUserAndKey, 0, 16)), any())) - .thenReturn(CompletableFuture.completedFuture(Subscriptions.GetResult.found(record))); - - when(SUBSCRIPTIONS.subscriptionCreated(any(), any(), any(), anyLong())) - .thenReturn(CompletableFuture.completedFuture(null)); + .thenReturn(Subscriptions.GetResult.found(record)); } @Test @@ -233,7 +230,7 @@ class SubscriptionControllerTest extends AbstractV1SubscriptionControllerTest { final Subscriptions.Record record = Subscriptions.Record.from( Arrays.copyOfRange(subscriberUserAndKey, 0, 16), dynamoItem); when(SUBSCRIPTIONS.get(eq(Arrays.copyOfRange(subscriberUserAndKey, 0, 16)), any())) - .thenReturn(CompletableFuture.completedFuture(Subscriptions.GetResult.found(record))); + .thenReturn(Subscriptions.GetResult.found(record)); final String level = String.valueOf(levelId); final String idempotencyKey = UUID.randomUUID().toString(); @@ -260,7 +257,7 @@ class SubscriptionControllerTest extends AbstractV1SubscriptionControllerTest { final Subscriptions.Record record = Subscriptions.Record.from( Arrays.copyOfRange(subscriberUserAndKey, 0, 16), dynamoItem); when(SUBSCRIPTIONS.get(eq(Arrays.copyOfRange(subscriberUserAndKey, 0, 16)), any())) - .thenReturn(CompletableFuture.completedFuture(Subscriptions.GetResult.found(record))); + .thenReturn(Subscriptions.GetResult.found(record)); final Response response = RESOURCE_EXTENSION .target(String.format("/v1/subscription/%s/create_payment_method", subscriberId)) @@ -305,8 +302,7 @@ class SubscriptionControllerTest extends AbstractV1SubscriptionControllerTest { Arrays.fill(subscriberUserAndKey, (byte) 1); final String subscriberId = Base64.getEncoder().encodeToString(subscriberUserAndKey); - when(SUBSCRIPTIONS.get(any(), any())).thenReturn(CompletableFuture.completedFuture( - Subscriptions.GetResult.NOT_STORED)); + when(SUBSCRIPTIONS.get(any(), any())).thenReturn(Subscriptions.GetResult.NOT_STORED); final Map dynamoItem = Map.of(Subscriptions.KEY_PASSWORD, b(new byte[16]), Subscriptions.KEY_CREATED_AT, n(Instant.now().getEpochSecond()), @@ -314,7 +310,7 @@ class SubscriptionControllerTest extends AbstractV1SubscriptionControllerTest { ); final Subscriptions.Record record = Subscriptions.Record.from( Arrays.copyOfRange(subscriberUserAndKey, 0, 16), dynamoItem); - when(SUBSCRIPTIONS.create(any(), any(), any())).thenReturn(CompletableFuture.completedFuture(record)); + when(SUBSCRIPTIONS.create(any(), any(), any())).thenReturn(record); final Response createResponse = RESOURCE_EXTENSION.target(String.format("/v1/subscription/%s", subscriberId)) .request() @@ -322,9 +318,7 @@ class SubscriptionControllerTest extends AbstractV1SubscriptionControllerTest { assertThat(createResponse.getStatus()).isEqualTo(200); // creating should be idempotent - when(SUBSCRIPTIONS.get(any(), any())).thenReturn(CompletableFuture.completedFuture( - Subscriptions.GetResult.found(record))); - when(SUBSCRIPTIONS.accessedAt(any(), any())).thenReturn(CompletableFuture.completedFuture(null)); + when(SUBSCRIPTIONS.get(any(), any())).thenReturn(Subscriptions.GetResult.found(record)); final Response idempotentCreateResponse = RESOURCE_EXTENSION.target( String.format("/v1/subscription/%s", subscriberId)) @@ -334,9 +328,8 @@ class SubscriptionControllerTest extends AbstractV1SubscriptionControllerTest { // when the manager returns `null`, it means there was a password mismatch from the storage layer `create`. // this could happen if there is a race between two concurrent `create` requests for the same user ID - when(SUBSCRIPTIONS.get(any(), any())).thenReturn(CompletableFuture.completedFuture( - Subscriptions.GetResult.NOT_STORED)); - when(SUBSCRIPTIONS.create(any(), any(), any())).thenReturn(CompletableFuture.completedFuture(null)); + when(SUBSCRIPTIONS.get(any(), any())).thenReturn(Subscriptions.GetResult.NOT_STORED); + when(SUBSCRIPTIONS.create(any(), any(), any())).thenReturn(null); final Response managerCreateNullResponse = RESOURCE_EXTENSION.target( String.format("/v1/subscription/%s", subscriberId)) @@ -350,8 +343,7 @@ class SubscriptionControllerTest extends AbstractV1SubscriptionControllerTest { final String mismatchedSubscriberId = Base64.getEncoder().encodeToString(subscriberUserAndMismatchedKey); // a password mismatch for an existing record - when(SUBSCRIPTIONS.get(any(), any())).thenReturn(CompletableFuture.completedFuture( - Subscriptions.GetResult.PASSWORD_MISMATCH)); + when(SUBSCRIPTIONS.get(any(), any())).thenReturn(Subscriptions.GetResult.PASSWORD_MISMATCH); final Response passwordMismatchResponse = RESOURCE_EXTENSION.target( String.format("/v1/subscription/%s", mismatchedSubscriberId)) @@ -380,8 +372,7 @@ class SubscriptionControllerTest extends AbstractV1SubscriptionControllerTest { final String subscriberId = Base64.getEncoder().encodeToString(subscriberUserAndKey); when(CLOCK.instant()).thenReturn(Instant.now()); - when(SUBSCRIPTIONS.get(any(), any())).thenReturn(CompletableFuture.completedFuture( - Subscriptions.GetResult.NOT_STORED)); + when(SUBSCRIPTIONS.get(any(), any())).thenReturn(Subscriptions.GetResult.NOT_STORED); final Map dynamoItem = Map.of(Subscriptions.KEY_PASSWORD, b(new byte[16]), Subscriptions.KEY_CREATED_AT, n(Instant.now().getEpochSecond()), @@ -389,8 +380,7 @@ class SubscriptionControllerTest extends AbstractV1SubscriptionControllerTest { ); final Subscriptions.Record record = Subscriptions.Record.from( Arrays.copyOfRange(subscriberUserAndKey, 0, 16), dynamoItem); - when(SUBSCRIPTIONS.create(any(), any(), any(Instant.class))) - .thenReturn(CompletableFuture.completedFuture(record)); + when(SUBSCRIPTIONS.create(any(), any(), any(Instant.class))).thenReturn(record); final Response createSubscriberResponse = RESOURCE_EXTENSION .target(String.format("/v1/subscription/%s", subscriberId)) @@ -400,7 +390,7 @@ class SubscriptionControllerTest extends AbstractV1SubscriptionControllerTest { assertThat(createSubscriberResponse.getStatus()).isEqualTo(200); when(SUBSCRIPTIONS.get(any(), any())) - .thenReturn(CompletableFuture.completedFuture(Subscriptions.GetResult.found(record))); + .thenReturn(Subscriptions.GetResult.found(record)); final String customerId = "some-customer-id"; final ProcessorCustomer customer = new ProcessorCustomer( @@ -414,9 +404,8 @@ class SubscriptionControllerTest extends AbstractV1SubscriptionControllerTest { final Subscriptions.Record recordWithCustomerId = Subscriptions.Record.from(record.user, dynamoItemWithProcessorCustomer); - when(SUBSCRIPTIONS.setProcessorAndCustomerId(any(Subscriptions.Record.class), any(), - any(Instant.class))) - .thenReturn(CompletableFuture.completedFuture(recordWithCustomerId)); + when(SUBSCRIPTIONS.setProcessorAndCustomerId(any(Subscriptions.Record.class), any(), any(Instant.class))) + .thenReturn(recordWithCustomerId); final String clientSecret = "some-client-secret"; when(STRIPE_MANAGER.createPaymentMethodSetupToken(customerId)) @@ -447,12 +436,12 @@ class SubscriptionControllerTest extends AbstractV1SubscriptionControllerTest { final Subscriptions.Record record = Subscriptions.Record.from( Arrays.copyOfRange(subscriberUserAndKey, 0, 16), dynamoItem); when(SUBSCRIPTIONS.create(any(), any(), any(Instant.class))) - .thenReturn(CompletableFuture.completedFuture(record)); + .thenReturn(record); // set up mocks when(CLOCK.instant()).thenReturn(Instant.now()); when(SUBSCRIPTIONS.get(any(), any())) - .thenReturn(CompletableFuture.completedFuture(Subscriptions.GetResult.found(record))); + .thenReturn(Subscriptions.GetResult.found(record)); final Response response = RESOURCE_EXTENSION .target(String.format("/v1/subscription/%s/level/%d/%s/%s", subscriberId, 5, "usd", "abcd")) @@ -486,17 +475,15 @@ class SubscriptionControllerTest extends AbstractV1SubscriptionControllerTest { final Subscriptions.Record record = Subscriptions.Record.from( Arrays.copyOfRange(subscriberUserAndKey, 0, 16), dynamoItem); when(SUBSCRIPTIONS.create(any(), any(), any(Instant.class))) - .thenReturn(CompletableFuture.completedFuture(record)); + .thenReturn(record); // set up mocks when(CLOCK.instant()).thenReturn(Instant.now()); when(SUBSCRIPTIONS.get(any(), any())) - .thenReturn(CompletableFuture.completedFuture(Subscriptions.GetResult.found(record))); + .thenReturn(Subscriptions.GetResult.found(record)); when(BRAINTREE_MANAGER.createSubscription(any(), any(), anyLong(), anyLong())) .thenReturn(new CustomerAwareSubscriptionPaymentProcessor.SubscriptionId("subscription")); - when(SUBSCRIPTIONS.subscriptionCreated(any(), any(), any(), anyLong())) - .thenReturn(CompletableFuture.completedFuture(null)); final Response response = RESOURCE_EXTENSION .target(String.format("/v1/subscription/%s/level/%d/%s/%s", subscriberId, levelId, "usd", "abcd")) @@ -536,12 +523,12 @@ class SubscriptionControllerTest extends AbstractV1SubscriptionControllerTest { final Subscriptions.Record record = Subscriptions.Record.from( Arrays.copyOfRange(subscriberUserAndKey, 0, 16), dynamoItem); when(SUBSCRIPTIONS.create(any(), any(), any(Instant.class))) - .thenReturn(CompletableFuture.completedFuture(record)); + .thenReturn(record); // set up mocks when(CLOCK.instant()).thenReturn(Instant.now()); when(SUBSCRIPTIONS.get(any(), any())) - .thenReturn(CompletableFuture.completedFuture(Subscriptions.GetResult.found(record))); + .thenReturn(Subscriptions.GetResult.found(record)); final Object subscriptionObj = new Object(); when(BRAINTREE_MANAGER.getSubscription(any())).thenReturn(subscriptionObj); @@ -552,8 +539,6 @@ class SubscriptionControllerTest extends AbstractV1SubscriptionControllerTest { if (expectUpdate) { when(BRAINTREE_MANAGER.updateSubscription(any(), any(), anyLong(), anyString())) .thenReturn(new CustomerAwareSubscriptionPaymentProcessor.SubscriptionId(updatedSubscriptionId)); - when(SUBSCRIPTIONS.subscriptionLevelChanged(any(), any(), anyLong(), anyString())) - .thenReturn(CompletableFuture.completedFuture(null)); } final String idempotencyKey = "abcd"; @@ -609,11 +594,11 @@ class SubscriptionControllerTest extends AbstractV1SubscriptionControllerTest { final Subscriptions.Record record = Subscriptions.Record.from( Arrays.copyOfRange(subscriberUserAndKey, 0, 16), dynamoItem); when(SUBSCRIPTIONS.create(any(), any(), any(Instant.class))) - .thenReturn(CompletableFuture.completedFuture(record)); + .thenReturn(record); when(CLOCK.instant()).thenReturn(Instant.now()); when(SUBSCRIPTIONS.get(any(), any())) - .thenReturn(CompletableFuture.completedFuture(Subscriptions.GetResult.found(record))); + .thenReturn(Subscriptions.GetResult.found(record)); final Object subscriptionObj = new Object(); when(BRAINTREE_MANAGER.getSubscription(any())).thenReturn(subscriptionObj); @@ -653,14 +638,11 @@ class SubscriptionControllerTest extends AbstractV1SubscriptionControllerTest { final Subscriptions.Record record = Subscriptions.Record.from(user, dynamoItem); when(SUBSCRIPTIONS.get(any(), any())) - .thenReturn(CompletableFuture.completedFuture(Subscriptions.GetResult.found(record))); + .thenReturn(Subscriptions.GetResult.found(record)); when(APPSTORE_MANAGER.validateTransaction(eq(originalTxId))) .thenReturn(99L); - - when(SUBSCRIPTIONS.setIapPurchase(any(), any(), anyString(), anyLong(), any())) - .thenReturn(CompletableFuture.completedFuture(null)); - + final Response response = RESOURCE_EXTENSION .target(String.format("/v1/subscription/%s/appstore/%s", subscriberId, originalTxId)) .request() @@ -695,15 +677,12 @@ class SubscriptionControllerTest extends AbstractV1SubscriptionControllerTest { ); final Subscriptions.Record record = Subscriptions.Record.from(user, dynamoItem); when(SUBSCRIPTIONS.get(any(), any())) - .thenReturn(CompletableFuture.completedFuture(Subscriptions.GetResult.found(record))); + .thenReturn(Subscriptions.GetResult.found(record)); final GooglePlayBillingManager.ValidatedToken validatedToken = mock(GooglePlayBillingManager.ValidatedToken.class); when(validatedToken.getLevel()).thenReturn(99L); when(PLAY_MANAGER.validateToken(eq(purchaseToken))).thenReturn(validatedToken); - 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, purchaseToken)) .request() @@ -739,16 +718,13 @@ class SubscriptionControllerTest extends AbstractV1SubscriptionControllerTest { 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))); + .thenReturn(Subscriptions.GetResult.found(record)); final GooglePlayBillingManager.ValidatedToken validatedToken = mock(GooglePlayBillingManager.ValidatedToken.class); when(validatedToken.getLevel()).thenReturn(99L); when(PLAY_MANAGER.validateToken(eq(newPurchaseToken))).thenReturn(validatedToken); - 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() @@ -776,14 +752,14 @@ class SubscriptionControllerTest extends AbstractV1SubscriptionControllerTest { when(CLOCK.instant()).thenReturn(Instant.now()); when(SUBSCRIPTIONS.get(any(), any())) - .thenReturn(CompletableFuture.completedFuture(Subscriptions.GetResult.found(Subscriptions.Record.from( + .thenReturn(Subscriptions.GetResult.found(Subscriptions.Record.from( Arrays.copyOfRange(subscriberUserAndKey, 0, 16), 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(new ProcessorCustomer("customer", PaymentProvider.STRIPE).toDynamoBytes()), - Subscriptions.KEY_SUBSCRIPTION_ID, s("subscriptionId")))))); + Subscriptions.KEY_SUBSCRIPTION_ID, s("subscriptionId"))))); when(STRIPE_MANAGER.getReceiptItem(any())) .thenThrow(new SubscriptionChargeFailurePaymentRequiredException( PaymentProvider.STRIPE, @@ -831,7 +807,7 @@ class SubscriptionControllerTest extends AbstractV1SubscriptionControllerTest { when(CLOCK.instant()).thenReturn(Instant.now()); when(SUBSCRIPTIONS.get(any(), any())) - .thenReturn(CompletableFuture.completedFuture(Subscriptions.GetResult.found(record))); + .thenReturn(Subscriptions.GetResult.found(record)); when(BRAINTREE_MANAGER.getReceiptItem(subscriptionId)).thenReturn( new CustomerAwareSubscriptionPaymentProcessor.ReceiptItem( "itemId", diff --git a/service/src/test/java/org/whispersystems/textsecuregcm/storage/SubscriptionsTest.java b/service/src/test/java/org/whispersystems/textsecuregcm/storage/SubscriptionsTest.java index 00fad9845..a03174e5d 100644 --- a/service/src/test/java/org/whispersystems/textsecuregcm/storage/SubscriptionsTest.java +++ b/service/src/test/java/org/whispersystems/textsecuregcm/storage/SubscriptionsTest.java @@ -6,6 +6,7 @@ package org.whispersystems.textsecuregcm.storage; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.assertj.core.api.Fail.fail; import static org.whispersystems.textsecuregcm.storage.Subscriptions.GetResult.Type.FOUND; import static org.whispersystems.textsecuregcm.storage.Subscriptions.GetResult.Type.NOT_STORED; @@ -16,8 +17,6 @@ import java.time.Duration; import java.time.Instant; import java.util.Base64; import java.util.Optional; -import java.util.concurrent.CompletableFuture; -import java.util.concurrent.ExecutionException; import java.util.function.Consumer; import javax.annotation.Nonnull; import org.assertj.core.api.Condition; @@ -34,7 +33,6 @@ import org.whispersystems.textsecuregcm.util.TestRandomUtil; class SubscriptionsTest { private static final long NOW_EPOCH_SECONDS = 1_500_000_000L; - private static final Duration DEFAULT_TIMEOUT = Duration.ofSeconds(3); @RegisterExtension static final DynamoDbExtension DYNAMO_DB_EXTENSION = new DynamoDbExtension(Tables.SUBSCRIPTIONS); @@ -52,7 +50,7 @@ class SubscriptionsTest { customer = Base64.getEncoder().encodeToString(TestRandomUtil.nextBytes(16)); created = Instant.ofEpochSecond(NOW_EPOCH_SECONDS); subscriptions = new Subscriptions( - Tables.SUBSCRIPTIONS.tableName(), DYNAMO_DB_EXTENSION.getDynamoDbAsyncClient()); + Tables.SUBSCRIPTIONS.tableName(), DYNAMO_DB_EXTENSION.getDynamoDbClient()); } @Test @@ -62,124 +60,112 @@ class SubscriptionsTest { Instant created1 = Instant.ofEpochSecond(NOW_EPOCH_SECONDS); Instant created2 = Instant.ofEpochSecond(NOW_EPOCH_SECONDS + 1); - CompletableFuture getFuture = subscriptions.get(user, password1); - assertThat(getFuture).succeedsWithin(DEFAULT_TIMEOUT).satisfies(getResult -> { - assertThat(getResult.type).isEqualTo(NOT_STORED); - assertThat(getResult.record).isNull(); - }); + GetResult getResult = subscriptions.get(user, password1); + assertThat(getResult.type).isEqualTo(NOT_STORED); + assertThat(getResult.record).isNull(); - getFuture = subscriptions.get(user, password2); - assertThat(getFuture).succeedsWithin(DEFAULT_TIMEOUT).satisfies(getResult -> { - assertThat(getResult.type).isEqualTo(NOT_STORED); - assertThat(getResult.record).isNull(); - }); + getResult = subscriptions.get(user, password2); + assertThat(getResult.type).isEqualTo(NOT_STORED); + assertThat(getResult.record).isNull(); - CompletableFuture createFuture = + Subscriptions.Record create = subscriptions.create(user, password1, created1); Consumer recordRequirements = checkFreshlyCreatedRecord(user, password1, created1); - assertThat(createFuture).succeedsWithin(DEFAULT_TIMEOUT).satisfies(recordRequirements); + assertThat(create).satisfies(recordRequirements); // password check fails so this should return null - createFuture = subscriptions.create(user, password2, created2); - assertThat(createFuture).succeedsWithin(DEFAULT_TIMEOUT).isNull(); + create = subscriptions.create(user, password2, created2); + assertThat(create).isNull(); // password check matches, but the record already exists so nothing should get updated - createFuture = subscriptions.create(user, password1, created2); - assertThat(createFuture).succeedsWithin(DEFAULT_TIMEOUT).satisfies(recordRequirements); + create = subscriptions.create(user, password1, created2); + assertThat(create).satisfies(recordRequirements); } @Test void testGet() { byte[] wrongUser = TestRandomUtil.nextBytes(16); byte[] wrongPassword = TestRandomUtil.nextBytes(16); - assertThat(subscriptions.create(user, password, created)).succeedsWithin(DEFAULT_TIMEOUT); - assertThat(subscriptions.get(user, password)).succeedsWithin(DEFAULT_TIMEOUT).satisfies(getResult -> { + subscriptions.create(user, password, created); + + assertThat(subscriptions.get(user, password)).satisfies(getResult -> { assertThat(getResult.type).isEqualTo(FOUND); assertThat(getResult.record).isNotNull().satisfies(checkFreshlyCreatedRecord(user, password, created)); }); - assertThat(subscriptions.get(user, wrongPassword)).succeedsWithin(DEFAULT_TIMEOUT) - .satisfies(getResult -> { + assertThat(subscriptions.get(user, wrongPassword)).satisfies(getResult -> { assertThat(getResult.type).isEqualTo(PASSWORD_MISMATCH); assertThat(getResult.record).isNull(); }); - assertThat(subscriptions.get(wrongUser, password)).succeedsWithin(DEFAULT_TIMEOUT) - .satisfies(getResult -> { + assertThat(subscriptions.get(wrongUser, password)).satisfies(getResult -> { assertThat(getResult.type).isEqualTo(NOT_STORED); assertThat(getResult.record).isNull(); }); } @Test - void testSetCustomerIdAndProcessor() throws Exception { + void testSetCustomerIdAndProcessor() { Instant subscriptionUpdated = Instant.ofEpochSecond(NOW_EPOCH_SECONDS + 1); - assertThat(subscriptions.create(user, password, created)).succeedsWithin(DEFAULT_TIMEOUT); + assertThat(subscriptions.create(user, password, created)).isNotNull(); - final CompletableFuture getUser = subscriptions.get(user, password); - assertThat(getUser).succeedsWithin(DEFAULT_TIMEOUT); - final Record userRecord = getUser.get().record; + final GetResult getUser = subscriptions.get(user, password); + final Record userRecord = getUser.record; assertThat(subscriptions.setProcessorAndCustomerId(userRecord, new ProcessorCustomer(customer, PaymentProvider.STRIPE), - subscriptionUpdated)).succeedsWithin(DEFAULT_TIMEOUT) - .hasFieldOrPropertyWithValue("processorCustomer", + subscriptionUpdated)).hasFieldOrPropertyWithValue("processorCustomer", Optional.of(new ProcessorCustomer(customer, PaymentProvider.STRIPE))); final Condition clientError409Condition = new Condition<>(e -> e instanceof ClientErrorException cee && cee.getResponse().getStatus() == 409, "Client error: 409"); // changing the customer ID is not permitted - assertThat( + + assertThatThrownBy(() -> subscriptions.setProcessorAndCustomerId(userRecord, new ProcessorCustomer(customer + "1", PaymentProvider.STRIPE), - subscriptionUpdated)).failsWithin(DEFAULT_TIMEOUT) - .withThrowableOfType(ExecutionException.class) - .withCauseInstanceOf(ClientErrorException.class) - .extracting(Throwable::getCause) + subscriptionUpdated)) + .isInstanceOf(ClientErrorException.class) .satisfies(clientError409Condition); // calling setProcessorAndCustomerId() with the same customer ID is also an error - assertThat( + assertThatThrownBy(() -> subscriptions.setProcessorAndCustomerId(userRecord, new ProcessorCustomer(customer, PaymentProvider.STRIPE), - subscriptionUpdated)).failsWithin(DEFAULT_TIMEOUT) - .withThrowableOfType(ExecutionException.class) - .withCauseInstanceOf(ClientErrorException.class) - .extracting(Throwable::getCause) + subscriptionUpdated)) + .isInstanceOf(ClientErrorException.class) .satisfies(clientError409Condition); assertThat(subscriptions.getSubscriberUserByProcessorCustomer( new ProcessorCustomer(customer, PaymentProvider.STRIPE))) - .succeedsWithin(DEFAULT_TIMEOUT). - isEqualTo(user); + .isEqualTo(user); } @Test - void testLookupByCustomerId() throws Exception { + void testLookupByCustomerId() { Instant subscriptionUpdated = Instant.ofEpochSecond(NOW_EPOCH_SECONDS + 1); - assertThat(subscriptions.create(user, password, created)).succeedsWithin(DEFAULT_TIMEOUT); + subscriptions.create(user, password, created); - final CompletableFuture getUser = subscriptions.get(user, password); - assertThat(getUser).succeedsWithin(DEFAULT_TIMEOUT); - final Record userRecord = getUser.get().record; + final GetResult getUser = subscriptions.get(user, password); + final Record userRecord = getUser.record; - assertThat(subscriptions.setProcessorAndCustomerId(userRecord, + subscriptions.setProcessorAndCustomerId(userRecord, new ProcessorCustomer(customer, PaymentProvider.STRIPE), - subscriptionUpdated)).succeedsWithin(DEFAULT_TIMEOUT); + subscriptionUpdated); assertThat(subscriptions.getSubscriberUserByProcessorCustomer( - new ProcessorCustomer(customer, PaymentProvider.STRIPE))). - succeedsWithin(DEFAULT_TIMEOUT). - isEqualTo(user); + new ProcessorCustomer(customer, PaymentProvider.STRIPE))) + .isEqualTo(user); } @Test void testSetCanceledAt() { Instant canceled = Instant.ofEpochSecond(NOW_EPOCH_SECONDS + 42); - assertThat(subscriptions.create(user, password, created)).succeedsWithin(DEFAULT_TIMEOUT); - assertThat(subscriptions.setCanceledAt(user, canceled)).succeedsWithin(DEFAULT_TIMEOUT); - assertThat(subscriptions.get(user, password)).succeedsWithin(DEFAULT_TIMEOUT).satisfies(getResult -> { + subscriptions.create(user, password, created); + subscriptions.setCanceledAt(user, canceled); + + assertThat(subscriptions.get(user, password)).satisfies(getResult -> { assertThat(getResult).isNotNull(); assertThat(getResult.type).isEqualTo(FOUND); assertThat(getResult.record).isNotNull().satisfies(record -> { @@ -195,10 +181,9 @@ class SubscriptionsTest { String subscriptionId = Base64.getEncoder().encodeToString(TestRandomUtil.nextBytes(16)); Instant subscriptionCreated = Instant.ofEpochSecond(NOW_EPOCH_SECONDS + 1); long level = 42; - assertThat(subscriptions.create(user, password, created)).succeedsWithin(DEFAULT_TIMEOUT); - assertThat(subscriptions.subscriptionCreated(user, subscriptionId, subscriptionCreated, level)). - succeedsWithin(DEFAULT_TIMEOUT); - assertThat(subscriptions.get(user, password)).succeedsWithin(DEFAULT_TIMEOUT).satisfies(getResult -> { + subscriptions.create(user, password, created); + subscriptions.subscriptionCreated(user, subscriptionId, subscriptionCreated, level); + assertThat(subscriptions.get(user, password)).satisfies(getResult -> { assertThat(getResult).isNotNull(); assertThat(getResult.type).isEqualTo(FOUND); assertThat(getResult.record).isNotNull().satisfies(record -> { @@ -217,17 +202,15 @@ class SubscriptionsTest { Instant subscriptionCreated = Instant.ofEpochSecond(NOW_EPOCH_SECONDS + 1); Instant canceledAt = subscriptionCreated.plusSeconds(1); long level = 42; - assertThat(subscriptions.create(user, password, created)).succeedsWithin(DEFAULT_TIMEOUT); - assertThat(subscriptions.subscriptionCreated(user, subscriptionId, subscriptionCreated, level)) - .succeedsWithin(DEFAULT_TIMEOUT); + subscriptions.create(user, password, created); + subscriptions.subscriptionCreated(user, subscriptionId, subscriptionCreated, level); - assertThat(subscriptions.setCanceledAt(user, canceledAt)).succeedsWithin(DEFAULT_TIMEOUT); - assertThat(subscriptions.get(user, password).join().record.canceledAt).isEqualTo(canceledAt); + subscriptions.setCanceledAt(user, canceledAt); + assertThat(subscriptions.get(user, password).record.canceledAt).isEqualTo(canceledAt); - assertThat(subscriptions.subscriptionCreated(user, subscriptionId, subscriptionCreated, level)) - .succeedsWithin(DEFAULT_TIMEOUT); + subscriptions.subscriptionCreated(user, subscriptionId, subscriptionCreated, level); - assertThat(subscriptions.get(user, password)).succeedsWithin(DEFAULT_TIMEOUT).satisfies(getResult -> { + assertThat(subscriptions.get(user, password)).satisfies(getResult -> { assertThat(getResult).isNotNull(); assertThat(getResult.type).isEqualTo(FOUND); assertThat(getResult.record).isNotNull().satisfies(record -> { @@ -246,12 +229,10 @@ class SubscriptionsTest { Instant at = Instant.ofEpochSecond(NOW_EPOCH_SECONDS + 500); long level = 1776; String updatedSubscriptionId = "new"; - assertThat(subscriptions.create(user, password, created)).succeedsWithin(DEFAULT_TIMEOUT); - assertThat(subscriptions.subscriptionCreated(user, "original", created, level - 1)).succeedsWithin( - DEFAULT_TIMEOUT); - assertThat(subscriptions.subscriptionLevelChanged(user, at, level, updatedSubscriptionId)).succeedsWithin( - DEFAULT_TIMEOUT); - assertThat(subscriptions.get(user, password)).succeedsWithin(DEFAULT_TIMEOUT).satisfies(getResult -> { + subscriptions.create(user, password, created); + subscriptions.subscriptionCreated(user, "original", created, level - 1); + subscriptions.subscriptionLevelChanged(user, at, level, updatedSubscriptionId); + assertThat(subscriptions.get(user, password)).satisfies(getResult -> { assertThat(getResult).isNotNull(); assertThat(getResult.type).isEqualTo(FOUND); assertThat(getResult.record).isNotNull().satisfies(record -> { @@ -269,16 +250,14 @@ class SubscriptionsTest { Instant canceledAt = at.plusSeconds(100); long level = 1776; String updatedSubscriptionId = "new"; - assertThat(subscriptions.create(user, password, created)).succeedsWithin(DEFAULT_TIMEOUT); - assertThat(subscriptions.subscriptionCreated(user, "original", created, level - 1)) - .succeedsWithin(DEFAULT_TIMEOUT); + subscriptions.create(user, password, created); + subscriptions.subscriptionCreated(user, "original", created, level - 1); - assertThat(subscriptions.setCanceledAt(user, canceledAt)).succeedsWithin(DEFAULT_TIMEOUT); - assertThat(subscriptions.get(user, password).join().record.canceledAt).isEqualTo(canceledAt); + subscriptions.setCanceledAt(user, canceledAt); + assertThat(subscriptions.get(user, password).record.canceledAt).isEqualTo(canceledAt); - assertThat(subscriptions.subscriptionLevelChanged(user, at, level, updatedSubscriptionId)) - .succeedsWithin(DEFAULT_TIMEOUT); - assertThat(subscriptions.get(user, password)).succeedsWithin(DEFAULT_TIMEOUT).satisfies(getResult -> { + subscriptions.subscriptionLevelChanged(user, at, level, updatedSubscriptionId); + assertThat(subscriptions.get(user, password)).satisfies(getResult -> { assertThat(getResult).isNotNull(); assertThat(getResult.type).isEqualTo(FOUND); assertThat(getResult.record).isNotNull().satisfies(record -> { @@ -297,13 +276,12 @@ class SubscriptionsTest { long level = 100; ProcessorCustomer pc = new ProcessorCustomer("customerId", PaymentProvider.GOOGLE_PLAY_BILLING); - Record record = subscriptions.create(user, password, created).join(); + Record record = assertThat(subscriptions.create(user, password, created)).isNotNull().actual(); // Should be able to set a fresh subscription - assertThat(subscriptions.setIapPurchase(record, pc, "subscriptionId", level, at)) - .succeedsWithin(DEFAULT_TIMEOUT); + subscriptions.setIapPurchase(record, pc, "subscriptionId", level, at); - record = subscriptions.get(user, password).join().record; + record = subscriptions.get(user, password).record; assertThat(record.subscriptionLevel).isEqualTo(level); assertThat(record.subscriptionLevelChangedAt).isEqualTo(at); assertThat(record.subscriptionCreatedAt).isEqualTo(at); @@ -312,31 +290,29 @@ class SubscriptionsTest { // should be able to update the level Instant nextAt = at.plus(Duration.ofSeconds(10)); long nextLevel = level + 1; - assertThat(subscriptions.setIapPurchase(record, pc, "subscriptionId", nextLevel, nextAt)) - .succeedsWithin(DEFAULT_TIMEOUT); + subscriptions.setIapPurchase(record, pc, "subscriptionId", nextLevel, nextAt); - record = subscriptions.get(user, password).join().record; + record = subscriptions.get(user, password).record; assertThat(record.subscriptionLevel).isEqualTo(nextLevel); assertThat(record.subscriptionLevelChangedAt).isEqualTo(nextAt); assertThat(record.subscriptionCreatedAt).isEqualTo(at); assertThat(record.getProcessorCustomer().orElseThrow()).isEqualTo(pc); nextAt = nextAt.plus(Duration.ofSeconds(10)); - nextLevel = level + 1; + nextLevel = nextLevel + 1; pc = new ProcessorCustomer("newCustomerId", PaymentProvider.STRIPE); try { - subscriptions.setIapPurchase(record, pc, "subscriptionId", nextLevel, nextAt).join(); + subscriptions.setIapPurchase(record, pc, "subscriptionId", nextLevel, nextAt); fail("should not be able to change the processor for an existing subscription record"); - } catch (IllegalArgumentException e) { + } catch (IllegalArgumentException _) { } // should be able to change the customerId of an existing record if the processor matches pc = new ProcessorCustomer("newCustomerId", PaymentProvider.GOOGLE_PLAY_BILLING); - assertThat(subscriptions.setIapPurchase(record, pc, "subscriptionId", nextLevel, nextAt)) - .succeedsWithin(DEFAULT_TIMEOUT); + subscriptions.setIapPurchase(record, pc, "subscriptionId", nextLevel, nextAt); - record = subscriptions.get(user, password).join().record; + record = subscriptions.get(user, password).record; assertThat(record.subscriptionLevel).isEqualTo(nextLevel); assertThat(record.subscriptionLevelChangedAt).isEqualTo(nextAt); assertThat(record.subscriptionCreatedAt).isEqualTo(at); @@ -350,26 +326,23 @@ class SubscriptionsTest { long level = 100; ProcessorCustomer pc = new ProcessorCustomer("customerId", PaymentProvider.GOOGLE_PLAY_BILLING); - Record record = subscriptions.create(user, password, created).join(); + Record record = assertThat(subscriptions.create(user, password, created)).isNotNull().actual(); // Should be able to set a fresh subscription - assertThat(subscriptions.setIapPurchase(record, pc, "subscriptionId", level, at)) - .succeedsWithin(DEFAULT_TIMEOUT); + subscriptions.setIapPurchase(record, pc, "subscriptionId", level, at); - assertThat(subscriptions.setCanceledAt(record.user, canceledAt)) - .succeedsWithin(DEFAULT_TIMEOUT); + subscriptions.setCanceledAt(record.user, canceledAt); - record = subscriptions.get(user, password).join().record; + record = subscriptions.get(user, password).record; assertThat(record.canceledAt).isEqualTo(canceledAt); // should be able to update the level Instant nextAt = at.plus(Duration.ofSeconds(10)); long nextLevel = level + 1; - assertThat(subscriptions.setIapPurchase(record, pc, "subscriptionId", nextLevel, nextAt)) - .succeedsWithin(DEFAULT_TIMEOUT); + subscriptions.setIapPurchase(record, pc, "subscriptionId", nextLevel, nextAt); // Resetting the level should clear the "canceled at" timestamp - record = subscriptions.get(user, password).join().record; + record = subscriptions.get(user, password).record; assertThat(record.canceledAt).isNull(); }