openvibers/openvibe.live Commits

Bot analytics rows are kept 7 days (they are in the rollups by then; ~1.15M rows), and a first stream gets one follow-up with its recording and the channel address (10 of 41 first-time streamers came back) (#308)

OP OpenVibers committed 22 days ago · d89e2312ae · 5 files, +177 −4

Changes

ddiff --git a/server/analytics/store.js b/server/analytics/store.js
iindex ad82ee8b..3fff7c87 100644
---- a/server/analytics/store.js
++++ b/server/analytics/store.js
@@@ -19,6 +19,26 @@ function processDb() {
});
}
++// Crawler and bot rows are two thirds of the raw table (2026-10-22: 1.65M of 2.47M rows in 30 days, 1.49M of them bot
++// API calls; the table was 1.1 GB, 70% of ov_live). They are counted in the hourly and daily rollups when their day
++// closes (the tracker recomputes only today and yesterday), so after a week only the admin's bot breakdown would read
++// them: they are kept 7 days, the window its suspicious-sessions check already uses, and people's events 30.
++const BOT_RAW_DAYS = 7;
++const DAY_MS = 24 * 60 * 60 * 1000;
++const sqlTime = (ms) => new Date(ms).toISOString().replace('T', ' ').slice(0, 19);
++
++/** Bot rows older than BOT_RAW_DAYS, in batches as the shared prune does. */
++async function pruneBotEvents(db, { days = BOT_RAW_DAYS, now = Date.now, batch = 5000 } = {}) {
++ const cutoff = sqlTime(now() - days * DAY_MS);
++ let removed = 0;
++ for (;;) {
++ const n = await db.exec(`DELETE FROM analytics_events WHERE id IN (SELECT id FROM analytics_events WHERE is_bot = 1 AND created_at < $1 LIMIT ${batch})`, [cutoff]);
++ removed += n;
++ if (n < batch) break;
++ }
++ return { removed, cutoff };
++}
++
function createAnalyticsStore({ db = processDb(), drill = false, timers = !drill } = {}) {
const tracker = new AnalyticsTrackerPg(db, 'live', { retention: false, timers });
let closing;
@@@ -26,7 +46,11 @@ function createAnalyticsStore({ db = processDb(), drill = false, timers = !drill
tracker,
// The owner's migration made the table; the runtime role has no DDL rights.
async ready() { await db.value('SELECT 1 FROM analytics_events LIMIT 1'); },
-- prune: () => pruneRawEventsPg(db, { days: analyticsModule.retention.MAX_DAYS }),
++ async prune() {
++ const all = await pruneRawEventsPg(db, { days: analyticsModule.retention.MAX_DAYS });
++ const bots = await pruneBotEvents(db);
++ return { ...all, bots };
++ },
close() {
if (!closing) closing = Promise.resolve(tracker.destroy());
return closing;
@@@ -34,4 +58,4 @@ function createAnalyticsStore({ db = processDb(), drill = false, timers = !drill
};
}
--module.exports = { createAnalyticsStore };
++module.exports = { createAnalyticsStore, pruneBotEvents, BOT_RAW_DAYS };
ddiff --git a/server/index.js b/server/index.js
iindex 7a86090d..9fb0c2d4 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -1631,14 +1631,22 @@ async function start() {
},
{ initialDelayMs: 3 * 60 * 1000, jitterMs: 60 * 1000 });
-- // 8b2. Raw analytics retention (ADR-021): events older than 30 days go, in bounded batches;
-- // hourly/daily rollups stay. Nightly, first run a few minutes after boot.
++ // 8b2. Raw analytics retention (ADR-021): events older than 30 days go, bot events older than 7, in bounded
++ // batches; hourly/daily rollups stay. Nightly, first run a few minutes after boot.
require('./utils/jobs').every('analytics-prune', 24 * 60 * 60 * 1000, async () => {
const out = await analyticsStore.prune();
const deleted = out.deleted || out.removed;
if (deleted) console.log(`[Analytics] pruned ${deleted} raw events older than ${out.cutoff}`);
++ if (out.bots && out.bots.removed) console.log(`[Analytics] pruned ${out.bots.removed} bot events older than ${out.bots.cutoff} (kept in the rollups)`);
}, { initialDelayMs: 5 * 60 * 1000, jitterMs: 60 * 1000 });
++ // 8b2b. The first stream's follow-up (server/streaming/first-stream-followup.js): one notification after someone's
++ // first stream ends, with the recording and their channel address. Every 5 minutes.
++ require('./utils/jobs').every('first-stream-followup', 5 * 60 * 1000, async () => {
++ const out = await require('./streaming/first-stream-followup').run();
++ if (out.sent) console.log(`[FirstStream] ${out.sent} first-stream follow-up(s) sent`);
++ }, { initialDelayMs: 4 * 60 * 1000, jitterMs: 30 * 1000 });
++
// 8b3. Media requests whose OpenCoins charge never got an answer: charged again with the same
// key (a replay if it landed) and refunded, so no viewer pays for a request that failed.
require('./utils/jobs').every('media-charge-reconcile', 5 * 60 * 1000, () => require('./media/media-queue').reconcileCharges(),
ddiff --git a/server/streaming/first-stream-followup.js b/server/streaming/first-stream-followup.js
nnew file mode 100644
iindex 00000000..9a4bee55
---- /dev/null
++++ b/server/streaming/first-stream-followup.js
@@@ -0,0 +1,69 @@
++'use strict';
++/**
++ * The first stream's follow-up (plan 2026-10-22, track C3). Of the 41 people whose first stream was in the last 30
++ * days (to 2026-10-22), 10 streamed again within 14 days; most first streams ran 1 to 10 minutes. After someone's
++ * first stream ends, they get one notification on the network: how long it ran, where the recording is (a public VOD
++ * on OpenVibe.Video once Media has it, else the after-show report, else their channel), and the channel address to
++ * share before going live again.
++ *
++ * run() every 5 minutes (server/index.js): first streams that ended 5 minutes to 2 hours ago, each notified once
++ * (app_state `first_stream_followup:<user id>`). A stream under 30 seconds is a test and gets nothing.
++ */
++const db = require('../db/database');
++
++const SETTLE_MS = 5 * 60 * 1000; // Media finalises the recording after the stream ends
++const LOOKBACK_MS = 2 * 60 * 60 * 1000;
++const MIN_SECONDS = 30;
++const RECAP_MIN_SECONDS = 8 * 60; // server/recap/recap.js MIN_DURATION_SEC
++const LIVE_URL = 'https://openvibe.live';
++const sqlTime = (ms) => new Date(ms).toISOString().replace('T', ' ').slice(0, 19);
++const stateKey = (userId) => `first_stream_followup:${userId}`;
++
++function minutes(seconds) {
++ const m = Math.round(seconds / 60);
++ return m < 1 ? 'Under a minute' : m === 1 ? '1 minute' : `${m} minutes`;
++}
++
++/** The notification for a first stream: { title, message, url }. */
++function messageFor({ username, seconds, vodUrl }) {
++ const channel = `${LIVE_URL}/@${encodeURIComponent(username)}`;
++ const where = vodUrl ? 'The recording is on OpenVibe.Video, ready to watch, clip and share.'
++ : seconds >= RECAP_MIN_SECONDS ? 'Your after-show report is ready, and the recording follows on your channel.'
++ : 'The recording appears on your channel once it is ready.';
++ return {
++ title: 'Your first stream is done 🎉',
++ message: `${minutes(seconds)} live. ${where} Next time, share ${channel.replace('https://', '')} before you go live so people can find you.`,
++ url: vodUrl || null,
++ };
++}
++
++async function run({ now = Date.now, push = null, vodFor = null } = {}) {
++ const send = push || require('../utils/notify').pushNotification;
++ const findVod = vodFor || (async (stream) => {
++ const [s] = await require('../media-proxy/lookups').attachPublicVods([{ id: stream.id }]);
++ return s && s.vod_id ? require('../web/video-url').videoUrl('vod', s.vod_id) : null;
++ });
++ const rows = await db.all(`SELECT s.id, s.user_id, s.duration_seconds, s.started_at, s.ended_at, u.username
++ FROM streams s JOIN users u ON u.id = s.user_id
++ WHERE COALESCE(s.is_live, 0) = 0 AND s.ended_at IS NOT NULL AND s.ended_at >= ? AND s.ended_at <= ?
++ AND NOT EXISTS (SELECT 1 FROM streams o WHERE o.user_id = s.user_id AND o.id < s.id)
++ ORDER BY s.id LIMIT 50`, [sqlTime(now() - LOOKBACK_MS), sqlTime(now() - SETTLE_MS)]);
++ let sent = 0;
++ for (const s of rows) {
++ if (await db.getState(stateKey(s.user_id))) continue;
++ const seconds = Number(s.duration_seconds) > 0 ? Number(s.duration_seconds)
++ : Math.max(0, Math.round((Date.parse(`${String(s.ended_at).replace(' ', 'T')}Z`) - Date.parse(`${String(s.started_at).replace(' ', 'T')}Z`)) / 1000)) || 0;
++ // Recorded first, so a crash after this never sends twice (a missed one is better than a repeated one).
++ await db.setState(stateKey(s.user_id), String(s.id));
++ if (seconds < MIN_SECONDS) continue;
++ let vodUrl = null;
++ try { vodUrl = await findVod(s); } catch { /* Media unreachable: the channel link instead */ }
++ const m = messageFor({ username: s.username, seconds, vodUrl });
++ const url = m.url || (seconds >= RECAP_MIN_SECONDS ? `${LIVE_URL}/recap/${s.id}` : `${LIVE_URL}/@${encodeURIComponent(s.username)}`);
++ send({ user_id: s.user_id, type: 'STREAM_SUMMARY', category: 'stream', title: m.title, message: m.message, url });
++ sent++;
++ }
++ return { checked: rows.length, sent };
++}
++
++module.exports = { run, messageFor, stateKey, MIN_SECONDS };
ddiff --git a/test/analytics-store.test.js b/test/analytics-store.test.js
iindex 9b53ee59..e8577981 100644
---- a/test/analytics-store.test.js
++++ b/test/analytics-store.test.js
@@@ -12,6 +12,16 @@ async function main() {
assert.ok(store.tracker, 'analytics tracker opens on the migrated PostgreSQL database');
assert.ok(await db.getDb().maybe("SELECT 1 FROM information_schema.tables WHERE table_name = 'analytics_events'"));
assert.ok('removed' in await store.prune());
++
++ // Bot rows go after 7 days (they are in the rollups by then); people's rows stay the 30 days ADR-021 allows.
++ const d = db.getDb();
++ const at = (daysAgo) => new Date(Date.now() - daysAgo * 86400000).toISOString().replace('T', ' ').slice(0, 19);
++ const ins = (isBot, daysAgo, path) => d.exec("INSERT INTO analytics_events (service, event_type, path, is_bot, created_at) VALUES ('live', 'api_call', $1, $2, $3)", [path, isBot, at(daysAgo)]);
++ await ins(1, 10, '/t/bot-old'); await ins(1, 2, '/t/bot-new'); await ins(0, 10, '/t/person-old'); await ins(0, 40, '/t/person-older');
++ const out = await store.prune();
++ assert.ok(out.bots && out.bots.removed >= 1, JSON.stringify(out));
++ const left = (await d.many("SELECT path FROM analytics_events WHERE path LIKE '/t/%' ORDER BY path")).map((r) => r.path);
++ assert.deepStrictEqual(left, ['/t/bot-new', '/t/person-old'], 'a 10-day-old bot row and a 40-day-old row go; a 2-day-old bot row and a 10-day-old person\'s row stay');
} finally {
await store.close();
}
ddiff --git a/test/first-stream-followup.test.js b/test/first-stream-followup.test.js
nnew file mode 100644
iindex 00000000..f1fa092c
---- /dev/null
++++ b/test/first-stream-followup.test.js
@@@ -0,0 +1,62 @@
++'use strict';
++// The first stream's follow-up (server/streaming/first-stream-followup.js): one notification after someone's first
++// stream ends (settled for 5 minutes), never twice, never for a second stream or a test under 30 seconds; it links the
++// public recording when Media has it, else the after-show report (8 minutes and up), else the channel.
++const assert = require('assert');
++const db = require('../server/db/database');
++const followup = require('../server/streaming/first-stream-followup');
++
++const sqlTime = (ms) => new Date(ms).toISOString().replace('T', ' ').slice(0, 19);
++
++(async () => {
++ await db.initDb();
++ const NOW = Date.now();
++ const mkUser = async (name) => (await db.createUser({ username: name, email: `${name}@x`, password_hash: 'x', display_name: name, stream_key: name.padEnd(32, '0') })).lastInsertRowid
++ || (await db.get('SELECT id FROM users WHERE username = ?', [name])).id;
++ const mkStream = async (userId, { endedAgoMin, seconds, live = false }) => {
++ const ended = NOW - endedAgoMin * 60000;
++ const r = await db.run(`INSERT INTO streams (user_id, title, protocol, is_live, started_at, ended_at, duration_seconds) VALUES (?, 't', 'webrtc', ?, ?, ?, ?) RETURNING id`,
++ [userId, live ? 1 : 0, sqlTime(ended - seconds * 1000), live ? null : sqlTime(ended), seconds]);
++ return Number(r.lastInsertRowid);
++ };
++
++ const ana = await mkUser('fs_ana'); // first stream, 6 minutes, recording on Video
++ const bo = await mkUser('fs_bo'); // first stream, 20 minutes, no public recording yet: the report
++ const cy = await mkUser('fs_cy'); // a 12-second test
++ const di = await mkUser('fs_di'); // ended 2 minutes ago: not settled yet
++ const ed = await mkUser('fs_ed'); // a second stream: their first was days ago
++ const anaStream = await mkStream(ana, { endedAgoMin: 10, seconds: 360 });
++ const boStream = await mkStream(bo, { endedAgoMin: 30, seconds: 1200 });
++ await mkStream(cy, { endedAgoMin: 15, seconds: 12 });
++ await mkStream(di, { endedAgoMin: 2, seconds: 600 });
++ await mkStream(ed, { endedAgoMin: 5000, seconds: 900 });
++ await mkStream(ed, { endedAgoMin: 20, seconds: 900 });
++
++ const sent = [];
++ const vodFor = async (s) => (s.id === anaStream ? 'https://openvibe.video/v/77' : null);
++ const out = await followup.run({ now: () => NOW, push: (n) => sent.push(n), vodFor });
++ assert.strictEqual(out.sent, 2, JSON.stringify(sent));
++ const byUser = Object.fromEntries(sent.map((n) => [n.user_id, n]));
++ assert.deepStrictEqual(Object.keys(byUser).map(Number).sort(), [ana, bo].map(Number).sort());
++ assert.strictEqual(byUser[ana].url, 'https://openvibe.video/v/77');
++ assert.match(byUser[ana].message, /^6 minutes live\. The recording is on OpenVibe\.Video/);
++ assert.match(byUser[ana].message, /share openvibe\.live\/@fs_ana before you go live/);
++ assert.strictEqual(byUser[bo].url, `https://openvibe.live/recap/${boStream}`, 'no public recording yet: the after-show report');
++ assert.strictEqual(byUser[ana].type, 'STREAM_SUMMARY');
++ assert.ok(!/viewer/i.test(sent.map((n) => n.message).join(' ')), 'no viewer counts (peak_viewers is not a person count)');
++
++ // Never twice; the test stream was marked too, so it never comes back either.
++ assert.strictEqual((await followup.run({ now: () => NOW, push: (n) => sent.push(n), vodFor })).sent, 0);
++ assert.ok(await db.getState(followup.stateKey(cy)), 'a test stream is marked, so a later run does not reconsider it');
++
++ // Once settled, the stream that had just ended is notified too.
++ const later = await followup.run({ now: () => NOW + 4 * 60000, push: (n) => sent.push(n), vodFor });
++ assert.strictEqual(later.sent, 1);
++ assert.strictEqual(sent[sent.length - 1].user_id, di);
++
++ // Short first stream with no recording: the channel.
++ const m = followup.messageFor({ username: 'zed', seconds: 90, vodUrl: null });
++ assert.match(m.message, /^2 minutes live\. The recording appears on your channel/);
++ console.log('first-stream follow-up: once, settled, recording/report/channel links, no test streams');
++ process.exit(0);
++})().catch((e) => { console.error(e); process.exit(1); });