# Replay Events from Any Point (/learn/events-store/tutorials/replay-events)



Events Store supports six subscription start positions that control where a subscriber begins reading from the event stream. This tutorial demonstrates each replay strategy with practical examples.

## Subscription Start Positions [#subscription-start-positions]

<Mermaid
  chart="graph LR
  STORE{{&#x22;Events Store channel<br/>orders.events&#x22;}}
  E1[(&#x22;seq=1<br/>10:00&#x22;)]
  E2[(&#x22;seq=2<br/>10:01&#x22;)]
  E3[(&#x22;seq=3<br/>10:02&#x22;)]
  E4[(&#x22;seq=4<br/>10:03&#x22;)]
  E5[(&#x22;seq=5<br/>10:04&#x22;)]
  SF[&#x22;StartFromFirst&#x22;]
  SS[&#x22;StartAtSequence(3)&#x22;]
  SL[&#x22;StartFromLast&#x22;]
  SN[&#x22;StartNewOnly&#x22;]

  STORE --- E1 --- E2 --- E3 --- E4 --- E5
  SF -. &#x22;starts at seq=1&#x22; .-> E1
  SS -. &#x22;starts at seq=3&#x22; .-> E3
  SL -. &#x22;starts at seq=5&#x22; .-> E5
  SN -. &#x22;waits for seq=6+&#x22; .-> E5

  class STORE store
  class E1,E2,E3,E4,E5 data
  class SF,SS,SL,SN client"
/>

*Each start position drops a subscriber at a different point in the stored stream: `StartFromFirst` rewinds to seq=1, `StartAtSequence` resumes at a chosen offset, `StartFromLast` catches the latest, and `StartNewOnly` ignores history and waits for new events.*

| Start Position     | Enum Value | Description                                            |
| ------------------ | ---------- | ------------------------------------------------------ |
| `StartNewOnly`     | 1          | Only events published **after** subscribing            |
| `StartFromFirst`   | 2          | Replay **all** events from the beginning               |
| `StartFromLast`    | 3          | Start from the **most recent** stored event            |
| `StartAtSequence`  | 4          | Start from a **specific sequence number**              |
| `StartAtTime`      | 5          | Start from a **specific timestamp** (Unix nanoseconds) |
| `StartAtTimeDelta` | 6          | Start from **N seconds ago**                           |

## 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>
    ### Seed the Event Stream [#seed-the-event-stream]

    Publish 10 order events to create a history for replay.

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

        import (
            "context"
            "fmt"
            "log"
            "time"

            "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()

            for i := 1; i <= 10; i++ {
                body := fmt.Sprintf(`{"orderId":"ORD-%04d","status":"created","total":%.2f}`,
                    i, float64(i)*29.99)
                result, err := client.SendEventStore(ctx, kubemq.NewEvent().
                    SetChannel("orders.events").
                    SetBody([]byte(body)),
                )
                if err != nil {
                    log.Fatal(err)
                }
                log.Printf("Stored seq=%s: ORD-%04d", result.EventID, i)
                time.Sleep(1 * time.Second)
            }
        }
        ```
      </Tab>

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

        with PubSubClient(address="localhost:50000") as client:
            for i in range(1, 11):
                body = json.dumps({"orderId": f"ORD-{i:04d}", "status": "created",
                                   "total": round(i * 29.99, 2)})
                result = client.publish_event_store(
                    EventStoreMessage(channel="orders.events", body=body.encode("utf-8"))
                )
                print(f"Stored: ORD-{i:04d}")
                time.sleep(1)
        ```
      </Tab>

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

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

        for (let i = 1; i <= 10; i++) {
          await client.sendEventStore(
            createEventStoreMessage({
              channel: 'orders.events',
              body: JSON.stringify({
                orderId: `ORD-${String(i).padStart(4, '0')}`,
                status: 'created',
                total: +(i * 29.99).toFixed(2),
              }),
            })
          );
          console.log(`Stored: ORD-${String(i).padStart(4, '0')}`);
          await new Promise((r) => setTimeout(r, 1000));
        }
        ```
      </Tab>

      <Tab value="Java">
        ```java title="SeedOrders.java"
        PubSubClient client = PubSubClient.builder()
            .address("localhost:50000")
            .clientId("seeder")
            .build();

        for (int i = 1; i <= 10; i++) {
            String body = String.format(
                "{\"orderId\":\"ORD-%04d\",\"status\":\"created\",\"total\":%.2f}", i, i * 29.99);
            client.sendEventsStoreMessage(
                EventStoreMessage.builder()
                    .channel("orders.events")
                    .body(body.getBytes())
                    .build());
            System.out.printf("Stored: ORD-%04d%n", i);
            Thread.sleep(1000);
        }
        client.close();
        ```
      </Tab>

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

        for (var i = 1; i <= 10; i++)
        {
            var body = $"{{\"orderId\":\"ORD-{i:D4}\",\"status\":\"created\",\"total\":{i * 29.99:F2}}}";
            await client.SendEventStoreAsync(new EventStoreMessage
            {
                Channel = "orders.events",
                Body = Encoding.UTF8.GetBytes(body),
            });
            Console.WriteLine($"Stored: ORD-{i:D4}");
            await Task.Delay(1000);
        }
        ```
      </Tab>

      <Tab value="Kotlin">
        ```kotlin title="SeedOrders.kt"
        val client = KubeMQClient.pubSub {
            address = "localhost:50000"
            clientId = "seeder"
        }

        client.use {
            for (i in 1..10) {
                val body = """{"orderId":"ORD-${"%04d".format(i)}","status":"created","total":${"%.2f".format(i * 29.99)}}"""
                client.sendEventStore(eventStoreMessage {
                    channel = "orders.events"
                    this.body = body.toByteArray()
                })
                println("Stored: ORD-${"%04d".format(i)}")
                delay(1000)
            }
        }
        ```
      </Tab>

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

        for (int i = 1; i <= 10; ++i) {
            kubemq::EventStoreMessage msg;
            msg.set_channel("orders.events");
            msg.set_body("{\"orderId\":\"ORD-" + std::to_string(i) + "\",\"status\":\"created\"}");
            client->SendEventStore(msg);
            std::cout << "Stored: ORD-" << i << std::endl;
            std::this_thread::sleep_for(std::chrono::seconds(1));
        }
        ```
      </Tab>

      <Tab value="Rust">
        ```rust title="seed_orders.rs"
        use kubemq::prelude::*;
        use kubemq::EventStoreBuilder;
        use std::time::Duration;

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

            for i in 1..=10 {
                let body = format!(
                    r#"{{"orderId":"ORD-{:04}","status":"created","total":{:.2}}}"#,
                    i,
                    i as f64 * 29.99
                );
                let event = EventStoreBuilder::new()
                    .channel("orders.events")
                    .body(body.into_bytes())
                    .build();
                let result = client.send_event_store(event).await?;
                println!("Stored seq=ORD-{:04}: sent={}", i, result.sent);
                tokio::time::sleep(Duration::from_secs(1)).await;
            }

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

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

        client = KubeMQ::PubSubClient.new(address: 'localhost:50000', client_id: 'seeder')

        (1..10).each do |i|
          body = { orderId: format('ORD-%04d', i), status: 'created',
                   total: (i * 29.99).round(2) }.to_json
          result = client.send_event_store(
            KubeMQ::PubSub::EventStoreMessage.new(channel: 'orders.events', body: body)
          )
          puts "Stored: ORD-#{format('%04d', i)} sent=#{result.sent}"
          sleep 1
        end

        client.close
        ```
      </Tab>

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

        for i <- 1..10 do
          body =
            Jason.encode!(%{
              orderId: "ORD-#{String.pad_leading(Integer.to_string(i), 4, "0")}",
              status: "created",
              total: Float.round(i * 29.99, 2)
            })

          {:ok, _} =
            KubeMQ.Client.send_event_store(
              client,
              KubeMQ.EventStore.new(channel: "orders.events", body: body)
            )

          IO.puts("Stored: ORD-#{String.pad_leading(Integer.to_string(i), 4, "0")}")
          Process.sleep(1_000)
        end

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

  <Step>
    ### Replay from Beginning (StartFromFirst) [#replay-from-beginning-startfromfirst]

    Receive every event ever stored in the channel, starting from sequence 1.

    <Tabs groupId="language" items="['Go', 'Python', 'Node.js', 'Java', 'C#', 'Kotlin', 'C++', 'Rust', 'Ruby', 'Elixir']">
      <Tab value="Go">
        ```go title="replay_from_first.go"
        sub, err := client.SubscribeToEventsStore(ctx, "orders.events", "",
            kubemq.StartFromFirst(),
            kubemq.WithOnEvent(func(event *kubemq.Event) {
                fmt.Printf("[FromFirst] seq=%d body=%s\n",
                    event.Sequence, string(event.Body))
            }),
            kubemq.WithOnError(func(err error) { log.Println(err) }),
        )
        ```
      </Tab>

      <Tab value="Python">
        ```python title="replay_from_first.py"
        client.subscribe_to_events_store(
            subscription=EventsStoreSubscription(
                channel="orders.events",
                start_position=EventStoreStartPosition.StartFromFirst,
                on_receive_event_callback=lambda e: print(
                    f"[FromFirst] seq={e.sequence} body={e.body.decode('utf-8')}"
                ),
                on_error_callback=lambda e: print(f"Error: {e}"),
            ),
            cancel=CancellationToken(),
        )
        ```
      </Tab>

      <Tab value="Node.js">
        ```typescript title="replay_from_first.ts"
        client.subscribeToEventsStore({
          channel: 'orders.events',
          startPosition: EventStoreStartPosition.StartFromFirst,
          onEvent: (msg) =>
            console.log(`[FromFirst] seq=${msg.sequence} body=${new TextDecoder().decode(msg.body)}`),
          onError: (err) => console.error(err.message),
        });
        ```
      </Tab>

      <Tab value="Java">
        ```java title="ReplayFromFirst.java"
        client.subscribeToEventsStore(EventsStoreSubscription.builder()
            .channel("orders.events")
            .startPosition(EventStoreStartPosition.StartFromFirst)
            .onReceiveEventCallback(event ->
                System.out.printf("[FromFirst] seq=%d body=%s%n",
                    event.getSequence(), new String(event.getBody())))
            .onErrorCallback(err -> System.err.println(err.getMessage()))
            .build());
        ```
      </Tab>

      <Tab value="C#">
        ```csharp title="ReplayFromFirst.cs"
        await foreach (var msg in client.SubscribeToEventsStoreAsync(
            new EventsStoreSubscription
            {
                Channel = "orders.events",
                StartPosition = EventStoreStartPosition.StartFromFirst,
            }))
        {
            Console.WriteLine($"[FromFirst] seq={msg.Sequence} body={Encoding.UTF8.GetString(msg.Body.Span)}");
        }
        ```
      </Tab>

      <Tab value="Kotlin">
        ```kotlin title="ReplayFromFirst.kt"
        client.subscribeToEventsStore {
            channel = "orders.events"
            startPosition = StartPosition.StartFromFirst
        }.collect { msg ->
            println("[FromFirst] seq=${msg.sequence} body=${String(msg.body)}")
        }
        ```
      </Tab>

      <Tab value="C++">
        ```cpp title="replay_from_first.cc"
        client->SubscribeToEventsStore("orders.events", "",
            kubemq::StartPosition::StartFromFirst,
            [](const kubemq::EventStoreReceived& msg) {
                std::cout << "[FromFirst] seq=" << msg.sequence()
                          << " body=" << msg.body() << std::endl;
            },
            [](const std::string& err) { std::cerr << err << std::endl; });
        ```
      </Tab>

      <Tab value="Rust">
        ```rust title="replay_from_first.rs"
        use kubemq::EventsStoreSubscription;

        let sub = client
            .subscribe_to_events_store(
                "orders.events",
                "",
                EventsStoreSubscription::StartFromFirst,
                |event| {
                    Box::pin(async move {
                        println!(
                            "[FromFirst] seq={} body={}",
                            event.sequence,
                            String::from_utf8_lossy(&event.body)
                        );
                    })
                },
                None,
            )
            .await?;
        ```
      </Tab>

      <Tab value="Ruby">
        ```ruby title="replay_from_first.rb"
        cancel = KubeMQ::CancellationToken.new

        sub = KubeMQ::PubSub::EventsStoreSubscription.new(
          channel: 'orders.events',
          start_position: KubeMQ::PubSub::EventStoreStartPosition::START_FROM_FIRST
        )
        client.subscribe_to_events_store(sub, cancellation_token: cancel,
                                         on_error: ->(e) { puts "Error: #{e.message}" }) do |event|
          puts "[FromFirst] seq=#{event.sequence} body=#{event.body}"
        end
        ```
      </Tab>

      <Tab value="Elixir">
        ```elixir title="replay_from_first.exs"
        {:ok, sub} =
          KubeMQ.Client.subscribe_to_events_store(client, "orders.events",
            start_at: :start_from_first,
            on_event: fn event ->
              IO.puts("[FromFirst] seq=#{event.sequence} body=#{event.body}")
            end
          )
        ```
      </Tab>
    </Tabs>

    **Output:** receives all 10 events (seq 1-10).
  </Step>

  <Step>
    ### Start from a Specific Sequence (StartAtSequence) [#start-from-a-specific-sequence-startatsequence]

    Resume from sequence number 7 to receive events 7-10 plus any new events.

    <Tabs groupId="language" items="['Go', 'Python', 'Node.js', 'Java', 'C#', 'Kotlin', 'C++', 'Rust', 'Ruby', 'Elixir']">
      <Tab value="Go">
        ```go title="replay_from_sequence.go"
        sub, err := client.SubscribeToEventsStore(ctx, "orders.events", "",
            kubemq.StartAtSequence(7),
            kubemq.WithOnEvent(func(event *kubemq.Event) {
                fmt.Printf("[AtSeq7] seq=%d body=%s\n",
                    event.Sequence, string(event.Body))
            }),
            kubemq.WithOnError(func(err error) { log.Println(err) }),
        )
        ```
      </Tab>

      <Tab value="Python">
        ```python title="replay_from_sequence.py"
        client.subscribe_to_events_store(
            subscription=EventsStoreSubscription(
                channel="orders.events",
                start_position=EventStoreStartPosition.StartAtSequence,
                start_position_value=7,
                on_receive_event_callback=lambda e: print(
                    f"[AtSeq7] seq={e.sequence} body={e.body.decode('utf-8')}"
                ),
                on_error_callback=lambda e: print(f"Error: {e}"),
            ),
            cancel=CancellationToken(),
        )
        ```
      </Tab>

      <Tab value="Node.js">
        ```typescript title="replay_from_sequence.ts"
        client.subscribeToEventsStore({
          channel: 'orders.events',
          startPosition: EventStoreStartPosition.StartAtSequence,
          startPositionValue: 7,
          onEvent: (msg) =>
            console.log(`[AtSeq7] seq=${msg.sequence} body=${new TextDecoder().decode(msg.body)}`),
          onError: (err) => console.error(err.message),
        });
        ```
      </Tab>

      <Tab value="Java">
        ```java title="ReplayFromSequence.java"
        client.subscribeToEventsStore(EventsStoreSubscription.builder()
            .channel("orders.events")
            .startPosition(EventStoreStartPosition.StartAtSequence)
            .startPositionValue(7)
            .onReceiveEventCallback(event ->
                System.out.printf("[AtSeq7] seq=%d body=%s%n",
                    event.getSequence(), new String(event.getBody())))
            .onErrorCallback(err -> System.err.println(err.getMessage()))
            .build());
        ```
      </Tab>

      <Tab value="C#">
        ```csharp title="ReplayFromSequence.cs"
        await foreach (var msg in client.SubscribeToEventsStoreAsync(
            new EventsStoreSubscription
            {
                Channel = "orders.events",
                StartPosition = EventStoreStartPosition.StartAtSequence,
                StartPositionValue = 7,
            }))
        {
            Console.WriteLine($"[AtSeq7] seq={msg.Sequence} body={Encoding.UTF8.GetString(msg.Body.Span)}");
        }
        ```
      </Tab>

      <Tab value="Kotlin">
        ```kotlin title="ReplayFromSequence.kt"
        client.subscribeToEventsStore {
            channel = "orders.events"
            startPosition = StartPosition.StartAtSequence
            startPositionValue = 7
        }.collect { msg ->
            println("[AtSeq7] seq=${msg.sequence} body=${String(msg.body)}")
        }
        ```
      </Tab>

      <Tab value="C++">
        ```cpp title="replay_from_sequence.cc"
        client->SubscribeToEventsStore("orders.events", "",
            kubemq::StartPosition::StartAtSequence, 7,
            [](const kubemq::EventStoreReceived& msg) {
                std::cout << "[AtSeq7] seq=" << msg.sequence()
                          << " body=" << msg.body() << std::endl;
            },
            [](const std::string& err) { std::cerr << err << std::endl; });
        ```
      </Tab>

      <Tab value="Rust">
        ```rust title="replay_from_sequence.rs"
        use kubemq::EventsStoreSubscription;

        let sub = client
            .subscribe_to_events_store(
                "orders.events",
                "",
                EventsStoreSubscription::StartAtSequence(7),
                |event| {
                    Box::pin(async move {
                        println!(
                            "[AtSeq7] seq={} body={}",
                            event.sequence,
                            String::from_utf8_lossy(&event.body)
                        );
                    })
                },
                None,
            )
            .await?;
        ```
      </Tab>

      <Tab value="Ruby">
        ```ruby title="replay_from_sequence.rb"
        cancel = KubeMQ::CancellationToken.new

        sub = KubeMQ::PubSub::EventsStoreSubscription.new(
          channel: 'orders.events',
          start_position: KubeMQ::PubSub::EventStoreStartPosition::START_AT_SEQUENCE,
          start_position_value: 7
        )
        client.subscribe_to_events_store(sub, cancellation_token: cancel,
                                         on_error: ->(e) { puts "Error: #{e.message}" }) do |event|
          puts "[AtSeq7] seq=#{event.sequence} body=#{event.body}"
        end
        ```
      </Tab>

      <Tab value="Elixir">
        ```elixir title="replay_from_sequence.exs"
        {:ok, sub} =
          KubeMQ.Client.subscribe_to_events_store(client, "orders.events",
            start_at: {:start_at_sequence, 7},
            on_event: fn event ->
              IO.puts("[AtSeq7] seq=#{event.sequence} body=#{event.body}")
            end
          )
        ```
      </Tab>
    </Tabs>

    **Output:** receives events with seq 7, 8, 9, 10, then waits for new events.
  </Step>

  <Step>
    ### Start from a Time Delta (StartAtTimeDelta) [#start-from-a-time-delta-startattimedelta]

    Receive events published in the last 30 seconds.

    <Tabs groupId="language" items="['Go', 'Python', 'Node.js', 'Java', 'C#', 'Kotlin', 'C++', 'Rust', 'Ruby', 'Elixir']">
      <Tab value="Go">
        ```go title="replay_time_delta.go"
        sub, err := client.SubscribeToEventsStore(ctx, "orders.events", "",
            kubemq.StartAtTimeDelta(30),
            kubemq.WithOnEvent(func(event *kubemq.Event) {
                fmt.Printf("[TimeDelta30s] seq=%d body=%s\n",
                    event.Sequence, string(event.Body))
            }),
            kubemq.WithOnError(func(err error) { log.Println(err) }),
        )
        ```
      </Tab>

      <Tab value="Python">
        ```python title="replay_time_delta.py"
        client.subscribe_to_events_store(
            subscription=EventsStoreSubscription(
                channel="orders.events",
                start_position=EventStoreStartPosition.StartAtTimeDelta,
                start_position_value=30,
                on_receive_event_callback=lambda e: print(
                    f"[TimeDelta30s] seq={e.sequence} body={e.body.decode('utf-8')}"
                ),
                on_error_callback=lambda e: print(f"Error: {e}"),
            ),
            cancel=CancellationToken(),
        )
        ```
      </Tab>

      <Tab value="Node.js">
        ```typescript title="replay_time_delta.ts"
        client.subscribeToEventsStore({
          channel: 'orders.events',
          startPosition: EventStoreStartPosition.StartAtTimeDelta,
          startPositionValue: 30,
          onEvent: (msg) =>
            console.log(`[TimeDelta30s] seq=${msg.sequence} body=${new TextDecoder().decode(msg.body)}`),
          onError: (err) => console.error(err.message),
        });
        ```
      </Tab>

      <Tab value="Java">
        ```java title="ReplayTimeDelta.java"
        client.subscribeToEventsStore(EventsStoreSubscription.builder()
            .channel("orders.events")
            .startPosition(EventStoreStartPosition.StartAtTimeDelta)
            .startPositionValue(30)
            .onReceiveEventCallback(event ->
                System.out.printf("[TimeDelta30s] seq=%d body=%s%n",
                    event.getSequence(), new String(event.getBody())))
            .onErrorCallback(err -> System.err.println(err.getMessage()))
            .build());
        ```
      </Tab>

      <Tab value="C#">
        ```csharp title="ReplayTimeDelta.cs"
        await foreach (var msg in client.SubscribeToEventsStoreAsync(
            new EventsStoreSubscription
            {
                Channel = "orders.events",
                StartPosition = EventStoreStartPosition.StartAtTimeDelta,
                StartPositionValue = 30,
            }))
        {
            Console.WriteLine($"[TimeDelta30s] seq={msg.Sequence} body={Encoding.UTF8.GetString(msg.Body.Span)}");
        }
        ```
      </Tab>

      <Tab value="Kotlin">
        ```kotlin title="ReplayTimeDelta.kt"
        client.subscribeToEventsStore {
            channel = "orders.events"
            startPosition = StartPosition.StartAtTimeDelta
            startPositionValue = 30
        }.collect { msg ->
            println("[TimeDelta30s] seq=${msg.sequence} body=${String(msg.body)}")
        }
        ```
      </Tab>

      <Tab value="C++">
        ```cpp title="replay_time_delta.cc"
        client->SubscribeToEventsStore("orders.events", "",
            kubemq::StartPosition::StartAtTimeDelta, 30,
            [](const kubemq::EventStoreReceived& msg) {
                std::cout << "[TimeDelta30s] seq=" << msg.sequence()
                          << " body=" << msg.body() << std::endl;
            },
            [](const std::string& err) { std::cerr << err << std::endl; });
        ```
      </Tab>

      <Tab value="Rust">
        ```rust title="replay_time_delta.rs"
        use kubemq::EventsStoreSubscription;
        use std::time::Duration;

        let sub = client
            .subscribe_to_events_store(
                "orders.events",
                "",
                EventsStoreSubscription::StartAtTimeDelta(Duration::from_secs(30)),
                |event| {
                    Box::pin(async move {
                        println!(
                            "[TimeDelta30s] seq={} body={}",
                            event.sequence,
                            String::from_utf8_lossy(&event.body)
                        );
                    })
                },
                None,
            )
            .await?;
        ```
      </Tab>

      <Tab value="Ruby">
        ```ruby title="replay_time_delta.rb"
        cancel = KubeMQ::CancellationToken.new

        sub = KubeMQ::PubSub::EventsStoreSubscription.new(
          channel: 'orders.events',
          start_position: KubeMQ::PubSub::EventStoreStartPosition::START_AT_TIME_DELTA,
          start_position_value: 30 # seconds
        )
        client.subscribe_to_events_store(sub, cancellation_token: cancel,
                                         on_error: ->(e) { puts "Error: #{e.message}" }) do |event|
          puts "[TimeDelta30s] seq=#{event.sequence} body=#{event.body}"
        end
        ```
      </Tab>

      <Tab value="Elixir">
        ```elixir title="replay_time_delta.exs"
        # The Elixir SDK expresses the time delta in milliseconds (30s = 30_000ms).
        {:ok, sub} =
          KubeMQ.Client.subscribe_to_events_store(client, "orders.events",
            start_at: {:start_at_time_delta, 30_000},
            on_event: fn event ->
              IO.puts("[TimeDelta30s] seq=#{event.sequence} body=#{event.body}")
            end
          )
        ```
      </Tab>
    </Tabs>

    **Output:** receives only events published within the last 30 seconds.
  </Step>

  <Step>
    ### Start from the Last Event (StartFromLast) [#start-from-the-last-event-startfromlast]

    Receive the most recently stored event, then all new events.

    <Tabs groupId="language" items="['Go', 'Python', 'Node.js', 'Java', 'C#', 'Kotlin', 'C++', 'Rust', 'Ruby', 'Elixir']">
      <Tab value="Go">
        ```go title="replay_from_last.go"
        sub, err := client.SubscribeToEventsStore(ctx, "orders.events", "",
            kubemq.StartFromLast(),
            kubemq.WithOnEvent(func(event *kubemq.Event) {
                fmt.Printf("[FromLast] seq=%d body=%s\n",
                    event.Sequence, string(event.Body))
            }),
            kubemq.WithOnError(func(err error) { log.Println(err) }),
        )
        ```
      </Tab>

      <Tab value="Python">
        ```python title="replay_from_last.py"
        client.subscribe_to_events_store(
            subscription=EventsStoreSubscription(
                channel="orders.events",
                start_position=EventStoreStartPosition.StartFromLast,
                on_receive_event_callback=lambda e: print(
                    f"[FromLast] seq={e.sequence} body={e.body.decode('utf-8')}"
                ),
                on_error_callback=lambda e: print(f"Error: {e}"),
            ),
            cancel=CancellationToken(),
        )
        ```
      </Tab>

      <Tab value="Node.js">
        ```typescript title="replay_from_last.ts"
        client.subscribeToEventsStore({
          channel: 'orders.events',
          startPosition: EventStoreStartPosition.StartFromLast,
          onEvent: (msg) =>
            console.log(`[FromLast] seq=${msg.sequence} body=${new TextDecoder().decode(msg.body)}`),
          onError: (err) => console.error(err.message),
        });
        ```
      </Tab>

      <Tab value="Java">
        ```java title="ReplayFromLast.java"
        client.subscribeToEventsStore(EventsStoreSubscription.builder()
            .channel("orders.events")
            .startPosition(EventStoreStartPosition.StartFromLast)
            .onReceiveEventCallback(event ->
                System.out.printf("[FromLast] seq=%d body=%s%n",
                    event.getSequence(), new String(event.getBody())))
            .onErrorCallback(err -> System.err.println(err.getMessage()))
            .build());
        ```
      </Tab>

      <Tab value="C#">
        ```csharp title="ReplayFromLast.cs"
        await foreach (var msg in client.SubscribeToEventsStoreAsync(
            new EventsStoreSubscription
            {
                Channel = "orders.events",
                StartPosition = EventStoreStartPosition.StartFromLast,
            }))
        {
            Console.WriteLine($"[FromLast] seq={msg.Sequence} body={Encoding.UTF8.GetString(msg.Body.Span)}");
        }
        ```
      </Tab>

      <Tab value="Kotlin">
        ```kotlin title="ReplayFromLast.kt"
        client.subscribeToEventsStore {
            channel = "orders.events"
            startPosition = StartPosition.StartFromLast
        }.collect { msg ->
            println("[FromLast] seq=${msg.sequence} body=${String(msg.body)}")
        }
        ```
      </Tab>

      <Tab value="C++">
        ```cpp title="replay_from_last.cc"
        client->SubscribeToEventsStore("orders.events", "",
            kubemq::StartPosition::StartFromLast,
            [](const kubemq::EventStoreReceived& msg) {
                std::cout << "[FromLast] seq=" << msg.sequence()
                          << " body=" << msg.body() << std::endl;
            },
            [](const std::string& err) { std::cerr << err << std::endl; });
        ```
      </Tab>

      <Tab value="Rust">
        ```rust title="replay_from_last.rs"
        use kubemq::EventsStoreSubscription;

        let sub = client
            .subscribe_to_events_store(
                "orders.events",
                "",
                EventsStoreSubscription::StartFromLast,
                |event| {
                    Box::pin(async move {
                        println!(
                            "[FromLast] seq={} body={}",
                            event.sequence,
                            String::from_utf8_lossy(&event.body)
                        );
                    })
                },
                None,
            )
            .await?;
        ```
      </Tab>

      <Tab value="Ruby">
        ```ruby title="replay_from_last.rb"
        cancel = KubeMQ::CancellationToken.new

        sub = KubeMQ::PubSub::EventsStoreSubscription.new(
          channel: 'orders.events',
          start_position: KubeMQ::PubSub::EventStoreStartPosition::START_FROM_LAST
        )
        client.subscribe_to_events_store(sub, cancellation_token: cancel,
                                         on_error: ->(e) { puts "Error: #{e.message}" }) do |event|
          puts "[FromLast] seq=#{event.sequence} body=#{event.body}"
        end
        ```
      </Tab>

      <Tab value="Elixir">
        ```elixir title="replay_from_last.exs"
        {:ok, sub} =
          KubeMQ.Client.subscribe_to_events_store(client, "orders.events",
            start_at: :start_from_last,
            on_event: fn event ->
              IO.puts("[FromLast] seq=#{event.sequence} body=#{event.body}")
            end
          )
        ```
      </Tab>
    </Tabs>

    **Output:** receives event with seq=10 (the last stored), then waits for new events.
  </Step>

  <Step>
    ### Start from a Specific Time (StartAtTime) [#start-from-a-specific-time-startattime]

    Receive events stored at or after a specific Unix timestamp in nanoseconds.

    <Tabs groupId="language" items="['Go', 'Python', 'Node.js', 'Java', 'C#', 'Kotlin', 'C++', 'Rust', 'Ruby', 'Elixir']">
      <Tab value="Go">
        ```go title="replay_from_time.go"
        targetTime := time.Now().Add(-5 * time.Minute).UnixNano()

        sub, err := client.SubscribeToEventsStore(ctx, "orders.events", "",
            kubemq.StartAtTime(targetTime),
            kubemq.WithOnEvent(func(event *kubemq.Event) {
                fmt.Printf("[AtTime] seq=%d body=%s\n",
                    event.Sequence, string(event.Body))
            }),
            kubemq.WithOnError(func(err error) { log.Println(err) }),
        )
        ```
      </Tab>

      <Tab value="Python">
        ```python title="replay_from_time.py"
        import time as time_mod

        target_time = int((time_mod.time() - 300) * 1_000_000_000)  # 5 min ago in nanos

        client.subscribe_to_events_store(
            subscription=EventsStoreSubscription(
                channel="orders.events",
                start_position=EventStoreStartPosition.StartAtTime,
                start_position_value=target_time,
                on_receive_event_callback=lambda e: print(
                    f"[AtTime] seq={e.sequence} body={e.body.decode('utf-8')}"
                ),
                on_error_callback=lambda e: print(f"Error: {e}"),
            ),
            cancel=CancellationToken(),
        )
        ```
      </Tab>

      <Tab value="Node.js">
        ```typescript title="replay_from_time.ts"
        const targetTime = (Date.now() - 5 * 60 * 1000) * 1_000_000; // 5 min ago in nanos

        client.subscribeToEventsStore({
          channel: 'orders.events',
          startPosition: EventStoreStartPosition.StartAtTime,
          startPositionValue: targetTime,
          onEvent: (msg) =>
            console.log(`[AtTime] seq=${msg.sequence} body=${new TextDecoder().decode(msg.body)}`),
          onError: (err) => console.error(err.message),
        });
        ```
      </Tab>

      <Tab value="Java">
        ```java title="ReplayFromTime.java"
        long targetTime = (System.currentTimeMillis() - 300_000) * 1_000_000L; // 5 min ago

        client.subscribeToEventsStore(EventsStoreSubscription.builder()
            .channel("orders.events")
            .startPosition(EventStoreStartPosition.StartAtTime)
            .startPositionValue(targetTime)
            .onReceiveEventCallback(event ->
                System.out.printf("[AtTime] seq=%d body=%s%n",
                    event.getSequence(), new String(event.getBody())))
            .onErrorCallback(err -> System.err.println(err.getMessage()))
            .build());
        ```
      </Tab>

      <Tab value="C#">
        ```csharp title="ReplayFromTime.cs"
        var targetTime = (DateTimeOffset.UtcNow.AddMinutes(-5)).ToUnixTimeMilliseconds() * 1_000_000;

        await foreach (var msg in client.SubscribeToEventsStoreAsync(
            new EventsStoreSubscription
            {
                Channel = "orders.events",
                StartPosition = EventStoreStartPosition.StartAtTime,
                StartPositionValue = targetTime,
            }))
        {
            Console.WriteLine($"[AtTime] seq={msg.Sequence} body={Encoding.UTF8.GetString(msg.Body.Span)}");
        }
        ```
      </Tab>

      <Tab value="Kotlin">
        ```kotlin title="ReplayFromTime.kt"
        val targetTime = (System.currentTimeMillis() - 300_000) * 1_000_000L

        client.subscribeToEventsStore {
            channel = "orders.events"
            startPosition = StartPosition.StartAtTime
            startPositionValue = targetTime
        }.collect { msg ->
            println("[AtTime] seq=${msg.sequence} body=${String(msg.body)}")
        }
        ```
      </Tab>

      <Tab value="C++">
        ```cpp title="replay_from_time.cc"
        auto now = std::chrono::system_clock::now();
        auto target = now - std::chrono::minutes(5);
        auto nanos = std::chrono::duration_cast<std::chrono::nanoseconds>(
            target.time_since_epoch()).count();

        client->SubscribeToEventsStore("orders.events", "",
            kubemq::StartPosition::StartAtTime, nanos,
            [](const kubemq::EventStoreReceived& msg) {
                std::cout << "[AtTime] seq=" << msg.sequence()
                          << " body=" << msg.body() << std::endl;
            },
            [](const std::string& err) { std::cerr << err << std::endl; });
        ```
      </Tab>

      <Tab value="Rust">
        ```rust title="replay_from_time.rs"
        use kubemq::EventsStoreSubscription;
        use std::time::{Duration, SystemTime};

        // The Rust SDK takes a SystemTime directly (5 minutes ago).
        let target_time = SystemTime::now() - Duration::from_secs(5 * 60);

        let sub = client
            .subscribe_to_events_store(
                "orders.events",
                "",
                EventsStoreSubscription::StartAtTime(target_time),
                |event| {
                    Box::pin(async move {
                        println!(
                            "[AtTime] seq={} body={}",
                            event.sequence,
                            String::from_utf8_lossy(&event.body)
                        );
                    })
                },
                None,
            )
            .await?;
        ```
      </Tab>

      <Tab value="Ruby">
        ```ruby title="replay_from_time.rb"
        cancel = KubeMQ::CancellationToken.new
        # The Ruby SDK takes a Unix timestamp in seconds (5 minutes ago).
        target_time = Time.now.to_i - 5 * 60

        sub = KubeMQ::PubSub::EventsStoreSubscription.new(
          channel: 'orders.events',
          start_position: KubeMQ::PubSub::EventStoreStartPosition::START_AT_TIME,
          start_position_value: target_time
        )
        client.subscribe_to_events_store(sub, cancellation_token: cancel,
                                         on_error: ->(e) { puts "Error: #{e.message}" }) do |event|
          puts "[AtTime] seq=#{event.sequence} body=#{event.body}"
        end
        ```
      </Tab>

      <Tab value="Elixir">
        ```elixir title="replay_from_time.exs"
        # The Elixir SDK takes a Unix timestamp in seconds (5 minutes ago).
        target_time = System.system_time(:second) - 5 * 60

        {:ok, sub} =
          KubeMQ.Client.subscribe_to_events_store(client, "orders.events",
            start_at: {:start_at_time, target_time},
            on_event: fn event ->
              IO.puts("[AtTime] seq=#{event.sequence} body=#{event.body}")
            end
          )
        ```
      </Tab>
    </Tabs>

    **Output:** receives events stored at or after the target timestamp.
  </Step>
</Steps>

## Choosing the Right Position [#choosing-the-right-position]

| Scenario                               | Recommended Position | Why                                 |
| -------------------------------------- | -------------------- | ----------------------------------- |
| Rebuild application state from scratch | `StartFromFirst`     | Replays the entire history          |
| Resume after a known checkpoint        | `StartAtSequence`    | Picks up exactly where you left off |
| Recover recent events after downtime   | `StartAtTimeDelta`   | Replays from a time window          |
| Monitor live activity only             | `StartNewOnly`       | Ignores history, lowest overhead    |
| Catch the latest event then go live    | `StartFromLast`      | Quick sync then real-time           |
| Point-in-time recovery                 | `StartAtTime`        | Precise timestamp-based replay      |

<Callout type="warn">
  `StartAtSequence` requires a value greater than 0. `StartAtTime` requires a Unix timestamp in nanoseconds greater than 0. `StartAtTimeDelta` requires a positive number of seconds. Providing 0 or negative values results in a validation error.
</Callout>

## Durable Replay Behavior [#durable-replay-behavior]

When a subscriber with a durable name reconnects, the start position parameter is **ignored** after the first connection. The store resumes from the last acknowledged sequence for that durable name. The durable name is `{channel}-{group}`.

To force a fresh replay, use a different `group` name or a different `clientId`.

## Next Steps [#next-steps]

* Implement [event sourcing](/learn/events-store/tutorials/event-sourcing) using replay from first
* Scale processing with [consumer groups](/learn/events-store/tutorials/consumer-groups)
* Configure [retention policies](/learn/events-store/how-to/configure-retention) to manage storage
* See [Events Store Reference](/learn/events-store/reference) for all subscription parameters