KurrentDB .NET Client

All events in KurrentDB must be wrapped in an EventData object. The core components are:

• eventId: A Uuid used to uniquely identify the event. If you attempt to append two events with the same Uuid to the same stream in quick succession, KurrentDB will only append the first one.
• type: A unique string identifying the event type. It is recommended to use a name that is not strictly tied to a specific code type to facilitate easier versioning later.
• data: The event payload as a byte array. While you can use any format, storing events as JSON objects is recommended to leverage KurrentDB features like projections.
• metadata (optional): A separate byte array for additional context (e.g., correlation IDs, timestamps).

Warning: Metadata must be a valid JSON object using string keys and string values only. Binary metadata is not supported in the current version.

var eventData = new EventData(
  Uuid.NewUuid(), // eventId
  "OrderPlaced",  // type
  "{\"orderId\": \"123\"}"u8.ToArray() // data
);

When using a catch-up subscription to process the $all stream, you should implement checkpointing to ensure your application can recover from crashes without reprocessing all events from the beginning.

A checkpoint is the position of an event in the $all stream that your application has successfully processed. To implement this, save the event's CommitPosition (or both CommitPosition and PreparePosition if using legacy TCP client transactions) to a persistent store like a database.

The KurrentDB .NET client provides a mechanism to notify your application at regular intervals so you can periodically save these checkpoints.

// Example of handling checkpoint notifications
await foreach (var message in subscription.Messages) {
  switch (message) {
    case StreamMessage.Event(var e):
      // Process the event
      Console.WriteLine($"{e.Event.EventType} @ {e.Event.Position.CommitPosition}");
      break;
    case StreamMessage.AllStreamCheckpointReached(var p):
      // Save commit position to a persistent store as a checkpoint
      Console.WriteLine($"checkpoint taken at {p.CommitPosition}");
      break;
  }
}

To implement optimistic concurrency, you can specify the expected state of a stream when appending. If the stream's actual state does not match your expectation, an exception is thrown.

Available StreamStates

• Any: No consistency check is performed.
• NoStream: The append succeeds only if the stream does not exist.
• StreamExists: The append succeeds only if the stream already exists.

Using Stream Revisions

You can also provide a specific stream revision (version number) to ensure you are appending to the version of the stream you last read. This prevents overwriting changes made by other clients in the interim.

// 1. Read the current state
var lastEvent = client
  .ReadStreamAsync(Direction.Forwards, "order-123", StreamPosition.Start)
  .LastAsync();

var orderPaid = new EventData(
  Uuid.NewUuid(), "OrderPaid", "{\"orderId\": \"123\"}"u8.ToArray()
);

// 2. Append using the version (revision) from the last event
await client.AppendToStreamAsync(
  "order-123",
  lastEvent.OriginalEventNumber.ToUInt64(),
  [orderPaid]
);

The .NET client generates traces for the following operations:

• append
• catch-up
• persistent subscription

Each trace includes metadata (attributes) to assist in debugging. Key attributes include:

You can start a subscription from a specific point in time rather than the beginning.

Important: Positions are exclusive. You will only receive events that occur after the provided position.

• Streams: Use FromStream.After(StreamPosition.FromInt64(revision)) to start after a specific 64-bit unsigned integer revision.
• $all: Use FromAll.After(logPosition) where logPosition is the LogPosition returned from an append operation.
• Live updates only: Use FromStream.End or FromAll.End to skip historical events and only receive new ones.

// Stream from specific position
await using var subscription = client.SubscribeToStream(
  "order-123",
  FromStream.After(StreamPosition.FromInt64(20))
);

// $all from specific position (using result from an Append operation)
await using var subscription = client.SubscribeToAll(
  FromAll.After(result.LogPosition)
);

// Live updates only
await using var subscription = client.SubscribeToStream("order-123", FromStream.End);

Once a group exists, use SubscribeToStream or SubscribeToAll to start receiving messages.

Important: Buffer Size The bufferSize parameter is critical. It defines how many outstanding (unacknowledged) messages the server allows for this client.

• Too small: The subscription stays idle waiting for ACKs.
• Too large: Wastes resources and may cause timeouts during processing.
• Default: 10

// Connecting to a specific stream
await using var subscription = client.SubscribeToStream(
  "order-123",
  "subscription-group",
  cancellationToken: ct
);

await foreach (var message in subscription.Messages) {
  switch (message) {
    case PersistentSubscriptionMessage.SubscriptionConfirmation(var subscriptionId):
      Console.WriteLine($"Subscription {subscriptionId} to stream started");
      break;
    case PersistentSubscriptionMessage.Event(var resolvedEvent, _):
      await HandleEvent(resolvedEvent);
      await subscription.Ack(resolvedEvent);
      break;
  }
}

Events are written to KurrentDB as byte arrays. You typically serialize an object (e.g., using System.Text.Json) and wrap it in an EventData structure.

To append an event to a stream, use AppendToStreamAsync. To ensure idempotent writes, you should specify the stream revision (e.g., using StreamState.Any).

Note: If you are using user-defined server-side projections, ensure your events are serialized in JSON format.

using System.Text.Json;

// 1. Define your event payload
public class OrderCreated {
    public string? OrderId { get; set; }
}

var evt = new OrderCreated {
    OrderId = Guid.NewGuid().ToString("N"),
};

// 2. Wrap in EventData
var orderCreated = new EventData(
    Uuid.NewUuid(),
    "OrderCreated",
    JsonSerializer.SerializeToUtf8Bytes(evt)
);

// 3. Append to a stream
await client.AppendToStreamAsync("order-123", StreamState.Any, [orderCreated]);