openvibers/openvibe.network Commits

Event deliveries in admin (WS-F task 3): the owner sees OpenVibe.Events' dead, failing and pending deliveries and replays them (one, or all shown for a subscription) through Events' operator API with a self-signed 5-minute events.delivery.admin token; every replay is written to the audit log

OP OpenVibers committed 25 Sept 2026 · eff91e6d15 · 5 files, +198 −2





Changes
ddiff --git a/package.json b/package.json
iindex baf74d06..62000b46 100644
---- a/package.json
++++ b/package.json
@@@ -6,7 +6,7 @@
    "scripts": {
      "start": "node server/index.js",
      "dev": "NODE_ENV=development node server/index.js",
--    "test": "node test/registry-bootstrap.test.js && node test/nginx-generator-live.test.js && node test/admin-refresh.test.js && node test/history-sso.test.js && node test/sso-check.test.js && node test/fedcm.test.js && node test/domains.test.js && node test/frame.test.js && node test/github-integration.test.js && node test/revocation.test.js && node test/usernames.test.js && node test/avatar.test.js && node test/roadmap-baseline.test.js && node test/identity-subjects.test.js && node test/principals.test.js && node test/modules.test.js && node test/module-events.test.js && node test/registry-ecosystem.test.js && node test/registry-exposure.test.js && node test/registry-manifest-fixture.test.js && node test/staff-claims.test.js && node test/profile-events.test.js && node test/grants-admin.test.js && node test/registry-topics.test.js && node test/service-principal.test.js && node test/oauth-pkce.test.js && node test/developer-projects.test.js && node test/developer-defaults.test.js && node test/public-cors.test.js && node test/observability.test.js && node test/status.test.js && node test/events-consumer.test.js && node test/security-session.test.js && node test/security-redirects.test.js && node test/analytics-privacy.test.js && node test/data-ownership.test.js && node test/unknown-paths.test.js && node test/oidc-discovery.test.js && node test/refresh-tokens.test.js && node test/provider-secrets.test.js && node test/first-party-origins.test.js && node test/updates.test.js && node test/static-headers.test.js && node test/library-tags.test.js && node test/moderation-audit.test.js",
++    "test": "node test/registry-bootstrap.test.js && node test/nginx-generator-live.test.js && node test/admin-refresh.test.js && node test/history-sso.test.js && node test/sso-check.test.js && node test/fedcm.test.js && node test/domains.test.js && node test/frame.test.js && node test/github-integration.test.js && node test/revocation.test.js && node test/usernames.test.js && node test/avatar.test.js && node test/roadmap-baseline.test.js && node test/identity-subjects.test.js && node test/principals.test.js && node test/modules.test.js && node test/module-events.test.js && node test/registry-ecosystem.test.js && node test/registry-exposure.test.js && node test/registry-manifest-fixture.test.js && node test/staff-claims.test.js && node test/profile-events.test.js && node test/grants-admin.test.js && node test/events-ops.test.js && node test/registry-topics.test.js && node test/service-principal.test.js && node test/oauth-pkce.test.js && node test/developer-projects.test.js && node test/developer-defaults.test.js && node test/public-cors.test.js && node test/observability.test.js && node test/status.test.js && node test/events-consumer.test.js && node test/security-session.test.js && node test/security-redirects.test.js && node test/analytics-privacy.test.js && node test/data-ownership.test.js && node test/unknown-paths.test.js && node test/oidc-discovery.test.js && node test/refresh-tokens.test.js && node test/provider-secrets.test.js && node test/first-party-origins.test.js && node test/updates.test.js && node test/static-headers.test.js && node test/library-tags.test.js && node test/moderation-audit.test.js",
      "reset-db": "node server/reset-db.js",
      "grant-admin": "node server/grant-admin.js"
    },
ddiff --git a/public/admin.html b/public/admin.html
iindex fbaffc45..74906419 100644
---- a/public/admin.html
++++ b/public/admin.html
@@@ -214,6 +214,7 @@ body{min-height:100%;color:var(--text);font-family:'Inter',-apple-system,BlinkMa
                  <button class="admin-tab-btn" onclick="showTab('audit',this)"><i class="fa-solid fa-clipboard-list"></i> Audit</button>
                  <button class="admin-tab-btn" onclick="showTab('modlog',this)"><i class="fa-solid fa-gavel"></i> Moderation log</button>
                  <button class="admin-tab-btn owner-only" onclick="showTab('grants',this)"><i class="fa-solid fa-key"></i> Service grants</button>
++                <button class="admin-tab-btn owner-only" onclick="showTab('eventsops',this)"><i class="fa-solid fa-route"></i> Event deliveries</button>
                  <button class="admin-tab-btn" onclick="showTab('analytics',this)"><i class="fa-solid fa-chart-pie"></i> Analytics</button>
                  <button class="admin-tab-btn owner-only" onclick="showTab('domains',this)"><i class="fa-solid fa-globe"></i> Domains</button>
                  <button class="admin-tab-btn owner-only" onclick="showTab('deploy',this)"><i class="fa-solid fa-server"></i> Deploy</button>
@@@ -425,6 +426,19 @@ body{min-height:100%;color:var(--text);font-family:'Inter',-apple-system,BlinkMa
          <div id="grant-changes"></div>
      </div>
  
++    <!-- Event deliveries (WS-F task 3): OpenVibe.Events' dead-letter queue and replays -->
++    <div class="section" id="sec-eventsops">
++        <p class="muted" style="margin:0 0 10px">Deliveries OpenVibe.Events could not make. A dead delivery stopped retrying; replay it once its consumer is fixed. Replays are audited.</p>
++        <div style="display:flex;flex-wrap:wrap;gap:8px;align-items:end;margin-bottom:10px">
++            <label class="muted" style="font-size:12px;display:flex;flex-direction:column;gap:2px">Status
++                <select id="dlq-status" onchange="loadDeliveries()"><option value="dead">Dead</option><option value="failed">Failing (retrying)</option><option value="pending">Pending</option></select></label>
++            <label class="muted" style="font-size:12px;display:flex;flex-direction:column;gap:2px">Subscription<input id="dlq-sub" placeholder="sub_…" maxlength="40" style="width:240px"></label>
++            <button class="btn" type="button" onclick="loadDeliveries()">Show</button>
++        </div>
++        <div id="dlq-counts" class="muted" style="font-size:12px;margin-bottom:8px"></div>
++        <div id="dlq-content"><div class="loading">Loading...</div></div>
++    </div>
++
      <!-- Analytics -->
      <div class="section" id="sec-analytics">
          <div class="sub-tabs" id="analytics-sub-tabs">
@@@ -820,7 +834,7 @@ let currentTab = 'dashboard';
  // Tabs restricted to the network owner (API keys / money / infrastructure / core settings).
  // NOTE: 'ai' is NOT here — admins may use its analytics/explorer tools; only the
  // AI configuration form inside it is owner-gated (see loadAi).
--const OWNER_ONLY_TABS = new Set(['deploy', 'domains', 'payments', 'cashouts', 'storage', 'email', 'settings', 'grants']);
++const OWNER_ONLY_TABS = new Set(['deploy', 'domains', 'payments', 'cashouts', 'storage', 'email', 'settings', 'grants', 'eventsops']);
  const tabLoaders = {
      dashboard: loadDashboard,
      users: loadUsers,
@@@ -842,6 +856,7 @@ const tabLoaders = {
      audit: loadAudit,
      modlog: () => loadModLog(true),
      grants: () => loadGrants(),
++    eventsops: () => loadDeliveries(),
      analytics: loadAnalytics,
      deploy: loadDeployOverview,
      domains: loadDomains,
@@@ -3606,6 +3621,35 @@ async function exportModLog() {
          setTimeout(() => URL.revokeObjectURL(url), 5000);
      } catch (err) { alert(err.message || 'Export failed'); }
  }
++let _deliveries = [];
++async function loadDeliveries() {
++    const c = document.getElementById('dlq-content');
++    const q = new URLSearchParams({ status: document.getElementById('dlq-status').value, limit: '200' });
++    const sub = document.getElementById('dlq-sub').value.trim();
++    if (sub) q.set('subscription_id', sub);
++    c.innerHTML = '<div class="loading">Loading...</div>';
++    try {
++        const d = await api(`/api/admin/events/deliveries?${q}`);
++        _deliveries = d.deliveries || [];
++        const counts = d.counts || {};
++        document.getElementById('dlq-counts').textContent = Object.entries(counts).map(([k, v]) => `${k}: ${v}`).join(' · ');
++        const bySub = {};
++        for (const x of _deliveries) (bySub[x.subscription_id] = bySub[x.subscription_id] || []).push(x.event_id);
++        const replayAll = Object.entries(bySub).map(([s, ids]) => `<button class="btn btn-sm" data-sub="${esc(s)}" onclick="replayDeliveries(this.dataset.sub, null)">Replay ${ids.length} for ${esc(s.slice(0, 14))}…</button>`).join(' ');
++        c.innerHTML = _deliveries.length ? `<div style="margin-bottom:8px">${replayAll}</div><table class="admin-table"><thead><tr><th>Event</th><th>Subscription</th><th>Seq</th><th>Attempts</th><th>Last answer</th><th></th></tr></thead><tbody>${_deliveries.map(x => `
++            <tr><td style="font-size:11px"><code>${esc(x.event_id)}</code></td><td style="font-size:11px"><code>${esc(x.subscription_id)}</code></td><td>${esc(String(x.seq))}</td><td>${esc(String(x.attempt))}</td>
++            <td style="font-size:11px;max-width:320px;overflow:hidden;text-overflow:ellipsis;white-space:nowrap" title="${esc(x.last_error || '')}">${esc(x.last_status ? String(x.last_status) : '-')} ${esc(x.last_error || '')}</td>
++            <td><button class="btn btn-sm" data-sub="${esc(x.subscription_id)}" data-evt="${esc(x.event_id)}" onclick="replayDeliveries(this.dataset.sub, [this.dataset.evt])">Replay</button></td></tr>`).join('')}</tbody></table>` : '<p class="muted">Nothing here.</p>';
++    } catch (err) {
++        c.innerHTML = `<p class="muted">Deliveries could not be loaded: ${esc(err.message || 'error')}</p>`;
++    }
++}
++async function replayDeliveries(sub, ids) {
++    const eventIds = ids || _deliveries.filter(x => x.subscription_id === sub).map(x => x.event_id);
++    if (!eventIds.length || !confirm(`Replay ${eventIds.length} deliver${eventIds.length === 1 ? 'y' : 'ies'} to ${sub}?`)) return;
++    try { const r = await api('/api/admin/events/replay', { method: 'POST', body: { subscription_id: sub, event_ids: eventIds } }); alert(`Queued ${r.queued}`); loadDeliveries(); }
++    catch (err) { alert(err.message || 'Replay failed'); }
++}
  let _grants = [];
  async function loadGrants() {
      const c = document.getElementById('grants-content');
ddiff --git a/server/admin/events-ops.js b/server/admin/events-ops.js
nnew file mode 100644
iindex 00000000..582cc2b9
---- /dev/null
++++ b/server/admin/events-ops.js
@@@ -0,0 +1,84 @@
++'use strict';
++/**
++ * The Events delivery operator view (roadmap WS-F task 3): Network's admin reads OpenVibe.Events'
++ * dead-letter queue and replays deliveries through Events' own operator API, with a self-signed
++ * 5-minute service token (sub svc:network, aud openvibe.events, cap events.delivery.admin; Network
++ * is the token issuer). Owner-only; every replay is written to audit_log.
++ *
++ *   GET  /api/admin/events/deliveries?status=dead|failed|pending|delivered&subscription_id=&after_seq=&limit=
++ *   POST /api/admin/events/replay      { subscription_id, event_ids: [...] } or { subscription_id, from_seq }
++ *
++ * Off (503) without OV_EVENTS_INTERNAL_URL.
++ */
++const crypto = require('crypto');
++const express = require('express');
++const { serviceAuth, assertValid } = require('openvibe-contracts');
++const { requireOwner } = require('../auth/owner-guard');
++
++const TTL_S = 300;
++const SUB_RE = /^sub_[0-9A-HJKMNP-TV-Z]{26}$/;
++const EVT_RE = /^evt_[0-9A-HJKMNP-TV-Z]{26}$/;
++
++function createEventsOps({ db, eventsUrl, privateKey, issuer, fetchImpl = globalThis.fetch }) {
++    const base = String(eventsUrl || '').replace(/\/+$/, '');
++    let cached = null;
++    function token() {
++        const now = Math.floor(Date.now() / 1000);
++        if (cached && cached.exp - 60 > now) return cached.token;
++        const claims = {
++            iss: issuer, sub: 'svc:network', actor_type: 'service', aud: ['openvibe.events'], cap: ['events.delivery.admin'],
++            iat: now, exp: now + TTL_S, jti: `tok_${crypto.randomBytes(12).toString('hex')}`,
++        };
++        assertValid('identity.service-token-claims@1', claims);
++        cached = { token: serviceAuth.signServiceToken(claims, privateKey), exp: claims.exp };
++        return cached.token;
++    }
++    async function call(method, path, body) {
++        const res = await fetchImpl(`${base}${path}`, {
++            method, headers: { Authorization: `Bearer ${token()}`, Accept: 'application/json', ...(body ? { 'Content-Type': 'application/json' } : {}) },
++            body: body ? JSON.stringify(body) : undefined, signal: AbortSignal.timeout(10000),
++        });
++        const data = await res.json().catch(() => ({}));
++        return { status: res.status, data };
++    }
++
++    const r = express.Router();
++    r.use(requireOwner);
++    r.use((req, res, next) => (base ? next() : res.status(503).json({ ok: false, error: 'OV_EVENTS_INTERNAL_URL is not set' })));
++    r.get('/deliveries', async (req, res) => {
++        const q = new URLSearchParams();
++        const status = ['dead', 'failed', 'pending', 'delivered'].includes(req.query.status) ? req.query.status : 'dead';
++        q.set('status', status);
++        if (SUB_RE.test(String(req.query.subscription_id || ''))) q.set('subscription_id', req.query.subscription_id);
++        const after = parseInt(req.query.after_seq, 10);
++        if (Number.isSafeInteger(after) && after > 0) q.set('after_seq', String(after));
++        q.set('limit', String(Math.min(Math.max(parseInt(req.query.limit, 10) || 100, 1), 500)));
++        try {
++            const out = await call('GET', `/api/v1/deliveries?${q}`);
++            res.set('Cache-Control', 'private, no-store');
++            res.status(out.status === 200 ? 200 : 502).json(out.status === 200 ? { ok: true, ...out.data } : { ok: false, error: `Events answered ${out.status}`, detail: out.data && (out.data.detail || out.data.title) });
++        } catch (err) {
++            res.status(502).json({ ok: false, error: `Events did not answer: ${err.message}` });
++        }
++    });
++    r.post('/replay', express.json({ limit: '64kb' }), async (req, res) => {
++        const b = req.body || {};
++        if (!SUB_RE.test(String(b.subscription_id || ''))) return res.status(400).json({ ok: false, error: 'subscription_id is required' });
++        const ids = Array.isArray(b.event_ids) ? b.event_ids.map(String) : null;
++        const fromSeq = Number.isSafeInteger(b.from_seq) && b.from_seq >= 0 ? b.from_seq : null;
++        if ((ids === null) === (fromSeq === null)) return res.status(400).json({ ok: false, error: 'pass exactly one of event_ids or from_seq' });
++        if (ids && (!ids.length || ids.length > 1000 || !ids.every((id) => EVT_RE.test(id)))) return res.status(400).json({ ok: false, error: 'event_ids: 1 to 1000 event ids' });
++        try {
++            const out = await call('POST', '/api/v1/deliveries/replay', ids ? { subscription_id: b.subscription_id, event_ids: ids } : { subscription_id: b.subscription_id, from_seq: fromSeq });
++            if (out.status !== 200) return res.status(out.status === 404 ? 404 : 502).json({ ok: false, error: `Events answered ${out.status}`, detail: out.data && (out.data.detail || out.data.title) });
++            db.prepare('INSERT INTO audit_log (user_id, action, details) VALUES (?, ?, ?)').run(req.user.id, 'events_replay',
++                JSON.stringify({ subscription_id: b.subscription_id, event_ids: ids ? ids.slice(0, 50) : undefined, event_count: ids ? ids.length : undefined, from_seq: fromSeq ?? undefined, queued: out.data.queued }));
++            res.json({ ok: true, queued: out.data.queued });
++        } catch (err) {
++            res.status(502).json({ ok: false, error: `Events did not answer: ${err.message}` });
++        }
++    });
++    return r;
++}
++
++module.exports = { createEventsOps };
ddiff --git a/server/index.js b/server/index.js
iindex a1f2cea5..beec0791 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -557,6 +557,8 @@ app.get('/api/admin/ssh-info', requireAuth, requireAdmin, (req, res) => {
  app.use('/api/setup', createSetupRoutes(db, config));
  
  // Admin panel API
++// OpenVibe.Events' dead-letter queue and replays, owner-only (WS-F task 3).
++app.use('/api/admin/events', requireAuth, require('./admin/events-ops').createEventsOps({ db, eventsUrl: config.eventsInternalUrl, privateKey, issuer: config.jwt.issuer }));
  // Service grants: owner-only, audited, network.principal_grant.changed (WS-D task 3).
  app.use('/api/admin/grants', requireAuth, require('./identity/grants-admin').router(db));
  app.use('/api/admin', createAdminRoutes(db, notificationService, emailService, requireAuth));
ddiff --git a/test/events-ops.test.js b/test/events-ops.test.js
nnew file mode 100644
iindex 00000000..e2c2bb4c
---- /dev/null
++++ b/test/events-ops.test.js
@@@ -0,0 +1,66 @@
++'use strict';
++// The Events delivery operator view (WS-F task 3): owner-only; Network signs a 5-minute token with
++// events.delivery.admin for openvibe.events; the dead-letter list passes through; a replay needs a
++// subscription and exactly one of event_ids or from_seq, reaches Events, and is audited.
++const assert = require('assert');
++const fs = require('fs');
++const os = require('os');
++const path = require('path');
++const crypto = require('crypto');
++const express = require('express');
++const { initDb } = require('../server/db/database');
++const { createEventsOps } = require('../server/admin/events-ops');
++
++const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-eventsops-'));
++const log = console.log; console.log = () => {};
++const db = initDb(path.join(dir, 'network.db'));
++console.log = log;
++const { privateKey, publicKey } = crypto.generateKeyPairSync('rsa', { modulusLength: 2048, privateKeyEncoding: { type: 'pkcs8', format: 'pem' }, publicKeyEncoding: { type: 'spki', format: 'pem' } });
++db.prepare("INSERT INTO users (id, username, password_hash, role) VALUES (1, 'goosely', 'x', 'admin'), (2, 'adminb', 'x', 'admin')").run();
++const SUB = 'sub_01M3D4CMQ6JX9ZR8SS34M1FTC4', EVT = 'evt_01M3D4GT16WXDXGZYRE97JRTKS';
++
++(async () => {
++    const seen = [];
++    const events = express();
++    events.use(express.json());
++    events.use((req, res, next) => { seen.push({ method: req.method, url: req.url, auth: req.headers.authorization, body: req.body }); next(); });
++    events.get('/api/v1/deliveries', (req, res) => res.json({ deliveries: [{ event_id: EVT, subscription_id: SUB, seq: 9, status: 'dead', attempt: 8, last_status: 500, last_error: 'boom' }], counts: { dead: 1 } }));
++    events.post('/api/v1/deliveries/replay', (req, res) => res.json({ subscription_id: req.body.subscription_id, queued: (req.body.event_ids || [1]).length }));
++    const ev = await new Promise((r) => { const s = events.listen(0, '127.0.0.1', () => r(s)); });
++
++    let as = 1;
++    const app = express();
++    app.use((req, res, next) => { req.user = db.prepare('SELECT * FROM users WHERE id = ?').get(as); next(); });
++    app.use('/api/admin/events', createEventsOps({ db, eventsUrl: `http://127.0.0.1:${ev.address().port}`, privateKey, issuer: 'https://openvibe.network' }));
++    const srv = await new Promise((r) => { const s = app.listen(0, '127.0.0.1', () => r(s)); });
++    const base = `http://127.0.0.1:${srv.address().port}`;
++    const call = async (method, p, body) => { const r = await fetch(base + p, { method, headers: { 'content-type': 'application/json' }, body: body ? JSON.stringify(body) : undefined }); return { status: r.status, body: await r.json() }; };
++
++    let r = await call('GET', '/api/admin/events/deliveries');
++    assert.strictEqual(r.status, 200);
++    assert.deepStrictEqual(r.body.deliveries.map((d) => d.event_id), [EVT]);
++    assert.ok(seen[0].url.startsWith('/api/v1/deliveries?status=dead'));
++    const claims = JSON.parse(Buffer.from(seen[0].auth.split('.')[1], 'base64url').toString());
++    assert.deepStrictEqual([claims.sub, claims.aud, claims.cap], ['svc:network', ['openvibe.events'], ['events.delivery.admin']]);
++    assert.ok(claims.exp - claims.iat <= 300);
++    assert.ok(require('jsonwebtoken').verify(seen[0].auth.slice(7), publicKey, { algorithms: ['RS256'] }), 'signed with Network\'s key');
++
++    assert.strictEqual((await call('POST', '/api/admin/events/replay', { subscription_id: SUB })).status, 400, 'event_ids or from_seq');
++    assert.strictEqual((await call('POST', '/api/admin/events/replay', { subscription_id: SUB, event_ids: [EVT], from_seq: 1 })).status, 400, 'not both');
++    assert.strictEqual((await call('POST', '/api/admin/events/replay', { subscription_id: SUB, event_ids: ['nope'] })).status, 400);
++    r = await call('POST', '/api/admin/events/replay', { subscription_id: SUB, event_ids: [EVT] });
++    assert.deepStrictEqual(r.body, { ok: true, queued: 1 });
++    assert.deepStrictEqual(seen.at(-1).body, { subscription_id: SUB, event_ids: [EVT] });
++    const audit = db.prepare("SELECT user_id, details FROM audit_log WHERE action = 'events_replay'").all();
++    assert.strictEqual(audit.length, 1); assert.strictEqual(audit[0].user_id, 1);
++    assert.deepStrictEqual(JSON.parse(audit[0].details).event_ids, [EVT]);
++
++    as = 2;
++    const n = seen.length;
++    assert.strictEqual((await call('GET', '/api/admin/events/deliveries')).status, 403, 'another admin is not the owner');
++    assert.strictEqual(seen.length, n, 'and nothing reached Events');
++
++    srv.close(); ev.close();
++    fs.rmSync(dir, { recursive: true, force: true });
++    console.log('events ops: all checks passed');
++})().catch((err) => { console.error(err); process.exit(1); });