From b8767af462b92f9b748e8d97cc6f915349c03bb6 Mon Sep 17 00:00:00 2001 From: John Simons Date: Wed, 29 Jul 2026 12:50:27 +1000 Subject: [PATCH] Refactor monolithic IErrorMessageDataStore into specialized interfaces This change decomposes the single, large `IErrorMessageDataStore` interface into several smaller, more focused interfaces such as `IMessagesViewDataStore`, `IFailedMessageQueryDataStore`, `IFailedMessageLifecycleDataStore`, `IFailedMessageRetryDataStore`, `IEditFailedMessagesDataStore`, and `INotificationsDataStore`. This refactoring improves the separation of concerns, enhances testability, and aligns the codebase with the Interface Segregation and Single Responsibility Principles. Existing implementations and consumers across both EFCore and RavenDB persistence layers, as well as application logic, have been updated to utilize these new, more specific contracts. --- .../When_email_notifications_are_enabled.cs | 4 +- .../When_a_retry_fails_to_be_sent.cs | 4 +- .../Abstractions/BasePersistence.cs | 7 +- .../EditFailedMessagesDataStore.cs | 7 + .../Implementation/ErrorMessagesDataStore.cs | 163 ----------------- .../FailedErrorImportDataStore.cs | 45 +++++ .../FailedMessageLifecycleDataStore.cs | 24 +++ .../FailedMessageQueryDataStore.cs | 30 ++++ .../FailedMessageRetryDataStore.cs | 18 ++ .../Implementation/GroupsDataStore.cs | 23 +++ .../Implementation/MessagesViewDataStore.cs | 22 +++ .../Implementation/NotificationsDataStore.cs | 7 + .../Editing/EditFailedMessagesDataStore.cs | 11 ++ .../Editing/NotificationsDataStore.cs | 11 ++ .../ErrorMessagesDataStore.cs | 166 ++---------------- .../EventLogDataStore.cs | 5 +- .../FailedErrorImportDataStore.cs | 17 ++ .../RavenPersistence.cs | 9 +- .../Recoverability/GroupsDataStore.cs | 108 +++++++++++- .../Expiration/MessageExpiryTests.cs | 8 +- .../FailedErrorImportCustomCheckTests.cs | 2 +- .../FailedErrorImportDedupeTests.cs | 2 +- .../ErrorMessageDataStoreTests.cs | 12 +- .../EFCore/FailedErrorImportTests.cs | 3 +- .../PersistenceTestBase.cs | 10 +- .../Recoverability/EditMessageTests.cs | 10 +- .../RetryConfirmationProcessorTests.cs | 2 +- .../ReturnToSenderDequeuerTests.cs | 75 +------- .../RetryStateTests.cs | 24 +-- .../IEditFailedMessagesDataStore.cs | 9 + .../IErrorMessageDatastore.cs | 75 -------- .../IFailedErrorImportDataStore.cs | 1 + .../IFailedMessageLifecycleDataStore.cs | 19 ++ .../IFailedMessageQueryDataStore.cs | 20 +++ .../IFailedMessageRetryDataStore.cs | 13 ++ .../IGroupsDataStore.cs | 13 +- .../IMessagesViewDataStore.cs | 17 ++ .../INotificationsDataStore.cs | 9 + .../MessageFailures/ArchiveScopeAuditTests.cs | 2 +- .../AsyncRangeAndQueueAuditTests.cs | 46 ++--- .../EditFailedMessagesControllerAuditTests.cs | 42 +---- .../GetAuditCountsForEndpointApi.cs | 4 +- .../Messages/GetAllMessagesApi.cs | 4 +- .../Messages/GetAllMessagesForEndpointApi.cs | 4 +- .../Messages/MessagesByConversationApi.cs | 4 +- .../CompositeViews/Messages/SearchApi.cs | 4 +- .../Messages/SearchEndpointApi.cs | 4 +- .../EventLog/AuditEventLogWriter.cs | 6 +- .../Api/ArchiveMessagesController.cs | 4 +- .../Api/EditFailedMessagesController.cs | 7 +- .../Api/GetAllErrorsController.cs | 10 +- .../Api/GetErrorByIdController.cs | 6 +- .../Handlers/ArchiveMessageHandler.cs | 6 +- .../LegacyMessageFailureResolvedHandler.cs | 10 +- .../Handlers/MessageFailureResolvedHandler.cs | 6 +- .../UnArchiveMessagesByRangeHandler.cs | 2 +- .../Handlers/UnArchiveMessagesHandler.cs | 2 +- .../Api/NotificationsController.cs | 2 +- .../Email/SendEmailNotificationHandler.cs | 2 +- .../Operations/ErrorIngestion.cs | 2 +- .../Operations/ErrorIngestionFaultPolicy.cs | 4 +- .../API/FailureGroupsController.cs | 2 +- .../Recoverability/Editing/EditHandler.cs | 2 +- .../MessageFailedPublisher.cs | 6 +- .../Retrying/FailedMessageRetryCleaner.cs | 6 +- .../Handlers/PendingRetriesHandler.cs | 8 +- .../Retrying/Handlers/RetriesHandler.cs | 4 +- .../Retrying/Infrastructure/ReturnToSender.cs | 4 +- .../Infrastructure/ReturnToSenderDequeuer.cs | 6 +- 69 files changed, 592 insertions(+), 634 deletions(-) create mode 100644 src/ServiceControl.Persistence.EFCore/Implementation/EditFailedMessagesDataStore.cs delete mode 100644 src/ServiceControl.Persistence.EFCore/Implementation/ErrorMessagesDataStore.cs create mode 100644 src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageLifecycleDataStore.cs create mode 100644 src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageQueryDataStore.cs create mode 100644 src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageRetryDataStore.cs create mode 100644 src/ServiceControl.Persistence.EFCore/Implementation/MessagesViewDataStore.cs create mode 100644 src/ServiceControl.Persistence.EFCore/Implementation/NotificationsDataStore.cs create mode 100644 src/ServiceControl.Persistence.RavenDB/Editing/EditFailedMessagesDataStore.cs create mode 100644 src/ServiceControl.Persistence.RavenDB/Editing/NotificationsDataStore.cs create mode 100644 src/ServiceControl.Persistence/IEditFailedMessagesDataStore.cs delete mode 100644 src/ServiceControl.Persistence/IErrorMessageDatastore.cs create mode 100644 src/ServiceControl.Persistence/IFailedMessageLifecycleDataStore.cs create mode 100644 src/ServiceControl.Persistence/IFailedMessageQueryDataStore.cs create mode 100644 src/ServiceControl.Persistence/IFailedMessageRetryDataStore.cs create mode 100644 src/ServiceControl.Persistence/IMessagesViewDataStore.cs create mode 100644 src/ServiceControl.Persistence/INotificationsDataStore.cs diff --git a/src/ServiceControl.AcceptanceTests/Monitoring/CustomChecks/When_email_notifications_are_enabled.cs b/src/ServiceControl.AcceptanceTests/Monitoring/CustomChecks/When_email_notifications_are_enabled.cs index 35c56ed7d9..20d1e6b30d 100644 --- a/src/ServiceControl.AcceptanceTests/Monitoring/CustomChecks/When_email_notifications_are_enabled.cs +++ b/src/ServiceControl.AcceptanceTests/Monitoring/CustomChecks/When_email_notifications_are_enabled.cs @@ -60,11 +60,11 @@ await Define(c => } } - class SetupNotificationSettings(IErrorMessageDataStore errorMessageDataStore) : IHostedService + class SetupNotificationSettings(INotificationsDataStore notificationsDataStore) : IHostedService { public async Task StartAsync(CancellationToken cancellationToken) { - using var notificationsManager = await errorMessageDataStore.CreateNotificationsManager(); + using var notificationsManager = await notificationsDataStore.CreateNotificationsManager(); var settings = await notificationsManager.LoadSettings(); settings.Email = new EmailNotifications diff --git a/src/ServiceControl.AcceptanceTests/Recoverability/MessageFailures/When_a_retry_fails_to_be_sent.cs b/src/ServiceControl.AcceptanceTests/Recoverability/MessageFailures/When_a_retry_fails_to_be_sent.cs index e0cf79a6fa..80dd123b4f 100644 --- a/src/ServiceControl.AcceptanceTests/Recoverability/MessageFailures/When_a_retry_fails_to_be_sent.cs +++ b/src/ServiceControl.AcceptanceTests/Recoverability/MessageFailures/When_a_retry_fails_to_be_sent.cs @@ -31,7 +31,7 @@ public async Task SubsequentBatchesShouldBeProcessed(CancellationToken cancellat CustomizeHostBuilder = hostBuilder => { - hostBuilder.Services.AddSingleton(provider => new FakeReturnToSender(provider.GetRequiredService(), provider.GetRequiredService())); + hostBuilder.Services.AddSingleton(provider => new FakeReturnToSender(provider.GetRequiredService(), provider.GetRequiredService())); }; await Define() @@ -149,7 +149,7 @@ public class MyContext : ScenarioContext public class MessageThatWillFail : ICommand; - public class FakeReturnToSender(IErrorMessageDataStore errorMessageStore, MyContext myContext) + public class FakeReturnToSender(IFailedMessageRetryDataStore errorMessageStore, MyContext myContext) : ReturnToSender(errorMessageStore, NullLogger.Instance) { public override Task HandleMessage(MessageContext message, IMessageDispatcher sender, string errorQueueTransportAddress, CancellationToken cancellationToken = default) diff --git a/src/ServiceControl.Persistence.EFCore/Abstractions/BasePersistence.cs b/src/ServiceControl.Persistence.EFCore/Abstractions/BasePersistence.cs index 2ead0819d0..cb2a8b91a5 100644 --- a/src/ServiceControl.Persistence.EFCore/Abstractions/BasePersistence.cs +++ b/src/ServiceControl.Persistence.EFCore/Abstractions/BasePersistence.cs @@ -39,7 +39,12 @@ protected static void RegisterDataStores(IServiceCollection services, EFPersiste services.AddSingleton(); services.AddSingleton(); - services.AddSingleton(); + services.AddSingleton(); + services.AddSingleton(); + services.AddSingleton(); + services.AddSingleton(); + services.AddSingleton(); + services.AddSingleton(); services.AddSingleton(); services.AddSingleton(); services.AddSingleton(); diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/EditFailedMessagesDataStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/EditFailedMessagesDataStore.cs new file mode 100644 index 0000000000..5b37f6b540 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Implementation/EditFailedMessagesDataStore.cs @@ -0,0 +1,7 @@ +namespace ServiceControl.Persistence.EFCore.Implementation; + +public class EditFailedMessagesDataStore : IEditFailedMessagesDataStore +{ + public Task CreateEditFailedMessageManager() => + throw new NotImplementedException(); +} diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/ErrorMessagesDataStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/ErrorMessagesDataStore.cs deleted file mode 100644 index 0b0930a5bf..0000000000 --- a/src/ServiceControl.Persistence.EFCore/Implementation/ErrorMessagesDataStore.cs +++ /dev/null @@ -1,163 +0,0 @@ -namespace ServiceControl.Persistence.EFCore.Implementation; - -using System.Text.Json; -using Entities; -using Microsoft.Extensions.DependencyInjection; -using NServiceBus; -using Persistence.UnitOfWork; -using ServiceControl.CompositeViews.Messages; -using ServiceControl.EventLog; -using ServiceControl.MessageFailures; -using ServiceControl.MessageFailures.Api; -using ServiceControl.Operations; -using ServiceControl.Persistence.EFCore.Abstractions; -using ServiceControl.Persistence.EFCore.Implementation.UnitOfWork; -using ServiceControl.Persistence.EFCore.Infrastructure; -using ServiceControl.Persistence.Infrastructure; -using ServiceControl.Recoverability; - -public class ErrorMessagesDataStore( - IServiceScopeFactory scopeFactory, - IBodyStoragePersistence bodyStorage, - BodyStorageSettings bodyStorageSettings, - TimeProvider timeProvider) : DataStoreBase(scopeFactory), IErrorMessageDataStore -{ - public Task>> GetAllMessages(PagingInfo pagingInfo, SortInfo sortInfo, bool includeSystemMessages, DateTimeRange? timeSentRange = null) => - throw new NotImplementedException(); - - public Task>> GetAllMessagesForEndpoint(string endpointName, PagingInfo pagingInfo, SortInfo sortInfo, bool includeSystemMessages, DateTimeRange? timeSentRange = null) => - throw new NotImplementedException(); - - public Task>> GetAllMessagesByConversation(string conversationId, PagingInfo pagingInfo, SortInfo sortInfo, bool includeSystemMessages) => - throw new NotImplementedException(); - - public Task>> GetAllMessagesForSearch(string searchTerms, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange? timeSentRange = null) => - throw new NotImplementedException(); - - public Task>> SearchEndpointMessages(string endpointName, string searchKeyword, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange? timeSentRange = null) => - throw new NotImplementedException(); - - // must set StatusChangedAt + LastModified - public Task FailedMessageMarkAsArchived(string failedMessageId) => - throw new NotImplementedException(); - - public Task FailedMessagesFetch(Guid[] ids) => - throw new NotImplementedException(); - - // Update-first, then insert. The dedupe key is deterministic, so a repeat failure updates the - // existing row and concurrent writers that both miss it race only on the insert. The loser of - // that race confirms the row is now present (the winner stored the same logical failure) and - // otherwise rethrows, so the caller never treats a message as stored when it is not. - public Task StoreFailedErrorImport(FailedErrorImport failure) => - ExecuteWithDbContext(async dbContext => - { - var uniqueMessageId = FailedErrorImport.DeriveKey(failure.Message.Headers, failure.Message.Id); - var body = failure.Message.Body ?? []; - var storeExternally = body.Length > bodyStorageSettings.MaxBodySizeToStore; - - if (storeExternally) - { - var contentType = failure.Message.Headers.GetValueOrDefault(Headers.ContentType) ?? "application/octet-stream"; - await bodyStorage.WriteBody(FailedErrorImportEntity.ExternalBodyId(uniqueMessageId), body, contentType); - } - - var failedAt = timeProvider.GetUtcNow().UtcDateTime; - var headersJson = JsonSerializer.Serialize(failure.Message.Headers, HeadersJsonContext.Default.DictionaryStringString); - byte[] storedBody = storeExternally ? [] : body; - - await dbContext.UpsertAsync([uniqueMessageId], () => new FailedErrorImportEntity - { - UniqueMessageId = uniqueMessageId, - FailedAt = failedAt, - MessageId = failure.Message.Id, - HeadersJson = headersJson, - Body = storedBody, - BodyStoredExternally = storeExternally, - ExceptionInfo = failure.ExceptionInfo - }, (entity) => - { - entity.FailedAt = failedAt; - entity.MessageId = failure.Message.Id; - entity.HeadersJson = headersJson; - entity.Body = storedBody; - entity.BodyStoredExternally = storeExternally; - entity.ExceptionInfo = failure.ExceptionInfo; - }); - }); - - public Task CreateEditFailedMessageManager() => - throw new NotImplementedException(); - - public Task> GetFailureGroupView(string groupId, string status, string modified) => - throw new NotImplementedException(); - - public Task> GetFailureGroupsByClassifier(string classifier) => - throw new NotImplementedException(); - - public Task>> ErrorGet(string status, string modified, string queueAddress, PagingInfo pagingInfo, SortInfo sortInfo) => - throw new NotImplementedException(); - - public Task ErrorsHead(string status, string modified, string queueAddress) => - throw new NotImplementedException(); - - public Task>> ErrorsByEndpointName(string status, string endpointName, string modified, PagingInfo pagingInfo, SortInfo sortInfo) => - throw new NotImplementedException(); - - public Task> ErrorsSummary() => - throw new NotImplementedException(); - - public Task ErrorLastBy(string failedMessageId) => - throw new NotImplementedException(); - - public Task ErrorBy(string failedMessageId) => - throw new NotImplementedException(); - - public Task CreateNotificationsManager() => - throw new NotImplementedException(); - - public Task EditComment(string groupId, string comment) => - throw new NotImplementedException(); - - public Task DeleteComment(string groupId) => - throw new NotImplementedException(); - - public Task>> GetGroupErrors(string groupId, string status, string modified, SortInfo sortInfo, PagingInfo pagingInfo) => - throw new NotImplementedException(); - - public Task GetGroupErrorsCount(string groupId, string status, string modified) => - throw new NotImplementedException(); - - public Task>> GetGroup(string groupId, string status, string modified) => - throw new NotImplementedException(); - - // must set StatusChangedAt + LastModified - public Task MarkMessageAsResolved(string failedMessageId) => - throw new NotImplementedException(); - - public Task ProcessPendingRetries(DateTime periodFrom, DateTime periodTo, string queueAddress, Func processCallback) => - throw new NotImplementedException(); - - // must set StatusChangedAt + LastModified - public Task UnArchiveMessagesByRange(DateTime from, DateTime to) => - throw new NotImplementedException(); - - // must set StatusChangedAt + LastModified - public Task UnArchiveMessages(IEnumerable failedMessageIds) => - throw new NotImplementedException(); - - // must set StatusChangedAt + LastModified - public Task RevertRetry(string messageUniqueId) => - throw new NotImplementedException(); - - public Task RemoveFailedMessageRetryDocument(string uniqueMessageId) => - throw new NotImplementedException(); - - public Task GetRetryPendingMessages(DateTime from, DateTime to, string queueAddress) => - throw new NotImplementedException(); - - public Task FetchFromFailedMessage(string uniqueMessageId) => - throw new NotImplementedException(); - - public Task StoreEventLogItem(EventLogItem logItem) => - throw new NotImplementedException(); -} diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/FailedErrorImportDataStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/FailedErrorImportDataStore.cs index 26cc3be1a6..3c0d9f6065 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/FailedErrorImportDataStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/FailedErrorImportDataStore.cs @@ -5,7 +5,9 @@ namespace ServiceControl.Persistence.EFCore.Implementation; using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; +using NServiceBus; using ServiceControl.Operations; +using ServiceControl.Persistence.EFCore.Abstractions; using ServiceControl.Persistence.EFCore.DbContexts; using ServiceControl.Persistence.EFCore.Entities; using ServiceControl.Persistence.EFCore.Implementation.UnitOfWork; @@ -14,6 +16,8 @@ namespace ServiceControl.Persistence.EFCore.Implementation; public class FailedErrorImportDataStore( IServiceScopeFactory scopeFactory, IBodyStoragePersistence bodyStorage, + BodyStorageSettings bodyStorageSettings, + TimeProvider timeProvider, ILogger logger) : DataStoreBase(scopeFactory), IFailedErrorImportDataStore { const int BatchSize = 100; @@ -21,6 +25,47 @@ public class FailedErrorImportDataStore( public Task QueryContainsFailedImports() => ExecuteWithDbContext(dbContext => dbContext.FailedErrorImports.AsNoTracking().AnyAsync()); + // Update-first, then insert. The dedupe key is deterministic, so a repeat failure updates the + // existing row and concurrent writers that both miss it race only on the insert. The loser of + // that race confirms the row is now present (the winner stored the same logical failure) and + // otherwise rethrows, so the caller never treats a message as stored when it is not. + public Task StoreFailedErrorImport(FailedErrorImport failure) => + ExecuteWithDbContext(async dbContext => + { + var uniqueMessageId = FailedErrorImport.DeriveKey(failure.Message.Headers, failure.Message.Id); + var body = failure.Message.Body ?? []; + var storeExternally = body.Length > bodyStorageSettings.MaxBodySizeToStore; + + if (storeExternally) + { + var contentType = failure.Message.Headers.GetValueOrDefault(Headers.ContentType) ?? "application/octet-stream"; + await bodyStorage.WriteBody(FailedErrorImportEntity.ExternalBodyId(uniqueMessageId), body, contentType); + } + + var failedAt = timeProvider.GetUtcNow().UtcDateTime; + var headersJson = JsonSerializer.Serialize(failure.Message.Headers, HeadersJsonContext.Default.DictionaryStringString); + byte[] storedBody = storeExternally ? [] : body; + + await dbContext.UpsertAsync([uniqueMessageId], () => new FailedErrorImportEntity + { + UniqueMessageId = uniqueMessageId, + FailedAt = failedAt, + MessageId = failure.Message.Id, + HeadersJson = headersJson, + Body = storedBody, + BodyStoredExternally = storeExternally, + ExceptionInfo = failure.ExceptionInfo + }, (entity) => + { + entity.FailedAt = failedAt; + entity.MessageId = failure.Message.Id; + entity.HeadersJson = headersJson; + entity.Body = storedBody; + entity.BodyStoredExternally = storeExternally; + entity.ExceptionInfo = failure.ExceptionInfo; + }); + }); + // Replays oldest-first. Successful imports delete their row; failures are left in place, so the // count of failures so far is exactly the offset to the next unseen row. This walks the whole // set once without retrying a failure within the same run. diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageLifecycleDataStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageLifecycleDataStore.cs new file mode 100644 index 0000000000..07f3b724d7 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageLifecycleDataStore.cs @@ -0,0 +1,24 @@ +namespace ServiceControl.Persistence.EFCore.Implementation; + +using Microsoft.Extensions.DependencyInjection; + +/// +/// Every operation here has to update both StatusChangedAt and LastModified. +/// +public class FailedMessageLifecycleDataStore(IServiceScopeFactory scopeFactory) : DataStoreBase(scopeFactory), IFailedMessageLifecycleDataStore +{ + public Task MarkAsArchived(string failedMessageId) => + throw new NotImplementedException(); + + public Task MarkAsResolved(string failedMessageId) => + throw new NotImplementedException(); + + public Task UnArchiveMessages(IEnumerable failedMessageIds) => + throw new NotImplementedException(); + + public Task UnArchiveMessagesByRange(DateTime from, DateTime to) => + throw new NotImplementedException(); + + public Task RevertRetry(string messageUniqueId) => + throw new NotImplementedException(); +} diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageQueryDataStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageQueryDataStore.cs new file mode 100644 index 0000000000..182882aa97 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageQueryDataStore.cs @@ -0,0 +1,30 @@ +namespace ServiceControl.Persistence.EFCore.Implementation; + +using Microsoft.Extensions.DependencyInjection; +using ServiceControl.MessageFailures; +using ServiceControl.MessageFailures.Api; +using ServiceControl.Persistence.Infrastructure; + +public class FailedMessageQueryDataStore(IServiceScopeFactory scopeFactory) : DataStoreBase(scopeFactory), IFailedMessageQueryDataStore +{ + public Task>> GetFailedMessages(string status, string modified, string queueAddress, PagingInfo pagingInfo, SortInfo sortInfo) => + throw new NotImplementedException(); + + public Task GetFailedMessagesStats(string status, string modified, string queueAddress) => + throw new NotImplementedException(); + + public Task>> GetFailedMessagesByEndpoint(string status, string endpointName, string modified, PagingInfo pagingInfo, SortInfo sortInfo) => + throw new NotImplementedException(); + + public Task> GetFailedMessagesSummary() => + throw new NotImplementedException(); + + public Task GetLatestFailedMessageView(string failedMessageId) => + throw new NotImplementedException(); + + public Task GetFailedMessage(string failedMessageId) => + throw new NotImplementedException(); + + public Task GetFailedMessagesByIds(Guid[] ids) => + throw new NotImplementedException(); +} diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageRetryDataStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageRetryDataStore.cs new file mode 100644 index 0000000000..002f365279 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageRetryDataStore.cs @@ -0,0 +1,18 @@ +namespace ServiceControl.Persistence.EFCore.Implementation; + +using Microsoft.Extensions.DependencyInjection; + +public class FailedMessageRetryDataStore(IServiceScopeFactory scopeFactory) : DataStoreBase(scopeFactory), IFailedMessageRetryDataStore +{ + public Task ProcessPendingRetries(DateTime periodFrom, DateTime periodTo, string queueAddress, Func processCallback) => + throw new NotImplementedException(); + + public Task GetRetryPendingMessages(DateTime from, DateTime to, string queueAddress) => + throw new NotImplementedException(); + + public Task RemoveFailedMessageRetry(string uniqueMessageId) => + throw new NotImplementedException(); + + public Task GetFailedMessageBody(string uniqueMessageId) => + throw new NotImplementedException(); +} diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/GroupsDataStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/GroupsDataStore.cs index 41b18e12df..0ef4ea6a5c 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/GroupsDataStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/GroupsDataStore.cs @@ -1,5 +1,7 @@ namespace ServiceControl.Persistence.EFCore.Implementation; +using ServiceControl.MessageFailures.Api; +using ServiceControl.Persistence.Infrastructure; using ServiceControl.Recoverability; public class GroupsDataStore : IGroupsDataStore @@ -7,6 +9,27 @@ public class GroupsDataStore : IGroupsDataStore public Task> GetFailureGroupsByClassifier(string classifier, string classifierFilter) => throw new NotImplementedException(); + public Task> GetArchivedFailureGroupsByClassifier(string classifier) => + throw new NotImplementedException(); + public Task GetCurrentForwardingBatch() => throw new NotImplementedException(); + + public Task>> GetGroup(string groupId, string status, string modified) => + throw new NotImplementedException(); + + public Task> GetFailureGroupView(string groupId, string status, string modified) => + throw new NotImplementedException(); + + public Task>> GetGroupErrors(string groupId, string status, string modified, SortInfo sortInfo, PagingInfo pagingInfo) => + throw new NotImplementedException(); + + public Task GetGroupErrorsCount(string groupId, string status, string modified) => + throw new NotImplementedException(); + + public Task EditComment(string groupId, string comment) => + throw new NotImplementedException(); + + public Task DeleteComment(string groupId) => + throw new NotImplementedException(); } diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/MessagesViewDataStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/MessagesViewDataStore.cs new file mode 100644 index 0000000000..c8439b1eca --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Implementation/MessagesViewDataStore.cs @@ -0,0 +1,22 @@ +namespace ServiceControl.Persistence.EFCore.Implementation; + +using ServiceControl.CompositeViews.Messages; +using ServiceControl.Persistence.Infrastructure; + +public class MessagesViewDataStore : IMessagesViewDataStore +{ + public Task>> GetAllMessages(PagingInfo pagingInfo, SortInfo sortInfo, bool includeSystemMessages, DateTimeRange? timeSentRange = null) => + throw new NotImplementedException(); + + public Task>> GetAllMessagesForEndpoint(string endpointName, PagingInfo pagingInfo, SortInfo sortInfo, bool includeSystemMessages, DateTimeRange? timeSentRange = null) => + throw new NotImplementedException(); + + public Task>> GetAllMessagesByConversation(string conversationId, PagingInfo pagingInfo, SortInfo sortInfo, bool includeSystemMessages) => + throw new NotImplementedException(); + + public Task>> GetAllMessagesForSearch(string searchTerms, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange? timeSentRange = null) => + throw new NotImplementedException(); + + public Task>> SearchEndpointMessages(string endpointName, string searchKeyword, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange? timeSentRange = null) => + throw new NotImplementedException(); +} diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/NotificationsDataStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/NotificationsDataStore.cs new file mode 100644 index 0000000000..c89c2c5666 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Implementation/NotificationsDataStore.cs @@ -0,0 +1,7 @@ +namespace ServiceControl.Persistence.EFCore.Implementation; + +public class NotificationsDataStore : INotificationsDataStore +{ + public Task CreateNotificationsManager() => + throw new NotImplementedException(); +} diff --git a/src/ServiceControl.Persistence.RavenDB/Editing/EditFailedMessagesDataStore.cs b/src/ServiceControl.Persistence.RavenDB/Editing/EditFailedMessagesDataStore.cs new file mode 100644 index 0000000000..5533487862 --- /dev/null +++ b/src/ServiceControl.Persistence.RavenDB/Editing/EditFailedMessagesDataStore.cs @@ -0,0 +1,11 @@ +namespace ServiceControl.Persistence.RavenDB.Editing +{ + using System.Threading.Tasks; + + class EditFailedMessagesDataStore(IRavenSessionProvider sessionProvider, ExpirationManager expirationManager) : IEditFailedMessagesDataStore + { + public async Task CreateEditFailedMessageManager() => + // the edit failed message manager manages the lifetime of the session + new EditFailedMessageManager(await sessionProvider.OpenSession(), expirationManager); + } +} diff --git a/src/ServiceControl.Persistence.RavenDB/Editing/NotificationsDataStore.cs b/src/ServiceControl.Persistence.RavenDB/Editing/NotificationsDataStore.cs new file mode 100644 index 0000000000..9e3c2f2fcc --- /dev/null +++ b/src/ServiceControl.Persistence.RavenDB/Editing/NotificationsDataStore.cs @@ -0,0 +1,11 @@ +namespace ServiceControl.Persistence.RavenDB.Editing +{ + using System.Threading.Tasks; + + class NotificationsDataStore(IRavenSessionProvider sessionProvider) : INotificationsDataStore + { + public async Task CreateNotificationsManager() => + // the notifications manager manages the lifetime of the session + new NotificationsManager(await sessionProvider.OpenSession()); + } +} diff --git a/src/ServiceControl.Persistence.RavenDB/ErrorMessagesDataStore.cs b/src/ServiceControl.Persistence.RavenDB/ErrorMessagesDataStore.cs index cc3f1b4fe5..91cd3f481d 100644 --- a/src/ServiceControl.Persistence.RavenDB/ErrorMessagesDataStore.cs +++ b/src/ServiceControl.Persistence.RavenDB/ErrorMessagesDataStore.cs @@ -31,7 +31,7 @@ class ErrorMessagesDataStore( IBodyStorage bodyStorage, ExpirationManager expirationManager, ILogger logger) - : IErrorMessageDataStore + : IMessagesViewDataStore, IFailedMessageQueryDataStore, IFailedMessageLifecycleDataStore, IFailedMessageRetryDataStore { public async Task>> GetAllMessages( PagingInfo pagingInfo, @@ -147,7 +147,7 @@ DateTimeRange timeSentRange return new QueryResult>(results, stats.ToQueryStatsInfo()); } - public async Task FailedMessageMarkAsArchived(string failedMessageId) + public async Task MarkAsArchived(string failedMessageId) { using var session = await sessionProvider.OpenSession(); var failedMessage = await session.LoadAsync(FailedMessageIdGenerator.MakeDocumentId(failedMessageId)); @@ -162,7 +162,7 @@ public async Task FailedMessageMarkAsArchived(string failedMessageId) await session.SaveChangesAsync(); } - public async Task FailedMessagesFetch(Guid[] ids) + public async Task GetFailedMessagesByIds(Guid[] ids) { using var session = await sessionProvider.OpenSession(); var docIds = ids.Select(g => FailedMessageIdGenerator.MakeDocumentId(g.ToString())); @@ -170,54 +170,7 @@ public async Task FailedMessagesFetch(Guid[] ids) return results.Values.Where(x => x != null).ToArray(); } - public async Task StoreFailedErrorImport(FailedErrorImport failure) - { - using var session = await sessionProvider.OpenSession(); - // This object's ID is generated externally, but is not in the RavenDB format - // Check that's true to make sure that if it already is that it doesn't get double-formatted - if (!failure.Id.StartsWith(CollectionName)) - { - failure.Id = MakeDocumentId(failure.Id); - } - await session.StoreAsync(failure); - - await session.SaveChangesAsync(); - } - - public async Task CreateEditFailedMessageManager() => - // the edit failed message manager manages the lifetime of the session - new EditFailedMessageManager(await sessionProvider.OpenSession(), expirationManager); - - public async Task> GetFailureGroupView(string groupId, string status, string modified) - { - using var session = await sessionProvider.OpenSession(); - var document = await session.Advanced - .AsyncDocumentQuery() - .Statistics(out var stats) - .WhereEquals(group => group.Id, groupId) - .FilterByStatusWhere(status) - .FilterByLastModifiedRange(modified) - .FirstOrDefaultAsync(); - - return new QueryResult(document, stats.ToQueryStatsInfo()); - } - - public async Task> GetFailureGroupsByClassifier(string classifier) - { - using var session = await sessionProvider.OpenSession(); - var groups = session - .Query() - .Where(v => v.Type == classifier); - - var results = await groups - .OrderByDescending(x => x.Last) - .Take(200) // only show 200 groups - .ToListAsync(); - - return results; - } - - public async Task>> ErrorGet( + public async Task>> GetFailedMessages( string status, string modified, string queueAddress, @@ -244,7 +197,7 @@ SortInfo sortInfo return new QueryResult>(results, stats.ToQueryStatsInfo()); } - public async Task ErrorsHead( + public async Task GetFailedMessagesStats( string status, string modified, string queueAddress @@ -261,7 +214,7 @@ string queueAddress return stats.ToQueryStatsInfo(); } - public async Task>> ErrorsByEndpointName( + public async Task>> GetFailedMessagesByEndpoint( string status, string endpointName, string modified, @@ -289,7 +242,7 @@ SortInfo sortInfo return new QueryResult>(results, stats.ToQueryStatsInfo()); } - public async Task> ErrorsSummary() + public async Task> GetFailedMessagesSummary() { using var session = await sessionProvider.OpenSession(); var facetResults = await session.Query() @@ -321,20 +274,14 @@ public async Task> ErrorsSummary() return results; } - public Task ErrorBy(string failedMessageId) => ErrorByDocumentId(FailedMessageIdGenerator.MakeDocumentId(failedMessageId)); - - async Task ErrorByDocumentId(string documentId) + public async Task GetFailedMessage(string failedMessageId) { using var session = await sessionProvider.OpenSession(); - var message = await session.LoadAsync(documentId); + var message = await session.LoadAsync(FailedMessageIdGenerator.MakeDocumentId(failedMessageId)); return message; } - public async Task CreateNotificationsManager() => - // the notifications manager manages the lifetime of the session - new NotificationsManager(await sessionProvider.OpenSession()); - - public async Task ErrorLastBy(string failedMessageId) + public async Task GetLatestFailedMessageView(string failedMessageId) { using var session = await sessionProvider.OpenSession(); var message = await session.LoadAsync(FailedMessageIdGenerator.MakeDocumentId(failedMessageId)); @@ -395,81 +342,7 @@ FailedMessageView Map(FailedMessage message, IAsyncDocumentSession session) } - public async Task EditComment(string groupId, string comment) - { - using var session = await sessionProvider.OpenSession(); - var groupComment = - await session.LoadAsync(GroupsDataStore.MakeId(groupId)) - ?? new GroupComment { Id = GroupsDataStore.MakeId(groupId) }; - - groupComment.Comment = comment; - - await session.StoreAsync(groupComment); - await session.SaveChangesAsync(); - } - - public async Task DeleteComment(string groupId) - { - using var session = await sessionProvider.OpenSession(); - session.Delete(GroupsDataStore.MakeId(groupId)); - await session.SaveChangesAsync(); - } - - public async Task>> GetGroupErrors( - string groupId, - string status, - string modified, - SortInfo sortInfo, - PagingInfo pagingInfo - ) - { - using var session = await sessionProvider.OpenSession(); - var query = session.Advanced - .AsyncDocumentQuery() - .Statistics(out var stats) - .WhereEquals(view => view.FailureGroupId, groupId) - .FilterByStatusWhere(status) - .FilterByLastModifiedRange(modified) - .Sort(sortInfo) - .Paging(pagingInfo) - .SelectFields() - .ToQueryable() - .TransformToFailedMessageView(); - - var results = await query - .ToListAsync(); - - return results.ToQueryResult(stats); - } - - public async Task GetGroupErrorsCount(string groupId, string status, string modified) - { - using var session = await sessionProvider.OpenSession(); - var queryResult = await session.Advanced - .AsyncDocumentQuery() - .WhereEquals(view => view.FailureGroupId, groupId) - .FilterByStatusWhere(status) - .FilterByLastModifiedRange(modified) - .GetQueryResultAsync(); - - return queryResult.ToQueryStatsInfo(); - } - - public async Task>> GetGroup(string groupId, string status, string modified) - { - using var session = await sessionProvider.OpenSession(); - var queryResult = await session.Advanced - .AsyncDocumentQuery() - .Statistics(out var stats) - .WhereEquals(group => group.Id, groupId) - .FilterByStatusWhere(status) - .FilterByLastModifiedRange(modified) - .ToListAsync(); - - return queryResult.ToQueryResult(stats); - } - - public async Task MarkMessageAsResolved(string failedMessageId) + public async Task MarkAsResolved(string failedMessageId) { var documentId = FailedMessageIdGenerator.MakeDocumentId(failedMessageId); @@ -606,7 +479,7 @@ public async Task RevertRetry(string messageUniqueId) await session.SaveChangesAsync(); } - public async Task RemoveFailedMessageRetryDocument(string uniqueMessageId) + public async Task RemoveFailedMessageRetry(string uniqueMessageId) { using var session = await sessionProvider.OpenSession(); await session.Advanced.RequestExecutor.ExecuteAsync(new DeleteDocumentCommand(RetryDocumentDataStore.MakeFailedMessageRetriesDocumentId(uniqueMessageId), null), session.Advanced.Context); @@ -632,7 +505,7 @@ public async Task GetRetryPendingMessages(DateTime from, DateTime to, record struct FailedMessageProjection(string UniqueMessageId); - public async Task FetchFromFailedMessage(string uniqueMessageId) + public async Task GetFailedMessageBody(string uniqueMessageId) { byte[] body = null; var result = await bodyStorage.TryFetch(uniqueMessageId) @@ -655,18 +528,5 @@ public async Task FetchFromFailedMessage(string uniqueMessageId) } return body; } - - public async Task StoreEventLogItem(EventLogItem logItem) - { - using var session = await sessionProvider.OpenSession(); - await session.StoreAsync(logItem); - - expirationManager.EnableExpiration(session, logItem); - - await session.SaveChangesAsync(); - } - - public static string MakeDocumentId(string id) => string.Join("/", CollectionName, id); - public const string CollectionName = "FailedErrorImports"; } } diff --git a/src/ServiceControl.Persistence.RavenDB/EventLogDataStore.cs b/src/ServiceControl.Persistence.RavenDB/EventLogDataStore.cs index f11461c94e..d3e2f5d850 100644 --- a/src/ServiceControl.Persistence.RavenDB/EventLogDataStore.cs +++ b/src/ServiceControl.Persistence.RavenDB/EventLogDataStore.cs @@ -6,12 +6,15 @@ using Persistence.Infrastructure; using Raven.Client.Documents; - class EventLogDataStore(IRavenSessionProvider sessionProvider) : IEventLogDataStore + class EventLogDataStore(IRavenSessionProvider sessionProvider, ExpirationManager expirationManager) : IEventLogDataStore { public async Task Add(EventLogItem logItem) { using var session = await sessionProvider.OpenSession(); await session.StoreAsync(logItem); + + expirationManager.EnableExpiration(session, logItem); + await session.SaveChangesAsync(); } diff --git a/src/ServiceControl.Persistence.RavenDB/FailedErrorImportDataStore.cs b/src/ServiceControl.Persistence.RavenDB/FailedErrorImportDataStore.cs index 10d63a5a1c..7b04debfe7 100644 --- a/src/ServiceControl.Persistence.RavenDB/FailedErrorImportDataStore.cs +++ b/src/ServiceControl.Persistence.RavenDB/FailedErrorImportDataStore.cs @@ -9,6 +9,20 @@ class FailedErrorImportDataStore(IRavenSessionProvider sessionProvider, ILogger logger) : IFailedErrorImportDataStore { + public async Task StoreFailedErrorImport(FailedErrorImport failure) + { + using var session = await sessionProvider.OpenSession(); + // This object's ID is generated externally, but is not in the RavenDB format + // Check that's true to make sure that if it already is that it doesn't get double-formatted + if (!failure.Id.StartsWith(CollectionName)) + { + failure.Id = MakeDocumentId(failure.Id); + } + await session.StoreAsync(failure); + + await session.SaveChangesAsync(); + } + public async Task ProcessFailedErrorImports(Func processMessage, CancellationToken cancellationToken) { var succeeded = 0; @@ -57,5 +71,8 @@ public async Task QueryContainsFailedImports() await using var ie = await session.Advanced.StreamAsync(query); return await ie.MoveNextAsync(); } + + public static string MakeDocumentId(string id) => string.Join("/", CollectionName, id); + public const string CollectionName = "FailedErrorImports"; } } \ No newline at end of file diff --git a/src/ServiceControl.Persistence.RavenDB/RavenPersistence.cs b/src/ServiceControl.Persistence.RavenDB/RavenPersistence.cs index 93a8fe63b7..e14b0421e4 100644 --- a/src/ServiceControl.Persistence.RavenDB/RavenPersistence.cs +++ b/src/ServiceControl.Persistence.RavenDB/RavenPersistence.cs @@ -1,6 +1,7 @@ namespace ServiceControl.Persistence.RavenDB; using CustomChecks; +using Editing; using MessageFailures; using MessageRedirects; using Microsoft.Extensions.DependencyInjection; @@ -56,7 +57,13 @@ public void AddPersistence(IServiceCollection services) services.AddSingleton(); services.AddSingleton(); - services.AddSingleton(); + services.AddSingleton(); + services.AddSingleton(p => p.GetRequiredService()); + services.AddSingleton(p => p.GetRequiredService()); + services.AddSingleton(p => p.GetRequiredService()); + services.AddSingleton(p => p.GetRequiredService()); + services.AddSingleton(); + services.AddSingleton(); services.AddSingleton(); services.AddSingleton(); services.AddSingleton(); diff --git a/src/ServiceControl.Persistence.RavenDB/Recoverability/GroupsDataStore.cs b/src/ServiceControl.Persistence.RavenDB/Recoverability/GroupsDataStore.cs index 6ae5c7879f..d517b37a83 100644 --- a/src/ServiceControl.Persistence.RavenDB/Recoverability/GroupsDataStore.cs +++ b/src/ServiceControl.Persistence.RavenDB/Recoverability/GroupsDataStore.cs @@ -1,4 +1,4 @@ -namespace ServiceControl.Persistence.RavenDB.Recoverability +namespace ServiceControl.Persistence.RavenDB.Recoverability { using System.Collections.Generic; using System.Linq; @@ -6,7 +6,10 @@ using System.Threading.Tasks; using Raven.Client.Documents; using Raven.Client.Documents.Linq; + using Raven.Client.Documents.Session; using ServiceControl.MessageFailures; + using ServiceControl.MessageFailures.Api; + using ServiceControl.Persistence.Infrastructure; using ServiceControl.Recoverability; class GroupsDataStore(IRavenSessionProvider sessionProvider) : IGroupsDataStore @@ -37,6 +40,21 @@ public async Task> GetFailureGroupsByClassifier(string c return groups; } + public async Task> GetArchivedFailureGroupsByClassifier(string classifier) + { + using var session = await sessionProvider.OpenSession(); + var groups = session + .Query() + .Where(v => v.Type == classifier); + + var results = await groups + .OrderByDescending(x => x.Last) + .Take(200) // only show 200 groups + .ToListAsync(); + + return results; + } + public async Task GetCurrentForwardingBatch() { using var session = await sessionProvider.OpenSession(); @@ -46,6 +64,94 @@ public async Task GetCurrentForwardingBatch() return nowForwarding == null ? null : await session.LoadAsync(nowForwarding.RetryBatchId); } + public async Task>> GetGroup(string groupId, string status, string modified) + { + using var session = await sessionProvider.OpenSession(); + var queryResult = await session.Advanced + .AsyncDocumentQuery() + .Statistics(out var stats) + .WhereEquals(group => group.Id, groupId) + .FilterByStatusWhere(status) + .FilterByLastModifiedRange(modified) + .ToListAsync(); + + return queryResult.ToQueryResult(stats); + } + + public async Task> GetFailureGroupView(string groupId, string status, string modified) + { + using var session = await sessionProvider.OpenSession(); + var document = await session.Advanced + .AsyncDocumentQuery() + .Statistics(out var stats) + .WhereEquals(group => group.Id, groupId) + .FilterByStatusWhere(status) + .FilterByLastModifiedRange(modified) + .FirstOrDefaultAsync(); + + return new QueryResult(document, stats.ToQueryStatsInfo()); + } + + public async Task>> GetGroupErrors( + string groupId, + string status, + string modified, + SortInfo sortInfo, + PagingInfo pagingInfo + ) + { + using var session = await sessionProvider.OpenSession(); + var query = session.Advanced + .AsyncDocumentQuery() + .Statistics(out var stats) + .WhereEquals(view => view.FailureGroupId, groupId) + .FilterByStatusWhere(status) + .FilterByLastModifiedRange(modified) + .Sort(sortInfo) + .Paging(pagingInfo) + .SelectFields() + .ToQueryable() + .TransformToFailedMessageView(); + + var results = await query + .ToListAsync(); + + return results.ToQueryResult(stats); + } + + public async Task GetGroupErrorsCount(string groupId, string status, string modified) + { + using var session = await sessionProvider.OpenSession(); + var queryResult = await session.Advanced + .AsyncDocumentQuery() + .WhereEquals(view => view.FailureGroupId, groupId) + .FilterByStatusWhere(status) + .FilterByLastModifiedRange(modified) + .GetQueryResultAsync(); + + return queryResult.ToQueryStatsInfo(); + } + + public async Task EditComment(string groupId, string comment) + { + using var session = await sessionProvider.OpenSession(); + var groupComment = + await session.LoadAsync(MakeId(groupId)) + ?? new GroupComment { Id = MakeId(groupId) }; + + groupComment.Comment = comment; + + await session.StoreAsync(groupComment); + await session.SaveChangesAsync(); + } + + public async Task DeleteComment(string groupId) + { + using var session = await sessionProvider.OpenSession(); + session.Delete(MakeId(groupId)); + await session.SaveChangesAsync(); + } + public static string MakeId(string groupId) => $"GroupComment/{groupId}"; } } diff --git a/src/ServiceControl.Persistence.Tests.RavenDB/Expiration/MessageExpiryTests.cs b/src/ServiceControl.Persistence.Tests.RavenDB/Expiration/MessageExpiryTests.cs index a0c3f8f080..b28475e17f 100644 --- a/src/ServiceControl.Persistence.Tests.RavenDB/Expiration/MessageExpiryTests.cs +++ b/src/ServiceControl.Persistence.Tests.RavenDB/Expiration/MessageExpiryTests.cs @@ -47,12 +47,12 @@ public async Task SingleMessageMarkedAsArchiveShouldExpire() Assert.That(error.Results, Has.Count.EqualTo(1), "Failed message should be available to query after ingestion"); - await ErrorStore.FailedMessageMarkAsArchived(error.Results.First().Id); + await FailedMessageLifecycleStore.MarkAsArchived(error.Results.First().Id); await WaitUntil(async () => (await GetAllMessages()).Results.Count == 0, "Archived message should be removed after archiving."); } - async Task>> GetAllMessages() => await ErrorStore.GetAllMessages(new PagingInfo(1, 10), new SortInfo(null, null), false); + async Task>> GetAllMessages() => await MessagesViewStore.GetAllMessages(new PagingInfo(1, 10), new SortInfo(null, null), false); [Test] public async Task AllMessagesInUnArchivedGroupShouldNotExpire() @@ -136,7 +136,7 @@ public async Task SingleMessageMarkedAsResolvedShouldExpire() Assert.That(errors.Results, Has.Count.EqualTo(1), "Failed message should be available to query after ingestion"); - await ErrorStore.MarkMessageAsResolved(errors.Results.First().Id); + await FailedMessageLifecycleStore.MarkAsResolved(errors.Results.First().Id); await WaitUntil(async () => (await GetAllMessages()).Results.Count == 0, "Archived message should be removed after archiving."); } @@ -246,7 +246,7 @@ public async Task EventLogItemShouldExpire() { await DisableExpiration(); - await ErrorStore.StoreEventLogItem(new EventLogItem()); + await EventLogDataStore.Add(new EventLogItem()); await CompleteDatabaseOperation(); diff --git a/src/ServiceControl.Persistence.Tests.RavenDB/Operations/FailedErrorImportCustomCheckTests.cs b/src/ServiceControl.Persistence.Tests.RavenDB/Operations/FailedErrorImportCustomCheckTests.cs index 2f778c3ebe..4231f209d7 100644 --- a/src/ServiceControl.Persistence.Tests.RavenDB/Operations/FailedErrorImportCustomCheckTests.cs +++ b/src/ServiceControl.Persistence.Tests.RavenDB/Operations/FailedErrorImportCustomCheckTests.cs @@ -37,7 +37,7 @@ public async Task Fail_if_failed_imports() { await session.StoreAsync(new FailedErrorImport { - Id = ServiceControl.Persistence.RavenDB.ErrorMessagesDataStore.MakeDocumentId(Guid.NewGuid().ToString()) + Id = ServiceControl.Persistence.RavenDB.FailedErrorImportDataStore.MakeDocumentId(Guid.NewGuid().ToString()) }); BlockToInspectDatabase(); diff --git a/src/ServiceControl.Persistence.Tests.RavenDB/Operations/FailedErrorImportDedupeTests.cs b/src/ServiceControl.Persistence.Tests.RavenDB/Operations/FailedErrorImportDedupeTests.cs index 323bd25008..7f9ec7694b 100644 --- a/src/ServiceControl.Persistence.Tests.RavenDB/Operations/FailedErrorImportDedupeTests.cs +++ b/src/ServiceControl.Persistence.Tests.RavenDB/Operations/FailedErrorImportDedupeTests.cs @@ -35,7 +35,7 @@ public async Task Repeated_failure_of_the_same_message_stores_one_document() } Task StoreFailure(IReadOnlyDictionary headers, string exceptionInfo) => - ErrorStore.StoreFailedErrorImport(new FailedErrorImport + FailedImportStore.StoreFailedErrorImport(new FailedErrorImport { Id = FailedErrorImport.DeriveKey(headers, "native-1").ToString(), Message = new FailedTransportMessage diff --git a/src/ServiceControl.Persistence.Tests.RavenDB/Recoverability/ErrorMessageDataStoreTests.cs b/src/ServiceControl.Persistence.Tests.RavenDB/Recoverability/ErrorMessageDataStoreTests.cs index 95bf8d2cc0..e54695a88d 100644 --- a/src/ServiceControl.Persistence.Tests.RavenDB/Recoverability/ErrorMessageDataStoreTests.cs +++ b/src/ServiceControl.Persistence.Tests.RavenDB/Recoverability/ErrorMessageDataStoreTests.cs @@ -15,13 +15,14 @@ [TestFixture] class ErrorMessageDataStoreTests : RavenPersistenceTestBase { - IErrorMessageDataStore store; + IMessagesViewDataStore messagesViewStore; + IFailedMessageQueryDataStore queryStore; FailedMessage processedMessage1, processedMessage2; [Test] public async Task GetAllMessages() { - var result = await store.GetAllMessages(new PagingInfo(1, 50), new SortInfo("", ""), false); + var result = await messagesViewStore.GetAllMessages(new PagingInfo(1, 50), new SortInfo("", ""), false); Assert.That(result.Results, Is.Not.Empty); } @@ -34,7 +35,7 @@ public async Task GetAllMessages() [TestCase("critical_time", "dsc", "b")] public async Task GetAllMessagesForEndpoint(string sort, string direction, string id) { - var result = await store.GetAllMessagesForEndpoint( + var result = await messagesViewStore.GetAllMessagesForEndpoint( "RamonAndTomek", new PagingInfo(1, 1), new SortInfo(sort, direction), @@ -49,7 +50,7 @@ public async Task GetAllMessagesForEndpoint(string sort, string direction, strin [Test] public async Task ErrorGet() { - var result = await store.ErrorGet(null, null, null, new PagingInfo(1, 50), new SortInfo("", "")); + var result = await queryStore.GetFailedMessages(null, null, null, new PagingInfo(1, 50), new SortInfo("", "")); Assert.That(result.Results, Is.Not.Empty); } @@ -61,7 +62,8 @@ public async Task GetStore() await CompleteDatabaseOperation(); - store = ServiceProvider.GetRequiredService(); + messagesViewStore = ServiceProvider.GetRequiredService(); + queryStore = ServiceProvider.GetRequiredService(); } async Task GenerateAndSaveFailedMessage() diff --git a/src/ServiceControl.Persistence.Tests/EFCore/FailedErrorImportTests.cs b/src/ServiceControl.Persistence.Tests/EFCore/FailedErrorImportTests.cs index 54eee84d0b..44b26831c9 100644 --- a/src/ServiceControl.Persistence.Tests/EFCore/FailedErrorImportTests.cs +++ b/src/ServiceControl.Persistence.Tests/EFCore/FailedErrorImportTests.cs @@ -248,13 +248,12 @@ public async Task A_missing_external_body_fails_the_re_import_without_blocking_t Assert.That(secondRun, Is.Empty, "the row with the missing body is retried and fails again"); } - IFailedErrorImportDataStore FailedImportStore => ServiceProvider.GetRequiredService(); Task StoreImport(Dictionary headers, byte[] body, string exceptionInfo = "boom", string nativeId = null) { nativeId ??= Guid.NewGuid().ToString(); - return ErrorStore.StoreFailedErrorImport(new FailedErrorImport + return FailedImportStore.StoreFailedErrorImport(new FailedErrorImport { Id = FailedErrorImport.DeriveKey(headers, nativeId).ToString(), Message = new FailedTransportMessage { Id = nativeId, Headers = headers, Body = body }, diff --git a/src/ServiceControl.Persistence.Tests/PersistenceTestBase.cs b/src/ServiceControl.Persistence.Tests/PersistenceTestBase.cs index dff4fbd232..c3a5ca861b 100644 --- a/src/ServiceControl.Persistence.Tests/PersistenceTestBase.cs +++ b/src/ServiceControl.Persistence.Tests/PersistenceTestBase.cs @@ -93,11 +93,16 @@ protected static async Task WaitUntil(Func> conditionChecker, string throw new Exception($"{condition} has not been meet in defined timespan: {timeout})"); } - protected IErrorMessageDataStore ErrorStore => ServiceProvider.GetRequiredService(); protected IRetryDocumentDataStore RetryStore => ServiceProvider.GetRequiredService(); protected IBodyStorage BodyStorage => ServiceProvider.GetRequiredService(); protected IRetryBatchesDataStore RetryBatchesStore => ServiceProvider.GetRequiredService(); - protected IErrorMessageDataStore ErrorMessageDataStore => ServiceProvider.GetRequiredService(); + protected IFailedMessageQueryDataStore FailedMessageQueryStore => ServiceProvider.GetRequiredService(); + protected IFailedMessageLifecycleDataStore FailedMessageLifecycleStore => ServiceProvider.GetRequiredService(); + protected IFailedMessageRetryDataStore FailedMessageRetryStore => ServiceProvider.GetRequiredService(); + protected IMessagesViewDataStore MessagesViewStore => ServiceProvider.GetRequiredService(); + protected IGroupsDataStore GroupsStore => ServiceProvider.GetRequiredService(); + protected IEditFailedMessagesDataStore EditFailedMessagesStore => ServiceProvider.GetRequiredService(); + protected INotificationsDataStore NotificationsStore => ServiceProvider.GetRequiredService(); protected IMessageRedirectsDataStore MessageRedirectsDataStore => ServiceProvider.GetRequiredService(); protected IMonitoringDataStore MonitoringDataStore => ServiceProvider.GetRequiredService(); protected IIngestionUnitOfWorkFactory UnitOfWorkFactory => ServiceProvider.GetRequiredService(); @@ -105,6 +110,7 @@ protected static async Task WaitUntil(Func> conditionChecker, string protected IArchiveMessages ArchiveMessages => ServiceProvider.GetRequiredService(); protected IIngestionUnitOfWorkFactory IngestionUnitOfWorkFactory => ServiceProvider.GetRequiredService(); protected IEventLogDataStore EventLogDataStore => ServiceProvider.GetRequiredService(); + protected IFailedErrorImportDataStore FailedImportStore => ServiceProvider.GetRequiredService(); protected IRetryDocumentDataStore RetryDocumentDataStore => ServiceProvider.GetRequiredService(); protected ILicensingDataStore LicensingDataStore => ServiceProvider.GetRequiredService(); protected IQueueAddressStore QueueAddressStore => ServiceProvider.GetRequiredService(); diff --git a/src/ServiceControl.Persistence.Tests/Recoverability/EditMessageTests.cs b/src/ServiceControl.Persistence.Tests/Recoverability/EditMessageTests.cs index ae9f7298be..edac10575f 100644 --- a/src/ServiceControl.Persistence.Tests/Recoverability/EditMessageTests.cs +++ b/src/ServiceControl.Persistence.Tests/Recoverability/EditMessageTests.cs @@ -55,9 +55,9 @@ public async Task Should_discard_edit_if_edited_message_not_unresolved(FailedMes var message = CreateEditMessage(failedMessageId); await handler.Handle(message, new TestableMessageHandlerContext()); - var failedMessage = await ErrorMessageDataStore.ErrorBy(failedMessageId); + var failedMessage = await FailedMessageQueryStore.GetFailedMessage(failedMessageId); - var editFailedMessagesManager = await ErrorMessageDataStore.CreateEditFailedMessageManager(); + var editFailedMessagesManager = await EditFailedMessagesStore.CreateEditFailedMessageManager(); var editOperation = await editFailedMessagesManager.GetCurrentEditingRequestId(failedMessageId); using (Assert.EnterMultipleScope()) @@ -76,7 +76,7 @@ public async Task Should_discard_edit_when_different_edit_already_exists() _ = await CreateAndStoreFailedMessage(failedMessageId); - using (var editFailedMessagesManager = await ErrorMessageDataStore.CreateEditFailedMessageManager()) + using (var editFailedMessagesManager = await EditFailedMessagesStore.CreateEditFailedMessageManager()) { _ = await editFailedMessagesManager.GetFailedMessage(failedMessageId); await editFailedMessagesManager.SetCurrentEditingRequestId(previousEdit); @@ -88,7 +88,7 @@ public async Task Should_discard_edit_when_different_edit_already_exists() // Act await handler.Handle(message, new TestableMessageHandlerContext()); - using (var editFailedMessagesManagerAssert = await ErrorMessageDataStore.CreateEditFailedMessageManager()) + using (var editFailedMessagesManagerAssert = await EditFailedMessagesStore.CreateEditFailedMessageManager()) { var failedMessage = await editFailedMessagesManagerAssert.GetFailedMessage(failedMessageId); var editId = await editFailedMessagesManagerAssert.GetCurrentEditingRequestId(failedMessageId); @@ -125,7 +125,7 @@ public async Task Should_dispatch_edited_message_when_first_edit() Assert.That(dispatchedMessage.Item1.Message.Headers["someKey"], Is.EqualTo("someValue")); } - using (var x = await ErrorMessageDataStore.CreateEditFailedMessageManager()) + using (var x = await EditFailedMessagesStore.CreateEditFailedMessageManager()) { var failedMessage2 = await x.GetFailedMessage(failedMessage.UniqueMessageId); Assert.That(failedMessage2, Is.Not.Null, "Edited failed message"); diff --git a/src/ServiceControl.Persistence.Tests/Recoverability/RetryConfirmationProcessorTests.cs b/src/ServiceControl.Persistence.Tests/Recoverability/RetryConfirmationProcessorTests.cs index 3fd4a136fb..d3111e676a 100644 --- a/src/ServiceControl.Persistence.Tests/Recoverability/RetryConfirmationProcessorTests.cs +++ b/src/ServiceControl.Persistence.Tests/Recoverability/RetryConfirmationProcessorTests.cs @@ -23,7 +23,7 @@ public async Task Setup() var domainEvents = new FakeDomainEvents(); Processor = new RetryConfirmationProcessor(domainEvents); - Handler = new LegacyMessageFailureResolvedHandler(ErrorMessageDataStore, domainEvents); + Handler = new LegacyMessageFailureResolvedHandler(FailedMessageRetryStore, FailedMessageLifecycleStore, domainEvents); await PersistenceTestsContext.InsertFailedMessages( new FailedMessage diff --git a/src/ServiceControl.Persistence.Tests/Recoverability/ReturnToSenderDequeuerTests.cs b/src/ServiceControl.Persistence.Tests/Recoverability/ReturnToSenderDequeuerTests.cs index d91b27da72..62224a020e 100644 --- a/src/ServiceControl.Persistence.Tests/Recoverability/ReturnToSenderDequeuerTests.cs +++ b/src/ServiceControl.Persistence.Tests/Recoverability/ReturnToSenderDequeuerTests.cs @@ -167,82 +167,15 @@ public Task Dispatch(TransportOperations outgoingMessages, TransportTransaction } } - class FakeErrorMessageDataStore : IErrorMessageDataStore + class FakeErrorMessageDataStore : IFailedMessageRetryDataStore { - public Task>> GetAllMessages(PagingInfo pagingInfo, SortInfo sortInfo, bool includeSystemMessages, - DateTimeRange timeSentRange = null) => - throw new NotImplementedException(); - - public Task>> GetAllMessagesForEndpoint(string endpointName, PagingInfo pagingInfo, SortInfo sortInfo, - bool includeSystemMessages, DateTimeRange timeSentRange = null) => - throw new NotImplementedException(); - - public Task>> GetAllMessagesByConversation(string conversationId, PagingInfo pagingInfo, SortInfo sortInfo, - bool includeSystemMessages) => - throw new NotImplementedException(); - - public Task>> GetAllMessagesForSearch(string searchTerms, PagingInfo pagingInfo, SortInfo sortInfo, - DateTimeRange timeSentRange = null) => - throw new NotImplementedException(); - - public Task>> SearchEndpointMessages(string endpointName, string searchKeyword, PagingInfo pagingInfo, SortInfo sortInfo, - DateTimeRange timeSentRange = null) => - throw new NotImplementedException(); - - public Task FailedMessageMarkAsArchived(string failedMessageId) => throw new NotImplementedException(); - - public Task FailedMessagesFetch(Guid[] ids) => throw new NotImplementedException(); - - public Task StoreFailedErrorImport(FailedErrorImport failure) => throw new NotImplementedException(); - - public Task CreateEditFailedMessageManager() => throw new NotImplementedException(); - - public Task> GetFailureGroupView(string groupId, string status, string modified) => throw new NotImplementedException(); - - public Task> GetFailureGroupsByClassifier(string classifier) => throw new NotImplementedException(); - - public Task>> ErrorGet(string status, string modified, string queueAddress, PagingInfo pagingInfo, SortInfo sortInfo) => throw new NotImplementedException(); - - public Task ErrorsHead(string status, string modified, string queueAddress) => throw new NotImplementedException(); - - public Task>> ErrorsByEndpointName(string status, string endpointName, string modified, PagingInfo pagingInfo, - SortInfo sortInfo) => - throw new NotImplementedException(); - - public Task> ErrorsSummary() => throw new NotImplementedException(); - - public Task ErrorLastBy(string failedMessageId) => throw new NotImplementedException(); - - public Task ErrorBy(string failedMessageId) => throw new NotImplementedException(); - - public Task CreateNotificationsManager() => throw new NotImplementedException(); - - public Task EditComment(string groupId, string comment) => throw new NotImplementedException(); - - public Task DeleteComment(string groupId) => throw new NotImplementedException(); - - public Task>> GetGroupErrors(string groupId, string status, string modified, SortInfo sortInfo, PagingInfo pagingInfo) => throw new NotImplementedException(); - - public Task GetGroupErrorsCount(string groupId, string status, string modified) => throw new NotImplementedException(); - - public Task>> GetGroup(string groupId, string status, string modified) => throw new NotImplementedException(); - - public Task MarkMessageAsResolved(string failedMessageId) => throw new NotImplementedException(); - public Task ProcessPendingRetries(DateTime periodFrom, DateTime periodTo, string queueAddress, Func processCallback) => throw new NotImplementedException(); - public Task UnArchiveMessagesByRange(DateTime from, DateTime to) => throw new NotImplementedException(); - - public Task UnArchiveMessages(IEnumerable failedMessageIds) => throw new NotImplementedException(); - - public Task RevertRetry(string messageUniqueId) => throw new NotImplementedException(); - - public Task RemoveFailedMessageRetryDocument(string uniqueMessageId) => throw new NotImplementedException(); - public Task GetRetryPendingMessages(DateTime from, DateTime to, string queueAddress) => throw new NotImplementedException(); - public Task FetchFromFailedMessage(string bodyId) => Task.FromResult(Encoding.UTF8.GetBytes(bodyId)); - public Task StoreEventLogItem(EventLogItem logItem) => throw new NotImplementedException(); + public Task RemoveFailedMessageRetry(string uniqueMessageId) => throw new NotImplementedException(); + + public Task GetFailedMessageBody(string bodyId) => Task.FromResult(Encoding.UTF8.GetBytes(bodyId)); } } } \ No newline at end of file diff --git a/src/ServiceControl.Persistence.Tests/RetryStateTests.cs b/src/ServiceControl.Persistence.Tests/RetryStateTests.cs index 4c11092471..2bd221eda8 100644 --- a/src/ServiceControl.Persistence.Tests/RetryStateTests.cs +++ b/src/ServiceControl.Persistence.Tests/RetryStateTests.cs @@ -71,7 +71,7 @@ public async Task When_the_dequeuer_is_created_then_the_error_address_is_cached( var transportCustomization = new TestTransportCustomization { TransportInfrastructure = transportInfrastructure }; - var testReturnToSenderDequeuer = new TestReturnToSenderDequeuer(new ReturnToSender(ErrorStore, NullLogger.Instance), ErrorStore, domainEvents, "TestEndpoint", + var testReturnToSenderDequeuer = new TestReturnToSenderDequeuer(new ReturnToSender(FailedMessageRetryStore, NullLogger.Instance), FailedMessageLifecycleStore, domainEvents, "TestEndpoint", errorQueueNameCache, transportCustomization); await testReturnToSenderDequeuer.StartAsync(new CancellationToken()); @@ -92,8 +92,8 @@ public async Task When_a_group_is_prepared_with_three_batches_and_SC_is_restarte RetryBatchesStore, domainEvents, new TestReturnToSenderDequeuer( - new ReturnToSender(ErrorStore, NullLogger.Instance), - ErrorStore, + new ReturnToSender(FailedMessageRetryStore, NullLogger.Instance), + FailedMessageLifecycleStore, domainEvents, "TestEndpoint", new ErrorQueueNameCache(), @@ -119,8 +119,8 @@ public async Task When_a_group_is_prepared_with_three_batches_and_SC_is_restarte RetryBatchesStore, domainEvents, new TestReturnToSenderDequeuer( - new ReturnToSender(ErrorStore, NullLogger.Instance), - ErrorStore, + new ReturnToSender(FailedMessageRetryStore, NullLogger.Instance), + FailedMessageLifecycleStore, domainEvents, "TestEndpoint", new ErrorQueueNameCache(), @@ -146,7 +146,7 @@ public async Task When_a_group_is_forwarded_the_status_is_Completed() var sender = new TestSender(); - var returnToSender = new TestReturnToSenderDequeuer(new ReturnToSender(ErrorStore, NullLogger.Instance), ErrorStore, domainEvents, "TestEndpoint", new ErrorQueueNameCache(), new TestTransportCustomization()); + var returnToSender = new TestReturnToSenderDequeuer(new ReturnToSender(FailedMessageRetryStore, NullLogger.Instance), FailedMessageLifecycleStore, domainEvents, "TestEndpoint", new ErrorQueueNameCache(), new TestTransportCustomization()); var processor = new RetryProcessor(RetryBatchesStore, domainEvents, returnToSender, retryManager, new Lazy(() => sender), new RecordingMessageActionAuditLog(), NullLogger.Instance); await processor.ProcessBatches(); // mark ready @@ -176,7 +176,7 @@ public async Task When_there_is_one_poison_message_it_is_removed_from_batch_and_ } }; - var returnToSender = new TestReturnToSenderDequeuer(new ReturnToSender(ErrorStore, NullLogger.Instance), ErrorStore, domainEvents, "TestEndpoint", new ErrorQueueNameCache(), new TestTransportCustomization()); + var returnToSender = new TestReturnToSenderDequeuer(new ReturnToSender(FailedMessageRetryStore, NullLogger.Instance), FailedMessageLifecycleStore, domainEvents, "TestEndpoint", new ErrorQueueNameCache(), new TestTransportCustomization()); var processor = new RetryProcessor(RetryBatchesStore, domainEvents, returnToSender, retryManager, new Lazy(() => sender), new RecordingMessageActionAuditLog(), NullLogger.Instance); bool c; @@ -213,11 +213,11 @@ public async Task When_a_group_has_one_batch_out_of_two_forwarded_the_status_is_ await CreateAFailedMessageAndMarkAsPartOfRetryBatch(retryManager, "Test-group", true, 1001); - var returnToSender = new ReturnToSender(ErrorStore, NullLogger.Instance); + var returnToSender = new ReturnToSender(FailedMessageRetryStore, NullLogger.Instance); var sender = new TestSender(); - var processor = new RetryProcessor(RetryBatchesStore, domainEvents, new TestReturnToSenderDequeuer(returnToSender, ErrorStore, domainEvents, "TestEndpoint", new ErrorQueueNameCache(), new TestTransportCustomization()), retryManager, new Lazy(() => sender), new RecordingMessageActionAuditLog(), NullLogger.Instance); + var processor = new RetryProcessor(RetryBatchesStore, domainEvents, new TestReturnToSenderDequeuer(returnToSender, FailedMessageLifecycleStore, domainEvents, "TestEndpoint", new ErrorQueueNameCache(), new TestTransportCustomization()), retryManager, new Lazy(() => sender), new RecordingMessageActionAuditLog(), NullLogger.Instance); await CompleteDatabaseOperation(); @@ -263,7 +263,7 @@ public async Task When_a_selection_is_staged_each_message_is_audited_as_a_batch( var audit = new RecordingMessageActionAuditLog(); var sender = new TestSender(); - var returnToSender = new TestReturnToSenderDequeuer(new ReturnToSender(ErrorStore, NullLogger.Instance), ErrorStore, domainEvents, "TestEndpoint", new ErrorQueueNameCache(), new TestTransportCustomization()); + var returnToSender = new TestReturnToSenderDequeuer(new ReturnToSender(FailedMessageRetryStore, NullLogger.Instance), FailedMessageLifecycleStore, domainEvents, "TestEndpoint", new ErrorQueueNameCache(), new TestTransportCustomization()); var processor = new RetryProcessor(RetryBatchesStore, domainEvents, returnToSender, retryManager, new Lazy(() => sender), audit, NullLogger.Instance); await processor.ProcessBatches(); // stage @@ -290,7 +290,7 @@ public async Task When_a_group_is_staged_each_message_is_audited_with_the_initia var audit = new RecordingMessageActionAuditLog(); var sender = new TestSender(); - var returnToSender = new TestReturnToSenderDequeuer(new ReturnToSender(ErrorStore, NullLogger.Instance), ErrorStore, domainEvents, "TestEndpoint", new ErrorQueueNameCache(), new TestTransportCustomization()); + var returnToSender = new TestReturnToSenderDequeuer(new ReturnToSender(FailedMessageRetryStore, NullLogger.Instance), FailedMessageLifecycleStore, domainEvents, "TestEndpoint", new ErrorQueueNameCache(), new TestTransportCustomization()); var processor = new RetryProcessor(RetryBatchesStore, domainEvents, returnToSender, retryManager, new Lazy(() => sender), audit, NullLogger.Instance); await processor.ProcessBatches(); // stage (emits per-message audit) @@ -416,7 +416,7 @@ class FakeApplicationLifetime : IHostApplicationLifetime class TestReturnToSenderDequeuer : ReturnToSenderDequeuer { - public TestReturnToSenderDequeuer(ReturnToSender returnToSender, IErrorMessageDataStore store, IDomainEvents domainEvents, string endpointName, + public TestReturnToSenderDequeuer(ReturnToSender returnToSender, IFailedMessageLifecycleDataStore store, IDomainEvents domainEvents, string endpointName, ErrorQueueNameCache cache, ITransportCustomization transportCustomization) : base(returnToSender, store, domainEvents, transportCustomization, null, new Settings { InstanceName = endpointName }, cache, NullLogger.Instance) { diff --git a/src/ServiceControl.Persistence/IEditFailedMessagesDataStore.cs b/src/ServiceControl.Persistence/IEditFailedMessagesDataStore.cs new file mode 100644 index 0000000000..2892d2bbe3 --- /dev/null +++ b/src/ServiceControl.Persistence/IEditFailedMessagesDataStore.cs @@ -0,0 +1,9 @@ +namespace ServiceControl.Persistence +{ + using System.Threading.Tasks; + + public interface IEditFailedMessagesDataStore + { + Task CreateEditFailedMessageManager(); + } +} diff --git a/src/ServiceControl.Persistence/IErrorMessageDatastore.cs b/src/ServiceControl.Persistence/IErrorMessageDatastore.cs deleted file mode 100644 index 28ad891425..0000000000 --- a/src/ServiceControl.Persistence/IErrorMessageDatastore.cs +++ /dev/null @@ -1,75 +0,0 @@ -namespace ServiceControl.Persistence -{ - using System; - using System.Collections.Generic; - using System.Threading.Tasks; - using CompositeViews.Messages; - using Infrastructure; - using MessageFailures.Api; - using ServiceControl.EventLog; - using ServiceControl.MessageFailures; - using ServiceControl.Operations; - using ServiceControl.Recoverability; - - public interface IErrorMessageDataStore - { - Task>> GetAllMessages(PagingInfo pagingInfo, SortInfo sortInfo, bool includeSystemMessages, DateTimeRange timeSentRange = null); - Task>> GetAllMessagesForEndpoint(string endpointName, PagingInfo pagingInfo, SortInfo sortInfo, bool includeSystemMessages, DateTimeRange timeSentRange = null); - Task>> GetAllMessagesByConversation(string conversationId, PagingInfo pagingInfo, SortInfo sortInfo, bool includeSystemMessages); - Task>> GetAllMessagesForSearch(string searchTerms, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange = null); - Task>> SearchEndpointMessages(string endpointName, string searchKeyword, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange = null); - Task FailedMessageMarkAsArchived(string failedMessageId); - Task FailedMessagesFetch(Guid[] ids); - Task StoreFailedErrorImport(FailedErrorImport failure); - Task CreateEditFailedMessageManager(); - Task> GetFailureGroupView(string groupId, string status, string modified); - Task> GetFailureGroupsByClassifier(string classifier); - - // GetAllErrorsController - Task>> ErrorGet(string status, string modified, string queueAddress, PagingInfo pagingInfo, SortInfo sortInfo); - Task ErrorsHead(string status, string modified, string queueAddress); - Task>> ErrorsByEndpointName(string status, string endpointName, string modified, PagingInfo pagingInfo, SortInfo sortInfo); - Task> ErrorsSummary(); - - // GetErrorByIdController - Task ErrorLastBy(string failedMessageId); - - //EditFailedMessagesController - // GetErrorByIdController - Task ErrorBy(string failedMessageId); - - //NotificationsController - Task CreateNotificationsManager(); - - // FailureGroupsController - Task EditComment(string groupId, string comment); - Task DeleteComment(string groupId); - Task>> GetGroupErrors(string groupId, string status, string modified, SortInfo sortInfo, PagingInfo pagingInfo); - Task GetGroupErrorsCount(string groupId, string status, string modified); - - Task>> GetGroup(string groupId, string status, string modified); - - // LegacyMessageFailureResolvedHandler - Task MarkMessageAsResolved(string failedMessageId); - - // MessageFailureResolvedHandler - Task ProcessPendingRetries(DateTime periodFrom, DateTime periodTo, string queueAddress, Func processCallback); - - // UnArchiveMessagesByRangeHandler - Task UnArchiveMessagesByRange(DateTime from, DateTime to); - - // UnArchiveMessagesHandler - Task UnArchiveMessages(IEnumerable failedMessageIds); - - // ReturnToSenderDequeuer.CaptureIfMessageSendingFails - Task RevertRetry(string messageUniqueId); - Task RemoveFailedMessageRetryDocument(string uniqueMessageId); - Task GetRetryPendingMessages(DateTime from, DateTime to, string queueAddress); - - // ReturnToSender.FetchFromFailedMessage - Task FetchFromFailedMessage(string uniqueMessageId); - - // AuditEventLogWriter - Task StoreEventLogItem(EventLogItem logItem); - } -} \ No newline at end of file diff --git a/src/ServiceControl.Persistence/IFailedErrorImportDataStore.cs b/src/ServiceControl.Persistence/IFailedErrorImportDataStore.cs index 54464bffcf..4e579a560f 100644 --- a/src/ServiceControl.Persistence/IFailedErrorImportDataStore.cs +++ b/src/ServiceControl.Persistence/IFailedErrorImportDataStore.cs @@ -7,6 +7,7 @@ public interface IFailedErrorImportDataStore { + Task StoreFailedErrorImport(FailedErrorImport failure); Task ProcessFailedErrorImports(Func processMessage, CancellationToken cancellationToken); Task QueryContainsFailedImports(); } diff --git a/src/ServiceControl.Persistence/IFailedMessageLifecycleDataStore.cs b/src/ServiceControl.Persistence/IFailedMessageLifecycleDataStore.cs new file mode 100644 index 0000000000..edae2a54ea --- /dev/null +++ b/src/ServiceControl.Persistence/IFailedMessageLifecycleDataStore.cs @@ -0,0 +1,19 @@ +namespace ServiceControl.Persistence +{ + using System; + using System.Collections.Generic; + using System.Threading.Tasks; + + /// + /// Transitions of . + /// Every operation here has to update both StatusChangedAt and LastModified. + /// + public interface IFailedMessageLifecycleDataStore + { + Task MarkAsArchived(string failedMessageId); + Task MarkAsResolved(string failedMessageId); + Task UnArchiveMessages(IEnumerable failedMessageIds); + Task UnArchiveMessagesByRange(DateTime from, DateTime to); + Task RevertRetry(string messageUniqueId); + } +} diff --git a/src/ServiceControl.Persistence/IFailedMessageQueryDataStore.cs b/src/ServiceControl.Persistence/IFailedMessageQueryDataStore.cs new file mode 100644 index 0000000000..66b7d8ed4a --- /dev/null +++ b/src/ServiceControl.Persistence/IFailedMessageQueryDataStore.cs @@ -0,0 +1,20 @@ +namespace ServiceControl.Persistence +{ + using System; + using System.Collections.Generic; + using System.Threading.Tasks; + using Infrastructure; + using MessageFailures.Api; + using ServiceControl.MessageFailures; + + public interface IFailedMessageQueryDataStore + { + Task>> GetFailedMessages(string status, string modified, string queueAddress, PagingInfo pagingInfo, SortInfo sortInfo); + Task GetFailedMessagesStats(string status, string modified, string queueAddress); + Task>> GetFailedMessagesByEndpoint(string status, string endpointName, string modified, PagingInfo pagingInfo, SortInfo sortInfo); + Task> GetFailedMessagesSummary(); + Task GetLatestFailedMessageView(string failedMessageId); + Task GetFailedMessage(string failedMessageId); + Task GetFailedMessagesByIds(Guid[] ids); + } +} diff --git a/src/ServiceControl.Persistence/IFailedMessageRetryDataStore.cs b/src/ServiceControl.Persistence/IFailedMessageRetryDataStore.cs new file mode 100644 index 0000000000..a55178edb6 --- /dev/null +++ b/src/ServiceControl.Persistence/IFailedMessageRetryDataStore.cs @@ -0,0 +1,13 @@ +namespace ServiceControl.Persistence +{ + using System; + using System.Threading.Tasks; + + public interface IFailedMessageRetryDataStore + { + Task ProcessPendingRetries(DateTime periodFrom, DateTime periodTo, string queueAddress, Func processCallback); + Task GetRetryPendingMessages(DateTime from, DateTime to, string queueAddress); + Task RemoveFailedMessageRetry(string uniqueMessageId); + Task GetFailedMessageBody(string uniqueMessageId); + } +} diff --git a/src/ServiceControl.Persistence/IGroupsDataStore.cs b/src/ServiceControl.Persistence/IGroupsDataStore.cs index 7f1fa1b02c..3180a7a67a 100644 --- a/src/ServiceControl.Persistence/IGroupsDataStore.cs +++ b/src/ServiceControl.Persistence/IGroupsDataStore.cs @@ -1,12 +1,23 @@ -namespace ServiceControl.Persistence +namespace ServiceControl.Persistence { using System.Collections.Generic; using System.Threading.Tasks; + using Infrastructure; + using MessageFailures.Api; using ServiceControl.Recoverability; public interface IGroupsDataStore { Task> GetFailureGroupsByClassifier(string classifier, string classifierFilter); + Task> GetArchivedFailureGroupsByClassifier(string classifier); Task GetCurrentForwardingBatch(); + + Task>> GetGroup(string groupId, string status, string modified); + Task> GetFailureGroupView(string groupId, string status, string modified); + Task>> GetGroupErrors(string groupId, string status, string modified, SortInfo sortInfo, PagingInfo pagingInfo); + Task GetGroupErrorsCount(string groupId, string status, string modified); + + Task EditComment(string groupId, string comment); + Task DeleteComment(string groupId); } } diff --git a/src/ServiceControl.Persistence/IMessagesViewDataStore.cs b/src/ServiceControl.Persistence/IMessagesViewDataStore.cs new file mode 100644 index 0000000000..281ca903ba --- /dev/null +++ b/src/ServiceControl.Persistence/IMessagesViewDataStore.cs @@ -0,0 +1,17 @@ +namespace ServiceControl.Persistence +{ + using System; + using System.Collections.Generic; + using System.Threading.Tasks; + using CompositeViews.Messages; + using Infrastructure; + + public interface IMessagesViewDataStore + { + Task>> GetAllMessages(PagingInfo pagingInfo, SortInfo sortInfo, bool includeSystemMessages, DateTimeRange timeSentRange = null); + Task>> GetAllMessagesForEndpoint(string endpointName, PagingInfo pagingInfo, SortInfo sortInfo, bool includeSystemMessages, DateTimeRange timeSentRange = null); + Task>> GetAllMessagesByConversation(string conversationId, PagingInfo pagingInfo, SortInfo sortInfo, bool includeSystemMessages); + Task>> GetAllMessagesForSearch(string searchTerms, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange = null); + Task>> SearchEndpointMessages(string endpointName, string searchKeyword, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange = null); + } +} diff --git a/src/ServiceControl.Persistence/INotificationsDataStore.cs b/src/ServiceControl.Persistence/INotificationsDataStore.cs new file mode 100644 index 0000000000..07beaf472a --- /dev/null +++ b/src/ServiceControl.Persistence/INotificationsDataStore.cs @@ -0,0 +1,9 @@ +namespace ServiceControl.Persistence +{ + using System.Threading.Tasks; + + public interface INotificationsDataStore + { + Task CreateNotificationsManager(); + } +} diff --git a/src/ServiceControl.UnitTests/MessageFailures/ArchiveScopeAuditTests.cs b/src/ServiceControl.UnitTests/MessageFailures/ArchiveScopeAuditTests.cs index e6772ee77a..f63deb85cf 100644 --- a/src/ServiceControl.UnitTests/MessageFailures/ArchiveScopeAuditTests.cs +++ b/src/ServiceControl.UnitTests/MessageFailures/ArchiveScopeAuditTests.cs @@ -51,7 +51,7 @@ public async Task Archived_message_is_audited_with_the_scope_of_the_originating_ { var audit = new RecordingMessageActionAuditLog(); var store = new AsyncRangeAndQueueAuditTests.StubErrorMessageDataStore { ErrorByResult = new FailedMessage { Status = FailedMessageStatus.Unresolved } }; - var handler = new ArchiveMessageHandler(store, new FakeDomainEvents(), audit); + var handler = new ArchiveMessageHandler(store, store, new FakeDomainEvents(), audit); var context = new TestableMessageHandlerContext { diff --git a/src/ServiceControl.UnitTests/MessageFailures/AsyncRangeAndQueueAuditTests.cs b/src/ServiceControl.UnitTests/MessageFailures/AsyncRangeAndQueueAuditTests.cs index 9c26e50c2f..e1bbbc56db 100644 --- a/src/ServiceControl.UnitTests/MessageFailures/AsyncRangeAndQueueAuditTests.cs +++ b/src/ServiceControl.UnitTests/MessageFailures/AsyncRangeAndQueueAuditTests.cs @@ -85,7 +85,7 @@ public async Task ArchiveMessage_audits_the_archived_message() { var audit = new RecordingMessageActionAuditLog(); var store = new StubErrorMessageDataStore { ErrorByResult = new FailedMessage { Status = FailedMessageStatus.Unresolved } }; - var handler = new ArchiveMessageHandler(store, new FakeDomainEvents(), audit); + var handler = new ArchiveMessageHandler(store, store, new FakeDomainEvents(), audit); var context = new TestableMessageHandlerContext { MessageHeaders = StampedHeaders("op-a") }; await handler.Handle(new ArchiveMessage { FailedMessageId = "m-1" }, context); @@ -105,7 +105,7 @@ public async Task ArchiveMessage_already_archived_is_not_audited() { var audit = new RecordingMessageActionAuditLog(); var store = new StubErrorMessageDataStore { ErrorByResult = new FailedMessage { Status = FailedMessageStatus.Archived } }; - var handler = new ArchiveMessageHandler(store, new FakeDomainEvents(), audit); + var handler = new ArchiveMessageHandler(store, store, new FakeDomainEvents(), audit); var context = new TestableMessageHandlerContext { MessageHeaders = StampedHeaders("op-a") }; await handler.Handle(new ArchiveMessage { FailedMessageId = "m-1" }, context); @@ -152,7 +152,7 @@ public async Task Unarchive_by_range_audits_each_message_with_bare_id() } } - internal sealed class StubErrorMessageDataStore : IErrorMessageDataStore + internal sealed class StubErrorMessageDataStore : IFailedMessageQueryDataStore, IFailedMessageLifecycleDataStore, IFailedMessageRetryDataStore { public string[] RetryPendingMessagesResult { get; set; } = []; public string[] UnArchiveByRangeResult { get; set; } = []; @@ -160,37 +160,21 @@ internal sealed class StubErrorMessageDataStore : IErrorMessageDataStore public FailedMessage ErrorByResult { get; set; } = new(); public Task GetRetryPendingMessages(DateTime from, DateTime to, string queueAddress) => Task.FromResult(RetryPendingMessagesResult); - public Task RemoveFailedMessageRetryDocument(string uniqueMessageId) => Task.CompletedTask; + public Task RemoveFailedMessageRetry(string uniqueMessageId) => Task.CompletedTask; public Task UnArchiveMessagesByRange(DateTime from, DateTime to) => Task.FromResult(UnArchiveByRangeResult); public Task UnArchiveMessages(IEnumerable failedMessageIds) => Task.FromResult(UnArchiveMessagesResult); - public Task ErrorBy(string failedMessageId) => Task.FromResult(ErrorByResult); - public Task FailedMessageMarkAsArchived(string failedMessageId) => Task.CompletedTask; - - public Task>> GetAllMessages(PagingInfo pagingInfo, SortInfo sortInfo, bool includeSystemMessages, DateTimeRange? timeSentRange = null) => throw new NotImplementedException(); - public Task>> GetAllMessagesForEndpoint(string endpointName, PagingInfo pagingInfo, SortInfo sortInfo, bool includeSystemMessages, DateTimeRange? timeSentRange = null) => throw new NotImplementedException(); - public Task>> GetAllMessagesByConversation(string conversationId, PagingInfo pagingInfo, SortInfo sortInfo, bool includeSystemMessages) => throw new NotImplementedException(); - public Task>> GetAllMessagesForSearch(string searchTerms, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange? timeSentRange = null) => throw new NotImplementedException(); - public Task>> SearchEndpointMessages(string endpointName, string searchKeyword, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange? timeSentRange = null) => throw new NotImplementedException(); - public Task FailedMessagesFetch(Guid[] ids) => throw new NotImplementedException(); - public Task StoreFailedErrorImport(FailedErrorImport failure) => throw new NotImplementedException(); - public Task CreateEditFailedMessageManager() => throw new NotImplementedException(); - public Task> GetFailureGroupView(string groupId, string status, string modified) => throw new NotImplementedException(); - public Task> GetFailureGroupsByClassifier(string classifier) => throw new NotImplementedException(); - public Task>> ErrorGet(string status, string modified, string queueAddress, PagingInfo pagingInfo, SortInfo sortInfo) => throw new NotImplementedException(); - public Task ErrorsHead(string status, string modified, string queueAddress) => throw new NotImplementedException(); - public Task>> ErrorsByEndpointName(string status, string endpointName, string modified, PagingInfo pagingInfo, SortInfo sortInfo) => throw new NotImplementedException(); - public Task> ErrorsSummary() => throw new NotImplementedException(); - public Task ErrorLastBy(string failedMessageId) => throw new NotImplementedException(); - public Task CreateNotificationsManager() => throw new NotImplementedException(); - public Task EditComment(string groupId, string comment) => throw new NotImplementedException(); - public Task DeleteComment(string groupId) => throw new NotImplementedException(); - public Task>> GetGroupErrors(string groupId, string status, string modified, SortInfo sortInfo, PagingInfo pagingInfo) => throw new NotImplementedException(); - public Task GetGroupErrorsCount(string groupId, string status, string modified) => throw new NotImplementedException(); - public Task>> GetGroup(string groupId, string status, string modified) => throw new NotImplementedException(); - public Task MarkMessageAsResolved(string failedMessageId) => throw new NotImplementedException(); + public Task GetFailedMessage(string failedMessageId) => Task.FromResult(ErrorByResult); + public Task MarkAsArchived(string failedMessageId) => Task.CompletedTask; + + public Task GetFailedMessagesByIds(Guid[] ids) => throw new NotImplementedException(); + public Task>> GetFailedMessages(string status, string modified, string queueAddress, PagingInfo pagingInfo, SortInfo sortInfo) => throw new NotImplementedException(); + public Task GetFailedMessagesStats(string status, string modified, string queueAddress) => throw new NotImplementedException(); + public Task>> GetFailedMessagesByEndpoint(string status, string endpointName, string modified, PagingInfo pagingInfo, SortInfo sortInfo) => throw new NotImplementedException(); + public Task> GetFailedMessagesSummary() => throw new NotImplementedException(); + public Task GetLatestFailedMessageView(string failedMessageId) => throw new NotImplementedException(); + public Task MarkAsResolved(string failedMessageId) => throw new NotImplementedException(); public Task ProcessPendingRetries(DateTime periodFrom, DateTime periodTo, string queueAddress, Func processCallback) => throw new NotImplementedException(); public Task RevertRetry(string messageUniqueId) => throw new NotImplementedException(); - public Task FetchFromFailedMessage(string uniqueMessageId) => throw new NotImplementedException(); - public Task StoreEventLogItem(EventLogItem logItem) => throw new NotImplementedException(); + public Task GetFailedMessageBody(string uniqueMessageId) => throw new NotImplementedException(); } } diff --git a/src/ServiceControl.UnitTests/MessageFailures/EditFailedMessagesControllerAuditTests.cs b/src/ServiceControl.UnitTests/MessageFailures/EditFailedMessagesControllerAuditTests.cs index 4cb8aa0738..e03faea222 100644 --- a/src/ServiceControl.UnitTests/MessageFailures/EditFailedMessagesControllerAuditTests.cs +++ b/src/ServiceControl.UnitTests/MessageFailures/EditFailedMessagesControllerAuditTests.cs @@ -24,7 +24,7 @@ namespace ServiceControl.UnitTests.MessageFailures; public class EditFailedMessagesControllerAuditTests { static EditFailedMessagesController Create(StubErrorMessageDataStore store, RecordingMessageActionAuditLog audit, bool allowMessageEditing = true) => - new(new Settings { AllowMessageEditing = allowMessageEditing }, store, new TestableMessageSession(), NullLogger.Instance, + new(new Settings { AllowMessageEditing = allowMessageEditing }, store, store, new TestableMessageSession(), NullLogger.Instance, new StubCurrentUserAccessor(new AuditUser("alice-sub", "Alice")), audit); static EditMessageModel ValidEdit() => new() { MessageBody = "body", MessageHeaders = [] }; @@ -59,43 +59,19 @@ public void Dispose() public Task SetFailedMessageAsResolved() => Task.CompletedTask; } - sealed class StubErrorMessageDataStore : IErrorMessageDataStore + sealed class StubErrorMessageDataStore : IFailedMessageQueryDataStore, IEditFailedMessagesDataStore { public FailedMessage? ErrorByResult { get; set; } public FakeEditFailedMessagesManager EditManager { get; } = new(); public Task CreateEditFailedMessageManager() => Task.FromResult(EditManager); - public Task ErrorBy(string failedMessageId) => Task.FromResult(ErrorByResult!); + public Task GetFailedMessage(string failedMessageId) => Task.FromResult(ErrorByResult!); - public Task>> GetAllMessages(PagingInfo pagingInfo, SortInfo sortInfo, bool includeSystemMessages, DateTimeRange? timeSentRange = null) => throw new NotImplementedException(); - public Task>> GetAllMessagesForEndpoint(string endpointName, PagingInfo pagingInfo, SortInfo sortInfo, bool includeSystemMessages, DateTimeRange? timeSentRange = null) => throw new NotImplementedException(); - public Task>> GetAllMessagesByConversation(string conversationId, PagingInfo pagingInfo, SortInfo sortInfo, bool includeSystemMessages) => throw new NotImplementedException(); - public Task>> GetAllMessagesForSearch(string searchTerms, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange? timeSentRange = null) => throw new NotImplementedException(); - public Task>> SearchEndpointMessages(string endpointName, string searchKeyword, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange? timeSentRange = null) => throw new NotImplementedException(); - public Task FailedMessageMarkAsArchived(string failedMessageId) => throw new NotImplementedException(); - public Task FailedMessagesFetch(Guid[] ids) => throw new NotImplementedException(); - public Task StoreFailedErrorImport(FailedErrorImport failure) => throw new NotImplementedException(); - public Task> GetFailureGroupView(string groupId, string status, string modified) => throw new NotImplementedException(); - public Task> GetFailureGroupsByClassifier(string classifier) => throw new NotImplementedException(); - public Task>> ErrorGet(string status, string modified, string queueAddress, PagingInfo pagingInfo, SortInfo sortInfo) => throw new NotImplementedException(); - public Task ErrorsHead(string status, string modified, string queueAddress) => throw new NotImplementedException(); - public Task>> ErrorsByEndpointName(string status, string endpointName, string modified, PagingInfo pagingInfo, SortInfo sortInfo) => throw new NotImplementedException(); - public Task> ErrorsSummary() => throw new NotImplementedException(); - public Task ErrorLastBy(string failedMessageId) => throw new NotImplementedException(); - public Task CreateNotificationsManager() => throw new NotImplementedException(); - public Task EditComment(string groupId, string comment) => throw new NotImplementedException(); - public Task DeleteComment(string groupId) => throw new NotImplementedException(); - public Task>> GetGroupErrors(string groupId, string status, string modified, SortInfo sortInfo, PagingInfo pagingInfo) => throw new NotImplementedException(); - public Task GetGroupErrorsCount(string groupId, string status, string modified) => throw new NotImplementedException(); - public Task>> GetGroup(string groupId, string status, string modified) => throw new NotImplementedException(); - public Task MarkMessageAsResolved(string failedMessageId) => throw new NotImplementedException(); - public Task ProcessPendingRetries(DateTime periodFrom, DateTime periodTo, string queueAddress, Func processCallback) => throw new NotImplementedException(); - public Task UnArchiveMessagesByRange(DateTime from, DateTime to) => throw new NotImplementedException(); - public Task UnArchiveMessages(IEnumerable failedMessageIds) => throw new NotImplementedException(); - public Task RevertRetry(string messageUniqueId) => throw new NotImplementedException(); - public Task RemoveFailedMessageRetryDocument(string uniqueMessageId) => throw new NotImplementedException(); - public Task GetRetryPendingMessages(DateTime from, DateTime to, string queueAddress) => throw new NotImplementedException(); - public Task FetchFromFailedMessage(string uniqueMessageId) => throw new NotImplementedException(); - public Task StoreEventLogItem(EventLogItem logItem) => throw new NotImplementedException(); + public Task GetFailedMessagesByIds(Guid[] ids) => throw new NotImplementedException(); + public Task>> GetFailedMessages(string status, string modified, string queueAddress, PagingInfo pagingInfo, SortInfo sortInfo) => throw new NotImplementedException(); + public Task GetFailedMessagesStats(string status, string modified, string queueAddress) => throw new NotImplementedException(); + public Task>> GetFailedMessagesByEndpoint(string status, string endpointName, string modified, PagingInfo pagingInfo, SortInfo sortInfo) => throw new NotImplementedException(); + public Task> GetFailedMessagesSummary() => throw new NotImplementedException(); + public Task GetLatestFailedMessageView(string failedMessageId) => throw new NotImplementedException(); } } diff --git a/src/ServiceControl/CompositeViews/AuditCounts/GetAuditCountsForEndpointApi.cs b/src/ServiceControl/CompositeViews/AuditCounts/GetAuditCountsForEndpointApi.cs index b8c051cd91..73899eff73 100644 --- a/src/ServiceControl/CompositeViews/AuditCounts/GetAuditCountsForEndpointApi.cs +++ b/src/ServiceControl/CompositeViews/AuditCounts/GetAuditCountsForEndpointApi.cs @@ -19,12 +19,12 @@ public record AuditCountsForEndpointContext(PagingInfo PagingInfo, string Endpoint) : ScatterGatherContext(PagingInfo); public class GetAuditCountsForEndpointApi( - IErrorMessageDataStore dataStore, + IMessagesViewDataStore dataStore, Settings settings, IHttpClientFactory httpClientFactory, IHttpContextAccessor httpContextAccessor, ILogger logger) - : ScatterGatherApi>(dataStore, settings, httpClientFactory, httpContextAccessor, logger) + : ScatterGatherApi>(dataStore, settings, httpClientFactory, httpContextAccessor, logger) { static readonly IList Empty = new List(0).AsReadOnly(); diff --git a/src/ServiceControl/CompositeViews/Messages/GetAllMessagesApi.cs b/src/ServiceControl/CompositeViews/Messages/GetAllMessagesApi.cs index a6c867331f..726d4f7365 100644 --- a/src/ServiceControl/CompositeViews/Messages/GetAllMessagesApi.cs +++ b/src/ServiceControl/CompositeViews/Messages/GetAllMessagesApi.cs @@ -9,9 +9,9 @@ namespace ServiceControl.CompositeViews.Messages using Persistence.Infrastructure; using ServiceBus.Management.Infrastructure.Settings; - public class GetAllMessagesApi : ScatterGatherApiMessageView + public class GetAllMessagesApi : ScatterGatherApiMessageView { - public GetAllMessagesApi(IErrorMessageDataStore dataStore, Settings settings, IHttpClientFactory httpClientFactory, IHttpContextAccessor httpContextAccessor, ILogger logger) + public GetAllMessagesApi(IMessagesViewDataStore dataStore, Settings settings, IHttpClientFactory httpClientFactory, IHttpContextAccessor httpContextAccessor, ILogger logger) : base(dataStore, settings, httpClientFactory, httpContextAccessor, logger) { } diff --git a/src/ServiceControl/CompositeViews/Messages/GetAllMessagesForEndpointApi.cs b/src/ServiceControl/CompositeViews/Messages/GetAllMessagesForEndpointApi.cs index 526be6056b..05e43e0e93 100644 --- a/src/ServiceControl/CompositeViews/Messages/GetAllMessagesForEndpointApi.cs +++ b/src/ServiceControl/CompositeViews/Messages/GetAllMessagesForEndpointApi.cs @@ -17,9 +17,9 @@ public record AllMessagesForEndpointContext( DateTimeRange TimeSentRange = null) : ScatterGatherApiMessageViewWithSystemMessagesContext(PagingInfo, SortInfo, IncludeSystemMessages, TimeSentRange); - public class GetAllMessagesForEndpointApi : ScatterGatherApiMessageView + public class GetAllMessagesForEndpointApi : ScatterGatherApiMessageView { - public GetAllMessagesForEndpointApi(IErrorMessageDataStore dataStore, Settings settings, IHttpClientFactory httpClientFactory, IHttpContextAccessor httpContextAccessor, ILogger logger) + public GetAllMessagesForEndpointApi(IMessagesViewDataStore dataStore, Settings settings, IHttpClientFactory httpClientFactory, IHttpContextAccessor httpContextAccessor, ILogger logger) : base(dataStore, settings, httpClientFactory, httpContextAccessor, logger) { } diff --git a/src/ServiceControl/CompositeViews/Messages/MessagesByConversationApi.cs b/src/ServiceControl/CompositeViews/Messages/MessagesByConversationApi.cs index 9848dbfb74..95ce512413 100644 --- a/src/ServiceControl/CompositeViews/Messages/MessagesByConversationApi.cs +++ b/src/ServiceControl/CompositeViews/Messages/MessagesByConversationApi.cs @@ -16,9 +16,9 @@ public record MessagesByConversationContext( string ConversationId) : ScatterGatherApiMessageViewWithSystemMessagesContext(PagingInfo, SortInfo, IncludeSystemMessages); - public class MessagesByConversationApi : ScatterGatherApiMessageView + public class MessagesByConversationApi : ScatterGatherApiMessageView { - public MessagesByConversationApi(IErrorMessageDataStore dataStore, Settings settings, IHttpClientFactory httpClientFactory, IHttpContextAccessor httpContextAccessor, ILogger logger) + public MessagesByConversationApi(IMessagesViewDataStore dataStore, Settings settings, IHttpClientFactory httpClientFactory, IHttpContextAccessor httpContextAccessor, ILogger logger) : base(dataStore, settings, httpClientFactory, httpContextAccessor, logger) { } diff --git a/src/ServiceControl/CompositeViews/Messages/SearchApi.cs b/src/ServiceControl/CompositeViews/Messages/SearchApi.cs index 51ad8798cb..8f9f1e8dd5 100644 --- a/src/ServiceControl/CompositeViews/Messages/SearchApi.cs +++ b/src/ServiceControl/CompositeViews/Messages/SearchApi.cs @@ -16,9 +16,9 @@ public record SearchApiContext( DateTimeRange TimeSentRange = null) : ScatterGatherApiMessageViewContext(PagingInfo, SortInfo, TimeSentRange); - public class SearchApi : ScatterGatherApiMessageView + public class SearchApi : ScatterGatherApiMessageView { - public SearchApi(IErrorMessageDataStore dataStore, Settings settings, IHttpClientFactory httpClientFactory, IHttpContextAccessor httpContextAccessor, ILogger logger) + public SearchApi(IMessagesViewDataStore dataStore, Settings settings, IHttpClientFactory httpClientFactory, IHttpContextAccessor httpContextAccessor, ILogger logger) : base(dataStore, settings, httpClientFactory, httpContextAccessor, logger) { } diff --git a/src/ServiceControl/CompositeViews/Messages/SearchEndpointApi.cs b/src/ServiceControl/CompositeViews/Messages/SearchEndpointApi.cs index 897ce47e4e..6acd23b653 100644 --- a/src/ServiceControl/CompositeViews/Messages/SearchEndpointApi.cs +++ b/src/ServiceControl/CompositeViews/Messages/SearchEndpointApi.cs @@ -17,9 +17,9 @@ public record SearchEndpointContext( DateTimeRange TimeSentRange = null) : ScatterGatherApiMessageViewContext(PagingInfo, SortInfo, TimeSentRange); - public class SearchEndpointApi : ScatterGatherApiMessageView + public class SearchEndpointApi : ScatterGatherApiMessageView { - public SearchEndpointApi(IErrorMessageDataStore dataStore, Settings settings, IHttpClientFactory httpClientFactory, IHttpContextAccessor httpContextAccessor, ILogger logger) + public SearchEndpointApi(IMessagesViewDataStore dataStore, Settings settings, IHttpClientFactory httpClientFactory, IHttpContextAccessor httpContextAccessor, ILogger logger) : base(dataStore, settings, httpClientFactory, httpContextAccessor, logger) { } diff --git a/src/ServiceControl/EventLog/AuditEventLogWriter.cs b/src/ServiceControl/EventLog/AuditEventLogWriter.cs index 84e1708e77..f44df336ed 100644 --- a/src/ServiceControl/EventLog/AuditEventLogWriter.cs +++ b/src/ServiceControl/EventLog/AuditEventLogWriter.cs @@ -13,7 +13,7 @@ /// class AuditEventLogWriter : IDomainHandler { - public AuditEventLogWriter(GlobalEventHandler broadcaster, IErrorMessageDataStore dataStore, EventLogMappings mappings) + public AuditEventLogWriter(GlobalEventHandler broadcaster, IEventLogDataStore dataStore, EventLogMappings mappings) { this.broadcaster = broadcaster; this.dataStore = dataStore; @@ -29,7 +29,7 @@ public async Task Handle(IDomainEvent message, CancellationToken cancellationTok var logItem = mappings.ApplyMapping(message); - await dataStore.StoreEventLogItem(logItem); + await dataStore.Add(logItem); await broadcaster.Broadcast(new EventLogItemAdded { @@ -46,7 +46,7 @@ await broadcaster.Broadcast(new EventLogItemAdded } readonly GlobalEventHandler broadcaster; - readonly IErrorMessageDataStore dataStore; + readonly IEventLogDataStore dataStore; readonly EventLogMappings mappings; } } \ No newline at end of file diff --git a/src/ServiceControl/MessageFailures/Api/ArchiveMessagesController.cs b/src/ServiceControl/MessageFailures/Api/ArchiveMessagesController.cs index d00451129b..8a7ff276ed 100644 --- a/src/ServiceControl/MessageFailures/Api/ArchiveMessagesController.cs +++ b/src/ServiceControl/MessageFailures/Api/ArchiveMessagesController.cs @@ -14,7 +14,7 @@ namespace ServiceControl.MessageFailures.Api [ApiController] [Route("api")] - public class ArchiveMessagesController(IMessageSession messageSession, IErrorMessageDataStore dataStore, ICurrentUserAccessor userAccessor, IMessageActionAuditLog auditLog) : ControllerBase + public class ArchiveMessagesController(IMessageSession messageSession, IGroupsDataStore dataStore, ICurrentUserAccessor userAccessor, IMessageActionAuditLog auditLog) : ControllerBase { [Authorize(Policy = Permissions.ErrorMessagesArchive)] [Route("errors/archive")] @@ -47,7 +47,7 @@ await auditLog.AuditedOperation(user, MessageActionKind.Archive, Permissions.Err [HttpGet] public async Task GetArchiveMessageGroups(string classifier = "Exception Type and Stack Trace") { - var results = await dataStore.GetFailureGroupsByClassifier(classifier); + var results = await dataStore.GetArchivedFailureGroupsByClassifier(classifier); Response.WithDeterministicEtag(EtagHelper.CalculateEtag(results)); diff --git a/src/ServiceControl/MessageFailures/Api/EditFailedMessagesController.cs b/src/ServiceControl/MessageFailures/Api/EditFailedMessagesController.cs index 20acc62019..f6906abad9 100644 --- a/src/ServiceControl/MessageFailures/Api/EditFailedMessagesController.cs +++ b/src/ServiceControl/MessageFailures/Api/EditFailedMessagesController.cs @@ -19,7 +19,8 @@ [Route("api")] public class EditFailedMessagesController( Settings settings, - IErrorMessageDataStore store, + IFailedMessageQueryDataStore store, + IEditFailedMessagesDataStore editStore, IMessageSession session, ILogger logger, ICurrentUserAccessor userAccessor, @@ -43,7 +44,7 @@ public async Task> Edit(string failedMessageId, } //HINT: This validation is the first one because we want to minimize the chance of two users concurrently execute an edit-retry. - var editManager = await store.CreateEditFailedMessageManager(); + var editManager = await editStore.CreateEditFailedMessageManager(); var editId = await editManager.GetCurrentEditingRequestId(failedMessageId); if (editId != null) { @@ -52,7 +53,7 @@ public async Task> Edit(string failedMessageId, return Ok(new EditRetryResponse { EditIgnored = true }); } - var failedMessage = await store.ErrorBy(failedMessageId); + var failedMessage = await store.GetFailedMessage(failedMessageId); if (failedMessage == null) { diff --git a/src/ServiceControl/MessageFailures/Api/GetAllErrorsController.cs b/src/ServiceControl/MessageFailures/Api/GetAllErrorsController.cs index 06c7260d26..15d6c81102 100644 --- a/src/ServiceControl/MessageFailures/Api/GetAllErrorsController.cs +++ b/src/ServiceControl/MessageFailures/Api/GetAllErrorsController.cs @@ -11,14 +11,14 @@ [ApiController] [Route("api")] - public class GetAllErrorsController(IErrorMessageDataStore store) : ControllerBase + public class GetAllErrorsController(IFailedMessageQueryDataStore store) : ControllerBase { [Authorize(Policy = Permissions.ErrorMessagesView)] [Route("errors")] [HttpGet] public async Task> ErrorsGet([FromQuery] PagingInfo pagingInfo, [FromQuery] SortInfo sortInfo, string status, string modified, string queueAddress) { - var results = await store.ErrorGet( + var results = await store.GetFailedMessages( status: status, modified: modified, queueAddress: queueAddress, @@ -36,7 +36,7 @@ public async Task> ErrorsGet([FromQuery] PagingInfo pag [HttpHead] public async Task ErrorsHead(string status, string modified, string queueAddress) { - var queryResult = await store.ErrorsHead( + var queryResult = await store.GetFailedMessagesStats( status: status, modified: modified, queueAddress: queueAddress @@ -50,7 +50,7 @@ public async Task ErrorsHead(string status, string modified, string queueAddress [HttpGet] public async Task> ErrorsByEndpointName([FromQuery] PagingInfo pagingInfo, [FromQuery] SortInfo sortInfo, string status, string modified, string endpointName) { - var results = await store.ErrorsByEndpointName( + var results = await store.GetFailedMessagesByEndpoint( status: status, endpointName: endpointName, modified: modified, @@ -66,6 +66,6 @@ public async Task> ErrorsByEndpointName([FromQuery] Pag [Authorize(Policy = Permissions.ErrorMessagesView)] [Route("errors/summary")] [HttpGet] - public async Task> ErrorsSummary() => await store.ErrorsSummary(); + public async Task> ErrorsSummary() => await store.GetFailedMessagesSummary(); } } \ No newline at end of file diff --git a/src/ServiceControl/MessageFailures/Api/GetErrorByIdController.cs b/src/ServiceControl/MessageFailures/Api/GetErrorByIdController.cs index bb419a014c..271420f2a5 100644 --- a/src/ServiceControl/MessageFailures/Api/GetErrorByIdController.cs +++ b/src/ServiceControl/MessageFailures/Api/GetErrorByIdController.cs @@ -8,14 +8,14 @@ [ApiController] [Route("api")] - public class GetErrorByIdController(IErrorMessageDataStore store) : ControllerBase + public class GetErrorByIdController(IFailedMessageQueryDataStore store) : ControllerBase { [Authorize(Policy = Permissions.ErrorMessagesView)] [Route("errors/{failedMessageId:required:minlength(1)}")] [HttpGet] public async Task> ErrorBy(string failedMessageId) { - var result = await store.ErrorBy(failedMessageId); + var result = await store.GetFailedMessage(failedMessageId); return result == null ? NotFound() : result; } @@ -25,7 +25,7 @@ public async Task> ErrorBy(string failedMessageId) [HttpGet] public async Task> ErrorLastBy(string failedMessageId) { - var result = await store.ErrorLastBy(failedMessageId); + var result = await store.GetLatestFailedMessageView(failedMessageId); return result == null ? NotFound() : result; } diff --git a/src/ServiceControl/MessageFailures/Handlers/ArchiveMessageHandler.cs b/src/ServiceControl/MessageFailures/Handlers/ArchiveMessageHandler.cs index 6169f2b4aa..1580c8b95f 100644 --- a/src/ServiceControl/MessageFailures/Handlers/ArchiveMessageHandler.cs +++ b/src/ServiceControl/MessageFailures/Handlers/ArchiveMessageHandler.cs @@ -9,13 +9,13 @@ using ServiceControl.Persistence; [Handler] - class ArchiveMessageHandler(IErrorMessageDataStore dataStore, IDomainEvents domainEvents, IMessageActionAuditLog auditLog) : IHandleMessages + class ArchiveMessageHandler(IFailedMessageQueryDataStore queryStore, IFailedMessageLifecycleDataStore lifecycleStore, IDomainEvents domainEvents, IMessageActionAuditLog auditLog) : IHandleMessages { public async Task Handle(ArchiveMessage message, IMessageHandlerContext context) { var failedMessageId = message.FailedMessageId; - var failedMessage = await dataStore.ErrorBy(failedMessageId); + var failedMessage = await queryStore.GetFailedMessage(failedMessageId); if (failedMessage.Status != FailedMessageStatus.Archived) { @@ -24,7 +24,7 @@ await domainEvents.Raise(new FailedMessageArchived FailedMessageId = failedMessageId }, context.CancellationToken); - await dataStore.FailedMessageMarkAsArchived(failedMessageId); + await lifecycleStore.MarkAsArchived(failedMessageId); var (user, operationId) = AuditHeaders.Read(context.MessageHeaders); if (!string.IsNullOrEmpty(operationId)) diff --git a/src/ServiceControl/MessageFailures/Handlers/LegacyMessageFailureResolvedHandler.cs b/src/ServiceControl/MessageFailures/Handlers/LegacyMessageFailureResolvedHandler.cs index 6ae338fdc6..a0f05157d3 100644 --- a/src/ServiceControl/MessageFailures/Handlers/LegacyMessageFailureResolvedHandler.cs +++ b/src/ServiceControl/MessageFailures/Handlers/LegacyMessageFailureResolvedHandler.cs @@ -11,7 +11,7 @@ /// This class handles legacy messages that mark a failed message as successfully retried. For further details go to message definitions. /// [Handler] - class LegacyMessageFailureResolvedHandler(IErrorMessageDataStore store, IDomainEvents domainEvents) : + class LegacyMessageFailureResolvedHandler(IFailedMessageRetryDataStore retryStore, IFailedMessageLifecycleDataStore lifecycleStore, IDomainEvents domainEvents) : IHandleMessages, IHandleMessages { @@ -38,9 +38,9 @@ await domainEvents.Raise(new MessageFailureResolvedByRetry async Task MarkAsResolvedByRetry(string primaryId, string[] messageAlternativeFailedMessageIds) { - await store.RemoveFailedMessageRetryDocument(primaryId); + await retryStore.RemoveFailedMessageRetry(primaryId); - var primaryUpdated = await store.MarkMessageAsResolved(primaryId); + var primaryUpdated = await lifecycleStore.MarkAsResolved(primaryId); if (primaryUpdated) { @@ -54,9 +54,9 @@ async Task MarkAsResolvedByRetry(string primaryId, string[] messageAlternativeFa foreach (var alternative in messageAlternativeFailedMessageIds.Where(x => x != primaryId)) { - await store.RemoveFailedMessageRetryDocument(alternative); + await retryStore.RemoveFailedMessageRetry(alternative); - var alternativeUpdated = await store.MarkMessageAsResolved(alternative); + var alternativeUpdated = await lifecycleStore.MarkAsResolved(alternative); if (alternativeUpdated) { diff --git a/src/ServiceControl/MessageFailures/Handlers/MessageFailureResolvedHandler.cs b/src/ServiceControl/MessageFailures/Handlers/MessageFailureResolvedHandler.cs index 031a66306a..9663593117 100644 --- a/src/ServiceControl/MessageFailures/Handlers/MessageFailureResolvedHandler.cs +++ b/src/ServiceControl/MessageFailures/Handlers/MessageFailureResolvedHandler.cs @@ -8,7 +8,7 @@ using Persistence; [Handler] - class MessageFailureResolvedHandler(IErrorMessageDataStore dataStore, IDomainEvents domainEvents) : + class MessageFailureResolvedHandler(IFailedMessageRetryDataStore retryStore, IFailedMessageLifecycleDataStore lifecycleStore, IDomainEvents domainEvents) : IHandleMessages, IHandleMessages { @@ -24,7 +24,7 @@ Task ProcessCallback(string id) return context.Send(m => m.FailedMessageId = id, sendOptions); } - return dataStore.ProcessPendingRetries( + return retryStore.ProcessPendingRetries( message.PeriodFrom, message.PeriodTo, message.QueueAddress, @@ -34,7 +34,7 @@ Task ProcessCallback(string id) public async Task Handle(MarkPendingRetryAsResolved message, IMessageHandlerContext context) { - _ = await dataStore.MarkMessageAsResolved(message.FailedMessageId); + _ = await lifecycleStore.MarkAsResolved(message.FailedMessageId); await domainEvents.Raise(new MessageFailureResolvedManually { diff --git a/src/ServiceControl/MessageFailures/Handlers/UnArchiveMessagesByRangeHandler.cs b/src/ServiceControl/MessageFailures/Handlers/UnArchiveMessagesByRangeHandler.cs index c7d2b6dbc6..a26b7680ab 100644 --- a/src/ServiceControl/MessageFailures/Handlers/UnArchiveMessagesByRangeHandler.cs +++ b/src/ServiceControl/MessageFailures/Handlers/UnArchiveMessagesByRangeHandler.cs @@ -10,7 +10,7 @@ using Persistence; [Handler] - class UnArchiveMessagesByRangeHandler(IErrorMessageDataStore dataStore, IDomainEvents domainEvents, IMessageActionAuditLog auditLog) : IHandleMessages + class UnArchiveMessagesByRangeHandler(IFailedMessageLifecycleDataStore dataStore, IDomainEvents domainEvents, IMessageActionAuditLog auditLog) : IHandleMessages { public async Task Handle(UnArchiveMessagesByRange message, IMessageHandlerContext context) { diff --git a/src/ServiceControl/MessageFailures/Handlers/UnArchiveMessagesHandler.cs b/src/ServiceControl/MessageFailures/Handlers/UnArchiveMessagesHandler.cs index 894cf17730..f4709d63d0 100644 --- a/src/ServiceControl/MessageFailures/Handlers/UnArchiveMessagesHandler.cs +++ b/src/ServiceControl/MessageFailures/Handlers/UnArchiveMessagesHandler.cs @@ -10,7 +10,7 @@ using Persistence; [Handler] - class UnArchiveMessagesHandler(IErrorMessageDataStore store, IDomainEvents domainEvents, IMessageActionAuditLog auditLog) + class UnArchiveMessagesHandler(IFailedMessageLifecycleDataStore store, IDomainEvents domainEvents, IMessageActionAuditLog auditLog) : IHandleMessages { public async Task Handle(UnArchiveMessages messages, IMessageHandlerContext context) diff --git a/src/ServiceControl/Notifications/Api/NotificationsController.cs b/src/ServiceControl/Notifications/Api/NotificationsController.cs index e42cfa5473..f25c3433da 100644 --- a/src/ServiceControl/Notifications/Api/NotificationsController.cs +++ b/src/ServiceControl/Notifications/Api/NotificationsController.cs @@ -12,7 +12,7 @@ [ApiController] [Route("api")] - public class NotificationsController(IErrorMessageDataStore store, Settings settings, EmailSender emailSender) : ControllerBase + public class NotificationsController(INotificationsDataStore store, Settings settings, EmailSender emailSender) : ControllerBase { [Authorize(Policy = Permissions.ErrorNotificationsView)] [Route("notifications/email")] diff --git a/src/ServiceControl/Notifications/Email/SendEmailNotificationHandler.cs b/src/ServiceControl/Notifications/Email/SendEmailNotificationHandler.cs index 5519662e4d..6c9c2c9550 100644 --- a/src/ServiceControl/Notifications/Email/SendEmailNotificationHandler.cs +++ b/src/ServiceControl/Notifications/Email/SendEmailNotificationHandler.cs @@ -9,7 +9,7 @@ using ServiceBus.Management.Infrastructure.Settings; [Handler] - class SendEmailNotificationHandler(IErrorMessageDataStore store, Settings settings, EmailThrottlingState throttlingState, EmailSender emailSender, ILogger logger) + class SendEmailNotificationHandler(INotificationsDataStore store, Settings settings, EmailThrottlingState throttlingState, EmailSender emailSender, ILogger logger) : IHandleMessages { public async Task Handle(SendEmailNotification message, IMessageHandlerContext context) diff --git a/src/ServiceControl/Operations/ErrorIngestion.cs b/src/ServiceControl/Operations/ErrorIngestion.cs index ec04de2e72..52ef5491f9 100644 --- a/src/ServiceControl/Operations/ErrorIngestion.cs +++ b/src/ServiceControl/Operations/ErrorIngestion.cs @@ -26,7 +26,7 @@ public ErrorIngestion( ITransportCustomization transportCustomization, TransportSettings transportSettings, Metrics metrics, - IErrorMessageDataStore dataStore, + IFailedErrorImportDataStore dataStore, ErrorIngestionCustomCheck.State ingestionState, ErrorIngestor ingestor, IIngestionUnitOfWorkFactory unitOfWorkFactory, diff --git a/src/ServiceControl/Operations/ErrorIngestionFaultPolicy.cs b/src/ServiceControl/Operations/ErrorIngestionFaultPolicy.cs index 78bd51b7cf..6719738a29 100644 --- a/src/ServiceControl/Operations/ErrorIngestionFaultPolicy.cs +++ b/src/ServiceControl/Operations/ErrorIngestionFaultPolicy.cs @@ -16,12 +16,12 @@ class ErrorIngestionFaultPolicy { - IErrorMessageDataStore store; + IFailedErrorImportDataStore store; string logPath; ImportFailureCircuitBreaker failureCircuitBreaker; - public ErrorIngestionFaultPolicy(IErrorMessageDataStore store, LoggingSettings loggingSettings, Func onCriticalError, ILogger logger) + public ErrorIngestionFaultPolicy(IFailedErrorImportDataStore store, LoggingSettings loggingSettings, Func onCriticalError, ILogger logger) { this.store = store; this.logger = logger; diff --git a/src/ServiceControl/Recoverability/API/FailureGroupsController.cs b/src/ServiceControl/Recoverability/API/FailureGroupsController.cs index 7a3964f9cb..e86b82dfbe 100644 --- a/src/ServiceControl/Recoverability/API/FailureGroupsController.cs +++ b/src/ServiceControl/Recoverability/API/FailureGroupsController.cs @@ -16,7 +16,7 @@ public class FailureGroupsController( IEnumerable classifiers, GroupFetcher fetcher, - IErrorMessageDataStore store, + IGroupsDataStore store, IRetryHistoryDataStore retryStore) : ControllerBase { diff --git a/src/ServiceControl/Recoverability/Editing/EditHandler.cs b/src/ServiceControl/Recoverability/Editing/EditHandler.cs index 1ff4ca428f..ae629a7f74 100644 --- a/src/ServiceControl/Recoverability/Editing/EditHandler.cs +++ b/src/ServiceControl/Recoverability/Editing/EditHandler.cs @@ -16,7 +16,7 @@ using ServiceControl.Persistence.MessageRedirects; [Handler] - class EditHandler(IErrorMessageDataStore store, IMessageRedirectsDataStore redirectsStore, IMessageDispatcher dispatcher, ErrorQueueNameCache errorQueueNameCache, IDomainEvents domainEvents, IMessageActionAuditLog auditLog, ILogger logger) + class EditHandler(IEditFailedMessagesDataStore store, IMessageRedirectsDataStore redirectsStore, IMessageDispatcher dispatcher, ErrorQueueNameCache errorQueueNameCache, IDomainEvents domainEvents, IMessageActionAuditLog auditLog, ILogger logger) : IHandleMessages { public async Task Handle(EditAndSend message, IMessageHandlerContext context) diff --git a/src/ServiceControl/Recoverability/ExternalIntegration/MessageFailedPublisher.cs b/src/ServiceControl/Recoverability/ExternalIntegration/MessageFailedPublisher.cs index f0db52c942..81d265bafe 100644 --- a/src/ServiceControl/Recoverability/ExternalIntegration/MessageFailedPublisher.cs +++ b/src/ServiceControl/Recoverability/ExternalIntegration/MessageFailedPublisher.cs @@ -10,9 +10,9 @@ namespace ServiceControl.Recoverability.ExternalIntegration class MessageFailedPublisher : EventPublisher { - readonly IErrorMessageDataStore dataStore; + readonly IFailedMessageQueryDataStore dataStore; - public MessageFailedPublisher(IErrorMessageDataStore dataStore) + public MessageFailedPublisher(IFailedMessageQueryDataStore dataStore) { this.dataStore = dataStore; } @@ -28,7 +28,7 @@ protected override DispatchContext CreateDispatchRequest(MessageFailed @event) protected override async Task> PublishEvents(IEnumerable contexts) { var ids = contexts.Select(x => x.FailedMessageId).ToArray(); - var results = await dataStore.FailedMessagesFetch(ids); + var results = await dataStore.GetFailedMessagesByIds(ids); return results.Select(x => x.ToEvent()); } diff --git a/src/ServiceControl/Recoverability/Retrying/FailedMessageRetryCleaner.cs b/src/ServiceControl/Recoverability/Retrying/FailedMessageRetryCleaner.cs index e2b55765a6..46abb2ec80 100644 --- a/src/ServiceControl/Recoverability/Retrying/FailedMessageRetryCleaner.cs +++ b/src/ServiceControl/Recoverability/Retrying/FailedMessageRetryCleaner.cs @@ -8,9 +8,9 @@ namespace ServiceControl.Recoverability class FailedMessageRetryCleaner : IDomainHandler { - readonly IErrorMessageDataStore dataStore; + readonly IFailedMessageRetryDataStore dataStore; - public FailedMessageRetryCleaner(IErrorMessageDataStore dataStore) + public FailedMessageRetryCleaner(IFailedMessageRetryDataStore dataStore) { this.dataStore = dataStore; } @@ -19,7 +19,7 @@ public Task Handle(MessageFailed message, CancellationToken cancellationToken) { if (message.RepeatedFailure) { - return dataStore.RemoveFailedMessageRetryDocument(message.FailedMessageId); + return dataStore.RemoveFailedMessageRetry(message.FailedMessageId); } return Task.CompletedTask; diff --git a/src/ServiceControl/Recoverability/Retrying/Handlers/PendingRetriesHandler.cs b/src/ServiceControl/Recoverability/Retrying/Handlers/PendingRetriesHandler.cs index 6177883caa..1190fea600 100644 --- a/src/ServiceControl/Recoverability/Retrying/Handlers/PendingRetriesHandler.cs +++ b/src/ServiceControl/Recoverability/Retrying/Handlers/PendingRetriesHandler.cs @@ -11,7 +11,7 @@ namespace ServiceControl.Recoverability class PendingRetriesHandler : IHandleMessages, IHandleMessages { - public PendingRetriesHandler(IErrorMessageDataStore dataStore) + public PendingRetriesHandler(IFailedMessageRetryDataStore dataStore) { this.dataStore = dataStore; } @@ -24,7 +24,7 @@ public async Task Handle(RetryPendingMessages message, IMessageHandlerContext co foreach (var id in ids) { - await dataStore.RemoveFailedMessageRetryDocument(id); + await dataStore.RemoveFailedMessageRetry(id); messageIds.Add(id); } @@ -35,7 +35,7 @@ public async Task Handle(RetryPendingMessagesById message, IMessageHandlerContex { foreach (var messageUniqueId in message.MessageUniqueIds) { - await dataStore.RemoveFailedMessageRetryDocument(messageUniqueId); + await dataStore.RemoveFailedMessageRetry(messageUniqueId); } await SendRetryMessagesById(context, message.MessageUniqueIds); @@ -55,6 +55,6 @@ static Task SendRetryMessagesById(IMessageHandlerContext context, string[] messa return context.Send(new RetryMessagesById { MessageUniqueIds = messageUniqueIds }, sendOptions); } - readonly IErrorMessageDataStore dataStore; + readonly IFailedMessageRetryDataStore dataStore; } } diff --git a/src/ServiceControl/Recoverability/Retrying/Handlers/RetriesHandler.cs b/src/ServiceControl/Recoverability/Retrying/Handlers/RetriesHandler.cs index 09d7ef11c3..d46f577c99 100644 --- a/src/ServiceControl/Recoverability/Retrying/Handlers/RetriesHandler.cs +++ b/src/ServiceControl/Recoverability/Retrying/Handlers/RetriesHandler.cs @@ -8,7 +8,7 @@ namespace ServiceControl.Recoverability using ServiceControl.Persistence; [Handler] - class RetriesHandler(RetriesGateway retries, IErrorMessageDataStore dataStore) : IHandleMessages, + class RetriesHandler(RetriesGateway retries, IFailedMessageRetryDataStore dataStore) : IHandleMessages, IHandleMessages, IHandleMessages, IHandleMessages, @@ -22,7 +22,7 @@ public Task Handle(MessageFailed message, IMessageHandlerContext context) { if (message.RepeatedFailure) { - return dataStore.RemoveFailedMessageRetryDocument(message.FailedMessageId); + return dataStore.RemoveFailedMessageRetry(message.FailedMessageId); } return Task.CompletedTask; diff --git a/src/ServiceControl/Recoverability/Retrying/Infrastructure/ReturnToSender.cs b/src/ServiceControl/Recoverability/Retrying/Infrastructure/ReturnToSender.cs index 3b27006b4b..1c87eed688 100644 --- a/src/ServiceControl/Recoverability/Retrying/Infrastructure/ReturnToSender.cs +++ b/src/ServiceControl/Recoverability/Retrying/Infrastructure/ReturnToSender.cs @@ -9,7 +9,7 @@ namespace ServiceControl.Recoverability using NServiceBus.Transport; using ServiceControl.Persistence; - class ReturnToSender(IErrorMessageDataStore errorMessageStore, ILogger logger) + class ReturnToSender(IFailedMessageRetryDataStore errorMessageStore, ILogger logger) { public virtual async Task HandleMessage(MessageContext message, IMessageDispatcher sender, string errorQueueTransportAddress, CancellationToken cancellationToken = default) { @@ -59,7 +59,7 @@ public virtual async Task HandleMessage(MessageContext message, IMessageDispatch async Task FetchFromFailedMessage(Dictionary outgoingHeaders, string messageId, string attemptMessageId) { var uniqueMessageId = outgoingHeaders["ServiceControl.Retry.UniqueMessageId"]; - byte[] body = await errorMessageStore.FetchFromFailedMessage(uniqueMessageId); + byte[] body = await errorMessageStore.GetFailedMessageBody(uniqueMessageId); if (body == null) { diff --git a/src/ServiceControl/Recoverability/Retrying/Infrastructure/ReturnToSenderDequeuer.cs b/src/ServiceControl/Recoverability/Retrying/Infrastructure/ReturnToSenderDequeuer.cs index d21d151738..82b59e0b36 100644 --- a/src/ServiceControl/Recoverability/Retrying/Infrastructure/ReturnToSenderDequeuer.cs +++ b/src/ServiceControl/Recoverability/Retrying/Infrastructure/ReturnToSenderDequeuer.cs @@ -16,7 +16,7 @@ class ReturnToSenderDequeuer : IHostedService { public ReturnToSenderDequeuer( ReturnToSender returnToSender, - IErrorMessageDataStore dataStore, + IFailedMessageLifecycleDataStore dataStore, IDomainEvents domainEvents, ITransportCustomization transportCustomization, TransportSettings transportSettings, @@ -180,7 +180,7 @@ async Task StopInternal() class CaptureIfMessageSendingFails { - public CaptureIfMessageSendingFails(IErrorMessageDataStore dataStore, IDomainEvents domainEvents, Action executeOnFailure, ILogger logger) + public CaptureIfMessageSendingFails(IFailedMessageLifecycleDataStore dataStore, IDomainEvents domainEvents, Action executeOnFailure, ILogger logger) { this.dataStore = dataStore; this.executeOnFailure = executeOnFailure; @@ -234,7 +234,7 @@ await domainEvents.Raise(new MessagesSubmittedForRetryFailed } readonly Action executeOnFailure; - readonly IErrorMessageDataStore dataStore; + readonly IFailedMessageLifecycleDataStore dataStore; readonly IDomainEvents domainEvents; readonly ILogger logger; }