From 520dde2c16ea275b83a58330fa0b6106d60f4b8c Mon Sep 17 00:00:00 2001 From: Flaminel Date: Fri, 9 Oct 2026 02:10:04 +0300 Subject: [PATCH] Remove MassTransit (#828) --- CLAUDE.md | 2 +- .../Json/RemovalRequestSerializationTests.cs | 116 ----- .../Cleanuparr.Api/Cleanuparr.Api.csproj | 1 - .../DependencyInjection/MainDI.cs | 53 +-- .../Json/CleanuparrJsonConfiguration.cs | 2 +- .../Entities/Arr/SearchItem.cs | 7 +- .../Consumers/DownloadRemoverConsumerTests.cs | 64 +-- .../Jobs/DownloadCleanerOrphanedFilesTests.cs | 2 +- .../Features/Jobs/DownloadCleanerTests.cs | 2 +- .../Features/Jobs/GenericHandlerTests.cs | 129 +++--- .../DownloadCleanerIntegrationTests.cs | 2 +- .../Integration/IntegrationTestFixture.cs | 17 +- .../MalwareBlockerIntegrationTests.cs | 2 +- .../QueueCleanerIntegrationTests.cs | 4 +- .../Features/Jobs/MalwareBlockerTests.cs | 53 +-- .../Features/Jobs/QueueCleanerTests.cs | 234 +++------- .../Jobs/TestHelpers/JobHandlerFixture.cs | 11 +- .../Features/Messaging/QueueWorkerTests.cs | 193 ++++++++ .../NotificationConsumerTests.cs | 25 +- .../NotificationPublisherTests.cs | 168 ++++--- .../TestHelpers/ChannelAssertions.cs | 20 + .../Cleanuparr.Infrastructure.csproj | 1 - .../Consumers/DownloadRemoverConsumer.cs | 22 +- .../Models/ArrRemovalTarget.cs | 3 - .../Models/LazyLibrarianRemovalTarget.cs | 3 - .../DownloadRemover/Models/RemovalTarget.cs | 8 - .../Features/Jobs/DownloadCleaner.cs | 7 +- .../Features/Jobs/GenericHandler.cs | 18 +- .../Features/Jobs/MalwareBlocker.cs | 7 +- .../Features/Jobs/QueueCleaner.cs | 7 +- .../Features/Messaging/IMessageHandler.cs | 14 + .../Features/Messaging/QueueWorker.cs | 57 +++ .../Consumers/NotificationConsumer.cs | 12 +- .../Notifications/NotificationPublisher.cs | 10 +- .../Logging/LoggingConfigManager.cs | 1 - .../deluge-seeding-rule-cleanup.spec.ts | 8 +- e2e/tests/helpers/api/malware-blocker.ts | 25 ++ e2e/tests/helpers/api/queue-cleaner.ts | 18 + e2e/tests/helpers/mocks/arr-stubs.ts | 28 ++ .../malware-blocker/malware-blocker.spec.ts | 58 +-- .../malware-blocker/removal-queue.spec.ts | 421 ++++++++++++++++++ 41 files changed, 1118 insertions(+), 717 deletions(-) delete mode 100644 code/backend/Cleanuparr.Api.Tests/Json/RemovalRequestSerializationTests.cs create mode 100644 code/backend/Cleanuparr.Infrastructure.Tests/Features/Messaging/QueueWorkerTests.cs create mode 100644 code/backend/Cleanuparr.Infrastructure.Tests/TestHelpers/ChannelAssertions.cs create mode 100644 code/backend/Cleanuparr.Infrastructure/Features/Messaging/IMessageHandler.cs create mode 100644 code/backend/Cleanuparr.Infrastructure/Features/Messaging/QueueWorker.cs create mode 100644 e2e/tests/malware-blocker/removal-queue.spec.ts diff --git a/CLAUDE.md b/CLAUDE.md index 616ec93e..e2d7666d 100644 --- a/CLAUDE.md +++ b/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) diff --git a/code/backend/Cleanuparr.Api.Tests/Json/RemovalRequestSerializationTests.cs b/code/backend/Cleanuparr.Api.Tests/Json/RemovalRequestSerializationTests.cs deleted file mode 100644 index 25295628..00000000 --- a/code/backend/Cleanuparr.Api.Tests/Json/RemovalRequestSerializationTests.cs +++ /dev/null @@ -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; - -/// -/// The in-memory bus serializes with these options, so a broken discriminator -/// only shows up at runtime. -/// -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(json, Options)!; - } - - [Fact] - public void ArrTarget_SurvivesRoundTrip() - { - QueueItemRemoveRequest result = RoundTrip(CreateRequest(new SearchItem { Id = 42 })); - - ArrRemovalTarget target = result.Target.ShouldBeOfType(); - 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(); - 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(); - 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; -} diff --git a/code/backend/Cleanuparr.Api/Cleanuparr.Api.csproj b/code/backend/Cleanuparr.Api/Cleanuparr.Api.csproj index 751d4120..2dbeaeb0 100644 --- a/code/backend/Cleanuparr.Api/Cleanuparr.Api.csproj +++ b/code/backend/Cleanuparr.Api/Cleanuparr.Api.csproj @@ -27,7 +27,6 @@ - all diff --git a/code/backend/Cleanuparr.Api/DependencyInjection/MainDI.cs b/code/backend/Cleanuparr.Api/DependencyInjection/MainDI.cs index 9f9f0c2b..224cb1f8 100644 --- a/code/backend/Cleanuparr.Api/DependencyInjection/MainDI.cs +++ b/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(); - config.AddConsumer(); + .AddMessageQueue() + .AddMessageQueue(); - config.UsingInMemory((context, cfg) => - { - cfg.ConfigureJsonSerializerOptions(options => - { - CleanuparrJsonConfiguration.ConfigureCore(options); + /// + /// Registers an unbounded in-process queue for , + /// drained one message at a time by a . + /// + private static IServiceCollection AddMessageQueue(this IServiceCollection services) + where THandler : class, IMessageHandler + { + Channel channel = Channel.CreateUnbounded(new UnboundedChannelOptions { SingleReader = true }); + + return services + .AddSingleton(channel) + .AddSingleton(channel.Writer) + .AddScoped, THandler>() + .AddHostedService>(); + } - return options; - }); - - cfg.ReceiveEndpoint("download-remover-queue", e => - { - e.ConfigureConsumer(context); - e.ConcurrentMessageLimit = 1; - e.PrefetchCount = 1; - }); - - cfg.ReceiveEndpoint("notification-queue", e => - { - e.ConfigureConsumer(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 diff --git a/code/backend/Cleanuparr.Api/Json/CleanuparrJsonConfiguration.cs b/code/backend/Cleanuparr.Api/Json/CleanuparrJsonConfiguration.cs index 1d48ce49..68fd31c7 100644 --- a/code/backend/Cleanuparr.Api/Json/CleanuparrJsonConfiguration.cs +++ b/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()); diff --git a/code/backend/Cleanuparr.Domain/Entities/Arr/SearchItem.cs b/code/backend/Cleanuparr.Domain/Entities/Arr/SearchItem.cs index 0183994c..193a339c 100644 --- a/code/backend/Cleanuparr.Domain/Entities/Arr/SearchItem.cs +++ b/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; } diff --git a/code/backend/Cleanuparr.Infrastructure.Tests/Features/DownloadRemover/Consumers/DownloadRemoverConsumerTests.cs b/code/backend/Cleanuparr.Infrastructure.Tests/Features/DownloadRemover/Consumers/DownloadRemoverConsumerTests.cs index a3939f8e..d5b0226a 100644 --- a/code/backend/Cleanuparr.Infrastructure.Tests/Features/DownloadRemover/Consumers/DownloadRemoverConsumerTests.cs +++ b/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()) .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()) .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()); // 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()) .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()) .Returns(Task.CompletedTask); // Act - await _consumer.Consume(context); + await _consumer.HandleAsync(request); // Assert Task 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 context = CreateConsumeContext(request); _queueItemRemover .RemoveQueueItemAsync(Arg.Any()) .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 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()) .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()) .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()) .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 CreateConsumeContext(QueueItemRemoveRequest message) - { - var context = Substitute.For>(); - context.Message.Returns(message); - return context; - } #endregion } diff --git a/code/backend/Cleanuparr.Infrastructure.Tests/Features/Jobs/DownloadCleanerOrphanedFilesTests.cs b/code/backend/Cleanuparr.Infrastructure.Tests/Features/Jobs/DownloadCleanerOrphanedFilesTests.cs index 2ede54c0..c151b047 100644 --- a/code/backend/Cleanuparr.Infrastructure.Tests/Features/Jobs/DownloadCleanerOrphanedFilesTests.cs +++ b/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, diff --git a/code/backend/Cleanuparr.Infrastructure.Tests/Features/Jobs/DownloadCleanerTests.cs b/code/backend/Cleanuparr.Infrastructure.Tests/Features/Jobs/DownloadCleanerTests.cs index f31040d1..9602eb31 100644 --- a/code/backend/Cleanuparr.Infrastructure.Tests/Features/Jobs/DownloadCleanerTests.cs +++ b/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, diff --git a/code/backend/Cleanuparr.Infrastructure.Tests/Features/Jobs/GenericHandlerTests.cs b/code/backend/Cleanuparr.Infrastructure.Tests/Features/Jobs/GenericHandlerTests.cs index 4bb175a0..dea6eaf2 100644 --- a/code/backend/Cleanuparr.Infrastructure.Tests/Features/Jobs/GenericHandlerTests.cs +++ b/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 Substitute.For>(), _fixture.DataContext, _fixture.Cache, - _fixture.MessageBus, + _fixture.RemovalQueue.Writer, _fixture.ArrClientFactory, _fixture.ArrQueueReader, _fixture.DownloadServiceFactory, @@ -293,7 +294,7 @@ public class GenericHandlerTests : IClassFixture 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 instance, record, isPack: false, removeFromClient: true, DeleteReason.FailedImport); // Assert - await _fixture.MessageBus.Received(1) - .Publish(Arg.Any(), Arg.Any()); + _fixture.RemovalQueue.Reader.Count.ShouldBe(1); await _fixture.EventPublisher.Received(1).PublishAsync( EventType.DownloadMarkedForDeletion, Arg.Any(), Arg.Any(), Arg.Any?>(), Arg.Any(), Arg.Any(), Arg.Any()); @@ -344,8 +344,8 @@ public class GenericHandlerTests : IClassFixture instance, record, isPack: false, removeFromClient: true, DeleteReason.FailedImport); // Assert - await _fixture.MessageBus.Received(1) - .Publish(Arg.Is(r => r.IsDryRun), Arg.Any()); + QueueItemRemoveRequest request = _fixture.RemovalQueue.ShouldHaveSingle(); + request.IsDryRun.ShouldBeTrue(); } [Fact] @@ -368,8 +368,7 @@ public class GenericHandlerTests : IClassFixture instance, record, isPack: false, removeFromClient: false, DeleteReason.Stalled); // Assert - await _fixture.MessageBus.Received(1) - .Publish(Arg.Any(), Arg.Any()); + _fixture.RemovalQueue.Reader.Count.ShouldBe(1); } [Fact] @@ -392,8 +391,7 @@ public class GenericHandlerTests : IClassFixture instance, record, isPack: false, removeFromClient: true, DeleteReason.FailedImport); // Assert - await _fixture.MessageBus.Received(1) - .Publish(Arg.Any(), Arg.Any()); + _fixture.RemovalQueue.Reader.Count.ShouldBe(1); } [Fact] @@ -411,45 +409,32 @@ public class GenericHandlerTests : IClassFixture }; 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(), Arg.Any())) - .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>(), + _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(), Arg.Any()) - .Returns(Task.FromException(new InvalidOperationException("bus is down"))); - - // Act & Assert - await Should.ThrowAsync(() => _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 _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(), Arg.Any()) - .Returns(Task.FromException(new InvalidOperationException("bus is down"))); - - // Act & Assert: nothing was removed, so a pending import still stands - await Should.ThrowAsync(() => _handler.PublicPublishQueueItemRemoveRequest( - instance, record, isPack: false, removeFromClient: true, DeleteReason.FailedImport)); - - _fixture.ForceImportService.DidNotReceive().Forget(Arg.Any(), Arg.Any()); - } - #endregion #region GetInitializedDownloadServicesAsync @@ -679,6 +638,32 @@ public class GenericHandlerTests : IClassFixture }; } + /// + /// Wraps a real channel writer and records, at the moment of the write, whether the given probe is true. + /// + private sealed class MarkOrderSpyChannelWriter : ChannelWriter + { + private readonly ChannelWriter _inner; + private readonly Func _probe; + + public bool? MarkWasSetAtWrite { get; private set; } + + public MarkOrderSpyChannelWriter(ChannelWriter inner, Func probe) + { + _inner = inner; + _probe = probe; + } + + public override bool TryWrite(QueueItemRemoveRequest item) + { + MarkWasSetAtWrite = _probe(); + return _inner.TryWrite(item); + } + + public override ValueTask WaitToWriteAsync(CancellationToken cancellationToken = default) => + _inner.WaitToWriteAsync(cancellationToken); + } + /// /// Concrete GenericHandler subclass exposing protected members for testing. /// @@ -706,14 +691,14 @@ public class GenericHandlerTests : IClassFixture ILogger logger, DataContext dataContext, IMemoryCache cache, - IBus messageBus, + ChannelWriter 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) { } diff --git a/code/backend/Cleanuparr.Infrastructure.Tests/Features/Jobs/Integration/DownloadCleanerIntegrationTests.cs b/code/backend/Cleanuparr.Infrastructure.Tests/Features/Jobs/Integration/DownloadCleanerIntegrationTests.cs index 7c499b30..df21cf0c 100644 --- a/code/backend/Cleanuparr.Infrastructure.Tests/Features/Jobs/Integration/DownloadCleanerIntegrationTests.cs +++ b/code/backend/Cleanuparr.Infrastructure.Tests/Features/Jobs/Integration/DownloadCleanerIntegrationTests.cs @@ -51,7 +51,7 @@ public class DownloadCleanerIntegrationTests : IDisposable Substitute.For>(), _fixture.DataContext, _fixture.Cache, - _fixture.MessageBus, + _fixture.RemovalQueue.Writer, _fixture.ArrClientFactory, _fixture.ArrQueueReader, _fixture.DownloadServiceFactory, diff --git a/code/backend/Cleanuparr.Infrastructure.Tests/Features/Jobs/Integration/IntegrationTestFixture.cs b/code/backend/Cleanuparr.Infrastructure.Tests/Features/Jobs/Integration/IntegrationTestFixture.cs index 1b2f8500..8d77dd9f 100644 --- a/code/backend/Cleanuparr.Infrastructure.Tests/Features/Jobs/Integration/IntegrationTestFixture.cs +++ b/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 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(); + RemovalQueue = Channel.CreateUnbounded(); ArrClientFactory = Substitute.For(); ArrClient = Substitute.For(); ArrQueueReader = Substitute.For(); @@ -112,10 +112,6 @@ public class IntegrationTestFixture : IDisposable DryRunInterceptor.IsDryRunEnabled().Returns(false); DryRunInterceptor.InterceptAsync(Arg.Any>(), Arg.Any()).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 GetCapturedRemoveRequests() { + while (RemovalQueue.Reader.TryRead(out QueueItemRemoveRequest? request)) + { + CapturedMessages.Add(request); + } + return CapturedMessages .OfType() .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(); + RemovalQueue = Channel.CreateUnbounded(); ArrClientFactory = Substitute.For(); ArrClient = Substitute.For(); ArrQueueReader = Substitute.For(); diff --git a/code/backend/Cleanuparr.Infrastructure.Tests/Features/Jobs/Integration/MalwareBlockerIntegrationTests.cs b/code/backend/Cleanuparr.Infrastructure.Tests/Features/Jobs/Integration/MalwareBlockerIntegrationTests.cs index 393765a8..d20b04f6 100644 --- a/code/backend/Cleanuparr.Infrastructure.Tests/Features/Jobs/Integration/MalwareBlockerIntegrationTests.cs +++ b/code/backend/Cleanuparr.Infrastructure.Tests/Features/Jobs/Integration/MalwareBlockerIntegrationTests.cs @@ -39,7 +39,7 @@ public class MalwareBlockerIntegrationTests : IDisposable Substitute.For>(), _fixture.DataContext, _fixture.Cache, - _fixture.MessageBus, + _fixture.RemovalQueue.Writer, _fixture.ArrClientFactory, _fixture.ArrQueueReader, _fixture.DownloadServiceFactory, diff --git a/code/backend/Cleanuparr.Infrastructure.Tests/Features/Jobs/Integration/QueueCleanerIntegrationTests.cs b/code/backend/Cleanuparr.Infrastructure.Tests/Features/Jobs/Integration/QueueCleanerIntegrationTests.cs index b16f9b1c..905181a1 100644 --- a/code/backend/Cleanuparr.Infrastructure.Tests/Features/Jobs/Integration/QueueCleanerIntegrationTests.cs +++ b/code/backend/Cleanuparr.Infrastructure.Tests/Features/Jobs/Integration/QueueCleanerIntegrationTests.cs @@ -42,7 +42,7 @@ public class QueueCleanerIntegrationTests : IDisposable Substitute.For>(), _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); diff --git a/code/backend/Cleanuparr.Infrastructure.Tests/Features/Jobs/MalwareBlockerTests.cs b/code/backend/Cleanuparr.Infrastructure.Tests/Features/Jobs/MalwareBlockerTests.cs index f3fb933c..646d6f16 100644 --- a/code/backend/Cleanuparr.Infrastructure.Tests/Features/Jobs/MalwareBlockerTests.cs +++ b/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(r => - r.DeleteReason == DeleteReason.AllFilesBlocked - ), - Arg.Any() - ); + 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(r => - r.DeleteReason == DeleteReason.AtLeastOneFileBlocked - ), - Arg.Any() - ); + 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(r => - r.ArrTarget().RemoveFromClient == false - ), - Arg.Any() - ); + 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(), - Arg.Any() - ); + _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(r => - r.SkipSearch == true && - r.DeleteReason == DeleteReason.AllFilesBlocked - ), - Arg.Any() - ); + 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(r => - r.LazyTarget().RemovedFromClient - && r.LazyTarget().Item.Books.Single().BookId == "OL7353617M" - && r.DeleteReason == DeleteReason.AllFilesBlocked - ), - Arg.Any() - ); + 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!); diff --git a/code/backend/Cleanuparr.Infrastructure.Tests/Features/Jobs/QueueCleanerTests.cs b/code/backend/Cleanuparr.Infrastructure.Tests/Features/Jobs/QueueCleanerTests.cs index a554c760..3e56e989 100644 --- a/code/backend/Cleanuparr.Infrastructure.Tests/Features/Jobs/QueueCleanerTests.cs +++ b/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(), - Arg.Any() - ); + _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(r => - r.SeriesItem().SearchType == SeriesSearchType.Season && - r.SeriesItem().Id == 1 - ), - Arg.Any() + List 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(r => - r.DeleteReason == DeleteReason.FailedImport - ), - Arg.Any() - ); + 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(), - Arg.Any() - ); + _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(r => - r.SkipSearch == true && - r.DeleteReason == DeleteReason.Stalled - ), - Arg.Any() - ); + 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(), - Arg.Any() - ); + _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(r => - r.Instance.ArrConfig.Type == InstanceType.Radarr && - r.ArrTarget().SearchItem.Id == 42 && - r.DeleteReason == DeleteReason.Stalled - ), - Arg.Any() - ); + 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(r => - r.Instance.ArrConfig.Type == InstanceType.Lidarr && - r.ArrTarget().SearchItem.Id == 123 && - r.DeleteReason == DeleteReason.SlowSpeed - ), - Arg.Any() - ); + 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(r => - r.Instance.ArrConfig.Type == InstanceType.Readarr && - r.ArrTarget().SearchItem.Id == 456 && - r.DeleteReason == DeleteReason.Stalled - ), - Arg.Any() - ); + 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(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() - ); + 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(r => - r.Instance.ArrConfig.Type == InstanceType.Whisparr && - r.ArrTarget().SearchItem.Id == 42 && // MovieId - r.DeleteReason == DeleteReason.Stalled - ), - Arg.Any() - ); + 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(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() - ); + 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(r => - r.DeleteReason == DeleteReason.FailedImport && - r.ArrTarget().ChangeCategory == true && - r.ArrTarget().RemoveFromClient == false - ), - Arg.Any() - ); + 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(r => - r.DeleteReason == DeleteReason.Stalled && - r.ArrTarget().ChangeCategory == true && - r.ArrTarget().RemoveFromClient == false - ), - Arg.Any() - ); + 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(r => - r.LazyTarget().RemovedFromClient - && r.LazyTarget().Item.Books.Single().BookId == "OL7353617M" - && r.DeleteReason == DeleteReason.Stalled - ), - Arg.Any() - ); + 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(), Arg.Any()); - await _fixture.MessageBus.Received(1).Publish( - Arg.Is(r => - !r.LazyTarget().RemovedFromClient - ), - Arg.Any() - ); + 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(), - Arg.Any() - ); - } - - [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(); - secondTorrent.Hash.Returns("torrent-hash-2"); - - DownloadClientConfig clientConfig = downloadService.ClientConfig; - IReadOnlyList 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(), Arg.Any>(), Arg.Any>()) - .Returns(decisions); - - _fixture.MessageBus - .Publish(Arg.Is(r => r.Target.DownloadId == "torrent-hash"), Arg.Any()) - .Returns(Task.FromException(new Exception("bus is down"))); - - var sut = CreateSut(); - - // Act - await sut.ExecuteAsync(); - - // Assert - await _fixture.MessageBus.Received(1).Publish( - Arg.Is(r => r.Target.DownloadId == "torrent-hash-2"), - Arg.Any() - ); + _fixture.RemovalQueue.Reader.Count.ShouldBe(0); } [Fact] diff --git a/code/backend/Cleanuparr.Infrastructure.Tests/Features/Jobs/TestHelpers/JobHandlerFixture.cs b/code/backend/Cleanuparr.Infrastructure.Tests/Features/Jobs/TestHelpers/JobHandlerFixture.cs index 278498b0..0c4e4a04 100644 --- a/code/backend/Cleanuparr.Infrastructure.Tests/Features/Jobs/TestHelpers/JobHandlerFixture.cs +++ b/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 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(); + RemovalQueue = Channel.CreateUnbounded(); ArrClientFactory = Substitute.For(); ArrQueueReader = Substitute.For(); DownloadServiceFactory = Substitute.For(); @@ -160,7 +161,7 @@ public class JobHandlerFixture : IDisposable { SubstituteHelper.ClearPendingArgSpecs(); // Recreate all substitutes to clear received call state - MessageBus = Substitute.For(); + RemovalQueue = Channel.CreateUnbounded(); ArrClientFactory = Substitute.For(); ArrQueueReader = Substitute.For(); DownloadServiceFactory = Substitute.For(); diff --git a/code/backend/Cleanuparr.Infrastructure.Tests/Features/Messaging/QueueWorkerTests.cs b/code/backend/Cleanuparr.Infrastructure.Tests/Features/Messaging/QueueWorkerTests.cs new file mode 100644 index 00000000..f6d97aed --- /dev/null +++ b/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 _channel = Channel.CreateUnbounded(); + private readonly Recorder _recorder = new(); + private readonly ServiceProvider _provider; + private readonly ILogger> _logger; + private readonly QueueWorker _worker; + + public QueueWorkerTests() + { + ServiceCollection services = new(); + services.AddSingleton(_recorder); + services.AddScoped(); + services.AddScoped, TestHandler>(); + _provider = services.BuildServiceProvider(); + + _logger = Substitute.For>>(); + _worker = new QueueWorker(_logger, _provider.GetRequiredService(), _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 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 HandledOrder = []; + public readonly List ObservedContextAtStart = []; + public readonly List ScopeIds = []; + public readonly HashSet ThrowIds = []; + public TaskCompletionSource? Gate; + public int Concurrent; + public int MaxConcurrent; + } + + private sealed class TestHandler : IMessageHandler + { + 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}"); + } + } + } +} diff --git a/code/backend/Cleanuparr.Infrastructure.Tests/Features/Notifications/NotificationConsumerTests.cs b/code/backend/Cleanuparr.Infrastructure.Tests/Features/Notifications/NotificationConsumerTests.cs index c898e1c7..ba61075a 100644 --- a/code/backend/Cleanuparr.Infrastructure.Tests/Features/Notifications/NotificationConsumerTests.cs +++ b/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 CreateConsumeContext(NotificationMessage message) - { - ConsumeContext context = Substitute.For>(); - 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()); - await _consumer.Consume(CreateConsumeContext(message)); + await _consumer.HandleAsync(message); _providerFactory.DidNotReceive().CreateProvider(Arg.Any()); } [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()) .ThrowsAsync(new InvalidOperationException("db locked")); - ConsumeContext ctx = CreateConsumeContext(message); - - await Should.NotThrowAsync(() => _consumer.Consume(ctx)); + await Should.NotThrowAsync(() => _consumer.HandleAsync(message)); _providerFactory.DidNotReceive().CreateProvider(Arg.Any()); } diff --git a/code/backend/Cleanuparr.Infrastructure.Tests/Features/Notifications/NotificationPublisherTests.cs b/code/backend/Cleanuparr.Infrastructure.Tests/Features/Notifications/NotificationPublisherTests.cs index 013cf833..2f678e82 100644 --- a/code/backend/Cleanuparr.Infrastructure.Tests/Features/Notifications/NotificationPublisherTests.cs +++ b/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 _logger; private readonly IDryRunInterceptor _dryRunInterceptor; - private readonly IBus _messageBus; + private readonly Channel _notificationQueue; private readonly NotificationPublisher _publisher; public NotificationPublisherTests() { _logger = Substitute.For>(); _dryRunInterceptor = Substitute.For(); - _messageBus = Substitute.For(); + _notificationQueue = Channel.CreateUnbounded(); _dryRunInterceptor.InterceptAsync(Arg.Any>(), Arg.Any()) .ReturnsForAnyArgs(ci => ci.ArgAt>(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( - m => m.EventType == NotificationEventType.StalledStrike && - m.Context.Data.ContainsKey("Strike type") && - m.Context.Data["Strike type"] == "Stalled"), Arg.Any()); + 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( - m => m.EventType == NotificationEventType.FailedImportStrike && - m.Context.Data["Strike count"] == "2"), Arg.Any()); + 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( - m => m.EventType == expectedEventType), Arg.Any()); + 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(), Arg.Any()); + _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( - m => m.Context.Data["Url"] == "http://sonarr.local/"), Arg.Any()); + 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( - m => m.EventType == NotificationEventType.QueueItemDeleted && - m.Context.Data["Reason"] == "Stalled" && - m.Context.Data["Removed from client?"] == "True" && - m.Context.Severity == EventSeverity.Important), Arg.Any()); + 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( - m => m.Context.Data["Removed from client?"] == "False" && - m.Context.Data["Reason"] == "AllFilesBlocked"), Arg.Any()); + 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( - 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()); + 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(m => captured = m), Arg.Any()) - .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( - m => m.Context.Data.ContainsKey("Url") && - m.Context.Data["Url"] == "https://qbit.external.com/"), Arg.Any()); + 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( - 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()); + 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(m => captured = m), Arg.Any()) - .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( - m => m.Context.Data.ContainsKey("Url") && - m.Context.Data["Url"] == "https://qbit.external.com/"), Arg.Any()); + 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( - 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()); + 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(m => captured = m), Arg.Any()) - .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( - m => m.Context.Severity == EventSeverity.Information), Arg.Any()); + 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(), Arg.Any()); + _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( - 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()); + 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( - 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()); + 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( - m => m.Context.Description == "Frankenstein" && m.Context.Data["Hash"] == "bookhash1"), Arg.Any()); + 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( - m => m.Context.Description == "Frankenstein" && m.Context.Data["Hash"] == "bookhash1"), Arg.Any()); + NotificationMessage m = _notificationQueue.ShouldHaveSingle(); + m.Context.Description.ShouldBe("Frankenstein"); + m.Context.Data["Hash"].ShouldBe("bookhash1"); } #endregion diff --git a/code/backend/Cleanuparr.Infrastructure.Tests/TestHelpers/ChannelAssertions.cs b/code/backend/Cleanuparr.Infrastructure.Tests/TestHelpers/ChannelAssertions.cs new file mode 100644 index 00000000..b5db4ee1 --- /dev/null +++ b/code/backend/Cleanuparr.Infrastructure.Tests/TestHelpers/ChannelAssertions.cs @@ -0,0 +1,20 @@ +using System.Threading.Channels; +using Shouldly; + +namespace Cleanuparr.Infrastructure.Tests.TestHelpers; + +/// +/// Assertions for channels used as in-process queues in tests. +/// +internal static class ChannelAssertions +{ + /// + /// Asserts the channel holds exactly one message and returns it. + /// + internal static T ShouldHaveSingle(this Channel channel) + { + channel.Reader.Count.ShouldBe(1); + channel.Reader.TryRead(out T? item).ShouldBeTrue(); + return item!; + } +} diff --git a/code/backend/Cleanuparr.Infrastructure/Cleanuparr.Infrastructure.csproj b/code/backend/Cleanuparr.Infrastructure/Cleanuparr.Infrastructure.csproj index c4c6cf8c..d8e73acc 100644 --- a/code/backend/Cleanuparr.Infrastructure/Cleanuparr.Infrastructure.csproj +++ b/code/backend/Cleanuparr.Infrastructure/Cleanuparr.Infrastructure.csproj @@ -13,7 +13,6 @@ - diff --git a/code/backend/Cleanuparr.Infrastructure/Features/DownloadRemover/Consumers/DownloadRemoverConsumer.cs b/code/backend/Cleanuparr.Infrastructure/Features/DownloadRemover/Consumers/DownloadRemoverConsumer.cs index 4b63267e..1c568166 100644 --- a/code/backend/Cleanuparr.Infrastructure/Features/DownloadRemover/Consumers/DownloadRemoverConsumer.cs +++ b/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 +/// +/// Removes one queued download removal request. +/// +public sealed class DownloadRemoverConsumer : IMessageHandler { private readonly ILogger _logger; private readonly IQueueItemRemover _queueItemRemover; @@ -24,15 +27,16 @@ public sealed class DownloadRemoverConsumer : IConsumer _dryRunActivity = dryRunActivity; } - public async Task Consume(ConsumeContext context) + /// + 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 try { - await _queueItemRemover.RemoveQueueItemAsync(context.Message); + await _queueItemRemover.RemoveQueueItemAsync(message); } finally { @@ -53,9 +57,9 @@ public sealed class DownloadRemoverConsumer : IConsumer 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 ); } } diff --git a/code/backend/Cleanuparr.Infrastructure/Features/DownloadRemover/Models/ArrRemovalTarget.cs b/code/backend/Cleanuparr.Infrastructure/Features/DownloadRemover/Models/ArrRemovalTarget.cs index 7704f1e5..4425cc58 100644 --- a/code/backend/Cleanuparr.Infrastructure/Features/DownloadRemover/Models/ArrRemovalTarget.cs +++ b/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 /// public bool ChangeCategory { get; init; } - [JsonIgnore] public override string DownloadId => Record.DownloadId; - [JsonIgnore] public override string Title => Record.Title; } diff --git a/code/backend/Cleanuparr.Infrastructure/Features/DownloadRemover/Models/LazyLibrarianRemovalTarget.cs b/code/backend/Cleanuparr.Infrastructure/Features/DownloadRemover/Models/LazyLibrarianRemovalTarget.cs index 1fc3665e..daeeb8ee 100644 --- a/code/backend/Cleanuparr.Infrastructure/Features/DownloadRemover/Models/LazyLibrarianRemovalTarget.cs +++ b/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 /// public required bool RemovedFromClient { get; init; } - [JsonIgnore] public override string DownloadId => Item.DownloadId; - [JsonIgnore] public override string Title => Item.Title; } diff --git a/code/backend/Cleanuparr.Infrastructure/Features/DownloadRemover/Models/RemovalTarget.cs b/code/backend/Cleanuparr.Infrastructure/Features/DownloadRemover/Models/RemovalTarget.cs index 30ced9c9..35a4d97a 100644 --- a/code/backend/Cleanuparr.Infrastructure/Features/DownloadRemover/Models/RemovalTarget.cs +++ b/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; } } diff --git a/code/backend/Cleanuparr.Infrastructure/Features/Jobs/DownloadCleaner.cs b/code/backend/Cleanuparr.Infrastructure/Features/Jobs/DownloadCleaner.cs index 7c12e15f..eaa1e132 100644 --- a/code/backend/Cleanuparr.Infrastructure/Features/Jobs/DownloadCleaner.cs +++ b/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 logger, DataContext dataContext, IMemoryCache cache, - IBus messageBus, + ChannelWriter 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 ) diff --git a/code/backend/Cleanuparr.Infrastructure/Features/Jobs/GenericHandler.cs b/code/backend/Cleanuparr.Infrastructure/Features/Jobs/GenericHandler.cs index 36dbedb5..84340c2f 100644 --- a/code/backend/Cleanuparr.Infrastructure/Features/Jobs/GenericHandler.cs +++ b/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 _logger; protected readonly DataContext _dataContext; protected readonly IMemoryCache _cache; - protected readonly IBus _messageBus; + protected readonly ChannelWriter _removalQueue; protected readonly IArrClientFactory _arrClientFactory; protected readonly IArrQueueReader _arrQueueReader; protected readonly IDownloadServiceFactory _downloadServiceFactory; @@ -42,7 +42,7 @@ public abstract class GenericHandler : IHandler ILogger logger, DataContext dataContext, IMemoryCache cache, - IBus messageBus, + ChannelWriter 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); diff --git a/code/backend/Cleanuparr.Infrastructure/Features/Jobs/MalwareBlocker.cs b/code/backend/Cleanuparr.Infrastructure/Features/Jobs/MalwareBlocker.cs index ae3b4005..abc30b69 100644 --- a/code/backend/Cleanuparr.Infrastructure/Features/Jobs/MalwareBlocker.cs +++ b/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 logger, DataContext dataContext, IMemoryCache cache, - IBus messageBus, + ChannelWriter 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 ) diff --git a/code/backend/Cleanuparr.Infrastructure/Features/Jobs/QueueCleaner.cs b/code/backend/Cleanuparr.Infrastructure/Features/Jobs/QueueCleaner.cs index e62d6b6a..70f8640c 100644 --- a/code/backend/Cleanuparr.Infrastructure/Features/Jobs/QueueCleaner.cs +++ b/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 logger, DataContext dataContext, IMemoryCache cache, - IBus messageBus, + ChannelWriter 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 ) diff --git a/code/backend/Cleanuparr.Infrastructure/Features/Messaging/IMessageHandler.cs b/code/backend/Cleanuparr.Infrastructure/Features/Messaging/IMessageHandler.cs new file mode 100644 index 00000000..689a85ec --- /dev/null +++ b/code/backend/Cleanuparr.Infrastructure/Features/Messaging/IMessageHandler.cs @@ -0,0 +1,14 @@ +namespace Cleanuparr.Infrastructure.Features.Messaging; + +/// +/// Handles one message taken from a queue. +/// +/// The queued message type. +public interface IMessageHandler +{ + /// + /// Handles a single queued message. + /// + /// The message to handle. + Task HandleAsync(TMessage message); +} diff --git a/code/backend/Cleanuparr.Infrastructure/Features/Messaging/QueueWorker.cs b/code/backend/Cleanuparr.Infrastructure/Features/Messaging/QueueWorker.cs new file mode 100644 index 00000000..5d45e606 --- /dev/null +++ b/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; + +/// +/// Drains a channel one message at a time, each in its own DI scope. +/// +/// The queued message type. +public sealed class QueueWorker : BackgroundService +{ + private readonly ILogger> _logger; + private readonly IServiceScopeFactory _scopeFactory; + private readonly Channel _channel; + + /// + /// Initializes a new instance of the class. + /// + /// The logger. + /// Creates a DI scope per message. + /// The channel to drain. + public QueueWorker( + ILogger> logger, + IServiceScopeFactory scopeFactory, + Channel channel) + { + _logger = logger; + _scopeFactory = scopeFactory; + _channel = channel; + } + + /// + 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 handler = scope.ServiceProvider.GetRequiredService>(); + await handler.HandleAsync(message); + } + catch (Exception exception) + { + _logger.LogError(exception, "failed to handle queued message | {MessageType}", typeof(TMessage).Name); + } + } +} diff --git a/code/backend/Cleanuparr.Infrastructure/Features/Notifications/Consumers/NotificationConsumer.cs b/code/backend/Cleanuparr.Infrastructure/Features/Notifications/Consumers/NotificationConsumer.cs index 20dc1367..fc051286 100644 --- a/code/backend/Cleanuparr.Infrastructure/Features/Notifications/Consumers/NotificationConsumer.cs +++ b/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 +/// +/// Sends one queued notification to its configured providers. +/// +public sealed class NotificationConsumer : IMessageHandler { private readonly NotificationService _notificationService; @@ -12,6 +15,7 @@ public sealed class NotificationConsumer : IConsumer _notificationService = notificationService; } - public Task Consume(ConsumeContext context) => - _notificationService.SendNotificationAsync(context.Message.EventType, context.Message.Context); + /// + public Task HandleAsync(NotificationMessage message) => + _notificationService.SendNotificationAsync(message.EventType, message.Context); } diff --git a/code/backend/Cleanuparr.Infrastructure/Features/Notifications/NotificationPublisher.cs b/code/backend/Cleanuparr.Infrastructure/Features/Notifications/NotificationPublisher.cs index 44524b32..f04a291a 100644 --- a/code/backend/Cleanuparr.Infrastructure/Features/Notifications/NotificationPublisher.cs +++ b/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 _logger; private readonly IDryRunInterceptor _dryRunInterceptor; - private readonly IBus _messageBus; + private readonly ChannelWriter _notificationQueue; public NotificationPublisher( ILogger logger, IDryRunInterceptor dryRunInterceptor, - IBus messageBus) + ChannelWriter 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)); } /// diff --git a/code/backend/Cleanuparr.Infrastructure/Logging/LoggingConfigManager.cs b/code/backend/Cleanuparr.Infrastructure/Logging/LoggingConfigManager.cs index acb155e0..052be638 100644 --- a/code/backend/Cleanuparr.Infrastructure/Logging/LoggingConfigManager.cs +++ b/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) diff --git a/e2e/tests/download-cleaner/deluge-seeding-rule-cleanup.spec.ts b/e2e/tests/download-cleaner/deluge-seeding-rule-cleanup.spec.ts index 790e4881..c4b69f70 100644 --- a/e2e/tests/download-cleaner/deluge-seeding-rule-cleanup.spec.ts +++ b/e2e/tests/download-cleaner/deluge-seeding-rule-cleanup.spec.ts @@ -65,10 +65,10 @@ async function stillPresent(infoHashes: string[]): Promise { * `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; diff --git a/e2e/tests/helpers/api/malware-blocker.ts b/e2e/tests/helpers/api/malware-blocker.ts index e5cde6f5..9b9e8226 100644 --- a/e2e/tests/helpers/api/malware-blocker.ts +++ b/e2e/tests/helpers/api/malware-blocker.ts @@ -10,4 +10,29 @@ export class MalwareBlockerApi { updateConfig(body: Record): Promise { return this.client.put('/api/configuration/malware_blocker', body); } + + async getJsonConfig(): Promise> { + 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): Promise { + const current = await this.getJsonConfig(); + const merged = { + ...current, + ...overrides, + sonarr: { + ...(current.sonarr as Record), + ...(overrides.sonarr as Record), + }, + }; + const res = await this.updateConfig(merged); + if (!res.ok) { + throw new Error(`PUT malware blocker config failed: ${res.status} ${await res.text()}`); + } + } } diff --git a/e2e/tests/helpers/api/queue-cleaner.ts b/e2e/tests/helpers/api/queue-cleaner.ts index 50dd9bfe..a973c415 100644 --- a/e2e/tests/helpers/api/queue-cleaner.ts +++ b/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> { + 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): Promise { + 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 { return this.client.get(`/api/queue-rules/${kind}`); } diff --git a/e2e/tests/helpers/mocks/arr-stubs.ts b/e2e/tests/helpers/mocks/arr-stubs.ts index b45deb1f..99e53a41 100644 --- a/e2e/tests/helpers/mocks/arr-stubs.ts +++ b/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 { + 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' }, diff --git a/e2e/tests/malware-blocker/malware-blocker.spec.ts b/e2e/tests/malware-blocker/malware-blocker.spec.ts index d82c6679..304f608e 100644 --- a/e2e/tests/malware-blocker/malware-blocker.spec.ts +++ b/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 { 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 { @@ -168,34 +146,8 @@ async function queueRequestSeen(arr: WireMockClient, timeoutMs: number): Promise return false; } -async function readDeletes(arr: WireMockClient): Promise { - 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, -): Promise { - 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 ?? {}) }, - }; - 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): Promise { + await new CleanuparrApi({ token }).malwareBlocker.patch(overrides); } test.describe.serial('Malware blocker — DeleteIfAnyFileBlocked', () => { diff --git a/e2e/tests/malware-blocker/removal-queue.spec.ts b/e2e/tests/malware-blocker/removal-queue.spec.ts new file mode 100644 index 00000000..c7e47d46 --- /dev/null +++ b/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 ), + })); +} + +async function waitUntilDeleted(arr: WireMockClient, ids: number[], timeoutMs = POLL_TIMEOUT_MS): Promise { + 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 { + 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 { + 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 }); + } + }); +});