JasperFx Software is around and ready to assist you with getting the best possible results using the Critter Stack.

The projections model in Marten and now Polecat has evolved quite a bit over the past decade. Consider this simple aggregated projection of data for our QuestParty in our tests:

public class QuestParty
{
    public List<string> Members { get; set; } = new();
    public IList<string> Slayed { get; } = new List<string>();
    public string Key { get; set; }
    public string Name { get; set; }
    // In this particular case, this is also the stream id for the quest events
    public Guid Id { get; set; }
    // These methods take in events and update the QuestParty
    public void Apply(MembersJoined joined) => Members.Fill(joined.Members);
    public void Apply(MembersDeparted departed) => Members.RemoveAll(x => departed.Members.Contains(x));
    public void Apply(QuestStarted started) => Name = started.Name;
    public override string ToString()
    {
        return $"Quest party '{Name}' is {Members.Join(", ")}";
    }
}

That type is mutable, but the projection library underneath Marten and Polecat happily supports projecting to immutable types as well.

Some people actually like the conventional method approach up above with the Apply, Create, and ShouldDelete methods. From the perspective of Marten’s or Polecat’s internals, it’s always been helpful because the projection subsystem “knows” in this case that the QuestParty is only applicable to the specific event types referenced in those methods, and when you call this code:

var party = await query
    .Events
    .AggregateStreamAsync<QuestParty>(streamId);

Marten and Polecat are able to quietly use extra SQL filters to limit the events fetched from the database to only the types utilized by the projected QuestParty aggregate.

Great, right? Except that some folks don’t like the naming conventions, just prefer explicit code, or do some clever things with subclasses on events that can confuse Marten or Polecat about the precedence of the event type handlers. To that end, Marten 8.0 introduced more options for explicit code. We can rewrite the projection part of the QuestParty above to a completely different class where you can add explicit code:

public class QuestPartyProjection: SingleStreamProjection<QuestParty, Guid>
{
    public QuestPartyProjection()
    {
        // This is *no longer necessary* in
        // the very most recent versions of Marten,
        // but used to be just to limit Marten's
        // querying of event types when doing live
        // or async projections
        IncludeType<MembersJoined>();
        IncludeType<MembersDeparted>();
        IncludeType<QuestStarted>();
    }
    public override QuestParty Evolve(QuestParty snapshot, Guid id, IEvent e)
    {
        snapshot ??= new QuestParty{ Id = id };
        switch (e.Data)
        {
            case MembersJoined j:
                // Small helper in JasperFx that prevents
                // double values
                snapshot.Members.Fill(j.Members);
                break;
            case MembersDeparted departed:
                snapshot.Members.RemoveAll(x => departed.Members.Contains(x));
                break;
        }
        return snapshot;
    }
}

There are several more items in that SingleStreamProjection base type like versioning or fine grained control over asynchronous projection behavior that might be valuable later, but for now, let’s look at a new feature in Marten and Polecat that let’s you use explicit code right in the single aggregate type:

public class QuestParty
{
    public List<string> Members { get; set; } = new();
    public IList<string> Slayed { get; } = new List<string>();
    public string Key { get; set; }
    public string Name { get; set; }
    // In this particular case, this is also the stream id for the quest events
    public Guid Id { get; set; }
    public void Evolve(IEvent e)
    {
        switch (e.Data)
        {
            case QuestStarted _:
                // Little goofy, but this let's Marten know that
                // the projection cares about that event type
                break;
            
            case MembersJoined j:
                // Small helper in JasperFx that prevents
                // double values
                Members.Fill(j.Members);
                break;
            case MembersDeparted departed:
                Members.RemoveAll(x => departed.Members.Contains(x));
                break;
        }
    }
    public override string ToString()
    {
        return $"Quest party '{Name}' is {Members.Join(", ")}";
    }
}

This is admittedly yet another convention method in terms of the method name and the possible arguments, but hopefully the switch statement approach is much more explicit for folks who prefer that. As an additional bonus, Marten is able to automatically register the event types via a source generator that the version of QuestParty just above is using automatically so that we get all the benefits of the event filtering without making users do extra explicit configuration.

Projecting to Immutable Views

Just for completeness, let’s look at alternative versions of QuestParty just to see what it looks like if you make the aggregate an immutable type. First up is the conventional method approach:

public sealed record QuestParty(Guid Id, List<string> Members)
{
    // These methods take in events and update the QuestParty
    public static QuestParty Create(QuestStarted started) => new(started.QuestId, []);
    public static QuestParty Apply(MembersJoined joined, QuestParty party) =>
        party with
        {
            Members = party.Members.Union(joined.Members).ToList()
        };
    public static QuestParty Apply(MembersDeparted departed, QuestParty party) =>
        party with
        {
            Members = party.Members.Where(x => !departed.Members.Contains(x)).ToList()
        };
    public static QuestParty Apply(MembersEscaped escaped, QuestParty party) =>
        party with
        {
            Members = party.Members.Where(x => !escaped.Members.Contains(x)).ToList()
        };
}

And with the Evolve approach:

public sealed record QuestParty(Guid Id, List<string> Members)
{
    public static QuestParty Evolve(QuestParty? party, IEvent e)
    {
        switch (e.Data)
        {
            case QuestStarted s:
                return new(s.QuestId, []);
            case MembersJoined joined:
                return party with         {
                    Members = party.Members.Union(joined.Members).ToList()
                };
            case MembersDeparted departed:
                return party with
                {
                    Members = party.Members.Where(x => !departed.Members.Contains(x)).ToList()
                };
            case MembersEscaped escaped:
                return party with
                {
                    Members = party.Members.Where(x => !escaped.Members.Contains(x)).ToList()
                };
        }
        return party;
    }

Summary

What do I recommend? Honestly, just whatever you prefer. This is a case where I’d like everyone to be happy with one of the available options. And yes, it’s not always good that there is more than one way to do the same thing in a framework, but I think we’re going to just keep all these options in the long run. It wasn’t shown here at all, but I think we’ll kill off the early options to define projections through a ton of inline Lambda functions within a fluent interface. That stuff can just die.

In the medium and longer term, we’re going to be utilizing more source generators across the entire Critter Stack as a way of both eliminating some explicit configuration requirements and to optimize our cold start times. I’m looking forward to getting much more into that work.

Just to level set everyone, there are two general categories of identifiers we use in software:

• “Surrogate” keys are data elements like Guid values, database auto numbering or sequences, or snowflake generated identifiers that have no real business meaning and just try to be unique values.
• “Natural” keys have some kind of business meaning and usually utilize some piece of existing information like email addresses or phone numbers. A natural key could also be an external supplied identifier from your clients. In fact, it’s quite common to have your own tracking identifier (usually a surrogate key) while also having to track a client or user’s own identification for the same business entity.

That very last sentence is where this post takes off. You see Marten can happily track event streams with either Guid identifiers (surrogate key) or string identifiers — or strong typed identifiers that wrap an inner Guid or string, but in this case that’s really the same thing, just with more style I guess. Likewise, in combination with Wolverine for our recommended “aggregate handler workflow” approach to building command handlers, we’ve only supported the stream id or key. Until now!

With the Marten 8.23 and Wolverine 5.18 releases last week (we’ve been very busy and there are newer releases now), you are now able to “tag” Marten (or Polecat!) event streams with a natural key in addition to its surrogate stream id and use that natural key in conjunction with Wolverine’s aggregate handler workflow.

Of course, if you use strings as the stream identifier you could already use natural keys, but let’s just focus on the case of Guid identified streams that are also tagged with some kind of natural key that will be supplied by users in the commands sent to the system.

First, to tag streams with natural keys in Marten, you have to have a strong typed identifier type for the natural key. Next, there’s a little bit of attribute decoration in the targeted document type of a single stream projection, i.e., the “write model” for an event stream. Here’s an example from the Marten documentation:

public record OrderNumber(string Value);
public record InvoiceNumber(string Value);
public class OrderAggregate
{
    public Guid Id { get; set; }
    [NaturalKey]
    public OrderNumber OrderNum { get; set; }
    public decimal TotalAmount { get; set; }
    public string CustomerName { get; set; }
    public bool IsComplete { get; set; }
    [NaturalKeySource]
    public void Apply(OrderCreated e)
    {
        OrderNum = e.OrderNumber;
        CustomerName = e.CustomerName;
    }
    public void Apply(OrderItemAdded e)
    {
        TotalAmount += e.Price;
    }
    [NaturalKeySource]
    public void Apply(OrderNumberChanged e)
    {
        OrderNum = e.NewOrderNumber;
    }
    public void Apply(OrderCompleted e)
    {
        IsComplete = true;
    }
}

In particular, see the usage of [NaturalKey] which should be self-explanatory. Also see the [NaturalKeySource] attribute that we’re using to mark when a natural key value might change. Marten is starting to use source generators for some projection internals (in place of some nasty, not entirely as efficient as it should have been, Expression-compiled-to-Lambda functions).

And that’s that, really. You’re now able to use the designated natural keys as the input to an “aggregate handler workflow” command handler with Wolverine. See Natural Keys from the Wolverine documentation for more information.

For a little more information:

• The natural keys are stored in a separate table, and when using FetchForWriting(), Marten is doing an inner join from the tag table for that natural key type to the mt_streams table in the Marten database
• You can change the natural key against the surrogate key
• We expect this to be most useful when you want to use the Guid surrogate keys for uniqueness in your own system, but you frequently receive a natural key from API users of your system — or at least this has been encountered by a couple different JasperFx Software customers.
• The natural key storage does have a unique value constraint on the “natural key” part of the storage
• Really only a curiosity, but this was done in the same wave of development as Marten’s new DCB support

It’s early so I should be too cocky, but JasperFx Software is having success in integrating SignalR with both Wolverine and Marten in our forthcoming CritterWatch product. In this post I’ll show you how we’re doing that from the server side C# code all the way down to the client side TypeScript.

Last week I did a live stream talking about many of the details and a way too early demonstration of CritterWatch, JasperFx Software‘s long planned management console for the “Critter Stack” tools (Marten, Wolverine, and soon to be Polecat).

A big technical wrinkle in the CritterWatch approach so far is our utilization of the SignalR messaging support built into Wolverine. Just like with external messaging brokers like Rabbit MQ or Azure Service Bus, Wolverine does a lot of work to remove the technical details of SignalR and let’s you focus on just writing your application code.

In some ways, CritterWatch is kind of a man in the middle between the intended CritterWatch user interface (Vue.js) and the Wolverine enabled applications in your system:

Note that Wolverine will be required for CritterWatch, but if you only today use Marten and want to use CritterWatch to manage just the event sourcing, know that you will be able to use a very minimalistic Wolverine setup just for communication to CritterWatch without having to migrate your entire messaging infrastructure to Wolverine. And for that matter, Wolverine now has a pretty robust HTTP transport for asynchronous messaging that would work fine for CritterWatch integration.

As I said earlier, CritterWatch is going to depend very heavily on two way WebSockets communication between the user interface and the CritterWatch server, and we’re utilizing Wolverine’s SignalR messaging transport (which was purposefully built for CritterWatch in the first place) to get that done. In the CritterWatch codebase, we have this little bit of Wolverine configuration:

    public static void AddCritterWatchServices(this WolverineOptions opts, NpgsqlDataSource postgresSource)
    {
        // Much more of course...
        opts.Services.AddWolverineHttp();
        
        opts.UseSignalR();
        
        // The publishing rule to route any message type that implements
        // a marker interface to the connected SignalR Hub
        opts.Publish(x =>
        {
            x.MessagesImplementing<ICritterStackWebSocketMessage>();
            x.ToSignalR();
        });


        // Really need this so we can handle messages in order for 
        // a particular service
        opts.MessagePartitioning.UseInferredMessageGrouping();
        opts.Policies.AllListeners(x => x.PartitionProcessingByGroupId(PartitionSlots.Five));
    }

And at the bottom of the ASP.Net Core application hosting CritterWatch, we’ll have this to configure the request pipeline:

builder.Services.AddWolverineHttp();
var app = builder.Build();
// Little bit more in the real code of course...
app.MapWolverineSignalRHub("/api/messages");
return await app.RunJasperFxCommands(args);

As you can infer from the Wolverine publishing rule above, we’re using a marker interface to let Wolverine “know” what messages should always be sent to SignalR:

/// <summary>
/// Marker interface for all messages that are sent to the CritterWatch web client
/// via web sockets
/// </summary>
public interface ICritterStackWebSocketMessage : ICritterWatchMessage, WebSocketMessage;

We also use that marker interface in a homegrown command line integration to generate TypeScript versions of all those messages with NJsonSchema as well as message types that go from the user interface to the CritterWatch server. Wolverine’s SignalR integration assumes that all messages sent to SignalR or received from SignalR are wrapped in a Cloud Events compliant JSON wrapper, but the only required members are type that should identify what type of message it is and data that holds the actual message body as JSON. To make this easier, when we generate the TypeScript code we also insert a little method like this that we can use to identify the message type sent from the client to the Wolverine powered back end:

export class CompactStreamResult implements WebsocketMessage {
    serviceName!: string;
    streamId!: string;
    success!: boolean;
    error!: string | undefined;
    queryId!: string | undefined;
    // THIS method is injected by our custom codegen
    // and helps us communicate with the server as
    // this matches Wolverine's internal identification of
    // this message
    get messageTypeName() : string{
        return "compact_stream_result";
    }
    
    // other stuff...
    init(_data?: any) {
        if (_data) {
            this.serviceName = _data["serviceName"];
            this.streamId = _data["streamId"];
            this.success = _data["success"];
            this.error = _data["error"];
            this.queryId = _data["queryId"];
        }
    }
    static fromJS(data: any): CompactStreamResult {
        data = typeof data === 'object' ? data : {};
        let result = new CompactStreamResult();
        result.init(data);
        return result;
    }
    toJSON(data?: any) {
        data = typeof data === 'object' ? data : {};
        data["serviceName"] = this.serviceName;
        data["streamId"] = this.streamId;
        data["success"] = this.success;
        data["error"] = this.error;
        data["queryId"] = this.queryId;
        return data;
    }
}

Most of the code above is generated by NJsonSchema, but our custom codegen inserts in the get messageTypeName() method, that we use in the client side code below to wrap up messages to send back up to our server:

  async function sendMessage(msg: WebsocketMessage) {
    if (conn.state === HubConnectionState.Connected) {
      const payload = 'toJSON' in msg ? (msg as any).toJSON() : msg
      const cloudEvent = JSON.stringify({
        id: crypto.randomUUID(),
        specversion: '1.0',
        type: msg.messageTypeName,
        source: 'Client',
        datacontenttype: 'application/json; charset=utf-8',
        time: new Date().toISOString(),
        data: payload,
      })
      await conn.invoke('ReceiveMessage', cloudEvent)
    }
  }

In the reverse direction, we receive the raw message from a connected WebSocket with the SignalR client, interrogate the expected CloudEvents wrapper, figure out what the message type is from there, deserialize the raw JSON to the right TypeScript type, and generally just relay that to a Pinia store where all the normal Vue.js + Pinia reactive user interface magic happens.

export function relayToStore(data: any){
    const servicesStore = useServicesStore();
    const dlqStore = useDlqStore();
    const metricsStore = useMetricsStore();
    const projectionsStore = useProjectionsStore();
    const durabilityStore = useDurabilityStore();
    const eventsStore = useEventsStore();
    const scheduledMessagesStore = useScheduledMessagesStore();
    const envelope = typeof data === 'string' ? JSON.parse(data) : data;
    switch (envelope.type){
        case "dead_letter_details":
            dlqStore.handleDeadLetterDetails(DeadLetterDetails.fromJS(envelope.data));
            break;
        case "dead_letter_queue_summary_results":
            dlqStore.handleDeadLetterQueueSummaryResults(DeadLetterQueueSummaryResults.fromJS(envelope.data));
            break;
        case "all_service_summaries":
            const allSummaries = AllServiceSummaries.fromJS(envelope.data);
            servicesStore.handleAllServiceSummaries(allSummaries);
            if (allSummaries.persistenceCounts) {
                for (const pc of allSummaries.persistenceCounts) {
                    durabilityStore.handlePersistenceCountsChanged(pc);
                }
            }
            if (allSummaries.metricsRollups) {
                metricsStore.handleAllMetricsRollups(allSummaries.metricsRollups);
            }
            break;
        case "summary_updated":
            servicesStore.handleSummaryUpdated(SummaryUpdated.fromJS(envelope.data));
            break;
        case "agent_and_node_state_changed":
            servicesStore.handleAgentAndNodeStateChanged(AgentAndNodeStateChanged.fromJS(envelope.data));
            break;
        case "service_summary_changed":
            servicesStore.handleServiceSummaryChanged(ServiceSummaryChanged.fromJS(envelope.data));
            break;
        case "metrics_rollup":
            metricsStore.handleMetricsRollup(MetricsRollup.fromJS(envelope.data));
            break;
        case "all_metrics_rollups":
            metricsStore.handleAllMetricsRollups(AllMetricsRollups.fromJS(envelope.data));
            break;
        case "shard_states_changed":
            projectionsStore.handleShardStatesChanged(ShardStatesChanged.fromJS(envelope.data));
            break;
        case "persistence_counts_changed":
            durabilityStore.handlePersistenceCountsChanged(PersistenceCountsChanged.fromJS(envelope.data));
            break;
        case "stream_details":
            eventsStore.handleStreamDetails(StreamDetails.fromJS(envelope.data));
            break;
        case "event_query_results":
            eventsStore.handleEventQueryResults(EventQueryResults.fromJS(envelope.data));
            break;
        case "compact_stream_result":
            eventsStore.handleCompactStreamResult(CompactStreamResult.fromJS(envelope.data));
            break;
        // *CASE ABOVE* -- do not remove this comment for the codegen please!
    }
}

And that’s really it. I omitted some of our custom codegen code (because it’s hokey), but it doesn’t do much more than find the message types in the .NET code that are marked as going to or coming from the Vue.js client and writes them as matching TypeScript types.

But wait, Marten gets into the act too!

With the Marten + Wolverine integration through this:

        opts.Services.AddMarten(m =>
        {
            // Other stuff...
            m.Projections.Add<ServiceSummaryProjection>(ProjectionLifecycle.Async);
        }).IntegrateWithWolverine(w =>
        {
            w.UseWolverineManagedEventSubscriptionDistribution = true;
        });

Marten can also get into the SignalR act through its support for “Side Effects” in projections. As a certain projection for ServiceSummary is updated with new events in CritterWatch, we can raise messages reflecting the new changes in state to notify our clients with code like this from a SingleStreamProjection:

    public override ValueTask RaiseSideEffects(IDocumentOperations operations, IEventSlice<ServiceSummary> slice)
    {
        var hasShardStates = slice.Events().Any(x => x.Data is ShardStatesUpdated);

        if (hasShardStates)
        {
            var shardEvent = slice.Events().Last(x => x.Data is ShardStatesUpdated).Data as ShardStatesUpdated;
            slice.PublishMessage(new ShardStatesChanged(slice.Snapshot.Id, shardEvent!.States));
        }

        if (slice.Events().All(x => x.Data is IImpactsAgentOrNodes || x.Data is ShardStatesUpdated))
        {
            if (!hasShardStates)
            {
                slice.PublishMessage(new AgentAndNodeStateChanged(slice.Snapshot.Id, slice.Snapshot.Nodes, slice.Snapshot.Agents));
            }
        }
        else
        {
            slice.PublishMessage(new ServiceSummaryChanged(slice.Snapshot));
        }

        return new ValueTask();
    }

The Marten projection itself knows absolutely nothing about where those messages will go or how, but Wolverine kicks in to help its Critter Stack sibling and it deals with all the message delivery. The message types above all implement the ICritterStackWebSocketMessage interface, so they will get routed by Wolverine to SignalR. To rewind, the workflow here is:

1. CritterWatch constantly receives messages from Wolverine applications with changes in state like new messaging endpoints being used, agents being reassigned, or nodes being started or shut down
2. CritterWatch saves any changes in state as events to Marten (or later to SQL Server backed Polecat)
3. The Marten async daemon processes those events to update CritterWatch’s ServiceSummary projection
4. As pages of events are applied to individual services, Marten calls that RaiseSideEffects() method to relay some state changes to Wolverine, which will..
5. Send those messages to SignalR based on Wolverine’s routing rules and on to the client side code which…
6. Relays the incoming messages to the proper Pinia store

Summary

I won’t say that using Wolverine for processing and sending messages via SignalR is justified in every application, but it more than pays off if you have a highly interactive application that sends any number of messages between the user interface and the server.

Sometime last week I said online that no project is truly a failure if you learned something valuable from that effort that could help a later project succeed. When I wrote that I was absolutely thinking about the work shown above and relating that to a failed effort of mine called Storyteller (Redux + early React.js + roll your own WebSockets support on the server) that went nowhere in the end, but taught me a lot of valuable lessons about using WebSockets in a highly interactive application that has directly informed my work on CritterWatch.

The Critter Stack had a big day today with releases for both Marten and Wolverine.

First up, we have Marten 8.22 that included:

• Lots of bug fixes, including several old LINQ related bugs and issues related to full text search that finally got addressed
• Some improvements for the newer Composite Projections feature as users start to use it in real project work. Hat tip to Anne Erdtsieck on this one (and a JasperFx client needing an addition to it as well)
• Some optimizations, including a potentially big one as Marten can now use a source generator to build some of the projection code that before depended on not perfectly efficient Expression compilation. This will impact “self aggregating” snapshot projections that use the Apply / Create / ShouldDelete conventions

Next, a giant Wolverine 5.16 release that brings:

• Many, many bug fixes
• Several small feature requests for our HTTP support
• Improved resiliency for Kafka especially but also for any usage of external message brokers with Wolverine. See Sending Error Handling. Plus better error handling for durable listener endpoints when the transactional inbox database is unavailable
• Wait, what? Wolverine has experimental support for CosmosDb as a transactional inbox/outbox and all of Wolverine’s declarative persistence helpers?
• The ability to mark some message handlers or HTTP endpoints as opting out of automatic transactional middleware (for a JasperFx client). See this, but it applies to all persistence options.
• Modular monolith usage improvements for a pair of JasperFx clients who are helping us stretch Wolverine to yet more use cases.
• More to come on this, but we’ve recently slipped in Sqlite and Oracle support for Wolverine