[SPIKE] Audits using EF

src/Directory.Packages.props

@@ -19,6 +19,10 @@
     <PackageVersion Include="HdrHistogram" Version="2.5.0" />
     <PackageVersion Include="Microsoft.AspNetCore.Mvc.Testing" Version="8.0.21" />
     <PackageVersion Include="Microsoft.AspNetCore.SignalR.Client" Version="8.0.21" />
+    <PackageVersion Include="Microsoft.EntityFrameworkCore" Version="8.0.11" />
+    <PackageVersion Include="Microsoft.EntityFrameworkCore.Design" Version="8.0.11" />
+    <PackageVersion Include="Microsoft.EntityFrameworkCore.Relational" Version="8.0.11" />
+    <PackageVersion Include="Microsoft.EntityFrameworkCore.SqlServer" Version="8.0.11" />
     <PackageVersion Include="Microsoft.Extensions.DependencyInjection" Version="8.0.1" />
     <PackageVersion Include="Microsoft.Extensions.DependencyModel" Version="8.0.2" />
     <PackageVersion Include="Microsoft.Extensions.Hosting" Version="8.0.1" />
@@ -50,6 +54,7 @@
     <PackageVersion Include="NServiceBus.Transport.Msmq.Sources" Version="3.0.2" />
     <PackageVersion Include="NServiceBus.Transport.SqlServer" Version="8.1.9" />
     <PackageVersion Include="NServiceBus.Transport.PostgreSql" Version="8.1.9" />
+    <PackageVersion Include="Npgsql.EntityFrameworkCore.PostgreSQL" Version="8.0.10" />
     <PackageVersion Include="NuGet.Versioning" Version="6.14.0" />
     <PackageVersion Include="NUnit" Version="4.4.0" />
     <PackageVersion Include="NUnit.Analyzers" Version="4.11.2" />
@@ -62,6 +67,7 @@
     <PackageVersion Include="Particular.LicensingComponent.Report" Version="1.0.0" />
     <PackageVersion Include="Particular.Obsoletes" Version="1.0.0" />
     <PackageVersion Include="Polly.Core" Version="8.5.2" />
+    <PackageVersion Include="Pomelo.EntityFrameworkCore.MySql" Version="8.0.2" />
     <PackageVersion Include="PropertyChanged.Fody" Version="4.1.0" />
     <PackageVersion Include="PropertyChanging.Fody" Version="1.30.3" />
     <PackageVersion Include="PublicApiGenerator" Version="11.5.4" />
@@ -78,6 +84,9 @@
     <PackageVersion Include="System.Reactive.Linq" Version="6.0.1" />
     <PackageVersion Include="System.Reflection.MetadataLoadContext" Version="8.0.1" />
     <PackageVersion Include="System.ServiceProcess.ServiceController" Version="8.0.1" />
+    <PackageVersion Include="Testcontainers.MsSql" Version="4.2.0" />
+    <PackageVersion Include="Testcontainers.MySql" Version="4.2.0" />
+    <PackageVersion Include="Testcontainers.PostgreSql" Version="4.2.0" />
     <PackageVersion Include="Validar.Fody" Version="1.9.0" />
     <PackageVersion Include="Yarp.ReverseProxy" Version="2.3.0" />
   </ItemGroup>

src/ProjectReferences.Persisters.Primary.props

@@ -2,6 +2,9 @@
 
   <ItemGroup Label="Persisters">
     <ProjectReference Include="..\ServiceControl.Persistence.RavenDB\ServiceControl.Persistence.RavenDB.csproj" ReferenceOutputAssembly="false" Private="false" />
+    <ProjectReference Include="..\ServiceControl.Persistence.Sql.SqlServer\ServiceControl.Persistence.Sql.SqlServer.csproj" ReferenceOutputAssembly="false" Private="false" />
+    <ProjectReference Include="..\ServiceControl.Persistence.Sql.PostgreSQL\ServiceControl.Persistence.Sql.PostgreSQL.csproj" ReferenceOutputAssembly="false" Private="false" />
+    <ProjectReference Include="..\ServiceControl.Persistence.Sql.MySQL\ServiceControl.Persistence.Sql.MySQL.csproj" ReferenceOutputAssembly="false" Private="false" />
   </ItemGroup>
 
 </Project>

src/ServiceControl.Audit.Persistence.Sql.Core/.editorconfig

@@ -0,0 +1,9 @@
+[*.cs]
+
+# Justification: ServiceControl app has no synchronization context
+dotnet_diagnostic.CA2007.severity = none
+
+# Disable style rules for auto-generated EF migrations
+[Migrations/**.cs]
+dotnet_diagnostic.IDE0065.severity = none
+generated_code = true

src/ServiceControl.Audit.Persistence.Sql.Core/Abstractions/AuditSqlPersisterSettings.cs

@@ -0,0 +1,17 @@
+namespace ServiceControl.Audit.Persistence.Sql.Core.Abstractions;
+
+public abstract class AuditSqlPersisterSettings : PersistenceSettings
+{
+    protected AuditSqlPersisterSettings(
+        TimeSpan auditRetentionPeriod,
+        bool enableFullTextSearchOnBodies,
+        int maxBodySizeToStore)
+        : base(auditRetentionPeriod, enableFullTextSearchOnBodies, maxBodySizeToStore)
+    {
+    }
+
+    public required string ConnectionString { get; set; }
+    public int CommandTimeout { get; set; } = 30;
+    public bool EnableSensitiveDataLogging { get; set; } = false;
+    public int MinBodySizeForCompression { get; set; } = 4096;
+}

src/ServiceControl.Audit.Persistence.Sql.Core/Abstractions/BaseAuditPersistence.cs

@@ -0,0 +1,21 @@
+namespace ServiceControl.Audit.Persistence.Sql.Core.Abstractions;
+
+using Implementation;
+using Implementation.UnitOfWork;
+using Infrastructure;
+using Microsoft.Extensions.DependencyInjection;
+using ServiceControl.Audit.Auditing.BodyStorage;
+using ServiceControl.Audit.Persistence.UnitOfWork;
+
+public abstract class BaseAuditPersistence
+{
+    protected static void RegisterDataStores(IServiceCollection services)
+    {
+        services.AddSingleton<MinimumRequiredStorageState>();
+        services.AddSingleton<FileSystemBodyStorageHelper>();
+        services.AddSingleton<IBodyStorage, EFBodyStorage>();
+        services.AddSingleton<IAuditDataStore, EFAuditDataStore>();
+        services.AddSingleton<IFailedAuditStorage, EFFailedAuditStorage>();
+        services.AddSingleton<IAuditIngestionUnitOfWorkFactory, AuditIngestionUnitOfWorkFactory>();
+    }
+}

src/ServiceControl.Audit.Persistence.Sql.Core/Abstractions/IAuditDatabaseMigrator.cs

@@ -0,0 +1,6 @@
+namespace ServiceControl.Audit.Persistence.Sql.Core.Abstractions;
+
+public interface IAuditDatabaseMigrator
+{
+    Task ApplyMigrations(CancellationToken cancellationToken = default);
+}

src/ServiceControl.Audit.Persistence.Sql.Core/Abstractions/MinimumRequiredStorageState.cs

@@ -0,0 +1,6 @@
+namespace ServiceControl.Audit.Persistence.Sql.Core.Abstractions;
+
+public class MinimumRequiredStorageState
+{
+    public bool CanIngestMore { get; set; } = true;
+}

src/ServiceControl.Audit.Persistence.Sql.Core/DbContexts/AuditDbContextBase.cs

@@ -0,0 +1,38 @@
+namespace ServiceControl.Audit.Persistence.Sql.Core.DbContexts;
+
+using Entities;
+using EntityConfigurations;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.Extensions.Logging;
+
+public abstract class AuditDbContextBase : DbContext
+{
+    protected AuditDbContextBase(DbContextOptions options) : base(options)
+    {
+    }
+
+    public DbSet<ProcessedMessageEntity> ProcessedMessages { get; set; }
+    public DbSet<FailedAuditImportEntity> FailedAuditImports { get; set; }
+    public DbSet<SagaSnapshotEntity> SagaSnapshots { get; set; }
+
+    protected override void OnConfiguring(DbContextOptionsBuilder optionsBuilder)
+    {
+        optionsBuilder.LogTo(Console.WriteLine, LogLevel.Warning)
+            .EnableDetailedErrors();
+    }
+
+    protected override void OnModelCreating(ModelBuilder modelBuilder)
+    {
+        base.OnModelCreating(modelBuilder);
+
+        modelBuilder.ApplyConfiguration(new ProcessedMessageConfiguration());
+        modelBuilder.ApplyConfiguration(new FailedAuditImportConfiguration());
+        modelBuilder.ApplyConfiguration(new SagaSnapshotConfiguration());
+
+        OnModelCreatingProvider(modelBuilder);
+    }
+
+    protected virtual void OnModelCreatingProvider(ModelBuilder modelBuilder)
+    {
+    }
+}

src/ServiceControl.Audit.Persistence.Sql.Core/Entities/FailedAuditImportEntity.cs

@@ -0,0 +1,8 @@
+namespace ServiceControl.Audit.Persistence.Sql.Core.Entities;
+
+public class FailedAuditImportEntity
+{
+    public Guid Id { get; set; }
+    public string MessageJson { get; set; } = null!;
+    public string? ExceptionInfo { get; set; }
+}

src/ServiceControl.Audit.Persistence.Sql.Core/Entities/ProcessedMessageEntity.cs

@@ -0,0 +1,35 @@
+namespace ServiceControl.Audit.Persistence.Sql.Core.Entities;
+
+public class ProcessedMessageEntity
+{
+    public long Id { get; set; }
+    public string UniqueMessageId { get; set; } = null!;
+
+    // JSON columns for complex nested data
+    public string HeadersJson { get; set; } = null!;
+
+    // Full-text search column (populated from headers and body)
+    public string? Body { get; set; }
+
+    // Denormalized fields for efficient querying
+    public string? MessageId { get; set; }
+    public string? MessageType { get; set; }
+    public DateTime? TimeSent { get; set; }
+    public DateTime ProcessedAt { get; set; }
+    public bool IsSystemMessage { get; set; }
+    public int Status { get; set; }
+    public string? ConversationId { get; set; }
+
+    // Endpoint details (denormalized from MessageMetadata)
+    public string? ReceivingEndpointName { get; set; }
+
+    // Performance metrics (stored as ticks for precision)
+    public long? CriticalTimeTicks { get; set; }
+    public long? ProcessingTimeTicks { get; set; }
+    public long? DeliveryTimeTicks { get; set; }
+
+    // Body storage info
+    public int BodySize { get; set; }
+    public string? BodyUrl { get; set; }
+    public bool BodyNotStored { get; set; }
+}

src/ServiceControl.Audit.Persistence.Sql.Core/Entities/SagaSnapshotEntity.cs

@@ -0,0 +1,18 @@
+namespace ServiceControl.Audit.Persistence.Sql.Core.Entities;
+
+using ServiceControl.SagaAudit;
+
+public class SagaSnapshotEntity
+{
+    public long Id { get; set; }
+    public Guid SagaId { get; set; }
+    public string? SagaType { get; set; }
+    public DateTime StartTime { get; set; }
+    public DateTime FinishTime { get; set; }
+    public SagaStateChangeStatus Status { get; set; }
+    public string? StateAfterChange { get; set; }
+    public string? InitiatingMessageJson { get; set; }
+    public string? OutgoingMessagesJson { get; set; }
+    public string? Endpoint { get; set; }
+    public DateTime ProcessedAt { get; set; }
+}

src/ServiceControl.Audit.Persistence.Sql.Core/EntityConfigurations/FailedAuditImportConfiguration.cs

@@ -0,0 +1,16 @@
+namespace ServiceControl.Audit.Persistence.Sql.Core.EntityConfigurations;
+
+using Entities;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.EntityFrameworkCore.Metadata.Builders;
+
+class FailedAuditImportConfiguration : IEntityTypeConfiguration<FailedAuditImportEntity>
+{
+    public void Configure(EntityTypeBuilder<FailedAuditImportEntity> builder)
+    {
+        builder.ToTable("FailedAuditImports");
+        builder.HasKey(e => e.Id);
+        builder.Property(e => e.Id).IsRequired();
+        builder.Property(e => e.MessageJson).IsRequired();
+    }
+}

src/ServiceControl.Audit.Persistence.Sql.Core/EntityConfigurations/ProcessedMessageConfiguration.cs

@@ -0,0 +1,59 @@
+namespace ServiceControl.Audit.Persistence.Sql.Core.EntityConfigurations;
+
+using Entities;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.EntityFrameworkCore.Metadata.Builders;
+
+class ProcessedMessageConfiguration : IEntityTypeConfiguration<ProcessedMessageEntity>
+{
+    public void Configure(EntityTypeBuilder<ProcessedMessageEntity> builder)
+    {
+        builder.ToTable("ProcessedMessages");
+        builder.HasKey(e => e.Id);
+        builder.Property(e => e.Id).ValueGeneratedOnAdd();
+        builder.Property(e => e.UniqueMessageId).HasMaxLength(200).IsRequired();
+
+        // JSON columns
+        builder.Property(e => e.HeadersJson).IsRequired();
+
+        // Full-text search column
+        builder.Property(e => e.Body); // Will be mapped to text/nvarchar(max) per database
+
+        // Denormalized query fields
+        builder.Property(e => e.MessageId).HasMaxLength(200);
+        builder.Property(e => e.MessageType).HasMaxLength(500);
+        builder.Property(e => e.ConversationId).HasMaxLength(200);
+        builder.Property(e => e.ReceivingEndpointName).HasMaxLength(500);
+        builder.Property(e => e.BodyUrl).HasMaxLength(500);
+        builder.Property(e => e.TimeSent);
+        builder.Property(e => e.ProcessedAt).IsRequired();
+        builder.Property(e => e.IsSystemMessage).IsRequired();
+        builder.Property(e => e.Status).IsRequired();
+        builder.Property(e => e.BodySize).IsRequired();
+        builder.Property(e => e.BodyNotStored).IsRequired();
+        builder.Property(e => e.CriticalTimeTicks);
+        builder.Property(e => e.ProcessingTimeTicks);
+        builder.Property(e => e.DeliveryTimeTicks);
+
+        // PRIMARY: Uniqueness index
+        builder.HasIndex(e => e.UniqueMessageId);
+
+        // COMPOSITE INDEXES: Based on IAuditDataStore query patterns
+
+        // GetMessages: includeSystemMessages, timeSent range, sort by ProcessedAt/TimeSent
+        builder.HasIndex(e => new { e.IsSystemMessage, e.TimeSent, e.ProcessedAt });
+
+        // QueryMessagesByReceivingEndpoint: endpoint + system messages + time range
+        builder.HasIndex(e => new { e.ReceivingEndpointName, e.IsSystemMessage, e.TimeSent, e.ProcessedAt });
+
+        // QueryMessagesByConversationId
+        builder.HasIndex(e => new { e.ConversationId, e.ProcessedAt });
+
+        // QueryAuditCounts: endpoint + system + processed at (date grouping)
+        builder.HasIndex(e => new { e.ReceivingEndpointName, e.IsSystemMessage, e.ProcessedAt });
+
+        // MessageId lookup (for body retrieval)
+        builder.HasIndex(e => e.MessageId);
+        builder.HasIndex(e => e.ProcessedAt);
+    }
+}

src/ServiceControl.Audit.Persistence.Sql.Core/EntityConfigurations/SagaSnapshotConfiguration.cs

@@ -0,0 +1,28 @@
+namespace ServiceControl.Audit.Persistence.Sql.Core.EntityConfigurations;
+
+using Entities;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.EntityFrameworkCore.Metadata.Builders;
+
+class SagaSnapshotConfiguration : IEntityTypeConfiguration<SagaSnapshotEntity>
+{
+    public void Configure(EntityTypeBuilder<SagaSnapshotEntity> builder)
+    {
+        builder.ToTable("SagaSnapshots");
+        builder.HasKey(e => e.Id);
+        builder.Property(e => e.Id).ValueGeneratedOnAdd();
+        builder.Property(e => e.SagaId).IsRequired();
+        builder.Property(e => e.SagaType).IsRequired();
+        builder.Property(e => e.StartTime).IsRequired();
+        builder.Property(e => e.FinishTime);
+        builder.Property(e => e.Status).IsRequired();
+        builder.Property(e => e.StateAfterChange).IsRequired();
+        builder.Property(e => e.InitiatingMessageJson).IsRequired();
+        builder.Property(e => e.OutgoingMessagesJson).IsRequired();
+        builder.Property(e => e.Endpoint).IsRequired();
+        builder.Property(e => e.ProcessedAt).IsRequired();
+
+        builder.HasIndex(e => e.SagaId);
+        builder.HasIndex(e => e.ProcessedAt);
+    }
+}

src/ServiceControl.Audit.Persistence.Sql.Core/FullTextSearch/IAuditFullTextSearchProvider.cs

@@ -0,0 +1,10 @@
+namespace ServiceControl.Audit.Persistence.Sql.Core.FullTextSearch;
+
+using Entities;
+
+public interface IAuditFullTextSearchProvider
+{
+    IQueryable<ProcessedMessageEntity> ApplyFullTextSearch(
+        IQueryable<ProcessedMessageEntity> query,
+        string searchTerms);
+}

src/ServiceControl.Audit.Persistence.Sql.Core/Implementation/EFAuditDataStore.cs

@@ -0,0 +1,39 @@
+namespace ServiceControl.Audit.Persistence.Sql.Core.Implementation;
+
+using ServiceControl.Audit.Auditing;
+using ServiceControl.Audit.Auditing.MessagesView;
+using ServiceControl.Audit.Infrastructure;
+using ServiceControl.Audit.Monitoring;
+using ServiceControl.SagaAudit;
+
+class EFAuditDataStore : IAuditDataStore
+{
+    static readonly QueryStatsInfo EmptyStats = new(string.Empty, 0);
+
+    public Task<QueryResult<IList<KnownEndpointsView>>> QueryKnownEndpoints(CancellationToken cancellationToken)
+        => Task.FromResult(new QueryResult<IList<KnownEndpointsView>>([], EmptyStats));
+
+    public Task<QueryResult<SagaHistory>> QuerySagaHistoryById(Guid input, CancellationToken cancellationToken)
+        => Task.FromResult(QueryResult<SagaHistory>.Empty());
+
+    public Task<QueryResult<IList<MessagesView>>> GetMessages(bool includeSystemMessages, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange? timeSentRange = null, CancellationToken cancellationToken = default)
+        => Task.FromResult(new QueryResult<IList<MessagesView>>([], EmptyStats));
+
+    public Task<QueryResult<IList<MessagesView>>> QueryMessages(string searchParam, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange? timeSentRange = null, CancellationToken cancellationToken = default)
+        => Task.FromResult(new QueryResult<IList<MessagesView>>([], EmptyStats));
+
+    public Task<QueryResult<IList<MessagesView>>> QueryMessagesByReceivingEndpointAndKeyword(string endpoint, string keyword, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange? timeSentRange = null, CancellationToken cancellationToken = default)
+        => Task.FromResult(new QueryResult<IList<MessagesView>>([], EmptyStats));
+
+    public Task<QueryResult<IList<MessagesView>>> QueryMessagesByReceivingEndpoint(bool includeSystemMessages, string endpointName, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange? timeSentRange = null, CancellationToken cancellationToken = default)
+        => Task.FromResult(new QueryResult<IList<MessagesView>>([], EmptyStats));
+
+    public Task<QueryResult<IList<MessagesView>>> QueryMessagesByConversationId(string conversationId, PagingInfo pagingInfo, SortInfo sortInfo, CancellationToken cancellationToken)
+        => Task.FromResult(new QueryResult<IList<MessagesView>>([], EmptyStats));
+
+    public Task<MessageBodyView> GetMessageBody(string messageId, CancellationToken cancellationToken)
+        => Task.FromResult(MessageBodyView.NoContent());
+
+    public Task<QueryResult<IList<AuditCount>>> QueryAuditCounts(string endpointName, CancellationToken cancellationToken)
+        => Task.FromResult(new QueryResult<IList<AuditCount>>([], EmptyStats));
+}

src/ServiceControl.Audit.Persistence.Sql.Core/Implementation/EFBodyStorage.cs

@@ -0,0 +1,17 @@
+namespace ServiceControl.Audit.Persistence.Sql.Core.Implementation;
+
+using Infrastructure;
+using ServiceControl.Audit.Auditing.BodyStorage;
+
+class EFBodyStorage(FileSystemBodyStorageHelper helper) : IBodyStorage
+{
+    public async Task Store(string bodyId, string contentType, int bodySize, Stream bodyStream, CancellationToken cancellationToken)
+    {
+        using var ms = new MemoryStream();
+        await bodyStream.CopyToAsync(ms, cancellationToken).ConfigureAwait(false);
+        await helper.WriteBodyAsync(bodyId, ms.ToArray(), contentType, cancellationToken).ConfigureAwait(false);
+    }
+
+    public Task<StreamResult> TryFetch(string bodyId, CancellationToken cancellationToken)
+        => Task.FromResult(new StreamResult { HasResult = false }); // No-op for initial test
+}