diff --git a/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260803010136_AddCustomChecks.Designer.cs b/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260804052012_AddArchiveEntities.Designer.cs similarity index 69% rename from src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260803010136_AddCustomChecks.Designer.cs rename to src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260804052012_AddArchiveEntities.Designer.cs index 5b84e3f762..eb43ef1626 100644 --- a/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260803010136_AddCustomChecks.Designer.cs +++ b/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260804052012_AddArchiveEntities.Designer.cs @@ -13,8 +13,8 @@ namespace ServiceControl.Persistence.EFCore.PostgreSql.Migrations { [DbContext(typeof(PostgreSqlServiceControlDbContext))] - [Migration("20260803010136_AddCustomChecks")] - partial class AddCustomChecks + [Migration("20260804052012_AddArchiveEntities")] + partial class AddArchiveEntities { /// protected override void BuildTargetModel(ModelBuilder modelBuilder) @@ -26,6 +26,70 @@ protected override void BuildTargetModel(ModelBuilder modelBuilder) NpgsqlModelBuilderExtensions.UseIdentityByDefaultColumns(modelBuilder); + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.ArchiveOperationEntity", b => + { + b.Property("RequestId") + .HasMaxLength(64) + .HasColumnType("character varying(64)") + .HasColumnName("request_id"); + + b.Property("ArchiveType") + .HasColumnType("integer") + .HasColumnName("archive_type"); + + b.Property("OperationType") + .HasColumnType("integer") + .HasColumnName("operation_type"); + + b.Property("CurrentBatch") + .HasColumnType("integer") + .HasColumnName("current_batch"); + + b.Property("GroupName") + .IsRequired() + .HasColumnType("text") + .HasColumnName("group_name"); + + b.Property("InitiatedById") + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("initiated_by_id"); + + b.Property("InitiatedByName") + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("initiated_by_name"); + + b.Property("NumberOfBatches") + .HasColumnType("integer") + .HasColumnName("number_of_batches"); + + b.Property("NumberOfMessagesProcessed") + .HasColumnType("integer") + .HasColumnName("number_of_messages_processed"); + + b.Property("OperationId") + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("operation_id"); + + b.Property("Started") + .HasColumnType("timestamp with time zone") + .HasColumnName("started"); + + b.Property("TotalNumberOfMessages") + .HasColumnType("integer") + .HasColumnName("total_number_of_messages"); + + b.HasKey("RequestId", "ArchiveType", "OperationType") + .HasName("pk_archive_operations"); + + b.HasIndex("Started") + .HasDatabaseName("ix_archive_operations_started"); + + b.ToTable("archive_operations", (string)null); + }); + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.CustomCheckEntity", b => { b.Property("Id") @@ -269,11 +333,6 @@ protected override void BuildTargetModel(ModelBuilder modelBuilder) .HasColumnType("integer") .HasColumnName("number_of_processing_attempts"); - b.Property("QueueAddress") - .HasMaxLength(450) - .HasColumnType("character varying(450)") - .HasColumnName("queue_address"); - b.Property("ReceivingEndpointHost") .HasMaxLength(450) .HasColumnType("character varying(450)") @@ -323,9 +382,6 @@ protected override void BuildTargetModel(ModelBuilder modelBuilder) b.HasIndex("FailingEndpointAddress") .HasDatabaseName("ix_failed_messages_failing_endpoint_address"); - b.HasIndex("QueueAddress") - .HasDatabaseName("ix_failed_messages_queue_address"); - b.HasIndex("ReceivingEndpointName") .HasDatabaseName("ix_failed_messages_receiving_endpoint_name"); @@ -370,6 +426,9 @@ protected override void BuildTargetModel(ModelBuilder modelBuilder) b.HasIndex("GroupId") .HasDatabaseName("ix_failed_message_groups_group_id"); + b.HasIndex("Type", "GroupId") + .HasDatabaseName("ix_failed_message_groups_type_group_id"); + b.ToTable("failed_message_groups", (string)null); }); @@ -379,17 +438,41 @@ protected override void BuildTargetModel(ModelBuilder modelBuilder) .HasColumnType("uuid") .HasColumnName("unique_message_id"); - b.Property("RetryId") - .HasMaxLength(450) - .HasColumnType("character varying(450)") - .HasColumnName("retry_id"); + b.Property("RetryBatchId") + .HasColumnType("uuid") + .HasColumnName("retry_batch_id"); + + b.Property("StageAttempts") + .HasColumnType("integer") + .HasColumnName("stage_attempts"); b.HasKey("UniqueMessageId") .HasName("pk_failed_message_retries"); + b.HasIndex("RetryBatchId") + .HasDatabaseName("ix_failed_message_retries_retry_batch_id"); + b.ToTable("failed_message_retries", (string)null); }); + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.GroupCommentEntity", b => + { + b.Property("GroupId") + .HasMaxLength(64) + .HasColumnType("character varying(64)") + .HasColumnName("group_id"); + + b.Property("Comment") + .IsRequired() + .HasColumnType("text") + .HasColumnName("comment"); + + b.HasKey("GroupId") + .HasName("pk_group_comments"); + + b.ToTable("group_comments", (string)null); + }); + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.KnownEndpointEntity", b => { b.Property("Id") @@ -422,6 +505,123 @@ protected override void BuildTargetModel(ModelBuilder modelBuilder) b.ToTable("known_endpoints", (string)null); }); + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.MessageRedirectEntity", b => + { + b.Property("FromPhysicalAddress") + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("from_physical_address"); + + b.Property("LastModified") + .HasColumnType("timestamp with time zone") + .HasColumnName("last_modified"); + + b.Property("ToPhysicalAddress") + .IsRequired() + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("to_physical_address"); + + b.HasKey("FromPhysicalAddress") + .HasName("pk_message_redirects"); + + b.ToTable("message_redirects", (string)null); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.RetryBatchEntity", b => + { + b.Property("Id") + .HasColumnType("uuid") + .HasColumnName("id"); + + b.Property("Classifier") + .HasColumnType("text") + .HasColumnName("classifier"); + + b.Property("Context") + .HasColumnType("text") + .HasColumnName("context"); + + b.Property("InitialBatchSize") + .HasColumnType("integer") + .HasColumnName("initial_batch_size"); + + b.Property("InitiatedById") + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("initiated_by_id"); + + b.Property("InitiatedByName") + .HasColumnType("text") + .HasColumnName("initiated_by_name"); + + b.Property("Last") + .HasColumnType("timestamp with time zone") + .HasColumnName("last"); + + b.Property("OperationId") + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("operation_id"); + + b.Property("Originator") + .HasColumnType("text") + .HasColumnName("originator"); + + b.Property("RequestId") + .IsRequired() + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("request_id"); + + b.Property("RetrySessionId") + .IsRequired() + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("retry_session_id"); + + b.Property("RetryType") + .HasColumnType("integer") + .HasColumnName("retry_type"); + + b.Property("StagingId") + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("staging_id"); + + b.Property("StartTime") + .HasColumnType("timestamp with time zone") + .HasColumnName("start_time"); + + b.Property("Status") + .HasColumnType("integer") + .HasColumnName("status"); + + b.HasKey("Id") + .HasName("pk_retry_batches"); + + b.HasIndex("Status", "RetrySessionId") + .HasDatabaseName("ix_retry_batches_status_retry_session_id"); + + b.ToTable("retry_batches", (string)null); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.RetryBatchNowForwardingEntity", b => + { + b.Property("Id") + .HasColumnType("integer") + .HasColumnName("id"); + + b.Property("RetryBatchId") + .HasColumnType("uuid") + .HasColumnName("retry_batch_id"); + + b.HasKey("Id") + .HasName("pk_retry_batch_now_forwarding"); + + b.ToTable("retry_batch_now_forwarding", (string)null); + }); + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.SubscriptionEntity", b => { b.Property("MessageType") diff --git a/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260803010136_AddCustomChecks.cs b/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260804052012_AddArchiveEntities.cs similarity index 51% rename from src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260803010136_AddCustomChecks.cs rename to src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260804052012_AddArchiveEntities.cs index c939f66397..fc11707147 100644 --- a/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260803010136_AddCustomChecks.cs +++ b/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260804052012_AddArchiveEntities.cs @@ -6,11 +6,33 @@ namespace ServiceControl.Persistence.EFCore.PostgreSql.Migrations { /// - public partial class AddCustomChecks : Migration + public partial class AddArchiveEntities : Migration { /// protected override void Up(MigrationBuilder migrationBuilder) { + migrationBuilder.CreateTable( + name: "archive_operations", + columns: table => new + { + request_id = table.Column(type: "character varying(64)", maxLength: 64, nullable: false), + archive_type = table.Column(type: "integer", nullable: false), + operation_type = table.Column(type: "integer", nullable: false), + group_name = table.Column(type: "text", nullable: false), + total_number_of_messages = table.Column(type: "integer", nullable: false), + number_of_messages_processed = table.Column(type: "integer", nullable: false), + number_of_batches = table.Column(type: "integer", nullable: false), + current_batch = table.Column(type: "integer", nullable: false), + started = table.Column(type: "timestamp with time zone", nullable: false), + initiated_by_id = table.Column(type: "character varying(450)", maxLength: 450, nullable: true), + initiated_by_name = table.Column(type: "character varying(450)", maxLength: 450, nullable: true), + operation_id = table.Column(type: "character varying(450)", maxLength: 450, nullable: true) + }, + constraints: table => + { + table.PrimaryKey("pk_archive_operations", x => new { x.request_id, x.archive_type, x.operation_type }); + }); + migrationBuilder.CreateTable( name: "custom_checks", columns: table => new @@ -30,6 +52,11 @@ protected override void Up(MigrationBuilder migrationBuilder) table.PrimaryKey("pk_custom_checks", x => x.id); }); + migrationBuilder.CreateIndex( + name: "ix_archive_operations_started", + table: "archive_operations", + column: "started"); + migrationBuilder.CreateIndex( name: "ix_custom_checks_reported_at", table: "custom_checks", @@ -44,6 +71,9 @@ protected override void Up(MigrationBuilder migrationBuilder) /// protected override void Down(MigrationBuilder migrationBuilder) { + migrationBuilder.DropTable( + name: "archive_operations"); + migrationBuilder.DropTable( name: "custom_checks"); } diff --git a/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/PostgreSqlServiceControlDbContextModelSnapshot.cs b/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/PostgreSqlServiceControlDbContextModelSnapshot.cs index 8184cac7ba..adee429be3 100644 --- a/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/PostgreSqlServiceControlDbContextModelSnapshot.cs +++ b/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/PostgreSqlServiceControlDbContextModelSnapshot.cs @@ -23,6 +23,70 @@ protected override void BuildModel(ModelBuilder modelBuilder) NpgsqlModelBuilderExtensions.UseIdentityByDefaultColumns(modelBuilder); + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.ArchiveOperationEntity", b => + { + b.Property("RequestId") + .HasMaxLength(64) + .HasColumnType("character varying(64)") + .HasColumnName("request_id"); + + b.Property("ArchiveType") + .HasColumnType("integer") + .HasColumnName("archive_type"); + + b.Property("OperationType") + .HasColumnType("integer") + .HasColumnName("operation_type"); + + b.Property("CurrentBatch") + .HasColumnType("integer") + .HasColumnName("current_batch"); + + b.Property("GroupName") + .IsRequired() + .HasColumnType("text") + .HasColumnName("group_name"); + + b.Property("InitiatedById") + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("initiated_by_id"); + + b.Property("InitiatedByName") + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("initiated_by_name"); + + b.Property("NumberOfBatches") + .HasColumnType("integer") + .HasColumnName("number_of_batches"); + + b.Property("NumberOfMessagesProcessed") + .HasColumnType("integer") + .HasColumnName("number_of_messages_processed"); + + b.Property("OperationId") + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("operation_id"); + + b.Property("Started") + .HasColumnType("timestamp with time zone") + .HasColumnName("started"); + + b.Property("TotalNumberOfMessages") + .HasColumnType("integer") + .HasColumnName("total_number_of_messages"); + + b.HasKey("RequestId", "ArchiveType", "OperationType") + .HasName("pk_archive_operations"); + + b.HasIndex("Started") + .HasDatabaseName("ix_archive_operations_started"); + + b.ToTable("archive_operations", (string)null); + }); + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.CustomCheckEntity", b => { b.Property("Id") diff --git a/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260803010143_AddCustomChecks.Designer.cs b/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260804052028_AddArchiveEntities.Designer.cs similarity index 70% rename from src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260803010143_AddCustomChecks.Designer.cs rename to src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260804052028_AddArchiveEntities.Designer.cs index c923ad9b0f..7e6a395cdf 100644 --- a/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260803010143_AddCustomChecks.Designer.cs +++ b/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260804052028_AddArchiveEntities.Designer.cs @@ -12,8 +12,8 @@ namespace ServiceControl.Persistence.EFCore.SqlServer.Migrations { [DbContext(typeof(SqlServerServiceControlDbContext))] - [Migration("20260803010143_AddCustomChecks")] - partial class AddCustomChecks + [Migration("20260804052028_AddArchiveEntities")] + partial class AddArchiveEntities { /// protected override void BuildTargetModel(ModelBuilder modelBuilder) @@ -25,6 +25,56 @@ protected override void BuildTargetModel(ModelBuilder modelBuilder) SqlServerModelBuilderExtensions.UseIdentityColumns(modelBuilder); + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.ArchiveOperationEntity", b => + { + b.Property("RequestId") + .HasMaxLength(64) + .HasColumnType("nvarchar(64)"); + + b.Property("ArchiveType") + .HasColumnType("int"); + + b.Property("OperationType") + .HasColumnType("int"); + + b.Property("CurrentBatch") + .HasColumnType("int"); + + b.Property("GroupName") + .IsRequired() + .HasColumnType("nvarchar(max)"); + + b.Property("InitiatedById") + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("InitiatedByName") + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("NumberOfBatches") + .HasColumnType("int"); + + b.Property("NumberOfMessagesProcessed") + .HasColumnType("int"); + + b.Property("OperationId") + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("Started") + .HasColumnType("datetime2"); + + b.Property("TotalNumberOfMessages") + .HasColumnType("int"); + + b.HasKey("RequestId", "ArchiveType", "OperationType"); + + b.HasIndex("Started"); + + b.ToTable("ArchiveOperations"); + }); + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.CustomCheckEntity", b => { b.Property("Id") @@ -217,10 +267,6 @@ protected override void BuildTargetModel(ModelBuilder modelBuilder) b.Property("NumberOfProcessingAttempts") .HasColumnType("int"); - b.Property("QueueAddress") - .HasMaxLength(450) - .HasColumnType("nvarchar(450)"); - b.Property("ReceivingEndpointHost") .HasMaxLength(450) .HasColumnType("nvarchar(450)"); @@ -258,8 +304,6 @@ protected override void BuildTargetModel(ModelBuilder modelBuilder) b.HasIndex("FailingEndpointAddress"); - b.HasIndex("QueueAddress"); - b.HasIndex("ReceivingEndpointName"); b.HasIndex("StatusChangedAt") @@ -294,6 +338,8 @@ protected override void BuildTargetModel(ModelBuilder modelBuilder) b.HasIndex("GroupId"); + b.HasIndex("Type", "GroupId"); + b.ToTable("FailedMessageGroups"); }); @@ -302,15 +348,34 @@ protected override void BuildTargetModel(ModelBuilder modelBuilder) b.Property("UniqueMessageId") .HasColumnType("uniqueidentifier"); - b.Property("RetryId") - .HasMaxLength(450) - .HasColumnType("nvarchar(450)"); + b.Property("RetryBatchId") + .HasColumnType("uniqueidentifier"); + + b.Property("StageAttempts") + .HasColumnType("int"); b.HasKey("UniqueMessageId"); + b.HasIndex("RetryBatchId"); + b.ToTable("FailedMessageRetries"); }); + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.GroupCommentEntity", b => + { + b.Property("GroupId") + .HasMaxLength(64) + .HasColumnType("nvarchar(64)"); + + b.Property("Comment") + .IsRequired() + .HasColumnType("nvarchar(max)"); + + b.HasKey("GroupId"); + + b.ToTable("GroupComments"); + }); + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.KnownEndpointEntity", b => { b.Property("Id") @@ -337,6 +402,99 @@ protected override void BuildTargetModel(ModelBuilder modelBuilder) b.ToTable("KnownEndpoints"); }); + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.MessageRedirectEntity", b => + { + b.Property("FromPhysicalAddress") + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("LastModified") + .HasColumnType("datetime2"); + + b.Property("ToPhysicalAddress") + .IsRequired() + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.HasKey("FromPhysicalAddress"); + + b.ToTable("MessageRedirects"); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.RetryBatchEntity", b => + { + b.Property("Id") + .HasColumnType("uniqueidentifier"); + + b.Property("Classifier") + .HasColumnType("nvarchar(max)"); + + b.Property("Context") + .HasColumnType("nvarchar(max)"); + + b.Property("InitialBatchSize") + .HasColumnType("int"); + + b.Property("InitiatedById") + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("InitiatedByName") + .HasColumnType("nvarchar(max)"); + + b.Property("Last") + .HasColumnType("datetime2"); + + b.Property("OperationId") + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("Originator") + .HasColumnType("nvarchar(max)"); + + b.Property("RequestId") + .IsRequired() + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("RetrySessionId") + .IsRequired() + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("RetryType") + .HasColumnType("int"); + + b.Property("StagingId") + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("StartTime") + .HasColumnType("datetime2"); + + b.Property("Status") + .HasColumnType("int"); + + b.HasKey("Id"); + + b.HasIndex("Status", "RetrySessionId"); + + b.ToTable("RetryBatches"); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.RetryBatchNowForwardingEntity", b => + { + b.Property("Id") + .HasColumnType("int"); + + b.Property("RetryBatchId") + .HasColumnType("uniqueidentifier"); + + b.HasKey("Id"); + + b.ToTable("RetryBatchNowForwarding"); + }); + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.SubscriptionEntity", b => { b.Property("MessageType") diff --git a/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260803010143_AddCustomChecks.cs b/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260804052028_AddArchiveEntities.cs similarity index 52% rename from src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260803010143_AddCustomChecks.cs rename to src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260804052028_AddArchiveEntities.cs index d74e29218a..b9b4044806 100644 --- a/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260803010143_AddCustomChecks.cs +++ b/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260804052028_AddArchiveEntities.cs @@ -6,11 +6,33 @@ namespace ServiceControl.Persistence.EFCore.SqlServer.Migrations { /// - public partial class AddCustomChecks : Migration + public partial class AddArchiveEntities : Migration { /// protected override void Up(MigrationBuilder migrationBuilder) { + migrationBuilder.CreateTable( + name: "ArchiveOperations", + columns: table => new + { + RequestId = table.Column(type: "nvarchar(64)", maxLength: 64, nullable: false), + ArchiveType = table.Column(type: "int", nullable: false), + OperationType = table.Column(type: "int", nullable: false), + GroupName = table.Column(type: "nvarchar(max)", nullable: false), + TotalNumberOfMessages = table.Column(type: "int", nullable: false), + NumberOfMessagesProcessed = table.Column(type: "int", nullable: false), + NumberOfBatches = table.Column(type: "int", nullable: false), + CurrentBatch = table.Column(type: "int", nullable: false), + Started = table.Column(type: "datetime2", nullable: false), + InitiatedById = table.Column(type: "nvarchar(450)", maxLength: 450, nullable: true), + InitiatedByName = table.Column(type: "nvarchar(450)", maxLength: 450, nullable: true), + OperationId = table.Column(type: "nvarchar(450)", maxLength: 450, nullable: true) + }, + constraints: table => + { + table.PrimaryKey("PK_ArchiveOperations", x => new { x.RequestId, x.ArchiveType, x.OperationType }); + }); + migrationBuilder.CreateTable( name: "CustomChecks", columns: table => new @@ -30,6 +52,11 @@ protected override void Up(MigrationBuilder migrationBuilder) table.PrimaryKey("PK_CustomChecks", x => x.Id); }); + migrationBuilder.CreateIndex( + name: "IX_ArchiveOperations_Started", + table: "ArchiveOperations", + column: "Started"); + migrationBuilder.CreateIndex( name: "IX_CustomChecks_ReportedAt", table: "CustomChecks", @@ -44,6 +71,9 @@ protected override void Up(MigrationBuilder migrationBuilder) /// protected override void Down(MigrationBuilder migrationBuilder) { + migrationBuilder.DropTable( + name: "ArchiveOperations"); + migrationBuilder.DropTable( name: "CustomChecks"); } diff --git a/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/SqlServerServiceControlDbContextModelSnapshot.cs b/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/SqlServerServiceControlDbContextModelSnapshot.cs index e2c68d1c5b..833fdf1a08 100644 --- a/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/SqlServerServiceControlDbContextModelSnapshot.cs +++ b/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/SqlServerServiceControlDbContextModelSnapshot.cs @@ -22,6 +22,56 @@ protected override void BuildModel(ModelBuilder modelBuilder) SqlServerModelBuilderExtensions.UseIdentityColumns(modelBuilder); + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.ArchiveOperationEntity", b => + { + b.Property("RequestId") + .HasMaxLength(64) + .HasColumnType("nvarchar(64)"); + + b.Property("ArchiveType") + .HasColumnType("int"); + + b.Property("OperationType") + .HasColumnType("int"); + + b.Property("CurrentBatch") + .HasColumnType("int"); + + b.Property("GroupName") + .IsRequired() + .HasColumnType("nvarchar(max)"); + + b.Property("InitiatedById") + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("InitiatedByName") + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("NumberOfBatches") + .HasColumnType("int"); + + b.Property("NumberOfMessagesProcessed") + .HasColumnType("int"); + + b.Property("OperationId") + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("Started") + .HasColumnType("datetime2"); + + b.Property("TotalNumberOfMessages") + .HasColumnType("int"); + + b.HasKey("RequestId", "ArchiveType", "OperationType"); + + b.HasIndex("Started"); + + b.ToTable("ArchiveOperations"); + }); + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.CustomCheckEntity", b => { b.Property("Id") diff --git a/src/ServiceControl.Persistence.EFCore/Abstractions/BasePersistence.cs b/src/ServiceControl.Persistence.EFCore/Abstractions/BasePersistence.cs index 34b80a3644..5b5603731b 100644 --- a/src/ServiceControl.Persistence.EFCore/Abstractions/BasePersistence.cs +++ b/src/ServiceControl.Persistence.EFCore/Abstractions/BasePersistence.cs @@ -8,11 +8,13 @@ namespace ServiceControl.Persistence.EFCore.Abstractions; using ServiceControl.Operations.BodyStorage; using ServiceControl.Persistence.EFCore.Implementation; using ServiceControl.Persistence.EFCore.Implementation.BodyStorage; +using ServiceControl.Persistence.EFCore.Implementation.Recoverability; using ServiceControl.Persistence.EFCore.Implementation.UnitOfWork; using ServiceControl.Persistence.EFCore.Infrastructure; using ServiceControl.Persistence.MessageRedirects; using ServiceControl.Persistence.Recoverability; using ServiceControl.Persistence.UnitOfWork; +using ServiceControl.Recoverability; public abstract class BasePersistence { @@ -33,6 +35,7 @@ protected static void RegisterDataStores(IServiceCollection services, EFPersiste services.AddHostedService(); + services.AddSingleton(); services.AddSingleton(); services.AddSingleton(); services.AddSingleton(); diff --git a/src/ServiceControl.Persistence.EFCore/DbContexts/ServiceControlDbContext.cs b/src/ServiceControl.Persistence.EFCore/DbContexts/ServiceControlDbContext.cs index 8973e49541..a53e3c8e42 100644 --- a/src/ServiceControl.Persistence.EFCore/DbContexts/ServiceControlDbContext.cs +++ b/src/ServiceControl.Persistence.EFCore/DbContexts/ServiceControlDbContext.cs @@ -20,6 +20,7 @@ public abstract class ServiceControlDbContext(DbContextOptions options) : DbCont public DbSet TrialMetadata { get; set; } public DbSet Subscriptions { get; set; } public DbSet EventLogItems { get; set; } + public DbSet ArchiveOperations { get; set; } protected override void OnConfiguring(DbContextOptionsBuilder optionsBuilder) => optionsBuilder.EnableDetailedErrors(); @@ -42,6 +43,7 @@ protected override void OnModelCreating(ModelBuilder modelBuilder) modelBuilder.ApplyConfiguration(new SubscriptionConfiguration()); modelBuilder.ApplyConfiguration(new TrialMetadataConfiguration()); modelBuilder.ApplyConfiguration(new EventLogItemConfiguration()); + modelBuilder.ApplyConfiguration(new ArchiveOperationConfiguration()); } public abstract bool IsDuplicateKeyException(DbUpdateException exception); diff --git a/src/ServiceControl.Persistence.EFCore/Entities/ArchiveOperationEntity.cs b/src/ServiceControl.Persistence.EFCore/Entities/ArchiveOperationEntity.cs new file mode 100644 index 0000000000..ab5a972706 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Entities/ArchiveOperationEntity.cs @@ -0,0 +1,47 @@ +namespace ServiceControl.Persistence.EFCore.Entities; + +using ServiceControl.Recoverability; + +/// +/// Persisted record of an in-progress archive or unarchive operation, enabling resume-after-crash. +/// The composite primary key (RequestId, ArchiveType, IsArchive) enforces one operation per +/// group/type/direction at a time — no separate unique index is needed. +/// +public class ArchiveOperationEntity +{ + /// The group id (or other request id) being archived/unarchived. + public required string RequestId { get; set; } + + /// Display name of the group, captured at operation start. + public required string GroupName { get; set; } + + /// The type of archive operation (FailureGroup, SingleMessage, etc.). + public ArchiveType ArchiveType { get; set; } + + /// Distinguishes archive from unarchive. + public ArchiveOperationType OperationType { get; set; } + + /// Total number of messages in the group at operation start. + public int TotalNumberOfMessages { get; set; } + + /// Number of messages processed so far (resume checkpoint). + public int NumberOfMessagesProcessed { get; set; } + + /// Total number of batches planned. + public int NumberOfBatches { get; set; } + + /// Current batch number (resume checkpoint, 0-based). + public int CurrentBatch { get; set; } + + /// When the operation started (UTC). + public DateTime Started { get; set; } + + /// Audit attribution: the id of the user who initiated the operation. + public string? InitiatedById { get; set; } + + /// Audit attribution: the name of the user who initiated the operation. + public string? InitiatedByName { get; set; } + + /// Audit attribution: the operation id used to correlate per-message audit entries. + public string? OperationId { get; set; } +} \ No newline at end of file diff --git a/src/ServiceControl.Persistence.EFCore/Entities/ArchiveOperationType.cs b/src/ServiceControl.Persistence.EFCore/Entities/ArchiveOperationType.cs new file mode 100644 index 0000000000..0cb7984ac8 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Entities/ArchiveOperationType.cs @@ -0,0 +1,7 @@ +namespace ServiceControl.Persistence.EFCore.Entities; + +public enum ArchiveOperationType +{ + Archive, + UnArchive +} \ No newline at end of file diff --git a/src/ServiceControl.Persistence.EFCore/EntityConfigurations/ArchiveOperationConfiguration.cs b/src/ServiceControl.Persistence.EFCore/EntityConfigurations/ArchiveOperationConfiguration.cs new file mode 100644 index 0000000000..a38d4e3867 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/EntityConfigurations/ArchiveOperationConfiguration.cs @@ -0,0 +1,32 @@ +namespace ServiceControl.Persistence.EFCore.EntityConfigurations; + +using Entities; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Metadata.Builders; + +class ArchiveOperationConfiguration : IEntityTypeConfiguration +{ + public void Configure(EntityTypeBuilder builder) + { + // Composite primary key — the natural key that distinguishes one operation from another. + // This also serves as the uniqueness constraint: one operation per (RequestId, ArchiveType, OperationType). + builder.HasKey(e => new { e.RequestId, e.ArchiveType, e.OperationType }); + + builder.Property(e => e.RequestId).HasMaxLength(64).IsRequired(); + builder.Property(e => e.GroupName).IsRequired(); + builder.Property(e => e.ArchiveType).IsRequired(); + builder.Property(e => e.OperationType).IsRequired(); + builder.Property(e => e.TotalNumberOfMessages).IsRequired(); + builder.Property(e => e.NumberOfMessagesProcessed).IsRequired(); + builder.Property(e => e.NumberOfBatches).IsRequired(); + builder.Property(e => e.CurrentBatch).IsRequired(); + builder.Property(e => e.Started).IsRequired(); + + builder.Property(e => e.InitiatedById).HasMaxLength(ColumnLengths.ShortTextLength); + builder.Property(e => e.InitiatedByName).HasMaxLength(ColumnLengths.ShortTextLength); + builder.Property(e => e.OperationId).HasMaxLength(ColumnLengths.ShortTextLength); + + // Non-unique index for cleanup / diagnostics queries + builder.HasIndex(e => e.Started); + } +} \ No newline at end of file diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageLifecycleDataStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageLifecycleDataStore.cs index 07f3b724d7..9c51049517 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageLifecycleDataStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageLifecycleDataStore.cs @@ -1,24 +1,141 @@ namespace ServiceControl.Persistence.EFCore.Implementation; +using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.DependencyInjection; +using ServiceControl.MessageFailures; /// /// 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 async Task MarkAsArchived(string failedMessageId) + { + await ExecuteWithDbContext(async dbContext => + { + if (!Guid.TryParse(failedMessageId, out var uniqueMessageId)) + { + return; + } - public Task MarkAsResolved(string failedMessageId) => - throw new NotImplementedException(); + var now = DateTime.UtcNow; - public Task UnArchiveMessages(IEnumerable failedMessageIds) => - throw new NotImplementedException(); + await dbContext.FailedMessages + .Where(fm => fm.UniqueMessageId == uniqueMessageId) + .ExecuteUpdateAsync(s => s + .SetProperty(fm => fm.Status, FailedMessageStatus.Archived) + .SetProperty(fm => fm.StatusChangedAt, now) + .SetProperty(fm => fm.LastModified, now)); + }); + } - public Task UnArchiveMessagesByRange(DateTime from, DateTime to) => - throw new NotImplementedException(); + public async Task MarkAsResolved(string failedMessageId) + { + return await ExecuteWithDbContext(async dbContext => + { + if (!Guid.TryParse(failedMessageId, out var uniqueMessageId)) + { + return false; + } - public Task RevertRetry(string messageUniqueId) => - throw new NotImplementedException(); -} + var now = DateTime.UtcNow; + + var affected = await dbContext.FailedMessages + .Where(fm => fm.UniqueMessageId == uniqueMessageId && fm.Status == FailedMessageStatus.Unresolved) + .ExecuteUpdateAsync(s => s + .SetProperty(fm => fm.Status, FailedMessageStatus.Resolved) + .SetProperty(fm => fm.StatusChangedAt, now) + .SetProperty(fm => fm.LastModified, now)); + + return affected > 0; + }); + } + + public async Task UnArchiveMessages(IEnumerable failedMessageIds) + { + var ids = failedMessageIds + .Select(id => Guid.TryParse(id, out var guid) ? guid : Guid.Empty) + .Where(guid => guid != Guid.Empty) + .ToList(); + + if (ids.Count == 0) + { + return []; + } + + return await ExecuteWithDbContext(async dbContext => + { + var now = DateTime.UtcNow; + + // Query which messages will actually be unarchived (must be Archived status) + var unarchivableIds = await dbContext.FailedMessages + .Where(fm => ids.Contains(fm.UniqueMessageId) && fm.Status == FailedMessageStatus.Archived) + .Select(fm => fm.UniqueMessageId) + .ToListAsync(); + + if (unarchivableIds.Count == 0) + { + return []; + } + + await dbContext.FailedMessages + .Where(fm => unarchivableIds.Contains(fm.UniqueMessageId) && fm.Status == FailedMessageStatus.Archived) + .ExecuteUpdateAsync(s => s + .SetProperty(fm => fm.Status, FailedMessageStatus.Unresolved) + .SetProperty(fm => fm.StatusChangedAt, now) + .SetProperty(fm => fm.LastModified, now)); + + return unarchivableIds.Select(id => id.ToString()).ToArray(); + }); + } + + public async Task UnArchiveMessagesByRange(DateTime from, DateTime to) + { + return await ExecuteWithDbContext(async dbContext => + { + var now = DateTime.UtcNow; + + // Query which messages will be unarchived (must be Archived and within the date range) + var unarchivableIds = await dbContext.FailedMessages + .Where(fm => fm.Status == FailedMessageStatus.Archived + && fm.LastTimeOfFailure >= from + && fm.LastTimeOfFailure <= to) + .Select(fm => fm.UniqueMessageId) + .ToListAsync(); + + if (unarchivableIds.Count == 0) + { + return []; + } + + await dbContext.FailedMessages + .Where(fm => unarchivableIds.Contains(fm.UniqueMessageId) && fm.Status == FailedMessageStatus.Archived) + .ExecuteUpdateAsync(s => s + .SetProperty(fm => fm.Status, FailedMessageStatus.Unresolved) + .SetProperty(fm => fm.StatusChangedAt, now) + .SetProperty(fm => fm.LastModified, now)); + + return unarchivableIds.Select(id => id.ToString()).ToArray(); + }); + } + + public async Task RevertRetry(string messageUniqueId) + { + await ExecuteWithDbContext(async dbContext => + { + if (!Guid.TryParse(messageUniqueId, out var uniqueMessageId)) + { + return; + } + + var now = DateTime.UtcNow; + + await dbContext.FailedMessages + .Where(fm => fm.UniqueMessageId == uniqueMessageId && fm.Status == FailedMessageStatus.RetryIssued) + .ExecuteUpdateAsync(s => s + .SetProperty(fm => fm.Status, FailedMessageStatus.Unresolved) + .SetProperty(fm => fm.StatusChangedAt, now) + .SetProperty(fm => fm.LastModified, now)); + }); + } +} \ No newline at end of file diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/MessageArchiver.cs b/src/ServiceControl.Persistence.EFCore/Implementation/MessageArchiver.cs deleted file mode 100644 index e56575df4d..0000000000 --- a/src/ServiceControl.Persistence.EFCore/Implementation/MessageArchiver.cs +++ /dev/null @@ -1,34 +0,0 @@ -namespace ServiceControl.Persistence.EFCore.Implementation; - -using ServiceControl.Infrastructure.Auth; -using ServiceControl.Persistence.Recoverability; -using ServiceControl.Recoverability; - -public class MessageArchiver : IArchiveMessages -{ - // must set StatusChangedAt + LastModified - public Task ArchiveAllInGroup(string groupId, AuditUser? initiatedBy = null, string? operationId = null) => - throw new NotImplementedException(); - - // must set StatusChangedAt + LastModified - public Task UnarchiveAllInGroup(string groupId, AuditUser? initiatedBy = null, string? operationId = null) => - throw new NotImplementedException(); - - public bool IsOperationInProgressFor(string groupId, ArchiveType archiveType) => - throw new NotImplementedException(); - - public bool IsArchiveInProgressFor(string groupId) => - throw new NotImplementedException(); - - public void DismissArchiveOperation(string groupId, ArchiveType archiveType) => - throw new NotImplementedException(); - - public Task StartArchiving(string groupId, ArchiveType archiveType) => - throw new NotImplementedException(); - - public Task StartUnarchiving(string groupId, ArchiveType archiveType) => - throw new NotImplementedException(); - - public IEnumerable GetArchivalOperations() => - throw new NotImplementedException(); -} diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/Recoverability/EFCoreArchivingManager.cs b/src/ServiceControl.Persistence.EFCore/Implementation/Recoverability/EFCoreArchivingManager.cs new file mode 100644 index 0000000000..7d9e29f268 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Implementation/Recoverability/EFCoreArchivingManager.cs @@ -0,0 +1,91 @@ +namespace ServiceControl.Persistence.EFCore.Implementation; + +using ServiceControl.Infrastructure.DomainEvents; +using ServiceControl.Persistence.EFCore.Entities; +using ServiceControl.Recoverability; + +/// +/// EFCore equivalent of the RavenDB . Wraps the shared +/// singleton to manage in-memory archive progress state. +/// +class EFCoreArchivingManager(IDomainEvents domainEvents, OperationsManager operationsManager) +{ + InMemoryArchive GetOrCreate(ArchiveType archiveType, string requestId) + { + var id = InMemoryArchive.MakeId(requestId, archiveType); + if (!operationsManager.ArchiveOperations.TryGetValue(id, out var summary)) + { + summary = new InMemoryArchive(requestId, archiveType, domainEvents); + operationsManager.ArchiveOperations[id] = summary; + } + + return summary; + } + + public Task StartArchiving(ArchiveOperationEntity operation) + { + var summary = GetOrCreate(operation.ArchiveType, operation.RequestId); + + summary.TotalNumberOfMessages = operation.TotalNumberOfMessages; + summary.NumberOfMessagesArchived = operation.NumberOfMessagesProcessed; + summary.Started = operation.Started; + summary.GroupName = operation.GroupName; + summary.NumberOfBatches = operation.NumberOfBatches; + summary.CurrentBatch = operation.CurrentBatch; + + return summary.Start(); + } + + public Task StartArchiving(string requestId, ArchiveType archiveType) + { + var summary = GetOrCreate(archiveType, requestId); + + summary.TotalNumberOfMessages = 0; + summary.NumberOfMessagesArchived = 0; + summary.Started = DateTime.UtcNow; + summary.GroupName = "Undefined"; + summary.NumberOfBatches = 0; + summary.CurrentBatch = 0; + + return summary.Start(); + } + + public InMemoryArchive? GetStatusForArchiveOperation(string requestId, ArchiveType archiveType) + { + operationsManager.ArchiveOperations.TryGetValue(InMemoryArchive.MakeId(requestId, archiveType), out var summary); + return summary; + } + + public Task BatchArchived(string requestId, ArchiveType archiveType, int numberOfMessagesArchivedInBatch) + { + var summary = GetOrCreate(archiveType, requestId); + return summary.BatchArchived(numberOfMessagesArchivedInBatch); + } + + public Task ArchiveOperationFinalizing(string requestId, ArchiveType archiveType) + { + var summary = GetOrCreate(archiveType, requestId); + return summary.FinalizeArchive(); + } + + public Task ArchiveOperationCompleted(string requestId, ArchiveType archiveType) + { + var summary = GetOrCreate(archiveType, requestId); + return summary.Complete(); + } + + public bool IsArchiveInProgressFor(string requestId) + { + return operationsManager.ArchiveOperations.Keys.Any(key => key.EndsWith($"/{requestId}")); + } + + public IEnumerable GetArchivalOperations() + { + return operationsManager.ArchiveOperations.Values; + } + + public void DismissArchiveOperation(string requestId, ArchiveType archiveType) + { + operationsManager.ArchiveOperations.Remove(InMemoryArchive.MakeId(requestId, archiveType)); + } +} \ No newline at end of file diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/Recoverability/EFCoreUnarchivingManager.cs b/src/ServiceControl.Persistence.EFCore/Implementation/Recoverability/EFCoreUnarchivingManager.cs new file mode 100644 index 0000000000..f9ec6a0996 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Implementation/Recoverability/EFCoreUnarchivingManager.cs @@ -0,0 +1,70 @@ +namespace ServiceControl.Persistence.EFCore.Implementation; + +using ServiceControl.Infrastructure.DomainEvents; +using ServiceControl.Persistence.EFCore.Entities; +using ServiceControl.Recoverability; + +/// +/// EFCore equivalent of the RavenDB . Wraps the shared +/// singleton to manage in-memory unarchive progress state. +/// +class EFCoreUnarchivingManager(IDomainEvents domainEvents, OperationsManager operationsManager) +{ + InMemoryUnarchive GetOrCreate(ArchiveType archiveType, string requestId) + { + var id = InMemoryUnarchive.MakeId(requestId, archiveType); + if (!operationsManager.UnarchiveOperations.TryGetValue(id, out var summary)) + { + summary = new InMemoryUnarchive(requestId, archiveType, domainEvents); + operationsManager.UnarchiveOperations[id] = summary; + } + + return summary; + } + + public Task StartUnarchiving(ArchiveOperationEntity operation) + { + var summary = GetOrCreate(operation.ArchiveType, operation.RequestId); + + summary.TotalNumberOfMessages = operation.TotalNumberOfMessages; + summary.NumberOfMessagesUnarchived = operation.NumberOfMessagesProcessed; + summary.Started = operation.Started; + summary.GroupName = operation.GroupName; + summary.NumberOfBatches = operation.NumberOfBatches; + summary.CurrentBatch = operation.CurrentBatch; + + return summary.Start(); + } + + public Task StartUnarchiving(string requestId, ArchiveType archiveType) + { + var summary = GetOrCreate(archiveType, requestId); + + summary.TotalNumberOfMessages = 0; + summary.NumberOfMessagesUnarchived = 0; + summary.Started = DateTime.UtcNow; + summary.GroupName = "Undefined"; + summary.NumberOfBatches = 0; + summary.CurrentBatch = 0; + + return summary.Start(); + } + + public Task BatchUnarchived(string requestId, ArchiveType archiveType, int numberOfMessagesUnarchivedInBatch) + { + var summary = GetOrCreate(archiveType, requestId); + return summary.BatchUnarchived(numberOfMessagesUnarchivedInBatch); + } + + public Task UnarchiveOperationFinalizing(string requestId, ArchiveType archiveType) + { + var summary = GetOrCreate(archiveType, requestId); + return summary.FinalizeUnarchive(); + } + + public Task UnarchiveOperationCompleted(string requestId, ArchiveType archiveType) + { + var summary = GetOrCreate(archiveType, requestId); + return summary.Complete(); + } +} \ No newline at end of file diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/Recoverability/MessageArchiver.cs b/src/ServiceControl.Persistence.EFCore/Implementation/Recoverability/MessageArchiver.cs new file mode 100644 index 0000000000..30f9b7f902 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Implementation/Recoverability/MessageArchiver.cs @@ -0,0 +1,339 @@ +namespace ServiceControl.Persistence.EFCore.Implementation.Recoverability; + +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; +using ServiceControl.Infrastructure.Auth; +using ServiceControl.Infrastructure.DomainEvents; +using ServiceControl.MessageFailures; +using ServiceControl.Persistence.EFCore.DbContexts; +using ServiceControl.Persistence.EFCore.Entities; +using ServiceControl.Persistence.Recoverability; +using ServiceControl.Recoverability; + +public class MessageArchiver : IArchiveMessages +{ + public MessageArchiver( + IServiceScopeFactory scopeFactory, + OperationsManager operationsManager, + IDomainEvents domainEvents, + IMessageActionAuditLog auditLog, + TimeProvider timeProvider, + ILogger logger + ) + { + this.scopeFactory = scopeFactory; + this.operationsManager = operationsManager; + this.auditLog = auditLog; + this.timeProvider = timeProvider; + this.logger = logger; + this.domainEvents = domainEvents; + + archivingManager = new EFCoreArchivingManager(domainEvents, operationsManager); + unarchivingManager = new EFCoreUnarchivingManager(domainEvents, operationsManager); + } + + public async Task ArchiveAllInGroup(string groupId, AuditUser? initiatedBy = null, string? operationId = null) + { + logger.LogInformation("Archiving of {GroupId} started", groupId); + + ArchiveOperationEntity? operationEntity; + AuditUser auditUser; + string? auditOperationId; + + await using (var scope = scopeFactory.CreateAsyncScope()) + { + var dbContext = scope.ServiceProvider.GetRequiredService(); + + operationEntity = await GetOrCreateOperation(dbContext, groupId, ArchiveOperationType.Archive, initiatedBy, operationId); + if (operationEntity == null) + { + return; + } + + // Capture audit attribution from the persisted entity + auditUser = new AuditUser(operationEntity.InitiatedById ?? AuditUser.AnonymousValue, operationEntity.InitiatedByName ?? AuditUser.AnonymousValue); + auditOperationId = operationEntity.OperationId; + } + + // ── Start in-memory tracking ── + await archivingManager.StartArchiving(operationEntity); + // ── Batch loop ── + string[] batchIds; + do + { + await using var batchScope = scopeFactory.CreateAsyncScope(); + var batchDbContext = batchScope.ServiceProvider.GetRequiredService(); + + operationEntity = await batchDbContext.ArchiveOperations.FindAsync(groupId, ArchiveType.FailureGroup, ArchiveOperationType.Archive) + ?? throw new InvalidOperationException($"No in progress Archive Operation found for {groupId}"); + + batchIds = await UpdateGroupStatusAsync(batchDbContext, groupId, FailedMessageStatus.Unresolved, FailedMessageStatus.Archived, batchSize); + await archivingManager.BatchArchived(groupId, ArchiveType.FailureGroup, batchIds.Length); + + // Update progress tracking + operationEntity.CurrentBatch++; + operationEntity.NumberOfMessagesProcessed += batchIds.Length; + await batchDbContext.SaveChangesAsync(); + + // Raise batch domain event + await domainEvents.Raise(new FailedMessageGroupBatchArchived { FailedMessagesIds = batchIds }); + + // Per-message audit + AuditArchivedMessages(MessageActionKind.Archive, Permissions.ErrorRecoverabilityGroupsArchive, auditUser, auditOperationId, batchIds); + + logger.LogInformation("Archiving of {MessageCount} messages from group {GroupId} completed", batchIds.Length, groupId); + } while (batchIds.Length >= batchSize); + + // ── Finalize ── + logger.LogInformation("Archiving of group {GroupId} is complete", groupId); + await archivingManager.ArchiveOperationFinalizing(groupId, ArchiveType.FailureGroup); + await archivingManager.ArchiveOperationCompleted(groupId, ArchiveType.FailureGroup); + + // Delete the operation row + await using (var finalizeScope = scopeFactory.CreateAsyncScope()) + { + var finalizeDbContext = finalizeScope.ServiceProvider.GetRequiredService(); + await finalizeDbContext.ArchiveOperations + .Where(o => + o.ArchiveType == operationEntity.ArchiveType + && o.RequestId == operationEntity.RequestId + && o.OperationType == operationEntity.OperationType) + .ExecuteDeleteAsync(); + } + + await domainEvents.Raise(new FailedMessageGroupArchived + { + GroupId = groupId, + GroupName = operationEntity.GroupName, + MessagesCount = operationEntity.TotalNumberOfMessages + }); + + logger.LogInformation("Archiving of group {GroupId} completed", groupId); + } + + public async Task UnarchiveAllInGroup(string groupId, AuditUser? initiatedBy = null, string? operationId = null) + { + logger.LogInformation("Unarchiving of {GroupId} started", groupId); + + ArchiveOperationEntity? operationEntity; + AuditUser auditUser; + string? auditOperationId; + + // ── Load-or-create operation row ── + using (var scope = scopeFactory.CreateAsyncScope()) + { + var dbContext = scope.ServiceProvider.GetRequiredService(); + + operationEntity = await GetOrCreateOperation(dbContext, groupId, ArchiveOperationType.UnArchive, initiatedBy, operationId); + if (operationEntity == null) + { + return; + } + + // Capture audit attribution from the persisted entity + auditUser = new AuditUser(operationEntity.InitiatedById ?? AuditUser.AnonymousValue, operationEntity.InitiatedByName ?? AuditUser.AnonymousValue); + auditOperationId = operationEntity.OperationId; + } + + await unarchivingManager.StartUnarchiving(operationEntity); + string[] batchIds; + do + { + await using var batchScope = scopeFactory.CreateAsyncScope(); + var batchDbContext = batchScope.ServiceProvider.GetRequiredService(); + operationEntity = await batchDbContext.ArchiveOperations.FindAsync(groupId, ArchiveType.FailureGroup, ArchiveOperationType.UnArchive) + ?? throw new InvalidOperationException($"No in progress Unarchive Operation found for {groupId}"); + + batchIds = await UpdateGroupStatusAsync(batchDbContext, groupId, FailedMessageStatus.Archived, FailedMessageStatus.Unresolved, batchSize); + + await unarchivingManager.BatchUnarchived(groupId, ArchiveType.FailureGroup, batchIds.Length); + + // Update progress tracking + operationEntity.CurrentBatch++; + operationEntity.NumberOfMessagesProcessed += batchIds.Length; + await batchDbContext.SaveChangesAsync(); + + // Raise batch domain event + await domainEvents.Raise(new FailedMessageGroupBatchUnarchived { FailedMessagesIds = batchIds }); + + // Per-message audit + AuditArchivedMessages(MessageActionKind.Unarchive, Permissions.ErrorRecoverabilityGroupsUnarchive, auditUser, auditOperationId, batchIds); + + logger.LogInformation("Unarchiving of {MessageCount} messages from group {GroupId} completed", batchIds.Length, groupId); + } while (batchIds.Length >= batchSize); + + // ── Finalize ── + logger.LogInformation("Unarchiving of group {GroupId} is complete", groupId); + await unarchivingManager.UnarchiveOperationFinalizing(groupId, ArchiveType.FailureGroup); + await unarchivingManager.UnarchiveOperationCompleted(groupId, ArchiveType.FailureGroup); + + // Delete the operation row + await using (var finalizeScope = scopeFactory.CreateAsyncScope()) + { + var finalizeDbContext = finalizeScope.ServiceProvider.GetRequiredService(); + await finalizeDbContext.ArchiveOperations + .Where(o => + o.ArchiveType == operationEntity.ArchiveType + && o.RequestId == operationEntity.RequestId + && o.OperationType == operationEntity.OperationType) + .ExecuteDeleteAsync(); + } + + await domainEvents.Raise(new FailedMessageGroupUnarchived + { + GroupId = groupId, + GroupName = operationEntity.GroupName, + MessagesCount = operationEntity.TotalNumberOfMessages + }); + + logger.LogInformation("Unarchiving of group {GroupId} completed", groupId); + } + + public bool IsOperationInProgressFor(string groupId, ArchiveType archiveType) + => operationsManager.IsOperationInProgressFor(groupId, archiveType); + + public bool IsArchiveInProgressFor(string groupId) + => archivingManager.IsArchiveInProgressFor(groupId); + + public void DismissArchiveOperation(string groupId, ArchiveType archiveType) + => archivingManager.DismissArchiveOperation(groupId, archiveType); + + public Task StartArchiving(string groupId, ArchiveType archiveType) + => archivingManager.StartArchiving(groupId, archiveType); + + public Task StartUnarchiving(string groupId, ArchiveType archiveType) + => unarchivingManager.StartUnarchiving(groupId, archiveType); + + public IEnumerable GetArchivalOperations() + => archivingManager.GetArchivalOperations(); + + async Task GetOrCreateOperation(ServiceControlDbContext dbContext, string groupId, ArchiveOperationType operation, AuditUser? initiatedBy, string? operationId) + { + ArchiveOperationEntity? operationEntity = await dbContext.ArchiveOperations.FindAsync(groupId, ArchiveType.FailureGroup, operation); + + if (operationEntity != null) + { + logger.LogInformation("Resuming {OperationType} operation for group {GroupId} at batch {CurrentBatch}/{NumberOfBatches}", operation.ToString(), groupId, operationEntity.CurrentBatch, operationEntity.NumberOfBatches); + } + else + { + var targetStatus = operation == ArchiveOperationType.Archive ? FailedMessageStatus.Unresolved : FailedMessageStatus.Archived; + var (count, groupName) = await GetGroupDetails(dbContext, groupId, targetStatus); + if (count == 0) + { + logger.LogWarning("No messages to {OperationType} in group {GroupId}", operation.ToString(), groupId); + return operationEntity; + } + + operationEntity = new ArchiveOperationEntity + { + RequestId = groupId, + GroupName = groupName, + ArchiveType = ArchiveType.FailureGroup, + OperationType = operation, + TotalNumberOfMessages = count, + NumberOfMessagesProcessed = 0, + NumberOfBatches = (int)Math.Ceiling(count / (float)batchSize), + CurrentBatch = 0, + Started = timeProvider.GetUtcNow().UtcDateTime, + InitiatedById = initiatedBy?.Id, + InitiatedByName = initiatedBy?.Name, + OperationId = operationId + }; + + dbContext.ArchiveOperations.Add(operationEntity); + + try + { + await dbContext.SaveChangesAsync(); + logger.LogInformation("Group {GroupId} has been split into {NumberOfBatches} batches", groupId, operationEntity.NumberOfBatches); + } + catch (DbUpdateException ex) when (dbContext.IsDuplicateKeyException(ex)) + { + //Concurrency issue and process has already started it, nothing to do until restart. + return null; + } + } + + return operationEntity; + } + + /// + /// Emits one per-message audit entry for each message in a batch, correlated to the initiating + /// operation. Skipped when no OperationId was captured (e.g. legacy in-flight operations). + /// + void AuditArchivedMessages(MessageActionKind kind, string permission, AuditUser user, string? operationId, string[] messageIds) + { + if (string.IsNullOrEmpty(operationId)) + { + return; + } + + foreach (var messageId in messageIds) + { + auditLog.MessageAction(user, kind, permission, MessageActionScope.Group, messageId, operationId); + } + } + + async Task UpdateGroupStatusAsync(ServiceControlDbContext dbContext, string groupId, FailedMessageStatus fromStatus, FailedMessageStatus toStatus, int batchSize, CancellationToken cancellationToken = default) + { + var batchIds = await GetNextBatch(dbContext, groupId, fromStatus, batchSize) + .Select(x => x.UniqueMessageId) + .ToListAsync(cancellationToken); + + if (batchIds.Count > 0) + { + var now = timeProvider.GetUtcNow().UtcDateTime; + + // Bulk status change with re-asserted status filter + await dbContext.FailedMessages + .Where(fm => batchIds.Contains(fm.UniqueMessageId) && fm.Status == fromStatus) + .ExecuteUpdateAsync(s => s + .SetProperty(fm => fm.Status, toStatus) + .SetProperty(fm => fm.StatusChangedAt, now) + .SetProperty(fm => fm.LastModified, now), cancellationToken); + } + + return batchIds.Select(id => id.ToString()).ToArray(); + } + + static async Task<(int count, string groupName)> GetGroupDetails( + ServiceControlDbContext dbContext, string groupId, FailedMessageStatus status, CancellationToken cancellationToken = default) + { + var query = + from fmg in dbContext.FailedMessageGroups + join fm in dbContext.FailedMessages + on fmg.FailedMessageUniqueId equals fm.UniqueMessageId + where fmg.GroupId == groupId && fm.Status == status + select new { fmg.Title, fm.UniqueMessageId }; + + var count = await query.CountAsync(cancellationToken); + var groupName = await dbContext.FailedMessageGroups + .Where(fmg => fmg.GroupId == groupId) + .Select(fmg => fmg.Title) + .FirstOrDefaultAsync(cancellationToken) ?? "Undefined"; + + return (count, groupName); + } + + static IQueryable GetGroup(ServiceControlDbContext dbContext, string groupId, FailedMessageStatus status) => + from fmg in dbContext.FailedMessageGroups + join fm in dbContext.FailedMessages on fmg.FailedMessageUniqueId equals fm.UniqueMessageId + where fmg.GroupId == groupId && fm.Status == status + orderby fm.UniqueMessageId + select fm; + + static IQueryable GetNextBatch(ServiceControlDbContext dbContext, string groupId, FailedMessageStatus status, int batchSize) => + GetGroup(dbContext, groupId, status).Take(batchSize); + + readonly IServiceScopeFactory scopeFactory; + readonly OperationsManager operationsManager; + readonly IDomainEvents domainEvents; + readonly IMessageActionAuditLog auditLog; + readonly TimeProvider timeProvider; + readonly EFCoreArchivingManager archivingManager; + readonly EFCoreUnarchivingManager unarchivingManager; + readonly ILogger logger; + const int batchSize = 1000; +} \ No newline at end of file diff --git a/src/ServiceControl.Persistence.Tests/EFCore/RetentionSweepTests.cs b/src/ServiceControl.Persistence.Tests/EFCore/RetentionSweepTests.cs index 5919b446b6..35aa3d930a 100644 --- a/src/ServiceControl.Persistence.Tests/EFCore/RetentionSweepTests.cs +++ b/src/ServiceControl.Persistence.Tests/EFCore/RetentionSweepTests.cs @@ -147,6 +147,29 @@ public async Task Keeps_comments_of_groups_that_still_have_messages() Assert.That(await FindGroupComment(groupId), Is.Not.Null); } + [Test] + public async Task Archived_messages_are_swept_after_the_archiver_updates_the_timestamp() + { + var groupId = Guid.NewGuid().ToString(); + var messageId = await SeedFailedMessage(FailedMessageStatus.Unresolved, Now.AddDays(-40)); + await Store(new FailedMessageGroupEntity { FailedMessageUniqueId = messageId, GroupId = groupId, Title = "t", Type = "Message Type" }); + + await ArchiveMessages.ArchiveAllInGroup(groupId); + + var archived = await FindFailedMessage(messageId); + Assert.That(archived, Is.Not.Null); + Assert.That(archived!.Status, Is.EqualTo(FailedMessageStatus.Archived)); + Assert.That(archived.StatusChangedAt, Is.EqualTo(Now), "the archiver should stamp the current fake time"); + + // The archiver reset the timestamp to Now, so the message is back inside the retention window. + // Only after the clock advances past the retention period can the sweeper remove it. + AdvanceClock(TimeSpan.FromDays(31)); + + await RunRetentionSweep(); + + Assert.That(await FindFailedMessage(messageId), Is.Null); + } + async Task SeedGroup(Guid uniqueMessageId) { var groupId = Guid.NewGuid().ToString(); diff --git a/src/ServiceControl.Persistence.Tests/Recoverability/ArchiveMessagesTests.cs b/src/ServiceControl.Persistence.Tests/Recoverability/ArchiveMessagesTests.cs new file mode 100644 index 0000000000..786de8335c --- /dev/null +++ b/src/ServiceControl.Persistence.Tests/Recoverability/ArchiveMessagesTests.cs @@ -0,0 +1,177 @@ +namespace ServiceControl.Persistence.Tests.Recoverability; + +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Extensions.DependencyInjection; +using NUnit.Framework; +using ServiceControl.Infrastructure.DomainEvents; +using ServiceControl.Persistence.Recoverability; +using ServiceControl.Recoverability; + +/// +/// Covers the in-memory state-management members of +/// (StartArchiving/StartUnarchiving, IsArchiveInProgressFor, +/// IsOperationInProgressFor, DismissArchiveOperation, GetArchivalOperations) +/// that are not exercised by the archive/unarchive loop tests. These members drive the +/// API controllers and the group fetcher and have no dedicated coverage otherwise. +/// +[TestFixture] +class ArchiveMessagesTests : PersistenceTestBase +{ + readonly CapturingDomainEvents events = new(); + + public ArchiveMessagesTests() => + RegisterServices = services => services.AddSingleton(events); + + [Test] + public async Task IsArchiveInProgressFor_is_false_when_no_operation_started() + { + Assert.That(ArchiveMessages.IsArchiveInProgressFor("group-1"), Is.False); + await Task.CompletedTask; + } + + [Test] + public async Task StartArchiving_makes_IsArchiveInProgressFor_true() + { + await ArchiveMessages.StartArchiving("group-1", ArchiveType.FailureGroup); + + Assert.That(ArchiveMessages.IsArchiveInProgressFor("group-1"), Is.True); + } + + [Test] + public async Task StartArchiving_emits_ArchiveOperationStarting() + { + await ArchiveMessages.StartArchiving("group-1", ArchiveType.FailureGroup); + + var starting = events.Raised.OfType().Single(); + using (Assert.EnterMultipleScope()) + { + Assert.That(starting.RequestId, Is.EqualTo("group-1")); + Assert.That(starting.ArchiveType, Is.EqualTo(ArchiveType.FailureGroup)); + } + } + + [Test] + public async Task StartArchiving_registers_operation_in_GetArchivalOperations() + { + await ArchiveMessages.StartArchiving("group-1", ArchiveType.FailureGroup); + + var op = ArchiveMessages.GetArchivalOperations().Single(); + using (Assert.EnterMultipleScope()) + { + Assert.That(op.RequestId, Is.EqualTo("group-1")); + Assert.That(op.ArchiveType, Is.EqualTo(ArchiveType.FailureGroup)); + Assert.That(op.GroupName, Is.EqualTo("Undefined")); + Assert.That(op.NeedsAcknowledgement(), Is.False, "an in-progress op does not need acknowledgement"); + } + } + + [Test] + public async Task StartArchiving_twice_for_same_group_keeps_a_single_operation() + { + await ArchiveMessages.StartArchiving("group-1", ArchiveType.FailureGroup); + await ArchiveMessages.StartArchiving("group-1", ArchiveType.FailureGroup); + + Assert.That(ArchiveMessages.GetArchivalOperations().Count(op => op.RequestId == "group-1"), Is.EqualTo(1)); + } + + [Test] + public async Task StartArchiving_for_different_groups_registers_each_independently() + { + await ArchiveMessages.StartArchiving("group-1", ArchiveType.FailureGroup); + await ArchiveMessages.StartArchiving("group-2", ArchiveType.FailureGroup); + + var ids = ArchiveMessages.GetArchivalOperations().Select(op => op.RequestId).ToArray(); + Assert.That(ids, Is.EquivalentTo(new[] { "group-1", "group-2" })); + using (Assert.EnterMultipleScope()) + { + Assert.That(ArchiveMessages.IsArchiveInProgressFor("group-1"), Is.True); + Assert.That(ArchiveMessages.IsArchiveInProgressFor("group-2"), Is.True); + } + } + + [Test] + public async Task DismissArchiveOperation_removes_the_operation() + { + await ArchiveMessages.StartArchiving("group-1", ArchiveType.FailureGroup); + + ArchiveMessages.DismissArchiveOperation("group-1", ArchiveType.FailureGroup); + + using (Assert.EnterMultipleScope()) + { + Assert.That(ArchiveMessages.IsArchiveInProgressFor("group-1"), Is.False); + Assert.That(ArchiveMessages.GetArchivalOperations(), Is.Empty); + } + } + + [Test] + public void DismissArchiveOperation_does_not_throw_when_no_operation_exists() + { + Assert.DoesNotThrow(() => ArchiveMessages.DismissArchiveOperation("group-unknown", ArchiveType.FailureGroup)); + } + + [Test] + public async Task StartUnarchiving_does_not_register_an_archive_operation() + { + await ArchiveMessages.StartUnarchiving("group-1", ArchiveType.FailureGroup); + + // GetArchivalOperations / IsArchiveInProgressFor track archive ops only, not unarchive ops. + using (Assert.EnterMultipleScope()) + { + Assert.That(ArchiveMessages.GetArchivalOperations(), Is.Empty); + Assert.That(ArchiveMessages.IsArchiveInProgressFor("group-1"), Is.False); + } + } + + [Test] + public async Task StartUnarchiving_emits_UnarchiveOperationStarting() + { + await ArchiveMessages.StartUnarchiving("group-1", ArchiveType.FailureGroup); + + var starting = events.Raised.OfType().Single(); + using (Assert.EnterMultipleScope()) + { + Assert.That(starting.RequestId, Is.EqualTo("group-1")); + Assert.That(starting.ArchiveType, Is.EqualTo(ArchiveType.FailureGroup)); + } + } + + [Test] + public async Task IsOperationInProgressFor_is_true_after_StartUnarchiving() + { + await ArchiveMessages.StartUnarchiving("group-1", ArchiveType.FailureGroup); + + Assert.That(ArchiveMessages.IsOperationInProgressFor("group-1", ArchiveType.FailureGroup), Is.True); + } + + [Test] + public async Task IsOperationInProgressFor_is_false_when_no_operation_started() + { + Assert.That(ArchiveMessages.IsOperationInProgressFor("group-1", ArchiveType.FailureGroup), Is.False); + await Task.CompletedTask; + } + + [Test] + public async Task DismissArchiveOperation_only_removes_the_targeted_group() + { + await ArchiveMessages.StartArchiving("group-1", ArchiveType.FailureGroup); + await ArchiveMessages.StartArchiving("group-2", ArchiveType.FailureGroup); + + ArchiveMessages.DismissArchiveOperation("group-1", ArchiveType.FailureGroup); + + var remaining = ArchiveMessages.GetArchivalOperations().Single(); + Assert.That(remaining.RequestId, Is.EqualTo("group-2")); + } + + sealed class CapturingDomainEvents : IDomainEvents + { + public System.Collections.Generic.List Raised { get; } = []; + + public Task Raise(T domainEvent, CancellationToken cancellationToken = default) where T : IDomainEvent + { + Raised.Add(domainEvent); + return Task.CompletedTask; + } + } +} \ No newline at end of file