Support for setting and looking up usernames

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

@@ -160,6 +160,7 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
     Accounts         accounts         = new Accounts(accountDatabase);
     PendingAccounts  pendingAccounts  = new PendingAccounts(accountDatabase);
     PendingDevices   pendingDevices   = new PendingDevices(accountDatabase);
+    Usernames        usernames        = new Usernames(accountDatabase);
     Keys             keys             = new Keys(keysDatabase);
     Messages         messages         = new Messages(messageDatabase);
     AbusiveHostRules abusiveHostRules = new AbusiveHostRules(abuseDatabase);
@@ -179,6 +180,7 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
     PendingAccountsManager     pendingAccountsManager     = new PendingAccountsManager(pendingAccounts, cacheClient);
     PendingDevicesManager      pendingDevicesManager      = new PendingDevicesManager (pendingDevices, cacheClient );
     AccountsManager            accountsManager            = new AccountsManager(accounts, directory, cacheClient);
+    UsernamesManager           usernamesManager           = new UsernamesManager(usernames, cacheClient);
     MessagesCache              messagesCache              = new MessagesCache(messagesClient, messages, accountsManager, config.getMessageCacheConfiguration().getPersistDelayMinutes());
     MessagesManager            messagesManager            = new MessagesManager(messages, messagesCache);
     DeadLetterHandler          deadLetterHandler          = new DeadLetterHandler(messagesManager);
@@ -232,7 +234,7 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
     AttachmentControllerV2 attachmentControllerV2 = new AttachmentControllerV2(rateLimiters, config.getAttachmentsConfiguration().getAccessKey(), config.getAttachmentsConfiguration().getAccessSecret(), config.getAttachmentsConfiguration().getRegion(), config.getAttachmentsConfiguration().getBucket());
     KeysController         keysController         = new KeysController(rateLimiters, keys, accountsManager, directoryQueue);
     MessageController      messageController      = new MessageController(rateLimiters, pushSender, receiptSender, accountsManager, messagesManager, apnFallbackManager);
-    ProfileController      profileController      = new ProfileController(rateLimiters, accountsManager, config.getCdnConfiguration());
+    ProfileController      profileController      = new ProfileController(rateLimiters, accountsManager, usernamesManager, config.getCdnConfiguration());
     StickerController      stickerController      = new StickerController(rateLimiters, config.getCdnConfiguration().getAccessKey(), config.getCdnConfiguration().getAccessSecret(), config.getCdnConfiguration().getRegion(), config.getCdnConfiguration().getBucket());
 
     AuthFilter<BasicCredentials, Account>                  accountAuthFilter                  = new BasicCredentialAuthFilter.Builder<Account>().setAuthenticator(accountAuthenticator).buildAuthFilter                                  ();
@@ -242,7 +244,7 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
                                                                                       DisabledPermittedAccount.class, disabledPermittedAccountAuthFilter)));
     environment.jersey().register(new PolymorphicAuthValueFactoryProvider.Binder<>(ImmutableSet.of(Account.class, DisabledPermittedAccount.class)));
 
-    environment.jersey().register(new AccountController(pendingAccountsManager, accountsManager, abusiveHostRules, rateLimiters, smsSender, directoryQueue, messagesManager, turnTokenGenerator, config.getTestDevices(), recaptchaClient, gcmSender, apnSender, backupCredentialsGenerator));
+    environment.jersey().register(new AccountController(pendingAccountsManager, accountsManager, usernamesManager, abusiveHostRules, rateLimiters, smsSender, directoryQueue, messagesManager, turnTokenGenerator, config.getTestDevices(), recaptchaClient, gcmSender, apnSender, backupCredentialsGenerator));
     environment.jersey().register(new DeviceController(pendingDevicesManager, accountsManager, messagesManager, directoryQueue, rateLimiters, config.getMaxDevices()));
     environment.jersey().register(new DirectoryController(rateLimiters, directory, directoryCredentialsGenerator));
     environment.jersey().register(new ProvisioningController(rateLimiters, pushSender));

service/src/main/java/org/whispersystems/textsecuregcm/configuration/RateLimitsConfiguration.java

@@ -71,6 +71,12 @@ public class RateLimitsConfiguration {
   @JsonProperty
   private RateLimitConfiguration stickerPack = new RateLimitConfiguration(50, 20 / (24.0 * 60.0));
 
+  @JsonProperty
+  private RateLimitConfiguration usernameLookup = new RateLimitConfiguration(100, 100 / (24.0 * 60.0));
+
+  @JsonProperty
+  private RateLimitConfiguration usernameSet = new RateLimitConfiguration(100, 100 / (24.0 * 60.0));
+
   public RateLimitConfiguration getAutoBlock() {
     return autoBlock;
   }
@@ -139,6 +145,14 @@ public class RateLimitsConfiguration {
     return stickerPack;
   }
 
+  public RateLimitConfiguration getUsernameLookup() {
+    return usernameLookup;
+  }
+
+  public RateLimitConfiguration getUsernameSet() {
+    return usernameSet;
+  }
+
   public static class RateLimitConfiguration {
     @JsonProperty
     private int bucketSize;

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

@@ -35,9 +35,9 @@ import org.whispersystems.textsecuregcm.auth.TurnTokenGenerator;
 import org.whispersystems.textsecuregcm.entities.AccountAttributes;
 import org.whispersystems.textsecuregcm.entities.AccountCreationResult;
 import org.whispersystems.textsecuregcm.entities.ApnRegistrationId;
+import org.whispersystems.textsecuregcm.entities.DeprecatedPin;
 import org.whispersystems.textsecuregcm.entities.DeviceName;
 import org.whispersystems.textsecuregcm.entities.GcmRegistrationId;
-import org.whispersystems.textsecuregcm.entities.DeprecatedPin;
 import org.whispersystems.textsecuregcm.entities.RegistrationLock;
 import org.whispersystems.textsecuregcm.entities.RegistrationLockFailure;
 import org.whispersystems.textsecuregcm.limits.RateLimiters;
@@ -55,6 +55,7 @@ import org.whispersystems.textsecuregcm.storage.AccountsManager;
 import org.whispersystems.textsecuregcm.storage.Device;
 import org.whispersystems.textsecuregcm.storage.MessagesManager;
 import org.whispersystems.textsecuregcm.storage.PendingAccountsManager;
+import org.whispersystems.textsecuregcm.storage.UsernamesManager;
 import org.whispersystems.textsecuregcm.util.Constants;
 import org.whispersystems.textsecuregcm.util.Hex;
 import org.whispersystems.textsecuregcm.util.Util;
@@ -102,6 +103,7 @@ public class AccountController {
 
   private final PendingAccountsManager             pendingAccounts;
   private final AccountsManager                    accounts;
+  private final UsernamesManager                   usernames;
   private final AbusiveHostRules                   abusiveHostRules;
   private final RateLimiters                       rateLimiters;
   private final SmsSender                          smsSender;
@@ -116,6 +118,7 @@ public class AccountController {
 
   public AccountController(PendingAccountsManager pendingAccounts,
                            AccountsManager accounts,
+                           UsernamesManager usernames,
                            AbusiveHostRules abusiveHostRules,
                            RateLimiters rateLimiters,
                            SmsSender smsSenderFactory,
@@ -130,6 +133,7 @@ public class AccountController {
   {
     this.pendingAccounts                   = pendingAccounts;
     this.accounts                          = accounts;
+    this.usernames                         = usernames;
     this.abusiveHostRules                  = abusiveHostRules;
     this.rateLimiters                      = rateLimiters;
     this.smsSender                         = smsSenderFactory;
@@ -517,6 +521,36 @@ public class AccountController {
     return new AccountCreationResult(account.getUuid());
   }
 
+  @DELETE
+  @Path("/username")
+  @Produces(MediaType.APPLICATION_JSON)
+  public void deleteUsername(@Auth Account account) {
+    usernames.delete(account.getUuid());
+  }
+
+  @PUT
+  @Path("/username/{username}")
+  @Produces(MediaType.APPLICATION_JSON)
+  public Response setUsername(@Auth Account account, @PathParam("username") String username) throws RateLimitExceededException {
+    rateLimiters.getUsernameSetLimiter().validate(account.getUuid().toString());
+
+    if (username == null || username.isEmpty()) {
+      return Response.status(Response.Status.BAD_REQUEST).build();
+    }
+
+    username = username.toLowerCase();
+
+    if (!username.matches("^[a-z0-9_]+$")) {
+      return Response.status(Response.Status.BAD_REQUEST).build();
+    }
+
+    if (!usernames.put(account.getUuid(), username)) {
+      return Response.status(Response.Status.CONFLICT).build();
+    }
+
+    return Response.ok().build();
+  }
+
   private CaptchaRequirement requiresCaptcha(String number, String transport, String forwardedFor,
                                              String requester,
                                              Optional<String>                 captchaToken,

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

@@ -23,6 +23,7 @@ import org.whispersystems.textsecuregcm.s3.PolicySigner;
 import org.whispersystems.textsecuregcm.s3.PostPolicyGenerator;
 import org.whispersystems.textsecuregcm.storage.Account;
 import org.whispersystems.textsecuregcm.storage.AccountsManager;
+import org.whispersystems.textsecuregcm.storage.UsernamesManager;
 import org.whispersystems.textsecuregcm.util.Pair;
 
 import javax.ws.rs.GET;
@@ -39,6 +40,7 @@ import java.security.SecureRandom;
 import java.time.ZoneOffset;
 import java.time.ZonedDateTime;
 import java.util.Optional;
+import java.util.UUID;
 
 import io.dropwizard.auth.Auth;
 
@@ -48,6 +50,7 @@ public class ProfileController {
 
   private final RateLimiters     rateLimiters;
   private final AccountsManager  accountsManager;
+  private final UsernamesManager usernamesManager;
 
   private final PolicySigner        policySigner;
   private final PostPolicyGenerator policyGenerator;
@@ -57,6 +60,7 @@ public class ProfileController {
 
   public ProfileController(RateLimiters rateLimiters,
                            AccountsManager accountsManager,
+                           UsernamesManager usernamesManager,
                            CdnConfiguration profilesConfiguration)
   {
     AWSCredentials         credentials         = new BasicAWSCredentials(profilesConfiguration.getAccessKey(), profilesConfiguration.getAccessSecret());
@@ -64,6 +68,7 @@ public class ProfileController {
 
     this.rateLimiters       = rateLimiters;
     this.accountsManager    = accountsManager;
+    this.usernamesManager   = usernamesManager;
     this.bucket             = profilesConfiguration.getBucket();
     this.s3client           = AmazonS3Client.builder()
                                             .withCredentials(credentialsProvider)
@@ -99,13 +104,52 @@ public class ProfileController {
     Optional<Account> accountProfile = accountsManager.get(identifier);
     OptionalAccess.verify(requestAccount, accessKey, accountProfile);
 
-    //noinspection ConstantConditions,OptionalGetWithoutIsPresent
+    Optional<String> username = Optional.empty();
+
+    if (!identifier.hasNumber()) {
+      //noinspection OptionalGetWithoutIsPresent
+      username = usernamesManager.get(accountProfile.get().getUuid());
+    }
+
     return new Profile(accountProfile.get().getProfileName(),
                        accountProfile.get().getAvatar(),
                        accountProfile.get().getIdentityKey(),
                        UnidentifiedAccessChecksum.generateFor(accountProfile.get().getUnidentifiedAccessKey()),
                        accountProfile.get().isUnrestrictedUnidentifiedAccess(),
-                       new UserCapabilities(accountProfile.get().isUuidAddressingSupported()));
+                       new UserCapabilities(accountProfile.get().isUuidAddressingSupported()),
+                       username.orElse(null),
+                       null);
+  }
+
+  @Timed
+  @GET
+  @Produces(MediaType.APPLICATION_JSON)
+  @Path("/username/{username}")
+  public Profile getProfileByUsername(@Auth Account account, @PathParam("username") String username) throws RateLimitExceededException {
+    rateLimiters.getUsernameLookupLimiter().validate(account.getUuid().toString());
+
+    username = username.toLowerCase();
+
+    Optional<UUID> uuid = usernamesManager.get(username);
+
+    if (!uuid.isPresent()) {
+      throw new WebApplicationException(Response.status(Response.Status.NOT_FOUND).build());
+    }
+
+    Optional<Account> accountProfile = accountsManager.get(uuid.get());
+
+    if (!accountProfile.isPresent()) {
+      throw new WebApplicationException(Response.status(Response.Status.NOT_FOUND).build());
+    }
+
+    return new Profile(accountProfile.get().getProfileName(),
+                       accountProfile.get().getAvatar(),
+                       accountProfile.get().getIdentityKey(),
+                       UnidentifiedAccessChecksum.generateFor(accountProfile.get().getUnidentifiedAccessKey()),
+                       accountProfile.get().isUnrestrictedUnidentifiedAccess(),
+                       new UserCapabilities(accountProfile.get().isUuidAddressingSupported()),
+                       username,
+                       accountProfile.get().getUuid());
   }
 
   @Timed

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

@@ -3,6 +3,8 @@ package org.whispersystems.textsecuregcm.entities;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.google.common.annotations.VisibleForTesting;
 
+import java.util.UUID;
+
 public class Profile {
 
   @JsonProperty
@@ -23,11 +25,17 @@ public class Profile {
   @JsonProperty
   private UserCapabilities capabilities;
 
+  @JsonProperty
+  private String username;
+
+  @JsonProperty
+  private UUID uuid;
+
   public Profile() {}
 
   public Profile(String name, String avatar, String identityKey,
                  String unidentifiedAccess, boolean unrestrictedUnidentifiedAccess,
-                 UserCapabilities capabilities)
+                 UserCapabilities capabilities, String username, UUID uuid)
   {
     this.name                           = name;
     this.avatar                         = avatar;
@@ -35,6 +43,8 @@ public class Profile {
     this.unidentifiedAccess             = unidentifiedAccess;
     this.unrestrictedUnidentifiedAccess = unrestrictedUnidentifiedAccess;
     this.capabilities                   = capabilities;
+    this.username                       = username;
+    this.uuid                           = uuid;
   }
 
   @VisibleForTesting
@@ -67,4 +77,13 @@ public class Profile {
     return capabilities;
   }
 
+  @VisibleForTesting
+  public String getUsername() {
+    return username;
+  }
+
+  @VisibleForTesting
+  public UUID getUuid() {
+    return uuid;
+  }
 }

service/src/main/java/org/whispersystems/textsecuregcm/limits/RateLimiters.java

@@ -43,6 +43,8 @@ public class RateLimiters {
 
   private final RateLimiter profileLimiter;
   private final RateLimiter stickerPackLimiter;
+  private final RateLimiter usernameLookupLimiter;
+  private final RateLimiter usernameSetLimiter;
 
   public RateLimiters(RateLimitsConfiguration config, ReplicatedJedisPool cacheClient) {
     this.smsDestinationLimiter = new RateLimiter(cacheClient, "smsDestination",
@@ -112,6 +114,14 @@ public class RateLimiters {
     this.stickerPackLimiter = new RateLimiter(cacheClient, "stickerPack",
                                               config.getStickerPack().getBucketSize(),
                                               config.getStickerPack().getLeakRatePerMinute());
+
+    this.usernameLookupLimiter = new RateLimiter(cacheClient, "usernameLookup",
+                                                 config.getUsernameLookup().getBucketSize(),
+                                                 config.getUsernameLookup().getLeakRatePerMinute());
+
+    this.usernameSetLimiter = new RateLimiter(cacheClient, "usernameSet",
+                                              config.getUsernameSet().getBucketSize(),
+                                              config.getUsernameSet().getLeakRatePerMinute());
   }
 
   public RateLimiter getAllocateDeviceLimiter() {
@@ -182,4 +192,12 @@ public class RateLimiters {
     return stickerPackLimiter;
   }
 
+  public RateLimiter getUsernameLookupLimiter() {
+    return usernameLookupLimiter;
+  }
+
+  public RateLimiter getUsernameSetLimiter() {
+    return usernameSetLimiter;
+  }
+
 }

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

@@ -0,0 +1,88 @@
+package org.whispersystems.textsecuregcm.storage;
+
+import com.codahale.metrics.MetricRegistry;
+import com.codahale.metrics.SharedMetricRegistries;
+import com.codahale.metrics.Timer;
+import org.jdbi.v3.core.JdbiException;
+import org.whispersystems.textsecuregcm.storage.mappers.AccountRowMapper;
+import org.whispersystems.textsecuregcm.util.Constants;
+
+import java.sql.SQLException;
+import java.util.Optional;
+import java.util.UUID;
+
+import static com.codahale.metrics.MetricRegistry.name;
+
+public class Usernames {
+
+  public static final String ID       = "id";
+  public static final String UID      = "uuid";
+  public static final String USERNAME = "username";
+
+  private final MetricRegistry metricRegistry     = SharedMetricRegistries.getOrCreate(Constants.METRICS_NAME);
+  private final Timer          createTimer        = metricRegistry.timer(name(Usernames.class, "create"       ));
+  private final Timer          deleteTimer        = metricRegistry.timer(name(Usernames.class, "delete"       ));
+  private final Timer          getByUsernameTimer = metricRegistry.timer(name(Usernames.class, "getByUsername"));
+  private final Timer          getByUuidTimer     = metricRegistry.timer(name(Usernames.class, "getByUuid"    ));
+
+  private final FaultTolerantDatabase database;
+
+  public Usernames(FaultTolerantDatabase database) {
+    this.database = database;
+    this.database.getDatabase().registerRowMapper(new AccountRowMapper());
+  }
+
+  public boolean put(UUID uuid, String username) {
+    return database.with(jdbi -> jdbi.withHandle(handle -> {
+      try (Timer.Context ignored = createTimer.time()) {
+        int modified = handle.createUpdate("INSERT INTO usernames (" + UID + ", " + USERNAME + ") VALUES (:uuid, :username) ON CONFLICT (" + UID + ") DO UPDATE SET " + USERNAME + " = EXCLUDED.username")
+                             .bind("uuid", uuid)
+                             .bind("username", username)
+                             .execute();
+
+        return modified > 0;
+      } catch (JdbiException e) {
+        if (e.getCause() instanceof SQLException) {
+          if (((SQLException)e.getCause()).getSQLState().equals("23505")) {
+            return false;
+          }
+        }
+
+        throw e;
+      }
+    }));
+  }
+
+  public void delete(UUID uuid) {
+    database.use(jdbi -> jdbi.useHandle(handle -> {
+      try (Timer.Context ignored = deleteTimer.time()) {
+        handle.createUpdate("DELETE FROM usernames WHERE " + UID + " = :uuid")
+              .bind("uuid", uuid)
+              .execute();
+      }
+    }));
+  }
+
+  public Optional<UUID> get(String username) {
+    return database.with(jdbi -> jdbi.withHandle(handle -> {
+      try (Timer.Context ignored = getByUsernameTimer.time()) {
+        return handle.createQuery("SELECT " + UID + " FROM usernames WHERE " + USERNAME + " = :username")
+                     .bind("username", username)
+                     .mapTo(UUID.class)
+                     .findFirst();
+      }
+    }));
+  }
+
+  public Optional<String> get(UUID uuid) {
+    return database.with(jdbi -> jdbi.withHandle(handle -> {
+      try (Timer.Context ignored = getByUuidTimer.time()) {
+        return handle.createQuery("SELECT " + USERNAME + " FROM usernames WHERE " + UID + " = :uuid")
+                     .bind("uuid", uuid)
+                     .mapTo(String.class)
+                     .findFirst();
+      }
+    }));
+  }
+
+}

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

@@ -0,0 +1,160 @@
+package org.whispersystems.textsecuregcm.storage;
+
+import com.codahale.metrics.MetricRegistry;
+import com.codahale.metrics.SharedMetricRegistries;
+import com.codahale.metrics.Timer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.whispersystems.textsecuregcm.redis.ReplicatedJedisPool;
+import org.whispersystems.textsecuregcm.util.Constants;
+
+import java.util.Optional;
+import java.util.UUID;
+
+import static com.codahale.metrics.MetricRegistry.name;
+import redis.clients.jedis.Jedis;
+import redis.clients.jedis.exceptions.JedisException;
+
+public class UsernamesManager {
+
+  private static final MetricRegistry metricRegistry        = SharedMetricRegistries.getOrCreate(Constants.METRICS_NAME);
+  private static final Timer          createTimer           = metricRegistry.timer(name(AccountsManager.class, "create"          ));
+  private static final Timer          deleteTimer           = metricRegistry.timer(name(AccountsManager.class, "delete"          ));
+  private static final Timer          getByUuidTimer        = metricRegistry.timer(name(AccountsManager.class, "getByUuid"       ));
+  private static final Timer          getByUsernameTimer    = metricRegistry.timer(name(AccountsManager.class, "getByUsername"   ));
+
+  private static final Timer          redisSetTimer         = metricRegistry.timer(name(AccountsManager.class, "redisSet"        ));
+  private static final Timer          redisUuidGetTimer     = metricRegistry.timer(name(AccountsManager.class, "redisUuidGet"    ));
+  private static final Timer          redisUsernameGetTimer = metricRegistry.timer(name(AccountsManager.class, "redisUsernameGet"));
+
+  private final Logger logger = LoggerFactory.getLogger(AccountsManager.class);
+
+  private final Usernames           usernames;
+  private final ReplicatedJedisPool cacheClient;
+
+  public UsernamesManager(Usernames usernames, ReplicatedJedisPool cacheClient) {
+    this.usernames   = usernames;
+    this.cacheClient = cacheClient;
+  }
+
+  public boolean put(UUID uuid, String username) {
+    try (Timer.Context ignored = createTimer.time()) {
+      if (databasePut(uuid, username)) {
+        redisSet(uuid, username);
+
+        return true;
+      }
+
+      return false;
+    }
+  }
+
+  public Optional<UUID> get(String username) {
+    try (Timer.Context ignored = getByUsernameTimer.time()) {
+      Optional<UUID> uuid = redisGet(username);
+
+      if (uuid.isPresent()) {
+        return uuid;
+      }
+
+      Optional<UUID> retrieved = databaseGet(username);
+      retrieved.ifPresent(retrievedUuid -> redisSet(retrievedUuid, username));
+
+      return retrieved;
+    }
+  }
+
+  public Optional<String> get(UUID uuid) {
+    try (Timer.Context ignored = getByUuidTimer.time()) {
+      Optional<String> username = redisGet(uuid);
+
+      if (username.isPresent()) {
+        return username;
+      }
+
+      Optional<String> retrieved = databaseGet(uuid);
+      retrieved.ifPresent(retrievedUsername -> redisSet(uuid, retrievedUsername));
+
+      return retrieved;
+    }
+  }
+
+  public void delete(UUID uuid) {
+    try (Timer.Context ignored = deleteTimer.time()) {
+      redisDelete(uuid);
+      databaseDelete(uuid);
+    }
+  }
+
+  private boolean databasePut(UUID uuid, String username) {
+    return usernames.put(uuid, username);
+  }
+
+  private Optional<UUID> databaseGet(String username) {
+    return usernames.get(username);
+  }
+
+  private void databaseDelete(UUID uuid) {
+    usernames.delete(uuid);
+  }
+
+  private Optional<String> databaseGet(UUID uuid) {
+    return usernames.get(uuid);
+  }
+
+  private void redisSet(UUID uuid, String username) {
+    try (Jedis         jedis   = cacheClient.getWriteResource();
+         Timer.Context ignored = redisSetTimer.time())
+    {
+      jedis.set(getUuidMapKey(uuid), username);
+      jedis.set(getUsernameMapKey(username), uuid.toString());
+    }
+  }
+
+  private Optional<UUID> redisGet(String username) {
+    try (Jedis         jedis   = cacheClient.getReadResource();
+         Timer.Context ignored = redisUsernameGetTimer.time())
+    {
+      String result = jedis.get(getUsernameMapKey(username));
+
+      if (result == null) return Optional.empty();
+      else                return Optional.of(UUID.fromString(result));
+    } catch (JedisException e) {
+      logger.warn("Redis get failure", e);
+      return Optional.empty();
+    }
+  }
+
+  private Optional<String> redisGet(UUID uuid) {
+    try (Jedis         jedis   = cacheClient.getReadResource();
+         Timer.Context ignored = redisUuidGetTimer.time())
+    {
+      return Optional.ofNullable(jedis.get(getUuidMapKey(uuid)));
+    } catch (JedisException e) {
+      logger.warn("Redis get failure", e);
+      return Optional.empty();
+    }
+  }
+
+  private void redisDelete(UUID uuid) {
+    try (Jedis         jedis   = cacheClient.getWriteResource();
+         Timer.Context ignored = redisUuidGetTimer.time())
+    {
+      Optional<String> username = redisGet(uuid);
+
+      if (username.isPresent()) {
+        jedis.del(getUsernameMapKey(username.get()));
+        jedis.del(getUuidMapKey(uuid));
+      }
+    }
+  }
+
+  private String getUuidMapKey(UUID uuid) {
+    return "UsernameByUuid::" + uuid.toString();
+  }
+
+  private String getUsernameMapKey(String username) {
+    return "UsernameByUsername::" + username;
+  }
+
+}

service/src/main/resources/accountsdb.xml

@@ -207,4 +207,20 @@
         <sql>CREATE UNIQUE INDEX CONCURRENTLY uuid_index ON accounts (uuid);</sql>
     </changeSet>
 
+    <changeSet id="9" author="moxie">
+        <createTable tableName="usernames">
+            <column name="id" type="bigint" autoIncrement="true">
+                <constraints nullable="false" primaryKey="true"/>
+            </column>
+
+            <column name="uuid" type="uuid">
+                <constraints nullable="false" unique="true"/>
+            </column>
+
+            <column name="username" type="text">
+                <constraints nullable="false" unique="true"/>
+            </column>
+        </createTable>
+    </changeSet>
+
 </databaseChangeLog>

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

@@ -37,6 +37,7 @@ import org.whispersystems.textsecuregcm.storage.Account;
 import org.whispersystems.textsecuregcm.storage.AccountsManager;
 import org.whispersystems.textsecuregcm.storage.MessagesManager;
 import org.whispersystems.textsecuregcm.storage.PendingAccountsManager;
+import org.whispersystems.textsecuregcm.storage.UsernamesManager;
 import org.whispersystems.textsecuregcm.tests.util.AuthHelper;
 import org.whispersystems.textsecuregcm.util.Hex;
 import org.whispersystems.textsecuregcm.util.SystemMapper;
@@ -86,6 +87,7 @@ public class AccountControllerTest {
   private        RateLimiter            smsVoiceIpLimiter      = mock(RateLimiter.class           );
   private        RateLimiter            smsVoicePrefixLimiter  = mock(RateLimiter.class);
   private        RateLimiter            autoBlockLimiter       = mock(RateLimiter.class);
+  private        RateLimiter            usernameSetLimiter     = mock(RateLimiter.class);
   private        SmsSender              smsSender              = mock(SmsSender.class             );
   private        DirectoryQueue         directoryQueue         = mock(DirectoryQueue.class);
   private        MessagesManager        storedMessages         = mock(MessagesManager.class       );
@@ -96,6 +98,7 @@ public class AccountControllerTest {
   private        RecaptchaClient        recaptchaClient        = mock(RecaptchaClient.class);
   private        GCMSender              gcmSender              = mock(GCMSender.class);
   private        APNSender              apnSender              = mock(APNSender.class);
+  private UsernamesManager              usernamesManager       = mock(UsernamesManager.class);
 
   private byte[] registration_lock_key = new byte[32];
   private ExternalServiceCredentialGenerator storageCredentialGenerator = new ExternalServiceCredentialGenerator(new byte[32], new byte[32], false);
@@ -109,6 +112,7 @@ public class AccountControllerTest {
                                                             .setTestContainerFactory(new GrizzlyWebTestContainerFactory())
                                                             .addResource(new AccountController(pendingAccountsManager,
                                                                                                accountsManager,
+                                                                                               usernamesManager,
                                                                                                abusiveHostRules,
                                                                                                rateLimiters,
                                                                                                smsSender,
@@ -135,6 +139,7 @@ public class AccountControllerTest {
     when(rateLimiters.getSmsVoiceIpLimiter()).thenReturn(smsVoiceIpLimiter);
     when(rateLimiters.getSmsVoicePrefixLimiter()).thenReturn(smsVoicePrefixLimiter);
     when(rateLimiters.getAutoBlockLimiter()).thenReturn(autoBlockLimiter);
+    when(rateLimiters.getUsernameSetLimiter()).thenReturn(usernameSetLimiter);
 
     when(timeProvider.getCurrentTimeMillis()).thenReturn(System.currentTimeMillis());
 
@@ -160,6 +165,9 @@ public class AccountControllerTest {
     when(accountsManager.get(eq(SENDER_OLD))).thenReturn(Optional.empty());
     when(accountsManager.get(eq(SENDER_PREAUTH))).thenReturn(Optional.empty());
 
+    when(usernamesManager.put(eq(AuthHelper.VALID_UUID), eq("n00bkiller"))).thenReturn(true);
+    when(usernamesManager.put(eq(AuthHelper.VALID_UUID), eq("takenusername"))).thenReturn(false);
+
     when(abusiveHostRules.getAbusiveHostRulesFor(eq(ABUSIVE_HOST))).thenReturn(Collections.singletonList(new AbusiveHostRule(ABUSIVE_HOST, true, Collections.emptyList())));
     when(abusiveHostRules.getAbusiveHostRulesFor(eq(RESTRICTED_HOST))).thenReturn(Collections.singletonList(new AbusiveHostRule(RESTRICTED_HOST, false, Collections.singletonList("+123"))));
     when(abusiveHostRules.getAbusiveHostRulesFor(eq(NICE_HOST))).thenReturn(Collections.emptyList());
@@ -820,4 +828,77 @@ public class AccountControllerTest {
     assertThat(response.getUuid()).isEqualTo(AuthHelper.VALID_UUID);
   }
 
+  @Test
+  public void testSetUsername() {
+    Response response =
+        resources.getJerseyTest()
+                 .target("/v1/accounts/username/n00bkiller")
+                 .request()
+                 .header("Authorization", AuthHelper.getAuthHeader(AuthHelper.VALID_NUMBER, AuthHelper.VALID_PASSWORD))
+                 .put(Entity.text(""));
+
+    assertThat(response.getStatus()).isEqualTo(200);
+  }
+
+  @Test
+  public void testSetTakenUsername() {
+    Response response =
+        resources.getJerseyTest()
+                 .target("/v1/accounts/username/takenusername")
+                 .request()
+                 .header("Authorization", AuthHelper.getAuthHeader(AuthHelper.VALID_NUMBER, AuthHelper.VALID_PASSWORD))
+                 .put(Entity.text(""));
+
+    assertThat(response.getStatus()).isEqualTo(409);
+  }
+
+  @Test
+  public void testSetInvalidUsername() {
+    Response response =
+        resources.getJerseyTest()
+                 .target("/v1/accounts/username/pаypal")
+                 .request()
+                 .header("Authorization", AuthHelper.getAuthHeader(AuthHelper.VALID_NUMBER, AuthHelper.VALID_PASSWORD))
+                 .put(Entity.text(""));
+
+    assertThat(response.getStatus()).isEqualTo(400);
+  }
+
+  @Test
+  public void testSetUsernameBadAuth() {
+    Response response =
+        resources.getJerseyTest()
+                 .target("/v1/accounts/username/n00bkiller")
+                 .request()
+                 .header("Authorization", AuthHelper.getAuthHeader(AuthHelper.VALID_NUMBER, AuthHelper.INVALID_PASSWORD))
+                 .put(Entity.text(""));
+
+    assertThat(response.getStatus()).isEqualTo(401);
+  }
+
+  @Test
+  public void testDeleteUsername() {
+    Response response =
+        resources.getJerseyTest()
+                 .target("/v1/accounts/username/")
+                 .request()
+                 .header("Authorization", AuthHelper.getAuthHeader(AuthHelper.VALID_NUMBER, AuthHelper.VALID_PASSWORD))
+                 .delete();
+
+    assertThat(response.getStatus()).isEqualTo(204);
+    verify(usernamesManager, times(1)).delete(eq(AuthHelper.VALID_UUID));
+  }
+
+  @Test
+  public void testDeleteUsernameBadAuth() {
+    Response response =
+        resources.getJerseyTest()
+                 .target("/v1/accounts/username/")
+                 .request()
+                 .header("Authorization", AuthHelper.getAuthHeader(AuthHelper.VALID_NUMBER, AuthHelper.INVALID_PASSWORD))
+                 .delete();
+
+    assertThat(response.getStatus()).isEqualTo(401);
+  }
+
 }

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

@@ -16,6 +16,7 @@ import org.whispersystems.textsecuregcm.limits.RateLimiter;
 import org.whispersystems.textsecuregcm.limits.RateLimiters;
 import org.whispersystems.textsecuregcm.storage.Account;
 import org.whispersystems.textsecuregcm.storage.AccountsManager;
+import org.whispersystems.textsecuregcm.storage.UsernamesManager;
 import org.whispersystems.textsecuregcm.tests.util.AuthHelper;
 import org.whispersystems.textsecuregcm.util.SystemMapper;
 
@@ -29,10 +30,12 @@ import static org.mockito.Mockito.*;
 
 public class ProfileControllerTest {
 
-  private static AccountsManager       accountsManager = mock(AccountsManager.class      );
+  private static AccountsManager  accountsManager     = mock(AccountsManager.class );
-  private static RateLimiters          rateLimiters    = mock(RateLimiters.class         );
+  private static UsernamesManager usernamesManager    = mock(UsernamesManager.class);
-  private static RateLimiter           rateLimiter     = mock(RateLimiter.class          );
+  private static RateLimiters     rateLimiters        = mock(RateLimiters.class    );
-  private static CdnConfiguration configuration   = mock(CdnConfiguration.class);
+  private static RateLimiter      rateLimiter         = mock(RateLimiter.class     );
+  private static RateLimiter      usernameRateLimiter = mock(RateLimiter.class     );
+  private static CdnConfiguration configuration       = mock(CdnConfiguration.class);
 
   static {
     when(configuration.getAccessKey()).thenReturn("accessKey");
@@ -49,12 +52,14 @@ public class ProfileControllerTest {
                                                                    .setTestContainerFactory(new GrizzlyWebTestContainerFactory())
                                                                    .addResource(new ProfileController(rateLimiters,
                                                                                                       accountsManager,
+                                                                                                      usernamesManager,
                                                                                                       configuration))
                                                                    .build();
 
   @Before
   public void setup() throws Exception {
     when(rateLimiters.getProfileLimiter()).thenReturn(rateLimiter);
+    when(rateLimiters.getUsernameLookupLimiter()).thenReturn(usernameRateLimiter);
 
     Account profileAccount = mock(Account.class);
 
@@ -62,6 +67,7 @@ public class ProfileControllerTest {
     when(profileAccount.getProfileName()).thenReturn("baz");
     when(profileAccount.getAvatar()).thenReturn("profiles/bang");
     when(profileAccount.getAvatarDigest()).thenReturn("buh");
+    when(profileAccount.getUuid()).thenReturn(AuthHelper.VALID_UUID_TWO);
     when(profileAccount.isEnabled()).thenReturn(true);
     when(profileAccount.isUuidAddressingSupported()).thenReturn(false);
 
@@ -75,15 +81,37 @@ public class ProfileControllerTest {
     when(capabilitiesAccount.isUuidAddressingSupported()).thenReturn(true);
 
     when(accountsManager.get(AuthHelper.VALID_NUMBER_TWO)).thenReturn(Optional.of(profileAccount));
+    when(accountsManager.get(AuthHelper.VALID_UUID_TWO)).thenReturn(Optional.of(profileAccount));
+    when(usernamesManager.get(AuthHelper.VALID_UUID_TWO)).thenReturn(Optional.of("n00bkiller"));
+    when(usernamesManager.get("n00bkiller")).thenReturn(Optional.of(AuthHelper.VALID_UUID_TWO));
     when(accountsManager.get(argThat((ArgumentMatcher<AmbiguousIdentifier>) identifier -> identifier != null && identifier.hasNumber() && identifier.getNumber().equals(AuthHelper.VALID_NUMBER_TWO)))).thenReturn(Optional.of(profileAccount));
+    when(accountsManager.get(argThat((ArgumentMatcher<AmbiguousIdentifier>) identifier -> identifier != null && identifier.hasUuid() && identifier.getUuid().equals(AuthHelper.VALID_UUID_TWO)))).thenReturn(Optional.of(profileAccount));
 
     when(accountsManager.get(AuthHelper.VALID_NUMBER)).thenReturn(Optional.of(capabilitiesAccount));
     when(accountsManager.get(argThat((ArgumentMatcher<AmbiguousIdentifier>) identifier -> identifier != null && identifier.hasNumber() && identifier.getNumber().equals(AuthHelper.VALID_NUMBER)))).thenReturn(Optional.of(capabilitiesAccount));
   }
 
+  @Test
+  public void testProfileGetByUuid() throws RateLimitExceededException {
+    Profile profile= resources.getJerseyTest()
+                              .target("/v1/profile/" + AuthHelper.VALID_UUID_TWO)
+                              .request()
+                              .header("Authorization", AuthHelper.getAuthHeader(AuthHelper.VALID_NUMBER, AuthHelper.VALID_PASSWORD))
+                              .get(Profile.class);
+
+    assertThat(profile.getIdentityKey()).isEqualTo("bar");
+    assertThat(profile.getName()).isEqualTo("baz");
+    assertThat(profile.getAvatar()).isEqualTo("profiles/bang");
+    assertThat(profile.getUsername()).isEqualTo("n00bkiller");
+
+    verify(accountsManager, times(1)).get(argThat((ArgumentMatcher<AmbiguousIdentifier>) identifier -> identifier != null && identifier.hasUuid() && identifier.getUuid().equals(AuthHelper.VALID_UUID_TWO)));
+    verify(usernamesManager, times(1)).get(eq(AuthHelper.VALID_UUID_TWO));
+    verify(rateLimiter, times(2)).validate(eq(AuthHelper.VALID_NUMBER));
+    reset(rateLimiter);
+  }
 
   @Test
-  public void testProfileGet() throws RateLimitExceededException {
+  public void testProfileGetByNumber() throws RateLimitExceededException {
     Profile profile= resources.getJerseyTest()
                               .target("/v1/profile/" + AuthHelper.VALID_NUMBER_TWO)
                               .request()
@@ -94,10 +122,32 @@ public class ProfileControllerTest {
     assertThat(profile.getName()).isEqualTo("baz");
     assertThat(profile.getAvatar()).isEqualTo("profiles/bang");
     assertThat(profile.getCapabilities().isUuid()).isFalse();
+    assertThat(profile.getUsername()).isNull();
+    assertThat(profile.getUuid()).isNull();;
 
     verify(accountsManager, times(1)).get(argThat((ArgumentMatcher<AmbiguousIdentifier>) identifier -> identifier != null && identifier.hasNumber() && identifier.getNumber().equals(AuthHelper.VALID_NUMBER_TWO)));
-    verify(rateLimiters, times(1)).getProfileLimiter();
+    verifyNoMoreInteractions(usernamesManager);
     verify(rateLimiter, times(1)).validate(eq(AuthHelper.VALID_NUMBER));
+    reset(rateLimiter);
+  }
+
+  @Test
+  public void testProfileGetByUsername() throws RateLimitExceededException {
+    Profile profile= resources.getJerseyTest()
+                              .target("/v1/profile/username/n00bkiller")
+                              .request()
+                              .header("Authorization", AuthHelper.getAuthHeader(AuthHelper.VALID_NUMBER, AuthHelper.VALID_PASSWORD))
+                              .get(Profile.class);
+
+    assertThat(profile.getIdentityKey()).isEqualTo("bar");
+    assertThat(profile.getName()).isEqualTo("baz");
+    assertThat(profile.getAvatar()).isEqualTo("profiles/bang");
+    assertThat(profile.getUsername()).isEqualTo("n00bkiller");
+    assertThat(profile.getUuid()).isEqualTo(AuthHelper.VALID_UUID_TWO);
+
+    verify(accountsManager, times(1)).get(eq(AuthHelper.VALID_UUID_TWO));
+    verify(usernamesManager, times(1)).get(eq("n00bkiller"));
+    verify(usernameRateLimiter, times(1)).validate(eq(AuthHelper.VALID_UUID.toString()));
   }
 
   @Test
@@ -110,6 +160,33 @@ public class ProfileControllerTest {
     assertThat(response.getStatus()).isEqualTo(401);
   }
 
+  @Test
+  public void testProfileGetByUsernameUnauthorized() throws Exception {
+    Response response = resources.getJerseyTest()
+                                 .target("/v1/profile/username/n00bkiller")
+                                 .request()
+                                 .get();
+
+    assertThat(response.getStatus()).isEqualTo(401);
+  }
+
+
+  @Test
+  public void testProfileGetByUsernameNotFound() throws RateLimitExceededException {
+    Response response = resources.getJerseyTest()
+                              .target("/v1/profile/username/n00bkillerzzzzz")
+                              .request()
+                              .header("Authorization", AuthHelper.getAuthHeader(AuthHelper.VALID_NUMBER, AuthHelper.VALID_PASSWORD))
+                              .get();
+
+    assertThat(response.getStatus()).isEqualTo(404);
+
+    verify(usernamesManager, times(1)).get(eq("n00bkillerzzzzz"));
+    verify(usernameRateLimiter, times(1)).validate(eq(AuthHelper.VALID_UUID.toString()));
+    reset(usernameRateLimiter);
+  }
+
+
   @Test
   public void testProfileGetDisabled() throws Exception {
     Response response = resources.getJerseyTest()

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

@@ -0,0 +1,191 @@
+package org.whispersystems.textsecuregcm.tests.storage;
+
+import org.junit.Test;
+import org.whispersystems.textsecuregcm.redis.ReplicatedJedisPool;
+import org.whispersystems.textsecuregcm.storage.Account;
+import org.whispersystems.textsecuregcm.storage.Accounts;
+import org.whispersystems.textsecuregcm.storage.AccountsManager;
+import org.whispersystems.textsecuregcm.storage.DirectoryManager;
+import org.whispersystems.textsecuregcm.storage.Usernames;
+import org.whispersystems.textsecuregcm.storage.UsernamesManager;
+
+import java.util.HashSet;
+import java.util.Optional;
+import java.util.UUID;
+
+import static junit.framework.TestCase.assertSame;
+import static junit.framework.TestCase.assertTrue;
+import static org.junit.Assert.assertEquals;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.*;
+import redis.clients.jedis.Jedis;
+import redis.clients.jedis.exceptions.JedisException;
+
+public class UsernamesManagerTest {
+
+  @Test
+  public void testGetByUsernameInCache() {
+    ReplicatedJedisPool cacheClient = mock(ReplicatedJedisPool.class);
+    Jedis               jedis       = mock(Jedis.class              );
+    Usernames           usernames   = mock(Usernames.class          );
+
+    UUID uuid = UUID.randomUUID();
+
+    when(cacheClient.getReadResource()).thenReturn(jedis);
+    when(jedis.get(eq("UsernameByUsername::n00bkiller"))).thenReturn(uuid.toString());
+
+    UsernamesManager usernamesManager = new UsernamesManager(usernames, cacheClient);
+    Optional<UUID>   retrieved        = usernamesManager.get("n00bkiller");
+
+    assertTrue(retrieved.isPresent());
+    assertEquals(retrieved.get(), uuid);
+
+    verify(jedis, times(1)).get(eq("UsernameByUsername::n00bkiller"));
+    verify(jedis, times(1)).close();
+    verifyNoMoreInteractions(jedis);
+    verifyNoMoreInteractions(usernames);
+  }
+
+  @Test
+  public void testGetByUuidInCache() {
+    ReplicatedJedisPool cacheClient = mock(ReplicatedJedisPool.class);
+    Jedis               jedis       = mock(Jedis.class              );
+    Usernames           usernames   = mock(Usernames.class          );
+
+    UUID uuid = UUID.randomUUID();
+
+    when(cacheClient.getReadResource()).thenReturn(jedis);
+    when(jedis.get(eq("UsernameByUuid::" + uuid.toString()))).thenReturn("n00bkiller");
+
+    UsernamesManager usernamesManager = new UsernamesManager(usernames, cacheClient);
+    Optional<String> retrieved        = usernamesManager.get(uuid);
+
+    assertTrue(retrieved.isPresent());
+    assertEquals(retrieved.get(), "n00bkiller");
+
+    verify(jedis, times(1)).get(eq("UsernameByUuid::" + uuid.toString()));
+    verify(jedis, times(1)).close();
+    verifyNoMoreInteractions(jedis);
+    verifyNoMoreInteractions(usernames);
+  }
+
+
+  @Test
+  public void testGetByUsernameNotInCache() {
+    ReplicatedJedisPool cacheClient = mock(ReplicatedJedisPool.class);
+    Jedis               jedis       = mock(Jedis.class              );
+    Usernames           usernames   = mock(Usernames.class          );
+
+    UUID                uuid             = UUID.randomUUID();
+
+
+    when(cacheClient.getReadResource()).thenReturn(jedis);
+    when(cacheClient.getWriteResource()).thenReturn(jedis);
+    when(jedis.get(eq("UsernameByUsername::n00bkiller"))).thenReturn(null);
+    when(usernames.get(eq("n00bkiller"))).thenReturn(Optional.of(uuid));
+
+    UsernamesManager usernamesManager = new UsernamesManager(usernames, cacheClient);
+    Optional<UUID>   retrieved        = usernamesManager.get("n00bkiller");
+
+    assertTrue(retrieved.isPresent());
+    assertSame(retrieved.get(), uuid);
+
+    verify(jedis, times(1)).get(eq("UsernameByUsername::n00bkiller"));
+    verify(jedis, times(1)).set(eq("UsernameByUsername::n00bkiller"), eq(uuid.toString()));
+    verify(jedis, times(1)).set(eq("UsernameByUuid::" + uuid.toString()), eq("n00bkiller"));
+    verify(jedis, times(2)).close();
+    verifyNoMoreInteractions(jedis);
+
+    verify(usernames, times(1)).get(eq("n00bkiller"));
+    verifyNoMoreInteractions(usernames);
+  }
+
+  @Test
+  public void testGetByUuidNotInCache() {
+    ReplicatedJedisPool cacheClient = mock(ReplicatedJedisPool.class);
+    Jedis               jedis       = mock(Jedis.class              );
+    Usernames           usernames   = mock(Usernames.class          );
+
+    UUID uuid = UUID.randomUUID();
+
+    when(cacheClient.getReadResource()).thenReturn(jedis);
+    when(cacheClient.getWriteResource()).thenReturn(jedis);
+    when(jedis.get(eq("UsernameByUuid::" + uuid.toString()))).thenReturn(null);
+    when(usernames.get(eq(uuid))).thenReturn(Optional.of("n00bkiller"));
+
+    UsernamesManager usernamesManager = new UsernamesManager(usernames, cacheClient);
+    Optional<String> retrieved        = usernamesManager.get(uuid);
+
+    assertTrue(retrieved.isPresent());
+    assertEquals(retrieved.get(), "n00bkiller");
+
+    verify(jedis, times(1)).get(eq("UsernameByUuid::" + uuid));
+    verify(jedis, times(1)).set(eq("UsernameByUuid::" + uuid), eq("n00bkiller"));
+    verify(jedis, times(1)).set(eq("UsernameByUsername::n00bkiller"), eq(uuid.toString()));
+    verify(jedis, times(2)).close();
+    verifyNoMoreInteractions(jedis);
+
+    verify(usernames, times(1)).get(eq(uuid));
+    verifyNoMoreInteractions(usernames);
+  }
+
+  @Test
+  public void testGetByUsernameBrokenCache() {
+    ReplicatedJedisPool cacheClient = mock(ReplicatedJedisPool.class);
+    Jedis               jedis       = mock(Jedis.class              );
+    Usernames           usernames   = mock(Usernames.class          );
+
+    UUID                uuid        = UUID.randomUUID();
+
+    when(cacheClient.getReadResource()).thenReturn(jedis);
+    when(cacheClient.getWriteResource()).thenReturn(jedis);
+    when(jedis.get(eq("UsernameByUsername::n00bkiller"))).thenThrow(new JedisException("Connection lost!"));
+    when(usernames.get(eq("n00bkiller"))).thenReturn(Optional.of(uuid));
+
+    UsernamesManager usernamesManager = new UsernamesManager(usernames, cacheClient);
+    Optional<UUID>   retrieved        = usernamesManager.get("n00bkiller");
+
+    assertTrue(retrieved.isPresent());
+    assertEquals(retrieved.get(), uuid);
+
+    verify(jedis, times(1)).get(eq("UsernameByUsername::n00bkiller"));
+    verify(jedis, times(1)).set(eq("UsernameByUsername::n00bkiller"), eq(uuid.toString()));
+    verify(jedis, times(1)).set(eq("UsernameByUuid::" + uuid.toString()), eq("n00bkiller"));
+    verify(jedis, times(2)).close();
+    verifyNoMoreInteractions(jedis);
+
+    verify(usernames, times(1)).get(eq("n00bkiller"));
+    verifyNoMoreInteractions(usernames);
+  }
+
+  @Test
+  public void testGetAccountByUuidBrokenCache() {
+    ReplicatedJedisPool cacheClient = mock(ReplicatedJedisPool.class);
+    Jedis               jedis       = mock(Jedis.class              );
+    Usernames           usernames   = mock(Usernames.class          );
+
+    UUID                uuid        = UUID.randomUUID();
+
+    when(cacheClient.getReadResource()).thenReturn(jedis);
+    when(cacheClient.getWriteResource()).thenReturn(jedis);
+    when(jedis.get(eq("UsernameByUuid::" + uuid))).thenThrow(new JedisException("Connection lost!"));
+    when(usernames.get(eq(uuid))).thenReturn(Optional.of("n00bkiller"));
+
+    UsernamesManager usernamesManager = new UsernamesManager(usernames, cacheClient);
+    Optional<String>   retrieved        = usernamesManager.get(uuid);
+
+    assertTrue(retrieved.isPresent());
+    assertEquals(retrieved.get(), "n00bkiller");
+
+    verify(jedis, times(1)).get(eq("UsernameByUuid::" + uuid));
+    verify(jedis, times(1)).set(eq("UsernameByUsername::n00bkiller"), eq(uuid.toString()));
+    verify(jedis, times(1)).set(eq("UsernameByUuid::" + uuid.toString()), eq("n00bkiller"));
+    verify(jedis, times(2)).close();
+    verifyNoMoreInteractions(jedis);
+
+    verify(usernames, times(1)).get(eq(uuid));
+    verifyNoMoreInteractions(usernames);
+  }
+
+}

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

@@ -0,0 +1,164 @@
+package org.whispersystems.textsecuregcm.tests.storage;
+
+import com.opentable.db.postgres.embedded.LiquibasePreparer;
+import com.opentable.db.postgres.junit.EmbeddedPostgresRules;
+import com.opentable.db.postgres.junit.PreparedDbRule;
+import org.jdbi.v3.core.Jdbi;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.whispersystems.textsecuregcm.configuration.CircuitBreakerConfiguration;
+import org.whispersystems.textsecuregcm.storage.FaultTolerantDatabase;
+import org.whispersystems.textsecuregcm.storage.Usernames;
+
+import java.io.IOException;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.Optional;
+import java.util.UUID;
+
+import static junit.framework.TestCase.assertTrue;
+import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
+import static org.junit.Assert.assertFalse;
+
+public class UsernamesTest {
+
+  @Rule
+  public PreparedDbRule db = EmbeddedPostgresRules.preparedDatabase(LiquibasePreparer.forClasspathLocation("accountsdb.xml"));
+
+  private Usernames usernames;
+
+  @Before
+  public void setupAccountsDao() {
+    FaultTolerantDatabase faultTolerantDatabase = new FaultTolerantDatabase("usernamesTest",
+                                                                            Jdbi.create(db.getTestDatabase()),
+                                                                            new CircuitBreakerConfiguration());
+
+    this.usernames = new Usernames(faultTolerantDatabase);
+  }
+
+  @Test
+  public void testPut() throws SQLException, IOException {
+    UUID   uuid     = UUID.randomUUID();
+    String username = "myusername";
+
+    assertTrue(usernames.put(uuid, username));
+
+    PreparedStatement statement = db.getTestDatabase().getConnection().prepareStatement("SELECT * FROM usernames WHERE uuid = ?");
+    verifyStoredState(statement, uuid, username);
+  }
+
+  @Test
+  public void testPutChange() throws SQLException, IOException {
+    UUID uuid = UUID.randomUUID();
+    String firstUsername = "myfirstusername";
+    String secondUsername = "mysecondusername";
+
+    assertTrue(usernames.put(uuid, firstUsername));
+
+    PreparedStatement statement = db.getTestDatabase().getConnection().prepareStatement("SELECT * FROM usernames WHERE uuid = ?");
+    verifyStoredState(statement, uuid, firstUsername);
+
+    assertTrue(usernames.put(uuid, secondUsername));
+
+    verifyStoredState(statement, uuid, secondUsername);
+  }
+
+  @Test
+  public void testPutConflict() throws SQLException {
+    UUID firstUuid = UUID.randomUUID();
+    UUID secondUuid = UUID.randomUUID();
+
+    String username = "myfirstusername";
+
+    assertTrue(usernames.put(firstUuid, username));
+    assertFalse(usernames.put(secondUuid, username));
+
+    PreparedStatement statement = db.getTestDatabase().getConnection().prepareStatement("SELECT * FROM usernames WHERE username = ?");
+    statement.setString(1, username);
+
+    ResultSet resultSet = statement.executeQuery();
+
+    assertTrue(resultSet.next());
+    assertThat(resultSet.getString("uuid")).isEqualTo(firstUuid.toString());
+    assertThat(resultSet.next()).isFalse();
+  }
+
+  @Test
+  public void testGetByUuid() {
+    UUID   uuid     = UUID.randomUUID();
+    String username = "myusername";
+
+    assertTrue(usernames.put(uuid, username));
+
+    Optional<String> retrieved = usernames.get(uuid);
+
+    assertTrue(retrieved.isPresent());
+    assertThat(retrieved.get()).isEqualTo(username);
+  }
+
+  @Test
+  public void testGetByUuidMissing() {
+    Optional<String> retrieved = usernames.get(UUID.randomUUID());
+    assertFalse(retrieved.isPresent());
+  }
+
+  @Test
+  public void testGetByUsername() {
+    UUID   uuid     = UUID.randomUUID();
+    String username = "myusername";
+
+    assertTrue(usernames.put(uuid, username));
+
+    Optional<UUID> retrieved = usernames.get(username);
+
+    assertTrue(retrieved.isPresent());
+    assertThat(retrieved.get()).isEqualTo(uuid);
+  }
+
+  @Test
+  public void testGetByUsernameMissing() {
+    Optional<UUID> retrieved = usernames.get("myusername");
+
+    assertFalse(retrieved.isPresent());
+  }
+
+
+  @Test
+  public void testDelete() {
+    UUID   uuid     = UUID.randomUUID();
+    String username = "myusername";
+
+    assertTrue(usernames.put(uuid, username));
+
+    Optional<UUID> retrieved = usernames.get(username);
+
+    assertTrue(retrieved.isPresent());
+    assertThat(retrieved.get()).isEqualTo(uuid);
+
+    usernames.delete(uuid);
+
+    assertThat(usernames.get(uuid).isPresent()).isFalse();
+  }
+
+  private void verifyStoredState(PreparedStatement statement, UUID uuid, String expectedUsername)
+      throws SQLException, IOException
+  {
+    statement.setObject(1, uuid);
+
+    ResultSet resultSet = statement.executeQuery();
+
+    if (resultSet.next()) {
+      String data = resultSet.getString("username");
+      assertThat(data).isNotEmpty();
+      assertThat(data).isEqualTo(expectedUsername);
+    } else {
+      throw new AssertionError("No data");
+    }
+
+    assertThat(resultSet.next()).isFalse();
+  }
+
+
+}