# Events Store (/integrations/watermill/how-to/events-store)



## Overview [#overview]

The **EventsStore** pattern (`PatternEventsStore`) maps Watermill's `message.Publisher` and
`message.Subscriber` onto KubeMQ's persistent events. Messages are written to durable
storage and assigned a monotonically increasing **sequence** number. Delivery is
**at-least-once**: the broker tracks each consumer's offset and advances it automatically as
messages are delivered. Because the stream is retained, a subscriber can **replay** history
from any point — the foundation for event sourcing, audit trails, and rebuilding read models.

For what the EventsStore pattern guarantees at the broker level, see
[Events Store](/learn/events-store). This page documents how the plugin exposes it through
the Watermill interfaces. EventsStore is the persistent sibling of the fire-and-forget
[Events](/integrations/watermill/how-to/events) pattern: the publisher API is identical — you still call
`pub.Publish(topic, msg)` — and the difference lives entirely on the subscriber, which
declares *where in the stream to begin* via a start option.

<Callout type="info">
  The Watermill plugin talks to KubeMQ over the native gRPC API on port `50000` — it is a Go
  client library, not an HTTP connector, and there is no enable flag. Start a broker before
  running any example (see [Getting Started](/integrations/watermill/tutorials/getting-started)).
</Callout>

All code on this page uses the plugin package:

```go
import kubemq "github.com/kubemq-io/watermill-kubemq/pkg/kubemq"
```

## Publisher and subscriber configuration [#publisher-and-subscriber-configuration]

An EventsStore publisher just sets `Pattern: kubemq.PatternEventsStore`. The subscriber adds
two EventsStore-specific concerns: an optional `ConsumerGroup` (same competing-consumers
semantics as Events) and an `EventsStoreStartOption` that selects the replay position.

```go title="events-store/basic-pubsub/main.go (excerpt)"
pub, err := kubemq.NewPublisher(kubemq.PublisherConfig{
	Address: "localhost:50000",
	Pattern: kubemq.PatternEventsStore,
	Logger:  logger,
})

sub, err := kubemq.NewSubscriber(kubemq.SubscriberConfig{
	Address:                "localhost:50000",
	Pattern:                kubemq.PatternEventsStore,
	EventsStoreStartOption: kubemq.StartFromNew, // default: only events published after subscribing
	Logger:                 logger,
})
```

The start option is the single most important EventsStore setting. The six values are:

| Start option  | Constant             | Behavior                                                                  | Extra field required   |
| ------------- | -------------------- | ------------------------------------------------------------------------- | ---------------------- |
| New (default) | `StartFromNew`       | Only events published after the subscription is created                   | —                      |
| First         | `StartFromFirst`     | Replay every stored event from the beginning, then continue with new ones | —                      |
| Last          | `StartFromLast`      | Deliver the single most recent stored event, then continue with new ones  | —                      |
| Sequence      | `StartFromSequence`  | Replay starting at a specific sequence number                             | `EventsStoreSequence`  |
| Time          | `StartFromTime`      | Replay starting at an absolute timestamp                                  | `EventsStoreStartTime` |
| Time delta    | `StartFromTimeDelta` | Replay starting at `now − duration` (e.g. "the last 5 minutes")           | `EventsStoreTimeDelta` |

`StartFromNew` is the zero value, so a subscriber that omits `EventsStoreStartOption` behaves
like a fresh live subscription — comparable to Events, but with durability and offset tracking
underneath.

### Start-option validation [#start-option-validation]

The three parameterized options carry a required companion field, and
`SubscriberConfig.Validate` rejects the config at `NewSubscriber` time if it is missing or out
of range:

* `StartFromSequence` requires `EventsStoreSequence > 0` — otherwise `NewSubscriber` returns `EventsStoreSequence must be > 0 for StartFromSequence`.
* `StartFromTime` requires a non-zero `EventsStoreStartTime` — otherwise `EventsStoreStartTime must be set for StartFromTime`.
* `StartFromTimeDelta` requires `EventsStoreTimeDelta > 0` — otherwise `EventsStoreTimeDelta must be > 0 for StartFromTimeDelta`.

These checks run before any broker connection is used, so a misconfigured replay fails fast
rather than silently delivering nothing.

## Replay examples [#replay-examples]

The repository ships one runnable example per start option. The pattern is always the same:
publish some messages, then subscribe with a start option and observe which messages are
replayed.

**Replay all from the beginning** — `events-store/start-from-first` publishes 5 messages,
then subscribes with `StartFromFirst` and receives all 5 in sequence order:

```go title="events-store/start-from-first/main.go (excerpt)"
// 5 messages were already published to the channel above.
sub, err := kubemq.NewSubscriber(kubemq.SubscriberConfig{
	Address:                "localhost:50000",
	Pattern:                kubemq.PatternEventsStore,
	EventsStoreStartOption: kubemq.StartFromFirst, // replay everything
	Logger:                 logger,
})
if err != nil {
	log.Fatal(err)
}
defer sub.Close()

msgs, err := sub.Subscribe(ctx, "watermill-es.start-from-first")
if err != nil {
	log.Fatal(err)
}

received := 0
for received < 5 {
	select {
	case msg := <-msgs:
		seq := msg.Metadata.Get("_kubemq_sequence")
		fmt.Printf("Received: Payload=%s, Sequence=%s\n", string(msg.Payload), seq)
		msg.Ack()
		received++
	case <-ctx.Done():
		log.Fatal("Timeout waiting for messages")
	}
}
```

**Replay from sequence N** — `events-store/start-from-sequence` sets `StartFromSequence` with
`EventsStoreSequence: 3`, so only messages with sequence 3, 4, 5 are delivered:

```go title="events-store/start-from-sequence/main.go (excerpt)"
sub, err := kubemq.NewSubscriber(kubemq.SubscriberConfig{
	Address:                "localhost:50000",
	Pattern:                kubemq.PatternEventsStore,
	EventsStoreStartOption: kubemq.StartFromSequence,
	EventsStoreSequence:    3, // replay messages with sequence >= 3
	Logger:                 logger,
})
```

**Replay from a point in time** — `events-store/start-from-time` records a cutoff timestamp,
publishes messages before and after it, then subscribes with `StartFromTime` and
`EventsStoreStartTime: cutoff` to receive only the post-cutoff messages:

```go title="events-store/start-from-time/main.go (excerpt)"
cutoff := time.Now() // messages published before this are ignored
// ... publish more messages after the cutoff ...

sub, err := kubemq.NewSubscriber(kubemq.SubscriberConfig{
	Address:                "localhost:50000",
	Pattern:                kubemq.PatternEventsStore,
	EventsStoreStartOption: kubemq.StartFromTime,
	EventsStoreStartTime:   cutoff,
	Logger:                 logger,
})
```

**Replay a recent window** — `events-store/start-from-time-delta` uses a *relative* offset.
With `EventsStoreTimeDelta: 5 * time.Second`, only messages stored within the last 5 seconds
are replayed:

```go title="events-store/start-from-time-delta/main.go (excerpt)"
sub, err := kubemq.NewSubscriber(kubemq.SubscriberConfig{
	Address:                "localhost:50000",
	Pattern:                kubemq.PatternEventsStore,
	EventsStoreStartOption: kubemq.StartFromTimeDelta,
	EventsStoreTimeDelta:   5 * time.Second, // "the last 5 seconds"
	Logger:                 logger,
})
```

The `events-store/consumer-group` example combines replay with competing consumers: two
subscribers share `ConsumerGroup: "es-workers"` and `StartFromFirst`, and the stored messages
are distributed between them for load-balanced reprocessing.

## Delivered metadata [#delivered-metadata]

For every EventsStore message, the subscriber copies KubeMQ's storage metadata onto the
Watermill message's `Metadata` map before handing it to your handler:

| Metadata key        | Value                                           | Use                                         |
| ------------------- | ----------------------------------------------- | ------------------------------------------- |
| `_kubemq_sequence`  | The message's sequence number                   | Ordering, checkpointing, "resume from here" |
| `_kubemq_timestamp` | The storage timestamp, formatted as RFC3339Nano | Audit, time-based reasoning                 |

Read them with `msg.Metadata.Get(...)`:

```go
seq := msg.Metadata.Get("_kubemq_sequence")   // e.g. "3"
ts := msg.Metadata.Get("_kubemq_timestamp")   // e.g. "2026-06-01T12:34:56.789012345Z"
fmt.Printf("seq=%s ts=%s payload=%s\n", seq, ts, string(msg.Payload))
```

A common pattern is to persist the highest `_kubemq_sequence` your service has processed, then
resume after a restart with `StartFromSequence` set to that value plus one.

<Callout type="info">
  These keys are populated only for EventsStore messages. Events messages do not carry a
  sequence or storage timestamp, because nothing is stored.
</Callout>

## Streaming publish [#streaming-publish]

Like Events, the EventsStore publisher opens a persistent gRPC streaming handle
(`SendEventStoreStream`) at construction time and sends every `Publish` over that single
long-lived stream. Setting `DisableStreaming: true` falls back to a synchronous
per-message send (`SendEventStore`); the non-streaming path also surfaces a per-message
result whose `Sent` flag the plugin checks before reporting success. For most workloads,
leave streaming enabled.

## When to choose Events vs EventsStore [#when-to-choose-events-vs-eventsstore]

Events and EventsStore look almost identical in code — the only difference is the `Pattern`
value and the EventsStore start option — but they serve opposite needs:

| Use **Events** when...                             | Use **EventsStore** when...                   |
| -------------------------------------------------- | --------------------------------------------- |
| You want the lowest latency                        | You need durability and replay                |
| Missing a message is acceptable                    | Every event must be recoverable               |
| Subscribers are always connected                   | Subscribers may be offline and catch up later |
| Real-time notifications, live metrics, log streams | Event sourcing, audit trails, stream replay   |

If you also need explicit per-message acknowledgment with redelivery, that is the Queues
pattern — see [Queues](/integrations/watermill/how-to/queues).

## Run the examples [#run-the-examples]

With a broker running on `localhost:50000`, run the EventsStore examples from the repository root:

```bash
go run ./examples/events-store/basic-pubsub/main.go
go run ./examples/events-store/start-from-first/main.go
go run ./examples/events-store/start-from-sequence/main.go
go run ./examples/events-store/start-from-time/main.go
go run ./examples/events-store/start-from-time-delta/main.go
go run ./examples/events-store/consumer-group/main.go
```

## Next steps [#next-steps]

<Cards>
  <Card title="Events" href="/integrations/watermill/how-to/events" description="The fire-and-forget sibling — at-most-once delivery, fan-out, and consumer groups." />

  <Card title="Queues" href="/integrations/watermill/how-to/queues" description="Reliable point-to-point delivery with explicit ack/nack, DLQ, and competing consumers." />

  <Card title="Commands & Queries" href="/integrations/watermill/how-to/commands-queries" description="EventsStore backs CQRS domain events — pair it with Queues for commands." />
</Cards>