Close websockets on account deletion

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

@@ -426,7 +426,7 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
         deletedAccountsLockDynamoDbClient, config.getDeletedAccountsLockDynamoDbConfiguration().getTableName());
     AccountsManager accountsManager = new AccountsManager(accounts, cacheCluster,
         deletedAccountsManager, directoryQueue, keysDynamoDb, messagesManager, usernamesManager, profilesManager,
-        pendingAccountsManager, secureStorageClient, secureBackupClient);
+        pendingAccountsManager, secureStorageClient, secureBackupClient, clientPresenceManager);
     RemoteConfigsManager remoteConfigsManager = new RemoteConfigsManager(remoteConfigs);
     DeadLetterHandler          deadLetterHandler          = new DeadLetterHandler(accountsManager, messagesManager);
     DispatchManager            dispatchManager            = new DispatchManager(pubSubClientFactory, Optional.of(deadLetterHandler));

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

@@ -29,7 +29,9 @@ import org.slf4j.LoggerFactory;
 import org.whispersystems.textsecuregcm.auth.AuthenticationCredentials;
 import org.whispersystems.textsecuregcm.controllers.AccountController;
 import org.whispersystems.textsecuregcm.entities.AccountAttributes;
+import org.whispersystems.textsecuregcm.push.ClientPresenceManager;
 import org.whispersystems.textsecuregcm.redis.FaultTolerantRedisCluster;
+import org.whispersystems.textsecuregcm.redis.RedisOperation;
 import org.whispersystems.textsecuregcm.securebackup.SecureBackupClient;
 import org.whispersystems.textsecuregcm.securestorage.SecureStorageClient;
 import org.whispersystems.textsecuregcm.sqs.DirectoryQueue;
@@ -73,6 +75,7 @@ public class AccountsManager {
   private final StoredVerificationCodeManager pendingAccounts;
   private final SecureStorageClient       secureStorageClient;
   private final SecureBackupClient        secureBackupClient;
+  private final ClientPresenceManager clientPresenceManager;
   private final ObjectMapper              mapper;
 
   public enum DeletionReason {
@@ -95,7 +98,8 @@ public class AccountsManager {
       final ProfilesManager profilesManager,
       final StoredVerificationCodeManager pendingAccounts,
       final SecureStorageClient secureStorageClient,
-      final SecureBackupClient secureBackupClient) {
+      final SecureBackupClient secureBackupClient,
+      final ClientPresenceManager clientPresenceManager) {
     this.accounts = accounts;
     this.cacheCluster = cacheCluster;
     this.deletedAccountsManager = deletedAccountsManager;
@@ -107,8 +111,8 @@ public class AccountsManager {
     this.pendingAccounts = pendingAccounts;
     this.secureStorageClient = secureStorageClient;
     this.secureBackupClient  = secureBackupClient;
+    this.clientPresenceManager = clientPresenceManager;
     this.mapper              = SystemMapper.getMapper();
-
   }
 
   public Account create(final String number,
@@ -354,6 +358,10 @@ public class AccountsManager {
         redisDelete(account);
         dynamoDelete(account);
 
+        RedisOperation.unchecked(() ->
+            account.getDevices().forEach(device ->
+                clientPresenceManager.displacePresence(account.getUuid(), device.getId())));
+
         return account.getUuid();
       });
     } catch (final RuntimeException | InterruptedException e) {

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

@@ -20,6 +20,7 @@ import io.lettuce.core.resource.ClientResources;
 import io.micrometer.core.instrument.Metrics;
 import java.util.Optional;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import net.sourceforge.argparse4j.inf.Namespace;
 import net.sourceforge.argparse4j.inf.Subparser;
 import org.jdbi.v3.core.Jdbi;
@@ -28,6 +29,7 @@ import org.slf4j.LoggerFactory;
 import org.whispersystems.textsecuregcm.WhisperServerConfiguration;
 import org.whispersystems.textsecuregcm.auth.ExternalServiceCredentialGenerator;
 import org.whispersystems.textsecuregcm.metrics.PushLatencyManager;
+import org.whispersystems.textsecuregcm.push.ClientPresenceManager;
 import org.whispersystems.textsecuregcm.redis.FaultTolerantRedisCluster;
 import org.whispersystems.textsecuregcm.securebackup.SecureBackupClient;
 import org.whispersystems.textsecuregcm.securestorage.SecureStorageClient;
@@ -168,10 +170,14 @@ public class DeleteUserCommand extends EnvironmentCommand<WhisperServerConfigura
           configuration.getMessageCacheConfiguration().getRedisClusterConfiguration(), redisClusterClientResources);
       FaultTolerantRedisCluster metricsCluster = new FaultTolerantRedisCluster("metrics_cluster",
           configuration.getMetricsClusterConfiguration(), redisClusterClientResources);
+      FaultTolerantRedisCluster clientPresenceCluster    = new FaultTolerantRedisCluster("client_presence_cluster",
+          configuration.getClientPresenceClusterConfiguration(), redisClusterClientResources);
       SecureBackupClient secureBackupClient = new SecureBackupClient(backupCredentialsGenerator, backupServiceExecutor,
           configuration.getSecureBackupServiceConfiguration());
       SecureStorageClient secureStorageClient = new SecureStorageClient(storageCredentialsGenerator,
           storageServiceExecutor, configuration.getSecureStorageServiceConfiguration());
+      ClientPresenceManager clientPresenceManager = new ClientPresenceManager(clientPresenceCluster,
+          Executors.newSingleThreadScheduledExecutor(), keyspaceNotificationDispatchExecutor);
       MessagesCache messagesCache = new MessagesCache(messageInsertCacheCluster, messageReadDeleteCluster,
           keyspaceNotificationDispatchExecutor);
       PushLatencyManager pushLatencyManager = new PushLatencyManager(metricsCluster);
@@ -191,7 +197,7 @@ public class DeleteUserCommand extends EnvironmentCommand<WhisperServerConfigura
       StoredVerificationCodeManager pendingAccountsManager = new StoredVerificationCodeManager(pendingAccounts);
       AccountsManager accountsManager = new AccountsManager(accounts, cacheCluster,
           deletedAccountsManager, directoryQueue, keysDynamoDb, messagesManager, usernamesManager, profilesManager,
-          pendingAccountsManager, secureStorageClient, secureBackupClient);
+          pendingAccountsManager, secureStorageClient, secureBackupClient, clientPresenceManager);
 
       for (String user : users) {
         Optional<Account> account = accountsManager.get(user);

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

@@ -21,12 +21,14 @@ import io.micrometer.core.instrument.Metrics;
 import java.util.Optional;
 import java.util.UUID;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import net.sourceforge.argparse4j.inf.Namespace;
 import net.sourceforge.argparse4j.inf.Subparser;
 import org.jdbi.v3.core.Jdbi;
 import org.whispersystems.textsecuregcm.WhisperServerConfiguration;
 import org.whispersystems.textsecuregcm.auth.ExternalServiceCredentialGenerator;
 import org.whispersystems.textsecuregcm.metrics.PushLatencyManager;
+import org.whispersystems.textsecuregcm.push.ClientPresenceManager;
 import org.whispersystems.textsecuregcm.redis.FaultTolerantRedisCluster;
 import org.whispersystems.textsecuregcm.securebackup.SecureBackupClient;
 import org.whispersystems.textsecuregcm.securestorage.SecureStorageClient;
@@ -171,10 +173,14 @@ public class SetUserDiscoverabilityCommand extends EnvironmentCommand<WhisperSer
           configuration.getMessageCacheConfiguration().getRedisClusterConfiguration(), redisClusterClientResources);
       FaultTolerantRedisCluster metricsCluster = new FaultTolerantRedisCluster("metrics_cluster",
           configuration.getMetricsClusterConfiguration(), redisClusterClientResources);
+      FaultTolerantRedisCluster clientPresenceCluster = new FaultTolerantRedisCluster("client_presence",
+          configuration.getClientPresenceClusterConfiguration(), redisClusterClientResources);
       SecureBackupClient secureBackupClient = new SecureBackupClient(backupCredentialsGenerator, backupServiceExecutor,
           configuration.getSecureBackupServiceConfiguration());
       SecureStorageClient secureStorageClient = new SecureStorageClient(storageCredentialsGenerator,
           storageServiceExecutor, configuration.getSecureStorageServiceConfiguration());
+      ClientPresenceManager clientPresenceManager = new ClientPresenceManager(clientPresenceCluster,
+          Executors.newSingleThreadScheduledExecutor(), keyspaceNotificationDispatchExecutor);
       MessagesCache messagesCache = new MessagesCache(messageInsertCacheCluster, messageReadDeleteCluster,
           keyspaceNotificationDispatchExecutor);
       PushLatencyManager pushLatencyManager = new PushLatencyManager(metricsCluster);
@@ -194,7 +200,7 @@ public class SetUserDiscoverabilityCommand extends EnvironmentCommand<WhisperSer
       StoredVerificationCodeManager pendingAccountsManager = new StoredVerificationCodeManager(pendingAccounts);
       AccountsManager accountsManager = new AccountsManager(accounts, cacheCluster,
           deletedAccountsManager, directoryQueue, keysDynamoDb, messagesManager, usernamesManager, profilesManager,
-          pendingAccountsManager, secureStorageClient, secureBackupClient);
+          pendingAccountsManager, secureStorageClient, secureBackupClient, clientPresenceManager);
 
       Optional<Account> maybeAccount;
 

service/src/test/java/org/whispersystems/textsecuregcm/storage/AccountsManagerConcurrentModificationIntegrationTest.java

@@ -36,6 +36,7 @@ import org.mockito.ArgumentCaptor;
 import org.whispersystems.textsecuregcm.auth.AuthenticationCredentials;
 import org.whispersystems.textsecuregcm.entities.AccountAttributes;
 import org.whispersystems.textsecuregcm.entities.SignedPreKey;
+import org.whispersystems.textsecuregcm.push.ClientPresenceManager;
 import org.whispersystems.textsecuregcm.securebackup.SecureBackupClient;
 import org.whispersystems.textsecuregcm.securestorage.SecureStorageClient;
 import org.whispersystems.textsecuregcm.sqs.DirectoryQueue;
@@ -122,7 +123,8 @@ class AccountsManagerConcurrentModificationIntegrationTest {
           mock(ProfilesManager.class),
           mock(StoredVerificationCodeManager.class),
           mock(SecureStorageClient.class),
-          mock(SecureBackupClient.class)
+          mock(SecureBackupClient.class),
+          mock(ClientPresenceManager.class)
       );
     }
   }

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

@@ -39,6 +39,7 @@ import org.mockito.stubbing.Answer;
 import org.whispersystems.textsecuregcm.configuration.dynamic.DynamicConfiguration;
 import org.whispersystems.textsecuregcm.entities.AccountAttributes;
 import org.whispersystems.textsecuregcm.entities.SignedPreKey;
+import org.whispersystems.textsecuregcm.push.ClientPresenceManager;
 import org.whispersystems.textsecuregcm.securebackup.SecureBackupClient;
 import org.whispersystems.textsecuregcm.securestorage.SecureStorageClient;
 import org.whispersystems.textsecuregcm.sqs.DirectoryQueue;
@@ -111,7 +112,8 @@ class AccountsManagerTest {
         profilesManager,
         mock(StoredVerificationCodeManager.class),
         mock(SecureStorageClient.class),
-        mock(SecureBackupClient.class)
+        mock(SecureBackupClient.class),
+        mock(ClientPresenceManager.class)
     );
   }