openvibers/openvibe.sdk Commits

SDK 0.37.0: openvibe-sdk/account-data startSubscriptions() — the two account-event subscriptions created at boot when missing, the half every adopter had copied (#25)

OP OpenVibers committed 9 Oct 2026 · 8ffde14a98 · 11 files, +125 −12

Changes

ddiff --git a/CHANGELOG.md b/CHANGELOG.md
iindex a77c9e3..c1774f3 100644
---- a/CHANGELOG.md
++++ b/CHANGELOG.md
@@@ -3,6 +3,14 @@
  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.
  
++## 0.37.0 (2026-10-08)
++
++- **`openvibe-sdk/account-data`: `startSubscriptions()`**, the boot-time half every adopter had copied. It creates any
++  missing OpenVibe.Events subscription for the two account topics (or `topics`) to the service's loopback endpoint with
++  its own token (audience openvibe.events, scope events.subscription.manage), retried with backoff in the background,
++  and leaves an existing one alone, so a restart creates nothing. It returns null when the Events URL, the delivery
++  secret or the client secret is unset; `stop()` ends the retries and `done` says whether every topic was subscribed.
++
  ## 0.36.0 (2026-10-08)
  
  - **`openvibe-sdk/account-data` (new, node): a service's share of account export and deletion (ADR-033).** Six services
ddiff --git a/README.md b/README.md
iindex d1ea3d4..c4ba6a4 100644
---- a/README.md
++++ b/README.md
@@@ -2,7 +2,7 @@
  
  > Supported browser and server clients for the OpenVibe platform.
  
--**Status:** alpha, v0.36.0 (roadmap Wave 2; developer apps from Wave 20; Media objects v2 from Wave 4; the Tools platform API; Search, Chat and AI clients from WS-F task 4; per-actor limits from WS-R task 4). Every release is tagged (see [CHANGELOG.md](CHANGELOG.md)). Tested against local stub servers and the built-in mock platform only, never against the live platform. Services pin release tags; each repository's `STATUS.json` names the one it uses.
++**Status:** alpha, v0.37.0 (roadmap Wave 2; developer apps from Wave 20; Media objects v2 from Wave 4; the Tools platform API; Search, Chat and AI clients from WS-F task 4; per-actor limits from WS-R task 4). Every release is tagged (see [CHANGELOG.md](CHANGELOG.md)). Tested against local stub servers and the built-in mock platform only, never against the live platform. Services pin release tags; each repository's `STATUS.json` names the one it uses.
  **Plan:** OpenVibe End-to-End Realignment & Implementation Plan, revision 3 (20 Sep 2026), §3.2; roadmap §30.  
  **License:** MIT ([LICENSE](LICENSE)). This package is a library that apps outside the network embed, so it uses MIT. The OpenVibe services themselves stay AGPL-3.0.
  
@@@ -174,7 +174,7 @@ Until OpenVibe.Tools serves the gateway facade (Tools S6), point the jobs client
  | `openvibe-sdk/bot` | both | `createBotClient(client, {actingSubject?})`: OpenVibe.Bot robots. `robots.list/get/create/update/delete/pairingCode`, `robots.operators.list/add/remove`, `robots.devices`, `robots.streaming.get/set`, `robots.audit`/`iterateAudit`, `robots.estop`/`clearEstop`, `robots.command` (`bot.command@1`), `devices.rotate/revoke`, public `profiles.list/get` and `kits.list/get`, `pair` (the agent's one-time code, never retried), `as(subject)`; a service acts for a person with `X-OV-Subject` (capabilities `bot.robot.read`/`manage`/`control`, `bot.device.connect`) |
  | `openvibe-sdk/search` | both | `createSearchClient(client, {actingSubject?})`: `query(text, {owner, type, lang, filter, facets, limit, cursor})`, `iterate(text, {…, max})`, `suggest(text, {owner, type, limit})`, `document(owner, type, id)` (null when missing or not visible) |
  | `openvibe-sdk/limits` | node | `createActorLimiter({limits: {minute, hour, day}, actor?, maxActors?, onLimited?})` → `limits(name, {minute?, hour?, day?})` Express middleware: per-actor fixed windows at a capability boundary (service principal, else the signed-in person, else `ip:`), 429 problem+json `rate_limited` with Retry-After before the handler; `defaultActor(req)`, `.stats()` |
--| `openvibe-sdk/account-data` | node | `createAccountData({db, service, tables: [{table, subject, value?, file?, columns?, erase?}], extraExport?, extraErase?})` → `.apply(event, {send})`, `.consumer({secrets, send, onEvent?})` (a POST /internal/events handler), `.exportPart(subject)`, `.erase(subjects)`, `.ensureSchema()`; `createNetworkSender({networkInternalUrl, clientId, clientSecret})`: a service's share of account export and deletion (ADR-033), idempotent per export/deletion id |
++| `openvibe-sdk/account-data` | node | `createAccountData({db, service, tables: [{table, subject, value?, file?, columns?, erase?}], extraExport?, extraErase?})` → `.apply(event, {send})`, `.consumer({secrets, send, onEvent?})` (a POST /internal/events handler), `.exportPart(subject)`, `.erase(subjects)`, `.ensureSchema()`; `createNetworkSender({networkInternalUrl, clientId, clientSecret})`; `startSubscriptions({eventsUrl, endpoint, secret, networkInternalUrl, clientId, clientSecret})` (the two Events subscriptions, created at boot when missing): a service's share of account export and deletion (ADR-033), idempotent per export/deletion id |
  | `openvibe-sdk/chat` | both | `createChatClient(client)`: `global.send/history/search`, `rooms.*` (list, create, get, update, messages, send, deleteMessage, join, leave, read, members, setMember), `dms.*` (list, create, get, messages, send, deleteMessage, read, unread, blocks, block, unblock), `exportMine()` |
  | `openvibe-sdk/ai` | both | `createAiClient(client)`: `runs.create(workflow, input, {wait, idempotencyKey, …})`, `runs.get/list/cancel/retry/citations/addCitations/waitFor`, `chat/generate/summarize/classify/extract/enrich/embed` |
  | `openvibe-sdk/jobs` | both | `createJobsClient(client, {baseUrl?})`: `submit` (Idempotency-Key), `get`, `cancel`, `retry`, `reference`, `unreference`, `events` (SSE, reattaches), `wait`, `file` (raw Response); `isTerminal()`. Default origin: the Tools gateway's jobs facade |
ddiff --git a/STATUS.json b/STATUS.json
iindex e11bfd2..8cd2a48 100644
---- a/STATUS.json
++++ b/STATUS.json
@@@ -3,7 +3,7 @@
    "stage": "alpha",
    "code": true,
    "wave": "W2",
--  "version": "0.36.0",
++  "version": "0.37.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",
ddiff --git a/browser/openvibe-sdk.mjs b/browser/openvibe-sdk.mjs
iindex f0616a6..6547da7 100644
---- a/browser/openvibe-sdk.mjs
++++ b/browser/openvibe-sdk.mjs
@@@ -1,4 +1,4 @@
--// Generated by scripts/browser-bundle.js from browser.js (openvibe-sdk 0.36.0, MIT). Do not edit.
++// Generated by scripts/browser-bundle.js from browser.js (openvibe-sdk 0.37.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.36.0';
++const SDK_VERSION = '0.37.0';
  
  module.exports = {
      createClient,
ddiff --git a/esm/account-data.mjs b/esm/account-data.mjs
iindex 194a80d..88313a3 100644
---- a/esm/account-data.mjs
++++ b/esm/account-data.mjs
@@@ -2,6 +2,6 @@
  import cjs from '../src/account-data.js';
  
  export const {
--    createAccountData, createNetworkSender, ACCOUNT_DATA_SCHEMA, TOPICS,
++    createAccountData, createNetworkSender, startSubscriptions, ACCOUNT_DATA_SCHEMA, TOPICS,
  } = cjs;
  export default cjs;
ddiff --git a/package-lock.json b/package-lock.json
iindex 6292497..f6130ee 100644
---- a/package-lock.json
++++ b/package-lock.json
@@@ -1,12 +1,12 @@
  {
    "name": "openvibe-sdk",
--  "version": "0.36.0",
++  "version": "0.37.0",
    "lockfileVersion": 3,
    "requires": true,
    "packages": {
      "": {
        "name": "openvibe-sdk",
--      "version": "0.36.0",
++      "version": "0.37.0",
        "license": "MIT",
        "devDependencies": {
          "@electric-sql/pglite": "^0.5.8",
ddiff --git a/package.json b/package.json
iindex d2a4af7..b93ec1a 100644
---- a/package.json
++++ b/package.json
@@@ -1,6 +1,6 @@
  {
    "name": "openvibe-sdk",
--  "version": "0.36.0",
++  "version": "0.37.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",
ddiff --git a/src/account-data.js b/src/account-data.js
iindex f082e92..151b010 100644
---- a/src/account-data.js
++++ b/src/account-data.js
@@@ -17,6 +17,8 @@
   *   await accountData.ensureSchema();                    // or ACCOUNT_DATA_SCHEMA in a migration
   *   const send = createNetworkSender({ networkInternalUrl, clientId, clientSecret });
   *   app.use('/internal/events', accountData.consumer({ secrets: [process.env.FOOD_EVENTS_SECRET], send }));
++ *   startSubscriptions({ eventsUrl, endpoint: 'http://127.0.0.1:4970/internal/events', secret, networkInternalUrl,
++ *       clientId, clientSecret });                      // the two subscriptions, created at boot when missing
   *   // or, inside a consumer the service already has:  const outcome = await accountData.apply(event, { send });
   *
   * A table entry:
@@@ -275,4 +277,64 @@ function createNetworkSender({ networkInternalUrl, clientId, clientSecret, fetch
      };
  }
  
--module.exports = { createAccountData, createNetworkSender, ACCOUNT_DATA_SCHEMA, TOPICS };
++/**
++ * The boot-time half: create any missing Events subscription for `topics` (default the two account topics) to this
++ * service's loopback `endpoint`, idempotently, retried with backoff in the background. Off (null) when the Events URL,
++ * the delivery secret or the client secret is unset. The token is this service's own, audience openvibe.events, scope
++ * events.subscription.manage. A subscription already there for the same topic and endpoint is left alone, so a
++ * restart creates nothing; Events answers 409 for a racing duplicate, which counts as done.
++ *
++ *   const subs = startSubscriptions({ eventsUrl, endpoint: `http://127.0.0.1:${port}/internal/events`, secret,
++ *       networkInternalUrl, clientId, clientSecret });
++ *   // graceful stop: subs && subs.stop()
++ *
++ * `done` resolves true once every topic is subscribed, false when the retries ran out or stop() came first.
++ */
++function startSubscriptions({
++    eventsUrl, endpoint, secret, topics = TOPICS, networkInternalUrl, clientId, clientSecret,
++    fetch: fetchImpl = globalThis.fetch, log = console, delays = [0, 10_000, 60_000, 5 * 60_000, 15 * 60_000],
++} = {}) {
++    if (!eventsUrl || !endpoint || !secret || !clientId || !clientSecret || !networkInternalUrl) return null;
++    const base = String(eventsUrl).replace(/\/+$/, '');
++    const tokens = createServiceTokenClient({ tokenUrl: `${String(networkInternalUrl).replace(/\/+$/, '')}/oauth/token`, clientId, clientSecret, audience: 'openvibe.events', scope: 'events.subscription.manage', fetch: fetchImpl });
++    const call = async (method, path, body) => {
++        const res = await fetchImpl(`${base}${path}`, {
++            method,
++            headers: { Accept: 'application/json', ...(body ? { 'Content-Type': 'application/json' } : {}), ...(await tokens.authHeaders()) },
++            body: body ? JSON.stringify(body) : undefined,
++            signal: AbortSignal.timeout(15000),
++        });
++        if (res.status === 401) tokens.invalidate();
++        return { status: res.status, ok: res.ok, body: await res.json().catch(() => ({})) };
++    };
++    const attempt = async () => {
++        const listed = await call('GET', '/api/v1/subscriptions');
++        if (!listed.ok) throw new Error(`listing subscriptions: ${listed.status}`);
++        const mine = (listed.body.subscriptions || []).filter((s) => s.endpoint === endpoint);
++        for (const topic of topics) {
++            if (mine.some((s) => s.topic_pattern === topic)) continue;
++            const r = await call('POST', '/api/v1/subscriptions', { topic_pattern: topic, endpoint, secret });
++            if (!r.ok && r.status !== 409) throw new Error(`subscribing to ${topic}: ${r.status} ${r.body.code || ''}`.trim());
++            if (r.ok) log.log(`[Events] subscription created: ${r.body.id} (${topic} → ${endpoint})`);
++        }
++    };
++    let i = 0;
++    let timer = null;
++    let stopped = false;
++    let settle;
++    const done = new Promise((resolve) => { settle = resolve; });
++    const schedule = (ms) => { timer = setTimeout(run, ms); if (timer.unref) timer.unref(); };
++    function run() {
++        timer = null;
++        if (stopped) return;
++        attempt().then(() => settle(true), (err) => {
++            if (stopped) return;
++            if (++i < delays.length) schedule(delays[i]);
++            else { log.warn(`[Events] subscriptions not created: ${err.message}`); settle(false); }
++        });
++    }
++    schedule(delays[0] || 0);
++    return { topics: [...topics], endpoint, done, stop() { stopped = true; if (timer) clearTimeout(timer); timer = null; settle(false); } };
++}
++
++module.exports = { createAccountData, createNetworkSender, startSubscriptions, ACCOUNT_DATA_SCHEMA, TOPICS };
ddiff --git a/src/core/index.js b/src/core/index.js
iindex b726da2..c06364a 100644
---- a/src/core/index.js
++++ b/src/core/index.js
@@@ -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.36.0';
++const SDK_VERSION = '0.37.0';
  
  module.exports = {
      createClient,
ddiff --git a/test/account-data.test.js b/test/account-data.test.js
iindex bcfccb4..61ba5c7 100644
---- a/test/account-data.test.js
++++ b/test/account-data.test.js
@@@ -8,7 +8,7 @@ const http = require('node:http');
  const { run, stubServer } = require('./helpers');
  const { createDb } = require('../src/db');
  const { signDeliveryHeaders } = require('../src/events');
--const { createAccountData, createNetworkSender, ACCOUNT_DATA_SCHEMA, TOPICS } = require('../src/account-data');
++const { createAccountData, createNetworkSender, startSubscriptions, ACCOUNT_DATA_SCHEMA, TOPICS } = require('../src/account-data');
  
  const A = 'usr_01JZ0000000000000000000AAA';
  const B = 'usr_01JZ0000000000000000000BBB';
@@@ -173,4 +173,40 @@ run([
              await db.close();
          }
      }],
++    ['startSubscriptions: the missing topics are created once with the events token, a failure is retried, a restart creates nothing', async () => {
++        const subs = [];
++        let failListing = 1;
++        const tokenAsks = [];
++        const stub = await stubServer(async (req, res, body) => {
++            res.setHeader('Content-Type', 'application/json');
++            if (req.url === '/oauth/token') { tokenAsks.push(Object.fromEntries(new URLSearchParams(body.toString()))); return res.end(JSON.stringify({ access_token: 'evtok', expires_in: 300 })); }
++            assert.equal(req.headers.authorization, 'Bearer evtok');
++            if (req.method === 'GET') {
++                if (failListing-- > 0) { res.statusCode = 502; return res.end('{}'); }
++                return res.end(JSON.stringify({ subscriptions: subs }));
++            }
++            const b = JSON.parse(body.toString());
++            const sub = { id: `sub_${subs.length + 1}`, topic_pattern: b.topic_pattern, endpoint: b.endpoint };
++            subs.push(sub);
++            res.statusCode = 201;
++            return res.end(JSON.stringify(sub));
++        });
++        const opts = { eventsUrl: `${stub.url}/`, networkInternalUrl: stub.url, endpoint: 'http://127.0.0.1:4970/internal/events', secret: SECRET, clientId: 'food', clientSecret: 'secret', log: quiet, delays: [0, 5] };
++        try {
++            assert.equal(startSubscriptions({ ...opts, secret: '' }), null, 'off without the delivery secret');
++            assert.equal(startSubscriptions({ ...opts, eventsUrl: '' }), null, 'off without the Events URL');
++            const first = startSubscriptions(opts);
++            assert.equal(await first.done, true);
++            assert.deepEqual(subs.map((x) => [x.topic_pattern, x.endpoint]), TOPICS.map((t) => [t, opts.endpoint]));
++            assert.deepEqual([tokenAsks[0].audience, tokenAsks[0].scope, tokenAsks[0].client_id], ['openvibe.events', 'events.subscription.manage', 'food']);
++            const again = startSubscriptions(opts);
++            assert.equal(await again.done, true);
++            assert.equal(subs.length, 2, 'a restart creates nothing');
++            const stopped = startSubscriptions({ ...opts, delays: [60_000] });
++            stopped.stop();
++            assert.equal(await stopped.done, false, 'stop() before the first attempt');
++        } finally {
++            await stub.close();
++        }
++    }],
  ]);
ddiff --git a/types/account-data.d.ts b/types/account-data.d.ts
iindex f961e4e..06534d6 100644
---- a/types/account-data.d.ts
++++ b/types/account-data.d.ts
@@@ -44,5 +44,12 @@ export function createAccountData(opts: {
      log?: { log(msg: string): void; warn(msg: string): void };
  }): AccountData;
  export function createNetworkSender(opts: { networkInternalUrl: string; clientId: string; clientSecret: string; fetch?: typeof fetch; timeoutMs?: number }): NetworkSend;
++export interface Subscriptions { topics: string[]; endpoint: string; done: Promise<boolean>; stop(): void }
++/** Create any missing Events subscription for `topics` (default TOPICS) at boot; null when a setting is missing. */
++export function startSubscriptions(opts: {
++    eventsUrl: string; endpoint: string; secret: string; topics?: readonly string[];
++    networkInternalUrl: string; clientId: string; clientSecret: string;
++    fetch?: typeof fetch; log?: { log(msg: string): void; warn(msg: string): void }; delays?: number[];
++}): Subscriptions | null;
  export const ACCOUNT_DATA_SCHEMA: string;
  export const TOPICS: readonly string[];