LlamaCasty/Services/Encoder/FramePump.cs

using System;
using System.Threading;
using System.Threading.Tasks;
using ytLive.Models;
using ytLive.Services.Compositor;

namespace ytLive.Services.Encoder;

/// <summary>
/// The live frame producer (TASK 4 ship step 5): the bridge between the capture
/// managers + compositor and the encoder. While live it snapshots the active
/// scene each tick, resolves every element to its latest frame, composites it
/// into the tier's output frame, and paces frames into the encoder at the tier's
/// FPS. All collaborators are constructor-injected seams (scene, resolver,
/// options, encoder factory, pacing delay) so the pump stays free of WPF and of
/// the capture managers and is fully hermetic in tests.
///
/// The RTMP URL comes from the options provider: until the live-stream create
/// flow lands (TASK 5) it yields null, so go-live runs the existing visual flow
/// without actually pushing.
/// </summary>
public sealed class FramePump : IDisposable
{
    private readonly Func<Scene?> _sceneProvider;
    private readonly Func<SceneElement, VideoFrame?> _frameResolver;
    private readonly Func<CompositorOptions> _compositorOptions;
    private readonly Func<EncoderOptions?> _encoderOptions;
    private readonly Func<IFfmpegEncoder> _encoderFactory;
    private readonly Action<string>? _log;
    private readonly Func<TimeSpan, CancellationToken, Task> _pacingDelay;
    private readonly Func<(VideoFrame? Frame, SocialBarPosition Position)>? _socialBar;
    private readonly TransitionService? _transition;
    private readonly Func<Scene?>? _fromSceneProvider;
    private readonly SceneCompositor _compositor = new();

    private readonly object _gate = new();
    private IFfmpegEncoder? _encoder;
    private CancellationTokenSource? _cts;
    private Task? _pumpTask;
    private bool _started;

    /// <summary>Forwards the encoder's parsed health — ship step 6 binds this to the bottom bar.</summary>
    public event EventHandler<StreamHealth>? HealthUpdated;

    /// <summary>Raised when the encoder cannot start or dies mid-stream. The pump stops itself.</summary>
    public event EventHandler<string>? Failed;

    public FramePump(
        Func<Scene?> sceneProvider,
        Func<SceneElement, VideoFrame?> frameResolver,
        Func<CompositorOptions> compositorOptions,
        Func<EncoderOptions?> encoderOptions,
        Func<IFfmpegEncoder> encoderFactory,
        Action<string>? log = null,
        Func<TimeSpan, CancellationToken, Task>? pacingDelay = null,
        Func<(VideoFrame? Frame, SocialBarPosition Position)>? socialBar = null,
        TransitionService? transition = null,
        Func<Scene?>? fromSceneProvider = null)
    {
        _sceneProvider = sceneProvider ?? throw new ArgumentNullException(nameof(sceneProvider));
        _frameResolver = frameResolver ?? throw new ArgumentNullException(nameof(frameResolver));
        _compositorOptions = compositorOptions ?? throw new ArgumentNullException(nameof(compositorOptions));
        _encoderOptions = encoderOptions ?? throw new ArgumentNullException(nameof(encoderOptions));
        _encoderFactory = encoderFactory ?? throw new ArgumentNullException(nameof(encoderFactory));
        _log = log;
        _pacingDelay = pacingDelay ?? ((delay, ct) => Task.Delay(delay, ct));
        _socialBar = socialBar;
        _transition = transition;
        _fromSceneProvider = fromSceneProvider;
    }

    public bool IsRunning { get; private set; }

    /// <summary>Never throws: failures are logged and surfaced via <see cref="Failed"/>,
    /// so the VM can fire-and-forget it from a sync command handler.</summary>
    public async Task StartAsync(CancellationToken cancellationToken = default)
    {
        lock (_gate)
        {
            if (_started) return;
            _started = true;
        }

        IFfmpegEncoder? encoder = null;
        try
        {
            var options = _encoderOptions();
            if (options == null)
            {
                _log?.Invoke("FramePump: no output configured (neither streaming nor recording) — encoder skipped");
                lock (_gate) _started = false;
                return;
            }

            encoder = _encoderFactory();
            encoder.HealthUpdated += OnHealthUpdated;
            encoder.ProcessFailed += OnProcessFailed;
            await encoder.StartAsync(options, cancellationToken);

            lock (_gate)
            {
                _encoder = encoder;
            }

            // IsRunning must be true before the loop starts: the loop reads it on
            // its first iteration, and with a completed-task delay it can run
            // synchronously on this thread before PumpAsync even returns.
            IsRunning = true;
            _cts = new CancellationTokenSource();
            _pumpTask = PumpAsync(options, _cts.Token);
            _log?.Invoke($"FramePump started ({options.Width}×{options.Height} @ {options.Fps} fps)");
        }
        catch (Exception ex)
        {
            _log?.Invoke($"FramePump: start failed: {ex.Message}");
            if (encoder != null)
            {
                encoder.HealthUpdated -= OnHealthUpdated;
                encoder.ProcessFailed -= OnProcessFailed;
                try
                {
                    encoder.Dispose();
                }
                catch (Exception disposeEx)
                {
                    _log?.Invoke($"FramePump: disposing failed encoder: {disposeEx.Message}");
                }
            }
            lock (_gate)
            {
                _started = false;
                IsRunning = false;
            }
            Failed?.Invoke(this, ex.Message);
        }
    }

    public async Task StopAsync(CancellationToken cancellationToken = default)
    {
        IFfmpegEncoder? encoder;
        Task? pump;
        lock (_gate)
        {
            if (!_started && _encoder == null) return;
            _started = false;
            IsRunning = false;
            encoder = _encoder;
            pump = _pumpTask;
            _cts?.Cancel();
        }

        // Stop the encoder BEFORE awaiting the pump: closing its stdin unblocks a
        // write stuck on pipe backpressure, otherwise the pump could await forever.
        if (encoder != null)
        {
            try
            {
                await encoder.StopAsync(cancellationToken);
            }
            catch (Exception ex)
            {
                _log?.Invoke($"FramePump: encoder stop failed: {ex.Message}");
            }
        }

        if (pump != null)
        {
            try
            {
                await pump;
            }
            catch (Exception ex)
            {
                _log?.Invoke($"FramePump: pump loop faulted during stop: {ex.Message}");
            }
        }

        if (encoder != null)
        {
            encoder.HealthUpdated -= OnHealthUpdated;
            encoder.ProcessFailed -= OnProcessFailed;
            try
            {
                encoder.Dispose();
            }
            catch (Exception ex)
            {
                _log?.Invoke($"FramePump: encoder dispose failed: {ex.Message}");
            }
        }

        lock (_gate)
        {
            _encoder = null;
            _cts = null;
            _pumpTask = null;
        }
        _log?.Invoke("FramePump stopped");
    }

    public void Dispose()
    {
        try
        {
            StopAsync().GetAwaiter().GetResult();
        }
        catch (Exception ex)
        {
            _log?.Invoke($"FramePump: dispose failed: {ex.Message}");
        }
    }

    private async Task PumpAsync(EncoderOptions options, CancellationToken ct)
    {
        var interval = TimeSpan.FromSeconds(1d / Math.Max(1, options.Fps));
        var lastTick = System.Diagnostics.Stopwatch.StartNew();

        try
        {
            while (!ct.IsCancellationRequested)
            {
                var scene = _sceneProvider();
                if (scene != null)
                {
                    var compositorOptions = _compositorOptions();
                    VideoFrame? socialBarFrame = null;
                    var socialBarTop = 0;
                    if (_socialBar != null)
                    {
                        var (barFrame, position) = _socialBar();
                        socialBarFrame = barFrame;
                        if (barFrame != null)
                            socialBarTop = position == SocialBarPosition.Top
                                ? 0
                                : compositorOptions.SourceRectHeight - barFrame.Height;
                    }

                    VideoFrame frame;
                    if (_transition is { Active: true } transition && _fromSceneProvider != null)
                    {
                        var fromScene = _fromSceneProvider();
                        var fromFrame = fromScene != null
                            ? _compositor.Render(fromScene, _frameResolver, null, compositorOptions, socialBarFrame, socialBarTop)
                            : _compositor.Render(scene, _frameResolver, null, compositorOptions, socialBarFrame, socialBarTop);
                        var toFrame = _compositor.Render(scene, _frameResolver, null, compositorOptions, socialBarFrame, socialBarTop);
                        frame = transition.BlendFrame(toFrame);
                        transition.Tick(lastTick.Elapsed.TotalMilliseconds);
                        lastTick.Restart();
                    }
                    else
                    {
                        frame = _compositor.Render(scene, _frameResolver, null, compositorOptions, socialBarFrame, socialBarTop);
                        lastTick.Restart();
                    }

                    IFfmpegEncoder? encoder;
                    lock (_gate) encoder = _encoder;
                    if (encoder == null) break;
                    await encoder.SubmitFrameAsync(frame, ct);
                }
                await _pacingDelay(interval, ct);
            }
        }
        catch (OperationCanceledException)
        {
            // normal stop
        }
        catch (Exception ex)
        {
            // A failure while the pump is supposed to run (encoder died under us,
            // scene provider faulted, ...) stops the pump and surfaces once.
            if (ct.IsCancellationRequested)
            {
                _log?.Invoke($"FramePump: pump exited during stop: {ex.Message}");
            }
            else
            {
                _log?.Invoke($"FramePump: pump loop faulted: {ex.Message}");
                Failed?.Invoke(this, ex.Message);
            }
        }
        finally
        {
            lock (_gate) IsRunning = false;
        }
    }

    private void OnHealthUpdated(object? sender, StreamHealth health) => HealthUpdated?.Invoke(this, health);

    private void OnProcessFailed(object? sender, string message)
    {
        _log?.Invoke($"FramePump: encoder process failed: {message}");
        Failed?.Invoke(this, message);
        _ = StopAsync();
    }
}