openvibers/openvibe.sdk Commits
SDK 0.25.2: openvibe-sdk/realtime follows Events' opaque cursors (ADR-042): it resumes from the last SSE id exactly as Events sent it and dedupes on the seq in the message data (0.25.1 parsed a cursor id as a number, so resume and the repeat filter stopped advancing); subscription.lastEventId is the raw id. A cursor-id resume test.
Changes
@@@ -8,6 +8,13 @@ release may change an API and says so here.++## 0.25.2 (2026-09-29)++++- `openvibe-sdk/realtime` follows Events' opaque cursors (ADR-042): it resumes from the last SSE id exactly as Events++ sent it (a cursor, or a seq from an older Events) and dedupes on the `seq` in the message data. With cursor ids the++ previous release parsed the id as a number, so its resume position and repeat filter stopped advancing.++ `subscription.lastEventId` is now that raw id (a string).++## 0.25.1 (2026-09-29)- `verifyServiceToken({ jwks })` also takes a JWKS client the service already holds (`jwksClient(url)`, or one a test@@@ -1,4 +1,4 @@--// Generated by scripts/browser-bundle.js from browser.js (openvibe-sdk 0.25.1, MIT). Do not edit.++// Generated by scripts/browser-bundle.js from browser.js (openvibe-sdk 0.25.2, 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, search, chat, ai, jobs, tools, projects, geo). No server-only code: no service tokens, JWT verification, events, identity./* eslint-disable */@@@ -41,7 +41,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.25.1';++const SDK_VERSION = '0.25.2';module.exports = {createClient,@@@ -987,7 +987,10 @@ 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;-- const state = { lastSeq: lastEventId != null && lastEventId !== '' ? Number(lastEventId) : null, closed: false, transport: null, connected: false };++ // 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.++ 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 stop = new AbortController();let es = null;let retryMs = reconnectDelayMs;@@@ -1005,11 +1008,12 @@ 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(id !== undefined && id !== '' ? id : msg && msg.seq);++ const seq = Number(msg && msg.seq != null ? msg.seq : id);if (Number.isFinite(seq)) {if (state.lastSeq != null && seq <= state.lastSeq) return; // already deliveredstate.lastSeq = seq;}++ if (id !== undefined && id !== '') state.lastId = String(id);try { onEvent(msg && msg.event, { seq }); } catch (err) { report(err); }}@@@ -1043,7 +1047,7 @@ function subscribe(topics, onEvent, opts = {}) {stop.signal.addEventListener('abort', abort, { once: true });try {const headers = { Accept: 'text/event-stream', 'Cache-Control': 'no-cache' };-- if (state.lastSeq != null) headers['Last-Event-ID'] = String(state.lastSeq);++ if (state.lastId != null) headers['Last-Event-ID'] = state.lastId;const t = await bearer();if (t) headers.Authorization = `Bearer ${t}`;const init = { headers, signal: ctrl.signal, cache: 'no-store' };@@@ -1082,7 +1086,7 @@ function subscribe(topics, onEvent, opts = {}) {const open = () => {if (state.closed) return;const target = new URL(u.toString());-- if (state.lastSeq != null) target.searchParams.set('last_event_id', String(state.lastSeq));++ if (state.lastId != null) target.searchParams.set('last_event_id', state.lastId);es = new ES(target.toString(), { withCredentials });es.onopen = () => { failures = 0; state.connected = true; if (onOpen) { try { onOpen(); } catch (err) { report(err); } } };es.onmessage = (m) => dispatch('message', m.data, m.lastEventId);@@@ -1112,7 +1116,7 @@ function subscribe(topics, onEvent, opts = {}) {stop.abort();if (es) { try { es.close(); } catch { /* already closed */ } }},-- get lastEventId() { return state.lastSeq; },++ get lastEventId() { return state.lastId; },get transport() { return state.transport; },get connected() { return state.connected; },get closed() { return state.closed; },@@@ -1,12 +1,12 @@{"name": "openvibe-sdk",-- "version": "0.25.1",++ "version": "0.25.2","lockfileVersion": 3,"requires": true,"packages": {"": {"name": "openvibe-sdk",-- "version": "0.25.1",++ "version": "0.25.2","license": "MIT","devDependencies": {"@electric-sql/pglite": "^0.5.8",@@@ -1,6 +1,6 @@{"name": "openvibe-sdk",-- "version": "0.25.1",++ "version": "0.25.2","description": "Supported browser and server clients for the OpenVibe platform: auth (incl. developer apps), registry discovery, identity, user modules, events, realtime, media, community, 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.25.1';++const SDK_VERSION = '0.25.2';module.exports = {createClient,@@@ -35,7 +35,10 @@ 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;-- const state = { lastSeq: lastEventId != null && lastEventId !== '' ? Number(lastEventId) : null, closed: false, transport: null, connected: false };++ // 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.++ 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 stop = new AbortController();let es = null;let retryMs = reconnectDelayMs;@@@ -53,11 +56,12 @@ 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(id !== undefined && id !== '' ? id : msg && msg.seq);++ const seq = Number(msg && msg.seq != null ? msg.seq : id);if (Number.isFinite(seq)) {if (state.lastSeq != null && seq <= state.lastSeq) return; // already deliveredstate.lastSeq = seq;}++ if (id !== undefined && id !== '') state.lastId = String(id);try { onEvent(msg && msg.event, { seq }); } catch (err) { report(err); }}@@@ -91,7 +95,7 @@ function subscribe(topics, onEvent, opts = {}) {stop.signal.addEventListener('abort', abort, { once: true });try {const headers = { Accept: 'text/event-stream', 'Cache-Control': 'no-cache' };-- if (state.lastSeq != null) headers['Last-Event-ID'] = String(state.lastSeq);++ if (state.lastId != null) headers['Last-Event-ID'] = state.lastId;const t = await bearer();if (t) headers.Authorization = `Bearer ${t}`;const init = { headers, signal: ctrl.signal, cache: 'no-store' };@@@ -130,7 +134,7 @@ function subscribe(topics, onEvent, opts = {}) {const open = () => {if (state.closed) return;const target = new URL(u.toString());-- if (state.lastSeq != null) target.searchParams.set('last_event_id', String(state.lastSeq));++ if (state.lastId != null) target.searchParams.set('last_event_id', state.lastId);es = new ES(target.toString(), { withCredentials });es.onopen = () => { failures = 0; state.connected = true; if (onOpen) { try { onOpen(); } catch (err) { report(err); } } };es.onmessage = (m) => dispatch('message', m.data, m.lastEventId);@@@ -160,7 +164,7 @@ function subscribe(topics, onEvent, opts = {}) {stop.abort();if (es) { try { es.close(); } catch { /* already closed */ } }},-- get lastEventId() { return state.lastSeq; },++ get lastEventId() { return state.lastId; },get transport() { return state.transport; },get connected() { return state.connected; },get closed() { return state.closed; },@@@ -40,13 +40,35 @@ run([assert.equal(connections[1].lastEventId, '2');assert.equal(connections[0].auth, 'Bearer user-jwt');assert.equal(new URL(connections[0].url, 'http://x').searchParams.get('topics'), 'live.stream.*,network.notification.*');-- assert.equal(sub.lastEventId, 5);++ assert.equal(sub.lastEventId, '5', 'the raw SSE id, as Events sent it');assert.equal(opens, 2);sub.close();await sub.done;await srv.close();}],++ ['cursor ids (ADR-042): resumes with the cursor Events sent, dedupes on the seq in the data', async () => {++ const cur = (seq) => `c1.0.${Buffer.from([seq]).toString('base64url')}`;++ const cframe = (seq) => `id: ${cur(seq)}\ndata: ${JSON.stringify({ seq, event: { event_type: 'live.stream.started', event_id: `e${seq}` } })}\n\n`;++ const seen = [];++ const srv = await stubServer((req, res) => {++ seen.push(req.headers['last-event-id']);++ res.writeHead(200, { 'Content-Type': 'text/event-stream' });++ res.write('retry: 10\n\n');++ if (seen.length === 1) { res.write(cframe(1)); res.write(cframe(2)); res.end(); }++ else { res.write(cframe(2)); res.write(cframe(3)); }++ });++ const got = [];++ const sub = subscribe('live.*', (event, { seq }) => got.push([seq, event.event_id]), { url: srv.url, token: 't', reconnectDelayMs: 10 });++ await waitFor(() => got.length === 3);++ assert.deepEqual(got, [[1, 'e1'], [2, 'e2'], [3, 'e3']], 'the replayed event is not delivered twice');++ assert.equal(seen[1], cur(2), 'the reconnect resumes from the cursor');++ assert.equal(sub.lastEventId, cur(3));++ sub.close();++ await sub.done;++ await srv.close();++ }],++['starts from a saved lastEventId; stops on a fatal 401', async () => {const srv = await stubServer((req, res) => {res.writeHead(401, { 'Content-Type': 'application/problem+json' });@@@ -23,7 +23,8 @@ export interface SubscribeOptions {}export interface RealtimeSubscription {close(): void;-- readonly lastEventId: number | null;++ /** The last SSE id as Events sent it: an opaque cursor (ADR-042), or a seq from an older Events. Save it to resume. */++ readonly lastEventId: string | null;readonly transport: 'eventsource' | 'fetch';readonly connected: boolean;readonly closed: boolean;
@@ -8,6 +8,13 @@ release may change an API and says so here.
@@ -8,6 +8,13 @@ release may change an API and says so here.
+## 0.25.2 (2026-09-29)
+
+- `openvibe-sdk/realtime` follows Events' opaque cursors (ADR-042): it resumes from the last SSE id exactly as Events
+ sent it (a cursor, or a seq from an older Events) and dedupes on the `seq` in the message data. With cursor ids the
+ previous release parsed the id as a number, so its resume position and repeat filter stopped advancing.
+ `subscription.lastEventId` is now that raw id (a string).
+
## 0.25.1 (2026-09-29)
## 0.25.1 (2026-09-29)
- `verifyServiceToken({ jwks })` also takes a JWKS client the service already holds (`jwksClient(url)`, or one a test
- `verifyServiceToken({ jwks })` also takes a JWKS client the service already holds (`jwksClient(url)`, or one a test
@@ -1,4 +1,4 @@
@@ -1,4 +1,4 @@
-// Generated by scripts/browser-bundle.js from browser.js (openvibe-sdk 0.25.1, MIT). Do not edit.
+// Generated by scripts/browser-bundle.js from browser.js (openvibe-sdk 0.25.2, 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, search, chat, ai, jobs, tools, projects, geo). No server-only code: no service tokens, JWT verification, events, identity.
// (auth, registry, modules, realtime, media, community, search, chat, ai, jobs, tools, projects, geo). No server-only code: no service tokens, JWT verification, events, identity.
/* eslint-disable */
/* eslint-disable */
@@ -41,7 +41,7 @@ const { paginate, offsetPager } = __load(7);
@@ -41,7 +41,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.25.1';
+const SDK_VERSION = '0.25.2';
module.exports = {
module.exports = {
createClient,
createClient,
@@ -987,7 +987,10 @@ function subscribe(topics, onEvent, opts = {}) {
@@ -987,7 +987,10 @@ 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;
- const state = { lastSeq: lastEventId != null && lastEventId !== '' ? Number(lastEventId) : null, closed: false, transport: null, connected: false };
+ // 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.
+ 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 stop = new AbortController();
const stop = new AbortController();
let es = null;
let es = null;
let retryMs = reconnectDelayMs;
let retryMs = reconnectDelayMs;
@@ -1005,11 +1008,12 @@ function subscribe(topics, onEvent, opts = {}) {
@@ -1005,11 +1008,12 @@ 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(id !== undefined && id !== '' ? id : msg && msg.seq);
+ const seq = Number(msg && msg.seq != null ? msg.seq : id);
if (Number.isFinite(seq)) {
if (Number.isFinite(seq)) {
if (state.lastSeq != null && seq <= state.lastSeq) return; // already delivered
if (state.lastSeq != null && seq <= state.lastSeq) return; // already delivered
state.lastSeq = seq;
state.lastSeq = seq;
}
}
+ if (id !== undefined && id !== '') state.lastId = String(id);
try { onEvent(msg && msg.event, { seq }); } catch (err) { report(err); }
try { onEvent(msg && msg.event, { seq }); } catch (err) { report(err); }
}
}
@@ -1043,7 +1047,7 @@ function subscribe(topics, onEvent, opts = {}) {
@@ -1043,7 +1047,7 @@ function subscribe(topics, onEvent, opts = {}) {
stop.signal.addEventListener('abort', abort, { once: true });
stop.signal.addEventListener('abort', abort, { once: true });
try {
try {
const headers = { Accept: 'text/event-stream', 'Cache-Control': 'no-cache' };
const headers = { Accept: 'text/event-stream', 'Cache-Control': 'no-cache' };
- if (state.lastSeq != null) headers['Last-Event-ID'] = String(state.lastSeq);
+ if (state.lastId != null) headers['Last-Event-ID'] = state.lastId;
const t = await bearer();
const t = await bearer();
if (t) headers.Authorization = `Bearer ${t}`;
if (t) headers.Authorization = `Bearer ${t}`;
const init = { headers, signal: ctrl.signal, cache: 'no-store' };
const init = { headers, signal: ctrl.signal, cache: 'no-store' };
@@ -1082,7 +1086,7 @@ function subscribe(topics, onEvent, opts = {}) {
@@ -1082,7 +1086,7 @@ function subscribe(topics, onEvent, opts = {}) {
const open = () => {
const open = () => {
if (state.closed) return;
if (state.closed) return;
const target = new URL(u.toString());
const target = new URL(u.toString());
- if (state.lastSeq != null) target.searchParams.set('last_event_id', String(state.lastSeq));
+ if (state.lastId != null) target.searchParams.set('last_event_id', state.lastId);
es = new ES(target.toString(), { withCredentials });
es = new ES(target.toString(), { withCredentials });
es.onopen = () => { failures = 0; state.connected = true; if (onOpen) { try { onOpen(); } catch (err) { report(err); } } };
es.onopen = () => { failures = 0; state.connected = true; if (onOpen) { try { onOpen(); } catch (err) { report(err); } } };
es.onmessage = (m) => dispatch('message', m.data, m.lastEventId);
es.onmessage = (m) => dispatch('message', m.data, m.lastEventId);
@@ -1112,7 +1116,7 @@ function subscribe(topics, onEvent, opts = {}) {
@@ -1112,7 +1116,7 @@ function subscribe(topics, onEvent, opts = {}) {
stop.abort();
stop.abort();
if (es) { try { es.close(); } catch { /* already closed */ } }
if (es) { try { es.close(); } catch { /* already closed */ } }
},
},
- get lastEventId() { return state.lastSeq; },
+ get lastEventId() { return state.lastId; },
get transport() { return state.transport; },
get transport() { return state.transport; },
get connected() { return state.connected; },
get connected() { return state.connected; },
get closed() { return state.closed; },
get closed() { return state.closed; },
@@ -1,12 +1,12 @@
@@ -1,12 +1,12 @@
{
{
"name": "openvibe-sdk",
"name": "openvibe-sdk",
- "version": "0.25.1",
+ "version": "0.25.2",
"lockfileVersion": 3,
"lockfileVersion": 3,
"requires": true,
"requires": true,
"packages": {
"packages": {
"": {
"": {
"name": "openvibe-sdk",
"name": "openvibe-sdk",
- "version": "0.25.1",
+ "version": "0.25.2",
"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.25.1",
+ "version": "0.25.2",
"description": "Supported browser and server clients for the OpenVibe platform: auth (incl. developer apps), registry discovery, identity, user modules, events, realtime, media, community, 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, 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.25.1';
+const SDK_VERSION = '0.25.2';
module.exports = {
module.exports = {
createClient,
createClient,
@@ -35,7 +35,10 @@ function subscribe(topics, onEvent, opts = {}) {
@@ -35,7 +35,10 @@ 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;
- const state = { lastSeq: lastEventId != null && lastEventId !== '' ? Number(lastEventId) : null, closed: false, transport: null, connected: false };
+ // 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.
+ 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 stop = new AbortController();
const stop = new AbortController();
let es = null;
let es = null;
let retryMs = reconnectDelayMs;
let retryMs = reconnectDelayMs;
@@ -53,11 +56,12 @@ function subscribe(topics, onEvent, opts = {}) {
@@ -53,11 +56,12 @@ 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(id !== undefined && id !== '' ? id : msg && msg.seq);
+ const seq = Number(msg && msg.seq != null ? msg.seq : id);
if (Number.isFinite(seq)) {
if (Number.isFinite(seq)) {
if (state.lastSeq != null && seq <= state.lastSeq) return; // already delivered
if (state.lastSeq != null && seq <= state.lastSeq) return; // already delivered
state.lastSeq = seq;
state.lastSeq = seq;
}
}
+ if (id !== undefined && id !== '') state.lastId = String(id);
try { onEvent(msg && msg.event, { seq }); } catch (err) { report(err); }
try { onEvent(msg && msg.event, { seq }); } catch (err) { report(err); }
}
}
@@ -91,7 +95,7 @@ function subscribe(topics, onEvent, opts = {}) {
@@ -91,7 +95,7 @@ function subscribe(topics, onEvent, opts = {}) {
stop.signal.addEventListener('abort', abort, { once: true });
stop.signal.addEventListener('abort', abort, { once: true });
try {
try {
const headers = { Accept: 'text/event-stream', 'Cache-Control': 'no-cache' };
const headers = { Accept: 'text/event-stream', 'Cache-Control': 'no-cache' };
- if (state.lastSeq != null) headers['Last-Event-ID'] = String(state.lastSeq);
+ if (state.lastId != null) headers['Last-Event-ID'] = state.lastId;
const t = await bearer();
const t = await bearer();
if (t) headers.Authorization = `Bearer ${t}`;
if (t) headers.Authorization = `Bearer ${t}`;
const init = { headers, signal: ctrl.signal, cache: 'no-store' };
const init = { headers, signal: ctrl.signal, cache: 'no-store' };
@@ -130,7 +134,7 @@ function subscribe(topics, onEvent, opts = {}) {
@@ -130,7 +134,7 @@ function subscribe(topics, onEvent, opts = {}) {
const open = () => {
const open = () => {
if (state.closed) return;
if (state.closed) return;
const target = new URL(u.toString());
const target = new URL(u.toString());
- if (state.lastSeq != null) target.searchParams.set('last_event_id', String(state.lastSeq));
+ if (state.lastId != null) target.searchParams.set('last_event_id', state.lastId);
es = new ES(target.toString(), { withCredentials });
es = new ES(target.toString(), { withCredentials });
es.onopen = () => { failures = 0; state.connected = true; if (onOpen) { try { onOpen(); } catch (err) { report(err); } } };
es.onopen = () => { failures = 0; state.connected = true; if (onOpen) { try { onOpen(); } catch (err) { report(err); } } };
es.onmessage = (m) => dispatch('message', m.data, m.lastEventId);
es.onmessage = (m) => dispatch('message', m.data, m.lastEventId);
@@ -160,7 +164,7 @@ function subscribe(topics, onEvent, opts = {}) {
@@ -160,7 +164,7 @@ function subscribe(topics, onEvent, opts = {}) {
stop.abort();
stop.abort();
if (es) { try { es.close(); } catch { /* already closed */ } }
if (es) { try { es.close(); } catch { /* already closed */ } }
},
},
- get lastEventId() { return state.lastSeq; },
+ get lastEventId() { return state.lastId; },
get transport() { return state.transport; },
get transport() { return state.transport; },
get connected() { return state.connected; },
get connected() { return state.connected; },
get closed() { return state.closed; },
get closed() { return state.closed; },
@@ -40,13 +40,35 @@ run([
@@ -40,13 +40,35 @@ run([
assert.equal(connections[1].lastEventId, '2');
assert.equal(connections[1].lastEventId, '2');
assert.equal(connections[0].auth, 'Bearer user-jwt');
assert.equal(connections[0].auth, 'Bearer user-jwt');
assert.equal(new URL(connections[0].url, 'http://x').searchParams.get('topics'), 'live.stream.*,network.notification.*');
assert.equal(new URL(connections[0].url, 'http://x').searchParams.get('topics'), 'live.stream.*,network.notification.*');
- assert.equal(sub.lastEventId, 5);
+ assert.equal(sub.lastEventId, '5', 'the raw SSE id, as Events sent it');
assert.equal(opens, 2);
assert.equal(opens, 2);
sub.close();
sub.close();
await sub.done;
await sub.done;
await srv.close();
await srv.close();
}],
}],
+ ['cursor ids (ADR-042): resumes with the cursor Events sent, dedupes on the seq in the data', async () => {
+ const cur = (seq) => `c1.0.${Buffer.from([seq]).toString('base64url')}`;
+ const cframe = (seq) => `id: ${cur(seq)}\ndata: ${JSON.stringify({ seq, event: { event_type: 'live.stream.started', event_id: `e${seq}` } })}\n\n`;
+ const seen = [];
+ const srv = await stubServer((req, res) => {
+ seen.push(req.headers['last-event-id']);
+ res.writeHead(200, { 'Content-Type': 'text/event-stream' });
+ res.write('retry: 10\n\n');
+ if (seen.length === 1) { res.write(cframe(1)); res.write(cframe(2)); res.end(); }
+ else { res.write(cframe(2)); res.write(cframe(3)); }
+ });
+ const got = [];
+ const sub = subscribe('live.*', (event, { seq }) => got.push([seq, event.event_id]), { url: srv.url, token: 't', reconnectDelayMs: 10 });
+ await waitFor(() => got.length === 3);
+ assert.deepEqual(got, [[1, 'e1'], [2, 'e2'], [3, 'e3']], 'the replayed event is not delivered twice');
+ assert.equal(seen[1], cur(2), 'the reconnect resumes from the cursor');
+ assert.equal(sub.lastEventId, cur(3));
+ sub.close();
+ await sub.done;
+ await srv.close();
+ }],
+
['starts from a saved lastEventId; stops on a fatal 401', async () => {
['starts from a saved lastEventId; stops on a fatal 401', async () => {
const srv = await stubServer((req, res) => {
const srv = await stubServer((req, res) => {
res.writeHead(401, { 'Content-Type': 'application/problem+json' });
res.writeHead(401, { 'Content-Type': 'application/problem+json' });
@@ -23,7 +23,8 @@ export interface SubscribeOptions {
@@ -23,7 +23,8 @@ export interface SubscribeOptions {
}
}
export interface RealtimeSubscription {
export interface RealtimeSubscription {
close(): void;
close(): void;
- readonly lastEventId: number | null;
+ /** The last SSE id as Events sent it: an opaque cursor (ADR-042), or a seq from an older Events. Save it to resume. */
+ readonly lastEventId: string | null;
readonly transport: 'eventsource' | 'fetch';
readonly transport: 'eventsource' | 'fetch';
readonly connected: boolean;
readonly connected: boolean;
readonly closed: boolean;
readonly closed: boolean;