Fluens.Migrations 0.9.1

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(). Upgrade helpers for already-deployed schemas: AddMintSeqColumns(), AddDeadLetterModuleNameColumn() (see Upgrading an existing database below).

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.

dead_letters.module_name is the deliberate exception to "an added column carries a default": in a freshly created table it is required and has no default, because every dead letter must record the module that owns it (ADR 20261006184926). Tables created before these columns existed are brought up to date with the upgrade helpers below.

Upgrading an existing database

Every schema change to a Fluens table ships a protected upgrade helper on FluensMigration. Call it from your own numbered migration - Fluens never runs a migration itself and keeps no version table, so you stay the owner of your migration history. Each helper is listed in the changelog under the release that needs it, and a host that skips one is refused at startup by the messaging schema verification (see the Fluens.Messaging README).

• AddMintSeqColumns(schema) - for outbox and inbox tables created before 0.7.7. Adds the bigint mint_seq column (non-null, default 0) to <schema>.outbox and <schema>.inbox and rebuilds ix_outbox_pending and ix_inbox_pending. Call it once per module schema.
• AddDeadLetterModuleNameColumn(moduleNamesBySchema, schema = "shared") - for a dead_letters table created before 0.8.5. Adds the module_name column, backfills every row (a row whose original_schema is a key of the map gets the mapped module name, every other row keeps its original_schema), makes the column required and creates ix_dead_letters_module_failed_at and ix_dead_letters_module_replayed_at.

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

    public override void Up()
    {
        AddMintSeqColumns("orders");
        AddMintSeqColumns("billing");

        // original_schema -> owning module name
        AddDeadLetterModuleNameColumn(new Dictionary<string, string>
        {
            ["orders"] = "orders",
            ["billing"] = "billing",
        });
    }

    public override void Down() { }
}

Give the migration the next version number in your own sequence. A helper is never removed or reshaped once published, so a migration that calls it keeps working on every later release.

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.