openvibers/openvibe.shared Commits

Shared 3.0.1: the bell and the release watcher resume from Events' cursor and dedupe by event_id (#57)

OP OpenVibers committed 10 Oct 2026 · 392dc371ab · 9 files, +54 −29

Plan T7, ADR-042 decision 7: nothing in Shared reads Events' numeric `seq` as a position any more, so Events can stop accepting one.

- **`notification-live.js` (the bell's realtime feed):**
  - It resumes from the opaque cursor: `last_event_id` is the last SSE id Events sent, handed back as it came. Before, it was a number read from the message data, which a cursor-only Events no longer accepts.
  - A replay after a reconnect is dropped by `event_id` (the last 256 heard), not by position.
  - `state()` reports `lastId` in place of `lastSeq`; `restart()` forgets both the cursor and the ids.
- **`release-watch.js`:** it already resumed from the cursor (2.21.2); now it drops a replay by `event_id` alone. Its 5 KB brotli budget still holds.
- **Older pins:** a bell on an older pin keeps working after Events changes. Its reconnect starts at the head, and the count poll (every 2 min while open, 15 s otherwise) fills in.

3.0.1.

Tests: the fake streams send cursor ids (the format Events emits; the clients never parse them).
- `notification-live.test.js`:
  - a replay of the same event is ignored;
  - `lastId` is the cursor;
  - every reconnect (failure, `online`, shown again) resumes from it.
- `notification-live-chrome.test.js` (real Chrome against a fake Network and Events): the fake Events emits and resumes from cursors, and replays only for a cursor.
- `release-offline-resume.test.js`: the resume URL carries the cursor.

`ov test`: 17 reached, all green.

Changes

ddiff --git a/CHANGELOG.md b/CHANGELOG.md
iindex 1de3fce..ac4eca5 100644
---- a/CHANGELOG.md
++++ b/CHANGELOG.md
@@@ -4,6 +4,16 @@ All notable changes to `openvibe-shared`. Versions follow [semver](https://semve
  breaking change to any exported module, browser global or served file name is a new major.
  A release is the git tag `vX.Y.Z`; consumers pin the tag's tarball (see README).
  
++## 3.0.1 — 2026-10-10
++
++- `notification-live.js` (the bell's realtime feed) resumes from Events' opaque cursor: `last_event_id` is the last
++  SSE id Events sent, handed back as it came, never a number it read from the message. A replay after a reconnect is
++  dropped by its `event_id` (the last 256 heard), not by position. `state()` reports `lastId` in place of `lastSeq`.
++- `release-watch.js` drops a replayed release message by its `event_id` alone (it already resumed from the cursor).
++- With these, nothing in Shared reads Events' numeric seq as a position, so Events can stop accepting one (plan T7,
++  ADR-042 decision 7). A bell on an older pin keeps working after that; a reconnect then starts at the head, and the
++  count poll fills in.
++
  ## 3.0.0 — 2026-10-10
  
  - Removed the SQLite configuration store. `createConfigStore` now requires an openvibe-sdk/db handle and returns a promise of the ready PostgreSQL store. `fromRows`, `adminRoutes`, `configSchema`, and other non-SQLite exports remain.
ddiff --git a/STATUS.json b/STATUS.json
iindex 7db7f05..7fb4383 100644
---- a/STATUS.json
++++ b/STATUS.json
@@@ -4,7 +4,7 @@
    "code": true,
    "wave": "W2",
    "package": "openvibe-shared",
--  "version": "3.0.0",
++  "version": "3.0.1",
    "plan": "OpenVibe End-to-End Realignment & Implementation Plan, revision 3 (20 Sep 2026)",
    "deployed": false,
    "deployment": "none of its own: a package released as git tags; OpenVibe.Network serves /shared/* from its pinned tag, and a site can serve its own pinned copy (openvibe-shared/serve)",
ddiff --git a/notification-live.js b/notification-live.js
iindex fdb708a..1e95d0e 100644
---- a/notification-live.js
++++ b/notification-live.js
@@@ -3,10 +3,11 @@
   * ADR-005 amendment 2). notification-ui.js loads it on demand when a site turns it on (navbar notificationsRealtime).
   * For every (re)connect it asks Network for a realtime ticket (POST <api>/api/v1/realtime/ticket: Bearer token, or the
   * cookie on openvibe.network itself) and opens Events' stream with it, without credentials:
-- *   <stream_url>?topics=network.notification.*&ticket=<ticket>[&last_event_id=<seq>]
++ *   <stream_url>?topics=network.notification.*&ticket=<ticket>[&last_event_id=<cursor>]
   * Events streams only the person's own network.notification.created events (subject visibility); one whose subject is
   * not the ticket's is ignored all the same. Each calls onNotification(payload); `event: gap` calls onGap.
-- * A ticket opens one stream: errors close it and reconnect with a fresh ticket from the cursor, backing off 2 s to 15 min
++ * The cursor is the last SSE id Events sent (opaque, handed back as it came); an event already heard (by event_id) is
++ * skipped. A ticket opens one stream: errors close it and reconnect with a fresh ticket from the cursor, backing off 2 s to 15 min
   * with jitter. Stops (the caller's polling is all there is) after 10 failures in a row, on a Content-Security-Policy
   * refusal, when signed out (401/403) or when Network has realtime off (404/503). Hidden 5 min: closed; shown: resumed.
   * OVNotificationLive.create({ ticketUrl, token, credentials, onNotification, onGap, onState }) → { start, stop, restart, state }
@@@ -17,10 +18,12 @@
      const TYPE_RE = /^network\.notification\./;
      const SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
      const MAX_FAILURES = 10;   // 2 s, 4 s, … 512 s: about 9 to 17 minutes of retries before polling only
++    const SEEN_MAX = 256;      // event ids remembered for dedupe
  
      function create(o) {
--        const rt = { state: 'off', failures: 0, lastSeq: null, subject: null, origin: null, events: 0, ignored: 0, gaps: 0, tickets: 0 };
++        const rt = { state: 'off', failures: 0, lastId: null, subject: null, origin: null, events: 0, ignored: 0, gaps: 0, tickets: 0 };
          let es = null; let stopped = true; let gen = 0;
++        const seen = new Set();
          const t = {};   // timers: retry, hide
          const later = (k, f, ms) => { root.clearTimeout(t[k]); t[k] = root.setTimeout(() => { t[k] = null; f(); }, ms); };
          const call = (f, ...a) => { try { if (typeof f === 'function') f(...a); } catch { /* the page's handler must not break the feed */ } };
@@@ -54,7 +57,7 @@
                  if (!/^https?:$/.test(u.protocol)) throw new Error('url');
                  u.searchParams.set('topics', tk.topics.join(','));
                  u.searchParams.set('ticket', tk.ticket);
--                if (rt.lastSeq != null) u.searchParams.set('last_event_id', String(rt.lastSeq));
++                if (rt.lastId != null) u.searchParams.set('last_event_id', rt.lastId);
              } catch { fail(); return; }
              rt.tickets++; rt.origin = u.origin;
              rt.subject = tk.subject;
@@@ -73,9 +76,14 @@
          }
          function heard(e) {
              let m; try { m = JSON.parse(e.data); } catch { return; }
--            const seq = m && typeof m.seq === 'number' ? m.seq : Number(e.lastEventId);
--            if (Number.isFinite(seq)) { if (rt.lastSeq != null && seq <= rt.lastSeq) return; rt.lastSeq = seq; }
              const ev = m && m.event;
++            const id = ev && typeof ev.event_id === 'string' ? ev.event_id : null;
++            if (id) {
++                if (seen.has(id)) return;   // already heard (a replay after a reconnect)
++                seen.add(id);
++                if (seen.size > SEEN_MAX) seen.delete(seen.values().next().value);
++            }
++            if (e.lastEventId) rt.lastId = String(e.lastEventId);
              // Events already sends only this person's own; anything else is never shown.
              if (!ev || !TYPE_RE.test(String(ev.event_type)) || !ev.subject || ev.subject.type !== 'user' || ev.subject.id !== rt.subject) { rt.ignored++; return; }
              rt.events++;
@@@ -107,7 +115,7 @@
              start() { if (!stopped) return api; stopped = false; set('off'); open(); return api; },
              stop() { stopped = true; gen++; close(); Object.keys(t).forEach((k) => { root.clearTimeout(t[k]); t[k] = null; }); set('off'); return api; },
              /** Another account (or a new token): forget the cursor and the subject, start again. */
--            restart() { api.stop(); rt.lastSeq = null; rt.subject = null; rt.failures = 0; return api.start(); },
++            restart() { api.stop(); rt.lastId = null; seen.clear(); rt.subject = null; rt.failures = 0; return api.start(); },
              state: () => ({ ...rt }),
          };
          return api;
ddiff --git a/package-lock.json b/package-lock.json
iindex c699835..63b868c 100644
---- a/package-lock.json
++++ b/package-lock.json
@@@ -1,12 +1,12 @@
  {
    "name": "openvibe-shared",
--  "version": "3.0.0",
++  "version": "3.0.1",
    "lockfileVersion": 3,
    "requires": true,
    "packages": {
      "": {
        "name": "openvibe-shared",
--      "version": "3.0.0",
++      "version": "3.0.1",
        "license": "MIT",
        "dependencies": {
          "jsonwebtoken": "^9.0.0"
ddiff --git a/package.json b/package.json
iindex 8ba2778..3899ab8 100644
---- a/package.json
++++ b/package.json
@@@ -1,6 +1,6 @@
  {
    "name": "openvibe-shared",
--  "version": "3.0.0",
++  "version": "3.0.1",
    "description": "Shared navbar, footer, themes, icons, legal pages, SEO, auth and analytics for every OpenVibe site",
    "main": "index.js",
    "bin": {
ddiff --git a/release-watch.js b/release-watch.js
iindex 39c01cf..4f51030 100644
---- a/release-watch.js
++++ b/release-watch.js
@@@ -181,7 +181,7 @@
      // back off 30 s to 15 min, after 6 failures only polling is left; online reconnects at once.
      const TOPIC = 'host.release.published';
      const HEX = /^[0-9a-f]{7,40}$/;
--    const rt = { state: 'off', service: null, url: null, events: 0, ignored: 0, checks: 0, failures: 0, lastSeq: null };
++    const rt = { state: 'off', service: null, url: null, events: 0, ignored: 0, checks: 0, failures: 0, lastId: null };
      let es = null; let queued = null; let again = false; let seen = [];
      const t = {};   // timers: run (the jittered check), cool (30 s after it), retry, hide
      const later = (k, f, ms) => { root.clearTimeout(t[k]); t[k] = root.setTimeout(() => { t[k] = null; f(); }, ms); };
@@@ -229,7 +229,6 @@
          let m; try { m = JSON.parse(e.data); } catch { return; }
          const ev = m && m.event; const p = ev && ev.payload;
          if (!p || ev.event_type !== TOPIC) return;
--        if (typeof m.seq === 'number') { if (rt.lastSeq != null && m.seq <= rt.lastSeq) return; rt.lastSeq = m.seq; }
          rt.lastId = e.lastEventId || rt.lastId;
          if (p.service !== rt.service || typeof p.release !== 'string') return;
          // Ids repeat across origins (Sites' placeholders): a named origin must be ours.
ddiff --git a/test/notification-live-chrome.test.js b/test/notification-live-chrome.test.js
iindex c5a4347..58430e7 100644
---- a/test/notification-live-chrome.test.js
++++ b/test/notification-live-chrome.test.js
@@@ -5,6 +5,8 @@
  // fresh ticket and nothing is lost or repeated; a gap is reported; an event for another person yields nothing.
  // Events' own refusal of a guessed user:<other> topic and of other people's events is OpenVibe.Events
  // test/realtime-tickets.test.js. Skipped (exit 0) when Chrome is not installed or OV_SKIP_BROWSER=1.
++// Events' opaque cursor for a position (the format Events emits; the client never parses it).
++const cur = (seq) => `c1.0.${Buffer.from(String(seq), 'utf8').toString('base64url')}`;
  const assert = require('assert');
  const fs = require('fs');
  const os = require('os');
@@@ -41,7 +43,7 @@ const PAGE = `<!doctype html><html><head><meta charset="utf-8"><title>bell</titl
  
  function sendEvent(res, e) {
      const event = { event_id: `evt_${e.seq}`, event_type: 'network.notification.created', visibility: 'subject', subject: { type: 'user', id: e.subject }, payload: { notification_id: e.id, unread_count: e.seq } };
--    res.write(`id: ${e.seq}\ndata: ${JSON.stringify({ seq: e.seq, event })}\n\n`);
++    res.write(`id: ${cur(e.seq)}\ndata: ${JSON.stringify({ seq: e.seq, event })}\n\n`);
  }
  /** A new notification: stored, then streamed to its person (or, `leak`, to everyone: a misbehaving server). */
  function publish(subject, id, { leak = false } = {}) {
@@@ -75,7 +77,8 @@ const server = http.createServer((req, res) => {
          res.writeHead(200, { 'content-type': 'text/event-stream', 'cache-control': 'no-cache' });
          res.write('retry: 3000\n: connected user\n\n');
          if (last != null) {
--            let cursor = Number(last);
++            let cursor = Number(Buffer.from(String(last).split('.')[2] || '', 'base64url').toString('utf8'));
++            if (cur(cursor) !== last) cursor = log.length;   // not a cursor: no replay, like Events
              if (cursor < oldest - 1) { res.write(`event: gap\ndata: ${JSON.stringify({ reason: 'retention', from_seq: cursor + 1, to_seq: oldest - 1 })}\n\n`); cursor = oldest - 1; }
              for (const e of log) if (e.seq > cursor && e.seq >= oldest && e.subject === subject) sendEvent(res, e);
          }
@@@ -117,7 +120,7 @@ const server = http.createServer((req, res) => {
          let v = await until((x) => x.heard.length === 2, 'a1 and a2');
          assert.deepStrictEqual(v.heard, ['a1', 'a2']);
          assert.strictEqual(v.state.ignored, 1, 'another person\'s event yields nothing');
--        assert.strictEqual(v.state.lastSeq, 3);
++        assert.strictEqual(v.state.lastId, cur(3));
  
          // The connection drops (Events restarts); while it is down two more notifications are stored.
          dropAll();
@@@ -127,7 +130,7 @@ const server = http.createServer((req, res) => {
          v = await until((x) => x.heard.length === 4 && x.state.state === 'open', 'the resume');
          assert.deepStrictEqual(v.heard, ['a1', 'a2', 'a3', 'a4'], 'the missed ones, in order, nothing twice');
          const resumed = streamRequests[streamRequests.length - 1];
--        assert.deepStrictEqual([resumed.ticket, resumed.last], ['tk-2', '3'], 'a fresh ticket and the cursor');
++        assert.deepStrictEqual([resumed.ticket, resumed.last], ['tk-2', cur(3)], 'a fresh ticket and the cursor');
          assert.strictEqual(new Set(streamRequests.filter((r) => issued.has(r.ticket) === false).map((r) => r.ticket)).size, streamRequests.length, 'a ticket is never used twice');
          publish(ALICE, 'a5');
          await until((x) => x.heard.length === 5, 'live again');
ddiff --git a/test/notification-live.test.js b/test/notification-live.test.js
iindex 5023df3..560af70 100644
---- a/test/notification-live.test.js
++++ b/test/notification-live.test.js
@@@ -4,13 +4,15 @@
  //   - it asks Network for a realtime ticket (Bearer, no cookies across sites) and opens Events' stream with it,
  //     topics=network.notification.*, never with credentials; the ticket is never kept;
  //   - the person's own network.notification.created calls onNotification; another person's (a guessed topic),
--//     other types and replayed seqs do nothing;
++//     other types and a replayed event (its event_id again) do nothing;
  //   - a disconnect reconnects with a fresh ticket and last_event_id; a gap is reported; hidden 5 min closes and
  //     showing resumes; 401/403, 503 and a CSP refusal stop it; ten failures in a row stop it, `online` revives it;
  //   - notification-ui with realtime on loads it from beside itself, re-reads the count on an event (once per burst,
  //     and again when a poll was already in flight), re-reads the lists on a gap, and polls every 2 min while the
  //     stream is open, every 15 s otherwise (CSP-blocked included); with realtime off nothing of it runs.
  const assert = require('assert');
++// Events' opaque cursor for a position (the format Events emits; the client never parses it).
++const cur = (seq) => `c1.0.${Buffer.from(String(seq), 'utf8').toString('base64url')}`;
  const fs = require('fs');
  const path = require('path');
  const vm = require('vm');
@@@ -37,7 +39,7 @@ function page({ url = 'https://openvibe.live/', eventSource = true, random = 0.5
          addEventListener(type, f) { (this.listeners[type] = this.listeners[type] || []).push(f); }
          close() { this.closed = true; }
          opened() { if (this.onopen) this.onopen({}); }
--        send(seq, event) { if (this.onmessage) this.onmessage({ data: JSON.stringify({ seq, event }), lastEventId: String(seq) }); }
++        send(seq, event) { if (this.onmessage) this.onmessage({ data: JSON.stringify({ seq, event }), lastEventId: cur(seq) }); }
          emit(type, data) { (this.listeners[type] || []).forEach((f) => f({ data: JSON.stringify(data) })); }
          fail() { if (this.onerror) this.onerror({}); }
      }
@@@ -148,13 +150,14 @@ const created = (subject, over = {}) => ({
          assert.strictEqual(live.state().state, 'open');
  
          // The person's own notification is heard; another person's (a guessed topic), other types and replays are not.
--        s1.send(10, created(ALICE));
++        const first = created(ALICE);
++        s1.send(10, first);
          s1.send(11, created(BOB));
          s1.send(12, { ...created(ALICE), event_type: 'live.stream.started' });
          s1.send(13, { ...created(ALICE), subject: { type: 'stream', id: ALICE } });
--        s1.send(10, created(ALICE));
++        s1.send(10, first);
          assert.deepStrictEqual(heard.map((h) => h[1]), [ALICE]);
--        assert.deepStrictEqual([live.state().events, live.state().ignored, live.state().lastSeq], [1, 3, 13]);
++        assert.deepStrictEqual([live.state().events, live.state().ignored, live.state().lastId], [1, 3, cur(13)], 'the cursor is the last SSE id, as Events sent it');
          assert.ok(!JSON.stringify(live.state()).includes('T1'), 'the ticket is not kept');
  
          // A gap is reported.
@@@ -171,7 +174,7 @@ const created = (subject, over = {}) => ({
          assert.strictEqual(p.tickets().length, 2, 'a new ticket for the reconnect');
          const s2 = p.es;
          assert.notStrictEqual(s2, s1);
--        assert.deepStrictEqual([new URL(s2.url).searchParams.get('ticket'), new URL(s2.url).searchParams.get('last_event_id')], ['hdr.T2.sig', '13']);
++        assert.deepStrictEqual([new URL(s2.url).searchParams.get('ticket'), new URL(s2.url).searchParams.get('last_event_id')], ['hdr.T2.sig', cur(13)]);
          s2.opened();
          s2.send(14, created(ALICE));
          assert.deepStrictEqual(heard.map((h) => h[0]).length, 2);
@@@ -188,7 +191,7 @@ const created = (subject, over = {}) => ({
          const n = p.out.es.length;
          p.dispatch('online'); await p.settle();
          assert.strictEqual(p.out.es.length, n + 1, 'back online: a new stream');
--        assert.strictEqual(new URL(p.es.url).searchParams.get('last_event_id'), '14', 'from the cursor');
++        assert.strictEqual(new URL(p.es.url).searchParams.get('last_event_id'), cur(14), 'from the cursor');
          p.es.opened();
  
          // Hidden for 5 minutes: closed; shown: resumed from the cursor with a fresh ticket.
@@@ -200,7 +203,7 @@ const created = (subject, over = {}) => ({
          const t3 = p.tickets().length;
          p.setHidden(false); await p.settle();
          assert.strictEqual(p.tickets().length, t3 + 1);
--        assert.strictEqual(new URL(p.es.url).searchParams.get('last_event_id'), '14');
++        assert.strictEqual(new URL(p.es.url).searchParams.get('last_event_id'), cur(14));
          p.es.opened();
  
          // A Content-Security-Policy refusal of Events: closed, and nothing reopens it.
ddiff --git a/test/release-offline-resume.test.js b/test/release-offline-resume.test.js
iindex 8ae538c..ea00782 100644
---- a/test/release-offline-resume.test.js
++++ b/test/release-offline-resume.test.js
@@@ -11,6 +11,8 @@
  //   - a tab away past the mixed-version window reloads only when safe (never under a focused field);
  //   - a hidden tab keeps its stream closed until it is shown again, then resumes from the cursor.
  // Lines `[metric] name=value` are read by OpenVibe.Host's release-acceptance runner.
++// Events' opaque cursor for a position (the format Events emits; the client never parses it).
++const cur = (seq) => `c1.0.${Buffer.from(String(seq), 'utf8').toString('base64url')}`;
  const assert = require('assert');
  const { openPage, manifestFor } = require('../release-compat');
  
@@@ -36,7 +38,7 @@ function fakeEventSource() {
          addEventListener(type, f) { (this.listeners[type] = this.listeners[type] || []).push(f); }
          close() { this.closed = true; }
          opened() { if (this.onopen) this.onopen({}); }
--        send(seq, event) { if (this.onmessage) this.onmessage({ data: JSON.stringify({ seq, event }), lastEventId: String(seq) }); }
++        send(seq, event) { if (this.onmessage) this.onmessage({ data: JSON.stringify({ seq, event }), lastEventId: cur(seq) }); }
          fail() { if (this.onerror) this.onerror({}); }
      }
      return { FakeEventSource, all };
@@@ -90,7 +92,7 @@ const metrics = { 'offline-resume.reloads-while-offline': 0, 'offline-resume.pro
          const s0 = p.es.all[0];
          s0.opened();
          s0.send(7, released('other', B, 1));
--        assert.strictEqual(p.rt().lastSeq, 7);
++        assert.strictEqual(p.rt().lastId, cur(7));
          // The network drops.
          p.state.offline = true;
          s0.fail();
@@@ -123,7 +125,7 @@ const metrics = { 'offline-resume.reloads-while-offline': 0, 'offline-resume.pro
          // Not reconnected at once: the stream waits out its backoff, and that wait is the delay.
          if (!reconnected) metrics['offline-resume.stream-reconnect-delay-ms'] = Math.max(metrics['offline-resume.stream-reconnect-delay-ms'], retry);
          assert.ok(reconnected, `back online after ${failures} failures: the release stream reconnects at once`);
--        assert.match(p.es.all[p.es.all.length - 1].url, /last_event_id=7$/, 'from its cursor');
++        assert.ok(p.es.all[p.es.all.length - 1].url.endsWith(`last_event_id=${cur(7)}`), 'from its cursor');
          assert.strictEqual(p.rt().failures, 0);
          p.settleStyles(true); await p.settle();
          metrics['offline-resume.manifest-reads-on-resume'] = Math.max(metrics['offline-resume.manifest-reads-on-resume'], p.manifestReads() - reads);
@@@ -188,7 +190,7 @@ const metrics = { 'offline-resume.reloads-while-offline': 0, 'offline-resume.pro
          assert.strictEqual(p.es.all.length, 1, 'online while hidden: no stream yet');
          p.setHidden(false); await p.settle();
          assert.strictEqual(p.es.all.length, 2, 'shown: one stream');
--        assert.match(p.es.all[1].url, /last_event_id=11$/);
++        assert.ok(p.es.all[1].url.endsWith(`last_event_id=${cur(11)}`));
          p.stop();
      }