Cleanuparr/code/backend/Cleanuparr.Infrastructure/Features/Jobs/QueueCleaner.cs

using Cleanuparr.Domain.Entities.Arr.Queue;
using Cleanuparr.Domain.Enums;
using Cleanuparr.Infrastructure.Events.Interfaces;
using Cleanuparr.Infrastructure.Features.Arr.ForceImport;
using Cleanuparr.Infrastructure.Features.Arr.Interfaces;
using Cleanuparr.Infrastructure.Features.Context;
using Cleanuparr.Infrastructure.Features.DownloadClient;
using Cleanuparr.Infrastructure.Features.DownloadRemover.Models;
using Cleanuparr.Infrastructure.Features.LazyLibrarian;
using Cleanuparr.Infrastructure.Helpers;
using Cleanuparr.Infrastructure.Interceptors;
using Cleanuparr.Infrastructure.Services.Interfaces;
using Cleanuparr.Persistence;
using Cleanuparr.Persistence.Models.Configuration;
using Cleanuparr.Persistence.Models.Configuration.Arr;
using Cleanuparr.Persistence.Models.Configuration.General;
using Cleanuparr.Persistence.Models.Configuration.QueueCleaner;
using System.Threading.Channels;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Caching.Memory;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using LogContext = Serilog.Context.LogContext;

namespace Cleanuparr.Infrastructure.Features.Jobs;

public sealed class QueueCleaner : GenericHandler
{
    private readonly IConnectivityChecker _connectivityChecker;
    private readonly ILazyLibrarianEvaluator _lazyLibrarianService;

    public QueueCleaner(
        ILogger<QueueCleaner> logger,
        DataContext dataContext,
        IMemoryCache cache,
        ChannelWriter<QueueItemRemoveRequest> removalQueue,
        IArrClientFactory arrClientFactory,
        IArrQueueReader arrQueueReader,
        IDownloadServiceFactory downloadServiceFactory,
        IEventPublisher eventPublisher,
        IDryRunInterceptor dryRunInterceptor,
        IConnectivityChecker connectivityChecker,
        IForceImportService forceImportService,
        [FromKeyedServices(ILazyLibrarianEvaluator.QueueCleanerKey)] ILazyLibrarianEvaluator lazyLibrarianService
    ) : base(
        logger, dataContext, cache, removalQueue,
        arrClientFactory, arrQueueReader, downloadServiceFactory, eventPublisher, dryRunInterceptor,
        forceImportService
    )
    {
        _connectivityChecker = connectivityChecker;
        _lazyLibrarianService = lazyLibrarianService;
    }

    protected override async Task ExecuteInternalAsync(CancellationToken cancellationToken = default)
    {
        GeneralConfig generalConfig = ContextProvider.Get<GeneralConfig>(nameof(GeneralConfig));

        if (!await _connectivityChecker.IsOnlineAsync(generalConfig, cancellationToken))
        {
            _logger.LogWarning($"skip {nameof(QueueCleaner)} run | no internet connectivity detected");
            return;
        }

        List<StallRule> stallRules = await _dataContext.StallRules
            .Where(r => r.Enabled)
            .OrderByDescending(r => r.MaxCompletionPercentage)
            .ThenByDescending(r => r.MinCompletionPercentage)
            .AsNoTracking()
            .ToListAsync();
            
        if (stallRules.Count is 0)
        {
            _logger.LogDebug("No active stall rules found");
        }
            
        ContextProvider.Set(nameof(StallRule), stallRules);
            
        List<SlowRule> slowRules = await _dataContext.SlowRules
            .Where(r => r.Enabled)
            .OrderByDescending(r => r.MaxCompletionPercentage)
            .ThenByDescending(r => r.MinCompletionPercentage)
            .AsNoTracking()
            .ToListAsync();
            
        if (slowRules.Count is 0)
        {
            _logger.LogDebug("No active slow rules found");
        }
            
        ContextProvider.Set(nameof(SlowRule), slowRules);
        
        var sonarrConfig = ContextProvider.Get<ArrConfig>(nameof(InstanceType.Sonarr));
        var radarrConfig = ContextProvider.Get<ArrConfig>(nameof(InstanceType.Radarr));
        var lidarrConfig = ContextProvider.Get<ArrConfig>(nameof(InstanceType.Lidarr));
        var readarrConfig = ContextProvider.Get<ArrConfig>(nameof(InstanceType.Readarr));
        var whisparrConfig = ContextProvider.Get<ArrConfig>(nameof(InstanceType.Whisparr));
        var sportarrConfig = ContextProvider.Get<ArrConfig>(nameof(InstanceType.Sportarr));
        ArrConfig lazyLibrarianConfig = ContextProvider.Get<ArrConfig>(nameof(InstanceType.LazyLibrarian));

        await ProcessArrConfigAsync(sonarrConfig);
        await ProcessArrConfigAsync(radarrConfig);
        await ProcessArrConfigAsync(lidarrConfig);
        await ProcessArrConfigAsync(readarrConfig);
        await ProcessArrConfigAsync(whisparrConfig);
        await ProcessArrConfigAsync(sportarrConfig);
        await ProcessArrConfigAsync(lazyLibrarianConfig);
    }

    protected override async Task ProcessInstanceAsync(ArrInstance instance)
    {
        List<string> ignoredDownloads = ContextProvider.Get<GeneralConfig>(nameof(GeneralConfig)).IgnoredDownloads;
        QueueCleanerConfig queueCleanerConfig = ContextProvider.Get<QueueCleanerConfig>();
        ignoredDownloads.AddRange(queueCleanerConfig.IgnoredDownloads);
        
        using var _ = LogContext.PushProperty(LogProperties.Category, instance.ArrConfig.Type.ToString());
        using var _2 = LogContext.PushProperty(LogProperties.InstanceName, instance.Name);

        // push to context
        ContextProvider.Set(ContextProvider.Keys.ArrInstanceUrl, instance.ExternalOrInternalUrl);
        ContextProvider.Set(nameof(InstanceType), instance.ArrConfig.Type);
        ContextProvider.Set(ContextProvider.Keys.ArrInstanceId, instance.Id);
        ContextProvider.Set(ContextProvider.Keys.Version, instance.Version);

        IReadOnlyList<IDownloadService> downloadServices = await GetInitializedDownloadServicesAsync();

        if (instance.ArrConfig.Type is InstanceType.LazyLibrarian)
        {
            IReadOnlyList<LazyLibrarianRemovalDecision> decisions =
                await _lazyLibrarianService.EvaluateAsync(instance, downloadServices, ignoredDownloads);

            await ProcessLazyLibrarianDecisionsAsync(instance, decisions);
            return;
        }

        IArrClient arrClient = _arrClientFactory.GetClient(instance.ArrConfig.Type, instance.Version);
        List<DownloadClientConfig> configuredClients = ContextProvider.Get<List<DownloadClientConfig>>(nameof(DownloadClientConfig));
        bool hasEnabledTorrentClients = configuredClients
            .Where(x => x.Type == DownloadClientType.Torrent)
            .Any(x => x.Enabled);
        bool hasEnabledUsenetClients = configuredClients
            .Where(x => x.Type == DownloadClientType.Usenet)
            .Any(x => x.Enabled);

        HashSet<string> queuedDownloadIds = new(StringComparer.InvariantCultureIgnoreCase);

        List<QueueRecord> queueRecords = await _arrQueueReader.ReadAllAsync(arrClient, instance);

        List<IGrouping<string, QueueRecord>> groups = queueRecords
            .GroupBy(x => x.DownloadId)
            .ToList();

        foreach (IGrouping<string, QueueRecord> group in groups)
        {
            QueueRecord record = group.First();
            queuedDownloadIds.Add(record.DownloadId);

            if (!arrClient.IsRecordValid(record))
            {
                continue;
            }

            if (record.IsIgnored(ignoredDownloads))
            {
                _logger.LogInformation("skip | download is ignored | {name}", record.Title);
                continue;
            }

            _logger.LogDebug("processing | {title} | {id}", record.Title, record.DownloadId);

            bool hasContentId = arrClient.HasContentId(record);

            if (!hasContentId)
            {
                if (!queueCleanerConfig.ProcessNoContentId)
                {
                    _logger.LogInformation("skip | item is missing the content id | {title}", record.Title);
                    continue;
                }

                _logger.LogDebug("item is missing the content id | {title}", record.Title);
            }

            string downloadRemovalKey = CacheKeys.DownloadMarkedForRemoval(record.DownloadId, instance.Url);

            if (_cache.TryGetValue(downloadRemovalKey, out bool _))
            {
                _logger.LogDebug("skip | already marked for removal | {title}", record.Title);
                continue;
            }

            // push record to context
            ContextProvider.Set(nameof(QueueRecord), record);

            (DownloadCheckResult downloadCheckResult, DownloadClientConfig? foundInClient) = await FindInClientsAsync<IQueueCheckCapable, DownloadCheckResult>(
                record,
                downloadServices,
                downloadService => downloadService.ShouldRemoveFromArrQueueAsync(record.DownloadId, ignoredDownloads),
                x => x.Found
            );

            if (downloadCheckResult.ShouldRemove)
            {
                bool changeCategory = downloadCheckResult.ChangeCategory;
                bool removeFromClient = !changeCategory && (!downloadCheckResult.IsPrivate || downloadCheckResult.DeleteFromClient);

                await PublishQueueItemRemoveRequest(
                    instance,
                    record,
                    group.Count() > 1,
                    removeFromClient,
                    downloadCheckResult.DeleteReason,
                    skipSearch: !hasContentId,
                    downloadClient: foundInClient,
                    changeCategory: changeCategory
                );

                continue;
            }

            // Runs before the client check below: a torrent missing from the client can still be imported.
            ForceImportOutcome forceImport = await _forceImportService.TryImportAsync(arrClient, instance, record);

            // A deferred import must not collect strikes while the arr works.
            if (forceImport is not ForceImportOutcome.NotApplicable)
            {
                continue;
            }

            // Skip failed import check if the download is not found in any client of its protocol and skipIfNotFoundInClient is enabled
            bool skipIfNotFoundInClient = GetClientType(record) switch
            {
                DownloadClientType.Torrent => hasEnabledTorrentClients,
                DownloadClientType.Usenet => hasEnabledUsenetClients,
                _ => false
            };

            if (skipIfNotFoundInClient && !downloadCheckResult.Found && queueCleanerConfig.FailedImport.SkipIfNotFoundInClient)
            {
                _logger.LogInformation("skip | download not found in any client | {Title}", record.Title);
                continue;
            }

            // Failed import check
            bool shouldRemoveFromArr = await arrClient
                .ShouldRemoveFromQueue(instance.ArrConfig.Type, record, downloadCheckResult.IsPrivate, instance.ArrConfig.FailedImportMaxStrikes);

            if (shouldRemoveFromArr)
            {
                bool changeCategory = queueCleanerConfig.FailedImport.ChangeCategory && GetClientType(record) is DownloadClientType.Torrent;
                bool removeFromClient = !changeCategory && (!downloadCheckResult.IsPrivate || queueCleanerConfig.FailedImport.DeletePrivate);

                await PublishQueueItemRemoveRequest(
                    instance,
                    record,
                    group.Count() > 1,
                    removeFromClient,
                    DeleteReason.FailedImport,
                    skipSearch: !hasContentId,
                    downloadClient: foundInClient,
                    changeCategory: changeCategory
                );

                continue;
            }

            _logger.LogDebug("skip | {title}", record.Title);
        }

        // A download the arr dropped from its queue has landed only if the arr's history says so.
        await _forceImportService.ReconcileAsync(arrClient, instance, queuedDownloadIds);
    }
}