diff --git a/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260729053645_AddSubscriptions.Designer.cs b/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260729053645_AddSubscriptions.Designer.cs new file mode 100644 index 0000000000..4feb704b4c --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260729053645_AddSubscriptions.Designer.cs @@ -0,0 +1,383 @@ +// +using System; +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("20260729053645_AddSubscriptions")] + partial class AddSubscriptions + { + /// + 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.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.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") + .IsRequired() + .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("QueueAddress") + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("queue_address"); + + 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("QueueAddress") + .HasDatabaseName("ix_failed_messages_queue_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.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("RetryId") + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("retry_id"); + + b.HasKey("UniqueMessageId") + .HasName("pk_failed_message_retries"); + + b.ToTable("failed_message_retries", (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.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.TrialMetadataEntity", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("integer") + .HasColumnName("id"); + + NpgsqlPropertyBuilderExtensions.UseIdentityByDefaultColumn(b.Property("Id")); + + b.Property("TrialEndDate") + .HasColumnType("date") + .HasColumnName("trial_end_date"); + + b.HasKey("Id") + .HasName("pk_trial_metadata"); + + b.ToTable("trial_metadata", (string)null); + + b.HasData( + new + { + Id = 1 + }); + }); + + 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"); + }); +#pragma warning restore 612, 618 + } + } +} diff --git a/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260729053645_AddSubscriptions.cs b/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260729053645_AddSubscriptions.cs new file mode 100644 index 0000000000..18de802cd8 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260729053645_AddSubscriptions.cs @@ -0,0 +1,34 @@ +using Microsoft.EntityFrameworkCore.Migrations; + +#nullable disable + +namespace ServiceControl.Persistence.EFCore.PostgreSql.Migrations +{ + /// + public partial class AddSubscriptions : Migration + { + /// + protected override void Up(MigrationBuilder migrationBuilder) + { + migrationBuilder.CreateTable( + name: "subscriptions", + columns: table => new + { + message_type = table.Column(type: "character varying(200)", maxLength: 200, nullable: false), + transport_address = table.Column(type: "character varying(200)", maxLength: 200, nullable: false), + endpoint = table.Column(type: "character varying(450)", maxLength: 450, nullable: false) + }, + constraints: table => + { + table.PrimaryKey("pk_subscriptions", x => new { x.message_type, x.transport_address }); + }); + } + + /// + protected override void Down(MigrationBuilder migrationBuilder) + { + migrationBuilder.DropTable( + name: "subscriptions"); + } + } +} diff --git a/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/PostgreSqlServiceControlDbContextModelSnapshot.cs b/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/PostgreSqlServiceControlDbContextModelSnapshot.cs index 5887803530..770e5515d5 100644 --- a/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/PostgreSqlServiceControlDbContextModelSnapshot.cs +++ b/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/PostgreSqlServiceControlDbContextModelSnapshot.cs @@ -316,6 +316,30 @@ protected override void BuildModel(ModelBuilder modelBuilder) b.ToTable("known_endpoints", (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.TrialMetadataEntity", b => { b.Property("Id") diff --git a/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260729053622_AddSubscriptions.Designer.cs b/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260729053622_AddSubscriptions.Designer.cs new file mode 100644 index 0000000000..2f268a7915 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260729053622_AddSubscriptions.Designer.cs @@ -0,0 +1,312 @@ +// +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("20260729053622_AddSubscriptions")] + partial class AddSubscriptions + { + /// + 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.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.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") + .IsRequired() + .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("QueueAddress") + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + 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("QueueAddress"); + + 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.ToTable("FailedMessageGroups"); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.FailedMessageRetryEntity", b => + { + b.Property("UniqueMessageId") + .HasColumnType("uniqueidentifier"); + + b.Property("RetryId") + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.HasKey("UniqueMessageId"); + + b.ToTable("FailedMessageRetries"); + }); + + 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.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.TrialMetadataEntity", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("int"); + + SqlServerPropertyBuilderExtensions.UseIdentityColumn(b.Property("Id")); + + b.Property("TrialEndDate") + .HasColumnType("date"); + + b.HasKey("Id"); + + b.ToTable("TrialMetadata"); + + b.HasData( + new + { + Id = 1 + }); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.FailedMessageGroupEntity", b => + { + b.HasOne("ServiceControl.Persistence.EFCore.Entities.FailedMessageEntity", null) + .WithMany() + .HasForeignKey("FailedMessageUniqueId") + .OnDelete(DeleteBehavior.Cascade) + .IsRequired(); + }); +#pragma warning restore 612, 618 + } + } +} diff --git a/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260729053622_AddSubscriptions.cs b/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260729053622_AddSubscriptions.cs new file mode 100644 index 0000000000..9cb0ecdda4 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260729053622_AddSubscriptions.cs @@ -0,0 +1,34 @@ +using Microsoft.EntityFrameworkCore.Migrations; + +#nullable disable + +namespace ServiceControl.Persistence.EFCore.SqlServer.Migrations +{ + /// + public partial class AddSubscriptions : Migration + { + /// + protected override void Up(MigrationBuilder migrationBuilder) + { + migrationBuilder.CreateTable( + name: "Subscriptions", + columns: table => new + { + MessageType = table.Column(type: "nvarchar(200)", maxLength: 200, nullable: false), + TransportAddress = table.Column(type: "nvarchar(200)", maxLength: 200, nullable: false), + Endpoint = table.Column(type: "nvarchar(450)", maxLength: 450, nullable: false) + }, + constraints: table => + { + table.PrimaryKey("PK_Subscriptions", x => new { x.MessageType, x.TransportAddress }); + }); + } + + /// + protected override void Down(MigrationBuilder migrationBuilder) + { + migrationBuilder.DropTable( + name: "Subscriptions"); + } + } +} diff --git a/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/SqlServerServiceControlDbContextModelSnapshot.cs b/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/SqlServerServiceControlDbContextModelSnapshot.cs index e7082075b4..2539cbd49d 100644 --- a/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/SqlServerServiceControlDbContextModelSnapshot.cs +++ b/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/SqlServerServiceControlDbContextModelSnapshot.cs @@ -253,6 +253,26 @@ protected override void BuildModel(ModelBuilder modelBuilder) b.ToTable("KnownEndpoints"); }); + 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.TrialMetadataEntity", b => { b.Property("Id") diff --git a/src/ServiceControl.Persistence.EFCore/Abstractions/EFPersistenceConfigurationBase.cs b/src/ServiceControl.Persistence.EFCore/Abstractions/EFPersistenceConfigurationBase.cs index b14cf2b24d..4d752ad296 100644 --- a/src/ServiceControl.Persistence.EFCore/Abstractions/EFPersistenceConfigurationBase.cs +++ b/src/ServiceControl.Persistence.EFCore/Abstractions/EFPersistenceConfigurationBase.cs @@ -26,6 +26,7 @@ public abstract class EFPersistenceConfigurationBase : IPersistenceConfiguration const string MaxBodySizeToStoreKey = "MaxBodySizeToStore"; const string ErrorRetentionPeriodKey = "ErrorRetentionPeriod"; const string EnableFullTextSearchOnBodiesKey = "EnableFullTextSearchOnBodies"; + const string SubscriptionCacheDurationKey = "SubscriptionCacheDuration"; public PersistenceSettings CreateSettings(SettingsRootNamespace settingsRootNamespace) { @@ -36,6 +37,7 @@ public PersistenceSettings CreateSettings(SettingsRootNamespace settingsRootName settings.CommandTimeout = SettingsReader.Read(settingsRootNamespace, CommandTimeoutKey, EFPersisterSettings.DefaultCommandTimeout); settings.ErrorRetentionPeriod = GetRequiredSetting(settingsRootNamespace, ErrorRetentionPeriodKey); settings.EnableFullTextSearchOnBodies = SettingsReader.Read(settingsRootNamespace, EnableFullTextSearchOnBodiesKey, true); + settings.SubscriptionCacheDuration = SettingsReader.Read(settingsRootNamespace, SubscriptionCacheDurationKey, EFPersisterSettings.DefaultSubscriptionCacheDuration); return settings; } diff --git a/src/ServiceControl.Persistence.EFCore/Abstractions/EFPersisterSettings.cs b/src/ServiceControl.Persistence.EFCore/Abstractions/EFPersisterSettings.cs index 1311e327e2..5eed5da073 100644 --- a/src/ServiceControl.Persistence.EFCore/Abstractions/EFPersisterSettings.cs +++ b/src/ServiceControl.Persistence.EFCore/Abstractions/EFPersisterSettings.cs @@ -6,6 +6,8 @@ public abstract class EFPersisterSettings : PersistenceSettings public const int DefaultCommandTimeout = 30; + public static readonly TimeSpan DefaultSubscriptionCacheDuration = TimeSpan.FromSeconds(60); + public required string ConnectionString { get; set; } public int CommandTimeout { get; set; } = DefaultCommandTimeout; public TimeSpan ErrorRetentionPeriod { get; set; } @@ -14,4 +16,9 @@ public abstract class EFPersisterSettings : PersistenceSettings public int MaxRetryDelayInSeconds { get; set; } = 30; public bool EnableSensitiveDataLogging { get; set; } public bool EnableRetryOnFailure { get; set; } = true; + + /// + /// How long subscriber lookups are cached for. disables caching. + /// + public TimeSpan SubscriptionCacheDuration { get; set; } = DefaultSubscriptionCacheDuration; } diff --git a/src/ServiceControl.Persistence.EFCore/DbContexts/ServiceControlDbContext.cs b/src/ServiceControl.Persistence.EFCore/DbContexts/ServiceControlDbContext.cs index 045b0f1414..fa10d2e455 100644 --- a/src/ServiceControl.Persistence.EFCore/DbContexts/ServiceControlDbContext.cs +++ b/src/ServiceControl.Persistence.EFCore/DbContexts/ServiceControlDbContext.cs @@ -13,6 +13,7 @@ public abstract class ServiceControlDbContext(DbContextOptions options) : DbCont public DbSet FailedMessageRetries { get; set; } public DbSet FailedErrorImports { get; set; } public DbSet TrialMetadata { get; set; } + public DbSet Subscriptions { get; set; } protected override void OnConfiguring(DbContextOptionsBuilder optionsBuilder) => optionsBuilder.EnableDetailedErrors(); @@ -27,6 +28,7 @@ protected override void OnModelCreating(ModelBuilder modelBuilder) modelBuilder.ApplyConfiguration(new FailedMessageGroupConfiguration()); modelBuilder.ApplyConfiguration(new FailedMessageRetryConfiguration()); modelBuilder.ApplyConfiguration(new KnownEndpointConfiguration()); + modelBuilder.ApplyConfiguration(new SubscriptionConfiguration()); modelBuilder.ApplyConfiguration(new TrialMetadataConfiguration()); } diff --git a/src/ServiceControl.Persistence.EFCore/Entities/SubscriptionEntity.cs b/src/ServiceControl.Persistence.EFCore/Entities/SubscriptionEntity.cs new file mode 100644 index 0000000000..ac9db9c398 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Entities/SubscriptionEntity.cs @@ -0,0 +1,8 @@ +namespace ServiceControl.Persistence.EFCore.Entities; + +public class SubscriptionEntity +{ + public required string MessageType { get; set; } + public required string TransportAddress { get; set; } + public required string Endpoint { get; set; } +} diff --git a/src/ServiceControl.Persistence.EFCore/EntityConfigurations/ColumnLengths.cs b/src/ServiceControl.Persistence.EFCore/EntityConfigurations/ColumnLengths.cs index 140d945469..69c8244b27 100644 --- a/src/ServiceControl.Persistence.EFCore/EntityConfigurations/ColumnLengths.cs +++ b/src/ServiceControl.Persistence.EFCore/EntityConfigurations/ColumnLengths.cs @@ -5,4 +5,8 @@ static class ColumnLengths // Indexed and short-by-nature values get a length so that SQL Server can index them, // nvarchar(max) columns cannot be index key columns. public const int ShortTextLength = 450; + + // The subscriptions key spans two columns and SQL Server caps a clustered index key at 900 bytes, + // so both have to stay well under ShortTextLength. Matches NServiceBus.Persistence.Sql. + public const int SubscriptionKeyLength = 200; } diff --git a/src/ServiceControl.Persistence.EFCore/EntityConfigurations/SubscriptionConfiguration.cs b/src/ServiceControl.Persistence.EFCore/EntityConfigurations/SubscriptionConfiguration.cs new file mode 100644 index 0000000000..01a58a3c5f --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/EntityConfigurations/SubscriptionConfiguration.cs @@ -0,0 +1,16 @@ +namespace ServiceControl.Persistence.EFCore.EntityConfigurations; + +using Entities; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Metadata.Builders; + +class SubscriptionConfiguration : IEntityTypeConfiguration +{ + public void Configure(EntityTypeBuilder builder) + { + builder.HasKey(e => new { e.MessageType, e.TransportAddress }); + builder.Property(e => e.MessageType).IsRequired().HasMaxLength(ColumnLengths.SubscriptionKeyLength); + builder.Property(e => e.TransportAddress).IsRequired().HasMaxLength(ColumnLengths.SubscriptionKeyLength); + builder.Property(e => e.Endpoint).IsRequired().HasMaxLength(ColumnLengths.ShortTextLength); + } +} diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/SubscriptionStorage.cs b/src/ServiceControl.Persistence.EFCore/Implementation/SubscriptionStorage.cs index 6b4d8c6e93..e910c4d8c0 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/SubscriptionStorage.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/SubscriptionStorage.cs @@ -1,20 +1,122 @@ namespace ServiceControl.Persistence.EFCore.Implementation; +using System.Collections.Concurrent; +using Abstractions; +using Entities; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.DependencyInjection; +using NServiceBus; using NServiceBus.Extensibility; +using NServiceBus.Settings; using NServiceBus.Unicast.Subscriptions; using NServiceBus.Unicast.Subscriptions.MessageDrivenSubscriptions; +using ServiceControl.Infrastructure; -public class SubscriptionStorage : IServiceControlSubscriptionStorage +public class SubscriptionStorage : DataStoreBase, IServiceControlSubscriptionStorage { - public Task Initialize() => - throw new NotImplementedException(); + public SubscriptionStorage(IServiceScopeFactory scopeFactory, IReadOnlySettings settings, ReceiveAddresses receiveAddresses, EFPersisterSettings persisterSettings, TimeProvider timeProvider) + : this(scopeFactory, settings.EndpointName(), receiveAddresses.MainReceiveAddress, settings.GetAvailableTypes().Implementing().Select(e => new MessageType(e)).ToArray(), persisterSettings.SubscriptionCacheDuration, timeProvider) + { + } - public Task Subscribe(Subscriber subscriber, MessageType messageType, ContextBag context, CancellationToken cancellationToken = default) => - throw new NotImplementedException(); + public SubscriptionStorage(IServiceScopeFactory scopeFactory, string endpointName, string localAddress, MessageType[] locallyHandledEventTypes, TimeSpan cacheDuration, TimeProvider timeProvider) + : base(scopeFactory) + { + localEndpointName = endpointName; + localSubscriber = new Subscriber(localAddress, endpointName); + locallyHandledTypeNames = locallyHandledEventTypes.Select(eventType => eventType.TypeName).ToHashSet(StringComparer.Ordinal); + this.cacheDuration = cacheDuration; + this.timeProvider = timeProvider; + } - public Task Unsubscribe(Subscriber subscriber, MessageType messageType, ContextBag context, CancellationToken cancellationToken = default) => - throw new NotImplementedException(); + // Subscriptions are read on demand, so there is nothing to prime at startup. + public Task Initialize() => Task.CompletedTask; - public Task> GetSubscriberAddressesForMessage(IEnumerable messageTypes, ContextBag context, CancellationToken cancellationToken = default) => - throw new NotImplementedException(); + public async Task Subscribe(Subscriber subscriber, MessageType messageType, ContextBag context, CancellationToken cancellationToken = default) + { + if (subscriber.Endpoint == localEndpointName) + { + return; + } + + var typeName = messageType.TypeName; + var transportAddress = subscriber.TransportAddress; + //When the subscriber is running V6 and UseLegacyMessageDrivenSubscriptionMode is enabled at the subscriber the 'subscriber.Endpoint' value is null + var endpoint = subscriber.Endpoint ?? transportAddress.Split('@').First(); + + await ExecuteWithDbContext(dbContext => dbContext.UpsertAsync( + [typeName, transportAddress], + () => new SubscriptionEntity { MessageType = typeName, TransportAddress = transportAddress, Endpoint = endpoint }, + entity => entity.Endpoint = endpoint, + cancellationToken)); + + InvalidateCache(typeName); + } + + public async Task Unsubscribe(Subscriber subscriber, MessageType messageType, ContextBag context, CancellationToken cancellationToken = default) + { + var typeName = messageType.TypeName; + var transportAddress = subscriber.TransportAddress; + + await ExecuteWithDbContext(dbContext => dbContext.Subscriptions + .Where(subscription => subscription.MessageType == typeName && subscription.TransportAddress == transportAddress) + .ExecuteDeleteAsync(cancellationToken)); + + InvalidateCache(typeName); + } + + public async Task> GetSubscriberAddressesForMessage(IEnumerable messageTypes, ContextBag context, CancellationToken cancellationToken = default) + { + var typeNames = messageTypes.Select(messageType => messageType.TypeName).ToArray(); + var key = string.Join(",", typeNames); + + if (cache.TryGetValue(key, out var cached) && timeProvider.GetUtcNow() - cached.Stored < cacheDuration) + { + return cached.Subscribers; + } + + var subscribers = await LoadSubscribers(typeNames, cancellationToken); + cache[key] = new CacheItem(timeProvider.GetUtcNow(), typeNames, subscribers); + + return subscribers; + } + + async Task LoadSubscribers(string[] typeNames, CancellationToken cancellationToken) + { + var stored = await ExecuteWithDbContext(dbContext => dbContext.Subscriptions + .AsNoTracking() + .Where(subscription => typeNames.Contains(subscription.MessageType)) + .Select(subscription => new { subscription.TransportAddress, subscription.Endpoint }) + .Distinct() + .ToArrayAsync(cancellationToken)); + + var subscribers = stored.Select(subscription => new Subscriber(subscription.TransportAddress, subscription.Endpoint)); + + if (locallyHandledTypeNames.Overlaps(typeNames)) + { + subscribers = subscribers.Append(localSubscriber); + } + + return subscribers.ToArray(); + } + + void InvalidateCache(string typeName) + { + foreach (var entry in cache) + { + if (entry.Value.TypeNames.Contains(typeName, StringComparer.Ordinal)) + { + cache.TryRemove(entry.Key, out _); + } + } + } + + readonly string localEndpointName; + readonly Subscriber localSubscriber; + readonly HashSet locallyHandledTypeNames; + readonly TimeSpan cacheDuration; + readonly TimeProvider timeProvider; + readonly ConcurrentDictionary cache = new(); + + record CacheItem(DateTimeOffset Stored, string[] TypeNames, Subscriber[] Subscribers); } diff --git a/src/ServiceControl.Persistence.Tests/EFCore/SubscriptionStorageTests.cs b/src/ServiceControl.Persistence.Tests/EFCore/SubscriptionStorageTests.cs new file mode 100644 index 0000000000..b94715c9b1 --- /dev/null +++ b/src/ServiceControl.Persistence.Tests/EFCore/SubscriptionStorageTests.cs @@ -0,0 +1,258 @@ +namespace ServiceControl.Persistence.Tests; + +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.DependencyInjection; +using NServiceBus.Extensibility; +using NServiceBus.Unicast.Subscriptions; +using NServiceBus.Unicast.Subscriptions.MessageDrivenSubscriptions; +using NUnit.Framework; +using ServiceControl.Persistence.EFCore.DbContexts; +using ServiceControl.Persistence.EFCore.Entities; +using ServiceControl.Persistence.EFCore.Implementation; + +class SubscriptionStorageTests : PersistenceTestBase +{ + const string LocalEndpointName = "Particular.ServiceControl"; + const string LocalAddress = "Particular.ServiceControl@local"; + + static readonly MessageType SampleEventV1 = new(typeof(SampleEvent).FullName, new Version(1, 0, 0)); + static readonly MessageType SampleEventV2 = new(typeof(SampleEvent).FullName, new Version(2, 0, 0)); + static readonly MessageType OtherEvent = new(typeof(AnotherSampleEvent).FullName, new Version(1, 0, 0)); + + [Test] + public async Task Subscribe_stores_the_subscriber() + { + var storage = CreateStorage(); + + await Subscribe(storage, new Subscriber("SalesAddress", "Sales"), SampleEventV1); + + var subscribers = await GetSubscribers(storage, SampleEventV1); + + Assert.That(subscribers, Has.Count.EqualTo(1)); + using (Assert.EnterMultipleScope()) + { + Assert.That(subscribers[0].TransportAddress, Is.EqualTo("SalesAddress")); + Assert.That(subscribers[0].Endpoint, Is.EqualTo("Sales")); + } + } + + [Test] + public async Task Subscribing_twice_stores_a_single_row() + { + var storage = CreateStorage(); + var subscriber = new Subscriber("SalesAddress", "Sales"); + + await Subscribe(storage, subscriber, SampleEventV1); + await Subscribe(storage, subscriber, SampleEventV1); + + Assert.That(await GetStoredSubscriptions(), Has.Count.EqualTo(1)); + } + + [Test] + public async Task Resubscribing_updates_the_endpoint_name() + { + var storage = CreateStorage(); + + await Subscribe(storage, new Subscriber("SalesAddress", "Sales"), SampleEventV1); + await Subscribe(storage, new Subscriber("SalesAddress", "Sales.Renamed"), SampleEventV1); + + var stored = await GetStoredSubscriptions(); + + Assert.That(stored, Has.Count.EqualTo(1)); + Assert.That(stored[0].Endpoint, Is.EqualTo("Sales.Renamed")); + } + + [Test] + public async Task Unsubscribe_removes_only_the_matching_subscription() + { + var storage = CreateStorage(); + var sales = new Subscriber("SalesAddress", "Sales"); + var shipping = new Subscriber("ShippingAddress", "Shipping"); + + await Subscribe(storage, sales, SampleEventV1); + await Subscribe(storage, shipping, SampleEventV1); + await Subscribe(storage, sales, OtherEvent); + + await storage.Unsubscribe(sales, SampleEventV1, new ContextBag(), CancellationToken.None); + + using (Assert.EnterMultipleScope()) + { + Assert.That((await GetSubscribers(storage, SampleEventV1)).Select(s => s.TransportAddress), Is.EqualTo(new[] { "ShippingAddress" })); + Assert.That((await GetSubscribers(storage, OtherEvent)).Select(s => s.TransportAddress), Is.EqualTo(new[] { "SalesAddress" })); + } + } + + [Test] + public async Task Unsubscribing_something_that_was_never_subscribed_is_a_no_op() + { + var storage = CreateStorage(); + + await storage.Unsubscribe(new Subscriber("SalesAddress", "Sales"), SampleEventV1, new ContextBag(), CancellationToken.None); + + Assert.That(await GetStoredSubscriptions(), Is.Empty); + } + + [Test] + public async Task Should_return_subscriptions_for_other_versions_of_the_same_message_type() + { + var storage = CreateStorage(); + + await Subscribe(storage, new Subscriber("V1SubscriberAddress", "V1Subscriber"), SampleEventV1); + + var subscribers = await GetSubscribers(storage, SampleEventV2); + + Assert.That(subscribers, Has.Count.EqualTo(1)); + Assert.That(subscribers[0].TransportAddress, Is.EqualTo("V1SubscriberAddress")); + } + + [Test] + public async Task Subscriber_is_returned_once_when_subscribed_to_several_types_in_the_hierarchy() + { + var storage = CreateStorage(); + var sales = new Subscriber("SalesAddress", "Sales"); + + await Subscribe(storage, sales, SampleEventV1); + await Subscribe(storage, sales, OtherEvent); + + var subscribers = await GetSubscribers(storage, SampleEventV1, OtherEvent); + + Assert.That(subscribers, Has.Count.EqualTo(1)); + } + + [Test] + public async Task Endpoint_is_derived_from_the_transport_address_when_the_subscriber_does_not_supply_one() + { + var storage = CreateStorage(); + + await Subscribe(storage, new Subscriber("Sales@MachineName", null), SampleEventV1); + + var subscribers = await GetSubscribers(storage, SampleEventV1); + + Assert.That(subscribers[0].Endpoint, Is.EqualTo("Sales")); + } + + [Test] + public async Task Subscriptions_from_the_local_endpoint_are_ignored() + { + var storage = CreateStorage(); + + await Subscribe(storage, new Subscriber(LocalAddress, LocalEndpointName), SampleEventV1); + + Assert.That(await GetStoredSubscriptions(), Is.Empty); + } + + [Test] + public async Task Locally_handled_event_types_resolve_to_the_local_address() + { + var storage = CreateStorage(locallyHandledEventTypes: [SampleEventV1]); + + var subscribers = await GetSubscribers(storage, SampleEventV2); + + Assert.That(subscribers, Has.Count.EqualTo(1)); + using (Assert.EnterMultipleScope()) + { + Assert.That(subscribers[0].TransportAddress, Is.EqualTo(LocalAddress)); + Assert.That(subscribers[0].Endpoint, Is.EqualTo(LocalEndpointName)); + } + } + + [Test] + public async Task Locally_handled_event_types_do_not_leak_into_other_message_types() + { + var storage = CreateStorage(locallyHandledEventTypes: [SampleEventV1]); + + Assert.That(await GetSubscribers(storage, OtherEvent), Is.Empty); + } + + [Test] + public async Task Subscriber_lookups_are_served_from_the_cache_until_it_expires() + { + var storage = CreateStorage(cacheDuration: TimeSpan.FromSeconds(60)); + + Assert.That(await GetSubscribers(storage, SampleEventV1), Is.Empty); + + await StoreSubscriptionDirectly(SampleEventV1.TypeName, "SalesAddress", "Sales"); + + Assert.That(await GetSubscribers(storage, SampleEventV1), Is.Empty, "the cached, empty result should still be served"); + + PersistenceTestsContext.FakeTime.Advance(TimeSpan.FromSeconds(60)); + + Assert.That(await GetSubscribers(storage, SampleEventV1), Has.Count.EqualTo(1)); + } + + [Test] + public async Task Subscribing_evicts_the_cached_lookup_for_that_message_type() + { + var storage = CreateStorage(cacheDuration: TimeSpan.FromSeconds(60)); + + Assert.That(await GetSubscribers(storage, SampleEventV1), Is.Empty); + Assert.That(await GetSubscribers(storage, OtherEvent), Is.Empty); + + await Subscribe(storage, new Subscriber("SalesAddress", "Sales"), SampleEventV1); + + using (Assert.EnterMultipleScope()) + { + Assert.That(await GetSubscribers(storage, SampleEventV1), Has.Count.EqualTo(1)); + Assert.That(await GetSubscribers(storage, OtherEvent), Is.Empty); + } + } + + [Test] + public async Task Unsubscribing_evicts_the_cached_lookup_for_that_message_type() + { + var storage = CreateStorage(cacheDuration: TimeSpan.FromSeconds(60)); + var sales = new Subscriber("SalesAddress", "Sales"); + + await Subscribe(storage, sales, SampleEventV1); + Assert.That(await GetSubscribers(storage, SampleEventV1), Has.Count.EqualTo(1)); + + await storage.Unsubscribe(sales, SampleEventV1, new ContextBag(), CancellationToken.None); + + Assert.That(await GetSubscribers(storage, SampleEventV1), Is.Empty); + } + + SubscriptionStorage CreateStorage(TimeSpan cacheDuration = default, MessageType[] locallyHandledEventTypes = null) => + new(ServiceProvider.GetRequiredService(), + LocalEndpointName, + LocalAddress, + locallyHandledEventTypes ?? [], + cacheDuration, + PersistenceTestsContext.FakeTime); + + static Task Subscribe(SubscriptionStorage storage, Subscriber subscriber, MessageType messageType) => + storage.Subscribe(subscriber, messageType, new ContextBag(), CancellationToken.None); + + static async Task> GetSubscribers(SubscriptionStorage storage, params MessageType[] messageTypes) + { + var subscribers = await storage.GetSubscriberAddressesForMessage(messageTypes, new ContextBag(), CancellationToken.None); + + return subscribers.ToList(); + } + + async Task> GetStoredSubscriptions() + { + await using var scope = ServiceProvider.CreateAsyncScope(); + var dbContext = scope.ServiceProvider.GetRequiredService(); + + return await dbContext.Subscriptions.AsNoTracking().ToListAsync(); + } + + async Task StoreSubscriptionDirectly(string messageType, string transportAddress, string endpoint) + { + await using var scope = ServiceProvider.CreateAsyncScope(); + var dbContext = scope.ServiceProvider.GetRequiredService(); + + dbContext.Subscriptions.Add(new SubscriptionEntity { MessageType = messageType, TransportAddress = transportAddress, Endpoint = endpoint }); + + await dbContext.SaveChangesAsync(); + } + + class SampleEvent; + + class AnotherSampleEvent; +}