openvibers/openvibe.events Commits
Events: the control API — suspend and resume a person's subscription from the Services console (#73)
Changes
@@@ -151,6 +151,12 @@ node scripts/events-archive.js restore --month 2026-01 [--hold-days 30]- OVRN: present exactly when `openvibe-contracts`' `contracts.resources.nameOf` composes one — `ovrn:events:<prj_…>:subscription/sub_…` for a project-scoped subscription. A project-less subscription is a first-party one (a service consumer with no project): it has no project segment, so no name, and only the unscoped caller sees it at all. An event is not a resource (`evt_` is what `common.resource-name@1` refuses by design) and a project's queue is a derived carrier class, not a stored row, so `events.queue` joins the index only when Events stores a queue row.- Errors: `400 resources.bad_query` for a value the index cannot honour (a project that is not a `prj_` id, a `limit` outside 1-1000, a cursor it did not issue) and `404 resources.unknown_resource` for an OVRN that names nothing it composes, both problem+json. Answers carry `Cache-Control: private, max-age=60`. The capability is `first-party`: a developer app's token is a `403`, as on every first-party route, and the two routes are not per-actor limited — they are one fan-out call each, not a browser's traffic.++**Control** (ADR-048, plan T13 step 10, contracts 0.168.0, `server/api/resource-control.js`): `POST /api/v1/resources/control` takes a `common.resource-control-request@1` from a first-party service holding `events.resource.control` (OpenVibe.Services, acting for the person in `on_behalf_of`), naming a subscription by its OVRN.++- **Actions**: `suspend` disables it (nothing is delivered until it resumes; the events it matches are kept and can be replayed) and `resume` enables it. Any other action, delete included, answers `control.unsupported_action`.++- **Who**: only the subscription's consumer, only inside the OVRN's project; anyone else's is `resources.unknown_resource`.++- **Dry runs and repeats**: `dry_run` decides without applying. The first answer per (caller, project, idempotency key) is kept in `events_control_log` (migration 0008) and returned for a repeat; another request under the same key is `409 control.idempotency_conflict`.++- **Test**: `test/resource-control.test.js`.++## The fabric: carriers, the planner and explainHow an event type is carried is its delivery policy (`events.delivery-policy@1`, [ADR-042](https://github.com/OpenVibers/OpenVibe.Contracts/blob/main/docs/adr/ADR-042-events-fabric.md) decision 1). The carrier class — TOPIC (fan-out, no ack), QUEUE (competing consumers, ack) or STREAM (retained, ordered per key) — is a derived, internal label, never in the envelope. PostgreSQL stays the record: every event and every delivery is committed before any carrier sees it, and the worker's lease is the only thing that makes a send happen. A publish commits, then the chosen carrier is signalled; a lost or duplicated signal changes latency, never the outcome.nnew file mode 100644@@@ -0,0 +1,12 @@++-- phase: expand++-- ADR-048 control operations (plan T13 step 10, server/api/resource-control.js): the first answer to each++-- (caller, project, idempotency_key), returned again for a repeat so a retried control request never applies twice.++CREATE TABLE events_control_log (++ caller text COLLATE "C" NOT NULL, -- svc:… that sent it++ project_id text COLLATE "C" NOT NULL,++ idempotency_key text COLLATE "C" NOT NULL,++ digest text COLLATE "C" NOT NULL, -- sha256 of the request without its delivery fields++ response jsonb NOT NULL, -- the common.resource-control-result@1 answered++ created_at bigint NOT NULL,++ PRIMARY KEY (caller, project_id, idempotency_key)++);@@@ -12,7 +12,7 @@"dotenv": "^16.4.7","express": "^4.21.2","iovalkey": "^0.4.0",-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.143.0",++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.168.0","openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.44.0","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.1.3","pg": "^8.23.0"@@@ -828,9 +828,9 @@}},"node_modules/openvibe-contracts": {-- "version": "0.143.0",-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.143.0",-- "integrity": "sha512-Cdbf+9RoPAUBibpM2zXQvtsiKMugb55dKubTt8pFZvAmxsLtGvX+zaC++yL1vOSub8FX3+GgqcGtoJS6tklrUg==",++ "version": "0.168.0",++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.168.0",++ "integrity": "sha512-v59y6iRgSSHozx98grx5CRwPHN8UCj0L88HHJpJVBw/Cnzd2QzvcQQZIPlOMqH9K1QEEYeFUIJMiCIMJez0b5A==","license": "AGPL-3.0-only","dependencies": {"ajv": "^8.17.1",@@@ -24,7 +24,7 @@"dotenv": "^16.4.7","express": "^4.21.2","iovalkey": "^0.4.0",-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.143.0",++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.168.0","openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.44.0","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.1.3","pg": "^8.23.0"nnew file mode 100644@@@ -0,0 +1,92 @@++'use strict';++/**++ * Events' control API (ADR-048, plan T13 step 10): POST /api/v1/resources/control — one++ * common.resource-control-request@1 from a first-party service (OpenVibe.Services) holding events.resource.control,++ * acting for the person in on_behalf_of, on an events.subscription named by its OVRN. The answer is a++ * common.resource-control-result@1 (200 for every outcome; a malformed request is a 400 problem).++ *++ * suspend disable the subscription: nothing is delivered to it until it resumes (events it matches are kept as++ * they always are, and can be replayed)++ * resume enable it again++ *++ * Any other action is refused `control.unsupported_action` (Events has no deletion of a subscription). Only the++ * subscription's consumer may act on it (on_behalf_of names them; another person's subscription is refused as not++ * found, never confirmed to exist), and only inside the project the OVRN names. `dry_run` decides and explains without++ * applying. The first answer per (caller, project, idempotency_key) is kept in events_control_log and returned for a++ * repeat; the same key with another request is refused `control.idempotency_conflict`.++ */++const crypto = require('crypto');++const express = require('express');++const contracts = require('openvibe-contracts');++++const SERVICE = 'events';++const CAPABILITY = 'events.resource.control';++const ACTIONS = { suspend: false, resume: true };++const USER_SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;++++function canonical(v) {++ if (Array.isArray(v)) return `[${v.map(canonical).join(',')}]`;++ if (v && typeof v === 'object') return `{${Object.keys(v).sort().filter((k) => v[k] !== undefined).map((k) => `${JSON.stringify(k)}:${canonical(v[k])}`).join(',')}}`;++ return JSON.stringify(v);++}++const digestOf = (request) => crypto.createHash('sha256')++ .update(canonical({ ...request, idempotency_key: undefined, confirmation_id: undefined, trace_id: undefined })).digest('hex');++++function resourceControlRouter({ store, auth, now = () => Date.now() }) {++ const router = express.Router();++ const guard = auth.requireCap(CAPABILITY, { requireService: true });++ const json = express.json({ limit: '16kb' });++ const db = store.db;++ const iso = () => new Date(now()).toISOString();++ const result = (request, state, extra = {}) => ({ action: request.action, ...(request.resource ? { resource: request.resource } : {}), state, at: iso(), ...extra });++ const problem = (status, code, detail) => ({ type: `https://openvibe.network/problems/${code}`, title: code, status, code, detail });++++ async function decide(request) {++ const parsed = contracts.resources.parse(request.resource || '');++ if (!parsed || parsed.service !== SERVICE || parsed.type !== 'subscription') {++ return result(request, 'refused', { problem: problem(404, 'resources.unknown_resource', `no subscription named ${request.resource || request.resource_kind}`) });++ }++ if (!(request.action in ACTIONS)) {++ return result(request, 'refused', { problem: problem(422, 'control.unsupported_action', `a subscription takes suspend and resume, not ${request.action}`) });++ }++ const person = request.on_behalf_of;++ if (!person || person.type !== 'user' || !USER_SUBJECT_RE.test(String(person.id || ''))) {++ return result(request, 'refused', { problem: problem(403, 'control.no_person', 'on_behalf_of must name the person whose subscription it is') });++ }++ const row = await store.getSubscription(parsed.id);++ if (!row || row.consumer !== person.id || row.project_id !== parsed.project_id || parsed.project_id !== request.project_id) {++ return result(request, 'refused', { problem: problem(404, 'resources.unknown_resource', `no subscription named ${request.resource} of this person in this project`) });++ }++ const target = ACTIONS[request.action];++ const enabled = Boolean(Number(row.enabled));++ const status = (on) => (on ? 'active' : 'disabled');++ if (enabled === target) return result(request, 'done', { result: { status: status(enabled), changed: false } });++ if (request.dry_run) return result(request, 'done', { result: { dry_run: true, would: target ? 'enable the subscription' : 'disable the subscription' } });++ const updated = await store.setSubscriptionEnabled(row.id, target);++ return result(request, 'done', { result: { status: status(Boolean(Number(updated.enabled))), changed: true } });++ }++++ router.post('/api/v1/resources/control', guard, json, async (req, res, next) => {++ try {++ const request = req.body;++ const check = contracts.resources.checkControlRequest(request || {});++ if (!check.valid) return contracts.http.sendProblem(res, 400, 'control.bad_request', { detail: JSON.stringify(check.errors).slice(0, 500), ctx: req.ov });++ const caller = req.principal.sub;++ const digest = digestOf(request);++ const seen = await db.prepare('SELECT digest, response FROM events_control_log WHERE caller = ? AND project_id = ? AND idempotency_key = ?')++ .get(caller, request.project_id, request.idempotency_key);++ if (seen) {++ if (seen.digest !== digest) return contracts.http.sendProblem(res, 409, 'control.idempotency_conflict', { detail: 'this idempotency_key was used for another request', ctx: req.ov });++ return res.json(typeof seen.response === 'string' ? JSON.parse(seen.response) : seen.response);++ }++ const answer = await decide(request);++ await db.prepare(`INSERT INTO events_control_log (caller, project_id, idempotency_key, digest, response, created_at)++ VALUES (?, ?, ?, ?, ?, ?) ON CONFLICT DO NOTHING`).run(caller, request.project_id, request.idempotency_key, digest, JSON.stringify(answer), now());++ return res.json(answer);++ } catch (err) { return next(err); }++ });++++ return router;++}++++module.exports = { resourceControlRouter, CAPABILITY, digestOf };@@@ -12,6 +12,7 @@ const { readRouter } = require('./api/read');const { policiesRouter } = require('./api/policies');const { placementRouter } = require('./api/placement');const { resourcesRouter } = require('./api/resources');++const { resourceControlRouter } = require('./api/resource-control');const { mountLimits } = require('./limits');const { createLimits } = require('./actor-limits');const { renderHome, renderNotFound, HOME_CSP } = require('./home');@@@ -95,6 +96,8 @@ function createApp({ config, store, auth, keys, worker, realtime, metrics, dnsLo// The authority resource index (ADR-048, capability events.resource.read): the subscriptions Events owns,// for OpenVibe.Services' fan-out. First-party like the routes above; no per-actor limit of its own.app.use(resourcesRouter({ store, auth }));++ // The control API (ADR-048, plan T13 step 10): OpenVibe.Services suspends or resumes a person's subscription.++ app.use(resourceControlRouter({ store, auth, now: () => (clock ? clock.now() : Date.now()) }));// Discovery for the product domain (openvibe.events): robots.txt, sitemap.xml, llms.txt, llms-full.txt.app.use(createDiscoveryRoutes({ config, releaseLastmod: release.full().released_at === release.full().booted_at ? undefined : release.full().released_at }));@@@ -46,6 +46,8 @@ const CAPS = Object.freeze({// so capabilities.check() applies the contracts rule (exact id or a `family.*` grant); the local fallback// below only covers an id the installed contracts do not know yet.resourceRead: 'events.resource.read',++ // The control API (ADR-048, plan T13 step 10; active since contracts 0.168.0): server/api/resource-control.js.++ resourceControl: 'events.resource.control',});// ── Capabilities ───────────────────────────────────────────nnew file mode 100644@@@ -0,0 +1,76 @@++'use strict';++/**++ * Events' control API (ADR-048, plan T13 step 10): OpenVibe.Services suspends and resumes a person's subscription by++ * its OVRN. Only a first-party service holding events.resource.control, only for the subscription's consumer and inside++ * the OVRN's project; delete and other actions refused; dry runs apply nothing; a repeat returns the first answer, the++ * same key with another request is a conflict; every answer is a common.resource-control-result@1.++ */++const assert = require('assert');++const contracts = require('openvibe-contracts');++const { boot, request, serviceToken, userToken, suite } = require('./helpers');++++const t = suite('resource-control');++let h;++const owner = contracts.ids.newId('user');++const stranger = contracts.ids.newId('user');++const project = contracts.ids.newId('project');++const otherProject = contracts.ids.newId('project');++const sub = contracts.ids.newId('subscription');++const token = serviceToken('services', ['events.resource.control']);++let n = 0;++const key = () => `ctl-test-${++n}-${Date.now()}`;++const control = async (body, bearer = token) => request(h.base, 'POST', '/api/v1/resources/control', { token: bearer, body });++const ovrn = (prj = project) => contracts.resources.format({ service: 'events', project_id: prj, type: 'subscription', id: sub });++const ask = (action, extra = {}) => ({ action, resource: ovrn(), project_id: project, idempotency_key: key(), on_behalf_of: { type: 'user', id: owner }, ...extra });++const valid = (body) => {++ const v = contracts.validate('common.resource-control-result@1', body);++ assert.ok(v.valid, JSON.stringify(v.errors));++ return body;++};++const enabled = async () => Boolean(Number((await h.store.getSubscription(sub)).enabled));++++t('boot with a person\'s subscription in a project', async () => {++ h = await boot();++ await h.store.createSubscription({ id: sub, consumer: owner, topicPattern: 'live.stream.*', endpoint: 'https://consumer.openvibe.network/hook', secret: 'whsec_test', projectId: project, env: 'production' });++});++++t('only a first-party service with the capability may call it', async () => {++ assert.strictEqual((await control(ask('suspend'), serviceToken('services', ['events.resource.read']))).status, 403);++ assert.strictEqual((await control(ask('suspend'), userToken(owner))).status, 401, 'a user JWT is not a service token');++ assert.strictEqual((await control({ action: 'suspend' })).status, 400, 'not a control request');++});++++t('suspend, a dry run first, then a repeat answered from the log', async () => {++ const dry = valid((await control(ask('suspend', { dry_run: true }))).body);++ assert.deepStrictEqual([dry.state, dry.result.dry_run], ['done', true]);++ assert.strictEqual(await enabled(), true, 'a dry run applies nothing');++ const body = ask('suspend');++ const r = await control(body);++ assert.strictEqual(r.status, 200);++ valid(r.body);++ assert.deepStrictEqual([r.body.state, r.body.result.status, r.body.result.changed], ['done', 'disabled', true]);++ assert.strictEqual(await enabled(), false);++ assert.deepStrictEqual((await control(body)).body, r.body, 'the first answer, again');++ assert.strictEqual((await control({ ...body, action: 'resume' })).status, 409);++ const again = valid((await control(ask('suspend'))).body);++ assert.deepStrictEqual([again.result.status, again.result.changed], ['disabled', false]);++});++++t('resume; delete, another person and another project are refused', async () => {++ const r = valid((await control(ask('resume'))).body);++ assert.deepStrictEqual([r.state, r.result.status], ['done', 'active']);++ assert.strictEqual(await enabled(), true);++ const del = valid((await control(ask('delete'))).body);++ assert.deepStrictEqual([del.state, del.problem.code], ['refused', 'control.unsupported_action']);++ const theirs = valid((await control(ask('suspend', { on_behalf_of: { type: 'user', id: stranger } }))).body);++ assert.deepStrictEqual([theirs.state, theirs.problem.code], ['refused', 'resources.unknown_resource']);++ const elsewhere = valid((await control({ ...ask('suspend'), resource: ovrn(otherProject), project_id: otherProject })).body);++ assert.strictEqual(elsewhere.problem.code, 'resources.unknown_resource');++ const nobody = valid((await control(ask('suspend', { on_behalf_of: undefined }))).body);++ assert.strictEqual(nobody.problem.code, 'control.no_person');++ assert.strictEqual(await enabled(), true);++});++++t('done', async () => { await h.stop(); });++++t.run();
@@ -151,6 +151,12 @@ node scripts/events-archive.js restore --month 2026-01 [--hold-days 30]
@@ -151,6 +151,12 @@ node scripts/events-archive.js restore --month 2026-01 [--hold-days 30]
- OVRN: present exactly when `openvibe-contracts`' `contracts.resources.nameOf` composes one — `ovrn:events:<prj_…>:subscription/sub_…` for a project-scoped subscription. A project-less subscription is a first-party one (a service consumer with no project): it has no project segment, so no name, and only the unscoped caller sees it at all. An event is not a resource (`evt_` is what `common.resource-name@1` refuses by design) and a project's queue is a derived carrier class, not a stored row, so `events.queue` joins the index only when Events stores a queue row.
- OVRN: present exactly when `openvibe-contracts`' `contracts.resources.nameOf` composes one — `ovrn:events:<prj_…>:subscription/sub_…` for a project-scoped subscription. A project-less subscription is a first-party one (a service consumer with no project): it has no project segment, so no name, and only the unscoped caller sees it at all. An event is not a resource (`evt_` is what `common.resource-name@1` refuses by design) and a project's queue is a derived carrier class, not a stored row, so `events.queue` joins the index only when Events stores a queue row.
- Errors: `400 resources.bad_query` for a value the index cannot honour (a project that is not a `prj_` id, a `limit` outside 1-1000, a cursor it did not issue) and `404 resources.unknown_resource` for an OVRN that names nothing it composes, both problem+json. Answers carry `Cache-Control: private, max-age=60`. The capability is `first-party`: a developer app's token is a `403`, as on every first-party route, and the two routes are not per-actor limited — they are one fan-out call each, not a browser's traffic.
- Errors: `400 resources.bad_query` for a value the index cannot honour (a project that is not a `prj_` id, a `limit` outside 1-1000, a cursor it did not issue) and `404 resources.unknown_resource` for an OVRN that names nothing it composes, both problem+json. Answers carry `Cache-Control: private, max-age=60`. The capability is `first-party`: a developer app's token is a `403`, as on every first-party route, and the two routes are not per-actor limited — they are one fan-out call each, not a browser's traffic.
+**Control** (ADR-048, plan T13 step 10, contracts 0.168.0, `server/api/resource-control.js`): `POST /api/v1/resources/control` takes a `common.resource-control-request@1` from a first-party service holding `events.resource.control` (OpenVibe.Services, acting for the person in `on_behalf_of`), naming a subscription by its OVRN.
+- **Actions**: `suspend` disables it (nothing is delivered until it resumes; the events it matches are kept and can be replayed) and `resume` enables it. Any other action, delete included, answers `control.unsupported_action`.
+- **Who**: only the subscription's consumer, only inside the OVRN's project; anyone else's is `resources.unknown_resource`.
+- **Dry runs and repeats**: `dry_run` decides without applying. The first answer per (caller, project, idempotency key) is kept in `events_control_log` (migration 0008) and returned for a repeat; another request under the same key is `409 control.idempotency_conflict`.
+- **Test**: `test/resource-control.test.js`.
+
## The fabric: carriers, the planner and explain
## The fabric: carriers, the planner and explain
How an event type is carried is its delivery policy (`events.delivery-policy@1`, [ADR-042](https://github.com/OpenVibers/OpenVibe.Contracts/blob/main/docs/adr/ADR-042-events-fabric.md) decision 1). The carrier class — TOPIC (fan-out, no ack), QUEUE (competing consumers, ack) or STREAM (retained, ordered per key) — is a derived, internal label, never in the envelope. PostgreSQL stays the record: every event and every delivery is committed before any carrier sees it, and the worker's lease is the only thing that makes a send happen. A publish commits, then the chosen carrier is signalled; a lost or duplicated signal changes latency, never the outcome.
How an event type is carried is its delivery policy (`events.delivery-policy@1`, [ADR-042](https://github.com/OpenVibers/OpenVibe.Contracts/blob/main/docs/adr/ADR-042-events-fabric.md) decision 1). The carrier class — TOPIC (fan-out, no ack), QUEUE (competing consumers, ack) or STREAM (retained, ordered per key) — is a derived, internal label, never in the envelope. PostgreSQL stays the record: every event and every delivery is committed before any carrier sees it, and the worker's lease is the only thing that makes a send happen. A publish commits, then the chosen carrier is signalled; a lost or duplicated signal changes latency, never the outcome.
new file mode 100644
new file mode 100644
@@ -0,0 +1,12 @@
@@ -0,0 +1,12 @@
+-- phase: expand
+-- ADR-048 control operations (plan T13 step 10, server/api/resource-control.js): the first answer to each
+-- (caller, project, idempotency_key), returned again for a repeat so a retried control request never applies twice.
+CREATE TABLE events_control_log (
+ caller text COLLATE "C" NOT NULL, -- svc:… that sent it
+ project_id text COLLATE "C" NOT NULL,
+ idempotency_key text COLLATE "C" NOT NULL,
+ digest text COLLATE "C" NOT NULL, -- sha256 of the request without its delivery fields
+ response jsonb NOT NULL, -- the common.resource-control-result@1 answered
+ created_at bigint NOT NULL,
+ PRIMARY KEY (caller, project_id, idempotency_key)
+);
@@ -12,7 +12,7 @@
@@ -12,7 +12,7 @@
"dotenv": "^16.4.7",
"dotenv": "^16.4.7",
"express": "^4.21.2",
"express": "^4.21.2",
"iovalkey": "^0.4.0",
"iovalkey": "^0.4.0",
- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.143.0",
+ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.168.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.44.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.44.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.1.3",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.1.3",
"pg": "^8.23.0"
"pg": "^8.23.0"
@@ -828,9 +828,9 @@
@@ -828,9 +828,9 @@
}
}
},
},
"node_modules/openvibe-contracts": {
"node_modules/openvibe-contracts": {
- "version": "0.143.0",
- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.143.0",
- "integrity": "sha512-Cdbf+9RoPAUBibpM2zXQvtsiKMugb55dKubTt8pFZvAmxsLtGvX+zaC++yL1vOSub8FX3+GgqcGtoJS6tklrUg==",
+ "version": "0.168.0",
+ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.168.0",
+ "integrity": "sha512-v59y6iRgSSHozx98grx5CRwPHN8UCj0L88HHJpJVBw/Cnzd2QzvcQQZIPlOMqH9K1QEEYeFUIJMiCIMJez0b5A==",
"license": "AGPL-3.0-only",
"license": "AGPL-3.0-only",
"dependencies": {
"dependencies": {
"ajv": "^8.17.1",
"ajv": "^8.17.1",
@@ -24,7 +24,7 @@
@@ -24,7 +24,7 @@
"dotenv": "^16.4.7",
"dotenv": "^16.4.7",
"express": "^4.21.2",
"express": "^4.21.2",
"iovalkey": "^0.4.0",
"iovalkey": "^0.4.0",
- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.143.0",
+ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.168.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.44.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.44.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.1.3",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.1.3",
"pg": "^8.23.0"
"pg": "^8.23.0"
new file mode 100644
new file mode 100644
@@ -0,0 +1,92 @@
@@ -0,0 +1,92 @@
+'use strict';
+/**
+ * Events' control API (ADR-048, plan T13 step 10): POST /api/v1/resources/control — one
+ * common.resource-control-request@1 from a first-party service (OpenVibe.Services) holding events.resource.control,
+ * acting for the person in on_behalf_of, on an events.subscription named by its OVRN. The answer is a
+ * common.resource-control-result@1 (200 for every outcome; a malformed request is a 400 problem).
+ *
+ * suspend disable the subscription: nothing is delivered to it until it resumes (events it matches are kept as
+ * they always are, and can be replayed)
+ * resume enable it again
+ *
+ * Any other action is refused `control.unsupported_action` (Events has no deletion of a subscription). Only the
+ * subscription's consumer may act on it (on_behalf_of names them; another person's subscription is refused as not
+ * found, never confirmed to exist), and only inside the project the OVRN names. `dry_run` decides and explains without
+ * applying. The first answer per (caller, project, idempotency_key) is kept in events_control_log and returned for a
+ * repeat; the same key with another request is refused `control.idempotency_conflict`.
+ */
+const crypto = require('crypto');
+const express = require('express');
+const contracts = require('openvibe-contracts');
+
+const SERVICE = 'events';
+const CAPABILITY = 'events.resource.control';
+const ACTIONS = { suspend: false, resume: true };
+const USER_SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
+
+function canonical(v) {
+ if (Array.isArray(v)) return `[${v.map(canonical).join(',')}]`;
+ if (v && typeof v === 'object') return `{${Object.keys(v).sort().filter((k) => v[k] !== undefined).map((k) => `${JSON.stringify(k)}:${canonical(v[k])}`).join(',')}}`;
+ return JSON.stringify(v);
+}
+const digestOf = (request) => crypto.createHash('sha256')
+ .update(canonical({ ...request, idempotency_key: undefined, confirmation_id: undefined, trace_id: undefined })).digest('hex');
+
+function resourceControlRouter({ store, auth, now = () => Date.now() }) {
+ const router = express.Router();
+ const guard = auth.requireCap(CAPABILITY, { requireService: true });
+ const json = express.json({ limit: '16kb' });
+ const db = store.db;
+ const iso = () => new Date(now()).toISOString();
+ const result = (request, state, extra = {}) => ({ action: request.action, ...(request.resource ? { resource: request.resource } : {}), state, at: iso(), ...extra });
+ const problem = (status, code, detail) => ({ type: `https://openvibe.network/problems/${code}`, title: code, status, code, detail });
+
+ async function decide(request) {
+ const parsed = contracts.resources.parse(request.resource || '');
+ if (!parsed || parsed.service !== SERVICE || parsed.type !== 'subscription') {
+ return result(request, 'refused', { problem: problem(404, 'resources.unknown_resource', `no subscription named ${request.resource || request.resource_kind}`) });
+ }
+ if (!(request.action in ACTIONS)) {
+ return result(request, 'refused', { problem: problem(422, 'control.unsupported_action', `a subscription takes suspend and resume, not ${request.action}`) });
+ }
+ const person = request.on_behalf_of;
+ if (!person || person.type !== 'user' || !USER_SUBJECT_RE.test(String(person.id || ''))) {
+ return result(request, 'refused', { problem: problem(403, 'control.no_person', 'on_behalf_of must name the person whose subscription it is') });
+ }
+ const row = await store.getSubscription(parsed.id);
+ if (!row || row.consumer !== person.id || row.project_id !== parsed.project_id || parsed.project_id !== request.project_id) {
+ return result(request, 'refused', { problem: problem(404, 'resources.unknown_resource', `no subscription named ${request.resource} of this person in this project`) });
+ }
+ const target = ACTIONS[request.action];
+ const enabled = Boolean(Number(row.enabled));
+ const status = (on) => (on ? 'active' : 'disabled');
+ if (enabled === target) return result(request, 'done', { result: { status: status(enabled), changed: false } });
+ if (request.dry_run) return result(request, 'done', { result: { dry_run: true, would: target ? 'enable the subscription' : 'disable the subscription' } });
+ const updated = await store.setSubscriptionEnabled(row.id, target);
+ return result(request, 'done', { result: { status: status(Boolean(Number(updated.enabled))), changed: true } });
+ }
+
+ router.post('/api/v1/resources/control', guard, json, async (req, res, next) => {
+ try {
+ const request = req.body;
+ const check = contracts.resources.checkControlRequest(request || {});
+ if (!check.valid) return contracts.http.sendProblem(res, 400, 'control.bad_request', { detail: JSON.stringify(check.errors).slice(0, 500), ctx: req.ov });
+ const caller = req.principal.sub;
+ const digest = digestOf(request);
+ const seen = await db.prepare('SELECT digest, response FROM events_control_log WHERE caller = ? AND project_id = ? AND idempotency_key = ?')
+ .get(caller, request.project_id, request.idempotency_key);
+ if (seen) {
+ if (seen.digest !== digest) return contracts.http.sendProblem(res, 409, 'control.idempotency_conflict', { detail: 'this idempotency_key was used for another request', ctx: req.ov });
+ return res.json(typeof seen.response === 'string' ? JSON.parse(seen.response) : seen.response);
+ }
+ const answer = await decide(request);
+ await db.prepare(`INSERT INTO events_control_log (caller, project_id, idempotency_key, digest, response, created_at)
+ VALUES (?, ?, ?, ?, ?, ?) ON CONFLICT DO NOTHING`).run(caller, request.project_id, request.idempotency_key, digest, JSON.stringify(answer), now());
+ return res.json(answer);
+ } catch (err) { return next(err); }
+ });
+
+ return router;
+}
+
+module.exports = { resourceControlRouter, CAPABILITY, digestOf };
@@ -12,6 +12,7 @@ const { readRouter } = require('./api/read');
@@ -12,6 +12,7 @@ const { readRouter } = require('./api/read');
const { policiesRouter } = require('./api/policies');
const { policiesRouter } = require('./api/policies');
const { placementRouter } = require('./api/placement');
const { placementRouter } = require('./api/placement');
const { resourcesRouter } = require('./api/resources');
const { resourcesRouter } = require('./api/resources');
+const { resourceControlRouter } = require('./api/resource-control');
const { mountLimits } = require('./limits');
const { mountLimits } = require('./limits');
const { createLimits } = require('./actor-limits');
const { createLimits } = require('./actor-limits');
const { renderHome, renderNotFound, HOME_CSP } = require('./home');
const { renderHome, renderNotFound, HOME_CSP } = require('./home');
@@ -95,6 +96,8 @@ function createApp({ config, store, auth, keys, worker, realtime, metrics, dnsLo
@@ -95,6 +96,8 @@ function createApp({ config, store, auth, keys, worker, realtime, metrics, dnsLo
// The authority resource index (ADR-048, capability events.resource.read): the subscriptions Events owns,
// The authority resource index (ADR-048, capability events.resource.read): the subscriptions Events owns,
// for OpenVibe.Services' fan-out. First-party like the routes above; no per-actor limit of its own.
// for OpenVibe.Services' fan-out. First-party like the routes above; no per-actor limit of its own.
app.use(resourcesRouter({ store, auth }));
app.use(resourcesRouter({ store, auth }));
+ // The control API (ADR-048, plan T13 step 10): OpenVibe.Services suspends or resumes a person's subscription.
+ app.use(resourceControlRouter({ store, auth, now: () => (clock ? clock.now() : Date.now()) }));
// Discovery for the product domain (openvibe.events): robots.txt, sitemap.xml, llms.txt, llms-full.txt.
// Discovery for the product domain (openvibe.events): robots.txt, sitemap.xml, llms.txt, llms-full.txt.
app.use(createDiscoveryRoutes({ config, releaseLastmod: release.full().released_at === release.full().booted_at ? undefined : release.full().released_at }));
app.use(createDiscoveryRoutes({ config, releaseLastmod: release.full().released_at === release.full().booted_at ? undefined : release.full().released_at }));
@@ -46,6 +46,8 @@ const CAPS = Object.freeze({
@@ -46,6 +46,8 @@ const CAPS = Object.freeze({
// so capabilities.check() applies the contracts rule (exact id or a `family.*` grant); the local fallback
// so capabilities.check() applies the contracts rule (exact id or a `family.*` grant); the local fallback
// below only covers an id the installed contracts do not know yet.
// below only covers an id the installed contracts do not know yet.
resourceRead: 'events.resource.read',
resourceRead: 'events.resource.read',
+ // The control API (ADR-048, plan T13 step 10; active since contracts 0.168.0): server/api/resource-control.js.
+ resourceControl: 'events.resource.control',
});
});
// ── Capabilities ───────────────────────────────────────────
// ── Capabilities ───────────────────────────────────────────
new file mode 100644
new file mode 100644
@@ -0,0 +1,76 @@
@@ -0,0 +1,76 @@
+'use strict';
+/**
+ * Events' control API (ADR-048, plan T13 step 10): OpenVibe.Services suspends and resumes a person's subscription by
+ * its OVRN. Only a first-party service holding events.resource.control, only for the subscription's consumer and inside
+ * the OVRN's project; delete and other actions refused; dry runs apply nothing; a repeat returns the first answer, the
+ * same key with another request is a conflict; every answer is a common.resource-control-result@1.
+ */
+const assert = require('assert');
+const contracts = require('openvibe-contracts');
+const { boot, request, serviceToken, userToken, suite } = require('./helpers');
+
+const t = suite('resource-control');
+let h;
+const owner = contracts.ids.newId('user');
+const stranger = contracts.ids.newId('user');
+const project = contracts.ids.newId('project');
+const otherProject = contracts.ids.newId('project');
+const sub = contracts.ids.newId('subscription');
+const token = serviceToken('services', ['events.resource.control']);
+let n = 0;
+const key = () => `ctl-test-${++n}-${Date.now()}`;
+const control = async (body, bearer = token) => request(h.base, 'POST', '/api/v1/resources/control', { token: bearer, body });
+const ovrn = (prj = project) => contracts.resources.format({ service: 'events', project_id: prj, type: 'subscription', id: sub });
+const ask = (action, extra = {}) => ({ action, resource: ovrn(), project_id: project, idempotency_key: key(), on_behalf_of: { type: 'user', id: owner }, ...extra });
+const valid = (body) => {
+ const v = contracts.validate('common.resource-control-result@1', body);
+ assert.ok(v.valid, JSON.stringify(v.errors));
+ return body;
+};
+const enabled = async () => Boolean(Number((await h.store.getSubscription(sub)).enabled));
+
+t('boot with a person\'s subscription in a project', async () => {
+ h = await boot();
+ await h.store.createSubscription({ id: sub, consumer: owner, topicPattern: 'live.stream.*', endpoint: 'https://consumer.openvibe.network/hook', secret: 'whsec_test', projectId: project, env: 'production' });
+});
+
+t('only a first-party service with the capability may call it', async () => {
+ assert.strictEqual((await control(ask('suspend'), serviceToken('services', ['events.resource.read']))).status, 403);
+ assert.strictEqual((await control(ask('suspend'), userToken(owner))).status, 401, 'a user JWT is not a service token');
+ assert.strictEqual((await control({ action: 'suspend' })).status, 400, 'not a control request');
+});
+
+t('suspend, a dry run first, then a repeat answered from the log', async () => {
+ const dry = valid((await control(ask('suspend', { dry_run: true }))).body);
+ assert.deepStrictEqual([dry.state, dry.result.dry_run], ['done', true]);
+ assert.strictEqual(await enabled(), true, 'a dry run applies nothing');
+ const body = ask('suspend');
+ const r = await control(body);
+ assert.strictEqual(r.status, 200);
+ valid(r.body);
+ assert.deepStrictEqual([r.body.state, r.body.result.status, r.body.result.changed], ['done', 'disabled', true]);
+ assert.strictEqual(await enabled(), false);
+ assert.deepStrictEqual((await control(body)).body, r.body, 'the first answer, again');
+ assert.strictEqual((await control({ ...body, action: 'resume' })).status, 409);
+ const again = valid((await control(ask('suspend'))).body);
+ assert.deepStrictEqual([again.result.status, again.result.changed], ['disabled', false]);
+});
+
+t('resume; delete, another person and another project are refused', async () => {
+ const r = valid((await control(ask('resume'))).body);
+ assert.deepStrictEqual([r.state, r.result.status], ['done', 'active']);
+ assert.strictEqual(await enabled(), true);
+ const del = valid((await control(ask('delete'))).body);
+ assert.deepStrictEqual([del.state, del.problem.code], ['refused', 'control.unsupported_action']);
+ const theirs = valid((await control(ask('suspend', { on_behalf_of: { type: 'user', id: stranger } }))).body);
+ assert.deepStrictEqual([theirs.state, theirs.problem.code], ['refused', 'resources.unknown_resource']);
+ const elsewhere = valid((await control({ ...ask('suspend'), resource: ovrn(otherProject), project_id: otherProject })).body);
+ assert.strictEqual(elsewhere.problem.code, 'resources.unknown_resource');
+ const nobody = valid((await control(ask('suspend', { on_behalf_of: undefined }))).body);
+ assert.strictEqual(nobody.problem.code, 'control.no_person');
+ assert.strictEqual(await enabled(), true);
+});
+
+t('done', async () => { await h.stop(); });
+
+t.run();