Stats v2 (#806)

Author: ipavlidakis

Sources/StreamVideo/Controllers/CallController.swift

@@ -693,7 +693,7 @@ class CallController: @unchecked Sendable {
     private func observeStatsReporterUpdates() async {
         await webRTCCoordinator
             .stateAdapter
-            .$statsReporter
+            .$statsAdapter
             .compactMap { $0 }
             .sink { [weak disposableBag, weak self] statsReporter in
                 statsReporter

Sources/StreamVideo/Models/CallStatsReport.swift

@@ -22,6 +22,8 @@ public struct CallStatsReport: Sendable {
     public let participantsStats: ParticipantsStats
     /// The timestamp when the call stats report was generated.
     public let timestamp: Double
+
+    let trackToKindMap: [String: TrackType]
 }
 
 /// A struct representing statistics for participants in the call.

Sources/StreamVideo/Utils/AnyEncodable.swift

@@ -0,0 +1,91 @@
+//
+// Copyright © 2025 Stream.io Inc. All rights reserved.
+//
+
+import Foundation
+
+/// A type-erased `Encodable` value with optional equality support.
+///
+/// `AnyEncodable` enables encoding of values whose concrete type is
+/// unknown at compile time. It captures the encoding logic using a
+/// closure and optionally supports equality if initialized with an
+/// `Equatable` value.
+struct AnyEncodable: Encodable, Equatable, @unchecked Sendable {
+    /// Encodes the value to an `Encoder`. This closure is assigned during
+    /// initialization and stores the encoding logic for the wrapped value.
+    private let _encode: (Encoder) throws -> Void
+    /// Compares the stored value against another. If the original value was
+    /// `Equatable`, this closure performs type-safe equality comparison.
+    private let _isEqual: (Any) -> Bool
+
+    /// The underlying value, which must conform to `Encodable`.
+    ///
+    /// If equality support is desired, the value must also conform to
+    /// `Equatable` and be initialized using the corresponding initializer.
+    let value: Any
+
+    /// Initializes from an existing `AnyEncodable` to avoid re-wrapping.
+    ///
+    /// This initializer allows preserving the original encoding and equality
+    /// closures when passing or storing `AnyEncodable` values.
+    init(_ value: AnyEncodable) {
+        self = value
+    }
+
+    /// Initializes with an `Encodable` value, without equality support.
+    ///
+    /// The provided value is encoded using a closure captured at
+    /// initialization time. Equality comparison will always return false.
+    ///
+    /// - Parameter value: The value to wrap.
+    init<T: Encodable>(_ value: T) {
+        _encode = { encoder in
+            try value.encode(to: encoder)
+        }
+        _isEqual = { _ in false }
+
+        self.value = value
+    }
+
+    /// Initializes with a value that conforms to both `Encodable` and
+    /// `Equatable`, enabling equality support.
+    ///
+    /// This version stores a type-safe equality closure that will return
+    /// true only if the other value has the same type and is equal.
+    ///
+    /// - Parameter value: The value to wrap.
+    init<T: Encodable & Equatable>(_ value: T) {
+        _encode = { encoder in
+            try value.encode(to: encoder)
+        }
+        _isEqual = { other in
+            guard let other = other as? T else {
+                return false
+            }
+            return other == value
+        }
+
+        self.value = value
+    }
+
+    /// Encodes the stored value using the provided encoder.
+    ///
+    /// - Parameter encoder: The encoder to write data to.
+    /// - Throws: An error if the encoding process fails.
+    func encode(to encoder: Encoder) throws {
+        try _encode(encoder)
+    }
+
+    /// Compares two `AnyEncodable` instances for equality.
+    ///
+    /// This uses the stored equality closure. If the values were not
+    /// initialized with `Equatable` conformance, this always returns false.
+    ///
+    /// - Parameters:
+    ///   - lhs: The first `AnyEncodable` value.
+    ///   - rhs: The second `AnyEncodable` value.
+    /// - Returns: `true` if the wrapped values are equal.
+    static func == (lhs: AnyEncodable, rhs: AnyEncodable) -> Bool {
+        lhs._isEqual(rhs.value)
+    }
+}

Sources/StreamVideo/Utils/AudioSession/StreamAudioSession.swift

@@ -409,6 +409,27 @@ final class StreamAudioSession: @unchecked Sendable, ObservableObject {
     }
 }
 
+extension StreamAudioSession: Encodable {
+    // MARK: - Codable
+
+    enum CodingKeys: String, CodingKey {
+        case speakerOn
+        case device
+        case deviceIsExternal = "device.isExternal"
+        case deviceIsSpeaker = "device.isSpeaker"
+        case deviceIsReceiver = "device.isReceiver"
+    }
+
+    func encode(to encoder: Encoder) throws {
+        var container = encoder.container(keyedBy: CodingKeys.self)
+        try container.encode(activeCallSettings.speakerOn, forKey: .speakerOn)
+        try container.encode("\(audioSession.currentRoute)", forKey: .device)
+        try container.encode(audioSession.currentRoute.isExternal, forKey: .deviceIsExternal)
+        try container.encode(audioSession.currentRoute.isSpeaker, forKey: .deviceIsSpeaker)
+        try container.encode(audioSession.currentRoute.isReceiver, forKey: .deviceIsReceiver)
+    }
+}
+
 /// A key for dependency injection of an `AudioSessionProtocol` instance
 /// that represents the active call audio session.
 extension StreamAudioSession: InjectionKey {

Sources/StreamVideo/Utils/ConsumableBucket/ConsumableBucket.swift

@@ -0,0 +1,144 @@
+//
+// Copyright © 2025 Stream.io Inc. All rights reserved.
+//
+
+import Combine
+import Foundation
+
+/// A thread-safe buffer that collects items for later consumption.
+///
+/// Supports on-demand appending or reactive observation from a
+/// publisher source. Items can be consumed in batch, with optional
+/// flushing of the internal buffer.
+final class ConsumableBucket<Element>: @unchecked Sendable {
+
+    /// Represents the type of bucket: on-demand or publisher-driven.
+    enum BucketType {
+        /// A manually managed bucket where items are appended directly.
+        case onDemand
+        /// A bucket that observes and collects items from a publisher.
+        case observer(AnyPublisher<Element, Never>)
+    }
+
+    private let queue = UnfairQueue()
+
+    private var items: [Element] = []
+    private var cancellable: AnyCancellable?
+
+    /// Initializes the bucket with a transformed publisher source.
+    ///
+    /// - Parameters:
+    ///   - source: A publisher of source values.
+    ///   - transformer: A transformer to map source to bucket elements.
+    convenience init<Source, Transformer: ConsumableBucketItemTransformer>(
+        _ source: AnyPublisher<Source, Never>,
+        transformer: Transformer
+    ) where Transformer.Input == Source, Transformer.Output == Element {
+        self.init(
+            .observer(
+                source
+                    .compactMap { transformer.transform($0) }
+                    .eraseToAnyPublisher()
+            )
+        )
+    }
+
+    /// Initializes the bucket with a transformed publisher source and
+    /// optionally removes duplicates.
+    ///
+    /// - Parameters:
+    ///   - source: A publisher of equatable source values.
+    ///   - transformer: A transformer to map source to bucket elements.
+    ///   - removeDuplicates: Whether to remove duplicate values.
+    convenience init<Source: Equatable, Transformer: ConsumableBucketItemTransformer>(
+        _ source: AnyPublisher<Source, Never>,
+        transformer: Transformer,
+        removeDuplicates: Bool
+    ) where Transformer.Input == Source, Transformer.Output == Element {
+        if removeDuplicates {
+            self.init(
+                .observer(
+                    source
+                        .removeDuplicates()
+                        .compactMap { transformer.transform($0) }
+                        .eraseToAnyPublisher()
+                )
+            )
+        } else {
+            self.init(source, transformer: transformer)
+        }
+    }
+
+    /// Initializes the bucket with a publisher of bucket elements.
+    ///
+    /// - Parameter source: The publisher emitting elements to store.
+    convenience init(
+        _ source: AnyPublisher<Element, Never>
+    ) {
+        self.init(.observer(source))
+    }
+
+    /// Initializes the bucket for manual on-demand use.
+    convenience init() {
+        self.init(.onDemand)
+    }
+
+    private init(
+        _ bucketType: BucketType
+    ) {
+        switch bucketType {
+        case .onDemand:
+            break
+        case let .observer(publisher):
+            cancellable = publisher.sink { [weak self] in self?.process($0) }
+        }
+    }
+
+    /// Appends a single element to the bucket.
+    ///
+    /// - Parameter element: The item to be appended.
+    func append(_ element: Element) {
+        queue.sync { [weak self] in self?.items.append(element) }
+    }
+
+    /// Returns the items in the bucket.
+    ///
+    /// - Parameter flush: Whether to clear the buffer after returning.
+    /// - Returns: An array of stored elements.
+    func consume(flush: Bool = false) -> [Element] {
+        if flush {
+            return queue.sync {
+                let result = items
+                items = []
+                return result
+            }
+        } else {
+            return queue.sync { items }
+        }
+    }
+
+    /// Inserts elements at a specific position in the buffer.
+    ///
+    /// - Parameters:
+    ///   - items: The elements to insert.
+    ///   - index: The position at which to insert them.
+    func insert(_ items: [Element], at index: Int) {
+        guard !items.isEmpty else {
+            return
+        }
+        queue.sync {
+            self.items.insert(contentsOf: items, at: index)
+        }
+    }
+
+    // MARK: - Private Helpers
+
+    /// Processes an incoming item from the observed publisher.
+    ///
+    /// - Parameter newItem: The new item to append.
+    private func process(_ newItem: Element) {
+        queue.sync {
+            self.items.append(newItem)
+        }
+    }
+}

Sources/StreamVideo/Utils/ConsumableBucket/ConsumableBucketItemTransformer.swift

@@ -0,0 +1,25 @@
+//
+// Copyright © 2025 Stream.io Inc. All rights reserved.
+//
+
+import Foundation
+
+/// A transformer used by `ConsumableBucket` to map source values to
+/// consumable elements.
+///
+/// This protocol allows a bucket to observe one data type while
+/// storing another, through a transformation step. Implementors define
+/// the transformation logic between `Input` and `Output`.
+protocol ConsumableBucketItemTransformer {
+
+    /// The input type received from the upstream publisher.
+    associatedtype Input
+    /// The output type to be stored in the `ConsumableBucket`.
+    associatedtype Output
+
+    /// Transforms an input into a bucket-storable output value.
+    ///
+    /// - Parameter input: The incoming value to transform.
+    /// - Returns: The transformed output.
+    func transform(_ input: Input) -> Output
+}

Sources/StreamVideo/Utils/Extensions/Foundation/Date+millisecondsSince1970.swift

@@ -0,0 +1,16 @@
+//
+// Copyright © 2025 Stream.io Inc. All rights reserved.
+//
+
+import Foundation
+
+extension Date {
+    /// The number of milliseconds since 1970-01-01 00:00:00 UTC.
+    ///
+    /// This converts the date's time interval since the Unix epoch into a
+    /// rounded 64-bit integer value in milliseconds. Useful for precise
+    /// timestamping or encoding dates for transport in APIs.
+    var millisecondsSince1970: Int64 {
+        Int64((timeIntervalSince1970 * 1000.0).rounded())
+    }
+}