Remove MassTransit (#828)

CLAUDE.md

@@ -38,7 +38,7 @@ Cleanuparr is a tool for automating the cleanup of unwanted or blocked files in
 - **Database**: SQLite with Entity Framework Core
   - Three separate contexts: `DataContext`, `EventsContext`, `UsersContext`
 - **Key Libraries**:
-  - MassTransit (messaging)
+  - System.Threading.Channels (in-process queues)
   - Quartz.NET (scheduling)
   - Serilog (logging)
   - SignalR (real-time communication)

code/backend/Cleanuparr.Api.Tests/Json/RemovalRequestSerializationTests.cs

@@ -1,116 +0,0 @@
-using System.Text.Json;
-using Cleanuparr.Api.Json;
-using Cleanuparr.Domain.Entities.Arr;
-using Cleanuparr.Domain.Entities.Arr.Queue;
-using Cleanuparr.Domain.Enums;
-using Cleanuparr.Infrastructure.Features.DownloadRemover.Models;
-using Cleanuparr.Persistence.Models.Configuration.Arr;
-using Shouldly;
-using Xunit;
-
-namespace Cleanuparr.Api.Tests.Json;
-
-/// <summary>
-/// The in-memory bus serializes with these options, so a broken discriminator
-/// only shows up at runtime.
-/// </summary>
-public class RemovalRequestSerializationTests
-{
-    private static readonly JsonSerializerOptions Options = CreateOptions();
-
-    private static JsonSerializerOptions CreateOptions()
-    {
-        JsonSerializerOptions options = new();
-        CleanuparrJsonConfiguration.ConfigureCore(options);
-        return options;
-    }
-
-    private static QueueItemRemoveRequest CreateRequest(SearchItem searchItem)
-    {
-        return new QueueItemRemoveRequest
-        {
-            Instance = new ArrInstance
-            {
-                Name = "Test Instance",
-                Url = new Uri("http://sonarr.local"),
-                ApiKey = "test-api-key",
-                ArrConfig = new ArrConfig { Type = InstanceType.Sonarr },
-            },
-            Target = new ArrRemovalTarget
-            {
-                Record = new QueueRecord
-                {
-                    Id = 1,
-                    Title = "Test Record",
-                    Protocol = "torrent",
-                    DownloadId = "ABC123",
-                },
-                SearchItem = searchItem,
-                RemoveFromClient = true,
-                ChangeCategory = false,
-            },
-            DeleteReason = DeleteReason.Stalled,
-            JobRunId = Guid.NewGuid(),
-            IsDryRun = false,
-        };
-    }
-
-    private static QueueItemRemoveRequest RoundTrip(QueueItemRemoveRequest request)
-    {
-        string json = JsonSerializer.Serialize(request, Options);
-        return JsonSerializer.Deserialize<QueueItemRemoveRequest>(json, Options)!;
-    }
-
-    [Fact]
-    public void ArrTarget_SurvivesRoundTrip()
-    {
-        QueueItemRemoveRequest result = RoundTrip(CreateRequest(new SearchItem { Id = 42 }));
-
-        ArrRemovalTarget target = result.Target.ShouldBeOfType<ArrRemovalTarget>();
-        target.Record.DownloadId.ShouldBe("ABC123");
-        target.RemoveFromClient.ShouldBeTrue();
-        result.DeleteReason.ShouldBe(DeleteReason.Stalled);
-    }
-
-    [Fact]
-    public void BaseSearchItem_SurvivesRoundTrip()
-    {
-        QueueItemRemoveRequest result = RoundTrip(CreateRequest(new SearchItem { Id = 42 }));
-
-        SearchItem item = ArrTargetOf(result).SearchItem;
-        item.ShouldBeOfType<SearchItem>();
-        item.Id.ShouldBe(42);
-    }
-
-    [Fact]
-    public void SeriesSearchItem_KeepsItsDerivedType()
-    {
-        QueueItemRemoveRequest result = RoundTrip(CreateRequest(new SeriesSearchItem
-        {
-            Id = 100,
-            SeriesId = 10,
-            SearchType = SeriesSearchType.Episode,
-        }));
-
-        SeriesSearchItem item = ArrTargetOf(result).SearchItem.ShouldBeOfType<SeriesSearchItem>();
-        item.Id.ShouldBe(100);
-        item.SeriesId.ShouldBe(10);
-        item.SearchType.ShouldBe(SeriesSearchType.Episode);
-    }
-
-    [Fact]
-    public void Target_CarriesTheDiscriminator_AndOmitsDerivedMembers()
-    {
-        string json = JsonSerializer.Serialize(CreateRequest(new SearchItem { Id = 42 }), Options);
-
-        using JsonDocument document = JsonDocument.Parse(json);
-        JsonElement target = document.RootElement.GetProperty("Target");
-
-        target.GetProperty("$target").GetString().ShouldBe("arr");
-        target.TryGetProperty("DownloadId", out _).ShouldBeFalse();
-        target.TryGetProperty("Title", out _).ShouldBeFalse();
-    }
-
-    private static ArrRemovalTarget ArrTargetOf(QueueItemRemoveRequest request) =>
-        (ArrRemovalTarget)request.Target;
-}

code/backend/Cleanuparr.Api/Cleanuparr.Api.csproj

@@ -27,7 +27,6 @@
     </ItemGroup>
 
     <ItemGroup>
-        <PackageReference Include="MassTransit" Version="8.5.7" />
         <PackageReference Include="Microsoft.AspNetCore.Authentication.JwtBearer" Version="10.0.1" />
         <PackageReference Include="Microsoft.EntityFrameworkCore.Design" Version="10.0.1">
           <PrivateAssets>all</PrivateAssets>

code/backend/Cleanuparr.Api/DependencyInjection/MainDI.cs

@@ -1,12 +1,14 @@
-using Cleanuparr.Api.Json;
+using System.Threading.Channels;
 using Cleanuparr.Domain.Entities.Arr;
 using Cleanuparr.Infrastructure.Features.DownloadRemover.Consumers;
+using Cleanuparr.Infrastructure.Features.DownloadRemover.Models;
+using Cleanuparr.Infrastructure.Features.Messaging;
 using Cleanuparr.Infrastructure.Features.Notifications.Consumers;
+using Cleanuparr.Infrastructure.Features.Notifications.Models;
 using Cleanuparr.Infrastructure.Health;
 using Cleanuparr.Infrastructure.Http;
 using Cleanuparr.Infrastructure.Http.DynamicHttpClientSystem;
 using Cleanuparr.Shared.Helpers;
-using MassTransit;
 using Microsoft.Extensions.Caching.Memory;
 
 namespace Cleanuparr.Api.DependencyInjection;
@@ -22,38 +24,25 @@ public static class MainDI
             .AddHealthServices()
             .AddQuartzServices(configuration)
             .AddNotifications()
-            .AddMassTransit(config =>
-            {
-                config.DisableUsageTelemetry();
-                
-                config.AddConsumer<DownloadRemoverConsumer>();
-                config.AddConsumer<NotificationConsumer>();
+            .AddMessageQueue<QueueItemRemoveRequest, DownloadRemoverConsumer>()
+            .AddMessageQueue<NotificationMessage, NotificationConsumer>();
 
-                config.UsingInMemory((context, cfg) =>
-                {
-                    cfg.ConfigureJsonSerializerOptions(options =>
-                    {
-                        CleanuparrJsonConfiguration.ConfigureCore(options);
+    /// <summary>
+    /// Registers an unbounded in-process queue for <typeparamref name="TMessage"/>,
+    /// drained one message at a time by a <see cref="QueueWorker{TMessage}"/>.
+    /// </summary>
+    private static IServiceCollection AddMessageQueue<TMessage, THandler>(this IServiceCollection services)
+        where THandler : class, IMessageHandler<TMessage>
+    {
+        Channel<TMessage> channel = Channel.CreateUnbounded<TMessage>(new UnboundedChannelOptions { SingleReader = true });
+
+        return services
+            .AddSingleton(channel)
+            .AddSingleton(channel.Writer)
+            .AddScoped<IMessageHandler<TMessage>, THandler>()
+            .AddHostedService<QueueWorker<TMessage>>();
+    }
 
-                        return options;
-                    });
-                    
-                    cfg.ReceiveEndpoint("download-remover-queue", e =>
-                    {
-                        e.ConfigureConsumer<DownloadRemoverConsumer>(context);
-                        e.ConcurrentMessageLimit = 1;
-                        e.PrefetchCount = 1;
-                    });
-                    
-                    cfg.ReceiveEndpoint("notification-queue", e =>
-                    {
-                        e.ConfigureConsumer<NotificationConsumer>(context);
-                        e.ConcurrentMessageLimit = 1;
-                        e.PrefetchCount = 1;
-                    });
-                });
-            });
-    
     private static IServiceCollection AddHttpClients(this IServiceCollection services, IConfiguration configuration)
     {
         // Add the dynamic HTTP client system - this replaces all the previous static configurations

code/backend/Cleanuparr.Api/Json/CleanuparrJsonConfiguration.cs

@@ -7,7 +7,7 @@ namespace Cleanuparr.Api.Json;
 
 public static class CleanuparrJsonConfiguration
 {
-    public static void ConfigureCore(JsonSerializerOptions options)
+    private static void ConfigureCore(JsonSerializerOptions options)
     {
         options.PropertyNameCaseInsensitive = true;
         options.Converters.Add(new JsonStringEnumConverter());

code/backend/Cleanuparr.Domain/Entities/Arr/SearchItem.cs

@@ -1,10 +1,5 @@
-using System.Text.Json.Serialization;
+namespace Cleanuparr.Domain.Entities.Arr;
 
-namespace Cleanuparr.Domain.Entities.Arr;
-
-[JsonPolymorphic(TypeDiscriminatorPropertyName = "$item")]
-[JsonDerivedType(typeof(SearchItem), "base")]
-[JsonDerivedType(typeof(SeriesSearchItem), "series")]
 public class SearchItem
 {
     public long Id { get; set; }

code/backend/Cleanuparr.Infrastructure.Tests/Features/DownloadRemover/Consumers/DownloadRemoverConsumerTests.cs

@@ -8,7 +8,6 @@ using Cleanuparr.Infrastructure.Features.DownloadRemover.Models;
 using Cleanuparr.Infrastructure.Features.DryRun;
 using Cleanuparr.Infrastructure.Tests.TestHelpers;
 using Cleanuparr.Persistence.Models.Configuration.Arr;
-using MassTransit;
 using Microsoft.Extensions.Logging;
 using NSubstitute;
 using NSubstitute.ExceptionExtensions;
@@ -32,50 +31,47 @@ public class DownloadRemoverConsumerTests
         _consumer = new DownloadRemoverConsumer(_logger, _queueItemRemover, _dryRunActivity);
     }
 
-    #region Consume Tests
+    #region HandleAsync Tests
 
     [Fact]
-    public async Task Consume_CallsRemoveQueueItemAsync()
+    public async Task HandleAsync_CallsRemoveQueueItemAsync()
     {
         // Arrange
         var request = CreateRemoveRequest();
-        var context = CreateConsumeContext(request);
 
         _queueItemRemover
             .RemoveQueueItemAsync(Arg.Any<QueueItemRemoveRequest>())
             .Returns(Task.CompletedTask);
 
         // Act
-        await _consumer.Consume(context);
+        await _consumer.HandleAsync(request);
 
         // Assert
         await _queueItemRemover.Received(1).RemoveQueueItemAsync(request);
     }
 
     [Fact]
-    public async Task Consume_WhenRemoverThrows_LogsErrorAndDoesNotRethrow()
+    public async Task HandleAsync_WhenRemoverThrows_LogsErrorAndDoesNotRethrow()
     {
         // Arrange
         var request = CreateRemoveRequest();
-        var context = CreateConsumeContext(request);
 
         _queueItemRemover
             .RemoveQueueItemAsync(Arg.Any<QueueItemRemoveRequest>())
             .ThrowsAsync(new Exception("Remove failed"));
 
         // Act - Should not throw
-        await _consumer.Consume(context);
+        await _consumer.HandleAsync(request);
 
         // Assert
         _logger.HasLogContaining(LogLevel.Error, "failed to remove queue item").ShouldBeTrue();
     }
 
     [Fact]
-    public async Task Consume_PassesCorrectRequestToRemover()
+    public async Task HandleAsync_PassesCorrectRequestToRemover()
     {
         // Arrange
         var request = CreateRemoveRequest();
-        var context = CreateConsumeContext(request);
         QueueItemRemoveRequest? capturedRequest = null;
 
         _queueItemRemover
@@ -84,7 +80,7 @@ public class DownloadRemoverConsumerTests
             .AndDoes(ci => capturedRequest = ci.Arg<QueueItemRemoveRequest>());
 
         // Act
-        await _consumer.Consume(context);
+        await _consumer.HandleAsync(request);
 
         // Assert
         capturedRequest.ShouldNotBeNull();
@@ -95,11 +91,10 @@ public class DownloadRemoverConsumerTests
     }
 
     [Fact]
-    public async Task Consume_WithDryRunRequest_SetsStickyDryRunBeforeRemoving()
+    public async Task HandleAsync_WithDryRunRequest_SetsStickyDryRunBeforeRemoving()
     {
         // Arrange
         var request = CreateRemoveRequest() with { IsDryRun = true };
-        var context = CreateConsumeContext(request);
         bool observedDryRun = false;
 
         _queueItemRemover
@@ -108,43 +103,41 @@ public class DownloadRemoverConsumerTests
             .AndDoes(_ => observedDryRun = ContextProvider.IsDryRunSticky());
 
         // Act
-        await _consumer.Consume(context);
+        await _consumer.HandleAsync(request);
 
         // Assert
         observedDryRun.ShouldBeTrue();
     }
 
     [Fact]
-    public async Task Consume_WithDryRunRequest_RequestsAPurgeAfterRemoving()
+    public async Task HandleAsync_WithDryRunRequest_RequestsAPurgeAfterRemoving()
     {
         // Arrange
         var request = CreateRemoveRequest() with { IsDryRun = true };
-        var context = CreateConsumeContext(request);
 
         _queueItemRemover
             .RemoveQueueItemAsync(Arg.Any<QueueItemRemoveRequest>())
             .Returns(Task.CompletedTask);
 
         // Act
-        await _consumer.Consume(context);
+        await _consumer.HandleAsync(request);
 
         // Assert
         (await _dryRunActivity.WaitForPurgeRequestAsync(CancellationToken.None)).ShouldBeTrue();
     }
 
     [Fact]
-    public async Task Consume_WithLiveRequest_RequestsNoPurge()
+    public async Task HandleAsync_WithLiveRequest_RequestsNoPurge()
     {
         // Arrange
         var request = CreateRemoveRequest() with { IsDryRun = false };
-        var context = CreateConsumeContext(request);
 
         _queueItemRemover
             .RemoveQueueItemAsync(Arg.Any<QueueItemRemoveRequest>())
             .Returns(Task.CompletedTask);
 
         // Act
-        await _consumer.Consume(context);
+        await _consumer.HandleAsync(request);
 
         // Assert
         Task<bool> waitTask = _dryRunActivity.WaitForPurgeRequestAsync(CancellationToken.None).AsTask();
@@ -153,29 +146,27 @@ public class DownloadRemoverConsumerTests
     }
 
     [Fact]
-    public async Task Consume_WithDryRunRequest_WhenRemoverThrows_ExitsActivity()
+    public async Task HandleAsync_WithDryRunRequest_WhenRemoverThrows_ExitsActivity()
     {
         // Arrange
         QueueItemRemoveRequest request = CreateRemoveRequest() with { IsDryRun = true };
-        ConsumeContext<QueueItemRemoveRequest> context = CreateConsumeContext(request);
 
         _queueItemRemover
             .RemoveQueueItemAsync(Arg.Any<QueueItemRemoveRequest>())
             .ThrowsAsync(new Exception("Remove failed"));
 
         // Act
-        await _consumer.Consume(context);
+        await _consumer.HandleAsync(request);
 
         // Assert
         _dryRunActivity.IsActive.ShouldBeFalse();
     }
 
     [Fact]
-    public async Task Consume_WithLiveRequest_NeverEntersActivity()
+    public async Task HandleAsync_WithLiveRequest_NeverEntersActivity()
     {
         // Arrange
         QueueItemRemoveRequest request = CreateRemoveRequest() with { IsDryRun = false };
-        ConsumeContext<QueueItemRemoveRequest> context = CreateConsumeContext(request);
         bool activeDuringRemoval = false;
 
         _queueItemRemover
@@ -184,7 +175,7 @@ public class DownloadRemoverConsumerTests
             .AndDoes(_ => activeDuringRemoval = _dryRunActivity.IsActive);
 
         // Act
-        await _consumer.Consume(context);
+        await _consumer.HandleAsync(request);
 
         // Assert
         activeDuringRemoval.ShouldBeFalse();
@@ -192,7 +183,7 @@ public class DownloadRemoverConsumerTests
     }
 
     [Fact]
-    public async Task Consume_WithRemoveFromClientTrue_PassesCorrectly()
+    public async Task HandleAsync_WithRemoveFromClientTrue_PassesCorrectly()
     {
         // Arrange
         var request = new QueueItemRemoveRequest
@@ -208,14 +199,13 @@ public class DownloadRemoverConsumerTests
             JobRunId = Guid.NewGuid(),
             IsDryRun = false,
         };
-        var context = CreateConsumeContext(request);
 
         _queueItemRemover
             .RemoveQueueItemAsync(Arg.Any<QueueItemRemoveRequest>())
             .Returns(Task.CompletedTask);
 
         // Act
-        await _consumer.Consume(context);
+        await _consumer.HandleAsync(request);
 
         // Assert
         await _queueItemRemover.Received(1).RemoveQueueItemAsync(
@@ -225,7 +215,7 @@ public class DownloadRemoverConsumerTests
     }
 
     [Fact]
-    public async Task Consume_WithDifferentDeleteReasons_HandlesCorrectly()
+    public async Task HandleAsync_WithDifferentDeleteReasons_HandlesCorrectly()
     {
         // Arrange
         var request = new QueueItemRemoveRequest
@@ -241,14 +231,13 @@ public class DownloadRemoverConsumerTests
             JobRunId = Guid.NewGuid(),
             IsDryRun = false,
         };
-        var context = CreateConsumeContext(request);
 
         _queueItemRemover
             .RemoveQueueItemAsync(Arg.Any<QueueItemRemoveRequest>())
             .Returns(Task.CompletedTask);
 
         // Act
-        await _consumer.Consume(context);
+        await _consumer.HandleAsync(request);
 
         // Assert
         await _queueItemRemover.Received(1).RemoveQueueItemAsync(
@@ -257,7 +246,7 @@ public class DownloadRemoverConsumerTests
     }
 
     [Fact]
-    public async Task Consume_WithDifferentInstanceTypes_HandlesCorrectly()
+    public async Task HandleAsync_WithDifferentInstanceTypes_HandlesCorrectly()
     {
         // Arrange
         var request = new QueueItemRemoveRequest
@@ -273,14 +262,13 @@ public class DownloadRemoverConsumerTests
             JobRunId = Guid.NewGuid(),
             IsDryRun = false,
         };
-        var context = CreateConsumeContext(request);
 
         _queueItemRemover
             .RemoveQueueItemAsync(Arg.Any<QueueItemRemoveRequest>())
             .Returns(Task.CompletedTask);
 
         // Act
-        await _consumer.Consume(context);
+        await _consumer.HandleAsync(request);
 
         // Assert
         await _queueItemRemover.Received(1).RemoveQueueItemAsync(
@@ -330,12 +318,6 @@ public class DownloadRemoverConsumerTests
         };
     }
 
-    private static ConsumeContext<QueueItemRemoveRequest> CreateConsumeContext(QueueItemRemoveRequest message)
-    {
-        var context = Substitute.For<ConsumeContext<QueueItemRemoveRequest>>();
-        context.Message.Returns(message);
-        return context;
-    }
 
     #endregion
 }

code/backend/Cleanuparr.Infrastructure.Tests/Features/Jobs/DownloadCleanerOrphanedFilesTests.cs

@@ -47,7 +47,7 @@ public sealed class DownloadCleanerOrphanedFilesTests : IDisposable
         _logger,
         _fixture.DataContext,
         _fixture.Cache,
-        _fixture.MessageBus,
+        _fixture.RemovalQueue.Writer,
         _fixture.ArrClientFactory,
         _fixture.ArrQueueReader,
         _fixture.DownloadServiceFactory,

code/backend/Cleanuparr.Infrastructure.Tests/Features/Jobs/DownloadCleanerTests.cs

@@ -45,7 +45,7 @@ public class DownloadCleanerTests : IDisposable
             _logger,
             _fixture.DataContext,
             _fixture.Cache,
-            _fixture.MessageBus,
+            _fixture.RemovalQueue.Writer,
             _fixture.ArrClientFactory,
             _fixture.ArrQueueReader,
             _fixture.DownloadServiceFactory,

code/backend/Cleanuparr.Infrastructure.Tests/Features/Jobs/GenericHandlerTests.cs

@@ -10,6 +10,7 @@ using Cleanuparr.Infrastructure.Features.DownloadRemover.Models;
 using Cleanuparr.Infrastructure.Features.Jobs;
 using Cleanuparr.Infrastructure.Helpers;
 using Cleanuparr.Infrastructure.Tests.Features.Jobs.TestHelpers;
+using Cleanuparr.Infrastructure.Tests.TestHelpers;
 using Cleanuparr.Persistence;
 using Cleanuparr.Persistence.Models.Configuration;
 using Cleanuparr.Persistence.Models.Events;
@@ -18,7 +19,7 @@ using Cleanuparr.Persistence.Models.Configuration.DownloadCleaner;
 using Cleanuparr.Persistence.Models.Configuration.General;
 using Cleanuparr.Persistence.Models.Configuration.MalwareBlocker;
 using Cleanuparr.Persistence.Models.Configuration.QueueCleaner;
-using MassTransit;
+using System.Threading.Channels;
 using Microsoft.EntityFrameworkCore;
 using Cleanuparr.Infrastructure.Interceptors;
 using Microsoft.Extensions.Caching.Memory;
@@ -43,7 +44,7 @@ public class GenericHandlerTests : IClassFixture<JobHandlerFixture>
             Substitute.For<ILogger<GenericHandler>>(),
             _fixture.DataContext,
             _fixture.Cache,
-            _fixture.MessageBus,
+            _fixture.RemovalQueue.Writer,
             _fixture.ArrClientFactory,
             _fixture.ArrQueueReader,
             _fixture.DownloadServiceFactory,
@@ -293,7 +294,7 @@ public class GenericHandlerTests : IClassFixture<JobHandlerFixture>
             DeleteReason.FailedImport);
 
         // Assert
-        await _fixture.MessageBus.DidNotReceiveWithAnyArgs().Publish(default(object)!, default(CancellationToken));
+        _fixture.RemovalQueue.Reader.Count.ShouldBe(0);
     }
 
     [Fact]
@@ -316,8 +317,7 @@ public class GenericHandlerTests : IClassFixture<JobHandlerFixture>
             instance, record, isPack: false, removeFromClient: true, DeleteReason.FailedImport);
 
         // Assert
-        await _fixture.MessageBus.Received(1)
-            .Publish(Arg.Any<QueueItemRemoveRequest>(), Arg.Any<CancellationToken>());
+        _fixture.RemovalQueue.Reader.Count.ShouldBe(1);
         await _fixture.EventPublisher.Received(1).PublishAsync(
             EventType.DownloadMarkedForDeletion, Arg.Any<string>(), Arg.Any<EventSeverity>(),
             Arg.Any<Action<AppEvent>?>(), Arg.Any<Guid?>(), Arg.Any<Guid?>(), Arg.Any<bool?>());
@@ -344,8 +344,8 @@ public class GenericHandlerTests : IClassFixture<JobHandlerFixture>
             instance, record, isPack: false, removeFromClient: true, DeleteReason.FailedImport);
 
         // Assert
-        await _fixture.MessageBus.Received(1)
-            .Publish(Arg.Is<QueueItemRemoveRequest>(r => r.IsDryRun), Arg.Any<CancellationToken>());
+        QueueItemRemoveRequest request = _fixture.RemovalQueue.ShouldHaveSingle();
+        request.IsDryRun.ShouldBeTrue();
     }
 
     [Fact]
@@ -368,8 +368,7 @@ public class GenericHandlerTests : IClassFixture<JobHandlerFixture>
             instance, record, isPack: false, removeFromClient: false, DeleteReason.Stalled);
 
         // Assert
-        await _fixture.MessageBus.Received(1)
-            .Publish(Arg.Any<QueueItemRemoveRequest>(), Arg.Any<CancellationToken>());
+        _fixture.RemovalQueue.Reader.Count.ShouldBe(1);
     }
 
     [Fact]
@@ -392,8 +391,7 @@ public class GenericHandlerTests : IClassFixture<JobHandlerFixture>
             instance, record, isPack: false, removeFromClient: true, DeleteReason.FailedImport);
 
         // Assert
-        await _fixture.MessageBus.Received(1)
-            .Publish(Arg.Any<QueueItemRemoveRequest>(), Arg.Any<CancellationToken>());
+        _fixture.RemovalQueue.Reader.Count.ShouldBe(1);
     }
 
     [Fact]
@@ -411,45 +409,32 @@ public class GenericHandlerTests : IClassFixture<JobHandlerFixture>
         };
         QueueRecord record = NewRecord(seriesId: 1, episodeId: 2);
         string key = CacheKeys.DownloadMarkedForRemoval(record.DownloadId, instance.Url);
-        bool markedWhenPublished = false;
 
-        _fixture.MessageBus
-            .When(bus => bus.Publish(Arg.Any<QueueItemRemoveRequest>(), Arg.Any<CancellationToken>()))
-            .Do(_ => markedWhenPublished = _fixture.Cache.TryGetValue(key, out bool _));
+        MarkOrderSpyChannelWriter spyWriter = new(
+            _fixture.RemovalQueue.Writer,
+            () => _fixture.Cache.TryGetValue(key, out bool _));
+
+        TestHandler handler = new(
+            Substitute.For<ILogger<GenericHandler>>(),
+            _fixture.DataContext,
+            _fixture.Cache,
+            spyWriter,
+            _fixture.ArrClientFactory,
+            _fixture.ArrQueueReader,
+            _fixture.DownloadServiceFactory,
+            _fixture.EventPublisher,
+            _fixture.DryRunInterceptor,
+            _fixture.ForceImportService);
 
         // Act
-        await _handler.PublicPublishQueueItemRemoveRequest(
+        await handler.PublicPublishQueueItemRemoveRequest(
             instance, record, isPack: false, removeFromClient: true, DeleteReason.FailedImport);
 
-        // Assert
-        markedWhenPublished.ShouldBeTrue();
-    }
+        // Assert: the mark was already set at the moment of the write
+        spyWriter.MarkWasSetAtWrite.ShouldBe(true);
 
-    [Fact]
-    public async Task PublishQueueItemRemoveRequest_PublishThrows_RemovesMark()
-    {
-        // Arrange
-        ArrConfig arrConfig = new() { Type = InstanceType.Sonarr, Instances = [] };
-        ArrInstance instance = new()
-        {
-            Name = "s",
-            Url = new Uri("http://s"),
-            ApiKey = "k",
-            ArrConfig = arrConfig,
-            Version = 4f,
-        };
-        QueueRecord record = NewRecord(seriesId: 1, episodeId: 2);
-        string key = CacheKeys.DownloadMarkedForRemoval(record.DownloadId, instance.Url);
-
-        _fixture.MessageBus
-            .Publish(Arg.Any<QueueItemRemoveRequest>(), Arg.Any<CancellationToken>())
-            .Returns(Task.FromException(new InvalidOperationException("bus is down")));
-
-        // Act & Assert
-        await Should.ThrowAsync<InvalidOperationException>(() => _handler.PublicPublishQueueItemRemoveRequest(
-            instance, record, isPack: false, removeFromClient: true, DeleteReason.FailedImport));
-
-        _fixture.Cache.TryGetValue(key, out bool _).ShouldBeFalse();
+        // Assert: a successful write leaves the mark set afterwards too
+        _fixture.Cache.TryGetValue(key, out bool _).ShouldBeTrue();
     }
 
     [Fact]
@@ -475,32 +460,6 @@ public class GenericHandlerTests : IClassFixture<JobHandlerFixture>
         _fixture.ForceImportService.Received(1).Forget(instance, record.DownloadId);
     }
 
-    [Fact]
-    public async Task PublishQueueItemRemoveRequest_PublishThrows_KeepsThePendingForceImport()
-    {
-        // Arrange
-        ArrConfig arrConfig = new() { Type = InstanceType.Sonarr, Instances = [] };
-        ArrInstance instance = new()
-        {
-            Name = "s",
-            Url = new Uri("http://s"),
-            ApiKey = "k",
-            ArrConfig = arrConfig,
-            Version = 4f,
-        };
-        QueueRecord record = NewRecord(seriesId: 1, episodeId: 2);
-
-        _fixture.MessageBus
-            .Publish(Arg.Any<QueueItemRemoveRequest>(), Arg.Any<CancellationToken>())
-            .Returns(Task.FromException(new InvalidOperationException("bus is down")));
-
-        // Act & Assert: nothing was removed, so a pending import still stands
-        await Should.ThrowAsync<InvalidOperationException>(() => _handler.PublicPublishQueueItemRemoveRequest(
-            instance, record, isPack: false, removeFromClient: true, DeleteReason.FailedImport));
-
-        _fixture.ForceImportService.DidNotReceive().Forget(Arg.Any<ArrInstance>(), Arg.Any<string>());
-    }
-
     #endregion
 
     #region GetInitializedDownloadServicesAsync
@@ -679,6 +638,32 @@ public class GenericHandlerTests : IClassFixture<JobHandlerFixture>
         };
     }
 
+    /// <summary>
+    /// Wraps a real channel writer and records, at the moment of the write, whether the given probe is true.
+    /// </summary>
+    private sealed class MarkOrderSpyChannelWriter : ChannelWriter<QueueItemRemoveRequest>
+    {
+        private readonly ChannelWriter<QueueItemRemoveRequest> _inner;
+        private readonly Func<bool> _probe;
+
+        public bool? MarkWasSetAtWrite { get; private set; }
+
+        public MarkOrderSpyChannelWriter(ChannelWriter<QueueItemRemoveRequest> inner, Func<bool> probe)
+        {
+            _inner = inner;
+            _probe = probe;
+        }
+
+        public override bool TryWrite(QueueItemRemoveRequest item)
+        {
+            MarkWasSetAtWrite = _probe();
+            return _inner.TryWrite(item);
+        }
+
+        public override ValueTask<bool> WaitToWriteAsync(CancellationToken cancellationToken = default) =>
+            _inner.WaitToWriteAsync(cancellationToken);
+    }
+
     /// <summary>
     /// Concrete GenericHandler subclass exposing protected members for testing.
     /// </summary>
@@ -706,14 +691,14 @@ public class GenericHandlerTests : IClassFixture<JobHandlerFixture>
             ILogger<GenericHandler> logger,
             DataContext dataContext,
             IMemoryCache cache,
-            IBus messageBus,
+            ChannelWriter<QueueItemRemoveRequest> removalQueue,
             IArrClientFactory arrClientFactory,
             IArrQueueReader arrQueueReader,
             IDownloadServiceFactory downloadServiceFactory,
             IEventPublisher eventPublisher,
             IDryRunInterceptor dryRunInterceptor,
             IForceImportService forceImportService)
-            : base(logger, dataContext, cache, messageBus, arrClientFactory, arrQueueReader, downloadServiceFactory, eventPublisher, dryRunInterceptor, forceImportService)
+            : base(logger, dataContext, cache, removalQueue, arrClientFactory, arrQueueReader, downloadServiceFactory, eventPublisher, dryRunInterceptor, forceImportService)
         {
         }
 

code/backend/Cleanuparr.Infrastructure.Tests/Features/Jobs/Integration/DownloadCleanerIntegrationTests.cs

@@ -51,7 +51,7 @@ public class DownloadCleanerIntegrationTests : IDisposable
             Substitute.For<ILogger<DownloadCleaner>>(),
             _fixture.DataContext,
             _fixture.Cache,
-            _fixture.MessageBus,
+            _fixture.RemovalQueue.Writer,
             _fixture.ArrClientFactory,
             _fixture.ArrQueueReader,
             _fixture.DownloadServiceFactory,

code/backend/Cleanuparr.Infrastructure.Tests/Features/Jobs/Integration/IntegrationTestFixture.cs

@@ -24,7 +24,7 @@ using Cleanuparr.Persistence.Models.Configuration;
 using Cleanuparr.Persistence.Models.Configuration.Arr;
 using Cleanuparr.Persistence.Models.State;
 using Cleanuparr.Persistence.Providers;
-using MassTransit;
+using System.Threading.Channels;
 using Microsoft.Extensions.Caching.Memory;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Time.Testing;
@@ -49,7 +49,7 @@ public class IntegrationTestFixture : IDisposable
     public FakeTimeProvider TimeProvider { get; private set; }
 
     // Mocks
-    public IBus MessageBus { get; private set; }
+    public Channel<QueueItemRemoveRequest> RemovalQueue { get; private set; }
     public IArrClientFactory ArrClientFactory { get; private set; }
     public IArrClient ArrClient { get; private set; }
     public IArrQueueReader ArrQueueReader { get; private set; }
@@ -83,7 +83,7 @@ public class IntegrationTestFixture : IDisposable
         Cache = new MemoryCache(new MemoryCacheOptions());
         TimeProvider = new FakeTimeProvider();
 
-        MessageBus = Substitute.For<IBus>();
+        RemovalQueue = Channel.CreateUnbounded<QueueItemRemoveRequest>();
         ArrClientFactory = Substitute.For<IArrClientFactory>();
         ArrClient = Substitute.For<IArrClient>();
         ArrQueueReader = Substitute.For<IArrQueueReader>();
@@ -112,10 +112,6 @@ public class IntegrationTestFixture : IDisposable
         DryRunInterceptor.IsDryRunEnabled().Returns(false);
         DryRunInterceptor.InterceptAsync(Arg.Any<Func<Task>>(), Arg.Any<string?>()).ReturnsForAnyArgs(Task.CompletedTask);
 
-        MessageBus.Publish(default(QueueItemRemoveRequest)!, default)
-            .ReturnsForAnyArgs(Task.CompletedTask)
-            .AndDoes(ci => CapturedMessages.Add(ci[0]));
-
         // Seed a JobRun so EventPublisher FK constraints are satisfied
         JobRunId = Guid.NewGuid();
         EventsContext.JobRuns.Add(new JobRun { Id = JobRunId, Type = JobType.QueueCleaner });
@@ -174,6 +170,11 @@ public class IntegrationTestFixture : IDisposable
 
     public List<QueueItemRemoveRequest> GetCapturedRemoveRequests()
     {
+        while (RemovalQueue.Reader.TryRead(out QueueItemRemoveRequest? request))
+        {
+            CapturedMessages.Add(request);
+        }
+
         return CapturedMessages
             .OfType<QueueItemRemoveRequest>()
             .DistinctBy(r => r.Target.DownloadId)
@@ -252,7 +253,7 @@ public class IntegrationTestFixture : IDisposable
         CapturedMessages.Clear();
 
         // Recreate all NSubstitute mocks to clear received call state
-        MessageBus = Substitute.For<IBus>();
+        RemovalQueue = Channel.CreateUnbounded<QueueItemRemoveRequest>();
         ArrClientFactory = Substitute.For<IArrClientFactory>();
         ArrClient = Substitute.For<IArrClient>();
         ArrQueueReader = Substitute.For<IArrQueueReader>();

code/backend/Cleanuparr.Infrastructure.Tests/Features/Jobs/Integration/MalwareBlockerIntegrationTests.cs

@@ -39,7 +39,7 @@ public class MalwareBlockerIntegrationTests : IDisposable
             Substitute.For<ILogger<MalwareBlockerJob>>(),
             _fixture.DataContext,
             _fixture.Cache,
-            _fixture.MessageBus,
+            _fixture.RemovalQueue.Writer,
             _fixture.ArrClientFactory,
             _fixture.ArrQueueReader,
             _fixture.DownloadServiceFactory,

code/backend/Cleanuparr.Infrastructure.Tests/Features/Jobs/Integration/QueueCleanerIntegrationTests.cs

@@ -42,7 +42,7 @@ public class QueueCleanerIntegrationTests : IDisposable
             Substitute.For<ILogger<QueueCleanerJob>>(),
             _fixture.DataContext,
             _fixture.Cache,
-            _fixture.MessageBus,
+            _fixture.RemovalQueue.Writer,
             _fixture.ArrClientFactory,
             _fixture.ArrQueueReader,
             _fixture.DownloadServiceFactory,
@@ -84,7 +84,7 @@ public class QueueCleanerIntegrationTests : IDisposable
         // Act
         await sut.ExecuteAsync();
 
-        // Assert Phase 1: IBus received a remove request
+        // Assert Phase 1: removal queue received a remove request
         var removeRequests = _fixture.GetCapturedRemoveRequests();
         removeRequests.Count.ShouldBe(1);
 

code/backend/Cleanuparr.Infrastructure.Tests/Features/Jobs/MalwareBlockerTests.cs

@@ -50,7 +50,7 @@ public class MalwareBlockerTests : IDisposable
             _logger,
             _fixture.DataContext,
             _fixture.Cache,
-            _fixture.MessageBus,
+            _fixture.RemovalQueue.Writer,
             _fixture.ArrClientFactory,
             _fixture.ArrQueueReader,
             _fixture.DownloadServiceFactory,
@@ -371,12 +371,8 @@ public class MalwareBlockerTests : IDisposable
         await sut.ExecuteAsync();
 
         // Assert
-        await _fixture.MessageBus.Received(1).Publish(
-            Arg.Is<QueueItemRemoveRequest>(r =>
-                r.DeleteReason == DeleteReason.AllFilesBlocked
-            ),
-            Arg.Any<CancellationToken>()
-        );
+        QueueItemRemoveRequest r = _fixture.RemovalQueue.ShouldHaveSingle();
+        r.DeleteReason.ShouldBe(DeleteReason.AllFilesBlocked);
     }
 
     [Fact]
@@ -492,12 +488,8 @@ public class MalwareBlockerTests : IDisposable
         await sut.ExecuteAsync();
 
         // Assert
-        await _fixture.MessageBus.Received(1).Publish(
-            Arg.Is<QueueItemRemoveRequest>(r =>
-                r.DeleteReason == DeleteReason.AtLeastOneFileBlocked
-            ),
-            Arg.Any<CancellationToken>()
-        );
+        QueueItemRemoveRequest r = _fixture.RemovalQueue.ShouldHaveSingle();
+        r.DeleteReason.ShouldBe(DeleteReason.AtLeastOneFileBlocked);
     }
 
     [Fact]
@@ -602,12 +594,8 @@ public class MalwareBlockerTests : IDisposable
         await sut.ExecuteAsync();
 
         // Assert - RemoveFromClient should be false
-        await _fixture.MessageBus.Received(1).Publish(
-            Arg.Is<QueueItemRemoveRequest>(r =>
-                r.ArrTarget().RemoveFromClient == false
-            ),
-            Arg.Any<CancellationToken>()
-        );
+        QueueItemRemoveRequest r = _fixture.RemovalQueue.ShouldHaveSingle();
+        r.ArrTarget().RemoveFromClient.ShouldBeFalse();
     }
 
     [Fact]
@@ -697,10 +685,7 @@ public class MalwareBlockerTests : IDisposable
         // Assert
         _logger.HasLogContaining(LogLevel.Information, "skip | item is missing the content id").ShouldBeTrue();
 
-        await _fixture.MessageBus.DidNotReceive().Publish(
-            Arg.Any<QueueItemRemoveRequest>(),
-            Arg.Any<CancellationToken>()
-        );
+        _fixture.RemovalQueue.Reader.Count.ShouldBe(0);
     }
 
     [Fact]
@@ -759,13 +744,9 @@ public class MalwareBlockerTests : IDisposable
         await sut.ExecuteAsync();
 
         // Assert - SkipSearch must be true because the item has no content ID
-        await _fixture.MessageBus.Received(1).Publish(
-            Arg.Is<QueueItemRemoveRequest>(r =>
-                r.SkipSearch == true &&
-                r.DeleteReason == DeleteReason.AllFilesBlocked
-            ),
-            Arg.Any<CancellationToken>()
-        );
+        QueueItemRemoveRequest r = _fixture.RemovalQueue.ShouldHaveSingle();
+        r.SkipSearch.ShouldBeTrue();
+        r.DeleteReason.ShouldBe(DeleteReason.AllFilesBlocked);
     }
 
     #endregion
@@ -872,14 +853,10 @@ public class MalwareBlockerTests : IDisposable
 
         // Assert
         await downloadService.Received(1).DeleteDownload(torrent, true);
-        await _fixture.MessageBus.Received(1).Publish(
-            Arg.Is<QueueItemRemoveRequest>(r =>
-                r.LazyTarget().RemovedFromClient
-                && r.LazyTarget().Item.Books.Single().BookId == "OL7353617M"
-                && r.DeleteReason == DeleteReason.AllFilesBlocked
-            ),
-            Arg.Any<CancellationToken>()
-        );
+        QueueItemRemoveRequest r = _fixture.RemovalQueue.ShouldHaveSingle();
+        r.LazyTarget().RemovedFromClient.ShouldBeTrue();
+        r.LazyTarget().Item.Books.Single().BookId.ShouldBe("OL7353617M");
+        r.DeleteReason.ShouldBe(DeleteReason.AllFilesBlocked);
         _fixture.ArrClientFactory.DidNotReceiveWithAnyArgs().GetClient(default, default);
         await _fixture.ArrQueueReader.DidNotReceiveWithAnyArgs()
             .ReadAllAsync(default!, default!);

code/backend/Cleanuparr.Infrastructure.Tests/Features/Jobs/QueueCleanerTests.cs

@@ -1,4 +1,5 @@
 using Cleanuparr.Domain.Entities.Arr;
+using System.Threading.Channels;
 using Cleanuparr.Infrastructure.Features.Arr.ForceImport;
 using Cleanuparr.Infrastructure.Features.LazyLibrarian;
 using Cleanuparr.Domain.Entities.LazyLibrarian;
@@ -54,7 +55,7 @@ public class QueueCleanerTests : IDisposable
             _logger,
             _fixture.DataContext,
             _fixture.Cache,
-            _fixture.MessageBus,
+            _fixture.RemovalQueue.Writer,
             _fixture.ArrClientFactory,
             _fixture.ArrQueueReader,
             _fixture.DownloadServiceFactory,
@@ -432,10 +433,7 @@ public class QueueCleanerTests : IDisposable
         await sut.ExecuteAsync();
 
         // Assert
-        await _fixture.MessageBus.Received(1).Publish(
-            Arg.Any<QueueItemRemoveRequest>(),
-            Arg.Any<CancellationToken>()
-        );
+        _fixture.RemovalQueue.Reader.Count.ShouldBe(1);
     }
 
     [Fact]
@@ -862,12 +860,16 @@ public class QueueCleanerTests : IDisposable
         await sut.ExecuteAsync();
 
         // Assert - the pack flag true branch builds a season SearchItem instead of an episode one
-        await _fixture.MessageBus.Received(1).Publish(
-            Arg.Is<QueueItemRemoveRequest>(r =>
-                r.SeriesItem().SearchType == SeriesSearchType.Season &&
-                r.SeriesItem().Id == 1
-            ),
-            Arg.Any<CancellationToken>()
+        List<QueueItemRemoveRequest> requests = [];
+
+        while (_fixture.RemovalQueue.Reader.TryRead(out QueueItemRemoveRequest? request))
+        {
+            requests.Add(request);
+        }
+
+        requests.ShouldContain(r =>
+            r.SeriesItem().SearchType == SeriesSearchType.Season &&
+            r.SeriesItem().Id == 1
         );
     }
 
@@ -924,12 +926,8 @@ public class QueueCleanerTests : IDisposable
         await sut.ExecuteAsync();
 
         // Assert
-        await _fixture.MessageBus.Received(1).Publish(
-            Arg.Is<QueueItemRemoveRequest>(r =>
-                r.DeleteReason == DeleteReason.FailedImport
-            ),
-            Arg.Any<CancellationToken>()
-        );
+        QueueItemRemoveRequest r = _fixture.RemovalQueue.ShouldHaveSingle();
+        r.DeleteReason.ShouldBe(DeleteReason.FailedImport);
     }
 
     [Fact]
@@ -967,10 +965,7 @@ public class QueueCleanerTests : IDisposable
         // Assert
         _logger.HasLogContaining(LogLevel.Information, "skip | item is missing the content id").ShouldBeTrue();
 
-        await _fixture.MessageBus.DidNotReceive().Publish(
-            Arg.Any<QueueItemRemoveRequest>(),
-            Arg.Any<CancellationToken>()
-        );
+        _fixture.RemovalQueue.Reader.Count.ShouldBe(0);
     }
 
     [Fact]
@@ -1035,13 +1030,9 @@ public class QueueCleanerTests : IDisposable
         await sut.ExecuteAsync();
 
         // Assert - SkipSearch must be true because the item has no content ID
-        await _fixture.MessageBus.Received(1).Publish(
-            Arg.Is<QueueItemRemoveRequest>(r =>
-                r.SkipSearch == true &&
-                r.DeleteReason == DeleteReason.Stalled
-            ),
-            Arg.Any<CancellationToken>()
-        );
+        QueueItemRemoveRequest r = _fixture.RemovalQueue.ShouldHaveSingle();
+        r.SkipSearch.ShouldBeTrue();
+        r.DeleteReason.ShouldBe(DeleteReason.Stalled);
     }
 
     #endregion
@@ -1175,10 +1166,7 @@ public class QueueCleanerTests : IDisposable
         _logger.HasLogContaining(LogLevel.Debug, "skip removal request").ShouldBeTrue();
 
         // Verify no publish was made
-        await _fixture.MessageBus.DidNotReceive().Publish(
-            Arg.Any<QueueItemRemoveRequest>(),
-            Arg.Any<CancellationToken>()
-        );
+        _fixture.RemovalQueue.Reader.Count.ShouldBe(0);
     }
 
     [Fact]
@@ -1234,14 +1222,10 @@ public class QueueCleanerTests : IDisposable
         await sut.ExecuteAsync();
 
         // Assert - should publish QueueItemRemoveRequest (not SeriesSearchItem)
-        await _fixture.MessageBus.Received(1).Publish(
-            Arg.Is<QueueItemRemoveRequest>(r =>
-                r.Instance.ArrConfig.Type == InstanceType.Radarr &&
-                r.ArrTarget().SearchItem.Id == 42 &&
-                r.DeleteReason == DeleteReason.Stalled
-            ),
-            Arg.Any<CancellationToken>()
-        );
+        QueueItemRemoveRequest r = _fixture.RemovalQueue.ShouldHaveSingle();
+        r.Instance.ArrConfig.Type.ShouldBe(InstanceType.Radarr);
+        r.ArrTarget().SearchItem.Id.ShouldBe(42);
+        r.DeleteReason.ShouldBe(DeleteReason.Stalled);
     }
 
     [Fact]
@@ -1297,14 +1281,10 @@ public class QueueCleanerTests : IDisposable
         await sut.ExecuteAsync();
 
         // Assert - should publish QueueItemRemoveRequest with AlbumId
-        await _fixture.MessageBus.Received(1).Publish(
-            Arg.Is<QueueItemRemoveRequest>(r =>
-                r.Instance.ArrConfig.Type == InstanceType.Lidarr &&
-                r.ArrTarget().SearchItem.Id == 123 &&
-                r.DeleteReason == DeleteReason.SlowSpeed
-            ),
-            Arg.Any<CancellationToken>()
-        );
+        QueueItemRemoveRequest r = _fixture.RemovalQueue.ShouldHaveSingle();
+        r.Instance.ArrConfig.Type.ShouldBe(InstanceType.Lidarr);
+        r.ArrTarget().SearchItem.Id.ShouldBe(123);
+        r.DeleteReason.ShouldBe(DeleteReason.SlowSpeed);
     }
 
     [Fact]
@@ -1360,14 +1340,10 @@ public class QueueCleanerTests : IDisposable
         await sut.ExecuteAsync();
 
         // Assert - should publish QueueItemRemoveRequest with BookId
-        await _fixture.MessageBus.Received(1).Publish(
-            Arg.Is<QueueItemRemoveRequest>(r =>
-                r.Instance.ArrConfig.Type == InstanceType.Readarr &&
-                r.ArrTarget().SearchItem.Id == 456 &&
-                r.DeleteReason == DeleteReason.Stalled
-            ),
-            Arg.Any<CancellationToken>()
-        );
+        QueueItemRemoveRequest r = _fixture.RemovalQueue.ShouldHaveSingle();
+        r.Instance.ArrConfig.Type.ShouldBe(InstanceType.Readarr);
+        r.ArrTarget().SearchItem.Id.ShouldBe(456);
+        r.DeleteReason.ShouldBe(DeleteReason.Stalled);
     }
 
     [Fact]
@@ -1424,16 +1400,12 @@ public class QueueCleanerTests : IDisposable
         await sut.ExecuteAsync();
 
         // Assert - should publish QueueItemRemoveRequest
-        await _fixture.MessageBus.Received(1).Publish(
-            Arg.Is<QueueItemRemoveRequest>(r =>
-                r.Instance.ArrConfig.Type == InstanceType.Whisparr &&
-                r.ArrTarget().SearchItem.Id == 100 && // EpisodeId
-                r.SeriesItem().SeriesId == 10 &&
-                r.SeriesItem().SearchType == SeriesSearchType.Episode &&
-                r.DeleteReason == DeleteReason.Stalled
-            ),
-            Arg.Any<CancellationToken>()
-        );
+        QueueItemRemoveRequest r = _fixture.RemovalQueue.ShouldHaveSingle();
+        r.Instance.ArrConfig.Type.ShouldBe(InstanceType.Whisparr);
+        r.ArrTarget().SearchItem.Id.ShouldBe(100); // EpisodeId
+        r.SeriesItem().SeriesId.ShouldBe(10);
+        r.SeriesItem().SearchType.ShouldBe(SeriesSearchType.Episode);
+        r.DeleteReason.ShouldBe(DeleteReason.Stalled);
     }
 
     [Fact]
@@ -1489,14 +1461,10 @@ public class QueueCleanerTests : IDisposable
         await sut.ExecuteAsync();
 
         // Assert - should publish QueueItemRemoveRequest with MovieId
-        await _fixture.MessageBus.Received(1).Publish(
-            Arg.Is<QueueItemRemoveRequest>(r =>
-                r.Instance.ArrConfig.Type == InstanceType.Whisparr &&
-                r.ArrTarget().SearchItem.Id == 42 && // MovieId
-                r.DeleteReason == DeleteReason.Stalled
-            ),
-            Arg.Any<CancellationToken>()
-        );
+        QueueItemRemoveRequest r = _fixture.RemovalQueue.ShouldHaveSingle();
+        r.Instance.ArrConfig.Type.ShouldBe(InstanceType.Whisparr);
+        r.ArrTarget().SearchItem.Id.ShouldBe(42); // MovieId
+        r.DeleteReason.ShouldBe(DeleteReason.Stalled);
     }
 
     [Fact]
@@ -1566,16 +1534,12 @@ public class QueueCleanerTests : IDisposable
 
         // Assert - should publish QueueItemRemoveRequest with Season search type
         // because multiple records with the same download ID indicate a pack
-        await _fixture.MessageBus.Received(1).Publish(
-            Arg.Is<QueueItemRemoveRequest>(r =>
-                r.Instance.ArrConfig.Type == InstanceType.Whisparr &&
-                r.ArrTarget().SearchItem.Id == 3 && // SeasonNumber
-                r.SeriesItem().SeriesId == 10 &&
-                r.SeriesItem().SearchType == SeriesSearchType.Season &&
-                r.DeleteReason == DeleteReason.Stalled
-            ),
-            Arg.Any<CancellationToken>()
-        );
+        QueueItemRemoveRequest r = _fixture.RemovalQueue.ShouldHaveSingle();
+        r.Instance.ArrConfig.Type.ShouldBe(InstanceType.Whisparr);
+        r.ArrTarget().SearchItem.Id.ShouldBe(3); // SeasonNumber
+        r.SeriesItem().SeriesId.ShouldBe(10);
+        r.SeriesItem().SearchType.ShouldBe(SeriesSearchType.Season);
+        r.DeleteReason.ShouldBe(DeleteReason.Stalled);
     }
 
     #endregion
@@ -1646,14 +1610,10 @@ public class QueueCleanerTests : IDisposable
         await sut.ExecuteAsync();
 
         // Assert
-        await _fixture.MessageBus.Received(1).Publish(
-            Arg.Is<QueueItemRemoveRequest>(r =>
-                r.DeleteReason == DeleteReason.FailedImport &&
-                r.ArrTarget().ChangeCategory == true &&
-                r.ArrTarget().RemoveFromClient == false
-            ),
-            Arg.Any<CancellationToken>()
-        );
+        QueueItemRemoveRequest r = _fixture.RemovalQueue.ShouldHaveSingle();
+        r.DeleteReason.ShouldBe(DeleteReason.FailedImport);
+        r.ArrTarget().ChangeCategory.ShouldBeTrue();
+        r.ArrTarget().RemoveFromClient.ShouldBeFalse();
     }
 
     [Fact]
@@ -1711,14 +1671,10 @@ public class QueueCleanerTests : IDisposable
         await sut.ExecuteAsync();
 
         // Assert
-        await _fixture.MessageBus.Received(1).Publish(
-            Arg.Is<QueueItemRemoveRequest>(r =>
-                r.DeleteReason == DeleteReason.Stalled &&
-                r.ArrTarget().ChangeCategory == true &&
-                r.ArrTarget().RemoveFromClient == false
-            ),
-            Arg.Any<CancellationToken>()
-        );
+        QueueItemRemoveRequest r = _fixture.RemovalQueue.ShouldHaveSingle();
+        r.DeleteReason.ShouldBe(DeleteReason.Stalled);
+        r.ArrTarget().ChangeCategory.ShouldBeTrue();
+        r.ArrTarget().RemoveFromClient.ShouldBeFalse();
     }
 
     #endregion
@@ -1788,14 +1744,10 @@ public class QueueCleanerTests : IDisposable
 
         // Assert
         await downloadService.Received(1).DeleteDownload(torrent, true);
-        await _fixture.MessageBus.Received(1).Publish(
-            Arg.Is<QueueItemRemoveRequest>(r =>
-                r.LazyTarget().RemovedFromClient
-                && r.LazyTarget().Item.Books.Single().BookId == "OL7353617M"
-                && r.DeleteReason == DeleteReason.Stalled
-            ),
-            Arg.Any<CancellationToken>()
-        );
+        QueueItemRemoveRequest r = _fixture.RemovalQueue.ShouldHaveSingle();
+        r.LazyTarget().RemovedFromClient.ShouldBeTrue();
+        r.LazyTarget().Item.Books.Single().BookId.ShouldBe("OL7353617M");
+        r.DeleteReason.ShouldBe(DeleteReason.Stalled);
     }
 
     [Fact]
@@ -1811,12 +1763,8 @@ public class QueueCleanerTests : IDisposable
 
         // Assert
         await downloadService.DidNotReceive().DeleteDownload(Arg.Any<ITorrentItemWrapper>(), Arg.Any<bool>());
-        await _fixture.MessageBus.Received(1).Publish(
-            Arg.Is<QueueItemRemoveRequest>(r =>
-                !r.LazyTarget().RemovedFromClient
-            ),
-            Arg.Any<CancellationToken>()
-        );
+        QueueItemRemoveRequest r = _fixture.RemovalQueue.ShouldHaveSingle();
+        r.LazyTarget().RemovedFromClient.ShouldBeFalse();
     }
 
     [Fact]
@@ -1834,63 +1782,7 @@ public class QueueCleanerTests : IDisposable
         await sut.ExecuteAsync();
 
         // Assert
-        await _fixture.MessageBus.DidNotReceive().Publish(
-            Arg.Any<QueueItemRemoveRequest>(),
-            Arg.Any<CancellationToken>()
-        );
-    }
-
-    [Fact]
-    public async Task ProcessInstanceAsync_LazyLibrarian_KeepsGoingWhenOneRemovalCannotBePublished()
-    {
-        // Arrange: the first torrent is already deleted, so a failed publish must not skip the second book.
-        (IDownloadService downloadService, ITorrentItemWrapper torrent) = StubLazyLibrarianDecision();
-        LazyLibrarianQueueItem second = CreateBookItem() with { DownloadId = "torrent-hash-2" };
-
-        ITorrentItemWrapper secondTorrent = Substitute.For<ITorrentItemWrapper>();
-        secondTorrent.Hash.Returns("torrent-hash-2");
-
-        DownloadClientConfig clientConfig = downloadService.ClientConfig;
-        IReadOnlyList<LazyLibrarianRemovalDecision> decisions =
-        [
-            new LazyLibrarianRemovalDecision
-            {
-                Item = CreateBookItem(),
-                DeleteReason = DeleteReason.Stalled,
-                RemoveFromClient = true,
-                DownloadClient = clientConfig,
-                DownloadService = downloadService,
-                Torrent = torrent,
-            },
-            new LazyLibrarianRemovalDecision
-            {
-                Item = second,
-                DeleteReason = DeleteReason.Stalled,
-                RemoveFromClient = true,
-                DownloadClient = clientConfig,
-                DownloadService = downloadService,
-                Torrent = secondTorrent,
-            },
-        ];
-
-        _fixture.LazyLibrarianServiceQC
-            .EvaluateAsync(Arg.Any<ArrInstance>(), Arg.Any<IReadOnlyList<IDownloadService>>(), Arg.Any<IReadOnlyList<string>>())
-            .Returns(decisions);
-
-        _fixture.MessageBus
-            .Publish(Arg.Is<QueueItemRemoveRequest>(r => r.Target.DownloadId == "torrent-hash"), Arg.Any<CancellationToken>())
-            .Returns(Task.FromException(new Exception("bus is down")));
-
-        var sut = CreateSut();
-
-        // Act
-        await sut.ExecuteAsync();
-
-        // Assert
-        await _fixture.MessageBus.Received(1).Publish(
-            Arg.Is<QueueItemRemoveRequest>(r => r.Target.DownloadId == "torrent-hash-2"),
-            Arg.Any<CancellationToken>()
-        );
+        _fixture.RemovalQueue.Reader.Count.ShouldBe(0);
     }
 
     [Fact]

code/backend/Cleanuparr.Infrastructure.Tests/Features/Jobs/TestHelpers/JobHandlerFixture.cs

@@ -1,15 +1,16 @@
-using Cleanuparr.Infrastructure.Events.Interfaces;
+using System.Threading.Channels;
+using Cleanuparr.Infrastructure.Events.Interfaces;
 using Cleanuparr.Infrastructure.Features.Arr.Interfaces;
 using Cleanuparr.Infrastructure.Features.Context;
 using Cleanuparr.Infrastructure.Features.DownloadCleaner.Services;
 using Cleanuparr.Infrastructure.Features.DownloadClient;
+using Cleanuparr.Infrastructure.Features.DownloadRemover.Models;
 using Cleanuparr.Infrastructure.Features.Files;
 using Cleanuparr.Infrastructure.Features.Jobs;
 using Cleanuparr.Infrastructure.Features.LazyLibrarian;
 using Cleanuparr.Infrastructure.Features.MalwareBlocker;
 using Cleanuparr.Infrastructure.Interceptors;
 using Cleanuparr.Persistence;
-using MassTransit;
 using Microsoft.Extensions.Caching.Memory;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Time.Testing;
@@ -27,7 +28,7 @@ public class JobHandlerFixture : IDisposable
     public DataContext DataContext { get; private set; }
     public EventsContext EventsContext { get; private set; }
     public MemoryCache Cache { get; private set; }
-    public IBus MessageBus { get; private set; }
+    public Channel<QueueItemRemoveRequest> RemovalQueue { get; private set; }
     public IArrClientFactory ArrClientFactory { get; private set; }
     public IArrQueueReader ArrQueueReader { get; private set; }
     public IDownloadServiceFactory DownloadServiceFactory { get; private set; }
@@ -55,7 +56,7 @@ public class JobHandlerFixture : IDisposable
         DataContext = TestDataContextFactory.Create();
         EventsContext = TestDataContextFactory.CreateEvents();
         Cache = new MemoryCache(new MemoryCacheOptions());
-        MessageBus = Substitute.For<IBus>();
+        RemovalQueue = Channel.CreateUnbounded<QueueItemRemoveRequest>();
         ArrClientFactory = Substitute.For<IArrClientFactory>();
         ArrQueueReader = Substitute.For<IArrQueueReader>();
         DownloadServiceFactory = Substitute.For<IDownloadServiceFactory>();
@@ -160,7 +161,7 @@ public class JobHandlerFixture : IDisposable
     {
         SubstituteHelper.ClearPendingArgSpecs();
         // Recreate all substitutes to clear received call state
-        MessageBus = Substitute.For<IBus>();
+        RemovalQueue = Channel.CreateUnbounded<QueueItemRemoveRequest>();
         ArrClientFactory = Substitute.For<IArrClientFactory>();
         ArrQueueReader = Substitute.For<IArrQueueReader>();
         DownloadServiceFactory = Substitute.For<IDownloadServiceFactory>();

code/backend/Cleanuparr.Infrastructure.Tests/Features/Messaging/QueueWorkerTests.cs

@@ -0,0 +1,193 @@
+using System.Threading.Channels;
+using Cleanuparr.Infrastructure.Features.Context;
+using Cleanuparr.Infrastructure.Features.Messaging;
+using Cleanuparr.Infrastructure.Tests.TestHelpers;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Logging;
+using NSubstitute;
+using Shouldly;
+using Xunit;
+
+namespace Cleanuparr.Infrastructure.Tests.Features.Messaging;
+
+public class QueueWorkerTests : IAsyncDisposable
+{
+    private const string ContextKey = "QueueWorkerTests.Value";
+
+    private readonly Channel<TestMessage> _channel = Channel.CreateUnbounded<TestMessage>();
+    private readonly Recorder _recorder = new();
+    private readonly ServiceProvider _provider;
+    private readonly ILogger<QueueWorker<TestMessage>> _logger;
+    private readonly QueueWorker<TestMessage> _worker;
+
+    public QueueWorkerTests()
+    {
+        ServiceCollection services = new();
+        services.AddSingleton(_recorder);
+        services.AddScoped<ScopedMarker>();
+        services.AddScoped<IMessageHandler<TestMessage>, TestHandler>();
+        _provider = services.BuildServiceProvider();
+
+        _logger = Substitute.For<ILogger<QueueWorker<TestMessage>>>();
+        _worker = new QueueWorker<TestMessage>(_logger, _provider.GetRequiredService<IServiceScopeFactory>(), _channel);
+    }
+
+    public async ValueTask DisposeAsync()
+    {
+        await _worker.StopAsync(CancellationToken.None);
+        await _provider.DisposeAsync();
+    }
+
+    [Fact]
+    public async Task ProcessesMessages_InFifoOrder()
+    {
+        await _worker.StartAsync(CancellationToken.None);
+
+        await _channel.Writer.WriteAsync(new TestMessage(1));
+        await _channel.Writer.WriteAsync(new TestMessage(2));
+        await _channel.Writer.WriteAsync(new TestMessage(3));
+
+        await WaitUntilAsync(() => _recorder.HandledOrder.Count == 3);
+
+        _recorder.HandledOrder.ShouldBe([1, 2, 3]);
+    }
+
+    [Fact]
+    public async Task NeverRunsTwoHandlersAtOnce()
+    {
+        _recorder.Gate = new TaskCompletionSource();
+        await _worker.StartAsync(CancellationToken.None);
+
+        await _channel.Writer.WriteAsync(new TestMessage(1));
+        await _channel.Writer.WriteAsync(new TestMessage(2));
+
+        await WaitUntilAsync(() => Volatile.Read(ref _recorder.Concurrent) == 1);
+        // The second message must still be waiting behind the first.
+        _recorder.HandledOrder.ShouldBeEmpty();
+
+        _recorder.Gate.SetResult();
+        await WaitUntilAsync(() => _recorder.HandledOrder.Count == 2);
+
+        _recorder.MaxConcurrent.ShouldBe(1);
+    }
+
+    [Fact]
+    public async Task KeepsRunning_AfterAHandlerThrows_AndLogsTheError()
+    {
+        _recorder.ThrowIds.Add(1);
+        await _worker.StartAsync(CancellationToken.None);
+
+        await _channel.Writer.WriteAsync(new TestMessage(1));
+        await _channel.Writer.WriteAsync(new TestMessage(2));
+
+        await WaitUntilAsync(() => _recorder.HandledOrder.Count == 2);
+
+        _logger.HasLogContainingAtLeastOnce(LogLevel.Error, "failed to handle queued message").ShouldBeTrue();
+    }
+
+    [Fact]
+    public async Task CreatesANewScopePerMessage()
+    {
+        await _worker.StartAsync(CancellationToken.None);
+
+        await _channel.Writer.WriteAsync(new TestMessage(1));
+        await _channel.Writer.WriteAsync(new TestMessage(2));
+
+        await WaitUntilAsync(() => _recorder.ScopeIds.Count == 2);
+
+        _recorder.ScopeIds[0].ShouldNotBe(_recorder.ScopeIds[1]);
+    }
+
+    [Fact]
+    public async Task DoesNotLeak_ContextProviderValues_BetweenMessages()
+    {
+        await _worker.StartAsync(CancellationToken.None);
+
+        await _channel.Writer.WriteAsync(new TestMessage(1));
+        await _channel.Writer.WriteAsync(new TestMessage(2));
+
+        await WaitUntilAsync(() => _recorder.HandledOrder.Count == 2);
+
+        // Neither message should observe the value the previous message set.
+        _recorder.ObservedContextAtStart.ShouldAllBe(value => value == null);
+    }
+
+    [Fact]
+    public async Task Stops_WhenTheStoppingTokenCancels()
+    {
+        await _worker.StartAsync(CancellationToken.None);
+
+        await Should.NotThrowAsync(() => _worker.StopAsync(CancellationToken.None));
+    }
+
+    private static async Task WaitUntilAsync(Func<bool> condition)
+    {
+        DateTime deadline = DateTime.UtcNow.AddSeconds(5);
+
+        while (!condition() && DateTime.UtcNow < deadline)
+        {
+            await Task.Delay(10);
+        }
+
+        condition().ShouldBeTrue("condition was not met within the timeout");
+    }
+
+    public sealed record TestMessage(int Id);
+
+    private sealed class ScopedMarker
+    {
+        public readonly Guid Id = Guid.NewGuid();
+    }
+
+    private sealed class Recorder
+    {
+        public readonly List<int> HandledOrder = [];
+        public readonly List<object?> ObservedContextAtStart = [];
+        public readonly List<Guid> ScopeIds = [];
+        public readonly HashSet<int> ThrowIds = [];
+        public TaskCompletionSource? Gate;
+        public int Concurrent;
+        public int MaxConcurrent;
+    }
+
+    private sealed class TestHandler : IMessageHandler<TestMessage>
+    {
+        private readonly Recorder _recorder;
+        private readonly ScopedMarker _marker;
+
+        public TestHandler(Recorder recorder, ScopedMarker marker)
+        {
+            _recorder = recorder;
+            _marker = marker;
+        }
+
+        public Task HandleAsync(TestMessage message)
+        {
+            _recorder.ObservedContextAtStart.Add(ContextProvider.Get(ContextKey));
+            ContextProvider.Set(ContextKey, message.Id);
+
+            return HandleAsyncCore(message);
+        }
+
+        private async Task HandleAsyncCore(TestMessage message)
+        {
+            int concurrent = Interlocked.Increment(ref _recorder.Concurrent);
+            _recorder.MaxConcurrent = Math.Max(_recorder.MaxConcurrent, concurrent);
+
+            if (_recorder.Gate is not null)
+            {
+                await _recorder.Gate.Task;
+            }
+
+            _recorder.ScopeIds.Add(_marker.Id);
+            _recorder.HandledOrder.Add(message.Id);
+
+            Interlocked.Decrement(ref _recorder.Concurrent);
+
+            if (_recorder.ThrowIds.Contains(message.Id))
+            {
+                throw new InvalidOperationException($"boom {message.Id}");
+            }
+        }
+    }
+}

code/backend/Cleanuparr.Infrastructure.Tests/Features/Notifications/NotificationConsumerTests.cs

@@ -2,7 +2,6 @@ using Cleanuparr.Domain.Enums;
 using Cleanuparr.Infrastructure.Features.Notifications;
 using Cleanuparr.Infrastructure.Features.Notifications.Consumers;
 using Cleanuparr.Infrastructure.Features.Notifications.Models;
-using MassTransit;
 using Microsoft.Extensions.Logging;
 using NSubstitute;
 using NSubstitute.ExceptionExtensions;
@@ -53,15 +52,9 @@ public class NotificationConsumerTests
         };
     }
 
-    private static ConsumeContext<NotificationMessage> CreateConsumeContext(NotificationMessage message)
-    {
-        ConsumeContext<NotificationMessage> context = Substitute.For<ConsumeContext<NotificationMessage>>();
-        context.Message.Returns(message);
-        return context;
-    }
 
     [Fact]
-    public async Task Consume_WithConfiguredProviders_SendsToAll()
+    public async Task HandleAsync_WithConfiguredProviders_SendsToAll()
     {
         NotificationProviderDto providerDto1 = CreateProviderDto("Provider1");
         NotificationProviderDto providerDto2 = CreateProviderDto("Provider2");
@@ -75,14 +68,14 @@ public class NotificationConsumerTests
         _providerFactory.CreateProvider(providerDto1).Returns(provider1);
         _providerFactory.CreateProvider(providerDto2).Returns(provider2);
 
-        await _consumer.Consume(CreateConsumeContext(message));
+        await _consumer.HandleAsync(message);
 
         await provider1.Received(1).SendNotificationAsync(message.Context);
         await provider2.Received(1).SendNotificationAsync(message.Context);
     }
 
     [Fact]
-    public async Task Consume_WhenOneProviderThrows_OthersStillReceiveNotification()
+    public async Task HandleAsync_WhenOneProviderThrows_OthersStillReceiveNotification()
     {
         NotificationProviderDto providerDto1 = CreateProviderDto("Provider1");
         NotificationProviderDto providerDto2 = CreateProviderDto("Provider2");
@@ -99,35 +92,33 @@ public class NotificationConsumerTests
         _providerFactory.CreateProvider(providerDto1).Returns(provider1);
         _providerFactory.CreateProvider(providerDto2).Returns(provider2);
 
-        await _consumer.Consume(CreateConsumeContext(message));
+        await _consumer.HandleAsync(message);
 
         await provider2.Received(1).SendNotificationAsync(message.Context);
     }
 
     [Fact]
-    public async Task Consume_WhenNoProvidersConfigured_DoesNotCreateAnyProvider()
+    public async Task HandleAsync_WhenNoProvidersConfigured_DoesNotCreateAnyProvider()
     {
         NotificationMessage message = new(NotificationEventType.QueueItemDeleted, CreateContext(NotificationEventType.QueueItemDeleted));
 
         _configurationService.GetProvidersForEventAsync(NotificationEventType.QueueItemDeleted)
             .Returns(new List<NotificationProviderDto>());
 
-        await _consumer.Consume(CreateConsumeContext(message));
+        await _consumer.HandleAsync(message);
 
         _providerFactory.DidNotReceive().CreateProvider(Arg.Any<NotificationProviderDto>());
     }
 
     [Fact]
-    public async Task Consume_WhenProviderLookupThrows_DoesNotThrow()
+    public async Task HandleAsync_WhenProviderLookupThrows_DoesNotThrow()
     {
         NotificationMessage message = new(NotificationEventType.QueueItemDeleted, CreateContext(NotificationEventType.QueueItemDeleted));
 
         _configurationService.GetProvidersForEventAsync(Arg.Any<NotificationEventType>())
             .ThrowsAsync(new InvalidOperationException("db locked"));
 
-        ConsumeContext<NotificationMessage> ctx = CreateConsumeContext(message);
-
-        await Should.NotThrowAsync(() => _consumer.Consume(ctx));
+        await Should.NotThrowAsync(() => _consumer.HandleAsync(message));
 
         _providerFactory.DidNotReceive().CreateProvider(Arg.Any<NotificationProviderDto>());
     }

code/backend/Cleanuparr.Infrastructure.Tests/Features/Notifications/NotificationPublisherTests.cs

@@ -6,7 +6,7 @@ using Cleanuparr.Infrastructure.Features.Notifications.Models;
 using Cleanuparr.Infrastructure.Interceptors;
 using Cleanuparr.Infrastructure.Tests.TestHelpers;
 using Cleanuparr.Persistence.Models.Configuration.QueueCleaner;
-using MassTransit;
+using System.Threading.Channels;
 using Microsoft.Extensions.Logging;
 using NSubstitute;
 using NSubstitute.ExceptionExtensions;
@@ -19,14 +19,14 @@ public class NotificationPublisherTests
 {
     private readonly ILogger<NotificationPublisher> _logger;
     private readonly IDryRunInterceptor _dryRunInterceptor;
-    private readonly IBus _messageBus;
+    private readonly Channel<NotificationMessage> _notificationQueue;
     private readonly NotificationPublisher _publisher;
 
     public NotificationPublisherTests()
     {
         _logger = Substitute.For<ILogger<NotificationPublisher>>();
         _dryRunInterceptor = Substitute.For<IDryRunInterceptor>();
-        _messageBus = Substitute.For<IBus>();
+        _notificationQueue = Channel.CreateUnbounded<NotificationMessage>();
 
         _dryRunInterceptor.InterceptAsync(Arg.Any<Func<Task>>(), Arg.Any<string?>())
             .ReturnsForAnyArgs(ci => ci.ArgAt<Func<Task>>(0).Invoke());
@@ -34,7 +34,7 @@ public class NotificationPublisherTests
         _publisher = new NotificationPublisher(
             _logger,
             _dryRunInterceptor,
-            _messageBus);
+            _notificationQueue.Writer);
     }
 
     private void SetupContext(InstanceType instanceType = InstanceType.Sonarr)
@@ -86,10 +86,10 @@ public class NotificationPublisherTests
         await _publisher.NotifyStrike(StrikeType.Stalled, 1);
 
         // Assert
-        await _messageBus.Received(1).Publish(Arg.Is<NotificationMessage>(
-            m => m.EventType == NotificationEventType.StalledStrike &&
-                 m.Context.Data.ContainsKey("Strike type") &&
-                 m.Context.Data["Strike type"] == "Stalled"), Arg.Any<CancellationToken>());
+        NotificationMessage m = _notificationQueue.ShouldHaveSingle();
+        m.EventType.ShouldBe(NotificationEventType.StalledStrike);
+        m.Context.Data.ShouldContainKey("Strike type");
+        m.Context.Data["Strike type"].ShouldBe("Stalled");
     }
 
     [Fact]
@@ -102,9 +102,9 @@ public class NotificationPublisherTests
         await _publisher.NotifyStrike(StrikeType.FailedImport, 2);
 
         // Assert
-        await _messageBus.Received(1).Publish(Arg.Is<NotificationMessage>(
-            m => m.EventType == NotificationEventType.FailedImportStrike &&
-                 m.Context.Data["Strike count"] == "2"), Arg.Any<CancellationToken>());
+        NotificationMessage m = _notificationQueue.ShouldHaveSingle();
+        m.EventType.ShouldBe(NotificationEventType.FailedImportStrike);
+        m.Context.Data["Strike count"].ShouldBe("2");
     }
 
     [Theory]
@@ -127,8 +127,8 @@ public class NotificationPublisherTests
         await _publisher.NotifyStrike(strikeType, 1);
 
         // Assert
-        await _messageBus.Received(1).Publish(Arg.Is<NotificationMessage>(
-            m => m.EventType == expectedEventType), Arg.Any<CancellationToken>());
+        NotificationMessage m = _notificationQueue.ShouldHaveSingle();
+        m.EventType.ShouldBe(expectedEventType);
     }
 
     [Fact]
@@ -137,7 +137,7 @@ public class NotificationPublisherTests
         // Act & Assert
         await _publisher.NotifyStrike(StrikeType.DeadTorrent, 3);
 
-        await _messageBus.DidNotReceive().Publish(Arg.Any<NotificationMessage>(), Arg.Any<CancellationToken>());
+        _notificationQueue.Reader.Count.ShouldBe(0);
     }
 
     [Fact]
@@ -150,8 +150,8 @@ public class NotificationPublisherTests
         await _publisher.NotifyStrike(StrikeType.FailedImport, 1);
 
         // Assert
-        await _messageBus.Received(1).Publish(Arg.Is<NotificationMessage>(
-            m => m.Context.Data["Url"] == "http://sonarr.local/"), Arg.Any<CancellationToken>());
+        NotificationMessage m = _notificationQueue.ShouldHaveSingle();
+        m.Context.Data["Url"].ShouldBe("http://sonarr.local/");
     }
 
     #endregion
@@ -168,11 +168,11 @@ public class NotificationPublisherTests
         await _publisher.NotifyQueueItemDeleted(true, DeleteReason.Stalled);
 
         // Assert
-        await _messageBus.Received(1).Publish(Arg.Is<NotificationMessage>(
-            m => m.EventType == NotificationEventType.QueueItemDeleted &&
-                 m.Context.Data["Reason"] == "Stalled" &&
-                 m.Context.Data["Removed from client?"] == "True" &&
-                 m.Context.Severity == EventSeverity.Important), Arg.Any<CancellationToken>());
+        NotificationMessage m = _notificationQueue.ShouldHaveSingle();
+        m.EventType.ShouldBe(NotificationEventType.QueueItemDeleted);
+        m.Context.Data["Reason"].ShouldBe("Stalled");
+        m.Context.Data["Removed from client?"].ShouldBe("True");
+        m.Context.Severity.ShouldBe(EventSeverity.Important);
     }
 
     [Fact]
@@ -185,9 +185,9 @@ public class NotificationPublisherTests
         await _publisher.NotifyQueueItemDeleted(false, DeleteReason.AllFilesBlocked);
 
         // Assert
-        await _messageBus.Received(1).Publish(Arg.Is<NotificationMessage>(
-            m => m.Context.Data["Removed from client?"] == "False" &&
-                 m.Context.Data["Reason"] == "AllFilesBlocked"), Arg.Any<CancellationToken>());
+        NotificationMessage m = _notificationQueue.ShouldHaveSingle();
+        m.Context.Data["Removed from client?"].ShouldBe("False");
+        m.Context.Data["Reason"].ShouldBe("AllFilesBlocked");
     }
 
     #endregion
@@ -204,12 +204,12 @@ public class NotificationPublisherTests
         await _publisher.NotifyDownloadCleaned(2.5, TimeSpan.FromHours(48), "movies", CleanReason.MaxRatioReached);
 
         // Assert
-        await _messageBus.Received(1).Publish(Arg.Is<NotificationMessage>(
-            m => m.EventType == NotificationEventType.DownloadCleaned &&
-                 m.Context.Description == "Test Download" &&
-                 m.Context.Data["Category"] == "movies" &&
-                 m.Context.Data["Ratio"] == "2.5" &&
-                 m.Context.Data["Seeding hours"] == "48"), Arg.Any<CancellationToken>());
+        NotificationMessage m = _notificationQueue.ShouldHaveSingle();
+        m.EventType.ShouldBe(NotificationEventType.DownloadCleaned);
+        m.Context.Description.ShouldBe("Test Download");
+        m.Context.Data["Category"].ShouldBe("movies");
+        m.Context.Data["Ratio"].ShouldBe("2.5");
+        m.Context.Data["Seeding hours"].ShouldBe("48");
     }
 
     [Fact]
@@ -217,15 +217,11 @@ public class NotificationPublisherTests
     {
         // Arrange
         SetupDownloadCleanerContext();
-        NotificationMessage? captured = null;
-        _messageBus.Publish(Arg.Do<NotificationMessage>(m => captured = m), Arg.Any<CancellationToken>())
-            .Returns(Task.CompletedTask);
-
         // Act
         await _publisher.NotifyDownloadCleaned(1.0, TimeSpan.FromHours(24.7), "tv", CleanReason.MaxSeedTimeReached);
 
         // Assert
-        captured.ShouldNotBeNull();
+        NotificationMessage captured = _notificationQueue.ShouldHaveSingle();
         captured.Context.Data["Seeding hours"].ShouldBe("25");
     }
 
@@ -240,9 +236,9 @@ public class NotificationPublisherTests
         await _publisher.NotifyDownloadCleaned(2.5, TimeSpan.FromHours(48), "movies", CleanReason.MaxRatioReached);
 
         // Assert
-        await _messageBus.Received(1).Publish(Arg.Is<NotificationMessage>(
-            m => m.Context.Data.ContainsKey("Url") &&
-                 m.Context.Data["Url"] == "https://qbit.external.com/"), Arg.Any<CancellationToken>());
+        NotificationMessage m = _notificationQueue.ShouldHaveSingle();
+        m.Context.Data.ShouldContainKey("Url");
+        m.Context.Data["Url"].ShouldBe("https://qbit.external.com/");
     }
 
     #endregion
@@ -259,12 +255,12 @@ public class NotificationPublisherTests
         await _publisher.NotifyDownloadStopped(2.5, TimeSpan.FromHours(48), "movies", CleanReason.MaxRatioReached);
 
         // Assert
-        await _messageBus.Received(1).Publish(Arg.Is<NotificationMessage>(
-            m => m.EventType == NotificationEventType.DownloadStopped &&
-                 m.Context.Description == "Test Download is no longer seeding. It stays in the download client and its files stay on disk." &&
-                 m.Context.Data["Category"] == "movies" &&
-                 m.Context.Data["Ratio"] == "2.5" &&
-                 m.Context.Data["Seeding hours"] == "48"), Arg.Any<CancellationToken>());
+        NotificationMessage m = _notificationQueue.ShouldHaveSingle();
+        m.EventType.ShouldBe(NotificationEventType.DownloadStopped);
+        m.Context.Description.ShouldBe("Test Download is no longer seeding. It stays in the download client and its files stay on disk.");
+        m.Context.Data["Category"].ShouldBe("movies");
+        m.Context.Data["Ratio"].ShouldBe("2.5");
+        m.Context.Data["Seeding hours"].ShouldBe("48");
     }
 
     [Fact]
@@ -272,15 +268,11 @@ public class NotificationPublisherTests
     {
         // Arrange
         SetupDownloadCleanerContext();
-        NotificationMessage? captured = null;
-        _messageBus.Publish(Arg.Do<NotificationMessage>(m => captured = m), Arg.Any<CancellationToken>())
-            .Returns(Task.CompletedTask);
-
         // Act
         await _publisher.NotifyDownloadStopped(1.0, TimeSpan.FromHours(24.7), "tv", CleanReason.MaxSeedTimeReached);
 
         // Assert
-        captured.ShouldNotBeNull();
+        NotificationMessage captured = _notificationQueue.ShouldHaveSingle();
         captured.Context.Data["Seeding hours"].ShouldBe("25"); // Rounds to 25
     }
 
@@ -295,9 +287,9 @@ public class NotificationPublisherTests
         await _publisher.NotifyDownloadStopped(2.5, TimeSpan.FromHours(48), "movies", CleanReason.MaxRatioReached);
 
         // Assert
-        await _messageBus.Received(1).Publish(Arg.Is<NotificationMessage>(
-            m => m.Context.Data.ContainsKey("Url") &&
-                 m.Context.Data["Url"] == "https://qbit.external.com/"), Arg.Any<CancellationToken>());
+        NotificationMessage m = _notificationQueue.ShouldHaveSingle();
+        m.Context.Data.ShouldContainKey("Url");
+        m.Context.Data["Url"].ShouldBe("https://qbit.external.com/");
     }
 
     #endregion
@@ -314,11 +306,11 @@ public class NotificationPublisherTests
         await _publisher.NotifyCategoryChanged("tv-sonarr", "seeding", false);
 
         // Assert
-        await _messageBus.Received(1).Publish(Arg.Is<NotificationMessage>(
-            m => m.EventType == NotificationEventType.CategoryChanged &&
-                 m.Context.Title == "Category changed" &&
-                 m.Context.Data["Old category"] == "tv-sonarr" &&
-                 m.Context.Data["New category"] == "seeding"), Arg.Any<CancellationToken>());
+        NotificationMessage m = _notificationQueue.ShouldHaveSingle();
+        m.EventType.ShouldBe(NotificationEventType.CategoryChanged);
+        m.Context.Title.ShouldBe("Category changed");
+        m.Context.Data["Old category"].ShouldBe("tv-sonarr");
+        m.Context.Data["New category"].ShouldBe("seeding");
     }
 
     [Fact]
@@ -326,20 +318,16 @@ public class NotificationPublisherTests
     {
         // Arrange
         SetupDownloadCleanerContext();
-        NotificationMessage? captured = null;
-        _messageBus.Publish(Arg.Do<NotificationMessage>(m => captured = m), Arg.Any<CancellationToken>())
-            .Returns(Task.CompletedTask);
-
         // Act
         await _publisher.NotifyCategoryChanged("", "seeded", true);
 
         // Assert
-        captured.ShouldNotBeNull();
+        NotificationMessage captured = _notificationQueue.ShouldHaveSingle();
         captured.Context.Title.ShouldBe("Tag added");
-        captured.Context.Data.ContainsKey("Tag").ShouldBeTrue();
+        captured.Context.Data.ShouldContainKey("Tag");
         captured.Context.Data["Tag"].ShouldBe("seeded");
-        captured.Context.Data.ContainsKey("Old category").ShouldBeFalse();
-        captured.Context.Data.ContainsKey("New category").ShouldBeFalse();
+        captured.Context.Data.ShouldNotContainKey("Old category");
+        captured.Context.Data.ShouldNotContainKey("New category");
     }
 
     [Fact]
@@ -352,8 +340,8 @@ public class NotificationPublisherTests
         await _publisher.NotifyCategoryChanged("old", "new", false);
 
         // Assert
-        await _messageBus.Received(1).Publish(Arg.Is<NotificationMessage>(
-            m => m.Context.Severity == EventSeverity.Information), Arg.Any<CancellationToken>());
+        NotificationMessage m = _notificationQueue.ShouldHaveSingle();
+        m.Context.Severity.ShouldBe(EventSeverity.Information);
     }
 
     #endregion
@@ -387,7 +375,7 @@ public class NotificationPublisherTests
         await _publisher.NotifyStrike(StrikeType.FailedImport, 1);
 
         // Assert
-        await _messageBus.DidNotReceive().Publish(Arg.Any<NotificationMessage>(), Arg.Any<CancellationToken>());
+        _notificationQueue.Reader.Count.ShouldBe(0);
     }
 
     #endregion
@@ -489,15 +477,15 @@ public class NotificationPublisherTests
         await _publisher.NotifySearchItemGrabbed("Movie A", grabbedItems, InstanceType.Radarr, "http://radarr.local:7878");
 
         // Assert
-        await _messageBus.Received(1).Publish(Arg.Is<NotificationMessage>(
-            m => m.EventType == NotificationEventType.SearchItemGrabbed &&
-                 m.Context.Title == "Download grabbed" &&
-                 m.Context.Description == "Movie A" &&
-                 m.Context.Severity == EventSeverity.Information &&
-                 m.Context.Data["Item"] == "Movie A" &&
-                 m.Context.Data["Grabbed"] == "Movie.A.2024.1080p, Movie.A.2024.720p" &&
-                 m.Context.Data["Instance type"] == "Radarr" &&
-                 m.Context.Data["Url"] == "http://radarr.local:7878"), Arg.Any<CancellationToken>());
+        NotificationMessage m = _notificationQueue.ShouldHaveSingle();
+        m.EventType.ShouldBe(NotificationEventType.SearchItemGrabbed);
+        m.Context.Title.ShouldBe("Download grabbed");
+        m.Context.Description.ShouldBe("Movie A");
+        m.Context.Severity.ShouldBe(EventSeverity.Information);
+        m.Context.Data["Item"].ShouldBe("Movie A");
+        m.Context.Data["Grabbed"].ShouldBe("Movie.A.2024.1080p, Movie.A.2024.720p");
+        m.Context.Data["Instance type"].ShouldBe("Radarr");
+        m.Context.Data["Url"].ShouldBe("http://radarr.local:7878");
     }
 
     [Fact]
@@ -528,14 +516,14 @@ public class NotificationPublisherTests
         await _publisher.NotifyForceImported();
 
         // Assert
-        await _messageBus.Received(1).Publish(Arg.Is<NotificationMessage>(
-            m => m.EventType == NotificationEventType.ForceImported &&
-                 m.Context.Title == "Imported a download the arr had blocked" &&
-                 m.Context.Description == "Test Show" &&
-                 m.Context.Severity == EventSeverity.Important &&
-                 m.Context.Data["Hash"] == "abcd1234" &&
-                 m.Context.Data["Instance type"] == "Sonarr" &&
-                 m.Context.Data["Url"] == "http://sonarr.local/"), Arg.Any<CancellationToken>());
+        NotificationMessage m = _notificationQueue.ShouldHaveSingle();
+        m.EventType.ShouldBe(NotificationEventType.ForceImported);
+        m.Context.Title.ShouldBe("Imported a download the arr had blocked");
+        m.Context.Description.ShouldBe("Test Show");
+        m.Context.Severity.ShouldBe(EventSeverity.Important);
+        m.Context.Data["Hash"].ShouldBe("abcd1234");
+        m.Context.Data["Instance type"].ShouldBe("Sonarr");
+        m.Context.Data["Url"].ShouldBe("http://sonarr.local/");
     }
 
     [Fact]
@@ -577,8 +565,9 @@ public class NotificationPublisherTests
         await _publisher.NotifyQueueItemDeleted(true, DeleteReason.Stalled);
 
         // Assert
-        await _messageBus.Received(1).Publish(Arg.Is<NotificationMessage>(
-            m => m.Context.Description == "Frankenstein" && m.Context.Data["Hash"] == "bookhash1"), Arg.Any<CancellationToken>());
+        NotificationMessage m = _notificationQueue.ShouldHaveSingle();
+        m.Context.Description.ShouldBe("Frankenstein");
+        m.Context.Data["Hash"].ShouldBe("bookhash1");
     }
 
     [Fact]
@@ -592,8 +581,9 @@ public class NotificationPublisherTests
         await _publisher.NotifyStrike(StrikeType.Stalled, 1);
 
         // Assert
-        await _messageBus.Received(1).Publish(Arg.Is<NotificationMessage>(
-            m => m.Context.Description == "Frankenstein" && m.Context.Data["Hash"] == "bookhash1"), Arg.Any<CancellationToken>());
+        NotificationMessage m = _notificationQueue.ShouldHaveSingle();
+        m.Context.Description.ShouldBe("Frankenstein");
+        m.Context.Data["Hash"].ShouldBe("bookhash1");
     }
 
     #endregion

code/backend/Cleanuparr.Infrastructure.Tests/TestHelpers/ChannelAssertions.cs

@@ -0,0 +1,20 @@
+using System.Threading.Channels;
+using Shouldly;
+
+namespace Cleanuparr.Infrastructure.Tests.TestHelpers;
+
+/// <summary>
+/// Assertions for channels used as in-process queues in tests.
+/// </summary>
+internal static class ChannelAssertions
+{
+    /// <summary>
+    /// Asserts the channel holds exactly one message and returns it.
+    /// </summary>
+    internal static T ShouldHaveSingle<T>(this Channel<T> channel)
+    {
+        channel.Reader.Count.ShouldBe(1);
+        channel.Reader.TryRead(out T? item).ShouldBeTrue();
+        return item!;
+    }
+}

code/backend/Cleanuparr.Infrastructure/Cleanuparr.Infrastructure.csproj

@@ -13,7 +13,6 @@
       <PackageReference Include="FLM.QBittorrent" Version="1.0.3" />
       <PackageReference Include="FLM.Transmission" Version="1.0.3" />
       <PackageReference Include="Newtonsoft.Json" Version="13.0.3" />
-      <PackageReference Include="MassTransit.Abstractions" Version="8.5.7" />
       <PackageReference Include="Microsoft.IdentityModel.Protocols.OpenIdConnect" Version="8.7.0" />
       <PackageReference Include="Microsoft.IdentityModel.Tokens" Version="8.7.0" />
       <PackageReference Include="System.IdentityModel.Tokens.Jwt" Version="8.7.0" />

code/backend/Cleanuparr.Infrastructure/Features/DownloadRemover/Consumers/DownloadRemoverConsumer.cs

@@ -2,12 +2,15 @@ using Cleanuparr.Infrastructure.Features.Context;
 using Cleanuparr.Infrastructure.Features.DownloadRemover.Interfaces;
 using Cleanuparr.Infrastructure.Features.DownloadRemover.Models;
 using Cleanuparr.Infrastructure.Features.DryRun;
-using MassTransit;
+using Cleanuparr.Infrastructure.Features.Messaging;
 using Microsoft.Extensions.Logging;
 
 namespace Cleanuparr.Infrastructure.Features.DownloadRemover.Consumers;
 
-public sealed class DownloadRemoverConsumer : IConsumer<QueueItemRemoveRequest>
+/// <summary>
+/// Removes one queued download removal request.
+/// </summary>
+public sealed class DownloadRemoverConsumer : IMessageHandler<QueueItemRemoveRequest>
 {
     private readonly ILogger<DownloadRemoverConsumer> _logger;
     private readonly IQueueItemRemover _queueItemRemover;
@@ -24,15 +27,16 @@ public sealed class DownloadRemoverConsumer : IConsumer<QueueItemRemoveRequest>
         _dryRunActivity = dryRunActivity;
     }
 
-    public async Task Consume(ConsumeContext<QueueItemRemoveRequest> context)
+    /// <inheritdoc />
+    public async Task HandleAsync(QueueItemRemoveRequest message)
     {
         bool trackedDryRun = false;
 
         try
         {
-            ContextProvider.SetDryRun(context.Message.IsDryRun);
+            ContextProvider.SetDryRun(message.IsDryRun);
 
-            if (context.Message.IsDryRun)
+            if (message.IsDryRun)
             {
                 await _dryRunActivity.EnterAsync();
                 trackedDryRun = true;
@@ -40,7 +44,7 @@ public sealed class DownloadRemoverConsumer : IConsumer<QueueItemRemoveRequest>
 
             try
             {
-                await _queueItemRemover.RemoveQueueItemAsync(context.Message);
+                await _queueItemRemover.RemoveQueueItemAsync(message);
             }
             finally
             {
@@ -53,9 +57,9 @@ public sealed class DownloadRemoverConsumer : IConsumer<QueueItemRemoveRequest>
         catch (Exception exception)
         {
             _logger.LogError(exception,
-                "failed to remove queue item | {title} | {url}",
-                context.Message.Target.Title,
-                context.Message.Instance.Url
+                "failed to remove queue item | {Title} | {Url}",
+                message.Target.Title,
+                message.Instance.Url
             );
         }
     }

code/backend/Cleanuparr.Infrastructure/Features/DownloadRemover/Models/ArrRemovalTarget.cs

@@ -1,4 +1,3 @@
-using System.Text.Json.Serialization;
 using Cleanuparr.Domain.Entities.Arr;
 using Cleanuparr.Domain.Entities.Arr.Queue;
 
@@ -21,9 +20,7 @@ public sealed record ArrRemovalTarget : RemovalTarget
     /// </summary>
     public bool ChangeCategory { get; init; }
 
-    [JsonIgnore]
     public override string DownloadId => Record.DownloadId;
 
-    [JsonIgnore]
     public override string Title => Record.Title;
 }

code/backend/Cleanuparr.Infrastructure/Features/DownloadRemover/Models/LazyLibrarianRemovalTarget.cs

@@ -1,4 +1,3 @@
-using System.Text.Json.Serialization;
 using Cleanuparr.Domain.Entities.LazyLibrarian;
 
 namespace Cleanuparr.Infrastructure.Features.DownloadRemover.Models;
@@ -13,9 +12,7 @@ public sealed record LazyLibrarianRemovalTarget : RemovalTarget
     /// </summary>
     public required bool RemovedFromClient { get; init; }
 
-    [JsonIgnore]
     public override string DownloadId => Item.DownloadId;
 
-    [JsonIgnore]
     public override string Title => Item.Title;
 }

code/backend/Cleanuparr.Infrastructure/Features/DownloadRemover/Models/RemovalTarget.cs

@@ -1,16 +1,8 @@
-using System.Text.Json.Serialization;
-
 namespace Cleanuparr.Infrastructure.Features.DownloadRemover.Models;
 
-[JsonPolymorphic(TypeDiscriminatorPropertyName = "$target")]
-[JsonDerivedType(typeof(ArrRemovalTarget), "arr")]
-[JsonDerivedType(typeof(LazyLibrarianRemovalTarget), "lazyLibrarian")]
 public abstract record RemovalTarget
 {
-    // JsonIgnore does not reach an override. Annotate each one.
-    [JsonIgnore]
     public abstract string DownloadId { get; }
 
-    [JsonIgnore]
     public abstract string Title { get; }
 }

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

@@ -6,6 +6,7 @@ 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.Features.DownloadCleaner.Services;
 using Cleanuparr.Infrastructure.Helpers;
@@ -14,7 +15,7 @@ using Cleanuparr.Persistence;
 using Cleanuparr.Persistence.Models.Configuration.Arr;
 using Cleanuparr.Persistence.Models.Configuration.DownloadCleaner;
 using Cleanuparr.Persistence.Models.Configuration.General;
-using MassTransit;
+using System.Threading.Channels;
 using Microsoft.Extensions.Caching.Memory;
 using Microsoft.Extensions.Logging;
 using LogContext = Serilog.Context.LogContext;
@@ -35,7 +36,7 @@ public sealed class DownloadCleaner : GenericHandler
         ILogger<DownloadCleaner> logger,
         DataContext dataContext,
         IMemoryCache cache,
-        IBus messageBus,
+        ChannelWriter<QueueItemRemoveRequest> removalQueue,
         IArrClientFactory arrClientFactory,
         IArrQueueReader arrQueueReader,
         IDownloadServiceFactory downloadServiceFactory,
@@ -49,7 +50,7 @@ public sealed class DownloadCleaner : GenericHandler
         IForceImportService forceImportService,
         ILazyLibrarianService lazyLibrarianService
     ) : base(
-        logger, dataContext, cache, messageBus,
+        logger, dataContext, cache, removalQueue,
         arrClientFactory, arrQueueReader, downloadServiceFactory, eventPublisher, dryRunInterceptor,
         forceImportService
     )

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

@@ -18,7 +18,7 @@ using Cleanuparr.Persistence.Models.Configuration.DownloadCleaner;
 using Cleanuparr.Persistence.Models.Configuration.General;
 using Cleanuparr.Persistence.Models.Configuration.MalwareBlocker;
 using Cleanuparr.Persistence.Models.Configuration.QueueCleaner;
-using MassTransit;
+using System.Threading.Channels;
 using Microsoft.EntityFrameworkCore;
 using Microsoft.Extensions.Caching.Memory;
 using Microsoft.Extensions.Logging;
@@ -30,7 +30,7 @@ public abstract class GenericHandler : IHandler
     protected readonly ILogger<GenericHandler> _logger;
     protected readonly DataContext _dataContext;
     protected readonly IMemoryCache _cache;
-    protected readonly IBus _messageBus;
+    protected readonly ChannelWriter<QueueItemRemoveRequest> _removalQueue;
     protected readonly IArrClientFactory _arrClientFactory;
     protected readonly IArrQueueReader _arrQueueReader;
     protected readonly IDownloadServiceFactory _downloadServiceFactory;
@@ -42,7 +42,7 @@ public abstract class GenericHandler : IHandler
         ILogger<GenericHandler> logger,
         DataContext dataContext,
         IMemoryCache cache,
-        IBus messageBus,
+        ChannelWriter<QueueItemRemoveRequest> removalQueue,
         IArrClientFactory arrClientFactory,
         IArrQueueReader arrQueueReader,
         IDownloadServiceFactory downloadServiceFactory,
@@ -54,7 +54,7 @@ public abstract class GenericHandler : IHandler
         _forceImportService = forceImportService;
         _logger = logger;
         _cache = cache;
-        _messageBus = messageBus;
+        _removalQueue = removalQueue;
         _arrClientFactory = arrClientFactory;
         _arrQueueReader = arrQueueReader;
         _downloadServiceFactory = downloadServiceFactory;
@@ -195,15 +195,7 @@ public abstract class GenericHandler : IHandler
         string downloadRemovalKey = CacheKeys.DownloadMarkedForRemoval(target.DownloadId, instance.Url);
         _cache.Set(downloadRemovalKey, true);
 
-        try
-        {
-            await _messageBus.Publish(removeRequest);
-        }
-        catch
-        {
-            _cache.Remove(downloadRemovalKey);
-            throw;
-        }
+        await _removalQueue.WriteAsync(removeRequest);
 
         // The mark above lives only while the removal is in flight, and its absence must not read as an import.
         _forceImportService.Forget(instance, target.DownloadId);

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

@@ -5,6 +5,7 @@ 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.Features.MalwareBlocker;
 using Cleanuparr.Infrastructure.Helpers;
@@ -14,7 +15,7 @@ using Cleanuparr.Persistence.Models.Configuration;
 using Cleanuparr.Persistence.Models.Configuration.Arr;
 using Cleanuparr.Persistence.Models.Configuration.General;
 using Cleanuparr.Persistence.Models.Configuration.MalwareBlocker;
-using MassTransit;
+using System.Threading.Channels;
 using Microsoft.Extensions.Caching.Memory;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Logging;
@@ -31,7 +32,7 @@ public sealed class MalwareBlocker : GenericHandler
         ILogger<MalwareBlocker> logger,
         DataContext dataContext,
         IMemoryCache cache,
-        IBus messageBus,
+        ChannelWriter<QueueItemRemoveRequest> removalQueue,
         IArrClientFactory arrClientFactory,
         IArrQueueReader arrQueueReader,
         IDownloadServiceFactory downloadServiceFactory,
@@ -41,7 +42,7 @@ public sealed class MalwareBlocker : GenericHandler
         IForceImportService forceImportService,
         [FromKeyedServices(ILazyLibrarianEvaluator.MalwareBlockerKey)] ILazyLibrarianEvaluator lazyLibrarianService
     ) : base(
-        logger, dataContext, cache, messageBus,
+        logger, dataContext, cache, removalQueue,
         arrClientFactory, arrQueueReader, downloadServiceFactory, eventPublisher, dryRunInterceptor,
         forceImportService
     )

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

@@ -5,6 +5,7 @@ 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;
@@ -14,7 +15,7 @@ using Cleanuparr.Persistence.Models.Configuration;
 using Cleanuparr.Persistence.Models.Configuration.Arr;
 using Cleanuparr.Persistence.Models.Configuration.General;
 using Cleanuparr.Persistence.Models.Configuration.QueueCleaner;
-using MassTransit;
+using System.Threading.Channels;
 using Microsoft.EntityFrameworkCore;
 using Microsoft.Extensions.Caching.Memory;
 using Microsoft.Extensions.DependencyInjection;
@@ -32,7 +33,7 @@ public sealed class QueueCleaner : GenericHandler
         ILogger<QueueCleaner> logger,
         DataContext dataContext,
         IMemoryCache cache,
-        IBus messageBus,
+        ChannelWriter<QueueItemRemoveRequest> removalQueue,
         IArrClientFactory arrClientFactory,
         IArrQueueReader arrQueueReader,
         IDownloadServiceFactory downloadServiceFactory,
@@ -42,7 +43,7 @@ public sealed class QueueCleaner : GenericHandler
         IForceImportService forceImportService,
         [FromKeyedServices(ILazyLibrarianEvaluator.QueueCleanerKey)] ILazyLibrarianEvaluator lazyLibrarianService
     ) : base(
-        logger, dataContext, cache, messageBus,
+        logger, dataContext, cache, removalQueue,
         arrClientFactory, arrQueueReader, downloadServiceFactory, eventPublisher, dryRunInterceptor,
         forceImportService
     )

code/backend/Cleanuparr.Infrastructure/Features/Messaging/IMessageHandler.cs

@@ -0,0 +1,14 @@
+namespace Cleanuparr.Infrastructure.Features.Messaging;
+
+/// <summary>
+/// Handles one message taken from a <see cref="QueueWorker{TMessage}"/> queue.
+/// </summary>
+/// <typeparam name="TMessage">The queued message type.</typeparam>
+public interface IMessageHandler<in TMessage>
+{
+    /// <summary>
+    /// Handles a single queued message.
+    /// </summary>
+    /// <param name="message">The message to handle.</param>
+    Task HandleAsync(TMessage message);
+}

code/backend/Cleanuparr.Infrastructure/Features/Messaging/QueueWorker.cs

@@ -0,0 +1,57 @@
+using System.Threading.Channels;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+
+namespace Cleanuparr.Infrastructure.Features.Messaging;
+
+/// <summary>
+/// Drains a channel one message at a time, each in its own DI scope.
+/// </summary>
+/// <typeparam name="TMessage">The queued message type.</typeparam>
+public sealed class QueueWorker<TMessage> : BackgroundService
+{
+    private readonly ILogger<QueueWorker<TMessage>> _logger;
+    private readonly IServiceScopeFactory _scopeFactory;
+    private readonly Channel<TMessage> _channel;
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="QueueWorker{TMessage}"/> class.
+    /// </summary>
+    /// <param name="logger">The logger.</param>
+    /// <param name="scopeFactory">Creates a DI scope per message.</param>
+    /// <param name="channel">The channel to drain.</param>
+    public QueueWorker(
+        ILogger<QueueWorker<TMessage>> logger,
+        IServiceScopeFactory scopeFactory,
+        Channel<TMessage> channel)
+    {
+        _logger = logger;
+        _scopeFactory = scopeFactory;
+        _channel = channel;
+    }
+
+    /// <inheritdoc />
+    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+    {
+        await foreach (TMessage message in _channel.Reader.ReadAllAsync(stoppingToken))
+        {
+            await HandleAsync(message);
+        }
+    }
+
+    // A separate async method keeps each message's ContextProvider values out of the next one.
+    private async Task HandleAsync(TMessage message)
+    {
+        try
+        {
+            await using AsyncServiceScope scope = _scopeFactory.CreateAsyncScope();
+            IMessageHandler<TMessage> handler = scope.ServiceProvider.GetRequiredService<IMessageHandler<TMessage>>();
+            await handler.HandleAsync(message);
+        }
+        catch (Exception exception)
+        {
+            _logger.LogError(exception, "failed to handle queued message | {MessageType}", typeof(TMessage).Name);
+        }
+    }
+}

code/backend/Cleanuparr.Infrastructure/Features/Notifications/Consumers/NotificationConsumer.cs

@@ -1,9 +1,12 @@
+using Cleanuparr.Infrastructure.Features.Messaging;
 using Cleanuparr.Infrastructure.Features.Notifications.Models;
-using MassTransit;
 
 namespace Cleanuparr.Infrastructure.Features.Notifications.Consumers;
 
-public sealed class NotificationConsumer : IConsumer<NotificationMessage>
+/// <summary>
+/// Sends one queued notification to its configured providers.
+/// </summary>
+public sealed class NotificationConsumer : IMessageHandler<NotificationMessage>
 {
     private readonly NotificationService _notificationService;
 
@@ -12,6 +15,7 @@ public sealed class NotificationConsumer : IConsumer<NotificationMessage>
         _notificationService = notificationService;
     }
 
-    public Task Consume(ConsumeContext<NotificationMessage> context) =>
-        _notificationService.SendNotificationAsync(context.Message.EventType, context.Message.Context);
+    /// <inheritdoc />
+    public Task HandleAsync(NotificationMessage message) =>
+        _notificationService.SendNotificationAsync(message.EventType, message.Context);
 }

code/backend/Cleanuparr.Infrastructure/Features/Notifications/NotificationPublisher.cs

@@ -2,10 +2,10 @@
 using Cleanuparr.Domain.Entities.Arr.Queue;
 using Cleanuparr.Domain.Enums;
 using Cleanuparr.Infrastructure.Features.Context;
+using System.Threading.Channels;
 using Cleanuparr.Infrastructure.Features.Notifications.Models;
 using Cleanuparr.Infrastructure.Interceptors;
 using Cleanuparr.Persistence.Models.Configuration.QueueCleaner;
-using MassTransit;
 using Microsoft.Extensions.Logging;
 
 namespace Cleanuparr.Infrastructure.Features.Notifications;
@@ -14,16 +14,16 @@ public class NotificationPublisher : INotificationPublisher
 {
     private readonly ILogger<NotificationPublisher> _logger;
     private readonly IDryRunInterceptor _dryRunInterceptor;
-    private readonly IBus _messageBus;
+    private readonly ChannelWriter<NotificationMessage> _notificationQueue;
 
     public NotificationPublisher(
         ILogger<NotificationPublisher> logger,
         IDryRunInterceptor dryRunInterceptor,
-        IBus messageBus)
+        ChannelWriter<NotificationMessage> notificationQueue)
     {
         _logger = logger;
         _dryRunInterceptor = dryRunInterceptor;
-        _messageBus = messageBus;
+        _notificationQueue = notificationQueue;
     }
 
     public virtual async Task NotifyStrike(StrikeType strikeType, int strikeCount)
@@ -147,7 +147,7 @@ public class NotificationPublisher : INotificationPublisher
     private async Task SendNotificationInternalAsync((NotificationEventType eventType, NotificationContext context) parameters)
     {
         var (eventType, context) = parameters;
-        await _messageBus.Publish(new NotificationMessage(eventType, context));
+        await _notificationQueue.WriteAsync(new NotificationMessage(eventType, context));
     }
 
     /// <summary>

code/backend/Cleanuparr.Infrastructure/Logging/LoggingConfigManager.cs

@@ -112,7 +112,6 @@ public static class LoggingConfigManager
 
         // Apply standard overrides
         logConfig
-            .MinimumLevel.Override("MassTransit", LogEventLevel.Warning)
             .MinimumLevel.Override("Microsoft.Hosting.Lifetime", LogEventLevel.Information)
             .MinimumLevel.Override("Microsoft", LogEventLevel.Warning)
             .MinimumLevel.Override("Microsoft.AspNetCore.DataProtection", LogEventLevel.Error)

e2e/tests/download-cleaner/deluge-seeding-rule-cleanup.spec.ts

@@ -65,10 +65,10 @@ async function stillPresent(infoHashes: string[]): Promise<string[]> {
  * `DownloadCleaned` event happens. Two torrents and one cleanup cycle show this
  * behavior.
  *
- * A DownloadCleaned removal also publishes a notification to the message bus
- * (NotificationPublisher -> MassTransit -> NotificationConsumer), so the same
- * cycle doubles as e2e coverage for that async dispatch: a provider subscribed
- * to OnDownloadCleaned must receive a real delivery when the rule fires.
+ * A DownloadCleaned removal also queues a notification on the in-process
+ * notification queue, so the same cycle doubles as e2e coverage for that
+ * async dispatch: a provider subscribed to OnDownloadCleaned must receive a
+ * real delivery when the rule fires.
  */
 test.describe.serial('Deluge seeding rule cleanup', () => {
   let token: string;

e2e/tests/helpers/api/malware-blocker.ts

@@ -10,4 +10,29 @@ export class MalwareBlockerApi {
   updateConfig(body: Record<string, unknown>): Promise<Response> {
     return this.client.put('/api/configuration/malware_blocker', body);
   }
+
+  async getJsonConfig(): Promise<Record<string, unknown>> {
+    const res = await this.getConfig();
+    if (!res.ok) {
+      throw new Error(`GET malware blocker config failed: ${res.status} ${await res.text()}`);
+    }
+    return res.json();
+  }
+
+  /** Merges overrides onto the current config, nesting sonarr overrides onto the current sonarr block. */
+  async patch(overrides: Record<string, unknown>): Promise<void> {
+    const current = await this.getJsonConfig();
+    const merged = {
+      ...current,
+      ...overrides,
+      sonarr: {
+        ...(current.sonarr as Record<string, unknown>),
+        ...(overrides.sonarr as Record<string, unknown>),
+      },
+    };
+    const res = await this.updateConfig(merged);
+    if (!res.ok) {
+      throw new Error(`PUT malware blocker config failed: ${res.status} ${await res.text()}`);
+    }
+  }
 }

e2e/tests/helpers/api/queue-cleaner.ts

@@ -13,6 +13,24 @@ export class QueueCleanerApi {
     return this.client.put('/api/configuration/queue_cleaner', body);
   }
 
+  async getJsonConfig(): Promise<Record<string, unknown>> {
+    const res = await this.getConfig();
+    if (!res.ok) {
+      throw new Error(`GET queue cleaner config failed: ${res.status} ${await res.text()}`);
+    }
+    return res.json();
+  }
+
+  /** Merges overrides onto the current config. */
+  async patch(overrides: Record<string, unknown>): Promise<void> {
+    const current = await this.getJsonConfig();
+    const merged = { ...current, ...overrides };
+    const res = await this.updateConfig(merged);
+    if (!res.ok) {
+      throw new Error(`PUT queue cleaner config failed: ${res.status} ${await res.text()}`);
+    }
+  }
+
   listRules(kind: QueueRuleKind): Promise<Response> {
     return this.client.get(`/api/queue-rules/${kind}`);
   }

e2e/tests/helpers/mocks/arr-stubs.ts

@@ -107,6 +107,34 @@ export function arrQueueStub(records: QueueRecord[]): Mapping {
   };
 }
 
+/** Answers every unstubbed arr request with an empty 200, lowest priority. */
+export function arrCatchAllMapping(): Mapping {
+  return {
+    request: { method: 'ANY', urlPathPattern: '/api/v[0-9]+/.*' },
+    response: { status: 200, jsonBody: {} },
+    priority: 100,
+  };
+}
+
+export interface DeleteCall {
+  id: number;
+  removeFromClient: boolean;
+  blocklist: boolean;
+}
+
+/** Reads the queue DELETE calls an arr stub recorded, decoded from the request's query string. */
+export async function readDeletes(arr: WireMockClient): Promise<DeleteCall[]> {
+  const reqs = await arr.findRequests({ method: 'DELETE', urlPattern: '/api/v3/queue/.*' });
+  return reqs.map((r) => {
+    const parsed = new URL(`http://stub${r.url}`);
+    return {
+      id: Number(parsed.pathname.split('/').pop()),
+      removeFromClient: parsed.searchParams.get('removeFromClient') === 'true',
+      blocklist: parsed.searchParams.get('blocklist') === 'true',
+    };
+  });
+}
+
 export function arrCommandTriggerStub(commandId = 1): Mapping {
   return {
     request: { method: 'POST', urlPath: '/api/v3/command' },

e2e/tests/malware-blocker/malware-blocker.spec.ts

@@ -7,13 +7,13 @@ import {
   deleteDownloadClient,
   createArrInstance,
   deleteArrInstance,
-  getMalwareBlockerConfig,
-  updateMalwareBlockerConfig,
   triggerJob,
 } from '../helpers/app-api';
+import { CleanuparrApi } from '../helpers/api';
 import { ALL_CLIENTS, TorrentClientFixture } from '../helpers/torrent-clients';
 import { buildMultiFileTorrent, resetDirectory } from '../helpers/torrent-fixtures';
 import { WireMockClient, Mapping } from '../helpers/mocks/wiremock-client';
+import { arrHealthStub, arrCatchAllMapping, readDeletes } from '../helpers/mocks/arr-stubs';
 import { TEST_CONFIG, farFutureCron } from '../helpers/test-config';
 import { mkdirShared, writeFileShared } from '../helpers/shared-volume';
 import { buildHubConnection, waitForEvent } from '../helpers/api/signalr';
@@ -100,20 +100,6 @@ interface QueueStubRecord {
   episodeId: number;
 }
 
-interface DeleteCall {
-  id: number;
-  removeFromClient: boolean;
-  blocklist: boolean;
-}
-
-function arrHealthMapping(): Mapping {
-  return {
-    request: { method: 'GET', urlPathPattern: '/api/v[0-9]+/system/status' },
-    response: { status: 200, jsonBody: { version: '4.0.0.0', appName: 'Sonarr' } },
-    priority: 1,
-  };
-}
-
 function arrQueueMapping(records: QueueStubRecord[]): Mapping {
   return {
     request: { method: 'GET', urlPath: '/api/v3/queue' },
@@ -143,17 +129,9 @@ function arrQueueMapping(records: QueueStubRecord[]): Mapping {
   };
 }
 
-function arrCatchAllMapping(): Mapping {
-  return {
-    request: { method: 'ANY', urlPathPattern: '/api/v[0-9]+/.*' },
-    response: { status: 200, jsonBody: {} },
-    priority: 100,
-  };
-}
-
 async function stubArrQueue(arr: WireMockClient, records: QueueStubRecord[]): Promise<void> {
   await arr.resetAll();
-  await arr.stubMany([arrHealthMapping(), arrQueueMapping(records), arrCatchAllMapping()]);
+  await arr.stubMany([arrHealthStub(), arrQueueMapping(records), arrCatchAllMapping()]);
 }
 
 async function queueRequestSeen(arr: WireMockClient, timeoutMs: number): Promise<boolean> {
@@ -168,34 +146,8 @@ async function queueRequestSeen(arr: WireMockClient, timeoutMs: number): Promise
   return false;
 }
 
-async function readDeletes(arr: WireMockClient): Promise<DeleteCall[]> {
-  const reqs = await arr.findRequests({ method: 'DELETE', urlPattern: '/api/v3/queue/.*' });
-  return reqs.map((r) => {
-    const parsed = new URL(`http://stub${r.url}`);
-    return {
-      id: Number(parsed.pathname.split('/').pop()),
-      removeFromClient: parsed.searchParams.get('removeFromClient') === 'true',
-      blocklist: parsed.searchParams.get('blocklist') === 'true',
-    };
-  });
-}
-
-async function setMalwareBlocker(
-  token: string,
-  overrides: Record<string, unknown>,
-): Promise<void> {
-  const currentRes = await getMalwareBlockerConfig(token);
-  expect(currentRes.status).toBe(200);
-  const current = await currentRes.json();
-  const merged = {
-    ...current,
-    ...overrides,
-    sonarr: { ...(current.sonarr ?? {}), ...(overrides.sonarr as Record<string, unknown> ?? {}) },
-  };
-  const updateRes = await updateMalwareBlockerConfig(token, merged);
-  if (updateRes.status >= 300) {
-    throw new Error(`update malware blocker config: ${updateRes.status} ${await updateRes.text()}`);
-  }
+async function setMalwareBlocker(token: string, overrides: Record<string, unknown>): Promise<void> {
+  await new CleanuparrApi({ token }).malwareBlocker.patch(overrides);
 }
 
 test.describe.serial('Malware blocker — DeleteIfAnyFileBlocked', () => {

e2e/tests/malware-blocker/removal-queue.spec.ts

@@ -0,0 +1,421 @@
+import { test, expect } from '@playwright/test';
+import { join } from 'node:path';
+import {
+  loginAndGetToken,
+  createDownloadClient,
+  listDownloadClients,
+  deleteDownloadClient,
+  createArrInstance,
+  deleteArrInstance,
+  triggerJob,
+  createNotificationProvider,
+  deleteNotificationProvider,
+} from '../helpers/app-api';
+import { CleanuparrApi } from '../helpers/api';
+import { QBittorrentDriver } from '../helpers/torrent-clients/qbittorrent';
+import { buildSingleFileTorrent, resetDirectory } from '../helpers/torrent-fixtures';
+import { WireMockClient, Mapping } from '../helpers/mocks/wiremock-client';
+import { QueueRecord, arrQueueStub, arrHealthStub, arrCatchAllMapping, readDeletes } from '../helpers/mocks/arr-stubs';
+import { gotifyMessageStub } from '../helpers/mocks/notification-stubs';
+import { TEST_CONFIG, farFutureCron } from '../helpers/test-config';
+import { mkdirShared, writeFileShared } from '../helpers/shared-volume';
+
+/**
+ * Removal queue behaviour.
+ * Each item is a single-file `*.exe` torrent on one qBittorrent client.
+ * Distinct series and episode ids catch id mix-ups between queued messages.
+ */
+
+const HOST_DOWNLOADS = join(__dirname, '..', '..', 'test-data', 'downloads');
+const CLIENT_DOWNLOADS = '/downloads';
+const APP_DOWNLOADS = '/e2e-downloads';
+const BLOCKLIST_REL_PATH = 'malware-blocker/removal-queue-blocklist.txt';
+const APP_BLOCKLIST_PATH = `${APP_DOWNLOADS}/${BLOCKLIST_REL_PATH}`;
+const SCAN_DIR_SLUG = 'qbittorrent-removal-queue';
+
+const POLL_TIMEOUT_MS = 30_000;
+const SETTLE_MS = 3_000;
+
+interface BlockedItem {
+  id: number;
+  title: string;
+  infoHash: string;
+  seriesId: number;
+  episodeId: number;
+}
+
+function arrDeleteFailsMapping(queueId: number, status: number): Mapping {
+  return {
+    request: { method: 'DELETE', urlPath: `/api/v3/queue/${queueId}` },
+    response: { status, jsonBody: { message: 'arr unavailable' } },
+    priority: 1,
+  };
+}
+
+function queueRecordsFor(items: BlockedItem[]): QueueRecord[] {
+  return items.map((item) => ({
+    id: item.id,
+    downloadId: item.infoHash,
+    title: item.title,
+    protocol: 'torrent',
+    status: 'downloading',
+    trackedDownloadStatus: 'ok',
+    trackedDownloadState: 'downloading',
+    // Sonarr sends seriesId and episodeId outside the QueueRecord helper type.
+    // The malware blocker needs both nonzero.
+    ...( { seriesId: item.seriesId, episodeId: item.episodeId, seasonNumber: 1 } as unknown as Partial<QueueRecord> ),
+  }));
+}
+
+async function waitUntilDeleted(arr: WireMockClient, ids: number[], timeoutMs = POLL_TIMEOUT_MS): Promise<void> {
+  const start = Date.now();
+  while (Date.now() - start < timeoutMs) {
+    const deletes = await readDeletes(arr);
+    const seen = new Set(deletes.map((d) => d.id));
+    if (ids.every((id) => seen.has(id))) {
+      return;
+    }
+    await new Promise((r) => setTimeout(r, 500));
+  }
+  const deletes = await readDeletes(arr);
+  throw new Error(`timed out waiting for deletes ${JSON.stringify(ids)}, saw ${JSON.stringify(deletes)}`);
+}
+
+interface AppEventItem {
+  eventType: string;
+  itemHash: string;
+  itemTitle: string;
+  isDryRun: boolean;
+}
+
+/**
+ * Finds events by type and item title.
+ * The recent-events cap of 50 drops events from earlier specs.
+ */
+async function queryEvents(token: string, eventType: string, search: string): Promise<AppEventItem[]> {
+  const url = `${TEST_CONFIG.appUrl}/api/events?eventType=${eventType}&search=${encodeURIComponent(search)}&pageSize=50`;
+  const res = await fetch(url, { headers: { Authorization: `Bearer ${token}` } });
+  if (!res.ok) {
+    throw new Error(`events lookup failed: ${res.status} ${await res.text()}`);
+  }
+  const body = await res.json();
+  return body.items ?? [];
+}
+
+async function waitForEvents(
+  token: string,
+  eventType: string,
+  search: string,
+  timeoutMs = POLL_TIMEOUT_MS,
+): Promise<AppEventItem[]> {
+  const start = Date.now();
+  while (Date.now() - start < timeoutMs) {
+    const items = await queryEvents(token, eventType, search);
+    if (items.length > 0) {
+      return items;
+    }
+    await new Promise((r) => setTimeout(r, 500));
+  }
+  return [];
+}
+
+/** Adds a stopped single-file blocked torrent to qBittorrent. */
+async function seedBlockedTorrent(
+  driver: QBittorrentDriver,
+  hostScanDir: string,
+  fileName: string,
+): Promise<{ infoHash: string; name: string }> {
+  const fx = buildSingleFileTorrent(hostScanDir, fileName);
+  await driver.addTorrent({
+    metainfo: fx.metainfo,
+    savePath: CLIENT_DOWNLOADS,
+    name: fx.name,
+    infoHash: fx.infoHash,
+  });
+  return { infoHash: fx.infoHash, name: fx.name };
+}
+
+test.describe.serial('Malware blocker — removal queue', () => {
+  let token: string;
+  let api: CleanuparrApi;
+  const arr = new WireMockClient(TEST_CONFIG.mocks.arrAdminUrl);
+  const notify = new WireMockClient(TEST_CONFIG.mocks.notifyAdminUrl);
+  const driver = new QBittorrentDriver();
+  const hostScanDir = join(HOST_DOWNLOADS, SCAN_DIR_SLUG);
+
+  let sonarrId: string | undefined;
+  let clientId: string | undefined;
+  let notificationId: string | undefined;
+
+  test.beforeAll(async () => {
+    token = await loginAndGetToken();
+    api = new CleanuparrApi({ token });
+
+    for (const client of await listDownloadClients(token)) {
+      await deleteDownloadClient(token, client.id);
+    }
+
+    mkdirShared(join(HOST_DOWNLOADS, 'malware-blocker'));
+    writeFileShared(join(HOST_DOWNLOADS, BLOCKLIST_REL_PATH), '*.exe\n');
+
+    resetDirectory(hostScanDir);
+    await driver.ready();
+    await driver.clearAllTorrents();
+
+    await arr.waitReady();
+    await notify.waitReady();
+    await arr.resetAll();
+    await notify.resetAll();
+    await notify.stub(gotifyMessageStub());
+    await arr.stubMany([arrHealthStub(), arrQueueStub([]), arrCatchAllMapping()]);
+
+    const sonarrRes = await createArrInstance(token, 'sonarr', {
+      name: 'Removal Queue Stub',
+      url: TEST_CONFIG.mocks.arrUrl,
+      apiKey: 'removal-queue-e2e',
+      version: 4,
+    });
+    expect(sonarrRes.status).toBe(201);
+    sonarrId = (await sonarrRes.json()).id;
+
+    const clientRes = await createDownloadClient(token, {
+      enabled: true,
+      name: 'qBittorrent removal-queue e2e',
+      typeName: driver.typeName,
+      type: 'Torrent',
+      host: driver.cleanuparrHost,
+      username: driver.username ?? '',
+      password: driver.password ?? '',
+    });
+    expect(clientRes.status).toBeGreaterThanOrEqual(200);
+    expect(clientRes.status).toBeLessThan(300);
+    clientId = (await clientRes.json()).id;
+
+    const notificationRes = await createNotificationProvider(token, 'gotify', {
+      name: 'removal-queue-e2e',
+      serverUrl: TEST_CONFIG.mocks.notifyUrl,
+      applicationToken: 'removal-queue-gotify-token',
+      isEnabled: true,
+      onQueueItemDeleted: true,
+    });
+    expect(notificationRes.status).toBeLessThan(300);
+    notificationId = (await notificationRes.json()).id;
+
+    await api.malwareBlocker.patch({
+      enabled: true,
+      useAdvancedScheduling: true,
+      cronExpression: farFutureCron(),
+      deleteIfAnyFileBlocked: false,
+      ignorePrivate: false,
+      deletePrivate: true,
+      processNoContentId: false,
+      sonarr: {
+        enabled: true,
+        blocklistPath: APP_BLOCKLIST_PATH,
+        blocklistType: 'Blacklist',
+      },
+    });
+  });
+
+  test.afterAll(async () => {
+    await api.general.patch({ dryRun: false });
+    if (notificationId) {
+      await deleteNotificationProvider(token, notificationId);
+    }
+    if (clientId) {
+      await deleteDownloadClient(token, clientId);
+    }
+    if (sonarrId) {
+      await deleteArrInstance(token, 'sonarr', sonarrId);
+    }
+    await arr.resetAll();
+    await notify.resetAll();
+  });
+
+  test.beforeEach(async () => {
+    await notify.resetRequests();
+    await arr.resetRequests();
+  });
+
+  test('several removals in one job run arrive with their own data', async () => {
+    test.setTimeout(120_000);
+
+    const items: BlockedItem[] = [
+      { id: 7001, title: '', infoHash: '', seriesId: 101, episodeId: 201 },
+      { id: 7002, title: '', infoHash: '', seriesId: 102, episodeId: 202 },
+      { id: 7003, title: '', infoHash: '', seriesId: 103, episodeId: 203 },
+    ];
+
+    for (const item of items) {
+      const seeded = await seedBlockedTorrent(driver, hostScanDir, `several-${item.id}.exe`);
+      item.infoHash = seeded.infoHash;
+      item.title = seeded.name;
+    }
+
+    await arr.stubMany([arrHealthStub(), arrQueueStub(queueRecordsFor(items)), arrCatchAllMapping()]);
+
+    const trig = await triggerJob(token, 'MalwareBlocker');
+    expect(trig.ok, `triggerJob: ${trig.status}`).toBe(true);
+
+    const ids = items.map((i) => i.id);
+    await waitUntilDeleted(arr, ids);
+
+    const deletes = await readDeletes(arr);
+    for (const item of items) {
+      const call = deletes.find((d) => d.id === item.id);
+      expect(call, `expected a delete for queue id ${item.id}`).toBeDefined();
+    }
+
+    for (const item of items) {
+      const matches = await waitForEvents(token, 'QueueItemDeleted', item.title);
+      const match = matches.find((e) => e.itemHash.toLowerCase() === item.infoHash.toLowerCase());
+      expect(match, `expected a QueueItemDeleted event for ${item.infoHash}`).toBeDefined();
+      expect(match!.itemTitle).toBe(item.title);
+    }
+
+    await expect
+      .poll(
+        async () => {
+          const requests = await notify.findRequests({ method: 'POST', urlPath: '/message' });
+          return requests.length;
+        },
+        { timeout: POLL_TIMEOUT_MS },
+      )
+      .toBeGreaterThanOrEqual(items.length);
+
+    const notifications = await notify.findRequests({ method: 'POST', urlPath: '/message' });
+    for (const item of items) {
+      const match = notifications.find((n) => (n.body ?? '').includes(item.title));
+      expect(match, `expected a notification request mentioning ${item.title}`).toBeDefined();
+    }
+  });
+
+  test('a failed removal does not stop the queue, and the retry succeeds', async () => {
+    test.setTimeout(120_000);
+
+    const items: BlockedItem[] = [
+      { id: 8001, title: '', infoHash: '', seriesId: 111, episodeId: 211 },
+      { id: 8002, title: '', infoHash: '', seriesId: 112, episodeId: 212 },
+      { id: 8003, title: '', infoHash: '', seriesId: 113, episodeId: 213 },
+    ];
+
+    // A unique filename keeps each repeat's hash clear of earlier runs' events.
+    const runNonce = Date.now();
+
+    for (const item of items) {
+      const seeded = await seedBlockedTorrent(driver, hostScanDir, `failing-${item.id}-${runNonce}.exe`);
+      item.infoHash = seeded.infoHash;
+      item.title = seeded.name;
+    }
+
+    const failingId = items[0].id;
+    const healthyIds = items.slice(1).map((i) => i.id);
+
+    await arr.stubMany([
+      arrHealthStub(),
+      arrQueueStub(queueRecordsFor(items)),
+      arrDeleteFailsMapping(failingId, 500),
+      arrCatchAllMapping(),
+    ]);
+
+    const trig = await triggerJob(token, 'MalwareBlocker');
+    expect(trig.ok, `triggerJob: ${trig.status}`).toBe(true);
+
+    await waitUntilDeleted(arr, healthyIds);
+    await new Promise((r) => setTimeout(r, SETTLE_MS));
+
+    for (const id of healthyIds) {
+      const item = items.find((i) => i.id === id)!;
+      const matches = await waitForEvents(token, 'QueueItemDeleted', item.title);
+      const match = matches.find((e) => e.itemHash.toLowerCase() === item.infoHash.toLowerCase());
+      expect(match, `expected a QueueItemDeleted event for the healthy item ${item.infoHash}`).toBeDefined();
+    }
+
+    const notifications = await notify.findRequests({ method: 'POST', urlPath: '/message' });
+    for (const id of healthyIds) {
+      const item = items.find((i) => i.id === id)!;
+      const match = notifications.find((n) => (n.body ?? '').includes(item.title));
+      expect(match, `expected a notification request mentioning ${item.title}`).toBeDefined();
+    }
+
+    // The arr rejected the first delete with 500.
+    // The queue still removed the other two items.
+    const failingItem = items[0];
+    const failingMatches = await queryEvents(token, 'QueueItemDeleted', failingItem.title);
+    const failingEvent = failingMatches.find((e) => e.itemHash.toLowerCase() === failingItem.infoHash.toLowerCase());
+    expect(failingEvent, 'the failed item should not have a QueueItemDeleted event').toBeUndefined();
+
+    const health = await fetch(`${TEST_CONFIG.appUrl}/health`);
+    expect(health.ok, 'the app should stay healthy after a failed removal').toBe(true);
+
+    // The malware blocker skipped the file already; QueueCleaner retries all-skipped torrents.
+    const originalQueueCleanerConfig = await api.queueCleaner.getJsonConfig();
+    await arr.resetMappings();
+    await arr.resetRequests();
+    await notify.resetRequests();
+    await arr.stubMany([arrHealthStub(), arrQueueStub(queueRecordsFor(items)), arrCatchAllMapping()]);
+
+    try {
+      await api.queueCleaner.patch({
+        enabled: true,
+        useAdvancedScheduling: true,
+        cronExpression: farFutureCron(),
+      });
+
+      const retrig = await triggerJob(token, 'QueueCleaner');
+      expect(retrig.ok, `retrigger triggerJob: ${retrig.status}`).toBe(true);
+
+      await waitUntilDeleted(arr, [failingId]);
+
+      const matches = await waitForEvents(token, 'QueueItemDeleted', failingItem.title);
+      const match = matches.find((e) => e.itemHash.toLowerCase() === failingItem.infoHash.toLowerCase());
+      expect(match, `expected a QueueItemDeleted event for the retried item ${failingItem.infoHash}`).toBeDefined();
+
+      // Wait out the notification so it misses the next test's reset.
+      await expect
+        .poll(
+          async () => {
+            const requests = await notify.findRequests({ method: 'POST', urlPath: '/message' });
+            return requests.some((n) => (n.body ?? '').includes(failingItem.title));
+          },
+          { timeout: POLL_TIMEOUT_MS },
+        )
+        .toBe(true);
+    } finally {
+      await api.queueCleaner.updateConfig(originalQueueCleanerConfig);
+    }
+  });
+
+  test('a dry-run removal deletes nothing and notifies nothing', async () => {
+    test.setTimeout(120_000);
+
+    await api.general.patch({ dryRun: true });
+
+    try {
+      const item: BlockedItem = { id: 9001, title: '', infoHash: '', seriesId: 121, episodeId: 221 };
+      const seeded = await seedBlockedTorrent(driver, hostScanDir, `dry-run-${item.id}.exe`);
+      item.infoHash = seeded.infoHash;
+      item.title = seeded.name;
+
+      await arr.stubMany([arrHealthStub(), arrQueueStub(queueRecordsFor([item])), arrCatchAllMapping()]);
+
+      const trig = await triggerJob(token, 'MalwareBlocker');
+      expect(trig.ok, `triggerJob: ${trig.status}`).toBe(true);
+
+      const markedEvents = await waitForEvents(token, 'DownloadMarkedForDeletion', item.title);
+      const marked = markedEvents.find((e) => e.itemHash.toLowerCase() === item.infoHash.toLowerCase());
+      expect(marked, 'expected a dry-run DownloadMarkedForDeletion event').toBeDefined();
+      expect(marked!.isDryRun).toBe(true);
+
+      await new Promise((r) => setTimeout(r, SETTLE_MS));
+
+      const deletes = await readDeletes(arr);
+      expect(deletes.some((d) => d.id === item.id), 'dry run must not call arr DELETE').toBe(false);
+
+      const notifications = await notify.findRequests({ method: 'POST', urlPath: '/message' });
+      expect(notifications.length, 'dry run must not send a notification').toBe(0);
+    } finally {
+      await api.general.patch({ dryRun: false });
+    }
+  });
+});