C# Event Sourcing Example With Tactical.DDD and Aperture
Have you ever wanted to quickly get an example EventSourcing project up and running maybe with a sprinkle of DDD Tactical patterns inside but didn’t know which libraries to choose or where to start? I have a couple of times and at some point decided to create a minimalistic set of .NET-focused packages to make the boilerplate stuff a bit easier and move it out of the way.
I am going to assume you know what EventSourcing is and have some “working knowledge” under your belt.
The example project is hosted here which you can run locally if you have at least .NET 7 SDK installed and a running Postgres instance. The project in itself is already really minimalistic but nevertheless, I’m still going to omit non-important details in this post and concentrate only on the library-specific things.
Nuget Packages
The packages we will be using are:
GitHub links:
To actually see which package is installed in which project please consult the solution itself.
Your Web API / Web App host
Steps 1, 3, and 4 are all you need to actually add the event-store implementation dependencies to your solution, just make sure you have an existing Postgres database up and running and provide the correct NpgSql connection string.
In step 2 we are simply adding repository implementation for our example Account aggregate. (You don’t need one)
// 1. Use Postgres event-store implementation
using Tactical.DDD.EventSourcing.Postgres;
// ...
// 2. Add our simple account repository implementation
builder.Services.AddScoped<IAccountRepo, AccountRepo>();
// 3. Add the postgrers event-store implementation
builder.Services.
AddPostgresEventStore("Host=localhost;Database=your-db;Username=your-user");
// ...
// 4. Ensure event-store is created via migration
app.EnsurePostgresEventStoreCreated();Your DDD Aggregate Root
In order to add DDD / EventSourcing-specific behavior to your aggregate root simply extend AggregateRoot from Tactical DDD.
I won’t go into detail here about what for example the Apply method does or what OnXXX method is since I assume you know your way around these patterns (I might go a bit deeper in a different post), but all in all we have a simplistic aggregate that can produce a single AccountProvisioned event.
using Tactical.DDD;
namespace Bank24.Core.AccountAggregate;
public class Account : AggregateRoot<AccountId>
{
public Account(IEnumerable<DomainEvent> events) : base(events)
{
}
private Account()
{
}
public static Account FromHolder(AccountId id, string holder)
{
var acc = new Account();
acc.Apply(new AccountProvisioned(DateTime.Now, holder, id));
return acc;
}
public void On(AccountProvisioned @event)
{
Id = AccountId.Parse(@event.AccountId);
}
}The domain event itself
In order to mark something as a domain event so the libraries can understand it as such it is enough to extend DomainEvent record.
using Tactical.DDD;
namespace Bank24.Core.AccountAggregate;
public record AccountProvisioned : DomainEvent
{
public string AccountId { get; private set; }
public string AccountHolder { get; private set; }
public AccountProvisioned(
DateTime CreatedAt,
string accountHolder,
string accountId
) : base(CreatedAt)
{
AccountHolder = accountHolder;
AccountId = accountId;
}
}The repository
The repository itself is simply a wrapper around the provided event-store implementation.
using Tactical.DDD.EventSourcing;
namespace Bank24.Core.AccountAggregate;
public class AccountRepo : IAccountRepo
{
private readonly IEventStore _eventStore;
public AccountRepo(IEventStore eventStore)
{
_eventStore = eventStore;
}
public async Task SaveAsync(Account account)
{
await _eventStore.SaveEventsAsync(
nameof(Account),
account.Id,
account.Version,
account.DomainEvents,
// eg. request id, user ip etc ...
new Dictionary<string, string>());
}
}The projections
We are running the projection subsystem as a separate executable, Worker in this example which is what you should usually be doing.
Creating the projections themselves is as simple as extending one of the provided projection base types and implementing IHandle<DomainEvent> interface for each event you want your projection to handle.
There are two projection base types used here: PostgresProjection and Projection
PostgresProjection
This base type is provided along with our Postgres event-store implementation and it handily provides out-of-the-box offset tracking for your projections, meaning when the projection subsystem restarts your projection will continue where it left off. I won’t go into specifics as to how this is implemented, but essentially an offset tracking table following aperture_offset_yourprojectionname naming convention will be automatically created within your database.
using Aperture.Core;
using Bank24.Core.AccountAggregate;
using Tactical.DDD.EventSourcing.Postgres.Aperture;
namespace Bank24.ProjectionWorker.Projections;
public class AllAccountsProjection : PostgresProjection, IHandle<AccountProvisioned>
{
public AllAccountsProjection(ITrackOffset offsetTracker) : base(offsetTracker)
{
}
public Task HandleAsync(AccountProvisioned @event)
{
Console.WriteLine($"{nameof(AllAccountsProjection)} -> New account provisioned for: {@event.AccountHolder}");
return Task.CompletedTask;
}
}Projection
This type is the default projection base type which simply applies all events it receives without offset tracking. You can extend this type to implement your own (this is how PostgresProjection is implemented also)
using Aperture.Core;
using Bank24.Core.AccountAggregate;
namespace Bank24.ProjectionWorker.Projections;
public class FooProjection : Projection, IHandle<AccountProvisioned>
{
public Task HandleAsync(AccountProvisioned @event)
{
Console.WriteLine($"{nameof(FooProjection)} -> New account provisioned for: {@event.AccountHolder}");
return Task.CompletedTask;
}
}Setting up the projection subsystem
It is as easy as registering your projections as implementations for IProjectEvents interface and using the provided AddPostgresAperture service collection extension by giving it the connection string to your event-store database and some optional configuration (there are sane defaults).
using Aperture.Core;
using Bank24.ProjectionWorker;
using Bank24.ProjectionWorker.Projections;
using Tactical.DDD.EventSourcing.Postgres.Aperture;
IHost host = Host.CreateDefaultBuilder(args)
.ConfigureServices(services =>
{
services.AddHostedService<Worker>();
services.AddSingleton<IProjectEvents, AllAccountsProjection>();
services.AddSingleton<IProjectEvents, FooProjection>();
services.AddPostgresAperture(
"Host=localhost;Database=your-db;Username=your-user",
new PullEventStream.Config
{
PullInterval = TimeSpan.FromMilliseconds(200),
BatchSize = 500
});
services.AddAperture();
})
.Build();
host.Run();The Worker
The worker simply runs the projection subsystem agent while applying some additional optional configuration for it (which you can also do from Program.cs)
using Aperture.Core;
using Aperture.Polly;
namespace Bank24.ProjectionWorker;
public class Worker : BackgroundService
{
private readonly ILogger<Worker> _logger;
private readonly ApertureAgent _apertureAgent;
public Worker(ILogger<Worker> logger, ApertureAgent apertureAgent)
{
_logger = logger;
_apertureAgent = apertureAgent;
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
await _apertureAgent
.UseRestartWithBackOffSupervision()
.UseLogger(_logger)
.UseCancellationToken(stoppingToken)
.StartAsync();
}
}I still need to update the respective libraries with proper documentation (especially Aperture — the projection subsystem) and explain all the different options available for configuring it.
I also plan to provide adapters for EventStore and SQLServer at least for both libraries.
If you are a beginner I am sorry for being a bit vague — I might remedy this in a follow-up post.
Happy EventSourcing !
