Add the connection ID to the stream watcher onRewatch, extend tests

Author: aleksandar-apostolov

CLAUDE.md

@@ -204,21 +204,22 @@ fun stop(): Result<Unit>
 **Usage Flow:**
 1. Product SDK watches a channel: `watcher.watch(StreamCid.parse("messaging:general"))`
 2. Watcher adds CID to internal `ConcurrentHashMap<StreamCid, Unit>` registry
-3. Product SDK registers a rewatch listener: `watcher.subscribe(StreamCidRewatchListener { cids -> ... })`
+3. Product SDK registers a rewatch listener: `watcher.subscribe(StreamCidRewatchListener { cids, connectionId -> ... })`
 4. Call `watcher.start()` to begin monitoring connection state changes
-5. On `StreamConnectionState.Connected` event, watcher invokes all listeners with complete CID list
-6. Product SDK re-establishes server-side watches for each CID
+5. On `StreamConnectionState.Connected` event, watcher invokes all listeners with complete CID list AND the current connectionId
+6. Product SDK re-establishes server-side watches for each CID using the provided connectionId
 
 **Implementation Details:**
-- Thread-safe: Uses `ConcurrentHashMap` for CID registry (line 36 in `StreamCidWatcherImpl.kt`)
-- Async execution: Rewatch callbacks invoked on internal coroutine scope with `SupervisorJob + Dispatchers.Default` (line 45)
-- Error handling: Exceptions from rewatch callbacks are caught, logged, and surfaced via `StreamClientListener.onError` (lines 61-67)
+- Thread-safe: Uses `ConcurrentHashMap` for CID registry (line 52 in `StreamCidWatcherImpl.kt`)
+- Async execution: Rewatch callbacks invoked on internal coroutine scope with `SupervisorJob + Dispatchers.Default` (line 61)
+- Error handling: Exceptions from rewatch callbacks are caught, logged, and surfaced via `StreamClientListener.onError` (lines 77-82)
 - Idempotent: Multiple `watch()` calls with same CID only maintain one entry
-- Only triggers on `Connected` state when registry is non-empty (line 51)
+- Only triggers on `Connected` state when registry is non-empty (line 67)
+- Connection ID extracted from `Connected` state and passed to all listeners (line 70)
 
 **Test Coverage:**
 - Location: `stream-android-core/src/test/java/io/getstream/android/core/internal/watcher/StreamCidWatcherImplTest.kt`
-- 24 comprehensive test cases covering watch operations, lifecycle, state changes, error handling, and concurrency
+- 29 comprehensive test cases covering watch operations, lifecycle, state changes, error handling, concurrency, and connectionId verification
 - 100% instruction/branch/line coverage (verified via Kover)
 
 ## Configuration Defaults

stream-android-core/src/main/java/io/getstream/android/core/api/StreamClient.kt

@@ -292,7 +292,8 @@ public fun StreamClient(
         ),
     cidWatcher: StreamCidWatcher =
         StreamCidWatcher(
-            logProvider.taggedLogger("SCCidRewatcher"),
+            scope = scope,
+            logger = logProvider.taggedLogger("SCCidRewatcher"),
             streamRewatchSubscriptionManager =
                 StreamSubscriptionManager(
                     logger = logProvider.taggedLogger("SCRewatchSubscriptionManager")
@@ -366,6 +367,7 @@ public fun StreamClient(
         tokenManager = tokenManager,
         singleFlight = singleFlight,
         serialQueue = serialQueue,
+        cidWatcher = cidWatcher,
         connectionIdHolder = connectionIdHolder,
         logger = clientLogger,
         mutableConnectionState = mutableConnectionState,

stream-android-core/src/main/java/io/getstream/android/core/api/model/StreamCid.kt

@@ -18,25 +18,205 @@ package io.getstream.android.core.api.model
 
 import io.getstream.android.core.annotations.StreamPublishedApi
 
+/**
+ * Represents a Configuration Identifier (CID) in the Stream platform.
+ *
+ * A CID uniquely identifies a resource by combining a configuration type and a resource ID in the
+ * format `"configuration:identifier"`. This is the primary identifier used throughout the Stream
+ * SDK for referencing various resources such as channels, calls, feeds, and other configuration
+ * entities.
+ *
+ * ## Format
+ *
+ * CIDs follow the pattern: `"<configuration>:<identifier>"`
+ * - **configuration**: The resource configuration type (e.g., "messaging", "livestream", "call",
+ *   "feed")
+ * - **identifier**: The unique identifier for the resource within that configuration (e.g.,
+ *   "general", "support", "user-123")
+ *
+ * ## Examples
+ *
+ * ```kotlin
+ * // Parse from formatted string
+ * val cid1 = StreamCid.parse("messaging:general")
+ * println(cid1.configuration)  // "messaging"
+ * println(cid1.id)    // "general"
+ * println(cid1.formatted())  // "messaging:general"
+ *
+ * // Create from configuration and id
+ * val cid2 = StreamCid.fromTypeAndId("call", "video-room-1")
+ * println(cid2.formatted())  // "call:video-room-1"
+ *
+ * // Different configuration types
+ * val channelCid = StreamCid.parse("messaging:support")
+ * val callCid = StreamCid.parse("call:team-meeting")
+ * val feedCid = StreamCid.parse("feed:user-timeline")
+ *
+ * // Use in equality comparisons
+ * val cid3 = StreamCid.parse("messaging:general")
+ * println(cid1 == cid3)  // true (data class equality)
+ * ```
+ *
+ * ## Thread Safety
+ *
+ * This class is immutable and thread-safe. All instances are created via factory methods in the
+ * companion object, and the constructor is private to enforce validation.
+ *
+ * ## Validation
+ *
+ * Both [configuration] and [id] must be non-empty strings. Attempting to create a CID with empty
+ * values will throw [IllegalArgumentException].
+ *
+ * @property configuration The configuration type (non-empty, e.g., "messaging", "call", "feed")
+ * @property id The resource identifier within the configuration (non-empty, e.g., "general",
+ *   "support")
+ * @see parse Factory method for creating a CID from a formatted string
+ * @see fromTypeAndId Factory method for creating a CID from separate configuration and id
+ *   components
+ * @see formatted Returns the string representation in "configuration:id" format
+ */
 @ConsistentCopyVisibility
 @StreamPublishedApi
-public data class StreamCid private constructor(public val type: String, public val id: String) {
+public data class StreamCid
+private constructor(
+    /**
+     * The configuration type component of the CID.
+     *
+     * Common configuration types include:
+     * - `"messaging"` - Messaging channels
+     * - `"livestream"` - Live streaming channels
+     * - `"call"` - Audio/video calls
+     * - `"feed"` - Activity feeds
+     * - `"team"` - Team collaboration
+     *
+     * This value is guaranteed to be non-empty.
+     *
+     * @see id The resource identifier component
+     */
+    public val configuration: String,
+    /**
+     * The unique identifier for the resource within its [configuration].
+     *
+     * This can be any non-empty string that uniquely identifies the resource within its
+     * configuration type. Common patterns include:
+     * - User IDs: `"user-123"`
+     * - Resource names: `"general"`, `"support"`, `"random"`
+     * - Generated IDs: `"ch-abc123"`, `UUID.randomUUID().toString()`
+     *
+     * This value is guaranteed to be non-empty.
+     *
+     * @see configuration The configuration type component
+     */
+    public val id: String,
+) {
 
     public companion object {
 
+        /**
+         * Parses a formatted CID string into a [StreamCid] instance.
+         *
+         * This method parses a CID from its standard string representation `"configuration:id"` and
+         * creates a [StreamCid] instance with the extracted configuration and id components.
+         *
+         * ## Format Requirements
+         *
+         * The input string must:
+         * - Be non-empty
+         * - Contain exactly one colon (`:`) separator
+         * - Have non-empty configuration and id components on both sides of the colon
+         *
+         * ## Examples
+         *
+         * ```kotlin
+         * // Valid CIDs
+         * val cid1 = StreamCid.parse("messaging:general")
+         * val cid2 = StreamCid.parse("livestream:sports-2023")
+         * val cid3 = StreamCid.parse("call:team-meeting")
+         *
+         * // Invalid CIDs (will throw IllegalArgumentException)
+         * StreamCid.parse("")  // Empty string
+         * StreamCid.parse("messaging")  // Missing colon separator
+         * StreamCid.parse("messaging:")  // Empty id
+         * StreamCid.parse(":general")  // Empty configuration
+         * StreamCid.parse("messaging:general:extra")  // Too many colons
+         * ```
+         *
+         * @param cid The formatted CID string in "configuration:id" format
+         * @return A new [StreamCid] instance with the parsed configuration and id
+         * @throws IllegalArgumentException if the input is empty, doesn't contain exactly one
+         *   colon, or has empty configuration/id components
+         * @see fromTypeAndId Alternative factory method for creating from separate components
+         * @see formatted Returns the inverse transformation (CID → String)
+         */
         public fun parse(cid: String): StreamCid {
-            require(cid.isNotEmpty())
+            require(cid.isNotEmpty()) { "CID string cannot be empty" }
             val split = cid.split(":")
-            require(split.size == 2)
-            return StreamCid(split[0], split[1])
+            require(split.size == 2) {
+                "CID must be in format 'configuration:id' with exactly one colon separator, got: '$cid'"
+            }
+            return fromTypeAndId(split[0], split[1])
         }
 
+        /**
+         * Creates a [StreamCid] from separate configuration and id components.
+         *
+         * This method constructs a CID directly from its constituent parts without requiring a
+         * pre-formatted string. This is useful when you already have the configuration and id as
+         * separate values.
+         *
+         * ## Examples
+         *
+         * ```kotlin
+         * // Create CID from components
+         * val cid = StreamCid.fromTypeAndId(type = "messaging", id = "general")
+         * println(cid.formatted())  // "messaging:general"
+         *
+         * // Useful when building CIDs dynamically
+         * val configurationType = "call"
+         * val resourceId = UUID.randomUUID().toString()
+         * val dynamicCid = StreamCid.fromTypeAndId(configurationType, resourceId)
+         * ```
+         *
+         * @param type The configuration type (must be non-empty)
+         * @param id The resource identifier (must be non-empty)
+         * @return A new [StreamCid] instance with the given configuration and id
+         * @throws IllegalArgumentException if either [type] or [id] is empty
+         * @see parse Alternative factory method for parsing from formatted strings
+         */
         public fun fromTypeAndId(type: String, id: String): StreamCid {
-            require(type.isNotEmpty())
-            require(id.isNotEmpty())
+            require(type.isNotEmpty()) { "Configuration type cannot be empty" }
+            require(id.isNotEmpty()) { "Resource id cannot be empty" }
             return StreamCid(type, id)
         }
     }
 
-    public fun formatted(): String = "$type:$id"
+    /**
+     * Returns the formatted string representation of this CID.
+     *
+     * Converts the CID back to its canonical string format `"configuration:id"`. This is the
+     * inverse operation of [parse].
+     *
+     * ## Examples
+     *
+     * ```kotlin
+     * val cid = StreamCid.fromTypeAndId("messaging", "general")
+     * println(cid.formatted())  // "messaging:general"
+     *
+     * // Round-trip parsing
+     * val original = "livestream:sports"
+     * val parsed = StreamCid.parse(original)
+     * val formatted = parsed.formatted()
+     * println(original == formatted)  // true
+     * ```
+     *
+     * ## Use Cases
+     * - Logging and debugging: Display CIDs in human-readable format
+     * - API requests: Send CID as string parameter to backend
+     * - Serialization: Convert CID to JSON/XML string representation
+     * - UI display: Show resource identifiers to users
+     *
+     * @return The CID in "configuration:id" format
+     * @see parse The inverse operation that creates a CID from this string format
+     */
+    public fun formatted(): String = "$configuration:$id"
 }

stream-android-core/src/main/java/io/getstream/android/core/api/watcher/StreamCidRewatchListener.kt

@@ -52,11 +52,11 @@ import io.getstream.android.core.api.model.StreamCid
  *
  * // Register a rewatch listener
  * watcher.subscribe(
- *     listener = StreamCidRewatchListener { cids ->
- *         logger.i { "Re-watching ${cids.size} channels" }
+ *     listener = StreamCidRewatchListener { cids, connectionId ->
+ *         logger.i { "Re-watching ${cids.size} channels on connection $connectionId" }
  *         cids.forEach { cid ->
  *             // Re-establish server-side watch for each CID
- *             channelClient.watch(cid)
+ *             channelClient.watch(cid, connectionId)
  *         }
  *     }
  * )
@@ -85,13 +85,15 @@ public fun interface StreamCidRewatchListener {
      * - The list represents a snapshot of the watch registry at the time of invocation
      * - Modifications to the list do not affect the internal registry
      * - Multiple calls may occur for the same set of CIDs during reconnection scenarios
+     * - The connectionId reflects the current active connection at the moment of invocation
      *
      * ## Error Handling
      *
      * Exceptions thrown by this method are caught, logged, and surfaced via error callbacks. The
      * failure of one listener does not prevent other listeners from being notified.
      *
      * @param list A non-empty list of [StreamCid]s that require re-watching
+     * @param connectionId The connection ID of the current active WebSocket connection
      */
-    public fun onRewatch(list: List<StreamCid>)
+    public fun onRewatch(list: List<StreamCid>, connectionId: String)
 }

stream-android-core/src/main/java/io/getstream/android/core/api/watcher/StreamCidWatcher.kt

@@ -24,6 +24,7 @@ import io.getstream.android.core.api.socket.listeners.StreamClientListener
 import io.getstream.android.core.api.subscribe.StreamObservable
 import io.getstream.android.core.api.subscribe.StreamSubscriptionManager
 import io.getstream.android.core.internal.watcher.StreamCidWatcherImpl
+import kotlinx.coroutines.CoroutineScope
 
 /**
  * Manages a registry of watched resources (channels, conversations, streams) and automatically
@@ -136,13 +137,85 @@ public interface StreamCidWatcher :
     public fun clear(): Result<Unit>
 }
 
+/**
+ * Creates a new [StreamCidWatcher] instance with the provided dependencies.
+ *
+ * This factory method instantiates the default implementation ([StreamCidWatcherImpl]) and wires
+ * all required dependencies for monitoring connection state changes and triggering rewatch
+ * callbacks.
+ *
+ * ## Parameters
+ * - **scope**: The [CoroutineScope] used for launching async operations (rewatch callbacks). This
+ *   scope should typically use `SupervisorJob + Dispatchers.Default` to ensure callback failures
+ *   don't cancel the scope and to avoid blocking the main thread. The scope is NOT cancelled when
+ *   [StreamCidWatcher.stop] is called, allowing the component to be restarted.
+ * - **logger**: A [StreamLogger] instance for diagnostic output, tagged appropriately (e.g.,
+ *   "SCCidWatcher"). Used for logging state changes, rewatch events, and errors.
+ * - **streamRewatchSubscriptionManager**: Manages subscriptions to [StreamCidRewatchListener]. This
+ *   manager handles the list of listeners that will be invoked when connection state changes
+ *   require re-watching CIDs.
+ * - **streamClientSubscriptionManager**: Manages subscriptions to [StreamClientListener]. This is
+ *   used to subscribe to connection state changes (via [StreamClientListener.onState]) and to
+ *   surface errors (via [StreamClientListener.onError]) when rewatch callbacks fail.
+ *
+ * ## Lifecycle
+ *
+ * The returned watcher is created in a stopped state. You must call [StreamCidWatcher.start] to
+ * begin monitoring connection state changes. The component can be started and stopped multiple
+ * times throughout its lifetime.
+ *
+ * ## Usage Example
+ *
+ * ```kotlin
+ * val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
+ * val logger = StreamLogger.getLogger("MyApp.CidWatcher")
+ *
+ * val rewatchManager = StreamSubscriptionManager<StreamCidRewatchListener>(
+ *     logger = logger.withTag("RewatchSubscriptions")
+ * )
+ * val clientManager = StreamSubscriptionManager<StreamClientListener>(
+ *     logger = logger.withTag("ClientSubscriptions")
+ * )
+ *
+ * val watcher = StreamCidWatcher(
+ *     scope = scope,
+ *     logger = logger,
+ *     streamRewatchSubscriptionManager = rewatchManager,
+ *     streamClientSubscriptionManager = clientManager
+ * )
+ *
+ * // Register rewatch listener
+ * watcher.subscribe(StreamCidRewatchListener { cids, connectionId ->
+ *     println("Re-watching ${cids.size} channels on connection $connectionId")
+ *     // Re-establish server-side watches...
+ * })
+ *
+ * // Start monitoring
+ * watcher.start()
+ *
+ * // Watch channels
+ * watcher.watch(StreamCid.parse("messaging:general"))
+ * watcher.watch(StreamCid.parse("livestream:sports"))
+ * ```
+ *
+ * @param scope Coroutine scope for async operations (rewatch callback invocations)
+ * @param logger Logger for diagnostic output and error reporting
+ * @param streamRewatchSubscriptionManager Subscription manager for rewatch listeners
+ * @param streamClientSubscriptionManager Subscription manager for client state listeners
+ * @return A new [StreamCidWatcher] instance (implementation: [StreamCidWatcherImpl])
+ * @see StreamCidWatcher The interface contract
+ * @see StreamCidWatcherImpl The concrete implementation
+ * @see StreamCidRewatchListener Callback interface for rewatch notifications
+ */
 @StreamInternalApi
 public fun StreamCidWatcher(
+    scope: CoroutineScope,
     logger: StreamLogger,
     streamRewatchSubscriptionManager: StreamSubscriptionManager<StreamCidRewatchListener>,
     streamClientSubscriptionManager: StreamSubscriptionManager<StreamClientListener>,
 ): StreamCidWatcher =
     StreamCidWatcherImpl(
+        scope = scope,
         logger = logger,
         rewatchSubscriptions = streamRewatchSubscriptionManager,
         clientSubscriptions = streamClientSubscriptionManager,