openvibers/openvibe.live Commits

The AI's used moments are durable rows, not a capped list (item 151, C06) (#578)

OP OpenVibers committed 10 days ago · a14a4af027 · 5 files, +148 −36

The paste job and both clip cutters share one record of used moments. It was a JSON list in app_state capped at 800,
so on a busy week the oldest fell out and a repeat came back. Migration 0024 makes each moment a row of ai_moments:
the same-recording check is an indexed lookup, the scene, speech and footage checks read the last 14 days, rows go
after 90 days, and the old list moves in once. The API is unchanged.

Changes

ddiff --git a/migrations/0024_ai_moments.sql b/migrations/0024_ai_moments.sql
nnew file mode 100644
iindex 00000000..97d15254
---- /dev/null
++++ b/migrations/0024_ai_moments.sql
@@@ -0,0 +1,23 @@
++-- phase: expand
++-- The AI's used moments, durable (item 151, C06 of the clip-quality research). The paste job and both clip cutters
++-- share one record of every moment they turned into something, so none picks the same spot or the same scene twice
++-- (server/ai/moment-registry.js). It was a JSON list in app_state, capped at 800 entries: on a busy week the oldest
++-- fell out early and a repeat came back. Here every moment is a row: the same-recording check (no two within two
++-- minutes) is an indexed lookup, the scene and speech checks read the last 14 days, and rows go after 90 days.
++-- The entries in app_state 'ai_used_moments' move in once, on first use.
++CREATE TABLE IF NOT EXISTS ai_moments (
++ id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
++ kind text NOT NULL, -- 'clip' | 'paste' | 'moment'
++ vod_id text, -- the Media recording, as a string id
++ stream_id text,
++ at_sec integer, -- the moment's second in that recording or stream
++ sig text, -- the short scene signature
++ scene text, -- the normalized description (600 chars)
++ speech text, -- what was said around it, normalized (600 chars)
++ embedding jsonb, -- an optional frame embedding (at most 1024 numbers)
++ title text,
++ created_at timestamptz NOT NULL DEFAULT now()
++);
++CREATE INDEX IF NOT EXISTS ai_moments_vod ON ai_moments (vod_id, at_sec) WHERE vod_id IS NOT NULL;
++CREATE INDEX IF NOT EXISTS ai_moments_stream ON ai_moments (stream_id, at_sec) WHERE stream_id IS NOT NULL;
++CREATE INDEX IF NOT EXISTS ai_moments_created ON ai_moments (created_at DESC);
ddiff --git a/server/ai/moment-registry.js b/server/ai/moment-registry.js
iindex 6fd6c3f4..66087110 100644
---- a/server/ai/moment-registry.js
++++ b/server/ai/moment-registry.js
@@@ -9,7 +9,10 @@
* isUsed({ vod_id, stream_id, offset, sig }) same VOD within GAP_SEC, or same scene
* usedOffsets(vod_id, stream_id) → [sec…] to tell the picker to avoid
*
-- * State key `ai_used_moments` (capped). Legacy logs are imported once on first use.
++ * Durable since 11-03 (item 151, C06; migration 0024): every moment is a row of `ai_moments`. The same-recording check
++ * is an indexed lookup, the scene/speech/footage checks read the last SIG_TTL (14 days), and rows go after OFFSET_TTL
++ * (90 days). It was a JSON list in app_state 'ai_used_moments' capped at 800, so on a busy week the oldest fell out and
++ * a repeat came back; that list (or, before it, the two legacy logs) moves in once, on first use.
*/
'use strict';
@@@ -17,27 +20,34 @@ const db = require('../db/database');
const dedupe = require('./clip-dedupe');
const KEY = 'ai_used_moments';
--const MAX = 800;
++const MOVED = 'ai_moments_moved';
const GAP_SEC = 120; // two moments on the same VOD must be at least this far apart
const SIG_TTL_MS = 14 * 24 * 3600_000; // a scene signature blocks re-use for two weeks
const OFFSET_TTL_MS = 90 * 24 * 3600_000;
++const SCAN = 5000; // the most recent rows the scene check reads (two weeks is far fewer today)
--let _cache = null;
--async function _load() {
-- if (_cache) return _cache;
-- let list = [];
-- try { const l = JSON.parse(await db.getState(KEY) || '[]'); if (Array.isArray(l)) list = l; } catch { /* */ }
-- if (!list.length) list = await _importLegacy();
-- _cache = list;
-- return list;
++const idOf = (v) => (v == null || v === '' ? null : String(v));
++const iso = (ms) => new Date(ms).toISOString();
++
++let _ready = null;
++/** The table, with the old list moved in once. */
++function _load() {
++ if (!_ready) _ready = (async () => {
++ if (await db.getState(MOVED)) return;
++ let list = [];
++ try { const l = JSON.parse(await db.getState(KEY) || '[]'); if (Array.isArray(l)) list = l; } catch { /* */ }
++ if (!list.length) list = await _importLegacy();
++ for (const e of list.slice().reverse()) await _insert(e, e.ts || Date.now());
++ await db.setState(MOVED, new Date().toISOString());
++ })().catch((e) => { _ready = null; throw e; });
++ return _ready;
}
--async function _save(list) { _cache = list.slice(0, MAX); try { await db.setState(KEY, JSON.stringify(_cache)); } catch { /* */ } }
async function _importLegacy() {
const out = [];
try { for (const c of JSON.parse(await db.getState('auto_clip_log') || '[]') || []) out.push({ kind: 'clip', vod_id: c.vod_id || null, stream_id: c.stream_id || null, offset: (c.start_time || 0) + 16, sig: c.sig || null, title: c.title || null, ts: c.ts || Date.now() }); } catch { /* */ }
try { const h = JSON.parse(await db.getState('home_hero_moments') || '{}') || {}; for (const m of h.moments || []) out.push({ kind: 'paste', vod_id: m.vodId || null, stream_id: null, offset: m.offset || 0, sig: sig(m.title), title: m.title || null, ts: h.updated_at || Date.now() }); for (const s of h.usedSigs || []) out.push({ kind: 'paste', vod_id: null, stream_id: null, offset: null, sig: s, title: null, ts: h.updated_at || Date.now() }); } catch { /* */ }
-- return out.slice(0, MAX);
++ return out;
}
/** Coarse scene signature — the first five significant tokens of a description/title. */
@@@ -52,41 +62,62 @@ function _embedding(v) {
return out.every(Number.isFinite) ? out : null;
}
++async function _insert(e, ts = Date.now()) {
++ await db.run(`INSERT INTO ai_moments (kind, vod_id, stream_id, at_sec, sig, scene, speech, embedding, title, created_at)
++ VALUES (?, ?, ?, ?, ?, ?, ?, ?::jsonb, ?, ?::timestamptz)`, [
++ String(e.kind || 'moment').slice(0, 20), idOf(e.vod_id), idOf(e.stream_id), e.offset == null ? null : Math.floor(Number(e.offset)) || 0,
++ e.sig ? String(e.sig).slice(0, 200) : null, e.scene ? String(e.scene).slice(0, 600) : null, e.speech ? String(e.speech).slice(0, 600) : null,
++ e.embedding ? JSON.stringify(e.embedding) : null, e.title ? String(e.title).slice(0, 120) : null, iso(ts),
++ ]);
++}
++
++let _pruned = 0;
async function record({ kind, vod_id = null, stream_id = null, offset = null, sig: s = null, title = null, desc = null, speech = null, embedding = null }) {
-- const list = await _load();
++ await _load();
const scene = dedupe.normalize(desc || title);
-- list.unshift({
-- kind: kind || 'moment', vod_id: vod_id || null, stream_id: stream_id || null,
-- offset: offset == null ? null : Math.floor(offset),
++ await _insert({
++ kind: kind || 'moment', vod_id, stream_id, offset,
sig: s || sig(desc || title),
// Full normalized scene + what was said, so a re-worded description still matches and a
-- // static webcam with a NEW conversation does not. Legacy entries keep only `sig`.
-- scene: scene ? scene.slice(0, 600) : null,
-- speech: speech ? dedupe.normalize(speech).slice(0, 600) : null,
++ // static webcam with a NEW conversation does not.
++ scene, speech: speech ? dedupe.normalize(speech) : null,
// Optional frame embedding: the same footage in another upload scores near 1.
embedding: _embedding(embedding),
-- title: title ? String(title).slice(0, 120) : null, ts: Date.now(),
++ title,
});
-- await _save(list);
++ // Rows past 90 days go, at most once an hour.
++ if (Date.now() - _pruned > 3600_000) { _pruned = Date.now(); await db.run('DELETE FROM ai_moments WHERE created_at < ?::timestamptz', [iso(Date.now() - OFFSET_TTL_MS)]).catch(() => {}); }
}
/** Why a moment is blocked, or null when it is free. */
async function usedReason({ vod_id = null, stream_id = null, offset = null, sig: s = null, desc = null, title = null, speech = null, embedding = null } = {}, { gapSec = GAP_SEC } = {}) {
++ await _load();
const now = Date.now();
++ const v = idOf(vod_id);
++ const st = idOf(stream_id);
++ if (offset != null && (v || st)) {
++ const at = Math.floor(Number(offset)) || 0;
++ const hit = await db.get(`SELECT kind, at_sec, vod_id FROM ai_moments
++ WHERE created_at >= ?::timestamptz AND at_sec IS NOT NULL AND at_sec > ? AND at_sec < ?
++ AND ((?::text IS NOT NULL AND vod_id = ?::text) OR (?::text IS NOT NULL AND stream_id = ?::text))
++ ORDER BY created_at DESC LIMIT 1`,
++ [iso(now - OFFSET_TTL_MS), at - gapSec, at + gapSec, v, v, st, st]);
++ if (hit) return `${hit.kind} already made at ${hit.at_sec}s of this ${v && hit.vod_id === v ? 'VOD' : 'stream'}`;
++ }
const scene = s || sig(desc || title);
const full = dedupe.normalize(desc || title);
const said = dedupe.normalize(speech);
const emb = _embedding(embedding);
-- for (const e of await _load()) {
-- const age = now - (e.ts || 0);
-- if (age > OFFSET_TTL_MS) continue;
-- if (offset != null && e.offset != null && ((vod_id && e.vod_id === vod_id) || (stream_id && e.stream_id === stream_id)) && Math.abs(e.offset - offset) < gapSec) return `${e.kind} already made at ${Math.floor(e.offset)}s of this ${e.vod_id === vod_id ? 'VOD' : 'stream'}`;
-- if (age > SIG_TTL_MS) continue;
++ if (!scene && !full && !emb) return null;
++ const rows = await db.all(`SELECT kind, sig, scene, speech, embedding FROM ai_moments WHERE created_at >= ?::timestamptz
++ ORDER BY created_at DESC LIMIT ${SCAN}`, [iso(now - SIG_TTL_MS)]);
++ for (const e of rows) {
++ const eEmb = Array.isArray(e.embedding) ? e.embedding : null;
const shortMatch = scene && e.sig && e.sig === scene;
const sceneMatch = full && e.scene && dedupe.isSameScene(e.scene, full);
// Frame embeddings (when both sides have them) are independent evidence: the same footage
// is a repeat even if it was re-described or re-uploaded.
-- const visualMatch = emb && e.embedding && dedupe.isSameEmbedding(e.embedding, emb);
++ const visualMatch = emb && eEmb && dedupe.isSameEmbedding(eEmb, emb);
if (!shortMatch && !sceneMatch && !visualMatch) continue;
if (visualMatch) return `same footage as a recent ${e.kind}`;
// Same camera/description but a different conversation is a new moment, not a repeat.
@@@ -99,14 +130,34 @@ async function isUsed(m, o) { return !!await usedReason(m, o); }
/** Offsets already used on this VOD/stream (for the picker's "avoid these" list). */
async function usedOffsets(vod_id = null, stream_id = null) {
-- const now = Date.now();
-- return [...new Set((await _load()).filter(e => e.offset != null && now - (e.ts || 0) <= OFFSET_TTL_MS && ((vod_id && e.vod_id === vod_id) || (stream_id && e.stream_id === stream_id))).map(e => Math.floor(e.offset)))].sort((a, b) => a - b);
++ await _load();
++ const v = idOf(vod_id);
++ const st = idOf(stream_id);
++ if (!v && !st) return [];
++ const rows = await db.all(`SELECT DISTINCT at_sec FROM ai_moments WHERE created_at >= ?::timestamptz AND at_sec IS NOT NULL
++ AND ((?::text IS NOT NULL AND vod_id = ?::text) OR (?::text IS NOT NULL AND stream_id = ?::text)) ORDER BY at_sec`,
++ [iso(Date.now() - OFFSET_TTL_MS), v, v, st, st]);
++ return rows.map((r) => Number(r.at_sec));
}
++const _shape = (r) => r && ({ kind: r.kind, vod_id: r.vod_id, stream_id: r.stream_id, offset: r.at_sec == null ? null : Number(r.at_sec), sig: r.sig, title: r.title, ts: new Date(r.created_at).getTime() });
++
async function lastOfKind(kind, { stream_id = null, vod_id = null } = {}) {
-- return (await _load()).find(e => e.kind === kind && (!stream_id || e.stream_id === stream_id) && (!vod_id || e.vod_id === vod_id)) || null;
++ await _load();
++ const st = idOf(stream_id);
++ const v = idOf(vod_id);
++ return _shape(await db.get(`SELECT * FROM ai_moments WHERE kind = ? AND (?::text IS NULL OR stream_id = ?::text) AND (?::text IS NULL OR vod_id = ?::text)
++ ORDER BY created_at DESC, id DESC LIMIT 1`, [kind, st, st, v, v])) || null;
}
--async function recent(hours = 48) { const cut = Date.now() - hours * 3600_000; return (await _load()).filter(e => (e.ts || 0) >= cut); }
++async function recent(hours = 48) {
++ await _load();
++ return (await db.all('SELECT * FROM ai_moments WHERE created_at >= ?::timestamptz ORDER BY created_at DESC, id DESC LIMIT 1000', [iso(Date.now() - hours * 3600_000)])).map(_shape);
++}
--module.exports = { record, isUsed, usedReason, usedOffsets, lastOfKind, recent, sig, GAP_SEC, _reset: () => { _cache = null; } };
++module.exports = {
++ record, isUsed, usedReason, usedOffsets, lastOfKind, recent, sig, GAP_SEC,
++ _reset: () => { _ready = null; },
++ /** Tests: an empty registry (the move counts as done). */
++ _clear: async () => { await db.run('DELETE FROM ai_moments'); await db.setState(MOVED, new Date().toISOString()); _ready = null; },
++};
ddiff --git a/test/clip-dedupe.test.js b/test/clip-dedupe.test.js
iindex e9d145e4..70cc6546 100644
---- a/test/clip-dedupe.test.js
++++ b/test/clip-dedupe.test.js
@@@ -34,7 +34,7 @@ assert.ok(!dedupe.isSameEmbedding([1, 0, 0], [0, 1, 0]), 'different frame embedd
console.log('✅ clip-dedupe: Unicode, paraphrases, 0.8 Dice, boilerplate and embedding negatives');
// ── The shared registry ────────────────────────────────────
--await db.setState('ai_used_moments', '[]');
++await require('../server/ai/moment-registry')._clear();
registry._reset();
await registry.record({
kind: 'paste', vod_id: 30, stream_id: 5, offset: 100,
@@@ -86,7 +86,7 @@ await autoClip._internals.checkLiveStream(liveStream);
assert.strictEqual(created.length, 0, 'the live path never cuts a moment the registry already used');
registry._reset();
--await db.setState('ai_used_moments', '[]');
++await require('../server/ai/moment-registry')._clear();
registry._reset();
await autoClip._internals.checkLiveStream(liveStream);
assert.strictEqual(created.length, 1, 'a free moment is still cut (the check does not block everything)');
ddiff --git a/test/clip-freshness.test.js b/test/clip-freshness.test.js
iindex 9ccaa481..078ebd28 100644
---- a/test/clip-freshness.test.js
++++ b/test/clip-freshness.test.js
@@@ -142,7 +142,7 @@ async function reset() {
await raw.prepare('DELETE FROM analytics_events').run();
await raw.prepare('DELETE FROM streams').run();
await db.setState('auto_clip_log', '[]');
-- await db.setState('ai_used_moments', '[]');
++ await require('../server/ai/moment-registry')._clear();
require('../server/ai/moment-registry')._reset();
require('../server/ai/clip-policy')._resetArchiveReservation();
currentVods = [];
@@@ -265,7 +265,7 @@ assert.strictEqual(policy.archiveAllowed(clipLog, Date.now()), false, "today's a
// A cut that fails must not burn the day.
policy._resetArchiveReservation();
await db.setState('auto_clip_log', '[]');
--await db.setState('ai_used_moments', '[]');
++await require('../server/ai/moment-registry')._clear();
registry._reset();
media.createClip = async () => null;
assert.strictEqual(await archiveCut(72), null, 'the failed archival cut returns nothing');
ddiff --git a/test/moment-registry-durable.test.js b/test/moment-registry-durable.test.js
nnew file mode 100644
iindex 00000000..0d2c27ee
---- /dev/null
++++ b/test/moment-registry-durable.test.js
@@@ -0,0 +1,38 @@
++'use strict';
++// The AI's used moments are durable (item 151, C06; migration 0024): nothing falls out at 800 entries, the old JSON
++// list in app_state moves in once, the same-recording check finds a moment among thousands, and rows go after 90 days.
++const assert = require('assert');
++const db = require('../server/db/database');
++(async () => {
++ await db.initDb();
++ const reg = require('../server/ai/moment-registry');
++
++ // The old list moves in once: a row per entry, then the state is not read again.
++ await db.run('DELETE FROM ai_moments');
++ await db.setState('ai_moments_moved', '');
++ await db.setState('ai_used_moments', JSON.stringify([
++ { kind: 'clip', vod_id: 5001, stream_id: 77, offset: 600, sig: 'cat walks keyboard', title: 'Cat on keys', ts: Date.now() - 3600_000 },
++ { kind: 'paste', vod_id: 5002, offset: 60, sig: 'big win screen', title: 'Big win', ts: Date.now() - 7200_000 },
++ ]));
++ reg._reset();
++ assert.ok(await reg.isUsed({ vod_id: 5001, offset: 650 }), 'a moved entry blocks its spot');
++ assert.strictEqual(Number((await db.get('SELECT count(*)::int AS n FROM ai_moments')).n), 2);
++ await db.setState('ai_used_moments', JSON.stringify([{ kind: 'clip', vod_id: 9999, offset: 1, ts: Date.now() }]));
++ reg._reset();
++ assert.ok(!await reg.isUsed({ vod_id: 9999, offset: 1 }), 'moved once: the list is not read again');
++
++ // Nothing falls out at 800: the first of 900 moments still blocks its spot.
++ await reg._clear();
++ await reg.record({ kind: 'clip', vod_id: 1, offset: 100, title: 'The first one' });
++ for (let i = 0; i < 900; i++) await reg.record({ kind: 'clip', vod_id: 1000 + i, offset: 100, title: `Moment ${i} of a long week` });
++ assert.match(await reg.usedReason({ vod_id: 1, offset: 150 }), /^clip already made at 100s of this VOD$/);
++ assert.ok(!await reg.isUsed({ vod_id: 1, offset: 400 }), 'two minutes away is free');
++ assert.match(await reg.usedReason({ stream_id: null, vod_id: 1500, offset: 30 }) || '', /already made at 100s/);
++ assert.deepStrictEqual(await reg.usedOffsets(1), [100]);
++ assert.strictEqual((await reg.lastOfKind('clip', { vod_id: 1 })).title, 'The first one');
++
++ // Rows past 90 days no longer block, and are pruned.
++ await db.run("UPDATE ai_moments SET created_at = now() - interval '91 days' WHERE vod_id = '1'");
++ assert.ok(!await reg.isUsed({ vod_id: 1, offset: 120 }), 'past 90 days it is free');
++ console.log('moment registry durable: all checks passed');
++})().catch((e) => { console.error(e); process.exit(1); });