openvibers/openvibe.live Commits
Backfill script for the moderation audit log: Live's actions logged before live.moderation.action go to Network with their original time (same rule as logModerationAction), once each (moderation_events_backfill), only before a required --before cutoff
Changes
nnew file mode 100644@@@ -0,0 +1,52 @@++#!/usr/bin/env node++'use strict';++/**++ * One-time backfill (WS-D task 1): Live's moderation actions logged before live.moderation.action++ * existed go to OpenVibe.Network's moderation audit log, each with its original time. Same rule as++ * logModerationAction (server/db/database.js): tidying one's own messages, configuring one's own channel++ * and acting on oneself are not sent. Idempotent: moderation_events_backfill remembers every row sent,++ * and --before (required) keeps it off rows the live code already announced.++ *++ * Run on the host AS LIVE, with Live's environment (EVENTS_URL, the OAuth client secret), e.g.:++ * node scripts/backfill-moderation-events.js --before '2026-09-25 20:31:27' [--dry-run]++ * It queues into Live's outbox and relays until the queue is empty. Nothing secret is printed.++ */++const args = process.argv.slice(2);++const flag = (n) => { const i = args.indexOf(n); return i >= 0 ? args[i + 1] : null; };++const before = flag('--before');++const dryRun = args.includes('--dry-run');++if (!before || !/^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}$/.test(before)) {++ console.error("--before 'YYYY-MM-DD HH:MM:SS' (UTC, the deploy time of logModerationAction's event) is required");++ process.exit(2);++}++++(async () => {++ const db = require('../server/db/database');++ db.initDb();++ const d = db.getDb();++ d.exec('CREATE TABLE IF NOT EXISTS moderation_events_backfill (action_id INTEGER PRIMARY KEY, sent_at DATETIME DEFAULT CURRENT_TIMESTAMP)');++ const rows = d.prepare(`SELECT m.* FROM moderation_actions m LEFT JOIN moderation_events_backfill b ON b.action_id = m.id++ WHERE b.action_id IS NULL AND m.created_at < ? ORDER BY m.id`).all(before);++ console.log(`${rows.length} action(s) logged before ${before} not yet backfilled`);++ if (dryRun) { process.exit(0); }++ const streamEvents = require('../server/events/stream-events');++ const outbox = streamEvents.init({ intervalMs: 500 });++ if (!outbox) { console.error('the outbox is off (EVENTS_URL or the client secret is missing)'); process.exit(1); }++ let sent = 0, skipped = 0;++ for (const r of rows) {++ let details = {};++ try { details = JSON.parse(r.details || '{}'); } catch { /* keep {} */ }++ const at = new Date(`${String(r.created_at).replace(' ', 'T')}Z`);++ d.transaction(() => {++ const ok = db.announceModerationAction(r.id, { ...r, details }, { at: Number.isNaN(at.getTime()) ? null : at.toISOString() });++ d.prepare('INSERT INTO moderation_events_backfill (action_id) VALUES (?)').run(r.id);++ if (ok) sent++; else skipped++;++ })();++ }++ console.log(`queued ${sent}, not moderation ${skipped}; relaying…`);++ for (let i = 0; i < 60 && streamEvents.status().pending > 0; i++) { await outbox.flush(); await new Promise((res) => setTimeout(res, 500)); }++ const st = streamEvents.status();++ console.log(`pending ${st.pending}, rejected ${st.rejected}${st.last_error ? ` (last error: ${st.last_error})` : ''}`);++ outbox.stop && outbox.stop();++ process.exit(st.rejected ? 1 : 0);++})().catch((err) => { console.error(err.message); process.exit(1); });@@@ -6722,7 +6722,7 @@ function logModerationAction({ scope_type, scope_id, actor_user_id, target_user_return info;}--function _announceModerationAction(id, a) {++function _announceModerationAction(id, a, { at = null } = {}) {const type = String(a.action_type || '');if (!type || type.startsWith('self_') || type === 'channel_settings_update') return false;if (a.actor_user_id != null && a.actor_user_id === a.target_user_id) return false;@@@ -6738,6 +6738,7 @@ function _announceModerationAction(id, a) {const scopeId = a.scope_id == null || a.scope_id === '' ? null : (Number.isSafeInteger(Number(a.scope_id)) ? Number(a.scope_id) : String(a.scope_id).slice(0, 128));const details = a.details && typeof a.details === 'object' && !Array.isArray(a.details) ? a.details : {};events.enqueue({++ ...(at ? { timestamp: at } : {}),event_type: 'live.moderation.action',actor: actorSubject ? { type: 'user', id: actorSubject } : { type: 'service', id: 'live' },subject: { type: 'moderation_action', id: String(id) },@@@ -7636,7 +7637,7 @@ module.exports = {getHomeStatSeries, HOME_SERIES_KEYS, vibesStatsSince, _computeHomeStats,getVodAiState, getClipAiState,scheduleClipNotifyState, bumpClipNotifyNowState, markClipNotifiedState, getDueClipNotifies,-- getDb, initDb, run, get, all, close,++ getDb, initDb, run, get, all, close, announceModerationAction: _announceModerationAction,mergeChatMessageMetadata, getTimelineSpeechSince,getDonationGoalsForWidget, getAllDonationGoals, getActiveDonationGoals, getDonationGoalById,recordViewerSample, getViewerTrend, getReadingSeries, getHomePulse, getActiveGoalsForUsers,@@@ -72,6 +72,11 @@ const stub = http.createServer((req, res) => {assert.strictEqual(ban.priority, 'important'); assert.strictEqual(ban.visibility, 'internal');assert.strictEqual(end.payload.scope_id, 77);++ // The backfill keeps each action's original time.++ d.transaction(() => db.announceModerationAction(1, { scope_type: 'site', actor_user_id: 1, target_user_id: 2, action_type: 'site_ban', details: {} }, { at: '2026-04-13T20:56:38.000Z' }))();++ await outbox.flush();++ assert.strictEqual(published.at(-1).timestamp, '2026-04-13T20:56:38.000Z');++// A failed insert queues nothing (the event is in its transaction).const before = published.length;assert.throws(() => db.logModerationAction({ scope_type: 'site', actor_user_id: 1, target_user_id: 2, action_type: null }));
new file mode 100644
new file mode 100644
@@ -0,0 +1,52 @@
@@ -0,0 +1,52 @@
+#!/usr/bin/env node
+'use strict';
+/**
+ * One-time backfill (WS-D task 1): Live's moderation actions logged before live.moderation.action
+ * existed go to OpenVibe.Network's moderation audit log, each with its original time. Same rule as
+ * logModerationAction (server/db/database.js): tidying one's own messages, configuring one's own channel
+ * and acting on oneself are not sent. Idempotent: moderation_events_backfill remembers every row sent,
+ * and --before (required) keeps it off rows the live code already announced.
+ *
+ * Run on the host AS LIVE, with Live's environment (EVENTS_URL, the OAuth client secret), e.g.:
+ * node scripts/backfill-moderation-events.js --before '2026-09-25 20:31:27' [--dry-run]
+ * It queues into Live's outbox and relays until the queue is empty. Nothing secret is printed.
+ */
+const args = process.argv.slice(2);
+const flag = (n) => { const i = args.indexOf(n); return i >= 0 ? args[i + 1] : null; };
+const before = flag('--before');
+const dryRun = args.includes('--dry-run');
+if (!before || !/^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}$/.test(before)) {
+ console.error("--before 'YYYY-MM-DD HH:MM:SS' (UTC, the deploy time of logModerationAction's event) is required");
+ process.exit(2);
+}
+
+(async () => {
+ const db = require('../server/db/database');
+ db.initDb();
+ const d = db.getDb();
+ d.exec('CREATE TABLE IF NOT EXISTS moderation_events_backfill (action_id INTEGER PRIMARY KEY, sent_at DATETIME DEFAULT CURRENT_TIMESTAMP)');
+ const rows = d.prepare(`SELECT m.* FROM moderation_actions m LEFT JOIN moderation_events_backfill b ON b.action_id = m.id
+ WHERE b.action_id IS NULL AND m.created_at < ? ORDER BY m.id`).all(before);
+ console.log(`${rows.length} action(s) logged before ${before} not yet backfilled`);
+ if (dryRun) { process.exit(0); }
+ const streamEvents = require('../server/events/stream-events');
+ const outbox = streamEvents.init({ intervalMs: 500 });
+ if (!outbox) { console.error('the outbox is off (EVENTS_URL or the client secret is missing)'); process.exit(1); }
+ let sent = 0, skipped = 0;
+ for (const r of rows) {
+ let details = {};
+ try { details = JSON.parse(r.details || '{}'); } catch { /* keep {} */ }
+ const at = new Date(`${String(r.created_at).replace(' ', 'T')}Z`);
+ d.transaction(() => {
+ const ok = db.announceModerationAction(r.id, { ...r, details }, { at: Number.isNaN(at.getTime()) ? null : at.toISOString() });
+ d.prepare('INSERT INTO moderation_events_backfill (action_id) VALUES (?)').run(r.id);
+ if (ok) sent++; else skipped++;
+ })();
+ }
+ console.log(`queued ${sent}, not moderation ${skipped}; relaying…`);
+ for (let i = 0; i < 60 && streamEvents.status().pending > 0; i++) { await outbox.flush(); await new Promise((res) => setTimeout(res, 500)); }
+ const st = streamEvents.status();
+ console.log(`pending ${st.pending}, rejected ${st.rejected}${st.last_error ? ` (last error: ${st.last_error})` : ''}`);
+ outbox.stop && outbox.stop();
+ process.exit(st.rejected ? 1 : 0);
+})().catch((err) => { console.error(err.message); process.exit(1); });
@@ -6722,7 +6722,7 @@ function logModerationAction({ scope_type, scope_id, actor_user_id, target_user_
@@ -6722,7 +6722,7 @@ function logModerationAction({ scope_type, scope_id, actor_user_id, target_user_
return info;
return info;
}
}
-function _announceModerationAction(id, a) {
+function _announceModerationAction(id, a, { at = null } = {}) {
const type = String(a.action_type || '');
const type = String(a.action_type || '');
if (!type || type.startsWith('self_') || type === 'channel_settings_update') return false;
if (!type || type.startsWith('self_') || type === 'channel_settings_update') return false;
if (a.actor_user_id != null && a.actor_user_id === a.target_user_id) return false;
if (a.actor_user_id != null && a.actor_user_id === a.target_user_id) return false;
@@ -6738,6 +6738,7 @@ function _announceModerationAction(id, a) {
@@ -6738,6 +6738,7 @@ function _announceModerationAction(id, a) {
const scopeId = a.scope_id == null || a.scope_id === '' ? null : (Number.isSafeInteger(Number(a.scope_id)) ? Number(a.scope_id) : String(a.scope_id).slice(0, 128));
const scopeId = a.scope_id == null || a.scope_id === '' ? null : (Number.isSafeInteger(Number(a.scope_id)) ? Number(a.scope_id) : String(a.scope_id).slice(0, 128));
const details = a.details && typeof a.details === 'object' && !Array.isArray(a.details) ? a.details : {};
const details = a.details && typeof a.details === 'object' && !Array.isArray(a.details) ? a.details : {};
events.enqueue({
events.enqueue({
+ ...(at ? { timestamp: at } : {}),
event_type: 'live.moderation.action',
event_type: 'live.moderation.action',
actor: actorSubject ? { type: 'user', id: actorSubject } : { type: 'service', id: 'live' },
actor: actorSubject ? { type: 'user', id: actorSubject } : { type: 'service', id: 'live' },
subject: { type: 'moderation_action', id: String(id) },
subject: { type: 'moderation_action', id: String(id) },
@@ -7636,7 +7637,7 @@ module.exports = {
@@ -7636,7 +7637,7 @@ module.exports = {
getHomeStatSeries, HOME_SERIES_KEYS, vibesStatsSince, _computeHomeStats,
getHomeStatSeries, HOME_SERIES_KEYS, vibesStatsSince, _computeHomeStats,
getVodAiState, getClipAiState,
getVodAiState, getClipAiState,
scheduleClipNotifyState, bumpClipNotifyNowState, markClipNotifiedState, getDueClipNotifies,
scheduleClipNotifyState, bumpClipNotifyNowState, markClipNotifiedState, getDueClipNotifies,
- getDb, initDb, run, get, all, close,
+ getDb, initDb, run, get, all, close, announceModerationAction: _announceModerationAction,
mergeChatMessageMetadata, getTimelineSpeechSince,
mergeChatMessageMetadata, getTimelineSpeechSince,
getDonationGoalsForWidget, getAllDonationGoals, getActiveDonationGoals, getDonationGoalById,
getDonationGoalsForWidget, getAllDonationGoals, getActiveDonationGoals, getDonationGoalById,
recordViewerSample, getViewerTrend, getReadingSeries, getHomePulse, getActiveGoalsForUsers,
recordViewerSample, getViewerTrend, getReadingSeries, getHomePulse, getActiveGoalsForUsers,
@@ -72,6 +72,11 @@ const stub = http.createServer((req, res) => {
@@ -72,6 +72,11 @@ const stub = http.createServer((req, res) => {
assert.strictEqual(ban.priority, 'important'); assert.strictEqual(ban.visibility, 'internal');
assert.strictEqual(ban.priority, 'important'); assert.strictEqual(ban.visibility, 'internal');
assert.strictEqual(end.payload.scope_id, 77);
assert.strictEqual(end.payload.scope_id, 77);
+ // The backfill keeps each action's original time.
+ d.transaction(() => db.announceModerationAction(1, { scope_type: 'site', actor_user_id: 1, target_user_id: 2, action_type: 'site_ban', details: {} }, { at: '2026-04-13T20:56:38.000Z' }))();
+ await outbox.flush();
+ assert.strictEqual(published.at(-1).timestamp, '2026-04-13T20:56:38.000Z');
+
// A failed insert queues nothing (the event is in its transaction).
// A failed insert queues nothing (the event is in its transaction).
const before = published.length;
const before = published.length;
assert.throws(() => db.logModerationAction({ scope_type: 'site', actor_user_id: 1, target_user_id: 2, action_type: null }));
assert.throws(() => db.logModerationAction({ scope_type: 'site', actor_user_id: 1, target_user_id: 2, action_type: null }));