Getting Started with Event Sourcing and EventSourcing.Backbone
Tags: eventsourcing, cqrs, redis, kafka
Published: 2023-05-29 05:28:10


## Getting Started with Event Sourcing and EventSourcing.Backbone

This post is the first of a series about event sourcing and an exciting framework called ***EventSourcing.Backbone***.

In this post, we’ll explore the fundamentals of event sourcing and a Hello World sample of the framework.

## Understanding Event Sourcing

**Event sourcing** is an architectural pattern that **captures and persists state changes as a sequence of events**. It provides a historical log of events that can be used to **reconstruct the current state** of an application at **any given point in time**. This approach offers various benefits, such as auditability, scalability, and building complex workflows.

Event sourcing, when combined with the Command Query Responsibility Segregation (**CQRS**) pattern, offers even more advantages. CQRS **separates** the **read** and **writes** concerns of an application, enabling the generation of dedicated databases optimized for specific read or write needs. This separation of concerns allows for more agile and **flexible database schema designs**, as they are less critical to set up in advance.

By leveraging *EventSourcing.Backbone*, developers can implement event sourcing and CQRS together, resulting in a powerful architecture that promotes scalability, flexibility, and maintainability.

*You can check [this ](https://www.eventstore.com/event-sourcing)article in order to learn more about event sourcing.*

## Introducing EventSourcing.Backbone

One notable aspect of **EventSourcing.Backbone** is its **unique approach** to event sourcing, instead of inventing a new event source database, *EventSourcing.Backbone* leverages a combination of existing **message streams** like Kafka, Redis Stream, or similar technologies, along with **key-value databases** or services like Redis.

This architecture enables several benefits. Message streams, while excellent for handling event sequences and ensuring reliable message delivery, may not be optimal for heavy payloads. By combining key-value databases with message streams, *EventSourcing.Backbone* allows for message payload to be stored in the key-value database while the stream holds the sequence and metadata. This approach **improves performance** and facilitates compliance with **GDPR** standards by allowing the splitting of the personal data aspects of the messages into different keys easily.

It’s worth noting that *EventSourcing.Backbone* currently provides an SDK for the .NET ecosystem. While the framework may expand to other programming languages and frameworks in the future, at present, it is specifically focused on the .NET platform.

## Leveraging Existing Infrastructure

One of the major advantages of *EventSourcing.Backbone* is its compatibility with widely adopted message streaming platforms like **Kafka or Redis Stream**. These platforms provide robust message delivery guarantees and high throughput, making them ideal for handling event streams at scale.

Furthermore, *EventSourcing.Backbone* seamlessly integrates with popular key-value databases such as **Redis, Couchbase, or Amazon DynamoDB**. This integration allows developers to leverage the strengths of these databases for efficient storage and retrieval of auxiliary data related to events.

## Behind the scenes

_EventSourcing.Backbone_ is taking a **somewhat different approach**. It defines messages **contract via interfaces** (rather than data classes). This approach is friendly for versioning and easier for handling diverse message types. It can also help with GDPR by segregating personal data into different method parameters. In a nutshell, a message type build-out of each method parameter in a way that is transparent to the developer.

Each **producer** method **call** gets **serialized** as **key-value** pairs based on the **method parameters** (by default) and **stored** in the **key-value storage** (Redis Hash in our case). Then the **metadata** of the message is **pushed into** the **stream** (Redis Stream in our case).

The **consumer** is **listening** to the **stream** and **reconstructing** the **message into a method call** right into the subscriber interface.

The framework is capable of handling different message patterns and different strategies, more on this will be posted in future posts.

## Let’s have a taste of it.

![[Taken by Asaf Cohen](https://www.carmel-law.co.il/)](https://cdn-images-1.medium.com/max/4096/1*_RMxa-iJrV0yKTKGuQgfRg.jpeg)

**Prerequisites**:
In order to run the sample you need 
- [_Docker Desktop_](https://www.docker.com/products/docker-desktop/) (for hosting REDIS)
- [_.NET 7 SDK_](https://learn.microsoft.com/en-us/dotnet/core/install/) (or higher)

Get Redis up & running:

```bash
docker run -p 6379:6379 -it --rm --name redis-event-source-sample redislabs/rejson:latest
```

Create a solution with 3 projects (Producer, Consumer, Common Abstraction). or clone the following repo as a starting point:

```bash
git clone https://github.com/bnayae/HelloEventSourcing
```

**Defining mutual interest codes like the common contracts:**

Add the following NuGet packages into the project of the common abstraction:
- [_EventSourcing.Backbone.SrcGen_](https://www.nuget.org/packages/EventSourcing.Backbone.SrcGen/)
- [_EventSourcing.Backbone.Abstractions_](https://www.nuget.org/packages/EventSourcing.Backbone.Abstractions/)

The first step is to define the **events schema**.
“[_EventSourcing.Backbone_](https://www.nuget.org/packages/EventSourcing.Backbone.Abstractions/)” is taking the approach of defining an interface for the event’s data structuring.

Put the following code into the *event abstraction* project.
    
```cs
using EventSourcing.Backbone;
namespace EventSourcing.Demo;

[EventsContract(EventsContractType.Producer)]
[EventsContract(EventsContractType.Consumer)]
public interface IShipmentTracking
{
    ValueTask OrderPlacedAsync(User user, Product product, DateTimeOffset time);
    ValueTask PackingAsync(string email, int productId, DateTimeOffset time);
    ValueTask OnDeliveryAsync(string email, int productId, DateTimeOffset time);
    ValueTask OnReceivedAsync(string email, int productId, DateTimeOffset time);
}
```

The _EventsContract_ attribute is where the **magic** begins.
By decorating the interface with _EventsContract _a **compile-time source generator** will generate the code necessary for producing and consuming events (you don’t have to write boilerplate code).
And in a result will generate the code of the _IShipmentTrackingProducer_ interface for the producer and _IShipmentTrackingConsumer_ for the consumer.
>  Note: all methods should return _ValueTask_ (this is the convention), yet _void _will be fine but will result in _ValueTask _on the generated version of the interface (producer/consumer). furthermore, all generated methods will have the _Async_ suffix.

**Creating a producer:**

Add the following Nuget for using a **Redis-based producer** (using Redis Stream + Redis Hash).
- [_EventSourcing.Backbone.Channels.RedisProducerProvider_](https://www.nuget.org/packages/EventSourcing.Backbone.Channels.RedisProducerProvider)
_And don’t forget to add a reference to the common abstraction project we created earlier._

```cs
using EventSourcing.Backbone;
using EventSourcing.Demo;
    
IShipmentTrackingProducer producer = RedisProducerBuilder.Create()
                                    .Uri("hello.event-sourcing")
                                    .BuildShipmentTrackingProducer();
    
User user = new(1, "someone@gmail.com", "someone");
var product = new Product(1234, "Something you must have", 10000);
await producer.OrderPlacedAsync(user, product, DateTimeOffset.UtcNow);
```

That's all you need to have a working producer.
>  Take a look at _BuildShipmentTrackingProducer_ which is an extension method generated from the _IShipmentTracking_ interface and returns an _IShipmentTrackingProducer_ .

**Creating a consumer:**

Add the following Nuget for using a **Redis-based consumer** (using Redis Stream + Redis Hash).
- [_EventSourcing.Backbone.Channels.RedisConsumerProvider_](https://www.nuget.org/packages/EventSourcing.Backbone.Channels.RedisConsumerProvider)
_And don’t forget to add a reference to the common abstraction project we created earlier._

Add a class that implements the consumer (generated) interface:

```cs
using EventSourcing.Demo;
    
class Subscription : IShipmentTrackingConsumer
{
    public static readonly Subscription Instance = new Subscription();
    
    public ValueTask OrderPlacedAsync(
                        User user, Product product, DateTimeOffset time)
    {
        Console.WriteLine($"Order Placed: {user.name}, {product.name}, {product.id}");
            return ValueTask.CompletedTask;
    }
    
    public ValueTask PackingAsync(string email, int productId, DateTimeOffset time)
    {
        Console.WriteLine($"Packing: {email}, {productId}, {time}");
        return ValueTask.CompletedTask;
    }
    
    public ValueTask OnDeliveryAsync(string email, int productId, DateTimeOffset time)
    {
        Console.WriteLine($"On Delivery: {email}, {productId}, {time}");
        return ValueTask.CompletedTask;    
    }
    
    public ValueTask OnReceivedAsync(string email, int productId, DateTimeOffset time)
    {
        Console.WriteLine($"On Received: {email}, {productId}, {time}");
        return ValueTask.CompletedTask;   
    }
}
```

The final step is to attach this class to the event stream.

```cs
using EventSourcing.Demo;
using EventSourcing.Backbone;
    
IConsumerLifetime subscription = RedisConsumerBuilder.Create()
                                          .Uri(URIs.Default)
                                          .SubscribeShipmentTrackingConsumer(Subscription.Instance);
    
await subscription.Completion;
```

If you recall the producer was sending a single event of _OrderPlacedAsync_, 
To complete the picture we’ll add a combination of consumer & producer which will change the state along the shipping flow.

Create a new project and add both the **Redis-based** producer & consumer NuGets:
- [_EventSourcing.Backbone.Channels.RedisConsumerProvider_](https://www.nuget.org/packages/EventSourcing.Backbone.Channels.RedisConsumerProvider)
- [_EventSourcing.Backbone.Channels.RedisProducerProvider_](https://www.nuget.org/packages/EventSourcing.Backbone.Channels.RedisProducerProvider)

Add an implementation of _IShipmentTrackingConsumer_.

```cs
using EventSourcing.Backbone;
using EventSourcing.Demo;
    
class Subscription : IShipmentTrackingConsumer
{
    public static readonly Subscription Instance = new Subscription();
    
    private readonly IShipmentTrackingProducer _producer = RedisProducerBuilder.Create()
                                .Uri(URIs.Default)
                                .BuildShipmentTrackingProducer();
    
    
    public async ValueTask OrderPlacedAsync(User user, Product product, DateTimeOffset time)
    {
        await _producer.PackingAsync(user.email, product.id, DateTimeOffset.Now);
    }
    
    public async ValueTask PackingAsync(string email, int productId, DateTimeOffset time)
    {
        await _producer.OnDeliveryAsync(email, productId, DateTimeOffset.Now);
    }
    
    public async ValueTask OnDeliveryAsync(string email, int productId, DateTimeOffset time)
    {
        await _producer.OnReceivedAsync(email, productId, DateTimeOffset.Now);
    }
    
    public ValueTask OnReceivedAsync(string email, int productId, DateTimeOffset time)
    {
            return ValueTask.CompletedTask;
    }
}
```

This way we can handle the state’s transition.

The last step is to plug it into the stream:

```cs
using EventSourcing.Demo;
using EventSourcing.Backbone;
    
Console.WriteLine("Consuming and Producing Events");
    
IConsumerLifetime subscription = RedisConsumerBuilder.Create()
                                                .Uri(URIs.Default)
                                                .Group("transit")
                                                .SubscribeShipmentTrackingConsumer(Subscription.Instance);
    
await subscription.Completion;
```

Notice the _Group(“transit”)_ this one is important for separating the consumer group which deals with the state transition from the one that reports the current state.
>  Consumer groups distribute the workload of processing messages among multiple consumers, allowing parallel processing while ensuring each message is handled only once. This ensures efficient and reliable message processing, scalability, and fault tolerance when dealing with large volumes of data or high-throughput message streams.

Putting both consumers in the same consumer group (or under the default one) will result in messages either handled by the transit or by the reporter and lead to a mess.
>  The full code is available at:
[https://github.com/bnayae/HelloEventSourcing/tree/HelloWorld](https://github.com/bnayae/HelloEventSourcing/tree/HelloWorld)
_notice that it is under the **HelloWorld** branch_

## Conclusion

In this introductory post, we’ve explored the concept of **event sourcing** and introduced the unique aspects of _EventSourcing.Backbone_. This framework stands out by leveraging a combination of message streams and key-value databases to achieve better performance, support GDPR standards, and provide flexibility in event-sourcing implementations.

_EventSourcing.Backbone_ integrates seamlessly with popular message-streaming platforms and key-value databases, enabling developers to leverage their strengths for scalable and efficient event sourcing.

When combined with CQRS, event sourcing can enhance database schema design, making it more agile and flexible, and enabling the generation of dedicated databases optimized for specific needs.

In future posts, we’ll drill down to more advanced patterns, concerns, pros & cons, and best practices.