Contact Discovery Service

config/sample.yml

@@ -28,9 +28,22 @@ cache: # Redis server configuration for cache cluster
   url:
   replicaUrls:
 
-directory: # Redis server configuration for directory cluster
+directory:
+  redis: # Redis server configuration for directory cluster
     url:
     replicaUrls:
+  client: # Configuration for interfacing with Contact Discovery Service cluster
+    userAuthenticationTokenSharedSecret: # hex-encoded secret shared with CDS used to generate auth tokens for Signal users
+    userAuthenticationTokenUserIdSecret: # hex-encoded secret shared among Signal-Servers to obscure user phone numbers from CDS
+  sqs:
+    accessKey:      # AWS SQS accessKey
+    accessSecret:   # AWS SQS accessSecret
+    queueUrl:       # AWS SQS queue url
+  server:
+    replicationUrl:           # CDS replication endpoint base url
+    replicationPassword:      # CDS replication endpoint password
+    replicationCaCertificate: # CDS replication endpoint TLS certificate trust root
+
 
 messageCache: # Redis server configuration for message store cache
   url:

pom.xml

@@ -9,7 +9,7 @@
 
     <groupId>org.whispersystems.textsecure</groupId>
     <artifactId>TextSecureServer</artifactId>
-    <version>1.88</version>
+    <version>1.89-RC2</version>
 
     <properties>
         <dropwizard.version>1.3.1</dropwizard.version>
@@ -68,7 +68,12 @@
         <dependency>
             <groupId>com.amazonaws</groupId>
             <artifactId>aws-java-sdk-s3</artifactId>
-            <version>1.11.115</version>
+            <version>1.11.366</version>
+        </dependency>
+        <dependency>
+            <groupId>com.amazonaws</groupId>
+            <artifactId>aws-java-sdk-sqs</artifactId>
+            <version>1.11.362</version>
         </dependency>
         <dependency>
             <groupId>com.google.protobuf</groupId>

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

@@ -28,6 +28,7 @@ import org.whispersystems.textsecuregcm.configuration.RateLimitsConfiguration;
 import org.whispersystems.textsecuregcm.configuration.RedPhoneConfiguration;
 import org.whispersystems.textsecuregcm.configuration.RedisConfiguration;
 import org.whispersystems.textsecuregcm.configuration.AttachmentsConfiguration;
+import org.whispersystems.textsecuregcm.configuration.DirectoryConfiguration;
 import org.whispersystems.textsecuregcm.configuration.TestDeviceConfiguration;
 import org.whispersystems.textsecuregcm.configuration.TurnConfiguration;
 import org.whispersystems.textsecuregcm.configuration.TwilioConfiguration;
@@ -74,7 +75,7 @@ public class WhisperServerConfiguration extends Configuration {
   @NotNull
   @Valid
   @JsonProperty
-  private RedisConfiguration directory;
+  private DirectoryConfiguration directory;
 
   @NotNull
   @Valid
@@ -167,7 +168,7 @@ public class WhisperServerConfiguration extends Configuration {
     return cache;
   }
 
-  public RedisConfiguration getDirectoryConfiguration() {
+  public DirectoryConfiguration getDirectoryConfiguration() {
     return directory;
   }
 

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

@@ -29,6 +29,7 @@ import org.whispersystems.dropwizard.simpleauth.AuthDynamicFeature;
 import org.whispersystems.dropwizard.simpleauth.AuthValueFactoryProvider;
 import org.whispersystems.dropwizard.simpleauth.BasicCredentialAuthFilter;
 import org.whispersystems.textsecuregcm.auth.AccountAuthenticator;
+import org.whispersystems.textsecuregcm.auth.DirectoryCredentialsGenerator;
 import org.whispersystems.textsecuregcm.auth.FederatedPeerAuthenticator;
 import org.whispersystems.textsecuregcm.auth.TurnTokenGenerator;
 import org.whispersystems.textsecuregcm.controllers.AccountController;
@@ -67,10 +68,14 @@ import org.whispersystems.textsecuregcm.redis.ReplicatedJedisPool;
 import org.whispersystems.textsecuregcm.s3.UrlSigner;
 import org.whispersystems.textsecuregcm.sms.SmsSender;
 import org.whispersystems.textsecuregcm.sms.TwilioSmsSender;
+import org.whispersystems.textsecuregcm.sqs.ContactDiscoveryQueueSender;
 import org.whispersystems.textsecuregcm.storage.Account;
 import org.whispersystems.textsecuregcm.storage.Accounts;
 import org.whispersystems.textsecuregcm.storage.AccountsManager;
+import org.whispersystems.textsecuregcm.storage.DirectoryReconciliationClient;
 import org.whispersystems.textsecuregcm.storage.DirectoryManager;
+import org.whispersystems.textsecuregcm.storage.DirectoryReconciler;
+import org.whispersystems.textsecuregcm.storage.DirectoryReconciliationCache;
 import org.whispersystems.textsecuregcm.storage.Keys;
 import org.whispersystems.textsecuregcm.storage.Messages;
 import org.whispersystems.textsecuregcm.storage.MessagesCache;
@@ -159,7 +164,7 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
     Messages        messages        = messagedb.onDemand(Messages.class);
 
     RedisClientFactory cacheClientFactory         = new RedisClientFactory(config.getCacheConfiguration().getUrl(), config.getCacheConfiguration().getReplicaUrls()                                                              );
-    RedisClientFactory directoryClientFactory     = new RedisClientFactory(config.getDirectoryConfiguration().getUrl(), config.getDirectoryConfiguration().getReplicaUrls()                                                      );
+    RedisClientFactory directoryClientFactory     = new RedisClientFactory(config.getDirectoryConfiguration().getRedisConfiguration().getUrl(), config.getDirectoryConfiguration().getRedisConfiguration().getReplicaUrls()      );
     RedisClientFactory messagesClientFactory      = new RedisClientFactory(config.getMessageCacheConfiguration().getRedisConfiguration().getUrl(), config.getMessageCacheConfiguration().getRedisConfiguration().getReplicaUrls());
     RedisClientFactory pushSchedulerClientFactory = new RedisClientFactory(config.getPushScheduler().getUrl(), config.getPushScheduler().getReplicaUrls()                                                                        );
 
@@ -193,6 +198,14 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
     ReceiptSender            receiptSender       = new ReceiptSender(accountsManager, pushSender, federatedClientManager);
     TurnTokenGenerator       turnTokenGenerator  = new TurnTokenGenerator(config.getTurnConfiguration());
 
+    ContactDiscoveryQueueSender cdsSender = new ContactDiscoveryQueueSender(config.getDirectoryConfiguration().getSqsConfiguration());
+
+    DirectoryCredentialsGenerator directoryCredentialsGenerator = new DirectoryCredentialsGenerator(config.getDirectoryConfiguration().getDirectoryClientConfiguration().getUserAuthenticationTokenSharedSecret(),
+                                                                                                    config.getDirectoryConfiguration().getDirectoryClientConfiguration().getUserAuthenticationTokenUserIdSecret());
+    DirectoryReconciliationCache  directoryReconciliationCache  = new DirectoryReconciliationCache(cacheClient);
+    DirectoryReconciliationClient directoryReconciliationClient = new DirectoryReconciliationClient(config.getDirectoryConfiguration().getDirectoryServerConfiguration());
+    DirectoryReconciler           directoryReconciler           = new DirectoryReconciler(directoryReconciliationClient, directoryReconciliationCache, directory, accounts);
+
     messagesCache.setPubSubManager(pubSubManager, pushSender);
 
     apnSender.setApnFallbackManager(apnFallbackManager);
@@ -200,6 +213,7 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
     environment.lifecycle().manage(pubSubManager);
     environment.lifecycle().manage(pushSender);
     environment.lifecycle().manage(messagesCache);
+    environment.lifecycle().manage(directoryReconciler);
 
     AttachmentController attachmentController = new AttachmentController(rateLimiters, federatedClientManager, urlSigner);
     KeysController       keysController       = new KeysController(rateLimiters, keys, accountsManager, federatedClientManager);
@@ -216,9 +230,9 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
                                                              .buildAuthFilter()));
     environment.jersey().register(new AuthValueFactoryProvider.Binder());
 
-    environment.jersey().register(new AccountController(pendingAccountsManager, accountsManager, rateLimiters, smsSender, messagesManager, turnTokenGenerator, config.getTestDevices()));
+    environment.jersey().register(new AccountController(pendingAccountsManager, accountsManager, rateLimiters, smsSender, cdsSender, messagesManager, turnTokenGenerator, config.getTestDevices()));
-    environment.jersey().register(new DeviceController(pendingDevicesManager, accountsManager, messagesManager, rateLimiters, config.getMaxDevices()));
+    environment.jersey().register(new DeviceController(pendingDevicesManager, accountsManager, messagesManager, cdsSender, rateLimiters, config.getMaxDevices()));
-    environment.jersey().register(new DirectoryController(rateLimiters, directory));
+    environment.jersey().register(new DirectoryController(rateLimiters, directory, directoryCredentialsGenerator));
     environment.jersey().register(new FederationControllerV1(accountsManager, attachmentController, messageController));
     environment.jersey().register(new FederationControllerV2(accountsManager, attachmentController, messageController, keysController));
     environment.jersey().register(new ProvisioningController(rateLimiters, pushSender));

src/main/java/org/whispersystems/textsecuregcm/auth/AuthorizationToken.java

@@ -1,29 +0,0 @@
-package org.whispersystems.textsecuregcm.auth;
-
-
-import com.fasterxml.jackson.annotation.JsonProperty;
-import org.apache.commons.codec.DecoderException;
-import org.apache.commons.codec.binary.Hex;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.whispersystems.textsecuregcm.util.Util;
-
-import javax.crypto.Mac;
-import javax.crypto.spec.SecretKeySpec;
-import java.security.InvalidKeyException;
-import java.security.MessageDigest;
-import java.security.NoSuchAlgorithmException;
-import java.util.concurrent.TimeUnit;
-
-public class AuthorizationToken {
-
-  @JsonProperty
-  private String token;
-
-  public AuthorizationToken(String token) {
-    this.token = token;
-  }
-
-  public AuthorizationToken() {}
-
-}

src/main/java/org/whispersystems/textsecuregcm/auth/AuthorizationTokenGenerator.java

@@ -1,90 +0,0 @@
-package org.whispersystems.textsecuregcm.auth;
-
-import org.apache.commons.codec.DecoderException;
-import org.apache.commons.codec.binary.Hex;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.whispersystems.textsecuregcm.util.Util;
-
-import javax.crypto.Mac;
-import javax.crypto.spec.SecretKeySpec;
-import java.security.InvalidKeyException;
-import java.security.MessageDigest;
-import java.security.NoSuchAlgorithmException;
-import java.util.concurrent.TimeUnit;
-
-public class AuthorizationTokenGenerator {
-
-  private final Logger logger = LoggerFactory.getLogger(AuthorizationTokenGenerator.class);
-
-  private final byte[] key;
-
-  public AuthorizationTokenGenerator(byte[] key) {
-    this.key = key;
-  }
-
-  public AuthorizationToken generateFor(String number) {
-    try {
-      Mac    mac                = Mac.getInstance("HmacSHA256");
-      long   currentTimeSeconds = System.currentTimeMillis() / 1000;
-      String prefix             = number + ":"  + currentTimeSeconds;
-
-      mac.init(new SecretKeySpec(key, "HmacSHA256"));
-      String output = Hex.encodeHexString(Util.truncate(mac.doFinal(prefix.getBytes()), 10));
-      String token  = prefix + ":" + output;
-
-      return new AuthorizationToken(token);
-    } catch (NoSuchAlgorithmException | InvalidKeyException e) {
-      throw new AssertionError(e);
-    }
-  }
-
-
-  public boolean isValid(String token, String number, long currentTimeMillis) {
-    String[] parts = token.split(":");
-
-    if (parts.length != 3) {
-      return false;
-    }
-
-    if (!number.equals(parts[0])) {
-      return false;
-    }
-
-    if (!isValidTime(parts[1], currentTimeMillis)) {
-      return false;
-    }
-
-    return isValidSignature(parts[0] + ":" + parts[1], parts[2]);
-  }
-
-  private boolean isValidTime(String timeString, long currentTimeMillis) {
-    try {
-      long tokenTime = Long.parseLong(timeString);
-      long ourTime   = TimeUnit.MILLISECONDS.toSeconds(currentTimeMillis);
-
-      return TimeUnit.SECONDS.toHours(Math.abs(ourTime - tokenTime)) < 24;
-    } catch (NumberFormatException e) {
-      logger.warn("Number Format", e);
-      return false;
-    }
-  }
-
-  private boolean isValidSignature(String prefix, String suffix) {
-    try {
-      Mac hmac = Mac.getInstance("HmacSHA256");
-      hmac.init(new SecretKeySpec(key, "HmacSHA256"));
-
-      byte[] ourSuffix   = Util.truncate(hmac.doFinal(prefix.getBytes()), 10);
-      byte[] theirSuffix = Hex.decodeHex(suffix.toCharArray());
-
-      return MessageDigest.isEqual(ourSuffix, theirSuffix);
-    } catch (NoSuchAlgorithmException | InvalidKeyException e) {
-      throw new AssertionError(e);
-    } catch (DecoderException e) {
-      logger.warn("Authorizationtoken", e);
-      return false;
-    }
-  }
-
-}

src/main/java/org/whispersystems/textsecuregcm/auth/DirectoryCredentials.java

@@ -0,0 +1,28 @@
+package org.whispersystems.textsecuregcm.auth;
+
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+public class DirectoryCredentials {
+
+  @JsonProperty
+  private String username;
+
+  @JsonProperty
+  private String password;
+
+  public DirectoryCredentials(String username, String password) {
+    this.username = username;
+    this.password = password;
+  }
+
+  public DirectoryCredentials() {}
+
+  public String getUsername() {
+    return username;
+  }
+
+  public String getPassword() {
+    return password;
+  }
+}

src/main/java/org/whispersystems/textsecuregcm/auth/DirectoryCredentialsGenerator.java

@@ -0,0 +1,105 @@
+package org.whispersystems.textsecuregcm.auth;
+
+import com.google.common.base.Optional;
+import org.apache.commons.codec.DecoderException;
+import org.apache.commons.codec.binary.Hex;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.whispersystems.textsecuregcm.util.Util;
+
+import javax.crypto.Mac;
+import javax.crypto.spec.SecretKeySpec;
+import java.security.InvalidKeyException;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.concurrent.TimeUnit;
+
+public class DirectoryCredentialsGenerator {
+
+  private final Logger logger = LoggerFactory.getLogger(DirectoryCredentialsGenerator.class);
+
+  private final byte[] key;
+  private final byte[] userIdKey;
+
+  public DirectoryCredentialsGenerator(byte[] key, byte[] userIdKey) {
+    this.key       = key;
+    this.userIdKey = userIdKey;
+  }
+
+  public DirectoryCredentials generateFor(String number) {
+    Mac    mac                = getMacInstance();
+    String username           = getUserId(number, mac);
+    long   currentTimeSeconds = System.currentTimeMillis() / 1000;
+    String prefix             = username + ":"  + currentTimeSeconds;
+    String output             = Hex.encodeHexString(Util.truncate(getHmac(key, prefix.getBytes(), mac), 10));
+    String token              = prefix + ":" + output;
+
+    return new DirectoryCredentials(username, token);
+  }
+
+
+  public boolean isValid(String token, String number, long currentTimeMillis) {
+    String[] parts = token.split(":");
+    Mac      mac   = getMacInstance();
+
+    if (parts.length != 3) {
+      return false;
+    }
+
+    if (!getUserId(number, mac).equals(parts[0])) {
+      return false;
+    }
+
+    if (!isValidTime(parts[1], currentTimeMillis)) {
+      return false;
+    }
+
+    return isValidSignature(parts[0] + ":" + parts[1], parts[2], mac);
+  }
+
+  private String getUserId(String number, Mac mac) {
+      return Hex.encodeHexString(Util.truncate(getHmac(userIdKey, number.getBytes(), mac), 10));
+  }
+
+  private boolean isValidTime(String timeString, long currentTimeMillis) {
+    try {
+      long tokenTime = Long.parseLong(timeString);
+      long ourTime   = TimeUnit.MILLISECONDS.toSeconds(currentTimeMillis);
+
+      return TimeUnit.SECONDS.toHours(Math.abs(ourTime - tokenTime)) < 24;
+    } catch (NumberFormatException e) {
+      logger.warn("Number Format", e);
+      return false;
+    }
+  }
+
+  private boolean isValidSignature(String prefix, String suffix, Mac mac) {
+    try {
+      byte[] ourSuffix   = Util.truncate(getHmac(key, prefix.getBytes(), mac), 10);
+      byte[] theirSuffix = Hex.decodeHex(suffix.toCharArray());
+
+      return MessageDigest.isEqual(ourSuffix, theirSuffix);
+    } catch (DecoderException e) {
+      logger.warn("DirectoryCredentials", e);
+      return false;
+    }
+  }
+
+  private Mac getMacInstance() {
+    try {
+      return Mac.getInstance("HmacSHA256");
+    } catch (NoSuchAlgorithmException e) {
+      throw new AssertionError(e);
+    }
+  }
+
+  private byte[] getHmac(byte[] key, byte[] input, Mac mac) {
+    try {
+      mac.init(new SecretKeySpec(key, "HmacSHA256"));
+      return mac.doFinal(input);
+    } catch (InvalidKeyException e) {
+      throw new AssertionError(e);
+    }
+  }
+
+}

src/main/java/org/whispersystems/textsecuregcm/configuration/DirectoryClientConfiguration.java

@@ -0,0 +1,42 @@
+/**
+ * Copyright (C) 2018 Open WhisperSystems
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package org.whispersystems.textsecuregcm.configuration;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.commons.codec.DecoderException;
+import org.apache.commons.codec.binary.Hex;
+import org.hibernate.validator.constraints.NotEmpty;
+
+public class DirectoryClientConfiguration {
+
+  @NotEmpty
+  @JsonProperty
+  private String userAuthenticationTokenSharedSecret;
+
+  @NotEmpty
+  @JsonProperty
+  private String userAuthenticationTokenUserIdSecret;
+
+  public byte[] getUserAuthenticationTokenSharedSecret() throws DecoderException {
+    return Hex.decodeHex(userAuthenticationTokenSharedSecret.toCharArray());
+  }
+
+  public byte[] getUserAuthenticationTokenUserIdSecret() throws DecoderException {
+    return Hex.decodeHex(userAuthenticationTokenUserIdSecret.toCharArray());
+  }
+
+}

src/main/java/org/whispersystems/textsecuregcm/configuration/DirectoryConfiguration.java

@@ -0,0 +1,62 @@
+/**
+ * Copyright (C) 2018 Open WhisperSystems
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package org.whispersystems.textsecuregcm.configuration;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import javax.validation.Valid;
+import javax.validation.constraints.NotNull;
+
+public class DirectoryConfiguration {
+
+  @JsonProperty
+  @NotNull
+  @Valid
+  private RedisConfiguration redis;
+    
+  @JsonProperty
+  @NotNull
+  @Valid
+  private SqsConfiguration sqs;
+    
+  @JsonProperty
+  @NotNull
+  @Valid
+  private DirectoryClientConfiguration client;
+
+  @JsonProperty
+  @NotNull
+  @Valid
+  private DirectoryServerConfiguration server;
+
+  public RedisConfiguration getRedisConfiguration() {
+    return redis;
+  }
+
+  public SqsConfiguration getSqsConfiguration() {
+    return sqs;
+  }
+
+  public DirectoryClientConfiguration getDirectoryClientConfiguration() {
+    return client;
+  }
+
+  public DirectoryServerConfiguration getDirectoryServerConfiguration() {
+    return server;
+  }
+
+}

src/main/java/org/whispersystems/textsecuregcm/configuration/DirectoryServerConfiguration.java

@@ -0,0 +1,47 @@
+/**
+ * Copyright (C) 2018 Open WhisperSystems
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package org.whispersystems.textsecuregcm.configuration;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.hibernate.validator.constraints.NotEmpty;
+
+public class DirectoryServerConfiguration {
+
+  @NotEmpty
+  @JsonProperty
+  private String replicationUrl;
+
+  @NotEmpty
+  @JsonProperty
+  private String replicationPassword;
+
+  @NotEmpty
+  @JsonProperty
+  private String replicationCaCertificate;
+
+  public String getReplicationUrl() {
+    return replicationUrl;
+  }
+
+  public String getReplicationPassword() {
+    return replicationPassword;
+  }
+
+  public String getReplicationCaCertificate() {
+    return replicationCaCertificate;
+  }
+}

src/main/java/org/whispersystems/textsecuregcm/configuration/SqsConfiguration.java

@@ -0,0 +1,48 @@
+/**
+ * Copyright (C) 2018 Open WhisperSystems
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package org.whispersystems.textsecuregcm.configuration;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.hibernate.validator.constraints.NotEmpty;
+
+public class SqsConfiguration {
+  @NotEmpty
+  @JsonProperty
+  private String accessKey;
+
+  @NotEmpty
+  @JsonProperty
+  private String accessSecret;
+
+  @NotEmpty
+  @JsonProperty
+  private String queueUrl;
+
+  public String getAccessKey() {
+    return accessKey;
+  }
+
+  public String getAccessSecret() {
+    return accessSecret;
+  }
+
+  public String getQueueUrl() {
+    return queueUrl;
+  }
+}
+
+   

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

@@ -38,6 +38,7 @@ import org.whispersystems.textsecuregcm.entities.RegistrationLockFailure;
 import org.whispersystems.textsecuregcm.limits.RateLimiters;
 import org.whispersystems.textsecuregcm.sms.SmsSender;
 import org.whispersystems.textsecuregcm.sms.TwilioSmsSender;
+import org.whispersystems.textsecuregcm.sqs.ContactDiscoveryQueueSender;
 import org.whispersystems.textsecuregcm.storage.Account;
 import org.whispersystems.textsecuregcm.storage.AccountsManager;
 import org.whispersystems.textsecuregcm.storage.Device;
@@ -81,6 +82,7 @@ public class AccountController {
   private final AccountsManager                       accounts;
   private final RateLimiters                          rateLimiters;
   private final SmsSender                             smsSender;
+  private final ContactDiscoveryQueueSender           cdsSender;
   private final MessagesManager                       messagesManager;
   private final TurnTokenGenerator                    turnTokenGenerator;
   private final Map<String, Integer>                  testDevices;
@@ -89,6 +91,7 @@ public class AccountController {
                            AccountsManager accounts,
                            RateLimiters rateLimiters,
                            SmsSender smsSenderFactory,
+                           ContactDiscoveryQueueSender cdsSender,
                            MessagesManager messagesManager,
                            TurnTokenGenerator turnTokenGenerator,
                            Map<String, Integer> testDevices)
@@ -97,6 +100,7 @@ public class AccountController {
     this.accounts           = accounts;
     this.rateLimiters       = rateLimiters;
     this.smsSender          = smsSenderFactory;
+    this.cdsSender          = cdsSender;
     this.messagesManager    = messagesManager;
     this.testDevices        = testDevices;
     this.turnTokenGenerator = turnTokenGenerator;
@@ -339,6 +343,7 @@ public class AccountController {
     if (accounts.create(account)) {
       newUserMeter.mark();
     }
+    cdsSender.addRegisteredUser(number);
 
     messagesManager.clear(number);
     pendingAccounts.remove(number);

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

@@ -30,6 +30,7 @@ import org.whispersystems.textsecuregcm.entities.DeviceInfo;
 import org.whispersystems.textsecuregcm.entities.DeviceInfoList;
 import org.whispersystems.textsecuregcm.entities.DeviceResponse;
 import org.whispersystems.textsecuregcm.limits.RateLimiters;
+import org.whispersystems.textsecuregcm.sqs.ContactDiscoveryQueueSender;
 import org.whispersystems.textsecuregcm.storage.Account;
 import org.whispersystems.textsecuregcm.storage.AccountsManager;
 import org.whispersystems.textsecuregcm.storage.Device;
@@ -71,15 +72,19 @@ public class DeviceController {
   private final RateLimiters          rateLimiters;
   private final Map<String, Integer>  maxDeviceConfiguration;
 
+  private final ContactDiscoveryQueueSender cdsSender;
+
   public DeviceController(PendingDevicesManager pendingDevices,
                           AccountsManager accounts,
                           MessagesManager messages,
+                          ContactDiscoveryQueueSender cdsSender,
                           RateLimiters rateLimiters,
                           Map<String, Integer> maxDeviceConfiguration)
   {
     this.pendingDevices         = pendingDevices;
     this.accounts               = accounts;
     this.messages               = messages;
+    this.cdsSender              = cdsSender;
     this.rateLimiters           = rateLimiters;
     this.maxDeviceConfiguration = maxDeviceConfiguration;
   }
@@ -108,6 +113,9 @@ public class DeviceController {
 
     account.removeDevice(deviceId);
     accounts.update(account);
+    if (!account.isActive()) {
+      cdsSender.deleteRegisteredUser(account.getNumber());
+    }
     messages.clear(account.getNumber(), deviceId);
   }
 

src/main/java/org/whispersystems/textsecuregcm/controllers/DirectoryController.java

@@ -21,8 +21,12 @@ import com.codahale.metrics.MetricRegistry;
 import com.codahale.metrics.SharedMetricRegistries;
 import com.codahale.metrics.annotation.Timed;
 import com.google.common.base.Optional;
+import org.apache.commons.codec.DecoderException;
+import org.hibernate.validator.constraints.NotEmpty;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.whispersystems.textsecuregcm.auth.DirectoryCredentialsGenerator;
+import org.whispersystems.textsecuregcm.configuration.DirectoryConfiguration;
 import org.whispersystems.textsecuregcm.entities.ClientContact;
 import org.whispersystems.textsecuregcm.entities.ClientContactTokens;
 import org.whispersystems.textsecuregcm.entities.ClientContacts;
@@ -58,12 +62,54 @@ public class DirectoryController {
 
   private final RateLimiters                  rateLimiters;
   private final DirectoryManager              directory;
+  private final DirectoryCredentialsGenerator userTokenGenerator;
 
-  public DirectoryController(RateLimiters rateLimiters, DirectoryManager directory) {
+  public DirectoryController(RateLimiters rateLimiters,
+                             DirectoryManager directory,
+                             DirectoryCredentialsGenerator userTokenGenerator)
+  {
     this.directory          = directory;
     this.rateLimiters       = rateLimiters;
+    this.userTokenGenerator = userTokenGenerator;
   }
 
+  @Timed
+  @GET
+  @Path("/auth")
+  @Produces(MediaType.APPLICATION_JSON)
+  public Response getAuthToken(@Auth Account account) {
+    return Response.ok().entity(userTokenGenerator.generateFor(account.getNumber())).build();
+  }
+
+  @Timed
+  @PUT
+  @Path("/feedback/ok")
+  public Response setFeedbackOk(@Auth Account account) {
+    return Response.ok().build();
+  }
+
+  @Timed
+  @PUT
+  @Path("/feedback/mismatch")
+  public Response setFeedbackMismatch(@Auth Account account) {
+    return Response.ok().build();
+  }
+
+  @Timed
+  @PUT
+  @Path("/feedback/attestation-error")
+  public Response setFeedbackAttestationError(@Auth Account account) {
+    return Response.ok().build();
+  }
+
+  @Timed
+  @PUT
+  @Path("/feedback/unexpected-error")
+  public Response setFeedbackUnexpectedError(@Auth Account account) {
+    return Response.ok().build();
+  }
+
+
   @Timed
   @GET
   @Path("/{token}")

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

@@ -0,0 +1,55 @@
+/**
+ * Copyright (C) 2018 Open WhisperSystems
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package org.whispersystems.textsecuregcm.entities;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.util.List;
+
+public class DirectoryReconciliationRequest {
+
+  @JsonProperty
+  private String fromNumber;
+
+  @JsonProperty
+  private String toNumber;
+
+  @JsonProperty
+  private List<String> numbers;
+
+  public DirectoryReconciliationRequest() {
+  }
+
+  public DirectoryReconciliationRequest(String fromNumber, String toNumber, List<String> numbers) {
+    this.fromNumber = fromNumber;
+    this.toNumber   = toNumber;
+    this.numbers    = numbers;
+  }
+
+  public String getFromNumber() {
+    return fromNumber;
+  }
+
+  public String getToNumber() {
+    return toNumber;
+  }
+
+  public List<String> getNumbers() {
+    return numbers;
+  }
+
+}

src/main/java/org/whispersystems/textsecuregcm/entities/DirectoryReconciliationResponse.java

@@ -0,0 +1,45 @@
+/*
+ * Copyright (C) 2018 Open WhisperSystems
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package org.whispersystems.textsecuregcm.entities;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.hibernate.validator.constraints.NotEmpty;
+
+public class DirectoryReconciliationResponse {
+
+  @JsonProperty
+  @NotEmpty
+  private Status status;
+
+  public DirectoryReconciliationResponse() {
+  }
+
+  public DirectoryReconciliationResponse(Status status) {
+    this.status = status;
+  }
+
+  public Status getStatus() {
+    return status;
+  }
+
+  public enum Status {
+    OK,
+    MISSING,
+  }
+
+}

src/main/java/org/whispersystems/textsecuregcm/sqs/ContactDiscoveryQueueSender.java

@@ -0,0 +1,93 @@
+/**
+ * Copyright (C) 2018 Open WhisperSystems
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package org.whispersystems.textsecuregcm.sqs;
+
+import com.amazonaws.AmazonClientException;
+import com.amazonaws.AmazonServiceException;
+import com.amazonaws.auth.AWSCredentials;
+import com.amazonaws.auth.AWSStaticCredentialsProvider;
+import com.amazonaws.auth.BasicAWSCredentials;
+import com.amazonaws.services.sqs.AmazonSQS;
+import com.amazonaws.services.sqs.AmazonSQSClientBuilder;
+import com.amazonaws.services.sqs.model.SendMessageRequest;
+import com.amazonaws.services.sqs.model.MessageAttributeValue;
+
+import com.codahale.metrics.Meter;
+import com.codahale.metrics.MetricRegistry;
+import com.codahale.metrics.SharedMetricRegistries;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.whispersystems.textsecuregcm.configuration.DirectoryConfiguration;
+import org.whispersystems.textsecuregcm.configuration.SqsConfiguration;
+import org.whispersystems.textsecuregcm.util.Constants;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static com.codahale.metrics.MetricRegistry.name;
+
+public class ContactDiscoveryQueueSender {
+
+  private static final Logger  logger = LoggerFactory.getLogger(ContactDiscoveryQueueSender.class);
+
+  private final MetricRegistry metricRegistry    = SharedMetricRegistries.getOrCreate(Constants.METRICS_NAME);
+  private final Meter          serviceErrorMeter = metricRegistry.meter(name(ContactDiscoveryQueueSender.class, "serviceError"));
+  private final Meter          clientErrorMeter  = metricRegistry.meter(name(ContactDiscoveryQueueSender.class, "clientError"));
+
+  private final String         queueUrl;
+  private final AmazonSQS      sqs;
+
+  public ContactDiscoveryQueueSender(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().withCredentials(credentialsProvider).build();
+  }
+
+  public void addRegisteredUser(String user) {
+    sendMessage("add", user);
+  }
+
+  public void deleteRegisteredUser(String user) {
+    sendMessage("delete", user);
+  }
+
+  private void sendMessage(String action, String user) {
+    final Map<String, MessageAttributeValue> messageAttributes = new HashMap<>();
+    messageAttributes.put("id", new MessageAttributeValue().withDataType("String").withStringValue(user));
+    messageAttributes.put("action", new MessageAttributeValue().withDataType("String").withStringValue(action));
+    SendMessageRequest sendMessageRequest = new SendMessageRequest()
+            .withQueueUrl(queueUrl)
+            .withMessageBody("-")
+            .withMessageDeduplicationId(user + action)
+            .withMessageGroupId(user)
+            .withMessageAttributes(messageAttributes);
+    try {
+      sqs.sendMessage(sendMessageRequest);
+    } catch (AmazonServiceException ex) {
+      serviceErrorMeter.mark();
+      logger.warn("sqs service error: ", ex);
+    } catch (AmazonClientException ex) {
+      clientErrorMeter.mark();
+      logger.warn("sqs client error: ", ex);
+    } catch (Throwable t) {
+      logger.warn("sqs unexpected error: ", t);
+    }
+  }
+
+}

src/main/java/org/whispersystems/textsecuregcm/storage/Accounts.java

@@ -66,7 +66,7 @@ public abstract class Accounts {
   abstract Account get(@Bind("number") String number);
 
   @SqlQuery("SELECT COUNT(DISTINCT " + NUMBER + ") from accounts")
-  abstract long getCount();
+  public abstract long getCount();
 
   @Mapper(AccountMapper.class)
   @SqlQuery("SELECT * FROM accounts OFFSET :offset LIMIT :limit")
@@ -76,6 +76,14 @@ public abstract class Accounts {
   @SqlQuery("SELECT * FROM accounts")
   public abstract Iterator<Account> getAll();
 
+  @Mapper(AccountMapper.class)
+  @SqlQuery("SELECT * FROM accounts ORDER BY " + NUMBER + " LIMIT :limit")
+  public abstract List<Account> getAllFrom(@Bind("limit") int length);
+
+  @Mapper(AccountMapper.class)
+  @SqlQuery("SELECT * FROM accounts WHERE " + NUMBER + " > :from ORDER BY " + NUMBER + " LIMIT :limit")
+  public abstract List<Account> getAllFrom(@Bind("from") String from, @Bind("limit") int length);
+
   @SqlQuery("SELECT COUNT(*) FROM accounts a, json_array_elements(a.data->'devices') devices WHERE devices->>'id' = '1' AND (devices->>'gcmId') is not null AND (devices->>'lastSeen')\\:\\:bigint >= :since")
   public abstract int getAndroidActiveSinceCount(@Bind("since") long since);
 

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

@@ -0,0 +1,269 @@
+/**
+ * Copyright (C) 2018 Open WhisperSystems
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package org.whispersystems.textsecuregcm.storage;
+
+import com.codahale.metrics.Meter;
+import com.codahale.metrics.MetricRegistry;
+import com.codahale.metrics.SharedMetricRegistries;
+import com.codahale.metrics.Timer;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Optional;
+import io.dropwizard.lifecycle.Managed;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.whispersystems.textsecuregcm.entities.ClientContact;
+import org.whispersystems.textsecuregcm.entities.DirectoryReconciliationRequest;
+import org.whispersystems.textsecuregcm.entities.DirectoryReconciliationResponse;
+import org.whispersystems.textsecuregcm.storage.DirectoryManager.BatchOperationHandle;
+import org.whispersystems.textsecuregcm.util.Constants;
+import org.whispersystems.textsecuregcm.util.Hex;
+import org.whispersystems.textsecuregcm.util.Util;
+
+import javax.ws.rs.ProcessingException;
+import java.security.SecureRandom;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static com.codahale.metrics.MetricRegistry.name;
+
+public class DirectoryReconciler implements Managed, Runnable {
+
+  private static final Logger logger = LoggerFactory.getLogger(DirectoryReconciler.class);
+
+  private static final MetricRegistry metricRegistry      = SharedMetricRegistries.getOrCreate(Constants.METRICS_NAME);
+  private static final Timer          readChunkTimer      = metricRegistry.timer(name(DirectoryReconciler.class, "readChunk"));
+  private static final Timer          sendChunkTimer      = metricRegistry.timer(name(DirectoryReconciler.class, "sendChunk"));
+  private static final Meter          sendChunkErrorMeter = metricRegistry.meter(name(DirectoryReconciler.class, "sendChunkError"));
+
+  private static final long   WORKER_TTL_MS              = 120_000L;
+  private static final long   PERIOD                     = 86400_000L;
+  private static final long   MAXIMUM_CHUNK_INTERVAL     = 30_000L;
+  private static final long   DEFAULT_CHUNK_INTERVAL     = 10_000L;
+  private static final long   MINIMUM_CHUNK_INTERVAL     = 500L;
+  private static final long   ACCELERATED_CHUNK_INTERVAL = 10L;
+  private static final int    CHUNK_SIZE                 = 1000;
+  private static final double JITTER_MAX                 = 0.20;
+
+  private final Accounts                      readOnlyAccounts;
+  private final DirectoryManager              directoryManager;
+  private final DirectoryReconciliationClient reconciliationClient;
+  private final DirectoryReconciliationCache  reconciliationCache;
+  private final String                        workerId;
+  private final SecureRandom                  random;
+
+  private boolean running;
+  private boolean finished;
+
+  public DirectoryReconciler(DirectoryReconciliationClient reconciliationClient,
+                             DirectoryReconciliationCache reconciliationCache,
+                             DirectoryManager directoryManager,
+                             Accounts readOnlyAccounts) {
+    this.readOnlyAccounts     = readOnlyAccounts;
+    this.directoryManager     = directoryManager;
+    this.reconciliationClient = reconciliationClient;
+    this.reconciliationCache  = reconciliationCache;
+    this.random               = new SecureRandom();
+    this.workerId             = generateWorkerId(random);
+  }
+
+  private static String generateWorkerId(SecureRandom random) {
+    byte[] workerIdBytes = new byte[16];
+    random.nextBytes(workerIdBytes);
+    return Hex.toString(workerIdBytes);
+  }
+
+  @Override
+  public synchronized void start() {
+    running = true;
+    new Thread(this).start();
+  }
+
+  @Override
+  public synchronized void stop() {
+    running = false;
+    notifyAll();
+    while (!finished) {
+      Util.wait(this);
+    }
+  }
+
+  @Override
+  public void run() {
+    long delayMs = DEFAULT_CHUNK_INTERVAL;
+
+    while (sleepWhileRunning(getDelayWithJitter(delayMs))) {
+      try {
+        delayMs = DEFAULT_CHUNK_INTERVAL;
+        delayMs = getBoundedChunkInterval(PERIOD * CHUNK_SIZE / getAccountCount());
+        delayMs = doPeriodicWork(delayMs);
+      } catch (Throwable t) {
+        logger.warn("error in directory reconciliation: ", t);
+      }
+    }
+
+    synchronized (this) {
+      finished = true;
+      notifyAll();
+    }
+  }
+
+  @VisibleForTesting
+  public long doPeriodicWork(long intervalMs) {
+    long nextIntervalTimeMs = System.currentTimeMillis() + intervalMs;
+
+    if (reconciliationCache.claimActiveWork(workerId, WORKER_TTL_MS)) {
+      if (processChunk()) {
+        if (!reconciliationCache.isAccelerated()) {
+          long timeUntilNextIntervalMs = getTimeUntilNextInterval(nextIntervalTimeMs);
+          reconciliationCache.claimActiveWork(workerId, timeUntilNextIntervalMs);
+          return timeUntilNextIntervalMs;
+        } else {
+          return ACCELERATED_CHUNK_INTERVAL;
+        }
+      }
+    }
+    return intervalMs;
+  }
+
+  @VisibleForTesting
+  public long getAccountCount() {
+    Optional<Long> cachedCount = reconciliationCache.getCachedAccountCount();
+
+    if (cachedCount.isPresent()) {
+      return cachedCount.get();
+    }
+
+    long count = readOnlyAccounts.getCount();
+    reconciliationCache.setCachedAccountCount(count);
+    return count;
+  }
+
+  private synchronized boolean sleepWhileRunning(long delayMs) {
+    long startTimeMs = System.currentTimeMillis();
+    while (running && delayMs > 0) {
+      Util.wait(this, delayMs);
+
+      long nowMs = System.currentTimeMillis();
+      delayMs -= Math.abs(nowMs - startTimeMs);
+    }
+    return running;
+  }
+
+  private long getTimeUntilNextInterval(long nextIntervalTimeMs) {
+    long nextIntervalMs = nextIntervalTimeMs - System.currentTimeMillis();
+    return getBoundedChunkInterval(nextIntervalMs);
+  }
+
+  private long getBoundedChunkInterval(long intervalMs) {
+    return Math.max(Math.min(intervalMs, MAXIMUM_CHUNK_INTERVAL), MINIMUM_CHUNK_INTERVAL);
+  }
+
+  private long getDelayWithJitter(long delayMs) {
+    long randomJitterMs = (long) (random.nextDouble() * JITTER_MAX * delayMs);
+    return delayMs + randomJitterMs;
+  }
+
+  private boolean processChunk() {
+    Optional<String> fromNumber    = reconciliationCache.getLastNumber();
+    List<Account>    chunkAccounts = readChunk(fromNumber, CHUNK_SIZE);
+
+    writeChunktoDirectoryCache(chunkAccounts);
+
+    DirectoryReconciliationRequest  request           = createChunkRequest(fromNumber, chunkAccounts);
+    DirectoryReconciliationResponse sendChunkResponse = sendChunk(request);
+
+    if (sendChunkResponse.getStatus() == DirectoryReconciliationResponse.Status.MISSING ||
+        request.getToNumber() == null) {
+      reconciliationCache.clearAccelerate();
+    }
+
+    if (sendChunkResponse.getStatus() == DirectoryReconciliationResponse.Status.OK) {
+      reconciliationCache.setLastNumber(Optional.fromNullable(request.getToNumber()));
+    } else if (sendChunkResponse.getStatus() == DirectoryReconciliationResponse.Status.MISSING) {
+      reconciliationCache.setLastNumber(Optional.absent());
+    }
+
+    return sendChunkResponse.getStatus() == DirectoryReconciliationResponse.Status.OK;
+  }
+
+  private List<Account> readChunk(Optional<String> fromNumber, int chunkSize) {
+    try (Timer.Context timer = readChunkTimer.time()) {
+      Optional<List<Account>> chunkAccounts;
+
+      if (fromNumber.isPresent()) {
+        chunkAccounts = Optional.fromNullable(readOnlyAccounts.getAllFrom(fromNumber.get(), chunkSize));
+      } else {
+        chunkAccounts = Optional.fromNullable(readOnlyAccounts.getAllFrom(chunkSize));
+      }
+
+      return chunkAccounts.or(Collections::emptyList);
+    }
+  }
+
+  private void writeChunktoDirectoryCache(List<Account> accounts) {
+    if (accounts.isEmpty()) {
+      return;
+    }
+
+    BatchOperationHandle batchOperation = directoryManager.startBatchOperation();
+    try {
+      for (Account account : accounts) {
+        if (account.isActive()) {
+          byte[]        token         = Util.getContactToken(account.getNumber());
+          ClientContact clientContact = new ClientContact(token, null, account.isVoiceSupported(), account.isVideoSupported());
+
+          directoryManager.add(batchOperation, clientContact);
+        } else {
+          directoryManager.remove(batchOperation, account.getNumber());
+        }
+      }
+    } finally {
+      directoryManager.stopBatchOperation(batchOperation);
+    }
+  }
+
+  private DirectoryReconciliationRequest createChunkRequest(Optional<String> fromNumber, List<Account> accounts) {
+    List<String> numbers = accounts.stream()
+                                   .filter(Account::isActive)
+                                   .map(Account::getNumber)
+                                   .collect(Collectors.toList());
+
+    Optional<String> toNumber = Optional.absent();
+    if (!accounts.isEmpty()) {
+      toNumber = Optional.of(accounts.get(accounts.size() - 1).getNumber());
+    }
+
+    return new DirectoryReconciliationRequest(fromNumber.orNull(), toNumber.orNull(), numbers);
+  }
+
+  private DirectoryReconciliationResponse sendChunk(DirectoryReconciliationRequest request) {
+    try (Timer.Context timer = sendChunkTimer.time()) {
+      DirectoryReconciliationResponse response = reconciliationClient.sendChunk(request);
+      if (response.getStatus() != DirectoryReconciliationResponse.Status.OK) {
+        sendChunkErrorMeter.mark();
+        logger.warn("reconciliation error: " + response.getStatus());
+      }
+      return response;
+    } catch (ProcessingException ex) {
+      sendChunkErrorMeter.mark();
+      logger.warn("request error: ", ex);
+      throw new ProcessingException(ex);
+    }
+  }
+
+}

src/main/java/org/whispersystems/textsecuregcm/storage/DirectoryReconciliationCache.java

@@ -0,0 +1,118 @@
+/**
+ * Copyright (C) 2018 Open WhisperSystems
+ * <p>
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ * <p>
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ * <p>
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package org.whispersystems.textsecuregcm.storage;
+
+import com.google.common.base.Optional;
+import org.whispersystems.textsecuregcm.redis.LuaScript;
+import org.whispersystems.textsecuregcm.redis.ReplicatedJedisPool;
+import redis.clients.jedis.Jedis;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+
+public class DirectoryReconciliationCache {
+
+  private static final String ACTIVE_WORKER_KEY = "directory_reconciliation_active_worker";
+  private static final String LAST_NUMBER_KEY   = "directory_reconciliation_last_number";
+  private static final String CACHED_COUNT_KEY  = "directory_reconciliation_cached_count";
+  private static final String ACCELERATE_KEY    = "directory_reconciliation_accelerate";
+
+  private static final long CACHED_COUNT_TTL_MS = 21600_000L;
+  private static final long LAST_NUMBER_TTL_MS  = 86400_000L;
+
+  private final ReplicatedJedisPool jedisPool;
+  private final UnlockOperation     unlockOperation;
+
+  public DirectoryReconciliationCache(ReplicatedJedisPool jedisPool) throws IOException {
+    this.jedisPool       = jedisPool;
+    this.unlockOperation = new UnlockOperation(jedisPool);
+  }
+
+  public void clearAccelerate() {
+    try (Jedis jedis = jedisPool.getWriteResource()) {
+      jedis.del(ACCELERATE_KEY);
+    }
+  }
+
+  public boolean isAccelerated() {
+    try (Jedis jedis = jedisPool.getWriteResource()) {
+      return "1".equals(jedis.get(ACCELERATE_KEY));
+    }
+  }
+
+  public boolean claimActiveWork(String workerId, long ttlMs) {
+    unlockOperation.unlock(ACTIVE_WORKER_KEY, workerId);
+    try (Jedis jedis = jedisPool.getWriteResource()) {
+      return "OK".equals(jedis.set(ACTIVE_WORKER_KEY, workerId, "NX", "PX", ttlMs));
+    }
+  }
+
+  public Optional<String> getLastNumber() {
+    try (Jedis jedis = jedisPool.getWriteResource()) {
+      return Optional.fromNullable(jedis.get(LAST_NUMBER_KEY));
+    }
+  }
+
+  public void setLastNumber(Optional<String> lastNumber) {
+    try (Jedis jedis = jedisPool.getWriteResource()) {
+      if (lastNumber.isPresent()) {
+        jedis.psetex(LAST_NUMBER_KEY, LAST_NUMBER_TTL_MS, lastNumber.get());
+      } else {
+        jedis.del(LAST_NUMBER_KEY);
+      }
+    }
+  }
+
+  public Optional<Long> getCachedAccountCount() {
+    try (Jedis jedis = jedisPool.getWriteResource()) {
+      Optional<String> cachedAccountCount = Optional.fromNullable(jedis.get(CACHED_COUNT_KEY));
+      if (!cachedAccountCount.isPresent()) {
+        return Optional.absent();
+      }
+
+      try {
+        return Optional.of(Long.parseUnsignedLong(cachedAccountCount.get()));
+      } catch (NumberFormatException ex) {
+        return Optional.absent();
+      }
+    }
+  }
+
+  public void setCachedAccountCount(long accountCount) {
+    try (Jedis jedis = jedisPool.getWriteResource()) {
+      jedis.psetex(CACHED_COUNT_KEY, CACHED_COUNT_TTL_MS, Long.toString(accountCount));
+    }
+  }
+
+  public static class UnlockOperation {
+
+    private final LuaScript luaScript;
+
+    UnlockOperation(ReplicatedJedisPool jedisPool) throws IOException {
+      this.luaScript = LuaScript.fromResource(jedisPool, "lua/unlock.lua");
+    }
+
+    public boolean unlock(String key, String value) {
+      List<byte[]> keys = Arrays.asList(key.getBytes());
+      List<byte[]> args = Arrays.asList(value.getBytes());
+
+      return ((long) luaScript.execute(keys, args)) > 0;
+    }
+  }
+
+}

src/main/java/org/whispersystems/textsecuregcm/storage/DirectoryReconciliationClient.java

@@ -0,0 +1,95 @@
+/**
+ * Copyright (C) 2018 Open WhisperSystems
+ * <p>
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ * <p>
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ * <p>
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package org.whispersystems.textsecuregcm.storage;
+
+import org.bouncycastle.openssl.PEMReader;
+import org.glassfish.jersey.SslConfigurator;
+import org.glassfish.jersey.client.authentication.HttpAuthenticationFeature;
+import org.whispersystems.textsecuregcm.configuration.DirectoryServerConfiguration;
+import org.whispersystems.textsecuregcm.entities.DirectoryReconciliationRequest;
+import org.whispersystems.textsecuregcm.entities.DirectoryReconciliationResponse;
+
+import javax.net.ssl.SSLContext;
+import javax.ws.rs.client.Client;
+import javax.ws.rs.client.ClientBuilder;
+import javax.ws.rs.client.Entity;
+import javax.ws.rs.core.MediaType;
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.security.KeyStore;
+import java.security.KeyStoreException;
+import java.security.NoSuchAlgorithmException;
+import java.security.cert.CertificateException;
+import java.security.cert.X509Certificate;
+
+public class DirectoryReconciliationClient {
+
+  private final String replicationUrl;
+  private final Client client;
+
+  public DirectoryReconciliationClient(DirectoryServerConfiguration directoryServerConfiguration)
+      throws CertificateException
+  {
+    this.replicationUrl = directoryServerConfiguration.getReplicationUrl();
+    this.client         = initializeClient(directoryServerConfiguration);
+  }
+
+  public DirectoryReconciliationResponse sendChunk(DirectoryReconciliationRequest request) {
+    return client.target(replicationUrl)
+                 .path("/v1/directory/reconcile")
+                 .request(MediaType.APPLICATION_JSON_TYPE)
+                 .put(Entity.json(request), DirectoryReconciliationResponse.class);
+  }
+
+  private static Client initializeClient(DirectoryServerConfiguration directoryServerConfiguration)
+      throws CertificateException
+  {
+    KeyStore   trustStore = initializeKeyStore(directoryServerConfiguration.getReplicationCaCertificate());
+    SSLContext sslContext = SslConfigurator.newInstance()
+                                           .securityProtocol("TLSv1.2")
+                                           .trustStore(trustStore)
+                                           .createSSLContext();
+    return ClientBuilder.newBuilder()
+                        .register(HttpAuthenticationFeature.basic("signal", directoryServerConfiguration.getReplicationPassword().getBytes()))
+                        .sslContext(sslContext)
+                        .build();
+  }
+
+  private static KeyStore initializeKeyStore(String caCertificatePem)
+      throws CertificateException
+  {
+    try {
+      PEMReader       reader      = new PEMReader(new InputStreamReader(new ByteArrayInputStream(caCertificatePem.getBytes())));
+      X509Certificate certificate = (X509Certificate) reader.readObject();
+
+      if (certificate == null) {
+        throw new CertificateException("No certificate found in parsing!");
+      }
+
+      KeyStore keyStore = KeyStore.getInstance(KeyStore.getDefaultType());
+      keyStore.load(null);
+      keyStore.setCertificateEntry("ca", certificate);
+      return keyStore;
+    } catch (IOException | KeyStoreException ex) {
+      throw new CertificateException(ex);
+    } catch (NoSuchAlgorithmException ex) {
+      throw new AssertionError(ex);
+    }
+  }
+
+}

src/main/java/org/whispersystems/textsecuregcm/util/Util.java

@@ -134,6 +134,14 @@ public class Util {
     }
   }
 
+  public static void wait(Object object, long timeoutMs) {
+    try {
+      object.wait(timeoutMs);
+    } catch (InterruptedException e) {
+      throw new AssertionError(e);
+    }
+  }
+
   public static int hashCode(Object... objects) {
     return Arrays.hashCode(objects);
   }

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

@@ -75,7 +75,7 @@ public class DeleteUserCommand extends EnvironmentCommand<WhisperServerConfigura
 
       Accounts            accounts        = dbi.onDemand(Accounts.class);
       ReplicatedJedisPool cacheClient     = new RedisClientFactory(configuration.getCacheConfiguration().getUrl(), configuration.getCacheConfiguration().getReplicaUrls()).getRedisClientPool();
-      ReplicatedJedisPool redisClient     = new RedisClientFactory(configuration.getDirectoryConfiguration().getUrl(), configuration.getDirectoryConfiguration().getReplicaUrls()).getRedisClientPool();
+      ReplicatedJedisPool redisClient     = new RedisClientFactory(configuration.getDirectoryConfiguration().getRedisConfiguration().getUrl(), configuration.getDirectoryConfiguration().getRedisConfiguration().getReplicaUrls()).getRedisClientPool();
       DirectoryManager    directory       = new DirectoryManager(redisClient);
       AccountsManager     accountsManager = new AccountsManager(accounts, directory, cacheClient);
 

src/main/java/org/whispersystems/textsecuregcm/workers/DirectoryCommand.java

@@ -74,7 +74,7 @@ public class DirectoryCommand extends EnvironmentCommand<WhisperServerConfigurat
 
       Accounts            accounts        = dbi.onDemand(Accounts.class);
       ReplicatedJedisPool cacheClient     = new RedisClientFactory(configuration.getCacheConfiguration().getUrl(), configuration.getCacheConfiguration().getReplicaUrls()).getRedisClientPool();
-      ReplicatedJedisPool redisClient     = new RedisClientFactory(configuration.getDirectoryConfiguration().getUrl(), configuration.getDirectoryConfiguration().getReplicaUrls()).getRedisClientPool();
+      ReplicatedJedisPool redisClient     = new RedisClientFactory(configuration.getDirectoryConfiguration().getRedisConfiguration().getUrl(), configuration.getDirectoryConfiguration().getRedisConfiguration().getReplicaUrls()).getRedisClientPool();
       DirectoryManager    directory       = new DirectoryManager(redisClient);
       AccountsManager     accountsManager = new AccountsManager(accounts, directory, cacheClient);
 //      FederatedClientManager federatedClientManager = new FederatedClientManager(environment,

src/main/resources/lua/unlock.lua

@@ -0,0 +1,8 @@
+-- keys: lock_key
+-- argv: lock_value
+
+if redis.call("GET", KEYS[1]) == ARGV[1] then
+    return redis.call("DEL", KEYS[1])
+else
+    return 0
+end

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

@@ -18,6 +18,7 @@ import org.whispersystems.textsecuregcm.limits.RateLimiters;
 import org.whispersystems.textsecuregcm.mappers.RateLimitExceededExceptionMapper;
 import org.whispersystems.textsecuregcm.providers.TimeProvider;
 import org.whispersystems.textsecuregcm.sms.SmsSender;
+import org.whispersystems.textsecuregcm.sqs.ContactDiscoveryQueueSender;
 import org.whispersystems.textsecuregcm.storage.Account;
 import org.whispersystems.textsecuregcm.storage.AccountsManager;
 import org.whispersystems.textsecuregcm.storage.MessagesManager;
@@ -49,6 +50,7 @@ public class AccountControllerTest {
   private        RateLimiter            rateLimiter            = mock(RateLimiter.class           );
   private        RateLimiter            pinLimiter             = mock(RateLimiter.class           );
   private        SmsSender              smsSender              = mock(SmsSender.class             );
+  private        ContactDiscoveryQueueSender cdsSender         = mock(ContactDiscoveryQueueSender.class);
   private        MessagesManager        storedMessages         = mock(MessagesManager.class       );
   private        TimeProvider           timeProvider           = mock(TimeProvider.class          );
   private        TurnTokenGenerator     turnTokenGenerator     = mock(TurnTokenGenerator.class);
@@ -65,6 +67,7 @@ public class AccountControllerTest {
                                                                                                accountsManager,
                                                                                                rateLimiters,
                                                                                                smsSender,
+                                                                                               cdsSender,
                                                                                                storedMessages,
                                                                                                turnTokenGenerator,
                                                                                                new HashMap<>()))
@@ -137,6 +140,7 @@ public class AccountControllerTest {
     assertThat(response.getStatus()).isEqualTo(204);
 
     verify(accountsManager, times(1)).create(isA(Account.class));
+    verify(cdsSender, times(1)).addRegisteredUser(eq(SENDER));
   }
 
   @Test

src/test/java/org/whispersystems/textsecuregcm/tests/controllers/DeviceControllerTest.java

@@ -29,10 +29,8 @@ import org.whispersystems.textsecuregcm.entities.DeviceResponse;
 import org.whispersystems.textsecuregcm.limits.RateLimiter;
 import org.whispersystems.textsecuregcm.limits.RateLimiters;
 import org.whispersystems.textsecuregcm.mappers.DeviceLimitExceededExceptionMapper;
-import org.whispersystems.textsecuregcm.storage.Account;
+import org.whispersystems.textsecuregcm.sqs.ContactDiscoveryQueueSender;
-import org.whispersystems.textsecuregcm.storage.AccountsManager;
+import org.whispersystems.textsecuregcm.storage.*;
-import org.whispersystems.textsecuregcm.storage.MessagesManager;
-import org.whispersystems.textsecuregcm.storage.PendingDevicesManager;
 import org.whispersystems.textsecuregcm.tests.util.AuthHelper;
 import org.whispersystems.textsecuregcm.util.VerificationCode;
 
@@ -55,10 +53,11 @@ public class DeviceControllerTest {
     public DumbVerificationDeviceController(PendingDevicesManager pendingDevices,
                                             AccountsManager accounts,
                                             MessagesManager messages,
+                                            ContactDiscoveryQueueSender cdsSender,
                                             RateLimiters rateLimiters,
                                             Map<String, Integer> deviceConfiguration)
     {
-      super(pendingDevices, accounts, messages, rateLimiters, deviceConfiguration);
+      super(pendingDevices, accounts, messages, cdsSender, rateLimiters, deviceConfiguration);
     }
 
     @Override
@@ -70,10 +69,12 @@ public class DeviceControllerTest {
   private PendingDevicesManager pendingDevicesManager = mock(PendingDevicesManager.class);
   private AccountsManager       accountsManager       = mock(AccountsManager.class       );
   private MessagesManager       messagesManager       = mock(MessagesManager.class);
+  private ContactDiscoveryQueueSender cdsSender       = mock(ContactDiscoveryQueueSender.class);
   private RateLimiters          rateLimiters          = mock(RateLimiters.class          );
   private RateLimiter           rateLimiter           = mock(RateLimiter.class           );
   private Account               account               = mock(Account.class               );
   private Account               maxedAccount          = mock(Account.class);
+  private Device                masterDevice          = mock(Device.class);
 
   private Map<String, Integer>  deviceConfiguration   = new HashMap<String, Integer>() {{
 
@@ -88,6 +89,7 @@ public class DeviceControllerTest {
                                                             .addResource(new DumbVerificationDeviceController(pendingDevicesManager,
                                                                                                               accountsManager,
                                                                                                               messagesManager,
+                                                                                                              cdsSender,
                                                                                                               rateLimiters,
                                                                                                               deviceConfiguration))
                                                             .build();
@@ -101,9 +103,13 @@ public class DeviceControllerTest {
     when(rateLimiters.getAllocateDeviceLimiter()).thenReturn(rateLimiter);
     when(rateLimiters.getVerifyDeviceLimiter()).thenReturn(rateLimiter);
 
+    when(masterDevice.getId()).thenReturn(1L);
+
     when(account.getNextDeviceId()).thenReturn(42L);
     when(account.getNumber()).thenReturn(AuthHelper.VALID_NUMBER);
 //    when(maxedAccount.getActiveDeviceCount()).thenReturn(6);
+    when(account.getAuthenticatedDevice()).thenReturn(Optional.of(masterDevice));
+    when(account.isActive()).thenReturn(false);
 
     when(pendingDevicesManager.getCodeForNumber(AuthHelper.VALID_NUMBER)).thenReturn(Optional.of(new StoredVerificationCode("5678901", System.currentTimeMillis())));
     when(pendingDevicesManager.getCodeForNumber(AuthHelper.VALID_NUMBER_TWO)).thenReturn(Optional.of(new StoredVerificationCode("1112223", System.currentTimeMillis() - TimeUnit.MINUTES.toMillis(31))));
@@ -195,4 +201,16 @@ public class DeviceControllerTest {
     assertEquals(response.getStatus(), 422);
     verifyNoMoreInteractions(messagesManager);
   }
+
+  @Test
+  public void removeDeviceTest() throws Exception {
+    Response response = resources.getJerseyTest()
+            .target("/v1/devices/12345")
+            .request()
+            .header("Authorization", AuthHelper.getAuthHeader(AuthHelper.VALID_NUMBER, AuthHelper.VALID_PASSWORD))
+            .delete();
+
+    assertEquals(204, response.getStatus());
+    verify(cdsSender).deleteRegisteredUser(eq(AuthHelper.VALID_NUMBER));
+  }
 }

src/test/java/org/whispersystems/textsecuregcm/tests/controllers/DirectoryControllerTest.java

@@ -1,5 +1,6 @@
 package org.whispersystems.textsecuregcm.tests.controllers;
 
+import com.google.common.base.Optional;
 import org.glassfish.jersey.test.grizzly.GrizzlyWebTestContainerFactory;
 import org.junit.Before;
 import org.junit.Rule;
@@ -7,6 +8,10 @@ import org.junit.Test;
 import org.mockito.invocation.InvocationOnMock;
 import org.mockito.stubbing.Answer;
 import org.whispersystems.dropwizard.simpleauth.AuthValueFactoryProvider;
+import org.whispersystems.textsecuregcm.auth.DirectoryCredentials;
+import org.whispersystems.textsecuregcm.auth.DirectoryCredentialsGenerator;
+import org.whispersystems.textsecuregcm.configuration.DirectoryConfiguration;
+import org.whispersystems.textsecuregcm.configuration.DirectoryClientConfiguration;
 import org.whispersystems.textsecuregcm.controllers.DirectoryController;
 import org.whispersystems.textsecuregcm.entities.ClientContactTokens;
 import org.whispersystems.textsecuregcm.limits.RateLimiter;
@@ -24,15 +29,19 @@ import java.util.List;
 import io.dropwizard.testing.junit.ResourceTestRule;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.anyListOf;
+import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Matchers.anyList;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
 public class DirectoryControllerTest {
 
-  private final RateLimiters     rateLimiters     = mock(RateLimiters.class    );
+  private final RateLimiters                  rateLimiters                  = mock(RateLimiters.class);
-  private final RateLimiter      rateLimiter      = mock(RateLimiter.class     );
+  private final RateLimiter                   rateLimiter                   = mock(RateLimiter.class);
   private final DirectoryManager              directoryManager              = mock(DirectoryManager.class);
+  private final DirectoryCredentialsGenerator directoryCredentialsGenerator = mock(DirectoryCredentialsGenerator.class);
+
+  private final DirectoryCredentials validCredentials = new DirectoryCredentials("username", "password");
 
   @Rule
   public final ResourceTestRule resources = ResourceTestRule.builder()
@@ -40,7 +49,8 @@ public class DirectoryControllerTest {
                                                             .addProvider(new AuthValueFactoryProvider.Binder())
                                                             .setTestContainerFactory(new GrizzlyWebTestContainerFactory())
                                                             .addResource(new DirectoryController(rateLimiters,
-                                                                                                 directoryManager))
+                                                                                                 directoryManager,
+                                                                                                 directoryCredentialsGenerator))
                                                             .build();
 
 
@@ -56,6 +66,19 @@ public class DirectoryControllerTest {
         return response;
       }
     });
+    when(directoryCredentialsGenerator.generateFor(eq(AuthHelper.VALID_NUMBER))).thenReturn(validCredentials);
+  }
+
+  @Test
+  public void testGetAuthToken() {
+    DirectoryCredentials token =
+            resources.getJerseyTest()
+                     .target("/v1/directory/auth")
+                     .request()
+                     .header("Authorization", AuthHelper.getAuthHeader(AuthHelper.VALID_NUMBER, AuthHelper.VALID_PASSWORD))
+                     .get(DirectoryCredentials.class);
+    assertThat(token.getUsername()).isEqualTo(validCredentials.getUsername());
+    assertThat(token.getPassword()).isEqualTo(validCredentials.getPassword());
   }
 
   @Test

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

@@ -0,0 +1,242 @@
+package org.whispersystems.textsecuregcm.tests.storage;
+
+import com.google.common.base.Optional;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+import org.whispersystems.textsecuregcm.entities.ClientContact;
+import org.whispersystems.textsecuregcm.entities.DirectoryReconciliationRequest;
+import org.whispersystems.textsecuregcm.entities.DirectoryReconciliationResponse;
+import org.whispersystems.textsecuregcm.storage.Account;
+import org.whispersystems.textsecuregcm.storage.Accounts;
+import org.whispersystems.textsecuregcm.storage.DirectoryManager;
+import org.whispersystems.textsecuregcm.storage.DirectoryManager.BatchOperationHandle;
+import org.whispersystems.textsecuregcm.storage.DirectoryReconciler;
+import org.whispersystems.textsecuregcm.storage.DirectoryReconciliationCache;
+import org.whispersystems.textsecuregcm.storage.DirectoryReconciliationClient;
+import org.whispersystems.textsecuregcm.util.Util;
+
+import java.util.Arrays;
+import java.util.Collections;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
+import static org.mockito.Mockito.when;
+
+public class DirectoryReconcilerTest {
+
+  private static final String VALID_NUMBER    = "valid";
+  private static final String INACTIVE_NUMBER = "inactive";
+
+  private static final long ACCOUNT_COUNT = 0L;
+  private static final long INTERVAL_MS   = 30_000L;
+
+  private final Account                       account              = mock(Account.class);
+  private final Account                       inactiveAccount      = mock(Account.class);
+  private final Accounts                      accounts             = mock(Accounts.class);
+  private final BatchOperationHandle          batchOperationHandle = mock(BatchOperationHandle.class);
+  private final DirectoryManager              directoryManager     = mock(DirectoryManager.class);
+  private final DirectoryReconciliationClient reconciliationClient = mock(DirectoryReconciliationClient.class);
+  private final DirectoryReconciliationCache  reconciliationCache  = mock(DirectoryReconciliationCache.class);
+  private final DirectoryReconciler           directoryReconciler  = new DirectoryReconciler(reconciliationClient, reconciliationCache, directoryManager, accounts);
+
+  private final DirectoryReconciliationResponse successResponse  = new DirectoryReconciliationResponse(DirectoryReconciliationResponse.Status.OK);
+  private final DirectoryReconciliationResponse notFoundResponse = new DirectoryReconciliationResponse(DirectoryReconciliationResponse.Status.MISSING);
+
+  @Before
+  public void setup() {
+    when(account.getNumber()).thenReturn(VALID_NUMBER);
+    when(account.isActive()).thenReturn(true);
+    when(account.isVideoSupported()).thenReturn(true);
+    when(account.isVoiceSupported()).thenReturn(true);
+    when(inactiveAccount.getNumber()).thenReturn(INACTIVE_NUMBER);
+    when(inactiveAccount.isActive()).thenReturn(false);
+
+    when(directoryManager.startBatchOperation()).thenReturn(batchOperationHandle);
+
+    when(accounts.getAllFrom(anyInt())).thenReturn(Arrays.asList(account, inactiveAccount));
+    when(accounts.getAllFrom(eq(VALID_NUMBER), anyInt())).thenReturn(Arrays.asList(inactiveAccount));
+    when(accounts.getAllFrom(eq(INACTIVE_NUMBER), anyInt())).thenReturn(Collections.emptyList());
+    when(accounts.getCount()).thenReturn(ACCOUNT_COUNT);
+
+    when(reconciliationClient.sendChunk(any())).thenReturn(successResponse);
+
+    when(reconciliationCache.getLastNumber()).thenReturn(Optional.absent());
+    when(reconciliationCache.claimActiveWork(any(), anyLong())).thenReturn(true);
+    when(reconciliationCache.isAccelerated()).thenReturn(false);
+  }
+
+  @Test
+  public void testGetUncachedAccountCount() {
+    when(reconciliationCache.getCachedAccountCount()).thenReturn(Optional.absent());
+
+    long accountCount = directoryReconciler.getAccountCount();
+
+    assertThat(accountCount).isEqualTo(ACCOUNT_COUNT);
+
+    verify(accounts, times(1)).getCount();
+
+    verify(reconciliationCache, times(1)).getCachedAccountCount();
+    verify(reconciliationCache, times(1)).setCachedAccountCount(eq(ACCOUNT_COUNT));
+
+    verifyNoMoreInteractions(directoryManager);
+    verifyNoMoreInteractions(accounts);
+    verifyNoMoreInteractions(reconciliationClient);
+    verifyNoMoreInteractions(reconciliationCache);
+  }
+
+  @Test
+  public void testGetCachedAccountCount() {
+    when(reconciliationCache.getCachedAccountCount()).thenReturn(Optional.of(ACCOUNT_COUNT));
+
+    long accountCount = directoryReconciler.getAccountCount();
+
+    assertThat(accountCount).isEqualTo(ACCOUNT_COUNT);
+
+    verify(reconciliationCache, times(1)).getCachedAccountCount();
+
+    verifyNoMoreInteractions(directoryManager);
+    verifyNoMoreInteractions(accounts);
+    verifyNoMoreInteractions(reconciliationClient);
+    verifyNoMoreInteractions(reconciliationCache);
+  }
+
+  @Test
+  public void testValid() {
+    long delayMs = directoryReconciler.doPeriodicWork(INTERVAL_MS);
+
+    assertThat(delayMs).isLessThanOrEqualTo(INTERVAL_MS);
+
+    verify(accounts, times(1)).getAllFrom(anyInt());
+
+    ArgumentCaptor<DirectoryReconciliationRequest> request = ArgumentCaptor.forClass(DirectoryReconciliationRequest.class);
+    verify(reconciliationClient, times(1)).sendChunk(request.capture());
+
+    assertThat(request.getValue().getFromNumber()).isNull();
+    assertThat(request.getValue().getToNumber()).isEqualTo(INACTIVE_NUMBER);
+    assertThat(request.getValue().getNumbers()).isEqualTo(Arrays.asList(VALID_NUMBER));
+
+    ArgumentCaptor<ClientContact> addedContact = ArgumentCaptor.forClass(ClientContact.class);
+    verify(directoryManager, times(1)).startBatchOperation();
+    verify(directoryManager, times(1)).add(eq(batchOperationHandle), addedContact.capture());
+    verify(directoryManager, times(1)).remove(eq(batchOperationHandle), eq(INACTIVE_NUMBER));
+    verify(directoryManager, times(1)).stopBatchOperation(eq(batchOperationHandle));
+
+    assertThat(addedContact.getValue().getToken()).isEqualTo(Util.getContactToken(VALID_NUMBER));
+
+    verify(reconciliationCache, times(1)).getLastNumber();
+    verify(reconciliationCache, times(1)).setLastNumber(eq(Optional.of(INACTIVE_NUMBER)));
+    verify(reconciliationCache, times(1)).isAccelerated();
+    verify(reconciliationCache, times(2)).claimActiveWork(any(), anyLong());
+
+    verifyNoMoreInteractions(accounts);
+    verifyNoMoreInteractions(directoryManager);
+    verifyNoMoreInteractions(reconciliationClient);
+    verifyNoMoreInteractions(reconciliationCache);
+  }
+
+  @Test
+  public void testInProgress() {
+    when(reconciliationCache.getLastNumber()).thenReturn(Optional.of(VALID_NUMBER));
+
+    long delayMs = directoryReconciler.doPeriodicWork(INTERVAL_MS);
+
+    assertThat(delayMs).isLessThanOrEqualTo(INTERVAL_MS);
+
+    verify(accounts, times(1)).getAllFrom(eq(VALID_NUMBER), anyInt());
+
+    ArgumentCaptor<DirectoryReconciliationRequest> request = ArgumentCaptor.forClass(DirectoryReconciliationRequest.class);
+    verify(reconciliationClient, times(1)).sendChunk(request.capture());
+
+    assertThat(request.getValue().getFromNumber()).isEqualTo(VALID_NUMBER);
+    assertThat(request.getValue().getToNumber()).isEqualTo(INACTIVE_NUMBER);
+    assertThat(request.getValue().getNumbers()).isEqualTo(Collections.emptyList());
+
+    verify(directoryManager, times(1)).startBatchOperation();
+    verify(directoryManager, times(1)).remove(eq(batchOperationHandle), eq(INACTIVE_NUMBER));
+    verify(directoryManager, times(1)).stopBatchOperation(eq(batchOperationHandle));
+
+    verify(reconciliationCache, times(1)).getLastNumber();
+    verify(reconciliationCache, times(1)).setLastNumber(eq(Optional.of(INACTIVE_NUMBER)));
+    verify(reconciliationCache, times(1)).isAccelerated();
+    verify(reconciliationCache, times(2)).claimActiveWork(any(), anyLong());
+
+    verifyNoMoreInteractions(accounts);
+    verifyNoMoreInteractions(directoryManager);
+    verifyNoMoreInteractions(reconciliationClient);
+    verifyNoMoreInteractions(reconciliationCache);
+  }
+
+  @Test
+  public void testLastChunk() {
+    when(reconciliationCache.getLastNumber()).thenReturn(Optional.of(INACTIVE_NUMBER));
+
+    long delayMs = directoryReconciler.doPeriodicWork(INTERVAL_MS);
+
+    assertThat(delayMs).isLessThanOrEqualTo(INTERVAL_MS);
+
+    verify(accounts, times(1)).getAllFrom(eq(INACTIVE_NUMBER), anyInt());
+
+    ArgumentCaptor<DirectoryReconciliationRequest> request = ArgumentCaptor.forClass(DirectoryReconciliationRequest.class);
+    verify(reconciliationClient, times(1)).sendChunk(request.capture());
+
+    assertThat(request.getValue().getFromNumber()).isEqualTo(INACTIVE_NUMBER);
+    assertThat(request.getValue().getToNumber()).isNull();
+    assertThat(request.getValue().getNumbers()).isEqualTo(Collections.emptyList());
+
+    verify(reconciliationCache, times(1)).getLastNumber();
+    verify(reconciliationCache, times(1)).setLastNumber(eq(Optional.absent()));
+    verify(reconciliationCache, times(1)).clearAccelerate();
+    verify(reconciliationCache, times(1)).isAccelerated();
+    verify(reconciliationCache, times(2)).claimActiveWork(any(), anyLong());
+
+    verifyNoMoreInteractions(accounts);
+    verifyNoMoreInteractions(directoryManager);
+    verifyNoMoreInteractions(reconciliationClient);
+    verifyNoMoreInteractions(reconciliationCache);
+  }
+
+  @Test
+  public void testNotFound() {
+    when(reconciliationClient.sendChunk(any())).thenReturn(notFoundResponse);
+
+    long delayMs = directoryReconciler.doPeriodicWork(INTERVAL_MS);
+
+    assertThat(delayMs).isLessThanOrEqualTo(INTERVAL_MS);
+
+    verify(accounts, times(1)).getAllFrom(anyInt());
+
+    ArgumentCaptor<DirectoryReconciliationRequest> request = ArgumentCaptor.forClass(DirectoryReconciliationRequest.class);
+    verify(reconciliationClient, times(1)).sendChunk(request.capture());
+
+    assertThat(request.getValue().getFromNumber()).isNull();
+    assertThat(request.getValue().getToNumber()).isEqualTo(INACTIVE_NUMBER);
+    assertThat(request.getValue().getNumbers()).isEqualTo(Arrays.asList(VALID_NUMBER));
+
+    ArgumentCaptor<ClientContact> addedContact = ArgumentCaptor.forClass(ClientContact.class);
+    verify(directoryManager, times(1)).startBatchOperation();
+    verify(directoryManager, times(1)).add(eq(batchOperationHandle), addedContact.capture());
+    verify(directoryManager, times(1)).remove(eq(batchOperationHandle), eq(INACTIVE_NUMBER));
+    verify(directoryManager, times(1)).stopBatchOperation(eq(batchOperationHandle));
+
+    assertThat(addedContact.getValue().getToken()).isEqualTo(Util.getContactToken(VALID_NUMBER));
+
+    verify(reconciliationCache, times(1)).getLastNumber();
+    verify(reconciliationCache, times(1)).setLastNumber(eq(Optional.absent()));
+    verify(reconciliationCache, times(1)).clearAccelerate();
+    verify(reconciliationCache, times(1)).claimActiveWork(any(), anyLong());
+
+    verifyNoMoreInteractions(accounts);
+    verifyNoMoreInteractions(directoryManager);
+    verifyNoMoreInteractions(reconciliationClient);
+    verifyNoMoreInteractions(reconciliationCache);
+  }
+
+}