openvibers/openvibe.media Commits

Add per-app webhook send tallies and an operator webhooks section

OP OpenVibers committed 2 Oct 2026 · 3c2d617892 · 5 files, +101 −5

Asked: Plan T4 (lane C, Media cleanup), audit step 9 first slice (audit: /home/workstation/openvibe/agents/jobs/media-t4-media-cleanup-audit-only-192107.out.md, section 3 and table row 9). Outbound tenant webhooks (C-83, README.md "Outbound webhooks to tenants ... stay until every consumer reads events") …
Harness job media-plan-t4-lane-c-media-092552 (plan T4).

Changes

ddiff --git a/README.md b/README.md
iindex 7cccddf..9131145 100644
---- a/README.md
++++ b/README.md
@@@ -134,7 +134,8 @@ other OpenVibe services call the API.
subject, in every tenant: uploads in progress, copies, derivatives, visibility, lifecycle and their own
usage per tenant and namespace (`GET /api/v2/me/objects`, `/objects/:id`, `/usage`). Read-only: changes
stay with the apps. Network staff with `staff.site.view` get the operator views at `/me/ops` (failed
--jobs, missing media, backfill, tiering, usage recompute with `staff.site.configure`). Details in
++jobs, missing media, backfill, tiering, outbound webhooks — who is configured and what this process
++has sent — and usage recompute with `staff.site.configure`). Details in
[docs/object-model.md](docs/object-model.md#object-explorer).
## Tenancy & auth
ddiff --git a/docs/object-model.md b/docs/object-model.md
iindex 0c54fcf..be448e1 100644
---- a/docs/object-model.md
++++ b/docs/object-model.md
@@@ -461,6 +461,7 @@ cheap and renders in a drill.
- `backfill`: the last object backfill that changed something; rows with no object yet; objects whose owner subject is unresolved; whether upload reservations were seeded.
- `tiering`: providers and policy; VODs and clips per provider; objects per canonical copy (native or projected); VODs eligible to offload; R2 decisions of the last 24 h and recent refusals and failures; the sweep's state in this process; and `native_objects`: the [object tiering](#tiering-of-native-objects)'s activation gate and policy, native objects in R2, how many are eligible to promote now, its decisions of the last 24 h (dry runs included) and the most recent ones, and its last sweep.
- `namespaces`: every namespace with quota and usage snapshot.
++- `webhooks`: the apps with a configured outbound webhook (app id and host only — never the URL or secret) and, per app, what this process has sent since it loaded (`sent`, `failed`, the last event and attempt/ok times, the last error). In-memory: counts reset on restart and scope to the app when the report is scoped.
## Readiness
ddiff --git a/server/me/ops.js b/server/me/ops.js
iindex cff5a0d..f258759 100644
---- a/server/me/ops.js
++++ b/server/me/ops.js
@@@ -2,9 +2,10 @@
* OpenVibe.Media — operator views over the object platform (roadmap WS-G task 12;
* docs/object-model.md#operator-views).
*
-- * report({ appId, limit }) failed jobs, missing media, backfill status, tiering diagnostics and the
-- * namespaces' usage snapshot; appId narrows everything that has a tenant
-- * (the tiering sweep and the provider switches are service-wide)
++ * report({ appId, limit }) failed jobs, missing media, backfill status, tiering diagnostics, the
++ * namespaces' usage snapshot and the outbound-webhooks section; appId
++ * narrows everything that has a tenant (the tiering sweep and the provider
++ * switches are service-wide)
* recompute({ appId }) refresh the namespaces' usage snapshot from the rows (namespaces.reconcile),
* the one write here
*
@@@ -18,6 +19,7 @@
const db = require('../db/database');
const namespaces = require('../objects/namespaces');
const copyReport = require('../objects/copy-report');
++const webhookStats = require('../webhooks');
const iso = require('./explorer').iso;
const clip = (s, n) => (s == null ? null : String(s).slice(0, n));
@@@ -204,6 +206,29 @@ async function namespaceSnapshot(appId) {
})));
}
++// ── Webhooks ─────────────────────────────────────────────────
++
++/** Who is configured to receive outbound webhooks (host only, never the URL or secret) and what this
++ * process has sent them since it loaded. */
++async function webhooks(appId) {
++ const s = scoped(appId);
++ const configured = (await db.all(`SELECT app_id, webhook_url FROM apps WHERE COALESCE(webhook_url, '') != ''${s.sql} ORDER BY app_id`, s.params))
++ .map((r) => {
++ let host = null;
++ try { host = new URL(r.webhook_url).host; } catch { /* an unparseable URL shows as null */ }
++ return { app_id: r.app_id, host };
++ });
++ const all = webhookStats.stats();
++ const sends = appId
++ ? { since: all.since, by_app: all.by_app[appId] ? { [appId]: all.by_app[appId] } : {} }
++ : all;
++ return {
++ configured,
++ sends,
++ note: 'Outbound webhooks (C-83) retire when every configured app reads OpenVibe.Events; sends counts reset on restart.',
++ };
++}
++
/** The whole operator report (see the header). `limit` bounds each list (1-200, default 50). */
async function report({ appId = null, limit = 50 } = {}) {
const n = Math.min(Math.max(parseInt(limit, 10) || 50, 1), 200);
@@@ -216,6 +241,7 @@ async function report({ appId = null, limit = 50 } = {}) {
backfill: await backfill(appId),
tiering: await tiering(appId, n),
namespaces: await namespaceSnapshot(appId),
++ webhooks: await webhooks(appId),
};
}
ddiff --git a/server/webhooks.js b/server/webhooks.js
iindex 5be7d8f..dc0c380 100644
---- a/server/webhooks.js
++++ b/server/webhooks.js
@@@ -21,6 +21,27 @@ const MAX_ATTEMPTS = 3;
const RETRY_BASE_MS = 3000;
const TIMEOUT_MS = 10000;
++// In-memory tally of what this process actually tried to send, per app (C-83: who still gets webhooks
++// is what decides when the outbound path can retire). Never holds the URL, secret, body or signature.
++const SINCE = new Date().toISOString();
++const tallies = new Map();
++
++function tallyFor(appId) {
++ let t = tallies.get(appId);
++ if (!t) { t = { sent: 0, failed: 0, last_event: null, last_attempt_at: null, last_ok_at: null, last_error: null }; tallies.set(appId, t); }
++ return t;
++}
++
++/** Fresh copy of the in-process tallies: { since, by_app: { [app_id]: {...} } }. */
++function stats() {
++ const by_app = {};
++ for (const [appId, t] of tallies) by_app[appId] = { ...t };
++ return { since: SINCE, by_app };
++}
++
++/** Test-only: forget every tally. */
++function _resetStats() { tallies.clear(); }
++
function sign(secret, rawBody) {
return 'sha256=' + crypto.createHmac('sha256', String(secret || '')).update(rawBody).digest('hex');
}
@@@ -62,12 +83,19 @@ async function sendWebhook(appOrId, event, data, { eventId = null } = {}) {
const rawBody = JSON.stringify(body);
const signature = sign(app.webhook_secret, rawBody);
++ const t = tallyFor(app.app_id);
for (let attempt = 1; attempt <= MAX_ATTEMPTS; attempt++) {
++ t.last_event = event;
++ t.last_attempt_at = new Date().toISOString();
try {
await _post(app.webhook_url, rawBody, signature);
++ t.sent++;
++ t.last_ok_at = new Date().toISOString();
return true;
} catch (err) {
if (attempt === MAX_ATTEMPTS) {
++ t.failed++;
++ t.last_error = String((err && err.message) || err).slice(0, 200);
console.warn(`[Webhooks] ${event} → ${app.app_id} failed after ${MAX_ATTEMPTS} attempts: ${err.message}`);
return false;
}
@@@ -101,4 +129,4 @@ async function announce(appId, event, { change = null, payload }) {
return { data, eventId };
}
--module.exports = { sendWebhook, announce, sign };
++module.exports = { sendWebhook, announce, sign, stats, _resetStats };
ddiff --git a/test/me-explorer.test.js b/test/me-explorer.test.js
iindex 68f532f..b8d2105 100644
---- a/test/me-explorer.test.js
++++ b/test/me-explorer.test.js
@@@ -37,6 +37,11 @@ const http = require('http');
await db.upsertApp({ app_id: 'live', name: 'OpenVibe.Live', api_key: 'live-key' });
await db.upsertApp({ app_id: 'tools', name: 'OpenVibe.Tools', api_key: 'tools-key' });
++ // The `live` app has an outbound webhook (tools has none): the operator report must show the host
++ // only and never the URL or the secret.
++ const HOOK_URL = 'https://hooks.example.test/ov-private-hook-path';
++ const HOOK_SECRET = 'whsec_TESTONLY-MUST-NOT-LEAK';
++ await db.run("UPDATE apps SET webhook_url = ?, webhook_secret = ? WHERE app_id = 'live'", [HOOK_URL, HOOK_SECRET]);
for (const a of ['live', 'tools']) await db.ensureRootNamespace(await db.getApp(a));
await namespaces.ensure(await db.getApp('live'), 'live.avatars');
@@@ -334,8 +339,15 @@ const http = require('http');
assert.ok(rep.backfill.unprojected && rep.backfill.owner_subject.missing >= 1);
assert.ok(rep.tiering.objects.some(o => o.origin === 'native') && rep.tiering.sweep && 'pending_offload' in rep.tiering);
assert.ok(rep.namespaces.some(n => n.namespace === 'live.avatars'));
++ // Webhooks section: host only, no URL path and no secret anywhere in the payload; counts are
++ // this process's own (none of these calls sends one).
++ assert.deepStrictEqual(rep.webhooks.configured, [{ app_id: 'live', host: 'hooks.example.test' }], 'configured webhooks, host only');
++ assert.ok(rep.webhooks.sends.since && !Object.keys(rep.webhooks.sends.by_app).length, 'no sends from the explorer itself');
++ assert.ok(rep.webhooks.note.includes('OpenVibe.Events') && rep.webhooks.note.includes('reset on restart'));
++ assert.ok(!r.text.includes(HOOK_URL) && !r.text.includes('ov-private-hook-path') && !r.text.includes(HOOK_SECRET), 'the report never carries the webhook URL or secret');
r = await call('GET', '/api/v2/me/ops?app=tools', { token: 'admin' });
assert.deepStrictEqual([r.json.scope, r.json.jobs.failed.total, r.json.namespaces.map(n => n.namespace)], ['tools', 0, ['tools']]);
++ assert.deepStrictEqual(r.json.webhooks.configured, [], 'tools has no webhook');
r = await call('GET', '/me/ops', { cookie: 'admin' });
assert.ok(r.status === 200 && r.text.includes('mjob_TESTFAILED') && r.text.includes('action="/me/ops/recompute"') && r.text.includes('noindex'));
console.log('✅ operator views: staff.site.view only (not users, not global mods); failed jobs, missing media, backfill, tiering, namespaces');
@@@ -364,12 +376,40 @@ const http = require('http');
assert.strictEqual(r.status, 200, r.text);
assert.strictEqual(r.json.scope, 'live');
assert.ok(r.json.namespaces.every(n => n.app_id === 'live'));
++ assert.deepStrictEqual(r.json.webhooks.configured, [{ app_id: 'live', host: 'hooks.example.test' }], 'the app-key twin shows its own webhook');
++ assert.ok(r.json.webhooks.sends.by_app === undefined || Object.keys(r.json.webhooks.sends.by_app).every(a => a === 'live'), 'sends narrowed to the app');
assert.strictEqual((await call('GET', '/api/v1/live/admin/storage/ops', { token: 'live-key', headers: { 'x-ov-user-id': '5' } })).status, 403, 'never acting for a user');
assert.strictEqual((await call('GET', '/api/v1/live/admin/storage/ops', { token: 'ana' })).status, 401);
r = await call('POST', '/api/v1/tools/admin/storage/ops/recompute', { token: 'tools-key' });
assert.deepStrictEqual([r.status, r.json.scope, r.json.namespaces], [200, 'tools', 1]);
console.log('✅ /api/v1/:app/admin/storage/ops: the same report for the app key, scoped to its tenant');
++ // ── Webhook tallies: one success, one failure ──
++ {
++ const hooks = require('../server/webhooks');
++ hooks._resetStats();
++ await db.run("UPDATE apps SET webhook_url = 'https://hooks-down.example.test/never' WHERE app_id = 'tools'");
++ const realFetch = global.fetch, realSetTimeout = global.setTimeout;
++ global.setTimeout = (fn) => realSetTimeout(fn, 0); // retries must not slow the suite
++ global.fetch = async (url) => ({ ok: String(url).includes('hooks.example.test'), status: String(url).includes('hooks.example.test') ? 200 : 500 });
++ try {
++ assert.strictEqual(await hooks.sendWebhook('live', 'vod.ready', { id: 1 }), true);
++ assert.strictEqual(await hooks.sendWebhook('tools', 'clip.ready', { id: 2 }), false);
++ } finally {
++ global.fetch = realFetch; global.setTimeout = realSetTimeout;
++ await db.run("UPDATE apps SET webhook_url = NULL, webhook_secret = NULL WHERE app_id = 'tools'");
++ }
++ const st = hooks.stats();
++ assert.deepStrictEqual([st.by_app.live.sent, st.by_app.live.failed, st.by_app.live.last_event], [1, 0, 'vod.ready']);
++ assert.ok(st.by_app.live.last_ok_at && st.by_app.live.last_attempt_at && !st.by_app.live.last_error);
++ assert.deepStrictEqual([st.by_app.tools.sent, st.by_app.tools.failed, st.by_app.tools.last_error, st.by_app.tools.last_event], [0, 1, 'HTTP 500', 'clip.ready']);
++ st.by_app.live.sent = 999;
++ assert.strictEqual(hooks.stats().by_app.live.sent, 1, 'stats() is a fresh copy');
++ hooks._resetStats();
++ assert.deepStrictEqual(hooks.stats().by_app, {});
++ console.log('✅ webhook tallies: a 200 counts sent, a 500 after retries counts failed with its error; stats() copies');
++ }
++
// ── Routes under /api/v2/me are the explorer's only ──
r = await call('GET', '/api/v2/me/namespaces', { token: 'ana' });
assert.deepStrictEqual([r.status, r.json.code], [404, 'media.not_found']);