Snapshots & sagas — för långa strömmar och fleraggregatsflöden

När behövs snapshots?

Ett aggregat med 50 events laddas på under 10 ms. Ett aggregat med 50 000 events tar sekunder. När en ström blir så lång att rehydreringen syns i svarstiderna är det dags för en snapshot.

Snapshot-tabell i MSSQL

CREATE TABLE dbo.Snapshots (
    StreamId    NVARCHAR(200)  NOT NULL,
    Version     INT            NOT NULL,
    Type        NVARCHAR(400)  NOT NULL,
    PayloadJson NVARCHAR(MAX)  NOT NULL,
    CreatedAt   DATETIMEOFFSET NOT NULL,
    CONSTRAINT PK_Snapshots PRIMARY KEY (StreamId, Version)
);

CREATE INDEX IX_Snapshots_StreamId_VersionDesc
    ON dbo.Snapshots (StreamId, Version DESC);

Snapshot-stöd i aggregatet

public interface ISnapshotable<TSnapshot>
{
    TSnapshot CreateSnapshot();
    void RestoreFromSnapshot(TSnapshot snapshot);
}

public sealed record AccountSnapshot(
    Guid Id,
    string Currency,
    decimal Balance,
    AccountState State,
    int Version);

public partial class Account : ISnapshotable<AccountSnapshot>
{
    public AccountSnapshot CreateSnapshot() =>
        new(Id.Value, Balance.Currency, Balance.Amount, _state, Version);

    public void RestoreFromSnapshot(AccountSnapshot s)
    {
        Id = new AccountId(s.Id);
        Balance = new Money(s.Balance, s.Currency);
        _state = s.State;
        SetVersion(s.Version);   // exponerad protected setter på AggregateRoot
    }
}

Repository med snapshot-stöd

public async Task<Account?> LoadAsync(AccountId id, CancellationToken ct)
{
    var streamId = $"account-{id.Value}";

    // 1) Försök ladda senaste snapshot
    var snapshot = await _snapshotStore.LoadLatestAsync<AccountSnapshot>(streamId, ct);
    var account = new Account();
    var fromVersion = StreamVersion.Start;

    if (snapshot is not null)
    {
        account.RestoreFromSnapshot(snapshot.Payload);
        fromVersion = snapshot.Version + 1;     // läs bara events EFTER snapshot
    }

    // 2) Läs återstående events från strömmen
    var events = await _eventStore.LoadFromAsync(streamId, fromVersion, ct);
    if (snapshot is null && events.Count == 0) return null;

    account.LoadFromHistory(events.Select(_typeMap.Deserialize));
    return account;
}

public async Task SaveAsync(Account account, CancellationToken ct)
{
    await _eventStore.AppendAsync(/* ... */);

    // Snapshot var 100:e event
    if (account.Version % 100 == 0)
        await _snapshotStore.SaveAsync($"account-{account.Id.Value}",
                                        account.Version,
                                        account.CreateSnapshot(), ct);
}

Projection rebuild — bygg om read modeller från noll

När du lägger till en ny read model eller upptäcker en bug i en projektion vill du köra om hela historiken. Eftersom event store-strömmen är auktoritativ är detta alltid möjligt.

public async Task RebuildAsync(CancellationToken ct)
{
    // 1) Töm read model och checkpoint
    await using var conn = new SqlConnection(_cs);
    await conn.ExecuteAsync("DELETE FROM dbo.AccountSummary");
    await conn.ExecuteAsync("DELETE FROM dbo.ProjectionCheckpoints WHERE ProjectionName = 'AccountSummary'");

    // 2) Låt vanliga projection-loopen ta vid — den läser från position 0 igen
    _log.LogInformation("Rebuild scheduled. BackgroundService will catch up.");
}

Sagas — flöden över flera aggregat

Vissa affärsoperationer berör flera aggregat. Klassiskt exempel: "överför pengar från konto A till konto B" — två skrivningar, två strömmar, behöver vara en logisk enhet trots att vi inte har distribuerade transaktioner.

Lösningen: en saga (även kallad process manager) lyssnar på events och skickar nästa command när rätt steg är klart. Vid fel: skicka compensating commands i omvänd ordning.

Lyckad saga

// 1. Användare:        POST /transfers
// 2. TransferAggregate.Start  → TransferStarted (egen stream)
// 3. Saga reagerar:    SendCommand(WithdrawMoneyCommand, account A)
// 4. Account A:        Withdraw → MoneyWithdrawn
// 5. Saga reagerar:    SendCommand(DepositMoneyCommand, account B)
// 6. Account B:        Deposit → MoneyDeposited
// 7. Saga reagerar:    SendCommand(CompleteTransferCommand)
// 8. TransferAggregate.Complete → TransferCompleted

Misslyckad saga (deposit på B fail)

// 5'. Account B:       Deposit → kastar (kontot stängt)
// 6'. Saga reagerar:   SendCommand(RevertWithdrawalCommand, account A)
// 7'. Account A:       RevertWithdrawal → MoneyWithdrawalReverted
// 8'. Saga reagerar:   SendCommand(FailTransferCommand, reason)
// 9'. TransferAggregate.Fail → TransferFailed

Sagaimplementation

public sealed class MoneyTransferSaga : BackgroundService
{
    private readonly IStreamStore _store;
    private readonly AggregateCommandQueue _queue;
    private readonly IEventTypeMap _typeMap;
    private readonly string _cs;
    private const string Name = "MoneyTransferSaga";

    protected override async Task ExecuteAsync(CancellationToken ct)
    {
        while (!ct.IsCancellationRequested)
        {
            var fromPos = await ReadCheckpointAsync(ct);
            var page = await _store.ReadAllForwards(fromPos + 1, 100, ct);

            if (page.Messages.Length == 0)
            {
                await Task.Delay(500, ct);
                continue;
            }

            foreach (var m in page.Messages)
            {
                var clrType = _typeMap.TryResolve(m.Type);
                if (clrType is null) { await WriteCheckpointAsync(m.Position, ct); continue; }

                var json = await m.GetJsonData(ct);
                var evt = (IDomainEvent)JsonSerializer.Deserialize(json, clrType)!;

                switch (evt)
                {
                    case TransferStarted s:
                        await _queue.EnqueueAsync(new WithdrawMoneyCommand(
                            Guid.NewGuid(), s.FromAccount, s.Amount, s.Currency,
                            $"Transfer {s.TransferId}"), ct);
                        break;

                    case MoneyWithdrawn w when w.Reference.StartsWith("Transfer "):
                        var transferId = ExtractTransferId(w.Reference);
                        var toAccount = await LookupTargetAsync(transferId, ct);
                        await _queue.EnqueueAsync(new DepositMoneyCommand(
                            Guid.NewGuid(), toAccount, w.Amount, w.Currency,
                            $"Transfer {transferId}"), ct);
                        break;

                    case MoneyDeposited d when d.Reference.StartsWith("Transfer "):
                        var tid = ExtractTransferId(d.Reference);
                        await _queue.EnqueueAsync(new CompleteTransferCommand(
                            Guid.NewGuid(), tid), ct);
                        break;
                }

                await WriteCheckpointAsync(m.Position, ct);
            }
        }
    }
}

Projection-typer — en översikt

Sammanfattning

• Snapshots = cache av aggregattillstånd. Bygg in stöd från dag ett, aktivera när strömmar växer.
• Strömmen är sanning — snapshots och read models kan alltid byggas om.
• Sagas koordinerar fleraggregatsflöden via events + commands, med kompensationer vid fel.
• En saga är strukturellt en projektion som skickar commands i stället för att skriva till en tabell.
• Undvik sagas om operationen ryms i ett aggregat — det är enklare och atomäre.

← Föregående: Command-kö & Undo Nästa: Blazor-klient →