Implement media storage and processing features: update configuration in .env.example and appsettings files, enhance Docker setup for media storage, and add media-related services and database entities. Include ffmpeg for media handling and adjust Kestrel settings for large file uploads.

2026-07-24 08:33:11 +03:00
parent 1dd6991174
commit e15ecbdb29
47 changed files with 2599 additions and 1 deletions
@@ -0,0 +1,163 @@
+using Microsoft.EntityFrameworkCore;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
+using TeleWave.Application.Common.Interfaces;
+using TeleWave.Domain.Media;
+
+namespace TeleWave.Infrastructure.Media;
+
+/// <summary>
+/// Единственный потребитель очереди обработки: по одному ассету за раз прогоняет через ffmpeg.
+/// На старте восстанавливает прерванные задачи (Pending/Processing) — переживает рестарт/краш.
+/// БД-контекст держится короткими отрезками (пометить статус), сам транскод идёт вне scope, чтобы
+/// не держать соединение открытым минутами.
+/// </summary>
+public sealed class MediaProcessingBackgroundService(
+    IMediaProcessingQueue queue,
+    IServiceScopeFactory scopeFactory,
+    MediaPathResolver paths,
+    IMediaProcessor processor,
+    IOptions<StorageOptions> storageOptions,
+    ILogger<MediaProcessingBackgroundService> logger
+) : BackgroundService
+{
+    private readonly StorageOptions _storage = storageOptions.Value;
+
+    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+    {
+        paths.EnsureDirectories();
+        await RecoverPendingAsync(stoppingToken);
+
+        await foreach (var assetId in queue.DequeueAllAsync(stoppingToken))
+        {
+            try
+            {
+                await ProcessAsync(assetId, stoppingToken);
+            }
+            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
+            {
+                break;
+            }
+            catch (Exception ex)
+            {
+                logger.LogError(ex, "Необработанная ошибка обработки ассета {AssetId}", assetId);
+            }
+        }
+    }
+
+    private async Task RecoverPendingAsync(CancellationToken cancellationToken)
+    {
+        await using var scope = scopeFactory.CreateAsyncScope();
+        var db = scope.ServiceProvider.GetRequiredService<IAppDbContext>();
+
+        var pending = await db.MediaAssets
+            .Where(x =>
+                x.Status == MediaAssetStatus.Pending || x.Status == MediaAssetStatus.Processing
+            )
+            .ToListAsync(cancellationToken);
+
+        foreach (var asset in pending.Where(x => x.Status == MediaAssetStatus.Processing))
+            asset.ResetToPending();
+
+        if (pending.Count > 0)
+            await db.SaveChangesAsync(cancellationToken);
+
+        foreach (var asset in pending)
+            queue.Enqueue(asset.Id);
+    }
+
+    private async Task ProcessAsync(Guid assetId, CancellationToken cancellationToken)
+    {
+        var extension = await BeginProcessingAsync(assetId, cancellationToken);
+        if (extension is null)
+            return;
+
+        try
+        {
+            var result = await processor.ProcessAsync(assetId, extension, cancellationToken);
+            await CompleteAsync(assetId, result, cancellationToken);
+
+            if (!_storage.KeepOriginals)
+                DeleteOriginal(assetId, extension);
+
+            logger.LogInformation(
+                "Ассет {AssetId} обработан: {Segments} сегментов, {Seconds:0.#}с",
+                assetId,
+                result.SegmentCount,
+                result.Duration.TotalSeconds
+            );
+        }
+        catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
+        {
+            throw;
+        }
+        catch (Exception ex)
+        {
+            logger.LogError(ex, "Обработка ассета {AssetId} провалилась", assetId);
+            await FailAsync(assetId, ex.Message, CancellationToken.None);
+        }
+    }
+
+    /// <summary>Помечает ассет Processing и возвращает его расширение, либо null если обрабатывать нечего.</summary>
+    private async Task<string?> BeginProcessingAsync(Guid assetId, CancellationToken cancellationToken)
+    {
+        await using var scope = scopeFactory.CreateAsyncScope();
+        var db = scope.ServiceProvider.GetRequiredService<IAppDbContext>();
+
+        var asset = await db.MediaAssets.FirstOrDefaultAsync(x => x.Id == assetId, cancellationToken);
+        if (asset is null || asset.Status is MediaAssetStatus.Ready or MediaAssetStatus.Failed)
+            return null;
+
+        asset.MarkProcessing();
+        await db.SaveChangesAsync(cancellationToken);
+        return asset.OriginalExtension;
+    }
+
+    private async Task CompleteAsync(
+        Guid assetId,
+        MediaProcessingResult result,
+        CancellationToken cancellationToken
+    )
+    {
+        await using var scope = scopeFactory.CreateAsyncScope();
+        var db = scope.ServiceProvider.GetRequiredService<IAppDbContext>();
+
+        var asset = await db.MediaAssets.FirstOrDefaultAsync(x => x.Id == assetId, cancellationToken);
+        if (asset is null)
+            return;
+
+        asset.MarkReady(
+            result.Duration,
+            result.SegmentSeconds,
+            result.SegmentCount,
+            result.Width,
+            result.Height,
+            result.VideoCodec,
+            result.AudioCodec,
+            result.RelativePath
+        );
+        await db.SaveChangesAsync(cancellationToken);
+    }
+
+    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(x => x.Id == assetId, cancellationToken);
+        if (asset is null)
+            return;
+
+        asset.MarkFailed(error);
+        await db.SaveChangesAsync(cancellationToken);
+    }
+
+    private void DeleteOriginal(Guid assetId, string extension)
+    {
+        var original = paths.OriginalPath(assetId, extension);
+        if (File.Exists(original))
+            File.Delete(original);
+    }
+}