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,
    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>
    public async Task<Stromsitzung> StartenAsync(
        Guid dateiId, Guid benutzer, DeviceProfile profil,
        int? hoehe, int? tonspur, double ab, CancellationToken ct = default)
    {
        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);

        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);

        var anfrage = new Stromanfrage(
            datei.Path, ordner, gelesen, plan, tonspur, Math.Max(0, ab), Hardware(), stufe);

        // **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, Math.Max(0, ab),
            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);

        log.LogInformation(
            "Strom {Kennung}: {Art} von {Datei} ab {Ab}s{Hoehe}.",
            kennung, plan.Art, Path.GetFileName(datei.Path), sitzung.Ab,
            plan.Hoehe is { } h ? $", auf {h}p" : "");

        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>Welche Beschleunigung benutzt wird.</summary>
    ///
    /// <remarks>
    /// <b><c>auto</c> prüft, statt zu behaupten.</b> Liegt kein
    /// <c>/dev/dri/renderD128</c> vor — auf einem Mac, in einem Container ohne
    /// durchgereichtes Gerät —, läuft es in Software weiter. Langsamer, aber
    /// es läuft; ein ffmpeg, das mit „No VA display found" abbricht, wäre für
    /// den Zuschauer dasselbe wie ein kaputter Server.
    /// </remarks>
    private Beschleunigung Hardware()
    {
        var gewuenscht = Environment.GetEnvironmentVariable("FLARO_HWACCEL")?.Trim().ToLowerInvariant();

        return gewuenscht switch
        {
            "none" or "software" => Beschleunigung.Software,
            "vaapi" => Beschleunigung.Vaapi,
            _ => File.Exists("/dev/dri/renderD128") ? Beschleunigung.Vaapi : Beschleunigung.Software,
        };
    }

    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) return;

            sitzung.Fehler = text.Trim();
            log.LogWarning("Strom {Kennung}: ffmpeg endete mit {Code}: {Fehler}",
                sitzung.Kennung, 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>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.
        try
        {
            var wurzel = Path.Combine(pfade.Transcode, "stroeme");
            if (Directory.Exists(wurzel)) Directory.Delete(wurzel, recursive: true);
        }
        catch (IOException) { }
    }
}