openvibers/openvibe.sdk Commits
openvibe-sdk 0.42.0: an Events position is an opaque cursor only (#39)
Changes
@@@ -3,6 +3,28 @@All notable changes to `openvibe-sdk`. The package follows semver; while it is `0.x`, a minorrelease may change an API and says so here.++## 0.42.0 (2026-10-10)++++An Events position is an opaque cursor and nothing else (ADR-042 decision 7, plan T7). **Breaking** for a caller that++still passes a number:++++- **`pull()` / `iterate()`** take `after` (a page's `next_cursor`, or `latest_cursor` for the head) and nothing++ numeric. `afterSeq` throws a `TypeError` instead of silently reading from the start. Without `after`, a page starts++ at the oldest retained event, and no `after_seq` is ever sent. `iterate()` ends on the page whose `next_cursor` is++ `latest_cursor`, or on one that does not move the position. Pages carry `next_cursor` and `latest_cursor` only++ (types: `next_after_seq` and `latest_seq` are gone).++- **Checkpoints** store the opaque cursor. `setCheckpoint(topic, cursor, { carrier })` rejects anything but a cursor++ string, and `getCheckpoint()` answers `{ consumer, topic, cursor, carrier, updated_at }` with `cursor` as stored,++ or null when none is.++- **Realtime** resumes from the cursor exactly as Events sent it (`lastEventId` is a string; a bare number is no++ longer read as a position), and it dedupes by `event_id` (the last 512), never by position. The callback's++ `{ cursor }` is what to save; `seq` is informational and null when absent.++- **`parseDelivery()`** answers `seq: null` when a delivery carries none (it was `NaN`). Dedupe deliveries by++ `event.event_id`.++- **Mock Events** (`openvibe-sdk/testing`) behaves like Events after the change: `after_seq` is a 400, pages have no++ numeric fields, checkpoints hold cursors, the realtime `id:` is the cursor and a bare-number `Last-Event-ID` is++ ignored.++## 0.41.0 (2026-10-10)- **`createResourceIndex().list({ owner })`.** The index passes `owner` (a `usr_…`/`agt_…` subject id) to every@@@ -46,8 +46,8 @@ const community = createCommunityClient(client);const { pastes } = await community.pastes.list({ limit: 20 });// Live updates, resumable across reloads.--const sub = subscribe(['live.stream.*'], (event, { seq }) => render(event), {-- client, lastEventId: localStorage.getItem('seq'), onGap: () => refetchEverything(),++const sub = subscribe(['live.stream.*'], (event, { cursor }) => { render(event); localStorage.setItem('cursor', cursor); }, {++ client, lastEventId: localStorage.getItem('cursor'), onGap: () => refetchEverything(),});```@@@ -108,7 +108,7 @@ const { createAppEvents } = require('openvibe-sdk/events');const events = createAppEvents(client, { projectId: unverifiedClaims.project_id, appId: process.env.OV_CLIENT_ID });await events.publish({ event_type: 'order.shipped', subject: { type: 'order', id: 'o1' }, payload: { n: 1 } });// -> app.p01k5….order.shipped, source app-01k5…, actor { type: 'app', id: 'app_…' } (onBehalfOf: the person instead)--const page = await events.pull({ topic: 'order.*', platformTopics: ['live.stream.*'], afterSeq }); // + public first-party events++const page = await events.pull({ topic: 'order.*', platformTopics: ['live.stream.*'], after: saved }); // + public first-party events; save page.next_cursorawait events.subscribe({ topicPattern: '*', endpoint: 'https://hooks.example.com/openvibe' }); // public https only```@@@ -217,7 +217,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 `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.++- **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`, or `latest_cursor` to start at the head; a position is never a number, and `afterSeq` throws) 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.41.0",++ "version": "0.42.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",@@@ -12,8 +12,8 @@"core client: retries, timeouts, problem+json errors, trace propagation","auth: service and app tokens, PKCE, offline verification, the Network signing keys (createNetworkKeys), revocation cutoffs","registry discovery and contracts version checks",-- "events: publish, pull, app events, webhook verification (v2), outbox and inbox",-- "realtime SSE",++ "events: publish, pull and checkpoints by opaque cursor only, app events, webhook verification (v2), outbox and inbox",++ "realtime SSE (resumes from the opaque cursor, dedupes by event_id)","media files and objects v2","community, bot, search, chat, ai, jobs, tools, projects, modules, identity, openre, geo and vip clients","bot (openvibe-sdk/bot): OpenVibe.Bot robots, operators, devices, streaming, commands and audit, the public kits and profiles, and pairing",@@@ -1,4 +1,4 @@--// Generated by scripts/browser-bundle.js from browser.js (openvibe-sdk 0.41.0, MIT). Do not edit.++// Generated by scripts/browser-bundle.js from browser.js (openvibe-sdk 0.42.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.41.0';++const SDK_VERSION = '0.42.0';module.exports = {createClient,@@@ -955,7 +955,7 @@ function (module, exports) {/*** openvibe-sdk/realtime: the browser realtime plane inside OpenVibe.Events (ADR-005), over SSE.*-- * const sub = subscribe(['live.stream.*'], (event, { seq, cursor }) => { … save cursor … }, {++ * const sub = subscribe(['live.stream.*'], (event, { cursor }) => { … save cursor … }, {* client, // or url: 'https://openvibe.events'* lastEventId: savedCursor, // resume after a reload (the `cursor` the callback was given)* onGap: (gap) => refetchState(),@@@ -965,13 +965,14 @@ function (module, exports) {* Transport: EventSource when the environment has one and no Bearer token is needed (cookies ride* along with withCredentials); otherwise fetch with a streamed body (Node, or a token). Either* way it resumes from the last position it saw (Last-Event-ID: Events' opaque cursor, the `cursor` each-- * callback gets; an older Events sends a bare seq), skips anything it already delivered,++ * callback gets, never parsed), skips an event it already delivered (by event_id),* and reports an `event: gap` (events missed beyond retention or replay limits) through onGap.* Browser-safe.*/const { OpenVibeError } = __load(3);const DEFAULT_ORIGIN = 'https://openvibe.events';++const SEEN_MAX = 512; // event ids remembered for dedupeconst FATAL = new Set([400, 401, 403, 404]);const sleep = (ms, signal) => new Promise((resolve) => {const t = setTimeout(resolve, ms);@@@ -989,10 +990,11 @@ function subscribe(topics, onEvent, opts = {}) {const fetchImpl = opts.fetch || (typeof globalThis.fetch === 'function' ? globalThis.fetch.bind(globalThis) : null);const ES = opts.EventSource !== undefined ? opts.EventSource : globalThis.EventSource;-- // The position to resume from is the last SSE id as Events sent it: an opaque cursor (ADR-042) or, from an older-- // Events, a bare seq. Dedupe compares the numeric seq the message data carries.++ // The position to resume from is the last SSE id Events sent: an opaque cursor (ADR-042), handed back as it came.++ // A reconnect replays strictly after it; the ids of the last events delivered catch anything a transport hands++ // over twice (dedupe by event_id, never by position).const initial = lastEventId != null && lastEventId !== '' ? String(lastEventId) : null;-- const state = { lastId: initial, lastSeq: initial != null && /^\d+$/.test(initial) ? Number(initial) : null, closed: false, transport: null, connected: false };++ const state = { lastId: initial, seen: new Set(), closed: false, transport: null, connected: false };const stop = new AbortController();let es = null;let retryMs = reconnectDelayMs;@@@ -1010,12 +1012,14 @@ function subscribe(topics, onEvent, opts = {}) {if (type !== 'message' || !data) return;let msg;try { msg = JSON.parse(data); } catch { report(new OpenVibeError({ code: 'sdk.bad_response', message: 'realtime: undecodable message' })); return; }-- const seq = Number(msg && msg.seq != null ? msg.seq : id);-- if (Number.isFinite(seq)) {-- if (state.lastSeq != null && seq <= state.lastSeq) return; // already delivered-- state.lastSeq = seq;++ const eventId = msg && msg.event && typeof msg.event.event_id === 'string' ? msg.event.event_id : null;++ if (eventId) {++ if (state.seen.has(eventId)) return; // already delivered++ state.seen.add(eventId);++ if (state.seen.size > SEEN_MAX) state.seen.delete(state.seen.values().next().value);}if (id !== undefined && id !== '') state.lastId = String(id);++ const seq = msg && msg.seq != null && Number.isFinite(Number(msg.seq)) ? Number(msg.seq) : null; // informationaltry { onEvent(msg && msg.event, { seq, cursor: id !== undefined && id !== '' ? String(id) : null }); } catch (err) { report(err); }}@@@ -1096,7 +1100,7 @@ function subscribe(topics, onEvent, opts = {}) {es.onerror = () => {state.connected = false;// CONNECTING (0): the browser reconnects by itself and sends Last-Event-ID.-- // CLOSED (2): it gave up (HTTP error); reconnect ourselves from the last seq.++ // CLOSED (2): it gave up (HTTP error); reconnect ourselves from the last cursor.if (es.readyState === 2 && !state.closed) {report(new OpenVibeError({ code: 'sdk.network_error', retryable: true, message: 'realtime: stream closed' }));failures++;@@@ -1,12 +1,12 @@{"name": "openvibe-sdk",-- "version": "0.41.0",++ "version": "0.42.0","lockfileVersion": 3,"requires": true,"packages": {"": {"name": "openvibe-sdk",-- "version": "0.41.0",++ "version": "0.42.0","license": "MIT","devDependencies": {"@electric-sql/pglite": "^0.5.8",@@@ -1,6 +1,6 @@{"name": "openvibe-sdk",-- "version": "0.41.0",++ "version": "0.42.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.41.0';++const SDK_VERSION = '0.42.0';module.exports = {createClient,@@@ -112,7 +112,8 @@ function verifyDeliveryV2(rawBody, headers, secret, { toleranceSec = V2_TOLERANC/*** Verify and parse one delivery: { event, seq, subscriptionId, attempt } or null when it does not-- * verify. `headers` is req.headers (or a Fetch Headers).++ * verify. `headers` is req.headers (or a Fetch Headers). `seq` is informational (null when the delivery++ * carries none): dedupe by event.event_id, never by seq.** X-OpenVibe-Signature-V2 present it must verify and be within ±toleranceSec (default 300) of* `now`; a bad or stale v2 is null, never a fallback to v1@@@ -130,9 +131,10 @@ function parseDelivery(rawBody, headers, secret, { requireV2 = false, toleranceSlet body;try { body = JSON.parse(toBuffer(rawBody).toString('utf8')); } catch { return null; }if (!body || !body.event) return null;++ const seq = body.seq ?? get('x-openvibe-seq');return {event: body.event,-- seq: Number(body.seq ?? get('x-openvibe-seq')),++ seq: seq == null || seq === '' || !Number.isFinite(Number(seq)) ? null : Number(seq),subscriptionId: get('x-openvibe-subscription-id') || null,attempt: Number(get('x-openvibe-delivery-attempt')) || 1,};@@@ -168,8 +170,8 @@ function createEventsClient(client, { source, baseUrl } = {}) {}/**-- * publish(envelope | envelope[], { traceparent }) -> { event_id, seq, duplicate } | { results }.-- * Safe to retry: Events stores an event_id once and answers a repeat with the stored seq.++ * publish(envelope | envelope[], { traceparent }) -> { event_id, seq, cursor, duplicate } | { results }.++ * Safe to retry: Events stores an event_id once and answers a repeat with the stored position.*/async function publish(input, { traceparent } = {}) {const span = startSpan(traceparent || client.traceparent());@@@ -182,49 +184,45 @@ function createEventsClient(client, { source, baseUrl } = {}) {}/**-- * One page: { events: [{ seq, cursor, event }], next_cursor, next_after_seq, latest_seq, latest_cursor, gap? }.++ * One page: { events: [{ seq, cursor, event }], next_cursor, latest_cursor, gap? }.* Pass `after` — a page's opaque `next_cursor` — to resume exactly where the previous page ended, or-- * `latest_cursor` to start at the head without history; `afterSeq`, the numeric position, remains for-- * one release.++ * `latest_cursor` to start at the head without history; without it the page starts at the oldest retained++ * event. A position is only ever an opaque cursor (ADR-042): the numeric `afterSeq` is gone, and passing it++ * throws rather than silently reading from the start.*/-- function pull({ topic = '*', after, afterSeq = 0, limit } = {}) {++ function pull({ topic = '*', after, limit, afterSeq } = {}) {++ if (afterSeq !== undefined) throw new TypeError("pull: afterSeq is retired; pass after (a page's next_cursor, or latest_cursor for the head)");return call({ path: '/api/v1/events', query: {topic: Array.isArray(topic) ? topic.join(',') : topic, limit,-- ...(after == null ? { after_seq: afterSeq } : { after }),++ ...(after == null || after === '' ? {} : { after: String(after) }),} });}/**-- * Async iterator over { seq, cursor, event } from `after` (an opaque cursor) or `afterSeq` up to-- * the current head.++ * Async iterator over { seq, cursor, event } from `after` (an opaque cursor; the oldest retained event++ * without one) 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_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++ * page.next_cursor there as your durable cursor: 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, 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;++ async function* iterate({ topic, after, limit, onGap, onPage, maxPages = Infinity, afterSeq } = {}) {++ if (afterSeq !== undefined) throw new TypeError('iterate: afterSeq is retired; pass after (a saved next_cursor)');++ let cursor = after == null || after === '' ? null : String(after);for (let pages = 0; pages < maxPages; pages++) {-- const page = await pull({ topic, ...(useCursor ? { after: cursor } : { afterSeq: cursor }), limit });++ const page = await pull({ topic, ...(cursor == null ? {} : { after: cursor }), limit });if (page.gap && onGap) await onGap(page.gap);for (const item of page.events || []) yield item;if (onPage) await onPage(page);-- 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;++ const next = page.next_cursor;++ // The head: the page reached latest_cursor (both name a position in the same epoch, so equal strings are++ // the same position; nothing is parsed), or it did not move the position at all.++ if (next == null || next === cursor || next === page.latest_cursor) return;cursor = next;}}@@@ -246,10 +244,15 @@ function createEventsClient(client, { source, baseUrl } = {}) {publish,pull,iterate,-- /** { seq, event } or null. */++ /** { seq, cursor, event } or null. */get: (eventId) => orNull(call({ path: `/api/v1/events/${encodeURIComponent(eventId)}` })),++ /** { consumer, topic, cursor, carrier, updated_at }: the stored opaque cursor, or cursor null when none is stored. */getCheckpoint: (topic) => call({ path: '/api/v1/checkpoints', query: { topic } }),-- setCheckpoint: (topic, cursor) => call({ method: 'PUT', path: '/api/v1/checkpoints', json: { topic, cursor } }),++ /** Store an opaque cursor (a page's next_cursor) for a topic pattern; `carrier` records the carrier it is on. */++ setCheckpoint: (topic, cursor, { carrier } = {}) => {++ if (typeof cursor !== 'string' || !cursor) return Promise.reject(new TypeError("setCheckpoint: cursor is an opaque cursor string (a page's next_cursor)"));++ return call({ method: 'PUT', path: '/api/v1/checkpoints', json: { topic, cursor, ...(carrier ? { carrier } : {}) } });++ },subscriptions,subscribe: subscriptions.create,/** Operators (events.delivery.admin): { deliveries, counts }. */@@@ -359,7 +362,7 @@ function createAppEvents(client, { projectId, appId, onBehalfOf, baseUrl } = {})iterate: ({ topic: t = '*', platformTopics, ...rest } = {}) => events.iterate({ ...rest, topic: readTopics(t, platformTopics) }),get: (eventId) => events.get(eventId),getCheckpoint: (t) => events.getCheckpoint(topic(t)),-- setCheckpoint: (t, cursor) => events.setCheckpoint(topic(t), cursor),++ setCheckpoint: (t, cursor, opts) => events.setCheckpoint(topic(t), cursor, opts),subscriptions: {create: ({ topicPattern = '*', ...rest } = {}) => events.subscriptions.create({ ...rest, topicPattern: topic(topicPattern) }),list: () => events.subscriptions.list(),@@@ -2,7 +2,7 @@/*** openvibe-sdk/realtime: the browser realtime plane inside OpenVibe.Events (ADR-005), over SSE.*-- * const sub = subscribe(['live.stream.*'], (event, { seq, cursor }) => { … save cursor … }, {++ * const sub = subscribe(['live.stream.*'], (event, { cursor }) => { … save cursor … }, {* client, // or url: 'https://openvibe.events'* lastEventId: savedCursor, // resume after a reload (the `cursor` the callback was given)* onGap: (gap) => refetchState(),@@@ -12,13 +12,14 @@* Transport: EventSource when the environment has one and no Bearer token is needed (cookies ride* along with withCredentials); otherwise fetch with a streamed body (Node, or a token). Either* way it resumes from the last position it saw (Last-Event-ID: Events' opaque cursor, the `cursor` each-- * callback gets; an older Events sends a bare seq), skips anything it already delivered,++ * callback gets, never parsed), skips an event it already delivered (by event_id),* and reports an `event: gap` (events missed beyond retention or replay limits) through onGap.* Browser-safe.*/const { OpenVibeError } = require('./core/errors');const DEFAULT_ORIGIN = 'https://openvibe.events';++const SEEN_MAX = 512; // event ids remembered for dedupeconst FATAL = new Set([400, 401, 403, 404]);const sleep = (ms, signal) => new Promise((resolve) => {const t = setTimeout(resolve, ms);@@@ -36,10 +37,11 @@ function subscribe(topics, onEvent, opts = {}) {const fetchImpl = opts.fetch || (typeof globalThis.fetch === 'function' ? globalThis.fetch.bind(globalThis) : null);const ES = opts.EventSource !== undefined ? opts.EventSource : globalThis.EventSource;-- // The position to resume from is the last SSE id as Events sent it: an opaque cursor (ADR-042) or, from an older-- // Events, a bare seq. Dedupe compares the numeric seq the message data carries.++ // The position to resume from is the last SSE id Events sent: an opaque cursor (ADR-042), handed back as it came.++ // A reconnect replays strictly after it; the ids of the last events delivered catch anything a transport hands++ // over twice (dedupe by event_id, never by position).const initial = lastEventId != null && lastEventId !== '' ? String(lastEventId) : null;-- const state = { lastId: initial, lastSeq: initial != null && /^\d+$/.test(initial) ? Number(initial) : null, closed: false, transport: null, connected: false };++ const state = { lastId: initial, seen: new Set(), closed: false, transport: null, connected: false };const stop = new AbortController();let es = null;let retryMs = reconnectDelayMs;@@@ -57,12 +59,14 @@ function subscribe(topics, onEvent, opts = {}) {if (type !== 'message' || !data) return;let msg;try { msg = JSON.parse(data); } catch { report(new OpenVibeError({ code: 'sdk.bad_response', message: 'realtime: undecodable message' })); return; }-- const seq = Number(msg && msg.seq != null ? msg.seq : id);-- if (Number.isFinite(seq)) {-- if (state.lastSeq != null && seq <= state.lastSeq) return; // already delivered-- state.lastSeq = seq;++ const eventId = msg && msg.event && typeof msg.event.event_id === 'string' ? msg.event.event_id : null;++ if (eventId) {++ if (state.seen.has(eventId)) return; // already delivered++ state.seen.add(eventId);++ if (state.seen.size > SEEN_MAX) state.seen.delete(state.seen.values().next().value);}if (id !== undefined && id !== '') state.lastId = String(id);++ const seq = msg && msg.seq != null && Number.isFinite(Number(msg.seq)) ? Number(msg.seq) : null; // informationaltry { onEvent(msg && msg.event, { seq, cursor: id !== undefined && id !== '' ? String(id) : null }); } catch (err) { report(err); }}@@@ -143,7 +147,7 @@ function subscribe(topics, onEvent, opts = {}) {es.onerror = () => {state.connected = false;// CONNECTING (0): the browser reconnects by itself and sends Last-Event-ID.-- // CLOSED (2): it gave up (HTTP error); reconnect ourselves from the last seq.++ // CLOSED (2): it gave up (HTTP error); reconnect ourselves from the last cursor.if (es.readyState === 2 && !state.closed) {report(new OpenVibeError({ code: 'sdk.network_error', retryable: true, message: 'realtime: stream closed' }));failures++;@@@ -458,12 +458,13 @@ 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);++ // Like Events: a position is an opaque cursor; the numeric after_seq is refused, never read as "from the start".++ if (url.searchParams.has('after_seq')) return problem(400, 'events.bad_request', "after_seq is retired: pass after=<cursor> (a page's next_cursor, or latest_cursor for the head)");const afterCursor = url.searchParams.get('after');-- const after = afterCursor ? eventSeqFromCursor(afterCursor) : Number(url.searchParams.get('after_seq') || 0);++ const after = afterCursor ? eventSeqFromCursor(afterCursor) : 0;const limit = Number(url.searchParams.get('limit') || 100);-- 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');-- }++ if (after === null) return problem(400, 'events.bad_request', 'after must be an opaque cursor');++ if (!Number.isInteger(limit) || limit < 1 || limit > 1000) return problem(400, 'events.bad_request', 'limit must be 1..1000');const pats = patterns.map((x) => [x, topicRegex(x)]);const out = [];const page = {};@@@ -480,7 +481,7 @@ function createMockPlatform(opts = {}) {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, next_cursor: eventCursor(cursor), latest_seq: lastSeq, latest_cursor: eventCursor(lastSeq) });++ return json(200, { ...page, events: out, next_cursor: eventCursor(cursor), latest_cursor: eventCursor(lastSeq) });}if ((r = path.match(/^\/api\/v1\/events\/([^/]+)$/)) && req.method === 'GET') {const who = eventsPrincipal(req, 'events.event.read', 'events.app.read');@@@ -495,15 +496,15 @@ function createMockPlatform(opts = {}) {const b = req.method === 'GET' ? {} : await req.json().catch(() => ({}));const topic = req.method === 'GET' ? url.searchParams.get('topic') || '' : b.topic;if (!appRules.isValidPattern(topic)) return problem(400, 'events.bad_request', 'topic (pattern) is required');-- if (req.method !== 'GET' && (!Number.isInteger(b.cursor) || b.cursor < 0)) return problem(400, 'events.bad_request', 'topic (pattern) and cursor (integer >= 0) are required');++ if (req.method !== 'GET' && (typeof b.cursor !== 'string' || eventSeqFromCursor(b.cursor) === null)) return problem(400, 'events.bad_request', 'topic (pattern) and cursor (an opaque cursor) are required');const scopeErr = scopeError(who.principal, [topic]);if (scopeErr) return problem(403, 'events.topic_not_allowed', scopeErr);const k = `${who.consumer}|${topic}`;if (req.method === 'GET') {const cp = checkpoints.get(k);-- return json(200, { consumer: who.consumer, topic, cursor: cp ? cp.cursor : 0, updated_at: cp ? cp.updated_at : null });++ return json(200, { consumer: who.consumer, topic, cursor: cp ? cp.cursor : null, carrier: cp ? cp.carrier : null, updated_at: cp ? cp.updated_at : null });}-- const cp = { cursor: b.cursor, updated_at: new Date().toISOString() };++ const cp = { cursor: b.cursor, carrier: typeof b.carrier === 'string' && b.carrier ? b.carrier : null, updated_at: new Date().toISOString() };checkpoints.set(k, cp);return json(200, { consumer: who.consumer, topic, ...cp });}@@@ -572,8 +573,9 @@ function createMockPlatform(opts = {}) {if (/^app:/.test(String(claims.sub)) && claims.env === 'sandbox') return problem(401, 'token.sandbox_refused', 'sandbox tokens are not accepted here');viewer = { kind: 'service' };} else if (claims) viewer = { kind: 'user', subject: claims.subject_id };++ // Like Events: Last-Event-ID is an opaque cursor; anything else (a bare number included) is ignored.const raw = req.headers.get('last-event-id') ?? url.searchParams.get('last_event_id');-- const last = raw != null && /^\d+$/.test(raw) ? Number(raw) : null;++ const last = raw != null ? eventSeqFromCursor(String(raw).trim()) : null;const enc = new TextEncoder();let conn;const body = new ReadableStream({@@@ -585,7 +587,7 @@ function createMockPlatform(opts = {}) {push(e) {if (e.seq <= conn.lastSeq || !pats.some((re) => re.test(e.event.event_type)) || !visible(viewer, e)) return;conn.lastSeq = e.seq;-- send(`id: ${e.seq}\ndata: ${JSON.stringify({ seq: e.seq, event: e.event })}\n\n`);++ send(`id: ${eventCursor(e.seq)}\ndata: ${JSON.stringify({ seq: e.seq, event: e.event })}\n\n`);},close() { streams.delete(conn); try { controller.close(); } catch { /* closed */ } },};@@@ -134,7 +134,6 @@ run([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');@@@ -158,11 +157,11 @@ run([assert.equal((await live.get(prod.event_id)).event.event_id, prod.event_id);// Checkpoints are per app and scoped the same way.-- const cp = await a.setCheckpoint('*', 5);++ const cp = await a.setCheckpoint('*', orderPage.next_cursor);assert.equal(cp.consumer, `app:${A}`);assert.equal(cp.topic, `app.${K1}.*`);-- assert.equal((await a.getCheckpoint('*')).cursor, 5);-- assert.equal((await r.getCheckpoint('*')).cursor, 0, 'another app has its own');++ assert.equal((await a.getCheckpoint('*')).cursor, orderPage.next_cursor);++ assert.equal((await r.getCheckpoint('*')).cursor, null, 'another app has its own');await assert.rejects(b.events.getCheckpoint(`app.${K1}.*`), { status: 403, code: 'events.topic_not_allowed' });}],@@@ -65,12 +65,11 @@ run([}],['pull + iterate walk the cursor across pages; onPage gives the durable cursor', async () => {-- const { platform, events } = setup();++ const { platform, client, events } = setup();for (let i = 1; i <= 5; i++) platform.publishEvent({ event_type: i % 2 ? 'media.vod.ready' : 'live.stream.started', source: 'media', actor, subject: { type: 'vod', id: String(i) } });-- const page = await events.pull({ topic: 'media.vod.*', afterSeq: 0, limit: 2 });-- assert.deepEqual(page.events.map((e) => e.seq), [1, 3]);-- assert.equal(page.next_after_seq, 3);-- assert.equal(page.latest_seq, 5);++ const page = await events.pull({ topic: 'media.vod.*', limit: 2 });++ assert.deepEqual(page.events.map((e) => e.seq), [1, 3], 'without after, from the oldest retained event');++ assert.ok(!('next_after_seq' in page) && !('latest_seq' in page), 'no numeric position on the page');assert.match(page.next_cursor, /^c1\./, 'the page carries an opaque next_cursor');assert.match(page.latest_cursor, /^c1\./, 'and the head as an opaque latest_cursor');assert.deepStrictEqual((await events.pull({ topic: 'media.vod.*', after: page.latest_cursor })).events, [], 'nothing after the head');@@@ -79,24 +78,32 @@ run([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]);++ // The numeric position is gone: passing it throws instead of reading from the start, and Events refuses it.++ assert.throws(() => events.pull({ topic: 'media.vod.*', afterSeq: 3 }), /afterSeq is retired/);++ await assert.rejects(async () => { for await (const e of events.iterate({ topic: 'media.vod.*', afterSeq: 0 })) void e; }, /afterSeq is retired/);++ await assert.rejects(client.json({ service: 'events', audience: 'openvibe.events', path: '/api/v1/events', query: { topic: 'media.vod.*', after_seq: 3 } }),++ { status: 400, code: 'events.bad_request' }, 'after_seq is refused, never read as "from the start"');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_cursor) })) {++ for await (const { seq, event } of events.iterate({ topic: ['media.vod.*'], 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.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');++ assert.equal(cursors.at(-1), (await events.pull({ topic: 'media.vod.*', limit: 1 })).latest_cursor, 'iterate ends on the head');++ assert.equal(cursors.length, 2, 'and stops there: the last page reached latest_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);++ assert.equal((await events.getCheckpoint('media.vod.*')).cursor, null, 'none stored yet');++ await events.setCheckpoint('media.vod.*', page.next_cursor, { carrier: 'pg' });++ const cp = await events.getCheckpoint('media.vod.*');++ assert.equal(cp.cursor, page.next_cursor, 'the opaque cursor comes back as it was stored');++ assert.equal(cp.carrier, 'pg');++ await assert.rejects(events.setCheckpoint('media.vod.*', 5), /opaque cursor string/, 'a number is not a position');}],['iterate: onPage advances the cursor even on a page that matches no events', async () => {@@@ -106,8 +113,7 @@ run([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');++ assert.equal(pages[0].next_cursor, pages[0].latest_cursor, 'the cursor advanced past the scanned, non-matching event to the head');// 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' } });@@@ -117,10 +123,12 @@ run([}],['iterate reports a retention gap through onGap', async () => {++ const asked = [];const srv = await stubServer((req, res) => {-- const after = Number(new URL(req.url, 'http://x').searchParams.get('after_seq'));-- if (after === 0) return send(res, 200, { gap: { from_seq: 1, to_seq: 9 }, events: [{ seq: 10, event: { event_type: 'a.b.c' } }], next_after_seq: 10, latest_seq: 11 });-- return send(res, 200, { events: [{ seq: 11, event: { event_type: 'a.b.c' } }], next_after_seq: 11, latest_seq: 11 });++ const q = new URL(req.url, 'http://x').searchParams;++ asked.push([q.get('after'), q.get('after_seq')]);++ if (!q.get('after')) return send(res, 200, { gap: { from_seq: 1, to_seq: 9 }, events: [{ seq: 10, cursor: 'c1.0.MTA', event: { event_type: 'a.b.c' } }], next_cursor: 'c1.0.MTA', latest_cursor: 'c1.0.MTE' });++ return send(res, 200, { events: [{ seq: 11, cursor: 'c1.0.MTE', event: { event_type: 'a.b.c' } }], next_cursor: 'c1.0.MTE', latest_cursor: 'c1.0.MTE' });});const events = createEventsClient(createClient({ baseUrls: { events: srv.url }, token: 't' }), { source: 'live' });const gaps = [];@@@ -128,6 +136,7 @@ run([for await (const e of events.iterate({ topic: 'a.*', onGap: (g) => gaps.push(g) })) seqs.push(e.seq);assert.deepEqual(gaps, [{ from_seq: 1, to_seq: 9 }]);assert.deepEqual(seqs, [10, 11]);++ assert.deepEqual(asked, [[null, null], ['c1.0.MTA', null]], 'no after_seq is ever sent; the second page reads on from the cursor and the head ends it');await srv.close();}],@@@ -226,20 +235,22 @@ run([const { platform, events } = setup();for (let i = 1; i <= 6; i++) platform.publishEvent({ event_type: i === 3 ? 'live.other.thing' : 'media.vod.ready', source: 'media', actor, subject: { type: 'vod', id: String(i) } });const log = [];-- let saved = 0;++ let saved = null;++ // The mock's cursor format, decoded only to label the log readably; a consumer never parses a cursor.++ const at = (c) => Number(Buffer.from(String(c).split('.')[2], 'base64url').toString('utf8'));const consume = async (failAt) => {-- for await (const { seq } of events.iterate({ topic: 'media.vod.*', afterSeq: saved, limit: 3, onPage: (p) => { log.push(`page ${p.next_after_seq}`); saved = p.next_after_seq; } })) {++ for await (const { seq } of events.iterate({ topic: 'media.vod.*', after: saved, limit: 3, onPage: (p) => { log.push(`page ${at(p.next_cursor)}`); saved = p.next_cursor; } })) {if (seq === failAt) throw new Error(`crash at ${seq}`);log.push(`item ${seq}`);}};await assert.rejects(consume(2), /crash at 2/);assert.deepEqual(log, ['item 1']);-- assert.equal(saved, 0, 'a crash inside the first page saves nothing');++ assert.equal(saved, null, 'a crash inside the first page saves nothing');log.length = 0;await assert.rejects(consume(5), /crash at 5/);assert.deepEqual(log, ['item 1', 'item 2', 'item 4', 'page 4'], 'page 1 (seq 1-4; 3 did not match) was saved after its last item; page 2 never was');-- assert.equal(saved, 4);++ assert.equal(at(saved), 4);log.length = 0;await consume(null);assert.deepEqual(log, ['item 5', 'item 6', 'page 6'], 'resumed from the saved cursor: nothing skipped');@@@ -251,11 +262,13 @@ run([['mock Events retention: pull and realtime report the pruned range as a gap', async () => {const { platform, events } = setup();for (let i = 1; i <= 5; i++) platform.publishEvent({ event_type: 'media.vod.ready', source: 'media', actor, subject: { type: 'vod', id: String(i) }, visibility: 'public' });++ // Cursors handed out before the prune (the mock's cursor of position n; a consumer never builds one).++ const cur = (n) => `c1.0.${Buffer.from(String(n), 'utf8').toString('base64url')}`;assert.equal(platform.pruneEvents(3), 3);-- const page = await events.pull({ topic: 'media.vod.*', afterSeq: 1 });++ const page = await events.pull({ topic: 'media.vod.*', after: cur(1) });assert.deepEqual(page.gap, { from_seq: 2, to_seq: 3 });assert.deepEqual(page.events.map((e) => e.seq), [4, 5]);-- assert.equal((await events.pull({ topic: 'media.vod.*', afterSeq: 3 })).gap, undefined);++ assert.equal((await events.pull({ topic: 'media.vod.*', after: cur(3) })).gap, undefined);const gaps = [];const seqs = [];for await (const e of events.iterate({ topic: 'media.vod.*', onGap: (g) => gaps.push(g) })) seqs.push(e.seq);@@@ -263,12 +276,12 @@ run([assert.deepEqual(seqs, [4, 5]);const rtGaps = [];const got = [];-- const sub = subscribe('media.vod.*', (ev, { seq }) => got.push(seq), { client: createClient({ fetch: platform.fetch }), fetch: platform.fetch, transport: 'fetch', lastEventId: 1, onGap: (g) => rtGaps.push(g) });++ const sub = subscribe('media.vod.*', (ev, { seq }) => got.push(seq), { client: createClient({ fetch: platform.fetch }), fetch: platform.fetch, transport: 'fetch', lastEventId: cur(1), onGap: (g) => rtGaps.push(g) });await waitFor(() => got.length === 2);sub.close();assert.deepEqual(rtGaps, [{ reason: 'retention', from_seq: 2, to_seq: 3, latest_seq: 5 }]);platform.publishEvent({ event_type: 'media.vod.ready', source: 'media', actor, subject: { type: 'vod', id: '6' } });-- assert.equal((await events.pull({ afterSeq: 5 })).events[0].seq, 6, 'seqs keep counting after a prune');++ assert.equal((await events.pull({ after: cur(5) })).events[0].seq, 6, 'positions keep counting after a prune');}],['deliverEvents: signed deliveries to a local endpoint, retried in order, dead after max attempts', async () => {@@@ -319,25 +332,33 @@ run([await worker.stop();await srv.close();}],-- ['iterate keeps paging when the server sends only cursors, and a string afterSeq stays numeric', async () => {++ ['iterate pages by cursor alone: the unchanged cursor or latest_cursor ends it', async () => {const asked = [];-- // A server past the compatibility release: no next_after_seq / latest_seq, the same cursor at the head.++ // A server that omits latest_cursor: the page that does not move the position ends the walk.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 };++ const body = u.searchParams.has('after') ? pages[u.searchParams.get('after')] : pages[''];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);++ for await (const item of events.iterate({})) 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.doesNotMatch(asked[0], /after/, 'no position on the first page: the oldest retained event');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');++ // With latest_cursor, the page that reaches the head ends it without one more request.++ asked.length = 0;++ pages[''] = { ...pages[''], latest_cursor: 'c2' };++ pages.c1 = { ...pages.c1, latest_cursor: 'c2' };++ seen.length = 0;++ for await (const item of events.iterate({})) seen.push(item.event.id);++ assert.deepEqual(seen, ['a', 'b']);++ assert.equal(asked.length, 2, 'the page whose next_cursor is latest_cursor is the last');}],]);@@@ -93,9 +93,9 @@ run([res.end(JSON.stringify({ status: 401, code: 'token.bad_signature', detail: 'bad', type: 't', title: 'x' }));});const errors = [];-- const sub = subscribe('live.*', () => {}, { url: `${srv.url}/realtime/stream`, lastEventId: 41, onError: (e) => errors.push(e), reconnectDelayMs: 10 });++ const sub = subscribe('live.*', () => {}, { url: `${srv.url}/realtime/stream`, lastEventId: 'c1.0.NDE', onError: (e) => errors.push(e), reconnectDelayMs: 10 });await sub.done;-- assert.equal(srv.requests[0].headers['last-event-id'], '41');++ assert.equal(srv.requests[0].headers['last-event-id'], 'c1.0.NDE', 'the saved cursor is handed back as it is');assert.ok(srv.requests[0].url.startsWith('/realtime/stream?topics=live.*'));assert.equal(errors[0].code, 'token.bad_signature');assert.equal(sub.closed, true);@@@ -112,22 +112,22 @@ run([}const got = [];const gaps = [];-- const sub = subscribe(['live.stream.*'], (event, { seq }) => got.push(seq), { url: 'https://events.example', EventSource: FakeES, onGap: (g) => gaps.push(g), reconnectDelayMs: 5 });++ const sub = subscribe(['live.stream.*'], (event, { seq, cursor }) => got.push([seq, cursor]), { url: 'https://events.example', EventSource: FakeES, onGap: (g) => gaps.push(g), reconnectDelayMs: 5 });await waitFor(() => made.length === 1);assert.equal(sub.transport, 'eventsource');const es = made[0];assert.deepEqual(es.init, { withCredentials: true });assert.equal(new URL(es.url).searchParams.get('last_event_id'), null);es.onopen();-- es.onmessage({ data: JSON.stringify({ seq: 7, event: { event_type: 'live.stream.started' } }), lastEventId: '7' });-- es.onmessage({ data: JSON.stringify({ seq: 7, event: {} }), lastEventId: '7' });++ es.onmessage({ data: JSON.stringify({ seq: 7, event: { event_id: 'evt_7', event_type: 'live.stream.started' } }), lastEventId: 'c1.0.Nw' });++ es.onmessage({ data: JSON.stringify({ seq: 7, event: { event_id: 'evt_7', event_type: 'live.stream.started' } }), lastEventId: 'c1.0.Nw' });es.listeners.gap({ data: JSON.stringify({ reason: 'replay_limit', from_seq: 8, to_seq: 9 }) });-- assert.deepEqual(got, [7]);++ assert.deepEqual(got, [[7, 'c1.0.Nw']], 'a repeated event_id is delivered once');assert.equal(gaps[0].reason, 'replay_limit');es.readyState = 2;es.onerror();await waitFor(() => made.length === 2);-- assert.equal(new URL(made[1].url).searchParams.get('last_event_id'), '7');++ assert.equal(new URL(made[1].url).searchParams.get('last_event_id'), 'c1.0.Nw', 'it resumes from the opaque cursor');sub.close();assert.equal(made[1].closedByClient, true);}],@@@ -141,7 +141,8 @@ run([pub('live.stream.updated');const got = [];const realtime = createRealtimeClient(client, { fetch: platform.fetch, reconnectDelayMs: 10 });-- const sub = realtime.subscribe('live.stream.*', (event, { seq }) => got.push([seq, event.event_type]), { lastEventId: 0 });++ // Resume from the cursor of position 0 (the mock's single epoch): everything retained is replayed.++ const sub = realtime.subscribe('live.stream.*', (event, { seq }) => got.push([seq, event.event_type]), { lastEventId: 'c1.0.MA' });await waitFor(() => got.length === 2);assert.deepEqual(got, [[1, 'live.stream.started'], [3, 'live.stream.updated']], 'internal events never reach an anonymous browser');pub('live.stream.ended');@@@ -151,7 +152,7 @@ run([await waitFor(() => got.length === 4);assert.deepEqual(got.map((g) => g[0]), [1, 3, 4, 5]);const reconnect = platform.stats.requests.filter((r) => r.url.includes('/realtime/stream')).at(-1);-- assert.equal(reconnect.headers['last-event-id'], '4');++ assert.equal(reconnect.headers['last-event-id'], 'c1.0.NA', 'the reconnect hands back the opaque cursor of the last event');sub.close();await sub.done;}],@@@ -2,10 +2,14 @@ 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; }++/** `seq` is informational; the position is `cursor`. */++export interface PublishResult { event_id: string; seq: number; cursor: string; duplicate: boolean; }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_cursor: string; next_after_seq: number; latest_seq: number; gap?: Gap; }++/** A position is only ever an opaque cursor: `next_cursor` to read on, `latest_cursor` for the head. */++export interface EventsPage { events: StoredEvent[]; next_cursor: string; latest_cursor: string; gap?: Gap; }++/** A stored checkpoint: `cursor` is the opaque cursor as it was stored, null when none is. */++export interface Checkpoint { consumer: string; topic: string; cursor: string | null; carrier: string | null; updated_at: string | null; }export interface Subscription {id: string;consumer: string;@@@ -23,12 +27,14 @@ 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[]; after?: string; afterSeq?: number; limit?: number }): Promise<EventsPage>;++ /** Without `after`, from the oldest retained event. */++ pull(opts?: { topic?: string | string[]; after?: string; 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>;++ iterate(opts?: { topic?: string | string[]; after?: string; 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 }>;++ getCheckpoint(topic: string): Promise<Checkpoint>;++ /** `cursor` is an opaque cursor string (a page's next_cursor); anything else rejects. */++ setCheckpoint(topic: string, cursor: string, opts?: { carrier?: string }): Promise<Checkpoint>;subscriptions: {create(input: CreateSubscription): Promise<Subscription>;list(): Promise<Subscription[]>;@@@ -65,11 +71,12 @@ 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[]; 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>;++ pull(opts?: { topic?: string | string[]; platformTopics?: string[]; after?: string; limit?: number }): Promise<EventsPage>;++ iterate(opts?: { topic?: string | string[]; platformTopics?: string[]; after?: string; 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 }>;++ getCheckpoint(topic: string): Promise<Checkpoint>;++ /** `cursor` is an opaque cursor string (a page's next_cursor); anything else rejects. */++ setCheckpoint(topic: string, cursor: string, opts?: { carrier?: string }): Promise<Checkpoint>;subscriptions: {/** topicPattern is project-relative (default '*'); the endpoint must be public https. */create(input: Partial<CreateSubscription> & { endpoint: string }): Promise<Subscription>;@@@ -111,7 +118,7 @@ export interface ParseDeliveryOptions extends DeliveryV2Options {/** Refuse deliveries without X-OpenVibe-Signature-V2 (v1-only). Default false. A present v2 header must always verify. */requireV2?: boolean;}--export declare function parseDelivery(rawBody: RawBody, headers: Record<string, any> | Headers, secret: string, opts?: ParseDeliveryOptions): { event: EventEnvelope; seq: number; subscriptionId: string | null; attempt: number } | null;++export declare function parseDelivery(rawBody: RawBody, headers: Record<string, any> | Headers, secret: string, opts?: ParseDeliveryOptions): { event: EventEnvelope; seq: number | null; subscriptionId: string | null; attempt: number } | null;/** The subset of a better-sqlite3 Database the outbox and inbox use. */export interface SqliteDatabase {@@@ -1,13 +1,15 @@import type { OpenVibeClient, EventEnvelope, FetchLike, TokenContext } from './core';export interface RealtimeGap { reason: 'retention' | 'replay_limit' | 'cursor_ahead' | string; from_seq: number; to_seq: number; latest_seq?: number; }++/** `cursor` is the position to save; `seq` is informational. */++export interface RealtimeMeta { seq: number | null; cursor: string | null; }export interface SubscribeOptions {client?: OpenVibeClient;/** Events origin or the full …/realtime/stream URL; default from the client's registry, else https://openvibe.events */url?: string;baseUrl?: string;-- /** Resume after this seq (e.g. saved before a reload). */-- lastEventId?: number | string | null;++ /** Resume after this opaque cursor (the `cursor` a callback was given, saved before a reload). */++ lastEventId?: string | null;onGap?: (gap: RealtimeGap) => void;onOpen?: () => void;onError?: (err: Error) => void;@@@ -30,8 +32,8 @@ export interface RealtimeSubscription {readonly closed: boolean;readonly done: Promise<void>;}--export declare function subscribe(topics: string | string[], onEvent: (event: EventEnvelope, meta: { seq: number }) => void, opts?: SubscribeOptions): RealtimeSubscription;--export declare function createRealtimeClient(client: OpenVibeClient, defaults?: Omit<SubscribeOptions, 'client'>): { subscribe(topics: string | string[], onEvent: (event: EventEnvelope, meta: { seq: number }) => void, opts?: SubscribeOptions): RealtimeSubscription };++export declare function subscribe(topics: string | string[], onEvent: (event: EventEnvelope, meta: RealtimeMeta) => void, opts?: SubscribeOptions): RealtimeSubscription;++export declare function createRealtimeClient(client: OpenVibeClient, defaults?: Omit<SubscribeOptions, 'client'>): { subscribe(topics: string | string[], onEvent: (event: EventEnvelope, meta: RealtimeMeta) => void, opts?: SubscribeOptions): RealtimeSubscription };export interface SSEMessage { event: string; data: string; id: string | undefined; }/** WHATWG event-stream parser over a fetch body (or any async iterable of chunks). */export declare function parseSSE(body: ReadableStream<Uint8Array> | AsyncIterable<Uint8Array | string>, opts?: { onRetry?: (ms: number) => void }): AsyncGenerator<SSEMessage, void, unknown>;
@@ -3,6 +3,28 @@
@@ -3,6 +3,28 @@
All notable changes to `openvibe-sdk`. The package follows semver; while it is `0.x`, a minor
All notable changes to `openvibe-sdk`. The package follows semver; while it is `0.x`, a minor
release may change an API and says so here.
release may change an API and says so here.
+## 0.42.0 (2026-10-10)
+
+An Events position is an opaque cursor and nothing else (ADR-042 decision 7, plan T7). **Breaking** for a caller that
+still passes a number:
+
+- **`pull()` / `iterate()`** take `after` (a page's `next_cursor`, or `latest_cursor` for the head) and nothing
+ numeric. `afterSeq` throws a `TypeError` instead of silently reading from the start. Without `after`, a page starts
+ at the oldest retained event, and no `after_seq` is ever sent. `iterate()` ends on the page whose `next_cursor` is
+ `latest_cursor`, or on one that does not move the position. Pages carry `next_cursor` and `latest_cursor` only
+ (types: `next_after_seq` and `latest_seq` are gone).
+- **Checkpoints** store the opaque cursor. `setCheckpoint(topic, cursor, { carrier })` rejects anything but a cursor
+ string, and `getCheckpoint()` answers `{ consumer, topic, cursor, carrier, updated_at }` with `cursor` as stored,
+ or null when none is.
+- **Realtime** resumes from the cursor exactly as Events sent it (`lastEventId` is a string; a bare number is no
+ longer read as a position), and it dedupes by `event_id` (the last 512), never by position. The callback's
+ `{ cursor }` is what to save; `seq` is informational and null when absent.
+- **`parseDelivery()`** answers `seq: null` when a delivery carries none (it was `NaN`). Dedupe deliveries by
+ `event.event_id`.
+- **Mock Events** (`openvibe-sdk/testing`) behaves like Events after the change: `after_seq` is a 400, pages have no
+ numeric fields, checkpoints hold cursors, the realtime `id:` is the cursor and a bare-number `Last-Event-ID` is
+ ignored.
+
## 0.41.0 (2026-10-10)
## 0.41.0 (2026-10-10)
- **`createResourceIndex().list({ owner })`.** The index passes `owner` (a `usr_…`/`agt_…` subject id) to every
- **`createResourceIndex().list({ owner })`.** The index passes `owner` (a `usr_…`/`agt_…` subject id) to every
@@ -46,8 +46,8 @@ const community = createCommunityClient(client);
@@ -46,8 +46,8 @@ const community = createCommunityClient(client);
const { pastes } = await community.pastes.list({ limit: 20 });
const { pastes } = await community.pastes.list({ limit: 20 });
// Live updates, resumable across reloads.
// Live updates, resumable across reloads.
-const sub = subscribe(['live.stream.*'], (event, { seq }) => render(event), {
- client, lastEventId: localStorage.getItem('seq'), onGap: () => refetchEverything(),
+const sub = subscribe(['live.stream.*'], (event, { cursor }) => { render(event); localStorage.setItem('cursor', cursor); }, {
+ client, lastEventId: localStorage.getItem('cursor'), onGap: () => refetchEverything(),
});
});
```
```
@@ -108,7 +108,7 @@ const { createAppEvents } = require('openvibe-sdk/events');
@@ -108,7 +108,7 @@ const { createAppEvents } = require('openvibe-sdk/events');
const events = createAppEvents(client, { projectId: unverifiedClaims.project_id, appId: process.env.OV_CLIENT_ID });
const events = createAppEvents(client, { projectId: unverifiedClaims.project_id, appId: process.env.OV_CLIENT_ID });
await events.publish({ event_type: 'order.shipped', subject: { type: 'order', id: 'o1' }, payload: { n: 1 } });
await events.publish({ event_type: 'order.shipped', subject: { type: 'order', id: 'o1' }, payload: { n: 1 } });
// -> app.p01k5….order.shipped, source app-01k5…, actor { type: 'app', id: 'app_…' } (onBehalfOf: the person instead)
// -> app.p01k5….order.shipped, source app-01k5…, actor { type: 'app', id: 'app_…' } (onBehalfOf: the person instead)
-const page = await events.pull({ topic: 'order.*', platformTopics: ['live.stream.*'], afterSeq }); // + public first-party events
+const page = await events.pull({ topic: 'order.*', platformTopics: ['live.stream.*'], after: saved }); // + public first-party events; save page.next_cursor
await events.subscribe({ topicPattern: '*', endpoint: 'https://hooks.example.com/openvibe' }); // public https only
await events.subscribe({ topicPattern: '*', endpoint: 'https://hooks.example.com/openvibe' }); // public https only
```
```
@@ -217,7 +217,7 @@ Server-only subpaths are declared `"browser": null` in the exports map, so a bun
@@ -217,7 +217,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 `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.
+- **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`, or `latest_cursor` to start at the head; a position is never a number, and `afterSeq` throws) 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.41.0",
+ "version": "0.42.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",
@@ -12,8 +12,8 @@
@@ -12,8 +12,8 @@
"core client: retries, timeouts, problem+json errors, trace propagation",
"core client: retries, timeouts, problem+json errors, trace propagation",
"auth: service and app tokens, PKCE, offline verification, the Network signing keys (createNetworkKeys), revocation cutoffs",
"auth: service and app tokens, PKCE, offline verification, the Network signing keys (createNetworkKeys), revocation cutoffs",
"registry discovery and contracts version checks",
"registry discovery and contracts version checks",
- "events: publish, pull, app events, webhook verification (v2), outbox and inbox",
- "realtime SSE",
+ "events: publish, pull and checkpoints by opaque cursor only, app events, webhook verification (v2), outbox and inbox",
+ "realtime SSE (resumes from the opaque cursor, dedupes by event_id)",
"media files and objects v2",
"media files and objects v2",
"community, bot, search, chat, ai, jobs, tools, projects, modules, identity, openre, geo and vip clients",
"community, bot, search, chat, ai, jobs, tools, projects, modules, identity, openre, geo and vip clients",
"bot (openvibe-sdk/bot): OpenVibe.Bot robots, operators, devices, streaming, commands and audit, the public kits and profiles, and pairing",
"bot (openvibe-sdk/bot): OpenVibe.Bot robots, operators, devices, streaming, commands and audit, the public kits and profiles, and pairing",
@@ -1,4 +1,4 @@
@@ -1,4 +1,4 @@
-// Generated by scripts/browser-bundle.js from browser.js (openvibe-sdk 0.41.0, MIT). Do not edit.
+// Generated by scripts/browser-bundle.js from browser.js (openvibe-sdk 0.42.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.41.0';
+const SDK_VERSION = '0.42.0';
module.exports = {
module.exports = {
createClient,
createClient,
@@ -955,7 +955,7 @@ function (module, exports) {
@@ -955,7 +955,7 @@ function (module, exports) {
/**
/**
* openvibe-sdk/realtime: the browser realtime plane inside OpenVibe.Events (ADR-005), over SSE.
* openvibe-sdk/realtime: the browser realtime plane inside OpenVibe.Events (ADR-005), over SSE.
*
*
- * const sub = subscribe(['live.stream.*'], (event, { seq, cursor }) => { … save cursor … }, {
+ * const sub = subscribe(['live.stream.*'], (event, { cursor }) => { … save cursor … }, {
* client, // or url: 'https://openvibe.events'
* client, // or url: 'https://openvibe.events'
* lastEventId: savedCursor, // resume after a reload (the `cursor` the callback was given)
* lastEventId: savedCursor, // resume after a reload (the `cursor` the callback was given)
* onGap: (gap) => refetchState(),
* onGap: (gap) => refetchState(),
@@ -965,13 +965,14 @@ function (module, exports) {
@@ -965,13 +965,14 @@ function (module, exports) {
* Transport: EventSource when the environment has one and no Bearer token is needed (cookies ride
* Transport: EventSource when the environment has one and no Bearer token is needed (cookies ride
* along with withCredentials); otherwise fetch with a streamed body (Node, or a token). Either
* along with withCredentials); otherwise fetch with a streamed body (Node, or a token). Either
* way it resumes from the last position it saw (Last-Event-ID: Events' opaque cursor, the `cursor` each
* way it resumes from the last position it saw (Last-Event-ID: Events' opaque cursor, the `cursor` each
- * callback gets; an older Events sends a bare seq), skips anything it already delivered,
+ * callback gets, never parsed), skips an event it already delivered (by event_id),
* and reports an `event: gap` (events missed beyond retention or replay limits) through onGap.
* and reports an `event: gap` (events missed beyond retention or replay limits) through onGap.
* Browser-safe.
* Browser-safe.
*/
*/
const { OpenVibeError } = __load(3);
const { OpenVibeError } = __load(3);
const DEFAULT_ORIGIN = 'https://openvibe.events';
const DEFAULT_ORIGIN = 'https://openvibe.events';
+const SEEN_MAX = 512; // event ids remembered for dedupe
const FATAL = new Set([400, 401, 403, 404]);
const FATAL = new Set([400, 401, 403, 404]);
const sleep = (ms, signal) => new Promise((resolve) => {
const sleep = (ms, signal) => new Promise((resolve) => {
const t = setTimeout(resolve, ms);
const t = setTimeout(resolve, ms);
@@ -989,10 +990,11 @@ function subscribe(topics, onEvent, opts = {}) {
@@ -989,10 +990,11 @@ function subscribe(topics, onEvent, opts = {}) {
const fetchImpl = opts.fetch || (typeof globalThis.fetch === 'function' ? globalThis.fetch.bind(globalThis) : null);
const fetchImpl = opts.fetch || (typeof globalThis.fetch === 'function' ? globalThis.fetch.bind(globalThis) : null);
const ES = opts.EventSource !== undefined ? opts.EventSource : globalThis.EventSource;
const ES = opts.EventSource !== undefined ? opts.EventSource : globalThis.EventSource;
- // The position to resume from is the last SSE id as Events sent it: an opaque cursor (ADR-042) or, from an older
- // Events, a bare seq. Dedupe compares the numeric seq the message data carries.
+ // The position to resume from is the last SSE id Events sent: an opaque cursor (ADR-042), handed back as it came.
+ // A reconnect replays strictly after it; the ids of the last events delivered catch anything a transport hands
+ // over twice (dedupe by event_id, never by position).
const initial = lastEventId != null && lastEventId !== '' ? String(lastEventId) : null;
const initial = lastEventId != null && lastEventId !== '' ? String(lastEventId) : null;
- const state = { lastId: initial, lastSeq: initial != null && /^\d+$/.test(initial) ? Number(initial) : null, closed: false, transport: null, connected: false };
+ const state = { lastId: initial, seen: new Set(), closed: false, transport: null, connected: false };
const stop = new AbortController();
const stop = new AbortController();
let es = null;
let es = null;
let retryMs = reconnectDelayMs;
let retryMs = reconnectDelayMs;
@@ -1010,12 +1012,14 @@ function subscribe(topics, onEvent, opts = {}) {
@@ -1010,12 +1012,14 @@ function subscribe(topics, onEvent, opts = {}) {
if (type !== 'message' || !data) return;
if (type !== 'message' || !data) return;
let msg;
let msg;
try { msg = JSON.parse(data); } catch { report(new OpenVibeError({ code: 'sdk.bad_response', message: 'realtime: undecodable message' })); return; }
try { msg = JSON.parse(data); } catch { report(new OpenVibeError({ code: 'sdk.bad_response', message: 'realtime: undecodable message' })); return; }
- const seq = Number(msg && msg.seq != null ? msg.seq : id);
- if (Number.isFinite(seq)) {
- if (state.lastSeq != null && seq <= state.lastSeq) return; // already delivered
- state.lastSeq = seq;
+ const eventId = msg && msg.event && typeof msg.event.event_id === 'string' ? msg.event.event_id : null;
+ if (eventId) {
+ if (state.seen.has(eventId)) return; // already delivered
+ state.seen.add(eventId);
+ if (state.seen.size > SEEN_MAX) state.seen.delete(state.seen.values().next().value);
}
}
if (id !== undefined && id !== '') state.lastId = String(id);
if (id !== undefined && id !== '') state.lastId = String(id);
+ const seq = msg && msg.seq != null && Number.isFinite(Number(msg.seq)) ? Number(msg.seq) : null; // informational
try { onEvent(msg && msg.event, { seq, cursor: id !== undefined && id !== '' ? String(id) : null }); } catch (err) { report(err); }
try { onEvent(msg && msg.event, { seq, cursor: id !== undefined && id !== '' ? String(id) : null }); } catch (err) { report(err); }
}
}
@@ -1096,7 +1100,7 @@ function subscribe(topics, onEvent, opts = {}) {
@@ -1096,7 +1100,7 @@ function subscribe(topics, onEvent, opts = {}) {
es.onerror = () => {
es.onerror = () => {
state.connected = false;
state.connected = false;
// CONNECTING (0): the browser reconnects by itself and sends Last-Event-ID.
// CONNECTING (0): the browser reconnects by itself and sends Last-Event-ID.
- // CLOSED (2): it gave up (HTTP error); reconnect ourselves from the last seq.
+ // CLOSED (2): it gave up (HTTP error); reconnect ourselves from the last cursor.
if (es.readyState === 2 && !state.closed) {
if (es.readyState === 2 && !state.closed) {
report(new OpenVibeError({ code: 'sdk.network_error', retryable: true, message: 'realtime: stream closed' }));
report(new OpenVibeError({ code: 'sdk.network_error', retryable: true, message: 'realtime: stream closed' }));
failures++;
failures++;
@@ -1,12 +1,12 @@
@@ -1,12 +1,12 @@
{
{
"name": "openvibe-sdk",
"name": "openvibe-sdk",
- "version": "0.41.0",
+ "version": "0.42.0",
"lockfileVersion": 3,
"lockfileVersion": 3,
"requires": true,
"requires": true,
"packages": {
"packages": {
"": {
"": {
"name": "openvibe-sdk",
"name": "openvibe-sdk",
- "version": "0.41.0",
+ "version": "0.42.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.41.0",
+ "version": "0.42.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.41.0';
+const SDK_VERSION = '0.42.0';
module.exports = {
module.exports = {
createClient,
createClient,
@@ -112,7 +112,8 @@ function verifyDeliveryV2(rawBody, headers, secret, { toleranceSec = V2_TOLERANC
@@ -112,7 +112,8 @@ function verifyDeliveryV2(rawBody, headers, secret, { toleranceSec = V2_TOLERANC
/**
/**
* Verify and parse one delivery: { event, seq, subscriptionId, attempt } or null when it does not
* Verify and parse one delivery: { event, seq, subscriptionId, attempt } or null when it does not
- * verify. `headers` is req.headers (or a Fetch Headers).
+ * verify. `headers` is req.headers (or a Fetch Headers). `seq` is informational (null when the delivery
+ * carries none): dedupe by event.event_id, never by seq.
*
*
* X-OpenVibe-Signature-V2 present it must verify and be within ±toleranceSec (default 300) of
* X-OpenVibe-Signature-V2 present it must verify and be within ±toleranceSec (default 300) of
* `now`; a bad or stale v2 is null, never a fallback to v1
* `now`; a bad or stale v2 is null, never a fallback to v1
@@ -130,9 +131,10 @@ function parseDelivery(rawBody, headers, secret, { requireV2 = false, toleranceS
@@ -130,9 +131,10 @@ function parseDelivery(rawBody, headers, secret, { requireV2 = false, toleranceS
let body;
let body;
try { body = JSON.parse(toBuffer(rawBody).toString('utf8')); } catch { return null; }
try { body = JSON.parse(toBuffer(rawBody).toString('utf8')); } catch { return null; }
if (!body || !body.event) return null;
if (!body || !body.event) return null;
+ const seq = body.seq ?? get('x-openvibe-seq');
return {
return {
event: body.event,
event: body.event,
- seq: Number(body.seq ?? get('x-openvibe-seq')),
+ seq: seq == null || seq === '' || !Number.isFinite(Number(seq)) ? null : Number(seq),
subscriptionId: get('x-openvibe-subscription-id') || null,
subscriptionId: get('x-openvibe-subscription-id') || null,
attempt: Number(get('x-openvibe-delivery-attempt')) || 1,
attempt: Number(get('x-openvibe-delivery-attempt')) || 1,
};
};
@@ -168,8 +170,8 @@ function createEventsClient(client, { source, baseUrl } = {}) {
@@ -168,8 +170,8 @@ function createEventsClient(client, { source, baseUrl } = {}) {
}
}
/**
/**
- * publish(envelope | envelope[], { traceparent }) -> { event_id, seq, duplicate } | { results }.
- * Safe to retry: Events stores an event_id once and answers a repeat with the stored seq.
+ * publish(envelope | envelope[], { traceparent }) -> { event_id, seq, cursor, duplicate } | { results }.
+ * Safe to retry: Events stores an event_id once and answers a repeat with the stored position.
*/
*/
async function publish(input, { traceparent } = {}) {
async function publish(input, { traceparent } = {}) {
const span = startSpan(traceparent || client.traceparent());
const span = startSpan(traceparent || client.traceparent());
@@ -182,49 +184,45 @@ function createEventsClient(client, { source, baseUrl } = {}) {
@@ -182,49 +184,45 @@ function createEventsClient(client, { source, baseUrl } = {}) {
}
}
/**
/**
- * One page: { events: [{ seq, cursor, event }], next_cursor, next_after_seq, latest_seq, latest_cursor, gap? }.
+ * One page: { events: [{ seq, cursor, event }], next_cursor, latest_cursor, gap? }.
* Pass `after` — a page's opaque `next_cursor` — to resume exactly where the previous page ended, or
* Pass `after` — a page's opaque `next_cursor` — to resume exactly where the previous page ended, or
- * `latest_cursor` to start at the head without history; `afterSeq`, the numeric position, remains for
- * one release.
+ * `latest_cursor` to start at the head without history; without it the page starts at the oldest retained
+ * event. A position is only ever an opaque cursor (ADR-042): the numeric `afterSeq` is gone, and passing it
+ * throws rather than silently reading from the start.
*/
*/
- function pull({ topic = '*', after, afterSeq = 0, limit } = {}) {
+ function pull({ topic = '*', after, limit, afterSeq } = {}) {
+ if (afterSeq !== undefined) throw new TypeError("pull: afterSeq is retired; pass after (a page's next_cursor, or latest_cursor for the head)");
return call({ path: '/api/v1/events', query: {
return call({ path: '/api/v1/events', query: {
topic: Array.isArray(topic) ? topic.join(',') : topic, limit,
topic: Array.isArray(topic) ? topic.join(',') : topic, limit,
- ...(after == null ? { after_seq: afterSeq } : { after }),
+ ...(after == null || after === '' ? {} : { after: String(after) }),
} });
} });
}
}
/**
/**
- * Async iterator over { seq, cursor, event } from `after` (an opaque cursor) or `afterSeq` up to
- * the current head.
+ * Async iterator over { seq, cursor, event } from `after` (an opaque cursor; the oldest retained event
+ * without one) 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_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
+ * page.next_cursor there as your durable cursor: it also moves past events that did not match your
* topics — a page with no matching events still calls onPage and advances it — and
* 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
* a crash (or a break/throw in your loop) before onPage leaves the cursor on the
* last page you finished, so nothing is skipped.
* last page you finished, so nothing is skipped.
*/
*/
- 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;
+ async function* iterate({ topic, after, limit, onGap, onPage, maxPages = Infinity, afterSeq } = {}) {
+ if (afterSeq !== undefined) throw new TypeError('iterate: afterSeq is retired; pass after (a saved next_cursor)');
+ let cursor = after == null || after === '' ? null : String(after);
for (let pages = 0; pages < maxPages; pages++) {
for (let pages = 0; pages < maxPages; pages++) {
- const page = await pull({ topic, ...(useCursor ? { after: cursor } : { afterSeq: cursor }), limit });
+ const page = await pull({ topic, ...(cursor == null ? {} : { after: 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 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;
+ const next = page.next_cursor;
+ // The head: the page reached latest_cursor (both name a position in the same epoch, so equal strings are
+ // the same position; nothing is parsed), or it did not move the position at all.
+ if (next == null || next === cursor || next === page.latest_cursor) return;
cursor = next;
cursor = next;
}
}
}
}
@@ -246,10 +244,15 @@ function createEventsClient(client, { source, baseUrl } = {}) {
@@ -246,10 +244,15 @@ function createEventsClient(client, { source, baseUrl } = {}) {
publish,
publish,
pull,
pull,
iterate,
iterate,
- /** { seq, event } or null. */
+ /** { seq, cursor, event } or null. */
get: (eventId) => orNull(call({ path: `/api/v1/events/${encodeURIComponent(eventId)}` })),
get: (eventId) => orNull(call({ path: `/api/v1/events/${encodeURIComponent(eventId)}` })),
+ /** { consumer, topic, cursor, carrier, updated_at }: the stored opaque cursor, or cursor null when none is stored. */
getCheckpoint: (topic) => call({ path: '/api/v1/checkpoints', query: { topic } }),
getCheckpoint: (topic) => call({ path: '/api/v1/checkpoints', query: { topic } }),
- setCheckpoint: (topic, cursor) => call({ method: 'PUT', path: '/api/v1/checkpoints', json: { topic, cursor } }),
+ /** Store an opaque cursor (a page's next_cursor) for a topic pattern; `carrier` records the carrier it is on. */
+ setCheckpoint: (topic, cursor, { carrier } = {}) => {
+ if (typeof cursor !== 'string' || !cursor) return Promise.reject(new TypeError("setCheckpoint: cursor is an opaque cursor string (a page's next_cursor)"));
+ return call({ method: 'PUT', path: '/api/v1/checkpoints', json: { topic, cursor, ...(carrier ? { carrier } : {}) } });
+ },
subscriptions,
subscriptions,
subscribe: subscriptions.create,
subscribe: subscriptions.create,
/** Operators (events.delivery.admin): { deliveries, counts }. */
/** Operators (events.delivery.admin): { deliveries, counts }. */
@@ -359,7 +362,7 @@ function createAppEvents(client, { projectId, appId, onBehalfOf, baseUrl } = {})
@@ -359,7 +362,7 @@ function createAppEvents(client, { projectId, appId, onBehalfOf, baseUrl } = {})
iterate: ({ topic: t = '*', platformTopics, ...rest } = {}) => events.iterate({ ...rest, topic: readTopics(t, platformTopics) }),
iterate: ({ topic: t = '*', platformTopics, ...rest } = {}) => events.iterate({ ...rest, topic: readTopics(t, platformTopics) }),
get: (eventId) => events.get(eventId),
get: (eventId) => events.get(eventId),
getCheckpoint: (t) => events.getCheckpoint(topic(t)),
getCheckpoint: (t) => events.getCheckpoint(topic(t)),
- setCheckpoint: (t, cursor) => events.setCheckpoint(topic(t), cursor),
+ setCheckpoint: (t, cursor, opts) => events.setCheckpoint(topic(t), cursor, opts),
subscriptions: {
subscriptions: {
create: ({ topicPattern = '*', ...rest } = {}) => events.subscriptions.create({ ...rest, topicPattern: topic(topicPattern) }),
create: ({ topicPattern = '*', ...rest } = {}) => events.subscriptions.create({ ...rest, topicPattern: topic(topicPattern) }),
list: () => events.subscriptions.list(),
list: () => events.subscriptions.list(),
@@ -2,7 +2,7 @@
@@ -2,7 +2,7 @@
/**
/**
* openvibe-sdk/realtime: the browser realtime plane inside OpenVibe.Events (ADR-005), over SSE.
* openvibe-sdk/realtime: the browser realtime plane inside OpenVibe.Events (ADR-005), over SSE.
*
*
- * const sub = subscribe(['live.stream.*'], (event, { seq, cursor }) => { … save cursor … }, {
+ * const sub = subscribe(['live.stream.*'], (event, { cursor }) => { … save cursor … }, {
* client, // or url: 'https://openvibe.events'
* client, // or url: 'https://openvibe.events'
* lastEventId: savedCursor, // resume after a reload (the `cursor` the callback was given)
* lastEventId: savedCursor, // resume after a reload (the `cursor` the callback was given)
* onGap: (gap) => refetchState(),
* onGap: (gap) => refetchState(),
@@ -12,13 +12,14 @@
@@ -12,13 +12,14 @@
* Transport: EventSource when the environment has one and no Bearer token is needed (cookies ride
* Transport: EventSource when the environment has one and no Bearer token is needed (cookies ride
* along with withCredentials); otherwise fetch with a streamed body (Node, or a token). Either
* along with withCredentials); otherwise fetch with a streamed body (Node, or a token). Either
* way it resumes from the last position it saw (Last-Event-ID: Events' opaque cursor, the `cursor` each
* way it resumes from the last position it saw (Last-Event-ID: Events' opaque cursor, the `cursor` each
- * callback gets; an older Events sends a bare seq), skips anything it already delivered,
+ * callback gets, never parsed), skips an event it already delivered (by event_id),
* and reports an `event: gap` (events missed beyond retention or replay limits) through onGap.
* and reports an `event: gap` (events missed beyond retention or replay limits) through onGap.
* Browser-safe.
* Browser-safe.
*/
*/
const { OpenVibeError } = require('./core/errors');
const { OpenVibeError } = require('./core/errors');
const DEFAULT_ORIGIN = 'https://openvibe.events';
const DEFAULT_ORIGIN = 'https://openvibe.events';
+const SEEN_MAX = 512; // event ids remembered for dedupe
const FATAL = new Set([400, 401, 403, 404]);
const FATAL = new Set([400, 401, 403, 404]);
const sleep = (ms, signal) => new Promise((resolve) => {
const sleep = (ms, signal) => new Promise((resolve) => {
const t = setTimeout(resolve, ms);
const t = setTimeout(resolve, ms);
@@ -36,10 +37,11 @@ function subscribe(topics, onEvent, opts = {}) {
@@ -36,10 +37,11 @@ function subscribe(topics, onEvent, opts = {}) {
const fetchImpl = opts.fetch || (typeof globalThis.fetch === 'function' ? globalThis.fetch.bind(globalThis) : null);
const fetchImpl = opts.fetch || (typeof globalThis.fetch === 'function' ? globalThis.fetch.bind(globalThis) : null);
const ES = opts.EventSource !== undefined ? opts.EventSource : globalThis.EventSource;
const ES = opts.EventSource !== undefined ? opts.EventSource : globalThis.EventSource;
- // The position to resume from is the last SSE id as Events sent it: an opaque cursor (ADR-042) or, from an older
- // Events, a bare seq. Dedupe compares the numeric seq the message data carries.
+ // The position to resume from is the last SSE id Events sent: an opaque cursor (ADR-042), handed back as it came.
+ // A reconnect replays strictly after it; the ids of the last events delivered catch anything a transport hands
+ // over twice (dedupe by event_id, never by position).
const initial = lastEventId != null && lastEventId !== '' ? String(lastEventId) : null;
const initial = lastEventId != null && lastEventId !== '' ? String(lastEventId) : null;
- const state = { lastId: initial, lastSeq: initial != null && /^\d+$/.test(initial) ? Number(initial) : null, closed: false, transport: null, connected: false };
+ const state = { lastId: initial, seen: new Set(), closed: false, transport: null, connected: false };
const stop = new AbortController();
const stop = new AbortController();
let es = null;
let es = null;
let retryMs = reconnectDelayMs;
let retryMs = reconnectDelayMs;
@@ -57,12 +59,14 @@ function subscribe(topics, onEvent, opts = {}) {
@@ -57,12 +59,14 @@ function subscribe(topics, onEvent, opts = {}) {
if (type !== 'message' || !data) return;
if (type !== 'message' || !data) return;
let msg;
let msg;
try { msg = JSON.parse(data); } catch { report(new OpenVibeError({ code: 'sdk.bad_response', message: 'realtime: undecodable message' })); return; }
try { msg = JSON.parse(data); } catch { report(new OpenVibeError({ code: 'sdk.bad_response', message: 'realtime: undecodable message' })); return; }
- const seq = Number(msg && msg.seq != null ? msg.seq : id);
- if (Number.isFinite(seq)) {
- if (state.lastSeq != null && seq <= state.lastSeq) return; // already delivered
- state.lastSeq = seq;
+ const eventId = msg && msg.event && typeof msg.event.event_id === 'string' ? msg.event.event_id : null;
+ if (eventId) {
+ if (state.seen.has(eventId)) return; // already delivered
+ state.seen.add(eventId);
+ if (state.seen.size > SEEN_MAX) state.seen.delete(state.seen.values().next().value);
}
}
if (id !== undefined && id !== '') state.lastId = String(id);
if (id !== undefined && id !== '') state.lastId = String(id);
+ const seq = msg && msg.seq != null && Number.isFinite(Number(msg.seq)) ? Number(msg.seq) : null; // informational
try { onEvent(msg && msg.event, { seq, cursor: id !== undefined && id !== '' ? String(id) : null }); } catch (err) { report(err); }
try { onEvent(msg && msg.event, { seq, cursor: id !== undefined && id !== '' ? String(id) : null }); } catch (err) { report(err); }
}
}
@@ -143,7 +147,7 @@ function subscribe(topics, onEvent, opts = {}) {
@@ -143,7 +147,7 @@ function subscribe(topics, onEvent, opts = {}) {
es.onerror = () => {
es.onerror = () => {
state.connected = false;
state.connected = false;
// CONNECTING (0): the browser reconnects by itself and sends Last-Event-ID.
// CONNECTING (0): the browser reconnects by itself and sends Last-Event-ID.
- // CLOSED (2): it gave up (HTTP error); reconnect ourselves from the last seq.
+ // CLOSED (2): it gave up (HTTP error); reconnect ourselves from the last cursor.
if (es.readyState === 2 && !state.closed) {
if (es.readyState === 2 && !state.closed) {
report(new OpenVibeError({ code: 'sdk.network_error', retryable: true, message: 'realtime: stream closed' }));
report(new OpenVibeError({ code: 'sdk.network_error', retryable: true, message: 'realtime: stream closed' }));
failures++;
failures++;
@@ -458,12 +458,13 @@ function createMockPlatform(opts = {}) {
@@ -458,12 +458,13 @@ 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);
+ // Like Events: a position is an opaque cursor; the numeric after_seq is refused, never read as "from the start".
+ if (url.searchParams.has('after_seq')) return problem(400, 'events.bad_request', "after_seq is retired: pass after=<cursor> (a page's next_cursor, or latest_cursor for the head)");
const afterCursor = url.searchParams.get('after');
const afterCursor = url.searchParams.get('after');
- const after = afterCursor ? eventSeqFromCursor(afterCursor) : Number(url.searchParams.get('after_seq') || 0);
+ const after = afterCursor ? eventSeqFromCursor(afterCursor) : 0;
const limit = Number(url.searchParams.get('limit') || 100);
const limit = Number(url.searchParams.get('limit') || 100);
- 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');
- }
+ if (after === null) return problem(400, 'events.bad_request', 'after must be an opaque cursor');
+ if (!Number.isInteger(limit) || limit < 1 || limit > 1000) return problem(400, 'events.bad_request', 'limit must be 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 = {};
@@ -480,7 +481,7 @@ function createMockPlatform(opts = {}) {
@@ -480,7 +481,7 @@ function createMockPlatform(opts = {}) {
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 (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, next_cursor: eventCursor(cursor), latest_seq: lastSeq, latest_cursor: eventCursor(lastSeq) });
+ return json(200, { ...page, events: out, next_cursor: eventCursor(cursor), latest_cursor: eventCursor(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');
@@ -495,15 +496,15 @@ function createMockPlatform(opts = {}) {
@@ -495,15 +496,15 @@ function createMockPlatform(opts = {}) {
const b = req.method === 'GET' ? {} : await req.json().catch(() => ({}));
const b = req.method === 'GET' ? {} : await req.json().catch(() => ({}));
const topic = req.method === 'GET' ? url.searchParams.get('topic') || '' : b.topic;
const topic = req.method === 'GET' ? url.searchParams.get('topic') || '' : b.topic;
if (!appRules.isValidPattern(topic)) return problem(400, 'events.bad_request', 'topic (pattern) is required');
if (!appRules.isValidPattern(topic)) return problem(400, 'events.bad_request', 'topic (pattern) is required');
- if (req.method !== 'GET' && (!Number.isInteger(b.cursor) || b.cursor < 0)) return problem(400, 'events.bad_request', 'topic (pattern) and cursor (integer >= 0) are required');
+ if (req.method !== 'GET' && (typeof b.cursor !== 'string' || eventSeqFromCursor(b.cursor) === null)) return problem(400, 'events.bad_request', 'topic (pattern) and cursor (an opaque cursor) are required');
const scopeErr = scopeError(who.principal, [topic]);
const scopeErr = scopeError(who.principal, [topic]);
if (scopeErr) return problem(403, 'events.topic_not_allowed', scopeErr);
if (scopeErr) return problem(403, 'events.topic_not_allowed', scopeErr);
const k = `${who.consumer}|${topic}`;
const k = `${who.consumer}|${topic}`;
if (req.method === 'GET') {
if (req.method === 'GET') {
const cp = checkpoints.get(k);
const cp = checkpoints.get(k);
- return json(200, { consumer: who.consumer, topic, cursor: cp ? cp.cursor : 0, updated_at: cp ? cp.updated_at : null });
+ return json(200, { consumer: who.consumer, topic, cursor: cp ? cp.cursor : null, carrier: cp ? cp.carrier : null, updated_at: cp ? cp.updated_at : null });
}
}
- const cp = { cursor: b.cursor, updated_at: new Date().toISOString() };
+ const cp = { cursor: b.cursor, carrier: typeof b.carrier === 'string' && b.carrier ? b.carrier : null, updated_at: new Date().toISOString() };
checkpoints.set(k, cp);
checkpoints.set(k, cp);
return json(200, { consumer: who.consumer, topic, ...cp });
return json(200, { consumer: who.consumer, topic, ...cp });
}
}
@@ -572,8 +573,9 @@ function createMockPlatform(opts = {}) {
@@ -572,8 +573,9 @@ function createMockPlatform(opts = {}) {
if (/^app:/.test(String(claims.sub)) && claims.env === 'sandbox') return problem(401, 'token.sandbox_refused', 'sandbox tokens are not accepted here');
if (/^app:/.test(String(claims.sub)) && claims.env === 'sandbox') return problem(401, 'token.sandbox_refused', 'sandbox tokens are not accepted here');
viewer = { kind: 'service' };
viewer = { kind: 'service' };
} else if (claims) viewer = { kind: 'user', subject: claims.subject_id };
} else if (claims) viewer = { kind: 'user', subject: claims.subject_id };
+ // Like Events: Last-Event-ID is an opaque cursor; anything else (a bare number included) is ignored.
const raw = req.headers.get('last-event-id') ?? url.searchParams.get('last_event_id');
const raw = req.headers.get('last-event-id') ?? url.searchParams.get('last_event_id');
- const last = raw != null && /^\d+$/.test(raw) ? Number(raw) : null;
+ const last = raw != null ? eventSeqFromCursor(String(raw).trim()) : null;
const enc = new TextEncoder();
const enc = new TextEncoder();
let conn;
let conn;
const body = new ReadableStream({
const body = new ReadableStream({
@@ -585,7 +587,7 @@ function createMockPlatform(opts = {}) {
@@ -585,7 +587,7 @@ function createMockPlatform(opts = {}) {
push(e) {
push(e) {
if (e.seq <= conn.lastSeq || !pats.some((re) => re.test(e.event.event_type)) || !visible(viewer, e)) return;
if (e.seq <= conn.lastSeq || !pats.some((re) => re.test(e.event.event_type)) || !visible(viewer, e)) return;
conn.lastSeq = e.seq;
conn.lastSeq = e.seq;
- send(`id: ${e.seq}\ndata: ${JSON.stringify({ seq: e.seq, event: e.event })}\n\n`);
+ send(`id: ${eventCursor(e.seq)}\ndata: ${JSON.stringify({ seq: e.seq, event: e.event })}\n\n`);
},
},
close() { streams.delete(conn); try { controller.close(); } catch { /* closed */ } },
close() { streams.delete(conn); try { controller.close(); } catch { /* closed */ } },
};
};
@@ -134,7 +134,6 @@ run([
@@ -134,7 +134,6 @@ run([
assert.match(orderPage.events[0].cursor, /^c1\./, 'a page item carries its cursor');
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.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.*', 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');
@@ -158,11 +157,11 @@ run([
@@ -158,11 +157,11 @@ run([
assert.equal((await live.get(prod.event_id)).event.event_id, prod.event_id);
assert.equal((await live.get(prod.event_id)).event.event_id, prod.event_id);
// Checkpoints are per app and scoped the same way.
// Checkpoints are per app and scoped the same way.
- const cp = await a.setCheckpoint('*', 5);
+ const cp = await a.setCheckpoint('*', orderPage.next_cursor);
assert.equal(cp.consumer, `app:${A}`);
assert.equal(cp.consumer, `app:${A}`);
assert.equal(cp.topic, `app.${K1}.*`);
assert.equal(cp.topic, `app.${K1}.*`);
- assert.equal((await a.getCheckpoint('*')).cursor, 5);
- assert.equal((await r.getCheckpoint('*')).cursor, 0, 'another app has its own');
+ assert.equal((await a.getCheckpoint('*')).cursor, orderPage.next_cursor);
+ assert.equal((await r.getCheckpoint('*')).cursor, null, 'another app has its own');
await assert.rejects(b.events.getCheckpoint(`app.${K1}.*`), { status: 403, code: 'events.topic_not_allowed' });
await assert.rejects(b.events.getCheckpoint(`app.${K1}.*`), { status: 403, code: 'events.topic_not_allowed' });
}],
}],
@@ -65,12 +65,11 @@ run([
@@ -65,12 +65,11 @@ run([
}],
}],
['pull + iterate walk the cursor across pages; onPage gives the durable cursor', async () => {
['pull + iterate walk the cursor across pages; onPage gives the durable cursor', async () => {
- const { platform, events } = setup();
+ const { platform, client, events } = setup();
for (let i = 1; i <= 5; i++) platform.publishEvent({ event_type: i % 2 ? 'media.vod.ready' : 'live.stream.started', source: 'media', actor, subject: { type: 'vod', id: String(i) } });
for (let i = 1; i <= 5; i++) platform.publishEvent({ event_type: i % 2 ? 'media.vod.ready' : 'live.stream.started', source: 'media', actor, subject: { type: 'vod', id: String(i) } });
- const page = await events.pull({ topic: 'media.vod.*', afterSeq: 0, limit: 2 });
- assert.deepEqual(page.events.map((e) => e.seq), [1, 3]);
- assert.equal(page.next_after_seq, 3);
- assert.equal(page.latest_seq, 5);
+ const page = await events.pull({ topic: 'media.vod.*', limit: 2 });
+ assert.deepEqual(page.events.map((e) => e.seq), [1, 3], 'without after, from the oldest retained event');
+ assert.ok(!('next_after_seq' in page) && !('latest_seq' in page), 'no numeric position on the page');
assert.match(page.next_cursor, /^c1\./, 'the page carries an opaque next_cursor');
assert.match(page.next_cursor, /^c1\./, 'the page carries an opaque next_cursor');
assert.match(page.latest_cursor, /^c1\./, 'and the head as an opaque latest_cursor');
assert.match(page.latest_cursor, /^c1\./, 'and the head as an opaque latest_cursor');
assert.deepStrictEqual((await events.pull({ topic: 'media.vod.*', after: page.latest_cursor })).events, [], 'nothing after the head');
assert.deepStrictEqual((await events.pull({ topic: 'media.vod.*', after: page.latest_cursor })).events, [], 'nothing after the head');
@@ -79,24 +78,32 @@ run([
@@ -79,24 +78,32 @@ run([
const rest = await events.pull({ topic: 'media.vod.*', after: page.next_cursor, limit: 2 });
const rest = await events.pull({ topic: 'media.vod.*', after: page.next_cursor, limit: 2 });
assert.deepEqual(rest.events.map((e) => e.seq), [5]);
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);
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]);
+ // The numeric position is gone: passing it throws instead of reading from the start, and Events refuses it.
+ assert.throws(() => events.pull({ topic: 'media.vod.*', afterSeq: 3 }), /afterSeq is retired/);
+ await assert.rejects(async () => { for await (const e of events.iterate({ topic: 'media.vod.*', afterSeq: 0 })) void e; }, /afterSeq is retired/);
+ await assert.rejects(client.json({ service: 'events', audience: 'openvibe.events', path: '/api/v1/events', query: { topic: 'media.vod.*', after_seq: 3 } }),
+ { status: 400, code: 'events.bad_request' }, 'after_seq is refused, never read as "from the start"');
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_cursor) })) {
+ for await (const { seq, event } of events.iterate({ topic: ['media.vod.*'], 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.match(cursors.at(-1), /^c1\./);
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');
+ assert.equal(cursors.at(-1), (await events.pull({ topic: 'media.vod.*', limit: 1 })).latest_cursor, 'iterate ends on the head');
+ assert.equal(cursors.length, 2, 'and stops there: the last page reached latest_cursor');
const resumed = [];
const resumed = [];
for await (const { seq } of events.iterate({ topic: 'media.vod.*', after: page.next_cursor, limit: 2 })) resumed.push(seq);
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.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);
- assert.equal((await events.getCheckpoint('media.vod.*')).cursor, 5);
+ assert.equal((await events.getCheckpoint('media.vod.*')).cursor, null, 'none stored yet');
+ await events.setCheckpoint('media.vod.*', page.next_cursor, { carrier: 'pg' });
+ const cp = await events.getCheckpoint('media.vod.*');
+ assert.equal(cp.cursor, page.next_cursor, 'the opaque cursor comes back as it was stored');
+ assert.equal(cp.carrier, 'pg');
+ await assert.rejects(events.setCheckpoint('media.vod.*', 5), /opaque cursor string/, 'a number is not a position');
}],
}],
['iterate: onPage advances the cursor even on a page that matches no events', async () => {
['iterate: onPage advances the cursor even on a page that matches no events', async () => {
@@ -106,8 +113,7 @@ run([
@@ -106,8 +113,7 @@ run([
for await (const e of events.iterate({ topic: 'media.vod.*', onPage: (p) => pages.push(p) })) void e;
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.equal(pages.length, 1, 'the empty page still called onPage');
assert.deepEqual(pages[0].events, []);
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');
+ assert.equal(pages[0].next_cursor, pages[0].latest_cursor, 'the cursor advanced past the scanned, non-matching event to the head');
// Resuming from that cursor does not replay the scanned event: the new match is seen once.
// Resuming from that cursor does not replay the scanned event: the new match is seen once.
const saved = pages[0].next_cursor;
const saved = pages[0].next_cursor;
platform.publishEvent({ event_type: 'media.vod.ready', source: 'media', actor, subject: { type: 'vod', id: '2' } });
platform.publishEvent({ event_type: 'media.vod.ready', source: 'media', actor, subject: { type: 'vod', id: '2' } });
@@ -117,10 +123,12 @@ run([
@@ -117,10 +123,12 @@ run([
}],
}],
['iterate reports a retention gap through onGap', async () => {
['iterate reports a retention gap through onGap', async () => {
+ const asked = [];
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'));
- if (after === 0) return send(res, 200, { gap: { from_seq: 1, to_seq: 9 }, events: [{ seq: 10, event: { event_type: 'a.b.c' } }], next_after_seq: 10, latest_seq: 11 });
- return send(res, 200, { events: [{ seq: 11, event: { event_type: 'a.b.c' } }], next_after_seq: 11, latest_seq: 11 });
+ const q = new URL(req.url, 'http://x').searchParams;
+ asked.push([q.get('after'), q.get('after_seq')]);
+ if (!q.get('after')) return send(res, 200, { gap: { from_seq: 1, to_seq: 9 }, events: [{ seq: 10, cursor: 'c1.0.MTA', event: { event_type: 'a.b.c' } }], next_cursor: 'c1.0.MTA', latest_cursor: 'c1.0.MTE' });
+ return send(res, 200, { events: [{ seq: 11, cursor: 'c1.0.MTE', event: { event_type: 'a.b.c' } }], next_cursor: 'c1.0.MTE', latest_cursor: 'c1.0.MTE' });
});
});
const events = createEventsClient(createClient({ baseUrls: { events: srv.url }, token: 't' }), { source: 'live' });
const events = createEventsClient(createClient({ baseUrls: { events: srv.url }, token: 't' }), { source: 'live' });
const gaps = [];
const gaps = [];
@@ -128,6 +136,7 @@ run([
@@ -128,6 +136,7 @@ run([
for await (const e of events.iterate({ topic: 'a.*', onGap: (g) => gaps.push(g) })) seqs.push(e.seq);
for await (const e of events.iterate({ topic: 'a.*', onGap: (g) => gaps.push(g) })) seqs.push(e.seq);
assert.deepEqual(gaps, [{ from_seq: 1, to_seq: 9 }]);
assert.deepEqual(gaps, [{ from_seq: 1, to_seq: 9 }]);
assert.deepEqual(seqs, [10, 11]);
assert.deepEqual(seqs, [10, 11]);
+ assert.deepEqual(asked, [[null, null], ['c1.0.MTA', null]], 'no after_seq is ever sent; the second page reads on from the cursor and the head ends it');
await srv.close();
await srv.close();
}],
}],
@@ -226,20 +235,22 @@ run([
@@ -226,20 +235,22 @@ run([
const { platform, events } = setup();
const { platform, events } = setup();
for (let i = 1; i <= 6; i++) platform.publishEvent({ event_type: i === 3 ? 'live.other.thing' : 'media.vod.ready', source: 'media', actor, subject: { type: 'vod', id: String(i) } });
for (let i = 1; i <= 6; i++) platform.publishEvent({ event_type: i === 3 ? 'live.other.thing' : 'media.vod.ready', source: 'media', actor, subject: { type: 'vod', id: String(i) } });
const log = [];
const log = [];
- let saved = 0;
+ let saved = null;
+ // The mock's cursor format, decoded only to label the log readably; a consumer never parses a cursor.
+ const at = (c) => Number(Buffer.from(String(c).split('.')[2], 'base64url').toString('utf8'));
const consume = async (failAt) => {
const consume = async (failAt) => {
- for await (const { seq } of events.iterate({ topic: 'media.vod.*', afterSeq: saved, limit: 3, onPage: (p) => { log.push(`page ${p.next_after_seq}`); saved = p.next_after_seq; } })) {
+ for await (const { seq } of events.iterate({ topic: 'media.vod.*', after: saved, limit: 3, onPage: (p) => { log.push(`page ${at(p.next_cursor)}`); saved = p.next_cursor; } })) {
if (seq === failAt) throw new Error(`crash at ${seq}`);
if (seq === failAt) throw new Error(`crash at ${seq}`);
log.push(`item ${seq}`);
log.push(`item ${seq}`);
}
}
};
};
await assert.rejects(consume(2), /crash at 2/);
await assert.rejects(consume(2), /crash at 2/);
assert.deepEqual(log, ['item 1']);
assert.deepEqual(log, ['item 1']);
- assert.equal(saved, 0, 'a crash inside the first page saves nothing');
+ assert.equal(saved, null, 'a crash inside the first page saves nothing');
log.length = 0;
log.length = 0;
await assert.rejects(consume(5), /crash at 5/);
await assert.rejects(consume(5), /crash at 5/);
assert.deepEqual(log, ['item 1', 'item 2', 'item 4', 'page 4'], 'page 1 (seq 1-4; 3 did not match) was saved after its last item; page 2 never was');
assert.deepEqual(log, ['item 1', 'item 2', 'item 4', 'page 4'], 'page 1 (seq 1-4; 3 did not match) was saved after its last item; page 2 never was');
- assert.equal(saved, 4);
+ assert.equal(at(saved), 4);
log.length = 0;
log.length = 0;
await consume(null);
await consume(null);
assert.deepEqual(log, ['item 5', 'item 6', 'page 6'], 'resumed from the saved cursor: nothing skipped');
assert.deepEqual(log, ['item 5', 'item 6', 'page 6'], 'resumed from the saved cursor: nothing skipped');
@@ -251,11 +262,13 @@ run([
@@ -251,11 +262,13 @@ run([
['mock Events retention: pull and realtime report the pruned range as a gap', async () => {
['mock Events retention: pull and realtime report the pruned range as a gap', async () => {
const { platform, events } = setup();
const { platform, events } = setup();
for (let i = 1; i <= 5; i++) platform.publishEvent({ event_type: 'media.vod.ready', source: 'media', actor, subject: { type: 'vod', id: String(i) }, visibility: 'public' });
for (let i = 1; i <= 5; i++) platform.publishEvent({ event_type: 'media.vod.ready', source: 'media', actor, subject: { type: 'vod', id: String(i) }, visibility: 'public' });
+ // Cursors handed out before the prune (the mock's cursor of position n; a consumer never builds one).
+ const cur = (n) => `c1.0.${Buffer.from(String(n), 'utf8').toString('base64url')}`;
assert.equal(platform.pruneEvents(3), 3);
assert.equal(platform.pruneEvents(3), 3);
- const page = await events.pull({ topic: 'media.vod.*', afterSeq: 1 });
+ const page = await events.pull({ topic: 'media.vod.*', after: cur(1) });
assert.deepEqual(page.gap, { from_seq: 2, to_seq: 3 });
assert.deepEqual(page.gap, { from_seq: 2, to_seq: 3 });
assert.deepEqual(page.events.map((e) => e.seq), [4, 5]);
assert.deepEqual(page.events.map((e) => e.seq), [4, 5]);
- assert.equal((await events.pull({ topic: 'media.vod.*', afterSeq: 3 })).gap, undefined);
+ assert.equal((await events.pull({ topic: 'media.vod.*', after: cur(3) })).gap, undefined);
const gaps = [];
const gaps = [];
const seqs = [];
const seqs = [];
for await (const e of events.iterate({ topic: 'media.vod.*', onGap: (g) => gaps.push(g) })) seqs.push(e.seq);
for await (const e of events.iterate({ topic: 'media.vod.*', onGap: (g) => gaps.push(g) })) seqs.push(e.seq);
@@ -263,12 +276,12 @@ run([
@@ -263,12 +276,12 @@ run([
assert.deepEqual(seqs, [4, 5]);
assert.deepEqual(seqs, [4, 5]);
const rtGaps = [];
const rtGaps = [];
const got = [];
const got = [];
- const sub = subscribe('media.vod.*', (ev, { seq }) => got.push(seq), { client: createClient({ fetch: platform.fetch }), fetch: platform.fetch, transport: 'fetch', lastEventId: 1, onGap: (g) => rtGaps.push(g) });
+ const sub = subscribe('media.vod.*', (ev, { seq }) => got.push(seq), { client: createClient({ fetch: platform.fetch }), fetch: platform.fetch, transport: 'fetch', lastEventId: cur(1), onGap: (g) => rtGaps.push(g) });
await waitFor(() => got.length === 2);
await waitFor(() => got.length === 2);
sub.close();
sub.close();
assert.deepEqual(rtGaps, [{ reason: 'retention', from_seq: 2, to_seq: 3, latest_seq: 5 }]);
assert.deepEqual(rtGaps, [{ reason: 'retention', from_seq: 2, to_seq: 3, latest_seq: 5 }]);
platform.publishEvent({ event_type: 'media.vod.ready', source: 'media', actor, subject: { type: 'vod', id: '6' } });
platform.publishEvent({ event_type: 'media.vod.ready', source: 'media', actor, subject: { type: 'vod', id: '6' } });
- assert.equal((await events.pull({ afterSeq: 5 })).events[0].seq, 6, 'seqs keep counting after a prune');
+ assert.equal((await events.pull({ after: cur(5) })).events[0].seq, 6, 'positions keep counting after a prune');
}],
}],
['deliverEvents: signed deliveries to a local endpoint, retried in order, dead after max attempts', async () => {
['deliverEvents: signed deliveries to a local endpoint, retried in order, dead after max attempts', async () => {
@@ -319,25 +332,33 @@ run([
@@ -319,25 +332,33 @@ 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 () => {
+ ['iterate pages by cursor alone: the unchanged cursor or latest_cursor ends it', async () => {
const asked = [];
const asked = [];
- // A server past the compatibility release: no next_after_seq / latest_seq, the same cursor at the head.
+ // A server that omits latest_cursor: the page that does not move the position ends the walk.
const pages = { '': { events: [{ seq: 1, cursor: 'c1', event: { id: 'a' } }], next_cursor: 'c1' },
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' },
c1: { events: [{ seq: 2, cursor: 'c2', event: { id: 'b' } }], next_cursor: 'c2' },
c2: { events: [], next_cursor: 'c2' } };
c2: { events: [], next_cursor: 'c2' } };
const fetch = async (url) => {
const fetch = async (url) => {
const u = new URL(url);
const u = new URL(url);
asked.push(u.search);
asked.push(u.search);
- const body = u.searchParams.has('after') ? pages[u.searchParams.get('after')] : { ...pages[''], next_after_seq: 1, latest_seq: 2 };
+ const body = u.searchParams.has('after') ? pages[u.searchParams.get('after')] : pages[''];
return new Response(JSON.stringify(body), { status: 200, headers: { 'content-type': 'application/json' } });
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 events = createEventsClient(createClient({ fetch }), { source: 'live', baseUrl: 'http://events.test' });
const seen = [];
const seen = [];
- for await (const item of events.iterate({ afterSeq: '0' })) seen.push(item.event.id);
+ for await (const item of events.iterate({})) seen.push(item.event.id);
assert.deepEqual(seen, ['a', 'b'], 'every page up to the head');
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.doesNotMatch(asked[0], /after/, 'no position on the first page: the oldest retained event');
assert.match(asked[1], /after=c1/);
assert.match(asked[1], /after=c1/);
assert.match(asked[2], /after=c2/);
assert.match(asked[2], /after=c2/);
assert.equal(asked.length, 3, 'the unchanged cursor at the head ends it');
assert.equal(asked.length, 3, 'the unchanged cursor at the head ends it');
+ // With latest_cursor, the page that reaches the head ends it without one more request.
+ asked.length = 0;
+ pages[''] = { ...pages[''], latest_cursor: 'c2' };
+ pages.c1 = { ...pages.c1, latest_cursor: 'c2' };
+ seen.length = 0;
+ for await (const item of events.iterate({})) seen.push(item.event.id);
+ assert.deepEqual(seen, ['a', 'b']);
+ assert.equal(asked.length, 2, 'the page whose next_cursor is latest_cursor is the last');
}],
}],
]);
]);
@@ -93,9 +93,9 @@ run([
@@ -93,9 +93,9 @@ run([
res.end(JSON.stringify({ status: 401, code: 'token.bad_signature', detail: 'bad', type: 't', title: 'x' }));
res.end(JSON.stringify({ status: 401, code: 'token.bad_signature', detail: 'bad', type: 't', title: 'x' }));
});
});
const errors = [];
const errors = [];
- const sub = subscribe('live.*', () => {}, { url: `${srv.url}/realtime/stream`, lastEventId: 41, onError: (e) => errors.push(e), reconnectDelayMs: 10 });
+ const sub = subscribe('live.*', () => {}, { url: `${srv.url}/realtime/stream`, lastEventId: 'c1.0.NDE', onError: (e) => errors.push(e), reconnectDelayMs: 10 });
await sub.done;
await sub.done;
- assert.equal(srv.requests[0].headers['last-event-id'], '41');
+ assert.equal(srv.requests[0].headers['last-event-id'], 'c1.0.NDE', 'the saved cursor is handed back as it is');
assert.ok(srv.requests[0].url.startsWith('/realtime/stream?topics=live.*'));
assert.ok(srv.requests[0].url.startsWith('/realtime/stream?topics=live.*'));
assert.equal(errors[0].code, 'token.bad_signature');
assert.equal(errors[0].code, 'token.bad_signature');
assert.equal(sub.closed, true);
assert.equal(sub.closed, true);
@@ -112,22 +112,22 @@ run([
@@ -112,22 +112,22 @@ run([
}
}
const got = [];
const got = [];
const gaps = [];
const gaps = [];
- const sub = subscribe(['live.stream.*'], (event, { seq }) => got.push(seq), { url: 'https://events.example', EventSource: FakeES, onGap: (g) => gaps.push(g), reconnectDelayMs: 5 });
+ const sub = subscribe(['live.stream.*'], (event, { seq, cursor }) => got.push([seq, cursor]), { url: 'https://events.example', EventSource: FakeES, onGap: (g) => gaps.push(g), reconnectDelayMs: 5 });
await waitFor(() => made.length === 1);
await waitFor(() => made.length === 1);
assert.equal(sub.transport, 'eventsource');
assert.equal(sub.transport, 'eventsource');
const es = made[0];
const es = made[0];
assert.deepEqual(es.init, { withCredentials: true });
assert.deepEqual(es.init, { withCredentials: true });
assert.equal(new URL(es.url).searchParams.get('last_event_id'), null);
assert.equal(new URL(es.url).searchParams.get('last_event_id'), null);
es.onopen();
es.onopen();
- es.onmessage({ data: JSON.stringify({ seq: 7, event: { event_type: 'live.stream.started' } }), lastEventId: '7' });
- es.onmessage({ data: JSON.stringify({ seq: 7, event: {} }), lastEventId: '7' });
+ es.onmessage({ data: JSON.stringify({ seq: 7, event: { event_id: 'evt_7', event_type: 'live.stream.started' } }), lastEventId: 'c1.0.Nw' });
+ es.onmessage({ data: JSON.stringify({ seq: 7, event: { event_id: 'evt_7', event_type: 'live.stream.started' } }), lastEventId: 'c1.0.Nw' });
es.listeners.gap({ data: JSON.stringify({ reason: 'replay_limit', from_seq: 8, to_seq: 9 }) });
es.listeners.gap({ data: JSON.stringify({ reason: 'replay_limit', from_seq: 8, to_seq: 9 }) });
- assert.deepEqual(got, [7]);
+ assert.deepEqual(got, [[7, 'c1.0.Nw']], 'a repeated event_id is delivered once');
assert.equal(gaps[0].reason, 'replay_limit');
assert.equal(gaps[0].reason, 'replay_limit');
es.readyState = 2;
es.readyState = 2;
es.onerror();
es.onerror();
await waitFor(() => made.length === 2);
await waitFor(() => made.length === 2);
- assert.equal(new URL(made[1].url).searchParams.get('last_event_id'), '7');
+ assert.equal(new URL(made[1].url).searchParams.get('last_event_id'), 'c1.0.Nw', 'it resumes from the opaque cursor');
sub.close();
sub.close();
assert.equal(made[1].closedByClient, true);
assert.equal(made[1].closedByClient, true);
}],
}],
@@ -141,7 +141,8 @@ run([
@@ -141,7 +141,8 @@ run([
pub('live.stream.updated');
pub('live.stream.updated');
const got = [];
const got = [];
const realtime = createRealtimeClient(client, { fetch: platform.fetch, reconnectDelayMs: 10 });
const realtime = createRealtimeClient(client, { fetch: platform.fetch, reconnectDelayMs: 10 });
- const sub = realtime.subscribe('live.stream.*', (event, { seq }) => got.push([seq, event.event_type]), { lastEventId: 0 });
+ // Resume from the cursor of position 0 (the mock's single epoch): everything retained is replayed.
+ const sub = realtime.subscribe('live.stream.*', (event, { seq }) => got.push([seq, event.event_type]), { lastEventId: 'c1.0.MA' });
await waitFor(() => got.length === 2);
await waitFor(() => got.length === 2);
assert.deepEqual(got, [[1, 'live.stream.started'], [3, 'live.stream.updated']], 'internal events never reach an anonymous browser');
assert.deepEqual(got, [[1, 'live.stream.started'], [3, 'live.stream.updated']], 'internal events never reach an anonymous browser');
pub('live.stream.ended');
pub('live.stream.ended');
@@ -151,7 +152,7 @@ run([
@@ -151,7 +152,7 @@ run([
await waitFor(() => got.length === 4);
await waitFor(() => got.length === 4);
assert.deepEqual(got.map((g) => g[0]), [1, 3, 4, 5]);
assert.deepEqual(got.map((g) => g[0]), [1, 3, 4, 5]);
const reconnect = platform.stats.requests.filter((r) => r.url.includes('/realtime/stream')).at(-1);
const reconnect = platform.stats.requests.filter((r) => r.url.includes('/realtime/stream')).at(-1);
- assert.equal(reconnect.headers['last-event-id'], '4');
+ assert.equal(reconnect.headers['last-event-id'], 'c1.0.NA', 'the reconnect hands back the opaque cursor of the last event');
sub.close();
sub.close();
await sub.done;
await sub.done;
}],
}],
@@ -2,10 +2,14 @@ import type { OpenVibeClient, EventEnvelope } from './core';
@@ -2,10 +2,14 @@ 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; }
+/** `seq` is informational; the position is `cursor`. */
+export interface PublishResult { event_id: string; seq: number; cursor: string; duplicate: boolean; }
export interface StoredEvent { seq: number; cursor: string; 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_cursor: string; next_after_seq: number; latest_seq: number; gap?: Gap; }
+/** A position is only ever an opaque cursor: `next_cursor` to read on, `latest_cursor` for the head. */
+export interface EventsPage { events: StoredEvent[]; next_cursor: string; latest_cursor: string; gap?: Gap; }
+/** A stored checkpoint: `cursor` is the opaque cursor as it was stored, null when none is. */
+export interface Checkpoint { consumer: string; topic: string; cursor: string | null; carrier: string | null; updated_at: string | null; }
export interface Subscription {
export interface Subscription {
id: string;
id: string;
consumer: string;
consumer: string;
@@ -23,12 +27,14 @@ export interface EventsClient {
@@ -23,12 +27,14 @@ 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[]; after?: string; afterSeq?: number; limit?: number }): Promise<EventsPage>;
+ /** Without `after`, from the oldest retained event. */
+ pull(opts?: { topic?: string | string[]; after?: string; limit?: number }): Promise<EventsPage>;
/** onPage runs after every item of that page was yielded and handled: save page.next_cursor there. */
/** 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>;
+ iterate(opts?: { topic?: string | string[]; after?: string; 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 }>;
- setCheckpoint(topic: string, cursor: number): Promise<{ consumer: string; topic: string; cursor: number }>;
+ getCheckpoint(topic: string): Promise<Checkpoint>;
+ /** `cursor` is an opaque cursor string (a page's next_cursor); anything else rejects. */
+ setCheckpoint(topic: string, cursor: string, opts?: { carrier?: string }): Promise<Checkpoint>;
subscriptions: {
subscriptions: {
create(input: CreateSubscription): Promise<Subscription>;
create(input: CreateSubscription): Promise<Subscription>;
list(): Promise<Subscription[]>;
list(): Promise<Subscription[]>;
@@ -65,11 +71,12 @@ export interface AppEventsClient {
@@ -65,11 +71,12 @@ 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[]; 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>;
+ pull(opts?: { topic?: string | string[]; platformTopics?: string[]; after?: string; limit?: number }): Promise<EventsPage>;
+ iterate(opts?: { topic?: string | string[]; platformTopics?: string[]; after?: string; 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 }>;
- setCheckpoint(topic: string, cursor: number): Promise<{ consumer: string; topic: string; cursor: number }>;
+ getCheckpoint(topic: string): Promise<Checkpoint>;
+ /** `cursor` is an opaque cursor string (a page's next_cursor); anything else rejects. */
+ setCheckpoint(topic: string, cursor: string, opts?: { carrier?: string }): Promise<Checkpoint>;
subscriptions: {
subscriptions: {
/** topicPattern is project-relative (default '*'); the endpoint must be public https. */
/** topicPattern is project-relative (default '*'); the endpoint must be public https. */
create(input: Partial<CreateSubscription> & { endpoint: string }): Promise<Subscription>;
create(input: Partial<CreateSubscription> & { endpoint: string }): Promise<Subscription>;
@@ -111,7 +118,7 @@ export interface ParseDeliveryOptions extends DeliveryV2Options {
@@ -111,7 +118,7 @@ export interface ParseDeliveryOptions extends DeliveryV2Options {
/** Refuse deliveries without X-OpenVibe-Signature-V2 (v1-only). Default false. A present v2 header must always verify. */
/** Refuse deliveries without X-OpenVibe-Signature-V2 (v1-only). Default false. A present v2 header must always verify. */
requireV2?: boolean;
requireV2?: boolean;
}
}
-export declare function parseDelivery(rawBody: RawBody, headers: Record<string, any> | Headers, secret: string, opts?: ParseDeliveryOptions): { event: EventEnvelope; seq: number; subscriptionId: string | null; attempt: number } | null;
+export declare function parseDelivery(rawBody: RawBody, headers: Record<string, any> | Headers, secret: string, opts?: ParseDeliveryOptions): { event: EventEnvelope; seq: number | null; subscriptionId: string | null; attempt: number } | null;
/** The subset of a better-sqlite3 Database the outbox and inbox use. */
/** The subset of a better-sqlite3 Database the outbox and inbox use. */
export interface SqliteDatabase {
export interface SqliteDatabase {
@@ -1,13 +1,15 @@
@@ -1,13 +1,15 @@
import type { OpenVibeClient, EventEnvelope, FetchLike, TokenContext } from './core';
import type { OpenVibeClient, EventEnvelope, FetchLike, TokenContext } from './core';
export interface RealtimeGap { reason: 'retention' | 'replay_limit' | 'cursor_ahead' | string; from_seq: number; to_seq: number; latest_seq?: number; }
export interface RealtimeGap { reason: 'retention' | 'replay_limit' | 'cursor_ahead' | string; from_seq: number; to_seq: number; latest_seq?: number; }
+/** `cursor` is the position to save; `seq` is informational. */
+export interface RealtimeMeta { seq: number | null; cursor: string | null; }
export interface SubscribeOptions {
export interface SubscribeOptions {
client?: OpenVibeClient;
client?: OpenVibeClient;
/** Events origin or the full …/realtime/stream URL; default from the client's registry, else https://openvibe.events */
/** Events origin or the full …/realtime/stream URL; default from the client's registry, else https://openvibe.events */
url?: string;
url?: string;
baseUrl?: string;
baseUrl?: string;
- /** Resume after this seq (e.g. saved before a reload). */
- lastEventId?: number | string | null;
+ /** Resume after this opaque cursor (the `cursor` a callback was given, saved before a reload). */
+ lastEventId?: string | null;
onGap?: (gap: RealtimeGap) => void;
onGap?: (gap: RealtimeGap) => void;
onOpen?: () => void;
onOpen?: () => void;
onError?: (err: Error) => void;
onError?: (err: Error) => void;
@@ -30,8 +32,8 @@ export interface RealtimeSubscription {
@@ -30,8 +32,8 @@ export interface RealtimeSubscription {
readonly closed: boolean;
readonly closed: boolean;
readonly done: Promise<void>;
readonly done: Promise<void>;
}
}
-export declare function subscribe(topics: string | string[], onEvent: (event: EventEnvelope, meta: { seq: number }) => void, opts?: SubscribeOptions): RealtimeSubscription;
-export declare function createRealtimeClient(client: OpenVibeClient, defaults?: Omit<SubscribeOptions, 'client'>): { subscribe(topics: string | string[], onEvent: (event: EventEnvelope, meta: { seq: number }) => void, opts?: SubscribeOptions): RealtimeSubscription };
+export declare function subscribe(topics: string | string[], onEvent: (event: EventEnvelope, meta: RealtimeMeta) => void, opts?: SubscribeOptions): RealtimeSubscription;
+export declare function createRealtimeClient(client: OpenVibeClient, defaults?: Omit<SubscribeOptions, 'client'>): { subscribe(topics: string | string[], onEvent: (event: EventEnvelope, meta: RealtimeMeta) => void, opts?: SubscribeOptions): RealtimeSubscription };
export interface SSEMessage { event: string; data: string; id: string | undefined; }
export interface SSEMessage { event: string; data: string; id: string | undefined; }
/** WHATWG event-stream parser over a fetch body (or any async iterable of chunks). */
/** WHATWG event-stream parser over a fetch body (or any async iterable of chunks). */
export declare function parseSSE(body: ReadableStream<Uint8Array> | AsyncIterable<Uint8Array | string>, opts?: { onRetry?: (ms: number) => void }): AsyncGenerator<SSEMessage, void, unknown>;
export declare function parseSSE(body: ReadableStream<Uint8Array> | AsyncIterable<Uint8Array | string>, opts?: { onRetry?: (ms: number) => void }): AsyncGenerator<SSEMessage, void, unknown>;