Observing Append Operations

IEventSequence exposes an AppendOperations observable that emits after every Append or AppendMany call on that sequence. Any code can subscribe to this observable to react to appended events in real time — without polling the event log.

AppendOperations

C#

using Cratis.Chronicle.EventSequences;

public interface IObservingAppendsEventSequence
{
    IObservable<IEnumerable<AppendedEventWithResult>> AppendOperations { get; }
}

TypeScript

import { AppendedEventWithResult } from '@cratis/chronicle';

// eventLog.appendOperations is an AsyncIterable<AppendedEventWithResult[]> - a hot,
// multicast stream: iterating it yields every batch of events appended through this
// specific event log instance from the moment you start iterating (never for
// transactional appends, and never replayed for a late subscriber).
type ObservingAppendsShape = AsyncIterable<AppendedEventWithResult[]>;

The observable emits a collection of AppendedEventWithResult after each operation:

• A single-event Append emits a collection containing one element.
• A batch AppendMany emits the full batch as one collection.

Each AppendedEventWithResult pairs the appended event with its result:

Member	Type	Description
Event	AppendedEvent	The appended event, including context and deserialized content
Event.Content	object	The deserialized event object
Event.Context	EventContext	Metadata: event source, sequence number, correlation ID, causation chain
Result	AppendResult	Success flag, sequence number, violations, and errors

Subscribers receive the notification after the operation has completed, whether it succeeded or failed. This observable does not fire for transactional appends through ITransactionalEventSequence.

Subscribing

Inject IEventSequence (or IEventLog) and subscribe to AppendOperations:

C#

using Cratis.Chronicle.EventSequences;

public class ObservingAppendsMonitor(IEventLog eventLog) : IDisposable
{
    readonly IDisposable _subscription = eventLog.AppendOperations.Subscribe(OnAppended);

    static void OnAppended(IEnumerable<AppendedEventWithResult> operations)
    {
        foreach (var item in operations)
        {
            Console.WriteLine($"Event {item.Event.Content.GetType().Name} appended: success={item.Result.IsSuccess}");
        }
    }

    public void Dispose() => _subscription.Dispose();
}

TypeScript

import { IEventLog } from '@cratis/chronicle';

async function monitorAppends(eventLog: IEventLog): Promise<void> {
    for await (const operations of eventLog.appendOperations) {
        for (const item of operations) {
            console.log(`Event ${item.event.eventType.id.value} appended: success=${item.result.isSuccess}`);
        }
    }
}

Always dispose the subscription when you no longer need it to avoid resource leaks.

Integration Testing

The most common use of AppendOperations in application code is through the IEventAppendCollection helper provided by Cratis.Chronicle.XUnit.Integration. It subscribes internally and provides a ready-to-assert collection of AppendedEventWithResult entries. See Chronicle.Testing.EventAppendCollection for full details.

Waiting for Observer Completion After Append

When you need to wait until observers affected by an append operation have completed, use WaitForCompletion() from the Cratis.Chronicle.Observation namespace.

If any affected observer fails while catching up, the returned result includes failed partitions.

Append

C#

using Cratis.Chronicle.Events;
using Cratis.Chronicle.EventSequences;
using Cratis.Chronicle.Observation;

[EventType]
public record ObservingAppendsSomeEvent(string Data);

public class ObservingAppendsCompletionWaiter(IEventLog eventLog)
{
    public async Task AppendAndWait(EventSourceId eventSourceId)
    {
        var appendResult = await eventLog.Append(eventSourceId, new ObservingAppendsSomeEvent("example"));
        var completion = await appendResult.WaitForCompletion();

        if (!completion.IsSuccess)
        {
            foreach (var failedPartition in completion.FailedPartitions)
            {
                Console.WriteLine($"Observer {failedPartition.ObserverId} failed partition {failedPartition.Partition}");
            }
        }
    }
}

TypeScript

import { eventType, IEventLog } from '@cratis/chronicle';

@eventType()
class ObservingAppendsSomeEvent {
    constructor(readonly data: string = '') {}
}

async function appendAndWait(eventLog: IEventLog, eventSourceId: string): Promise<void> {
    const appendResult = await eventLog.append(eventSourceId, new ObservingAppendsSomeEvent('example'));
    const completion = await appendResult.waitForCompletion();

    if (!completion.isSuccess) {
        for (const failedPartition of completion.failedPartitions) {
            console.log(`Observer ${failedPartition.observerId} failed partition ${failedPartition.partition}`);
        }
    }
}

AppendMany

C#

using Cratis.Chronicle.Events;
using Cratis.Chronicle.EventSequences;
using Cratis.Chronicle.Observation;

[EventType]
public record ObservingAppendsFirstEvent(string Data);

[EventType]
public record ObservingAppendsSecondEvent(string Data);

public class ObservingAppendsBatchCompletionWaiter(IEventLog eventLog)
{
    public async Task AppendManyAndWait(EventSourceId eventSourceId)
    {
        var appendManyResult = await eventLog.AppendMany(eventSourceId, new object[]
        {
            new ObservingAppendsFirstEvent("first"),
            new ObservingAppendsSecondEvent("second")
        });

        var completion = await appendManyResult.WaitForCompletion();
        if (!completion.IsSuccess)
        {
            // Inspect failed partitions from affected observers
        }
    }
}

TypeScript

import { eventType, IEventLog } from '@cratis/chronicle';

@eventType()
class ObservingAppendsFirstEvent {
    constructor(readonly data: string = '') {}
}

@eventType()
class ObservingAppendsSecondEvent {
    constructor(readonly data: string = '') {}
}

async function appendManyAndWait(eventLog: IEventLog, eventSourceId: string): Promise<void> {
    const appendManyResults = await eventLog.appendMany(eventSourceId, [
        new ObservingAppendsFirstEvent('first'),
        new ObservingAppendsSecondEvent('second')
    ]);

    for (const appendResult of appendManyResults) {
        const completion = await appendResult.waitForCompletion();
        if (!completion.isSuccess) {
            // Inspect completion.failedPartitions from affected observers.
        }
    }
}