Cursor-Based Stream Pagination

Cursor-based pagination for streams differs from database pagination in one critical way: the dataset is growing while the client reads it. In a static dataset, the cursor is a stable reference point — row 500 is always row 500. In a stream, new data arrives continuously, and the cursor must track a position in an append-only sequence. The cursor is typically a timestamp, sequence number, or log offset — a value that monotonically increases and uniquely identifies a position in the stream.

Stream cursors must handle three edge cases. First, cursor expiry: if the client disconnects and reconnects hours later, the cursor position may no longer be available (log rotation, event compaction). The server should return an error with the earliest available cursor so the client can resume from the oldest available position. Second, cursor gaps: if events are deleted or compacted, the cursor may point to a position between two available events. The server should advance to the next available event. Third, concurrent writes: new events arriving while the client reads a chunk should not appear in the current response — they belong to the next chunk.

interface StreamPosition {
  cursor: string;          // Opaque cursor value
  cursorType: "timestamp" | "sequence" | "offset";
  isExpired: boolean;      // True if cursor is beyond retention
  earliestAvailable?: string; // Fallback if cursor expired
}

function resolveStreamCursor(
  requestedCursor: string | undefined,
  streamMeta: { earliest: string; latest: string }
): StreamPosition {
  if (!requestedCursor) {
    // No cursor — start from latest (tail mode)
    return {
      cursor: streamMeta.latest,
      cursorType: "sequence",
      isExpired: false,
    };
  }

  if (requestedCursor < streamMeta.earliest) {
    // Cursor expired — data has been rotated/compacted
    return {
      cursor: streamMeta.earliest,
      cursorType: "sequence",
      isExpired: true,
      earliestAvailable: streamMeta.earliest,
    };
  }

  return {
    cursor: requestedCursor,
    cursorType: "sequence",
    isExpired: false,
  };
}

// Resource with cursor expiry handling
server.resource(
  "event-stream",
  new ResourceTemplate("events://stream{?cursor,limit}", { list: undefined }),
  async (uri, params) => {
    const position = resolveStreamCursor(params.cursor, await getStreamMeta());
    if (position.isExpired) {
      return {
        contents: [{
          uri: uri.href,
          mimeType: "application/json",
          text: JSON.stringify({
            error: "cursor_expired",
            message: "Requested position is beyond retention window",
            earliestAvailable: position.earliestAvailable,
          }),
        }],
      };
    }
    // ... fetch events from cursor position
  }
);