Command-kö & ångra — single-writer och compensating events

Två betydelser av "command"

Vi använder CQRS-betydelsen på serversidan. På klientsidan i Blazor (lektion 10) bygger vi en GoF-liknande undo-stack som genererar compensating commands.

Varför en kö?

Även om optimistic concurrency gör att vi aldrig tappar data, så är det dyrt att rulla tillbaka och försöka igen. Bättre att i de fall vi vet hur — undvika konflikten helt: serialisera skrivningarna per aggregat. En kö per StreamId garanterar att två commands mot samma aggregat aldrig körs parallellt.

Implementation med Channel<T>

.NET 10 har System.Threading.Channels som är gjort för exakt detta — en lock-fri, asynkron producer/consumer-kö.

public interface ICommandEnvelope
{
    string AggregateStreamId { get; }     // partitioneringsnyckel
    Guid MessageId { get; }
}

public sealed record EnqueuedCommand(
    ICommandEnvelope Command,
    TaskCompletionSource<object?> Completion);

public sealed class AggregateCommandQueue
{
    private readonly Dictionary<string, Channel<EnqueuedCommand>> _channels = new();
    private readonly object _gate = new();
    private readonly IServiceScopeFactory _scopes;
    private readonly ILogger<AggregateCommandQueue> _log;

    public AggregateCommandQueue(IServiceScopeFactory scopes, ILogger<AggregateCommandQueue> log)
        => (_scopes, _log) = (scopes, log);

    public Task<object?> EnqueueAsync(ICommandEnvelope cmd, CancellationToken ct)
    {
        var ch = GetOrCreate(cmd.AggregateStreamId);
        var completion = new TaskCompletionSource<object?>(TaskCreationOptions.RunContinuationsAsynchronously);
        ch.Writer.TryWrite(new EnqueuedCommand(cmd, completion));
        return completion.Task.WaitAsync(ct);
    }

    private Channel<EnqueuedCommand> GetOrCreate(string streamId)
    {
        lock (_gate)
        {
            if (_channels.TryGetValue(streamId, out var existing)) return existing;

            var ch = Channel.CreateUnbounded<EnqueuedCommand>(
                new UnboundedChannelOptions { SingleReader = true });
            _channels[streamId] = ch;

            _ = Task.Run(() => PumpAsync(streamId, ch.Reader));
            return ch;
        }
    }

    private async Task PumpAsync(string streamId, ChannelReader<EnqueuedCommand> reader)
    {
        await foreach (var item in reader.ReadAllAsync())
        {
            try
            {
                using var scope = _scopes.CreateScope();
                var mediator = scope.ServiceProvider.GetRequiredService<IMediator>();
                var result = await mediator.Send(item.Command);
                item.Completion.TrySetResult(result);
            }
            catch (Exception ex)
            {
                _log.LogError(ex, "Command failed on stream {Stream}", streamId);
                item.Completion.TrySetException(ex);
            }
        }
    }
}

Registrering

builder.Services.AddSingleton<AggregateCommandQueue>();

app.MapPost("/accounts/{id:guid}/deposit", async (
    Guid id, DepositRequest body,
    AggregateCommandQueue queue,
    [FromHeader(Name = "Idempotency-Key")] Guid? idempotencyKey,
    CancellationToken ct) =>
{
    var cmd = new DepositMoneyCommand(
        MessageId: idempotencyKey ?? Guid.NewGuid(),
        Account: new AccountId(id),
        Amount: body.Amount,
        Currency: body.Currency,
        Reference: body.Reference);

    var result = await queue.EnqueueAsync(cmd, ct);
    return Results.Ok(result);
});

Ångra i en event-sourced värld

Den kardinalsynd man vill begå när man kommer från CRUD-tänk: "jag tar bara bort eventet". Detta får man aldrig göra — andra projektioner, granskningsrapporter och historiska analyser har redan sett eventet.

Den korrekta vägen: emittera ett nytt event som kompenserar det förra. Eventet beskriver vad som hände — "insättningen återställdes", inte "insättningen togs bort".

public sealed record MoneyDepositReverted(
    Guid EventId,
    DateTimeOffset OccurredAt,
    AccountId Account,
    Guid OriginalEventId,
    decimal Amount,
    string Currency,
    string Reason) : IDomainEvent;

public partial class Account
{
    public void RevertDeposit(Guid originalEventId, string reason)
    {
        if (_state == AccountState.Closed)
            throw new InvalidOperationException("Cannot revert on a closed account.");

        var original = _committedDeposits.FirstOrDefault(d => d.EventId == originalEventId)
            ?? throw new InvalidOperationException("Original deposit not found in this stream.");

        Raise(new MoneyDepositReverted(
            EventId: Guid.NewGuid(),
            OccurredAt: DateTimeOffset.UtcNow,
            Account: Id,
            OriginalEventId: originalEventId,
            Amount: original.Amount,
            Currency: original.Currency,
            Reason: reason));
    }

    private void Apply(MoneyDepositReverted e) =>
        Balance = Balance.Subtract(new Money(e.Amount, e.Currency));
}

Compensating command

public sealed record RevertDepositCommand(
    Guid MessageId,
    AccountId Account,
    Guid OriginalEventId,
    string Reason) : ICommand<Unit>, ICommandEnvelope
{
    public string AggregateStreamId => $"account-{Account.Value}";
}

public sealed class RevertDepositHandler : IRequestHandler<RevertDepositCommand, Unit>
{
    private readonly IAccountRepository _repo;
    public RevertDepositHandler(IAccountRepository repo) => _repo = repo;

    public async Task<Unit> Handle(RevertDepositCommand cmd, CancellationToken ct)
    {
        var account = await _repo.LoadAsync(cmd.Account, ct)
            ?? throw new NotFoundException(cmd.Account);
        account.RevertDeposit(cmd.OriginalEventId, cmd.Reason);
        await _repo.SaveAsync(account, ct);
        return Unit.Value;
    }
}

Undo-stack på klientsidan (förhandstitt)

Server-sidan vet inget om "senaste handlingen". Klienten håller en stack med commands den har skickat — och när användaren trycker Ctrl+Z genereras motsvarande compensating command.

public sealed class UndoStack
{
    private readonly Stack<ICommandEnvelope> _history = new();
    private readonly Stack<ICommandEnvelope> _redo = new();

    public void Record(ICommandEnvelope executed, ICommandEnvelope compensation)
    {
        _history.Push(compensation);
        _redo.Clear();
    }

    public ICommandEnvelope? PopUndo() => _history.Count > 0 ? _history.Pop() : null;
}

Detaljerna kring Blazor-integrationen finns i lektion 10.

Felscenarier i kön

Sammanfattning

• En Channel<T> per StreamId serialiserar commands mot ett aggregat — single-writer-principen.
• Olika aggregat körs parallellt, samma aggregat strikt sekventiellt.
• "Undo" i ES = compensating event (MoneyDepositReverted), inte radering.
• Klienten håller undo-stacken och genererar kompensationer som vanliga commands.
• Polly + idempotency + kö = robust skrivkedja över flera processer.

← Föregående: Transaktioner & concurrency Nästa: Snapshots & Sagas →