PouryaKit.Persistence.Marten 0.1.0-preview.2

PouryaKit.Persistence.Marten

Streams on Marten over PostgreSQL: the Kit's second write model. A Stream is an aggregate whose stored state is the events it appended. A behaviour applies its event to the aggregate's state and raises it, the Unit of Work appends what was raised, and loading the Stream folds its stored events through the same Apply methods.

Install

dotnet add package PouryaKit.Persistence.Marten --prerelease

When to use it

• An aggregate's history is the point: a ledger, an account, an audit of every change.
• You register a Stream Module, beside EF Core Modules or without them.
• You load and start Streams in a handler, and handle a concurrent append as a conflict.

A Host with a message store adds PouryaKit.Messaging.Wolverine.Marten, which commits a Stream's events with what it sends. PouryaKit.Persistence.Postgres is the other write model.

Quick start

Declare the write model in the Module's Composition Root:

module.AddPouryaKitStreams(schema: "ledger");

Write behaviour as events, as the Proving Sample's Ledger does:

internal sealed class Account : StreamAggregate
{
    public string Holder { get; private set; } = "";

    public decimal Balance { get; private set; }

    public void Post(decimal amount)
    {
        var posted = new EntryPostedDomainEvent(new AccountId(Id), amount);
        Apply(posted);
        Raise(posted);
    }

    public void Apply(EntryPostedDomainEvent posted) => Balance += posted.Amount;
}

Load and change a Stream in the handler through IStreams:

internal sealed class PostEntryCommandHandler(IStreams streams) : CommandHandler<PostEntryCommand>
{
    public override async Task<Result> Handle(
        PostEntryCommand command,
        CancellationToken cancellationToken
    )
    {
        if (
            await streams.LoadAsync<Account>(new AccountId(command.AccountId), cancellationToken)
            is not { } account
        )
        {
            return AccountErrors.NotFound;
        }

        account.Post(command.Amount);
        return Result.Success();
    }
}

What you get

• A Stream Module that is still a Module: the same Gateway, pipeline and Notification Handlers.
• One commit owner per write model, so each Operation commits through exactly one.
• A concurrent append refused as a conflict Error, with nothing appended.
• A Stream that belongs to the tenant of the Operation that wrote it.

Learn more

• Streams
• Stream
• Stream or EF Core aggregate
• Limits