# Events Store Reference (/learn/events-store/reference)



This reference documents every aspect of the KubeMQ Events Store messaging pattern.

## Message Structure [#message-structure]

### Event Store Message (Publish) [#event-store-message-publish]

<TypeTable
  type="{
  EventID: {
    type: 'string',
    description: 'Unique identifier. Auto-generated (NUID) if empty.',
    default: 'Auto-generated',
  },
  Channel: {
    type: 'string',
    description: 'Target channel name. Must follow channel naming rules.',
    required: true,
  },
  ClientID: {
    type: 'string',
    description: 'Publishing client identifier. Must match ^[a-zA-Z0-9_-]+$.',
    required: true,
  },
  Metadata: {
    type: 'string',
    description: 'Text metadata. At least one of Body or Metadata must be provided.',
    default: '&#x22;&#x22;',
  },
  Body: {
    type: 'bytes',
    description: 'Binary payload. At least one of Body or Metadata must be provided.',
    default: 'null',
  },
  Tags: {
    type: 'map<string, string>',
    description: 'Key-value pairs for filtering and routing.',
    default: '{}',
  },
  Store: {
    type: 'boolean',
    description: 'Must be true for Events Store. Set false for plain Events.',
    default: 'true',
    required: true,
  },
}"
/>

### Event Store Receive (Subscription Callback) [#event-store-receive-subscription-callback]

<TypeTable
  type="{
  EventID: {
    type: 'string',
    description: 'Unique identifier of the received event.',
  },
  Channel: {
    type: 'string',
    description: 'Channel the event was published to.',
  },
  Metadata: {
    type: 'string',
    description: 'Text metadata from the publisher.',
  },
  Body: {
    type: 'bytes',
    description: 'Binary payload from the publisher.',
  },
  Tags: {
    type: 'map<string, string>',
    description: 'Key-value pairs from the publisher.',
  },
  Sequence: {
    type: 'uint64',
    description: 'Monotonically increasing sequence number assigned by the server.',
  },
  Timestamp: {
    type: 'int64',
    description: 'Server-assigned timestamp in Unix nanoseconds.',
  },
}"
/>

<Callout type="info">
  The `Sequence` and `Timestamp` fields are assigned by the server at persistence time and are not set by the publisher. They are available only on received events.
</Callout>

### Send Result [#send-result]

<TypeTable
  type="{
  EventID: {
    type: 'string',
    description: 'The event identifier (auto-generated if not provided).',
  },
  Sent: {
    type: 'boolean',
    description: 'true if the event was stored successfully.',
  },
  Error: {
    type: 'string',
    description: 'Error message if the store operation failed. Empty on success.',
  },
}"
/>

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

<TypeTable
  type="{
  StartNewOnly: {
    type: 'enum (1)',
    description: 'Only events published after subscribing. No value required.',
  },
  StartFromFirst: {
    type: 'enum (2)',
    description: 'Replay all events from the beginning. No value required.',
  },
  StartFromLast: {
    type: 'enum (3)',
    description: 'Start from the most recent stored event. No value required.',
  },
  StartAtSequence: {
    type: 'enum (4)',
    description: 'Start from a specific sequence number. Value: sequence > 0.',
    required: true,
  },
  StartAtTime: {
    type: 'enum (5)',
    description: 'Start from a specific timestamp. Value: Unix nanos > 0.',
    required: true,
  },
  StartAtTimeDelta: {
    type: 'enum (6)',
    description: 'Start from N seconds ago. Value: seconds > 0.',
    required: true,
  },
}"
/>

## Subscribe Request [#subscribe-request]

<TypeTable
  type="{
  Channel: {
    type: 'string',
    description: 'Exact channel name (no wildcards).',
    required: true,
  },
  Group: {
    type: 'string',
    description: 'Consumer group name for load-balanced delivery.',
  },
  ClientID: {
    type: 'string',
    description: 'Subscriber identifier. Must match ^[a-zA-Z0-9_-]+$.',
    required: true,
  },
  SubscribeType: {
    type: 'enum',
    description: 'Must be EventsStore.',
    required: true,
  },
  EventsStoreTypeData: {
    type: 'enum (1-6)',
    description: 'Start position. Cannot be 0 (Undefined).',
    required: true,
  },
  EventsStoreTypeValue: {
    type: 'int64',
    description: 'Value for StartAtSequence, StartAtTime, or StartAtTimeDelta.',
  },
}"
/>

## Channel Naming Rules [#channel-naming-rules]

| Rule            | Constraint                | Error Code |
| --------------- | ------------------------- | ---------- |
| Required        | Cannot be empty           | 102        |
| No trailing dot | Cannot end with `.`       | 119        |
| No whitespace   | Cannot contain spaces     | 108        |
| No wildcards    | Cannot contain `*` or `>` | 107        |

**Valid channel regex:** `^[^\s*>]+[^.]$`

<Callout type="warn">
  Events Store does **not** support wildcard subscriptions. Both publish and subscribe channel names must be exact. For wildcard support, use plain [Events](/learn/events).
</Callout>

## Durable Subscriptions [#durable-subscriptions]

Every Events Store subscription creates a durable name:

```text
DurableName = "{channel}-{group}"
```

* On first connection, the `EventsStoreTypeData` determines the start position
* On subsequent connections with the same durable name, the start position is **ignored** — the subscription resumes from the last tracked position
* To force a fresh replay, use a different `group` name

## Consumer Groups [#consumer-groups]

When multiple subscribers specify the same `Group` on the same channel:

* Each event is delivered to exactly **one** member (round-robin)
* The group's position is tracked durably
* Adding or removing members rebalances delivery automatically
* Groups are independent per channel

## Routing (Multicast) [#routing-multicast]

| Character | Purpose                            | Example                                           |
| --------- | ---------------------------------- | ------------------------------------------------- |
| `;`       | Separate channels of the same type | `audit;compliance`                                |
| `:`       | Specify target pattern type        | `events_store:audit;events:notify;queues:process` |

For Events Store publish, channels without a prefix default to `events_store`. Routed messages are tagged with `X-KUBEMQ-ROUTED=true`.

## Storage Configuration [#storage-configuration]

<TypeTable
  type="{
  'Store.MaxRetention': {
    type: 'int (minutes)',
    description: 'Maximum message age. 0 = unlimited.',
    default: '1440 (24h)',
  },
  'Store.MaxQueueSize': {
    type: 'int (bytes)',
    description: 'Maximum bytes per channel. Oldest removed when exceeded.',
    default: '0 (unlimited)',
  },
  'Store.MaxMessages': {
    type: 'int',
    description: 'Maximum messages per channel. Oldest removed when exceeded.',
    default: '0 (unlimited)',
  },
  'Store.MaxPurgeInactive': {
    type: 'int (minutes)',
    description: 'Time before empty inactive channels are purged.',
    default: '1440 (24h)',
  },
  'Store.CleanStore': {
    type: 'boolean',
    description: 'Remove all stored data on server start.',
    default: 'false',
  },
  'Store.StorePath': {
    type: 'string',
    description: 'Directory for persistent storage.',
    default: './store',
  },
}"
/>

## File Store Configuration [#file-store-configuration]

<TypeTable
  type="{
  'Broker.WriteBufferSize': {
    type: 'int (bytes)',
    description: 'Write buffer size for the file store.',
    default: '2 MB',
  },
  'Broker.ReadBufferSize': {
    type: 'int (bytes)',
    description: 'Read buffer size for the file store.',
    default: '2 MB',
  },
  'Broker.DiskSyncSeconds': {
    type: 'int',
    description: 'Seconds between automatic disk syncs.',
    default: '60',
  },
  'Broker.SliceMaxBytes': {
    type: 'int (bytes)',
    description: 'Maximum bytes per file slice before rotation.',
    default: '64 MB',
  },
  'Broker.SliceMaxMessages': {
    type: 'int',
    description: 'Maximum messages per file slice.',
    default: '0 (unlimited)',
  },
  'Broker.ParallelRecovery': {
    type: 'int',
    description: 'Goroutines for store recovery on startup.',
    default: '2',
  },
}"
/>

## Storage Utilization Thresholds [#storage-utilization-thresholds]

| Utilization | Level    | Polling Interval | Publishing                |
| ----------- | -------- | ---------------- | ------------------------- |
| 0-80%       | Normal   | 5 seconds        | Allowed                   |
| 80-90%      | Warning  | 3 seconds        | Allowed (warnings logged) |
| 90-95%      | Critical | 2 seconds        | Allowed (errors logged)   |
| Above 95%   | Disabled | 1 second         | **Blocked**               |

Publishing automatically resumes when utilization drops below 95%.

## Transport Protocols [#transport-protocols]

### gRPC [#grpc]

* **Publish:** `SendEvent(pb.Event)` with `Store=true`
* **Publish Stream:** `SendEventsStream()` — bidirectional streaming, result sent for every store event
* **Subscribe:** `SubscribeToEvents(pb.Subscribe)` with `SubscribeType=EventsStore`
* Default port: `50000`

### REST [#rest]

* **Publish:** `POST /send` with `isEvents=false`
* **Subscribe:** WebSocket with `subscribe_type=events_store`
* Default port: `9090`

## Delivery Semantics [#delivery-semantics]

| Aspect             | Behavior                                              |
| ------------------ | ----------------------------------------------------- |
| Delivery guarantee | **At-least-once**                                     |
| Persistence        | Disk-backed file store                                |
| Ordering           | Events delivered in sequence order per channel        |
| Sequence numbers   | Monotonically increasing, starting from 1 per channel |
| Acknowledgment     | Asynchronous publish ack                              |
| Durable names      | Position tracked across reconnections                 |
| Replay             | Yes (6 start positions)                               |

## Events Store vs Events [#events-store-vs-events]

| Feature                    | Events Store                 | Events           |
| -------------------------- | ---------------------------- | ---------------- |
| Persistence                | Yes (disk-backed)            | No (memory only) |
| Replay                     | Yes (6 start positions)      | No               |
| Delivery guarantee         | At-least-once                | At-most-once     |
| Wildcard subscriptions     | No                           | Yes (`*`, `>`)   |
| Consumer groups            | Durable                      | Ephemeral        |
| Sequence numbers           | Yes                          | No               |
| Server timestamps          | Yes                          | No               |
| Latency                    | Slightly higher (disk write) | Lowest           |
| Storage utilization limits | Yes (80/90/95% thresholds)   | No               |

## SDK Quick Reference [#sdk-quick-reference]

<Tabs groupId="language" items="['Go', 'Python', 'Node.js', 'Java', 'C#', 'Kotlin', 'C++', 'Rust', 'Ruby', 'Elixir']">
  <Tab value="Go">
    ```go
    client.SendEventStore(ctx, event)
    client.SendEventsStoreStream(ctx, opts...)
    client.SubscribeToEventsStore(ctx, channel, group, startPos, opts...)
    ```
  </Tab>

  <Tab value="Python">
    ```python
    client.publish_event_store(message)
    client.open_events_store_stream(on_result, on_error)
    client.subscribe_to_events_store(subscription, cancel)
    ```
  </Tab>

  <Tab value="Node.js">
    ```typescript
    client.sendEventStore(message)
    client.sendEventsStoreStream(opts)
    client.subscribeToEventsStore(opts)
    ```
  </Tab>

  <Tab value="Java">
    ```java
    client.sendEventsStoreMessage(message)
    client.openEventsStoreStream(onResult, onError)
    client.subscribeToEventsStore(subscription)
    ```
  </Tab>

  <Tab value="C#">
    ```csharp
    client.SendEventStoreAsync(message)
    client.OpenEventsStoreStream(onResult, onError)
    client.SubscribeToEventsStoreAsync(subscription)
    ```
  </Tab>

  <Tab value="Kotlin">
    ```kotlin
    client.sendEventStore(message)
    client.openEventsStoreStream(onResult, onError)
    client.subscribeToEventsStore { ... }
    ```
  </Tab>

  <Tab value="C++">
    ```cpp
    client->SendEventStore(message)
    client->OpenEventsStoreStream(onResult, onError)
    client->SubscribeToEventsStore(channel, group, startPos, onEvent, onError)
    ```
  </Tab>

  <Tab value="Rust">
    ```rust
    client.send_event_store(event).await
    client.send_event_store_stream().await       // returns a stream; stream.send(event)
    client.subscribe_to_events_store(channel, group, EventsStoreSubscription::StartFromFirst, on_event, None).await
    ```
  </Tab>

  <Tab value="Ruby">
    ```ruby
    client.send_event_store(msg)
    client.create_events_store_sender              # sender.publish(msg)
    client.subscribe_to_events_store(subscription, cancellation_token:, on_error:) { |event| ... }
    ```
  </Tab>

  <Tab value="Elixir">
    ```elixir
    KubeMQ.Client.send_event_store(client, event)
    KubeMQ.Client.send_event_store(client, event)  # stream-style: send in rapid succession
    KubeMQ.Client.subscribe_to_events_store(client, channel, start_at: :start_new_only, on_event: fn event -> ... end)
    ```
  </Tab>
</Tabs>

## Error Codes [#error-codes]

### Validation Errors [#validation-errors]

| Code | Error                     | Description                                                        |
| ---- | ------------------------- | ------------------------------------------------------------------ |
| 101  | Invalid ClientID          | ClientID is empty                                                  |
| 102  | Invalid Channel           | Channel is empty                                                   |
| 107  | Invalid Wildcards         | Channel contains `*` or `>`                                        |
| 108  | Invalid Whitespace        | Channel contains spaces                                            |
| 110  | Invalid Message           | Both `Body` and `Metadata` are empty                               |
| 111  | Invalid Subscription Type | `EventsStoreTypeData` is `Undefined` (0)                           |
| 112  | Invalid Sequence Value    | `StartAtSequence` value is 0 or negative                           |
| 113  | Invalid Time Value        | `StartAtTime` value is 0 or negative                               |
| 114  | Invalid Time Delta Value  | `StartAtTimeDelta` value is 0 or negative                          |
| 118  | Wrong Subscribe Type      | Event store parameters set but subscribe type is not events\_store |
| 119  | Invalid Channel Separator | Channel ends with `.`                                              |

### Runtime Errors [#runtime-errors]

| Error                        | Cause                                | Resolution                                   |
| ---------------------------- | ------------------------------------ | -------------------------------------------- |
| `ErrShutdownMode`            | Server is shutting down              | Reconnect after server restart               |
| `ErrConnectionNoAvailable`   | Connection is down                   | SDK auto-reconnects; check server health     |
| Storage utilization exceeded | Disk usage above 95%                 | Free disk space or adjust retention settings |
| Authorization denied         | Casbin policy rejected the operation | Verify client permissions                    |

## Related [#related]

* [Getting Started with Events Store](/learn/events-store/getting-started)
* [Events Reference](/learn/events/reference) for the non-persistent variant
* Core Concepts for foundational KubeMQ concepts