CompletionStage<Void> reclaimAccount(final Account existingAccount,
      final Account accountToCreate,
      final Collection<TransactWriteItem> additionalWriteItems) {

    if (!existingAccount.getUuid().equals(accountToCreate.getUuid()) ||
        !existingAccount.getNumber().equals(accountToCreate.getNumber())) {

      throw new IllegalArgumentException("reclaimed accounts must match");
    }

    return AsyncTimerUtil.record(RECLAIM_TIMER, () -> {

      accountToCreate.setVersion(existingAccount.getVersion());

      // Carry over the old backup id commitment. If the new account claimer cannot does not have the secret used to
      // generate their backup-id, this credential is useless, however if they can produce the same credential they
      // won't be rate-limited for setting their backup-id.
      accountToCreate.setBackupCredentialRequest(existingAccount.getBackupCredentialRequest());

      // Carry over the old SVR3 share-set. This is required for an account to restore information from SVR. The share-
      // set is not a secret, if the new account claimer does not have the SVR3 pin, it is useless.
      accountToCreate.setSvr3ShareSet(existingAccount.getSvr3ShareSet());

      final List<TransactWriteItem> writeItems = new ArrayList<>();

      // If we're reclaiming an account that already has a username, we'd like to give the re-registering client
      // an opportunity to reclaim their original username and link. We do this by:
      //   1. marking the usernameHash as reserved for the aci
      //   2. saving the username link id, but not the encrypted username. The link will be broken until the client
      //      reclaims their username
      //
      // If we partially reclaim the account but fail (for example, we update the account but the client goes away
      // before creation is finished), we might be reclaiming the account we already reclaimed. In that case, we
      // should copy over the reserved username and link verbatim
      if (existingAccount.getReservedUsernameHash().isPresent() &&
          existingAccount.getUsernameLinkHandle() != null &&
          existingAccount.getUsernameHash().isEmpty() &&
          existingAccount.getEncryptedUsername().isEmpty()) {
        // reclaiming a partially reclaimed account
        accountToCreate.setReservedUsernameHash(existingAccount.getReservedUsernameHash().get());
        accountToCreate.setUsernameLinkHandle(existingAccount.getUsernameLinkHandle());
      } else if (existingAccount.getUsernameHash().isPresent()) {
        // reclaiming an account with a username
        final byte[] usernameHash = existingAccount.getUsernameHash().get();
        final long expirationTime = clock.instant().plus(USERNAME_RECLAIM_TTL).getEpochSecond();
        accountToCreate.setReservedUsernameHash(usernameHash);
        accountToCreate.setUsernameLinkHandle(existingAccount.getUsernameLinkHandle());

        writeItems.add(TransactWriteItem.builder()
            .put(Put.builder()
                .tableName(usernamesConstraintTableName)
                .item(Map.of(
                    UsernameTable.KEY_USERNAME_HASH, AttributeValues.fromByteArray(usernameHash),
                    UsernameTable.ATTR_ACCOUNT_UUID, AttributeValues.fromUUID(accountToCreate.getUuid()),
                    UsernameTable.ATTR_TTL, AttributeValues.fromLong(expirationTime),
                    UsernameTable.ATTR_CONFIRMED, AttributeValues.fromBool(false),
                    UsernameTable.ATTR_RECLAIMABLE, AttributeValues.fromBool(true)))
                .conditionExpression("attribute_not_exists(#username_hash) OR (#ttl < :now) OR #uuid = :uuid")
                .expressionAttributeNames(Map.of(
                    "#username_hash", UsernameTable.KEY_USERNAME_HASH,
                    "#ttl", UsernameTable.ATTR_TTL,
                    "#uuid", UsernameTable.ATTR_ACCOUNT_UUID))
                .expressionAttributeValues(Map.of(
                    ":now", AttributeValues.fromLong(clock.instant().getEpochSecond()),
                    ":uuid", AttributeValues.fromUUID(accountToCreate.getUuid())))
                .returnValuesOnConditionCheckFailure(ReturnValuesOnConditionCheckFailure.ALL_OLD)
                .build())
            .build());
      }
      writeItems.add(UpdateAccountSpec.forAccount(accountsTableName, accountToCreate).transactItem());
      writeItems.addAll(additionalWriteItems);

      return asyncClient.transactWriteItems(TransactWriteItemsRequest.builder().transactItems(writeItems).build())
          .thenApply(response -> {
            accountToCreate.setVersion(accountToCreate.getVersion() + 1);
            return (Void) null;
          })
          .exceptionally(throwable -> {
            final Throwable unwrapped = ExceptionUtils.unwrap(throwable);
            if (unwrapped instanceof TransactionCanceledException te) {
              if (te.cancellationReasons().stream().anyMatch(Accounts::conditionalCheckFailed)) {
                throw new ContestedOptimisticLockException();
              }
            }
            // rethrow
            throw CompletableFutureUtils.errorAsCompletionException(throwable);
          });
    });
  }
@Test
  void testReclaimAccount() {
    final String e164 = "+14151112222";
    final Device device = generateDevice(DEVICE_ID_1);
    final UUID existingUuid = UUID.randomUUID();
    final UUID existingPni = UUID.randomUUID();
    final Account existingAccount = generateAccount(e164, existingUuid, existingPni, List.of(device));

    createAccount(existingAccount);

    final byte[] usernameHash = randomBytes(32);
    final byte[] encryptedUsername = randomBytes(16);

    // Set up the existing account to have a username hash
    accounts.confirmUsernameHash(existingAccount, usernameHash, encryptedUsername).join();
    final UUID usernameLinkHandle = existingAccount.getUsernameLinkHandle();

    verifyStoredState(e164, existingAccount.getUuid(), existingAccount.getPhoneNumberIdentifier(), usernameHash, existingAccount, true);

    assertPhoneNumberConstraintExists(e164, existingUuid);
    assertPhoneNumberIdentifierConstraintExists(existingPni, existingUuid);

    assertDoesNotThrow(() -> accounts.update(existingAccount));

    final UUID secondUuid = UUID.randomUUID();

    final Device secondDevice = generateDevice(DEVICE_ID_1);
    final Account secondAccount = generateAccount(e164, secondUuid, UUID.randomUUID(), List.of(secondDevice));

    reclaimAccount(secondAccount);

    // usernameHash should be unset
    verifyStoredState("+14151112222", existingUuid, existingPni, null, secondAccount, true);

    // username should become 'reclaimable'
    Map<String, AttributeValue> item = readAccount(existingUuid);
    Account result = Accounts.fromItem(item);
    assertThat(AttributeValues.getUUID(item, Accounts.ATTR_USERNAME_LINK_UUID, null))
        .isEqualTo(usernameLinkHandle)
        .isEqualTo(result.getUsernameLinkHandle());
    assertThat(result.getUsernameHash()).isEmpty();
    assertThat(result.getEncryptedUsername()).isEmpty();
    assertArrayEquals(result.getReservedUsernameHash().get(), usernameHash);

    // should keep the same usernameLink, now encryptedUsername should be set
    accounts.confirmUsernameHash(result, usernameHash, encryptedUsername).join();
    item = readAccount(existingUuid);
    result = Accounts.fromItem(item);
    assertThat(AttributeValues.getUUID(item, Accounts.ATTR_USERNAME_LINK_UUID, null))
        .isEqualTo(usernameLinkHandle)
        .isEqualTo(result.getUsernameLinkHandle());
    assertArrayEquals(result.getEncryptedUsername().get(), encryptedUsername);
    assertArrayEquals(result.getUsernameHash().get(), usernameHash);
    assertThat(result.getReservedUsernameHash()).isEmpty();

    assertPhoneNumberConstraintExists("+14151112222", existingUuid);
    assertPhoneNumberIdentifierConstraintExists(existingPni, existingUuid);

    Account invalidAccount = generateAccount("+14151113333", existingUuid, UUID.randomUUID(), List.of(generateDevice(DEVICE_ID_1)));

    assertThatThrownBy(() -> createAccount(invalidAccount));
  }