[Enhancement] Extend CallJoinIntercepting with will-join and did-join hooks

DemoApp/Sources/Components/Debug/Items/FeatureFlags/Components/CallJoinInterceptor.swift

@@ -33,7 +33,7 @@ extension AppEnvironment {
     }
 
     /// Default to enabled in debug builds so the pipeline is exercised during development.
-    nonisolated(unsafe) static var callJoinInterceptor: CallJoinInterceptor = .none
+    nonisolated(unsafe) static var callJoinInterceptor: CallJoinInterceptor = .synchronised
 }
 
 extension DebugMenu {

DemoApp/Sources/Components/DemoCallJoinInterceptor/DemoCallJoinInterceptor.swift

@@ -21,6 +21,17 @@ final class DemoCallJoinInterceptor: CallJoinIntercepting {
 
     private let hasOtherReadyParticipants: CurrentValueSubject<Bool, Never> = .init(false)
 
+    /// Creates the interceptor and starts tracking the active ringing call.
+    ///
+    /// The readiness handshake is only meaningful for ringing (1:1/group)
+    /// calls, so the interceptor observes `streamVideo.state.ringingCall` from
+    /// construction. Each time the ringing call changes it (re)wires the
+    /// custom-event observation in `didUpdate(ringingCall:)`. The current user
+    /// id is cached here because it's used both to emit the local readiness
+    /// marker and to ignore the marker echoed back from ourselves.
+    ///
+    /// - Parameter customEventKey: The custom-event key used to exchange the
+    ///   readiness marker between participants.
     init(
         customEventKey: String = "participant.ready"
     ) {
@@ -42,6 +53,18 @@ final class DemoCallJoinInterceptor: CallJoinIntercepting {
 
     // MARK: - CallJoinIntercepting
 
+    /// Mutes incoming media while the call is still connecting.
+    ///
+    /// Called by the SDK when the WebRTC layer enters its joining stage. The
+    /// demo silences remote audio so the local user doesn't hear other
+    /// participants until the readiness handshake completes and the call is
+    /// fully joined, avoiding leaking audio during the "preparing" window.
+    ///
+    /// - Parameter call: The call being prepared for joining.
+    func callWillJoin(_ call: Call) async {
+        await disableMediaTracksAndCallSettings(call)
+    }
+
     /// Sends the local readiness marker and waits until another participant
     /// sends the same signal.
     ///
@@ -67,8 +90,31 @@ final class DemoCallJoinInterceptor: CallJoinIntercepting {
             .nextValue()
     }
 
+    /// Restores incoming media once the call is fully joined.
+    ///
+    /// Counterpart to ``callWillJoin(_:)``. Called by the SDK after the call
+    /// has transitioned into the joined state, it re-enables the remote audio
+    /// that was muted during preparation and stops the deactivation observer so
+    /// the user can finally hear the other participants.
+    ///
+    /// - Parameter call: The call that just became active.
+    func callDidJoin(_ call: Call) async {
+        await restoreMediaTracksAndCallSettings(call)
+    }
+
     // MARK: - Private Helpers
 
+    /// Rewires the readiness observation whenever the ringing call changes.
+    ///
+    /// Any previous observation is cancelled first so a stale subscription
+    /// never leaks across calls. When a new ringing call is present, it
+    /// subscribes to that call's custom events, keeps only the readiness marker
+    /// emitted by *other* participants (filtering out our own echo), and flips
+    /// `hasOtherReadyParticipants` to `true` — which is what
+    /// ``callReadyToJoin(_:)`` is waiting on. When the ringing call becomes
+    /// `nil` it simply leaves the observation cancelled.
+    ///
+    /// - Parameter ringingCall: The currently ringing call, or `nil` if none.
     private func didUpdate(ringingCall: Call?) {
         cancelCustomEventObservation()
 
@@ -92,6 +138,12 @@ final class DemoCallJoinInterceptor: CallJoinIntercepting {
         log.debug("Call presence events observation has started.")
     }
 
+    /// Tears down the readiness observation and resets its state.
+    ///
+    /// Used when the ringing call changes or disappears. It cancels the custom
+    /// event subscription and resets `hasOtherReadyParticipants` to `false` so
+    /// a later join doesn't inherit a stale "ready" signal from a previous
+    /// call. The early return keeps the call idempotent when nothing is active.
     private func cancelCustomEventObservation() {
         guard customEventCancellable != nil else {
             return
@@ -101,4 +153,46 @@ final class DemoCallJoinInterceptor: CallJoinIntercepting {
         hasOtherReadyParticipants.send(false)
         log.debug("Call presence events observation was cancelled.")
     }
+
+    /// Continuously mutes remote participants' audio while preparing.
+    ///
+    /// A simple one-shot mute isn't enough: participants (and their audio
+    /// tracks) can arrive after the join begins. This subscribes to the
+    /// participants list and disables the audio track of every participant
+    /// other than the local user as they appear, so newly published remote
+    /// audio stays muted too. The subscription is keyed in the disposable bag
+    /// so ``restoreMediaTracksAndCallSettings(_:)`` can stop it once joined.
+    ///
+    /// - Parameter call: The call whose remote audio should be muted.
+    private func disableMediaTracksAndCallSettings(_ call: Call) async {
+        let currentUser = call.state.localParticipant
+
+        call
+            .state
+            .$participants
+            .map { array in array.filter { $0.sessionId != currentUser?.sessionId } }
+            .map { $0.compactMap { $0.audioTrack } }
+            .filter { $0.isEmpty == false }
+            .sink { $0.forEach { $0.isEnabled = false } }
+            .store(in: disposableBag, key: "tracks-deactivation")
+    }
+
+    /// Re-enables remote participants' audio after the call has joined.
+    ///
+    /// Reverses ``disableMediaTracksAndCallSettings(_:)``: it turns the audio
+    /// track of every remote participant back on for a one-time restore, then
+    /// removes the keyed subscription that was muting newly arriving tracks so
+    /// the demo stops interfering with normal call audio.
+    ///
+    /// - Parameter call: The call whose remote audio should be restored.
+    private func restoreMediaTracksAndCallSettings(_ call: Call) async {
+        let currentUser = call.state.localParticipant
+        call
+            .state
+            .participants
+            .filter { $0.sessionId != currentUser?.sessionId && $0.audioTrack != nil }
+            .forEach { $0.audioTrack?.isEnabled = true }
+
+        disposableBag.remove("tracks-deactivation")
+    }
 }

Sources/StreamVideo/CallStateMachine/Stages/Call+JoinedStage.swift

@@ -19,6 +19,7 @@ extension Call.StateMachine.Stage {
         JoinedStage(
             .init(
                 call: context.call,
+                input: context.input,
                 output: .joined(response)
             )
         )
@@ -77,11 +78,22 @@ extension Call.StateMachine.Stage {
         ///
         /// The method subscribes to call settings and capability updates so the
         /// call controller and media managers stay aligned with live call state.
+        /// Once those subscriptions are in place it notifies the optional join
+        /// interceptor via ``CallJoinIntercepting/callDidJoin(_:)``, allowing it
+        /// to undo any temporary setup applied during the preparing stage. The
+        /// interceptor is recovered from the join input that travelled through
+        /// the state machine context.
         private func execute() {
+            let joinInterceptor: CallJoinIntercepting? = {
+                guard case let .join(input) = context.input else { return nil }
+                return input.joinInterceptor
+            }()
+
             Task(disposableBag: disposableBag) { [weak self] in
                 guard let self, let call = context.call else { return }
                 await subscribeToCallSettingsUpdates(on: call)
                 await subscribeToOwnCapabilitiesChanges(on: call)
+                await joinInterceptor?.callDidJoin(call)
             }
         }
 

Sources/StreamVideo/CallStateMachine/Stages/Call+JoiningStage.swift

@@ -141,6 +141,10 @@ extension Call.StateMachine.Stage {
         ) async throws -> JoinCallResponse {
             try Task.checkCancellation()
 
+            observeCallStage(call, joinCallInterceptor: input.joinInterceptor)
+
+            try Task.checkCancellation()
+
             let response = try await call.callController.joinCall(
                 create: input.create,
                 callSettings: input.callSettings,
@@ -254,5 +258,37 @@ extension Call.StateMachine.Stage {
                 throw CancellationError()
             }
         }
+
+        /// Bridges the WebRTC join lifecycle to the interceptor's preparation
+        /// hook.
+        ///
+        /// When an interceptor is provided, this subscribes to the call
+        /// controller's `stagePublisher` and, every time the underlying WebRTC
+        /// state machine reports the `.joining` stage, invokes
+        /// ``CallJoinIntercepting/callWillJoin(_:)``. This gives integrators
+        /// a chance to run setup work (e.g. muting remote tracks) while the peer
+        /// connection is still being negotiated. The subscription is stored in
+        /// the stage-scoped `disposableBag`, so it is torn down automatically
+        /// when the stage transitions away.
+        ///
+        /// - Parameters:
+        ///   - call: The call being joined.
+        ///   - joinCallInterceptor: The optional interceptor to notify. When
+        ///     `nil`, no observation is set up.
+        private func observeCallStage(
+            _ call: Call,
+            joinCallInterceptor: CallJoinIntercepting?
+        ) {
+            guard let joinCallInterceptor else {
+                return
+            }
+
+            call
+                .callController
+                .stagePublisher
+                .filter { $0 == .joining }
+                .sinkTask(storeIn: disposableBag) { _ in await joinCallInterceptor.callWillJoin(call) }
+                .store(in: disposableBag)
+        }
     }
 }

Sources/StreamVideo/Controllers/CallController.swift

@@ -61,6 +61,17 @@ class CallController: @unchecked Sendable {
         }
     }
 
+    /// Publishes the identifier of the WebRTC coordinator's current state
+    /// machine stage.
+    ///
+    /// A new value is emitted whenever the join/connection lifecycle advances
+    /// (for example into `.joining` or `.joined`). The call's join state
+    /// machine observes this to drive the optional ``CallJoinIntercepting``
+    /// preparation hook without reaching into WebRTC internals directly.
+    var stagePublisher: AnyPublisher<WebRTCCoordinator.StateMachine.Stage.ID, Never> {
+        webRTCCoordinator.stateMachine.publisher.map(\.id).eraseToAnyPublisher()
+    }
+
     private let user: User
     private let callId: String
     private let callType: String

Sources/StreamVideo/Utils/CallJoinIntercepting.swift

@@ -20,6 +20,21 @@
 /// active and completes the join flow.
 public protocol CallJoinIntercepting: Sendable {
 
+    /// Notifies the interceptor that the SDK has started preparing the call for
+    /// a join.
+    ///
+    /// This fires as soon as the underlying WebRTC state machine enters its
+    /// `.joining` stage, before the peer connection has finished negotiating.
+    /// Use it to apply setup that should be in place while the call is
+    /// connecting — for example muting remote media or tweaking local call
+    /// settings — so the user is never exposed to a half-connected experience.
+    ///
+    /// The hook is best-effort: the SDK does not wait for it before continuing
+    /// to negotiate the call.
+    ///
+    /// - Parameter call: The call that is being prepared for joining.
+    func callWillJoin(_ call: Call) async
+
     /// Performs application-specific work before the SDK completes a call join.
     ///
     /// Return normally to let the call continue into the joined state. Throw an
@@ -29,6 +44,33 @@
     /// - Parameter call: The call whose join response has already been applied
     ///   to local state and is about to become active.
     func callReadyToJoin(_ call: Call) async throws
+
+    /// Notifies the interceptor that the call has fully transitioned into the
+    /// joined state.
+    ///
+    /// This is the counterpart to ``callWillJoin(_:)`` and is the right place
+    /// to undo any temporary setup applied while the call was connecting — for
+    /// example restoring remote media that was muted during preparation.
+    ///
+    /// Like ``callWillJoin(_:)`` it is best-effort and does not block the
+    /// call lifecycle.
+    ///
+    /// - Parameter call: The call that just became active.
+    func callDidJoin(_ call: Call) async
+}
+
+/// Default no-op implementations that make the will-join and did-join hooks
+/// optional.
+///
+/// Only ``CallJoinIntercepting/callReadyToJoin(_:)`` must be implemented.
+/// Integrators that don't need to react to the will-join or did-join
+/// transitions can omit these methods, and the additions stay source-compatible
+/// with existing conformers.
+extension CallJoinIntercepting {
+
+    public func callWillJoin(_ call: Call) async { /* No-op by default. */ }
+
+    public func callDidJoin(_ call: Call) async { /* No-op by default. */ }
 }
 
 extension WebRTCTrace {

StreamVideoTests/CallStateMachine/CallStateMachine/Stages/CallStateMachine_JoinedStageTests.swift

@@ -97,6 +97,42 @@ final class CallStateMachineStageJoinedStage_Tests: StreamVideoTestCase, @unchec
                 .contains([.sendAudio]) ?? false
         }
     }
+
+    // MARK: - JoinInterceptor
+
+    func test_joiningTransition_joinInterceptorProvided_callDidJoinInvoked() async {
+        let joinInterceptor = JoinedTrackingInterceptor_Spy()
+        let context = Call.StateMachine.Stage.Context(
+            call: call,
+            input: .join(
+                .init(
+                    create: true,
+                    ring: false,
+                    notify: false,
+                    source: .inApp,
+                    deliverySubject: .init(nil),
+                    joinInterceptor: joinInterceptor
+                )
+            )
+        )
+        subject = .joined(context, response: response)
+
+        _ = subject.transition(from: .init(id: .joining, context: context))
+
+        await fulfillment {
+            joinInterceptor.recordedJoinedCIds == [self.call.cId]
+        }
+    }
+}
+
+private final class JoinedTrackingInterceptor_Spy: CallJoinIntercepting, @unchecked Sendable {
+    @Atomic private(set) var recordedJoinedCIds: [String] = []
+
+    func callReadyToJoin(_ call: Call) async throws {}
+
+    func callDidJoin(_ call: Call) async {
+        recordedJoinedCIds.append(call.cId)
+    }
 }
 
 extension Call.StateMachine.Stage.Context.Output {

StreamVideoTests/CallStateMachine/CallStateMachine/Stages/CallStateMachine_JoiningStageTests.swift

@@ -662,6 +662,40 @@ final class StreamCallStateMachineStageJoiningStage_Tests: StreamVideoTestCase,
         cancellable.cancel()
     }
 
+    func test_execute_whenWebRTCStageBecomesJoining_invokesInterceptorCallIsPreparing() async throws {
+        // The interceptor suspends in `callReadyToJoin`, keeping the stage in
+        // `.joining` so the stage-observation subscription stays alive while we
+        // assert that the preparing hook fired.
+        let joinInterceptor = NonCancellableJoinInterceptor_Spy()
+        callController.stageSubject.send(.joining)
+        callController.stub(for: .join, with: JoinCallResponse.dummy())
+        let context = Call.StateMachine.Stage.Context(
+            call: call,
+            input: .join(
+                .init(
+                    create: true,
+                    callSettings: .init(audioOn: false),
+                    options: .init(memberIds: [.unique]),
+                    ring: true,
+                    notify: false,
+                    source: .inApp,
+                    deliverySubject: .init(nil),
+                    joinInterceptor: joinInterceptor
+                )
+            )
+        )
+        subject.transition = { self.transitionedToStage = $0 }
+        subject.context = context
+
+        _ = subject.transition(from: .idle(.init()))
+
+        await fulfilmentInMainActor(timeout: defaultTimeout) {
+            joinInterceptor.recordedPreparingCIds == [self.call.cId]
+        }
+
+        joinInterceptor.resume()
+    }
+
     func test_execute_withRetries_whenJoinFailsAndThereAreAvailableRetries_transitionsToJoining() async throws {
         let context = Call.StateMachine.Stage.Context(
             call: call,
@@ -836,12 +870,17 @@ private final class CallJoinInterceptor_Spy: CallJoinIntercepting, @unchecked Se
 private final class NonCancellableJoinInterceptor_Spy: CallJoinIntercepting, @unchecked Sendable {
     private let onEntered: @Sendable () -> Void
     @Atomic private(set) var recordedCallCIds: [String] = []
+    @Atomic private(set) var recordedPreparingCIds: [String] = []
     @Atomic private var continuation: CheckedContinuation<Void, Never>?
 
     init(onEntered: @escaping @Sendable () -> Void = {}) {
         self.onEntered = onEntered
     }
 
+    func callWillJoin(_ call: Call) async {
+        recordedPreparingCIds.append(call.cId)
+    }
+
     func callReadyToJoin(_ call: Call) async throws {
         recordedCallCIds.append(call.cId)
         onEntered()