Split up saga and message audit stores

src/ServiceControl.Audit.Persistence.InMemory/InMemoryAuditDataStore.cs

@@ -11,7 +11,7 @@
     using ServiceControl.Audit.Infrastructure;
     using ServiceControl.SagaAudit;
 
-    class InMemoryAuditDataStore : IAuditDataStore
+    class InMemoryAuditDataStore : IAuditMessagesViewDataStore, ISagaHistoryDataStore
     {
         IBodyStorage bodyStorage;
         public List<FailedAuditImport> failedAuditImports;

src/ServiceControl.Audit.Persistence.InMemory/InMemoryPersistence.cs

@@ -10,7 +10,8 @@ public void AddPersistence(IServiceCollection services)
         {
             services.AddSingleton(persistenceSettings);
             services.AddSingleton<InMemoryAuditDataStore>();
-            services.AddSingleton<IAuditDataStore>(sp => sp.GetRequiredService<InMemoryAuditDataStore>());
+            services.AddSingleton<IAuditMessagesViewDataStore>(sp => sp.GetRequiredService<InMemoryAuditDataStore>());
+            services.AddSingleton<ISagaHistoryDataStore>(sp => sp.GetRequiredService<InMemoryAuditDataStore>());
             services.AddSingleton<IBodyStorage, InMemoryAttachmentsBodyStorage>();
             services.AddSingleton<IFailedAuditStorage, InMemoryFailedAuditStorage>();
             services.AddSingleton<IAuditIngestionUnitOfWorkFactory, InMemoryAuditIngestionUnitOfWorkFactory>();

src/ServiceControl.Audit.Persistence.RavenDB/RavenAuditDataStore.cs

@@ -16,7 +16,7 @@ namespace ServiceControl.Audit.Persistence.RavenDB
     using Transformers;
 
     class RavenAuditDataStore(IRavenSessionProvider sessionProvider, DatabaseConfiguration databaseConfiguration)
-        : IAuditDataStore
+        : IAuditMessagesViewDataStore, ISagaHistoryDataStore
     {
         public async Task<QueryResult<SagaHistory>> QuerySagaHistoryById(Guid input, CancellationToken cancellationToken = default)
         {

src/ServiceControl.Audit.Persistence.RavenDB/RavenPersistence.cs

@@ -25,7 +25,9 @@ public void AddPersistence(IServiceCollection services)
                 endpointConfiguration.AddCustomCheck<CheckRavenDBSearchEngine>();
             }
 
-            services.AddSingleton<IAuditDataStore, RavenAuditDataStore>();
+            services.AddSingleton<RavenAuditDataStore>();
+            services.AddSingleton<IAuditMessagesViewDataStore>(sp => sp.GetRequiredService<RavenAuditDataStore>());
+            services.AddSingleton<ISagaHistoryDataStore>(sp => sp.GetRequiredService<RavenAuditDataStore>());
             services.AddSingleton<IAuditIngestionUnitOfWorkFactory, RavenAuditIngestionUnitOfWorkFactory>();
             services.AddSingleton<IFailedAuditStorage, RavenFailedAuditStorage>();
             services.AddSingleton<MinimumRequiredStorageState>();

src/ServiceControl.Audit.Persistence.Tests.RavenDB/EmbeddedLifecycleTests.cs

@@ -37,7 +37,7 @@ public override async Task Setup()
         [Test]
         public async Task Verify_embedded_database()
         {
-            await DataStore.GetMessages(false, new PagingInfo(), new SortInfo("Id", "asc"), cancellationToken: TestContext.CurrentContext.CancellationToken);
+            await MessagesViewStore.GetMessages(false, new PagingInfo(), new SortInfo("Id", "asc"), cancellationToken: TestContext.CurrentContext.CancellationToken);
 
             using (Assert.EnterMultipleScope())
             {

src/ServiceControl.Audit.Persistence.Tests.RavenDB/PersistenceTestsConfiguration.cs

@@ -16,7 +16,9 @@
 
     class PersistenceTestsConfiguration
     {
-        public IAuditDataStore AuditDataStore { get; private set; }
+        public IAuditMessagesViewDataStore MessagesViewStore { get; private set; }
+
+        public ISagaHistoryDataStore SagaHistoryStore { get; private set; }
 
         public IFailedAuditStorage FailedAuditStorage { get; private set; }
 
@@ -78,7 +80,8 @@ public async Task Configure(Action<PersistenceSettings> setSettings)
             host = hostBuilder.Build();
             await host.StartAsync();
 
-            AuditDataStore = host.Services.GetRequiredService<IAuditDataStore>();
+            MessagesViewStore = host.Services.GetRequiredService<IAuditMessagesViewDataStore>();
+            SagaHistoryStore = host.Services.GetRequiredService<ISagaHistoryDataStore>();
             FailedAuditStorage = host.Services.GetRequiredService<IFailedAuditStorage>();
 
             var documentStoreProvider = host.Services.GetRequiredService<IRavenDocumentStoreProvider>();

src/ServiceControl.Audit.Persistence.Tests.RavenDB/ProcessedMessageIdTests.cs

@@ -33,7 +33,7 @@ public async Task Retains_existing_document_and_body_ids()
             var expectedId = $"ProcessedMessages-{started.UtcDateTime.Ticks}-{message.GetProcessingId()}";
             Assert.That(message.Id, Is.EqualTo(expectedId));
             Assert.That(message.MessageMetadata["BodyUrl"], Is.EqualTo($"/messages/{expectedId}/body"));
-            var body = await DataStore.GetMessageBody(expectedId);
+            var body = await MessagesViewStore.GetMessageBody(expectedId);
             Assert.That(body.HasContent, Is.True);
             body.StreamContent.Dispose();
         }

src/ServiceControl.Audit.Persistence.Tests.RavenDB/RetentionTests.cs

@@ -43,7 +43,7 @@ public async Task AuditMessageRetention()
             Assert.That(queryResultAfterExpiration.Results, Is.Empty);
 
             Task<QueryResult<IList<MessagesView>>> QueryMessages(CancellationToken cancellationToken) =>
-                DataStore.QueryMessages("MyMessageId", new PagingInfo(), new SortInfo("Id", "asc"), cancellationToken: cancellationToken);
+                MessagesViewStore.QueryMessages("MyMessageId", new PagingInfo(), new SortInfo("Id", "asc"), cancellationToken: cancellationToken);
         }
 
         [Test]
@@ -70,7 +70,7 @@ await IngestSagaAudits(
             Assert.That(queryResultAfterExpiration.Results, Is.Null);
 
             Task<QueryResult<SagaHistory>> QuerySagaHistory(CancellationToken cancellationToken) =>
-                DataStore.QuerySagaHistoryById(sagaId, cancellationToken);
+                SagaHistoryStore.QuerySagaHistoryById(sagaId, cancellationToken);
         }
 
         // The retention window starts at ingestion, so building the indexes of the fresh database must

src/ServiceControl.Audit.Persistence.Tests/AuditCountingTests.cs

@@ -46,9 +46,9 @@ public async Task ShouldCountAuditedMessages()
 
             await IngestProcessedMessagesAudits(messages);
 
-            var endpointA = (await DataStore.QueryAuditCounts("EndpointA", TestContext.CurrentContext.CancellationToken)).Results;
-            var endpointB = (await DataStore.QueryAuditCounts("EndpointB", TestContext.CurrentContext.CancellationToken)).Results;
-            var sysMsgEndpoint = (await DataStore.QueryAuditCounts("SystemEndpoint", TestContext.CurrentContext.CancellationToken)).Results;
+            var endpointA = (await MessagesViewStore.QueryAuditCounts("EndpointA", TestContext.CurrentContext.CancellationToken)).Results;
+            var endpointB = (await MessagesViewStore.QueryAuditCounts("EndpointB", TestContext.CurrentContext.CancellationToken)).Results;
+            var sysMsgEndpoint = (await MessagesViewStore.QueryAuditCounts("SystemEndpoint", TestContext.CurrentContext.CancellationToken)).Results;
 
             Assert.That(sysMsgEndpoint, Is.Empty);
 
@@ -83,7 +83,7 @@ public async Task Should_return_zero_throughput_entry_when_SendOnly()
             await IngestProcessedMessagesAudits(messages);
 
             // Act
-            var result = (await DataStore.QueryAuditCounts(sendOnlyEndpoint, TestContext.CurrentContext.CancellationToken)).Results;
+            var result = (await MessagesViewStore.QueryAuditCounts(sendOnlyEndpoint, TestContext.CurrentContext.CancellationToken)).Results;
 
             // Assert
             Assert.That(result, Is.Not.Empty, "Expected non-empty result for endpoint that only sent messages");

src/ServiceControl.Audit.Persistence.Tests/AuditDataStoreRegistrationTests.cs

@@ -0,0 +1,18 @@
+namespace ServiceControl.Audit.Persistence.Tests
+{
+    using Microsoft.Extensions.DependencyInjection;
+    using NUnit.Framework;
+
+    [TestFixture]
+    class AuditDataStoreRegistrationTests : PersistenceTestFixture
+    {
+        [Test]
+        public void Should_resolve_all_capabilities_from_one_store_instance()
+        {
+            var messagesViewStore = ServiceProvider.GetRequiredService<IAuditMessagesViewDataStore>();
+            var sagaHistoryStore = ServiceProvider.GetRequiredService<ISagaHistoryDataStore>();
+
+            Assert.That(sagaHistoryStore, Is.SameAs(messagesViewStore));
+        }
+    }
+}

src/ServiceControl.Audit.Persistence.Tests/AuditTests.cs

@@ -30,7 +30,7 @@ await IngestProcessedMessagesAudits(
                 message
             );
 
-            var queryResult = await DataStore.QueryMessages("MyMessageId", new PagingInfo(), new SortInfo("Id", "asc"), cancellationToken: TestContext.CurrentContext.CancellationToken);
+            var queryResult = await MessagesViewStore.QueryMessages("MyMessageId", new PagingInfo(), new SortInfo("Id", "asc"), cancellationToken: TestContext.CurrentContext.CancellationToken);
 
             Assert.That(queryResult.Results, Has.Count.EqualTo(1));
             Assert.That(queryResult.Results[0].MessageId, Is.EqualTo("MyMessageId"));
@@ -40,7 +40,7 @@ await IngestProcessedMessagesAudits(
         public async Task Handles_no_results_gracefully()
         {
             var nonExistingMessage = Guid.NewGuid().ToString();
-            var queryResult = await DataStore.QueryMessages(nonExistingMessage, new PagingInfo(), new SortInfo("Id", "asc"), cancellationToken: TestContext.CurrentContext.CancellationToken);
+            var queryResult = await MessagesViewStore.QueryMessages(nonExistingMessage, new PagingInfo(), new SortInfo("Id", "asc"), cancellationToken: TestContext.CurrentContext.CancellationToken);
 
             Assert.That(queryResult.Results, Is.Empty);
         }
@@ -57,7 +57,7 @@ await IngestProcessedMessagesAudits(
                 MakeMessage(conversationId: conversationId)
             );
 
-            var queryResult = await DataStore.QueryMessagesByConversationId(conversationId, new PagingInfo(),
+            var queryResult = await MessagesViewStore.QueryMessagesByConversationId(conversationId, new PagingInfo(),
                 new SortInfo("message_id", "asc"), TestContext.CurrentContext.CancellationToken);
 
             Assert.That(queryResult.Results, Has.Count.EqualTo(2));
@@ -72,7 +72,7 @@ await IngestProcessedMessagesAudits(
                 MakeMessage(messageType: "MyMessageType")
             );
 
-            var queryResult = await DataStore.QueryMessages("MyMessageType", new PagingInfo(),
+            var queryResult = await MessagesViewStore.QueryMessages("MyMessageType", new PagingInfo(),
                 new SortInfo("message_id", "asc"), cancellationToken: TestContext.CurrentContext.CancellationToken);
 
             Assert.That(queryResult.Results, Has.Count.EqualTo(2));
@@ -94,7 +94,7 @@ public async Task Can_roundtrip_message_body()
 
             var bodyId = GetBodyId(processedMessage);
 
-            var retrievedMessage = await DataStore.GetMessageBody(bodyId, TestContext.CurrentContext.CancellationToken);
+            var retrievedMessage = await MessagesViewStore.GetMessageBody(bodyId, TestContext.CurrentContext.CancellationToken);
 
             Assert.That(retrievedMessage, Is.Not.Null);
             using (Assert.EnterMultipleScope())
@@ -131,8 +131,8 @@ public async Task Message_body_validator_is_stable_across_reads()
 
             var bodyId = GetBodyId(processedMessage);
 
-            var first = await DataStore.GetMessageBody(bodyId, TestContext.CurrentContext.CancellationToken);
-            var second = await DataStore.GetMessageBody(bodyId, TestContext.CurrentContext.CancellationToken);
+            var first = await MessagesViewStore.GetMessageBody(bodyId, TestContext.CurrentContext.CancellationToken);
+            var second = await MessagesViewStore.GetMessageBody(bodyId, TestContext.CurrentContext.CancellationToken);
 
             Assert.That(first.ETag, Is.Not.Null.And.Not.Empty, "a body with no validator cannot be revalidated, so conditional GET is dead on it");
             Assert.That(second.ETag, Is.EqualTo(first.ETag), "the body did not change, so neither may its validator");
@@ -153,7 +153,7 @@ public async Task Does_respect_max_message_body()
 
             var bodyId = GetBodyId(processedMessage);
 
-            var retrievedMessage = await DataStore.GetMessageBody(bodyId, TestContext.CurrentContext.CancellationToken);
+            var retrievedMessage = await MessagesViewStore.GetMessageBody(bodyId, TestContext.CurrentContext.CancellationToken);
 
             Assert.That(retrievedMessage, Is.Not.Null);
             using (Assert.EnterMultipleScope())
@@ -180,7 +180,7 @@ public async Task Deduplicates_messages_in_same_batch()
 
             await configuration.CompleteDBOperation();
 
-            var queryResult = await DataStore.GetMessages(false, new PagingInfo(), new SortInfo("message_id", "asc"), cancellationToken: TestContext.CurrentContext.CancellationToken);
+            var queryResult = await MessagesViewStore.GetMessages(false, new PagingInfo(), new SortInfo("message_id", "asc"), cancellationToken: TestContext.CurrentContext.CancellationToken);
 
             Assert.That(queryResult.QueryStats.TotalCount, Is.EqualTo(1));
         }
@@ -204,7 +204,7 @@ public async Task Deduplicates_messages_in_different_batches()
 
             await configuration.CompleteDBOperation();
 
-            var queryResult = await DataStore.GetMessages(false, new PagingInfo(), new SortInfo("message_id", "asc"), cancellationToken: TestContext.CurrentContext.CancellationToken);
+            var queryResult = await MessagesViewStore.GetMessages(false, new PagingInfo(), new SortInfo("message_id", "asc"), cancellationToken: TestContext.CurrentContext.CancellationToken);
 
             Assert.That(queryResult.QueryStats.TotalCount, Is.EqualTo(1));
         }
@@ -227,7 +227,7 @@ public async Task Does_not_deduplicate_with_different_processing_started_header(
 
             await configuration.CompleteDBOperation();
 
-            var queryResult = await DataStore.GetMessages(false, new PagingInfo(), new SortInfo("message_id", "asc"), cancellationToken: TestContext.CurrentContext.CancellationToken);
+            var queryResult = await MessagesViewStore.GetMessages(false, new PagingInfo(), new SortInfo("message_id", "asc"), cancellationToken: TestContext.CurrentContext.CancellationToken);
 
             Assert.That(queryResult.QueryStats.TotalCount, Is.EqualTo(2));
         }

src/ServiceControl.Audit.Persistence.Tests/InMemory/PersistenceTestsConfiguration.cs

@@ -11,7 +11,9 @@
 
     class PersistenceTestsConfiguration
     {
-        public IAuditDataStore AuditDataStore { get; private set; }
+        public IAuditMessagesViewDataStore MessagesViewStore { get; private set; }
+
+        public ISagaHistoryDataStore SagaHistoryStore { get; private set; }
 
         public IFailedAuditStorage FailedAuditStorage { get; private set; }
 
@@ -48,7 +50,8 @@ public async Task Configure(Action<PersistenceSettings> setSettings)
             host = hostBuilder.Build();
             await host.StartAsync();
 
-            AuditDataStore = host.Services.GetRequiredService<IAuditDataStore>();
+            MessagesViewStore = host.Services.GetRequiredService<IAuditMessagesViewDataStore>();
+            SagaHistoryStore = host.Services.GetRequiredService<ISagaHistoryDataStore>();
             FailedAuditStorage = host.Services.GetRequiredService<IFailedAuditStorage>();
             BodyStorage = host.Services.GetService<IBodyStorage>();
             AuditIngestionUnitOfWorkFactory = host.Services.GetRequiredService<IAuditIngestionUnitOfWorkFactory>();

src/ServiceControl.Audit.Persistence.Tests/PersistenceTestFixture.cs

@@ -32,7 +32,9 @@ public virtual Task Cleanup()
 
         protected string PersisterName => configuration.Name;
 
-        protected IAuditDataStore DataStore => configuration.AuditDataStore;
+        protected IAuditMessagesViewDataStore MessagesViewStore => configuration.MessagesViewStore;
+
+        protected ISagaHistoryDataStore SagaHistoryStore => configuration.SagaHistoryStore;
 
         protected IFailedAuditStorage FailedAuditStorage => configuration.FailedAuditStorage;
 

src/ServiceControl.Audit.Persistence.Tests/SagaHistoryTests.cs

@@ -21,7 +21,7 @@ await IngestSagaAudits(
                 }
                 );
 
-            var queryResult = await DataStore.QuerySagaHistoryById(sagaId, TestContext.CurrentContext.CancellationToken);
+            var queryResult = await SagaHistoryStore.QuerySagaHistoryById(sagaId, TestContext.CurrentContext.CancellationToken);
 
             using (Assert.EnterMultipleScope())
             {
@@ -36,7 +36,7 @@ await IngestSagaAudits(
         public async Task Handles_no_results_gracefully()
         {
             var nonExistentSagaId = Guid.NewGuid();
-            var queryResult = await DataStore.QuerySagaHistoryById(nonExistentSagaId, TestContext.CurrentContext.CancellationToken);
+            var queryResult = await SagaHistoryStore.QuerySagaHistoryById(nonExistentSagaId, TestContext.CurrentContext.CancellationToken);
 
             Assert.That(queryResult.Results, Is.Null);
         }
@@ -53,7 +53,7 @@ await IngestSagaAudits(
                 new SagaSnapshot { SagaId = sagaId }
             );
 
-            var queryResult = await DataStore.QuerySagaHistoryById(sagaId, TestContext.CurrentContext.CancellationToken);
+            var queryResult = await SagaHistoryStore.QuerySagaHistoryById(sagaId, TestContext.CurrentContext.CancellationToken);
 
             Assert.That(queryResult.Results.Changes, Has.Count.EqualTo(2));
         }

src/ServiceControl.Audit.Persistence/IAuditDataStore.cs → src/ServiceControl.Audit.Persistence/IAuditMessagesViewDataStore.cs

@@ -7,12 +7,9 @@
     using ServiceControl.Audit.Auditing;
     using ServiceControl.Audit.Auditing.MessagesView;
     using ServiceControl.Audit.Infrastructure;
-    using ServiceControl.Audit.Monitoring;
-    using ServiceControl.SagaAudit;
 
-    public interface IAuditDataStore
+    public interface IAuditMessagesViewDataStore
     {
-        Task<QueryResult<SagaHistory>> QuerySagaHistoryById(Guid input, CancellationToken cancellationToken = default);
         Task<QueryResult<IList<MessagesView>>> GetMessages(bool includeSystemMessages, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange = null, CancellationToken cancellationToken = default);
         Task<QueryResult<IList<MessagesView>>> QueryMessages(string searchParam, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange = null, CancellationToken cancellationToken = default);
         Task<QueryResult<IList<MessagesView>>> QueryMessagesByReceivingEndpointAndKeyword(string endpoint, string keyword, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange = null, CancellationToken cancellationToken = default);

src/ServiceControl.Audit.Persistence/ISagaHistoryDataStore.cs

@@ -0,0 +1,13 @@
+namespace ServiceControl.Audit.Persistence
+{
+    using System;
+    using System.Threading;
+    using System.Threading.Tasks;
+    using ServiceControl.Audit.Auditing.MessagesView;
+    using ServiceControl.SagaAudit;
+
+    public interface ISagaHistoryDataStore
+    {
+        Task<QueryResult<SagaHistory>> QuerySagaHistoryById(Guid input, CancellationToken cancellationToken = default);
+    }
+}

src/ServiceControl.Audit/Auditing/MessagesView/GetMessages2Controller.cs

@@ -12,7 +12,7 @@ namespace ServiceControl.Audit.Auditing.MessagesView;
 
 [ApiController]
 [Route("api")]
-public class GetMessages2Controller(IAuditDataStore dataStore) : ControllerBase
+public class GetMessages2Controller(IAuditMessagesViewDataStore dataStore) : ControllerBase
 {
     [Authorize(Policy = Permissions.AuditMessageView)]
     [Route("messages2")]

src/ServiceControl.Audit/Auditing/MessagesView/GetMessagesController.cs

@@ -13,7 +13,7 @@ namespace ServiceControl.Audit.Auditing.MessagesView
 
     [ApiController]
     [Route("api")]
-    public class GetMessagesController(IAuditDataStore dataStore) : ControllerBase
+    public class GetMessagesController(IAuditMessagesViewDataStore dataStore) : ControllerBase
     {
         [Authorize(Policy = Permissions.AuditMessageView)]
         [Route("messages")]

src/ServiceControl.Audit/Auditing/MessagesView/MessagesConversationController.cs

@@ -12,7 +12,7 @@ namespace ServiceControl.Audit.Auditing.MessagesView
 
     [ApiController]
     [Route("api")]
-    public class MessagesConversationController(IAuditDataStore dataStore) : ControllerBase
+    public class MessagesConversationController(IAuditMessagesViewDataStore dataStore) : ControllerBase
     {
         [Authorize(Policy = Permissions.AuditMessageView)]
         [Route("conversations/{conversationId}")]

src/ServiceControl.Audit/SagaAudit/SagasController.cs

@@ -13,7 +13,7 @@ namespace ServiceControl.Audit.SagaAudit
 
     [ApiController]
     [Route("api")]
-    public class SagasController(IAuditDataStore dataStore) : ControllerBase
+    public class SagasController(ISagaHistoryDataStore dataStore) : ControllerBase
     {
         [Authorize(Policy = Permissions.AuditSagaView)]
         [Route("sagas/{id}")]