using System.Collections.Concurrent;
using System.Linq.Expressions;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using TeleWave.Application.Common.Interfaces;
using TeleWave.Domain.Media;

namespace TeleWave.Infrastructure.Media;

/// <summary>
/// Каркас воркера, разбирающего очередь ассетов из БД. Источник истины — статус: единственный
/// диспетчер последовательно и атомарно захватывает следующий Pending (помечает Processing), поэтому
/// два воркера никогда не возьмут один ассет; сама работа идёт в фоне с ограничением по числу слотов.
/// Рестарт/краш ничего не теряет — прерванные Processing сбрасываются в Pending на старте.
/// БД-контекст держится короткими отрезками (пометить статус), сама работа идёт вне scope, чтобы не
/// держать соединение минутами.
///
/// Пространство ассетов делится по <see cref="HandlesGenerated"/>: ТВ-заставки (Source=Generated)
/// рендерит один воркер, всё остальное транскодирует другой, и пересечься они не могут.
/// </summary>
internal abstract class MediaClaimingBackgroundService<TJob>(
    IServiceScopeFactory scopeFactory,
    ILogger logger
) : BackgroundService
    where TJob : class
{
    // Период опроса вынесен в неродовой класс: статическое поле обобщённого типа существует
    // отдельной копией на каждую замкнутую специализацию, то есть выглядит общей константой,
    // не будучи ею. Здесь значение одинаковое и вреда нет, но читателя это вводит в заблуждение.
    private static TimeSpan IdlePoll => MediaClaimingDefaults.IdlePoll;

    /// <summary>true — воркер обслуживает только сгенерированные ассеты, false — только остальные.</summary>
    protected abstract bool HandlesGenerated { get; }

    /// <summary>Сколько работ выполняется одновременно. 1 — строго последовательно.</summary>
    protected virtual int MaxParallel => 1;

    /// <summary>Что писать в лог при сбое цикла (не самой работы).</summary>
    protected abstract string LoopErrorMessage { get; }

    /// <summary>Ожидание сигнала о новой работе — у каждого воркера своя очередь-будильник.</summary>
    protected abstract ValueTask WaitForWorkAsync(CancellationToken cancellationToken);

    /// <summary>Что из захваченного ассета нужно воркеру: работа идёт уже без БД-контекста.</summary>
    protected abstract TJob ToJob(MediaAsset asset);

    /// <summary>Сама работа над захваченным (уже Processing) ассетом.</summary>
    protected abstract Task ProcessAsync(TJob job, CancellationToken cancellationToken);

    /// <summary>Разовая подготовка перед первым тиком (например, создание каталогов).</summary>
    protected virtual void OnStarting() { }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        OnStarting();
        await ResetInterruptedAsync(stoppingToken);

        // Слоты параллелизма: не запускаем больше MaxParallel работ одновременно.
        using var slots = new SemaphoreSlim(MaxParallel, MaxParallel);
        var inFlight = new ConcurrentDictionary<Task, byte>();

        while (!stoppingToken.IsCancellationRequested)
        {
            try
            {
                // Захватываем и раздаём по слотам всю накопившуюся работу из БД.
                while (!stoppingToken.IsCancellationRequested)
                {
                    await slots.WaitAsync(stoppingToken);

                    // Слот уже захвачен — любой сбой захвата ассета (транзиентная ошибка БД и т.п.)
                    // обязан вернуть слот, иначе после нескольких ошибок семафор исчерпается и
                    // диспетчер зависнет навсегда (сервис формально жив, но ничего не обрабатывает).
                    TJob? claim;
                    try
                    {
                        claim = await ClaimNextAsync(stoppingToken);
                    }
                    catch
                    {
                        slots.Release();
                        throw;
                    }

                    if (claim is not { } job)
                    {
                        slots.Release();
                        break;
                    }

                    var task = Task.Run(
                        async () =>
                        {
                            try
                            {
                                await ProcessAsync(job, stoppingToken);
                            }
                            finally
                            {
                                slots.Release();
                            }
                        },
                        CancellationToken.None
                    );
                    inFlight[task] = 0;
                    _ = task.ContinueWith(t => inFlight.TryRemove(t, out _), TaskScheduler.Default);
                }

                await WaitForNextRoundAsync(stoppingToken);
            }
            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
            {
                break;
            }
            catch (Exception ex)
            {
                logger.LogError(ex, "{Message}", LoopErrorMessage);
                await Task.Delay(TimeSpan.FromSeconds(5), stoppingToken);
            }
        }

        // Даём уже запущенным работам корректно завершиться (или отмениться) на остановке.
        try
        {
            await Task.WhenAll(inFlight.Keys.ToArray());
        }
        catch
        {
            // Ошибки/отмена отдельных задач уже залогированы внутри ProcessAsync.
        }
    }

    /// <summary>
    /// Работы нет — ждём сигнала о новой либо истечения периода опроса. Тайм-аут здесь штатный
    /// выход, а не сбой: отмена по остановке сервиса пробрасывается наружу, отмена по тайм-ауту
    /// гасится и приводит к обычной перепроверке БД.
    /// </summary>
    private async Task WaitForNextRoundAsync(CancellationToken stoppingToken)
    {
        using var wake = CancellationTokenSource.CreateLinkedTokenSource(stoppingToken);
        wake.CancelAfter(IdlePoll);
        try
        {
            await WaitForWorkAsync(wake.Token);
        }
        catch (OperationCanceledException) when (!stoppingToken.IsCancellationRequested)
        {
            // Тайм-аут опроса — просто перепроверяем БД.
        }
    }

    /// <summary>Помечает ассет провалившимся. Вызывается воркером из его обработчика ошибок.</summary>
    protected async Task FailAsync(
        Guid assetId,
        string error,
        CancellationToken cancellationToken
    ) => await WithAssetAsync(assetId, asset => asset.MarkFailed(error), cancellationToken);

    /// <summary>Находит ассет в свежем scope, применяет к нему изменение и сохраняет.</summary>
    protected async Task WithAssetAsync(
        Guid assetId,
        Action<MediaAsset> change,
        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;

        change(asset);
        await db.SaveChangesAsync(cancellationToken);
    }

    /// <summary>
    /// Выполняет что-то на scoped-сервисе в отдельном scope — для сборки данных под работу.
    /// Scope живёт только на время вызова: соединение с БД не удерживается на весь рендер/транскод.
    /// </summary>
    protected async Task<T> WithScopeAsync<TService, T>(Func<TService, Task<T>> use)
        where TService : notnull
    {
        await using var scope = scopeFactory.CreateAsyncScope();
        return await use(scope.ServiceProvider.GetRequiredService<TService>());
    }

    /// <summary>Сброс прерванных рестартом задач (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(Owned(MediaAssetStatus.Processing))
            .ToListAsync(cancellationToken);
        if (interrupted.Count == 0)
            return;

        foreach (var asset in interrupted)
            asset.ResetToPending();
        await db.SaveChangesAsync(cancellationToken);
    }

    /// <summary>
    /// Атомарно захватывает самый ранний Pending этого воркера: помечает его Processing и возвращает
    /// job, либо null если работы нет. Вызывается только диспетчером последовательно, поэтому две
    /// работы не возьмут один ассет.
    /// </summary>
    private async Task<TJob?> ClaimNextAsync(CancellationToken cancellationToken)
    {
        await using var scope = scopeFactory.CreateAsyncScope();
        var db = scope.ServiceProvider.GetRequiredService<IAppDbContext>();

        var asset = await db
            .MediaAssets.Where(Owned(MediaAssetStatus.Pending))
            .OrderBy(x => x.CreatedAt)
            .FirstOrDefaultAsync(cancellationToken);
        if (asset is null)
            return default;

        asset.MarkProcessing();
        await db.SaveChangesAsync(cancellationToken);
        return ToJob(asset);
    }

    /// <summary>Ассеты этого воркера в заданном статусе — предикат переводится в SQL.</summary>
    private Expression<Func<MediaAsset, bool>> Owned(MediaAssetStatus status) =>
        HandlesGenerated
            ? x => x.Status == status && x.Source == MediaSource.Generated
            : x => x.Status == status && x.Source != MediaSource.Generated;
}

/// <summary>Общие для всех воркеров значения — одно на процесс, а не на специализацию.</summary>
internal static class MediaClaimingDefaults
{
    /// <summary>Периодически перепроверяем БД, даже если сигнал не пришёл — страховка на любой случай.</summary>
    public static readonly TimeSpan IdlePoll = TimeSpan.FromSeconds(30);
}
