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 logger, DataContext dataContext, IMemoryCache cache, ChannelWriter 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(nameof(GeneralConfig)); if (!await _connectivityChecker.IsOnlineAsync(generalConfig, cancellationToken)) { _logger.LogWarning($"skip {nameof(QueueCleaner)} run | no internet connectivity detected"); return; } List 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 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(nameof(InstanceType.Sonarr)); var radarrConfig = ContextProvider.Get(nameof(InstanceType.Radarr)); var lidarrConfig = ContextProvider.Get(nameof(InstanceType.Lidarr)); var readarrConfig = ContextProvider.Get(nameof(InstanceType.Readarr)); var whisparrConfig = ContextProvider.Get(nameof(InstanceType.Whisparr)); var sportarrConfig = ContextProvider.Get(nameof(InstanceType.Sportarr)); ArrConfig lazyLibrarianConfig = ContextProvider.Get(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 ignoredDownloads = ContextProvider.Get(nameof(GeneralConfig)).IgnoredDownloads; QueueCleanerConfig queueCleanerConfig = ContextProvider.Get(); 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 downloadServices = await GetInitializedDownloadServicesAsync(); if (instance.ArrConfig.Type is InstanceType.LazyLibrarian) { IReadOnlyList decisions = await _lazyLibrarianService.EvaluateAsync(instance, downloadServices, ignoredDownloads); await ProcessLazyLibrarianDecisionsAsync(instance, decisions); return; } IArrClient arrClient = _arrClientFactory.GetClient(instance.ArrConfig.Type, instance.Version); List configuredClients = ContextProvider.Get>(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 queuedDownloadIds = new(StringComparer.InvariantCultureIgnoreCase); List queueRecords = await _arrQueueReader.ReadAllAsync(arrClient, instance); List> groups = queueRecords .GroupBy(x => x.DownloadId) .ToList(); foreach (IGrouping 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( 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); } }