CQRS and Event Sourcing with Polecat and SQL Server

Uncategorized 5 Minutes

If you’re already familiar with Marten and Wolverine, this is all old news except for the part where we’re using SQL Server. If you’re brand new to the “Critter Stack,” Event Sourcing, or CQRS, hang around! And just so you know, JasperFx Software is completely ready to support our clients using Polecat.

All of the sample code in this blog post can be found in the Wolverine codebase on GitHub here.

With the advent of Polecat going 1.0 last week, you now have a robust solution for Event Sourcing using SQL Server 2025 as the backing store. If you’re reading this, you’re surely involved in software development and that means that your job at some point has been dictated by some kind of issue tracking tool, so let’s use that as our example system and pretend we’re creating an incident tracking system for our help desk folks as shown below:

To get started, I’m a fan of using the Event Storming technique to identify some of the meaningful events we should capture in our system and start to identify possible commands within our system:

Having at least some initial thoughts about the shape of our system, let’s start a new web service project in .NET with:

dotnet new webapi

Then add both Polecat (for persistence) and Wolverine (for both HTTP endpoints and asynchronous messaging) with:

dotnet add package WolverineFx.Polecat
dotnet add package WolverineFx.Http

And now, let’s jump into our Program file to wire up Polecat to an existing SQL Server database and configure Wolverine as well:

using Polecat;
using Polecat.Projections;
using PolecatIncidentService;
using Wolverine;
using Wolverine.Http;
using Wolverine.Polecat;
var builder = WebApplication.CreateBuilder(args);
builder.Services.AddOpenApi();
builder.Services.AddPolecat(opts =>
    {
        var connectionString = builder.Configuration.GetConnectionString("SqlServer")
                               ??
                               "Server=localhost,1434;User Id=sa;Password=P@55w0rd;Timeout=5;MultipleActiveResultSets=True;Initial Catalog=master;Encrypt=False";
        opts.ConnectionString = connectionString;
        opts.DatabaseSchemaName = "incidents";
        // We'll talk about this soon...
        opts.Projections.Snapshot<Incident>(SnapshotLifecycle.Inline);
    })
// For Marten users, *this* is the default for Polecat!
//.UseLightweightSessions()
    .IntegrateWithWolverine(x => x.UseWolverineManagedEventSubscriptionDistribution = true);
builder.Host.UseWolverine(opts => { opts.Policies.AutoApplyTransactions(); });
builder.Services.AddWolverineHttp();
var app = builder.Build();
if (app.Environment.IsDevelopment())
{
    app.MapOpenApi();
}
// Adding Wolverine.HTTP
app.MapWolverineEndpoints();
// This gets you a lot of CLI goodness from the 
// greater JasperFx / Critter Stack ecosystem
// and will soon feed quite a bit of AI assisted development as well
return await app.RunJasperFxCommands(args);
// For test bootstrapping in case you want to work w/
// more than one system at a time
public partial class Program
{
}

Our events are just going to be some immutable records like this:

public record LogIncident(
    Guid CustomerId,
    Contact Contact,
    string Description,
    Guid LoggedBy
);
public record CategoriseIncident(
    IncidentCategory Category,
    Guid CategorisedBy,
    int Version
);
public record CloseIncident(
    Guid ClosedBy,
    int Version
);

It’s not mandatory to use immutable types, but you might as well and it’s just idiomatic.

Let’s start with our LogIncident use case and build out an HTTP endpoint that creates a new “event stream” for events related to a single, logical Incident:

public static class LogIncidentEndpoint
{
    [WolverinePost("/api/incidents")]
    public static (CreationResponse<Guid>, IStartStream) Post(LogIncident command)
    {
        var (customerId, contact, description, loggedBy) = command;
        var logged = new IncidentLogged(customerId, contact, description, loggedBy);
        var start = PolecatOps.StartStream<Incident>(logged);
        var response = new CreationResponse<Guid>("/api/incidents/" + start.StreamId, start.StreamId);
        return (response, start);
    }
}

Polecat does support “Dynamic Consistency Boundary” event sourcing as well, but that’s not where I think most people should start, and I’ll get to that in a later post I keep putting off…

With some help from Alba, another JasperFx supported library, we can write both unit tests for the business logic (such as it is) and do an end to end test through the HTTP endpoint like this:

public class when_logging_an_incident : IntegrationContext
{
    public when_logging_an_incident(AppFixture fixture) : base(fixture)
    {
    }
    [Fact]
    public void unit_test()
    {
        var contact = new Contact(ContactChannel.Email);
        var command = new LogIncident(Guid.NewGuid(), contact, "It's broken", Guid.NewGuid());
        // Pure function FTW!
        var (response, startStream) = LogIncidentEndpoint.Post(command);
        // Should only have the one event
        startStream.Events.ShouldBe([
            new IncidentLogged(command.CustomerId, command.Contact, command.Description, command.LoggedBy)
        ]);
    }
    [Fact]
    public async Task happy_path_end_to_end()
    {
        var contact = new Contact(ContactChannel.Email);
        var command = new LogIncident(Guid.NewGuid(), contact, "It's broken", Guid.NewGuid());
        // Log a new incident first
        var initial = await Scenario(x =>
        {
            x.Post.Json(command).ToUrl("/api/incidents");
            x.StatusCodeShouldBe(201);
        });
        // Read the response body by deserialization
        var response = initial.ReadAsJson<CreationResponse<Guid>>();
        // Reaching into Polecat to build the current state of the new Incident
        await using var session = Store.LightweightSession();
        var incident = await session.Events.FetchLatest<Incident>(response.Value);
        incident!.Status.ShouldBe(IncidentStatus.Pending);
    }
}

Now, to build out a command handler for potentially categorizing an event, we’ll need to:

  1. Know the current state of the logical Incident by rolling up the events into some kind of representation of the state so that we can “decide” which if any events should be appended at this time. In Event Sourcing terms, I’d refer to this as the “write model.”
  2. The command type itself
  3. Validation logic for the input
  4. Like I said earlier, decide which events should be published
  5. Do some metadata correlation for observability. It’s not obvious from the code, but in the sample below Wolverine & Marten are tracking the events captured against the correlation id of the current HTTP request
  6. Establish transactional boundaries, including any outbound messaging that might be taking place in response to the events that are being appended. This is something that Wolverine does for Polecat (and Marten) in command handlers. This includes the transactional outbox support in Wolverine.
  7. Create protections against concurrent writes to any given Incident stream, which Wolverine and Polecat do for you in the next endpoint by applying optimistic concurrency checks to guarantee that no other thread changed the Incident since this CategoriseIncident command was issued by the caller

That’s actually quite a bit of responsibility for the command handler, but not to worry, Wolverine and Polecat are going to keep your code nice and simple. Hopefully even a pure function “Decider” for the business logic in many cases. Before I get into the command handler, here’s what the “projection” that gives us the current state of the Incident by applying events:

public class Incident
{
    public Guid Id { get; set; }
    // Polecat will set this itself for optimistic concurrency
    public int Version { get; set; }
    public IncidentStatus Status { get; set; } = IncidentStatus.Pending;
    public IncidentCategory? Category { get; set; }
    public bool HasOutstandingResponseToCustomer { get; set; } = false;
    public Incident()
    {
    }
    public void Apply(IncidentLogged _) { }
    public void Apply(IncidentCategorised e) => Category = e.Category;
    public void Apply(AgentRespondedToIncident _) => HasOutstandingResponseToCustomer = false;
    public void Apply(CustomerRespondedToIncident _) => HasOutstandingResponseToCustomer = true;
    public void Apply(IncidentResolved _) => Status = IncidentStatus.Resolved;
    public void Apply(ResolutionAcknowledgedByCustomer _) => Status = IncidentStatus.ResolutionAcknowledgedByCustomer;
    public void Apply(IncidentClosed _) => Status = IncidentStatus.Closed;
    public bool ShouldDelete(Archived @event) => true;
}

And finally, the command handler:

public record CategoriseIncident(
    IncidentCategory Category,
    Guid CategorisedBy,
    int Version
);
public static class CategoriseIncidentEndpoint
{
    public static ProblemDetails Validate(Incident incident)
    {
        return incident.Status == IncidentStatus.Closed
            ? new ProblemDetails { Detail = "Incident is already closed" }
            : WolverineContinue.NoProblems;
    }
    [EmptyResponse]
    [WolverinePost("/api/incidents/{incidentId:guid}/category")]
    public static IncidentCategorised Post(
        CategoriseIncident command,
        [Aggregate("incidentId")] Incident incident)
    {
        return new IncidentCategorised(incident.Id, command.Category, command.CategorisedBy);
    }
}

And I admit that that’s a lot of code thrown at you all at once, and maybe even a lot of new concepts. For further reading, see:

Unknown's avatar

Published by jeremydmiller

Published

Leave a comment