EventStore.Tools.PositionRepository.Gprc 1.4.6

Position Repository for KurrentDB

A lightweight library for persisting and retrieving the last processed position in a KurrentDB (formerly EventStoreDB) stream. Useful when you need to resume a subscription from where you left off after a restart.

The position is saved to a dedicated KurrentDB stream (capped at 1 event). To avoid excessive writes during fast processing, saves are batched on a configurable interval (default: 1 second).

Packages

Installation

gRPC package (recommended)

PM> Install-Package EventStore.Tools.PositionRepository.Gprc

dotnet add package EventStore.Tools.PositionRepository.Gprc

Legacy TCP package

PM> Install-Package EventStore.Tools.PositionRepository

dotnet add package EventStore.Tools.PositionRepository

Usage

Create the repository

var client = new KurrentDBClient(KurrentDBClientSettings.Create("esdb://localhost:2113?tls=false"));

var positionRepo = new PositionRepository(
 positionStreamName: "my-subscription-position",
 positionEventType: "PositionStored",
 client: client);

Save the current position

positionRepo.Set(resolvedEvent.OriginalPosition.Value);

Resume from the last saved position

if (positionRepo.TryGet(out var position))
{
 await client.SubscribeToAllAsync(position, eventAppeared);
}

Optional: disable the timer and save on every event

// Pass interval: 0 to save immediately on every Set() call
var positionRepo = new PositionRepository("my-position", "PositionStored", client, interval: 0);

Building Projections with GrpcSyncService

GrpcSyncService is a ready-made subscription host included in the gRPC package. It manages the KurrentDB catch-up subscription loop, handles reconnections with backoff, and persists the position automatically. You implement only the business logic for the events you care about.

How it works

• Subscribes to the $all stream from the last saved position
• Skips system events ($-prefixed), Position*, and CloudEvent* event types automatically
• Dispatches each recognised event type to its registered handler (data, metadata both as JsonNode)
• Saves the position after every handled event; also saves checkpoint positions between events
• Fires CatchedUpEvent once the subscription has replayed all historical events and gone live
• Reconnects automatically if the subscription drops

Pattern 1 — Composition (recommended)

Register one ISyncEventHandler per event type and let GrpcSyncService dispatch to them. This is the simplest path and works well with dependency injection.

// 1. Implement a handler for each event type you care about
public class OrderPlacedHandler : ISyncEventHandler
{
 public string EventType => "OrderPlacedV1";

 public void Handle(JsonNode data, JsonNode metadata)
 {
 var orderId = data["orderId"]?.GetValue<string>();
 // update your read model / projection
 }
}

// 2. Register everything in DI (e.g. in Program.cs or a Startup class)
services.AddSingleton<KurrentDBClient>(_ =>
 new KurrentDBClient(KurrentDBClientSettings.Create("esdb://localhost:2113?tls=false")));

services.AddSingleton<IPositionRepository>(sp =>
 new PositionRepository("orders-projection-position", "PositionStored",
 sp.GetRequiredService<KurrentDBClient>()));

services.AddSingleton<ISyncEventHandler, OrderPlacedHandler>();
services.AddSingleton<ISyncEventHandler, OrderCancelledHandler>();

services.AddSingleton<ISyncService>(sp =>
{
 var posRepo = sp.GetRequiredService<IPositionRepository>();
 var client = sp.GetRequiredService<KurrentDBClient>();
 var handlers = sp.GetServices<ISyncEventHandler>();
 var logger = sp.GetRequiredService<ILogger<GrpcSyncService>>();

 var catchUpFrom = posRepo.TryGet(out var saved)
 ? FromAll.After(saved)
 : FromAll.Start;

 return new GrpcSyncService(posRepo, client, catchUpFrom, handlers, logger);
});

// 3. Start/stop via the ISyncService interface
var syncService = sp.GetRequiredService<ISyncService>();
await syncService.StartAsync(cancellationToken);

// React when catch-up is complete (all historical events replayed)
syncService.CatchedUpEvent += (_, _) =>
{
 Console.WriteLine("Projection is live.");
 return Task.CompletedTask;
};

Pattern 2 — Inheritance

Subclass GrpcSyncService when you need cross-cutting concerns applied to every event before dispatch — for example tenant filtering, custom logging context, or bridging to a different serialiser.

public class MyProjectionService : GrpcSyncService
{
 public MyProjectionService(
 IPositionRepository positionRepository,
 KurrentDBClient conn,
 FromAll catchUpFrom,
 IEnumerable<ISyncEventHandler> handlers,
 ILogger<MyProjectionService> logger)
 : base(positionRepository, conn, catchUpFrom, handlers, logger) { }

 // Override to intercept the raw ResolvedEvent before any dispatch
 protected override Task EventAppeared(ResolvedEvent evt)
 {
 // add context, filter, transform — then call base to dispatch normally
 using var scope = _logger.BeginScope(new { evt.Event.EventType });
 return base.EventAppeared(evt);
 }

 // Override to intercept after parsing but just before the handler is called
 protected override void ProcessEventAppeared(string eventType, JsonNode data, JsonNode metadata)
 {
 // e.g. enrich data, log, or call a different dispatch mechanism
 base.ProcessEventAppeared(eventType, data, metadata);
 }
}

Running as an IHostedService

GrpcSyncService implements ISyncService which maps directly onto the IHostedService lifecycle. Wire it up as a hosted service:

public class ProjectionWorker : BackgroundService
{
 private readonly ISyncService _syncService;

 public ProjectionWorker(ISyncService syncService) => _syncService = syncService;

 protected override Task ExecuteAsync(CancellationToken stoppingToken)
 => _syncService.StartAsync(stoppingToken);

 public override Task StopAsync(CancellationToken cancellationToken)
 => _syncService.StopAsync(cancellationToken);
}

// Register in Program.cs
services.AddHostedService<ProjectionWorker>();

Cutting a Release

Releases are published to NuGet via GitHub Actions and triggered manually.

Prerequisites

• A NUGET_API_KEY secret must be set in the repository under Settings → Secrets and variables → Actions

Publish EventStore.Tools.PositionRepository.Gprc

1. Go to Actions → Publish EventStore.Tools.PositionRepository.Gprc → Run workflow
2. Leave the version field empty to auto-bump the patch (e.g. 1.4.4 → 1.4.5), or enter a specific version (e.g. 1.5.0) for a minor/major bump
3. Click Run workflow

The workflow will:

• Update the version in the .csproj
• Build and pack the NuGet package
• Push it to NuGet.org
• Commit the version bump and create a gprc-<version> tag

Publish EventStore.Tools.PositionRepository (legacy)

Same steps as above but use Actions → Publish EventStore.Tools.PositionRepository.
A legacy-<version> tag is created on completion.