diff --git a/src/Mocha/src/Mocha.Transport.Postgres/Configurations/PostgresTransportConfiguration.cs b/src/Mocha/src/Mocha.Transport.Postgres/Configurations/PostgresTransportConfiguration.cs index 3436df9b042..87e1bddab3a 100644 --- a/src/Mocha/src/Mocha.Transport.Postgres/Configurations/PostgresTransportConfiguration.cs +++ b/src/Mocha/src/Mocha.Transport.Postgres/Configurations/PostgresTransportConfiguration.cs @@ -59,6 +59,12 @@ public PostgresTransportConfiguration() /// public bool? AutoProvision { get; set; } + /// + /// Gets or sets whether transport schema migrations run at startup. Defaults to true. + /// When false, startup skips migration without checking the schema. + /// + public bool AutoMigrate { get; set; } = true; + /// /// Gets or sets the bus-level defaults applied to all auto-provisioned queues and topics. /// diff --git a/src/Mocha/src/Mocha.Transport.Postgres/Connection/PostgresConnectionManager.cs b/src/Mocha/src/Mocha.Transport.Postgres/Connection/PostgresConnectionManager.cs index 8a3cfda41c6..f37afa904c3 100644 --- a/src/Mocha/src/Mocha.Transport.Postgres/Connection/PostgresConnectionManager.cs +++ b/src/Mocha/src/Mocha.Transport.Postgres/Connection/PostgresConnectionManager.cs @@ -50,8 +50,7 @@ public async Task EnsureMigratedAsync(CancellationToken cancellationToken) } await using var connection = await OpenConnectionAsync(cancellationToken); - var migrator = new PostgresSchemaMigrator(_schemaOptions); - await migrator.MigrateAsync(connection); + await PostgresTransportSchema.MigrateAsync(connection, _schemaOptions, cancellationToken); _isMigrated = true; _logger.SchemaMigrated(); diff --git a/src/Mocha/src/Mocha.Transport.Postgres/Connection/PostgresSchemaMigrator.cs b/src/Mocha/src/Mocha.Transport.Postgres/Connection/PostgresSchemaMigrator.cs deleted file mode 100644 index b745fd25a65..00000000000 --- a/src/Mocha/src/Mocha.Transport.Postgres/Connection/PostgresSchemaMigrator.cs +++ /dev/null @@ -1,115 +0,0 @@ -using Npgsql; - -namespace Mocha.Transport.Postgres; - -/// -/// Manages database schema migrations for the PostgreSQL messaging transport. -/// Uses PostgreSQL advisory locks to prevent concurrent migration attempts. -/// -internal sealed class PostgresSchemaMigrator -{ - private const int LockId = 958913715; - - private readonly IReadOnlyPostgresSchemaOptions _settings; - - public PostgresSchemaMigrator(IReadOnlyPostgresSchemaOptions settings) - { - _settings = settings; - } - - public async Task MigrateAsync(NpgsqlConnection connection) - { - await using var transaction = await connection.BeginTransactionAsync(CancellationToken.None); - - await AcquireLockAsync(connection); - await EnsureMigrationsTableAsync(connection, _settings); - - var migrations = new List - { - new("2026-03-06_InitialSchema", PostgresSchemaSql.InitialSchema(_settings)), - new("2026-03-06_AddTransportIndex", PostgresSchemaSql.AddTransportIndex(_settings)), - new("2026-03-06_AddConsumerManagement", PostgresSchemaSql.AddConsumerManagement(_settings)) - }; - - var applied = await GetAppliedMigrationsAsync(connection, _settings); - - foreach (var migration in migrations) - { - if (!applied.Contains(migration.Id)) - { - await ApplyMigrationAsync(connection, _settings, migration); - } - } - - await transaction.CommitAsync(); - } - - private static async Task AcquireLockAsync(NpgsqlConnection connection) - { - await using var cmd = connection.CreateCommand(); - cmd.CommandText = "SELECT pg_advisory_xact_lock(@lockId);"; - cmd.Parameters.AddWithValue("lockId", LockId); - - await cmd.ExecuteNonQueryAsync(); - } - - private static async Task EnsureMigrationsTableAsync( - NpgsqlConnection connection, - IReadOnlyPostgresSchemaOptions settings) - { - await using var cmd = connection.CreateCommand(); - - cmd.CommandText = $""" - CREATE SCHEMA IF NOT EXISTS {settings.Schema}; - - CREATE TABLE IF NOT EXISTS {settings.MigrationsTable} - ( - migration_id text NOT NULL PRIMARY KEY, - applied_on timestamptz NOT NULL DEFAULT (now() at time zone 'utc') - ); - """; - - await cmd.ExecuteNonQueryAsync(); - } - - private static async Task> GetAppliedMigrationsAsync( - NpgsqlConnection connection, - IReadOnlyPostgresSchemaOptions settings) - { - await using var cmd = connection.CreateCommand(); - cmd.CommandText = $"SELECT migration_id FROM {settings.MigrationsTable};"; - - await using var reader = await cmd.ExecuteReaderAsync(); - var applied = new HashSet(); - while (await reader.ReadAsync()) - { - applied.Add(reader.GetString(0)); - } - - return applied; - } - - private static async Task ApplyMigrationAsync( - NpgsqlConnection connection, - IReadOnlyPostgresSchemaOptions settings, - MigrationInfo migration) - { - await using (var cmd = connection.CreateCommand()) - { - cmd.CommandText = migration.Sql; - await cmd.ExecuteNonQueryAsync(); - } - - await using (var insertCmd = connection.CreateCommand()) - { - insertCmd.CommandText = $""" - INSERT INTO {settings.MigrationsTable} (migration_id) - VALUES (@id); - """; - insertCmd.Parameters.AddWithValue("id", migration.Id); - await insertCmd.ExecuteNonQueryAsync(); - } - } - - private record MigrationInfo(string Id, string Sql); -} diff --git a/src/Mocha/src/Mocha.Transport.Postgres/Connection/PostgresSchemaSql.Migrations.cs b/src/Mocha/src/Mocha.Transport.Postgres/Connection/PostgresSchemaSql.Migrations.cs new file mode 100644 index 00000000000..d36356f0653 --- /dev/null +++ b/src/Mocha/src/Mocha.Transport.Postgres/Connection/PostgresSchemaSql.Migrations.cs @@ -0,0 +1,154 @@ +namespace Mocha.Transport.Postgres; + +internal static partial class PostgresSchemaSql +{ + private static string Bootstrap(IReadOnlyPostgresSchemaOptions s) + => $""" + SELECT pg_advisory_xact_lock({LockId}); + CREATE SCHEMA IF NOT EXISTS {s.Schema}; + + CREATE TABLE IF NOT EXISTS {s.MigrationsTable} + ( + migration_id text NOT NULL PRIMARY KEY, + applied_on timestamptz NOT NULL DEFAULT (now() at time zone 'utc') + ); + """; + + public static string InitialSchema(IReadOnlyPostgresSchemaOptions s) => + $""" + CREATE SCHEMA IF NOT EXISTS {s.Schema}; + + CREATE SEQUENCE IF NOT EXISTS {s.TopologySequence} AS bigint; + + CREATE TABLE IF NOT EXISTS {s.TopicTable} + ( + id bigint NOT NULL PRIMARY KEY DEFAULT nextval({Literal(s.TopologySequence)}), + updated timestamptz NOT NULL DEFAULT (now() at time zone 'utc'), + name text NOT NULL + ); + + CREATE UNIQUE INDEX IF NOT EXISTS {TopicUniqueIndex(s)} ON {s.TopicTable} (name) INCLUDE (id); + ALTER TABLE {s.TopicTable} ADD CONSTRAINT {TopicUniqueConstraint(s)} UNIQUE USING INDEX {TopicUniqueIndex(s)}; + + CREATE TABLE IF NOT EXISTS {s.QueueTable} + ( + id bigint NOT NULL PRIMARY KEY DEFAULT nextval({Literal(s.TopologySequence)}), + updated timestamptz NOT NULL DEFAULT (now() at time zone 'utc'), + name text NOT NULL + ); + + CREATE UNIQUE INDEX IF NOT EXISTS {QueueUniqueIndex(s)} ON {s.QueueTable} (name) INCLUDE (id); + ALTER TABLE {s.QueueTable} ADD CONSTRAINT {QueueUniqueConstraint(s)} UNIQUE USING INDEX {QueueUniqueIndex(s)}; + + CREATE TABLE IF NOT EXISTS {s.QueueSubscriptionTable} + ( + id bigint NOT NULL PRIMARY KEY DEFAULT nextval({Literal(s.TopologySequence)}), + updated timestamptz NOT NULL DEFAULT (now() at time zone 'utc'), + source_id bigint NOT NULL REFERENCES {s.TopicTable} (id) ON DELETE CASCADE, + destination_id bigint NOT NULL REFERENCES {s.QueueTable} (id) ON DELETE CASCADE + ); + + CREATE UNIQUE INDEX IF NOT EXISTS {QueueSubscriptionUniqueIndex(s)} + ON {s.QueueSubscriptionTable} (source_id, destination_id); + ALTER TABLE {s.QueueSubscriptionTable} + ADD CONSTRAINT {QueueSubscriptionUniqueConstraint(s)} UNIQUE USING INDEX {QueueSubscriptionUniqueIndex(s)}; + + CREATE INDEX IF NOT EXISTS {QueueSubscriptionSourceIndex(s)} + ON {s.QueueSubscriptionTable} (source_id) INCLUDE (id, destination_id); + CREATE INDEX IF NOT EXISTS {QueueSubscriptionDestinationIndex(s)} + ON {s.QueueSubscriptionTable} (destination_id) INCLUDE (id, source_id); + + CREATE TABLE IF NOT EXISTS {s.MessageTable} + ( + transport_message_id uuid NOT NULL PRIMARY KEY DEFAULT gen_random_uuid(), + body bytea NOT NULL, + headers jsonb, + queue_id bigint NOT NULL REFERENCES {s.QueueTable} (id) ON DELETE CASCADE, + sent_time timestamptz NOT NULL DEFAULT (now() at time zone 'utc'), + scheduled_time timestamptz, + expiration_time timestamptz, + delivery_count int NOT NULL DEFAULT 0, + max_delivery_count int NOT NULL DEFAULT 10, + last_delivered timestamptz, + consumer_id uuid, + error_reason jsonb + ); + + CREATE INDEX IF NOT EXISTS {MessageQueueIndex(s)} + ON {s.MessageTable} (queue_id) INCLUDE (transport_message_id); + """; + + private static string AddTransportIndex(IReadOnlyPostgresSchemaOptions s) => + $""" + CREATE INDEX IF NOT EXISTS {MessageTransportQueueIndex(s)} + ON {s.MessageTable} (transport_message_id, queue_id); + + CREATE INDEX IF NOT EXISTS {MessageExpirationScheduledIndex(s)} + ON {s.MessageTable} (expiration_time, scheduled_time); + + CREATE INDEX IF NOT EXISTS {MessageSentTimeIndex(s)} + ON {s.MessageTable} (sent_time); + """; + + private static string AddConsumerManagement(IReadOnlyPostgresSchemaOptions s) => + $""" + CREATE TABLE IF NOT EXISTS {s.ConsumersTable} + ( + id uuid NOT NULL PRIMARY KEY, + service_name text NOT NULL, + created_at timestamptz NOT NULL DEFAULT now(), + updated_at timestamptz NOT NULL DEFAULT now() + ); + + ALTER TABLE {s.QueueTable} + ADD COLUMN IF NOT EXISTS consumer_id uuid REFERENCES {s.ConsumersTable}(id) ON DELETE CASCADE; + + CREATE INDEX IF NOT EXISTS {ConsumersUpdatedAtIndex(s)} + ON {s.ConsumersTable} (updated_at); + + CREATE INDEX IF NOT EXISTS {QueueConsumerIdIndex(s)} + ON {s.QueueTable} (consumer_id); + """; + + private static string TopicUniqueIndex(IReadOnlyPostgresSchemaOptions options) + => options.TablePrefix + "topic_uqx"; + + private static string TopicUniqueConstraint(IReadOnlyPostgresSchemaOptions options) + => options.TablePrefix + "unique_topic"; + + private static string QueueUniqueIndex(IReadOnlyPostgresSchemaOptions options) + => options.TablePrefix + "queue_uqx"; + + private static string QueueUniqueConstraint(IReadOnlyPostgresSchemaOptions options) + => options.TablePrefix + "unique_queue"; + + private static string QueueSubscriptionUniqueIndex(IReadOnlyPostgresSchemaOptions options) + => options.TablePrefix + "queue_subscription_uqx"; + + private static string QueueSubscriptionUniqueConstraint(IReadOnlyPostgresSchemaOptions options) + => options.TablePrefix + "unique_queue_subscription"; + + private static string QueueSubscriptionSourceIndex(IReadOnlyPostgresSchemaOptions options) + => options.TablePrefix + "queue_subscription_source_ndx"; + + private static string QueueSubscriptionDestinationIndex(IReadOnlyPostgresSchemaOptions options) + => options.TablePrefix + "queue_subscription_dest_ndx"; + + private static string MessageQueueIndex(IReadOnlyPostgresSchemaOptions options) + => options.TablePrefix + "message_queue_ndx"; + + private static string MessageTransportQueueIndex(IReadOnlyPostgresSchemaOptions options) + => options.TablePrefix + "message_transport_queue_ndx"; + + private static string MessageExpirationScheduledIndex(IReadOnlyPostgresSchemaOptions options) + => options.TablePrefix + "message_expiration_scheduled_ndx"; + + private static string MessageSentTimeIndex(IReadOnlyPostgresSchemaOptions options) + => options.TablePrefix + "message_sent_time_ndx"; + + private static string ConsumersUpdatedAtIndex(IReadOnlyPostgresSchemaOptions options) + => options.TablePrefix + "consumers_updated_at_ndx"; + + private static string QueueConsumerIdIndex(IReadOnlyPostgresSchemaOptions options) + => options.TablePrefix + "queue_consumer_id_ndx"; +} diff --git a/src/Mocha/src/Mocha.Transport.Postgres/Connection/PostgresSchemaSql.cs b/src/Mocha/src/Mocha.Transport.Postgres/Connection/PostgresSchemaSql.cs index 7106e57f0d1..028c3530a43 100644 --- a/src/Mocha/src/Mocha.Transport.Postgres/Connection/PostgresSchemaSql.cs +++ b/src/Mocha/src/Mocha.Transport.Postgres/Connection/PostgresSchemaSql.cs @@ -1,103 +1,59 @@ namespace Mocha.Transport.Postgres; -/// -/// Contains the SQL migration scripts for the PostgreSQL messaging transport schema. -/// -internal static class PostgresSchemaSql +internal static partial class PostgresSchemaSql { - public static string InitialSchema(IReadOnlyPostgresSchemaOptions s) => - $""" - CREATE SCHEMA IF NOT EXISTS {s.Schema}; - - CREATE SEQUENCE IF NOT EXISTS {s.TopologySequence} AS bigint; - - CREATE TABLE IF NOT EXISTS {s.TopicTable} - ( - id bigint NOT NULL PRIMARY KEY DEFAULT nextval('{s.TopologySequence}'), - updated timestamptz NOT NULL DEFAULT (now() at time zone 'utc'), - name text NOT NULL - ); - - CREATE UNIQUE INDEX IF NOT EXISTS {s.TablePrefix}topic_uqx ON {s.TopicTable} (name) INCLUDE (id); - ALTER TABLE {s.TopicTable} ADD CONSTRAINT {s.TablePrefix}unique_topic UNIQUE USING INDEX {s.TablePrefix}topic_uqx; - - CREATE TABLE IF NOT EXISTS {s.QueueTable} - ( - id bigint NOT NULL PRIMARY KEY DEFAULT nextval('{s.TopologySequence}'), - updated timestamptz NOT NULL DEFAULT (now() at time zone 'utc'), - name text NOT NULL - ); - - CREATE UNIQUE INDEX IF NOT EXISTS {s.TablePrefix}queue_uqx ON {s.QueueTable} (name) INCLUDE (id); - ALTER TABLE {s.QueueTable} ADD CONSTRAINT {s.TablePrefix}unique_queue UNIQUE USING INDEX {s.TablePrefix}queue_uqx; - - CREATE TABLE IF NOT EXISTS {s.QueueSubscriptionTable} - ( - id bigint NOT NULL PRIMARY KEY DEFAULT nextval('{s.TopologySequence}'), - updated timestamptz NOT NULL DEFAULT (now() at time zone 'utc'), - source_id bigint NOT NULL REFERENCES {s.TopicTable} (id) ON DELETE CASCADE, - destination_id bigint NOT NULL REFERENCES {s.QueueTable} (id) ON DELETE CASCADE - ); - - CREATE UNIQUE INDEX IF NOT EXISTS {s.TablePrefix}queue_subscription_uqx - ON {s.QueueSubscriptionTable} (source_id, destination_id); - ALTER TABLE {s.QueueSubscriptionTable} - ADD CONSTRAINT {s.TablePrefix}unique_queue_subscription UNIQUE USING INDEX {s.TablePrefix}queue_subscription_uqx; - - CREATE INDEX IF NOT EXISTS {s.TablePrefix}queue_subscription_source_ndx - ON {s.QueueSubscriptionTable} (source_id) INCLUDE (id, destination_id); - CREATE INDEX IF NOT EXISTS {s.TablePrefix}queue_subscription_dest_ndx - ON {s.QueueSubscriptionTable} (destination_id) INCLUDE (id, source_id); - - CREATE TABLE IF NOT EXISTS {s.MessageTable} - ( - transport_message_id uuid NOT NULL PRIMARY KEY DEFAULT gen_random_uuid(), - body bytea NOT NULL, - headers jsonb, - queue_id bigint NOT NULL REFERENCES {s.QueueTable} (id) ON DELETE CASCADE, - sent_time timestamptz NOT NULL DEFAULT (now() at time zone 'utc'), - scheduled_time timestamptz, - expiration_time timestamptz, - delivery_count int NOT NULL DEFAULT 0, - max_delivery_count int NOT NULL DEFAULT 10, - last_delivered timestamptz, - consumer_id uuid, - error_reason jsonb - ); - - CREATE INDEX IF NOT EXISTS {s.TablePrefix}message_queue_ndx - ON {s.MessageTable} (queue_id) INCLUDE (transport_message_id); - """; - - public static string AddTransportIndex(IReadOnlyPostgresSchemaOptions s) => - $""" - CREATE INDEX IF NOT EXISTS {s.TablePrefix}message_transport_queue_ndx - ON {s.MessageTable} (transport_message_id, queue_id); - - CREATE INDEX IF NOT EXISTS {s.TablePrefix}message_expiration_scheduled_ndx - ON {s.MessageTable} (expiration_time, scheduled_time); - - CREATE INDEX IF NOT EXISTS {s.TablePrefix}message_sent_time_ndx - ON {s.MessageTable} (sent_time); - """; - - public static string AddConsumerManagement(IReadOnlyPostgresSchemaOptions s) => - $""" - CREATE TABLE IF NOT EXISTS {s.ConsumersTable} - ( - id uuid NOT NULL PRIMARY KEY, - service_name text NOT NULL, - created_at timestamptz NOT NULL DEFAULT now(), - updated_at timestamptz NOT NULL DEFAULT now() - ); - - ALTER TABLE {s.QueueTable} - ADD COLUMN IF NOT EXISTS consumer_id uuid REFERENCES {s.ConsumersTable}(id) ON DELETE CASCADE; - - CREATE INDEX IF NOT EXISTS {s.TablePrefix}consumers_updated_at_ndx - ON {s.ConsumersTable} (updated_at); - - CREATE INDEX IF NOT EXISTS {s.TablePrefix}queue_consumer_id_ndx - ON {s.QueueTable} (consumer_id); - """; + private const int LockId = 958913715; + + public static string GenerateMigrationsSql(IReadOnlyPostgresSchemaOptions s) + => $""" + {Bootstrap(s)} + + {Migration(s, "2026-03-06_InitialSchema", InitialSchema(s))} + + {Migration(s, "2026-03-06_AddTransportIndex", AddTransportIndex(s))} + + {Migration(s, "2026-03-06_AddConsumerManagement", AddConsumerManagement(s))} + + """.ReplaceLineEndings("\n"); + + private static string Migration(IReadOnlyPostgresSchemaOptions s, string id, string sql) + { + var migrationIdLiteral = Literal(id); + + return DoBlock( + $""" + BEGIN + IF NOT EXISTS (SELECT 1 FROM {s.MigrationsTable} WHERE migration_id = {migrationIdLiteral}) THEN + {Indent(sql, 8)} + INSERT INTO {s.MigrationsTable} (migration_id) VALUES ({migrationIdLiteral}); + END IF; + END; + """); + } + + private static string DoBlock(string body) + { + var delimiter = "$mocha_migration$"; + var suffix = 0; + while (body.Contains(delimiter, StringComparison.Ordinal)) + { + suffix++; + delimiter = $"$mocha_migration_{suffix}$"; + } + + return $""" + DO {delimiter} + {body} + {delimiter}; + """; + } + + private static string Indent(string sql, int spaces) + { + var indentation = new string(' ', spaces); + return indentation + sql.ReplaceLineEndings("\n" + indentation); + } + + private static string Literal(string value) + => "E'" + value.Replace("\\", "\\\\").Replace("'", "''") + "'"; } diff --git a/src/Mocha/src/Mocha.Transport.Postgres/Descriptors/IPostgresMessagingTransportDescriptor.cs b/src/Mocha/src/Mocha.Transport.Postgres/Descriptors/IPostgresMessagingTransportDescriptor.cs index 5bf7c1c3b0c..d0300aebf59 100644 --- a/src/Mocha/src/Mocha.Transport.Postgres/Descriptors/IPostgresMessagingTransportDescriptor.cs +++ b/src/Mocha/src/Mocha.Transport.Postgres/Descriptors/IPostgresMessagingTransportDescriptor.cs @@ -81,6 +81,12 @@ public interface IPostgresMessagingTransportDescriptor /// The descriptor for method chaining. IPostgresMessagingTransportDescriptor AutoProvision(bool autoProvision = true); + /// + /// Sets whether transport schema migrations run at startup. Defaults to true. + /// When false, startup skips migration without checking the schema. + /// + IPostgresMessagingTransportDescriptor AutoMigrate(bool autoMigrate); + /// new IPostgresMessagingTransportDescriptor Name(string name); diff --git a/src/Mocha/src/Mocha.Transport.Postgres/Descriptors/PostgresMessagingTransportDescriptor.cs b/src/Mocha/src/Mocha.Transport.Postgres/Descriptors/PostgresMessagingTransportDescriptor.cs index bf3fd95cd4e..7665c1aa08f 100644 --- a/src/Mocha/src/Mocha.Transport.Postgres/Descriptors/PostgresMessagingTransportDescriptor.cs +++ b/src/Mocha/src/Mocha.Transport.Postgres/Descriptors/PostgresMessagingTransportDescriptor.cs @@ -146,6 +146,13 @@ public IPostgresMessagingTransportDescriptor AutoProvision(bool autoProvision = return this; } + /// + public IPostgresMessagingTransportDescriptor AutoMigrate(bool autoMigrate) + { + Configuration.AutoMigrate = autoMigrate; + return this; + } + /// public IPostgresMessagingTransportDescriptor ConnectionString(string connectionString) { diff --git a/src/Mocha/src/Mocha.Transport.Postgres/Descriptors/PostgresQueueDescriptor.cs b/src/Mocha/src/Mocha.Transport.Postgres/Descriptors/PostgresQueueDescriptor.cs index e41070c7bcf..7495893f83f 100644 --- a/src/Mocha/src/Mocha.Transport.Postgres/Descriptors/PostgresQueueDescriptor.cs +++ b/src/Mocha/src/Mocha.Transport.Postgres/Descriptors/PostgresQueueDescriptor.cs @@ -9,8 +9,7 @@ internal sealed class PostgresQueueDescriptor : MessagingDescriptorBase , IPostgresQueueDescriptor { - private PostgresQueueDescriptor(IMessagingConfigurationContext context, string name) - : base(context) + private PostgresQueueDescriptor(IMessagingConfigurationContext context, string name) : base(context) { Configuration = new PostgresQueueDescriptorConfiguration(name); } diff --git a/src/Mocha/src/Mocha.Transport.Postgres/PostgresMessagingTransport.cs b/src/Mocha/src/Mocha.Transport.Postgres/PostgresMessagingTransport.cs index 763daea4b92..2c9c1e6af46 100644 --- a/src/Mocha/src/Mocha.Transport.Postgres/PostgresMessagingTransport.cs +++ b/src/Mocha/src/Mocha.Transport.Postgres/PostgresMessagingTransport.cs @@ -14,6 +14,7 @@ public sealed class PostgresMessagingTransport : MessagingTransport private readonly Action _configure; private readonly PostgresBackgroundTaskScheduler _backgroundTasks = new(); private IReadOnlyPostgresSchemaOptions _schemaOptions = null!; + private bool _autoMigrate; /// /// Creates a new PostgreSQL transport with the specified configuration delegate. @@ -64,6 +65,7 @@ protected override void OnAfterInitialized(IMessagingSetupContext context) } _schemaOptions = configuration.SchemaOptions; + _autoMigrate = configuration.AutoMigrate; var connectionManagerLogger = context.Services.GetRequiredService>(); ConnectionManager = new PostgresConnectionManager( @@ -113,9 +115,8 @@ protected override void OnAfterInitialized(IMessagingSetupContext context) } /// - /// Ensures database migrations are run, registers the consumer, starts the notification listener, - /// provisions topology resources, and starts background tasks before the transport's endpoints - /// begin processing messages. + /// Prepares the configured schema and topology and registers the consumer before + /// the transport's endpoints begin processing messages. /// /// The configuration context for the current startup phase. /// A token to cancel the startup. @@ -123,7 +124,11 @@ protected override async ValueTask OnBeforeStartAsync( IMessagingConfigurationContext context, CancellationToken cancellationToken) { - await ConnectionManager.EnsureMigratedAsync(cancellationToken); + if (_autoMigrate) + { + await ConnectionManager.EnsureMigratedAsync(cancellationToken); + } + await ConsumerManager.RegisterAsync(cancellationToken); await NotificationListener.StartAsync(cancellationToken); diff --git a/src/Mocha/src/Mocha.Transport.Postgres/PostgresTransportSchema.cs b/src/Mocha/src/Mocha.Transport.Postgres/PostgresTransportSchema.cs new file mode 100644 index 00000000000..26416bdd9ec --- /dev/null +++ b/src/Mocha/src/Mocha.Transport.Postgres/PostgresTransportSchema.cs @@ -0,0 +1,67 @@ +using System.Data; +using System.Transactions; +using Npgsql; + +namespace Mocha.Transport.Postgres; + +public static class PostgresTransportSchema +{ + /// + /// Applies pending transport migrations in a transaction on an open, caller-owned connection. + /// The connection must have no active or ambient transaction and remains open after the operation. + /// + /// The open connection to the database to migrate. + /// The schema and table naming used by the transport. + /// A token to cancel migration, including waiting for its lock. + public static async Task MigrateAsync( + NpgsqlConnection connection, + IReadOnlyPostgresSchemaOptions schemaOptions, + CancellationToken cancellationToken) + { + ArgumentNullException.ThrowIfNull(connection); + ArgumentNullException.ThrowIfNull(schemaOptions); + cancellationToken.ThrowIfCancellationRequested(); + + if (connection.State != ConnectionState.Open) + { + throw ThrowHelper.MigrationConnectionMustBeOpen(); + } + + if (Transaction.Current is not null) + { + throw ThrowHelper.MigrationTransactionNotSupported(null); + } + + var sql = PostgresSchemaSql.GenerateMigrationsSql(schemaOptions); + NpgsqlTransaction transaction; + try + { + transaction = await connection.BeginTransactionAsync(cancellationToken); + } + catch (InvalidOperationException exception) + { + throw ThrowHelper.MigrationTransactionNotSupported(exception); + } + + await using (transaction) + { + await using var command = connection.CreateCommand(); + command.Transaction = transaction; + command.CommandText = sql; + await command.ExecuteNonQueryAsync(cancellationToken); + await transaction.CommitAsync(cancellationToken); + } + } + + /// + /// Returns an idempotent forward migration script with transaction boundaries and LF line endings. + /// Generation is synchronous and does not access a database or file. + /// + /// The schema and table naming used by the transport. + public static string GenerateMigrationsSql(IReadOnlyPostgresSchemaOptions schemaOptions) + { + ArgumentNullException.ThrowIfNull(schemaOptions); + + return "BEGIN;\n" + PostgresSchemaSql.GenerateMigrationsSql(schemaOptions) + "COMMIT;\n"; + } +} diff --git a/src/Mocha/src/Mocha.Transport.Postgres/ThrowHelper.cs b/src/Mocha/src/Mocha.Transport.Postgres/ThrowHelper.cs new file mode 100644 index 00000000000..815bada1668 --- /dev/null +++ b/src/Mocha/src/Mocha.Transport.Postgres/ThrowHelper.cs @@ -0,0 +1,10 @@ +namespace Mocha.Transport.Postgres; + +internal static class ThrowHelper +{ + public static InvalidOperationException MigrationConnectionMustBeOpen() + => new("Transport migration requires an open PostgreSQL connection."); + + public static InvalidOperationException MigrationTransactionNotSupported(Exception? innerException) + => new("Transport migration requires a connection without an active or ambient transaction.", innerException); +} diff --git a/src/Mocha/test/Mocha.Transport.Postgres.Tests/Behaviors/AutoProvisionIntegrationTests.cs b/src/Mocha/test/Mocha.Transport.Postgres.Tests/Behaviors/AutoProvisionIntegrationTests.cs index c73026a81d3..a69da3eb74d 100644 --- a/src/Mocha/test/Mocha.Transport.Postgres.Tests/Behaviors/AutoProvisionIntegrationTests.cs +++ b/src/Mocha/test/Mocha.Transport.Postgres.Tests/Behaviors/AutoProvisionIntegrationTests.cs @@ -143,8 +143,7 @@ public async Task ExplicitTopology_Should_Deliver_When_PreProvisionedAndAutoProv await using (var conn = new NpgsqlConnection(db.ConnectionString)) { await conn.OpenAsync(TestContext.Current.CancellationToken); - var migrator = new PostgresSchemaMigrator(schemaOptions); - await migrator.MigrateAsync(conn); + await PostgresTransportSchema.MigrateAsync(conn, schemaOptions, TestContext.Current.CancellationToken); } // Pre-provision topology resources directly in the database @@ -206,8 +205,7 @@ public async Task ExplicitTopology_Should_Deliver_When_MixedAutoProvision() await using (var conn = new NpgsqlConnection(db.ConnectionString)) { await conn.OpenAsync(TestContext.Current.CancellationToken); - var migrator = new PostgresSchemaMigrator(schemaOptions); - await migrator.MigrateAsync(conn); + await PostgresTransportSchema.MigrateAsync(conn, schemaOptions, TestContext.Current.CancellationToken); } // Pre-provision only the topic (with auto-provision disabled for it) diff --git a/src/Mocha/test/Mocha.Transport.Postgres.Tests/Connection/PostgresTransportSchemaTests.cs b/src/Mocha/test/Mocha.Transport.Postgres.Tests/Connection/PostgresTransportSchemaTests.cs new file mode 100644 index 00000000000..313c45b70ea --- /dev/null +++ b/src/Mocha/test/Mocha.Transport.Postgres.Tests/Connection/PostgresTransportSchemaTests.cs @@ -0,0 +1,350 @@ +using Microsoft.Extensions.DependencyInjection; +using Mocha.Transport.Postgres.Tests.Helpers; +using Npgsql; + +namespace Mocha.Transport.Postgres.Tests.Connection; + +[Collection("Postgres")] +public class PostgresTransportSchemaTests(PostgresFixture fixture) +{ + private readonly PostgresSchemaOptions _schemaOptions = new() { Schema = "messaging" }; + private static CancellationToken CancellationToken => TestContext.Current.CancellationToken; + + [Fact] + public async Task StartAsync_Should_CreateSchema_When_AutoMigrateEnabledAndSchemaPermissionsGranted() + { + // arrange + await using var database = await fixture.CreateDatabaseAsync(); + await using var connection = await OpenConnectionAsync(database.ConnectionString); + var services = new ServiceCollection(); + services.AddMessageBus() + .AddPostgres(transport => + { + transport.ConnectionString(database.ConnectionString); + transport.ExtendWith(extension => extension.Configuration.SchemaOptions = _schemaOptions); + }); + var provider = services.BuildServiceProvider(); + var runtime = (MessagingRuntime)provider.GetRequiredService(); + await using var transport = runtime.Transports.OfType().Single(); + await using var bus = new TestBus(provider, runtime); + + // act + await runtime.StartAsync(CancellationToken); + + // assert + var migrationHistory = await ReadMigrationHistoryAsync(connection); + + Assert.Equal( + ["2026-03-06_AddConsumerManagement", "2026-03-06_AddTransportIndex", "2026-03-06_InitialSchema"], + migrationHistory); + } + + [Fact] + public async Task StartAsync_Should_Fail_When_AutoMigrateEnabledWithoutSchemaPermissions() + { + // arrange + await using var database = await fixture.CreateDatabaseAsync(); + await using var connection = await OpenConnectionAsync(database.ConnectionString); + var role = $"mocha_runtime_{Guid.NewGuid():N}"; + await ExecuteNonQueryAsync(connection, $"CREATE ROLE {role} LOGIN PASSWORD 'runtime-test';"); + + try + { + await ExecuteNonQueryAsync(connection, + $""" + GRANT CONNECT ON DATABASE "{database.DatabaseName}" TO {role}; + """); + var settings = new NpgsqlConnectionStringBuilder(database.ConnectionString) + { + Username = role, + Password = "runtime-test", + Pooling = false + }; + var services = new ServiceCollection(); + services.AddMessageBus() + .AddPostgres(transport => + { + transport.ConnectionString(settings.ConnectionString); + transport.ExtendWith(extension => extension.Configuration.SchemaOptions = _schemaOptions); + }); + var provider = services.BuildServiceProvider(); + var runtime = (MessagingRuntime)provider.GetRequiredService(); + await using var transport = runtime.Transports.OfType().Single(); + await using var bus = new TestBus(provider, runtime); + + // act + var exception = await Assert.ThrowsAsync( + () => runtime.StartAsync(CancellationToken).AsTask()); + + // assert + Assert.Equal(PostgresErrorCodes.InsufficientPrivilege, exception.SqlState); + } + finally + { + await ExecuteNonQueryAsync(connection, $"DROP OWNED BY {role}; DROP ROLE {role};"); + } + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task MigrateAsync_Should_CreateResources_When_DatabaseIsEmpty(bool useMigrationScript) + { + // arrange + await using var database = await fixture.CreateDatabaseAsync(); + await using var connection = await OpenConnectionAsync(database.ConnectionString); + + // act + await ApplyMigrationsAsync(connection, useMigrationScript); + + // assert + var migrationHistory = await ReadMigrationHistoryAsync(connection); + var tables = await ReadTableNamesAsync(connection); + var sequences = await ReadSequenceNamesAsync(connection); + + Assert.Equal( + ["2026-03-06_AddConsumerManagement", "2026-03-06_AddTransportIndex", "2026-03-06_InitialSchema"], + migrationHistory); + Assert.Equal( + [ + "mocha_consumers", + "mocha_message", + "mocha_migrations", + "mocha_queue", + "mocha_queue_subscription", + "mocha_topic" + ], + tables); + Assert.Equal(["mocha_topology_seq"], sequences); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task MigrateAsync_Should_LeaveResourcesUnchanged_When_RunTwice(bool useMigrationScript) + { + // arrange + await using var database = await fixture.CreateDatabaseAsync(); + await using var connection = await OpenConnectionAsync(database.ConnectionString); + await ApplyMigrationsAsync(connection, useMigrationScript); + var tables = await ReadTableNamesAsync(connection); + var migrationHistory = await ReadMigrationHistoryAsync(connection); + + // act + await ApplyMigrationsAsync(connection, useMigrationScript); + + // assert + var repeatedTables = await ReadTableNamesAsync(connection); + var repeatedMigrationHistory = await ReadMigrationHistoryAsync(connection); + + Assert.Equal(tables, repeatedTables); + Assert.Equal(migrationHistory, repeatedMigrationHistory); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task MigrateAsync_Should_ApplyPendingMigrations_When_OnlyInitialMigrationIsRecorded(bool useMigrationScript) + { + // arrange + await using var database = await fixture.CreateDatabaseAsync(); + await using var connection = await OpenConnectionAsync(database.ConnectionString); + await ExecuteNonQueryAsync(connection, PostgresSchemaSql.InitialSchema(_schemaOptions)); + await ExecuteNonQueryAsync(connection, + """ + CREATE TABLE messaging.mocha_migrations ( + migration_id text PRIMARY KEY, + applied_on timestamptz NOT NULL DEFAULT now()); + INSERT INTO messaging.mocha_migrations (migration_id) VALUES ('2026-03-06_InitialSchema'); + """); + var tables = await ReadTableNamesAsync(connection); + + // act + await ApplyMigrationsAsync(connection, useMigrationScript); + + // assert + var migrationHistory = await ReadMigrationHistoryAsync(connection); + var migratedTables = await ReadTableNamesAsync(connection); + var messageIndexes = await ReadMessageIndexNamesAsync(connection); + var queueColumns = await ReadQueueColumnNamesAsync(connection); + + Assert.Equal( + ["2026-03-06_AddConsumerManagement", "2026-03-06_AddTransportIndex", "2026-03-06_InitialSchema"], + migrationHistory); + Assert.Equal(["mocha_consumers"], migratedTables.Except(tables)); + Assert.Equal( + [ + "mocha_message_expiration_scheduled_ndx", + "mocha_message_pkey", + "mocha_message_queue_ndx", + "mocha_message_sent_time_ndx", + "mocha_message_transport_queue_ndx" + ], + messageIndexes); + Assert.Equal( + [ + "id", + "updated", + "name", + "consumer_id" + ], + queueColumns); + } + + [Fact] + public async Task PublishAsync_Should_Deliver_When_ScriptAppliedAndAutoMigrateDisabled() + { + // arrange + var recorder = new MessageRecorder(); + await using var database = await fixture.CreateDatabaseAsync(); + await using var connection = await OpenConnectionAsync(database.ConnectionString); + var script = PostgresTransportSchema.GenerateMigrationsSql(_schemaOptions); + await ExecuteNonQueryAsync(connection, script); + var role = $"mocha_runtime_{Guid.NewGuid():N}"; + await ExecuteNonQueryAsync(connection, $"CREATE ROLE {role} LOGIN PASSWORD 'runtime-test';"); + + try + { + await ExecuteNonQueryAsync(connection, + $""" + GRANT CONNECT ON DATABASE "{database.DatabaseName}" TO {role}; + GRANT USAGE ON SCHEMA messaging TO {role}; + GRANT SELECT, INSERT, UPDATE, DELETE ON + messaging.mocha_topic, messaging.mocha_queue, messaging.mocha_queue_subscription, + messaging.mocha_message, messaging.mocha_consumers TO {role}; + GRANT USAGE ON SEQUENCE messaging.mocha_topology_seq TO {role}; + """); + var settings = new NpgsqlConnectionStringBuilder(database.ConnectionString) + { + Username = role, + Password = "runtime-test", + Pooling = false + }; + await using var bus = await new ServiceCollection() + .AddSingleton(recorder) + .AddMessageBus() + .AddEventHandler() + .AddPostgres(transport => + { + transport.ConnectionString(settings.ConnectionString).AutoMigrate(false); + transport.ExtendWith(extension => extension.Configuration.SchemaOptions = _schemaOptions); + }) + .BuildTestBusAsync(); + using var scope = bus.Provider.CreateScope(); + var messageBus = scope.ServiceProvider.GetRequiredService(); + + // act + await messageBus.PublishAsync(new OrderCreated { OrderId = "ORD-1" }, CancellationToken); + var delivered = await recorder.WaitAsync(TimeSpan.FromSeconds(30)); + + // assert + Assert.True(delivered, "Handler did not receive the event."); + var order = Assert.Single(recorder.Messages.Cast()); + Assert.Equal("ORD-1", order.OrderId); + } + finally + { + await ExecuteNonQueryAsync(connection, $"DROP OWNED BY {role}; DROP ROLE {role};"); + } + } + + [Fact] + public async Task MigrateAsync_Should_RollBack_When_MigrationFails() + { + // arrange + await using var database = await fixture.CreateDatabaseAsync(); + await using var connection = await OpenConnectionAsync(database.ConnectionString); + // The conflicting view deliberately makes index creation fail after earlier resources have been created. + await ExecuteNonQueryAsync(connection, + """ + CREATE SCHEMA messaging; + CREATE VIEW messaging.mocha_message AS SELECT 1 AS sentinel; + """); + + // act + await Assert.ThrowsAsync( + () => PostgresTransportSchema.MigrateAsync(connection, _schemaOptions, CancellationToken)); + + // assert + var relations = await ReadRelationNamesAsync(connection); + + Assert.Equal(["mocha_message"], relations); + } + + private static async Task OpenConnectionAsync(string connectionString) + { + var settings = new NpgsqlConnectionStringBuilder(connectionString) { Pooling = false, Enlist = false }; + var connection = new NpgsqlConnection(settings.ConnectionString); + await connection.OpenAsync(CancellationToken); + return connection; + } + + private Task ApplyMigrationsAsync(NpgsqlConnection connection, bool useMigrationScript) + { + if (useMigrationScript) + { + return ExecuteNonQueryAsync(connection, PostgresTransportSchema.GenerateMigrationsSql(_schemaOptions)); + } + + return PostgresTransportSchema.MigrateAsync(connection, _schemaOptions, CancellationToken); + } + + private static async Task ExecuteNonQueryAsync(NpgsqlConnection connection, string sql) + { + await using var command = connection.CreateCommand(); + command.CommandText = sql; + await command.ExecuteNonQueryAsync(CancellationToken); + } + + private static async Task ExecuteScalarAsync(NpgsqlConnection connection, string sql) + { + await using var command = connection.CreateCommand(); + command.CommandText = sql; + var result = await command.ExecuteScalarAsync(CancellationToken); + return (T)result!; + } + + private static Task ReadMigrationHistoryAsync(NpgsqlConnection connection) + => ExecuteScalarAsync(connection, + "SELECT array_agg(migration_id ORDER BY migration_id) FROM messaging.mocha_migrations;"); + + private static Task ReadTableNamesAsync(NpgsqlConnection connection) + => ExecuteScalarAsync(connection, + """ + SELECT array_agg(table_name ORDER BY table_name) + FROM information_schema.tables + WHERE table_schema = 'messaging' AND table_type = 'BASE TABLE'; + """); + + private static Task ReadSequenceNamesAsync(NpgsqlConnection connection) + => ExecuteScalarAsync(connection, + """ + SELECT array_agg(sequence_name ORDER BY sequence_name) + FROM information_schema.sequences + WHERE sequence_schema = 'messaging'; + """); + + private static Task ReadMessageIndexNamesAsync(NpgsqlConnection connection) + => ExecuteScalarAsync(connection, + """ + SELECT array_agg(indexname ORDER BY indexname) + FROM pg_indexes + WHERE schemaname = 'messaging' AND tablename = 'mocha_message'; + """); + + private static Task ReadQueueColumnNamesAsync(NpgsqlConnection connection) + => ExecuteScalarAsync(connection, + """ + SELECT array_agg(column_name ORDER BY ordinal_position) + FROM information_schema.columns + WHERE table_schema = 'messaging' AND table_name = 'mocha_queue'; + """); + + private static Task ReadRelationNamesAsync(NpgsqlConnection connection) + => ExecuteScalarAsync(connection, + """ + SELECT array_agg(relname ORDER BY relname) + FROM pg_class + WHERE relnamespace = 'messaging'::regnamespace; + """); +} diff --git a/website/content/docs/mocha/transports/postgres.md b/website/content/docs/mocha/transports/postgres.md index 5b4e76df1dc..2f5432e1e43 100644 --- a/website/content/docs/mocha/transports/postgres.md +++ b/website/content/docs/mocha/transports/postgres.md @@ -46,7 +46,7 @@ var app = builder.Build(); app.Run(); ``` -`.AddPostgres(connectionString)` creates an `NpgsqlDataSource` from the connection string, runs schema migrations on first use, and provisions topics, queues, and subscriptions for your registered handlers. +`.AddPostgres(connectionString)` creates an `NpgsqlDataSource` from the connection string, runs schema migrations at transport startup, and provisions topics, queues, and subscriptions for your registered handlers. ## Register with .NET Aspire @@ -243,12 +243,36 @@ Changing `Schema` and `TablePrefix` shifts all table names accordingly. For exam ## Schema migration -The transport runs migrations automatically on first use. Migrations are protected by a PostgreSQL advisory lock (`pg_advisory_xact_lock`) to prevent concurrent migration attempts from multiple service instances starting simultaneously. +The transport runs migrations automatically at startup by default. Migrations are protected by a PostgreSQL advisory lock (`pg_advisory_xact_lock`) to prevent concurrent migration attempts from multiple service instances starting simultaneously. Each migration is tracked in the migrations table and is idempotent - running the same migration twice has no effect. The migration creates the schema if it does not exist, then applies each pending migration in order within a single transaction. > [!WARNING] -> The advisory lock ID is fixed. If you run multiple independent Mocha transports in the same PostgreSQL cluster with different table prefixes, they share the same advisory lock. This is safe - it serializes migrations but does not block normal message operations. +> The advisory lock ID is fixed. Independent Mocha transports in the same database share this lock, including transports with different table prefixes. The lock serializes migrations without blocking normal message operations. + +To set up the database schema at a different point in time, disable automatic migration: + +```csharp +builder.Services.AddMessageBus().AddPostgres(transport => +{ + transport.ConnectionString(connectionString); + transport.AutoMigrate(false); +}); +``` + +Call `PostgresTransportSchema.MigrateAsync` before the transport is started. Use an open connection with schema modification permissions and the same schema options as the transport: + +```csharp +// Match the transport's schema and table prefix. +var options = new PostgresSchemaOptions(); + +await using var connection = new NpgsqlConnection(connectionString); +await connection.OpenAsync(cancellationToken); +await PostgresTransportSchema.MigrateAsync( + connection, + options, + cancellationToken); +``` # Configure queues @@ -336,7 +360,7 @@ If the queue already declares `AutoDelete(false)` explicitly, `Temporary()` fail # Control auto-provisioning -By default, the transport auto-provisions all topology resources (topics, queues, subscriptions) in the database at startup. In environments where database schema is managed externally - for example by Flyway, Liquibase, or a CI/CD pipeline - you can disable auto-provisioning so the transport expects resources to already exist. +By default, the transport auto-provisions topology resources (topics, queues, subscriptions) as rows at startup. Disable auto-provisioning when these rows are managed externally. Schema migration is controlled separately by `AutoMigrate`, as described in [Schema migration](#schema-migration). ## Disable globally