replicate uuids to contact discovery

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

@@ -357,7 +357,7 @@ public class AccountController {
     accounts.update(account);
 
     if (!wasAccountEnabled && account.isEnabled()) {
-      directoryQueue.addRegisteredUser(account.getNumber());
+      directoryQueue.addRegisteredUser(account.getUuid(), account.getNumber());
     }
   }
 
@@ -373,7 +373,7 @@ public class AccountController {
     accounts.update(account);
 
     if (!account.isEnabled()) {
-      directoryQueue.deleteRegisteredUser(account.getNumber());
+      directoryQueue.deleteRegisteredUser(account.getUuid(), account.getNumber());
     }
   }
 
@@ -393,7 +393,7 @@ public class AccountController {
     accounts.update(account);
 
     if (!wasAccountEnabled && account.isEnabled()) {
-      directoryQueue.addRegisteredUser(account.getNumber());
+      directoryQueue.addRegisteredUser(account.getUuid(), account.getNumber());
     }
   }
 
@@ -409,7 +409,7 @@ public class AccountController {
     accounts.update(account);
 
     if (!account.isEnabled()) {
-      directoryQueue.deleteRegisteredUser(account.getNumber());
+      directoryQueue.deleteRegisteredUser(account.getUuid(), account.getNumber());
     }
   }
 
@@ -617,9 +617,9 @@ public class AccountController {
     }
 
     if (account.isEnabled()) {
-      directoryQueue.addRegisteredUser(number);
+      directoryQueue.addRegisteredUser(account.getUuid(), number);
     } else {
-      directoryQueue.deleteRegisteredUser(number);
+      directoryQueue.deleteRegisteredUser(account.getUuid(), number);
     }
 
     messagesManager.clear(number);

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

@@ -113,7 +113,7 @@ public class DeviceController {
     accounts.update(account);
 
     if (!account.isEnabled()) {
-      directoryQueue.deleteRegisteredUser(account.getNumber());
+      directoryQueue.deleteRegisteredUser(account.getUuid(), account.getNumber());
     }
 
     messages.clear(account.getNumber(), deviceId);

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

@@ -107,7 +107,7 @@ public class KeysController {
       accounts.update(account);
 
       if (!wasAccountEnabled && account.isEnabled()) {
-        directoryQueue.addRegisteredUser(account.getNumber());
+        directoryQueue.addRegisteredUser(account.getUuid(), account.getNumber());
       }
     }
 
@@ -173,7 +173,7 @@ public class KeysController {
     accounts.update(account);
 
     if (!wasAccountEnabled && account.isEnabled()) {
-      directoryQueue.addRegisteredUser(account.getNumber());
+      directoryQueue.addRegisteredUser(account.getUuid(), account.getNumber());
     }
   }
 

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

@@ -29,15 +29,19 @@ public class DirectoryReconciliationRequest {
   @JsonProperty
   private UUID toUuid;
 
+  @JsonProperty
+  private List<UUID> uuids;
+
   @JsonProperty
   private List<String> numbers;
 
   public DirectoryReconciliationRequest() {
   }
 
-  public DirectoryReconciliationRequest(UUID fromUuid, UUID toUuid, List<String> numbers) {
+  public DirectoryReconciliationRequest(UUID fromUuid, UUID toUuid, List<UUID> uuids, List<String> numbers) {
     this.fromUuid = fromUuid;
     this.toUuid   = toUuid;
+    this.uuids    = uuids;
     this.numbers  = numbers;
   }
 
@@ -49,6 +53,10 @@ public class DirectoryReconciliationRequest {
     return toUuid;
   }
 
+  public List<UUID> getUuids() {
+    return uuids;
+  }
+
   public List<String> getNumbers() {
     return numbers;
   }

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

@@ -35,6 +35,7 @@ import org.whispersystems.textsecuregcm.util.Constants;
 
 import java.util.HashMap;
 import java.util.Map;
+import java.util.UUID;
 
 import static com.codahale.metrics.MetricRegistry.name;
 
@@ -52,28 +53,29 @@ public class DirectoryQueue {
   public DirectoryQueue(SqsConfiguration sqsConfig) {
     final AWSCredentials               credentials         = new BasicAWSCredentials(sqsConfig.getAccessKey(), sqsConfig.getAccessSecret());
     final AWSStaticCredentialsProvider credentialsProvider = new AWSStaticCredentialsProvider(credentials);
-    
+
     this.queueUrl = sqsConfig.getQueueUrl();
     this.sqs      = AmazonSQSClientBuilder.standard().withRegion(sqsConfig.getRegion()).withCredentials(credentialsProvider).build();
   }
 
-  public void addRegisteredUser(String user) {
+  public void addRegisteredUser(UUID uuid, String number) {
-    sendMessage("add", user);
+    sendMessage("add", uuid, number);
   }
 
-  public void deleteRegisteredUser(String user) {
+  public void deleteRegisteredUser(UUID uuid, String number) {
-    sendMessage("delete", user);
+    sendMessage("delete", uuid, number);
   }
 
-  private void sendMessage(String action, String user) {
+  private void sendMessage(String action, UUID uuid, String number) {
     final Map<String, MessageAttributeValue> messageAttributes = new HashMap<>();
-    messageAttributes.put("id", new MessageAttributeValue().withDataType("String").withStringValue(user));
+    messageAttributes.put("id", new MessageAttributeValue().withDataType("String").withStringValue(number));
+    messageAttributes.put("uuid", new MessageAttributeValue().withDataType("String").withStringValue(uuid.toString()));
     messageAttributes.put("action", new MessageAttributeValue().withDataType("String").withStringValue(action));
     SendMessageRequest sendMessageRequest = new SendMessageRequest()
             .withQueueUrl(queueUrl)
             .withMessageBody("-")
-            .withMessageDeduplicationId(user + action)
+            .withMessageDeduplicationId(UUID.randomUUID().toString())
-            .withMessageGroupId(user)
+            .withMessageGroupId(number)
             .withMessageAttributes(messageAttributes);
     try {
       sqs.sendMessage(sendMessageRequest);

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

@@ -68,7 +68,7 @@ public class AccountCleaner implements AccountDatabaseCrawlerListener {
           accountUpdateCount++;
           accountsManager.update(account);
 
-          directoryQueue.deleteRegisteredUser(account.getNumber());
+          directoryQueue.deleteRegisteredUser(account.getUuid(), account.getNumber());
         }
       }
     }

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

@@ -30,6 +30,8 @@ import org.whispersystems.textsecuregcm.util.Constants;
 import org.whispersystems.textsecuregcm.util.Util;
 
 import javax.ws.rs.ProcessingException;
+
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
@@ -57,7 +59,7 @@ public class DirectoryReconciler implements AccountDatabaseCrawlerListener {
   public void onCrawlStart() { }
 
   public void onCrawlEnd(Optional<UUID> fromUuid) {
-    DirectoryReconciliationRequest  request  = new DirectoryReconciliationRequest(fromUuid.orElse(null), null, Collections.emptyList());
+    DirectoryReconciliationRequest  request  = new DirectoryReconciliationRequest(fromUuid.orElse(null), null, Collections.emptyList(), Collections.emptyList());
     DirectoryReconciliationResponse response = sendChunk(request);
   }
 
@@ -93,10 +95,14 @@ public class DirectoryReconciler implements AccountDatabaseCrawlerListener {
 
   @SuppressWarnings("OptionalUsedAsFieldOrParameterType")
   private DirectoryReconciliationRequest createChunkRequest(Optional<UUID> fromUuid, List<Account> accounts) {
-    List<String> numbers = accounts.stream()
+    List<UUID>   uuids   = new ArrayList<>(accounts.size());
-                                   .filter(Account::isEnabled)
+    List<String> numbers = new ArrayList<>(accounts.size());
-                                   .map(Account::getNumber)
+    for (Account account : accounts) {
-                                   .collect(Collectors.toList());
+      if (account.isEnabled()) {
+        uuids.add(account.getUuid());
+        numbers.add(account.getNumber());
+      }
+    }
 
     Optional<UUID> toUuid = Optional.empty();
 
@@ -104,7 +110,7 @@ public class DirectoryReconciler implements AccountDatabaseCrawlerListener {
       toUuid = Optional.of(accounts.get(accounts.size() - 1).getUuid());
     }
 
-    return new DirectoryReconciliationRequest(fromUuid.orElse(null), toUuid.orElse(null), numbers);
+    return new DirectoryReconciliationRequest(fromUuid.orElse(null), toUuid.orElse(null), uuids, numbers);
   }
 
   private DirectoryReconciliationResponse sendChunk(DirectoryReconciliationRequest request) {

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

@@ -59,7 +59,7 @@ public class PushFeedbackProcessor implements AccountDatabaseCrawlerListener {
         accountsManager.update(account);
 
         if (!account.isEnabled()) {
-          directoryQueue.deleteRegisteredUser(account.getNumber());
+          directoryQueue.deleteRegisteredUser(account.getUuid(), account.getNumber());
         }
       }
     }

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

@@ -88,7 +88,7 @@ public class DeleteUserCommand extends EnvironmentCommand<WhisperServerConfigura
             device.get().setAuthenticationCredentials(new AuthenticationCredentials(Base64.encodeBytes(random)));
 
             accountsManager.update(account.get());
-            directoryQueue.deleteRegisteredUser(account.get().getNumber());
+            directoryQueue.deleteRegisteredUser(account.get().getUuid(), account.get().getNumber());
 
             logger.warn("Removed " + account.get().getNumber());
           } else {

service/src/test/java/org/whispersystems/textsecuregcm/tests/controllers/AccountControllerTest.java

@@ -475,7 +475,7 @@ public class AccountControllerTest {
     assertThat(result.getUuid()).isNotNull();
 
     verify(accountsManager, times(1)).create(isA(Account.class));
-    verify(directoryQueue, times(1)).deleteRegisteredUser(eq(SENDER));
+    verify(directoryQueue, times(1)).deleteRegisteredUser(notNull(), eq(SENDER));
   }
 
   @Test

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

@@ -31,6 +31,7 @@ import java.util.HashSet;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Optional;
+import java.util.UUID;
 import java.util.concurrent.TimeUnit;
 
 import static org.junit.Assert.assertFalse;
@@ -64,6 +65,7 @@ public class AccountCleanerTest {
     when(deletedDisabledAccount.getLastSeen()).thenReturn(System.currentTimeMillis() - TimeUnit.DAYS.toMillis(1000));
     when(deletedDisabledAccount.getMasterDevice()).thenReturn(Optional.of(deletedDisabledDevice));
     when(deletedDisabledAccount.getNumber()).thenReturn("+14151231234");
+    when(deletedDisabledAccount.getUuid()).thenReturn(UUID.randomUUID());
 
     when(undeletedDisabledDevice.isEnabled()).thenReturn(false);
     when(undeletedDisabledDevice.getGcmId()).thenReturn("foo");
@@ -71,6 +73,7 @@ public class AccountCleanerTest {
     when(undeletedDisabledAccount.getLastSeen()).thenReturn(System.currentTimeMillis() - TimeUnit.DAYS.toMillis(366));
     when(undeletedDisabledAccount.getMasterDevice()).thenReturn(Optional.of(undeletedDisabledDevice));
     when(undeletedDisabledAccount.getNumber()).thenReturn("+14152222222");
+    when(undeletedDisabledAccount.getUuid()).thenReturn(UUID.randomUUID());
 
     when(undeletedEnabledDevice.isEnabled()).thenReturn(true);
     when(undeletedEnabledDevice.getApnId()).thenReturn("bar");
@@ -78,6 +81,7 @@ public class AccountCleanerTest {
     when(undeletedEnabledAccount.getMasterDevice()).thenReturn(Optional.of(undeletedEnabledDevice));
     when(undeletedEnabledAccount.getNumber()).thenReturn("+14153333333");
     when(undeletedEnabledAccount.getLastSeen()).thenReturn(System.currentTimeMillis() - TimeUnit.DAYS.toMillis(364));
+    when(undeletedEnabledAccount.getUuid()).thenReturn(UUID.randomUUID());
   }
 
   @Test
@@ -93,7 +97,7 @@ public class AccountCleanerTest {
     verify(deletedDisabledDevice, never()).setFetchesMessages(anyBoolean());
 
     verify(accountsManager, never()).update(eq(deletedDisabledAccount));
-    verify(directoryQueue, never()).deleteRegisteredUser(eq("+14151231234"));
+    verify(directoryQueue, never()).deleteRegisteredUser(notNull(), eq("+14151231234"));
 
 
     verify(undeletedDisabledDevice, times(1)).setGcmId(isNull());
@@ -102,7 +106,7 @@ public class AccountCleanerTest {
     verify(undeletedDisabledDevice, times(1)).setFetchesMessages(eq(false));
 
     verify(accountsManager, times(1)).update(eq(undeletedDisabledAccount));
-    verify(directoryQueue, times(1)).deleteRegisteredUser(eq("+14152222222"));
+    verify(directoryQueue, times(1)).deleteRegisteredUser(notNull(), eq("+14152222222"));
 
     verify(undeletedEnabledDevice, never()).setGcmId(any());
     verify(undeletedEnabledDevice, never()).setApnId(any());
@@ -110,7 +114,7 @@ public class AccountCleanerTest {
     verify(undeletedEnabledDevice, never()).setFetchesMessages(anyBoolean());
 
     verify(accountsManager, never()).update(eq(undeletedEnabledAccount));
-    verify(directoryQueue, never()).deleteRegisteredUser(eq("+14153333333"));
+    verify(directoryQueue, never()).deleteRegisteredUser(notNull(), eq("+14153333333"));
 
     verifyNoMoreInteractions(accountsManager);
     verifyNoMoreInteractions(directoryQueue);
@@ -139,7 +143,7 @@ public class AccountCleanerTest {
     verify(undeletedDisabledDevice, times(AccountCleaner.MAX_ACCOUNT_UPDATES_PER_CHUNK)).setFetchesMessages(eq(false));
 
     verify(accountsManager, times(AccountCleaner.MAX_ACCOUNT_UPDATES_PER_CHUNK)).update(eq(undeletedDisabledAccount));
-    verify(directoryQueue, times(AccountCleaner.MAX_ACCOUNT_UPDATES_PER_CHUNK)).deleteRegisteredUser(eq("+14152222222"));
+    verify(directoryQueue, times(AccountCleaner.MAX_ACCOUNT_UPDATES_PER_CHUNK)).deleteRegisteredUser(notNull(), eq("+14152222222"));
 
     verify(deletedDisabledDevice, never()).setGcmId(any());
     verify(deletedDisabledDevice, never()).setApnId(any());

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

@@ -72,11 +72,12 @@ public class DirectoryReconcilerTest {
     when(reconciliationClient.sendChunk(any())).thenReturn(successResponse);
     directoryReconciler.onCrawlChunk(Optional.of(VALID_UUID), Arrays.asList(activeAccount, inactiveAccount));
 
-    verify(activeAccount, times(2)).getNumber();
+    verify(activeAccount, atLeastOnce()).getUuid();
-    verify(activeAccount, times(2)).isEnabled();
+    verify(activeAccount, atLeastOnce()).getNumber();
-    verify(inactiveAccount, times(1)).getUuid();
+    verify(activeAccount, atLeastOnce()).isEnabled();
-    verify(inactiveAccount, times(1)).getNumber();
+    verify(inactiveAccount, atLeastOnce()).getUuid();
-    verify(inactiveAccount, times(2)).isEnabled();
+    verify(inactiveAccount, atLeastOnce()).getNumber();
+    verify(inactiveAccount, atLeastOnce()).isEnabled();
 
     ArgumentCaptor<DirectoryReconciliationRequest> request = ArgumentCaptor.forClass(DirectoryReconciliationRequest.class);
     verify(reconciliationClient, times(1)).sendChunk(request.capture());