Add receipt redemption API to chat server

service/src/main/java/org/whispersystems/textsecuregcm/WhisperServerConfiguration.java

@@ -38,6 +38,7 @@ import org.whispersystems.textsecuregcm.configuration.PushConfiguration;
 import org.whispersystems.textsecuregcm.configuration.RateLimitsConfiguration;
 import org.whispersystems.textsecuregcm.configuration.RecaptchaConfiguration;
 import org.whispersystems.textsecuregcm.configuration.RecaptchaV2Configuration;
+import org.whispersystems.textsecuregcm.configuration.RedeemedReceiptsDynamoDbConfiguration;
 import org.whispersystems.textsecuregcm.configuration.RedisClusterConfiguration;
 import org.whispersystems.textsecuregcm.configuration.RedisConfiguration;
 import org.whispersystems.textsecuregcm.configuration.RemoteConfigConfiguration;
@@ -154,6 +155,11 @@ public class WhisperServerConfiguration extends Configuration {
   @JsonProperty
   private DynamoDbConfiguration deletedAccountsLockDynamoDb;
 
+  @Valid
+  @NotNull
+  @JsonProperty
+  private RedeemedReceiptsDynamoDbConfiguration redeemedReceiptsDynamoDb;
+
   @Valid
   @NotNull
   @JsonProperty
@@ -392,6 +398,10 @@ public class WhisperServerConfiguration extends Configuration {
     return deletedAccountsLockDynamoDb;
   }
 
+  public RedeemedReceiptsDynamoDbConfiguration getRedeemedReceiptsDynamoDbConfiguration() {
+    return redeemedReceiptsDynamoDb;
+  }
+
   public DatabaseConfiguration getAbuseDatabaseConfiguration() {
     return abuseDatabase;
   }

service/src/main/java/org/whispersystems/textsecuregcm/WhisperServerService.java

@@ -56,6 +56,8 @@ import org.jdbi.v3.core.Jdbi;
 import org.signal.zkgroup.ServerSecretParams;
 import org.signal.zkgroup.auth.ServerZkAuthOperations;
 import org.signal.zkgroup.profiles.ServerZkProfileOperations;
+import org.signal.zkgroup.receipts.ReceiptCredentialPresentation;
+import org.signal.zkgroup.receipts.ServerZkReceiptOperations;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.whispersystems.dispatch.DispatchManager;
@@ -175,6 +177,7 @@ import org.whispersystems.textsecuregcm.storage.ProfilesManager;
 import org.whispersystems.textsecuregcm.storage.PubSubManager;
 import org.whispersystems.textsecuregcm.storage.PushChallengeDynamoDb;
 import org.whispersystems.textsecuregcm.storage.PushFeedbackProcessor;
+import org.whispersystems.textsecuregcm.storage.RedeemedReceiptsManager;
 import org.whispersystems.textsecuregcm.storage.RemoteConfigs;
 import org.whispersystems.textsecuregcm.storage.RemoteConfigsManager;
 import org.whispersystems.textsecuregcm.storage.ReportMessageDynamoDb;
@@ -208,6 +211,7 @@ import org.whispersystems.websocket.setup.WebSocketEnvironment;
 import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
 import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
 import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient;
 import software.amazon.awssdk.services.dynamodb.DynamoDbClient;
 import software.amazon.awssdk.services.s3.S3Client;
 
@@ -286,8 +290,7 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
     environment.getObjectMapper().setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.NONE);
     environment.getObjectMapper().setVisibility(PropertyAccessor.FIELD, JsonAutoDetect.Visibility.ANY);
 
-    ProfileBadgeConverter profileBadgeConverter = new ConfiguredProfileBadgeConverter(
-        Clock.systemUTC(), config.getBadges());
+    ProfileBadgeConverter profileBadgeConverter = new ConfiguredProfileBadgeConverter(clock, config.getBadges());
 
     JdbiFactory jdbiFactory = new JdbiFactory(DefaultNameStrategy.CHECK_EMPTY);
     Jdbi        accountJdbi = jdbiFactory.build(environment, config.getAccountsDatabaseConfiguration(), "accountdb");
@@ -333,6 +336,10 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
         .withCredentials(InstanceProfileCredentialsProvider.getInstance())
         .build();
 
+    DynamoDbAsyncClient redeemedReceiptsDynamoDbClient = DynamoDbFromConfig.asyncClient(
+        config.getRedeemedReceiptsDynamoDbConfiguration(),
+        software.amazon.awssdk.auth.credentials.InstanceProfileCredentialsProvider.create());
+
     DeletedAccounts deletedAccounts = new DeletedAccounts(deletedAccountsDynamoDbClient,
         config.getDeletedAccountsDynamoDbConfiguration().getTableName(),
         config.getDeletedAccountsDynamoDbConfiguration().getNeedsReconciliationIndexName());
@@ -437,6 +444,11 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
     ProvisioningManager        provisioningManager        = new ProvisioningManager(pubSubManager);
     TorExitNodeManager         torExitNodeManager         = new TorExitNodeManager(recurringJobExecutor, config.getTorExitNodeListConfiguration());
     AsnManager                 asnManager                 = new AsnManager(recurringJobExecutor, config.getAsnTableConfiguration());
+    RedeemedReceiptsManager redeemedReceiptsManager = new RedeemedReceiptsManager(
+        clock,
+        config.getRedeemedReceiptsDynamoDbConfiguration().getTableName(),
+        redeemedReceiptsDynamoDbClient,
+        config.getRedeemedReceiptsDynamoDbConfiguration().getExpirationTime());
 
     AccountAuthenticator                  accountAuthenticator                  = new AccountAuthenticator(accountsManager);
     DisabledPermittedAccountAuthenticator disabledPermittedAccountAuthenticator = new DisabledPermittedAccountAuthenticator(accountsManager);
@@ -535,6 +547,7 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
     ServerSecretParams zkSecretParams = new ServerSecretParams(config.getZkConfig().getServerSecret());
     ServerZkProfileOperations zkProfileOperations = new ServerZkProfileOperations(zkSecretParams);
     ServerZkAuthOperations zkAuthOperations = new ServerZkAuthOperations(zkSecretParams);
+    ServerZkReceiptOperations zkReceiptOperations = new ServerZkReceiptOperations(zkSecretParams);
 
     AuthFilter<BasicCredentials, AuthenticatedAccount> accountAuthFilter = new BasicCredentialAuthFilter.Builder<AuthenticatedAccount>().setAuthenticator(
         accountAuthenticator).buildAuthFilter();
@@ -587,7 +600,8 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
         new ChallengeController(rateLimitChallengeManager),
         new DeviceController(pendingDevicesManager, accountsManager, messagesManager, keysDynamoDb, rateLimiters, config.getMaxDevices()),
         new DirectoryController(directoryCredentialsGenerator),
-        new DonationController(donationExecutor, config.getDonationConfiguration()),
+        new DonationController(clock, zkReceiptOperations, redeemedReceiptsManager, accountsManager, config.getBadges(),
+            ReceiptCredentialPresentation::new, donationExecutor, config.getDonationConfiguration()),
         new MessageController(rateLimiters, messageSender, receiptSender, accountsManager, messagesManager, unsealedSenderRateLimiter, apnFallbackManager, dynamicConfigurationManager, rateLimitChallengeManager, reportMessageManager, metricsCluster, declinedMessageReceiptExecutor, multiRecipientMessageExecutor),
         new PaymentsController(currencyManager, paymentsCredentialsGenerator),
         new ProfileController(clock, rateLimiters, accountsManager, profilesManager, usernamesManager, dynamicConfigurationManager, profileBadgeConverter, config.getBadges(), cdnS3Client, profileCdnPolicyGenerator, profileCdnPolicySigner, config.getCdnConfiguration().getBucket(), zkProfileOperations),

service/src/main/java/org/whispersystems/textsecuregcm/configuration/BadgesConfiguration.java

@@ -6,24 +6,32 @@
 package org.whispersystems.textsecuregcm.configuration;
 
 import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.annotation.JsonSetter;
 import com.fasterxml.jackson.annotation.Nulls;
+import io.dropwizard.validation.ValidationMethod;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
+import java.util.Set;
+import java.util.stream.Collectors;
 import javax.validation.Valid;
 import javax.validation.constraints.NotNull;
 
 public class BadgesConfiguration {
   private final List<BadgeConfiguration> badges;
   private final List<String> badgeIdsEnabledForAll;
+  private final Map<Long, String> receiptLevels;
 
   @JsonCreator
   public BadgesConfiguration(
       @JsonProperty("badges") @JsonSetter(nulls = Nulls.AS_EMPTY) final List<BadgeConfiguration> badges,
-      @JsonProperty("badgeIdsEnabledForAll") @JsonSetter(nulls = Nulls.AS_EMPTY) final List<String> badgeIdsEnabledForAll) {
+      @JsonProperty("badgeIdsEnabledForAll") @JsonSetter(nulls = Nulls.AS_EMPTY) final List<String> badgeIdsEnabledForAll,
+      @JsonProperty("receiptLevels") @JsonSetter(nulls = Nulls.AS_EMPTY) final Map<Long, String> receiptLevels) {
     this.badges = Objects.requireNonNull(badges);
     this.badgeIdsEnabledForAll = Objects.requireNonNull(badgeIdsEnabledForAll);
+    this.receiptLevels = Objects.requireNonNull(receiptLevels);
   }
 
   @Valid
@@ -37,4 +45,17 @@ public class BadgesConfiguration {
   public List<String> getBadgeIdsEnabledForAll() {
     return badgeIdsEnabledForAll;
   }
+
+  @Valid
+  @NotNull
+  public Map<Long, String> getReceiptLevels() {
+    return receiptLevels;
+  }
+
+  @JsonIgnore
+  @ValidationMethod(message = "contains receipt level mappings that are not configured badges")
+  public boolean isAllReceiptLevelsConfigured() {
+    final Set<String> badgeNames = badges.stream().map(BadgeConfiguration::getId).collect(Collectors.toSet());
+    return badgeNames.containsAll(receiptLevels.values());
+  }
 }

service/src/main/java/org/whispersystems/textsecuregcm/configuration/RedeemedReceiptsDynamoDbConfiguration.java

@@ -0,0 +1,21 @@
+/*
+ * Copyright 2021 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.whispersystems.textsecuregcm.configuration;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import java.time.Duration;
+import javax.validation.constraints.NotNull;
+
+public class RedeemedReceiptsDynamoDbConfiguration extends DynamoDbConfiguration {
+
+  private Duration expirationTime;
+
+  @NotNull
+  @JsonProperty
+  public Duration getExpirationTime() {
+    return expirationTime;
+  }
+}

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

@@ -18,13 +18,22 @@ import java.net.http.HttpRequest;
 import java.net.http.HttpResponse;
 import java.net.http.HttpResponse.BodyHandlers;
 import java.nio.charset.StandardCharsets;
+import java.time.Clock;
 import java.time.Duration;
+import java.time.Instant;
 import java.util.Base64;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
 import java.util.concurrent.Executor;
+import java.util.concurrent.ForkJoinPool;
+import java.util.concurrent.ForkJoinPool.ManagedBlocker;
+import java.util.function.Function;
+import javax.annotation.Nonnull;
 import javax.validation.Valid;
 import javax.ws.rs.Consumes;
 import javax.ws.rs.POST;
@@ -32,28 +41,64 @@ import javax.ws.rs.Path;
 import javax.ws.rs.Produces;
 import javax.ws.rs.core.MediaType;
 import javax.ws.rs.core.Response;
+import javax.ws.rs.core.Response.Status;
+import org.signal.zkgroup.InvalidInputException;
+import org.signal.zkgroup.VerificationFailedException;
+import org.signal.zkgroup.receipts.ReceiptCredentialPresentation;
+import org.signal.zkgroup.receipts.ReceiptSerial;
+import org.signal.zkgroup.receipts.ServerZkReceiptOperations;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.whispersystems.textsecuregcm.auth.AuthenticatedAccount;
+import org.whispersystems.textsecuregcm.configuration.BadgesConfiguration;
 import org.whispersystems.textsecuregcm.configuration.DonationConfiguration;
 import org.whispersystems.textsecuregcm.entities.ApplePayAuthorizationRequest;
 import org.whispersystems.textsecuregcm.entities.ApplePayAuthorizationResponse;
+import org.whispersystems.textsecuregcm.entities.RedeemReceiptRequest;
 import org.whispersystems.textsecuregcm.http.FaultTolerantHttpClient;
 import org.whispersystems.textsecuregcm.http.FormDataBodyPublisher;
+import org.whispersystems.textsecuregcm.storage.Account;
+import org.whispersystems.textsecuregcm.storage.AccountBadge;
+import org.whispersystems.textsecuregcm.storage.AccountsManager;
+import org.whispersystems.textsecuregcm.storage.RedeemedReceiptsManager;
 import org.whispersystems.textsecuregcm.util.SystemMapper;
 
 @Path("/v1/donation")
 public class DonationController {
 
+  public interface ReceiptCredentialPresentationFactory {
+    ReceiptCredentialPresentation build(byte[] bytes) throws InvalidInputException;
+  }
+
   private final Logger logger = LoggerFactory.getLogger(DonationController.class);
 
+  private final Clock clock;
+  private final ServerZkReceiptOperations serverZkReceiptOperations;
+  private final RedeemedReceiptsManager redeemedReceiptsManager;
+  private final AccountsManager accountsManager;
+  private final BadgesConfiguration badgesConfiguration;
+  private final ReceiptCredentialPresentationFactory receiptCredentialPresentationFactory;
   private final URI uri;
   private final String apiKey;
   private final String description;
   private final Set<String> supportedCurrencies;
   private final FaultTolerantHttpClient httpClient;
 
-  public DonationController(final Executor executor, final DonationConfiguration configuration) {
+  public DonationController(
+      @Nonnull final Clock clock,
+      @Nonnull final ServerZkReceiptOperations serverZkReceiptOperations,
+      @Nonnull final RedeemedReceiptsManager redeemedReceiptsManager,
+      @Nonnull final AccountsManager accountsManager,
+      @Nonnull final BadgesConfiguration badgesConfiguration,
+      @Nonnull final ReceiptCredentialPresentationFactory receiptCredentialPresentationFactory,
+      @Nonnull final Executor httpClientExecutor,
+      @Nonnull final DonationConfiguration configuration) {
+    this.clock = Objects.requireNonNull(clock);
+    this.serverZkReceiptOperations = Objects.requireNonNull(serverZkReceiptOperations);
+    this.redeemedReceiptsManager = Objects.requireNonNull(redeemedReceiptsManager);
+    this.accountsManager = Objects.requireNonNull(accountsManager);
+    this.badgesConfiguration = Objects.requireNonNull(badgesConfiguration);
+    this.receiptCredentialPresentationFactory = Objects.requireNonNull(receiptCredentialPresentationFactory);
     this.uri = URI.create(configuration.getUri());
     this.apiKey = configuration.getApiKey();
     this.description = configuration.getDescription();
@@ -64,12 +109,82 @@ public class DonationController {
         .withVersion(Version.HTTP_2)
         .withConnectTimeout(Duration.ofSeconds(10))
         .withRedirect(Redirect.NEVER)
-        .withExecutor(executor)
+        .withExecutor(Objects.requireNonNull(httpClientExecutor))
         .withName("donation")
         .withSecurityProtocol(FaultTolerantHttpClient.SECURITY_PROTOCOL_TLS_1_3)
         .build();
   }
 
+  @Timed
+  @POST
+  @Path("/redeem-receipt")
+  @Consumes(MediaType.APPLICATION_JSON)
+  @Produces({MediaType.APPLICATION_JSON, MediaType.TEXT_PLAIN})
+  public CompletionStage<Response> redeemReceipt(
+      @Auth final AuthenticatedAccount auth,
+      @Valid final RedeemReceiptRequest request) {
+    return CompletableFuture.supplyAsync(() -> {
+      ReceiptCredentialPresentation receiptCredentialPresentation;
+      try {
+        receiptCredentialPresentation = receiptCredentialPresentationFactory.build(
+            request.getReceiptCredentialPresentation());
+      } catch (InvalidInputException e) {
+        return CompletableFuture.completedFuture(Response.status(Status.BAD_REQUEST).entity("invalid receipt credential presentation").type(MediaType.TEXT_PLAIN_TYPE).build());
+      }
+      try {
+        serverZkReceiptOperations.verifyReceiptCredentialPresentation(receiptCredentialPresentation);
+      } catch (VerificationFailedException e) {
+        return CompletableFuture.completedFuture(Response.status(Status.BAD_REQUEST).entity("receipt credential presentation verification failed").type(MediaType.TEXT_PLAIN_TYPE).build());
+      }
+
+      final ReceiptSerial receiptSerial = receiptCredentialPresentation.getReceiptSerial();
+      final Instant receiptExpiration = Instant.ofEpochSecond(receiptCredentialPresentation.getReceiptExpirationTime());
+      final long receiptLevel = receiptCredentialPresentation.getReceiptLevel();
+      final String badgeId = badgesConfiguration.getReceiptLevels().get(receiptLevel);
+      if (badgeId == null) {
+        return CompletableFuture.completedFuture(Response.serverError().entity("server does not recognize the requested receipt level").type(MediaType.TEXT_PLAIN_TYPE).build());
+      }
+      final CompletionStage<Boolean> putStage = redeemedReceiptsManager.put(
+          receiptSerial, receiptExpiration.getEpochSecond(), receiptLevel, auth.getAccount().getUuid());
+      return putStage.thenApplyAsync(receiptMatched -> {
+        if (!receiptMatched) {
+          return Response.status(Status.BAD_REQUEST).entity("receipt serial is already redeemed").type(MediaType.TEXT_PLAIN_TYPE).build();
+        }
+
+        try {
+          ForkJoinPool.managedBlock(new ManagedBlocker() {
+            boolean done = false;
+
+            @Override
+            public boolean block() {
+              final Optional<Account> optionalAccount = accountsManager.get(auth.getAccount().getUuid());
+              optionalAccount.ifPresent(account -> {
+                accountsManager.update(account, a -> {
+                  a.addBadge(clock, new AccountBadge(badgeId, receiptExpiration, request.isVisible()));
+                  if (request.isPrimary()) {
+                    a.makeBadgePrimaryIfExists(clock, badgeId);
+                  }
+                });
+              });
+              done = true;
+              return true;
+            }
+
+            @Override
+            public boolean isReleasable() {
+              return done;
+            }
+          });
+        } catch (InterruptedException e) {
+          Thread.currentThread().interrupt();
+          return Response.serverError().build();
+        }
+
+        return Response.ok().build();
+      });
+    }).thenCompose(Function.identity());
+  }
+
   @Timed
   @POST
   @Path("/authorize-apple-pay")

service/src/main/java/org/whispersystems/textsecuregcm/entities/DeliveryCertificate.java

@@ -5,48 +5,20 @@
 
 package org.whispersystems.textsecuregcm.entities;
 
+import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
-import com.fasterxml.jackson.core.JsonGenerator;
-import com.fasterxml.jackson.core.JsonParser;
-import com.fasterxml.jackson.databind.DeserializationContext;
-import com.fasterxml.jackson.databind.JsonDeserializer;
-import com.fasterxml.jackson.databind.JsonSerializer;
-import com.fasterxml.jackson.databind.SerializerProvider;
-import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
-import com.fasterxml.jackson.databind.annotation.JsonSerialize;
-import com.google.common.annotations.VisibleForTesting;
-import java.io.IOException;
-import java.util.Base64;
 
 public class DeliveryCertificate {
 
-  @JsonProperty
-  @JsonSerialize(using = ByteArraySerializer.class)
-  @JsonDeserialize(using = ByteArrayDeserializer.class)
-  private byte[] certificate;
+  private final byte[] certificate;
 
-  public DeliveryCertificate(byte[] certificate) {
+  @JsonCreator
+  public DeliveryCertificate(
+      @JsonProperty("certificate") byte[] certificate) {
     this.certificate = certificate;
   }
 
-  public DeliveryCertificate() {}
-
-  @VisibleForTesting
   public byte[] getCertificate() {
     return certificate;
   }
-
-  public static class ByteArraySerializer extends JsonSerializer<byte[]> {
-    @Override
-    public void serialize(byte[] bytes, JsonGenerator jsonGenerator, SerializerProvider serializerProvider) throws IOException {
-      jsonGenerator.writeString(Base64.getEncoder().encodeToString(bytes));
-    }
-  }
-
-  public static class ByteArrayDeserializer extends JsonDeserializer<byte[]> {
-    @Override
-    public byte[] deserialize(JsonParser jsonParser, DeserializationContext deserializationContext) throws IOException {
-      return Base64.getDecoder().decode(jsonParser.getValueAsString());
-    }
-  }
 }

service/src/main/java/org/whispersystems/textsecuregcm/entities/GroupCredentials.java

@@ -1,32 +1,21 @@
 /*
- * Copyright 2013-2020 Signal Messenger, LLC
+ * Copyright 2021 Signal Messenger, LLC
  * SPDX-License-Identifier: AGPL-3.0-only
  */
 
 package org.whispersystems.textsecuregcm.entities;
 
+import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
-import com.fasterxml.jackson.core.JsonGenerator;
-import com.fasterxml.jackson.core.JsonParser;
-import com.fasterxml.jackson.databind.DeserializationContext;
-import com.fasterxml.jackson.databind.JsonDeserializer;
-import com.fasterxml.jackson.databind.JsonSerializer;
-import com.fasterxml.jackson.databind.SerializerProvider;
-import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
-import com.fasterxml.jackson.databind.annotation.JsonSerialize;
-
-import java.io.IOException;
-import java.util.Base64;
 import java.util.List;
 
 public class GroupCredentials {
 
-  @JsonProperty
-  private List<GroupCredential> credentials;
+  private final List<GroupCredential> credentials;
 
-  public GroupCredentials() {}
-
-  public GroupCredentials(List<GroupCredential> credentials) {
+  @JsonCreator
+  public GroupCredentials(
+      @JsonProperty("credentials") List<GroupCredential> credentials) {
     this.credentials = credentials;
   }
 
@@ -36,18 +25,14 @@ public class GroupCredentials {
 
   public static class GroupCredential {
 
-    @JsonProperty
-    @JsonSerialize(using = ByteArraySerializer.class)
-    @JsonDeserialize(using = ByteArrayDeserializer.class)
-    private byte[] credential;
+    private final byte[] credential;
+    private final int redemptionTime;
 
-    @JsonProperty
-    private int redemptionTime;
-
-    public GroupCredential() {}
-
-    public GroupCredential(byte[] credential, int redemptionTime) {
-      this.credential     = credential;
+    @JsonCreator
+    public GroupCredential(
+        @JsonProperty("credential") byte[] credential,
+        @JsonProperty("redemptionTime") int redemptionTime) {
+      this.credential = credential;
       this.redemptionTime = redemptionTime;
     }
 
@@ -59,19 +44,4 @@ public class GroupCredentials {
       return redemptionTime;
     }
   }
-
-  public static class ByteArraySerializer extends JsonSerializer<byte[]> {
-    @Override
-    public void serialize(byte[] bytes, JsonGenerator jsonGenerator, SerializerProvider serializerProvider) throws IOException {
-      jsonGenerator.writeString(Base64.getEncoder().encodeToString(bytes));
-    }
-  }
-
-  public static class ByteArrayDeserializer extends JsonDeserializer<byte[]> {
-    @Override
-    public byte[] deserialize(JsonParser jsonParser, DeserializationContext deserializationContext) throws IOException {
-      return Base64.getDecoder().decode(jsonParser.getValueAsString());
-    }
-  }
-
 }

service/src/main/java/org/whispersystems/textsecuregcm/entities/RedeemReceiptRequest.java

@@ -0,0 +1,43 @@
+/*
+ * Copyright 2021 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.whispersystems.textsecuregcm.entities;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import javax.validation.constraints.NotEmpty;
+import org.signal.zkgroup.receipts.ReceiptCredentialPresentation;
+import org.whispersystems.textsecuregcm.util.ExactlySize;
+
+public class RedeemReceiptRequest {
+
+  private final byte[] receiptCredentialPresentation;
+  private final boolean visible;
+  private final boolean primary;
+
+  @JsonCreator
+  public RedeemReceiptRequest(
+      @JsonProperty("receiptCredentialPresentation") byte[] receiptCredentialPresentation,
+      @JsonProperty("visible") boolean visible,
+      @JsonProperty("primary") boolean primary) {
+    this.receiptCredentialPresentation = receiptCredentialPresentation;
+    this.visible = visible;
+    this.primary = primary;
+  }
+
+  @NotEmpty
+  @ExactlySize({ReceiptCredentialPresentation.SIZE})
+  public byte[] getReceiptCredentialPresentation() {
+    return receiptCredentialPresentation;
+  }
+
+  public boolean isVisible() {
+    return visible;
+  }
+
+  public boolean isPrimary() {
+    return primary;
+  }
+}

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

@@ -353,6 +353,30 @@ public class Account {
     purgeStaleBadges(clock);
   }
 
+  public void makeBadgePrimaryIfExists(Clock clock, String badgeId) {
+    requireNotStale();
+
+    // early exit if it's already the first item in the list
+    if (!badges.isEmpty() && Objects.equals(badges.get(0).getId(), badgeId)) {
+      purgeStaleBadges(clock);
+      return;
+    }
+
+    int indexOfBadge = -1;
+    for (int i = 1; i < badges.size(); i++) {
+      if (Objects.equals(badgeId, badges.get(i).getId())) {
+        indexOfBadge = i;
+        break;
+      }
+    }
+
+    if (indexOfBadge != -1) {
+      badges.add(0, badges.remove(indexOfBadge));
+    }
+
+    purgeStaleBadges(clock);
+  }
+
   public void removeBadge(Clock clock, String id) {
     requireNotStale();
 

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

@@ -0,0 +1,99 @@
+/*
+ * Copyright 2021 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.whispersystems.textsecuregcm.storage;
+
+import java.time.Clock;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Map;
+import java.util.Objects;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import javax.annotation.Nonnull;
+import org.signal.zkgroup.receipts.ReceiptSerial;
+import org.whispersystems.textsecuregcm.util.AttributeValues;
+import org.whispersystems.textsecuregcm.util.UUIDUtil;
+import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient;
+import software.amazon.awssdk.services.dynamodb.model.AttributeValue;
+import software.amazon.awssdk.services.dynamodb.model.ReturnValue;
+import software.amazon.awssdk.services.dynamodb.model.UpdateItemRequest;
+
+public class RedeemedReceiptsManager {
+
+  public static final String KEY_SERIAL = "S";
+  public static final String KEY_TTL = "E";
+  public static final String KEY_RECEIPT_EXPIRATION = "G";
+  public static final String KEY_RECEIPT_LEVEL = "L";
+  public static final String KEY_ACCOUNT_UUID = "U";
+  public static final String KEY_REDEMPTION_TIME = "R";
+
+  private final Clock clock;
+  private final String table;
+  private final DynamoDbAsyncClient client;
+  private final Duration expirationTime;
+
+  public RedeemedReceiptsManager(
+      @Nonnull final Clock clock,
+      @Nonnull final String table,
+      @Nonnull final DynamoDbAsyncClient client,
+      @Nonnull final Duration expirationTime) {
+    this.clock = Objects.requireNonNull(clock);
+    this.table = Objects.requireNonNull(table);
+    this.client = Objects.requireNonNull(client);
+    this.expirationTime = Objects.requireNonNull(expirationTime);
+  }
+
+  /**
+   * Returns true either if it's able to insert a new redeemed receipt entry with the {@code receiptExpiration}, {@code
+   * receiptLevel}, and {@code accountUuid} provided or if an existing entry already exists with the same values thereby
+   * allowing idempotent request processing.
+   */
+  public CompletableFuture<Boolean> put(
+      @Nonnull final ReceiptSerial receiptSerial,
+      final long receiptExpiration,
+      final long receiptLevel,
+      @Nonnull final UUID accountUuid) {
+
+    // fail early if given bad inputs
+    Objects.requireNonNull(receiptSerial);
+    Objects.requireNonNull(accountUuid);
+
+    final Instant now = clock.instant();
+    final Instant rowExpiration = now.plus(expirationTime);
+    final AttributeValue serialAttributeValue = AttributeValues.b(receiptSerial.serialize());
+
+    final UpdateItemRequest updateItemRequest = UpdateItemRequest.builder()
+        .tableName(table)
+        .key(Map.of(KEY_SERIAL, serialAttributeValue))
+        .returnValues(ReturnValue.ALL_NEW)
+        .updateExpression("SET #ttl = if_not_exists(#ttl, :ttl), "
+            + "#receipt_expiration = if_not_exists(#receipt_expiration, :receipt_expiration), "
+            + "#receipt_level = if_not_exists(#receipt_level, :receipt_level), "
+            + "#account_uuid = if_not_exists(#account_uuid, :account_uuid), "
+            + "#redemption_time = if_not_exists(#redemption_time, :redemption_time)")
+        .expressionAttributeNames(Map.of(
+            "#ttl", KEY_TTL,
+            "#receipt_expiration", KEY_RECEIPT_EXPIRATION,
+            "#receipt_level", KEY_RECEIPT_LEVEL,
+            "#account_uuid", KEY_ACCOUNT_UUID,
+            "#redemption_time", KEY_REDEMPTION_TIME))
+        .expressionAttributeValues(Map.of(
+            ":ttl", AttributeValues.n(rowExpiration.getEpochSecond()),
+            ":receipt_expiration", AttributeValues.n(receiptExpiration),
+            ":receipt_level", AttributeValues.n(receiptLevel),
+            ":account_uuid", AttributeValues.b(accountUuid),
+            ":redemption_time", AttributeValues.n(now.getEpochSecond())))
+        .build();
+    return client.updateItem(updateItemRequest).thenApply(updateItemResponse -> {
+      final Map<String, AttributeValue> attributes = updateItemResponse.attributes();
+      final long ddbReceiptExpiration = Long.parseLong(attributes.get(KEY_RECEIPT_EXPIRATION).n());
+      final long ddbReceiptLevel = Long.parseLong(attributes.get(KEY_RECEIPT_LEVEL).n());
+      final UUID ddbAccountUuid = UUIDUtil.fromByteBuffer(attributes.get(KEY_ACCOUNT_UUID).b().asByteBuffer());
+      return ddbReceiptExpiration == receiptExpiration && ddbReceiptLevel == receiptLevel &&
+          Objects.equals(ddbAccountUuid, accountUuid);
+    });
+  }
+}

service/src/main/java/org/whispersystems/textsecuregcm/util/AttributeValues.java

@@ -15,6 +15,26 @@ import java.util.UUID;
 /** AwsAV provides static helper methods for working with AWS AttributeValues. */
 public class AttributeValues {
 
+  // Clear-type methods
+
+  public static AttributeValue b(byte[] value) {
+    return AttributeValue.builder().b(SdkBytes.fromByteArray(value)).build();
+  }
+
+  public static AttributeValue b(ByteBuffer value) {
+    return AttributeValue.builder().b(SdkBytes.fromByteBuffer(value)).build();
+  }
+
+  public static AttributeValue b(UUID value) {
+    return b(UUIDUtil.toByteBuffer(value));
+  }
+
+  public static AttributeValue n(long value) {
+    return AttributeValue.builder().n(String.valueOf(value)).build();
+  }
+
+  // More opinionated methods
+
   public static AttributeValue fromString(String value) {
     return AttributeValue.builder().s(value).build();
   }

service/src/main/java/org/whispersystems/textsecuregcm/util/DynamoDbFromConfig.java

@@ -2,22 +2,21 @@ package org.whispersystems.textsecuregcm.util;
 
 import org.whispersystems.textsecuregcm.configuration.DynamoDbConfiguration;
 import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
-import software.amazon.awssdk.core.client.config.ClientAsyncConfiguration;
 import software.amazon.awssdk.core.client.config.ClientOverrideConfiguration;
-import software.amazon.awssdk.core.client.config.SdkAdvancedAsyncClientOption;
 import software.amazon.awssdk.regions.Region;
 import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient;
 import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClientBuilder;
 import software.amazon.awssdk.services.dynamodb.DynamoDbClient;
-import java.util.concurrent.Executor;
 
 public class DynamoDbFromConfig {
+
   private static ClientOverrideConfiguration clientOverrideConfiguration(DynamoDbConfiguration config) {
     return ClientOverrideConfiguration.builder()
         .apiCallTimeout(config.getClientExecutionTimeout())
         .apiCallAttemptTimeout(config.getClientRequestTimeout())
         .build();
   }
+
   public static DynamoDbClient client(DynamoDbConfiguration config, AwsCredentialsProvider credentialsProvider) {
     return DynamoDbClient.builder()
         .region(Region.of(config.getRegion()))
@@ -25,17 +24,13 @@ public class DynamoDbFromConfig {
         .overrideConfiguration(clientOverrideConfiguration(config))
         .build();
   }
-  public static DynamoDbAsyncClient asyncClient(DynamoDbConfiguration config, AwsCredentialsProvider credentialsProvider, Executor executor) {
+
+  public static DynamoDbAsyncClient asyncClient(
+      DynamoDbConfiguration config, AwsCredentialsProvider credentialsProvider) {
     DynamoDbAsyncClientBuilder builder = DynamoDbAsyncClient.builder()
         .region(Region.of(config.getRegion()))
         .credentialsProvider(credentialsProvider)
         .overrideConfiguration(clientOverrideConfiguration(config));
-    if (executor != null) {
-      builder.asyncConfiguration(ClientAsyncConfiguration.builder()
-          .advancedOption(SdkAdvancedAsyncClientOption.FUTURE_COMPLETION_EXECUTOR,
-              executor)
-          .build());
-    }
     return builder.build();
   }
 }

service/src/main/java/org/whispersystems/textsecuregcm/util/ExactlySize.java

@@ -5,19 +5,24 @@
 
 package org.whispersystems.textsecuregcm.util;
 
-import javax.validation.Constraint;
-import javax.validation.Payload;
-import java.lang.annotation.Documented;
-import java.lang.annotation.Retention;
-import java.lang.annotation.Target;
-
-import static java.lang.annotation.ElementType.*;
+import static java.lang.annotation.ElementType.ANNOTATION_TYPE;
+import static java.lang.annotation.ElementType.FIELD;
+import static java.lang.annotation.ElementType.METHOD;
 import static java.lang.annotation.ElementType.PARAMETER;
 import static java.lang.annotation.RetentionPolicy.RUNTIME;
 
+import java.lang.annotation.Documented;
+import java.lang.annotation.Retention;
+import java.lang.annotation.Target;
+import javax.validation.Constraint;
+import javax.validation.Payload;
+
 @Target({ FIELD, METHOD, PARAMETER, ANNOTATION_TYPE })
 @Retention(RUNTIME)
-@Constraint(validatedBy = ExactlySizeValidator.class)
+@Constraint(validatedBy = {
+    ExactlySizeValidatorForString.class,
+    ExactlySizeValidatorForArraysOfByte.class,
+})
 @Documented
 public @interface ExactlySize {
 

service/src/main/java/org/whispersystems/textsecuregcm/util/ExactlySizeValidator.java

@@ -1,34 +1,29 @@
 /*
- * Copyright 2013-2020 Signal Messenger, LLC
+ * Copyright 2021 Signal Messenger, LLC
  * SPDX-License-Identifier: AGPL-3.0-only
  */
 
 package org.whispersystems.textsecuregcm.util;
 
+import java.util.Arrays;
+import java.util.Set;
+import java.util.stream.Collectors;
 import javax.validation.ConstraintValidator;
 import javax.validation.ConstraintValidatorContext;
 
+public abstract class ExactlySizeValidator<T> implements ConstraintValidator<ExactlySize, T> {
 
-public class ExactlySizeValidator implements ConstraintValidator<ExactlySize, String> {
-
-  private int[] permittedSizes;
+  private Set<Integer> permittedSizes;
 
   @Override
-  public void initialize(ExactlySize exactlySize) {
-    this.permittedSizes = exactlySize.value();
+  public void initialize(ExactlySize annotation) {
+    permittedSizes = Arrays.stream(annotation.value()).boxed().collect(Collectors.toSet());
   }
 
   @Override
-  public boolean isValid(String object, ConstraintValidatorContext constraintContext) {
-    int objectLength;
-
-    if (object == null) objectLength = 0;
-    else                objectLength = object.length();
-
-    for (int permittedSize : permittedSizes) {
-      if (permittedSize == objectLength) return true;
-    }
-
-    return false;
+  public boolean isValid(T value, ConstraintValidatorContext context) {
+    return permittedSizes.contains(size(value));
   }
+
+  protected abstract int size(T value);
 }

service/src/main/java/org/whispersystems/textsecuregcm/util/ExactlySizeValidatorForArraysOfByte.java

@@ -0,0 +1,14 @@
+/*
+ * Copyright 2021 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.whispersystems.textsecuregcm.util;
+
+public class ExactlySizeValidatorForArraysOfByte extends ExactlySizeValidator<byte[]> {
+
+  @Override
+  protected int size(final byte[] value) {
+    return value == null ? 0 : value.length;
+  }
+}

service/src/main/java/org/whispersystems/textsecuregcm/util/ExactlySizeValidatorForString.java

@@ -0,0 +1,15 @@
+/*
+ * Copyright 2013-2020 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.whispersystems.textsecuregcm.util;
+
+
+public class ExactlySizeValidatorForString extends ExactlySizeValidator<String> {
+
+  @Override
+  protected int size(final String value) {
+    return value == null ? 0 : value.length();
+  }
+}