Add rate limiting for download removal (#141)

code/Common/Configuration/General/SearchConfig.cs

@@ -0,0 +1,12 @@
+using Microsoft.Extensions.Configuration;
+
+namespace Common.Configuration.General;
+
+public sealed record SearchConfig
+{
+    [ConfigurationKeyName("SEARCH_ENABLED")]
+    public bool SearchEnabled { get; init; } = true;
+    
+    [ConfigurationKeyName("SEARCH_DELAY")]
+    public ushort SearchDelay { get; init; } = 30;
+}

code/Executable/DependencyInjection/ConfigurationDI.cs

@@ -13,6 +13,7 @@ public static class ConfigurationDI
     public static IServiceCollection AddConfiguration(this IServiceCollection services, IConfiguration configuration) =>
         services
             .Configure<DryRunConfig>(configuration)
+            .Configure<SearchConfig>(configuration)
             .Configure<QueueCleanerConfig>(configuration.GetSection(QueueCleanerConfig.SectionName))
             .Configure<ContentBlockerConfig>(configuration.GetSection(ContentBlockerConfig.SectionName))
             .Configure<DownloadCleanerConfig>(configuration.GetSection(DownloadCleanerConfig.SectionName))

code/Executable/DependencyInjection/MainDI.cs

@@ -1,11 +1,14 @@
 using System.Net;
 using Common.Configuration.General;
 using Common.Helpers;
+using Domain.Models.Arr;
 using Infrastructure.Services;
 using Infrastructure.Verticals.DownloadClient.Deluge;
+using Infrastructure.Verticals.DownloadRemover.Consumers;
 using Infrastructure.Verticals.Notifications.Consumers;
 using Infrastructure.Verticals.Notifications.Models;
 using MassTransit;
+using MassTransit.Configuration;
 using Microsoft.Extensions.Options;
 using Polly;
 using Polly.Extensions.Http;
@@ -27,6 +30,9 @@ public static class MainDI
             .AddNotifications(configuration)
             .AddMassTransit(config =>
             {
+                config.AddConsumer<DownloadRemoverConsumer<SearchItem>>();
+                config.AddConsumer<DownloadRemoverConsumer<SonarrSearchItem>>();
+                
                 config.AddConsumer<NotificationConsumer<FailedImportStrikeNotification>>();
                 config.AddConsumer<NotificationConsumer<StalledStrikeNotification>>();
                 config.AddConsumer<NotificationConsumer<SlowStrikeNotification>>();
@@ -36,6 +42,14 @@ public static class MainDI
 
                 config.UsingInMemory((context, cfg) =>
                 {
+                    cfg.ReceiveEndpoint("download-remover-queue", e =>
+                    {
+                        e.ConfigureConsumer<DownloadRemoverConsumer<SearchItem>>(context);
+                        e.ConfigureConsumer<DownloadRemoverConsumer<SonarrSearchItem>>(context);
+                        e.ConcurrentMessageLimit = 1;
+                        e.PrefetchCount = 1;
+                    });
+                    
                     cfg.ReceiveEndpoint("notification-queue", e =>
                     {
                         e.ConfigureConsumer<NotificationConsumer<FailedImportStrikeNotification>>(context);

code/Executable/DependencyInjection/ServicesDI.cs

@@ -11,6 +11,8 @@ using Infrastructure.Verticals.DownloadClient;
 using Infrastructure.Verticals.DownloadClient.Deluge;
 using Infrastructure.Verticals.DownloadClient.QBittorrent;
 using Infrastructure.Verticals.DownloadClient.Transmission;
+using Infrastructure.Verticals.DownloadRemover;
+using Infrastructure.Verticals.DownloadRemover.Interfaces;
 using Infrastructure.Verticals.Files;
 using Infrastructure.Verticals.ItemStriker;
 using Infrastructure.Verticals.QueueCleaner;
@@ -26,9 +28,11 @@ public static class ServicesDI
             .AddTransient<SonarrClient>()
             .AddTransient<RadarrClient>()
             .AddTransient<LidarrClient>()
+            .AddTransient<ArrClientFactory>()
             .AddTransient<QueueCleaner>()
             .AddTransient<ContentBlocker>()
             .AddTransient<DownloadCleaner>()
+            .AddTransient<IQueueItemRemover, QueueItemRemover>()
             .AddTransient<IFilenameEvaluator, FilenameEvaluator>()
             .AddTransient<IHardLinkFileService, HardLinkFileService>()
             .AddTransient<UnixHardLinkFileService>()

code/Executable/appsettings.Development.json

@@ -11,6 +11,8 @@
       "Path": ""
     }
   },
+  "SEARCH_ENABLED": true,
+  "SEARCH_DELAY": 5,
   "Triggers": {
     "QueueCleaner": "0/10 * * * * ?",
     "ContentBlocker": "0/10 * * * * ?",

code/Executable/appsettings.json

@@ -11,6 +11,8 @@
       "Path": ""
     }
   },
+  "SEARCH_ENABLED": true,
+  "SEARCH_DELAY": 30,
   "Triggers": {
     "QueueCleaner": "0 0/5 * * * ?",
     "ContentBlocker": "0 0/5 * * * ?",

code/Infrastructure/Extensions/DelugeExtensions.cs

@@ -1,5 +1,6 @@
 using Domain.Models.Deluge.Response;
 using Infrastructure.Helpers;
+using Infrastructure.Services;
 
 namespace Infrastructure.Extensions;
 

code/Infrastructure/Extensions/QBitExtensions.cs

@@ -1,4 +1,5 @@
 using Infrastructure.Helpers;
+using Infrastructure.Services;
 using QBittorrent.Client;
 
 namespace Infrastructure.Extensions;

code/Infrastructure/Extensions/TransmissionExtensions.cs

@@ -1,4 +1,5 @@
 using Infrastructure.Helpers;
+using Infrastructure.Services;
 using Transmission.API.RPC.Entity;
 
 namespace Infrastructure.Extensions;

code/Infrastructure/Helpers/CacheKeys.cs

@@ -13,4 +13,6 @@ public static class CacheKeys
     public static string StrikeItem(string hash, StrikeType strikeType) => $"item_{hash}_{strikeType.ToString()}";
     
     public static string IgnoredDownloads(string name) => $"{name}_ignored";
+    
+    public static string DownloadMarkedForRemoval(string hash, Uri url) => $"remove_{hash.ToLowerInvariant()}_{url}";
 }

code/Infrastructure/Services/UriService.cs

@@ -1,6 +1,6 @@
 using System.Text.RegularExpressions;
 
-namespace Infrastructure.Helpers;
+namespace Infrastructure.Services;
 
 public static class UriService
 {

code/Infrastructure/Verticals/Arr/ArrClient.cs

@@ -157,7 +157,7 @@ public abstract class ArrClient : IArrClient
         }
     }
 
-    public abstract Task RefreshItemsAsync(ArrInstance arrInstance, HashSet<SearchItem>? items);
+    public abstract Task SearchItemsAsync(ArrInstance arrInstance, HashSet<SearchItem>? items);
 
     public virtual bool IsRecordValid(QueueRecord record)
     {

code/Infrastructure/Verticals/Arr/ArrClientFactory.cs

@@ -0,0 +1,31 @@
+using Domain.Enums;
+using Infrastructure.Verticals.Arr.Interfaces;
+
+namespace Infrastructure.Verticals.Arr;
+
+public sealed class ArrClientFactory
+{
+    private readonly ISonarrClient _sonarrClient;
+    private readonly IRadarrClient _radarrClient;
+    private readonly ILidarrClient _lidarrClient;
+
+    public ArrClientFactory(
+        SonarrClient sonarrClient,
+        RadarrClient radarrClient,
+        LidarrClient lidarrClient
+    )
+    {
+        _sonarrClient = sonarrClient;
+        _radarrClient = radarrClient;
+        _lidarrClient = lidarrClient;
+    }
+    
+    public IArrClient GetClient(InstanceType type) =>
+        type switch
+        {
+            InstanceType.Sonarr => _sonarrClient,
+            InstanceType.Radarr => _radarrClient,
+            InstanceType.Lidarr => _lidarrClient,
+            _ => throw new NotImplementedException($"instance type {type} is not yet supported")
+        };
+}

code/Infrastructure/Verticals/Arr/Interfaces/IArrClient.cs

@@ -13,7 +13,7 @@ public interface IArrClient
 
     Task DeleteQueueItemAsync(ArrInstance arrInstance, QueueRecord record, bool removeFromClient, DeleteReason deleteReason);
     
-    Task RefreshItemsAsync(ArrInstance arrInstance, HashSet<SearchItem>? items);
+    Task SearchItemsAsync(ArrInstance arrInstance, HashSet<SearchItem>? items);
     
     bool IsRecordValid(QueueRecord record);
 }

code/Infrastructure/Verticals/Arr/LidarrClient.cs

@@ -50,7 +50,7 @@ public class LidarrClient : ArrClient, ILidarrClient
         return query;
     }
 
-    public override async Task RefreshItemsAsync(ArrInstance arrInstance, HashSet<SearchItem>? items)
+    public override async Task SearchItemsAsync(ArrInstance arrInstance, HashSet<SearchItem>? items)
     {
         if (items?.Count is null or 0)
         {

code/Infrastructure/Verticals/Arr/RadarrClient.cs

@@ -50,7 +50,7 @@ public class RadarrClient : ArrClient, IRadarrClient
         return query;
     }
 
-    public override async Task RefreshItemsAsync(ArrInstance arrInstance, HashSet<SearchItem>? items)
+    public override async Task SearchItemsAsync(ArrInstance arrInstance, HashSet<SearchItem>? items)
     {
         if (items?.Count is null or 0)
         {

code/Infrastructure/Verticals/Arr/SonarrClient.cs

@@ -51,7 +51,7 @@ public class SonarrClient : ArrClient, ISonarrClient
         return query;
     }
 
-    public override async Task RefreshItemsAsync(ArrInstance arrInstance, HashSet<SearchItem>? items)
+    public override async Task SearchItemsAsync(ArrInstance arrInstance, HashSet<SearchItem>? items)
     {
         if (items?.Count is null or 0)
         {

code/Infrastructure/Verticals/ContentBlocker/ContentBlocker.cs

@@ -6,16 +6,20 @@ using Common.Configuration.DownloadClient;
 using Domain.Enums;
 using Domain.Models.Arr;
 using Domain.Models.Arr.Queue;
+using Infrastructure.Helpers;
 using Infrastructure.Providers;
 using Infrastructure.Verticals.Arr;
 using Infrastructure.Verticals.Arr.Interfaces;
 using Infrastructure.Verticals.Context;
 using Infrastructure.Verticals.DownloadClient;
+using Infrastructure.Verticals.DownloadRemover.Models;
 using Infrastructure.Verticals.Jobs;
 using Infrastructure.Verticals.Notifications;
+using MassTransit;
+using Microsoft.Extensions.Caching.Memory;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
-using Serilog.Context;
+using LogContext = Serilog.Context.LogContext;
 
 namespace Infrastructure.Verticals.ContentBlocker;
 
@@ -32,9 +36,9 @@ public sealed class ContentBlocker : GenericHandler
         IOptions<SonarrConfig> sonarrConfig,
         IOptions<RadarrConfig> radarrConfig,
         IOptions<LidarrConfig> lidarrConfig,
-        SonarrClient sonarrClient,
-        RadarrClient radarrClient,
-        LidarrClient lidarrClient,
+        IMemoryCache cache,
+        IBus messageBus,
+        ArrClientFactory arrClientFactory,
         ArrQueueIterator arrArrQueueIterator,
         BlocklistProvider blocklistProvider,
         DownloadServiceFactory downloadServiceFactory,
@@ -43,8 +47,7 @@ public sealed class ContentBlocker : GenericHandler
     ) : base(
         logger, downloadClientConfig,
         sonarrConfig, radarrConfig, lidarrConfig,
-        sonarrClient, radarrClient, lidarrClient,
-        arrArrQueueIterator, downloadServiceFactory,
+        cache, messageBus, arrClientFactory, arrArrQueueIterator, downloadServiceFactory,
         notifier
     )
     {
@@ -81,15 +84,10 @@ public sealed class ContentBlocker : GenericHandler
         
         using var _ = LogContext.PushProperty("InstanceName", instanceType.ToString());
 
-        HashSet<SearchItem> itemsToBeRefreshed = [];
-        IArrClient arrClient = GetClient(instanceType);
+        IArrClient arrClient = _arrClientFactory.GetClient(instanceType);
         BlocklistType blocklistType = _blocklistProvider.GetBlocklistType(instanceType);
         ConcurrentBag<string> patterns = _blocklistProvider.GetPatterns(instanceType);
         ConcurrentBag<Regex> regexes = _blocklistProvider.GetRegexes(instanceType);
-        
-        // push to context
-        ContextProvider.Set(nameof(ArrInstance) + nameof(ArrInstance.Url), instance.Url);
-        ContextProvider.Set(nameof(InstanceType), instanceType);
 
         await _arrArrQueueIterator.Iterate(arrClient, instance, async items =>
         {
@@ -117,9 +115,14 @@ public sealed class ContentBlocker : GenericHandler
                     _logger.LogInformation("skip | {title} | ignored", record.Title);
                     continue;
                 }
+
+                string downloadRemovalKey = CacheKeys.DownloadMarkedForRemoval(record.DownloadId, instance.Url);
                 
-                // push record to context
-                ContextProvider.Set(nameof(QueueRecord), record);
+                if (_cache.TryGetValue(downloadRemovalKey, out bool _))
+                {
+                    _logger.LogDebug("skip | already marked for removal | {title}", record.Title);
+                    continue;
+                }
                 
                 _logger.LogDebug("searching unwanted files for {title}", record.Title);
 
@@ -133,8 +136,6 @@ public sealed class ContentBlocker : GenericHandler
                 
                 _logger.LogDebug("all files are marked as unwanted | {hash}", record.Title);
                 
-                itemsToBeRefreshed.Add(GetRecordSearchItem(instanceType, record, group.Count() > 1));
-
                 bool removeFromClient = true;
                 
                 if (result.IsPrivate && !_config.DeletePrivate)
@@ -142,11 +143,16 @@ public sealed class ContentBlocker : GenericHandler
                     removeFromClient = false;
                 }
                 
-                await arrClient.DeleteQueueItemAsync(instance, record, removeFromClient, DeleteReason.AllFilesBlocked);
-                await _notifier.NotifyQueueItemDeleted(removeFromClient, DeleteReason.AllFilesBlocked);
+                await PublishQueueItemRemoveRequest(
+                    downloadRemovalKey,
+                    instanceType,
+                    instance,
+                    record,
+                    group.Count() > 1,
+                    removeFromClient,
+                    DeleteReason.AllFilesBlocked
+                );
             }
         });
-        
-        await arrClient.RefreshItemsAsync(instance, itemsToBeRefreshed);
     }
 }

code/Infrastructure/Verticals/DownloadCleaner/DownloadCleaner.cs

@@ -9,9 +9,11 @@ using Infrastructure.Verticals.Arr.Interfaces;
 using Infrastructure.Verticals.DownloadClient;
 using Infrastructure.Verticals.Jobs;
 using Infrastructure.Verticals.Notifications;
+using MassTransit;
+using Microsoft.Extensions.Caching.Memory;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
-using Serilog.Context;
+using LogContext = Serilog.Context.LogContext;
 
 namespace Infrastructure.Verticals.DownloadCleaner;
 
@@ -30,9 +32,9 @@ public sealed class DownloadCleaner : GenericHandler
         IOptions<SonarrConfig> sonarrConfig,
         IOptions<RadarrConfig> radarrConfig,
         IOptions<LidarrConfig> lidarrConfig,
-        SonarrClient sonarrClient,
-        RadarrClient radarrClient,
-        LidarrClient lidarrClient,
+        IMemoryCache cache,
+        IBus messageBus,
+        ArrClientFactory arrClientFactory,
         ArrQueueIterator arrArrQueueIterator,
         DownloadServiceFactory downloadServiceFactory,
         INotificationPublisher notifier,
@@ -40,8 +42,7 @@ public sealed class DownloadCleaner : GenericHandler
     ) : base(
         logger, downloadClientConfig,
         sonarrConfig, radarrConfig, lidarrConfig,
-        sonarrClient, radarrClient, lidarrClient,
-        arrArrQueueIterator, downloadServiceFactory,
+        cache, messageBus, arrClientFactory, arrArrQueueIterator, downloadServiceFactory,
         notifier
     )
     {
@@ -131,7 +132,7 @@ public sealed class DownloadCleaner : GenericHandler
     {
         using var _ = LogContext.PushProperty("InstanceName", instanceType.ToString());
         
-        IArrClient arrClient = GetClient(instanceType);
+        IArrClient arrClient = _arrClientFactory.GetClient(instanceType);
         
         await _arrArrQueueIterator.Iterate(arrClient, instance, async items =>
         {

code/Infrastructure/Verticals/DownloadRemover/Consumers/DownloadRemoverConsumer.cs

@@ -0,0 +1,39 @@
+using Domain.Models.Arr;
+using Infrastructure.Verticals.DownloadRemover.Interfaces;
+using Infrastructure.Verticals.DownloadRemover.Models;
+using MassTransit;
+using Microsoft.Extensions.Logging;
+
+namespace Infrastructure.Verticals.DownloadRemover.Consumers;
+
+public class DownloadRemoverConsumer<T> : IConsumer<QueueItemRemoveRequest<T>>
+    where T : SearchItem
+{
+    private readonly ILogger<DownloadRemoverConsumer<T>> _logger;
+    private readonly IQueueItemRemover _queueItemRemover;
+
+    public DownloadRemoverConsumer(
+        ILogger<DownloadRemoverConsumer<T>> logger,
+        IQueueItemRemover queueItemRemover
+    )
+    {
+        _logger = logger;
+        _queueItemRemover = queueItemRemover;
+    }
+
+    public async Task Consume(ConsumeContext<QueueItemRemoveRequest<T>> context)
+    {
+        try
+        {
+            await _queueItemRemover.RemoveQueueItemAsync(context.Message);
+        }
+        catch (Exception exception)
+        {
+            _logger.LogError(exception,
+                "failed to remove queue item| {title} | {url}",
+                context.Message.Record.Title,
+                context.Message.Instance.Url
+            );
+        }
+    }
+}

code/Infrastructure/Verticals/DownloadRemover/Interfaces/IQueueItemRemover.cs

@@ -0,0 +1,9 @@
+using Domain.Models.Arr;
+using Infrastructure.Verticals.DownloadRemover.Models;
+
+namespace Infrastructure.Verticals.DownloadRemover.Interfaces;
+
+public interface IQueueItemRemover
+{
+    Task RemoveQueueItemAsync<T>(QueueItemRemoveRequest<T> request) where T : SearchItem;
+}

code/Infrastructure/Verticals/DownloadRemover/Models/QueueItemRemoveRequest.cs

@@ -0,0 +1,22 @@
+using Common.Configuration.Arr;
+using Domain.Enums;
+using Domain.Models.Arr;
+using Domain.Models.Arr.Queue;
+
+namespace Infrastructure.Verticals.DownloadRemover.Models;
+
+public sealed record QueueItemRemoveRequest<T>
+    where T : SearchItem
+{
+    public required InstanceType InstanceType { get; init; }
+    
+    public required ArrInstance Instance { get; init; }
+    
+    public required T SearchItem { get; init; }
+    
+    public required QueueRecord Record { get; init; }
+    
+    public required bool RemoveFromClient { get; init; }
+    
+    public required DeleteReason DeleteReason { get; init; }
+}

code/Infrastructure/Verticals/DownloadRemover/QueueItemRemover.cs

@@ -0,0 +1,66 @@
+using Common.Configuration.Arr;
+using Common.Configuration.General;
+using Domain.Enums;
+using Domain.Models.Arr;
+using Domain.Models.Arr.Queue;
+using Infrastructure.Helpers;
+using Infrastructure.Verticals.Arr;
+using Infrastructure.Verticals.Context;
+using Infrastructure.Verticals.DownloadRemover.Interfaces;
+using Infrastructure.Verticals.DownloadRemover.Models;
+using Infrastructure.Verticals.Notifications;
+using Microsoft.Extensions.Caching.Memory;
+using Microsoft.Extensions.Options;
+
+namespace Infrastructure.Verticals.DownloadRemover;
+
+public sealed class QueueItemRemover : IQueueItemRemover
+{
+    private readonly SearchConfig _searchConfig;
+    private readonly IMemoryCache _cache;
+    private readonly ArrClientFactory _arrClientFactory;
+    private readonly INotificationPublisher _notifier;
+
+    public QueueItemRemover(
+        IOptions<SearchConfig> searchConfig,
+        IMemoryCache cache,
+        ArrClientFactory arrClientFactory,
+        INotificationPublisher notifier
+    )
+    {
+        _searchConfig = searchConfig.Value;
+        _cache = cache;
+        _arrClientFactory = arrClientFactory;
+        _notifier = notifier;
+    }
+
+    public async Task RemoveQueueItemAsync<T>(QueueItemRemoveRequest<T> request)
+        where T : SearchItem
+    {
+        try
+        {
+            var arrClient = _arrClientFactory.GetClient(request.InstanceType);
+            await arrClient.DeleteQueueItemAsync(request.Instance, request.Record, request.RemoveFromClient, request.DeleteReason);
+            
+            // push to context
+            ContextProvider.Set(nameof(QueueRecord), request.Record);
+            ContextProvider.Set(nameof(ArrInstance) + nameof(ArrInstance.Url), request.Instance.Url);
+            ContextProvider.Set(nameof(InstanceType), request.InstanceType);
+            await _notifier.NotifyQueueItemDeleted(request.RemoveFromClient, request.DeleteReason);
+
+            if (!_searchConfig.SearchEnabled)
+            {
+                return;
+            }
+
+            await arrClient.SearchItemsAsync(request.Instance, [request.SearchItem]);
+
+            // prevent tracker spamming
+            await Task.Delay(TimeSpan.FromSeconds(_searchConfig.SearchDelay));
+        }
+        finally
+        {
+            _cache.Remove(CacheKeys.DownloadMarkedForRemoval(request.Record.DownloadId, request.Instance.Url));
+        }
+    }
+}

code/Infrastructure/Verticals/Jobs/GenericHandler.cs

@@ -4,9 +4,11 @@ using Domain.Enums;
 using Domain.Models.Arr;
 using Domain.Models.Arr.Queue;
 using Infrastructure.Verticals.Arr;
-using Infrastructure.Verticals.Arr.Interfaces;
 using Infrastructure.Verticals.DownloadClient;
+using Infrastructure.Verticals.DownloadRemover.Models;
 using Infrastructure.Verticals.Notifications;
+using MassTransit;
+using Microsoft.Extensions.Caching.Memory;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
 
@@ -19,9 +21,9 @@ public abstract class GenericHandler : IHandler, IDisposable
     protected readonly SonarrConfig _sonarrConfig;
     protected readonly RadarrConfig _radarrConfig;
     protected readonly LidarrConfig _lidarrConfig;
-    protected readonly ISonarrClient _sonarrClient;
-    protected readonly IRadarrClient _radarrClient;
-    protected readonly ILidarrClient _lidarrClient;
+    protected readonly IMemoryCache _cache;
+    protected readonly IBus _messageBus;
+    protected readonly ArrClientFactory _arrClientFactory;
     protected readonly ArrQueueIterator _arrArrQueueIterator;
     protected readonly IDownloadService _downloadService;
     protected readonly INotificationPublisher _notifier;
@@ -32,9 +34,9 @@ public abstract class GenericHandler : IHandler, IDisposable
         IOptions<SonarrConfig> sonarrConfig,
         IOptions<RadarrConfig> radarrConfig,
         IOptions<LidarrConfig> lidarrConfig,
-        ISonarrClient sonarrClient,
-        IRadarrClient radarrClient,
-        ILidarrClient lidarrClient,
+        IMemoryCache cache,
+        IBus messageBus,
+        ArrClientFactory arrClientFactory,
         ArrQueueIterator arrArrQueueIterator,
         DownloadServiceFactory downloadServiceFactory,
         INotificationPublisher notifier
@@ -45,9 +47,9 @@ public abstract class GenericHandler : IHandler, IDisposable
         _sonarrConfig = sonarrConfig.Value;
         _radarrConfig = radarrConfig.Value;
         _lidarrConfig = lidarrConfig.Value;
-        _sonarrClient = sonarrClient;
-        _radarrClient = radarrClient;
-        _lidarrClient = lidarrClient;
+        _cache = cache;
+        _messageBus = messageBus;
+        _arrClientFactory = arrClientFactory;
         _arrArrQueueIterator = arrArrQueueIterator;
         _downloadService = downloadServiceFactory.CreateDownloadClient();
         _notifier = notifier;
@@ -93,16 +95,50 @@ public abstract class GenericHandler : IHandler, IDisposable
             }
         }
     }
-    
-    protected IArrClient GetClient(InstanceType type) =>
-        type switch
-        {
-            InstanceType.Sonarr => _sonarrClient,
-            InstanceType.Radarr => _radarrClient,
-            InstanceType.Lidarr => _lidarrClient,
-            _ => throw new NotImplementedException($"instance type {type} is not yet supported")
-        };
 
+    protected async Task PublishQueueItemRemoveRequest(
+        string downloadRemovalKey,
+        InstanceType instanceType,
+        ArrInstance instance,
+        QueueRecord record,
+        bool isPack,
+        bool removeFromClient,
+        DeleteReason deleteReason
+    )
+    {
+        if (instanceType is InstanceType.Sonarr)
+        {
+            QueueItemRemoveRequest<SonarrSearchItem> removeRequest = new()
+            {
+                InstanceType = instanceType,
+                Instance = instance,
+                Record = record,
+                SearchItem = (SonarrSearchItem)GetRecordSearchItem(instanceType, record, isPack),
+                RemoveFromClient = removeFromClient,
+                DeleteReason = deleteReason
+            };
+
+            await _messageBus.Publish(removeRequest);
+        }
+        else
+        {
+            QueueItemRemoveRequest<SearchItem> removeRequest = new()
+            {
+                InstanceType = instanceType,
+                Instance = instance,
+                Record = record,
+                SearchItem = GetRecordSearchItem(instanceType, record, isPack),
+                RemoveFromClient = removeFromClient,
+                DeleteReason = deleteReason
+            };
+
+            await _messageBus.Publish(removeRequest);
+        }
+        
+        _cache.Set(downloadRemovalKey, true);
+        _logger.LogInformation("item marked for removal | {title} | {url}", record.Title, instance.Url);
+    }
+    
     protected SearchItem GetRecordSearchItem(InstanceType type, QueueRecord record, bool isPack = false)
     {
         return type switch

code/Infrastructure/Verticals/QueueCleaner/QueueCleaner.cs

@@ -4,22 +4,27 @@ using Common.Configuration.QueueCleaner;
 using Domain.Enums;
 using Domain.Models.Arr;
 using Domain.Models.Arr.Queue;
+using Infrastructure.Helpers;
 using Infrastructure.Providers;
 using Infrastructure.Verticals.Arr;
 using Infrastructure.Verticals.Arr.Interfaces;
 using Infrastructure.Verticals.Context;
 using Infrastructure.Verticals.DownloadClient;
+using Infrastructure.Verticals.DownloadRemover.Models;
 using Infrastructure.Verticals.Jobs;
 using Infrastructure.Verticals.Notifications;
+using MassTransit;
+using Microsoft.Extensions.Caching.Memory;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
-using Serilog.Context;
+using LogContext = Serilog.Context.LogContext;
 
 namespace Infrastructure.Verticals.QueueCleaner;
 
 public sealed class QueueCleaner : GenericHandler
 {
     private readonly QueueCleanerConfig _config;
+    private readonly IMemoryCache _cache;
     private readonly IgnoredDownloadsProvider<QueueCleanerConfig> _ignoredDownloadsProvider;
 
     public QueueCleaner(
@@ -29,9 +34,9 @@ public sealed class QueueCleaner : GenericHandler
         IOptions<SonarrConfig> sonarrConfig,
         IOptions<RadarrConfig> radarrConfig,
         IOptions<LidarrConfig> lidarrConfig,
-        SonarrClient sonarrClient,
-        RadarrClient radarrClient,
-        LidarrClient lidarrClient,
+        IMemoryCache cache,
+        IBus messageBus,
+        ArrClientFactory arrClientFactory,
         ArrQueueIterator arrArrQueueIterator,
         DownloadServiceFactory downloadServiceFactory,
         INotificationPublisher notifier,
@@ -39,13 +44,13 @@ public sealed class QueueCleaner : GenericHandler
     ) : base(
         logger, downloadClientConfig,
         sonarrConfig, radarrConfig, lidarrConfig,
-        sonarrClient, radarrClient, lidarrClient,
-        arrArrQueueIterator, downloadServiceFactory,
+        cache, messageBus, arrClientFactory, arrArrQueueIterator, downloadServiceFactory,
         notifier
     )
     {
         _config = config.Value;
         _config.Validate();
+        _cache = cache;
         _ignoredDownloadsProvider = ignoredDownloadsProvider;
     }
     
@@ -55,8 +60,7 @@ public sealed class QueueCleaner : GenericHandler
         
         using var _ = LogContext.PushProperty("InstanceName", instanceType.ToString());
         
-        HashSet<SearchItem> itemsToBeRefreshed = [];
-        IArrClient arrClient = GetClient(instanceType);
+        IArrClient arrClient = _arrClientFactory.GetClient(instanceType);
         
         // push to context
         ContextProvider.Set(nameof(ArrInstance) + nameof(ArrInstance.Url), instance.Url);
@@ -90,6 +94,14 @@ public sealed class QueueCleaner : GenericHandler
                     continue;
                 }
                 
+                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);
 
@@ -116,8 +128,6 @@ public sealed class QueueCleaner : GenericHandler
                     _logger.LogInformation("skip | {title}", record.Title);
                     continue;
                 }
-                
-                itemsToBeRefreshed.Add(GetRecordSearchItem(instanceType, record, group.Count() > 1));
 
                 bool removeFromClient = true;
 
@@ -140,11 +150,16 @@ public sealed class QueueCleaner : GenericHandler
                     }
                 }
                 
-                await arrClient.DeleteQueueItemAsync(instance, record, removeFromClient, deleteReason);
-                await _notifier.NotifyQueueItemDeleted(removeFromClient, deleteReason);
+                await PublishQueueItemRemoveRequest(
+                    downloadRemovalKey,
+                    instanceType,
+                    instance,
+                    record,
+                    group.Count() > 1,
+                    removeFromClient,
+                    deleteReason
+                );
             }
         });
-        
-        await arrClient.RefreshItemsAsync(instance, itemsToBeRefreshed);
     }
 }

code/test/docker-compose.yml

@@ -186,6 +186,9 @@ services:
       - HTTP_MAX_RETRIES=0
       - HTTP_TIMEOUT=20
 
+      - SEARCH_ENABLED=true
+      - SEARCH_DELAY=5
+
       - TRIGGERS__QUEUECLEANER=0/30 * * * * ?
       - TRIGGERS__CONTENTBLOCKER=0/30 * * * * ?
       - TRIGGERS__DOWNLOADCLEANER=0/30 * * * * ?