Add plumbing for deleting accounts and all associated data.

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

@@ -292,12 +292,12 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
     DirectoryQueue             directoryQueue             = new DirectoryQueue(config.getDirectoryConfiguration().getSqsConfiguration());
     PendingAccountsManager     pendingAccountsManager     = new PendingAccountsManager(pendingAccounts, cacheCluster);
     PendingDevicesManager      pendingDevicesManager      = new PendingDevicesManager(pendingDevices, cacheCluster);
-    AccountsManager            accountsManager            = new AccountsManager(accounts, directory, cacheCluster);
     UsernamesManager           usernamesManager           = new UsernamesManager(usernames, reservedUsernames, cacheCluster);
     ProfilesManager            profilesManager            = new ProfilesManager(profiles, cacheCluster);
     MessagesCache              messagesCache              = new MessagesCache(messagesCacheCluster, keyspaceNotificationDispatchExecutor);
     PushLatencyManager         pushLatencyManager         = new PushLatencyManager(metricsCluster);
     MessagesManager            messagesManager            = new MessagesManager(messages, messagesCache, pushLatencyManager);
+    AccountsManager            accountsManager            = new AccountsManager(accounts, directory, cacheCluster, directoryQueue, keys, messagesManager, usernamesManager, profilesManager);
     RemoteConfigsManager       remoteConfigsManager       = new RemoteConfigsManager(remoteConfigs);
     FeatureFlagsManager        featureFlagsManager        = new FeatureFlagsManager(featureFlags, recurringJobExecutor);
     DeadLetterHandler          deadLetterHandler          = new DeadLetterHandler(accountsManager, messagesManager);

service/src/main/java/org/whispersystems/textsecuregcm/sqs/DirectoryQueue.java

@@ -70,6 +70,10 @@ public class DirectoryQueue {
     sendMessage(account.isEnabled() && account.isDiscoverableByPhoneNumber() ? "add" : "delete", account.getUuid(), account.getNumber());
   }
 
+  public void deleteAccount(final Account account) {
+    sendMessage("delete", account.getUuid(), account.getNumber());
+  }
+
   private void sendMessage(String action, UUID uuid, String number) {
     final Map<String, MessageAttributeValue> messageAttributes = new HashMap<>();
     messageAttributes.put("id", new MessageAttributeValue().withDataType("String").withStringValue(number));

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

@@ -29,6 +29,7 @@ import org.slf4j.LoggerFactory;
 import org.whispersystems.textsecuregcm.auth.AmbiguousIdentifier;
 import org.whispersystems.textsecuregcm.entities.ClientContact;
 import org.whispersystems.textsecuregcm.redis.FaultTolerantRedisCluster;
+import org.whispersystems.textsecuregcm.sqs.DirectoryQueue;
 import org.whispersystems.textsecuregcm.util.Constants;
 import org.whispersystems.textsecuregcm.util.SystemMapper;
 import org.whispersystems.textsecuregcm.util.Util;
@@ -51,19 +52,30 @@ public class AccountsManager {
   private static final Timer redisSetTimer       = metricRegistry.timer(name(AccountsManager.class, "redisSet"      ));
   private static final Timer redisNumberGetTimer = metricRegistry.timer(name(AccountsManager.class, "redisNumberGet"));
   private static final Timer redisUuidGetTimer   = metricRegistry.timer(name(AccountsManager.class, "redisUuidGet"  ));
+  private static final Timer redisDeleteTimer    = metricRegistry.timer(name(AccountsManager.class, "redisDelete"   ));
 
   private final Logger logger = LoggerFactory.getLogger(AccountsManager.class);
 
   private final Accounts                  accounts;
   private final FaultTolerantRedisCluster cacheCluster;
   private final DirectoryManager          directory;
+  private final DirectoryQueue            directoryQueue;
+  private final Keys                      keys;
+  private final MessagesManager           messagesManager;
+  private final UsernamesManager          usernamesManager;
+  private final ProfilesManager           profilesManager;
   private final ObjectMapper              mapper;
 
-  public AccountsManager(Accounts accounts, DirectoryManager directory, FaultTolerantRedisCluster cacheCluster) {
+  public AccountsManager(Accounts accounts, DirectoryManager directory, FaultTolerantRedisCluster cacheCluster, final DirectoryQueue directoryQueue, final Keys keys, final MessagesManager messagesManager, final UsernamesManager usernamesManager, final ProfilesManager profilesManager) {
-    this.accounts               = accounts;
+    this.accounts         = accounts;
-    this.directory              = directory;
+    this.directory        = directory;
-    this.cacheCluster           = cacheCluster;
+    this.cacheCluster     = cacheCluster;
-    this.mapper                 = SystemMapper.getMapper();
+    this.directoryQueue   = directoryQueue;
+    this.keys             = keys;
+    this.messagesManager  = messagesManager;
+    this.usernamesManager = usernamesManager;
+    this.profilesManager  = profilesManager;
+    this.mapper           = SystemMapper.getMapper();
   }
 
   public boolean create(Account account) {
@@ -125,6 +137,17 @@ public class AccountsManager {
     return accounts.getAllFrom(uuid, length);
   }
 
+  public void delete(final Account account) {
+    usernamesManager.delete(account.getUuid());
+    directoryQueue.deleteAccount(account);
+    directory.remove(account.getNumber());
+    profilesManager.deleteAll(account.getUuid());
+    keys.delete(account.getNumber());
+    messagesManager.clear(account.getNumber(), account.getUuid());
+    redisDelete(account);
+    databaseDelete(account);
+  }
+
   private void updateDirectory(Account account) {
     if (account.isEnabled()) {
       byte[]        token         = Util.getContactToken(account.getNumber());
@@ -194,6 +217,12 @@ public class AccountsManager {
     }
   }
 
+  private void redisDelete(final Account account) {
+    try (final Timer.Context ignored = redisDeleteTimer.time()) {
+      cacheCluster.useCluster(connection -> connection.sync().del(getAccountMapKey(account.getNumber()), getAccountEntityKey(account.getUuid())));
+    }
+  }
+
   private Optional<Account> databaseGet(String number) {
     return accounts.get(number);
   }
@@ -209,4 +238,8 @@ public class AccountsManager {
   private void databaseUpdate(Account account) {
     accounts.update(account);
   }
+
+  private void databaseDelete(final Account account) {
+    accounts.delete(account.getUuid());
+  }
 }

service/src/main/java/org/whispersystems/textsecuregcm/workers/DeleteUserCommand.java

@@ -11,7 +11,7 @@ import org.jdbi.v3.core.Jdbi;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.whispersystems.textsecuregcm.WhisperServerConfiguration;
-import org.whispersystems.textsecuregcm.auth.AuthenticationCredentials;
+import org.whispersystems.textsecuregcm.metrics.PushLatencyManager;
 import org.whispersystems.textsecuregcm.providers.RedisClientFactory;
 import org.whispersystems.textsecuregcm.redis.FaultTolerantRedisCluster;
 import org.whispersystems.textsecuregcm.redis.ReplicatedJedisPool;
@@ -19,13 +19,22 @@ import org.whispersystems.textsecuregcm.sqs.DirectoryQueue;
 import org.whispersystems.textsecuregcm.storage.Account;
 import org.whispersystems.textsecuregcm.storage.Accounts;
 import org.whispersystems.textsecuregcm.storage.AccountsManager;
-import org.whispersystems.textsecuregcm.storage.Device;
 import org.whispersystems.textsecuregcm.storage.DirectoryManager;
 import org.whispersystems.textsecuregcm.storage.FaultTolerantDatabase;
-import org.whispersystems.textsecuregcm.util.Base64;
+import org.whispersystems.textsecuregcm.storage.Keys;
+import org.whispersystems.textsecuregcm.storage.Messages;
+import org.whispersystems.textsecuregcm.storage.MessagesCache;
+import org.whispersystems.textsecuregcm.storage.MessagesManager;
+import org.whispersystems.textsecuregcm.storage.Profiles;
+import org.whispersystems.textsecuregcm.storage.ProfilesManager;
+import org.whispersystems.textsecuregcm.storage.ReservedUsernames;
+import org.whispersystems.textsecuregcm.storage.Usernames;
+import org.whispersystems.textsecuregcm.storage.UsernamesManager;
 
-import java.security.SecureRandom;
 import java.util.Optional;
+import java.util.concurrent.ExecutorService;
+
+import static com.codahale.metrics.MetricRegistry.name;
 
 public class DeleteUserCommand extends EnvironmentCommand<WhisperServerConfiguration> {
 
@@ -64,39 +73,40 @@ public class DeleteUserCommand extends EnvironmentCommand<WhisperServerConfigura
 
       JdbiFactory           jdbiFactory     = new JdbiFactory();
       Jdbi                  accountJdbi     = jdbiFactory.build(environment, configuration.getAccountsDatabaseConfiguration(), "accountdb");
+      Jdbi                  messageJdbi     = jdbiFactory.build(environment, configuration.getMessageStoreConfiguration(), "messagedb" );
       FaultTolerantDatabase accountDatabase = new FaultTolerantDatabase("account_database_delete_user", accountJdbi, configuration.getAccountsDatabaseConfiguration().getCircuitBreakerConfiguration());
+      FaultTolerantDatabase messageDatabase = new FaultTolerantDatabase("message_database", messageJdbi, configuration.getMessageStoreConfiguration().getCircuitBreakerConfiguration());
 
       FaultTolerantRedisCluster cacheCluster = new FaultTolerantRedisCluster("main_cache_cluster", configuration.getCacheClusterConfiguration());
 
-      Accounts            accounts        = new Accounts(accountDatabase);
+      ExecutorService keyspaceNotificationDispatchExecutor = environment.lifecycle().executorService(name(getClass(), "keyspaceNotification-%d")).maxThreads(4).build();
-      ReplicatedJedisPool redisClient     = new RedisClientFactory("directory_cache_delete_command", configuration.getDirectoryConfiguration().getRedisConfiguration().getUrl(), configuration.getDirectoryConfiguration().getRedisConfiguration().getReplicaUrls(), configuration.getDirectoryConfiguration().getRedisConfiguration().getCircuitBreakerConfiguration()).getRedisClientPool();
+
-      DirectoryQueue      directoryQueue  = new DirectoryQueue  (configuration.getDirectoryConfiguration().getSqsConfiguration());
+      Accounts                  accounts             = new Accounts(accountDatabase);
-      DirectoryManager    directory       = new DirectoryManager(redisClient                                                    );
+      Usernames                 usernames            = new Usernames(accountDatabase);
-      AccountsManager     accountsManager = new AccountsManager(accounts, directory, cacheCluster);
+      Profiles                  profiles             = new Profiles(accountDatabase);
+      ReservedUsernames         reservedUsernames    = new ReservedUsernames(accountDatabase);
+      Keys                      keys                 = new Keys(accountDatabase);
+      Messages                  messages             = new Messages(messageDatabase);
+      ReplicatedJedisPool       redisClient          = new RedisClientFactory("directory_cache_delete_command", configuration.getDirectoryConfiguration().getRedisConfiguration().getUrl(), configuration.getDirectoryConfiguration().getRedisConfiguration().getReplicaUrls(), configuration.getDirectoryConfiguration().getRedisConfiguration().getCircuitBreakerConfiguration()).getRedisClientPool();
+      FaultTolerantRedisCluster messagesCacheCluster = new FaultTolerantRedisCluster("messages_cluster", configuration.getMessageCacheConfiguration().getRedisClusterConfiguration());
+      FaultTolerantRedisCluster metricsCluster       = new FaultTolerantRedisCluster("metrics_cluster", configuration.getMetricsClusterConfiguration());
+      MessagesCache             messagesCache        = new MessagesCache(messagesCacheCluster, keyspaceNotificationDispatchExecutor);
+      PushLatencyManager        pushLatencyManager   = new PushLatencyManager(metricsCluster);
+      DirectoryQueue            directoryQueue       = new DirectoryQueue  (configuration.getDirectoryConfiguration().getSqsConfiguration());
+      DirectoryManager          directory            = new DirectoryManager(redisClient                                                    );
+      UsernamesManager          usernamesManager     = new UsernamesManager(usernames, reservedUsernames, cacheCluster);
+      ProfilesManager           profilesManager      = new ProfilesManager(profiles, cacheCluster);
+      MessagesManager           messagesManager      = new MessagesManager(messages, messagesCache, pushLatencyManager);
+      AccountsManager           accountsManager      = new AccountsManager(accounts, directory, cacheCluster, directoryQueue, keys, messagesManager, usernamesManager, profilesManager);
 
       for (String user: users) {
         Optional<Account> account = accountsManager.get(user);
 
         if (account.isPresent()) {
-          Optional<Device> device = account.get().getDevice(1);
+          accountsManager.delete(account.get());
-
+          logger.warn("Removed " + account.get().getNumber());
-          if (device.isPresent()) {
-            byte[] random = new byte[16];
-            new SecureRandom().nextBytes(random);
-
-            device.get().setGcmId(null);
-            device.get().setFetchesMessages(false);
-            device.get().setAuthenticationCredentials(new AuthenticationCredentials(Base64.encodeBytes(random)));
-
-            accountsManager.update(account.get());
-            directoryQueue.refreshRegisteredUser(account.get());
-
-            logger.warn("Removed " + account.get().getNumber());
-          } else {
-            logger.warn("No primary device found...");
-          }
         } else {
-          logger.warn("Account not found...");
+          logger.warn("Account not found");
         }
       }
     } catch (Exception ex) {

service/src/test/java/org/whispersystems/textsecuregcm/tests/storage/AccountsManagerTest.java

@@ -3,11 +3,17 @@ package org.whispersystems.textsecuregcm.tests.storage;
 import io.lettuce.core.RedisException;
 import io.lettuce.core.cluster.api.sync.RedisAdvancedClusterCommands;
 import org.junit.Test;
+import org.whispersystems.textsecuregcm.entities.Profile;
 import org.whispersystems.textsecuregcm.redis.FaultTolerantRedisCluster;
+import org.whispersystems.textsecuregcm.sqs.DirectoryQueue;
 import org.whispersystems.textsecuregcm.storage.Account;
 import org.whispersystems.textsecuregcm.storage.Accounts;
 import org.whispersystems.textsecuregcm.storage.AccountsManager;
 import org.whispersystems.textsecuregcm.storage.DirectoryManager;
+import org.whispersystems.textsecuregcm.storage.Keys;
+import org.whispersystems.textsecuregcm.storage.MessagesManager;
+import org.whispersystems.textsecuregcm.storage.ProfilesManager;
+import org.whispersystems.textsecuregcm.storage.UsernamesManager;
 import org.whispersystems.textsecuregcm.tests.util.RedisClusterHelper;
 
 import java.util.HashSet;
@@ -33,13 +39,18 @@ public class AccountsManagerTest {
     FaultTolerantRedisCluster                    cacheCluster     = RedisClusterHelper.buildMockRedisCluster(commands);
     Accounts                                     accounts         = mock(Accounts.class);
     DirectoryManager                             directoryManager = mock(DirectoryManager.class);
+    DirectoryQueue                               directoryQueue   = mock(DirectoryQueue.class);
+    Keys                                         keys             = mock(Keys.class);
+    MessagesManager                              messagesManager  = mock(MessagesManager.class);
+    UsernamesManager                             usernamesManager = mock(UsernamesManager.class);
+    ProfilesManager                              profilesManager  = mock(ProfilesManager.class);
 
     UUID uuid = UUID.randomUUID();
 
     when(commands.get(eq("AccountMap::+14152222222"))).thenReturn(uuid.toString());
     when(commands.get(eq("Account3::" + uuid.toString()))).thenReturn("{\"number\": \"+14152222222\", \"name\": \"test\"}");
 
-    AccountsManager   accountsManager = new AccountsManager(accounts, directoryManager, cacheCluster);
+    AccountsManager   accountsManager = new AccountsManager(accounts, directoryManager, cacheCluster, directoryQueue, keys, messagesManager, usernamesManager, profilesManager);
     Optional<Account> account         = accountsManager.get("+14152222222");
 
     assertTrue(account.isPresent());
@@ -58,12 +69,17 @@ public class AccountsManagerTest {
     FaultTolerantRedisCluster                    cacheCluster     = RedisClusterHelper.buildMockRedisCluster(commands);
     Accounts                                     accounts         = mock(Accounts.class);
     DirectoryManager                             directoryManager = mock(DirectoryManager.class);
+    DirectoryQueue                               directoryQueue   = mock(DirectoryQueue.class);
+    Keys                                         keys             = mock(Keys.class);
+    MessagesManager                              messagesManager  = mock(MessagesManager.class);
+    UsernamesManager                             usernamesManager = mock(UsernamesManager.class);
+    ProfilesManager                              profilesManager  = mock(ProfilesManager.class);
 
     UUID uuid = UUID.randomUUID();
 
     when(commands.get(eq("Account3::" + uuid.toString()))).thenReturn("{\"number\": \"+14152222222\", \"name\": \"test\"}");
 
-    AccountsManager   accountsManager = new AccountsManager(accounts, directoryManager, cacheCluster);
+    AccountsManager   accountsManager = new AccountsManager(accounts, directoryManager, cacheCluster, directoryQueue, keys, messagesManager, usernamesManager, profilesManager);
     Optional<Account> account         = accountsManager.get(uuid);
 
     assertTrue(account.isPresent());
@@ -83,13 +99,18 @@ public class AccountsManagerTest {
     FaultTolerantRedisCluster                    cacheCluster     = RedisClusterHelper.buildMockRedisCluster(commands);
     Accounts                                     accounts         = mock(Accounts.class);
     DirectoryManager                             directoryManager = mock(DirectoryManager.class);
+    DirectoryQueue                               directoryQueue   = mock(DirectoryQueue.class);
+    Keys                                         keys             = mock(Keys.class);
+    MessagesManager                              messagesManager  = mock(MessagesManager.class);
+    UsernamesManager                             usernamesManager = mock(UsernamesManager.class);
+    ProfilesManager                              profilesManager  = mock(ProfilesManager.class);
     UUID                                         uuid             = UUID.randomUUID();
     Account                                      account          = new Account("+14152222222", uuid, new HashSet<>(), new byte[16]);
 
     when(commands.get(eq("AccountMap::+14152222222"))).thenReturn(null);
     when(accounts.get(eq("+14152222222"))).thenReturn(Optional.of(account));
 
-    AccountsManager   accountsManager = new AccountsManager(accounts, directoryManager, cacheCluster);
+    AccountsManager   accountsManager = new AccountsManager(accounts, directoryManager, cacheCluster, directoryQueue, keys, messagesManager, usernamesManager, profilesManager);
     Optional<Account> retrieved       = accountsManager.get("+14152222222");
 
     assertTrue(retrieved.isPresent());
@@ -110,13 +131,18 @@ public class AccountsManagerTest {
     FaultTolerantRedisCluster                    cacheCluster     = RedisClusterHelper.buildMockRedisCluster(commands);
     Accounts                                     accounts         = mock(Accounts.class);
     DirectoryManager                             directoryManager = mock(DirectoryManager.class);
+    DirectoryQueue                               directoryQueue   = mock(DirectoryQueue.class);
+    Keys                                         keys             = mock(Keys.class);
+    MessagesManager                              messagesManager  = mock(MessagesManager.class);
+    UsernamesManager                             usernamesManager = mock(UsernamesManager.class);
+    ProfilesManager                              profilesManager  = mock(ProfilesManager.class);
     UUID                                         uuid             = UUID.randomUUID();
     Account                                      account          = new Account("+14152222222", uuid, new HashSet<>(), new byte[16]);
 
     when(commands.get(eq("Account3::" + uuid))).thenReturn(null);
     when(accounts.get(eq(uuid))).thenReturn(Optional.of(account));
 
-    AccountsManager   accountsManager = new AccountsManager(accounts, directoryManager, cacheCluster);
+    AccountsManager   accountsManager = new AccountsManager(accounts, directoryManager, cacheCluster, directoryQueue, keys, messagesManager, usernamesManager, profilesManager);
     Optional<Account> retrieved       = accountsManager.get(uuid);
 
     assertTrue(retrieved.isPresent());
@@ -137,13 +163,18 @@ public class AccountsManagerTest {
     FaultTolerantRedisCluster                    cacheCluster     = RedisClusterHelper.buildMockRedisCluster(commands);
     Accounts                                     accounts         = mock(Accounts.class);
     DirectoryManager                             directoryManager = mock(DirectoryManager.class);
+    DirectoryQueue                               directoryQueue   = mock(DirectoryQueue.class);
+    Keys                                         keys             = mock(Keys.class);
+    MessagesManager                              messagesManager  = mock(MessagesManager.class);
+    UsernamesManager                             usernamesManager = mock(UsernamesManager.class);
+    ProfilesManager                              profilesManager  = mock(ProfilesManager.class);
     UUID                                         uuid             = UUID.randomUUID();
     Account                                      account          = new Account("+14152222222", uuid, new HashSet<>(), new byte[16]);
 
     when(commands.get(eq("AccountMap::+14152222222"))).thenThrow(new RedisException("Connection lost!"));
     when(accounts.get(eq("+14152222222"))).thenReturn(Optional.of(account));
 
-    AccountsManager   accountsManager = new AccountsManager(accounts, directoryManager, cacheCluster);
+    AccountsManager   accountsManager = new AccountsManager(accounts, directoryManager, cacheCluster, directoryQueue, keys, messagesManager, usernamesManager, profilesManager);
     Optional<Account> retrieved       = accountsManager.get("+14152222222");
 
     assertTrue(retrieved.isPresent());
@@ -164,13 +195,18 @@ public class AccountsManagerTest {
     FaultTolerantRedisCluster                    cacheCluster     = RedisClusterHelper.buildMockRedisCluster(commands);
     Accounts                                     accounts         = mock(Accounts.class);
     DirectoryManager                             directoryManager = mock(DirectoryManager.class);
+    DirectoryQueue                               directoryQueue   = mock(DirectoryQueue.class);
+    Keys                                         keys             = mock(Keys.class);
+    MessagesManager                              messagesManager  = mock(MessagesManager.class);
+    UsernamesManager                             usernamesManager = mock(UsernamesManager.class);
+    ProfilesManager                              profilesManager  = mock(ProfilesManager.class);
     UUID                                         uuid             = UUID.randomUUID();
     Account                                      account          = new Account("+14152222222", uuid, new HashSet<>(), new byte[16]);
 
     when(commands.get(eq("Account3::" + uuid))).thenThrow(new RedisException("Connection lost!"));
     when(accounts.get(eq(uuid))).thenReturn(Optional.of(account));
 
-    AccountsManager   accountsManager = new AccountsManager(accounts, directoryManager, cacheCluster);
+    AccountsManager   accountsManager = new AccountsManager(accounts, directoryManager, cacheCluster, directoryQueue, keys, messagesManager, usernamesManager, profilesManager);
     Optional<Account> retrieved       = accountsManager.get(uuid);
 
     assertTrue(retrieved.isPresent());