openvibers/openvibe.network Commits

Creator analytics from events (roadmap WS-E task 6; Contracts v0.68.0, pinned from 0.67.0). Network consumes live.stream.ended (added to its topics) into creator_streams: one row per stream, replaced on redelivery, counts only (no viewer, chatter or IP column; pruned after 400 days). GET /api/v1/creators/:creator/analytics?days= gives everyone streams, minutes and peak viewers, daily and per stream. The creator, and services with network.analytics.creator.read (granted to Live for its dashboards), also get average viewers, chatters, messages and watch minutes. test/creator-analytics.test.js

OP OpenVibers committed 26 Sept 2026 · 45f6c4ce2b · 11 files, +260 −9





Changes
ddiff --git a/README.md b/README.md
iindex 794c6fd4..c8aa0b21 100644
---- a/README.md
++++ b/README.md
@@@ -521,6 +521,13 @@ Google-style account management supporting up to 5 accounts:
  
  ## Analytics (ADR-021)
  
++**Creator analytics** (roadmap WS-E task 6, Contracts 0.68.0; `server/analytics/creators.js`) are built from
++events, not from Live's tables. Every `live.stream.ended` becomes one `creator_streams` row, with the stream's
++totals from the event's `stats`: counts only, never who watched or chatted, and kept 400 days.
++`GET /api/v1/creators/:creator/analytics?days=` shows everyone streams, minutes and peak viewers. The creator,
++and services with `network.analytics.creator.read` (Live's dashboards), also get average viewers, chatters,
++messages and watch minutes.
++
  Network records one row per finished request in the `analytics_*` tables of `network.db` and rolls
  them up hourly and daily for the admin analytics pages (`/api/admin/analytics`, which also gathers
  Live, Tools, Games and Media). What a raw row may carry is bound by ADR-021 (OpenVibe.Contracts
ddiff --git a/package-lock.json b/package-lock.json
iindex 698d4f1f..38a03659 100644
---- a/package-lock.json
++++ b/package-lock.json
@@@ -20,7 +20,7 @@
          "helmet": "^8.1.0",
          "jsonwebtoken": "^9.0.0",
          "multer": "^1.4.5-lts.1",
--        "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.67.0",
++        "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.68.0",
          "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.11.0",
          "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.22.0",
          "uuid": "^11.1.0",
@@@ -1465,9 +1465,9 @@
        }
      },
      "node_modules/openvibe-contracts": {
--      "version": "0.67.0",
--      "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.67.0",
--      "integrity": "sha512-1S1Tq3v2O49VOz86Rf40j6u3bSR9V2Ejqz0SeZCc1DRYCy0WnG/e/8cd5676kPpqSEvnYre+HguQufnTqGCjEA==",
++      "version": "0.68.0",
++      "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.68.0",
++      "integrity": "sha512-5XJ/MwzJj4vQlDyMqB/zZf59Bw/Cq7M14UTBJ23uPzXLjRTTQx1jO6/tV76l8/j1C95iuM/Hm7gkFn72X843og==",
        "license": "AGPL-3.0-only",
        "dependencies": {
          "ajv": "^8.17.1",
ddiff --git a/package.json b/package.json
iindex 0e160e2d..a06afd33 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/operator-alerts.test.js && node test/follows.test.js && node test/follows-authority.test.js && node test/status-incidents.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/admin-role-events.test.js && node test/grants-admin.test.js && node test/events-ops.test.js && node test/deploy-drift.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/developer-export-tokens.test.js && node test/project-usage.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/notification-events.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 && node test/blocks.test.js && node test/site-config.test.js && node test/perf-budget.test.js && node test/shutdown.test.js && node test/loyalty-policy.test.js && node test/themes-review.test.js && node test/page-checks.test.js && node test/operator-checklist.test.js && node test/release-health.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/operator-alerts.test.js && node test/follows.test.js && node test/follows-authority.test.js && node test/creator-analytics.test.js && node test/status-incidents.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/admin-role-events.test.js && node test/grants-admin.test.js && node test/events-ops.test.js && node test/deploy-drift.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/developer-export-tokens.test.js && node test/project-usage.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/notification-events.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 && node test/blocks.test.js && node test/site-config.test.js && node test/perf-budget.test.js && node test/shutdown.test.js && node test/loyalty-policy.test.js && node test/themes-review.test.js && node test/page-checks.test.js && node test/operator-checklist.test.js && node test/release-health.test.js",
      "reset-db": "node server/reset-db.js",
      "grant-admin": "node server/grant-admin.js"
    },
@@@ -23,7 +23,7 @@
      "helmet": "^8.1.0",
      "jsonwebtoken": "^9.0.0",
      "multer": "^1.4.5-lts.1",
--    "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.67.0",
++    "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.68.0",
      "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.11.0",
      "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.22.0",
      "uuid": "^11.1.0",
ddiff --git a/server/analytics/creators.js b/server/analytics/creators.js
nnew file mode 100644
iindex 00000000..fc6202c0
---- /dev/null
++++ b/server/analytics/creators.js
@@@ -0,0 +1,137 @@
++'use strict';
++/**
++ * Creator analytics (roadmap WS-E task 6; Contracts 0.68.0 live.stream.ended `stats`,
++ * network.creator-analytics-result@1, network.analytics.creator.read).
++ *
++ * Built from events, not from Live's tables: every live.stream.ended (Network's Events consumer,
++ * server/notifications/events-consumer.js) becomes one creator_streams row, keyed by the stream id, so a
++ * redelivery replaces rather than adds. Only counts are kept, per stream and per creator subject: never who
++ * watched or chatted, no IP address, no viewer or chatter id (ADR-021; test/creator-analytics.test.js checks
++ * the columns). Rows older than 400 days are pruned.
++ *
++ *   GET /api/v1/creators/:creator/analytics?days=1-365   (:creator = usr_ subject or username)
++ *       everyone: streams, stream seconds and peak viewers, daily and per stream (the last 100)
++ *       the creator (signed in), or a service with network.analytics.creator.read (Live's dashboards):
++ *       also average viewers, unique chatters, messages and watch minutes (`full: true`)
++ */
++const express = require('express');
++const { http } = require('openvibe-contracts');
++
++const SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
++const NAME_RE = /^[A-Za-z0-9_]{1,64}$/;
++const KEEP_DAYS = 400;
++const COLUMNS = ['stream_id', 'creator_subject', 'title', 'category', 'started_at', 'ended_at', 'duration_seconds',
++    'peak_viewers', 'avg_viewers', 'unique_chatters', 'messages', 'watch_minutes', 'received_at'];
++
++function ensureSchema(db) {
++    db.exec(`CREATE TABLE IF NOT EXISTS creator_streams (
++        stream_id        INTEGER PRIMARY KEY,
++        creator_subject  TEXT NOT NULL,
++        title            TEXT,
++        category         TEXT,
++        started_at       TEXT NOT NULL,
++        ended_at         TEXT NOT NULL,
++        duration_seconds INTEGER NOT NULL DEFAULT 0,
++        peak_viewers     INTEGER,
++        avg_viewers      REAL,
++        unique_chatters  INTEGER,
++        messages         INTEGER,
++        watch_minutes    INTEGER,
++        received_at      TEXT NOT NULL
++    );
++    CREATE INDEX IF NOT EXISTS idx_creator_streams_creator ON creator_streams(creator_subject, started_at);`);
++}
++
++const int = (v) => (Number.isFinite(Number(v)) && Number(v) >= 0 ? Math.round(Number(v)) : null);
++
++/** One live.stream.ended envelope → 'analytics:recorded' | 'ignored:<why>'. Inside the consumer's inbox claim. */
++function record(db, event, { now = new Date().toISOString() } = {}) {
++    ensureSchema(db);
++    if (event.source !== 'live') return 'ignored:source';
++    const p = event.payload && typeof event.payload === 'object' ? event.payload : {};
++    const creator = p.channel && p.channel.subject && p.channel.subject.id;
++    const streamId = Number(p.stream_id);
++    if (!SUBJECT_RE.test(String(creator || '')) || !Number.isSafeInteger(streamId) || streamId < 1) return 'ignored:payload';
++    const started = Date.parse(p.started_at); const ended = Date.parse(p.ended_at);
++    if (!Number.isFinite(started) || !Number.isFinite(ended) || ended < started) return 'ignored:times';
++    const s = p.stats && typeof p.stats === 'object' ? p.stats : {};
++    db.prepare(`INSERT INTO creator_streams (stream_id, creator_subject, title, category, started_at, ended_at, duration_seconds, peak_viewers, avg_viewers, unique_chatters, messages, watch_minutes, received_at)
++                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
++                ON CONFLICT(stream_id) DO UPDATE SET creator_subject = excluded.creator_subject, title = excluded.title, category = excluded.category,
++                    started_at = excluded.started_at, ended_at = excluded.ended_at, duration_seconds = excluded.duration_seconds,
++                    peak_viewers = excluded.peak_viewers, avg_viewers = excluded.avg_viewers, unique_chatters = excluded.unique_chatters,
++                    messages = excluded.messages, watch_minutes = excluded.watch_minutes, received_at = excluded.received_at`)
++        .run(streamId, creator, p.title == null ? null : String(p.title).slice(0, 300), p.category == null ? null : String(p.category).slice(0, 100),
++            new Date(started).toISOString(), new Date(ended).toISOString(), int(p.duration_seconds) ?? Math.round((ended - started) / 1000),
++            int(s.peak_viewers), Number.isFinite(Number(s.avg_viewers)) && Number(s.avg_viewers) >= 0 ? Math.round(Number(s.avg_viewers) * 10) / 10 : null,
++            int(s.unique_chatters), int(s.messages), int(s.watch_minutes), now);
++    db.prepare('DELETE FROM creator_streams WHERE ended_at < ?').run(new Date(Date.parse(now) - KEEP_DAYS * 86400000).toISOString());
++    return 'analytics:recorded';
++}
++
++/** network.creator-analytics-result@1 for a creator over `days`; `full` adds the audience figures. */
++function query(db, creator, { days = 30, full = false, now = Date.now() } = {}) {
++    ensureSchema(db);
++    const d = Math.min(Math.max(parseInt(days, 10) || 30, 1), 365);
++    const since = new Date(now - d * 86400000).toISOString();
++    const rows = db.prepare('SELECT * FROM creator_streams WHERE creator_subject = ? AND started_at >= ? ORDER BY started_at DESC').all(creator, since);
++    const pick = (r) => {
++        const o = { stream_id: r.stream_id, title: r.title, category: r.category, started_at: r.started_at, ended_at: r.ended_at, duration_seconds: r.duration_seconds };
++        if (r.peak_viewers != null) o.peak_viewers = r.peak_viewers;
++        if (full) for (const k of ['avg_viewers', 'unique_chatters', 'messages', 'watch_minutes']) if (r[k] != null) o[k] = r[k];
++        return o;
++    };
++    const add = (acc, r) => {
++        acc.streams += 1; acc.stream_seconds += r.duration_seconds || 0; acc.peak_viewers = Math.max(acc.peak_viewers, r.peak_viewers || 0);
++        if (full) {
++            acc.unique_chatters += r.unique_chatters || 0; acc.messages += r.messages || 0; acc.watch_minutes += r.watch_minutes || 0;
++            acc._avgWeighted += (r.avg_viewers || 0) * (r.duration_seconds || 0);
++        }
++        return acc;
++    };
++    const blank = () => ({ streams: 0, stream_seconds: 0, peak_viewers: 0, ...(full ? { unique_chatters: 0, messages: 0, watch_minutes: 0, _avgWeighted: 0 } : {}) });
++    const finish = (acc) => {
++        if (full) { acc.avg_viewers = acc.stream_seconds ? Math.round((acc._avgWeighted / acc.stream_seconds) * 10) / 10 : 0; delete acc._avgWeighted; }
++        return acc;
++    };
++    const totals = finish(rows.reduce(add, blank()));
++    const byDay = new Map();
++    for (const r of rows) {
++        const day = r.started_at.slice(0, 10);
++        byDay.set(day, add(byDay.get(day) || blank(), r));
++    }
++    const daily = [...byDay.entries()].sort((a, b) => a[0].localeCompare(b[0])).map(([day, acc]) => ({ day, ...finish(acc) }));
++    return { creator, days: d, full: !!full, totals, daily, streams: rows.slice(0, 100).map(pick) };
++}
++
++function viewerOf(req) {
++    const h = req.headers.authorization || '';
++    const token = h.startsWith('Bearer ') ? h.slice(7) : (req.cookies && req.cookies.ov_token);
++    if (!token) return null;
++    try {
++        const out = require('../auth/session').verifySession(token, { db: req.app.locals.db, publicKey: req.app.locals.publicKey, config: req.app.locals.config });
++        return out && out.user ? out.user : null;
++    } catch { return null; }
++}
++
++/** Routes; fullGuard = principals.guard('network.analytics.creator.read', { legacy: false }). */
++function router({ fullGuard }) {
++    const r = express.Router();
++    r.use(http.middleware());
++    r.get('/:creator/analytics', (req, res) => {
++        const db = req.app.locals.db;
++        const key = String(req.params.creator || '');
++        const u = SUBJECT_RE.test(key) ? db.prepare('SELECT subject_id, is_anon FROM users WHERE subject_id = ?').get(key)
++            : NAME_RE.test(key) ? db.prepare('SELECT subject_id, is_anon FROM users WHERE username = ? COLLATE NOCASE').get(key) : null;
++        if (!u || u.is_anon || !SUBJECT_RE.test(String(u.subject_id || ''))) return http.sendProblem(res, 404, 'analytics.unknown_creator', { detail: 'no such creator', ctx: req.ov });
++        const answer = (full) => res.set('Cache-Control', full ? 'private, no-store' : 'public, max-age=60').json(query(db, u.subject_id, { days: req.query.days, full }));
++        const viewer = viewerOf(req);
++        if (viewer) return answer(viewer.subject_id === u.subject_id);
++        const h = String(req.headers.authorization || '');
++        if (h.startsWith('Bearer ')) return fullGuard(req, res, () => answer(true));   // a service token: full, or refused
++        return answer(false);
++    });
++    return r;
++}
++
++module.exports = { ensureSchema, record, query, router, COLUMNS, KEEP_DAYS };
ddiff --git a/server/db/database.js b/server/db/database.js
iindex f132ddd7..1b9b0d9a 100644
---- a/server/db/database.js
++++ b/server/db/database.js
@@@ -752,6 +752,8 @@ function initDb(dbPath) {
      require('../identity/follows').ensureSchema(db);
      // Incidents and maintenance on /status (server/status/incidents.js).
      require('../status/incidents').ensureSchema(db);
++    // Creator analytics from live.stream.ended (server/analytics/creators.js).
++    require('../analytics/creators').ensureSchema(db);
  
      console.log('[DB] Central database initialized');
      return db;
ddiff --git a/server/identity/principals.js b/server/identity/principals.js
iindex 1bbdd909..ca89e67d 100644
---- a/server/identity/principals.js
++++ b/server/identity/principals.js
@@@ -124,6 +124,8 @@ const DEFAULT_GRANTS = [
      // ADR-030 (WS-E task 4): Live rebuilds its follows projection from Network's graph.
      ['live', 'network.follows.read', SELF_AUDIENCE, []],
      ['live', 'network.follows.write', SELF_AUDIENCE, []],
++    // WS-E task 6: Live's creator dashboards read the full analytics built from its events.
++    ['live', 'network.analytics.creator.read', SELF_AUDIENCE, []],
      // WS-H task 11: Host relays the alerts firing on the production host (ovhost alerts relay).
      ['host', 'network.operator.alert', SELF_AUDIENCE, []],
      // WS-N task 12: ovhost incident / maintenance post to the status page.
ddiff --git a/server/index.js b/server/index.js
iindex 2aab60b8..ec633f4e 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -447,6 +447,9 @@ ecosystem.start();
  const libraryTags = require('./registry/library-tags').createLibraryTags({ onUpdate: require('./registry/exposure').setLibraryReleases, token: () => require('./integrations/github').tokenOf(db) });
  libraryTags.start();
  // Operator status: GET /status (server-rendered, noindex), /api/v1/status, /api/v1/status/slo.
++// Creator analytics (WS-E task 6): from live.stream.ended, counts only; the full figures for the creator or
++// network.analytics.creator.read (Live's dashboards).
++app.use('/api/v1/creators', rateLimit({ windowMs: 60_000, max: 300 }), require('./analytics/creators').router({ fullGuard: require('./identity/principals').guard('network.analytics.creator.read', { legacy: false }) }));
  // Incidents and maintenance (WS-N task 12): public list; staff admins or network.status.incident (ovhost) write.
  app.use('/api/v1/status/incidents', rateLimit({ windowMs: 60_000, max: 120 }), require('./status/incidents').router({ requireAuth, incidentGuard: require('./identity/principals').guard('network.status.incident', { legacy: false }) }));
  app.use(require('./status/routes').createStatusRoutes({ ecosystem }));
ddiff --git a/server/notifications/events-consumer.js b/server/notifications/events-consumer.js
iindex 054ed1ce..b88e1872 100644
---- a/server/notifications/events-consumer.js
++++ b/server/notifications/events-consumer.js
@@@ -55,7 +55,7 @@ const CONSUMER = 'network-notifications';
  const INBOX_TABLE = 'network_event_inbox';
  const AUDIT_TOPICS = require('../admin/moderation-audit').TOPICS;
  const USAGE_TOPICS = require('../developer/usage').TOPICS;
--const TOPICS = Object.freeze(['deals.watch.matched', 'trade.alert.triggered', 'live.stream.started', ...AUDIT_TOPICS, ...USAGE_TOPICS]);
++const TOPICS = Object.freeze(['deals.watch.matched', 'trade.alert.triggered', 'live.stream.started', 'live.stream.ended', ...AUDIT_TOPICS, ...USAGE_TOPICS]);
  const LIVE_STARTED_MAX_AGE_MS = 30 * 60 * 1000;
  const EVENT_ID_RE = /^evt_[0-9A-HJKMNP-TV-Z]{26}$/;
  
@@@ -236,6 +236,8 @@ function createEventsConsumer({ db, notifications, secrets, liveFollowers = null
              if (AUDIT_TOPICS.includes(event.event_type)) return moderationAudit ? moderationAudit.record(event) : 'ignored:audit_off';
              // Developer projects' usage rollups go to their dashboards (WS-N task 4), never to an inbox.
              if (USAGE_TOPICS.includes(event.event_type)) return projectUsage ? projectUsage.record(event) : 'ignored:usage_off';
++            // Creator analytics (WS-E task 6): each ended stream's totals, counts only.
++            if (event.event_type === 'live.stream.ended') return require('../analytics/creators').record(db, event);
              if (event.event_type === 'live.stream.started') {
                  const out = liveStreamStarted(event, prep);
                  if (typeof out === 'string') return out;
ddiff --git a/test/creator-analytics.test.js b/test/creator-analytics.test.js
nnew file mode 100644
iindex 00000000..2bdea6fc
---- /dev/null
++++ b/test/creator-analytics.test.js
@@@ -0,0 +1,98 @@
++'use strict';
++// Creator analytics (roadmap WS-E task 6; Contracts 0.68.0): live.stream.ended events become per-stream rows
++// (counts only; a redelivery replaces), the public sees streams, minutes and peak viewers, the creator and
++// services with network.analytics.creator.read see the rest, and nothing stored names a viewer, chatter or IP.
++//   node test/creator-analytics.test.js
++const assert = require('assert');
++const crypto = require('crypto');
++const fs = require('fs');
++const os = require('os');
++const path = require('path');
++const http = require('http');
++const express = require('express');
++const { ids, validate } = require('openvibe-contracts');
++const { signDeliveryHeaders } = require('openvibe-sdk/events');
++const { initDb } = require('../server/db/database');
++const { NotificationService } = require('../server/notifications/notification-service');
++const { createEventsConsumer } = require('../server/notifications/events-consumer');
++const creators = require('../server/analytics/creators');
++
++const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-creator-analytics-'));
++const log = console.log; console.log = () => {};
++const db = initDb(path.join(dir, 'network.db'));
++console.log = log;
++require('../server/identity/principals').ensureSchema(db);
++for (const c of ['live', 'tools']) db.prepare('UPDATE oauth_clients SET client_secret = ? WHERE client_id = ?').run(`${c}-secret`, c);
++const CAROL = ids.newId('user'), DAVE = ids.newId('user');
++db.prepare('INSERT INTO users (id, username, password_hash, subject_id, role) VALUES (20, ?, ?, ?, ?), (21, ?, ?, ?, ?)').run('carol', 'x', CAROL, 'streamer', 'dave', 'x', DAVE, 'user');
++
++const keys = crypto.generateKeyPairSync('rsa', { modulusLength: 2048, publicKeyEncoding: { type: 'spki', format: 'pem' }, privateKeyEncoding: { type: 'pkcs8', format: 'pem' } });
++const config = { internalKey: 'legacy-key', jwt: { issuer: 'https://openvibe.network', accessTokenExpiry: '1h' } };
++const { signToken } = require('../server/auth/routes');
++const SECRET = 's'.repeat(40);
++const consumer = createEventsConsumer({ db, notifications: new NotificationService(db), secrets: SECRET, log: { log() {}, warn() {}, error() {} } });
++const app = express();
++Object.assign(app.locals, { db, config, privateKey: keys.privateKey, publicKey: keys.publicKey });
++app.use(require('cookie-parser')());
++app.use(express.urlencoded({ extended: true }));
++app.use('/oauth', require('../server/auth/oauth-routes'));
++app.use('/internal/events', consumer.router);
++app.use('/api/v1/creators', creators.router({ fullGuard: require('../server/identity/principals').guard('network.analytics.creator.read', { legacy: false }) }));
++const server = http.createServer(app);
++
++const ended = (streamId, startedAt, minutes, stats) => ({
++    event_id: ids.newId('event'), event_type: 'live.stream.ended', version: 1, source: 'live',
++    actor: { type: 'user', id: CAROL }, subject: { type: 'stream', id: String(streamId), revision: 2 }, visibility: 'public', priority: 'important',
++    occurred_at: new Date().toISOString(),
++    payload: { stream_id: streamId, channel: { username: 'carol', display_name: 'Carol', url: 'https://openvibe.live/@carol', subject: { type: 'user', id: CAROL } }, title: `Stream ${streamId}`, category: null, protocol: 'rtmp', is_nsfw: false,
++        started_at: startedAt, ended_at: new Date(Date.parse(startedAt) + minutes * 60000).toISOString(), duration_seconds: minutes * 60, ...(stats ? { stats } : {}) },
++});
++
++(async () => {
++    await new Promise((r) => server.listen(0, '127.0.0.1', r));
++    const base = `http://127.0.0.1:${server.address().port}`;
++    const deliver = async (event) => { const raw = JSON.stringify({ event, seq: 1 }); const res = await fetch(`${base}/internal/events`, { method: 'POST', headers: { 'content-type': 'application/json', ...signDeliveryHeaders(raw, SECRET) }, body: raw }); return res.json(); };
++    const call = (p, headers = {}) => fetch(base + p, { headers }).then(async (r) => ({ status: r.status, cache: r.headers.get('cache-control'), body: await r.json() }));
++    const svc = async (client) => (await (await fetch(`${base}/oauth/token`, { method: 'POST', body: new URLSearchParams({ grant_type: 'client_credentials', client_id: client, client_secret: `${client}-secret`, audience: 'openvibe.network' }) })).json()).access_token;
++    try {
++        const day = (h) => new Date(Date.now() - h * 3600000).toISOString();
++        const e1 = ended(501, day(30), 60, { peak_viewers: 12, avg_viewers: 6.5, unique_chatters: 4, messages: 120, watch_minutes: 300 });
++        assert.strictEqual((await deliver(e1)).outcome, 'analytics:recorded');
++        assert.strictEqual((await deliver(ended(502, day(2), 30, { peak_viewers: 20, avg_viewers: 10, unique_chatters: 9, messages: 200, watch_minutes: 250 }))).outcome, 'analytics:recorded');
++        assert.strictEqual((await deliver(ended(503, day(1), 15))).outcome, 'analytics:recorded', 'an ended stream without stats still counts');
++        assert.strictEqual((await deliver({ ...ended(501, day(30), 60, { peak_viewers: 12, avg_viewers: 6.5, unique_chatters: 4, messages: 120, watch_minutes: 300 }) })).outcome, 'analytics:recorded', 'a new event for the same stream replaces it');
++        assert.strictEqual(db.prepare('SELECT COUNT(*) AS n FROM creator_streams').get().n, 3);
++        assert.strictEqual((await deliver({ ...ended(504, day(1), 5), source: 'media' })).outcome, 'ignored:source');
++
++        // Public: streams, minutes, peak. No audience figures.
++        let r = await call('/api/v1/creators/carol/analytics?days=30');
++        assert.strictEqual(r.status, 200);
++        assert.ok(validate('network.creator-analytics-result@1', r.body).valid, JSON.stringify(validate('network.creator-analytics-result@1', r.body).errors));
++        assert.deepStrictEqual([r.body.full, r.body.totals], [false, { streams: 3, stream_seconds: 6300, peak_viewers: 20 }]);
++        assert.ok(!r.body.streams.some((s) => 'messages' in s || 'unique_chatters' in s));
++        assert.match(r.cache, /public/);
++        assert.strictEqual(r.body.daily.reduce((n, d) => n + d.streams, 0), 3);
++
++        // The creator, and Live's service token: full figures. Another person: public only. A service without the grant: refused.
++        const creatorTok = { authorization: `Bearer ${signToken(db.prepare('SELECT * FROM users WHERE id = 20').get(), keys.privateKey, config)}` };
++        r = await call(`/api/v1/creators/${CAROL}/analytics`, creatorTok);
++        assert.deepStrictEqual([r.body.full, r.body.totals.messages, r.body.totals.unique_chatters, r.body.totals.watch_minutes], [true, 320, 13, 550]);
++        assert.strictEqual(r.body.totals.avg_viewers, Math.round(((6.5 * 3600 + 10 * 1800) / 6300) * 10) / 10, 'duration-weighted average');
++        assert.match(r.cache, /private/);
++        r = await call('/api/v1/creators/carol/analytics', { authorization: `Bearer ${signToken(db.prepare('SELECT * FROM users WHERE id = 21').get(), keys.privateKey, config)}` });
++        assert.strictEqual(r.body.full, false);
++        r = await call('/api/v1/creators/carol/analytics', { authorization: `Bearer ${await svc('live')}` });
++        assert.strictEqual(r.body.full, true);
++        assert.strictEqual((await call('/api/v1/creators/carol/analytics', { authorization: `Bearer ${await svc('tools')}` })).status, 403);
++        assert.strictEqual((await call('/api/v1/creators/nobody/analytics')).status, 404);
++
++        // Privacy (ADR-021): the table holds counts and the creator's subject only.
++        const cols = db.prepare('PRAGMA table_info(creator_streams)').all().map((c) => c.name);
++        assert.deepStrictEqual(cols, creators.COLUMNS);
++        assert.ok(!cols.some((c) => /ip|viewer_id|chatter_id|user_id|follower/.test(c)), 'no viewer, chatter or IP column');
++    } finally {
++        server.close();
++        fs.rmSync(dir, { recursive: true, force: true });
++    }
++    console.log('creator analytics: all checks passed');
++})().catch((e) => { console.error(e); process.exit(1); });
ddiff --git a/test/events-consumer.test.js b/test/events-consumer.test.js
iindex b62be8a5..72402389 100644
---- a/test/events-consumer.test.js
++++ b/test/events-consumer.test.js
@@@ -58,7 +58,7 @@ const rows = (userId) => db.prepare('SELECT * FROM notifications WHERE user_id =
  
  (async () => {
      // The subscriptions this consumer is meant for, and nothing else.
--    assert.deepStrictEqual(TOPICS, ['deals.watch.matched', 'trade.alert.triggered', 'live.stream.started', 'chat.moderation.action', 'live.moderation.action', 'community.moderation.action', 'tips.interaction.moderated', 'billing.staff.action',
++    assert.deepStrictEqual(TOPICS, ['deals.watch.matched', 'trade.alert.triggered', 'live.stream.started', 'live.stream.ended', 'chat.moderation.action', 'live.moderation.action', 'community.moderation.action', 'tips.interaction.moderated', 'billing.staff.action',
          ...['media', 'tools', 'games', 'wiki', 'blog', 'news', 'reviews', 'deals', 'coupons', 'trade', 'codes'].map((svc) => `${svc}.moderation.action`),
          'tools.usage.recorded', 'events.usage.recorded']);
      assert.deepStrictEqual(topicsFrom([]), TOPICS, 'scripts/subscribe-events.js subscribes every topic');
ddiff --git a/test/principals.test.js b/test/principals.test.js
iindex 4cb34250..03805678 100644
---- a/test/principals.test.js
++++ b/test/principals.test.js
@@@ -50,7 +50,7 @@ const server = http.createServer(app);
      const claims = serviceAuth.verifyServiceToken(t.body.access_token, { publicKey: keys.publicKey, issuer: ISSUER, audience: 'openvibe.network' });
      assert.ok(claims.ok, claims.reason);
      assert.strictEqual(claims.claims.sub, 'svc:live');
--    assert.deepStrictEqual(claims.claims.cap, ['identity.subject.resolve', 'network.coins.credit', 'network.coins.debit', 'network.follows.read', 'network.follows.write', 'network.modules.read', 'network.modules.write', 'network.notifications.push']);
++    assert.deepStrictEqual(claims.claims.cap, ['identity.subject.resolve', 'network.analytics.creator.read', 'network.coins.credit', 'network.coins.debit', 'network.follows.read', 'network.follows.write', 'network.modules.read', 'network.modules.write', 'network.notifications.push']);
      assert.ok(validate('identity.service-token-claims@1', claims.claims).valid);
      assert.ok(claims.claims.exp - claims.claims.iat <= 300, 'short-lived');
      const full = t.body.access_token;