CallLink profile sharing via ProfileKeySendJob.

app/src/main/java/org/thoughtcrime/securesms/WebRtcCallActivity.java

@@ -58,7 +58,7 @@ import org.signal.libsignal.protocol.IdentityKey;
 import org.thoughtcrime.securesms.components.TooltipPopup;
 import org.thoughtcrime.securesms.components.sensors.DeviceOrientationMonitor;
 import org.thoughtcrime.securesms.components.webrtc.CallLinkInfoSheet;
-import org.thoughtcrime.securesms.components.webrtc.CallLinkNullMessageSender;
+import org.thoughtcrime.securesms.components.webrtc.CallLinkProfileKeySender;
 import org.thoughtcrime.securesms.components.webrtc.CallParticipantsListUpdatePopupWindow;
 import org.thoughtcrime.securesms.components.webrtc.CallParticipantsState;
 import org.thoughtcrime.securesms.components.webrtc.CallStateUpdatePopupWindow;
@@ -778,7 +778,7 @@ public class WebRtcCallActivity extends BaseActivity implements SafetyNumberChan
     }
 
     if (state.isCallLink()) {
-      CallLinkNullMessageSender.onSendAnyway(new HashSet<>(changedRecipients));
+      CallLinkProfileKeySender.onSendAnyway(new HashSet<>(changedRecipients));
     }
 
     if (state.getGroupCallState().isConnected()) {

app/src/main/java/org/thoughtcrime/securesms/components/webrtc/CallLinkInfoSheet.kt

@@ -52,7 +52,6 @@ import com.google.android.material.dialog.MaterialAlertDialogBuilder
 import io.reactivex.rxjava3.android.schedulers.AndroidSchedulers
 import io.reactivex.rxjava3.core.BackpressureStrategy
 import io.reactivex.rxjava3.kotlin.subscribeBy
-import kotlinx.collections.immutable.ImmutableList
 import kotlinx.collections.immutable.toImmutableList
 import org.signal.core.ui.BottomSheets
 import org.signal.core.ui.Dividers
@@ -73,7 +72,6 @@ import org.thoughtcrime.securesms.compose.ComposeBottomSheetDialogFragment
 import org.thoughtcrime.securesms.database.CallLinkTable
 import org.thoughtcrime.securesms.dependencies.ApplicationDependencies
 import org.thoughtcrime.securesms.events.CallParticipant
-import org.thoughtcrime.securesms.events.WebRtcViewModel
 import org.thoughtcrime.securesms.recipients.Recipient
 import org.thoughtcrime.securesms.recipients.RecipientId
 import org.thoughtcrime.securesms.service.webrtc.links.CallLinkCredentials
@@ -128,11 +126,11 @@ class CallLinkInfoSheet : ComposeBottomSheetDialogFragment() {
       .toLiveData()
       .observeAsState()
 
-    val participants: ImmutableList<CallParticipant> = if (callParticipantsState?.callState == WebRtcViewModel.State.CALL_CONNECTED) {
-      listOf(CallParticipant(recipient = Recipient.self())) + (callParticipantsState?.allRemoteParticipants?.map { it } ?: emptyList())
-    } else {
-      emptyList()
-    }.toImmutableList()
+    val participants: List<CallParticipant> by webRtcCallViewModel.callParticipantsState
+      .toFlowable(BackpressureStrategy.LATEST)
+      .map { state -> state.allRemoteParticipants }
+      .toLiveData()
+      .observeAsState(emptyList())
 
     val onEditNameClicked: () -> Unit = remember(callLinkDetailsState) {
       {
@@ -249,7 +247,7 @@ private fun SheetPreview() {
 @Composable
 private fun Sheet(
   callLink: CallLinkTable.CallLink,
-  participants: ImmutableList<CallParticipant>,
+  participants: List<CallParticipant>,
   onShareLinkClicked: () -> Unit,
   onEditNameClicked: () -> Unit,
   onToggleAdminApprovalClicked: (Boolean) -> Unit,
@@ -334,6 +332,11 @@ private fun CallLinkMemberRow(
       .fillMaxWidth()
       .padding(Rows.defaultPadding())
   ) {
+    val recipient by Recipient.observable(callParticipant.recipient.id)
+      .toFlowable(BackpressureStrategy.LATEST)
+      .toLiveData()
+      .observeAsState(initial = callParticipant.recipient)
+
     if (LocalInspectionMode.current) {
       Spacer(
         modifier = Modifier
@@ -345,20 +348,20 @@ private fun CallLinkMemberRow(
         factory = ::AvatarImageView,
         modifier = Modifier.size(40.dp)
       ) {
-        it.setAvatarUsingProfile(callParticipant.recipient)
+        it.setAvatarUsingProfile(recipient)
       }
     }
 
     Spacer(modifier = Modifier.width(24.dp))
 
     Text(
-      text = callParticipant.recipient.getShortDisplayName(LocalContext.current),
+      text = recipient.getShortDisplayName(LocalContext.current),
       modifier = Modifier
         .weight(1f)
         .align(Alignment.CenterVertically)
     )
 
-    if (isSelfAdmin && !callParticipant.recipient.isSelf) {
+    if (isSelfAdmin && !recipient.isSelf) {
       Icon(
         imageVector = ImageVector.vectorResource(id = R.drawable.symbol_minus_circle_24),
         contentDescription = null,

app/src/main/java/org/thoughtcrime/securesms/components/webrtc/CallLinkProfileKeySender.kt

@@ -5,18 +5,21 @@
 
 package org.thoughtcrime.securesms.components.webrtc
 
+import org.signal.core.util.logging.Log
 import org.thoughtcrime.securesms.database.identity.IdentityRecordList
 import org.thoughtcrime.securesms.dependencies.ApplicationDependencies
-import org.thoughtcrime.securesms.jobs.NullMessageSendJob
+import org.thoughtcrime.securesms.jobs.ProfileKeySendJob
 import org.thoughtcrime.securesms.recipients.Recipient
 import org.thoughtcrime.securesms.recipients.RecipientId
 
 /**
- * In-Memory Cache that keeps track of whom we've sent NullMessages to. This is
+ * In-Memory Cache that keeps track of whom we've sent profile keys to. This is
  * something that really only needs to happen once so that profile information is
  * displayed correctly, so we maintain an application process scoped cache.
  */
-object CallLinkNullMessageSender {
+object CallLinkProfileKeySender {
+  private val TAG = Log.tag(CallLinkProfileKeySender::class.java)
+
   private val cache = hashSetOf<RecipientId>()
 
   /**
@@ -28,21 +31,26 @@ object CallLinkNullMessageSender {
     val toSendMessagesTo: Set<RecipientId> = recipientIds - cache
     cache += recipientIds
 
-    val jobs: List<NullMessageSendJob> = toSendMessagesTo.map { NullMessageSendJob(it) }
-    ApplicationDependencies.getJobManager().addAll(jobs)
+    if (toSendMessagesTo.isNotEmpty()) {
+      Log.i(TAG, "Sending profile key to $toSendMessagesTo users.")
+      val job = ProfileKeySendJob.createForCallLinks(toSendMessagesTo.toList())
+      ApplicationDependencies.getJobManager().add(job)
+    } else {
+      Log.i(TAG, "No users to send profile key to.")
+    }
   }
 
   /**
    * Given the set of recipients, for each unblocked recipient we don't distrust, send a NullMessage
    */
   fun onRecipientsUpdated(recipients: Set<Recipient>) {
-    val nonBlockedRecipients: List<Recipient> = recipients.filterNot { it.isBlocked }
+    val nonBlockedNonSelfRecipients: List<Recipient> = recipients.filterNot { it.isBlocked || it.isSelf }
 
     val identityRecords: IdentityRecordList = ApplicationDependencies
       .getProtocolStore()
       .aci()
       .identities()
-      .getIdentityRecords(nonBlockedRecipients)
+      .getIdentityRecords(nonBlockedNonSelfRecipients)
 
     val untrustedAndUnverifiedRecipients = if (identityRecords.isUntrusted(false) || identityRecords.isUnverified(false)) {
       (identityRecords.untrustedRecipients + identityRecords.unverifiedRecipients).toSet()
@@ -50,7 +58,8 @@ object CallLinkNullMessageSender {
       emptySet()
     }
 
-    val trustedRecipients: Set<RecipientId> = (nonBlockedRecipients - untrustedAndUnverifiedRecipients).map { it.id }.toSet()
+    val trustedRecipients: Set<RecipientId> = (nonBlockedNonSelfRecipients - untrustedAndUnverifiedRecipients).map { it.id }.toSet()
+
     onSendAnyway(trustedRecipients)
   }
 }

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

@@ -8,8 +8,8 @@ import com.annimon.stream.Stream;
 
 import org.signal.core.util.logging.Log;
 import org.thoughtcrime.securesms.database.SignalDatabase;
-import org.thoughtcrime.securesms.jobmanager.JsonJobData;
 import org.thoughtcrime.securesms.jobmanager.Job;
+import org.thoughtcrime.securesms.jobmanager.JsonJobData;
 import org.thoughtcrime.securesms.jobmanager.impl.DecryptionsDrainedConstraint;
 import org.thoughtcrime.securesms.jobmanager.impl.NetworkConstraint;
 import org.thoughtcrime.securesms.messages.GroupSendUtil;
@@ -39,6 +39,21 @@ public class ProfileKeySendJob extends BaseJob {
   private final long              threadId;
   private final List<RecipientId> recipients;
 
+  public static ProfileKeySendJob createForCallLinks(List<RecipientId> recipientIds) {
+    return new ProfileKeySendJob(
+        new Parameters.Builder()
+            .setQueue("ProfileKeySendJob__call_links")
+            .setMaxInstancesForQueue(Parameters.UNLIMITED)
+            .addConstraint(NetworkConstraint.KEY)
+            .addConstraint(DecryptionsDrainedConstraint.KEY)
+            .setLifespan(TimeUnit.DAYS.toMillis(1))
+            .setMaxAttempts(Parameters.UNLIMITED)
+            .build(),
+        -1L,
+        recipientIds
+    );
+  }
+
   /**
    * Suitable for a 1:1 conversation or a GV1 group only.
    *
@@ -99,11 +114,13 @@ public class ProfileKeySendJob extends BaseJob {
       throw new NotPushRegisteredException();
     }
 
-    Recipient conversationRecipient = SignalDatabase.threads().getRecipientForThreadId(threadId);
+    if (threadId > 0) {
+      Recipient conversationRecipient = SignalDatabase.threads().getRecipientForThreadId(threadId);
 
-    if (conversationRecipient == null) {
-      Log.w(TAG, "Thread no longer present");
-      return;
+      if (conversationRecipient == null) {
+        Log.w(TAG, "Thread no longer present");
+        return;
+      }
     }
 
     List<Recipient> destinations = Stream.of(recipients).map(Recipient::resolved).toList();

app/src/main/java/org/thoughtcrime/securesms/service/webrtc/CallLinkConnectedActionProcessor.kt

@@ -9,7 +9,7 @@ import org.signal.core.util.logging.Log
 import org.signal.ringrtc.CallException
 import org.signal.ringrtc.GroupCall
 import org.signal.ringrtc.PeekInfo
-import org.thoughtcrime.securesms.components.webrtc.CallLinkNullMessageSender
+import org.thoughtcrime.securesms.components.webrtc.CallLinkProfileKeySender
 import org.thoughtcrime.securesms.database.CallLinkTable
 import org.thoughtcrime.securesms.database.SignalDatabase
 import org.thoughtcrime.securesms.events.CallParticipant
@@ -46,6 +46,9 @@ class CallLinkConnectedActionProcessor(
 
     val callLinkRoomId: CallLinkRoomId = superState.callInfoState.callRecipient.requireCallLinkRoomId()
     val callLink: CallLinkTable.CallLink = SignalDatabase.callLinks.getCallLinkByRoomId(callLinkRoomId) ?: return superState
+    val joinedParticipants: Set<Recipient> = peekInfo.joinedMembers.map { Recipient.externalPush(ServiceId.ACI.from(it)) }.toSet()
+
+    CallLinkProfileKeySender.onRecipientsUpdated(joinedParticipants)
 
     if (callLink.credentials?.adminPassBytes == null) {
       Log.i(tag, "User is not an admin.")
@@ -55,8 +58,6 @@ class CallLinkConnectedActionProcessor(
     Log.i(tag, "Updating pending list with ${peekInfo.pendingUsers.size} entries.")
     val pendingParticipants: List<Recipient> = peekInfo.pendingUsers.map { Recipient.externalPush(ServiceId.ACI.from(it)) }
 
-    CallLinkNullMessageSender.onRecipientsUpdated(superState.callInfoState.remoteCallParticipants.map { it.recipient }.toSet())
-
     return superState.builder()
       .changeCallInfoState()
       .setCallLinkPendingParticipants(pendingParticipants)