Improve pinned participant state on rejoin with new session

generate-sfu.sh

@@ -0,0 +1,61 @@
+#!/usr/bin/env bash
+
+set -e
+
+REPO_URL="git@github.com:GetStream/protocol.git"
+REFERENCE_TYPE="branch"
+REFERENCE_VALUE="main"
+
+PROJECT_ROOT="$(dirname "$(realpath "$0")")/"
+BUILD_DIR="$PROJECT_ROOT/build"
+CLONE_DIR="$BUILD_DIR/protocol-repo"
+OUTPUT_CLIENT_PATH="$PROJECT_ROOT/stream-video-android-core/src/main/proto/video/sfu/"
+
+# Step 1: Delete OUTPUT_CLIENT_PATH if exists else create an empty directory
+echo "🧹 Preparing output directory: $OUTPUT_CLIENT_PATH"
+rm -rf "$OUTPUT_CLIENT_PATH"
+mkdir -p "$OUTPUT_CLIENT_PATH"
+
+# Step 2: Clone the repository with shallow depth
+echo "🚀 Cloning repository: $REPO_URL (Type: $REFERENCE_TYPE, Value: $REFERENCE_VALUE)..."
+rm -rf "$CLONE_DIR"
+mkdir -p "$BUILD_DIR"
+git clone --depth=1 --branch "$REFERENCE_VALUE" "$REPO_URL" "$CLONE_DIR"
+
+cd "$CLONE_DIR"
+
+# Step 3: Checkout to the correct branch, tag, or commit
+if [ "$REFERENCE_TYPE" == "branch" ]; then
+    git checkout "$REFERENCE_VALUE"
+elif [ "$REFERENCE_TYPE" == "tag" ]; then
+    git fetch --tags
+    git checkout "tags/$REFERENCE_VALUE"
+elif [ "$REFERENCE_TYPE" == "commit" ]; then
+    git fetch --depth=1 origin "$REFERENCE_VALUE"
+    git checkout "$REFERENCE_VALUE"
+else
+    echo "❌ ERROR: Invalid reference type '$REFERENCE_TYPE'. Use 'branch', 'tag', or 'commit'."
+    exit 1
+fi
+
+# Step 4: Copy content from CLONE_DIR/protobuf/video/sfu to OUTPUT_CLIENT_PATH
+echo "📦 Copying proto files..."
+SOURCE_PROTO_PATH="$CLONE_DIR/protobuf/video/sfu"
+
+if [ ! -d "$SOURCE_PROTO_PATH" ]; then
+    echo "❌ ERROR: Source proto directory does not exist: $SOURCE_PROTO_PATH"
+    exit 1
+fi
+
+cp -R "$SOURCE_PROTO_PATH/"* "$OUTPUT_CLIENT_PATH"
+
+# Step 5: Run Spotless (from project root, not inside cloned repo)
+echo "✨ Running Spotless..."
+cd "$PROJECT_ROOT"
+./gradlew spotlessApply
+
+# Step 6: Delete CLONE_DIR
+echo "🗑 Cleaning up cloned repo..."
+rm -rf "$CLONE_DIR"
+
+echo "✅ Done."

stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/Call.kt

@@ -137,7 +137,7 @@
         "fastReconnectDeadlineSeconds. This constant will be removed in a future release.",
     level = DeprecationLevel.WARNING,
 )
 const val sfuReconnectTimeoutMillis = 30_000
 
 /**
  * Outcome of a single reconnect attempt. Each reconnect method returns one of
@@ -264,8 +264,8 @@
      */
     private var isDestroyed = false
 
     /** Session handles all real time communication for video and audio */
     internal val session: MutableStateFlow<RtcSession?> = MutableStateFlow(null)
 
     var sessionId = UUID.randomUUID().toString()
     internal val unifiedSessionId = UUID.randomUUID().toString()
@@ -853,7 +853,7 @@
      * @param strategy the initial reconnection strategy requested by the caller.
      * @param reason a human-readable reason for logging / tracing.
      */
     internal suspend fun reconnect(
         strategy: WebsocketReconnectStrategy,
         reason: String,
     ) {
@@ -1870,7 +1870,7 @@
     /**
      * Should outlive both the call scope and the service scope and needs to be executed in the client-level scope.
      * Because the call scope or service scope may be cancelled or finished while the network request is still in flight
      * TODO: Run this in clientImpl.scope internally
      */
     suspend fun reject(reason: RejectReason? = null): Result<RejectCallResponse> {
         logger.d { "[reject] #ringing; rejectReason: $reason, call_id:$id" }
@@ -1937,9 +1937,13 @@
             sessionId,
         )
 
-    fun isServerPin(sessionId: String): Boolean = state._serverPins.value.containsKey(sessionId)
+    fun isServerPin(sessionId: String): Boolean = state.pinManager.serverPins.value.containsKey(
+        sessionId,
+    )
 
-    fun isLocalPin(sessionId: String): Boolean = state._localPins.value.containsKey(sessionId)
+    fun isLocalPin(sessionId: String): Boolean = state.pinManager.localPins.value.containsKey(
+        sessionId,
+    )
 
     fun hasCapability(vararg capability: OwnCapability): Boolean {
         val elements = capability.toList()

stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/CallState.kt

@@ -118,8 +118,8 @@
 import io.getstream.video.android.core.notifications.internal.telecom.jetpack.JetpackTelecomRepository
 import io.getstream.video.android.core.notifications.internal.telecom.jetpack.TelecomCall
 import io.getstream.video.android.core.permission.PermissionRequest
-import io.getstream.video.android.core.pinning.PinType
-import io.getstream.video.android.core.pinning.PinUpdateAtTime
+import io.getstream.video.android.core.pinning.PinEntry
+import io.getstream.video.android.core.pinning.PinManager
 import io.getstream.video.android.core.socket.common.scope.ClientScope
 import io.getstream.video.android.core.socket.common.scope.UserScope
 import io.getstream.video.android.core.sorting.SortPreset
@@ -167,6 +167,8 @@
 import java.util.concurrent.ConcurrentHashMap
 import java.util.concurrent.atomic.AtomicBoolean
 import java.util.concurrent.atomic.AtomicReference
+import kotlin.collections.map
+import kotlin.collections.toMutableList
 import kotlin.time.Duration
 import kotlin.time.DurationUnit
 import kotlin.time.toDuration
@@ -210,6 +212,9 @@
     public data object Disconnected : RealtimeConnection // normal disconnect by the app
 }
 
+internal typealias SessionId = String
+internal typealias IsPinned = Boolean
+
 /**
  * The CallState class keeps all state for a call
  * It's available on every call object
@@ -269,7 +274,7 @@
         it is RealtimeConnection.Reconnecting
     }
 
-    private val internalParticipants = ConcurrentHashMap<String, ParticipantState>()
+    private val internalParticipants = ConcurrentHashMap<SessionId, ParticipantState>()
     private val _participants = MutableStateFlow<Map<String, ParticipantState>>(emptyMap())
 
     /**
@@ -323,34 +328,31 @@
     private val _dominantSpeaker: MutableStateFlow<ParticipantState?> = MutableStateFlow(null)
     public val dominantSpeaker: StateFlow<ParticipantState?> = _dominantSpeaker
 
-    internal val _localPins: MutableStateFlow<Map<String, PinUpdateAtTime>> =
-        MutableStateFlow(emptyMap())
-    internal val _serverPins: MutableStateFlow<Map<String, PinUpdateAtTime>> =
-        MutableStateFlow(emptyMap())
+    internal val pinManager = PinManager() { internalParticipants }
 
     /**
-     * Combined pin state preserving the underlying [PinUpdateAtTime] (which carries
+     * Combined pin state preserving the underlying [PinEntry] (which carries
      * [io.getstream.video.android.core.pinning.PinType] and the original timestamp).
      * Internal; the sorter consumes this to apply local-pin > server-pin > recency.
      * Local pins win on key collision.
      */
-    internal val _pinnedParticipantsDetailed: StateFlow<Map<String, PinUpdateAtTime>> =
-        combine(_localPins, _serverPins) { local, server ->
-            val combined = mutableMapOf<String, PinUpdateAtTime>()
+    internal val _pinnedParticipantsDetailed: StateFlow<Map<SessionId, PinEntry>> =
+        combine(pinManager.localPins, pinManager.serverPins) { local, server ->
+            val combined = mutableMapOf<String, PinEntry>()
             combined.putAll(server)
             combined.putAll(local) // local overrides server on key collision
             combined.toMap()
         }.stateIn(scope, SharingStarted.Eagerly, emptyMap())
 
-    internal val _pinnedParticipants: StateFlow<Map<String, OffsetDateTime>> =
+    internal val _pinnedParticipants: StateFlow<Map<SessionId, OffsetDateTime>> =
         _pinnedParticipantsDetailed.mapState { detailed ->
             detailed.mapValues { it.value.at }
         }
 
     /**
      * Pinned participants, combined value both from server and local pins.
      */
-    val pinnedParticipants: StateFlow<Map<String, OffsetDateTime>> = _pinnedParticipants
+    val pinnedParticipants: StateFlow<Map<SessionId, OffsetDateTime>> = _pinnedParticipants
 
     val stats = CallStats(call, scope)
 
@@ -471,7 +473,7 @@
         replaceWith = ReplaceWith("participants"),
         level = DeprecationLevel.WARNING,
     )
     val sortedParticipants: Flow<List<ParticipantState>>
         get() = participants
 
     /**
@@ -755,7 +757,8 @@
      */
     val ccMode: StateFlow<ClosedCaptionMode> = closedCaptionManager.ccMode
 
-    private val pendingParticipantsJoined = ConcurrentHashMap<String, Participant>()
+    private val pendingParticipantsJoined =
+        ConcurrentHashMap<SessionId, Pair<Participant, IsPinned>>()
 
     /**
      * We re-create notification more than 1 times, so we don't want to
@@ -805,7 +808,7 @@
             }
 
             is PinsUpdatedEvent -> {
-                updateServerSidePins(event.pins)
+                pinManager.setServerPins(event.pins)
             }
 
             is UnblockedUserEvent -> {
@@ -1098,7 +1101,7 @@
                 } else {
                     _ringingState.value = RingingState.Outgoing(acceptedByCallee = true)
                 }
-                updateServerSidePins(
+                pinManager.setServerPins(
                     event.callState.pins.map {
                         PinUpdate(it.user_id, it.session_id)
                     },
@@ -1109,13 +1112,17 @@
                 try {
                     if (participants.value.size < 8) {
                         getOrCreateParticipant(event.participant)
+                        pinManager.onParticipantJoined(event)
                     } else {
-                        pendingParticipantsJoined[event.participant.session_id] = event.participant
+                        pendingParticipantsJoined[event.participant.session_id] = Pair(event.participant, event.isPinned)
                         participantsUpdate.schedule(scope, participantsUpdateConfig) {
                             logger.d {
                                 "[ParticipantJoinedEvent] #participants; #debounce; participants: ${participants.value.size}"
                             }
-                            getOrCreateParticipants(pendingParticipantsJoined.values.toList())
+                            val pendingParticipants = pendingParticipantsJoined.values.map { it.first }.toList()
+                            getOrCreateParticipants(pendingParticipants)
+                            val pendingParticipantsWithPin = pendingParticipantsJoined.values.toList()
+                            pinManager.onParticipantsJoined(pendingParticipantsWithPin)
                         }
                     }
                 } catch (e: Exception) {
@@ -1136,16 +1143,8 @@
                 if (current?.participant?.sessionId == sessionId) {
                     _screenSharingSession.value = null
                 }
-                if (_localPins.value.containsKey(sessionId)) {
-                    // Remove any pins for the participant
-                    unpin(sessionId)
-                }
 
-                if (_serverPins.value.containsKey(sessionId)) {
-                    scope.launch {
-                        call.unpinForEveryone(sessionId, event.participant.user_id)
-                    }
-                }
+                pinManager.onParticipantLeft(sessionId)
             }
 
             is SubscriberOfferEvent -> {
@@ -1295,19 +1294,6 @@
         }
     }
 
-    private fun updateServerSidePins(pins: List<PinUpdate>) {
-        // Update participants that are still in the call
-        val pinnedInCall = pins.filter {
-            internalParticipants.containsKey(it.sessionId)
-        }
-        _serverPins.value = pinnedInCall.associate {
-            Pair(
-                it.sessionId,
-                PinUpdateAtTime(it, OffsetDateTime.now(Clock.systemUTC()), PinType.Server),
-            )
-        }
-    }
-
     private fun updateRingingState(rejectReason: RejectReason? = null) {
         when (ringingState.value) {
             RingingState.TimeoutNoAnswer, RingingState.RejectedByAll -> {
@@ -1331,7 +1317,7 @@
             _session.value?.participants?.find { it.user.id == client.userId } != null
         val outgoingMembersCount = _members.value.filter { it.value.user.id != client.userId }.size
         val isCallEnded: Boolean = _endedAt.value != null
         val createdBySelf = createdBy?.id == client.userId
 
         ringingLogger.d { "Current: ${_ringingState.value}, call_id: ${call.cid}" }
 
@@ -1348,7 +1334,7 @@
         ringingLogger.d { "call_id: ${call.cid}, Flags: $ringingStateLogs" }
 
         // no members - call is empty, we can join
         val state: RingingState = if (hasActiveCall && !isJoinAndRingInProgress.get()) {
             /**
              * Normal join, not joinAndRing
              */
@@ -1701,19 +1687,11 @@
     }
 
     fun pin(userId: String, sessionId: String) {
-        val pins = _localPins.value.toMutableMap()
-        pins[sessionId] = PinUpdateAtTime(
-            PinUpdate(userId, sessionId),
-            OffsetDateTime.now(Clock.systemUTC()),
-            PinType.Local,
-        )
-        _localPins.value = pins
+        pinManager.pin(userId, sessionId)
     }
 
     fun unpin(sessionId: String) {
-        val pins = _localPins.value.toMutableMap()
-        pins.remove(sessionId)
-        _localPins.value = pins
+        pinManager.unpin(sessionId)
     }
 
     fun updateFromResponse(result: StopLiveResponse) {
@@ -1840,7 +1818,7 @@
     }
 
     @Deprecated("Use updateNotification(Int, Notification) instead")
     fun updateNotification(notification: Notification) {
         atomicNotification.set(notification)
     }
 
@@ -1861,12 +1839,12 @@
     private fun observeTelecomHold(repo: JetpackTelecomRepository) {
         telecomHoldObserverJob?.cancel()
 
         telecomHoldObserverJob = scope.launch(Dispatchers.Default) {
             repo.currentCall
                 .map { (it as? TelecomCall.Registered)?.isOnHold == true }
                 .distinctUntilChanged()
                 .filter { it }
                 .collect { isOnHold ->
                     when (ringingState.value) {
                         is RingingState.Active -> {
                             call.leave("call-on-hold")

stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/signal/socket/RTCEventMapper.kt

@@ -95,7 +95,7 @@ public object RTCEventMapper {
             }
 
             event.participant_joined != null -> with(event.participant_joined) {
-                ParticipantJoinedEvent(participant!!, call_cid)
+                ParticipantJoinedEvent(participant!!, call_cid, is_pinned)
             }
 
             event.participant_left != null -> with(event.participant_left) {

stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/utils/RetryableSignalingServiceDecorator.kt

@@ -27,6 +27,8 @@ import stream.video.sfu.signal.ICERestartResponse
 import stream.video.sfu.signal.ICETrickleResponse
 import stream.video.sfu.signal.SendAnswerRequest
 import stream.video.sfu.signal.SendAnswerResponse
+import stream.video.sfu.signal.SendMetricsRequest
+import stream.video.sfu.signal.SendMetricsResponse
 import stream.video.sfu.signal.SendStatsRequest
 import stream.video.sfu.signal.SendStatsResponse
 import stream.video.sfu.signal.SetPublisherRequest
@@ -117,6 +119,9 @@ internal class RetryableSignalingServiceDecorator(
     override suspend fun sendStats(sendStatsRequest: SendStatsRequest): SendStatsResponse =
         retryCall({ it.error }) { decorated.sendStats(sendStatsRequest) }
 
+    override suspend fun sendMetrics(sendMetricsRequest: SendMetricsRequest): SendMetricsResponse =
+        retryCall({ null }) { decorated.sendMetrics(sendMetricsRequest) }
+
     override suspend fun startNoiseCancellation(startNoiseCancellationRequest: StartNoiseCancellationRequest): StartNoiseCancellationResponse =
         retryCall({ it.error }) { decorated.startNoiseCancellation(startNoiseCancellationRequest) }
 

stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/events/SfuDataEvent.kt

@@ -126,9 +126,9 @@
 ) : SfuDataEvent()
 
 public data class ParticipantJoinedEvent(
-
     val participant: Participant,
     val callCid: String,
+    val isPinned: Boolean,
 ) : SfuDataEvent()
 
 public data class ParticipantLeftEvent(
@@ -168,6 +168,7 @@
     val pins: List<PinUpdate>,
 ) : SfuDataEvent()
 
+// Todo Rahul, should rename it to PinTarget in v2
 public data class PinUpdate(val userId: String, val sessionId: String)
 
 public data class UnknownEvent(val event: Any?) : SfuDataEvent()