From 0bc18fdf1b01be43161185a7a338303dfa8b3ff0 Mon Sep 17 00:00:00 2001 From: Andreas Bednarz <110360248+abparticular@users.noreply.github.com> Date: Thu, 13 Aug 2026 14:49:53 +1000 Subject: [PATCH] Initial implementation of IExternalIntegrationRequestsDataStore --- .../AuditQuery_Tests.cs | 1 - ...eControl.AcceptanceTests.PostgreSql.csproj | 3 + ...ceControl.AcceptanceTests.SqlServer.csproj | 3 + ...nalIntegrationDispatchRequests.Designer.cs | 892 ++++++++++++++++++ ..._AddExternalIntegrationDispatchRequests.cs | 36 + ...SqlServiceControlDbContextModelSnapshot.cs | 26 + ...nalIntegrationDispatchRequests.Designer.cs | 716 ++++++++++++++ ..._AddExternalIntegrationDispatchRequests.cs | 35 + ...verServiceControlDbContextModelSnapshot.cs | 22 + .../EFPersistenceConfigurationBase.cs | 2 + .../Abstractions/EFPersisterSettings.cs | 1 + .../DbContexts/ServiceControlDbContext.cs | 2 + ...xternalIntegrationDispatchRequestEntity.cs | 11 + ...IntegrationDispatchRequestConfiguration.cs | 17 + .../DispatchContextSerializer.cs | 33 + .../ExternalIntegrationRequestsDataStore.cs | 216 ++++- .../ExternalIntegrationRequestsDataStore.cs | 5 +- .../EFCore/DispatchContextSerializerTests.cs | 65 ++ ...rnalIntegrationRequestsDataStoreEFTests.cs | 115 +++ ...ternalIntegrationRequestsDataStoreTests.cs | 122 +++ .../PersistenceTestBase.cs | 1 + ...IntegrationsServiceCollectionExtensions.cs | 7 +- .../IntegrationEventWriter.cs | 19 +- 23 files changed, 2320 insertions(+), 30 deletions(-) create mode 100644 src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260811033225_AddExternalIntegrationDispatchRequests.Designer.cs create mode 100644 src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260811033225_AddExternalIntegrationDispatchRequests.cs create mode 100644 src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260811033208_AddExternalIntegrationDispatchRequests.Designer.cs create mode 100644 src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260811033208_AddExternalIntegrationDispatchRequests.cs create mode 100644 src/ServiceControl.Persistence.EFCore/Entities/ExternalIntegrationDispatchRequestEntity.cs create mode 100644 src/ServiceControl.Persistence.EFCore/EntityConfigurations/ExternalIntegrationDispatchRequestConfiguration.cs create mode 100644 src/ServiceControl.Persistence.EFCore/Implementation/DispatchContextSerializer.cs create mode 100644 src/ServiceControl.Persistence.Tests/EFCore/DispatchContextSerializerTests.cs create mode 100644 src/ServiceControl.Persistence.Tests/EFCore/ExternalIntegrationRequestsDataStoreEFTests.cs create mode 100644 src/ServiceControl.Persistence.Tests/ExternalIntegrationRequestsDataStoreTests.cs diff --git a/src/Particular.LicensingComponent.UnitTests/AuditQuery_Tests.cs b/src/Particular.LicensingComponent.UnitTests/AuditQuery_Tests.cs index b9318ecc88..19e0ccfa4c 100644 --- a/src/Particular.LicensingComponent.UnitTests/AuditQuery_Tests.cs +++ b/src/Particular.LicensingComponent.UnitTests/AuditQuery_Tests.cs @@ -6,7 +6,6 @@ using System.Text.Json.Nodes; using System.Threading; using System.Threading.Tasks; -using Contracts; using Microsoft.Extensions.Logging.Abstractions; using NUnit.Framework; using Particular.Approvals; diff --git a/src/ServiceControl.AcceptanceTests.PostgreSql/ServiceControl.AcceptanceTests.PostgreSql.csproj b/src/ServiceControl.AcceptanceTests.PostgreSql/ServiceControl.AcceptanceTests.PostgreSql.csproj index 74ed777f0b..451fa277b6 100644 --- a/src/ServiceControl.AcceptanceTests.PostgreSql/ServiceControl.AcceptanceTests.PostgreSql.csproj +++ b/src/ServiceControl.AcceptanceTests.PostgreSql/ServiceControl.AcceptanceTests.PostgreSql.csproj @@ -52,6 +52,9 @@ + + + diff --git a/src/ServiceControl.AcceptanceTests.SqlServer/ServiceControl.AcceptanceTests.SqlServer.csproj b/src/ServiceControl.AcceptanceTests.SqlServer/ServiceControl.AcceptanceTests.SqlServer.csproj index 5dda932f0e..48087c4e28 100644 --- a/src/ServiceControl.AcceptanceTests.SqlServer/ServiceControl.AcceptanceTests.SqlServer.csproj +++ b/src/ServiceControl.AcceptanceTests.SqlServer/ServiceControl.AcceptanceTests.SqlServer.csproj @@ -52,6 +52,9 @@ + + + diff --git a/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260811033225_AddExternalIntegrationDispatchRequests.Designer.cs b/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260811033225_AddExternalIntegrationDispatchRequests.Designer.cs new file mode 100644 index 0000000000..03b08fcd64 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260811033225_AddExternalIntegrationDispatchRequests.Designer.cs @@ -0,0 +1,892 @@ +// +using System; +using System.Collections.Generic; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Infrastructure; +using Microsoft.EntityFrameworkCore.Migrations; +using Microsoft.EntityFrameworkCore.Storage.ValueConversion; +using Npgsql.EntityFrameworkCore.PostgreSQL.Metadata; +using ServiceControl.Persistence.EFCore.PostgreSql; + +#nullable disable + +namespace ServiceControl.Persistence.EFCore.PostgreSql.Migrations +{ + [DbContext(typeof(PostgreSqlServiceControlDbContext))] + [Migration("20260811033225_AddExternalIntegrationDispatchRequests")] + partial class AddExternalIntegrationDispatchRequests + { + /// + protected override void BuildTargetModel(ModelBuilder modelBuilder) + { +#pragma warning disable 612, 618 + modelBuilder + .HasAnnotation("ProductVersion", "10.0.10") + .HasAnnotation("Relational:MaxIdentifierLength", 63); + + 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") + .HasColumnType("uuid") + .HasColumnName("id"); + + b.Property("Category") + .IsRequired() + .HasColumnType("text") + .HasColumnName("category"); + + b.Property("CustomCheckId") + .IsRequired() + .HasColumnType("text") + .HasColumnName("custom_check_id"); + + b.Property("FailureReason") + .HasColumnType("text") + .HasColumnName("failure_reason"); + + b.Property("OriginatingEndpointHost") + .IsRequired() + .HasColumnType("text") + .HasColumnName("originating_endpoint_host"); + + b.Property("OriginatingEndpointHostId") + .HasColumnType("uuid") + .HasColumnName("originating_endpoint_host_id"); + + b.Property("OriginatingEndpointName") + .IsRequired() + .HasColumnType("text") + .HasColumnName("originating_endpoint_name"); + + b.Property("ReportedAt") + .HasColumnType("timestamp with time zone") + .HasColumnName("reported_at"); + + b.Property("Status") + .HasColumnType("integer") + .HasColumnName("status"); + + b.HasKey("Id") + .HasName("pk_custom_checks"); + + b.HasIndex("ReportedAt") + .HasDatabaseName("ix_custom_checks_reported_at"); + + b.HasIndex("Status", "ReportedAt") + .HasDatabaseName("ix_custom_checks_status_reported_at"); + + b.ToTable("custom_checks", (string)null); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.EndpointSettingsEntity", b => + { + b.Property("Name") + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("name"); + + b.Property("TrackInstances") + .HasColumnType("boolean") + .HasColumnName("track_instances"); + + b.HasKey("Name") + .HasName("pk_endpoint_settings"); + + b.ToTable("endpoint_settings", (string)null); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.EventLogItemEntity", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("bigint") + .HasColumnName("id"); + + NpgsqlPropertyBuilderExtensions.UseIdentityByDefaultColumn(b.Property("Id")); + + b.Property("Category") + .IsRequired() + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("category"); + + b.Property("Description") + .IsRequired() + .HasColumnType("text") + .HasColumnName("description"); + + b.Property("EventType") + .IsRequired() + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("event_type"); + + b.Property("RaisedAt") + .HasColumnType("timestamp with time zone") + .HasColumnName("raised_at"); + + b.PrimitiveCollection>("RelatedTo") + .IsRequired() + .HasColumnType("text[]") + .HasColumnName("related_to"); + + b.Property("Severity") + .HasColumnType("integer") + .HasColumnName("severity"); + + b.HasKey("Id") + .HasName("pk_event_log_items"); + + b.HasIndex("RaisedAt", "Id") + .IsDescending() + .HasDatabaseName("ix_event_log_items_raised_at_id"); + + b.ToTable("EventLogItems", (string)null); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.ExternalIntegrationDispatchRequestEntity", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("bigint") + .HasColumnName("id"); + + NpgsqlPropertyBuilderExtensions.UseIdentityByDefaultColumn(b.Property("Id")); + + b.Property("DispatchContextJson") + .IsRequired() + .HasColumnType("text") + .HasColumnName("dispatch_context_json"); + + b.Property("DispatchContextTypeName") + .IsRequired() + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("dispatch_context_type_name"); + + b.HasKey("Id") + .HasName("pk_external_integration_dispatch_requests"); + + b.ToTable("ExternalIntegrationDispatchRequests", (string)null); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.FailedErrorImportEntity", b => + { + b.Property("UniqueMessageId") + .HasColumnType("uuid") + .HasColumnName("unique_message_id"); + + b.Property("Body") + .IsRequired() + .HasColumnType("bytea") + .HasColumnName("body"); + + b.Property("BodyStoredExternally") + .HasColumnType("boolean") + .HasColumnName("body_stored_externally"); + + b.Property("ExceptionInfo") + .IsRequired() + .HasColumnType("text") + .HasColumnName("exception_info"); + + b.Property("FailedAt") + .HasColumnType("timestamp with time zone") + .HasColumnName("failed_at"); + + b.Property("HeadersJson") + .IsRequired() + .HasColumnType("text") + .HasColumnName("headers_json"); + + b.Property("MessageId") + .IsRequired() + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("message_id"); + + b.HasKey("UniqueMessageId") + .HasName("pk_failed_error_imports"); + + b.HasIndex("FailedAt") + .HasDatabaseName("ix_failed_error_imports_failed_at"); + + b.ToTable("failed_error_imports", (string)null); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.FailedMessageEntity", b => + { + b.Property("UniqueMessageId") + .HasColumnType("uuid") + .HasColumnName("unique_message_id"); + + b.Property("BodyContentType") + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("body_content_type"); + + b.Property("BodySize") + .HasColumnType("integer") + .HasColumnName("body_size"); + + b.Property("BodyStoredExternally") + .HasColumnType("boolean") + .HasColumnName("body_stored_externally"); + + b.Property("BodyText") + .HasColumnType("text") + .HasColumnName("body_text"); + + b.Property("ConversationId") + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("conversation_id"); + + b.Property("ExceptionMessage") + .HasColumnType("text") + .HasColumnName("exception_message"); + + b.Property("ExceptionType") + .HasColumnType("text") + .HasColumnName("exception_type"); + + b.Property("FailingEndpointAddress") + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("failing_endpoint_address"); + + b.Property("FirstTimeOfFailure") + .HasColumnType("timestamp with time zone") + .HasColumnName("first_time_of_failure"); + + b.Property("HeadersJson") + .IsRequired() + .HasColumnType("text") + .HasColumnName("headers_json"); + + b.Property("IsSystemMessage") + .HasColumnType("boolean") + .HasColumnName("is_system_message"); + + b.Property("LastAttemptedAt") + .HasColumnType("timestamp with time zone") + .HasColumnName("last_attempted_at"); + + b.Property("LastModified") + .HasColumnType("timestamp with time zone") + .HasColumnName("last_modified"); + + b.Property("LastTimeOfFailure") + .HasColumnType("timestamp with time zone") + .HasColumnName("last_time_of_failure"); + + b.Property("MessageId") + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("message_id"); + + b.Property("MessageType") + .HasColumnType("text") + .HasColumnName("message_type"); + + b.Property("NumberOfProcessingAttempts") + .HasColumnType("integer") + .HasColumnName("number_of_processing_attempts"); + + b.Property("ReceivingEndpointHost") + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("receiving_endpoint_host"); + + b.Property("ReceivingEndpointHostId") + .HasColumnType("uuid") + .HasColumnName("receiving_endpoint_host_id"); + + b.Property("ReceivingEndpointName") + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("receiving_endpoint_name"); + + b.Property("SendingEndpointHost") + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("sending_endpoint_host"); + + b.Property("SendingEndpointHostId") + .HasColumnType("uuid") + .HasColumnName("sending_endpoint_host_id"); + + b.Property("SendingEndpointName") + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("sending_endpoint_name"); + + b.Property("Status") + .HasColumnType("integer") + .HasColumnName("status"); + + b.Property("StatusChangedAt") + .HasColumnType("timestamp with time zone") + .HasColumnName("status_changed_at"); + + b.Property("TimeSent") + .HasColumnType("timestamp with time zone") + .HasColumnName("time_sent"); + + b.HasKey("UniqueMessageId") + .HasName("pk_failed_messages"); + + b.HasIndex("ConversationId") + .HasDatabaseName("ix_failed_messages_conversation_id"); + + b.HasIndex("FailingEndpointAddress") + .HasDatabaseName("ix_failed_messages_failing_endpoint_address"); + + b.HasIndex("ReceivingEndpointName") + .HasDatabaseName("ix_failed_messages_receiving_endpoint_name"); + + b.HasIndex("StatusChangedAt") + .HasDatabaseName("ix_failed_messages_status_changed_at") + .HasFilter("status IN (2, 4)"); + + b.HasIndex("TimeSent") + .HasDatabaseName("ix_failed_messages_time_sent"); + + b.HasIndex("Status", "LastModified") + .HasDatabaseName("ix_failed_messages_status_last_modified"); + + b.ToTable("failed_messages", (string)null); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.FailedMessageGroupEntity", b => + { + b.Property("FailedMessageUniqueId") + .HasColumnType("uuid") + .HasColumnName("failed_message_unique_id"); + + b.Property("GroupId") + .HasMaxLength(64) + .HasColumnType("character varying(64)") + .HasColumnName("group_id"); + + b.Property("Title") + .IsRequired() + .HasColumnType("text") + .HasColumnName("title"); + + b.Property("Type") + .IsRequired() + .HasMaxLength(255) + .HasColumnType("character varying(255)") + .HasColumnName("type"); + + b.HasKey("FailedMessageUniqueId", "GroupId") + .HasName("pk_failed_message_groups"); + + 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); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.FailedMessageRetryEntity", b => + { + b.Property("UniqueMessageId") + .HasColumnType("uuid") + .HasColumnName("unique_message_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.HistoricRetryOperationEntity", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("bigint") + .HasColumnName("id"); + + NpgsqlPropertyBuilderExtensions.UseIdentityByDefaultColumn(b.Property("Id")); + + b.Property("CompletionTime") + .HasColumnType("timestamp with time zone") + .HasColumnName("completion_time"); + + b.Property("Failed") + .HasColumnType("boolean") + .HasColumnName("failed"); + + b.Property("NumberOfMessagesProcessed") + .HasColumnType("integer") + .HasColumnName("number_of_messages_processed"); + + b.Property("Originator") + .HasColumnType("text") + .HasColumnName("originator"); + + b.Property("RequestId") + .IsRequired() + .HasMaxLength(400) + .HasColumnType("character varying(400)") + .HasColumnName("request_id"); + + b.Property("RetryType") + .HasColumnType("integer") + .HasColumnName("retry_type"); + + b.Property("StartTime") + .HasColumnType("timestamp with time zone") + .HasColumnName("start_time"); + + b.HasKey("Id") + .HasName("pk_historic_retry_operations"); + + b.HasIndex("CompletionTime", "Id") + .IsDescending() + .HasDatabaseName("ix_historic_retry_operations_completion_time_id"); + + b.ToTable("historic_retry_operations", (string)null); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.KnownEndpointEntity", b => + { + b.Property("Id") + .HasColumnType("uuid") + .HasColumnName("id"); + + b.Property("Host") + .IsRequired() + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("host"); + + b.Property("HostId") + .HasColumnType("uuid") + .HasColumnName("host_id"); + + b.Property("Monitored") + .HasColumnType("boolean") + .HasColumnName("monitored"); + + b.Property("Name") + .IsRequired() + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("name"); + + b.HasKey("Id") + .HasName("pk_known_endpoints"); + + b.ToTable("known_endpoints", (string)null); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.LicensingEndpointEntity", b => + { + b.Property("NormalizedName") + .HasMaxLength(300) + .HasColumnType("character varying(300)") + .HasColumnName("normalized_name"); + + b.Property("ThroughputSource") + .HasColumnType("integer") + .HasColumnName("throughput_source"); + + b.PrimitiveCollection>("EndpointIndicators") + .IsRequired() + .HasColumnType("text[]") + .HasColumnName("endpoint_indicators"); + + b.Property("Name") + .IsRequired() + .HasMaxLength(300) + .HasColumnType("character varying(300)") + .HasColumnName("name"); + + b.Property("NormalizedSanitizedName") + .IsRequired() + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("normalized_sanitized_name"); + + b.Property("SanitizedName") + .IsRequired() + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("sanitized_name"); + + b.Property("Scope") + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("scope"); + + b.Property("UserIndicator") + .IsRequired() + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("user_indicator"); + + b.HasKey("NormalizedName", "ThroughputSource") + .HasName("pk_licensing_endpoints"); + + b.HasIndex("NormalizedSanitizedName") + .HasDatabaseName("ix_licensing_endpoints_normalized_sanitized_name"); + + b.ToTable("licensing_endpoints", (string)null); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.LicensingEndpointThroughputEntity", b => + { + b.Property("NormalizedName") + .HasMaxLength(300) + .HasColumnType("character varying(300)") + .HasColumnName("normalized_name"); + + b.Property("ThroughputSource") + .HasColumnType("integer") + .HasColumnName("throughput_source"); + + b.Property("DateUtc") + .HasColumnType("date") + .HasColumnName("date_utc"); + + b.Property("MessageCount") + .HasColumnType("bigint") + .HasColumnName("message_count"); + + b.HasKey("NormalizedName", "ThroughputSource", "DateUtc") + .HasName("pk_licensing_endpoint_throughput"); + + b.HasIndex("DateUtc") + .HasDatabaseName("ix_licensing_endpoint_throughput_date_utc"); + + b.ToTable("licensing_endpoint_throughput", (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.SettingEntity", b => + { + b.Property("Key") + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("key"); + + b.Property("Value") + .IsRequired() + .HasColumnType("text") + .HasColumnName("value"); + + b.HasKey("Key") + .HasName("pk_settings"); + + b.ToTable("settings", (string)null); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.SubscriptionEntity", b => + { + b.Property("MessageType") + .HasMaxLength(200) + .HasColumnType("character varying(200)") + .HasColumnName("message_type"); + + b.Property("TransportAddress") + .HasMaxLength(200) + .HasColumnType("character varying(200)") + .HasColumnName("transport_address"); + + b.Property("Endpoint") + .IsRequired() + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("endpoint"); + + b.HasKey("MessageType", "TransportAddress") + .HasName("pk_subscriptions"); + + b.ToTable("subscriptions", (string)null); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.UnacknowledgedRetryOperationEntity", b => + { + b.Property("RequestId") + .HasMaxLength(400) + .HasColumnType("character varying(400)") + .HasColumnName("request_id"); + + b.Property("RetryType") + .HasColumnType("integer") + .HasColumnName("retry_type"); + + b.Property("Classifier") + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("classifier"); + + b.Property("CompletionTime") + .HasColumnType("timestamp with time zone") + .HasColumnName("completion_time"); + + b.Property("Failed") + .HasColumnType("boolean") + .HasColumnName("failed"); + + b.Property("Last") + .HasColumnType("timestamp with time zone") + .HasColumnName("last"); + + b.Property("NumberOfMessagesProcessed") + .HasColumnType("integer") + .HasColumnName("number_of_messages_processed"); + + b.Property("Originator") + .HasColumnType("text") + .HasColumnName("originator"); + + b.Property("StartTime") + .HasColumnType("timestamp with time zone") + .HasColumnName("start_time"); + + b.HasKey("RequestId", "RetryType") + .HasName("pk_unacknowledged_retry_operations"); + + b.ToTable("unacknowledged_retry_operations", (string)null); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.FailedMessageGroupEntity", b => + { + b.HasOne("ServiceControl.Persistence.EFCore.Entities.FailedMessageEntity", null) + .WithMany() + .HasForeignKey("FailedMessageUniqueId") + .OnDelete(DeleteBehavior.Cascade) + .IsRequired() + .HasConstraintName("fk_failed_message_groups_failed_messages_failed_message_unique"); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.LicensingEndpointThroughputEntity", b => + { + b.HasOne("ServiceControl.Persistence.EFCore.Entities.LicensingEndpointEntity", null) + .WithMany() + .HasForeignKey("NormalizedName", "ThroughputSource") + .OnDelete(DeleteBehavior.Cascade) + .IsRequired() + .HasConstraintName("fk_licensing_endpoint_throughput_licensing_endpoints_normalize"); + }); +#pragma warning restore 612, 618 + } + } +} diff --git a/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260811033225_AddExternalIntegrationDispatchRequests.cs b/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260811033225_AddExternalIntegrationDispatchRequests.cs new file mode 100644 index 0000000000..21e999c399 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260811033225_AddExternalIntegrationDispatchRequests.cs @@ -0,0 +1,36 @@ +using Microsoft.EntityFrameworkCore.Migrations; +using Npgsql.EntityFrameworkCore.PostgreSQL.Metadata; + +#nullable disable + +namespace ServiceControl.Persistence.EFCore.PostgreSql.Migrations +{ + /// + public partial class AddExternalIntegrationDispatchRequests : Migration + { + /// + protected override void Up(MigrationBuilder migrationBuilder) + { + migrationBuilder.CreateTable( + name: "ExternalIntegrationDispatchRequests", + columns: table => new + { + id = table.Column(type: "bigint", nullable: false) + .Annotation("Npgsql:ValueGenerationStrategy", NpgsqlValueGenerationStrategy.IdentityByDefaultColumn), + dispatch_context_type_name = table.Column(type: "character varying(450)", maxLength: 450, nullable: false), + dispatch_context_json = table.Column(type: "text", nullable: false) + }, + constraints: table => + { + table.PrimaryKey("pk_external_integration_dispatch_requests", x => x.id); + }); + } + + /// + protected override void Down(MigrationBuilder migrationBuilder) + { + migrationBuilder.DropTable( + name: "ExternalIntegrationDispatchRequests"); + } + } +} diff --git a/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/PostgreSqlServiceControlDbContextModelSnapshot.cs b/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/PostgreSqlServiceControlDbContextModelSnapshot.cs index 2d746723fd..8b70bfa65c 100644 --- a/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/PostgreSqlServiceControlDbContextModelSnapshot.cs +++ b/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/PostgreSqlServiceControlDbContextModelSnapshot.cs @@ -207,6 +207,32 @@ protected override void BuildModel(ModelBuilder modelBuilder) b.ToTable("EventLogItems", (string)null); }); + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.ExternalIntegrationDispatchRequestEntity", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("bigint") + .HasColumnName("id"); + + NpgsqlPropertyBuilderExtensions.UseIdentityByDefaultColumn(b.Property("Id")); + + b.Property("DispatchContextJson") + .IsRequired() + .HasColumnType("text") + .HasColumnName("dispatch_context_json"); + + b.Property("DispatchContextTypeName") + .IsRequired() + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("dispatch_context_type_name"); + + b.HasKey("Id") + .HasName("pk_external_integration_dispatch_requests"); + + b.ToTable("ExternalIntegrationDispatchRequests", (string)null); + }); + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.FailedErrorImportEntity", b => { b.Property("UniqueMessageId") diff --git a/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260811033208_AddExternalIntegrationDispatchRequests.Designer.cs b/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260811033208_AddExternalIntegrationDispatchRequests.Designer.cs new file mode 100644 index 0000000000..f504489855 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260811033208_AddExternalIntegrationDispatchRequests.Designer.cs @@ -0,0 +1,716 @@ +// +using System; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Infrastructure; +using Microsoft.EntityFrameworkCore.Metadata; +using Microsoft.EntityFrameworkCore.Migrations; +using Microsoft.EntityFrameworkCore.Storage.ValueConversion; +using ServiceControl.Persistence.EFCore.SqlServer; + +#nullable disable + +namespace ServiceControl.Persistence.EFCore.SqlServer.Migrations +{ + [DbContext(typeof(SqlServerServiceControlDbContext))] + [Migration("20260811033208_AddExternalIntegrationDispatchRequests")] + partial class AddExternalIntegrationDispatchRequests + { + /// + protected override void BuildTargetModel(ModelBuilder modelBuilder) + { +#pragma warning disable 612, 618 + modelBuilder + .HasAnnotation("ProductVersion", "10.0.10") + .HasAnnotation("Relational:MaxIdentifierLength", 128); + + 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") + .HasColumnType("uniqueidentifier"); + + b.Property("Category") + .IsRequired() + .HasColumnType("nvarchar(max)"); + + b.Property("CustomCheckId") + .IsRequired() + .HasColumnType("nvarchar(max)"); + + b.Property("FailureReason") + .HasColumnType("nvarchar(max)"); + + b.Property("OriginatingEndpointHost") + .IsRequired() + .HasColumnType("nvarchar(max)"); + + b.Property("OriginatingEndpointHostId") + .HasColumnType("uniqueidentifier"); + + b.Property("OriginatingEndpointName") + .IsRequired() + .HasColumnType("nvarchar(max)"); + + b.Property("ReportedAt") + .HasColumnType("datetime2"); + + b.Property("Status") + .HasColumnType("int"); + + b.HasKey("Id"); + + b.HasIndex("ReportedAt"); + + b.HasIndex("Status", "ReportedAt"); + + b.ToTable("CustomChecks"); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.EndpointSettingsEntity", b => + { + b.Property("Name") + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("TrackInstances") + .HasColumnType("bit"); + + b.HasKey("Name"); + + b.ToTable("EndpointSettings"); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.EventLogItemEntity", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("bigint"); + + SqlServerPropertyBuilderExtensions.UseIdentityColumn(b.Property("Id")); + + b.Property("Category") + .IsRequired() + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("Description") + .IsRequired() + .HasColumnType("nvarchar(max)"); + + b.Property("EventType") + .IsRequired() + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("RaisedAt") + .HasColumnType("datetime2"); + + b.PrimitiveCollection("RelatedTo") + .IsRequired() + .HasColumnType("nvarchar(max)"); + + b.Property("Severity") + .HasColumnType("int"); + + b.HasKey("Id"); + + b.HasIndex("RaisedAt", "Id") + .IsDescending(); + + b.ToTable("EventLogItems", (string)null); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.ExternalIntegrationDispatchRequestEntity", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("bigint"); + + SqlServerPropertyBuilderExtensions.UseIdentityColumn(b.Property("Id")); + + b.Property("DispatchContextJson") + .IsRequired() + .HasColumnType("nvarchar(max)"); + + b.Property("DispatchContextTypeName") + .IsRequired() + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.HasKey("Id"); + + b.ToTable("ExternalIntegrationDispatchRequests", (string)null); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.FailedErrorImportEntity", b => + { + b.Property("UniqueMessageId") + .HasColumnType("uniqueidentifier"); + + b.Property("Body") + .IsRequired() + .HasColumnType("varbinary(max)"); + + b.Property("BodyStoredExternally") + .HasColumnType("bit"); + + b.Property("ExceptionInfo") + .IsRequired() + .HasColumnType("nvarchar(max)"); + + b.Property("FailedAt") + .HasColumnType("datetime2"); + + b.Property("HeadersJson") + .IsRequired() + .HasColumnType("nvarchar(max)"); + + b.Property("MessageId") + .IsRequired() + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.HasKey("UniqueMessageId"); + + b.HasIndex("FailedAt"); + + b.ToTable("FailedErrorImports"); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.FailedMessageEntity", b => + { + b.Property("UniqueMessageId") + .HasColumnType("uniqueidentifier"); + + b.Property("BodyContentType") + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("BodySize") + .HasColumnType("int"); + + b.Property("BodyStoredExternally") + .HasColumnType("bit"); + + b.Property("BodyText") + .HasColumnType("nvarchar(max)"); + + b.Property("ConversationId") + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("ExceptionMessage") + .HasColumnType("nvarchar(max)"); + + b.Property("ExceptionType") + .HasColumnType("nvarchar(max)"); + + b.Property("FailingEndpointAddress") + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("FirstTimeOfFailure") + .HasColumnType("datetime2"); + + b.Property("HeadersJson") + .IsRequired() + .HasColumnType("nvarchar(max)"); + + b.Property("IsSystemMessage") + .HasColumnType("bit"); + + b.Property("LastAttemptedAt") + .HasColumnType("datetime2"); + + b.Property("LastModified") + .HasColumnType("datetime2"); + + b.Property("LastTimeOfFailure") + .HasColumnType("datetime2"); + + b.Property("MessageId") + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("MessageType") + .HasColumnType("nvarchar(max)"); + + b.Property("NumberOfProcessingAttempts") + .HasColumnType("int"); + + b.Property("ReceivingEndpointHost") + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("ReceivingEndpointHostId") + .HasColumnType("uniqueidentifier"); + + b.Property("ReceivingEndpointName") + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("SendingEndpointHost") + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("SendingEndpointHostId") + .HasColumnType("uniqueidentifier"); + + b.Property("SendingEndpointName") + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("Status") + .HasColumnType("int"); + + b.Property("StatusChangedAt") + .HasColumnType("datetime2"); + + b.Property("TimeSent") + .HasColumnType("datetime2"); + + b.HasKey("UniqueMessageId"); + + b.HasIndex("ConversationId"); + + b.HasIndex("FailingEndpointAddress"); + + b.HasIndex("ReceivingEndpointName"); + + b.HasIndex("StatusChangedAt") + .HasFilter("[Status] IN (2, 4)"); + + b.HasIndex("TimeSent"); + + b.HasIndex("Status", "LastModified"); + + b.ToTable("FailedMessages"); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.FailedMessageGroupEntity", b => + { + b.Property("FailedMessageUniqueId") + .HasColumnType("uniqueidentifier"); + + b.Property("GroupId") + .HasMaxLength(64) + .HasColumnType("nvarchar(64)"); + + b.Property("Title") + .IsRequired() + .HasColumnType("nvarchar(max)"); + + b.Property("Type") + .IsRequired() + .HasMaxLength(255) + .HasColumnType("nvarchar(255)"); + + b.HasKey("FailedMessageUniqueId", "GroupId"); + + b.HasIndex("GroupId"); + + b.HasIndex("Type", "GroupId"); + + b.ToTable("FailedMessageGroups"); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.FailedMessageRetryEntity", b => + { + b.Property("UniqueMessageId") + .HasColumnType("uniqueidentifier"); + + 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.HistoricRetryOperationEntity", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("bigint"); + + SqlServerPropertyBuilderExtensions.UseIdentityColumn(b.Property("Id")); + + b.Property("CompletionTime") + .HasColumnType("datetime2"); + + b.Property("Failed") + .HasColumnType("bit"); + + b.Property("NumberOfMessagesProcessed") + .HasColumnType("int"); + + b.Property("Originator") + .HasColumnType("nvarchar(max)"); + + b.Property("RequestId") + .IsRequired() + .HasMaxLength(400) + .HasColumnType("nvarchar(400)"); + + b.Property("RetryType") + .HasColumnType("int"); + + b.Property("StartTime") + .HasColumnType("datetime2"); + + b.HasKey("Id"); + + b.HasIndex("CompletionTime", "Id") + .IsDescending(); + + b.ToTable("HistoricRetryOperations"); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.KnownEndpointEntity", b => + { + b.Property("Id") + .HasColumnType("uniqueidentifier"); + + b.Property("Host") + .IsRequired() + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("HostId") + .HasColumnType("uniqueidentifier"); + + b.Property("Monitored") + .HasColumnType("bit"); + + b.Property("Name") + .IsRequired() + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.HasKey("Id"); + + b.ToTable("KnownEndpoints"); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.LicensingEndpointEntity", b => + { + b.Property("NormalizedName") + .HasMaxLength(300) + .HasColumnType("nvarchar(300)"); + + b.Property("ThroughputSource") + .HasColumnType("int"); + + b.PrimitiveCollection("EndpointIndicators") + .IsRequired() + .HasColumnType("nvarchar(max)"); + + b.Property("Name") + .IsRequired() + .HasMaxLength(300) + .HasColumnType("nvarchar(300)"); + + b.Property("NormalizedSanitizedName") + .IsRequired() + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("SanitizedName") + .IsRequired() + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("Scope") + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("UserIndicator") + .IsRequired() + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.HasKey("NormalizedName", "ThroughputSource"); + + b.HasIndex("NormalizedSanitizedName"); + + b.ToTable("LicensingEndpoints"); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.LicensingEndpointThroughputEntity", b => + { + b.Property("NormalizedName") + .HasMaxLength(300) + .HasColumnType("nvarchar(300)"); + + b.Property("ThroughputSource") + .HasColumnType("int"); + + b.Property("DateUtc") + .HasColumnType("date"); + + b.Property("MessageCount") + .HasColumnType("bigint"); + + b.HasKey("NormalizedName", "ThroughputSource", "DateUtc"); + + b.HasIndex("DateUtc"); + + b.ToTable("LicensingEndpointThroughput"); + }); + + 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.SettingEntity", b => + { + b.Property("Key") + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("Value") + .IsRequired() + .HasColumnType("nvarchar(max)"); + + b.HasKey("Key"); + + b.ToTable("Settings"); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.SubscriptionEntity", b => + { + b.Property("MessageType") + .HasMaxLength(200) + .HasColumnType("nvarchar(200)"); + + b.Property("TransportAddress") + .HasMaxLength(200) + .HasColumnType("nvarchar(200)"); + + b.Property("Endpoint") + .IsRequired() + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.HasKey("MessageType", "TransportAddress"); + + b.ToTable("Subscriptions"); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.UnacknowledgedRetryOperationEntity", b => + { + b.Property("RequestId") + .HasMaxLength(400) + .HasColumnType("nvarchar(400)"); + + b.Property("RetryType") + .HasColumnType("int"); + + b.Property("Classifier") + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("CompletionTime") + .HasColumnType("datetime2"); + + b.Property("Failed") + .HasColumnType("bit"); + + b.Property("Last") + .HasColumnType("datetime2"); + + b.Property("NumberOfMessagesProcessed") + .HasColumnType("int"); + + b.Property("Originator") + .HasColumnType("nvarchar(max)"); + + b.Property("StartTime") + .HasColumnType("datetime2"); + + b.HasKey("RequestId", "RetryType"); + + b.ToTable("UnacknowledgedRetryOperations"); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.FailedMessageGroupEntity", b => + { + b.HasOne("ServiceControl.Persistence.EFCore.Entities.FailedMessageEntity", null) + .WithMany() + .HasForeignKey("FailedMessageUniqueId") + .OnDelete(DeleteBehavior.Cascade) + .IsRequired(); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.LicensingEndpointThroughputEntity", b => + { + b.HasOne("ServiceControl.Persistence.EFCore.Entities.LicensingEndpointEntity", null) + .WithMany() + .HasForeignKey("NormalizedName", "ThroughputSource") + .OnDelete(DeleteBehavior.Cascade) + .IsRequired(); + }); +#pragma warning restore 612, 618 + } + } +} diff --git a/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260811033208_AddExternalIntegrationDispatchRequests.cs b/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260811033208_AddExternalIntegrationDispatchRequests.cs new file mode 100644 index 0000000000..d4ab995bdd --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260811033208_AddExternalIntegrationDispatchRequests.cs @@ -0,0 +1,35 @@ +using Microsoft.EntityFrameworkCore.Migrations; + +#nullable disable + +namespace ServiceControl.Persistence.EFCore.SqlServer.Migrations +{ + /// + public partial class AddExternalIntegrationDispatchRequests : Migration + { + /// + protected override void Up(MigrationBuilder migrationBuilder) + { + migrationBuilder.CreateTable( + name: "ExternalIntegrationDispatchRequests", + columns: table => new + { + Id = table.Column(type: "bigint", nullable: false) + .Annotation("SqlServer:Identity", "1, 1"), + DispatchContextTypeName = table.Column(type: "nvarchar(450)", maxLength: 450, nullable: false), + DispatchContextJson = table.Column(type: "nvarchar(max)", nullable: false) + }, + constraints: table => + { + table.PrimaryKey("PK_ExternalIntegrationDispatchRequests", x => x.Id); + }); + } + + /// + protected override void Down(MigrationBuilder migrationBuilder) + { + migrationBuilder.DropTable( + name: "ExternalIntegrationDispatchRequests"); + } + } +} diff --git a/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/SqlServerServiceControlDbContextModelSnapshot.cs b/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/SqlServerServiceControlDbContextModelSnapshot.cs index 53ee50e29d..1d8dba4608 100644 --- a/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/SqlServerServiceControlDbContextModelSnapshot.cs +++ b/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/SqlServerServiceControlDbContextModelSnapshot.cs @@ -168,6 +168,28 @@ protected override void BuildModel(ModelBuilder modelBuilder) b.ToTable("EventLogItems", (string)null); }); + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.ExternalIntegrationDispatchRequestEntity", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("bigint"); + + SqlServerPropertyBuilderExtensions.UseIdentityColumn(b.Property("Id")); + + b.Property("DispatchContextJson") + .IsRequired() + .HasColumnType("nvarchar(max)"); + + b.Property("DispatchContextTypeName") + .IsRequired() + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.HasKey("Id"); + + b.ToTable("ExternalIntegrationDispatchRequests", (string)null); + }); + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.FailedErrorImportEntity", b => { b.Property("UniqueMessageId") diff --git a/src/ServiceControl.Persistence.EFCore/Abstractions/EFPersistenceConfigurationBase.cs b/src/ServiceControl.Persistence.EFCore/Abstractions/EFPersistenceConfigurationBase.cs index 16444defb9..7021685bfd 100644 --- a/src/ServiceControl.Persistence.EFCore/Abstractions/EFPersistenceConfigurationBase.cs +++ b/src/ServiceControl.Persistence.EFCore/Abstractions/EFPersistenceConfigurationBase.cs @@ -27,6 +27,7 @@ public abstract class EFPersistenceConfigurationBase : PersistenceConfiguration, const string ErrorRetentionPeriodKey = "ErrorRetentionPeriod"; const string EventsRetentionPeriodKey = "EventsRetentionPeriod"; const string SubscriptionCacheDurationKey = "SubscriptionCacheDuration"; + const string ExternalIntegrationsDispatchingBatchSizeKey = "ExternalIntegrationsDispatchingBatchSize"; public PersistenceSettings CreateSettings(SettingsRootNamespace settingsRootNamespace) { @@ -38,6 +39,7 @@ public PersistenceSettings CreateSettings(SettingsRootNamespace settingsRootName settings.ErrorRetentionPeriod = GetRequiredSetting(settingsRootNamespace, ErrorRetentionPeriodKey); settings.EventsRetentionPeriod = SettingsReader.Read(settingsRootNamespace, EventsRetentionPeriodKey, EFPersisterSettings.DefaultEventsRetentionPeriod); settings.SubscriptionCacheDuration = SettingsReader.Read(settingsRootNamespace, SubscriptionCacheDurationKey, EFPersisterSettings.DefaultSubscriptionCacheDuration); + settings.ExternalIntegrationsDispatchingBatchSize = SettingsReader.Read(settingsRootNamespace, ExternalIntegrationsDispatchingBatchSizeKey, 100); return settings; } diff --git a/src/ServiceControl.Persistence.EFCore/Abstractions/EFPersisterSettings.cs b/src/ServiceControl.Persistence.EFCore/Abstractions/EFPersisterSettings.cs index 7744f65dc0..9ad534b467 100644 --- a/src/ServiceControl.Persistence.EFCore/Abstractions/EFPersisterSettings.cs +++ b/src/ServiceControl.Persistence.EFCore/Abstractions/EFPersisterSettings.cs @@ -18,6 +18,7 @@ public abstract class EFPersisterSettings : PersistenceSettings public int MaxRetryDelayInSeconds { get; set; } = 30; public bool EnableSensitiveDataLogging { get; set; } public bool EnableRetryOnFailure { get; set; } = true; + public int ExternalIntegrationsDispatchingBatchSize { get; set; } = 100; /// /// How long subscriber lookups are cached for. disables caching. diff --git a/src/ServiceControl.Persistence.EFCore/DbContexts/ServiceControlDbContext.cs b/src/ServiceControl.Persistence.EFCore/DbContexts/ServiceControlDbContext.cs index a8f868db44..09fec536e4 100644 --- a/src/ServiceControl.Persistence.EFCore/DbContexts/ServiceControlDbContext.cs +++ b/src/ServiceControl.Persistence.EFCore/DbContexts/ServiceControlDbContext.cs @@ -26,6 +26,7 @@ public abstract class ServiceControlDbContext(DbContextOptions options) : DbCont public DbSet FailedMessageEdits { get; set; } public DbSet LicensingEndpoints { get; set; } public DbSet LicensingEndpointThroughput { get; set; } + public DbSet ExternalIntegrationDispatchRequests { get; set; } protected override void OnConfiguring(DbContextOptionsBuilder optionsBuilder) => optionsBuilder.EnableDetailedErrors(); @@ -54,6 +55,7 @@ protected override void OnModelCreating(ModelBuilder modelBuilder) modelBuilder.ApplyConfiguration(new ArchiveOperationConfiguration()); modelBuilder.ApplyConfiguration(new LicensingEndpointConfiguration()); modelBuilder.ApplyConfiguration(new LicensingEndpointThroughputConfiguration()); + modelBuilder.ApplyConfiguration(new ExternalIntegrationDispatchRequestConfiguration()); } public abstract bool IsDuplicateKeyException(DbUpdateException exception); diff --git a/src/ServiceControl.Persistence.EFCore/Entities/ExternalIntegrationDispatchRequestEntity.cs b/src/ServiceControl.Persistence.EFCore/Entities/ExternalIntegrationDispatchRequestEntity.cs new file mode 100644 index 0000000000..fb08e246e7 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Entities/ExternalIntegrationDispatchRequestEntity.cs @@ -0,0 +1,11 @@ +namespace ServiceControl.Persistence.EFCore.Entities; + +public class ExternalIntegrationDispatchRequestEntity +{ + // Auto-increment, so ordering by Id gives natural FIFO dispatch order. + public long Id { get; set; } + + public required string DispatchContextTypeName { get; set; } + + public required string DispatchContextJson { get; set; } +} diff --git a/src/ServiceControl.Persistence.EFCore/EntityConfigurations/ExternalIntegrationDispatchRequestConfiguration.cs b/src/ServiceControl.Persistence.EFCore/EntityConfigurations/ExternalIntegrationDispatchRequestConfiguration.cs new file mode 100644 index 0000000000..acce35b9bf --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/EntityConfigurations/ExternalIntegrationDispatchRequestConfiguration.cs @@ -0,0 +1,17 @@ +namespace ServiceControl.Persistence.EFCore.EntityConfigurations; + +using Entities; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Metadata.Builders; + +class ExternalIntegrationDispatchRequestConfiguration : IEntityTypeConfiguration +{ + public void Configure(EntityTypeBuilder builder) + { + builder.ToTable("ExternalIntegrationDispatchRequests"); + builder.HasKey(e => e.Id); + builder.Property(e => e.Id).ValueGeneratedOnAdd(); + builder.Property(e => e.DispatchContextTypeName).IsRequired().HasMaxLength(ColumnLengths.ShortTextLength); + builder.Property(e => e.DispatchContextJson).IsRequired(); + } +} diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/DispatchContextSerializer.cs b/src/ServiceControl.Persistence.EFCore/Implementation/DispatchContextSerializer.cs new file mode 100644 index 0000000000..7d77e1a6bc --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Implementation/DispatchContextSerializer.cs @@ -0,0 +1,33 @@ +namespace ServiceControl.Persistence.EFCore.Implementation; + +using System.Text.Json; + +// DispatchContext's concrete type (defined by whichever IEventPublisher created it, living in +// ServiceControl.csproj, which this project does not reference) is only known at runtime, so +// round-tripping it through a single JSON column needs the type name stored alongside the +// payload. This has to stay reflection-based (JsonSerializer.Serialize(obj, obj.GetType())) - +// unlike HeadersJson's always-Dictionary shape, a source-generated +// JsonSerializerContext isn't possible here since the type isn't known at this project's +// compile time. +static class DispatchContextSerializer +{ + public static (string TypeName, string Json) Serialize(object dispatchContext) + { + var type = dispatchContext.GetType(); + + // Versionless assembly-qualified name: writer and reader are always the same + // ServiceControl.dll, so version/culture/token would only make the stored value + // needlessly brittle across builds. + var typeName = $"{type.FullName}, {type.Assembly.GetName().Name}"; + var json = JsonSerializer.Serialize(dispatchContext, type); + + return (typeName, json); + } + + public static object Deserialize(string typeName, string json) + { + var type = Type.GetType(typeName, throwOnError: true)!; + + return JsonSerializer.Deserialize(json, type)!; + } +} diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/ExternalIntegrationRequestsDataStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/ExternalIntegrationRequestsDataStore.cs index 2387c15fbb..dac5c240e7 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/ExternalIntegrationRequestsDataStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/ExternalIntegrationRequestsDataStore.cs @@ -1,15 +1,221 @@ namespace ServiceControl.Persistence.EFCore.Implementation; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using NServiceBus; using ServiceControl.ExternalIntegrations; +using ServiceControl.Persistence.EFCore.Abstractions; +using ServiceControl.Persistence.EFCore.Entities; -public class ExternalIntegrationRequestsDataStore : IExternalIntegrationRequestsDataStore, IHostedService +// SQL has no equivalent of RavenDB's server-push Changes API, so new work is signalled +// in-process instead (there is no evidence of multiple concurrent processes of the same +// ServiceControl role sharing one database) and a PeriodicTimer is a safety-net fallback in +// case a signal is ever missed. +public class ExternalIntegrationRequestsDataStore( + IServiceScopeFactory scopeFactory, + EFPersisterSettings settings, + TimeProvider timeProvider, + CriticalError criticalError, + ILogger logger) + : DataStoreBase(scopeFactory), IExternalIntegrationRequestsDataStore, IHostedService, IAsyncDisposable { - public void Subscribe(Func callback) { } + static readonly TimeSpan PollingInterval = TimeSpan.FromSeconds(30); - public Task StoreDispatchRequest(IEnumerable dispatchRequests, CancellationToken cancellationToken = default) => Task.CompletedTask; + Func? callback; + Task? task; + bool isDisposed; - public Task StartAsync(CancellationToken cancellationToken = default) => Task.CompletedTask; + public async Task StoreDispatchRequest(IEnumerable dispatchRequests, CancellationToken cancellationToken = default) + { + await ExecuteWithDbContext(async (context, cancellationToken) => + { + foreach (var dispatchRequest in dispatchRequests) + { + if (dispatchRequest.Id is not null) + { + throw new ArgumentException("Items cannot have their Id property set", nameof(dispatchRequests)); + } - public Task StopAsync(CancellationToken cancellationToken = default) => Task.CompletedTask; + var (typeName, json) = DispatchContextSerializer.Serialize(dispatchRequest.DispatchContext); + + context.ExternalIntegrationDispatchRequests.Add(new ExternalIntegrationDispatchRequestEntity + { + DispatchContextTypeName = typeName, + DispatchContextJson = json + }); + } + + await context.SaveChangesAsync(cancellationToken); + }, cancellationToken); + + // Wakes the dispatch loop immediately instead of waiting for the next PeriodicTimer tick. + signal.Release(); + } + + public void Subscribe(Func callback) + { + if (this.callback is not null) + { + throw new InvalidOperationException("Subscription already exists."); + } + + this.callback = callback ?? throw new ArgumentNullException(nameof(callback)); + } + + public Task StartAsync(CancellationToken cancellationToken = default) + { + task = RunLoop(tokenSource.Token); + + return Task.CompletedTask; + } + + public async Task StopAsync(CancellationToken cancellationToken = default) => await DisposeAsync(); + + // Test hook: runs one full drain pass immediately, bypassing the timer/signal wait. Shares + // drainLock with the background loop so it cannot race a concurrently-woken pass into + // double-dispatching the same batch. + public Task DispatchNow(CancellationToken cancellationToken = default) => DrainAvailableBatches(cancellationToken); + + async Task RunLoop(CancellationToken cancellationToken) + { + using var timer = new PeriodicTimer(PollingInterval, timeProvider); + + while (!cancellationToken.IsCancellationRequested) + { + try + { + await DrainAvailableBatches(cancellationToken); + circuitBreaker.Success(); + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + break; + } + catch (Exception ex) + { + logger.LogError(ex, "An exception occurred when dispatching external integration events"); + + // Failure() itself is the retry backoff (it awaits a delay before returning), so + // looping straight back to DrainAvailableBatches retries the still-undispatched + // batch promptly instead of also waiting for the next signal/timer tick on top. + await circuitBreaker.Failure(ex, cancellationToken); + continue; + } + + try + { + await Task.WhenAny(signal.WaitAsync(cancellationToken), timer.WaitForNextTickAsync(cancellationToken).AsTask()); + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + break; + } + } + } + + async Task DrainAvailableBatches(CancellationToken cancellationToken) + { + // No subscriber yet (e.g. loop started before EventDispatcherHostedService.StartAsync + // ran Subscribe): nothing to hand batches to, so there is nothing to do this pass. + // Captured into a local so TryDispatchBatch below doesn't need to deal with it as nullable. + var subscribedCallback = callback; + + if (subscribedCallback is null) + { + return; + } + + await drainLock.WaitAsync(cancellationToken); + + try + { + while (await TryDispatchBatch(subscribedCallback, cancellationToken) && !cancellationToken.IsCancellationRequested) + { + } + } + finally + { + drainLock.Release(); + } + } + + // Select -> publish -> delete, each its own scope, so a DB transaction is never held open + // across the network call to publish on the bus (unlike Raven, which spans query, callback + // and delete in a single session). + async Task TryDispatchBatch(Func callback, CancellationToken cancellationToken) + { + var rows = await ExecuteWithDbContext((context, cancellationToken) => + context.ExternalIntegrationDispatchRequests + .AsNoTracking() + .OrderBy(row => row.Id) + .Take(settings.ExternalIntegrationsDispatchingBatchSize) + .ToListAsync(cancellationToken), cancellationToken); + + if (rows.Count == 0) + { + return false; + } + + var contexts = rows + .Select(row => DispatchContextSerializer.Deserialize(row.DispatchContextTypeName, row.DispatchContextJson)) + .ToArray(); + + logger.LogDebug("Dispatching {EventCount} events", contexts.Length); + + // Rows are only deleted once the callback has successfully consumed the batch: if it + // throws (e.g. a transient failure resolving one of the outgoing contract events), the + // rows are left in place so the same batch is retried on the next pass. This matches + // RavenDB, which never reaches its delete either when its callback throws. + await callback(contexts, cancellationToken); + + var ids = rows.Select(row => row.Id).ToArray(); + + await ExecuteWithDbContext((context, cancellationToken) => + context.ExternalIntegrationDispatchRequests + .Where(row => ids.Contains(row.Id)) + .ExecuteDeleteAsync(cancellationToken), cancellationToken); + + return rows.Count == settings.ExternalIntegrationsDispatchingBatchSize; + } + + public async ValueTask DisposeAsync() + { + if (isDisposed) + { + return; + } + + isDisposed = true; + + await tokenSource.CancelAsync(); + + if (task is not null) + { + try + { + await task; + } + catch (OperationCanceledException) + { + // expected during shutdown + } + } + + tokenSource.Dispose(); + signal.Dispose(); + drainLock.Dispose(); + } + + readonly SemaphoreSlim signal = new(0, int.MaxValue); + readonly SemaphoreSlim drainLock = new(1, 1); + readonly CancellationTokenSource tokenSource = new(); + + readonly RepeatedFailuresOverTimeCircuitBreaker circuitBreaker = new( + "ExternalIntegrationEventDispatcher", + TimeSpan.FromMinutes(5), + ex => criticalError.Raise("Repeated failures when dispatching external integration events.", ex), + logger, + timeToWaitWhenArmed: TimeSpan.FromSeconds(20)); } diff --git a/src/ServiceControl.Persistence.RavenDB/ExternalIntegrationRequestsDataStore.cs b/src/ServiceControl.Persistence.RavenDB/ExternalIntegrationRequestsDataStore.cs index 9b928db0ee..6478da32f1 100644 --- a/src/ServiceControl.Persistence.RavenDB/ExternalIntegrationRequestsDataStore.cs +++ b/src/ServiceControl.Persistence.RavenDB/ExternalIntegrationRequestsDataStore.cs @@ -170,10 +170,7 @@ public async Task StartAsync(CancellationToken cancellationToken = default) .Changes() .ForDocumentsStartingWith(KeyPrefix) .Where(c => c.Type == DocumentChangeTypes.Put) - .Subscribe(d => - { - signal.Set(); - }); + .Subscribe(_ => signal.Set()); } public async Task StopAsync(CancellationToken cancellationToken = default) => await DisposeAsync(); diff --git a/src/ServiceControl.Persistence.Tests/EFCore/DispatchContextSerializerTests.cs b/src/ServiceControl.Persistence.Tests/EFCore/DispatchContextSerializerTests.cs new file mode 100644 index 0000000000..c2c1a56314 --- /dev/null +++ b/src/ServiceControl.Persistence.Tests/EFCore/DispatchContextSerializerTests.cs @@ -0,0 +1,65 @@ +namespace ServiceControl.Persistence.Tests; + +using System; +using NUnit.Framework; +using ServiceControl.Persistence.EFCore.Implementation; + +class DispatchContextSerializerTests +{ + [Test] + public void Round_trips_a_simple_dispatch_context() + { + var original = new SimpleContext { Name = "endpoint-a", Count = 3 }; + + var (typeName, json) = DispatchContextSerializer.Serialize(original); + var result = (SimpleContext)DispatchContextSerializer.Deserialize(typeName, json); + + using (Assert.EnterMultipleScope()) + { + Assert.That(result.Name, Is.EqualTo(original.Name)); + Assert.That(result.Count, Is.EqualTo(original.Count)); + } + } + + // Two distinct shapes stored side by side (as happens once several IEventPublishers have + // written requests) must each resolve back to their own type, not get confused with each other. + [Test] + public void Round_trips_a_different_dispatch_context_shape_using_its_own_type() + { + var original = new OtherContext { FailedMessageId = Guid.NewGuid(), Tags = ["a", "b"] }; + + var (typeName, json) = DispatchContextSerializer.Serialize(original); + var result = (OtherContext)DispatchContextSerializer.Deserialize(typeName, json); + + using (Assert.EnterMultipleScope()) + { + Assert.That(result.FailedMessageId, Is.EqualTo(original.FailedMessageId)); + Assert.That(result.Tags, Is.EqualTo(original.Tags)); + } + } + + [Test] + public void Type_name_does_not_include_assembly_version_culture_or_public_key_token() + { + var (typeName, _) = DispatchContextSerializer.Serialize(new SimpleContext { Name = "x", Count = 1 }); + + using (Assert.EnterMultipleScope()) + { + Assert.That(typeName, Does.Not.Contain("Version=")); + Assert.That(typeName, Does.Not.Contain("Culture=")); + Assert.That(typeName, Does.Not.Contain("PublicKeyToken=")); + } + } + + public class SimpleContext + { + public string Name { get; set; } + public int Count { get; set; } + } + + public class OtherContext + { + public Guid FailedMessageId { get; set; } + public string[] Tags { get; set; } + } +} diff --git a/src/ServiceControl.Persistence.Tests/EFCore/ExternalIntegrationRequestsDataStoreEFTests.cs b/src/ServiceControl.Persistence.Tests/EFCore/ExternalIntegrationRequestsDataStoreEFTests.cs new file mode 100644 index 0000000000..61cdc98175 --- /dev/null +++ b/src/ServiceControl.Persistence.Tests/EFCore/ExternalIntegrationRequestsDataStoreEFTests.cs @@ -0,0 +1,115 @@ +namespace ServiceControl.Persistence.Tests; + +using System; +using System.Linq; +using System.Threading.Tasks; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using NUnit.Framework; +using ServiceControl.ExternalIntegrations; +using ServiceControl.Persistence.EFCore.Abstractions; +using ServiceControl.Persistence.EFCore.DbContexts; +using ServiceControl.Persistence.EFCore.Implementation; + +// DispatchNow bypasses the timer/signal wait, so these tests are deterministic instead of relying +// on WaitUntil(...) polling - the same test-determinism convention RetentionSweeper.SweepNow uses +// (see ErrorIngestionTestBase.RunRetentionSweep). +class ExternalIntegrationRequestsDataStoreEFTests : PersistenceTestBase +{ + EFPersisterSettings EFSettings => (EFPersisterSettings)PersistenceSettings; + + ExternalIntegrationRequestsDataStore Store => + ServiceProvider.GetServices().OfType().Single(); + + [Test] + public async Task DispatchNow_dispatches_available_rows_and_removes_them() + { + var dispatched = Array.Empty(); + ExternalIntegrationStore.Subscribe((contexts, cancellationToken) => + { + dispatched = contexts; + return Task.CompletedTask; + }); + + await ExternalIntegrationStore.StoreDispatchRequest([Request("one")]); + + await Store.DispatchNow(TestContext.CurrentContext.CancellationToken); + + Assert.That(((TestDispatchContext)dispatched.Single()).Value, Is.EqualTo("one")); + Assert.That(await CountRows(), Is.Zero); + } + + [Test] + public async Task DispatchNow_leaves_a_failed_batch_in_place_and_removes_it_once_a_retry_succeeds() + { + var attempts = 0; + ExternalIntegrationStore.Subscribe((_, cancellationToken) => + { + attempts++; + + if (attempts == 1) + { + throw new InvalidOperationException("Simulated callback failure"); + } + + return Task.CompletedTask; + }); + + await ExternalIntegrationStore.StoreDispatchRequest([Request("retry-me")]); + + Assert.ThrowsAsync(() => Store.DispatchNow(TestContext.CurrentContext.CancellationToken)); + Assert.That(await CountRows(), Is.EqualTo(1), "the row should survive a failed dispatch attempt"); + + await Store.DispatchNow(TestContext.CurrentContext.CancellationToken); + + Assert.That(attempts, Is.EqualTo(2)); + Assert.That(await CountRows(), Is.Zero, "the row should be removed once the retry succeeds"); + } + + [Test] + public async Task DispatchNow_drains_more_than_one_batch_when_the_backlog_exceeds_the_configured_batch_size() + { + EFSettings.ExternalIntegrationsDispatchingBatchSize = 2; + + var callbackInvocations = 0; + var dispatchedCount = 0; + ExternalIntegrationStore.Subscribe((contexts, cancellationToken) => + { + callbackInvocations++; + dispatchedCount += contexts.Length; + return Task.CompletedTask; + }); + + await ExternalIntegrationStore.StoreDispatchRequest(Enumerable.Range(0, 5).Select(i => Request($"item-{i}")).ToList()); + + await Store.DispatchNow(TestContext.CurrentContext.CancellationToken); + + Assert.That(dispatchedCount, Is.EqualTo(5)); + Assert.That(callbackInvocations, Is.EqualTo(3), "5 rows at a batch size of 2 should dispatch as 2 + 2 + 1"); + Assert.That(await CountRows(), Is.Zero); + } + + [Test] + public async Task DispatchNow_is_a_noop_when_nothing_is_stored() + { + ExternalIntegrationStore.Subscribe((_, cancellationToken) => throw new InvalidOperationException("Should not be called")); + + Assert.DoesNotThrowAsync(() => Store.DispatchNow(TestContext.CurrentContext.CancellationToken)); + } + + static ExternalIntegrationDispatchRequest Request(string value) => new() { DispatchContext = new TestDispatchContext { Value = value } }; + + async Task CountRows() + { + using var scope = ServiceProvider.CreateScope(); + var dbContext = scope.ServiceProvider.GetRequiredService(); + + return await dbContext.ExternalIntegrationDispatchRequests.CountAsync(TestContext.CurrentContext.CancellationToken); + } + + public class TestDispatchContext + { + public string Value { get; set; } + } +} diff --git a/src/ServiceControl.Persistence.Tests/ExternalIntegrationRequestsDataStoreTests.cs b/src/ServiceControl.Persistence.Tests/ExternalIntegrationRequestsDataStoreTests.cs new file mode 100644 index 0000000000..696cc86089 --- /dev/null +++ b/src/ServiceControl.Persistence.Tests/ExternalIntegrationRequestsDataStoreTests.cs @@ -0,0 +1,122 @@ +namespace ServiceControl.Persistence.Tests; + +using System; +using System.Collections.Concurrent; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using NUnit.Framework; +using ServiceControl.ExternalIntegrations; + +class ExternalIntegrationRequestsDataStoreTests : PersistenceTestBase +{ + // Matches the default ExternalIntegrationsDispatchingBatchSize on both backends' settings + // (EFPersisterSettings/RavenPersisterSettings) - neither test context overrides it. + const int DefaultBatchSize = 100; + + [Test] + public async Task Stored_request_is_dispatched_to_the_subscriber_and_removed() + { + var dispatched = new ConcurrentBag(); + ExternalIntegrationStore.Subscribe((contexts, _) => + { + foreach (var context in contexts) + { + dispatched.Add(context); + } + + return Task.CompletedTask; + }); + + await ExternalIntegrationStore.StoreDispatchRequest([Request("one")]); + + await WaitUntil(() => Task.FromResult(dispatched.Count == 1), "the stored request to be dispatched"); + + Assert.That(((TestDispatchContext)dispatched.Single()).Value, Is.EqualTo("one")); + } + + [Test] + public async Task More_requests_than_the_batch_size_are_dispatched_across_multiple_callback_invocations() + { + var callbackInvocations = 0; + var dispatched = new ConcurrentBag(); + ExternalIntegrationStore.Subscribe((contexts, _) => + { + Interlocked.Increment(ref callbackInvocations); + + foreach (var context in contexts) + { + dispatched.Add(context); + } + + return Task.CompletedTask; + }); + + var requests = Enumerable.Range(0, DefaultBatchSize + 1).Select(i => Request($"item-{i}")).ToList(); + await ExternalIntegrationStore.StoreDispatchRequest(requests); + + await WaitUntil(() => Task.FromResult(dispatched.Count == requests.Count), "every stored request to be dispatched"); + + Assert.That(callbackInvocations, Is.GreaterThanOrEqualTo(2), + "a backlog larger than the configured batch size should be dispatched across more than one callback invocation"); + } + + [Test] + public async Task Concurrent_stores_are_all_eventually_dispatched() + { + var dispatched = new ConcurrentBag(); + ExternalIntegrationStore.Subscribe((contexts, _) => + { + foreach (var context in contexts) + { + dispatched.Add(context); + } + + return Task.CompletedTask; + }); + + const int concurrentStores = 20; + await Task.WhenAll(Enumerable.Range(0, concurrentStores) + .Select(i => ExternalIntegrationStore.StoreDispatchRequest([Request($"concurrent-{i}")]))); + + await WaitUntil(() => Task.FromResult(dispatched.Count == concurrentStores), "every concurrently stored request to be dispatched"); + } + + [Test] + public async Task Callback_failure_does_not_stop_future_dispatch() + { + var attempts = 0; + var dispatched = new ConcurrentBag(); + ExternalIntegrationStore.Subscribe((contexts, _) => + { + if (Interlocked.Increment(ref attempts) == 1) + { + throw new InvalidOperationException("Simulated callback failure"); + } + + foreach (var context in contexts) + { + dispatched.Add(context); + } + + return Task.CompletedTask; + }); + + await ExternalIntegrationStore.StoreDispatchRequest([Request("retry-me")]); + + // The failed first attempt is not removed, so the same request is redelivered once the + // dispatcher recovers. Recovery is paced by RepeatedFailuresOverTimeCircuitBreaker's own + // ~20s "armed" backoff on both backends, hence the longer-than-usual timeout. + await WaitUntil(() => Task.FromResult(dispatched.Count == 1), + "the request to be redelivered and dispatched after the failed attempt", TimeSpan.FromSeconds(40)); + + Assert.That(attempts, Is.GreaterThanOrEqualTo(2)); + } + + static ExternalIntegrationDispatchRequest Request(string value) => new() { DispatchContext = new TestDispatchContext { Value = value } }; + + public class TestDispatchContext + { + public string Value { get; set; } + } +} diff --git a/src/ServiceControl.Persistence.Tests/PersistenceTestBase.cs b/src/ServiceControl.Persistence.Tests/PersistenceTestBase.cs index bee2026528..ae1d4068cc 100644 --- a/src/ServiceControl.Persistence.Tests/PersistenceTestBase.cs +++ b/src/ServiceControl.Persistence.Tests/PersistenceTestBase.cs @@ -115,4 +115,5 @@ protected static async Task WaitUntil(Func> conditionChecker, string protected ILicensingDataStore LicensingDataStore => ServiceProvider.GetRequiredService(); protected IQueueAddressStore QueueAddressStore => ServiceProvider.GetRequiredService(); protected IEndpointSettingsStore EndpointSettingsStore => ServiceProvider.GetRequiredService(); + protected IExternalIntegrationRequestsDataStore ExternalIntegrationStore => ServiceProvider.GetRequiredService(); } \ No newline at end of file diff --git a/src/ServiceControl/ExternalIntegrations/ExternalIntegrationsServiceCollectionExtensions.cs b/src/ServiceControl/ExternalIntegrations/ExternalIntegrationsServiceCollectionExtensions.cs index dad21a4e8b..4c5cf1cae8 100644 --- a/src/ServiceControl/ExternalIntegrations/ExternalIntegrationsServiceCollectionExtensions.cs +++ b/src/ServiceControl/ExternalIntegrations/ExternalIntegrationsServiceCollectionExtensions.cs @@ -4,10 +4,7 @@ namespace ServiceControl.ExternalIntegrations static class ExternalIntegrationsServiceCollectionExtensions { - public static void AddIntegrationEventPublisher(this IServiceCollection serviceCollection) - where T : class, IEventPublisher - { - serviceCollection.AddSingleton(); - } + public static void AddIntegrationEventPublisher(this IServiceCollection serviceCollection) where T : class, IEventPublisher + => serviceCollection.AddSingleton(); } } \ No newline at end of file diff --git a/src/ServiceControl/ExternalIntegrations/IntegrationEventWriter.cs b/src/ServiceControl/ExternalIntegrations/IntegrationEventWriter.cs index d09eb5af2b..a1c6eb7b84 100644 --- a/src/ServiceControl/ExternalIntegrations/IntegrationEventWriter.cs +++ b/src/ServiceControl/ExternalIntegrations/IntegrationEventWriter.cs @@ -8,18 +8,11 @@ using Microsoft.Extensions.Logging; using ServiceControl.Persistence; - class IntegrationEventWriter : IDomainHandler + class IntegrationEventWriter( + IExternalIntegrationRequestsDataStore store, + IEnumerable eventPublishers, + ILogger logger) : IDomainHandler { - public IntegrationEventWriter( - IExternalIntegrationRequestsDataStore store, - IEnumerable eventPublishers, - ILogger logger) - { - this.store = store; - this.eventPublishers = eventPublishers; - this.logger = logger; - } - public async Task Handle(IDomainEvent message, CancellationToken cancellationToken = default) { var dispatchContexts = eventPublishers @@ -41,9 +34,5 @@ public async Task Handle(IDomainEvent message, CancellationToken cancellationTok await store.StoreDispatchRequest(dispatchRequests, cancellationToken); } - - readonly IExternalIntegrationRequestsDataStore store; - readonly IEnumerable eventPublishers; - readonly ILogger logger; } } \ No newline at end of file