Flaro/server/src/Flaro.Api/Streamkoordinator.cs

using System.Collections.Concurrent;
using System.Diagnostics;
using Flaro.Core.Delivery;
using Flaro.Core.Library;
using Flaro.Data;
using Flaro.Media;
using Microsoft.EntityFrameworkCore;

namespace Flaro.Api;

/// <summary>Eine laufende Wiedergabe mit ihrem ffmpeg dahinter.</summary>
///
/// <param name="Ab">
/// Ab welcher Sekunde des Films diese Sitzung beginnt. <b>Die Zeitachse des
/// Spielers fängt bei null an</b>, nicht beim Film — wer ab Minute 30
/// einsteigt, dessen Spieler steht bei 0, während der Film bei 1800 steht.
/// Ohne diesen Wert ginge jeder gemeldete Stand um eine halbe Stunde daneben.
/// </param>
public sealed record Stromsitzung(
    Guid Kennung,
    Guid DateiId,
    Guid Benutzer,
    string Ordner,
    double Ab,
    int? Hoehe,
    int? Tonspur,
    Stromart Art,
    DateTimeOffset Begonnen)
{
    /// <summary>Wann zuletzt ein Segment abgeholt wurde.</summary>
    ///
    /// <remarks>
    /// <b>Daran hängt das Aufräumen.</b> Ein geschlossener Browser sagt
    /// niemandem Bescheid; bliebe die Sitzung stehen, liefe ffmpeg weiter und
    /// schriebe den ganzen Film auf den Kratzplatz.
    /// </remarks>
    public DateTimeOffset Gesehen { get; set; } = Begonnen;

    public string? Fehler { get; set; }
    public bool Beendet { get; set; }
}

/// <summary>Verwaltet laufende Ströme: starten, bedienen, aufräumen.</summary>
///
/// <remarks>
/// <para>
/// <b>Neben <see cref="PackageCoordinator"/>, nicht an seiner Stelle.</b> Der
/// baut eine fertige Datei zum Mitnehmen — mit Länge, Bereichsanfragen und
/// Fortsetzen nach einem Abbruch; das bleibt für die Apps richtig. Hier geht
/// es um etwas anderes: der erste Ton soll nach Sekunden kommen, nicht nach
/// Minuten.
/// </para>
/// <para>
/// <b>Eine Sitzung je Wiedergabe, nicht je Datei.</b> Zwei Leute können
/// denselben Film in verschiedenen Auflösungen und mit verschiedenen
/// Tonspuren sehen. Das nach Dateikennung zu führen hiesse, dass der zweite
/// bekommt, was der erste gewählt hat.
/// </para>
/// </remarks>
public sealed class Streamkoordinator(
    IServiceScopeFactory scopes,
    FlaroPaths pfade,
    FfmpegOptions ffmpeg,
    Schluesselbilder schluesselbilder,
    TimeProvider clock,
    ILogger<Streamkoordinator> log) : IDisposable
{
    private readonly ConcurrentDictionary<Guid, Eintrag> _sitzungen = new();

    private sealed record Eintrag(Stromsitzung Sitzung, Process Prozess, CancellationTokenSource Abbruch);

    /// <summary>Nach dieser Stille gilt eine Sitzung als verlassen.</summary>
    ///
    /// <remarks>
    /// <b>Grosszügiger als ein Segment lang.</b> Wer pausiert, holt eine
    /// Weile nichts ab; ihn nach zwanzig Sekunden abzuwürgen wäre
    /// Sparsamkeit an der falschen Stelle. Zwei Minuten reichen für eine
    /// Pause und sind kurz genug, dass ein geschlossener Browser nicht
    /// stundenlang Platten füllt.
    /// </remarks>
    public static readonly TimeSpan Geduld = TimeSpan.FromMinutes(2);

    /// <summary>Wie viele Ströme gleichzeitig laufen dürfen.</summary>
    ///
    /// <remarks>
    /// <b>Weil Umkodieren einen ganzen Rechner beschäftigt.</b> Vier
    /// gleichzeitige 4K-Läufe bringen jedes NAS zum Stehen — und dann
    /// ruckelt nicht einer, sondern alle vier.
    /// </remarks>
    private const int HoechstensGleichzeitig = 6;

    public int LaufendeAnzahl => _sitzungen.Count;

    public Stromsitzung? Sitzung(Guid kennung) => _sitzungen.GetValueOrDefault(kennung)?.Sitzung;

    /// <summary>Startet einen Strom und gibt die Sitzung zurück.</summary>
    /// <param name="erzwingen">
    /// Umkodieren, auch wenn das Profil etwas anderes sagt — der Spieler hat
    /// es mit dem ersten Plan probiert und ist gescheitert.
    /// </param>
    public async Task<Stromsitzung> StartenAsync(
        Guid dateiId, Guid benutzer, DeviceProfile profil,
        int? hoehe, int? tonspur, double ab, CancellationToken ct = default,
        bool erzwingen = false, double tonversatz = 0)
    {
        AbgelaufeneWegraeumen();

        if (_sitzungen.Count >= HoechstensGleichzeitig)
        {
            throw new InvalidOperationException(
                $"Es laufen bereits {HoechstensGleichzeitig} Wiedergaben, die umgerechnet werden müssen. "
                + "Warte, bis eine davon zu Ende ist.");
        }

        using var bereich = scopes.CreateScope();
        var db = bereich.ServiceProvider.GetRequiredService<FlaroContext>();

        var datei = await db.MediaFiles.AsNoTracking().FirstOrDefaultAsync(f => f.Id == dateiId, ct)
            ?? throw new InvalidOperationException("Diese Datei steht nicht mehr in der Mediathek.");

        if (!File.Exists(datei.Path))
            throw new InvalidOperationException($"„{Path.GetFileName(datei.Path)}“ liegt nicht mehr da.");

        var spuren = await db.MediaStreams.AsNoTracking()
            .Where(s => s.MediaFileId == dateiId).OrderBy(s => s.Index)
            .ToListAsync(ct);

        var gelesen = spuren.Select(s => new ProbedStream(
            s.Index, s.Kind, s.Codec, s.Language, s.Title,
            s.IsDefault, s.IsForced, s.Channels, s.Width, s.Height)).ToList();

        var plan = Stromentscheid.Fuer(datei.Container, gelesen, profil, hoehe, erzwingen);

        var kennung = Guid.CreateVersion7();
        var ordner = Path.Combine(pfade.Transcode, "stroeme", kennung.ToString("N"));
        Directory.CreateDirectory(ordner);

        // Die eingestellte Stufe aus demselben Bereich — ein Lesevorgang, und
        // er steht hier, weil nur hier feststeht, dass überhaupt kodiert wird.
        var stufe = await bereich.ServiceProvider
            .GetRequiredService<Wiedergabeeinstellungen>().StufeAsync(ct);

        // **Beim Kopieren wird auf das Schlüsselbild gerundet.**
        //
        // Ein kopierter Bildstrom lässt sich nicht mitten in einer Bildgruppe
        // aufschneiden; ffmpeg beginnt beim Schlüsselbild davor, während der
        // Ton neu gerechnet wird und genau beim Wunsch anfängt. Gemessen an
        // einer Quelle mit Schlüsselbildern alle fünf Sekunden, Sprung auf
        // Sekunde 12: Bild ab 0, Ton erst ab 1,98 — zwei Sekunden stummes
        // Bild, und die Uhr des Spielers zwei Sekunden daneben.
        //
        // Beim Umkodieren entfällt das: dort wird das Bild neu gerechnet und
        // schneidet auf die Sekunde genau. Dann vorzurunden hiesse, ohne Not
        // ein Stück zurückzuspringen.
        // **Gesprungen wird auf den Wunsch, gemeldet wird das Schlüsselbild.**
        //
        // Beim Kopieren springt ffmpeg ungenau — siehe `HlsArgumente` — und
        // landet damit auf dem Schlüsselbild vor dem Wunsch. Bild und Ton
        // fangen dort gemeinsam an; das ist die Zusage, und sie hängt an
        // keiner Rechnung hier.
        //
        // Wo genau das war, sagt die Suche. Sie ist dafür da, dem Spieler
        // eine ehrliche Uhr zu geben — nicht dafür, die Ausrichtung
        // herzustellen. Vergreift sie sich, geht die Uhr falsch; Bild und Ton
        // bleiben zusammen. Umgekehrt wäre es ein schlechter Tausch: einen
        // Sprungpunkt auf die Millisekunde zu treffen, ist zu viel verlangt.
        var gewuenscht = Math.Max(0, ab);
        var uhr = System.Diagnostics.Stopwatch.StartNew();
        var gemeldet = plan.BildKopieren
            ? await schluesselbilder.DavorAsync(datei.Path, gewuenscht, ct)
            : gewuenscht;
        var suche = uhr.ElapsedMilliseconds;

        var anfrage = new Stromanfrage(
            datei.Path, ordner, gelesen, plan, tonspur, gewuenscht, Hardware(), stufe, tonversatz);

        // **Die tatsächlich benutzte Spur, nicht die gewünschte.** Sonst
        // stünde im Menü bei keiner ein Haken, obwohl eine davon läuft — und
        // bei einer Spur, die es nicht mehr gibt, bei der falschen.
        var sitzung = new Stromsitzung(
            kennung, dateiId, benutzer, ordner, gemeldet,
            plan.Hoehe, HlsArgumente.Tonspur(gelesen, tonspur)?.Index, plan.Art, clock.GetUtcNow());

        var prozess = Starten(anfrage, sitzung);
        var abbruch = new CancellationTokenSource();
        _sitzungen[kennung] = new Eintrag(sitzung, prozess, abbruch);

        // **Mit Zeiten, und auch im Tagesprotokoll.**
        //
        // „Es dauert fünfzehn Sekunden" liess sich nicht beantworten: zwischen
        // dem Antippen und dem ersten Bild liegen die Schlüsselbildsuche, der
        // Start von ffmpeg und das erste Segment, und keine dieser Strecken
        // war irgendwo abzulesen. Eine Zeile je Wiedergabe ist kein Lärm —
        // sie ist genau das, wonach man im Protokoll sucht.
        var meldung =
            $"Wiedergabe: {plan.Art} von {Path.GetFileName(datei.Path)} ab {sitzung.Ab:0.#}s"
            + (plan.Hoehe is { } h ? $", auf {h}p" : "")
            + $" — Schlüsselbildsuche {suche} ms, Vorbereitung {uhr.ElapsedMilliseconds} ms.";

        log.LogInformation("{Meldung}", meldung);

        return sitzung;
    }

    /// <summary>Hält eine Sitzung am Leben — bei jedem abgeholten Segment.</summary>
    public void Lebenszeichen(Guid kennung)
    {
        if (_sitzungen.TryGetValue(kennung, out var eintrag))
            eintrag.Sitzung.Gesehen = clock.GetUtcNow();
    }

    /// <summary>Beendet eine Sitzung und räumt ihren Ordner weg.</summary>
    public void Beenden(Guid kennung)
    {
        if (!_sitzungen.TryRemove(kennung, out var eintrag)) return;
        Abraeumen(eintrag);
    }

    /// <summary>Beendet, was niemand mehr abholt.</summary>
    ///
    /// <remarks>
    /// <b>Beim Starten einer neuen Sitzung, nicht in einem eigenen Takt.</b>
    /// Ein Hintergrunddienst für einen Handgriff, der ohnehin selten
    /// geschieht, wäre ein Faden mehr, der schiefgehen kann.
    /// </remarks>
    public void AbgelaufeneWegraeumen()
    {
        var grenze = clock.GetUtcNow() - Geduld;
        foreach (var (kennung, eintrag) in _sitzungen)
        {
            if (eintrag.Sitzung.Gesehen > grenze) continue;
            if (!_sitzungen.TryRemove(kennung, out var weg)) continue;

            log.LogInformation("Strom {Kennung} wurde seit {Geduld} nicht abgeholt — beendet.",
                kennung, Geduld);
            Abraeumen(weg);
        }
    }

    /// <summary>Ob die Playlist schon etwas enthält, das sich abspielen lässt.</summary>
    ///
    /// <remarks>
    /// <b>Die Datei allein genügt nicht.</b> ffmpeg legt sie an, bevor das
    /// erste Segment fertig ist; ein Spieler, der sie dann bekommt, sieht
    /// eine leere Liste und gibt auf.
    /// </remarks>
    public static string? Playlist(string ordner)
    {
        var pfad = Path.Combine(ordner, HlsArgumente.Playlist);
        if (!File.Exists(pfad)) return null;

        try
        {
            // `ReadAllText` auf einer Datei, in die ffmpeg gerade schreibt:
            // deshalb ausdrücklich teilend geöffnet, sonst gäbe es hier je
            // nach Zeitpunkt eine Ausnahme statt einer Antwort.
            using var strom = new FileStream(
                pfad, FileMode.Open, FileAccess.Read, FileShare.ReadWrite | FileShare.Delete);
            using var leser = new StreamReader(strom);
            var text = leser.ReadToEnd();

            // Ohne Segment ist die Liste wertlos — der Spieler sähe eine
            // leere Playlist und gäbe auf.
            return text.Contains(".m4s", StringComparison.Ordinal) ? text : null;
        }
        catch (IOException) { return null; }
    }

    /// <summary>Ob die Playlist schon etwas enthält, das sich abspielen lässt.</summary>
    public static bool PlaylistBereit(string ordner) => Playlist(ordner) is not null;

    /// <summary>Das Gerät, über das VAAPI läuft.</summary>
    private const string Vaapigeraet = "/dev/dri/renderD128";

    /// <summary>Das Ergebnis der einmaligen Probe — <c>null</c>, solange keine lief.</summary>
    private bool? _vaapiLaeuft;

    /// <summary>Welche Beschleunigung benutzt wird.</summary>
    ///
    /// <remarks>
    /// <para>
    /// <b><c>auto</c> fragt ffmpeg, statt die Datei anzusehen.</b> Hier stand
    /// <c>File.Exists("/dev/dri/renderD128")</c>, und das war keine Prüfung,
    /// sondern eine Vermutung: auf dem NAS liegt das Gerät durchgereicht da,
    /// aber dahinter steht keine VA-Anzeige. ffmpeg bricht dann schon an der
    /// globalen Option ab — <c>No VA display found for device</c>, Rückgabe
    /// 234 —, und für den Zuschauer sieht das aus wie ein kaputter Server.
    /// </para>
    /// <para>
    /// <b>Einmal je Lauf.</b> Die Probe kostet einen ffmpeg-Start; sie vor
    /// jeder Wiedergabe zu wiederholen wäre Aufwand für eine Antwort, die
    /// sich nicht ändert. Wechselt die Hardware, wechselt auch der
    /// Container.
    /// </para>
    /// <para>
    /// <b><c>FLARO_HWACCEL=vaapi</c> überspringt sie</b> — wer es
    /// ausdrücklich verlangt, hat einen Grund, und ein Fehlschlag steht dann
    /// im Protokoll.
    /// </para>
    /// </remarks>
    private Beschleunigung Hardware()
    {
        var gewuenscht = Environment.GetEnvironmentVariable("FLARO_HWACCEL")?.Trim().ToLowerInvariant();

        return gewuenscht switch
        {
            "none" or "software" => Beschleunigung.Software,
            "vaapi" => Beschleunigung.Vaapi,
            _ => (_vaapiLaeuft ??= VaapiLaeuft()) ? Beschleunigung.Vaapi : Beschleunigung.Software,
        };
    }

    /// <summary>Lässt ffmpeg das Gerät einmal wirklich öffnen.</summary>
    private bool VaapiLaeuft()
    {
        if (!File.Exists(Vaapigeraet)) return false;

        try
        {
            var start = new ProcessStartInfo(ffmpeg.Ffmpeg)
            {
                RedirectStandardOutput = true,
                RedirectStandardError = true,
                UseShellExecute = false,
            };
            // Ein Bild aus dem Nichts, nirgendwohin. Es geht allein darum, ob
            // die Zeile vor dem ersten Bild durchgeht.
            foreach (var a in new[]
            {
                "-hide_banner", "-loglevel", "error",
                "-vaapi_device", Vaapigeraet,
                "-f", "lavfi", "-i", "nullsrc=s=64x64:d=0.1",
                "-frames:v", "1", "-f", "null", "-",
            })
            {
                start.ArgumentList.Add(a);
            }

            using var prozess = Process.Start(start);
            if (prozess is null) return false;

            // **Mit Frist.** Ein hängender Prozess hielte sonst die erste
            // Wiedergabe nach dem Start auf.
            if (!prozess.WaitForExit(5000))
            {
                try { prozess.Kill(entireProcessTree: true); } catch (InvalidOperationException) { }
                log.LogWarning("Die VAAPI-Probe antwortete nicht; es läuft in Software.");
                return false;
            }

            if (prozess.ExitCode == 0) return true;

            log.LogInformation(
                "VAAPI ist nicht benutzbar ({Code}); es läuft in Software. {Grund}",
                prozess.ExitCode, prozess.StandardError.ReadToEnd().Trim());
            return false;
        }
        catch (Exception fehler) when (fehler is System.ComponentModel.Win32Exception or IOException)
        {
            log.LogWarning(fehler, "Die VAAPI-Probe liess sich nicht starten; es läuft in Software.");
            return false;
        }
    }

    private Process Starten(Stromanfrage anfrage, Stromsitzung sitzung)
    {
        var argumente = HlsArgumente.Fuer(anfrage);

        var start = new ProcessStartInfo(ffmpeg.Ffmpeg)
        {
            RedirectStandardOutput = true,
            RedirectStandardError = true,
            UseShellExecute = false,
        };
        foreach (var a in argumente) start.ArgumentList.Add(a);

        var prozess = Process.Start(start)
            ?? throw new InvalidOperationException($"{ffmpeg.Ffmpeg} liess sich nicht starten.");

        // **Der Fehlerkanal wird gelesen, nicht nur umgeleitet.** Läuft die
        // Röhre voll, bleibt ffmpeg beim Schreiben stehen — und der Strom
        // hört mitten im Film auf, ohne dass irgendwo etwas stünde.
        _ = Task.Run(async () =>
        {
            var text = await prozess.StandardError.ReadToEndAsync();
            await prozess.WaitForExitAsync();

            sitzung.Beendet = true;
            if (prozess.ExitCode == 0)
            {
                // **Auch ein geglückter Lauf darf etwas zu sagen haben.**
                // Vorher wurde der Text weggeworfen, sobald der Rückgabewert 0
                // war — und genau die Meldungen, die einen Tonversatz
                // ankündigen, sind Warnungen, keine Fehler. Ein Film, der
                // schief klingt, aber durchläuft, hinterliess damit nirgends
                // eine Spur.
                if (text.Trim() is { Length: > 0 } warnungen)
                {
                    log.LogWarning("Strom {Kennung}: ffmpeg meldet {Warnungen}",
                        sitzung.Kennung, warnungen);
                }

                return;
            }

            sitzung.Fehler = text.Trim();
            log.LogError("Wiedergabe gescheitert: ffmpeg endete mit {Code}. {Fehler}",
                prozess.ExitCode, sitzung.Fehler);

        });

        _ = Task.Run(() => prozess.StandardOutput.ReadToEndAsync());
        return prozess;
    }

    private void Abraeumen(Eintrag eintrag)
    {
        try
        {
            if (!eintrag.Prozess.HasExited) eintrag.Prozess.Kill(entireProcessTree: true);
        }
        catch (Exception fehler) when (fehler is InvalidOperationException or NotSupportedException)
        {
            log.LogDebug(fehler, "ffmpeg zu {Kennung} liess sich nicht beenden.", eintrag.Sitzung.Kennung);
        }

        eintrag.Prozess.Dispose();
        eintrag.Abbruch.Dispose();

        try
        {
            if (Directory.Exists(eintrag.Sitzung.Ordner))
                Directory.Delete(eintrag.Sitzung.Ordner, recursive: true);
        }
        catch (IOException fehler)
        {
            log.LogWarning(fehler, "Der Ordner zu Strom {Kennung} blieb liegen.", eintrag.Sitzung.Kennung);
        }
    }

    /// <summary>Der Ordner, in dem die Segmente liegen.</summary>
    public string Wurzel => Path.Combine(pfade.Transcode, "stroeme");

    /// <summary>Räumt alles weg, was unter der Wurzel liegt.</summary>
    ///
    /// <remarks>
    /// <b>Beim Herunterfahren und beim Hochfahren.</b> Die Ordnernamen sind
    /// Sitzungskennungen aus dem Arbeitsspeicher; nach einem Start gibt es
    /// keine einzige Sitzung, also ist jeder Ordner verwaist. Bisher geschah
    /// das nur beim geordneten Ende — nach einem harten Abbruch blieb alles
    /// liegen, und niemand kam je wieder darauf zurück.
    /// </remarks>
    public long WurzelLeeren()
    {
        if (!Directory.Exists(Wurzel)) return 0;

        var frei = Belegung(Wurzel);
        try { Directory.Delete(Wurzel, recursive: true); }
        catch (IOException fehler)
        {
            log.LogDebug(fehler, "Der Stromordner {Pfad} liess sich nicht leeren.", Wurzel);
            return 0;
        }

        return frei;
    }

    /// <summary>Räumt Ordner weg, zu denen es keine Sitzung mehr gibt.</summary>
    ///
    /// <remarks>
    /// <b>Nicht bloss Symmetrie zum Herunterfahren.</b> Scheitert das
    /// Löschen in <c>Abraeumen</c> an einem <c>IOException</c>, ist die
    /// Sitzung aus dem Verzeichnis heraus und der Ordner bleibt für immer
    /// liegen. Die Altersprüfung nimmt dabei den Fall heraus, dass ein
    /// zweiter Prozess denselben Kratzplatz benutzt: ein aktiver Ordner
    /// bekommt alle paar Sekunden ein Segment geschrieben.
    /// </remarks>
    public long VerwaisteOrdnerWegraeumen(DateTimeOffset jetzt)
    {
        AbgelaufeneWegraeumen();
        if (!Directory.Exists(Wurzel)) return 0;

        long frei = 0;
        foreach (var ordner in Directory.EnumerateDirectories(Wurzel))
        {
            var name = Path.GetFileName(ordner);
            if (Guid.TryParse(name, out var kennung) && _sitzungen.ContainsKey(kennung)) continue;
            if (jetzt - new DateTimeOffset(Directory.GetLastWriteTimeUtc(ordner), TimeSpan.Zero) < Geduld) continue;

            var gross = Belegung(ordner);
            try { Directory.Delete(ordner, recursive: true); frei += gross; }
            catch (IOException fehler)
            {
                log.LogDebug(fehler, "Der verwaiste Ordner {Pfad} liess sich nicht löschen.", ordner);
            }
        }

        return frei;
    }

    /// <summary>Wie viele Bytes unter einem Ordner liegen.</summary>
    public static long Belegung(string ordner)
    {
        try
        {
            return Directory.Exists(ordner)
                ? Directory.EnumerateFiles(ordner, "*", SearchOption.AllDirectories)
                    .Sum(d => { try { return new FileInfo(d).Length; } catch (IOException) { return 0; } })
                : 0;
        }
        catch (Exception fehler) when (fehler is IOException or UnauthorizedAccessException)
        {
            return 0;
        }
    }

    /// <summary>Beim Herunterfahren bleibt nichts laufen und nichts liegen.</summary>
    public void Dispose()
    {
        foreach (var kennung in _sitzungen.Keys) Beenden(kennung);

        // **Auch, was ein Absturz hinterlassen hat.** Nach einem harten Ende
        // steht der Ordner noch da; ihn beim nächsten Herunterfahren zu
        // übersehen hiesse, dass der Kratzplatz nur wächst.
        WurzelLeeren();
    }
}