openvibers/openvibe.sdk Commits
Surface opaque event cursors in the SDK events clients (after / next_cursor; afterSeq kept one release); release 0.33.0 (#17)
Changes
@@@ -7,6 +7,21 @@ release may change an API and says so here.++## 0.33.0 (2026-10-05)++++- **`openvibe-sdk/events` surfaces the opaque `after`/`next_cursor` pull cursors (ADR-042 decision 7):** `pull()`++ now takes `after` — a page's opaque `next_cursor` — and sends it as the `after` query parameter; `afterSeq` (the++ numeric `after_seq`) stays for one release and is sent when no cursor is given. `StoredEvent` carries `cursor`++ and `EventsPage` carries `next_cursor`. `iterate()` starts from `after` or `afterSeq` and advances on++ `page.next_cursor` (falling back to `next_after_seq` while the compatibility field remains) after `onPage`, so a++ page that matched no events still advances the saved cursor; resuming from a saved `next_cursor` replays++ nothing. The mock platform (`openvibe-sdk/testing`) models the same page shape and accepts `after`, and++ `README.md`'s pagination note names the cursor. New assertions in `test/events.test.js` and++ `test/app-events.test.js` cover the round-trip, the empty page, and the `after_seq` fallback.++- `iterate()` keeps the mode the caller asked for (a string `afterSeq` such as `'5'` is still the numeric++ position) and finds the head without the numeric compatibility fields: once a server sends only cursors,++ the page whose `next_cursor` does not move ends the walk.++## 0.32.0 (2026-10-05)- **`openvibe-sdk/usage` gains `createUsageReporter` (plan T1 step 7):** the shared reporter Tools, AI, Events and@@@ -2,14 +2,14 @@> Supported browser and server clients for the OpenVibe platform.--**Status:** alpha, v0.32.0 (roadmap Wave 2; developer apps from Wave 20; Media objects v2 from Wave 4; the Tools platform API; Search, Chat and AI clients from WS-F task 4; per-actor limits from WS-R task 4). Every release is tagged (see [CHANGELOG.md](CHANGELOG.md)). Tested against local stub servers and the built-in mock platform only, never against the live platform. Services pin release tags; each repository's `STATUS.json` names the one it uses.++**Status:** alpha, v0.33.0 (roadmap Wave 2; developer apps from Wave 20; Media objects v2 from Wave 4; the Tools platform API; Search, Chat and AI clients from WS-F task 4; per-actor limits from WS-R task 4). Every release is tagged (see [CHANGELOG.md](CHANGELOG.md)). Tested against local stub servers and the built-in mock platform only, never against the live platform. Services pin release tags; each repository's `STATUS.json` names the one it uses.**Plan:** OpenVibe End-to-End Realignment & Implementation Plan, revision 3 (20 Sep 2026), §3.2; roadmap §30.**License:** MIT ([LICENSE](LICENSE)). This package is a library that apps outside the network embed, so it uses MIT. The OpenVibe services themselves stay AGPL-3.0.**If a capability is not in the SDK, it is not public.** Apps call services through `openvibe-sdk` and never build internal routes themselves. A route with no SDK wrapper is internal, even when you can reach it, and it can change without notice. To make a capability public, first define it in OpenVibe.Contracts, then wrap it here.```bash--npm install https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.32.0++npm install https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.33.0```It has no runtime dependencies. It needs Node ≥ 20, or any browser with `fetch`, Web Crypto and `TextDecoder`. There is no build step. The package is CommonJS with ESM entry points (`import` works). Each subpath has its own `.d.ts`. For a page with no bundler, `browser/openvibe-sdk.mjs` is one self-contained ES module of the browser-safe subpaths (see [Browser without a bundler](#browser-without-a-bundler)). It does not depend on `openvibe-contracts` at runtime: it copies the contract types it uses (from Contracts v0.86.0, a devDependency the tests check them against).@@@ -214,7 +214,7 @@ Server-only subpaths are declared `"browser": null` in the exports map, so a bun- Legacy `{ error: 'text' }` bodies become `http.<status>`.- Failures with no HTTP response use `sdk.timeout`, `sdk.deadline_exceeded`, `sdk.aborted`, `sdk.network_error` or `sdk.unknown_service`.- **Tracing.** Every call sends `traceparent` and `X-OpenVibe-Request-Id`. The request id stays the same across retries. `client.withContext({ traceparent, requestId })` or `client.fromRequest(req)` continues the caller's trace with a new span. The `traceparent` option also accepts a getter, for example one backed by AsyncLocalStorage.--- **Pagination.** `paginate(fetchPage, { cursor })` is an async iterator over `{ items, next }` pages. Media and Community use offsets. Events `iterate()` walks `after_seq` and calls `onPage(page)` only after every item of that page was handled, so saving `page.next_after_seq` there is crash-safe.++- **Pagination.** `paginate(fetchPage, { cursor })` is an async iterator over `{ items, next }` pages. Media and Community use offsets. Events `pull()` takes an opaque `after` cursor (a page's `next_cursor`; `afterSeq`, the numeric position, remains for one release) and `iterate()` walks it, calling `onPage(page)` only after every item of that page was handled, so saving `page.next_cursor` there — it advances even when a page matched nothing — is crash-safe.- **Streams and downloads.** `responseType: 'response'` resolves with the raw fetch `Response` (`data` and `response`, body unread). Error statuses are still read and thrown. The timeout covers the headers only, and `signal` still cancels the body. `parseSSE(res.body)` (from `openvibe-sdk/realtime`) iterates a `text/event-stream`.### Auth details@@@ -3,7 +3,7 @@"stage": "alpha","code": true,"wave": "W2",-- "version": "0.32.0",++ "version": "0.33.0","plan": "OpenVibe End-to-End Realignment & Implementation Plan, revision 3 (20 Sep 2026), §3.2; roadmap §30","deployed": false,"deployment": "none: a package released as git tags; services and apps pin a tag",@@@ -1,4 +1,4 @@--// Generated by scripts/browser-bundle.js from browser.js (openvibe-sdk 0.32.0, MIT). Do not edit.++// Generated by scripts/browser-bundle.js from browser.js (openvibe-sdk 0.33.0, MIT). Do not edit.// One self-contained ES module: core at the top level plus the browser-safe subpaths as namespaces// (auth, registry, modules, realtime, media, community, bot, search, chat, ai, jobs, tools, projects, geo). No server-only code: no service tokens, JWT verification, events, identity./* eslint-disable */@@@ -42,7 +42,7 @@ const { paginate, offsetPager } = __load(7);const semver = __load(6);/** The package version (test/version.test.js keeps it equal to package.json). */--const SDK_VERSION = '0.32.0';++const SDK_VERSION = '0.33.0';module.exports = {createClient,@@@ -1,12 +1,12 @@{"name": "openvibe-sdk",-- "version": "0.32.0",++ "version": "0.33.0","lockfileVersion": 3,"requires": true,"packages": {"": {"name": "openvibe-sdk",-- "version": "0.32.0",++ "version": "0.33.0","license": "MIT","devDependencies": {"@electric-sql/pglite": "^0.5.8",@@@ -1,6 +1,6 @@{"name": "openvibe-sdk",-- "version": "0.32.0",++ "version": "0.33.0","description": "Supported browser and server clients for the OpenVibe platform: auth (incl. developer apps), registry discovery, identity, user modules, events, realtime, media, community, Bot robots, search, chat, AI runs, Tools (registry, run API and jobs), VIP memberships and developer projects, the async PostgreSQL data layer (db) and the Valkey cache, limits store, queue and pub/sub, with deadlines, safe retries, problem errors and trace propagation.","main": "index.js","types": "types/index.d.ts",@@@ -8,7 +8,7 @@ const { paginate, offsetPager } = require('./paginate');const semver = require('./semver');/** The package version (test/version.test.js keeps it equal to package.json). */--const SDK_VERSION = '0.32.0';++const SDK_VERSION = '0.33.0';module.exports = {createClient,@@@ -181,32 +181,49 @@ function createEventsClient(client, { source, baseUrl } = {}) {});}-- /** One page: { events: [{ seq, event }], next_after_seq, latest_seq, gap? } */-- function pull({ topic = '*', afterSeq = 0, limit } = {}) {-- return call({ path: '/api/v1/events', query: { topic: Array.isArray(topic) ? topic.join(',') : topic, after_seq: afterSeq, limit } });++ /**++ * One page: { events: [{ seq, cursor, event }], next_cursor, next_after_seq, latest_seq, gap? }.++ * Pass `after` — a page's opaque `next_cursor` — to resume exactly where the previous page ended;++ * `afterSeq`, the numeric position, remains for one release.++ */++ function pull({ topic = '*', after, afterSeq = 0, limit } = {}) {++ return call({ path: '/api/v1/events', query: {++ topic: Array.isArray(topic) ? topic.join(',') : topic, limit,++ ...(after == null ? { after_seq: afterSeq } : { after }),++ } });}/**-- * Async iterator over { seq, event } from afterSeq up to the current head.++ * Async iterator over { seq, cursor, event } from `after` (an opaque cursor) or `afterSeq` up to++ * the current head.** onGap(gap) called before a page's items when retention already pruned part of the range* ({ from_seq, to_seq }): nothing can replay it, so resync derived state.* onPage(page) called AFTER every item of that page was yielded and handled (your loop body* ran for the page's last item and asked for the next one). Save-- * page.next_after_seq there as your durable cursor: it also moves past events-- * that did not match your topics, and a crash (or a break/throw in your loop)-- * before onPage leaves the cursor on the last page you finished, so nothing is-- * skipped. Pages with no matching events still call onPage.++ * page.next_cursor there as your durable cursor (page.next_after_seq while the++ * compatibility field remains): it also moves past events that did not match your++ * topics — a page with no matching events still calls onPage and advances it — and++ * a crash (or a break/throw in your loop) before onPage leaves the cursor on the++ * last page you finished, so nothing is skipped.*/-- async function* iterate({ topic, afterSeq = 0, limit, onGap, onPage, maxPages = Infinity } = {}) {-- let cursor = afterSeq;++ async function* iterate({ topic, after, afterSeq = 0, limit, onGap, onPage, maxPages = Infinity } = {}) {++ // The caller picks the mode (an opaque `after`, or the numeric `afterSeq`, which may arrive as a string), and++ // the iterator moves to cursors as soon as the server hands one back.++ let useCursor = after != null;++ let cursor = useCursor ? after : afterSeq;for (let pages = 0; pages < maxPages; pages++) {-- const page = await pull({ topic, afterSeq: cursor, limit });++ const page = await pull({ topic, ...(useCursor ? { after: cursor } : { afterSeq: cursor }), limit });if (page.gap && onGap) await onGap(page.gap);for (const item of page.events || []) yield item;if (onPage) await onPage(page);-- const next = page.next_after_seq;-- if (next == null || next === cursor || !(next < page.latest_seq)) return;++ const hasCursor = page.next_cursor != null;++ const next = hasCursor ? page.next_cursor : page.next_after_seq;++ // The head: a page that does not move the position, or (while the server still sends the numeric++ // compatibility fields) one that reached latest_seq. Without those fields the unchanged cursor ends it.++ if (next == null || next === cursor) return;++ if (page.next_after_seq != null && page.latest_seq != null && !(page.next_after_seq < page.latest_seq)) return;++ useCursor = hasCursor;cursor = next;}}@@@ -357,6 +357,15 @@ function createMockPlatform(opts = {}) {}// ── Events ──────────────────────────────────────────────++ // Opaque cursors (ADR-042 decision 7): `c1.<epoch>.<base64url(seq)>`. The mock keeps one epoch.++ const eventCursor = (seq) => `c1.0.${Buffer.from(String(seq), 'utf8').toString('base64url')}`;++ const eventSeqFromCursor = (input) => {++ if (typeof input !== 'string') return null;++ const m = /^c1\.0\.([A-Za-z0-9_-]+)$/.exec(input);++ if (!m) return null;++ const seq = Number(Buffer.from(m[1], 'base64url').toString('utf8'));++ return Number.isSafeInteger(seq) && seq >= 0 && eventCursor(seq) === input ? seq : null;++ };function visible(viewer, e) {if (e.project_id || (e.env || 'production') !== 'production') return false; // never streamedif (e.event.visibility === 'public') return true;@@@ -448,9 +457,12 @@ function createMockPlatform(opts = {}) {if (!patterns.length || patterns.length > 20 || !patterns.every(appRules.isValidPattern)) return problem(400, 'events.bad_topic', 'topic must be 1..20 comma-separated patterns');const scopeErr = scopeError(who.principal, patterns);if (scopeErr) return problem(403, 'events.topic_not_allowed', scopeErr);-- const after = Number(url.searchParams.get('after_seq') || 0);++ const afterCursor = url.searchParams.get('after');++ const after = afterCursor ? eventSeqFromCursor(afterCursor) : Number(url.searchParams.get('after_seq') || 0);const limit = Number(url.searchParams.get('limit') || 100);-- if (!Number.isInteger(after) || after < 0 || !Number.isInteger(limit) || limit < 1 || limit > 1000) return problem(400, 'events.bad_request', 'after_seq must be >= 0 and limit 1..1000');++ if (after === null || !Number.isInteger(after) || after < 0 || !Number.isInteger(limit) || limit < 1 || limit > 1000) {++ return problem(400, 'events.bad_request', afterCursor ? 'after must be an opaque cursor' : 'after_seq must be >= 0 and limit 1..1000');++ }const pats = patterns.map((x) => [x, topicRegex(x)]);const out = [];const page = {};@@@ -464,17 +476,17 @@ function createMockPlatform(opts = {}) {if (e.seq <= from) continue;if (out.length >= limit) break;cursor = e.seq;-- if (pats.some(([x, re]) => re.test(e.event.event_type) && readable(who.principal, e, x))) out.push({ seq: e.seq, event: e.event });++ if (pats.some(([x, re]) => re.test(e.event.event_type) && readable(who.principal, e, x))) out.push({ seq: e.seq, cursor: eventCursor(e.seq), event: e.event });}if (out.length < limit) cursor = Math.max(cursor, lastSeq);-- return json(200, { ...page, events: out, next_after_seq: cursor, latest_seq: lastSeq });++ return json(200, { ...page, events: out, next_after_seq: cursor, next_cursor: eventCursor(cursor), latest_seq: lastSeq });}if ((r = path.match(/^\/api\/v1\/events\/([^/]+)$/)) && req.method === 'GET') {const who = eventsPrincipal(req, 'events.event.read', 'events.app.read');if (who.res) return who.res;const e = events.find((x) => x.event.event_id === decodeURIComponent(r[1]));const ok = e && (who.principal.kind === 'app' ? appRules.visibleToApp(e, who.principal) : (e.env || 'production') === 'production');-- return ok ? json(200, { seq: e.seq, event: e.event }) : problem(404, 'events.not_found', 'no such event (or pruned by retention)');++ return ok ? json(200, { seq: e.seq, cursor: eventCursor(e.seq), event: e.event }) : problem(404, 'events.not_found', 'no such event (or pruned by retention)');}if (path === '/api/v1/checkpoints') {const who = eventsPrincipal(req, 'events.event.read', 'events.app.read');@@@ -130,6 +130,11 @@ run([assert.deepEqual(ids(await r.pull()), [sbx.event_id], 'another app of the project, same env');assert.deepEqual(ids(await a.pull({ topic: 'order.*' })), [sbx.event_id], 'a sandbox app never sees production events');++ const orderPage = await a.pull({ topic: 'order.*' });++ assert.match(orderPage.events[0].cursor, /^c1\./, 'a page item carries its cursor');++ assert.match(orderPage.next_cursor, /^c1\./, 'a page carries its opaque next_cursor');++ assert.deepEqual(ids(await a.pull({ topic: 'order.*', after: orderPage.next_cursor })), [], 'after resumes from the cursor');++ assert.deepEqual(ids(await a.pull({ topic: 'order.*', afterSeq: orderPage.next_after_seq })), [], 'afterSeq still works for one release');assert.deepEqual(ids(await a2.pull()), [prod.event_id], 'a production app never sees sandbox events');assert.deepEqual(ids(await b.pull()), [other.event_id]);assert.deepEqual(ids(await a.pull({ platformTopics: ['live.*'] })), [sbx.event_id, pub.event_id], 'public first-party only, never internal');@@@ -141,8 +146,10 @@ run([assert.equal(await b.get(sbx.event_id), null, 'another project\'s event does not exist for this app');assert.equal(await a2.get(sbx.event_id), null);const seen = [];-- for await (const item of a.iterate({ topic: '*', platformTopics: ['live.stream.*'] })) seen.push(item.event.event_id);++ const cursors = [];++ for await (const item of a.iterate({ topic: '*', platformTopics: ['live.stream.*'], onPage: (p) => cursors.push(p.next_cursor) })) seen.push(item.event.event_id);assert.deepEqual(seen, [sbx.event_id, pub.event_id]);++ assert.match(cursors.at(-1), /^c1\./);// First-party readers: never sandbox; app events only through an app.* pattern.assert.ok(!(await live.pull({ topic: '*' })).events.some((e) => e.event.event_type.startsWith('app.')));@@@ -71,20 +71,49 @@ run([assert.deepEqual(page.events.map((e) => e.seq), [1, 3]);assert.equal(page.next_after_seq, 3);assert.equal(page.latest_seq, 5);++ assert.match(page.next_cursor, /^c1\./, 'the page carries an opaque next_cursor');++ assert.match(page.events[0].cursor, /^c1\./, 'each event carries its cursor');++ // The opaque cursor round-trips: `after` resumes exactly where the page ended.++ const rest = await events.pull({ topic: 'media.vod.*', after: page.next_cursor, limit: 2 });++ assert.deepEqual(rest.events.map((e) => e.seq), [5]);++ assert.equal((await events.get(platform.state.events[0].event.event_id)).cursor, page.events[0].cursor);++ // afterSeq, the numeric position, still works for one release.++ assert.deepEqual((await events.pull({ topic: 'media.vod.*', afterSeq: page.next_after_seq, limit: 2 })).events.map((e) => e.seq), [5]);const seen = [];const cursors = [];-- for await (const { seq, event } of events.iterate({ topic: ['media.vod.*'], afterSeq: 0, limit: 2, onPage: (p) => cursors.push(p.next_after_seq) })) {++ for await (const { seq, event } of events.iterate({ topic: ['media.vod.*'], afterSeq: 0, limit: 2, onPage: (p) => cursors.push(p.next_cursor) })) {seen.push(seq);assert.equal(event.event_type, 'media.vod.ready');}assert.deepEqual(seen, [1, 3, 5]);-- assert.equal(cursors.at(-1), 5);++ assert.match(cursors.at(-1), /^c1\./);++ assert.equal(cursors.at(-1), (await events.pull({ topic: 'media.vod.*', afterSeq: 5 })).next_cursor, 'iterate ends on the page cursor');++ const resumed = [];++ for await (const { seq } of events.iterate({ topic: 'media.vod.*', after: page.next_cursor, limit: 2 })) resumed.push(seq);++ assert.deepEqual(resumed, [5], 'iterate resumes from an opaque cursor');assert.equal((await events.get(platform.state.events[0].event.event_id)).seq, 1);assert.equal(await events.get('evt_01J00000000000000000000000'), null);await events.setCheckpoint('media.vod.*', 5);assert.equal((await events.getCheckpoint('media.vod.*')).cursor, 5);}],++ ['iterate: onPage advances the cursor even on a page that matches no events', async () => {++ const { platform, events } = setup();++ platform.publishEvent({ event_type: 'live.other.thing', source: 'media', actor, subject: { type: 'vod', id: '1' } });++ const pages = [];++ for await (const e of events.iterate({ topic: 'media.vod.*', onPage: (p) => pages.push(p) })) void e;++ assert.equal(pages.length, 1, 'the empty page still called onPage');++ assert.deepEqual(pages[0].events, []);++ assert.equal(pages[0].next_after_seq, 1);++ assert.match(pages[0].next_cursor, /^c1\./, 'the cursor advanced past the scanned, non-matching event');++ // Resuming from that cursor does not replay the scanned event: the new match is seen once.++ const saved = pages[0].next_cursor;++ platform.publishEvent({ event_type: 'media.vod.ready', source: 'media', actor, subject: { type: 'vod', id: '2' } });++ const seen = [];++ for await (const e of events.iterate({ topic: 'media.vod.*', after: saved })) seen.push(e.seq);++ assert.deepEqual(seen, [2]);++ }],++['iterate reports a retention gap through onGap', async () => {const srv = await stubServer((req, res) => {const after = Number(new URL(req.url, 'http://x').searchParams.get('after_seq'));@@@ -288,4 +317,25 @@ run([await worker.stop();await srv.close();}],++ ['iterate keeps paging when the server sends only cursors, and a string afterSeq stays numeric', async () => {++ const asked = [];++ // A server past the compatibility release: no next_after_seq / latest_seq, the same cursor at the head.++ const pages = { '': { events: [{ seq: 1, cursor: 'c1', event: { id: 'a' } }], next_cursor: 'c1' },++ c1: { events: [{ seq: 2, cursor: 'c2', event: { id: 'b' } }], next_cursor: 'c2' },++ c2: { events: [], next_cursor: 'c2' } };++ const fetch = async (url) => {++ const u = new URL(url);++ asked.push(u.search);++ const body = u.searchParams.has('after') ? pages[u.searchParams.get('after')] : { ...pages[''], next_after_seq: 1, latest_seq: 2 };++ return new Response(JSON.stringify(body), { status: 200, headers: { 'content-type': 'application/json' } });++ };++ const events = createEventsClient(createClient({ fetch }), { source: 'live', baseUrl: 'http://events.test' });++ const seen = [];++ for await (const item of events.iterate({ afterSeq: '0' })) seen.push(item.event.id);++ assert.deepEqual(seen, ['a', 'b'], 'every page up to the head');++ assert.match(asked[0], /after_seq=0/, 'a string afterSeq is still the numeric position');++ assert.match(asked[1], /after=c1/);++ assert.match(asked[2], /after=c2/);++ assert.equal(asked.length, 3, 'the unchanged cursor at the head ends it');++ }],]);@@@ -3,9 +3,9 @@ import type { OpenVibeClient, EventEnvelope } from './core';/** What a producer passes: event_id, timestamp, source, version, payload and trace_id are filled in. */export type EventInput = Omit<EventEnvelope, 'event_id' | 'timestamp' | 'source' | 'version' | 'payload'> & Partial<Pick<EventEnvelope, 'event_id' | 'timestamp' | 'source' | 'version' | 'payload'>>;export interface PublishResult { event_id: string; seq: number; duplicate: boolean; }--export interface StoredEvent { seq: number; event: EventEnvelope; }++export interface StoredEvent { seq: number; cursor: string; event: EventEnvelope; }export interface Gap { from_seq: number; to_seq: number; reason?: string; latest_seq?: number; }--export interface EventsPage { events: StoredEvent[]; next_after_seq: number; latest_seq: number; gap?: Gap; }++export interface EventsPage { events: StoredEvent[]; next_cursor: string; next_after_seq: number; latest_seq: number; gap?: Gap; }export interface Subscription {id: string;consumer: string;@@@ -23,9 +23,9 @@ export interface EventsClient {prepare(envelope: EventInput, opts?: { traceId?: string; now?: number }): EventEnvelope;publish(envelope: EventInput, opts?: { traceparent?: string }): Promise<PublishResult>;publish(envelopes: EventInput[], opts?: { traceparent?: string }): Promise<{ results: PublishResult[] }>;-- pull(opts?: { topic?: string | string[]; afterSeq?: number; limit?: number }): Promise<EventsPage>;-- /** onPage runs after every item of that page was yielded and handled: save page.next_after_seq there. */-- iterate(opts?: { topic?: string | string[]; afterSeq?: number; limit?: number; maxPages?: number; onGap?: (gap: Gap) => void | Promise<void>; onPage?: (page: EventsPage) => void | Promise<void> }): AsyncGenerator<StoredEvent, void, unknown>;++ pull(opts?: { topic?: string | string[]; after?: string; afterSeq?: number; limit?: number }): Promise<EventsPage>;++ /** onPage runs after every item of that page was yielded and handled: save page.next_cursor there. */++ iterate(opts?: { topic?: string | string[]; after?: string; afterSeq?: number; limit?: number; maxPages?: number; onGap?: (gap: Gap) => void | Promise<void>; onPage?: (page: EventsPage) => void | Promise<void> }): AsyncGenerator<StoredEvent, void, unknown>;get(eventId: string): Promise<StoredEvent | null>;getCheckpoint(topic: string): Promise<{ consumer: string; topic: string; cursor: number; updated_at: string | null }>;setCheckpoint(topic: string, cursor: number): Promise<{ consumer: string; topic: string; cursor: number }>;@@@ -65,8 +65,8 @@ export interface AppEventsClient {publish(envelope: AppEventInput, opts?: { traceparent?: string }): Promise<PublishResult>;publish(envelopes: AppEventInput[], opts?: { traceparent?: string }): Promise<{ results: PublishResult[] }>;/** topic is project-relative (default '*'); platformTopics are first-party patterns (public events only). */-- pull(opts?: { topic?: string | string[]; platformTopics?: string[]; afterSeq?: number; limit?: number }): Promise<EventsPage>;-- iterate(opts?: { topic?: string | string[]; platformTopics?: string[]; afterSeq?: number; limit?: number; maxPages?: number; onGap?: (gap: Gap) => void | Promise<void>; onPage?: (page: EventsPage) => void | Promise<void> }): AsyncGenerator<StoredEvent, void, unknown>;++ pull(opts?: { topic?: string | string[]; platformTopics?: string[]; after?: string; afterSeq?: number; limit?: number }): Promise<EventsPage>;++ iterate(opts?: { topic?: string | string[]; platformTopics?: string[]; after?: string; afterSeq?: number; limit?: number; maxPages?: number; onGap?: (gap: Gap) => void | Promise<void>; onPage?: (page: EventsPage) => void | Promise<void> }): AsyncGenerator<StoredEvent, void, unknown>;get(eventId: string): Promise<StoredEvent | null>;getCheckpoint(topic: string): Promise<{ consumer: string; topic: string; cursor: number; updated_at: string | null }>;setCheckpoint(topic: string, cursor: number): Promise<{ consumer: string; topic: string; cursor: number }>;
@@ -7,6 +7,21 @@ release may change an API and says so here.
@@ -7,6 +7,21 @@ release may change an API and says so here.
+## 0.33.0 (2026-10-05)
+
+- **`openvibe-sdk/events` surfaces the opaque `after`/`next_cursor` pull cursors (ADR-042 decision 7):** `pull()`
+ now takes `after` — a page's opaque `next_cursor` — and sends it as the `after` query parameter; `afterSeq` (the
+ numeric `after_seq`) stays for one release and is sent when no cursor is given. `StoredEvent` carries `cursor`
+ and `EventsPage` carries `next_cursor`. `iterate()` starts from `after` or `afterSeq` and advances on
+ `page.next_cursor` (falling back to `next_after_seq` while the compatibility field remains) after `onPage`, so a
+ page that matched no events still advances the saved cursor; resuming from a saved `next_cursor` replays
+ nothing. The mock platform (`openvibe-sdk/testing`) models the same page shape and accepts `after`, and
+ `README.md`'s pagination note names the cursor. New assertions in `test/events.test.js` and
+ `test/app-events.test.js` cover the round-trip, the empty page, and the `after_seq` fallback.
+- `iterate()` keeps the mode the caller asked for (a string `afterSeq` such as `'5'` is still the numeric
+ position) and finds the head without the numeric compatibility fields: once a server sends only cursors,
+ the page whose `next_cursor` does not move ends the walk.
+
## 0.32.0 (2026-10-05)
## 0.32.0 (2026-10-05)
- **`openvibe-sdk/usage` gains `createUsageReporter` (plan T1 step 7):** the shared reporter Tools, AI, Events and
- **`openvibe-sdk/usage` gains `createUsageReporter` (plan T1 step 7):** the shared reporter Tools, AI, Events and
@@ -2,14 +2,14 @@
@@ -2,14 +2,14 @@
> Supported browser and server clients for the OpenVibe platform.
> Supported browser and server clients for the OpenVibe platform.
-**Status:** alpha, v0.32.0 (roadmap Wave 2; developer apps from Wave 20; Media objects v2 from Wave 4; the Tools platform API; Search, Chat and AI clients from WS-F task 4; per-actor limits from WS-R task 4). Every release is tagged (see [CHANGELOG.md](CHANGELOG.md)). Tested against local stub servers and the built-in mock platform only, never against the live platform. Services pin release tags; each repository's `STATUS.json` names the one it uses.
+**Status:** alpha, v0.33.0 (roadmap Wave 2; developer apps from Wave 20; Media objects v2 from Wave 4; the Tools platform API; Search, Chat and AI clients from WS-F task 4; per-actor limits from WS-R task 4). Every release is tagged (see [CHANGELOG.md](CHANGELOG.md)). Tested against local stub servers and the built-in mock platform only, never against the live platform. Services pin release tags; each repository's `STATUS.json` names the one it uses.
**Plan:** OpenVibe End-to-End Realignment & Implementation Plan, revision 3 (20 Sep 2026), §3.2; roadmap §30.
**Plan:** OpenVibe End-to-End Realignment & Implementation Plan, revision 3 (20 Sep 2026), §3.2; roadmap §30.
**License:** MIT ([LICENSE](LICENSE)). This package is a library that apps outside the network embed, so it uses MIT. The OpenVibe services themselves stay AGPL-3.0.
**License:** MIT ([LICENSE](LICENSE)). This package is a library that apps outside the network embed, so it uses MIT. The OpenVibe services themselves stay AGPL-3.0.
**If a capability is not in the SDK, it is not public.** Apps call services through `openvibe-sdk` and never build internal routes themselves. A route with no SDK wrapper is internal, even when you can reach it, and it can change without notice. To make a capability public, first define it in OpenVibe.Contracts, then wrap it here.
**If a capability is not in the SDK, it is not public.** Apps call services through `openvibe-sdk` and never build internal routes themselves. A route with no SDK wrapper is internal, even when you can reach it, and it can change without notice. To make a capability public, first define it in OpenVibe.Contracts, then wrap it here.
```bash
```bash
-npm install https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.32.0
+npm install https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.33.0
```
```
It has no runtime dependencies. It needs Node ≥ 20, or any browser with `fetch`, Web Crypto and `TextDecoder`. There is no build step. The package is CommonJS with ESM entry points (`import` works). Each subpath has its own `.d.ts`. For a page with no bundler, `browser/openvibe-sdk.mjs` is one self-contained ES module of the browser-safe subpaths (see [Browser without a bundler](#browser-without-a-bundler)). It does not depend on `openvibe-contracts` at runtime: it copies the contract types it uses (from Contracts v0.86.0, a devDependency the tests check them against).
It has no runtime dependencies. It needs Node ≥ 20, or any browser with `fetch`, Web Crypto and `TextDecoder`. There is no build step. The package is CommonJS with ESM entry points (`import` works). Each subpath has its own `.d.ts`. For a page with no bundler, `browser/openvibe-sdk.mjs` is one self-contained ES module of the browser-safe subpaths (see [Browser without a bundler](#browser-without-a-bundler)). It does not depend on `openvibe-contracts` at runtime: it copies the contract types it uses (from Contracts v0.86.0, a devDependency the tests check them against).
@@ -214,7 +214,7 @@ Server-only subpaths are declared `"browser": null` in the exports map, so a bun
@@ -214,7 +214,7 @@ Server-only subpaths are declared `"browser": null` in the exports map, so a bun
- Legacy `{ error: 'text' }` bodies become `http.<status>`.
- Legacy `{ error: 'text' }` bodies become `http.<status>`.
- Failures with no HTTP response use `sdk.timeout`, `sdk.deadline_exceeded`, `sdk.aborted`, `sdk.network_error` or `sdk.unknown_service`.
- Failures with no HTTP response use `sdk.timeout`, `sdk.deadline_exceeded`, `sdk.aborted`, `sdk.network_error` or `sdk.unknown_service`.
- **Tracing.** Every call sends `traceparent` and `X-OpenVibe-Request-Id`. The request id stays the same across retries. `client.withContext({ traceparent, requestId })` or `client.fromRequest(req)` continues the caller's trace with a new span. The `traceparent` option also accepts a getter, for example one backed by AsyncLocalStorage.
- **Tracing.** Every call sends `traceparent` and `X-OpenVibe-Request-Id`. The request id stays the same across retries. `client.withContext({ traceparent, requestId })` or `client.fromRequest(req)` continues the caller's trace with a new span. The `traceparent` option also accepts a getter, for example one backed by AsyncLocalStorage.
-- **Pagination.** `paginate(fetchPage, { cursor })` is an async iterator over `{ items, next }` pages. Media and Community use offsets. Events `iterate()` walks `after_seq` and calls `onPage(page)` only after every item of that page was handled, so saving `page.next_after_seq` there is crash-safe.
+- **Pagination.** `paginate(fetchPage, { cursor })` is an async iterator over `{ items, next }` pages. Media and Community use offsets. Events `pull()` takes an opaque `after` cursor (a page's `next_cursor`; `afterSeq`, the numeric position, remains for one release) and `iterate()` walks it, calling `onPage(page)` only after every item of that page was handled, so saving `page.next_cursor` there — it advances even when a page matched nothing — is crash-safe.
- **Streams and downloads.** `responseType: 'response'` resolves with the raw fetch `Response` (`data` and `response`, body unread). Error statuses are still read and thrown. The timeout covers the headers only, and `signal` still cancels the body. `parseSSE(res.body)` (from `openvibe-sdk/realtime`) iterates a `text/event-stream`.
- **Streams and downloads.** `responseType: 'response'` resolves with the raw fetch `Response` (`data` and `response`, body unread). Error statuses are still read and thrown. The timeout covers the headers only, and `signal` still cancels the body. `parseSSE(res.body)` (from `openvibe-sdk/realtime`) iterates a `text/event-stream`.
### Auth details
### Auth details
@@ -3,7 +3,7 @@
@@ -3,7 +3,7 @@
"stage": "alpha",
"stage": "alpha",
"code": true,
"code": true,
"wave": "W2",
"wave": "W2",
- "version": "0.32.0",
+ "version": "0.33.0",
"plan": "OpenVibe End-to-End Realignment & Implementation Plan, revision 3 (20 Sep 2026), §3.2; roadmap §30",
"plan": "OpenVibe End-to-End Realignment & Implementation Plan, revision 3 (20 Sep 2026), §3.2; roadmap §30",
"deployed": false,
"deployed": false,
"deployment": "none: a package released as git tags; services and apps pin a tag",
"deployment": "none: a package released as git tags; services and apps pin a tag",
@@ -1,4 +1,4 @@
@@ -1,4 +1,4 @@
-// Generated by scripts/browser-bundle.js from browser.js (openvibe-sdk 0.32.0, MIT). Do not edit.
+// Generated by scripts/browser-bundle.js from browser.js (openvibe-sdk 0.33.0, MIT). Do not edit.
// One self-contained ES module: core at the top level plus the browser-safe subpaths as namespaces
// One self-contained ES module: core at the top level plus the browser-safe subpaths as namespaces
// (auth, registry, modules, realtime, media, community, bot, search, chat, ai, jobs, tools, projects, geo). No server-only code: no service tokens, JWT verification, events, identity.
// (auth, registry, modules, realtime, media, community, bot, search, chat, ai, jobs, tools, projects, geo). No server-only code: no service tokens, JWT verification, events, identity.
/* eslint-disable */
/* eslint-disable */
@@ -42,7 +42,7 @@ const { paginate, offsetPager } = __load(7);
@@ -42,7 +42,7 @@ const { paginate, offsetPager } = __load(7);
const semver = __load(6);
const semver = __load(6);
/** The package version (test/version.test.js keeps it equal to package.json). */
/** The package version (test/version.test.js keeps it equal to package.json). */
-const SDK_VERSION = '0.32.0';
+const SDK_VERSION = '0.33.0';
module.exports = {
module.exports = {
createClient,
createClient,
@@ -1,12 +1,12 @@
@@ -1,12 +1,12 @@
{
{
"name": "openvibe-sdk",
"name": "openvibe-sdk",
- "version": "0.32.0",
+ "version": "0.33.0",
"lockfileVersion": 3,
"lockfileVersion": 3,
"requires": true,
"requires": true,
"packages": {
"packages": {
"": {
"": {
"name": "openvibe-sdk",
"name": "openvibe-sdk",
- "version": "0.32.0",
+ "version": "0.33.0",
"license": "MIT",
"license": "MIT",
"devDependencies": {
"devDependencies": {
"@electric-sql/pglite": "^0.5.8",
"@electric-sql/pglite": "^0.5.8",
@@ -1,6 +1,6 @@
@@ -1,6 +1,6 @@
{
{
"name": "openvibe-sdk",
"name": "openvibe-sdk",
- "version": "0.32.0",
+ "version": "0.33.0",
"description": "Supported browser and server clients for the OpenVibe platform: auth (incl. developer apps), registry discovery, identity, user modules, events, realtime, media, community, Bot robots, search, chat, AI runs, Tools (registry, run API and jobs), VIP memberships and developer projects, the async PostgreSQL data layer (db) and the Valkey cache, limits store, queue and pub/sub, with deadlines, safe retries, problem errors and trace propagation.",
"description": "Supported browser and server clients for the OpenVibe platform: auth (incl. developer apps), registry discovery, identity, user modules, events, realtime, media, community, Bot robots, search, chat, AI runs, Tools (registry, run API and jobs), VIP memberships and developer projects, the async PostgreSQL data layer (db) and the Valkey cache, limits store, queue and pub/sub, with deadlines, safe retries, problem errors and trace propagation.",
"main": "index.js",
"main": "index.js",
"types": "types/index.d.ts",
"types": "types/index.d.ts",
@@ -8,7 +8,7 @@ const { paginate, offsetPager } = require('./paginate');
@@ -8,7 +8,7 @@ const { paginate, offsetPager } = require('./paginate');
const semver = require('./semver');
const semver = require('./semver');
/** The package version (test/version.test.js keeps it equal to package.json). */
/** The package version (test/version.test.js keeps it equal to package.json). */
-const SDK_VERSION = '0.32.0';
+const SDK_VERSION = '0.33.0';
module.exports = {
module.exports = {
createClient,
createClient,
@@ -181,32 +181,49 @@ function createEventsClient(client, { source, baseUrl } = {}) {
@@ -181,32 +181,49 @@ function createEventsClient(client, { source, baseUrl } = {}) {
});
});
}
}
- /** One page: { events: [{ seq, event }], next_after_seq, latest_seq, gap? } */
- function pull({ topic = '*', afterSeq = 0, limit } = {}) {
- return call({ path: '/api/v1/events', query: { topic: Array.isArray(topic) ? topic.join(',') : topic, after_seq: afterSeq, limit } });
+ /**
+ * One page: { events: [{ seq, cursor, event }], next_cursor, next_after_seq, latest_seq, gap? }.
+ * Pass `after` — a page's opaque `next_cursor` — to resume exactly where the previous page ended;
+ * `afterSeq`, the numeric position, remains for one release.
+ */
+ function pull({ topic = '*', after, afterSeq = 0, limit } = {}) {
+ return call({ path: '/api/v1/events', query: {
+ topic: Array.isArray(topic) ? topic.join(',') : topic, limit,
+ ...(after == null ? { after_seq: afterSeq } : { after }),
+ } });
}
}
/**
/**
- * Async iterator over { seq, event } from afterSeq up to the current head.
+ * Async iterator over { seq, cursor, event } from `after` (an opaque cursor) or `afterSeq` up to
+ * the current head.
*
*
* onGap(gap) called before a page's items when retention already pruned part of the range
* onGap(gap) called before a page's items when retention already pruned part of the range
* ({ from_seq, to_seq }): nothing can replay it, so resync derived state.
* ({ from_seq, to_seq }): nothing can replay it, so resync derived state.
* onPage(page) called AFTER every item of that page was yielded and handled (your loop body
* onPage(page) called AFTER every item of that page was yielded and handled (your loop body
* ran for the page's last item and asked for the next one). Save
* ran for the page's last item and asked for the next one). Save
- * page.next_after_seq there as your durable cursor: it also moves past events
- * that did not match your topics, and a crash (or a break/throw in your loop)
- * before onPage leaves the cursor on the last page you finished, so nothing is
- * skipped. Pages with no matching events still call onPage.
+ * page.next_cursor there as your durable cursor (page.next_after_seq while the
+ * compatibility field remains): it also moves past events that did not match your
+ * topics — a page with no matching events still calls onPage and advances it — and
+ * a crash (or a break/throw in your loop) before onPage leaves the cursor on the
+ * last page you finished, so nothing is skipped.
*/
*/
- async function* iterate({ topic, afterSeq = 0, limit, onGap, onPage, maxPages = Infinity } = {}) {
- let cursor = afterSeq;
+ async function* iterate({ topic, after, afterSeq = 0, limit, onGap, onPage, maxPages = Infinity } = {}) {
+ // The caller picks the mode (an opaque `after`, or the numeric `afterSeq`, which may arrive as a string), and
+ // the iterator moves to cursors as soon as the server hands one back.
+ let useCursor = after != null;
+ let cursor = useCursor ? after : afterSeq;
for (let pages = 0; pages < maxPages; pages++) {
for (let pages = 0; pages < maxPages; pages++) {
- const page = await pull({ topic, afterSeq: cursor, limit });
+ const page = await pull({ topic, ...(useCursor ? { after: cursor } : { afterSeq: cursor }), limit });
if (page.gap && onGap) await onGap(page.gap);
if (page.gap && onGap) await onGap(page.gap);
for (const item of page.events || []) yield item;
for (const item of page.events || []) yield item;
if (onPage) await onPage(page);
if (onPage) await onPage(page);
- const next = page.next_after_seq;
- if (next == null || next === cursor || !(next < page.latest_seq)) return;
+ const hasCursor = page.next_cursor != null;
+ const next = hasCursor ? page.next_cursor : page.next_after_seq;
+ // The head: a page that does not move the position, or (while the server still sends the numeric
+ // compatibility fields) one that reached latest_seq. Without those fields the unchanged cursor ends it.
+ if (next == null || next === cursor) return;
+ if (page.next_after_seq != null && page.latest_seq != null && !(page.next_after_seq < page.latest_seq)) return;
+ useCursor = hasCursor;
cursor = next;
cursor = next;
}
}
}
}
@@ -357,6 +357,15 @@ function createMockPlatform(opts = {}) {
@@ -357,6 +357,15 @@ function createMockPlatform(opts = {}) {
}
}
// ── Events ──────────────────────────────────────────────
// ── Events ──────────────────────────────────────────────
+ // Opaque cursors (ADR-042 decision 7): `c1.<epoch>.<base64url(seq)>`. The mock keeps one epoch.
+ const eventCursor = (seq) => `c1.0.${Buffer.from(String(seq), 'utf8').toString('base64url')}`;
+ const eventSeqFromCursor = (input) => {
+ if (typeof input !== 'string') return null;
+ const m = /^c1\.0\.([A-Za-z0-9_-]+)$/.exec(input);
+ if (!m) return null;
+ const seq = Number(Buffer.from(m[1], 'base64url').toString('utf8'));
+ return Number.isSafeInteger(seq) && seq >= 0 && eventCursor(seq) === input ? seq : null;
+ };
function visible(viewer, e) {
function visible(viewer, e) {
if (e.project_id || (e.env || 'production') !== 'production') return false; // never streamed
if (e.project_id || (e.env || 'production') !== 'production') return false; // never streamed
if (e.event.visibility === 'public') return true;
if (e.event.visibility === 'public') return true;
@@ -448,9 +457,12 @@ function createMockPlatform(opts = {}) {
@@ -448,9 +457,12 @@ function createMockPlatform(opts = {}) {
if (!patterns.length || patterns.length > 20 || !patterns.every(appRules.isValidPattern)) return problem(400, 'events.bad_topic', 'topic must be 1..20 comma-separated patterns');
if (!patterns.length || patterns.length > 20 || !patterns.every(appRules.isValidPattern)) return problem(400, 'events.bad_topic', 'topic must be 1..20 comma-separated patterns');
const scopeErr = scopeError(who.principal, patterns);
const scopeErr = scopeError(who.principal, patterns);
if (scopeErr) return problem(403, 'events.topic_not_allowed', scopeErr);
if (scopeErr) return problem(403, 'events.topic_not_allowed', scopeErr);
- const after = Number(url.searchParams.get('after_seq') || 0);
+ const afterCursor = url.searchParams.get('after');
+ const after = afterCursor ? eventSeqFromCursor(afterCursor) : Number(url.searchParams.get('after_seq') || 0);
const limit = Number(url.searchParams.get('limit') || 100);
const limit = Number(url.searchParams.get('limit') || 100);
- if (!Number.isInteger(after) || after < 0 || !Number.isInteger(limit) || limit < 1 || limit > 1000) return problem(400, 'events.bad_request', 'after_seq must be >= 0 and limit 1..1000');
+ if (after === null || !Number.isInteger(after) || after < 0 || !Number.isInteger(limit) || limit < 1 || limit > 1000) {
+ return problem(400, 'events.bad_request', afterCursor ? 'after must be an opaque cursor' : 'after_seq must be >= 0 and limit 1..1000');
+ }
const pats = patterns.map((x) => [x, topicRegex(x)]);
const pats = patterns.map((x) => [x, topicRegex(x)]);
const out = [];
const out = [];
const page = {};
const page = {};
@@ -464,17 +476,17 @@ function createMockPlatform(opts = {}) {
@@ -464,17 +476,17 @@ function createMockPlatform(opts = {}) {
if (e.seq <= from) continue;
if (e.seq <= from) continue;
if (out.length >= limit) break;
if (out.length >= limit) break;
cursor = e.seq;
cursor = e.seq;
- if (pats.some(([x, re]) => re.test(e.event.event_type) && readable(who.principal, e, x))) out.push({ seq: e.seq, event: e.event });
+ if (pats.some(([x, re]) => re.test(e.event.event_type) && readable(who.principal, e, x))) out.push({ seq: e.seq, cursor: eventCursor(e.seq), event: e.event });
}
}
if (out.length < limit) cursor = Math.max(cursor, lastSeq);
if (out.length < limit) cursor = Math.max(cursor, lastSeq);
- return json(200, { ...page, events: out, next_after_seq: cursor, latest_seq: lastSeq });
+ return json(200, { ...page, events: out, next_after_seq: cursor, next_cursor: eventCursor(cursor), latest_seq: lastSeq });
}
}
if ((r = path.match(/^\/api\/v1\/events\/([^/]+)$/)) && req.method === 'GET') {
if ((r = path.match(/^\/api\/v1\/events\/([^/]+)$/)) && req.method === 'GET') {
const who = eventsPrincipal(req, 'events.event.read', 'events.app.read');
const who = eventsPrincipal(req, 'events.event.read', 'events.app.read');
if (who.res) return who.res;
if (who.res) return who.res;
const e = events.find((x) => x.event.event_id === decodeURIComponent(r[1]));
const e = events.find((x) => x.event.event_id === decodeURIComponent(r[1]));
const ok = e && (who.principal.kind === 'app' ? appRules.visibleToApp(e, who.principal) : (e.env || 'production') === 'production');
const ok = e && (who.principal.kind === 'app' ? appRules.visibleToApp(e, who.principal) : (e.env || 'production') === 'production');
- return ok ? json(200, { seq: e.seq, event: e.event }) : problem(404, 'events.not_found', 'no such event (or pruned by retention)');
+ return ok ? json(200, { seq: e.seq, cursor: eventCursor(e.seq), event: e.event }) : problem(404, 'events.not_found', 'no such event (or pruned by retention)');
}
}
if (path === '/api/v1/checkpoints') {
if (path === '/api/v1/checkpoints') {
const who = eventsPrincipal(req, 'events.event.read', 'events.app.read');
const who = eventsPrincipal(req, 'events.event.read', 'events.app.read');
@@ -130,6 +130,11 @@ run([
@@ -130,6 +130,11 @@ run([
assert.deepEqual(ids(await r.pull()), [sbx.event_id], 'another app of the project, same env');
assert.deepEqual(ids(await r.pull()), [sbx.event_id], 'another app of the project, same env');
assert.deepEqual(ids(await a.pull({ topic: 'order.*' })), [sbx.event_id], 'a sandbox app never sees production events');
assert.deepEqual(ids(await a.pull({ topic: 'order.*' })), [sbx.event_id], 'a sandbox app never sees production events');
+ const orderPage = await a.pull({ topic: 'order.*' });
+ assert.match(orderPage.events[0].cursor, /^c1\./, 'a page item carries its cursor');
+ assert.match(orderPage.next_cursor, /^c1\./, 'a page carries its opaque next_cursor');
+ assert.deepEqual(ids(await a.pull({ topic: 'order.*', after: orderPage.next_cursor })), [], 'after resumes from the cursor');
+ assert.deepEqual(ids(await a.pull({ topic: 'order.*', afterSeq: orderPage.next_after_seq })), [], 'afterSeq still works for one release');
assert.deepEqual(ids(await a2.pull()), [prod.event_id], 'a production app never sees sandbox events');
assert.deepEqual(ids(await a2.pull()), [prod.event_id], 'a production app never sees sandbox events');
assert.deepEqual(ids(await b.pull()), [other.event_id]);
assert.deepEqual(ids(await b.pull()), [other.event_id]);
assert.deepEqual(ids(await a.pull({ platformTopics: ['live.*'] })), [sbx.event_id, pub.event_id], 'public first-party only, never internal');
assert.deepEqual(ids(await a.pull({ platformTopics: ['live.*'] })), [sbx.event_id, pub.event_id], 'public first-party only, never internal');
@@ -141,8 +146,10 @@ run([
@@ -141,8 +146,10 @@ run([
assert.equal(await b.get(sbx.event_id), null, 'another project\'s event does not exist for this app');
assert.equal(await b.get(sbx.event_id), null, 'another project\'s event does not exist for this app');
assert.equal(await a2.get(sbx.event_id), null);
assert.equal(await a2.get(sbx.event_id), null);
const seen = [];
const seen = [];
- for await (const item of a.iterate({ topic: '*', platformTopics: ['live.stream.*'] })) seen.push(item.event.event_id);
+ const cursors = [];
+ for await (const item of a.iterate({ topic: '*', platformTopics: ['live.stream.*'], onPage: (p) => cursors.push(p.next_cursor) })) seen.push(item.event.event_id);
assert.deepEqual(seen, [sbx.event_id, pub.event_id]);
assert.deepEqual(seen, [sbx.event_id, pub.event_id]);
+ assert.match(cursors.at(-1), /^c1\./);
// First-party readers: never sandbox; app events only through an app.* pattern.
// First-party readers: never sandbox; app events only through an app.* pattern.
assert.ok(!(await live.pull({ topic: '*' })).events.some((e) => e.event.event_type.startsWith('app.')));
assert.ok(!(await live.pull({ topic: '*' })).events.some((e) => e.event.event_type.startsWith('app.')));
@@ -71,20 +71,49 @@ run([
@@ -71,20 +71,49 @@ run([
assert.deepEqual(page.events.map((e) => e.seq), [1, 3]);
assert.deepEqual(page.events.map((e) => e.seq), [1, 3]);
assert.equal(page.next_after_seq, 3);
assert.equal(page.next_after_seq, 3);
assert.equal(page.latest_seq, 5);
assert.equal(page.latest_seq, 5);
+ assert.match(page.next_cursor, /^c1\./, 'the page carries an opaque next_cursor');
+ assert.match(page.events[0].cursor, /^c1\./, 'each event carries its cursor');
+ // The opaque cursor round-trips: `after` resumes exactly where the page ended.
+ const rest = await events.pull({ topic: 'media.vod.*', after: page.next_cursor, limit: 2 });
+ assert.deepEqual(rest.events.map((e) => e.seq), [5]);
+ assert.equal((await events.get(platform.state.events[0].event.event_id)).cursor, page.events[0].cursor);
+ // afterSeq, the numeric position, still works for one release.
+ assert.deepEqual((await events.pull({ topic: 'media.vod.*', afterSeq: page.next_after_seq, limit: 2 })).events.map((e) => e.seq), [5]);
const seen = [];
const seen = [];
const cursors = [];
const cursors = [];
- for await (const { seq, event } of events.iterate({ topic: ['media.vod.*'], afterSeq: 0, limit: 2, onPage: (p) => cursors.push(p.next_after_seq) })) {
+ for await (const { seq, event } of events.iterate({ topic: ['media.vod.*'], afterSeq: 0, limit: 2, onPage: (p) => cursors.push(p.next_cursor) })) {
seen.push(seq);
seen.push(seq);
assert.equal(event.event_type, 'media.vod.ready');
assert.equal(event.event_type, 'media.vod.ready');
}
}
assert.deepEqual(seen, [1, 3, 5]);
assert.deepEqual(seen, [1, 3, 5]);
- assert.equal(cursors.at(-1), 5);
+ assert.match(cursors.at(-1), /^c1\./);
+ assert.equal(cursors.at(-1), (await events.pull({ topic: 'media.vod.*', afterSeq: 5 })).next_cursor, 'iterate ends on the page cursor');
+ const resumed = [];
+ for await (const { seq } of events.iterate({ topic: 'media.vod.*', after: page.next_cursor, limit: 2 })) resumed.push(seq);
+ assert.deepEqual(resumed, [5], 'iterate resumes from an opaque cursor');
assert.equal((await events.get(platform.state.events[0].event.event_id)).seq, 1);
assert.equal((await events.get(platform.state.events[0].event.event_id)).seq, 1);
assert.equal(await events.get('evt_01J00000000000000000000000'), null);
assert.equal(await events.get('evt_01J00000000000000000000000'), null);
await events.setCheckpoint('media.vod.*', 5);
await events.setCheckpoint('media.vod.*', 5);
assert.equal((await events.getCheckpoint('media.vod.*')).cursor, 5);
assert.equal((await events.getCheckpoint('media.vod.*')).cursor, 5);
}],
}],
+ ['iterate: onPage advances the cursor even on a page that matches no events', async () => {
+ const { platform, events } = setup();
+ platform.publishEvent({ event_type: 'live.other.thing', source: 'media', actor, subject: { type: 'vod', id: '1' } });
+ const pages = [];
+ for await (const e of events.iterate({ topic: 'media.vod.*', onPage: (p) => pages.push(p) })) void e;
+ assert.equal(pages.length, 1, 'the empty page still called onPage');
+ assert.deepEqual(pages[0].events, []);
+ assert.equal(pages[0].next_after_seq, 1);
+ assert.match(pages[0].next_cursor, /^c1\./, 'the cursor advanced past the scanned, non-matching event');
+ // Resuming from that cursor does not replay the scanned event: the new match is seen once.
+ const saved = pages[0].next_cursor;
+ platform.publishEvent({ event_type: 'media.vod.ready', source: 'media', actor, subject: { type: 'vod', id: '2' } });
+ const seen = [];
+ for await (const e of events.iterate({ topic: 'media.vod.*', after: saved })) seen.push(e.seq);
+ assert.deepEqual(seen, [2]);
+ }],
+
['iterate reports a retention gap through onGap', async () => {
['iterate reports a retention gap through onGap', async () => {
const srv = await stubServer((req, res) => {
const srv = await stubServer((req, res) => {
const after = Number(new URL(req.url, 'http://x').searchParams.get('after_seq'));
const after = Number(new URL(req.url, 'http://x').searchParams.get('after_seq'));
@@ -288,4 +317,25 @@ run([
@@ -288,4 +317,25 @@ run([
await worker.stop();
await worker.stop();
await srv.close();
await srv.close();
}],
}],
+ ['iterate keeps paging when the server sends only cursors, and a string afterSeq stays numeric', async () => {
+ const asked = [];
+ // A server past the compatibility release: no next_after_seq / latest_seq, the same cursor at the head.
+ const pages = { '': { events: [{ seq: 1, cursor: 'c1', event: { id: 'a' } }], next_cursor: 'c1' },
+ c1: { events: [{ seq: 2, cursor: 'c2', event: { id: 'b' } }], next_cursor: 'c2' },
+ c2: { events: [], next_cursor: 'c2' } };
+ const fetch = async (url) => {
+ const u = new URL(url);
+ asked.push(u.search);
+ const body = u.searchParams.has('after') ? pages[u.searchParams.get('after')] : { ...pages[''], next_after_seq: 1, latest_seq: 2 };
+ return new Response(JSON.stringify(body), { status: 200, headers: { 'content-type': 'application/json' } });
+ };
+ const events = createEventsClient(createClient({ fetch }), { source: 'live', baseUrl: 'http://events.test' });
+ const seen = [];
+ for await (const item of events.iterate({ afterSeq: '0' })) seen.push(item.event.id);
+ assert.deepEqual(seen, ['a', 'b'], 'every page up to the head');
+ assert.match(asked[0], /after_seq=0/, 'a string afterSeq is still the numeric position');
+ assert.match(asked[1], /after=c1/);
+ assert.match(asked[2], /after=c2/);
+ assert.equal(asked.length, 3, 'the unchanged cursor at the head ends it');
+ }],
]);
]);
@@ -3,9 +3,9 @@ import type { OpenVibeClient, EventEnvelope } from './core';
@@ -3,9 +3,9 @@ import type { OpenVibeClient, EventEnvelope } from './core';
/** What a producer passes: event_id, timestamp, source, version, payload and trace_id are filled in. */
/** What a producer passes: event_id, timestamp, source, version, payload and trace_id are filled in. */
export type EventInput = Omit<EventEnvelope, 'event_id' | 'timestamp' | 'source' | 'version' | 'payload'> & Partial<Pick<EventEnvelope, 'event_id' | 'timestamp' | 'source' | 'version' | 'payload'>>;
export type EventInput = Omit<EventEnvelope, 'event_id' | 'timestamp' | 'source' | 'version' | 'payload'> & Partial<Pick<EventEnvelope, 'event_id' | 'timestamp' | 'source' | 'version' | 'payload'>>;
export interface PublishResult { event_id: string; seq: number; duplicate: boolean; }
export interface PublishResult { event_id: string; seq: number; duplicate: boolean; }
-export interface StoredEvent { seq: number; event: EventEnvelope; }
+export interface StoredEvent { seq: number; cursor: string; event: EventEnvelope; }
export interface Gap { from_seq: number; to_seq: number; reason?: string; latest_seq?: number; }
export interface Gap { from_seq: number; to_seq: number; reason?: string; latest_seq?: number; }
-export interface EventsPage { events: StoredEvent[]; next_after_seq: number; latest_seq: number; gap?: Gap; }
+export interface EventsPage { events: StoredEvent[]; next_cursor: string; next_after_seq: number; latest_seq: number; gap?: Gap; }
export interface Subscription {
export interface Subscription {
id: string;
id: string;
consumer: string;
consumer: string;
@@ -23,9 +23,9 @@ export interface EventsClient {
@@ -23,9 +23,9 @@ export interface EventsClient {
prepare(envelope: EventInput, opts?: { traceId?: string; now?: number }): EventEnvelope;
prepare(envelope: EventInput, opts?: { traceId?: string; now?: number }): EventEnvelope;
publish(envelope: EventInput, opts?: { traceparent?: string }): Promise<PublishResult>;
publish(envelope: EventInput, opts?: { traceparent?: string }): Promise<PublishResult>;
publish(envelopes: EventInput[], opts?: { traceparent?: string }): Promise<{ results: PublishResult[] }>;
publish(envelopes: EventInput[], opts?: { traceparent?: string }): Promise<{ results: PublishResult[] }>;
- pull(opts?: { topic?: string | string[]; afterSeq?: number; limit?: number }): Promise<EventsPage>;
- /** onPage runs after every item of that page was yielded and handled: save page.next_after_seq there. */
- iterate(opts?: { topic?: string | string[]; afterSeq?: number; limit?: number; maxPages?: number; onGap?: (gap: Gap) => void | Promise<void>; onPage?: (page: EventsPage) => void | Promise<void> }): AsyncGenerator<StoredEvent, void, unknown>;
+ pull(opts?: { topic?: string | string[]; after?: string; afterSeq?: number; limit?: number }): Promise<EventsPage>;
+ /** onPage runs after every item of that page was yielded and handled: save page.next_cursor there. */
+ iterate(opts?: { topic?: string | string[]; after?: string; afterSeq?: number; limit?: number; maxPages?: number; onGap?: (gap: Gap) => void | Promise<void>; onPage?: (page: EventsPage) => void | Promise<void> }): AsyncGenerator<StoredEvent, void, unknown>;
get(eventId: string): Promise<StoredEvent | null>;
get(eventId: string): Promise<StoredEvent | null>;
getCheckpoint(topic: string): Promise<{ consumer: string; topic: string; cursor: number; updated_at: string | null }>;
getCheckpoint(topic: string): Promise<{ consumer: string; topic: string; cursor: number; updated_at: string | null }>;
setCheckpoint(topic: string, cursor: number): Promise<{ consumer: string; topic: string; cursor: number }>;
setCheckpoint(topic: string, cursor: number): Promise<{ consumer: string; topic: string; cursor: number }>;
@@ -65,8 +65,8 @@ export interface AppEventsClient {
@@ -65,8 +65,8 @@ export interface AppEventsClient {
publish(envelope: AppEventInput, opts?: { traceparent?: string }): Promise<PublishResult>;
publish(envelope: AppEventInput, opts?: { traceparent?: string }): Promise<PublishResult>;
publish(envelopes: AppEventInput[], opts?: { traceparent?: string }): Promise<{ results: PublishResult[] }>;
publish(envelopes: AppEventInput[], opts?: { traceparent?: string }): Promise<{ results: PublishResult[] }>;
/** topic is project-relative (default '*'); platformTopics are first-party patterns (public events only). */
/** topic is project-relative (default '*'); platformTopics are first-party patterns (public events only). */
- pull(opts?: { topic?: string | string[]; platformTopics?: string[]; afterSeq?: number; limit?: number }): Promise<EventsPage>;
- iterate(opts?: { topic?: string | string[]; platformTopics?: string[]; afterSeq?: number; limit?: number; maxPages?: number; onGap?: (gap: Gap) => void | Promise<void>; onPage?: (page: EventsPage) => void | Promise<void> }): AsyncGenerator<StoredEvent, void, unknown>;
+ pull(opts?: { topic?: string | string[]; platformTopics?: string[]; after?: string; afterSeq?: number; limit?: number }): Promise<EventsPage>;
+ iterate(opts?: { topic?: string | string[]; platformTopics?: string[]; after?: string; afterSeq?: number; limit?: number; maxPages?: number; onGap?: (gap: Gap) => void | Promise<void>; onPage?: (page: EventsPage) => void | Promise<void> }): AsyncGenerator<StoredEvent, void, unknown>;
get(eventId: string): Promise<StoredEvent | null>;
get(eventId: string): Promise<StoredEvent | null>;
getCheckpoint(topic: string): Promise<{ consumer: string; topic: string; cursor: number; updated_at: string | null }>;
getCheckpoint(topic: string): Promise<{ consumer: string; topic: string; cursor: number; updated_at: string | null }>;
setCheckpoint(topic: string, cursor: number): Promise<{ consumer: string; topic: string; cursor: number }>;
setCheckpoint(topic: string, cursor: number): Promise<{ consumer: string; topic: string; cursor: number }>;