Watch: the control API — Services suspends, resumes and deletes a person's watch by its OVRN (ADR-048) (#57)

ddiff --git a/README.md b/README.md
iindex 0438b27..9042f97 100644
---- a/README.md
++++ b/README.md
@@@ -296,6 +296,7 @@ an app or module may not.
  | `watch.observation.read` | `GET /api/v1/watches/:id/observations?before=&limit=`, `GET /api/v1/watches/:id/checks?before=&limit=` |
  | `watch.check.run` | `POST /api/v1/watches/:id/check` (404 unknown, 409 `watch.busy` while one runs, 409 `watch.disabled` for a paused watch, 422 `watch.carrier_unavailable`, or 422 `watch.event_awaited` for an event source) |
  | `watch.resource.read` | `GET /api/v1/resources?project=&kind=&cursor=&limit=`, `GET /api/v1/resources/:ovrn` (first-party service token only) |
++| `watch.resource.control` | `POST /api/v1/resources/control` (first-party service token only; ADR-048, [server/api/resource-control.js](server/api/resource-control.js)): a `common.resource-control-request@1` for a watch from OpenVibe.Services acting for its owner (`on_behalf_of`) — `suspend` (pause), `resume`, `delete`; other actions refused `control.unsupported_action`; only the owner's watch in the OVRN's project (otherwise `resources.unknown_resource`); `dry_run` applies nothing; the first answer per (caller, project, `idempotency_key`) is kept (`watch_control_log`, migration 0009) and a key reused for another request is 409. Loopback only: the public vhost 404s `/api/v1/resources*`. |
  
  The loopback-only authority index lists non-deleted `watch.watch` resources as
  `common.resource-list-result@1`. It orders watches by id, pages with an opaque cursor, and
ddiff --git a/migrations/0009_control_log.sql b/migrations/0009_control_log.sql
nnew file mode 100644
iindex 0000000..7436629
---- /dev/null
++++ b/migrations/0009_control_log.sql
@@@ -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 watch_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)
++);
ddiff --git a/package-lock.json b/package-lock.json
iindex 6894aaf..cd71ef1 100644
---- a/package-lock.json
++++ b/package-lock.json
@@@ -12,7 +12,7 @@
          "dotenv": "^16.4.7",
          "express": "^4.21.2",
          "fast-xml-parser": "^4.5.3",
--        "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.153.0",
++        "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.165.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"
@@@ -769,9 +769,9 @@
        }
      },
      "node_modules/openvibe-contracts": {
--      "version": "0.153.0",
--      "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.153.0",
--      "integrity": "sha512-WQ7p7ATKtTky9nI9yE9Agkm3rNKuyWljmrMCdZaqxg+FjVaziTqrFz0a3YMBTXJmNf5JTGXn/ioOmWjO95/lCA==",
++      "version": "0.165.0",
++      "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.165.0",
++      "integrity": "sha512-IfBo0pI0lAqSGvtjA7PWjzf2apWMOrY1wAvHB9miGUgVOsjlLbkE8ND8SSutExoLGmZRsf+pePcCw2QbGjTjIA==",
        "license": "AGPL-3.0-only",
        "dependencies": {
          "ajv": "^8.17.1",
ddiff --git a/package.json b/package.json
iindex d56583c..a44d82b 100644
---- a/package.json
++++ b/package.json
@@@ -24,7 +24,7 @@
      "dotenv": "^16.4.7",
      "express": "^4.21.2",
      "fast-xml-parser": "^4.5.3",
--    "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.153.0",
++    "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.165.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/api/resource-control.js b/server/api/resource-control.js
nnew file mode 100644
iindex 0000000..c13fe4a
---- /dev/null
++++ b/server/api/resource-control.js
@@@ -0,0 +1,101 @@
++'use strict';
++/**
++ * Watch's 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 watch.resource.control,
++ * acting for the person in on_behalf_of, on a watch.watch 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   pause the watch: nothing is checked until it resumes        (registry.patch → watch.watch.paused)
++ *   resume    make a paused watch active again                             (registry.patch → watch.watch.updated)
++ *   delete    remove it, exactly as DELETE /api/v1/watches/:id does        (registry.remove → watch.watch.removed)
++ *
++ * Any other action is refused `control.unsupported_action`. Only the watch's owner may act on it (on_behalf_of names
++ * them; a watch of another person 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 watch_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 = 'watch';
++const CAPABILITY = 'watch.resource.control';
++const ACTIONS = { suspend: 'paused', resume: 'active' };
++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({ db, auth, registry, now = () => Date.now() }) {
++    const router = express.Router();
++    const guard = auth.requireCap(CAPABILITY);
++    const json = express.json({ limit: '16kb' });
++    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, principal) {
++        const parsed = contracts.resources.parse(request.resource || '');
++        if (!parsed || parsed.service !== SERVICE || parsed.type !== 'watch') {
++            return result(request, 'refused', { problem: problem(404, 'resources.unknown_resource', `no watch named ${request.resource || request.resource_kind}`) });
++        }
++        if (!(request.action in ACTIONS) && request.action !== 'delete') {
++            return result(request, 'refused', { problem: problem(422, 'control.unsupported_action', `a watch takes suspend, resume and delete, 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 who owns the watch') });
++        }
++        const owner = { subject: person.id };
++        const row = await registry.get(parsed.id, owner);
++        if (!row || row.deleted_at || row.project_id !== parsed.project_id || parsed.project_id !== request.project_id) {
++            return result(request, 'refused', { problem: problem(404, 'resources.unknown_resource', `no watch named ${request.resource} of this person in this project`) });
++        }
++        if (request.action === 'delete') {
++            if (request.dry_run) return result(request, 'done', { result: { dry_run: true, would: 'delete the watch' } });
++            await registry.remove(row.id, owner);
++            return result(request, 'done', { result: { deleted: true } });
++        }
++        const target = ACTIONS[request.action];
++        if (row.status === target) return result(request, 'done', { result: { status: row.status, changed: false } });
++        if (request.action === 'resume' && row.status !== 'paused') {
++            return result(request, 'refused', { problem: problem(409, 'control.not_paused', `a ${row.status} watch is not resumed here; edit it on openvibe.watch`) });
++        }
++        if (request.dry_run) return result(request, 'done', { result: { dry_run: true, would: `make the watch ${target}` } });
++        const updated = await registry.patch(row.id, { status: target }, owner);
++        return result(request, 'done', { result: { status: updated.status, changed: true } });
++    }
++
++    router.post('/api/v1/resources/control', guard, json, async (req, res, next) => {
++        try {
++            if (req.principal.kind !== 'service' || !/^svc:/.test(req.principal.sub)) {
++                return contracts.http.sendProblem(res, 403, 'capability.denied', { detail: `${CAPABILITY} is first-party`, ctx: req.ov });
++            }
++            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 watch_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, req.principal);
++            await db.prepare(`INSERT INTO watch_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 };
ddiff --git a/server/app.js b/server/app.js
iindex 5ce9e0d..e1e0747 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -22,6 +22,7 @@ const { createSso } = require('./auth/sso');
  const { createPageRoutes } = require('./http/pages');
  const { watchesRouter } = require('./api/watches');
  const { resourceIndexRouter } = require('./api/resource-index');
++const { resourceControlRouter } = require('./api/resource-control');
  const { checksRouter } = require('./api/checks');
  const { createActorLimits } = require('./api/actor-limits');
  const { assetVersion, send, setRelease, SITE_NAME, TAGLINE } = require('./render/layout');
@@@ -125,6 +126,8 @@ function createApp({ config, db, registry, check, observations, scheduler, auth,
      const limits = createActorLimits({ config, now: limitsNow || (() => Date.now()), registry: metrics.registry, log });
      // OpenVibe.Services' resource index (watch.resource.read, service tokens only): read on loopback; the public vhost 404s it.
      app.use(resourceIndexRouter({ db, auth }));
++    // ...and its control API (watch.resource.control, ADR-048): suspend, resume and delete a person's watch.
++    app.use(resourceControlRouter({ db, auth, registry, now }));
      app.use(watchesRouter({ registry, auth, limits, webhookSources }));
      app.use(checksRouter({ registry, check, observations, auth, limits }));
      app.use('/api', (req, res) => http.sendProblem(res, 404, 'watch.not_found', { detail: `no route ${req.method} ${req.path}`, ctx: req.ov }));
ddiff --git a/test/resource-control.test.js b/test/resource-control.test.js
nnew file mode 100644
iindex 0000000..7834735
---- /dev/null
++++ b/test/resource-control.test.js
@@@ -0,0 +1,84 @@
++'use strict';
++/**
++ * Watch's control API (ADR-048, plan T13 step 10): OpenVibe.Services suspends, resumes and deletes a person's watch by
++ * its OVRN. Only a first-party service holding watch.resource.control, only for the watch's owner and inside the OVRN's
++ * project; 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, newUser, watchDef, suite } = require('./helpers');
++
++const t = suite('resource-control');
++let svc;
++const owner = newUser();
++const stranger = newUser();
++const project = contracts.ids.newId('project');
++const otherProject = contracts.ids.newId('project');
++const token = serviceToken('services', ['watch.resource.control']);
++let watch;
++let n = 0;
++const key = () => `ctl-test-${++n}-${Date.now()}`;
++const control = async (body, bearer = token) => request(svc.base, 'POST', '/api/v1/resources/control', { token: bearer, body });
++const ovrn = (w, prj = project) => contracts.resources.format({ service: 'watch', project_id: prj, type: 'watch', id: w.id });
++const ask = (action, extra = {}) => ({ action, resource: ovrn(watch), 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 statusOf = async () => (await svc.db.prepare('SELECT status, deleted_at FROM watches WHERE id = ?').get(watch.id));
++
++t('boot with a person\'s watch in a project', async () => {
++    svc = await boot();
++    watch = await svc.registry.create(watchDef({ name: 'controlled' }), { sub: owner });
++    await svc.db.prepare('UPDATE watches SET project_id = ? WHERE id = ?').run(project, watch.id);
++});
++
++t('only a first-party service with the capability may call it', async () => {
++    assert.strictEqual((await control(ask('suspend'), serviceToken('services', ['watch.resource.read']))).status, 403);
++    assert.strictEqual((await control(ask('suspend'), userToken(owner))).status, 403);
++    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 statusOf()).status, 'active', '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], ['done', 'paused']);
++    assert.strictEqual((await statusOf()).status, 'paused');
++    const again = await control(body);
++    assert.deepStrictEqual(again.body, r.body, 'the first answer, again');
++    const conflict = await control({ ...body, action: 'resume' });
++    assert.strictEqual(conflict.status, 409);
++});
++
++t('resume; unsupported actions, 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']);
++    const resize = valid((await control(ask('resize'))).body);
++    assert.deepStrictEqual([resize.state, resize.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(watch, otherProject), project_id: otherProject })).body);
++    assert.deepStrictEqual([elsewhere.state, elsewhere.problem.code], ['refused', 'resources.unknown_resource']);
++    const nobody = valid((await control(ask('suspend', { on_behalf_of: undefined }))).body);
++    assert.deepStrictEqual([nobody.state, nobody.problem.code], ['refused', 'control.no_person']);
++    assert.strictEqual((await statusOf()).status, 'active');
++});
++
++t('delete removes it as DELETE /api/v1/watches/:id does', async () => {
++    const r = valid((await control(ask('delete'))).body);
++    assert.deepStrictEqual([r.state, r.result.deleted], ['done', true]);
++    assert.ok((await statusOf()).deleted_at);
++    const gone = valid((await control(ask('resume'))).body);
++    assert.strictEqual(gone.problem.code, 'resources.unknown_resource');
++});
++
++t('done', async () => { await svc.stop(); });
++
++t.run();