# Event Sourcing Pattern (/learn/events-store/tutorials/event-sourcing)



Event sourcing stores every state change as an immutable event rather than overwriting the current state. KubeMQ Events Store is a natural fit because it provides persistent, sequenced, and replayable event streams.

## What You Will Build [#what-you-will-build]

An order management service that:

* Stores every order state change (created, paid, shipped, delivered) as an event
* Reconstructs the current order state by replaying the full event history
* Supports checkpoint-based recovery to avoid full replays

<Mermaid
  chart="graph LR
    OS[&#x22;Order Service&#x22;]
    ES[(&#x22;Event Store<br/>channel: order.ORD-1001&#x22;)]
    E1[&#x22;seq=1: order.created&#x22;]
    E2[&#x22;seq=2: order.paid&#x22;]
    E3[&#x22;seq=3: order.shipped&#x22;]
    RS[&#x22;State Rebuilder&#x22;]
    STATE[&#x22;Order: ORD-1001<br/>Status: shipped<br/>Total: $149.99&#x22;]

    OS -- &#x22;store events&#x22; --> ES
    ES --- E1 --- E2 --- E3
    ES -- &#x22;replay all&#x22; --> RS
    RS -- compute --> STATE

    class OS,RS client
    class ES store
    class E1,E2,E3 store
    class STATE data"
/>

*Every order state change is appended to the event store; replaying the full sequence reconstructs the current order state.*

## Prerequisites [#prerequisites]

* KubeMQ server running on `localhost:50000`
* SDK installed ([Getting Started](/learn/events-store/getting-started))

## Step-by-Step [#step-by-step]

<Steps>
  <Step>
    ### Define the Event Schema [#define-the-event-schema]

    Each event includes:

    * **type** — the event kind (e.g., `order.created`, `order.paid`)
    * **orderId** — the aggregate identifier
    * **data** — event-specific payload
    * **timestamp** — when the state change occurred

    ```json
    {
      "type": "order.created",
      "orderId": "ORD-1001",
      "data": { "customer": "C-500", "items": [{"sku": "WIDGET-A", "qty": 2, "price": 49.99}] },
      "timestamp": "2026-03-26T10:00:00Z"
    }
    ```
  </Step>

  <Step>
    ### Publish Order Events [#publish-order-events]

    Store a series of state changes for an order. Each event is immutable once stored.

    <Tabs groupId="language" items="['Go', 'Python', 'Node.js', 'Java', 'C#', 'Kotlin', 'C++', 'Rust', 'Ruby', 'Elixir']">
      <Tab value="Go">
        ```go title="order_service.go"
        package main

        import (
            "context"
            "fmt"
            "log"

            "github.com/kubemq-io/kubemq-go/v2"
        )

        func main() {
            ctx := context.Background()
            client, err := kubemq.NewClient(ctx,
                kubemq.WithAddress("localhost", 50000),
            )
            if err != nil {
                log.Fatal(err)
            }
            defer client.Close()

            channel := "order.ORD-1001"
            events := []string{
                `{"type":"order.created","orderId":"ORD-1001","data":{"customer":"C-500","total":149.99}}`,
                `{"type":"order.paid","orderId":"ORD-1001","data":{"method":"credit_card","txId":"TX-789"}}`,
                `{"type":"order.shipped","orderId":"ORD-1001","data":{"carrier":"fedex","tracking":"FX-123"}}`,
            }

            for _, body := range events {
                result, err := client.SendEventStore(ctx, kubemq.NewEvent().
                    SetChannel(channel).
                    SetBody([]byte(body)),
                )
                if err != nil {
                    log.Fatal(err)
                }
                log.Printf("Stored: %s (seq: %s)", body[:40], result.EventID)
            }
        }
        ```
      </Tab>

      <Tab value="Python">
        ```python title="order_service.py"
        import json
        from kubemq import PubSubClient, EventStoreMessage

        events = [
            {"type": "order.created", "orderId": "ORD-1001",
             "data": {"customer": "C-500", "total": 149.99}},
            {"type": "order.paid", "orderId": "ORD-1001",
             "data": {"method": "credit_card", "txId": "TX-789"}},
            {"type": "order.shipped", "orderId": "ORD-1001",
             "data": {"carrier": "fedex", "tracking": "FX-123"}},
        ]

        with PubSubClient(address="localhost:50000") as client:
            for event in events:
                result = client.publish_event_store(
                    EventStoreMessage(
                        channel="order.ORD-1001",
                        body=json.dumps(event).encode("utf-8"),
                    )
                )
                print(f"Stored: {event['type']} (ID: {result.id})")
        ```
      </Tab>

      <Tab value="Node.js">
        ```typescript title="order_service.ts"
        import { KubeMQClient, createEventStoreMessage } from 'kubemq-js';

        const client = await KubeMQClient.create({ address: 'localhost:50000' });

        const events = [
          { type: 'order.created', orderId: 'ORD-1001', data: { customer: 'C-500', total: 149.99 } },
          { type: 'order.paid', orderId: 'ORD-1001', data: { method: 'credit_card', txId: 'TX-789' } },
          { type: 'order.shipped', orderId: 'ORD-1001', data: { carrier: 'fedex', tracking: 'FX-123' } },
        ];

        for (const event of events) {
          await client.sendEventStore(
            createEventStoreMessage({
              channel: 'order.ORD-1001',
              body: JSON.stringify(event),
            })
          );
          console.log(`Stored: ${event.type}`);
        }
        ```
      </Tab>

      <Tab value="Java">
        ```java title="OrderService.java"
        PubSubClient client = PubSubClient.builder()
            .address("localhost:50000")
            .clientId("order-service")
            .build();

        String[] events = {
            "{\"type\":\"order.created\",\"orderId\":\"ORD-1001\",\"data\":{\"customer\":\"C-500\",\"total\":149.99}}",
            "{\"type\":\"order.paid\",\"orderId\":\"ORD-1001\",\"data\":{\"method\":\"credit_card\"}}",
            "{\"type\":\"order.shipped\",\"orderId\":\"ORD-1001\",\"data\":{\"carrier\":\"fedex\"}}"
        };

        for (String body : events) {
            client.sendEventsStoreMessage(EventStoreMessage.builder()
                .channel("order.ORD-1001")
                .body(body.getBytes())
                .build());
            System.out.println("Stored event");
        }
        client.close();
        ```
      </Tab>

      <Tab value="C#">
        ```csharp title="OrderService.cs"
        await using var client = new KubeMQClient(new KubeMQClientOptions());
        await client.ConnectAsync();

        string[] events = {
            "{\"type\":\"order.created\",\"orderId\":\"ORD-1001\",\"data\":{\"customer\":\"C-500\",\"total\":149.99}}",
            "{\"type\":\"order.paid\",\"orderId\":\"ORD-1001\",\"data\":{\"method\":\"credit_card\"}}",
            "{\"type\":\"order.shipped\",\"orderId\":\"ORD-1001\",\"data\":{\"carrier\":\"fedex\"}}"
        };

        foreach (var body in events)
        {
            await client.SendEventStoreAsync(new EventStoreMessage
            {
                Channel = "order.ORD-1001",
                Body = Encoding.UTF8.GetBytes(body),
            });
            Console.WriteLine("Stored event");
        }
        ```
      </Tab>

      <Tab value="Kotlin">
        ```kotlin title="OrderService.kt"
        val client = KubeMQClient.pubSub {
            address = "localhost:50000"
            clientId = "order-service"
        }

        val events = listOf(
            """{"type":"order.created","orderId":"ORD-1001","data":{"customer":"C-500","total":149.99}}""",
            """{"type":"order.paid","orderId":"ORD-1001","data":{"method":"credit_card"}}""",
            """{"type":"order.shipped","orderId":"ORD-1001","data":{"carrier":"fedex"}}""",
        )

        client.use {
            for (body in events) {
                client.sendEventStore(eventStoreMessage {
                    channel = "order.ORD-1001"
                    this.body = body.toByteArray()
                })
                println("Stored event")
            }
        }
        ```
      </Tab>

      <Tab value="C++">
        ```cpp title="order_service.cc"
        kubemq::ClientOptions options;
        options.set_address("localhost", 50000);
        options.set_client_id("order-service");
        auto client = kubemq::Client::Create(options).value();

        std::vector<std::string> events = {
            R"({"type":"order.created","orderId":"ORD-1001","data":{"customer":"C-500","total":149.99}})",
            R"({"type":"order.paid","orderId":"ORD-1001","data":{"method":"credit_card"}})",
            R"({"type":"order.shipped","orderId":"ORD-1001","data":{"carrier":"fedex"}})"
        };

        for (const auto& body : events) {
            kubemq::EventStoreMessage msg;
            msg.set_channel("order.ORD-1001");
            msg.set_body(body);
            client->SendEventStore(msg);
            std::cout << "Stored event" << std::endl;
        }
        ```
      </Tab>

      <Tab value="Rust">
        ```rust title="order_service.rs"
        use kubemq::prelude::*;
        use kubemq::EventStoreBuilder;

        #[tokio::main]
        async fn main() -> kubemq::Result<()> {
            let client = KubemqClient::builder()
                .host("localhost")
                .port(50000)
                .build()
                .await?;

            let channel = "order.ORD-1001";
            let events = [
                r#"{"type":"order.created","orderId":"ORD-1001","data":{"customer":"C-500","total":149.99}}"#,
                r#"{"type":"order.paid","orderId":"ORD-1001","data":{"method":"credit_card","txId":"TX-789"}}"#,
                r#"{"type":"order.shipped","orderId":"ORD-1001","data":{"carrier":"fedex","tracking":"FX-123"}}"#,
            ];

            for body in events {
                let event = EventStoreBuilder::new()
                    .channel(channel)
                    .body(body.as_bytes().to_vec())
                    .build();
                let result = client.send_event_store(event).await?;
                println!("Stored: id={}, sent={}", result.id, result.sent);
            }

            client.close().await?;
            Ok(())
        }
        ```
      </Tab>

      <Tab value="Ruby">
        ```ruby title="order_service.rb"
        require 'kubemq'

        client = KubeMQ::PubSubClient.new(address: 'localhost:50000', client_id: 'order-service')

        channel = 'order.ORD-1001'
        events = [
          '{"type":"order.created","orderId":"ORD-1001","data":{"customer":"C-500","total":149.99}}',
          '{"type":"order.paid","orderId":"ORD-1001","data":{"method":"credit_card","txId":"TX-789"}}',
          '{"type":"order.shipped","orderId":"ORD-1001","data":{"carrier":"fedex","tracking":"FX-123"}}'
        ]

        events.each do |body|
          result = client.send_event_store(
            KubeMQ::PubSub::EventStoreMessage.new(channel: channel, body: body)
          )
          puts "Stored event: sent=#{result.sent}"
        end

        client.close
        ```
      </Tab>

      <Tab value="Elixir">
        ```elixir title="order_service.exs"
        channel = "order.ORD-1001"
        {:ok, client} = KubeMQ.Client.start_link(address: "localhost:50000", client_id: "order-service")

        events = [
          ~s({"type":"order.created","orderId":"ORD-1001","data":{"customer":"C-500","total":149.99}}),
          ~s({"type":"order.paid","orderId":"ORD-1001","data":{"method":"credit_card","txId":"TX-789"}}),
          ~s({"type":"order.shipped","orderId":"ORD-1001","data":{"carrier":"fedex","tracking":"FX-123"}})
        ]

        for body <- events do
          {:ok, result} =
            KubeMQ.Client.send_event_store(client, KubeMQ.EventStore.new(channel: channel, body: body))

          IO.puts("Stored event: sent=#{result.sent}")
        end

        KubeMQ.Client.close(client)
        ```
      </Tab>
    </Tabs>
  </Step>

  <Step>
    ### Rebuild State from Event History [#rebuild-state-from-event-history]

    Subscribe with `StartFromFirst` to replay all events and compute the current order state.

    <Tabs groupId="language" items="['Go', 'Python', 'Node.js', 'Java', 'C#', 'Kotlin', 'C++', 'Rust', 'Ruby', 'Elixir']">
      <Tab value="Go">
        ```go title="state_rebuilder.go"
        package main

        import (
            "context"
            "encoding/json"
            "fmt"
            "log"
            "sync"
            "time"

            "github.com/kubemq-io/kubemq-go/v2"
        )

        type OrderState struct {
            OrderID  string
            Status   string
            Customer string
            Total    float64
            Carrier  string
            Tracking string
        }

        func main() {
            ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
            defer cancel()

            client, err := kubemq.NewClient(ctx,
                kubemq.WithAddress("localhost", 50000),
            )
            if err != nil {
                log.Fatal(err)
            }
            defer client.Close()

            var mu sync.Mutex
            order := OrderState{OrderID: "ORD-1001"}

            sub, err := client.SubscribeToEventsStore(ctx, "order.ORD-1001", "",
                kubemq.StartFromFirst(),
                kubemq.WithOnEvent(func(event *kubemq.Event) {
                    mu.Lock()
                    defer mu.Unlock()

                    var evt map[string]interface{}
                    json.Unmarshal(event.Body, &evt)
                    eventType := evt["type"].(string)
                    data := evt["data"].(map[string]interface{})

                    switch eventType {
                    case "order.created":
                        order.Status = "created"
                        order.Customer = data["customer"].(string)
                        order.Total = data["total"].(float64)
                    case "order.paid":
                        order.Status = "paid"
                    case "order.shipped":
                        order.Status = "shipped"
                        order.Carrier = data["carrier"].(string)
                    }

                    fmt.Printf("seq=%d %s -> status=%s\n", event.Sequence, eventType, order.Status)
                }),
                kubemq.WithOnError(func(err error) { log.Println("Error:", err) }),
            )
            if err != nil {
                log.Fatal(err)
            }
            defer sub.Unsubscribe()

            <-ctx.Done()
            fmt.Printf("\nOrder State: %+v\n", order)
        }
        ```
      </Tab>

      <Tab value="Python">
        ```python title="state_rebuilder.py"
        import json
        import time
        from kubemq import (
            PubSubClient, EventsStoreSubscription,
            EventStoreStartPosition, CancellationToken,
        )

        order = {"orderId": "ORD-1001", "status": "unknown"}

        def on_event(event):
            global order
            evt = json.loads(event.body.decode("utf-8"))
            event_type = evt["type"]
            data = evt.get("data", {})

            if event_type == "order.created":
                order["status"] = "created"
                order["customer"] = data.get("customer")
                order["total"] = data.get("total")
            elif event_type == "order.paid":
                order["status"] = "paid"
            elif event_type == "order.shipped":
                order["status"] = "shipped"
                order["carrier"] = data.get("carrier")

            print(f"seq={event.sequence} {event_type} -> status={order['status']}")

        with PubSubClient(address="localhost:50000") as client:
            client.subscribe_to_events_store(
                subscription=EventsStoreSubscription(
                    channel="order.ORD-1001",
                    start_position=EventStoreStartPosition.StartFromFirst,
                    on_receive_event_callback=on_event,
                    on_error_callback=lambda e: print(f"Error: {e}"),
                ),
                cancel=CancellationToken(),
            )
            time.sleep(5)
            print(f"\nOrder State: {order}")
        ```
      </Tab>

      <Tab value="Node.js">
        ```typescript title="state_rebuilder.ts"
        import { KubeMQClient, EventStoreStartPosition } from 'kubemq-js';

        const client = await KubeMQClient.create({ address: 'localhost:50000' });

        const order: Record<string, unknown> = { orderId: 'ORD-1001', status: 'unknown' };

        client.subscribeToEventsStore({
          channel: 'order.ORD-1001',
          startFrom: EventStoreStartPosition.StartFromFirst,
          onEvent: (msg) => {
            const evt = JSON.parse(new TextDecoder().decode(msg.body));
            const { type, data } = evt;

            if (type === 'order.created') {
              Object.assign(order, { status: 'created', customer: data.customer, total: data.total });
            } else if (type === 'order.paid') {
              order.status = 'paid';
            } else if (type === 'order.shipped') {
              Object.assign(order, { status: 'shipped', carrier: data.carrier });
            }
            console.log(`seq=${msg.sequence} ${type} -> status=${order.status}`);
          },
          onError: (err) => console.error('Error:', err.message),
        });

        setTimeout(() => console.log('\nOrder State:', order), 5000);
        ```
      </Tab>

      <Tab value="Java">
        ```java title="StateRebuilder.java"
        PubSubClient client = PubSubClient.builder()
            .address("localhost:50000")
            .clientId("state-rebuilder")
            .build();

        Map<String, Object> order = new ConcurrentHashMap<>();
        order.put("orderId", "ORD-1001");

        client.subscribeToEventsStore(EventsStoreSubscription.builder()
            .channel("order.ORD-1001")
            .startPosition(EventStoreStartPosition.StartFromFirst)
            .onReceiveEventCallback(event -> {
                var evt = new ObjectMapper().readTree(event.getBody());
                String type = evt.get("type").asText();
                if ("order.created".equals(type)) {
                    order.put("status", "created");
                    order.put("total", evt.get("data").get("total").asDouble());
                } else if ("order.paid".equals(type)) {
                    order.put("status", "paid");
                } else if ("order.shipped".equals(type)) {
                    order.put("status", "shipped");
                }
                System.out.printf("seq=%d %s -> status=%s%n",
                    event.getSequence(), type, order.get("status"));
            })
            .onErrorCallback(err -> System.err.println(err.getMessage()))
            .build());

        Thread.sleep(5000);
        System.out.println("\nOrder State: " + order);
        client.close();
        ```
      </Tab>

      <Tab value="C#">
        ```csharp title="StateRebuilder.cs"
        await using var client = new KubeMQClient(new KubeMQClientOptions());
        await client.ConnectAsync();

        var order = new Dictionary<string, object> { ["orderId"] = "ORD-1001" };

        await foreach (var msg in client.SubscribeToEventsStoreAsync(
            new EventsStoreSubscription
            {
                Channel = "order.ORD-1001",
                StartPosition = EventStoreStartPosition.StartFromFirst,
            }))
        {
            var evt = JsonSerializer.Deserialize<JsonElement>(msg.Body.Span);
            var type = evt.GetProperty("type").GetString()!;

            order["status"] = type switch
            {
                "order.created" => "created",
                "order.paid" => "paid",
                "order.shipped" => "shipped",
                _ => order.GetValueOrDefault("status", "unknown")!
            };
            Console.WriteLine($"seq={msg.Sequence} {type} -> status={order["status"]}");
        }
        ```
      </Tab>

      <Tab value="Kotlin">
        ```kotlin title="StateRebuilder.kt"
        val client = KubeMQClient.pubSub {
            address = "localhost:50000"
            clientId = "state-rebuilder"
        }

        data class OrderState(var status: String = "unknown", var total: Double = 0.0)
        val order = OrderState()

        client.use {
            client.subscribeToEventsStore {
                channel = "order.ORD-1001"
                startPosition = StartPosition.StartFromFirst
            }.collect { msg ->
                val evt = JSONObject(String(msg.body))
                val type = evt.getString("type")
                when (type) {
                    "order.created" -> { order.status = "created"; order.total = evt.getJSONObject("data").getDouble("total") }
                    "order.paid" -> order.status = "paid"
                    "order.shipped" -> order.status = "shipped"
                }
                println("seq=${msg.sequence} $type -> status=${order.status}")
            }
        }
        ```
      </Tab>

      <Tab value="C++">
        ```cpp title="state_rebuilder.cc"
        kubemq::ClientOptions options;
        options.set_address("localhost", 50000);
        auto client = kubemq::Client::Create(options).value();

        std::string status = "unknown";

        client->SubscribeToEventsStore("order.ORD-1001", "",
            kubemq::StartPosition::StartFromFirst,
            [&status](const kubemq::EventStoreReceived& msg) {
                auto body = msg.body();
                if (body.find("order.created") != std::string::npos) status = "created";
                else if (body.find("order.paid") != std::string::npos) status = "paid";
                else if (body.find("order.shipped") != std::string::npos) status = "shipped";
                std::cout << "seq=" << msg.sequence() << " -> status=" << status << std::endl;
            },
            [](const std::string& err) { std::cerr << err << std::endl; });
        ```
      </Tab>

      <Tab value="Rust">
        ```rust title="state_rebuilder.rs"
        use kubemq::prelude::*;
        use kubemq::EventsStoreSubscription;
        use serde_json::Value;
        use std::sync::{Arc, Mutex};
        use std::time::Duration;

        #[tokio::main]
        async fn main() -> kubemq::Result<()> {
            let client = KubemqClient::builder()
                .host("localhost")
                .port(50000)
                .build()
                .await?;

            let status = Arc::new(Mutex::new(String::from("unknown")));
            let status_cb = status.clone();

            // Replay every stored event from the beginning to rebuild state.
            let sub = client
                .subscribe_to_events_store(
                    "order.ORD-1001",
                    "",
                    EventsStoreSubscription::StartFromFirst,
                    move |event| {
                        let status = status_cb.clone();
                        Box::pin(async move {
                            let evt: Value = serde_json::from_slice(&event.body).unwrap_or(Value::Null);
                            let event_type = evt["type"].as_str().unwrap_or("");
                            let mut s = status.lock().unwrap();
                            match event_type {
                                "order.created" => *s = "created".into(),
                                "order.paid" => *s = "paid".into(),
                                "order.shipped" => *s = "shipped".into(),
                                _ => {}
                            }
                            println!("seq={} {} -> status={}", event.sequence, event_type, *s);
                        })
                    },
                    None,
                )
                .await?;

            tokio::time::sleep(Duration::from_secs(5)).await;
            println!("Order State: status={}", *status.lock().unwrap());

            sub.unsubscribe().await;
            client.close().await?;
            Ok(())
        }
        ```
      </Tab>

      <Tab value="Ruby">
        ```ruby title="state_rebuilder.rb"
        require 'kubemq'
        require 'json'

        client = KubeMQ::PubSubClient.new(address: 'localhost:50000', client_id: 'state-rebuilder')
        cancel = KubeMQ::CancellationToken.new

        order = { 'orderId' => 'ORD-1001', 'status' => 'unknown' }

        # Replay all stored events from the first to rebuild current state.
        sub = KubeMQ::PubSub::EventsStoreSubscription.new(
          channel: 'order.ORD-1001',
          start_position: KubeMQ::PubSub::EventStoreStartPosition::START_FROM_FIRST
        )
        client.subscribe_to_events_store(sub, cancellation_token: cancel, on_error: lambda { |e|
          puts "Error: #{e.message}"
        }) do |event|
          evt = JSON.parse(event.body)
          case evt['type']
          when 'order.created' then order['status'] = 'created'
          when 'order.paid'    then order['status'] = 'paid'
          when 'order.shipped' then order['status'] = 'shipped'
          end
          puts "seq=#{event.sequence} #{evt['type']} -> status=#{order['status']}"
        end

        sleep 5
        puts "Order State: #{order}"
        cancel.cancel
        client.close
        ```
      </Tab>

      <Tab value="Elixir">
        ```elixir title="state_rebuilder.exs"
        {:ok, client} = KubeMQ.Client.start_link(address: "localhost:50000", client_id: "state-rebuilder")

        {:ok, agent} = Agent.start_link(fn -> "unknown" end)

        # Replay all stored events from the first to rebuild current state.
        {:ok, sub} =
          KubeMQ.Client.subscribe_to_events_store(client, "order.ORD-1001",
            start_at: :start_from_first,
            on_event: fn event ->
              evt = Jason.decode!(event.body)

              status =
                case evt["type"] do
                  "order.created" -> "created"
                  "order.paid" -> "paid"
                  "order.shipped" -> "shipped"
                  _ -> Agent.get(agent, & &1)
                end

              Agent.update(agent, fn _ -> status end)
              IO.puts("seq=#{event.sequence} #{evt["type"]} -> status=#{status}")
            end
          )

        Process.sleep(5_000)
        IO.puts("Order State: status=#{Agent.get(agent, & &1)}")

        KubeMQ.Subscription.cancel(sub)
        KubeMQ.Client.close(client)
        ```
      </Tab>
    </Tabs>

    **Expected output:**

    ```text
    seq=1 order.created -> status=created
    seq=2 order.paid -> status=paid
    seq=3 order.shipped -> status=shipped

    Order State: {orderId=ORD-1001, status=shipped, customer=C-500, total=149.99}
    ```
  </Step>

  <Step>
    ### Checkpoint-Based Recovery [#checkpoint-based-recovery]

    For aggregates with long histories, save the last processed sequence number as a checkpoint. On restart, subscribe from that sequence instead of replaying the entire history.

    <Mermaid
      chart="graph LR
    DB[(&#x22;Checkpoint DB<br/>seq=50, state=...&#x22;)]
    APP[&#x22;Application&#x22;]
    ES[(&#x22;Event Store&#x22;)]

    DB -- load --> APP
    APP -- &#x22;StartAtSequence 51&#x22; --> ES
    ES -. &#x22;events 51-75&#x22; .-> APP
    APP -- &#x22;save seq=75&#x22; --> DB

    class APP client
    class ES store
    class DB data"
    />

    *A checkpoint records the last processed sequence so restarts replay only the unprocessed tail instead of the full history.*

    The subscriber uses `StartAtSequence(lastCheckpoint + 1)` to pick up only unprocessed events.
  </Step>
</Steps>

## Event Sourcing Best Practices [#event-sourcing-best-practices]

### Channel Per Aggregate [#channel-per-aggregate]

Use one channel per aggregate instance (e.g., `order.ORD-1001`, `order.ORD-1002`). This provides independent sequencing and clean replay per entity.

### Immutable Events [#immutable-events]

Never modify or delete events. The event stream is the source of truth. To correct an error, append a compensating event (e.g., an `order.refunded` event).

### Snapshots for Performance [#snapshots-for-performance]

For aggregates with long histories, periodically save a **snapshot** (the computed state at a sequence number). On startup, load the snapshot and replay only events after the snapshot sequence.

### Event Schema Versioning [#event-schema-versioning]

Include a `version` field in your event schema. When the schema changes, handle both old and new formats in your event handler.

<Callout type="info">
  For event sourcing, configure retention to unlimited (`Store.MaxRetention=0`) or set it longer than your maximum replay window. See [Configure Retention](/learn/events-store/how-to/configure-retention).
</Callout>

## Next Steps [#next-steps]

* Configure [retention policies](/learn/events-store/how-to/configure-retention) for long-lived streams
* Scale processing with [consumer groups](/learn/events-store/tutorials/consumer-groups)
* Build an [audit trail](/learn/events-store/scenarios/audit-trail) with Events Store
* See the [Events Store Reference](/learn/events-store/reference) for configuration options