Fluens.Migrations 0.8.5

Fluens.Migrations

Provider-neutral FluentMigrator base classes and helpers for inbox/outbox tables, dead letters, saga tables, HiLo sequences, and auditable/deletable columns. Zero Fluens dependencies.

Installation

dotnet add package Fluens.Migrations

# Plus exactly one provider package:
dotnet add package Fluens.Migrations.PostgreSql
# or
dotnet add package Fluens.Migrations.SqlServer

Usage

Runner registration

The provider is opted into from the AddFluens callback on FluentMigrator's own ConfigureRunner pipeline — never through a bespoke top-level method. rb.AddPostgres() / rb.AddSqlServer() (FluentMigrator's own provider registration) and .AddFluens(f => f.AddPostgres()) / .AddFluens(f => f.AddSqlServer()) (the Fluens conventions opt-in) are distinct extension targets — no ambiguity:

services.AddFluentMigratorCore()
    .ConfigureRunner(rb => rb
        .AddPostgres()                             // FluentMigrator's own provider registration
        .WithGlobalConnectionString(connectionString)
        .ScanIn(typeof(CreateOrdersSchema).Assembly).For.Migrations()
        .AddFluens(f => f.AddPostgres()));          // Fluens provider-conventions opt-in

Migration classes

Every FluensMigration subclass needs a forwarding constructor — Conventions is resolved by FluentMigrator's own constructor-injection support when the runner instantiates the migration, so a single, unchanged subclass creates the same schema on both providers:

public class CreateOrdersSchema : FluensMigration
{
    public CreateOrdersSchema(IMigrationProviderConventions conventions) : base(conventions) { }

    public override void Up()
    {
        Create.Schema("orders");

        Create.Table("orders").InSchema("orders")
            .WithColumn("id").AsGuid().PrimaryKey()
            .WithColumn("number").AsString(50).NotNullable()
            .WithColumn("total").AsDecimal().NotNullable()
            .WithAuditableAndDeletableColumns()
            .WithVersionColumn();

        CreateMessagingTables("orders"); // outbox + inbox + dead_letters
        CreateHiLoSequence("orders_hilo", "orders");
    }

    public override void Down()
    {
        DeleteMessagingTables("orders");
        DeleteHiLoSequence("orders_hilo", "orders");
        Delete.Table("orders").InSchema("orders");
    }
}

Helpers: WithAuditableColumns(), WithDeletableColumn(), WithAuditableAndDeletableColumns(), WithVersionColumn(), CreateOutboxTable(), CreateOutboxDeliveryTable(), CreateInboxTable(), CreateIngressOutboxTable(), CreateDeadLettersTable(), CreateSagaTable(), CreateMessagingTables(), CreateHiLoSequence(), DeleteOutboxTable(), DeleteOutboxDeliveryTable(), DeleteInboxTable(), DeleteIngressOutboxTable(), DeleteDeadLettersTable(), DeleteSagaTable(), DeleteMessagingTables(), DeleteHiLoSequence().

Messaging schema (ADR 20260530000001 / ADR 20260530000002 / ADR 20260729000001)

All four messaging tables carry the serialized envelope blob (non-nullable; ADR 20260530000001 — replaces the bare context column) plus the denormalized routing columns (partition_key, priority, and on outbox also available_at) so the partial pending indexes can sort / filter on real columns. Each table has a partial pending index emitted via Conventions.ApplyFilter(<index>.WithOptions(), "<predicate>") — the registered provider package's .Filter(...) extension — so the hot fetch scans only in-flight rows:

• outbox — the terminal marker is sent_at (renamed from the legacy published_at flag; ADR 20260530000002). Message-TTL columns expires_at (non-nullable, computed from (available_at ?? created_at) at write) and expired_at (nullable, set in place on TTL expiry — the row is never deleted, only marked). Also carries mint_seq (bigint, non-nullable, default 0 — a provider-neutral monotonic mint-time key derived from id's embedded GUIDv7 timestamp; ADR 20260729000001). Partial ix_outbox_pending (WHERE sent_at IS NULL) over (sent_at, priority DESC, mint_seq).
• inbox — composite primary key (message_id, handler_type) (one row per subscribing handler). Also carries mint_seq (same shape as outbox, derived from message_id) and adds next_retry_at (stage-2 retry schedule). Partial ix_inbox_pending (WHERE processed_at IS NULL) over (handler_type, partition_key, priority DESC, mint_seq).
• ingress_outbox — primary key message_id (boundary dedup). Partial ix_ingress_pending (WHERE processed_at IS NULL) over received_at.
• dead_letters — adds the owning module_name (string(100), NOT NULL, no default), the why-columns handler_type, partition_key, failure_code, exception_type, attempt_history (Conventions.JsonColumnType — jsonb on PostgreSQL, nvarchar(max) on SQL Server), and the replay marker replayed_at. Two module-leading indexes serve the per-module operator queries (query, replay, housekeeping): ix_dead_letters_module_failed_at on (module_name ASC, failed_at DESC) and ix_dead_letters_module_replayed_at on (module_name ASC, replayed_at ASC); DeleteDeadLettersTable drops both, then the table.

No upgrade (AddColumn) migration ships for mint_seq — CreateOutboxTable/CreateInboxTable are create-table helpers only, so a consumer upgrading an already-deployed schema authors their own AddColumn("mint_seq").AsInt64().NotNullable().WithDefaultValue(0) step plus the corresponding index rebuild (ADR 20260729000001).

dead_letters.module_name is the deliberate exception to "an added column carries a default": it is required, has no default and has no backward compatibility, because every dead letter must record the module that owns it. A consumer upgrading an already-deployed dead_letters table adds the column and the two indexes in its own migration (ADR 20261006184926).

Saga table (ADR 20261006184927)

CreateSagaTable(tableName, schema) creates the SagaData base schema for a Fluens.Messaging.Sagas saga and returns the table syntax, so you chain your business columns straight on. The base columns are id (Guid, primary key), created_at (NOT NULL), completed_at (nullable) and version, whose type comes from Conventions.SagaVersionColumnType because the column type EF Core uses for the uint concurrency token differs per provider (bigint on both PostgreSQL and SQL Server). Three indexes back the SQL-side saga operator queries:

• ix_<tableName>_active — created_at DESC, partial WHERE completed_at IS NULL (via Conventions.ApplyFilter)
• ix_<tableName>_created_at — created_at DESC
• ix_<tableName>_completed_at — completed_at ASC

public override void Up()
{
    CreateSagaTable("order_fulfillment", "orders")
        .WithColumn("order_id").AsGuid().NotNullable()
        .WithColumn("customer_email").AsString(200).NotNullable()
        .WithColumn("payment_captured").AsBoolean().NotNullable();
}

public override void Down() => DeleteSagaTable("order_fulfillment", "orders");

DeleteSagaTable drops the three indexes, then the table. The indexes come from this helper only — the EF model of a consumer's saga table declares none.

CreateOutboxDeliveryTable(schema) creates the per-destination delivery table (one row per (outbox_id, destination_app) pair) with a unique ix_outbox_delivery_outbox_dest index on (outbox_id, destination_app) for re-fan-out idempotency, an ix_outbox_delivery_destination_app index, and the partial ix_delivery_pending (WHERE delivered_at IS NULL AND expired_at IS NULL) over (expires_at, destination_app); CreateMessagingTables creates it alongside outbox/inbox in the module schema. CreateIngressOutboxTable(schema = "shared") creates the single staging table into which a peer-delivered batch is written atomically before acking.

Provider-specific column types

Use Fluens.Migrations.PostgreSql or Fluens.Migrations.SqlServer for provider-specific column types — independent of which IMigrationProviderConventions is registered for the run:

using Fluens.Migrations.PostgreSql; // AsCitext(), AsTimestampTz()

Create.Table("customers").InSchema("orders")
    .WithColumn("id").AsInt32().PrimaryKey().Identity()
    .WithColumn("email").AsCitext().NotNullable()         // case-insensitive text
    .WithColumn("created_at").AsTimestampTz().NotNullable(); // timestamp with time zone

using Fluens.Migrations.SqlServer; // AsNVarCharMax(), AsDateTimeOffset2()

Create.Table("customers").InSchema("orders")
    .WithColumn("id").AsInt32().PrimaryKey().Identity()
    .WithColumn("notes").AsNVarCharMax().Nullable()             // unbounded Unicode text
    .WithColumn("created_at").AsDateTimeOffset2().NotNullable(); // timestamp with time zone

Helpers: Fluens.Migrations.PostgreSql.AsCitext() / .AsTimestampTz(); Fluens.Migrations.SqlServer.AsNVarCharMax() / .AsDateTimeOffset2().

License

This project is licensed under the MIT License.