Building an Event Sourced Application with NATS JetStream for Aggregate Playback and Filtering

Building an Event Sourced Application with NATS JetStream for Aggregate Playback and Filtering

Creating an event-sourced application using NATS JetStream that allows for individual aggregate playback, filtering by event type, and subscription to all events for a given aggregate, while also organizing aggregates by domain, requires a careful design of your stream and consumer configuration. This detailed guide will walk you through the necessary steps and considerations.

Core Concepts: Event Sourcing and Aggregates

Before diving into the specifics of NATS JetStream, it's crucial to understand the core concepts of event sourcing and aggregates. Event sourcing is a pattern where the state of an application is determined by a sequence of events. Instead of storing the current state directly, you store each change as an immutable event. The current state can then be derived by replaying these events. An aggregate, in the context of domain-driven design, is a cluster of domain objects that can be treated as a single unit. In your case, an "item" is an aggregate.

NATS JetStream Stream Design

The foundation of your event-sourced application in NATS JetStream is the stream design. You'll need to structure your stream to efficiently support your requirements for aggregate playback, filtering, and subscription. Here's a recommended approach:

Stream Structure

Instead of a single monolithic stream, it's best to create a stream per domain. This allows you to isolate concerns and manage data more effectively. For example, you might have a stream named "inventory" for inventory-related events and another stream named "orders" for order-related events. Within each stream, you will publish events with a subject that encodes the aggregate ID and event type. The subject structure should follow a consistent pattern, such as:

domain.aggregateType.aggregateId.eventType

For example, for an item with ID 123 in the inventory domain, you might have subjects like:

• inventory.item.123.created
• inventory.item.123.updated
• inventory.item.123.priceChanged
• inventory.item.123.stockAdjusted

This subject structure provides several advantages:

• Domain Isolation: Events for different domains are kept separate, improving organization and scalability.
• Aggregate Identification: The subject clearly identifies the aggregate type (e.g., item) and the specific aggregate instance (e.g., 123).
• Event Type Filtering: The event type (e.g., created, updated) is part of the subject, enabling easy filtering.

Stream Configuration

When creating your streams, consider the following configuration options:

• Storage: Choose the appropriate storage type (e.g., file or memory) based on your persistence requirements. File storage is recommended for production environments.
• Retention Policy: Determine how long events should be retained. You might choose a time-based retention policy (e.g., retain events for 30 days) or a size-based policy (e.g., retain the last 100,000 events).
• Replication: Configure replication to ensure high availability and fault tolerance.
• Subjects: Define the subjects that the stream will accept. You can use wildcards to match multiple subjects. For example, inventory.item.*.* would match all events for all items in the inventory domain.

Aggregate Playback

To play back all events for a specific aggregate, you'll need to create a consumer that is configured to filter by the aggregate ID. Here's how to achieve this:

Creating a Consumer for Aggregate Playback

You'll create a consumer with a filter that matches the specific aggregate ID. For example, to play back all events for item 123 in the inventory domain, you would create a consumer with the following subject filter:

inventory.item.123.*

This filter will ensure that the consumer only receives events related to item 123. You can then iterate through the received events to reconstruct the state of the aggregate at any point in time.

Replaying Events

When replaying events, you'll typically start from the beginning of the stream and apply each event in order. This process will reconstruct the aggregate's state. You can also choose to replay events from a specific point in time by using the consumer's start_time option.

Filtering by Event Type

To further filter events by type, you can refine the subject filter of your consumer. For example, to retrieve only the updated events for item 123, you would use the following subject filter:

inventory.item.123.updated

This will ensure that the consumer only receives events of the updated type for the specified aggregate. You can combine this with aggregate filtering to achieve very specific event retrieval.

Subscribing to All Events for an Aggregate

To subscribe to all events for a given aggregate, you'll create a consumer that matches all event types for that aggregate. For example, to subscribe to all events for item 123 in the inventory domain, you would use the following subject filter:

inventory.item.123.*

This consumer will receive all events related to item 123 as they are published to the stream. This allows you to build real-time views of your aggregates or trigger downstream processes based on aggregate changes.

Organizing Aggregates by Domain

As mentioned earlier, organizing aggregates by domain is crucial for scalability and maintainability. By creating separate streams for each domain, you can:

• Isolate Concerns: Each domain has its own stream, preventing interference between different parts of your application.
• Improve Scalability: You can scale each domain independently, allowing you to optimize resource usage.
• Enhance Maintainability: Domain-specific streams make it easier to understand and maintain your application.

The subject structure domain.aggregateType.aggregateId.eventType ensures that events are correctly routed to the appropriate stream and can be easily filtered based on domain, aggregate type, aggregate ID, and event type.

Implementation Details

Here are some implementation details to consider:

Publishing Events

When publishing events, ensure that you use the correct subject format. You'll need to construct the subject string dynamically based on the domain, aggregate type, aggregate ID, and event type. For example, in Python, you might have a function like this:

        
def publish_event(js, domain, aggregate_type, aggregate_id, event_type, event_data):
    subject = f"{domain}.{aggregate_type}.{aggregate_id}.{event_type}"
    js.publish(subject, json.dumps(event_data).encode())
        
    

Creating Consumers

When creating consumers, you'll need to specify the subject filter and other consumer options. You can use the NATS client library to create consumers programmatically. For example, in Python, you might have a function like this:

        
def create_consumer(js, stream_name, subject_filter, consumer_name):
    consumer = js.add_consumer(
        stream_name,
        durable_name=consumer_name,
        ack_policy=nats.AckPolicy.ALL,
        filter_subject=subject_filter
    )
    return consumer
        
    

Handling Events

When handling events, you'll need to deserialize the event data and apply it to your aggregate. You'll also need to handle errors and ensure that events are processed in the correct order. You can use the consumer's next_batch() method to retrieve multiple events at once and process them in a batch.

Example Scenario

Let's consider a concrete example. Suppose you have an e-commerce application with two domains: inventory and orders. In the inventory domain, you have item aggregates, and in the orders domain, you have order aggregates.

Here's how you might structure your streams and subjects:

• Inventory Stream:
  • Subjects:
    • inventory.item.123.created
    • inventory.item.123.updated
    • inventory.item.456.created
    • inventory.item.456.stockAdjusted
• Orders Stream:
  • Subjects:
    • orders.order.789.created
    • orders.order.789.itemAdded
    • orders.order.101.created
    • orders.order.101.shipped

To play back all events for item 123, you would create a consumer with the subject filter inventory.item.123.*. To subscribe to all created events for orders, you would create a consumer with the subject filter orders.order.*.created. To get all events for order 789, you would use the subject filter orders.order.789.*.

Benefits of This Approach

This approach offers several benefits:

• Flexibility: You can easily filter events based on domain, aggregate type, aggregate ID, and event type.
• Scalability: You can scale each domain independently.
• Maintainability: The clear subject structure makes it easier to understand and maintain your application.
• Real-time Updates: You can subscribe to events and receive real-time updates for your aggregates.
• Auditability: All events are stored in the stream, providing a complete audit trail of changes to your aggregates.

Conclusion

Building an event-sourced application with NATS JetStream requires careful planning and design. By structuring your streams and subjects appropriately, you can achieve efficient aggregate playback, filtering, and subscription. The key is to use a consistent subject structure that encodes the domain, aggregate type, aggregate ID, and event type. This will allow you to leverage the powerful filtering capabilities of NATS JetStream and build a robust and scalable event-sourced application.