Merge streaming-overhaul into dev (Opus low-data streaming, windowed streaming, HW-accel-off stabilization)

DeepDrftAPI/Services/Opus/IOpusTranscodeQueue.cs

@@ -0,0 +1,18 @@
+namespace DeepDrftAPI.Services.Opus;
+
+/// <summary>
+/// The enqueue seam for the background Opus transcode (OQ6 / §3.1a). <see cref="UnifiedTrackService"/>
+/// depends only on this thin interface — not on the worker — so adding the background derive to the
+/// upload/replace paths costs one small dependency, not the whole transcode graph. Enqueuing is
+/// non-blocking and best-effort: a freshly uploaded track is already persisted and playable losslessly
+/// before anything is enqueued, and the transcode runs off the request thread.
+/// </summary>
+public interface IOpusTranscodeQueue
+{
+    /// <summary>
+    /// Schedules a background Opus derive for the track identified by <paramref name="entryKey"/>. Returns
+    /// immediately. A dropped or failed enqueue must not affect the caller — the track remains
+    /// lossless-only and eligible for backfill.
+    /// </summary>
+    void Enqueue(string entryKey);
+}

DeepDrftAPI/Services/Opus/OpusTranscodeBackgroundService.cs

@@ -0,0 +1,72 @@
+using System.Threading.Channels;
+using DeepDrftContent.Processors.Opus;
+
+namespace DeepDrftAPI.Services.Opus;
+
+/// <summary>
+/// The background worker behind <see cref="IOpusTranscodeQueue"/> (OQ6 / §3.1a). An unbounded in-process
+/// channel buffers EntryKeys enqueued by the upload and replace-audio paths; a single hosted loop drains
+/// them one at a time and runs <see cref="OpusTranscodeService.TranscodeAndStoreAsync"/> for each. Serial
+/// by design — a transcode is CPU-heavy (§3.1), so running them concurrently would starve request
+/// handling; one-at-a-time keeps the derive strictly off the hot path without saturating the host.
+///
+/// This worker IS the queue (implements <see cref="IOpusTranscodeQueue"/>) so enqueue and drain share one
+/// channel with no extra indirection. It is registered as a singleton and surfaced under both the
+/// interface and <see cref="IHostedService"/>.
+/// </summary>
+public sealed class OpusTranscodeBackgroundService : BackgroundService, IOpusTranscodeQueue
+{
+    private readonly Channel<string> _channel =
+        Channel.CreateUnbounded<string>(new UnboundedChannelOptions { SingleReader = true });
+
+    private readonly OpusTranscodeService _transcodeService;
+    private readonly ILogger<OpusTranscodeBackgroundService> _logger;
+
+    public OpusTranscodeBackgroundService(
+        OpusTranscodeService transcodeService,
+        ILogger<OpusTranscodeBackgroundService> logger)
+    {
+        _transcodeService = transcodeService;
+        _logger = logger;
+    }
+
+    public void Enqueue(string entryKey)
+    {
+        if (string.IsNullOrWhiteSpace(entryKey))
+            return;
+
+        if (!_channel.Writer.TryWrite(entryKey))
+        {
+            // Unbounded writer only rejects after Complete(), i.e. during shutdown. The track stays
+            // lossless-only and is eligible for backfill, so a dropped enqueue is non-fatal — log it.
+            _logger.LogWarning("Opus transcode: could not enqueue {EntryKey} (queue closed).", entryKey);
+        }
+    }
+
+    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+    {
+        await foreach (var entryKey in _channel.Reader.ReadAllAsync(stoppingToken))
+        {
+            try
+            {
+                await _transcodeService.TranscodeAndStoreAsync(entryKey, stoppingToken);
+            }
+            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
+            {
+                break; // host shutting down
+            }
+            catch (Exception ex)
+            {
+                // TranscodeAndStoreAsync already swallows expected failures; this guards the loop against
+                // anything unexpected so one bad track never kills the worker.
+                _logger.LogError(ex, "Opus transcode: unhandled failure draining {EntryKey}; worker continues.", entryKey);
+            }
+        }
+    }
+
+    public override Task StopAsync(CancellationToken cancellationToken)
+    {
+        _channel.Writer.TryComplete();
+        return base.StopAsync(cancellationToken);
+    }
+}

DeepDrftAPI/Services/UnifiedTrackService.cs

@@ -1,6 +1,8 @@
+using DeepDrftAPI.Services.Opus;
 using DeepDrftContent;
 using DeepDrftContent.Constants;
 using DeepDrftContent.Processors;
+using DeepDrftContent.Processors.Opus;
 using DeepDrftData;
 using DeepDrftModels.DTOs;
 using DeepDrftModels.Enums;
@@ -39,6 +41,8 @@ public class UnifiedTrackService
     private readonly ITrackService _sqlTrackService;
     private readonly FileDb _fileDatabase;
     private readonly WaveformProfileService _waveformProfileService;
+    private readonly IOpusTranscodeQueue _opusTranscodeQueue;
+    private readonly TrackFormatResolver _formatResolver;
     private readonly ILogger<UnifiedTrackService> _logger;
 
     public UnifiedTrackService(
@@ -46,12 +50,16 @@ public class UnifiedTrackService
         ITrackService sqlTrackService,
         FileDb fileDatabase,
         WaveformProfileService waveformProfileService,
+        IOpusTranscodeQueue opusTranscodeQueue,
+        TrackFormatResolver formatResolver,
         ILogger<UnifiedTrackService> logger)
     {
         _contentTrackContentService = contentTrackContentService;
         _sqlTrackService = sqlTrackService;
         _fileDatabase = fileDatabase;
         _waveformProfileService = waveformProfileService;
+        _opusTranscodeQueue = opusTranscodeQueue;
+        _formatResolver = formatResolver;
         _logger = logger;
     }
 
@@ -219,6 +227,11 @@ public class UnifiedTrackService
         // frontend, so a failure here is logged and swallowed — never fails the upload.
         await TryStoreWaveformDatumsAsync(unpersisted.EntryKey, ct);
 
+        // Schedule the low-data Opus derive (OQ6 / §3.1a): the track is persisted and lossless-playable
+        // NOW; the transcode + seek-index build run on a background worker. Non-blocking and best-effort
+        // — the upload response never waits on it, and a transcode failure leaves the track lossless-only.
+        _opusTranscodeQueue.Enqueue(unpersisted.EntryKey);
+
         return saveResult;
     }
 
@@ -297,6 +310,11 @@ public class UnifiedTrackService
             return Result.CreateFailResult("Audio replaced but duration metadata could not be updated.");
         }
 
+        // The stale Opus artifact (if any) no longer matches the new source. Schedule a background
+        // regenerate — the transcode service overwrites the prior artifacts in place keyed by the same
+        // EntryKey. Best-effort, off the request thread, mirrors the waveform regen above.
+        _opusTranscodeQueue.Enqueue(entryKey);
+
         return Result.CreatePassResult();
     }
 
@@ -379,6 +397,69 @@ public class UnifiedTrackService
         return ResultContainer<(int, int)>.CreatePassResult((updated, skipped));
     }
 
+    /// <summary>
+    /// Backfill-Opus (18.5, OQ4): enqueue a background Opus derive for every non-deleted track that lacks a
+    /// complete Opus artifact (missing audio OR missing sidecar — a half-derived track is treated as missing
+    /// and re-derived). Mirrors the duration-backfill posture: enumerate SQL rows, check each against the
+    /// <c>track-opus</c> vault, schedule the misses. Enqueue-only and non-blocking — the actual transcodes run
+    /// on the shared background worker, serially (the same queue the upload/replace paths feed), so this
+    /// returns as soon as the misses are scheduled rather than waiting on CPU-heavy transcodes. Idempotent:
+    /// a re-run only enqueues tracks still missing Opus, and already-queued/in-flight derives simply overwrite
+    /// in place. Returns (enqueued, skipped) — skipped = tracks that already have a complete Opus artifact.
+    /// </summary>
+    public async Task<ResultContainer<(int Enqueued, int Skipped)>> BackfillOpusAsync(CancellationToken ct)
+    {
+        var all = await _sqlTrackService.GetAll();
+        if (!all.Success || all.Value is null)
+        {
+            var error = all.Messages.FirstOrDefault()?.Message ?? "Unknown error";
+            _logger.LogError("BackfillOpusAsync: failed to load tracks: {Error}", error);
+            return ResultContainer<(int, int)>.CreateFailResult($"Could not load tracks: {error}");
+        }
+
+        var enqueued = 0;
+        var skipped = 0;
+        foreach (var track in all.Value)
+        {
+            ct.ThrowIfCancellationRequested();
+
+            if (await _formatResolver.HasOpusAsync(track.EntryKey))
+            {
+                skipped++;
+                continue;
+            }
+
+            _opusTranscodeQueue.Enqueue(track.EntryKey);
+            enqueued++;
+        }
+
+        _logger.LogInformation("BackfillOpusAsync complete: {Enqueued} enqueued, {Skipped} already had Opus.",
+            enqueued, skipped);
+        return ResultContainer<(int, int)>.CreatePassResult((enqueued, skipped));
+    }
+
+    /// <summary>
+    /// Per-track Opus (re)derive trigger (18.5): schedule a background transcode for one track. Returns false
+    /// only when the track id is unknown; the enqueue itself is non-blocking and best-effort, like the bulk
+    /// backfill. Re-runnable — overwrites any prior artifact in place.
+    /// </summary>
+    public async Task<Result> EnqueueOpusAsync(string entryKey, CancellationToken ct)
+    {
+        var lookup = await _sqlTrackService.GetByEntryKey(entryKey);
+        if (!lookup.Success)
+        {
+            var error = lookup.Messages.FirstOrDefault()?.Message ?? "Unknown error";
+            _logger.LogError("EnqueueOpusAsync: lookup failed for {EntryKey}: {Error}", entryKey, error);
+            return Result.CreateFailResult("Failed to load track.");
+        }
+
+        if (lookup.Value is null)
+            return Result.CreateFailResult(TrackNotFoundMessage);
+
+        _opusTranscodeQueue.Enqueue(entryKey);
+        return Result.CreatePassResult();
+    }
+
     /// <summary>
     /// Delete a track's SQL row, then its vault entry. SQL is the source of truth: a SQL delete
     /// failure fails the operation (and leaves the vault untouched), but a subsequent vault delete