Add plumbing and configuration to migrate pending accounts/devices to DynamoDB.

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

@@ -167,6 +167,16 @@ public class WhisperServerConfiguration extends Configuration {
   @JsonProperty
   private DynamoDbConfiguration reportMessageDynamoDb;
 
+  @Valid
+  @NotNull
+  @JsonProperty
+  private DynamoDbConfiguration pendingAccountsDynamoDb;
+
+  @Valid
+  @NotNull
+  @JsonProperty
+  private DynamoDbConfiguration pendingDevicesDynamoDb;
+
   @Valid
   @NotNull
   @JsonProperty
@@ -465,6 +475,14 @@ public class WhisperServerConfiguration extends Configuration {
     return reportMessageDynamoDb;
   }
 
+  public DynamoDbConfiguration getPendingAccountsDynamoDbConfiguration() {
+    return pendingAccountsDynamoDb;
+  }
+
+  public DynamoDbConfiguration getPendingDevicesDynamoDbConfiguration() {
+    return pendingDevicesDynamoDb;
+  }
+
   public MonitoredS3ObjectConfiguration getTorExitNodeListConfiguration() {
     return torExitNodeList;
   }

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

@@ -184,6 +184,7 @@ import org.whispersystems.textsecuregcm.storage.ReportMessageManager;
 import org.whispersystems.textsecuregcm.storage.ReservedUsernames;
 import org.whispersystems.textsecuregcm.storage.Usernames;
 import org.whispersystems.textsecuregcm.storage.UsernamesManager;
+import org.whispersystems.textsecuregcm.storage.VerificationCodeStoreDynamoDb;
 import org.whispersystems.textsecuregcm.util.AsnManager;
 import org.whispersystems.textsecuregcm.util.Constants;
 import org.whispersystems.textsecuregcm.util.DynamoDbFromConfig;
@@ -368,6 +369,12 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
     DynamoDbClient migrationRetryAccountsDynamoDb = DynamoDbFromConfig.client(config.getMigrationRetryAccountsDynamoDbConfiguration(),
         software.amazon.awssdk.auth.credentials.InstanceProfileCredentialsProvider.create());
 
+    DynamoDbClient pendingAccountsDynamoDbClient = DynamoDbFromConfig.client(config.getPendingAccountsDynamoDbConfiguration(),
+        software.amazon.awssdk.auth.credentials.InstanceProfileCredentialsProvider.create());
+
+    DynamoDbClient pendingDevicesDynamoDbClient = DynamoDbFromConfig.client(config.getPendingDevicesDynamoDbConfiguration(),
+        software.amazon.awssdk.auth.credentials.InstanceProfileCredentialsProvider.create());
+
     MigrationDeletedAccounts migrationDeletedAccounts = new MigrationDeletedAccounts(recentlyDeletedAccountsDynamoDb, config.getMigrationDeletedAccountsDynamoDbConfiguration().getTableName());
     MigrationRetryAccounts migrationRetryAccounts = new MigrationRetryAccounts(migrationRetryAccountsDynamoDb, config.getMigrationRetryAccountsDynamoDbConfiguration().getTableName());
 
@@ -384,6 +391,8 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
     RemoteConfigs     remoteConfigs     = new RemoteConfigs(accountDatabase);
     PushChallengeDynamoDb pushChallengeDynamoDb = new PushChallengeDynamoDb(pushChallengeDynamoDbClient, config.getPushChallengeDynamoDbConfiguration().getTableName());
     ReportMessageDynamoDb reportMessageDynamoDb = new ReportMessageDynamoDb(reportMessageDynamoDbClient, config.getReportMessageDynamoDbConfiguration().getTableName());
+    VerificationCodeStoreDynamoDb pendingAccountsDynamoDb = new VerificationCodeStoreDynamoDb(pendingAccountsDynamoDbClient, config.getPendingAccountsDynamoDbConfiguration().getTableName());
+    VerificationCodeStoreDynamoDb pendingDevicesDynamoDb = new VerificationCodeStoreDynamoDb(pendingDevicesDynamoDbClient, config.getPendingDevicesDynamoDbConfiguration().getTableName());
 
     RedisClientFactory  pubSubClientFactory = new RedisClientFactory("pubsub_cache", config.getPubsubCacheConfiguration().getUrl(), config.getPubsubCacheConfiguration().getReplicaUrls(), config.getPubsubCacheConfiguration().getCircuitBreakerConfiguration());
     ReplicatedJedisPool pubsubClient        = pubSubClientFactory.getRedisClientPool();
@@ -440,8 +449,8 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
     SecureStorageClient        secureStorageClient        = new SecureStorageClient(storageCredentialsGenerator, storageServiceExecutor, config.getSecureStorageServiceConfiguration());
     ClientPresenceManager      clientPresenceManager      = new ClientPresenceManager(clientPresenceCluster, recurringJobExecutor, keyspaceNotificationDispatchExecutor);
     DirectoryQueue             directoryQueue             = new DirectoryQueue(config.getDirectoryConfiguration().getSqsConfiguration());
-    PendingAccountsManager     pendingAccountsManager     = new PendingAccountsManager(pendingAccounts);
+    PendingAccountsManager     pendingAccountsManager     = new PendingAccountsManager(pendingAccounts, pendingAccountsDynamoDb, dynamicConfigurationManager);
-    PendingDevicesManager      pendingDevicesManager      = new PendingDevicesManager(pendingDevices);
+    PendingDevicesManager      pendingDevicesManager      = new PendingDevicesManager(pendingDevices, pendingDevicesDynamoDb, dynamicConfigurationManager);
     UsernamesManager           usernamesManager           = new UsernamesManager(usernames, reservedUsernames, cacheCluster);
     ProfilesManager            profilesManager            = new ProfilesManager(profiles, cacheCluster);
     MessagesCache              messagesCache              = new MessagesCache(messagesCluster, messagesCluster, keyspaceNotificationDispatchExecutor);

service/src/main/java/org/whispersystems/textsecuregcm/configuration/dynamic/DynamicConfiguration.java

@@ -51,6 +51,14 @@ public class DynamicConfiguration {
   @Valid
   private DynamicRateLimitChallengeConfiguration rateLimitChallenge = new DynamicRateLimitChallengeConfiguration();
 
+  @JsonProperty
+  @Valid
+  private DynamicVerificationCodeStoreMigrationConfiguration pendingAccountsMigration = new DynamicVerificationCodeStoreMigrationConfiguration();
+
+  @JsonProperty
+  @Valid
+  private DynamicVerificationCodeStoreMigrationConfiguration pendingDevicesMigration = new DynamicVerificationCodeStoreMigrationConfiguration();
+
   public Optional<DynamicExperimentEnrollmentConfiguration> getExperimentEnrollmentConfiguration(
       final String experimentName) {
     return Optional.ofNullable(experiments.get(experimentName));
@@ -101,4 +109,12 @@ public class DynamicConfiguration {
   public DynamicRateLimitChallengeConfiguration getRateLimitChallengeConfiguration() {
     return rateLimitChallenge;
   }
+
+  public DynamicVerificationCodeStoreMigrationConfiguration getPendingAccountsMigrationConfiguration() {
+    return pendingAccountsMigration;
+  }
+
+  public DynamicVerificationCodeStoreMigrationConfiguration getPendingDevicesMigrationConfiguration() {
+    return pendingDevicesMigration;
+  }
 }

service/src/main/java/org/whispersystems/textsecuregcm/configuration/dynamic/DynamicVerificationCodeStoreMigrationConfiguration.java

@@ -0,0 +1,51 @@
+/*
+ * Copyright 2013-2021 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.whispersystems.textsecuregcm.configuration.dynamic;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import javax.validation.constraints.NotNull;
+
+public class DynamicVerificationCodeStoreMigrationConfiguration {
+
+  public enum WriteDestination {
+    POSTGRES,
+    DYNAMODB
+  }
+
+  @JsonProperty
+  @NotNull
+  private WriteDestination writeDestination = WriteDestination.POSTGRES;
+
+  @JsonProperty
+  private boolean readPostgres = true;
+
+  @JsonProperty
+  private boolean readDynamoDb = false;
+
+  public WriteDestination getWriteDestination() {
+    return writeDestination;
+  }
+
+  public void setWriteDestination(final WriteDestination writeDestination) {
+    this.writeDestination = writeDestination;
+  }
+
+  public boolean isReadPostgres() {
+    return readPostgres;
+  }
+
+  public void setReadPostgres(final boolean readPostgres) {
+    this.readPostgres = readPostgres;
+  }
+
+  public boolean isReadDynamoDb() {
+    return readDynamoDb;
+  }
+
+  public void setReadDynamoDb(final boolean readDynamoDb) {
+    this.readDynamoDb = readDynamoDb;
+  }
+}

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

@@ -10,20 +10,46 @@ import org.whispersystems.textsecuregcm.auth.StoredVerificationCode;
 public class PendingAccountsManager {
 
   private final PendingAccounts pendingAccounts;
+  private final VerificationCodeStoreDynamoDb pendingAccountsDynamoDb;
+  private final DynamicConfigurationManager dynamicConfigurationManager;
+
+  public PendingAccountsManager(
+      final PendingAccounts pendingAccounts,
+      final VerificationCodeStoreDynamoDb pendingAccountsDynamoDb,
+      final DynamicConfigurationManager dynamicConfigurationManager) {
 
-  public PendingAccountsManager(PendingAccounts pendingAccounts) {
     this.pendingAccounts = pendingAccounts;
+    this.pendingAccountsDynamoDb = pendingAccountsDynamoDb;
+    this.dynamicConfigurationManager = dynamicConfigurationManager;
   }
 
   public void store(String number, StoredVerificationCode code) {
-    pendingAccounts.insert(number, code);
+    switch (dynamicConfigurationManager.getConfiguration().getPendingAccountsMigrationConfiguration().getWriteDestination()) {
+
+      case POSTGRES:
+        pendingAccounts.insert(number, code);
+        break;
+
+      case DYNAMODB:
+        pendingAccountsDynamoDb.insert(number, code);
+        break;
+    }
   }
 
   public void remove(String number) {
     pendingAccounts.remove(number);
+    pendingAccountsDynamoDb.remove(number);
   }
 
   public Optional<StoredVerificationCode> getCodeForNumber(String number) {
-    return pendingAccounts.findForNumber(number);
+    final Optional<StoredVerificationCode> maybeCodeFromPostgres =
+        dynamicConfigurationManager.getConfiguration().getPendingAccountsMigrationConfiguration().isReadPostgres()
+            ? pendingAccounts.findForNumber(number)
+            : Optional.empty();
+
+    return maybeCodeFromPostgres.or(
+        () -> dynamicConfigurationManager.getConfiguration().getPendingAccountsMigrationConfiguration().isReadDynamoDb()
+            ? pendingAccountsDynamoDb.findForNumber(number)
+            : Optional.empty());
   }
 }

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

@@ -10,20 +10,45 @@ import org.whispersystems.textsecuregcm.auth.StoredVerificationCode;
 public class PendingDevicesManager {
 
   private final PendingDevices pendingDevices;
+  private final VerificationCodeStoreDynamoDb pendingDevicesDynamoDb;
+  private final DynamicConfigurationManager dynamicConfigurationManager;
+
+  public PendingDevicesManager(
+      final PendingDevices pendingDevices,
+      final VerificationCodeStoreDynamoDb pendingDevicesDynamoDb,
+      final DynamicConfigurationManager dynamicConfigurationManager) {
 
-  public PendingDevicesManager(PendingDevices pendingDevices) {
     this.pendingDevices = pendingDevices;
+    this.pendingDevicesDynamoDb = pendingDevicesDynamoDb;
+    this.dynamicConfigurationManager = dynamicConfigurationManager;
   }
 
   public void store(String number, StoredVerificationCode code) {
-    pendingDevices.insert(number, code);
+    switch (dynamicConfigurationManager.getConfiguration().getPendingDevicesMigrationConfiguration().getWriteDestination()) {
+      case POSTGRES:
+        pendingDevices.insert(number, code);
+        break;
+
+      case DYNAMODB:
+        pendingDevicesDynamoDb.insert(number, code);
+        break;
+    }
   }
 
   public void remove(String number) {
     pendingDevices.remove(number);
+    pendingDevicesDynamoDb.remove(number);
   }
 
   public Optional<StoredVerificationCode> getCodeForNumber(String number) {
-    return pendingDevices.findForNumber(number);
+    final Optional<StoredVerificationCode> maybeCodeFromPostgres =
+        dynamicConfigurationManager.getConfiguration().getPendingDevicesMigrationConfiguration().isReadPostgres()
+            ? pendingDevices.findForNumber(number)
+            : Optional.empty();
+
+    return maybeCodeFromPostgres.or(
+        () -> dynamicConfigurationManager.getConfiguration().getPendingDevicesMigrationConfiguration().isReadDynamoDb()
+            ? pendingDevicesDynamoDb.findForNumber(number)
+            : Optional.empty());
   }
 }