Flaro/Packages/HaloKit/Sources/HaloPlayback/Prefetch/StreamCache.swift

import Foundation
import OSLog

/// Holt die Schnipsel eines Streams, bevor der Player sie verlangt.
///
/// **Warum es das gibt.** AVPlayer entscheidet selbst, wie weit er puffert,
/// und bei HLS lässt Apple uns nicht dazwischen. Wir müssen aber auch nicht:
/// steht zwischen Player und Server ein eigener Speicher, kann der so weit
/// vorauslaufen, wie er will. Der Player holt dann von der Platte, und ob die
/// Leitung gerade steht, merkt er nicht.
///
/// Genau das ist der Unterschied zwischen „Puffer" und „Vorrat": ein Puffer
/// ist das, was der Player im Arbeitsspeicher hält, und den bestimmt Apple.
/// Der Vorrat hier ist das, was **wir** schon haben, und den bestimmen wir.
///
/// **Streng der Reihe nach.** Der Server erzeugt die Schnipsel gerade erst.
/// Wer Nummer 40 verlangt, während er bei 12 ist, sieht für Jellyfin aus wie
/// jemand, der gesprungen ist — und dann beginnt das Umrechnen von vorn, an
/// der falschen Stelle. Deshalb wird nie übersprungen, sondern nur zügiger
/// geholt, als abgespielt wird.
///
/// **Der Vorsprung ist begrenzt**, und zwar in Sekunden Film, nicht in
/// Schnipseln: die Zahl soll dasselbe bedeuten, egal wie der Server schneidet.
public actor StreamCache {
    /// Ein Schnipsel — Daten **und** wofür sie sich ausgeben.
    public struct Segment: Sendable {
        public let data: Data
        public let contentType: String
    }

    /// Wie weit vorausgelaufen wird.
    public struct Policy: Sendable {
        /// Wie viel Film im Voraus geholt werden soll.
        public let lookahead: Duration
        /// Wie viel Platz der Vorrat höchstens belegen darf.
        public let diskLimit: Int

        public init(lookahead: Duration = .seconds(300), diskLimit: Int = 768 * 1_024 * 1_024) {
            self.lookahead = lookahead
            self.diskLimit = diskLimit
        }
    }

    private let policy: Policy
    private let session: URLSession
    private let directory: URL
    private let logger = Logger(subsystem: "app.halo", category: "StreamCache")

    /// Woher die Schnipsel wirklich kommen.
    private var upstream: URL?
    private var headers: [String: String] = [:]
    /// Die Liste, in der Reihenfolge des Servers.
    private var segments: [String] = []
    private var targetDuration: Duration = .seconds(6)

    /// Was schon auf der Platte liegt — Adresse auf Datei **und Sorte**.
    ///
    /// **Die Sorte gehört dazu.** Ein Schnipsel ist nicht zwangsläufig Bild:
    /// die Untertitelspur liefert WebVTT, der Ton kann AAC sein. Wer die
    /// Sorte wegwirft und später etwas behauptet, liefert dem Player einen
    /// Text als Video — der gibt ihn dem falschen Parser, findet keine Box,
    /// und es entsteht lautlos gar nichts.
    private var stored: [String: (ort: URL, type: String)] = [:]
    private var storedBytes = 0
    /// Welchen Schnipsel der Player zuletzt verlangt hat.
    private var lastRequested = 0
    /// Bis wohin geholt wurde.
    private var fetched = 0

    private var prefetchTask: Task<Void, Never>?
    /// Ob gerade eine Anfrage des Players offen ist.
    private var isServingPlayer = false

    public init(policy: Policy = Policy()) {
        self.policy = policy

        let configuration = URLSessionConfiguration.ephemeral
        // Der Vorrat ist unser Zwischenspeicher; ein zweiter darunter
        // verdoppelte nur den Platzbedarf.
        configuration.urlCache = nil
        configuration.httpMaximumConnectionsPerHost = 4
        configuration.timeoutIntervalForRequest = 30
        self.session = URLSession(configuration: configuration)

        self.directory = FileManager.default.temporaryDirectory
            .appending(path: "halo-stream-\(UUID().uuidString)", directoryHint: .isDirectory)
        try? FileManager.default.createDirectory(at: directory, withIntermediateDirectories: true)
    }

    // MARK: - Einrichten

    /// Nimmt die Abspielliste entgegen und beginnt vorauszulaufen.
    ///
    /// **Alles wird auf eine vollständige Adresse umgeschrieben.** Relative
    /// Angaben gelten je nach Liste gegen eine andere Grundadresse — die
    /// Untertitelspur liegt woanders als das Bild. Wer das erst beim Abholen
    /// auflöst, löst es irgendwann gegen die falsche auf. Hier steht die
    /// Grundadresse noch fest.
    public func start(upstream url: URL, headers: [String: String] = [:]) async throws -> String {
        self.headers = headers
        let liste = try await load(url)
        return rewrite(liste, base: url, adopting: !liste.isMaster)
    }

    /// Eine Liste, die der Player eine Ebene tiefer angefordert hat.
    ///
    /// `prefetch` sagt, ob dahinter Bild liegt. Bei einer Nebenspur — also
    /// Untertiteln — steht es auf `false`: sonst lüde der Vorrat Untertitel
    /// statt Bild, und der Film bliebe stehen, während die Kachel meldet, es
    /// sei reichlich geladen.
    public func playlist(at address: String, prefetch: Bool) async -> String? {
        guard let url = URL(string: address), let liste = try? await load(url) else { return nil }
        return rewrite(liste, base: url, adopting: prefetch && !liste.isMaster)
    }

    private func rewrite(_ liste: MediaPlaylist, base: URL, adopting: Bool) -> String {
        if adopting { adopt(liste, base: base) }

        return liste.rewritingReferences { adresse, art in
            let absolut = URL(string: adresse, relativeTo: base)?.absoluteString ?? adresse
            switch art {
            case .level:
                return LocalStreamServer.localPlaylistPath(for: absolut, prefetch: true)
            case .rendition:
                return LocalStreamServer.localPlaylistPath(for: absolut, prefetch: false)
            case .map, .segment:
                return LocalStreamServer.localPath(for: absolut)
            }
        }
    }

    /// Übernimmt eine Liste von Schnipseln als die, der wir vorauslaufen.
    private func adopt(_ liste: MediaPlaylist, base: URL) {
        // **Nur bei einem echten Wechsel von vorn.** Der Player holt dieselbe
        // Liste immer wieder — nach einem Sprung, beim Stufenwechsel, von
        // sich aus. Jedes Mal zurückzusetzen hiesse, den Vorrat jedes Mal zu
        // vergessen und von Schnipsel null neu zu zählen.
        let neueQuelle = upstream != base
        upstream = base
        segments = liste.segments.map {
            URL(string: $0, relativeTo: base)?.absoluteString ?? $0
        }
        if let dauer = liste.targetDuration { targetDuration = dauer }

        if neueQuelle {
            lastRequested = 0
            fetched = 0
        }
        beginPrefetching()
    }

    private func load(_ url: URL) async throws -> MediaPlaylist {
        var request = URLRequest(url: url)
        for (name, wert) in headers { request.setValue(wert, forHTTPHeaderField: name) }
        let (daten, _) = try await session.data(for: request)
        return MediaPlaylist(String(decoding: daten, as: UTF8.self))
    }

    // MARK: - Ausliefern

    /// Die Daten zu einer Adresse — aus dem Vorrat, sonst frisch geholt.
    public func data(for address: String) async -> Segment? {
        if let index = segments.firstIndex(of: address) {
            lastRequested = max(lastRequested, index)
            trimBehind()
        }

        if let eintrag = stored[address], let daten = try? Data(contentsOf: eintrag.ort) {
            return Segment(data: daten, contentType: eintrag.type)
        }

        // **Der Player hat Vorrang.** Fehlt ihm etwas, wartet er darauf — und
        // der Vorrat darf ihm dabei nicht die Leitung wegnehmen. Rechnet der
        // Server gerade um, bedient er ohnehin einen Schnipsel nach dem
        // anderen: zwei Anfragen gleichzeitig machen die dringende langsamer,
        // nicht schneller. Genau so kam es zu „Stream abgerissen", sobald das
        // Umrechnen teuer wurde.
        isServingPlayer = true
        defer { isServingPlayer = false }
        return await fetch(address)
    }

    // MARK: - Vorauslaufen

    private func beginPrefetching() {
        prefetchTask?.cancel()
        prefetchTask = Task { [weak self] in
            while !Task.isCancelled {
                guard let self else { return }
                let weiter = await self.prefetchNext()
                if !weiter { try? await Task.sleep(for: .milliseconds(500)) }
            }
        }
    }

    /// Holt den nächsten Schnipsel, falls einer fehlt und Platz ist.
    ///
    /// Gibt `false` zurück, wenn gerade nichts zu tun ist — dann wartet die
    /// Schleife, statt zu drehen.
    private func prefetchNext() async -> Bool {
        // Solange der Player wartet, wird nicht vorgeladen.
        guard !isServingPlayer else { return false }

        // **Ist der Player vorausgesprungen, dort weitermachen.**
        //
        // Sonst holt der Vorrat brav weiter Schnipsel hinter der
        // Abspielstelle — und der eine, auf den gerade gewartet wird, kommt
        // als letzter. Das ist genau der Fall, in dem es nach einem Sprung
        // lange lädt, obwohl vorn alles bereitliegt.
        if lastRequested > fetched { fetched = lastRequested }
        guard fetched < segments.count else { return false }

        // Der Vorsprung in Schnipseln, aus dem Vorsprung in Sekunden.
        let vorsprung = Int((policy.lookahead.seconds / max(targetDuration.seconds, 1)).rounded())
        guard fetched - lastRequested < vorsprung else { return false }
        guard storedBytes < policy.diskLimit else { return false }

        let pfad = segments[fetched]
        if stored[pfad] != nil {
            fetched += 1
            return true
        }

        guard await fetch(pfad) != nil else {
            // Nicht weiterzählen: beim nächsten Durchgang noch einmal
            // versuchen. Ein Schnipsel, den der Server noch nicht fertig hat,
            // ist kein Fehler — nur ein „gleich".
            return false
        }
        fetched += 1
        return true
    }

    @discardableResult
    private func fetch(_ address: String) async -> Segment? {
        guard let ziel = URL(string: address) else { return nil }

        do {
            var anfrage = URLRequest(url: ziel)
            for (name, wert) in headers { anfrage.setValue(wert, forHTTPHeaderField: name) }
            let (daten, antwort) = try await session.data(for: anfrage)
            let http = antwort as? HTTPURLResponse
            guard http?.statusCode ?? 200 < 400 else { return nil }

            // **Die Sorte kommt von der Gegenstelle, nicht von uns.** Sie
            // weiss, was sie geschickt hat; wir wüssten es nur aus der
            // Endung, und die kann fehlen oder lügen. Erst wenn sie nichts
            // sagt, raten wir anhand der Adresse.
            let sorte = http?.value(forHTTPHeaderField: "Content-Type")
                ?? Self.guessedType(for: address)
            store(daten, type: sorte, at: address)
            return Segment(data: daten, contentType: sorte)
        } catch {
            return nil
        }
    }

    /// Die Sorte aus der Endung — der Notnagel.
    static func guessedType(for address: String) -> String {
        let pfad = URL(string: address)?.path.lowercased() ?? address.lowercased()
        if pfad.hasSuffix(".vtt") { return "text/vtt" }
        if pfad.hasSuffix(".ts") { return "video/MP2T" }
        if pfad.hasSuffix(".aac") { return "audio/aac" }
        if pfad.hasSuffix(".m3u8") { return "application/vnd.apple.mpegurl" }
        return "video/mp4"
    }

    private func store(_ daten: Data, type: String, at path: String) {
        let ort = directory.appending(path: UUID().uuidString)
        guard (try? daten.write(to: ort, options: .atomic)) != nil else { return }
        stored[path] = (ort, type)
        storedBytes += daten.count
    }

    /// Wirft weg, was hinter der Abspielstelle liegt.
    ///
    /// **Nicht alles dahinter**, sondern mit Abstand: ein kurzer Rücksprung
    /// ist üblich, und ihn jedes Mal neu zu holen wäre teurer als die paar
    /// Megabyte.
    private func trimBehind() {
        let grenze = lastRequested - 12
        guard grenze > 0 else { return }

        for index in 0..<min(grenze, segments.count) {
            let pfad = segments[index]
            guard let eintrag = stored.removeValue(forKey: pfad) else { continue }
            let größe = (try? FileManager.default.attributesOfItem(atPath: eintrag.ort.path))
                .flatMap { $0[.size] as? Int } ?? 0
            storedBytes = max(0, storedBytes - größe)
            try? FileManager.default.removeItem(at: eintrag.ort)
        }
    }

    // MARK: - Abräumen

    public func teardown() {
        prefetchTask?.cancel()
        prefetchTask = nil
        stored.removeAll()
        storedBytes = 0
        try? FileManager.default.removeItem(at: directory)
    }

    /// Wie viel Film im Voraus bereitliegt — für die Anzeige.
    public var bufferedAhead: Duration {
        let voraus = max(0, fetched - lastRequested)
        return .seconds(Double(voraus) * targetDuration.seconds)
    }
}