Handle split contacts in storage service when in PNP mode.

app/src/main/java/org/thoughtcrime/securesms/database/RecipientTable.kt

@@ -2236,6 +2236,7 @@ open class RecipientTable(context: Context, databaseHelper: SignalDatabase) : Da
 
     if (update(id, contentValues)) {
       Log.i(TAG, "[WithSplit] Newly marked $id as unregistered.")
+      markNeedsSync(id)
       ApplicationDependencies.getDatabaseObserver().notifyRecipientChanged(id)
     }
 
@@ -2254,10 +2255,33 @@ open class RecipientTable(context: Context, databaseHelper: SignalDatabase) : Da
 
     if (update(id, contentValues)) {
       Log.i(TAG, "[WithoutSplit] Newly marked $id as unregistered.")
+      markNeedsSync(id)
       ApplicationDependencies.getDatabaseObserver().notifyRecipientChanged(id)
     }
   }
 
+  /**
+   * Removes the target recipient's E164+PNI, then creates a new recipient with that E164+PNI.
+   * Done so we can match a split contact during storage sync.
+   */
+  fun splitForStorageSync(storageId: ByteArray) {
+    check(FeatureFlags.phoneNumberPrivacy())
+
+    val record = getByStorageId(storageId)!!
+    check(record.serviceId != null && record.pni != null && record.serviceId != record.pni)
+
+    writableDatabase
+      .update(TABLE_NAME)
+      .values(
+        PNI_COLUMN to null,
+        PHONE to null
+      )
+      .where("$ID = ?", record.id)
+      .run()
+
+    getAndPossiblyMerge(record.pni, record.pni, record.e164)
+  }
+
   fun bulkUpdatedRegisteredStatus(registered: Map<RecipientId, ServiceId?>, unregistered: Collection<RecipientId>) {
     writableDatabase.withinTransaction {
       val registeredWithServiceId: Set<RecipientId> = getRegisteredWithServiceIds()

app/src/main/java/org/thoughtcrime/securesms/storage/ContactRecordProcessor.java

@@ -19,9 +19,15 @@ import org.whispersystems.signalservice.api.storage.SignalContactRecord;
 import org.whispersystems.signalservice.api.util.OptionalUtil;
 import org.whispersystems.signalservice.internal.storage.protos.ContactRecord.IdentityState;
 
+import java.io.IOException;
+import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.TreeMap;
+import java.util.TreeSet;
 
 public class ContactRecordProcessor extends DefaultStorageRecordProcessor<SignalContactRecord> {
 
@@ -47,6 +53,69 @@ public class ContactRecordProcessor extends DefaultStorageRecordProcessor<Signal
     this.selfE164       = selfE164;
   }
 
+  /**
+   * For contact records specifically, we have some extra work that needs to be done before we process all of the records.
+   *
+   * We have to look and see if there is an unregistered ACI-only record and another E164/PNI-only record that points to the
+   * same local contact row.
+   *
+   * If so, we actually want to mimic the split and turn them into two separate contact rows locally. The reasons are nuanced,
+   * but the TL;DR is that we want to split unregistered users into separate rows so that a user could re-register and get a
+   * different ACI.
+   */
+  @Override
+  public void process(@NonNull Collection<SignalContactRecord> remoteRecords, @NonNull StorageKeyGenerator keyGenerator) throws IOException {
+    if (!FeatureFlags.phoneNumberPrivacy()) {
+      super.process(remoteRecords, keyGenerator);
+      return;
+    }
+
+    List<SignalContactRecord> unregisteredAciOnly = new ArrayList<>();
+    List<SignalContactRecord> pniE164Only         = new ArrayList<>();
+
+    for (SignalContactRecord remoteRecord : remoteRecords) {
+      if (isInvalid(remoteRecord)) {
+        continue;
+      }
+
+      if (remoteRecord.getUnregisteredTimestamp() > 0 && remoteRecord.getServiceId() != null && !remoteRecord.getPni().isPresent() && !remoteRecord.getNumber().isPresent()) {
+        unregisteredAciOnly.add(remoteRecord);
+      } else if (remoteRecord.getServiceId() != null && remoteRecord.getServiceId().equals(remoteRecord.getPni().orElse(null))) {
+        pniE164Only.add(remoteRecord);
+      }
+    }
+
+    if (unregisteredAciOnly.isEmpty() || pniE164Only.isEmpty()) {
+      super.process(remoteRecords, keyGenerator);
+      return;
+    }
+
+    Log.i(TAG, "We have some unregistered ACI-only contacts as well as some PNI-only contacts. Need to do an intersection to detect any possible required splits.");
+
+    TreeSet<SignalContactRecord> localMatches = new TreeSet<>(this);
+
+    for (SignalContactRecord aciOnly : unregisteredAciOnly) {
+      Optional<SignalContactRecord> localMatch = getMatching(aciOnly, keyGenerator);
+
+      if (localMatch.isPresent()) {
+        localMatches.add(localMatch.get());
+      }
+    }
+
+    for (SignalContactRecord pniOnly : pniE164Only) {
+      Optional<SignalContactRecord> localMatch = getMatching(pniOnly, keyGenerator);
+
+      if (localMatch.isPresent() && localMatches.contains(localMatch.get())) {
+        Log.w(TAG, "Found a situation where we need to split our local record in two in order to match the remote state.");
+
+        SignalDatabase.recipients().splitForStorageSync(localMatch.get().getId().getRaw());
+      }
+    }
+
+
+    super.process(remoteRecords, keyGenerator);
+  }
+
   /**
    * Error cases:
    * - You can't have a contact record without an address component.