Implement asynchronous bumper rendering and add integration tests: refactor BumperAsset to support async rendering, introduce BumperRenderBackgroundService for processing, and create TeleWave.Integration.Tests with Testcontainers-Postgres for comprehensive testing of scheduling and bumper generation scenarios. Update documentation and ensure tests skip if Docker is not available.

2026-07-25 23:42:40 +03:00
parent 6c18a9da79
commit d6ad8e11a4
22 changed files with 1967 additions and 197 deletions
@@ -158,8 +158,10 @@ public static class DependencyInjection
        services.AddSingleton<IAudioProbe, FfprobeAudioProbe>();
        services.AddSingleton<IMediaProcessor, FfmpegMediaProcessor>();
        services.AddSingleton<IMediaProcessingQueue, MediaProcessingQueue>();
+        services.AddSingleton<IBumperRenderQueue, BumperRenderQueue>();

        services.AddHostedService<MediaProcessingBackgroundService>();
        services.AddHostedService<InboxScannerBackgroundService>();
+        services.AddHostedService<BumperRenderBackgroundService>();
    }
 }
@@ -0,0 +1,273 @@
+using Microsoft.EntityFrameworkCore;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
+using TeleWave.Application.Broadcast.Bumpers;
+using TeleWave.Application.Broadcast.Scheduling;
+using TeleWave.Application.Common.Interfaces;
+using TeleWave.Application.Streaming;
+using TeleWave.Domain.Media;
+
+namespace TeleWave.Infrastructure.Media;
+
+/// <summary>
+/// Асинхронно рендерит ТВ-заставки расписания: планировщик лишь создаёт ассет (Source=Generated) в
+/// статусе Pending и кэш-строку <see cref="Domain.Broadcast.BumperAsset"/>, а сам ffmpeg крутится здесь,
+/// вне тика планировщика и его транзакции. Источник истины — статус в БД (последовательно берём
+/// следующий Pending c Source=Generated, помечаем Processing), поэтому рестарт/краш ничего не теряет
+/// (прерванные Processing сбрасываются в Pending на старте). До готовности ассета плейлист отдаёт филлер.
+/// </summary>
+public sealed class BumperRenderBackgroundService(
+    IBumperRenderQueue queue,
+    IServiceScopeFactory scopeFactory,
+    IBumperRenderer renderer,
+    IBumperTemplateStorage bumperStorage,
+    IImageStore imageStore,
+    IOptions<BumperOptions> bumperOptions,
+    IOptions<StreamingOptions> streamingOptions,
+    ILogger<BumperRenderBackgroundService> logger
+) : BackgroundService
+{
+    private static readonly TimeSpan IdlePoll = TimeSpan.FromSeconds(30);
+    private readonly BumperOptions _bumper = bumperOptions.Value;
+    private readonly int _segmentSeconds = Math.Max(1, streamingOptions.Value.SegmentSeconds);
+
+    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+    {
+        await ResetInterruptedAsync(stoppingToken);
+
+        while (!stoppingToken.IsCancellationRequested)
+        {
+            try
+            {
+                // Разбираем всю накопившуюся работу из БД.
+                while (!stoppingToken.IsCancellationRequested)
+                {
+                    var assetId = await ClaimNextAsync(stoppingToken);
+                    if (assetId is not { } id)
+                        break;
+                    await RenderClaimedAsync(id, stoppingToken);
+                }
+
+                using var wake = CancellationTokenSource.CreateLinkedTokenSource(stoppingToken);
+                wake.CancelAfter(IdlePoll);
+                try
+                {
+                    await queue.WaitAsync(wake.Token);
+                }
+                catch (OperationCanceledException) when (!stoppingToken.IsCancellationRequested)
+                {
+                    // Тайм-аут опроса — просто перепроверяем БД.
+                }
+            }
+            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
+            {
+                break;
+            }
+            catch (Exception ex)
+            {
+                logger.LogError(ex, "Ошибка цикла рендера заставок");
+                await Task.Delay(TimeSpan.FromSeconds(5), stoppingToken);
+            }
+        }
+    }
+
+    /// <summary>Сброс прерванных рестартом заставок (Generated Processing → Pending) на старте.</summary>
+    private async Task ResetInterruptedAsync(CancellationToken cancellationToken)
+    {
+        await using var scope = scopeFactory.CreateAsyncScope();
+        var db = scope.ServiceProvider.GetRequiredService<IAppDbContext>();
+
+        var interrupted = await db
+            .MediaAssets.Where(x =>
+                x.Status == MediaAssetStatus.Processing && x.Source == MediaSource.Generated
+            )
+            .ToListAsync(cancellationToken);
+        if (interrupted.Count == 0)
+            return;
+
+        foreach (var asset in interrupted)
+            asset.ResetToPending();
+        await db.SaveChangesAsync(cancellationToken);
+    }
+
+    /// <summary>Атомарно захватывает самую раннюю Pending-заставку (Generated): Pending → Processing.</summary>
+    private async Task<Guid?> ClaimNextAsync(CancellationToken cancellationToken)
+    {
+        await using var scope = scopeFactory.CreateAsyncScope();
+        var db = scope.ServiceProvider.GetRequiredService<IAppDbContext>();
+
+        var asset = await db
+            .MediaAssets.Where(x =>
+                x.Status == MediaAssetStatus.Pending && x.Source == MediaSource.Generated
+            )
+            .OrderBy(x => x.CreatedAt)
+            .FirstOrDefaultAsync(cancellationToken);
+        if (asset is null)
+            return null;
+
+        asset.MarkProcessing();
+        await db.SaveChangesAsync(cancellationToken);
+        return asset.Id;
+    }
+
+    private async Task RenderClaimedAsync(Guid assetId, CancellationToken cancellationToken)
+    {
+        try
+        {
+            await using var scope = scopeFactory.CreateAsyncScope();
+            var db = scope.ServiceProvider.GetRequiredService<IAppDbContext>();
+
+            var spec = await BuildSpecAsync(db, assetId, cancellationToken);
+            if (spec is null)
+            {
+                await FailAsync(assetId, "Не удалось восстановить спецификацию заставки", cancellationToken);
+                return;
+            }
+
+            var render = await renderer.RenderAsync(assetId, spec, cancellationToken);
+
+            var asset = await db.MediaAssets.FirstOrDefaultAsync(
+                a => a.Id == assetId,
+                cancellationToken
+            );
+            if (asset is null)
+                return;
+            asset.MarkReady(
+                render.Duration,
+                render.SegmentSeconds,
+                render.SegmentCount,
+                render.Width,
+                render.Height,
+                "h264",
+                "aac",
+                render.RelativePath
+            );
+            await db.SaveChangesAsync(cancellationToken);
+        }
+        catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
+        {
+            throw;
+        }
+        catch (Exception ex)
+        {
+            logger.LogError(ex, "Рендер заставки {AssetId} провалился", assetId);
+            await FailAsync(assetId, ex.Message, CancellationToken.None);
+        }
+    }
+
+    /// <summary>Восстанавливает <see cref="BumperRenderSpec"/> по кэш-строке заставки (канал/блок/подблок).</summary>
+    private async Task<BumperRenderSpec?> BuildSpecAsync(
+        IAppDbContext db,
+        Guid assetId,
+        CancellationToken cancellationToken
+    )
+    {
+        var cache = await db
+            .BumperAssets.AsNoTracking()
+            .Where(b => b.MediaAssetId == assetId)
+            .OrderByDescending(b => b.CreatedAt)
+            .FirstOrDefaultAsync(cancellationToken);
+        if (cache is null)
+            return null;
+
+        var channel = await db
+            .Channels.AsNoTracking()
+            .Include(c => c.BumperTemplates)
+                .ThenInclude(t => t.Variants)
+            .FirstOrDefaultAsync(c => c.Id == cache.ChannelId, cancellationToken);
+        var template = channel?.BumperTemplates.FirstOrDefault(t => t.Id == cache.TemplateId);
+        var variant = template?.Variants.FirstOrDefault(v => v.Id == cache.VariantId);
+        if (channel is null || template is null || variant is null)
+            return null;
+
+        var names = await db
+            .Shows.AsNoTracking()
+            .Where(s => s.Id == cache.FromShowId || s.Id == cache.ToShowId)
+            .Select(s => new { s.Id, s.Name })
+            .ToDictionaryAsync(s => s.Id, s => s.Name, cancellationToken);
+        var fromName = names.GetValueOrDefault(cache.FromShowId, "…");
+        var toName = names.GetValueOrDefault(cache.ToShowId, "…");
+
+        // Постер шоу-получателя как фон — только для «Сейчас/Далее».
+        string? posterPath = null;
+        if (variant.Kind == Domain.Broadcast.BumperTextKind.NowNext)
+            posterPath = await ResolveShowPosterAsync(db, cache.ToShowId, cancellationToken);
+
+        var bgPath = await ResolveTemplateBackgroundAsync(db, template.BackgroundImageId, cancellationToken);
+        var aligned = BumperDuration.Aligned(
+            BumperDuration.TemplateSeconds(template),
+            _segmentSeconds
+        );
+        var audioPath = bumperStorage.AudioPath(template.Id, template.AudioExtension);
+
+        return BumperSpecFactory.Build(
+            _bumper,
+            channel.BumperFont,
+            template,
+            variant,
+            aligned,
+            fromName,
+            toName,
+            audioPath,
+            posterPath,
+            bgPath
+        );
+    }
+
+    private async Task<string?> ResolveShowPosterAsync(
+        IAppDbContext db,
+        Guid showId,
+        CancellationToken cancellationToken
+    )
+    {
+        var posterImageId = await db
+            .Shows.AsNoTracking()
+            .Where(s => s.Id == showId && s.PosterImageId != null)
+            .Select(s => s.PosterImageId!.Value)
+            .FirstOrDefaultAsync(cancellationToken);
+        if (posterImageId == Guid.Empty)
+            return null;
+        return await ResolveImagePathAsync(db, posterImageId, cancellationToken);
+    }
+
+    private async Task<string?> ResolveTemplateBackgroundAsync(
+        IAppDbContext db,
+        Guid? backgroundImageId,
+        CancellationToken cancellationToken
+    )
+    {
+        if (backgroundImageId is not { } bgId)
+            return null;
+        return await ResolveImagePathAsync(db, bgId, cancellationToken);
+    }
+
+    private async Task<string?> ResolveImagePathAsync(
+        IAppDbContext db,
+        Guid imageId,
+        CancellationToken cancellationToken
+    )
+    {
+        var ext = await db
+            .Images.AsNoTracking()
+            .Where(i => i.Id == imageId)
+            .Select(i => i.FileExtension)
+            .FirstOrDefaultAsync(cancellationToken);
+        return ext is null ? null : imageStore.ResolvePath(imageId, ext);
+    }
+
+    private async Task FailAsync(Guid assetId, string error, CancellationToken cancellationToken)
+    {
+        await using var scope = scopeFactory.CreateAsyncScope();
+        var db = scope.ServiceProvider.GetRequiredService<IAppDbContext>();
+        var asset = await db.MediaAssets.FirstOrDefaultAsync(
+            a => a.Id == assetId,
+            cancellationToken
+        );
+        if (asset is null)
+            return;
+        asset.MarkFailed(error);
+        await db.SaveChangesAsync(cancellationToken);
+    }
+}
@@ -0,0 +1,20 @@
+using System.Threading.Channels;
+using TeleWave.Application.Common.Interfaces;
+
+namespace TeleWave.Infrastructure.Media;
+
+/// <summary>Сигнальная очередь-будильник поверх Channel (id ассета — лишь сигнал; работу берём из БД).</summary>
+public sealed class BumperRenderQueue : IBumperRenderQueue
+{
+    private readonly Channel<Guid> _channel = Channel.CreateUnbounded<Guid>(
+        new UnboundedChannelOptions { SingleReader = true }
+    );
+
+    public void Enqueue(Guid assetId) => _channel.Writer.TryWrite(assetId);
+
+    public async ValueTask WaitAsync(CancellationToken cancellationToken)
+    {
+        await _channel.Reader.ReadAsync(cancellationToken);
+        while (_channel.Reader.TryRead(out _)) { }
+    }
+}
@@ -130,7 +130,9 @@ public sealed class MediaProcessingBackgroundService(
        var db = scope.ServiceProvider.GetRequiredService<IAppDbContext>();

        var interrupted = await db
-            .MediaAssets.Where(x => x.Status == MediaAssetStatus.Processing)
+            .MediaAssets.Where(x =>
+                x.Status == MediaAssetStatus.Processing && x.Source != MediaSource.Generated
+            )
            .ToListAsync(cancellationToken);
        if (interrupted.Count == 0)
            return;
@@ -152,8 +154,11 @@ public sealed class MediaProcessingBackgroundService(
        await using var scope = scopeFactory.CreateAsyncScope();
        var db = scope.ServiceProvider.GetRequiredService<IAppDbContext>();

+        // Generated-ассеты (ТВ-заставки) обслуживает BumperRenderBackgroundService — их не берём.
        var asset = await db
-            .MediaAssets.Where(x => x.Status == MediaAssetStatus.Pending)
+            .MediaAssets.Where(x =>
+                x.Status == MediaAssetStatus.Pending && x.Source != MediaSource.Generated
+            )
            .OrderBy(x => x.CreatedAt)
            .FirstOrDefaultAsync(cancellationToken);
        if (asset is null)
@@ -0,0 +1,52 @@
+using System;
+using Microsoft.EntityFrameworkCore.Migrations;
+
+#nullable disable
+
+namespace TeleWave.Infrastructure.Migrations
+{
+    /// <inheritdoc />
+    public partial class BumperAssetRenderContext : Migration
+    {
+        /// <inheritdoc />
+        protected override void Up(MigrationBuilder migrationBuilder)
+        {
+            migrationBuilder.AddColumn<Guid>(
+                name: "ChannelId",
+                table: "BumperAssets",
+                type: "uuid",
+                nullable: false,
+                defaultValue: new Guid("00000000-0000-0000-0000-000000000000"));
+
+            migrationBuilder.AddColumn<Guid>(
+                name: "TemplateId",
+                table: "BumperAssets",
+                type: "uuid",
+                nullable: false,
+                defaultValue: new Guid("00000000-0000-0000-0000-000000000000"));
+
+            migrationBuilder.AddColumn<Guid>(
+                name: "VariantId",
+                table: "BumperAssets",
+                type: "uuid",
+                nullable: false,
+                defaultValue: new Guid("00000000-0000-0000-0000-000000000000"));
+        }
+
+        /// <inheritdoc />
+        protected override void Down(MigrationBuilder migrationBuilder)
+        {
+            migrationBuilder.DropColumn(
+                name: "ChannelId",
+                table: "BumperAssets");
+
+            migrationBuilder.DropColumn(
+                name: "TemplateId",
+                table: "BumperAssets");
+
+            migrationBuilder.DropColumn(
+                name: "VariantId",
+                table: "BumperAssets");
+        }
+    }
+}
@@ -164,6 +164,9 @@ namespace TeleWave.Infrastructure.Migrations
                    b.Property<Guid>("Id")
                        .HasColumnType("uuid");

+                    b.Property<Guid>("ChannelId")
+                        .HasColumnType("uuid");
+
                    b.Property<DateTimeOffset>("CreatedAt")
                        .HasColumnType("timestamp with time zone");

@@ -178,9 +181,15 @@ namespace TeleWave.Infrastructure.Migrations
                        .HasMaxLength(128)
                        .HasColumnType("character varying(128)");

+                    b.Property<Guid>("TemplateId")
+                        .HasColumnType("uuid");
+
                    b.Property<Guid>("ToShowId")
                        .HasColumnType("uuid");

+                    b.Property<Guid>("VariantId")
+                        .HasColumnType("uuid");
+
                    b.HasKey("Id");

                    b.HasIndex("FromShowId", "ToShowId", "Signature");