Events Store

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. This page documents how the plugin exposes it through the Watermill interfaces. EventsStore is the persistent sibling of the fire-and-forget 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.

All code on this page uses the plugin package:

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

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.

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:

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

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

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:

// 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:

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:

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:

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

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

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

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.

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

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:

If you also need explicit per-message acknowledgment with redelivery, that is the Queues pattern — see Queues.

Run the examples

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

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