Pin coordinator join to a specific SFU via sfu_id

demo-app/src/main/kotlin/io/getstream/video/android/util/StreamVideoInitHelper.kt

@@ -33,6 +33,7 @@ import io.getstream.video.android.core.StreamVideo
 import io.getstream.video.android.core.StreamVideoBuilder
 import io.getstream.video.android.core.call.CallType
 import io.getstream.video.android.core.internal.ExperimentalStreamVideoApi
+import io.getstream.video.android.core.internal.InternalStreamVideoApi
 import io.getstream.video.android.core.logging.LoggingLevel
 import io.getstream.video.android.core.moderations.ModerationConfig
 import io.getstream.video.android.core.moderations.ModerationWarningConfig
@@ -81,13 +82,16 @@ public enum class InitializedState {
  * @property secret API secret from the local coordinator — used to sign a JWT for [userId].
  * @property userId User ID to connect as.
  * @property token Pre-generated JWT token. If null, one is generated from [secret].
+ * @property sfuId Coordinator edge pin (`?sfu_id=` / `WithPinToSFUID`). When set,
+ * every join / rejoin / migrate asks the coordinator for this SFU.
  */
 data class LocalDevConfig(
     val coordinatorAddress: String,
     val apiKey: String,
     val userId: String,
     val secret: String? = null,
     val token: String? = null,
+    val sfuId: String? = null,
 ) {
     init {
         require(secret != null || token != null) {
@@ -196,6 +200,7 @@ object StreamVideoInitHelper {
                     token = localCfg.resolveToken(),
                     loggingLevel = LoggingLevel(priority = Priority.VERBOSE),
                     localCoordinatorAddress = localCfg.coordinatorAddress,
+                    sfuId = localCfg.sfuId,
                     localTokenProvider = object : TokenProvider {
                         override suspend fun loadToken(): String = localCfg.resolveToken()
                     },
@@ -367,14 +372,15 @@ object StreamVideoInitHelper {
     }
 
     /** Sets up and returns the [StreamVideo] required to connect to the API. */
-    @OptIn(ExperimentalStreamVideoApi::class)
+    @OptIn(ExperimentalStreamVideoApi::class, InternalStreamVideoApi::class)
     private fun initializeStreamVideo(
         context: Context,
         apiKey: ApiKey,
         user: User,
         token: String,
         loggingLevel: LoggingLevel,
         localCoordinatorAddress: String? = null,
+        sfuId: String? = null,
         localTokenProvider: TokenProvider? = null,
     ): StreamVideo {
         val callServiceConfigRegistry = CallServiceConfigRegistry()
@@ -506,6 +512,8 @@ object StreamVideoInitHelper {
             telecomConfig = TelecomConfig(context.packageName),
             connectOnInit = false,
             rejectCallWhenBusy = false,
-        ).build()
+        ).apply {
+            sfuId?.takeIf { it.isNotBlank() }?.let(::forceSfuId)
+        }.build()
     }
 }

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

@@ -172,6 +172,7 @@
 
     private var apiUrl: String? = null
     private var wssUrl: String? = null
+    private var sfuId: String? = null
 
     /**
      * Set the API URL to be used for the video client.
@@ -193,6 +194,16 @@
         wssUrl = value
     }
 
+    /**
+     * Set the SFU id the coordinator should pin every join, rejoin and migrate to.
+     *
+     * For testing purposes only.
+     */
+    @InternalStreamVideoApi
+    public fun forceSfuId(value: String): StreamVideoBuilder = apply {
+        sfuId = value
+    }
+
     /**
      * Builds the [StreamVideo] client.
      *
@@ -218,7 +229,7 @@
             throw IllegalArgumentException("The API key cannot be blank")
         }
 
         if (user.type == UserType.Authenticated && token.isBlank()) {
             throw IllegalArgumentException("The token cannot be blank for authenticated users")
         }
 
@@ -261,6 +272,7 @@
             tokenProvider = tokenProvider,
             lifecycle = lifecycle,
             tokenRepository = tokenRepository,
+            pinnedSfuId = sfuId,
         )
 
         val deviceTokenStorage = DeviceTokenStorage(context)

stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/internal/module/CoordinatorConnectionModule.kt

@@ -60,6 +60,7 @@ internal class CoordinatorConnectionModule(
     override val apiKey: ApiKey,
     override val lifecycle: Lifecycle,
     override val tracer: Tracer = Tracer("coordinator"),
+    pinnedSfuId: String? = null,
 ) : ConnectionModuleDeclaration<ProductvideoApi, CoordinatorSocketConnection, OkHttpClient, UserToken> {
     // Internals
     private val authInterceptor = CoordinatorAuthInterceptor(apiKey, tokenRepository)
@@ -75,6 +76,11 @@ internal class CoordinatorConnectionModule(
     override val http: OkHttpClient = OkHttpClient.Builder().addInterceptor(
         HeadersInterceptor(HeadersUtil()),
     )
+        .apply {
+            if (!pinnedSfuId.isNullOrBlank()) {
+                addInterceptor(CoordinatorSfuPinInterceptor(pinnedSfuId))
+            }
+        }
         .addInterceptor(authInterceptor).addInterceptor(
             HttpLoggingInterceptor {
                 streamLog(tag = "Video:Http") { it }

stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/internal/module/CoordinatorSfuPinInterceptor.kt

@@ -0,0 +1,52 @@
+/*
+ * Copyright (c) 2014-2026 Stream.io Inc. All rights reserved.
+ *
+ * Licensed under the Stream License;
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    https://github.com/GetStream/stream-video-android/blob/main/LICENSE
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.getstream.video.android.core.internal.module
+
+import okhttp3.Interceptor
+import okhttp3.Response
+
+/**
+ * Adds `?sfu_id=` on coordinator join requests when a local-dev pin is configured.
+ *
+ * The coordinator reads this query via `WithPinToSFUID`. It is not part of the
+ * published OpenAPI join body, so this interceptor keeps
+ * [io.getstream.android.video.generated.apis.ProductvideoApi.joinCall]
+ * binary-compatible instead of adding a Retrofit `@Query`.
+ */
+internal class CoordinatorSfuPinInterceptor(
+    private val pinnedSfuId: String?,
+) : Interceptor {
+    companion object {
+        const val QUERY_SFU_ID = "sfu_id"
+        const val JOIN_PATH_SUFFIX = "/join"
+    }
+
+    override fun intercept(chain: Interceptor.Chain): Response {
+        val request = chain.request()
+        val pin = pinnedSfuId?.takeIf { it.isNotBlank() }
+        if (pin == null ||
+            !request.url.encodedPath.endsWith(JOIN_PATH_SUFFIX) ||
+            request.url.queryParameter(QUERY_SFU_ID) != null
+        ) {
+            return chain.proceed(request)
+        }
+        val url = request.url.newBuilder()
+            .addQueryParameter(QUERY_SFU_ID, pin)
+            .build()
+        return chain.proceed(request.newBuilder().url(url).build())
+    }
+}

stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/internal/module/CoordinatorSfuPinInterceptorTest.kt

@@ -0,0 +1,105 @@
+/*
+ * Copyright (c) 2014-2026 Stream.io Inc. All rights reserved.
+ *
+ * Licensed under the Stream License;
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    https://github.com/GetStream/stream-video-android/blob/main/LICENSE
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.getstream.video.android.core.internal.module
+
+import com.google.common.truth.Truth.assertThat
+import okhttp3.MediaType.Companion.toMediaType
+import okhttp3.OkHttpClient
+import okhttp3.Protocol
+import okhttp3.Request
+import okhttp3.RequestBody.Companion.toRequestBody
+import okhttp3.Response
+import okhttp3.ResponseBody.Companion.toResponseBody
+import org.junit.Test
+
+class CoordinatorSfuPinInterceptorTest {
+
+    @Test
+    fun `adds sfu_id on coordinator join when a pin is configured`() {
+        val proceeded = intercept(
+            pinnedSfuId = "SFU-1",
+            url = "https://video.stream-io-api.com/video/call/default/abc/join?connection_id=c1",
+        )
+
+        assertThat(proceeded.url.queryParameter("sfu_id")).isEqualTo("SFU-1")
+        assertThat(proceeded.url.queryParameter("connection_id")).isEqualTo("c1")
+    }
+
+    @Test
+    fun `leaves non-join requests unchanged`() {
+        val proceeded = intercept(
+            pinnedSfuId = "SFU-1",
+            url = "https://video.stream-io-api.com/video/call/default/abc",
+        )
+
+        assertThat(proceeded.url.queryParameter("sfu_id")).isNull()
+    }
+
+    @Test
+    fun `leaves join unchanged when no pin is configured`() {
+        val proceeded = intercept(
+            pinnedSfuId = null,
+            url = "https://video.stream-io-api.com/video/call/default/abc/join",
+        )
+
+        assertThat(proceeded.url.queryParameter("sfu_id")).isNull()
+    }
+
+    @Test
+    fun `leaves join unchanged when the pin is blank`() {
+        val proceeded = intercept(
+            pinnedSfuId = "   ",
+            url = "https://video.stream-io-api.com/video/call/default/abc/join",
+        )
+
+        assertThat(proceeded.url.queryParameter("sfu_id")).isNull()
+    }
+
+    @Test
+    fun `does not duplicate an existing sfu_id`() {
+        val proceeded = intercept(
+            pinnedSfuId = "SFU-2",
+            url = "https://video.stream-io-api.com/video/call/default/abc/join?sfu_id=SFU-1",
+        )
+
+        assertThat(proceeded.url.queryParameterValues("sfu_id")).containsExactly("SFU-1")
+    }
+
+    private fun intercept(pinnedSfuId: String?, url: String): Request {
+        lateinit var proceeded: Request
+        val client = OkHttpClient.Builder()
+            .addInterceptor(CoordinatorSfuPinInterceptor(pinnedSfuId))
+            .addInterceptor { chain ->
+                proceeded = chain.request()
+                Response.Builder()
+                    .request(proceeded)
+                    .protocol(Protocol.HTTP_1_1)
+                    .code(200)
+                    .message("OK")
+                    .body("".toResponseBody())
+                    .build()
+            }
+            .build()
+        client.newCall(
+            Request.Builder()
+                .url(url)
+                .post("{}".toRequestBody("application/json".toMediaType()))
+                .build(),
+        ).execute().close()
+        return proceeded
+    }
+}