Winche.Events.WebSocket 7.0.0

Winche.Events

A Marten-backed event sourcing library for .NET 10. Provides a command-dispatch layer with typed handlers and optional real-time transport over WebSocket — domain code depends only on Winche.Events.Commands; Marten is wired up in the host.

Packages

Getting started

1. Define your domain

Aggregates, events, and commands are all plain records — no base classes required.

// Aggregate state
record Order(string Id, string Status, string[] Items)
{
    public static Order Empty => new("", "Pending", []);
}

// Events
record OrderPlaced(string[] Items);
record OrderConfirmed();
record OrderShipped();
record OrderCancelled();

// Commands — plain records, no base class
record PlaceOrder(string[] Items);
record ConfirmOrder();
record ShipOrder();
record CancelOrder();

2. Define a projection

Use Marten's SingleStreamProjection<TDoc, string> directly. Set opts.Schema.For<T>().Identity(x => x.Id) to make the identity explicit.

using JasperFx.Events.Projections;
using Marten.Events.Aggregation;
using JasperFxEvent = JasperFx.Events.IEvent;

class OrderProjection : SingleStreamProjection<Order, string>
{
    public override Order? Evolve(Order? snapshot, string id, JasperFxEvent @event)
    {
        var state = snapshot ?? Order.Empty with { Id = id };
        return @event.Data switch
        {
            OrderPlaced p  => state with { Status = "Pending", Items = p.Items },
            OrderConfirmed => state with { Status = "Confirmed" },
            OrderShipped   => state with { Status = "Shipped"   },
            OrderCancelled => state with { Status = "Cancelled" },
            _              => state,
        };
    }
}

3. Define a command handler

All commands for an aggregate live in one CommandHandler<TAggregate>. Define one public Handle or HandleAsync method per command type. Return IEnumerable<object> — Marten stores whatever you return. Throw to reject the command.

using Winche.Events.Commands;

class OrderCommandHandler : CommandHandler<Order>
{
    public IEnumerable<object> Handle(Order? state, PlaceOrder cmd)
        => [new OrderPlaced(cmd.Items)];

    public IEnumerable<object> Handle(Order? state, ConfirmOrder _)
    {
        if (state?.Status == "Cancelled")
            throw new InvalidOperationException("Cannot confirm a cancelled order.");
        return [new OrderConfirmed()];
    }

    public IEnumerable<object> Handle(Order? state, ShipOrder _)
    {
        if (state?.Status != "Confirmed")
            throw new InvalidOperationException("Cannot ship an order that is not confirmed.");
        return [new OrderShipped()];
    }

    public IEnumerable<object> Handle(Order? state, CancelOrder _)
        => [new OrderCancelled()];
}

state is the current aggregate snapshot (null if the stream does not exist yet). Sync and async handlers can coexist in the same class.

4. Register services

using JasperFx.Events;
using JasperFx.Events.Projections;
using Marten;
using Winche.Events.Commands.DependencyInjection;

builder.Services.AddMarten(opts =>
{
    opts.Connection(connectionString);
    opts.Events.StreamIdentity = StreamIdentity.AsString;

    opts.Events.AddEventType<OrderPlaced>();
    opts.Events.AddEventType<OrderConfirmed>();
    opts.Events.AddEventType<OrderShipped>();
    opts.Events.AddEventType<OrderCancelled>();

    opts.Projections.Add<OrderProjection>(ProjectionLifecycle.Inline);
    opts.Schema.For<Order>().Identity(x => x.Id);
});

builder.Services.AddWincheEventsCommands(opts =>
{
    opts.AddCommandHandler<OrderCommandHandler>();
});

5. Dispatch commands

var dispatcher = provider.GetRequiredService<ICommandDispatcher>();

var result = await dispatcher.DispatchAsync<Order>("orders/123", new PlaceOrder(["Widget A"]));

result.Version        // new stream version after commit
result.Events         // server-enriched events (id, timestamp, version, sequence)

Dispatch flow:

1. Open a session (ReadCommitted)
2. Fetch stream state → load current aggregate snapshot
3. Call the handler → produce events
4. Append events and commit
5. Fetch newly appended events with full server metadata
6. Return DispatchResult

Optimistic concurrency — pass expectedVersion to reject the command if the stream has advanced since the command was created:

await dispatcher.DispatchAsync<Order>("orders/123", new ConfirmOrder(), expectedVersion: 1);

Commands (Winche.Events.Commands)

CommandHandler<TAggregate>

class MyHandler : CommandHandler<MyAggregate>
{
    // Sync handler
    public IEnumerable<object> Handle(MyAggregate? state, MyCommand cmd)
        => [new MyEvent()];

    // Async handler — optional CancellationToken as third parameter
    public async Task<IEnumerable<object>> HandleAsync(
        MyAggregate? state, MyOtherCommand cmd, CancellationToken ct)
    {
        var data = await _service.FetchAsync(ct);
        return [new MyOtherEvent(data)];
    }
}

Handler discovery rules:

• Method name must be Handle or HandleAsync
• Second parameter is the command type (any class or record)
• Optional third parameter CancellationToken is supported
• Both sync and async handlers can coexist in the same class

DispatchResult

public sealed record DispatchResult(IReadOnlyList<IEvent> Events, long Version);

Events are Marten-enriched — each carries server-assigned Id, Timestamp, Version, and Sequence. The Dart SDK uses these to replace pending optimistic-update events with server truth.

Runtime dispatch (transport layer)

Transport layers deserialize commands without knowing TAggregate at compile time. Use the non-generic overload:

await dispatcher.DispatchAsync(streamId, commandObject, ct);

The aggregate type is resolved automatically from registered handlers.

WebSocket transport (Winche.Events.WebSocket)

Single persistent connection per client. All operations are JSON messages on one HTTP/1.1 port — no proto files, no code generation, browser-compatible.

Message types

Wire format

// Client sends:
{
  "type": "DispatchRequest",
  "id": "uuid",
  "streamId": "orders/123",
  "commandType": "PlaceOrder",
  "payload": { "items": ["Widget A"] }
}

// Server responds:
{
  "type": "DispatchResponse",
  "id": "uuid",
  "version": 1,
  "events": [
    {
      "id": "server-uuid",
      "streamId": "orders/123",
      "eventType": "OrderPlaced",
      "data": { "items": ["Widget A"] },
      "version": 1,
      "timestamp": "2026-06-02T...",
      "sequence": 42
    }
  ]
}

Register (WebSocket)

builder.Services.AddWincheEventsCommands(opts =>
{
    opts.AddCommandHandler<OrderCommandHandler>();
});

builder.Services.AddWincheEventsWebSocket();

app.UseCors();
app.UseWebSockets();
app.MapWincheEventsWebSocket("/ws");

AddWincheEventsWebSocket must be called after AddWincheEventsCommands.

Authentication

Connection-time (recommended): Protect the HTTP upgrade with standard ASP.NET Core auth:

app.MapWincheEventsWebSocket("/ws").RequireAuthorization();

Per-message expiry check (optional):

builder.Services.AddWincheEventsWebSocket(opts =>
{
    opts.IsAuthorized = ctx =>
    {
        var exp = ctx.User.FindFirst("exp")?.Value;
        return exp is not null &&
               DateTimeOffset.FromUnixTimeSeconds(long.Parse(exp)) > DateTimeOffset.UtcNow;
    };
});

Run the WebSocket sample

$env:DOTNET_ENVIRONMENT = "Development"
dotnet run --project samples/Winche.Events.WebSocketSample
# WebSocket at ws://localhost:5002/ws

Integration sample (samples/Winche.Events.IntegrationSample)

A console app that runs end-to-end against a real Postgres database. Demonstrates four scenarios: happy path, optimistic concurrency conflict, multiple aggregates, and business rule enforcement.

# Set connection string in samples/Winche.Events.IntegrationSample/appsettings.Development.json
$env:DOTNET_ENVIRONMENT = "Development"
dotnet run --project samples/Winche.Events.IntegrationSample

Requirements

• .NET 10
• PostgreSQL (via Marten / Npgsql)