Replay from Time

Overview

Replaying from a timestamp lets a consumer recover a window of history without knowing exact sequence numbers — you reach for it after a deploy, an outage, or any gap where you know roughly when you went dark but not where you left off in the stream. It turns an Events Store channel into a rewindable log: resubscribe with a point in time and the broker replays every event stored at or after it, then hands off to live delivery.

The subscription's startFrom is set to EventStoreStartPosition.StartAtTimeDelta (a relative offset in seconds via startValue) or StartAtTime (an absolute Unix timestamp) — the broker compares this against the storage timestamp it assigned to each event, not any timestamp embedded in the payload. Because it's wall-clock based, the window is approximate rather than exact: pass a value far enough back to be safe.

Gotchas: clock skew between your subscriber's clock and the server's matters — favor a generous buffer over a precise cutoff. Storage timestamps reflect when the broker persisted the event, not when the producer created it, so under load the two can drift. And unlike sequence-based replay, a time-based start position has no way to guarantee "no gaps, no duplicates" across a network hiccup — use EventStoreStartPosition.StartAtSequence instead if you need exact resumption.

Prerequisites

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

Code

/**
 * Example: Replay Events from a Specific Timestamp
 *
 * Demonstrates subscribing to an event store starting from a specific
 * point in time. Useful for replaying events that occurred after a known
 * timestamp (e.g., "replay everything since last deployment").
 *
 * Prerequisites:
 *   - KubeMQ server running on localhost:50000
 *
 * Run: npx tsx examples/events-store/replay-from-time.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-replay-from-time-client',
  });

  try {
    await client.sendEventStore(
      createEventStoreMessage({
        channel: 'js-events-store.replay-from-time',
        body: 'Deployed v2.4.1 to production',
      }),
    );

    // Subscribe from 60 seconds ago — catches recent events.
    const subscription = client.subscribeToEventsStore({
      channel: 'js-events-store.replay-from-time',
      startFrom: EventStoreStartPosition.StartAtTimeDelta,
      startValue: 60, // seconds ago
      onEvent: (event) => {
        console.log(`[${event.timestamp.toISOString()}] ${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);

How It Works

• StartAtTimeDelta with startValue: 60 tells the server to replay all events published in the last 60 seconds from now.
• The event is published first; the StartAtTimeDelta subscription then replays it because it falls within the 60-second window.
• Each received EventStoreReceived carries a timestamp field (Date) set by the server at ingestion time — used here to confirm the replay time.
• For replaying from an absolute Unix timestamp in seconds instead, use StartAtTime with startValue: Math.floor(Date.now() / 1000).

Related

• Replay from Sequence — exact, gap-free resumption from a persisted checkpoint
• Pattern overview
• Node.js SDK Reference
• Persistent Pub/Sub
• Cancel Subscription