openvibers/openvibe.events Commits

Capability ids with three segments (events.event.publish, events.event.read, events.subscription.manage, events.delivery.admin); openvibe-contracts v0.6.0; contract check in CI

OP OpenVibers committed 23 Sept 2026 · 5082c385f8 · 25 files, +154 −155


Changes

ddiff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
iindex 1ba60fe..ae6aa5a 100644
---- a/.github/workflows/ci.yml
++++ b/.github/workflows/ci.yml
@@@ -14,6 +14,5 @@ jobs:
cache: npm
- run: npm ci --no-audit --no-fund
- run: npm test
-- # Add once the events.* capabilities and the events service manifest ship in openvibe-contracts
-- # (proposals in docs/capabilities-proposal/):
-- # - run: npx openvibe-contracts-check --service events --src server
++ - name: contract check (openvibe-contracts)
++ run: npx openvibe-contracts-check --service events --src server
ddiff --git a/README.md b/README.md
iindex 94411c3..7e92a56 100644
---- a/README.md
++++ b/README.md
@@@ -30,10 +30,10 @@ Services call with an OpenVibe.Network client-credentials token (`POST /oauth/to
| Capability | Routes |
|---|---|
--| `events.publish` | `POST /api/v1/events` |
--| `events.subscribe` | `/api/v1/subscriptions…` (own subscriptions only) |
--| `events.read` | `GET /api/v1/events`, `GET /api/v1/events/:id`, `/api/v1/checkpoints`, realtime as a service |
--| `events.admin` | `GET /api/v1/deliveries`, `POST /api/v1/deliveries/replay` |
++| `events.event.publish` | `POST /api/v1/events` |
++| `events.subscription.manage` | `/api/v1/subscriptions…` (own subscriptions only) |
++| `events.event.read` | `GET /api/v1/events`, `GET /api/v1/events/:id`, `/api/v1/checkpoints`, realtime as a service |
++| `events.delivery.admin` | `GET /api/v1/deliveries`, `POST /api/v1/deliveries/replay` |
These ids are not in `openvibe-contracts` yet; their manifests are proposed in [docs/capabilities-proposal/](docs/capabilities-proposal/) (with the service manifest). Until the contracts release defines them, `server/auth.js` grants them with the contracts rule (exact id or a `family.*` grant) and hands the decision to `capabilities.check()` as soon as contracts know the id.
@@@ -102,7 +102,7 @@ es.onmessage = (m) => { const { seq, event } = JSON.parse(m.data); };
es.addEventListener('gap', (m) => { /* events were missed: refetch state */ });
```
--- Auth: the Network `ov_token` cookie or a Bearer user JWT; a service token with `events.read`; or nobody (public events only, `REALTIME_ALLOW_ANONYMOUS`). An expired cookie degrades to anonymous; a bad Bearer is a 401.
++- Auth: the Network `ov_token` cookie or a Bearer user JWT; a service token with `events.event.read`; or nobody (public events only, `REALTIME_ALLOW_ANONYMOUS`). An expired cookie degrades to anonymous; a bad Bearer is a 401.
- Visibility: `public` events go to anyone subscribed to the topic; `subject` events only to the user whose subject id (`usr_…`) is the event's `actor.id` or its user `subject.id`; `internal` events never reach a browser. A guessed topic yields nothing.
- Resume: the SSE `id` is the seq, so the browser's automatic `Last-Event-ID` (or `?last_event_id=`) replays what was missed. A cursor older than retention first gets `event: gap` (`{ reason, from_seq, to_seq }`), as does a replay that hits `REALTIME_REPLAY_MAX`.
- Heartbeat comment every 25 s; at most 20 topics per connection and `REALTIME_MAX_CONNECTIONS` (2000) overall; CORS with credentials for `https://*.openvibe.*` only.
ddiff --git a/STATUS.json b/STATUS.json
iindex 7003d0e..9e3fdf5 100644
---- a/STATUS.json
++++ b/STATUS.json
@@@ -8,7 +8,7 @@
"port": 4300,
"domain": "events.openvibe.network",
"contracts": "openvibe-contracts v0.5.1",
-- "capabilities": ["events.publish", "events.subscribe", "events.read", "events.admin"],
++ "capabilities": ["events.event.publish", "events.subscription.manage", "events.event.read", "events.delivery.admin"],
"capabilitiesRegistered": false,
"deployed": false,
"updated": "2026-09-22"
ddiff --git a/docs/capabilities-proposal/events.admin.json b/docs/capabilities-proposal/events.admin.json
ddeleted file mode 100644
iindex 107719c..0000000
---- a/docs/capabilities-proposal/events.admin.json
++++ /dev/null
@@@ -1,20 +0,0 @@
--{
-- "id": "events.admin",
-- "version": "1.0",
-- "owner": "events",
-- "status": "active",
-- "visibility": "internal",
-- "description": "Operator access: inspect deliveries (including the dead-letter queue) and replay events to any subscription.",
-- "permissions": [
-- "events:admin"
-- ],
-- "resourceConstraints": [
-- "none"
-- ],
-- "quotaClass": "events-admin",
-- "events": [],
-- "implementedBy": [
-- "GET /api/v1/deliveries",
-- "POST /api/v1/deliveries/replay"
-- ]
--}
ddiff --git a/docs/capabilities-proposal/events.delivery.admin.json b/docs/capabilities-proposal/events.delivery.admin.json
nnew file mode 100644
iindex 0000000..79c9889
---- /dev/null
++++ b/docs/capabilities-proposal/events.delivery.admin.json
@@@ -0,0 +1,20 @@
++{
++ "id": "events.delivery.admin",
++ "version": "1.0",
++ "owner": "events",
++ "status": "active",
++ "visibility": "internal",
++ "description": "Operator access: inspect deliveries (including the dead-letter queue) and replay events to any subscription.",
++ "permissions": [
++ "events:admin"
++ ],
++ "resourceConstraints": [
++ "none"
++ ],
++ "quotaClass": "events-admin",
++ "events": [],
++ "implementedBy": [
++ "GET /api/v1/deliveries",
++ "POST /api/v1/deliveries/replay"
++ ]
++}
ddiff --git a/docs/capabilities-proposal/events.event.publish.json b/docs/capabilities-proposal/events.event.publish.json
nnew file mode 100644
iindex 0000000..b3bb1e2
---- /dev/null
++++ b/docs/capabilities-proposal/events.event.publish.json
@@@ -0,0 +1,20 @@
++{
++ "id": "events.event.publish",
++ "version": "1.0",
++ "owner": "events",
++ "status": "active",
++ "visibility": "internal",
++ "description": "Publish durable events (events.event-envelope@1) as the calling service: source must be the principal's service slug and event_type must use a prefix that source owns. Idempotent on event_id.",
++ "inputSchema": "events.event-envelope@1",
++ "permissions": [
++ "events:write"
++ ],
++ "resourceConstraints": [
++ "owner"
++ ],
++ "quotaClass": "events-publish",
++ "events": [],
++ "implementedBy": [
++ "POST /api/v1/events"
++ ]
++}
ddiff --git a/docs/capabilities-proposal/events.event.read.json b/docs/capabilities-proposal/events.event.read.json
nnew file mode 100644
iindex 0000000..5e9de8d
---- /dev/null
++++ b/docs/capabilities-proposal/events.event.read.json
@@@ -0,0 +1,24 @@
++{
++ "id": "events.event.read",
++ "version": "1.0",
++ "owner": "events",
++ "status": "active",
++ "visibility": "internal",
++ "description": "Read retained events of every visibility with a seq cursor (pull consumers), keep per-consumer checkpoints, and open realtime streams as a service principal.",
++ "outputSchema": "events.event-envelope@1",
++ "permissions": [
++ "events:read"
++ ],
++ "resourceConstraints": [
++ "none"
++ ],
++ "quotaClass": "events-read",
++ "events": [],
++ "implementedBy": [
++ "GET /api/v1/events",
++ "GET /api/v1/events/:event_id",
++ "GET /api/v1/checkpoints",
++ "PUT /api/v1/checkpoints",
++ "GET /realtime/stream"
++ ]
++}
ddiff --git a/docs/capabilities-proposal/events.publish.json b/docs/capabilities-proposal/events.publish.json
ddeleted file mode 100644
iindex dd2c32a..0000000
---- a/docs/capabilities-proposal/events.publish.json
++++ /dev/null
@@@ -1,20 +0,0 @@
--{
-- "id": "events.publish",
-- "version": "1.0",
-- "owner": "events",
-- "status": "active",
-- "visibility": "internal",
-- "description": "Publish durable events (events.event-envelope@1) as the calling service: source must be the principal's service slug and event_type must use a prefix that source owns. Idempotent on event_id.",
-- "inputSchema": "events.event-envelope@1",
-- "permissions": [
-- "events:write"
-- ],
-- "resourceConstraints": [
-- "owner"
-- ],
-- "quotaClass": "events-publish",
-- "events": [],
-- "implementedBy": [
-- "POST /api/v1/events"
-- ]
--}
ddiff --git a/docs/capabilities-proposal/events.read.json b/docs/capabilities-proposal/events.read.json
ddeleted file mode 100644
iindex 0dd2a0c..0000000
---- a/docs/capabilities-proposal/events.read.json
++++ /dev/null
@@@ -1,24 +0,0 @@
--{
-- "id": "events.read",
-- "version": "1.0",
-- "owner": "events",
-- "status": "active",
-- "visibility": "internal",
-- "description": "Read retained events of every visibility with a seq cursor (pull consumers), keep per-consumer checkpoints, and open realtime streams as a service principal.",
-- "outputSchema": "events.event-envelope@1",
-- "permissions": [
-- "events:read"
-- ],
-- "resourceConstraints": [
-- "none"
-- ],
-- "quotaClass": "events-read",
-- "events": [],
-- "implementedBy": [
-- "GET /api/v1/events",
-- "GET /api/v1/events/:event_id",
-- "GET /api/v1/checkpoints",
-- "PUT /api/v1/checkpoints",
-- "GET /realtime/stream"
-- ]
--}
ddiff --git a/docs/capabilities-proposal/events.subscribe.json b/docs/capabilities-proposal/events.subscribe.json
ddeleted file mode 100644
iindex 60e32a2..0000000
---- a/docs/capabilities-proposal/events.subscribe.json
++++ /dev/null
@@@ -1,23 +0,0 @@
--{
-- "id": "events.subscribe",
-- "version": "1.0",
-- "owner": "events",
-- "status": "active",
-- "visibility": "internal",
-- "description": "Create, list, disable and enable the calling service's own webhook subscriptions (topic pattern + allow-listed internal endpoint, HMAC-signed deliveries with retry and dead letters).",
-- "permissions": [
-- "events:subscribe"
-- ],
-- "resourceConstraints": [
-- "owner"
-- ],
-- "quotaClass": "events-subscribe",
-- "events": [],
-- "implementedBy": [
-- "POST /api/v1/subscriptions",
-- "GET /api/v1/subscriptions",
-- "GET /api/v1/subscriptions/:id",
-- "POST /api/v1/subscriptions/:id/disable",
-- "POST /api/v1/subscriptions/:id/enable"
-- ]
--}
ddiff --git a/docs/capabilities-proposal/events.subscription.manage.json b/docs/capabilities-proposal/events.subscription.manage.json
nnew file mode 100644
iindex 0000000..7c38ec9
---- /dev/null
++++ b/docs/capabilities-proposal/events.subscription.manage.json
@@@ -0,0 +1,23 @@
++{
++ "id": "events.subscription.manage",
++ "version": "1.0",
++ "owner": "events",
++ "status": "active",
++ "visibility": "internal",
++ "description": "Create, list, disable and enable the calling service's own webhook subscriptions (topic pattern + allow-listed internal endpoint, HMAC-signed deliveries with retry and dead letters).",
++ "permissions": [
++ "events:subscribe"
++ ],
++ "resourceConstraints": [
++ "owner"
++ ],
++ "quotaClass": "events-subscribe",
++ "events": [],
++ "implementedBy": [
++ "POST /api/v1/subscriptions",
++ "GET /api/v1/subscriptions",
++ "GET /api/v1/subscriptions/:id",
++ "POST /api/v1/subscriptions/:id/disable",
++ "POST /api/v1/subscriptions/:id/enable"
++ ]
++}
ddiff --git a/docs/capabilities-proposal/service-manifest.events.json b/docs/capabilities-proposal/service-manifest.events.json
iindex cd29c69..4db70a7 100644
---- a/docs/capabilities-proposal/service-manifest.events.json
++++ b/docs/capabilities-proposal/service-manifest.events.json
@@@ -14,10 +14,10 @@
"health": "/api/health",
"ready": "/api/ready",
"capabilities": [
-- "events.publish",
-- "events.subscribe",
-- "events.read",
-- "events.admin"
++ "events.event.publish",
++ "events.subscription.manage",
++ "events.event.read",
++ "events.delivery.admin"
],
"eventsProduced": [],
"eventsConsumed": [],
ddiff --git a/package-lock.json b/package-lock.json
iindex b96a408..aabab88 100644
---- a/package-lock.json
++++ b/package-lock.json
@@@ -12,7 +12,7 @@
"better-sqlite3": "^11.10.0",
"dotenv": "^16.4.7",
"express": "^4.21.2",
-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.5.1"
++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.6.0"
},
"engines": {
"node": ">=20.0.0"
@@@ -860,9 +860,9 @@
}
},
"node_modules/openvibe-contracts": {
-- "version": "0.5.1",
-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.5.1",
-- "integrity": "sha512-czQpWKEo8cY56Crj5l5Mv8r5vlIIj0vn/HMxv0OlIu5YTuSSlpvXyayYEfn3GQ5orDhaMIhOUlzmQMxUtWXaLA==",
++ "version": "0.6.0",
++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.6.0",
++ "integrity": "sha512-bRN/XG+mlvg0lnOc+9kVa5HgxBWTXy/xRfDn8Gnz5k41FT73zg6i463Le7yWkcLKhq6HI3NKlpBf3C/NRpSX9g==",
"license": "AGPL-3.0-only",
"dependencies": {
"ajv": "^8.17.1",
ddiff --git a/package.json b/package.json
iindex bd3fa28..7530bca 100644
---- a/package.json
++++ b/package.json
@@@ -17,7 +17,7 @@
"better-sqlite3": "^11.10.0",
"dotenv": "^16.4.7",
"express": "^4.21.2",
-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.5.1"
++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.6.0"
},
"engines": {
"node": ">=20.0.0"
ddiff --git a/server/api/publish.js b/server/api/publish.js
iindex dc59317..290dc6b 100644
---- a/server/api/publish.js
++++ b/server/api/publish.js
@@@ -1,6 +1,6 @@
'use strict';
/**
-- * POST /api/v1/events (service token, audience openvibe.events, capability events.publish)
++ * POST /api/v1/events (service token, audience openvibe.events, capability events.event.publish)
*
* body: <envelope> -> 201 { event_id, seq, duplicate: false } | 200 on a repeat
* body: { events: [...] } -> 201 { results: [{ event_id, seq, duplicate }] } (<= 100, atomic)
ddiff --git a/server/api/read.js b/server/api/read.js
iindex 59723df..1de6f0b 100644
---- a/server/api/read.js
++++ b/server/api/read.js
@@@ -1,6 +1,6 @@
'use strict';
/**
-- * Pull consumers (service token with events.read):
++ * Pull consumers (service token with events.event.read):
*
* GET /api/v1/events?topic=media.vod.*[,…]&after_seq=0&limit=100
* -> { events: [{ seq, event }], next_after_seq, latest_seq, gap? }
@@@ -10,7 +10,7 @@
* GET /api/v1/checkpoints?topic=… / PUT /api/v1/checkpoints { topic, cursor }
* a consumer's own stored cursor per topic pattern (consumer = calling principal)
*
-- * Operators (events.admin):
++ * Operators (events.delivery.admin):
*
* GET /api/v1/deliveries?status=dead&subscription_id=&after_seq=&limit=
* POST /api/v1/deliveries/replay { subscription_id, from_seq | event_ids: [...] }
ddiff --git a/server/api/subscriptions.js b/server/api/subscriptions.js
iindex 8afaa18..4b0df45 100644
---- a/server/api/subscriptions.js
++++ b/server/api/subscriptions.js
@@@ -1,6 +1,6 @@
'use strict';
/**
-- * Subscriptions (service token with events.subscribe; the consumer is always the calling service).
++ * Subscriptions (service token with events.subscription.manage; the consumer is always the calling service).
*
* POST /api/v1/subscriptions { topic_pattern, endpoint, secret?, retry_policy? }
* -> 201 subscription + secret (shown once)
ddiff --git a/server/app.js b/server/app.js
iindex 8971e55..2ea532a 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -50,10 +50,10 @@ function createApp({ config, store, auth, keys, worker, realtime, log = console
res.type('text/plain').send([
'OpenVibe.Events: durable events, subscriptions, signed delivery, dead letters and replay.',
'',
-- 'POST /api/v1/events publish (service token, events.publish)',
-- 'GET /api/v1/events pull with a cursor (events.read)',
-- ' /api/v1/subscriptions webhook subscriptions (events.subscribe)',
-- ' /api/v1/deliveries DLQ inspect and replay (events.admin)',
++ 'POST /api/v1/events publish (service token, events.event.publish)',
++ 'GET /api/v1/events pull with a cursor (events.event.read)',
++ ' /api/v1/subscriptions webhook subscriptions (events.subscription.manage)',
++ ' /api/v1/deliveries DLQ inspect and replay (events.delivery.admin)',
'GET /realtime/stream?topics=... server-sent events for browsers',
'GET /api/health, /api/ready',
'',
ddiff --git a/server/auth.js b/server/auth.js
iindex 4c56e29..2da0a91 100644
---- a/server/auth.js
++++ b/server/auth.js
@@@ -6,7 +6,7 @@
* openvibe.events), verified with openvibe-contracts serviceAuth.verifyServiceToken.
* - Browsers: Network user JWTs (RS256, same key), cookie `ov_token` or Bearer.
*
-- * Capability ids events.publish / events.subscribe / events.read / events.admin are proposed in
++ * Capability ids events.event.publish / events.subscription.manage / events.event.read / events.delivery.admin are proposed in
* docs/capabilities-proposal/ and are not in openvibe-contracts yet. capabilities.check() answers
* capability.unknown for an id it does not know; until the release that defines them, hasCap()
* decides with the same grant rule (exact id, or a `family.*` grant).
@@@ -16,10 +16,10 @@ const { serviceAuth, capabilities, http } = require('openvibe-contracts');
/** The capability ids this service enforces (manifests: docs/capabilities-proposal/). */
const CAPS = Object.freeze({
-- publish: 'events.publish',
-- subscribe: 'events.subscribe',
-- read: 'events.read',
-- admin: 'events.admin',
++ publish: 'events.event.publish',
++ subscribe: 'events.subscription.manage',
++ read: 'events.event.read',
++ admin: 'events.delivery.admin',
});
// ── Network public key ─────────────────────────────────────
ddiff --git a/server/realtime.js b/server/realtime.js
iindex e333cc9..5b0e5ea 100644
---- a/server/realtime.js
++++ b/server/realtime.js
@@@ -9,7 +9,7 @@
* public anyone subscribed to a matching topic (signed-out visitors too, unless disabled)
* subject only the user whose subject id is the event's actor.id or its subject.id (subject
* type user); a guessed topic yields nothing for anyone else
-- * internal service principals (token with events.read) only, never a browser
++ * internal service principals (token with events.event.read) only, never a browser
*
* Resume: `Last-Event-ID` (or ?last_event_id=) is a seq. Missed events are replayed first; when the
* cursor is older than retention an `event: gap` message comes first, so the client knows to
ddiff --git a/test/delivery.test.js b/test/delivery.test.js
iindex 268a911..75e4960 100644
---- a/test/delivery.test.js
++++ b/test/delivery.test.js
@@@ -5,9 +5,9 @@ const { boot, request, serviceToken, envelope, subscriber, suite, sleep } = requ
const t = suite('delivery');
let h;
--const live = serviceToken('live', ['events.publish']);
--const media = serviceToken('media', ['events.subscribe']);
--const admin = serviceToken('ops', ['events.admin']);
++const live = serviceToken('live', ['events.event.publish']);
++const media = serviceToken('media', ['events.subscription.manage']);
++const admin = serviceToken('ops', ['events.delivery.admin']);
async function subscribe(token, body) {
return request(h.base, 'POST', '/api/v1/subscriptions', { token, body });
@@@ -30,13 +30,13 @@ t('subscription API: create, list own, secret shown once', async () => {
const list = await request(h.base, 'GET', '/api/v1/subscriptions', { token: media });
assert.strictEqual(list.body.subscriptions.length, 1);
assert.strictEqual(list.body.subscriptions[0].secret, undefined);
-- const other = await request(h.base, 'GET', '/api/v1/subscriptions', { token: serviceToken('games', ['events.subscribe']) });
++ const other = await request(h.base, 'GET', '/api/v1/subscriptions', { token: serviceToken('games', ['events.subscription.manage']) });
assert.strictEqual(other.body.subscriptions.length, 0, 'consumers only see their own');
const dup = await subscribe(media, { topic_pattern: 'live.demo.*', endpoint: stub.url });
assert.strictEqual(dup.status, 409);
-- const foreign = await request(h.base, 'POST', `/api/v1/subscriptions/${r.body.id}/disable`, { token: serviceToken('games', ['events.subscribe']) });
++ const foreign = await request(h.base, 'POST', `/api/v1/subscriptions/${r.body.id}/disable`, { token: serviceToken('games', ['events.subscription.manage']) });
assert.strictEqual(foreign.status, 404, 'cannot disable another consumer\'s subscription');
-- const noCap = await subscribe(serviceToken('media', ['events.read']), { topic_pattern: 'x.y.z', endpoint: stub.url });
++ const noCap = await subscribe(serviceToken('media', ['events.event.read']), { topic_pattern: 'x.y.z', endpoint: stub.url });
assert.strictEqual(noCap.status, 403);
await request(h.base, 'POST', `/api/v1/subscriptions/${r.body.id}/disable`, { token: media });
await stub.close();
@@@ -54,11 +54,11 @@ t('SSRF: only http(s) endpoints on 127.0.0.1 or *.openvibe.* are accepted', asyn
assert.strictEqual(r.body.code, 'events.endpoint_not_allowed', endpoint);
}
for (const endpoint of ['http://127.0.0.1:3000/internal/events', 'https://openvibe.live/internal/events', 'https://live.openvibe.network/hook']) {
-- const r = await subscribe(serviceToken('ssrfok', ['events.subscribe']), { topic_pattern: 'nothing.matches.this', endpoint });
++ const r = await subscribe(serviceToken('ssrfok', ['events.subscription.manage']), { topic_pattern: 'nothing.matches.this', endpoint });
assert.strictEqual(r.status, 201, endpoint);
}
// http://2130706433/ is 127.0.0.1 once parsed; it is stored normalised.
-- const r = await subscribe(serviceToken('ssrfok', ['events.subscribe']), { topic_pattern: 'nothing.matches.that', endpoint: 'http://2130706433:9/x' });
++ const r = await subscribe(serviceToken('ssrfok', ['events.subscription.manage']), { topic_pattern: 'nothing.matches.that', endpoint: 'http://2130706433:9/x' });
assert.strictEqual(r.status, 201);
assert.strictEqual(r.body.endpoint, 'http://127.0.0.1:9/x');
});
@@@ -155,7 +155,7 @@ t('priority classes: critical before important before low', async () => {
const orders = [[], [], []];
const stubs = [];
for (let i = 0; i < 3; i++) stubs.push(await subscriber((c) => { orders[i].push(c.body.event.priority); return 204; }));
-- for (let i = 0; i < 3; i++) await subscribe(serviceToken(`prio${i}`, ['events.subscribe']), { topic_pattern: 'live.prio.*', endpoint: stubs[i].url });
++ for (let i = 0; i < 3; i++) await subscribe(serviceToken(`prio${i}`, ['events.subscription.manage']), { topic_pattern: 'live.prio.*', endpoint: stubs[i].url });
// Low first, critical last: the queue must still send critical first.
await publish(envelope('live', { event_type: 'live.prio.x', priority: 'low' }));
await publish(envelope('live', { event_type: 'live.prio.x', priority: 'important' }));
@@@ -184,7 +184,7 @@ t('backpressure: one in flight per subscription, <= maxInflight (20) overall', a
};
const bp = await boot({ deliveryFetch: fetchImpl });
for (let i = 0; i < 25; i++) {
-- const r = await request(bp.base, 'POST', '/api/v1/subscriptions', { token: serviceToken(`bp${i}`, ['events.subscribe']), body: { topic_pattern: 'live.bp.*', endpoint: 'http://127.0.0.1:9/hook' } });
++ const r = await request(bp.base, 'POST', '/api/v1/subscriptions', { token: serviceToken(`bp${i}`, ['events.subscription.manage']), body: { topic_pattern: 'live.bp.*', endpoint: 'http://127.0.0.1:9/hook' } });
assert.strictEqual(r.status, 201);
}
for (let i = 0; i < 3; i++) await request(bp.base, 'POST', '/api/v1/events', { token: live, body: envelope('live', { event_type: 'live.bp.x' }) });
ddiff --git a/test/outbox-inbox.test.js b/test/outbox-inbox.test.js
iindex 3def504..4367813 100644
---- a/test/outbox-inbox.test.js
++++ b/test/outbox-inbox.test.js
@@@ -11,7 +11,7 @@ const t = suite('outbox-inbox');
let h;
let network;
let tokenClient;
--const admin = serviceToken('ops', ['events.admin']);
++const admin = serviceToken('ops', ['events.delivery.admin']);
function countIn(id) {
return h.db.prepare('SELECT COUNT(*) AS n FROM events WHERE id = ?').get(id).n;
@@@ -29,7 +29,7 @@ t('boot Events + a stub Network token endpoint', async () => {
assert.strictEqual(p.get('audience'), 'openvibe.events');
issued++;
res.setHeader('Content-Type', 'application/json');
-- res.end(JSON.stringify({ access_token: serviceToken(p.get('client_id'), ['events.publish']), token_type: 'Bearer', expires_in: 300 }));
++ res.end(JSON.stringify({ access_token: serviceToken(p.get('client_id'), ['events.event.publish']), token_type: 'Bearer', expires_in: 300 }));
});
});
await new Promise(r => network.listen(0, '127.0.0.1', r));
@@@ -150,10 +150,10 @@ t('consumer crash mid-processing + redelivery + replay -> exactly one effect (cr
if (mode === 'crash-after-commit') { mode = 'ok'; return 'destroy'; } // effect committed, no response sent
return 204;
});
-- const media = serviceToken('media', ['events.subscribe']);
++ const media = serviceToken('media', ['events.subscription.manage']);
const sub = (await request(h.base, 'POST', '/api/v1/subscriptions', { token: media, body: { topic_pattern: 'live.tip.*', endpoint: stub.url } })).body;
secret = sub.secret;
-- const live = serviceToken('live', ['events.publish']);
++ const live = serviceToken('live', ['events.event.publish']);
const effects = (id) => consumerDb.prepare('SELECT COUNT(*) AS n FROM credits WHERE event_id = ?').get(id).n;
for (const first of ['crash-after-commit', 'crash-before-commit']) {
ddiff --git a/test/publish.test.js b/test/publish.test.js
iindex 1eb875d..57b7da0 100644
---- a/test/publish.test.js
++++ b/test/publish.test.js
@@@ -5,7 +5,7 @@ const { boot, request, serviceToken, envelope, suite } = require('./helpers');
const t = suite('publish');
let h;
--const live = serviceToken('live', ['events.publish']);
++const live = serviceToken('live', ['events.event.publish']);
t('boot', async () => { h = await boot(); });
@@@ -14,15 +14,15 @@ t('no token -> 401, wrong audience -> 401, missing capability -> 403', async ()
assert.strictEqual(r.status, 401);
assert.strictEqual(r.body.code, 'token.missing');
assert.match(r.headers.get('content-type'), /problem\+json/);
-- r = await request(h.base, 'POST', '/api/v1/events', { token: serviceToken('live', ['events.publish'], { aud: 'openvibe.media' }), body: envelope() });
++ r = await request(h.base, 'POST', '/api/v1/events', { token: serviceToken('live', ['events.event.publish'], { aud: 'openvibe.media' }), body: envelope() });
assert.strictEqual(r.status, 401);
assert.strictEqual(r.body.code, 'token.wrong_audience');
-- r = await request(h.base, 'POST', '/api/v1/events', { token: serviceToken('live', ['events.read']), body: envelope() });
++ r = await request(h.base, 'POST', '/api/v1/events', { token: serviceToken('live', ['events.event.read']), body: envelope() });
assert.strictEqual(r.status, 403);
assert.strictEqual(r.body.code, 'capability.denied');
});
--t('a family grant (events.*) covers events.publish', async () => {
++t('a family grant (events.*) covers events.event.publish', async () => {
const r = await request(h.base, 'POST', '/api/v1/events', { token: serviceToken('live', ['events.*']), body: envelope() });
assert.strictEqual(r.status, 201);
});
@@@ -73,14 +73,14 @@ t('event_type must use a prefix the source owns', async () => {
});
t('unknown source (not in the manifest map) -> 403', async () => {
-- const r = await request(h.base, 'POST', '/api/v1/events', { token: serviceToken('stranger', ['events.publish']), body: envelope('stranger') });
++ const r = await request(h.base, 'POST', '/api/v1/events', { token: serviceToken('stranger', ['events.event.publish']), body: envelope('stranger') });
assert.strictEqual(r.status, 403);
assert.strictEqual(r.body.code, 'events.unknown_source');
});
t('app principals cannot publish', async () => {
const r = await request(h.base, 'POST', '/api/v1/events', {
-- token: serviceToken('x', ['events.publish'], { sub: `app:${ids.newId('app')}` }), body: envelope('live'),
++ token: serviceToken('x', ['events.event.publish'], { sub: `app:${ids.newId('app')}` }), body: envelope('live'),
});
assert.strictEqual(r.status, 403);
});
@@@ -125,7 +125,7 @@ t('payload size limit', async () => {
t('loop guard: a trace bouncing between services is refused at 8 hops', async () => {
const trace = 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa';
-- const media = serviceToken('media', ['events.publish']);
++ const media = serviceToken('media', ['events.event.publish']);
let last;
for (let i = 0; i < 9; i++) {
const src = i % 2 ? 'media' : 'live';
ddiff --git a/test/pull-health.test.js b/test/pull-health.test.js
iindex 6477925..cf23a37 100644
---- a/test/pull-health.test.js
++++ b/test/pull-health.test.js
@@@ -9,8 +9,8 @@ const { boot, request, serviceToken, envelope, publicKey, suite, sleep } = requi
const t = suite('pull-health');
let h;
--const live = serviceToken('live', ['events.publish']);
--const reader = serviceToken('games', ['events.read']);
++const live = serviceToken('live', ['events.event.publish']);
++const reader = serviceToken('games', ['events.event.read']);
t('topic patterns: * spans one or more segments', () => {
const yes = [['media.vod.*', 'media.vod.ready'], ['media.vod.*', 'media.vod.clip.cut'], ['*.created', 'community.post.created'],
@@@ -23,12 +23,12 @@ t('topic patterns: * spans one or more segments', () => {
});
t('capability grants: exact or family wildcard; contracts decide ids they know', () => {
-- assert.ok(hasCap({ cap: ['events.publish'] }, 'events.publish'));
-- assert.ok(hasCap({ cap: ['events.*'] }, 'events.admin'));
-- assert.ok(!hasCap({ cap: ['events.pub*'] }, 'events.publish'));
-- assert.ok(!hasCap({ cap: ['event.*'] }, 'events.publish'));
-- assert.ok(!hasCap({}, 'events.read'));
-- assert.strictEqual(allows({ cap: ['events.read'] }, 'events.read').allowed, true);
++ assert.ok(hasCap({ cap: ['events.event.publish'] }, 'events.event.publish'));
++ assert.ok(hasCap({ cap: ['events.*'] }, 'events.delivery.admin'));
++ assert.ok(!hasCap({ cap: ['events.pub*'] }, 'events.event.publish'));
++ assert.ok(!hasCap({ cap: ['event.*'] }, 'events.event.publish'));
++ assert.ok(!hasCap({}, 'events.event.read'));
++ assert.strictEqual(allows({ cap: ['events.event.read'] }, 'events.event.read').allowed, true);
assert.strictEqual(allows({ cap: ['media.object.upload'] }, 'media.object.upload').allowed, true, 'known ids go through contracts');
});
@@@ -62,7 +62,7 @@ t('pull: cursor, topic filter, internal events included for services', async ()
r = await request(h.base, 'GET', `/api/v1/events/${id3}`, { token: reader });
assert.strictEqual(r.body.seq, s3);
r = await request(h.base, 'GET', '/api/v1/events?topic=live.*', { token: live });
-- assert.strictEqual(r.status, 403, 'events.read is required');
++ assert.strictEqual(r.status, 403, 'events.event.read is required');
r = await request(h.base, 'GET', '/api/v1/events?limit=5000', { token: reader });
assert.strictEqual(r.status, 400);
});
@@@ -84,7 +84,7 @@ t('checkpoints: per consumer and topic', async () => {
r = await request(h.base, 'GET', '/api/v1/checkpoints?topic=live.vod.*', { token: reader });
assert.strictEqual(r.body.cursor, 42);
assert.strictEqual(r.body.consumer, 'games');
-- r = await request(h.base, 'GET', '/api/v1/checkpoints?topic=live.vod.*', { token: serviceToken('tools', ['events.read']) });
++ r = await request(h.base, 'GET', '/api/v1/checkpoints?topic=live.vod.*', { token: serviceToken('tools', ['events.event.read']) });
assert.strictEqual(r.body.cursor, 0, 'another consumer has its own');
});
ddiff --git a/test/realtime.test.js b/test/realtime.test.js
iindex c95acd8..815020a 100644
---- a/test/realtime.test.js
++++ b/test/realtime.test.js
@@@ -5,8 +5,8 @@ const { boot, request, serviceToken, userToken, envelope, sse, sleep, suite } =
const t = suite('realtime');
let h;
--const live = serviceToken('live', ['events.publish']);
--const network = serviceToken('network', ['events.publish']);
++const live = serviceToken('live', ['events.event.publish']);
++const network = serviceToken('network', ['events.event.publish']);
const alice = ids.newId('user');
const bob = ids.newId('user');
const open = [];
@@@ -29,7 +29,7 @@ t('visibility: public to everyone, subject only to that user, internal only to s
const anon = await stream('topics=live.stream.*,network.notification.*');
const a = await stream('topics=live.stream.*,network.notification.*', { Cookie: `theme=x; ov_token=${userToken({ subjectId: alice })}` });
const b = await stream('topics=live.stream.*,network.notification.*', { Authorization: `Bearer ${userToken({ subjectId: bob })}` });
-- const svc = await stream('topics=live.stream.*,network.notification.*', { Authorization: `Bearer ${serviceToken('live', ['events.read'])}` });
++ const svc = await stream('topics=live.stream.*,network.notification.*', { Authorization: `Bearer ${serviceToken('live', ['events.event.read'])}` });
assert.strictEqual(anon.status, 200);
assert.match(anon.headers['content-type'], /text\/event-stream/);
await sleep(50);
@@@ -123,8 +123,8 @@ t('limits and auth errors', async () => {
assert.strictEqual(r.status, 400);
r = await request(h.base, 'GET', '/realtime/stream?topics=live.*', { headers: { Authorization: 'Bearer not.a.token' } });
assert.strictEqual(r.status, 401, 'a bad Bearer is refused, not downgraded');
-- r = await request(h.base, 'GET', '/realtime/stream?topics=live.*', { token: serviceToken('live', ['events.publish']) });
-- assert.strictEqual(r.status, 403, 'service tokens need events.read');
++ r = await request(h.base, 'GET', '/realtime/stream?topics=live.*', { token: serviceToken('live', ['events.event.publish']) });
++ assert.strictEqual(r.status, 403, 'service tokens need events.event.read');
const expired = await stream('topics=live.*', { Cookie: `ov_token=${userToken({ exp: Math.floor(Date.now() / 1000) - 3600 })}` });
await sleep(30);
assert.ok(expired.comments.includes('connected anonymous'), 'an expired cookie degrades to anonymous');