# Persistent Pub/Sub (/sdks/nodejs/tutorials/persistent-pubsub)



## Overview [#overview]

This tutorial builds a publisher and subscriber on a KubeMQ Events Store channel — reach for this pattern when a subscriber can't guarantee it's listening the instant a message is published. Plain events are fire-and-forget: publish with no one subscribed and the message is gone. Events Store persists every event to a durable, ordered log, so a subscriber connecting seconds or a full restart later still catches up — useful for anything needing a complete history, like an audit trail or event-sourced state.

The two calls involved: `sendEventStore()` publishes and returns a result confirming storage plus a broker-assigned sequence number, and `subscribeToEventsStore()` takes a required `startFrom` position telling the broker where to start — new events only, from the first stored event (`EventStoreStartPosition.StartFromFirst`, used here), or a given sequence or time. Production subscribers usually resume from a saved checkpoint instead of replaying from the beginning.

**Gotchas:** replaying from the first event on every restart replays the whole log, which gets costly on a busy channel — track the last `sequence` you processed instead. Starting from new-only has the opposite risk: anything published earlier is silently skipped, so don't rely on a fixed delay like this sample's to paper over that race in production. Persistence isn't consumer coordination: each independent subscriber gets its own full replay unless grouped with a consumer group.

## Prerequisites [#prerequisites]

* KubeMQ server running on `localhost:50000`
* Node.js SDK installed (`npm install kubemq-js`)

## Code [#code]

```typescript title="persistent-pubsub.ts"
/**
 * Example: Persistent Event Publish/Subscribe
 *
 * Demonstrates event store messaging where events are persisted and can be
 * replayed by subscribers. Unlike regular events, subscribers can connect
 * after events are published and still receive them.
 *
 * Prerequisites:
 *   - KubeMQ server running on localhost:50000
 *
 * Run: npx tsx examples/events-store/persistent-pubsub.ts
 */
import { KubeMQClient, createEventStoreMessage, EventStoreStartPosition } from 'kubemq-js';

async function main(): Promise<void> {
  const client = await KubeMQClient.create({
    address: 'localhost:50000',
    clientId: 'js-events-store-persistent-pubsub-client',
  });

  try {
    // Publish events first — they are persisted in the store.
    for (let i = 1; i <= 3; i++) {
      await client.sendEventStore(
        createEventStoreMessage({
          channel: 'js-events-store.persistent-pubsub',
          body: `User action #${i}: login from 192.168.1.${i}`,
          tags: { action: 'login', sequence: String(i) },
        }),
      );
      console.log(`Published event #${i}`);
    }

    // Subscribe starting from the first stored event — replays all three.
    const subscription = client.subscribeToEventsStore({
      channel: 'js-events-store.persistent-pubsub',
      startFrom: EventStoreStartPosition.StartFromFirst,
      onEvent: (event) => {
        console.log(`Received [seq=${event.sequence}]:`, new TextDecoder().decode(event.body));
      },
      onError: (err) => {
        console.error('Subscription error:', err.message);
      },
    });

    await new Promise((resolve) => setTimeout(resolve, 2000));
    subscription.cancel();
  } finally {
    await client.close();
  }
}

main().catch(console.error);

// Expected output:
// Published event #1
// Published event #2
// Published event #3
// Received [seq=<sequence>]: User action #1: login from 192.168.1.1
// Received [seq=<sequence>]: User action #2: login from 192.168.1.2
// Received [seq=<sequence>]: User action #3: login from 192.168.1.3

```

## How It Works [#how-it-works]

* `sendEventStore()` persists each event on the server — unlike `sendEvent()`, messages survive subscriber disconnection.
* `subscribeToEventsStore()` with `StartFromFirst` replays all stored events from the very beginning of the channel, even though the publisher finished before the subscriber connected.
* The `event.sequence` field is a server-assigned monotonic integer; it can be used to resume replay from a checkpoint (see [Replay from Sequence](/sdks/nodejs/how-to/events-store/replay-from-sequence)).
* The 2-second wait gives the subscriber time to receive all replayed events before `cancel()` closes the stream.

## Related [#related]

* [Pattern overview](/learn/events-store/getting-started)
* [Node.js SDK Reference](/sdks/nodejs/reference)
* [Cancel Subscription](/sdks/nodejs/how-to/events-store/cancel-subscription)
* [Consumer Group](/sdks/nodejs/how-to/events-store/consumer-group)