openvibers/openrestream Commits

OpenRestream: a definition's owner suspends and resumes it from OpenVibe.Services (openre.resource.control, contracts 0.176.0) (#69)

OP OpenVibers committed 29 days ago · 5368543289 · 6 files, +172 −5

A restream definition's owner can suspend and resume it from OpenVibe.Services ("Your resources"), through OpenRestream's control API (contracts 0.176.0, ADR-048 amendment: person-owned resources).

- **`POST /api/v1/resources/control`** (`server/registry/resource-control.js`, capability `openre.resource.control`, first-party `svc:` tokens only: the shared guard's owner pass-through does not apply here). It takes one `common.owned-resource-control-request@1` naming an `openre.stream` by its definition id and its owner, and answers `common.resource-control-result@1` without `resource`:
  - `suspend`: state `disabled`, through the same `store.definitions.update` as the owner's edit. New ingest is refused (`stream_disabled`); a session already live keeps going until it ends.
  - `resume`: state `active` again.
- **Refusals:**
  - another action: `control.unsupported_action`;
  - another person's definition, an archived one, or another kind: refused as unknown, never confirmed to exist;
  - a project-scoped request: 400.
- **dry_run** explains without applying.
- **Idempotency:** the first answer per (caller, key) is kept in `openre_control_log` (migration 0004, expand). The same key with another request gets 409.
- **Mounting:** the control router sits before the index (`openre.resource.read`). The public vhost already answers 404 for `/api/v1/resources*`.
- **Pin:** `openvibe-contracts` v0.143.0 → v0.176.0.

**Tests:** `test/resource-control.test.js`:
- a dry run, suspend and its repeat, the key conflict, suspend again with no change, resume;
- another person's definition, another action or kind, a person acting for someone else and a project request all refused;
- a read-only service token, the owner's own token and no token all refused.

**Checks:** the full suite passes 29/33. The 4 skipped need PostgreSQL (`npm run test:pg`), as before.

Changes

ddiff --git a/migrations/0004_control_log.sql b/migrations/0004_control_log.sql
nnew file mode 100644
iindex 0000000..a4f585f
---- /dev/null
++++ b/migrations/0004_control_log.sql
@@@ -0,0 +1,13 @@
++-- phase: expand
++-- ADR-048 amendment 2026-10-15 (server/registry/resource-control.js): the first answer to each control request
++-- OpenVibe.Services sent for a person's restream definition, per (caller, idempotency_key), so a retry is never
++-- applied twice. Never edited after it runs; a further change is a new migration.
++CREATE TABLE openre_control_log (
++ caller text COLLATE "C" NOT NULL, -- the service that asked (svc:services)
++ idempotency_key text COLLATE "C" NOT NULL,
++ owner_subject text COLLATE "C" NOT NULL, -- the definition's owner the request named
++ digest text COLLATE "C" NOT NULL, -- the request without its delivery fields
++ response text NOT NULL,
++ created_at bigint NOT NULL,
++ PRIMARY KEY (caller, idempotency_key)
++);
ddiff --git a/package-lock.json b/package-lock.json
iindex b56472c..3b119ea 100644
---- a/package-lock.json
++++ b/package-lock.json
@@@ -15,7 +15,7 @@
"iovalkey": "^0.4.0",
"mediasoup": "^3.14.14",
"node-media-server": "2.7.4",
-- "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.176.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",
@@@ -1638,9 +1638,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.176.0",
++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.176.0",
++ "integrity": "sha512-COLEoRbjs6he+Tb2LvazezNWvHzB6CEWflebrgEyMaTd1hhE5zWJAAL/mdvR0LOzVDAX6vb1GPdTw83AulLfCg==",
"license": "AGPL-3.0-only",
"dependencies": {
"ajv": "^8.17.1",
ddiff --git a/package.json b/package.json
iindex f380cb7..d67cd7b 100644
---- a/package.json
++++ b/package.json
@@@ -31,7 +31,7 @@
"iovalkey": "^0.4.0",
"mediasoup": "^3.14.14",
"node-media-server": "2.7.4",
-- "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.176.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",
ddiff --git a/server/app.js b/server/app.js
iindex e1c4bd4..908c003 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -135,6 +135,8 @@ function createApp({ rt, auth, keys, log = console, fetchImpl }) {
// Per-actor limits on a signed-in person's writes (server/api/actor-limits.js; roadmap WS-R task 4).
const actorLimits = require('./api/actor-limits').createOpenReActorLimits({ registry: metrics.registry, valkey });
app.use('/api/v1', apiJson, auth.middleware({ services: true }), actorLimits);
++ // Control first (openre.resource.control: suspend or resume a person's definition), then the index.
++ app.use('/api/v1/resources/control', require('./registry/resource-control').router({ db, store, guard: auth.guard('openre.resource.control') }));
app.use('/api/v1/resources', resourceIndex.router({ db, guard: auth.guard('openre.resource.read') }));
app.use('/api/v1', v1.router);
ddiff --git a/server/registry/resource-control.js b/server/registry/resource-control.js
nnew file mode 100644
iindex 0000000..733a383
---- /dev/null
++++ b/server/registry/resource-control.js
@@@ -0,0 +1,86 @@
++'use strict';
++/**
++ * OpenRestream's control API for person-owned restream definitions (ADR-048 amendment 2026-10-15, step 8 phase 2;
++ * capability openre.resource.control): POST /api/v1/resources/control takes one common.owned-resource-control-request@1
++ * from a first-party service (OpenVibe.Services) naming an openre.stream by its definition id and its owner, and
++ * answers a common.resource-control-result@1 (200 for every outcome, without `resource`: definitions have no OVRN; a
++ * malformed request is a 400 problem).
++ *
++ * suspend state disabled, as the owner's edit does: new ingest is refused (stream_disabled); a session already
++ * live keeps going until it ends
++ * resume state active again
++ *
++ * Any other action is refused `control.unsupported_action`. Only the owner OpenRestream holds may be named: a definition
++ * of another person (or an archived one) is refused as unknown, never confirmed to exist. dry_run decides and explains
++ * without applying. The first answer per (caller, idempotency_key) is kept in openre_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 CAPABILITY = 'openre.resource.control';
++const KIND = 'openre.stream';
++const TARGET = { suspend: 'disabled', resume: 'active' };
++
++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 router({ db, store, guard, now = () => Date.now() }) {
++ const r = express.Router();
++ const problemOf = (status, code, detail) => ({ type: `https://openvibe.network/problems/${code}`, title: code, status, code, detail });
++ const send = (req, res, status, code, detail) => contracts.http.sendProblem(res, status, code, { detail, ctx: req.ov });
++ const result = (request, state, extra = {}) => ({ action: request.action, state, at: new Date(now()).toISOString(), ...extra });
++
++ async function decide(request) {
++ if (request.owned.kind !== KIND) return result(request, 'refused', { problem: problemOf(404, 'resources.unknown_resource', `no stream ${request.owned.kind} ${request.owned.id}`) });
++ if (!(request.action in TARGET)) return result(request, 'refused', { problem: problemOf(422, 'control.unsupported_action', `a stream takes suspend and resume, not ${request.action}`) });
++ const d = await store.definitions.get(request.owned.id);
++ if (!d || d.state === 'archived' || d.owner_subject !== request.owner.id) {
++ return result(request, 'refused', { problem: problemOf(404, 'resources.unknown_resource', `no stream ${request.owned.id} of this person`) });
++ }
++ const target = TARGET[request.action];
++ if (d.state === target) return result(request, 'done', { result: { state: d.state, changed: false } });
++ if (request.dry_run) return result(request, 'done', { result: { dry_run: true, would: `make the stream ${target}` } });
++ const updated = await store.definitions.update(d.id, { state: target });
++ return result(request, 'done', { result: { state: updated.state, changed: true } });
++ }
++
++ // First-party only: the shared guard lets owners through on ordinary routes; control needs a service.
++ r.use((req, res, next) => {
++ const caller = req.caller || { kind: 'anonymous' };
++ if (caller.kind === 'service' && /^svc:/.test(String(caller.sub || ''))) return next();
++ if (caller.kind === 'anonymous') return send(req, res, 401, 'token.missing', 'call with a service token');
++ return send(req, res, 403, 'capability.denied', `${CAPABILITY} is first-party and requires a service token`);
++ });
++ r.post('/', guard, express.json({ limit: '16kb' }), async (req, res, next) => {
++ try {
++ const request = req.body;
++ if (!request || typeof request !== 'object' || request.owned === undefined) {
++ // OpenRestream holds no project resources: a project-scoped request names nothing here.
++ return send(req, res, 400, 'control.bad_request', 'restream definitions are person-owned: send common.owned-resource-control-request@1');
++ }
++ const check = contracts.resources.checkControlRequest(request);
++ if (!check.valid) return send(req, res, 400, 'control.bad_request', JSON.stringify(check.errors).slice(0, 500));
++ const caller = req.caller.sub;
++ const digest = digestOf(request);
++ const seen = await db.prepare('SELECT digest, response FROM openre_control_log WHERE caller = ? AND idempotency_key = ?').get(caller, request.idempotency_key);
++ if (seen) {
++ if (seen.digest !== digest) return send(req, res, 409, 'control.idempotency_conflict', 'this idempotency_key was used for another request');
++ return res.json(typeof seen.response === 'string' ? JSON.parse(seen.response) : seen.response);
++ }
++ const answer = await decide(request);
++ await db.prepare(`INSERT INTO openre_control_log (caller, idempotency_key, owner_subject, digest, response, created_at)
++ VALUES (?, ?, ?, ?, ?, ?) ON CONFLICT DO NOTHING`).run(caller, request.idempotency_key, request.owner.id, digest, JSON.stringify(answer), now());
++ return res.json(answer);
++ } catch (err) { return next(err); }
++ });
++ return r;
++}
++
++module.exports = { router, CAPABILITY, digestOf };
ddiff --git a/test/resource-control.test.js b/test/resource-control.test.js
nnew file mode 100644
iindex 0000000..55c6b26
---- /dev/null
++++ b/test/resource-control.test.js
@@@ -0,0 +1,66 @@
++'use strict';
++// ADR-048 amendment 2026-10-15: OpenVibe.Services suspends and resumes a person's restream definition through
++// OpenRestream's control API (openre.resource.control, common.owned-resource-control-request@1).
++const assert = require('assert');
++const contracts = require('openvibe-contracts');
++const { bootApi, request, serviceToken, userToken, suite, OWNER, OTHER } = require('./helpers');
++
++const t = suite('resource-control');
++const services = serviceToken('services', ['openre.resource.control']);
++let api, mine, theirs;
++let n = 0;
++const send = (body, token = services) => request(api.base, 'POST', '/api/v1/resources/control', { token, body });
++const req = (action, over = {}) => ({
++ action, owned: { kind: 'openre.stream', id: mine.id }, owner: { type: 'user', id: OWNER },
++ on_behalf_of: { type: 'user', id: OWNER }, idempotency_key: `idem_stream_${++n}_${Date.now()}`, ...over,
++});
++const stateOf = async (id) => (await api.rt.store.definitions.get(id)).state;
++
++t('suspend disables (a dry run first); a repeat answers the first result; resume makes it active', async () => {
++ api = await bootApi();
++ mine = (await api.rt.store.definitions.create({ owner_subject: OWNER, title: 'My stream' })).definition;
++ theirs = (await api.rt.store.definitions.create({ owner_subject: OTHER, title: 'Their stream' })).definition;
++ let r = await send(req('suspend', { dry_run: true }));
++ assert.strictEqual(r.status, 200, JSON.stringify(r.body));
++ assert.deepStrictEqual([r.body.state, r.body.result.dry_run], ['done', true]);
++ assert.strictEqual(await stateOf(mine.id), 'active');
++ const first = req('suspend');
++ r = await send(first);
++ assert.ok(contracts.resources.checkControlResult(r.body).valid, JSON.stringify(r.body));
++ assert.deepStrictEqual([r.body.state, r.body.result, r.body.resource], ['done', { state: 'disabled', changed: true }, undefined]);
++ assert.strictEqual(await stateOf(mine.id), 'disabled');
++ assert.deepStrictEqual((await send(first)).body, r.body, 'the same key answers the first result');
++ const conflict = await send({ ...first, dry_run: true });
++ assert.deepStrictEqual([conflict.status, conflict.body.code], [409, 'control.idempotency_conflict']);
++ assert.deepStrictEqual((await send(req('suspend'))).body.result, { state: 'disabled', changed: false });
++ r = await send(req('resume'));
++ assert.deepStrictEqual(r.body.result, { state: 'active', changed: true });
++ assert.strictEqual(await stateOf(mine.id), 'active');
++});
++
++t('only the owner OpenRestream holds; other actions and kinds are refused', async () => {
++ let r = await send(req('suspend', { owned: { kind: 'openre.stream', id: theirs.id } }));
++ assert.deepStrictEqual([r.body.state, r.body.problem.code], ['refused', 'resources.unknown_resource'], 'another person\'s stream is not confirmed to exist');
++ assert.strictEqual(await stateOf(theirs.id), 'active');
++ r = await send(req('delete'));
++ assert.strictEqual(r.body.problem.code, 'control.unsupported_action');
++ r = await send(req('stop', { owned: { kind: 'openre.session', id: 'ses_x' } }));
++ assert.strictEqual(r.body.problem.code, 'resources.unknown_resource');
++ r = await send(req('suspend', { on_behalf_of: { type: 'user', id: OTHER } }));
++ assert.deepStrictEqual([r.status, r.body.code], [400, 'control.bad_request'], 'a person acts only on what they own');
++ r = await send({ action: 'suspend', resource: 'ovrn:openre:prj_01J8ZQ4Y7N3M2K1H0G9F8E7D6C:stream/str_01J8ZQ4Y7N3M2K1H0G9F8E7D6C', project_id: 'prj_01J8ZQ4Y7N3M2K1H0G9F8E7D6C', idempotency_key: 'idem_project_1' });
++ assert.deepStrictEqual([r.status, r.body.code], [400, 'control.bad_request'], 'OpenRestream holds no project resources');
++});
++
++t('first-party only, with openre.resource.control', async () => {
++ let r = await send(req('suspend'), serviceToken('services', ['openre.resource.read']));
++ assert.strictEqual(r.status, 403);
++ r = await send(req('suspend'), userToken({ subjectId: OWNER }));
++ assert.strictEqual(r.status, 403, 'the owner\'s own token goes to the ordinary edit, not here');
++ r = await send(req('suspend'), null);
++ assert.strictEqual(r.status, 401);
++ assert.strictEqual(await stateOf(mine.id), 'active');
++ await api.close();
++});
++
++t.run();