tv-anarchy/Sources/TVAnarchyCore/StreamabilityMonitor.swift

import Foundation
import Observation

/// Occasional ping + bandwidth probes while Stream sourcing is selected.
@Observable
@MainActor
public final class StreamabilityMonitor {
    public private(set) var sample = StreamabilitySample.idle
    public private(set) var isRunning = false

    public var episodeDuration: Double?
    public var fileBytes: Int64?

    private weak var player: PlayerController?
    private var task: Task<Void, Never>?
    private let interval: TimeInterval = 45

    public init() {}

    public func attach(player: PlayerController) { self.player = player }

    public func syncFromPlayer() {
        guard let player else { return }
        episodeDuration = player.activeSnapshot.status.duration
    }

    /// Start or stop based on whether Stream mode is active.
    public func refreshActivation() {
        guard shouldMonitor else {
            stop()
            sample = .idle
            return
        }
        start()
    }

    public func start() {
        guard task == nil else { return }
        isRunning = true
        sample = .checking
        task = Task { [weak self] in
            guard let self else { return }
            await self.probeLoop()
        }
    }

    public func stop() {
        task?.cancel()
        task = nil
        isRunning = false
    }

    private var shouldMonitor: Bool {
        guard let player else { return false }
        return player.playbackMode == .stream
    }

    private func probeLoop() async {
        while !Task.isCancelled {
            await tick()
            try? await Task.sleep(for: .seconds(interval))
        }
    }

    private func tick() async {
        guard shouldMonitor else {
            stop()
            sample = .idle
            return
        }
        sample = .checking
        let policy = DevicesConfig.localStreamPolicy()
        let episode = episodeDuration ?? Double(StreamPolicy.typicalEpisodeSeconds)
        let buffer = policy.effectiveBufferSeconds(episodeDuration: episodeDuration)
        let required = StreamabilityAssessor.requiredKBps(
            bufferSeconds: buffer, episodeSeconds: episode, fileBytes: fileBytes)
        let hosts = DevicesConfig.storageSSHEndpoints()
        let bytes = fileBytes
        let result = await Task.detached {
            StreamabilityProbe.run(hosts: hosts, episodeSeconds: episode,
                                   bufferSeconds: buffer, fileBytes: bytes)
        }.value
        let level = StreamabilityAssessor.assess(
            pingMs: result.pingMs, bandwidthKBps: result.bandwidthKBps, requiredKBps: required)
        let detail = StreamabilityAssessor.detail(
            level: level, pingMs: result.pingMs, bandwidthKBps: result.bandwidthKBps,
            requiredKBps: required, bufferSeconds: buffer)
        sample = StreamabilitySample(
            level: level, pingMs: result.pingMs, bandwidthKBps: result.bandwidthKBps,
            requiredKBps: required, bufferSeconds: buffer, checkedAt: Date(), detail: detail)
    }
}