Implemented body storage as a separate table

Author: johnsimons

src/ServiceControl.Persistence.Sql.Core/DbContexts/ServiceControlDbContextBase.cs

@@ -18,6 +18,7 @@ protected ServiceControlDbContextBase(DbContextOptions options) : base(options)
     public DbSet<QueueAddressEntity> QueueAddresses { get; set; }
     public DbSet<KnownEndpointEntity> KnownEndpoints { get; set; }
     public DbSet<CustomCheckEntity> CustomChecks { get; set; }
+    public DbSet<MessageBodyEntity> MessageBodies { get; set; }
 
     protected override void OnModelCreating(ModelBuilder modelBuilder)
     {
@@ -31,6 +32,7 @@ protected override void OnModelCreating(ModelBuilder modelBuilder)
         modelBuilder.ApplyConfiguration(new QueueAddressConfiguration());
         modelBuilder.ApplyConfiguration(new KnownEndpointConfiguration());
         modelBuilder.ApplyConfiguration(new CustomCheckConfiguration());
+        modelBuilder.ApplyConfiguration(new MessageBodyConfiguration());
 
         OnModelCreatingProvider(modelBuilder);
     }

src/ServiceControl.Persistence.Sql.Core/Entities/MessageBodyEntity.cs

@@ -0,0 +1,10 @@
+namespace ServiceControl.Persistence.Sql.Core.Entities;
+
+public class MessageBodyEntity
+{
+    public string Id { get; set; } = null!;
+    public byte[] Body { get; set; } = null!;
+    public string ContentType { get; set; } = null!;
+    public int BodySize { get; set; }
+    public string? Etag { get; set; }
+}

src/ServiceControl.Persistence.Sql.Core/EntityConfigurations/MessageBodyConfiguration.cs

@@ -0,0 +1,19 @@
+namespace ServiceControl.Persistence.Sql.Core.EntityConfigurations;
+
+using Entities;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.EntityFrameworkCore.Metadata.Builders;
+
+class MessageBodyConfiguration : IEntityTypeConfiguration<MessageBodyEntity>
+{
+    public void Configure(EntityTypeBuilder<MessageBodyEntity> builder)
+    {
+        builder.ToTable("MessageBodies");
+        builder.HasKey(e => e.Id);
+        builder.Property(e => e.Id).HasMaxLength(200).IsRequired();
+        builder.Property(e => e.Body).IsRequired();
+        builder.Property(e => e.ContentType).HasMaxLength(200).IsRequired();
+        builder.Property(e => e.BodySize).IsRequired();
+        builder.Property(e => e.Etag).HasMaxLength(100);
+    }
+}

src/ServiceControl.Persistence.Sql.Core/Implementation/BodyStorage.cs

@@ -0,0 +1,44 @@
+namespace ServiceControl.Persistence.Sql.Core.Implementation;
+
+using System;
+using System.IO;
+using System.Threading.Tasks;
+using DbContexts;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.Extensions.DependencyInjection;
+using ServiceControl.Operations.BodyStorage;
+
+public class BodyStorage : IBodyStorage
+{
+    readonly IServiceProvider serviceProvider;
+
+    public BodyStorage(IServiceProvider serviceProvider)
+    {
+        this.serviceProvider = serviceProvider;
+    }
+
+    public async Task<MessageBodyStreamResult> TryFetch(string bodyId)
+    {
+        using var scope = serviceProvider.CreateScope();
+        var dbContext = scope.ServiceProvider.GetRequiredService<ServiceControlDbContextBase>();
+
+        // Try to fetch the body directly by ID
+        var messageBody = await dbContext.MessageBodies
+            .AsNoTracking()
+            .FirstOrDefaultAsync(mb => mb.Id == bodyId);
+
+        if (messageBody == null)
+        {
+            return new MessageBodyStreamResult { HasResult = false };
+        }
+
+        return new MessageBodyStreamResult
+        {
+            HasResult = true,
+            Stream = new MemoryStream(messageBody.Body),
+            ContentType = messageBody.ContentType,
+            BodySize = messageBody.BodySize,
+            Etag = messageBody.Etag
+        };
+    }
+}

src/ServiceControl.Persistence.Sql.MySQL/Migrations/20251209023203_AddMessageBodies.Designer.cs

@@ -0,0 +1,39 @@
+#nullable disable
+
+namespace ServiceControl.Persistence.Sql.MySQL.Migrations
+{
+    using Microsoft.EntityFrameworkCore.Migrations;
+    /// <inheritdoc />
+    public partial class AddMessageBodies : Migration
+    {
+        /// <inheritdoc />
+        protected override void Up(MigrationBuilder migrationBuilder)
+        {
+            migrationBuilder.CreateTable(
+                name: "MessageBodies",
+                columns: table => new
+                {
+                    Id = table.Column<string>(type: "varchar(200)", maxLength: 200, nullable: false)
+                        .Annotation("MySql:CharSet", "utf8mb4"),
+                    Body = table.Column<byte[]>(type: "longblob", nullable: false),
+                    ContentType = table.Column<string>(type: "varchar(200)", maxLength: 200, nullable: false)
+                        .Annotation("MySql:CharSet", "utf8mb4"),
+                    BodySize = table.Column<int>(type: "int", nullable: false),
+                    Etag = table.Column<string>(type: "varchar(100)", maxLength: 100, nullable: true)
+                        .Annotation("MySql:CharSet", "utf8mb4")
+                },
+                constraints: table =>
+                {
+                    table.PrimaryKey("PK_MessageBodies", x => x.Id);
+                })
+                .Annotation("MySql:CharSet", "utf8mb4");
+        }
+
+        /// <inheritdoc />
+        protected override void Down(MigrationBuilder migrationBuilder)
+        {
+            migrationBuilder.DropTable(
+                name: "MessageBodies");
+        }
+    }
+}