Interface IEventSource

Namespace

Stratara.Abstractions.EventSourcing

Assembly

Stratara.Abstractions.dll

Write-side façade for the event store. Command handlers append events through this service; the implementation tracks pending writes and flushes them on SaveChangesAsync(CancellationToken).

public interface IEventSource

Extension Methods

FieldChangedEventExtensions.ApplyPropertyChanged<TAggregate>(TAggregate, FieldChangedEvent<TAggregate>)

EventStream.ApplyEvents(object, IReadOnlyList<IEvent>)

Examples

Append events from a command handler and commit the unit-of-work:

public sealed class CreateOrderHandler(IEventSource events) : ICommandHandler<CreateOrder>
{
    public async Task HandleAsync(CreateOrder command, CancellationToken cancellationToken)
    {
        await events.CreateAsync<Order>(command.OrderId,
            new OrderCreated(command.OrderId, command.CustomerId, command.Amount),
            cancellationToken);
        await events.SaveChangesAsync(cancellationToken);
    }
}

Remarks

Every event is recorded under the tenant that owns it — its Subject — and that tenant is not simply the one in the caller's session. For each event the store takes the first of these that names a tenant: the Subject stated for that event with AppendOnBehalfOfAsync<TAggregate>(Guid, object, EventSubject, CancellationToken), or for a stream's first event with CreateOnBehalfOfAsync<TAggregate>(Guid, object, EventSubject, CancellationToken); the owner already resolved for the same stream earlier in the batch; the owner recorded on the stream's first event — its tenant, and its user where one was recorded; the TenantId of a creation event; and only then the tenant in the session. If none of them names a tenant, the append fails.

A stream therefore keeps the owner it was created with, user included, whatever session appends to it later; a stream whose first event names no user is given none. A first event that states no owner takes the tenant and the data-owner user in the session, so an aggregate created for another tenant states that tenant on its first event by implementing IAggregateCreationEvent, or is created with CreateOnBehalfOfAsync<TAggregate>(Guid, object, EventSubject, CancellationToken). A creation event names no user, so a stream it creates has none. A host can be warned of, or refuse, a first event that takes its owner from the session, with the EventSourcing:NewStreamOwnerFromSession option.

Methods

AppendAsync<TAggregate>(Guid, object, CancellationToken)

Append an event to an existing stream.

Task AppendAsync<TAggregate>(Guid streamId, object @event, CancellationToken cancellationToken = default) where TAggregate : notnull, new()

Parameters

streamId Guid

The stream id.

event object

The event payload.

cancellationToken CancellationToken

Propagated to the write-store transaction.

Returns

Task

Type Parameters

TAggregate

The aggregate type.

Remarks

The event takes the owner recorded on the stream, not the tenant in the session. On a stream that does not exist yet, the owner is resolved as for the first event of CreateAsync<TAggregate>(Guid, object, CancellationToken), including what the host's EventSourcing:NewStreamOwnerFromSession decides. Use AppendOnBehalfOfAsync<TAggregate>(Guid, object, EventSubject, CancellationToken) for an event whose owner differs from the stream's.

Exceptions

InvalidOperationException

No candidate names an owner, or the stream does not exist yet, its first event would take its owner from the session, and the host refuses that (EventSourcing:NewStreamOwnerFromSession = Refuse).

SessionRequiredException

No session context is set on the current scope.

AppendAtVersionAsync<TAggregate>(Guid, long, object, CancellationToken)

Append an event on the condition that the stream is still at expectedVersion, the version the caller decided on.

Task AppendAtVersionAsync<TAggregate>(Guid streamId, long expectedVersion, object @event, CancellationToken cancellationToken = default) where TAggregate : notnull, new()

Parameters

streamId Guid

The stream id.

expectedVersion long

The version the caller read; 0 for a stream that is expected not to exist yet.

event object

The event payload.

cancellationToken CancellationToken

Propagated to the write-store transaction.

Returns

Task

A task that completes once the event is staged.

Type Parameters

TAggregate

The aggregate type.

Remarks

The event is numbered expectedVersion + 1. If another writer has moved the stream past expectedVersion by the time of the save, SaveChangesAsync(CancellationToken) throws ConcurrencyException and records nothing of the batch, so a fact decided on the stream's earlier state is never recorded after one that contradicts it. Without the condition, AppendAsync<TAggregate>(Guid, object, CancellationToken) numbers the event after whatever the stream holds when the append is made.

A caller without a version of its own reads one with GetCurrentVersionAsync(Guid, CancellationToken) before it reads the state it decides on; a write that lands in between then surfaces as a conflict rather than going unnoticed. A host can apply the same condition to every append that follows a rebuild through IAggregationService with the EventSourcing:AppendAgainstAggregatedVersion option. The event takes the stream's owner, as in AppendAsync<TAggregate>(Guid, object, CancellationToken).

Exceptions

ArgumentOutOfRangeException

expectedVersion is negative, or above the stream's current version — a version the stream has not reached, which would leave a gap in its versions.

InvalidOperationException

Events for the stream are already staged in this batch, and expectedVersion is not the version they end at; or the owner cannot be resolved, as for AppendAsync<TAggregate>(Guid, object, CancellationToken).

NotSupportedException

The implementation does not support conditional appends.

SessionRequiredException

No session context is set on the current scope.

AppendOnBehalfOfAsync<TAggregate>(Guid, object, EventSubject, CancellationToken)

Append an event with an explicit Subject (data owner), overriding every other source — the stream's recorded owner, a creation event's tenant and the session — for this one event.

Task AppendOnBehalfOfAsync<TAggregate>(Guid streamId, object @event, EventSubject subject, CancellationToken cancellationToken = default) where TAggregate : notnull, new()

Parameters

streamId Guid

The stream id.

event object

The event payload.

subject EventSubject

The data owner to record. It names a tenant, or a user alone (ForUser(Guid)): stating the Subject also states that the stream, the event and the session are not to be consulted, so one that names neither fails the append rather than falling back to them.

cancellationToken CancellationToken

Propagated to the write-store transaction.

Returns

Task

Type Parameters

TAggregate

The aggregate type.

Remarks

Use this method for an event whose owner differs from the stream's. The actor recorded with the event stays the caller's session; only the owner changes, and only for this event — the next append to the stream takes the stream's owner again, in the same batch as in a later one. On a stream's first event the stated owner is the one the stream records.

The event's protected fields are encrypted under the stated owner's keys, so an erasure of the stream's owner does not reach them, and an erasure of the stated owner can leave the stream unable to rehydrate.

Exceptions

ArgumentException

subject names neither a tenant nor a user.

SessionRequiredException

No session context is set on the current scope.

AppendRangeAsync<TAggregate>(Guid, IEnumerable<object>, CancellationToken)

Append multiple events to an existing stream in order.

Task AppendRangeAsync<TAggregate>(Guid streamId, IEnumerable<object> events, CancellationToken cancellationToken = default) where TAggregate : notnull, new()

Parameters

streamId Guid

The stream id.

events IEnumerable<object>

The events to append, in order.

cancellationToken CancellationToken

Propagated to the write-store transaction.

Returns

Task

Type Parameters

TAggregate

The aggregate type.

Remarks

The events take the owner recorded on the stream, as in AppendAsync<TAggregate>(Guid, object, CancellationToken).

Exceptions

InvalidOperationException

No candidate names an owner, or the stream does not exist yet, its first event would take its owner from the session, and the host refuses that (EventSourcing:NewStreamOwnerFromSession = Refuse).

SessionRequiredException

No session context is set on the current scope.

AppendRangeAtVersionAsync<TAggregate>(Guid, long, IEnumerable<object>, CancellationToken)

Append multiple events in order, on the condition that the stream is still at expectedVersion.

Task AppendRangeAtVersionAsync<TAggregate>(Guid streamId, long expectedVersion, IEnumerable<object> events, CancellationToken cancellationToken = default) where TAggregate : notnull, new()

Parameters

streamId Guid

The stream id.

expectedVersion long

The version the caller read; 0 for a stream that is expected not to exist yet.

events IEnumerable<object>

The events to append, in order.

cancellationToken CancellationToken

Propagated to the write-store transaction.

Returns

Task

A task that completes once the events are staged.

Type Parameters

TAggregate

The aggregate type.

Remarks

The events are numbered from expectedVersion + 1 onward, and the condition is the one AppendAtVersionAsync<TAggregate>(Guid, long, object, CancellationToken) describes.

Exceptions

ArgumentOutOfRangeException

expectedVersion is negative, or above the stream's current version.

InvalidOperationException

Events for the stream are already staged in this batch, and expectedVersion is not the version they end at; or the owner cannot be resolved, as for AppendAsync<TAggregate>(Guid, object, CancellationToken).

NotSupportedException

The implementation does not support conditional appends.

SessionRequiredException

No session context is set on the current scope.

CreateAsync<TAggregate>(Guid, object, CancellationToken)

Create a new stream with the first event. Fails if the stream already exists or already has events staged in this batch.

Task CreateAsync<TAggregate>(Guid streamId, object @event, CancellationToken cancellationToken = default) where TAggregate : notnull, new()

Parameters

streamId Guid

The stream id.

event object

The creation event. Implement IAggregateCreationEvent on it to state the new stream's owner instead of taking the tenant in the session.

cancellationToken CancellationToken

Propagated to the write-store transaction.

Returns

Task

Type Parameters

TAggregate

The aggregate type the stream represents.

Remarks

The new stream's owner is the tenant the event carries when it is an IAggregateCreationEvent with a non-empty tenant, otherwise the tenant in the session — unless the host's EventSourcing:NewStreamOwnerFromSession warns of or refuses that. Every later event on the stream keeps that owner.

Exceptions

InvalidOperationException

The stream already exists or already has events staged in this batch; no candidate names an owner; or the first event would take its owner from the session and the host refuses that (EventSourcing:NewStreamOwnerFromSession = Refuse).

SessionRequiredException

No session context is set on the current scope.

CreateOnBehalfOfAsync<TAggregate>(Guid, object, EventSubject, CancellationToken)

Create a new stream with its first event, recorded for the owner subject states; every later event on the stream for which no owner is stated keeps it. Fails if the stream already exists.

Task CreateOnBehalfOfAsync<TAggregate>(Guid streamId, object @event, EventSubject subject, CancellationToken cancellationToken = default) where TAggregate : notnull, new()

Parameters

streamId Guid

The stream id.

event object

The creation event.

subject EventSubject

The owner to record. It names a tenant, or — through ForUser(Guid) — a user alone, for data that belongs to a person rather than to any of their tenants.

cancellationToken CancellationToken

Propagated to the write-store transaction.

Returns

Task

A task that completes once the event is staged.

Type Parameters

TAggregate

The aggregate type the stream represents.

Remarks

The explicit counterpart of CreateAsync<TAggregate>(Guid, object, CancellationToken) for an owner the event does not carry: a record created by an operator for another tenant, or a platform-owned record created under any session. The actor recorded with the event stays the caller's session. The default implementation checks ExistsAsync(Guid, CancellationToken) and appends through AppendOnBehalfOfAsync<TAggregate>(Guid, object, EventSubject, CancellationToken); it cannot see a stream staged in the same batch but not yet saved.

Exceptions

ArgumentException

subject names neither a tenant nor a user.

InvalidOperationException

The stream already exists.

SessionRequiredException

No session context is set on the current scope.

CreateRangeAsync<TAggregate>(Guid, IEnumerable<object>, CancellationToken)

Create a new stream with multiple events in order.

Task CreateRangeAsync<TAggregate>(Guid streamId, IEnumerable<object> events, CancellationToken cancellationToken = default) where TAggregate : notnull, new()

Parameters

streamId Guid

The stream id.

events IEnumerable<object>

The events to append, in order.

cancellationToken CancellationToken

Propagated to the write-store transaction.

Returns

Task

Type Parameters

TAggregate

The aggregate type the stream represents.

Remarks

The owner is resolved from the first event as in CreateAsync<TAggregate>(Guid, object, CancellationToken), and the events after it take the same owner.

Exceptions

InvalidOperationException

The stream already exists or already has events staged in this batch; no candidate names an owner; or the first event would take its owner from the session and the host refuses that (EventSourcing:NewStreamOwnerFromSession = Refuse).

SessionRequiredException

No session context is set on the current scope.

ExistsAsync(Guid, CancellationToken)

Returns true if the stream exists in the event store.

Task<bool> ExistsAsync(Guid streamId, CancellationToken cancellationToken = default)

Parameters

streamId Guid

cancellationToken CancellationToken

Returns

Task<bool>

GetCurrentVersionAsync(Guid, CancellationToken)

Returns the head version of the stream, or 0 if it does not exist.

Task<long> GetCurrentVersionAsync(Guid streamId, CancellationToken cancellationToken = default)

Parameters

streamId Guid

cancellationToken CancellationToken

Returns

Task<long>

SaveChangesAsync(CancellationToken)

Flush every pending append/create to the underlying write store. Throws ConcurrencyException if another writer committed first.

Task SaveChangesAsync(CancellationToken cancellationToken = default)

Parameters

cancellationToken CancellationToken

Honoured while the changes are written; a commit once begun runs to its end on the framework's stores, so a save the store committed is never reported as cancelled. A handover after the commit that the cancellation stops before the bundle is published or recorded surfaces as CommittedEventsNotPublishedException.

Returns

Task

Remarks

A save clears what was staged whether it succeeds or fails. After a failure, append the events again before saving again, as after a ConcurrencyException: a save with nothing staged writes and publishes nothing. The one exception is CommittedEventsNotPublishedException: its events are already recorded, and appending them again would record them twice.

Exceptions

OperationCanceledException

The save was cancelled before it committed; no events were written.

ConcurrencyException

Another writer beat this one to the stream's head version.

CommittedEventsNotPublishedException

The events were committed, but handing their bundle on failed afterwards. Do not append them again.

SessionRequiredException

No session context is set on the current scope.