ersatztv/ErsatzTV/Services/SchedulerService.cs

using System.Globalization;
using System.Threading.Channels;
using ErsatzTV.Application;
using ErsatzTV.Application.Channels;
using ErsatzTV.Application.Emby;
using ErsatzTV.Application.FFmpeg;
using ErsatzTV.Application.Graphics;
using ErsatzTV.Application.Jellyfin;
using ErsatzTV.Application.Maintenance;
using ErsatzTV.Application.MediaCollections;
using ErsatzTV.Application.MediaSources;
using ErsatzTV.Application.Playouts;
using ErsatzTV.Application.Plex;
using ErsatzTV.Core;
using ErsatzTV.Core.Domain;
using ErsatzTV.Core.Interfaces.FFmpeg;
using ErsatzTV.Core.Interfaces.Locking;
using ErsatzTV.Core.Scheduling;
using ErsatzTV.Infrastructure.Data;
using Microsoft.EntityFrameworkCore;

namespace ErsatzTV.Services;

public class SchedulerService : BackgroundService
{
    private readonly IEntityLocker _entityLocker;
    private readonly ILogger<SchedulerService> _logger;
    private readonly ChannelWriter<IScannerBackgroundServiceRequest> _scannerWorkerChannel;
    private readonly IServiceScopeFactory _serviceScopeFactory;
    private readonly SystemStartup _systemStartup;
    private readonly ChannelWriter<IBackgroundServiceRequest> _workerChannel;

    public SchedulerService(
        IServiceScopeFactory serviceScopeFactory,
        ChannelWriter<IBackgroundServiceRequest> workerChannel,
        ChannelWriter<IScannerBackgroundServiceRequest> scannerWorkerChannel,
        IEntityLocker entityLocker,
        SystemStartup systemStartup,
        ILogger<SchedulerService> logger)
    {
        _serviceScopeFactory = serviceScopeFactory;
        _workerChannel = workerChannel;
        _scannerWorkerChannel = scannerWorkerChannel;
        _entityLocker = entityLocker;
        _systemStartup = systemStartup;
        _logger = logger;
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        await Task.Yield();

        await _systemStartup.WaitForSearchIndex(stoppingToken);
        if (stoppingToken.IsCancellationRequested)
        {
            return;
        }

        try
        {
            _logger.LogInformation("Scheduler service started");

            DateTime firstRun = DateTime.Now;

            // run once immediately at startup
            if (!stoppingToken.IsCancellationRequested)
            {
                await QueueFFmpegCapabilitiesRefresh(stoppingToken);
                await DoWork(stoppingToken);
            }

            while (!stoppingToken.IsCancellationRequested)
            {
                int currentMinutes = DateTime.Now.TimeOfDay.Minutes;
                int toWait = currentMinutes < 30 ? 30 - currentMinutes : 60 - currentMinutes;
                _logger.LogDebug("Scheduler sleeping for {Minutes} minutes", toWait);

                try
                {
                    await Task.Delay(TimeSpan.FromMinutes(toWait), stoppingToken);
                }
                catch (Exception ex) when (ex is TaskCanceledException or OperationCanceledException)
                {
                    // do nothing
                }

                if (!stoppingToken.IsCancellationRequested)
                {
                    var roundedMinute = (int)(Math.Round(DateTime.Now.Minute / 5.0) * 5);
                    if (roundedMinute % 30 == 0)
                    {
                        // check for playouts to reset every 30 minutes
                        await ResetPlayouts(stoppingToken);
                    }

                    if (roundedMinute % 60 == 0 && DateTime.Now.Subtract(firstRun) > TimeSpan.FromHours(1))
                    {
                        // do other work every hour (on the hour)
                        await DoWork(stoppingToken);
                    }
                    else if (roundedMinute % 30 == 0)
                    {
                        // release memory every 30 minutes no matter what
                        await ReleaseMemory(stoppingToken);
                    }
                }
            }

            stoppingToken.ThrowIfCancellationRequested();
        }
        catch (Exception ex) when (ex is TaskCanceledException or OperationCanceledException)
        {
            _logger.LogInformation("Scheduler service shutting down");
        }
    }

    private async Task DoWork(CancellationToken cancellationToken)
    {
        try
        {
            await DeleteOrphanedSubtitles(cancellationToken);
            await RefreshMpegTsScripts(cancellationToken);
            await RefreshChannelGuideChannelList(cancellationToken);
            await BuildPlayouts(cancellationToken);
#if !DEBUG_NO_SYNC
            await ScanLocalMediaSources(cancellationToken);
            await ScanPlexMediaSources(cancellationToken);
            await ScanJellyfinMediaSources(cancellationToken);
            await ScanEmbyMediaSources(cancellationToken);
#endif
            await RefreshTraktLists(cancellationToken);
            await MatchTraktLists(cancellationToken);

            await RefreshGraphicsElements(cancellationToken);

            await ReleaseMemory(cancellationToken);
        }
        catch (Exception ex) when (ex is TaskCanceledException or OperationCanceledException)
        {
            // do nothing
        }
        catch (Exception ex)
        {
            _logger.LogWarning(ex, "Error during scheduler run");
        }
    }

    private async Task ResetPlayouts(CancellationToken cancellationToken)
    {
        try
        {
            using IServiceScope scope = _serviceScopeFactory.CreateScope();
            TvContext dbContext = scope.ServiceProvider.GetRequiredService<TvContext>();

            List<Playout> playouts = await dbContext.Playouts
                .AsNoTracking()
                .Filter(p => p.DailyRebuildTime != null)
                .Include(p => p.Channel)
                .ToListAsync(cancellationToken);

            foreach (Playout playout in playouts.OrderBy(p => decimal.Parse(
                         p.Channel.Number,
                         CultureInfo.InvariantCulture)))
            {
                DateTime now = DateTime.Now;
                DateTime target = DateTime.Today.Add(playout.DailyRebuildTime ?? TimeSpan.FromDays(7));
                // check absolute diff
                if (now.Subtract(target).Duration() < TimeSpan.FromMinutes(5))
                {
                    await _workerChannel.WriteAsync(
                        new BuildPlayout(playout.Id, PlayoutBuildMode.Reset),
                        cancellationToken);
                }
            }
        }
        catch (Exception ex)
        {
            _logger.LogWarning(ex, "Error during scheduler run");
        }
    }

    private async Task BuildPlayouts(CancellationToken cancellationToken)
    {
        using IServiceScope scope = _serviceScopeFactory.CreateScope();
        TvContext dbContext = scope.ServiceProvider.GetRequiredService<TvContext>();

        List<Playout> playouts = await dbContext.Playouts
            .AsNoTracking()
            .Include(p => p.Channel)
            .ToListAsync(cancellationToken);

        foreach (Playout playout in playouts.OrderBy(p => decimal.Parse(
                     p.Channel.Number,
                     CultureInfo.InvariantCulture)))
        {
            await _workerChannel.WriteAsync(
                new BuildPlayout(playout.Id, PlayoutBuildMode.Continue),
                cancellationToken);
        }
    }

    private ValueTask RefreshChannelGuideChannelList(CancellationToken cancellationToken) =>
        _workerChannel.WriteAsync(new RefreshChannelList(), cancellationToken);

    private async Task ScanLocalMediaSources(CancellationToken cancellationToken)
    {
        using IServiceScope scope = _serviceScopeFactory.CreateScope();
        TvContext dbContext = scope.ServiceProvider.GetRequiredService<TvContext>();

        foreach (int libraryId in dbContext.LocalMediaSources.SelectMany(ms => ms.Libraries).Map(l => l.Id))
        {
            if (_entityLocker.LockLibrary(libraryId))
            {
                try
                {
                    await _scannerWorkerChannel.WriteAsync(new ScanLocalLibraryIfNeeded(libraryId), cancellationToken);
                }
                catch
                {
                    // the message that would release this lock never got enqueued - release it here
                    _entityLocker.UnlockLibrary(libraryId);
                    throw;
                }
            }
        }
    }

    private async Task ScanPlexMediaSources(CancellationToken cancellationToken)
    {
        using IServiceScope scope = _serviceScopeFactory.CreateScope();
        TvContext dbContext = scope.ServiceProvider.GetRequiredService<TvContext>();

        var mediaSourceIds = new System.Collections.Generic.HashSet<int>();

        foreach (PlexLibrary library in dbContext.PlexLibraries.AsNoTracking().Filter(l => l.ShouldSyncItems))
        {
            mediaSourceIds.Add(library.MediaSourceId);

            if (_entityLocker.LockLibrary(library.Id))
            {
                // one LockLibrary ⇄ one release: when a networks scan follows, it (the last message)
                // carries the release; the library message runs with Unlock: false.
                bool networksFollow = library.MediaKind is LibraryMediaKind.Shows;

                try
                {
                    await _scannerWorkerChannel.WriteAsync(
                        new SynchronizePlexLibraryByIdIfNeeded(library.Id, Unlock: !networksFollow),
                        cancellationToken);

                    if (networksFollow)
                    {
                        await _scannerWorkerChannel.WriteAsync(
                            new SynchronizePlexNetworks(library.Id, false),
                            cancellationToken);
                    }
                }
                catch
                {
                    // if enqueueing the batch fails partway (the library message carries Unlock: false,
                    // so the not-yet-enqueued networks message was the sole releaser), release here
                    _entityLocker.UnlockLibrary(library.Id);
                    throw;
                }
            }
        }

        // lock the collections slot once for the whole provider batch: the LAST message owns the
        // single release (Unlock: true), the rest run Unlock: false. If the slot is already held
        // (an API scan or a prior tick's batch is running/queued), skip enqueuing entirely - never
        // enqueue unlocked, which would cross-release the holder's lock (#250). Skipping starves
        // nothing: the next tick retries. (see #235)
        // lock the collections slot once for the whole provider batch: the LAST message owns the
        // single release (Unlock: true), the rest run Unlock: false. If the slot is already held
        // (an API scan or a prior tick's batch is running/queued), skip enqueuing entirely - never
        // enqueue unlocked, which would cross-release the holder's lock (#250). Skipping starves
        // nothing: the next tick retries. (see #235)
        var plexSourceIds = mediaSourceIds.ToList();
        if (plexSourceIds.Count > 0 && _entityLocker.LockPlexCollections())
        {
            try
            {
                for (var i = 0; i < plexSourceIds.Count; i++)
                {
                    bool isLast = i == plexSourceIds.Count - 1;
                    await _scannerWorkerChannel.WriteAsync(
                        new SynchronizePlexCollections(plexSourceIds[i], false, false, Unlock: isLast),
                        cancellationToken);
                }
            }
            catch
            {
                // an enqueue threw before the last (releasing) message went out - compensate here
                _entityLocker.UnlockPlexCollections();
                throw;
            }
        }
    }

    private async Task ScanJellyfinMediaSources(CancellationToken cancellationToken)
    {
        using IServiceScope scope = _serviceScopeFactory.CreateScope();
        TvContext dbContext = scope.ServiceProvider.GetRequiredService<TvContext>();

        var mediaSourceIds = new System.Collections.Generic.HashSet<int>();

        foreach (JellyfinLibrary library in dbContext.JellyfinLibraries.AsNoTracking().Filter(l => l.ShouldSyncItems))
        {
            mediaSourceIds.Add(library.MediaSourceId);

            if (_entityLocker.LockLibrary(library.Id))
            {
                try
                {
                    await _scannerWorkerChannel.WriteAsync(
                        new SynchronizeJellyfinLibraryByIdIfNeeded(library.Id),
                        cancellationToken);
                }
                catch
                {
                    _entityLocker.UnlockLibrary(library.Id);
                    throw;
                }
            }
        }

        // lock-once per provider batch; last message owns the release; skip entirely if held (#235/#250)
        var jellyfinSourceIds = mediaSourceIds.ToList();
        if (jellyfinSourceIds.Count > 0 && _entityLocker.LockJellyfinCollections())
        {
            try
            {
                for (var i = 0; i < jellyfinSourceIds.Count; i++)
                {
                    bool isLast = i == jellyfinSourceIds.Count - 1;
                    await _scannerWorkerChannel.WriteAsync(
                        new SynchronizeJellyfinCollections(jellyfinSourceIds[i], false, false, Unlock: isLast),
                        cancellationToken);
                }
            }
            catch
            {
                _entityLocker.UnlockJellyfinCollections();
                throw;
            }
        }
    }

    private async Task ScanEmbyMediaSources(CancellationToken cancellationToken)
    {
        using IServiceScope scope = _serviceScopeFactory.CreateScope();
        TvContext dbContext = scope.ServiceProvider.GetRequiredService<TvContext>();

        var mediaSourceIds = new System.Collections.Generic.HashSet<int>();

        foreach (EmbyLibrary library in dbContext.EmbyLibraries.AsNoTracking().Filter(l => l.ShouldSyncItems))
        {
            mediaSourceIds.Add(library.MediaSourceId);

            if (_entityLocker.LockLibrary(library.Id))
            {
                try
                {
                    await _scannerWorkerChannel.WriteAsync(
                        new SynchronizeEmbyLibraryByIdIfNeeded(library.Id),
                        cancellationToken);
                }
                catch
                {
                    _entityLocker.UnlockLibrary(library.Id);
                    throw;
                }
            }
        }

        // lock-once per provider batch; last message owns the release; skip entirely if held (#235/#250)
        var embySourceIds = mediaSourceIds.ToList();
        if (embySourceIds.Count > 0 && _entityLocker.LockEmbyCollections())
        {
            try
            {
                for (var i = 0; i < embySourceIds.Count; i++)
                {
                    bool isLast = i == embySourceIds.Count - 1;
                    await _scannerWorkerChannel.WriteAsync(
                        new SynchronizeEmbyCollections(embySourceIds[i], false, false, Unlock: isLast),
                        cancellationToken);
                }
            }
            catch
            {
                _entityLocker.UnlockEmbyCollections();
                throw;
            }
        }
    }

    private async Task RefreshTraktLists(CancellationToken cancellationToken)
    {
        using IServiceScope scope = _serviceScopeFactory.CreateScope();
        TvContext dbContext = scope.ServiceProvider.GetRequiredService<TvContext>();

        DateTime target = DateTime.UtcNow.AddDays(-1);

        List<TraktList> traktLists = await dbContext.TraktLists
            .AsNoTracking()
            .Filter(tl => tl.AutoRefresh && (tl.LastUpdate == null || tl.LastUpdate <= target))
            .ToListAsync(cancellationToken);

        if (traktLists.Count != 0 && _entityLocker.LockTrakt())
        {
            TraktList last = traktLists.Last();
            foreach (TraktList list in traktLists)
            {
                await _workerChannel.WriteAsync(
                    AddTraktList.Existing(list.User, list.List, list == last),
                    cancellationToken);
            }
        }
    }

    private async Task MatchTraktLists(CancellationToken cancellationToken)
    {
        using IServiceScope scope = _serviceScopeFactory.CreateScope();
        TvContext dbContext = scope.ServiceProvider.GetRequiredService<TvContext>();

        DateTime target = DateTime.UtcNow.AddHours(-1);

        List<TraktList> traktLists = await dbContext.TraktLists
            .AsNoTracking()
            .Filter(tl => tl.LastMatch == null || tl.LastMatch <= target)
            .ToListAsync(cancellationToken);

        if (traktLists.Count != 0 && _entityLocker.LockTrakt())
        {
            TraktList last = traktLists.Last();
            foreach (TraktList list in traktLists)
            {
                await _workerChannel.WriteAsync(
                    new MatchTraktListItems(list.Id, list == last),
                    cancellationToken);
            }
        }
    }

    private async Task RefreshMpegTsScripts(CancellationToken _)
    {
        using IServiceScope scope = _serviceScopeFactory.CreateScope();
        var service = scope.ServiceProvider.GetRequiredService<IMpegTsScriptService>();
        await service.RefreshScripts();
    }

    private ValueTask RefreshGraphicsElements(CancellationToken cancellationToken) =>
        _workerChannel.WriteAsync(new RefreshGraphicsElements(), cancellationToken);

    private ValueTask DeleteOrphanedSubtitles(CancellationToken cancellationToken) =>
        _workerChannel.WriteAsync(new DeleteOrphanedSubtitles(), cancellationToken);

    private ValueTask ReleaseMemory(CancellationToken cancellationToken) =>
        _workerChannel.WriteAsync(new ReleaseMemory(false), cancellationToken);

    private ValueTask QueueFFmpegCapabilitiesRefresh(CancellationToken cancellationToken) =>
        _workerChannel.WriteAsync(new RefreshFFmpegCapabilities(), cancellationToken);
}