Reactively share profiles to those who should already have it.

app/src/main/java/org/thoughtcrime/securesms/conversation/ConversationActivity.java

@@ -567,7 +567,7 @@ public class ConversationActivity extends PassphraseRequiredActivity
 
       ApplicationDependencies.getJobManager()
                              .startChain(new RequestGroupV2InfoJob(groupId))
-                             .then(new GroupV2UpdateSelfProfileKeyJob(groupId))
+                             .then(GroupV2UpdateSelfProfileKeyJob.withoutLimits(groupId))
                              .enqueue();
 
       if (viewModel.getArgs().isFirstTimeInSelfCreatedGroup()) {

app/src/main/java/org/thoughtcrime/securesms/jobs/GroupV2UpdateSelfProfileKeyJob.java

@@ -11,6 +11,7 @@ import org.thoughtcrime.securesms.groups.GroupManager;
 import org.thoughtcrime.securesms.groups.GroupNotAMemberException;
 import org.thoughtcrime.securesms.jobmanager.Data;
 import org.thoughtcrime.securesms.jobmanager.Job;
+import org.thoughtcrime.securesms.jobmanager.impl.DecryptionsDrainedConstraint;
 import org.thoughtcrime.securesms.jobmanager.impl.NetworkConstraint;
 import org.whispersystems.signalservice.api.groupsv2.NoCredentialForRedemptionTimeException;
 import org.whispersystems.signalservice.api.push.exceptions.PushNetworkException;
@@ -36,8 +37,11 @@ public final class GroupV2UpdateSelfProfileKeyJob extends BaseJob {
 
   private final GroupId.V2 groupId;
 
-  public GroupV2UpdateSelfProfileKeyJob(@NonNull GroupId.V2 groupId) {
+  /**
-    this(new Parameters.Builder()
+   * Job will run regardless of how many times you enqueue it.
+   */
+  public static @NonNull GroupV2UpdateSelfProfileKeyJob withoutLimits(@NonNull GroupId.V2 groupId) {
+    return new GroupV2UpdateSelfProfileKeyJob(new Parameters.Builder()
                                                             .addConstraint(NetworkConstraint.KEY)
                                                             .setLifespan(TimeUnit.DAYS.toMillis(1))
                                                             .setMaxAttempts(Parameters.UNLIMITED)
@@ -46,6 +50,22 @@ public final class GroupV2UpdateSelfProfileKeyJob extends BaseJob {
                                               groupId);
   }
 
+  /**
+   * Only one instance will be enqueued per group, and it won't run until after decryptions are
+   * drained.
+   */
+  public static @NonNull GroupV2UpdateSelfProfileKeyJob withQueueLimits(@NonNull GroupId.V2 groupId) {
+    return new GroupV2UpdateSelfProfileKeyJob(new Parameters.Builder()
+                                                            .addConstraint(NetworkConstraint.KEY)
+                                                            .addConstraint(DecryptionsDrainedConstraint.KEY)
+                                                            .setLifespan(TimeUnit.DAYS.toMillis(1))
+                                                            .setMaxAttempts(Parameters.UNLIMITED)
+                                                            .setQueue(QUEUE + "_" + groupId.toString())
+                                                            .setMaxInstancesForQueue(1)
+                                                            .build(),
+                                              groupId);
+  }
+
   private GroupV2UpdateSelfProfileKeyJob(@NonNull Parameters parameters, @NonNull GroupId.V2 groupId) {
     super(parameters);
     this.groupId = groupId;

app/src/main/java/org/thoughtcrime/securesms/jobs/ProfileKeySendJob.java

@@ -13,6 +13,8 @@ import org.thoughtcrime.securesms.database.DatabaseFactory;
 import org.thoughtcrime.securesms.dependencies.ApplicationDependencies;
 import org.thoughtcrime.securesms.jobmanager.Data;
 import org.thoughtcrime.securesms.jobmanager.Job;
+import org.thoughtcrime.securesms.jobmanager.impl.DecryptionsDrainedConstraint;
+import org.thoughtcrime.securesms.jobmanager.impl.NetworkConstraint;
 import org.thoughtcrime.securesms.recipients.Recipient;
 import org.thoughtcrime.securesms.recipients.RecipientId;
 import org.thoughtcrime.securesms.recipients.RecipientUtil;
@@ -44,9 +46,12 @@ public class ProfileKeySendJob extends BaseJob {
 
   /**
    * Suitable for a 1:1 conversation or a GV1 group only.
+   *
+   * @param queueLimits True if you only want one of these to be run per person after decryptions
+   *                    are drained, otherwise false.
    */
   @WorkerThread
-  public static ProfileKeySendJob create(@NonNull Context context, long threadId) {
+  public static ProfileKeySendJob create(@NonNull Context context, long threadId, boolean queueLimits) {
     Recipient conversationRecipient = DatabaseFactory.getThreadDatabase(context).getRecipientForThreadId(threadId);
 
     if (conversationRecipient == null) {
@@ -62,11 +67,22 @@ public class ProfileKeySendJob extends BaseJob {
 
     recipients.remove(Recipient.self().getId());
 
+    if (queueLimits) {
       return new ProfileKeySendJob(new Parameters.Builder()
                                                  .setQueue(conversationRecipient.getId().toQueueKey())
+                                                 .addConstraint(NetworkConstraint.KEY)
                                                  .setLifespan(TimeUnit.DAYS.toMillis(1))
                                                  .setMaxAttempts(Parameters.UNLIMITED)
                                                  .build(), threadId, recipients);
+    } else {
+      return new ProfileKeySendJob(new Parameters.Builder()
+                                                 .setQueue("ProfileKeySendJob_" + conversationRecipient.getId().toQueueKey())
+                                                 .addConstraint(NetworkConstraint.KEY)
+                                                 .addConstraint(DecryptionsDrainedConstraint.KEY)
+                                                 .setLifespan(TimeUnit.DAYS.toMillis(1))
+                                                 .setMaxAttempts(Parameters.UNLIMITED)
+                                                 .build(), threadId, recipients);
+    }
   }
 
   private ProfileKeySendJob(@NonNull Parameters parameters, long threadId, @NonNull List<RecipientId> recipients) {

app/src/main/java/org/thoughtcrime/securesms/jobs/RefreshAttributesJob.java

@@ -27,21 +27,37 @@ public class RefreshAttributesJob extends BaseJob {
 
   private static final String TAG = Log.tag(RefreshAttributesJob.class);
 
+  private static final String KEY_FORCED = "forced";
+
+  private static volatile boolean hasRefreshedThisAppCycle;
+
+  private final boolean forced;
+
   public RefreshAttributesJob() {
+    this(true);
+  }
+
+  /**
+   * @param forced True if you want this job to run no matter what. False if you only want this job
+   *               to run if it hasn't run yet this app cycle.
+   */
+  public RefreshAttributesJob(boolean forced) {
     this(new Job.Parameters.Builder()
                            .addConstraint(NetworkConstraint.KEY)
                            .setQueue("RefreshAttributesJob")
                            .setMaxInstancesForFactory(2)
-                           .build());
+                           .build(),
+         forced);
   }
 
-  private RefreshAttributesJob(@NonNull Job.Parameters parameters) {
+  private RefreshAttributesJob(@NonNull Job.Parameters parameters, boolean forced) {
     super(parameters);
+    this.forced = forced;
   }
 
   @Override
   public @NonNull Data serialize() {
-    return Data.EMPTY;
+    return new Data.Builder().putBoolean(KEY_FORCED, forced).build();
   }
 
   @Override
@@ -56,6 +72,11 @@ public class RefreshAttributesJob extends BaseJob {
       return;
     }
 
+    if (!forced && hasRefreshedThisAppCycle) {
+      Log.d(TAG, "Already refreshed this app cycle. Skipping.");
+      return;
+    }
+
     int       registrationId              = TextSecurePreferences.getLocalRegistrationId(context);
     boolean   fetchesMessages             = TextSecurePreferences.isFcmDisabled(context);
     byte[]    unidentifiedAccessKey       = UnidentifiedAccess.deriveAccessKeyFrom(ProfileKeyUtil.getSelfProfileKey());
@@ -90,6 +111,8 @@ public class RefreshAttributesJob extends BaseJob {
                                               phoneNumberDiscoverable);
 
     ApplicationDependencies.getJobManager().add(new RefreshOwnProfileJob());
+
+    hasRefreshedThisAppCycle = true;
   }
 
   @Override
@@ -105,7 +128,7 @@ public class RefreshAttributesJob extends BaseJob {
   public static class Factory implements Job.Factory<RefreshAttributesJob> {
     @Override
     public @NonNull RefreshAttributesJob create(@NonNull Parameters parameters, @NonNull org.thoughtcrime.securesms.jobmanager.Data data) {
-      return new RefreshAttributesJob(parameters);
+      return new RefreshAttributesJob(parameters, data.getBooleanOrDefault(KEY_FORCED, true));
     }
   }
 }

app/src/main/java/org/thoughtcrime/securesms/jobs/RotateProfileKeyJob.java

@@ -56,7 +56,7 @@ public class RotateProfileKeyJob extends BaseJob {
     List<GroupId.V2> allGv2Groups = DatabaseFactory.getGroupDatabase(context).getAllGroupV2Ids();
 
     for (GroupId.V2 groupId : allGv2Groups) {
-      ApplicationDependencies.getJobManager().add(new GroupV2UpdateSelfProfileKeyJob(groupId));
+      ApplicationDependencies.getJobManager().add(GroupV2UpdateSelfProfileKeyJob.withoutLimits(groupId));
     }
   }
 

app/src/main/java/org/thoughtcrime/securesms/messages/MessageContentProcessor.java

@@ -56,6 +56,7 @@ import org.thoughtcrime.securesms.jobmanager.JobManager;
 import org.thoughtcrime.securesms.jobs.AttachmentDownloadJob;
 import org.thoughtcrime.securesms.jobs.AutomaticSessionResetJob;
 import org.thoughtcrime.securesms.jobs.GroupCallPeekJob;
+import org.thoughtcrime.securesms.jobs.GroupV2UpdateSelfProfileKeyJob;
 import org.thoughtcrime.securesms.jobs.MultiDeviceBlockedUpdateJob;
 import org.thoughtcrime.securesms.jobs.MultiDeviceConfigurationUpdateJob;
 import org.thoughtcrime.securesms.jobs.MultiDeviceContactUpdateJob;
@@ -64,7 +65,9 @@ import org.thoughtcrime.securesms.jobs.MultiDeviceKeysUpdateJob;
 import org.thoughtcrime.securesms.jobs.MultiDeviceStickerPackSyncJob;
 import org.thoughtcrime.securesms.jobs.PaymentLedgerUpdateJob;
 import org.thoughtcrime.securesms.jobs.PaymentTransactionCheckJob;
+import org.thoughtcrime.securesms.jobs.ProfileKeySendJob;
 import org.thoughtcrime.securesms.jobs.PushProcessMessageJob;
+import org.thoughtcrime.securesms.jobs.RefreshAttributesJob;
 import org.thoughtcrime.securesms.jobs.RefreshOwnProfileJob;
 import org.thoughtcrime.securesms.jobs.RequestGroupInfoJob;
 import org.thoughtcrime.securesms.jobs.RetrieveProfileJob;
@@ -88,6 +91,7 @@ import org.thoughtcrime.securesms.payments.MobileCoinPublicAddress;
 import org.thoughtcrime.securesms.ratelimit.RateLimitUtil;
 import org.thoughtcrime.securesms.recipients.Recipient;
 import org.thoughtcrime.securesms.recipients.RecipientId;
+import org.thoughtcrime.securesms.recipients.RecipientUtil;
 import org.thoughtcrime.securesms.ringrtc.RemotePeer;
 import org.thoughtcrime.securesms.service.webrtc.WebRtcData;
 import org.thoughtcrime.securesms.sms.IncomingEncryptedMessage;
@@ -263,6 +267,24 @@ public final class MessageContentProcessor {
 
         if (content.isNeedsReceipt()) {
           handleNeedsDeliveryReceipt(content, message);
+        } else {
+          Recipient sender = getMessageDestination(content, message);
+
+          if (RecipientUtil.shouldHaveProfileKey(context, sender)) {
+            Log.w(TAG, "Received an unsealed sender message from " + sender.getId() + ", but they should already have our profile key. Correcting.");
+
+            if (groupId.isPresent() && groupId.get().isV2()) {
+              Log.i(TAG, "Message was to a GV2 group. Ensuring our group profile keys are up to date.");
+              ApplicationDependencies.getJobManager().startChain(new RefreshAttributesJob(false))
+                                                     .then(GroupV2UpdateSelfProfileKeyJob.withQueueLimits(groupId.get().requireV2()))
+                                                     .enqueue();
+            } else {
+              Log.i(TAG, "Message was to a 1:1 or GV1 chat. Ensuring this user has our profile key.");
+              ApplicationDependencies.getJobManager().startChain(new RefreshAttributesJob(false))
+                                     .then(ProfileKeySendJob.create(context, DatabaseFactory.getThreadDatabase(context).getThreadIdFor(sender), true))
+                                     .enqueue();
+            }
+          }
         }
       } else if (content.getSyncMessage().isPresent()) {
         TextSecurePreferences.setMultiDevice(context, true);

app/src/main/java/org/thoughtcrime/securesms/recipients/RecipientUtil.java

@@ -11,6 +11,7 @@ import com.annimon.stream.Stream;
 import org.signal.core.util.logging.Log;
 import org.thoughtcrime.securesms.contacts.sync.DirectoryHelper;
 import org.thoughtcrime.securesms.database.DatabaseFactory;
+import org.thoughtcrime.securesms.database.GroupDatabase;
 import org.thoughtcrime.securesms.database.RecipientDatabase.RegisteredState;
 import org.thoughtcrime.securesms.database.ThreadDatabase;
 import org.thoughtcrime.securesms.dependencies.ApplicationDependencies;
@@ -265,6 +266,25 @@ public class RecipientUtil {
            threadRecipient.isForceSmsSelection();
   }
 
+  /**
+   * @return True if this recipient should already have your profile key, otherwise false.
+   */
+  public static boolean shouldHaveProfileKey(@NonNull Context context, @NonNull Recipient recipient) {
+    if (recipient.isBlocked()) {
+      return false;
+    }
+
+    if (recipient.isProfileSharing()) {
+      return true;
+    } else {
+      GroupDatabase groupDatabase = DatabaseFactory.getGroupDatabase(context);
+      return groupDatabase.getPushGroupsContainingMember(recipient.getId())
+                          .stream()
+                          .anyMatch(GroupDatabase.GroupRecord::isV2Group);
+
+    }
+  }
+
   @WorkerThread
   private static boolean isMessageRequestAccepted(@NonNull Context context, long threadId, @NonNull Recipient threadRecipient) {
     return threadRecipient.isSelf()                              ||

app/src/main/java/org/thoughtcrime/securesms/sms/MessageSender.java

@@ -91,7 +91,7 @@ public class MessageSender {
    */
   @WorkerThread
   public static void sendProfileKey(final Context context, final long threadId) {
-    ApplicationDependencies.getJobManager().add(ProfileKeySendJob.create(context, threadId));
+    ApplicationDependencies.getJobManager().add(ProfileKeySendJob.create(context, threadId, false));
   }
 
   public static long send(final Context context,