openvibers/openvibe.live Commits

Creator analytics through Network (roadmap WS-E task 6; Contracts v0.68.0, pinned from 0.56.0). live.stream.ended carries the stream's totals (stats: peak and average viewers, unique chatters, messages, watch minutes; counts only), computed in the transaction that ends the row. With ANALYTICS_SOURCE=network, /api/analytics/channel/:username and its dashboard read Network's creator analytics with Live's service token (network.analytics.creator.read) and merge Live's own per-stream extras (coins, new followers, clips). Unset, or with no subject or Network down, they use Live's tables as before. test/network-analytics.test.js

OP OpenVibers committed 26 Sept 2026 · 83ad224f32 · 7 files, +147 −9


Changes

ddiff --git a/package-lock.json b/package-lock.json
iindex a2687859..f66b2cc5 100644
---- a/package-lock.json
++++ b/package-lock.json
@@@ -25,7 +25,7 @@
"multer": "^1.4.5-lts.1",
"node-media-server": "^2.7.1",
"onnxruntime-node": "^1.20.0",
-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.56.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",
"sanitize-html": "^2.17.7",
@@@ -2814,9 +2814,9 @@
}
},
"node_modules/openvibe-contracts": {
-- "version": "0.56.0",
-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.56.0",
-- "integrity": "sha512-g8i+4dKCwaUyfBDsxYm+WodSn2UFtGgV0QZvj6hAqHCKK8+W1scyZ1/YyrFJTS4OWaMkaOXepkUFuUmBYVdD6A==",
++ "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 9288d2d3..7c54d83c 100644
---- a/package.json
++++ b/package.json
@@@ -44,7 +44,7 @@
"multer": "^1.4.5-lts.1",
"node-media-server": "^2.7.1",
"onnxruntime-node": "^1.20.0",
-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.56.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",
"sanitize-html": "^2.17.7",
ddiff --git a/server/analytics/network-analytics.js b/server/analytics/network-analytics.js
nnew file mode 100644
iindex 00000000..6ea263fc
---- /dev/null
++++ b/server/analytics/network-analytics.js
@@@ -0,0 +1,69 @@
++'use strict';
++/**
++ * Creator analytics read from OpenVibe.Network (roadmap WS-E task 6). Network builds them from Live's
++ * live.stream.ended events (Contracts 0.68.0 `stats`: counts only) at GET /api/v1/creators/:subject/analytics;
++ * Live reads them with its service token (network.analytics.creator.read, the full figures) when
++ * ANALYTICS_SOURCE=network, and merges what only Live knows per stream (coins earned, new followers, clips).
++ *
++ * summaryFor(userId, days) → the shape of db.getChannelAnalyticsSummary (plus source: 'network'), or null when
++ * Network is not the source, the channel has no Network subject, or Network did not answer (the caller then
++ * uses Live's own tables). all_time stays Live's (every stream it ever ran).
++ */
++const db = require('../db/database');
++
++const NETWORK_INTERNAL_URL = (process.env.OV_NETWORK_INTERNAL_URL || 'http://127.0.0.1:4000').replace(/\/+$/, '');
++const SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
++const stats = { network: 0, fallback: 0 };
++
++const enabled = () => process.env.ANALYTICS_SOURCE === 'network';
++
++async function summaryFor(userId, days, { fetchImpl = globalThis.fetch } = {}) {
++ if (!enabled()) return null;
++ const r0 = db.getDb().prepare("SELECT subject_id FROM linked_accounts WHERE service = 'network' AND user_id = ?").get(userId);
++ const subject = r0 && SUBJECT_RE.test(String(r0.subject_id || '')) ? r0.subject_id : null;
++ if (!subject) { stats.fallback++; return null; }
++ let body;
++ try {
++ const principal = require('../net/network-principal');
++ const res = await fetchImpl(`${NETWORK_INTERNAL_URL}/api/v1/creators/${subject}/analytics?days=${Math.min(Math.max(parseInt(days, 10) || 30, 1), 365)}`, {
++ headers: { Accept: 'application/json', ...(await principal.serviceHeaders('openvibe.network')) }, signal: AbortSignal.timeout(4000),
++ });
++ if (res.status !== 200) { stats.fallback++; return null; }
++ body = await res.json();
++ } catch { stats.fallback++; return null; }
++ if (!body || !body.full || !Array.isArray(body.streams)) { stats.fallback++; return null; }
++ const local = new Map();
++ const q = db.getDb().prepare('SELECT stream_id, new_followers, clips_created, coins_earned FROM stream_analytics WHERE stream_id = ?');
++ for (const s of body.streams) { const row = q.get(s.stream_id); if (row) local.set(s.stream_id, row); }
++ const streams = body.streams.map((s) => {
++ const l = local.get(s.stream_id) || {};
++ return {
++ id: s.stream_id, title: s.title, category: s.category, started_at: s.started_at, ended_at: s.ended_at, duration_seconds: s.duration_seconds,
++ peak_viewers: s.peak_viewers ?? null, viewer_count: null, avg_viewers: s.avg_viewers ?? null, unique_chatters: s.unique_chatters ?? null,
++ total_messages: s.messages ?? null, total_watch_minutes: s.watch_minutes ?? null,
++ new_followers: l.new_followers ?? null, clips_created: l.clips_created ?? null, coins_earned: l.coins_earned ?? null,
++ };
++ });
++ const sum = (k) => streams.reduce((n, s) => n + (Number(s[k]) || 0), 0);
++ const t = body.totals || {};
++ const base = db.getChannelAnalyticsSummary(userId, days); // all_time and follower count stay Live's
++ stats.network++;
++ return {
++ ...base,
++ source: 'network',
++ streams,
++ summary: {
++ total_streams: t.streams || 0,
++ total_duration_seconds: t.stream_seconds || 0,
++ peak_viewers: t.peak_viewers || 0,
++ avg_viewers_per_stream: t.avg_viewers || 0,
++ total_messages: t.messages || 0,
++ total_unique_chatters: t.unique_chatters || 0,
++ total_watch_minutes: t.watch_minutes || 0,
++ total_new_followers: sum('new_followers'),
++ total_clips: sum('clips_created'),
++ },
++ };
++}
++
++module.exports = { summaryFor, enabled, stats };
ddiff --git a/server/events/stream-events.js b/server/events/stream-events.js
iindex 31103d90..38dc2b1f 100644
---- a/server/events/stream-events.js
++++ b/server/events/stream-events.js
@@@ -53,6 +53,15 @@ function envelopeFor(kind, streamId) {
if (kind === 'ended') {
payload.ended_at = toIso(s.ended_at);
payload.duration_seconds = s.duration_seconds == null ? null : Number(s.duration_seconds);
++ // Contracts 0.68.0 (WS-E task 6): the stream's totals for creator analytics on Network. Counts only,
++ // computed here in the transaction that ends the row (stream_analytics is refreshed as a side effect).
++ try {
++ const a = db.computeAndCacheStreamAnalytics(s.id);
++ if (a) {
++ const n = (v) => Math.max(0, Math.round(Number(v) || 0));
++ payload.stats = { peak_viewers: n(a.peak_viewers), avg_viewers: Math.max(0, Math.round((Number(a.avg_viewers) || 0) * 10) / 10), unique_chatters: n(a.unique_chatters), messages: n(a.total_messages), watch_minutes: n(a.total_watch_minutes) };
++ }
++ } catch { /* the event goes without stats rather than not at all */ }
}
return {
event_type: `live.stream.${kind}`,
ddiff --git a/server/streaming/analytics-routes.js b/server/streaming/analytics-routes.js
iindex d557fde2..78bcf5cb 100644
---- a/server/streaming/analytics-routes.js
++++ b/server/streaming/analytics-routes.js
@@@ -20,13 +20,14 @@ const router = express.Router();
// ── Public: Channel summary stats ────────────────────────────
// Anyone can see aggregate stats for a channel
--router.get('/channel/:username', optionalAuth, (req, res) => {
++router.get('/channel/:username', optionalAuth, async (req, res) => {
try {
const channel = db.getChannelByUsername(req.params.username);
if (!channel) return res.status(404).json({ error: 'Channel not found' });
const days = Math.min(parseInt(req.query.days) || 30, 365);
-- const data = db.getChannelAnalyticsSummary(channel.user_id, days);
++ // WS-E task 6: from Network's creator analytics when ANALYTICS_SOURCE=network, else (or on any failure) Live's tables.
++ const data = (await require('../analytics/network-analytics').summaryFor(channel.user_id, days)) || db.getChannelAnalyticsSummary(channel.user_id, days);
// Public view — return summary + stream list (no revenue data)
res.json({
@@@ -154,7 +155,7 @@ router.get('/stream/:id', (req, res) => {
// ── Streamer-only: Detailed dashboard ────────────────────────
// Shows extra data like watch minutes, coins, followers over time
--router.get('/channel/:username/dashboard', optionalAuth, (req, res) => {
++router.get('/channel/:username/dashboard', optionalAuth, async (req, res) => {
try {
const channel = db.getChannelByUsername(req.params.username);
if (!channel) return res.status(404).json({ error: 'Channel not found' });
@@@ -164,7 +165,8 @@ router.get('/channel/:username/dashboard', optionalAuth, (req, res) => {
if (!isOwner) return res.status(403).json({ error: 'Access denied' });
const days = Math.min(parseInt(req.query.days) || 30, 365);
-- const data = db.getChannelAnalyticsSummary(channel.user_id, days);
++ // WS-E task 6: from Network's creator analytics when ANALYTICS_SOURCE=network, else (or on any failure) Live's tables.
++ const data = (await require('../analytics/network-analytics').summaryFor(channel.user_id, days)) || db.getChannelAnalyticsSummary(channel.user_id, days);
// Add streamer-only data: watch minutes, coins earned, follower growth
const streams = data.streams.map(s => ({
ddiff --git a/test/network-analytics.test.js b/test/network-analytics.test.js
nnew file mode 100644
iindex 00000000..0ac36c13
---- /dev/null
++++ b/test/network-analytics.test.js
@@@ -0,0 +1,54 @@
++'use strict';
++// Creator analytics from Network (roadmap WS-E task 6; server/analytics/network-analytics.js): with
++// ANALYTICS_SOURCE=network Live reads Network's figures with its service token and merges its own per-stream
++// extras; unset, no subject, or Network down → Live's own tables (null here).
++const assert = require('assert');
++const fs = require('fs');
++const os = require('os');
++const path = require('path');
++
++const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-netanalytics-'));
++process.env.DB_PATH = path.join(tmp, 'live.db');
++process.env.OV_NETWORK_INTERNAL_URL = 'http://network.test';
++const log = console.log; console.log = () => {}; console.warn = () => {};
++const db = require('../server/db/database');
++db.initDb();
++console.log = log;
++require('../server/net/network-principal').serviceHeaders = async () => ({ Authorization: 'Bearer svc' });
++const na = require('../server/analytics/network-analytics');
++const SUBJ = `usr_${'01JAA'.padEnd(26, '0')}`;
++const d = db.getDb();
++d.prepare("INSERT INTO users (id, username, password_hash) VALUES (7, 'carol', 'x'), (8, 'nolink', 'x')").run();
++d.prepare("INSERT INTO linked_accounts (user_id, service, service_user_id, subject_id) VALUES (7, 'network', '70', ?)").run(SUBJ);
++d.prepare('INSERT INTO streams (id, user_id, title) VALUES (501, 7, \'A\')').run();
++d.prepare('INSERT INTO stream_analytics (stream_id, new_followers, clips_created, coins_earned) VALUES (501, 3, 2, 40)').run();
++
++(async () => {
++ const calls = [];
++ let status = 200;
++ const body = { creator: SUBJ, days: 30, full: true, totals: { streams: 2, stream_seconds: 5400, peak_viewers: 20, avg_viewers: 7.7, unique_chatters: 13, messages: 320, watch_minutes: 550 }, daily: [],
++ streams: [{ stream_id: 501, title: 'A', category: null, started_at: '2026-09-25T10:00:00.000Z', ended_at: '2026-09-25T11:00:00.000Z', duration_seconds: 3600, peak_viewers: 12, avg_viewers: 6.5, unique_chatters: 4, messages: 120, watch_minutes: 300 },
++ { stream_id: 502, title: 'B', category: null, started_at: '2026-09-26T10:00:00.000Z', ended_at: '2026-09-26T10:30:00.000Z', duration_seconds: 1800, peak_viewers: 20, avg_viewers: 10, unique_chatters: 9, messages: 200, watch_minutes: 250 }] };
++ const fetchImpl = async (url, o) => { calls.push({ url, auth: o.headers.Authorization }); if (status instanceof Error) throw status; return { status, json: async () => body }; };
++
++ delete process.env.ANALYTICS_SOURCE;
++ assert.strictEqual(await na.summaryFor(7, 30, { fetchImpl }), null, 'off by default');
++ process.env.ANALYTICS_SOURCE = 'network';
++ const r = await na.summaryFor(7, 30, { fetchImpl });
++ assert.deepStrictEqual(calls.pop(), { url: `http://network.test/api/v1/creators/${SUBJ}/analytics?days=30`, auth: 'Bearer svc' });
++ assert.strictEqual(r.source, 'network');
++ assert.deepStrictEqual([r.summary.total_streams, r.summary.total_messages, r.summary.total_watch_minutes, r.summary.avg_viewers_per_stream], [2, 320, 550, 7.7]);
++ assert.deepStrictEqual([r.summary.total_new_followers, r.summary.total_clips], [3, 2], 'Live\'s own extras merged');
++ assert.deepStrictEqual([r.streams[0].id, r.streams[0].total_messages, r.streams[0].coins_earned], [501, 120, 40]);
++ assert.ok(r.all_time, 'all_time stays Live\'s');
++ assert.strictEqual(await na.summaryFor(8, 30, { fetchImpl }), null, 'no subject: Live\'s tables');
++ status = 503;
++ assert.strictEqual(await na.summaryFor(7, 30, { fetchImpl }), null, 'Network refused: Live\'s tables');
++ status = new Error('ECONNREFUSED');
++ assert.strictEqual(await na.summaryFor(7, 30, { fetchImpl }), null);
++ const routes = fs.readFileSync(path.join(__dirname, '..', 'server', 'streaming', 'analytics-routes.js'), 'utf8');
++ assert.strictEqual((routes.match(/network-analytics'\)\.summaryFor\(channel\.user_id, days\)\) \|\| db\.getChannelAnalyticsSummary/g) || []).length, 2, 'both channel analytics routes');
++ fs.rmSync(tmp, { recursive: true, force: true });
++ console.log('network analytics: all checks passed');
++ process.exit(0);
++})().catch((e) => { console.error(e); process.exit(1); });
ddiff --git a/test/stream-events.test.js b/test/stream-events.test.js
iindex 561a003f..a55890ef 100644
---- a/test/stream-events.test.js
++++ b/test/stream-events.test.js
@@@ -87,6 +87,10 @@ const stub = http.createServer((req, res) => {
assert.strictEqual(published[1].subject.revision, 2);
assert.ok('duration_seconds' in published[1].payload);
require('openvibe-contracts').assertValid('live.stream.ended@1', published[1].payload);
++ // Contracts 0.68.0: the stream's totals ride on the ended event (counts only) for creator analytics on Network.
++ const st = published[1].payload.stats;
++ assert.ok(st && ['peak_viewers', 'avg_viewers', 'unique_chatters', 'messages', 'watch_minutes'].every((k) => typeof st[k] === 'number' && st[k] >= 0), JSON.stringify(st));
++ assert.deepStrictEqual(Object.keys(st).sort(), ['avg_viewers', 'messages', 'peak_viewers', 'unique_chatters', 'watch_minutes'], 'no people in it');
// A rolled-back go-live leaves no event.
assert.throws(() => d.transaction(() => { db.createStream({ user_id: 501, title: 'rolled back' }); throw new Error('abort'); })());