openvibers/openvibe.live Commits
Unresolved lineage references for operators (roadmap D20, remaining item 2). Every unresolved answer of the canonical resolver is counted in lineage_unresolved by its inputs (canonical 'clip_id=7 slug=x') and reason, with detail, last caller, count and first/last seen, and kept 30 days: to another service through /internal/lineage/resolve, and Live's own clip-owner check. Reasons: conflict, source_unavailable, not_found, display_name_only; no_input is not kept. Staff read them at GET /api/admin/lineage/unresolved (?reason=, counts per reason). Recording never changes an answer. test/lineage-resolver.test.js
Changes
@@@ -52,6 +52,13 @@ const router = express.Router();router.use(requireAuth, permissions.requireAdmin);// ── Process diagnostics (event-loop delay, memory, sockets, jobs, queues, migrations) ──++// Unresolved lineage references (D20: the operator view). ?reason=conflict|source_unavailable|not_found|display_name_only…++router.get('/lineage/unresolved', (req, res) => {++ res.set('Cache-Control', 'no-store');++ const reason = typeof req.query.reason === 'string' && /^[a-z_]{1,60}$/.test(req.query.reason) ? req.query.reason : null;++ res.json(require('../lineage/unresolved').list({ reason, limit: req.query.limit }));++});++router.get('/diagnostics', (req, res) => {res.set('Cache-Control', 'no-store');res.json(require('../diagnostics').snapshot({@@@ -15,6 +15,7 @@ const express = require('express');const { http } = require('openvibe-contracts');const { guard } = require('../net/service-guard');const resolver = require('./resolver');++const unresolved = require('./unresolved');const router = express.Router();router.use(guard('live.lineage.resolve'));@@@ -23,7 +24,9 @@ async function answer(req, res, raw, flat) {const { input, error } = resolver.normalizeRequest(raw, { flat });if (error) return http.sendProblem(res, 400, 'lineage.invalid_request', { detail: error });try {-- res.set('Cache-Control', 'no-store').json(await resolver.resolve(input));++ const out = await resolver.resolve(input);++ unresolved.record(input, out, req.principal && req.principal.sub); // the operator view (D20)++ res.set('Cache-Control', 'no-store').json(out);} catch (err) {console.warn('[Lineage] resolve failed:', err.message);http.sendProblem(res, 500, 'lineage.failed', { detail: 'the resolver failed' });nnew file mode 100644@@@ -0,0 +1,85 @@++'use strict';++/**++ * Unresolved references, for operators (roadmap D20 remaining 2: "an operator view of unresolved++ * references"). Every unresolved answer of the canonical resolver — to another service through++ * /internal/lineage/resolve (./routes.js) or to Live's own clip-owner check (media-proxy/clips.js) — is++ * counted here by its inputs and reason: a conflict, a source that could not be read, a record whose++ * owner is gone, a display name offered alone. `no_input` is not a reference and is not kept.++ *++ * lineage_unresolved (ref, reason) → detail, last caller, count, first and last seen++ * ref the inputs, canonical and sorted: "clip_id=123 slug=alice" (ids and slugs only;++ * a display name is recorded as display_name=… because that is what was offered)++ *++ * Rows not seen for 30 days are dropped (checked at most once an hour, on a write). Staff read them at++ * GET /api/admin/lineage/unresolved (admin/routes.js); a failure to record never changes an answer.++ */++const db = require('../db/database');++++const KEEP_DAYS = 30;++const PRUNE_EVERY_MS = 3600 * 1000;++let ensured = false;++let lastPrune = 0;++++function ensureTable() {++ if (ensured) return;++ const d = db.getDb();++ d.exec(`CREATE TABLE IF NOT EXISTS lineage_unresolved (++ ref TEXT NOT NULL,++ reason TEXT NOT NULL,++ detail TEXT,++ last_caller TEXT,++ count INTEGER NOT NULL DEFAULT 1,++ first_at TEXT NOT NULL,++ last_at TEXT NOT NULL,++ PRIMARY KEY (ref, reason)++ )`);++ d.exec('CREATE INDEX IF NOT EXISTS idx_lineage_unresolved_last ON lineage_unresolved(last_at DESC)');++ ensured = true;++}++++/** A normalized request (resolver.normalizeRequest) → "k=v k=v", sorted; legacy_ids flattened. */++function refOf(input) {++ const parts = [];++ for (const [k, v] of Object.entries(input || {})) {++ if (v == null) continue;++ if (k === 'legacy_ids' && typeof v === 'object') { for (const [lk, lv] of Object.entries(v)) if (lv != null) parts.push(`${lk}=${lv}`); continue; }++ parts.push(`${k}=${String(v).replace(/\s+/g, ' ').slice(0, 80)}`);++ }++ return parts.sort().join(' ').slice(0, 500);++}++++/** Count one unresolved answer. result: { status, reason, detail }; caller: 'svc:media', 'live:clip-owner'… */++function record(input, result, caller, { now = Date.now() } = {}) {++ try {++ if (!result || result.status === 'resolved' || !result.reason || result.reason === 'no_input') return false;++ const ref = refOf(input);++ if (!ref) return false;++ ensureTable();++ const at = new Date(now).toISOString();++ db.run(`INSERT INTO lineage_unresolved (ref, reason, detail, last_caller, count, first_at, last_at) VALUES (?, ?, ?, ?, 1, ?, ?)++ ON CONFLICT (ref, reason) DO UPDATE SET detail = excluded.detail, last_caller = excluded.last_caller, count = count + 1, last_at = excluded.last_at`,++ [ref, String(result.reason).slice(0, 60), result.detail ? String(result.detail).slice(0, 500) : null, caller ? String(caller).slice(0, 60) : null, at, at]);++ if (now - lastPrune > PRUNE_EVERY_MS) {++ lastPrune = now;++ db.run('DELETE FROM lineage_unresolved WHERE last_at < ?', [new Date(now - KEEP_DAYS * 86400000).toISOString()]);++ }++ return true;++ } catch (err) {++ console.warn('[Lineage] could not record an unresolved reference:', err.message);++ return false;++ }++}++++/** For staff: the newest first, optionally one reason; plus counts per reason. */++function list({ reason = null, limit = 100 } = {}) {++ ensureTable();++ const n = Math.min(Math.max(parseInt(limit, 10) || 100, 1), 500);++ const rows = reason++ ? db.all('SELECT * FROM lineage_unresolved WHERE reason = ? ORDER BY last_at DESC LIMIT ?', [reason, n])++ : db.all('SELECT * FROM lineage_unresolved ORDER BY last_at DESC LIMIT ?', [n]);++ const byReason = {};++ for (const r of db.all('SELECT reason, COUNT(*) AS refs, SUM(count) AS answers FROM lineage_unresolved GROUP BY reason')) byReason[r.reason] = { refs: r.refs, answers: r.answers };++ return { unresolved: rows, by_reason: byReason, keep_days: KEEP_DAYS };++}++++module.exports = { record, list, refOf, KEEP_DAYS };@@@ -88,6 +88,7 @@ function withUserFields(clip) {async function clipChannelOwnerId(clip) {if (!clip || clip.id == null) return null;const r = await lineage.resolveOwner({ clip_id: String(clip.id) }, { records: { clip }, rules: ['stream_lookup', 'vod_parent'] });++ if (r.status !== 'resolved') require('../lineage/unresolved').record({ clip_id: String(clip.id) }, r, 'live:clip-owner');return r.status === 'resolved' ? r.userId : null;}@@@ -290,6 +290,17 @@ function checkContract(out, label) {assert.deepStrictEqual([res.status, res.json.status, res.json.reason], [200, 'unresolved', 'display_name_only']);res = await call('POST', '', { slug: 'mallory', vod_id: '42' });assert.deepStrictEqual([res.status, res.json.reason], [200, 'conflict']);++ await call('GET', 'slug=mallory&vod_id=42');++ // The operator view (D20 remaining 2): unresolved answers are counted by inputs and reason; resolved ones are not.++ const ops = require('../server/lineage/unresolved').list();++ const byRef = Object.fromEntries(ops.unresolved.map((u) => [`${u.ref}|${u.reason}`, u]));++ assert.deepStrictEqual([byRef['slug=mallory vod_id=42|conflict'].count, byRef['slug=mallory vod_id=42|conflict'].last_caller], [2, 'svc:community'], 'counted per ref and reason, GET and POST alike');++ assert.ok(byRef['display_name=Alice Wonder|display_name_only'], 'a display name offered alone is kept as offered');++ assert.ok(!Object.keys(byRef).some((k) => k.startsWith('clip_id=7 network_user_id=57') || k.startsWith('live_user_id=17 slug=alice')), 'resolved answers are not recorded');++ assert.ok(['conflict', 'display_name_only'].every((r) => ops.by_reason[r] && ops.by_reason[r].refs >= 1), JSON.stringify(ops.by_reason));++ assert.ok(ops.unresolved.some((u) => u.last_caller === 'live:clip-owner' && u.reason === 'source_unavailable'), "Live's own clip-owner check records too (Media down earlier in this test)");++ assert.ok(require('../server/lineage/unresolved').list({ reason: 'conflict' }).unresolved.every((u) => u.reason === 'conflict'));++ assert.strictEqual(require('../server/lineage/unresolved').record({}, { status: 'unresolved', reason: 'no_input' }, 'x'), false, 'no_input is not a reference');for (const [q, body] of [['username=alice', null], ['slug=Alice%20Wonder', null], ['', { owner_subject: '17' }], ['', { vod_id: '42', extra: 1 }], ['slug=a&slug=b', null]]) {res = await call(body ? 'POST' : 'GET', q, body);assert.deepStrictEqual([res.status, res.json.code], [400, 'lineage.invalid_request'], `${q || JSON.stringify(body)} is refused`);
@@ -52,6 +52,13 @@ const router = express.Router();
@@ -52,6 +52,13 @@ const router = express.Router();
router.use(requireAuth, permissions.requireAdmin);
router.use(requireAuth, permissions.requireAdmin);
// ── Process diagnostics (event-loop delay, memory, sockets, jobs, queues, migrations) ──
// ── Process diagnostics (event-loop delay, memory, sockets, jobs, queues, migrations) ──
+// Unresolved lineage references (D20: the operator view). ?reason=conflict|source_unavailable|not_found|display_name_only…
+router.get('/lineage/unresolved', (req, res) => {
+ res.set('Cache-Control', 'no-store');
+ const reason = typeof req.query.reason === 'string' && /^[a-z_]{1,60}$/.test(req.query.reason) ? req.query.reason : null;
+ res.json(require('../lineage/unresolved').list({ reason, limit: req.query.limit }));
+});
+
router.get('/diagnostics', (req, res) => {
router.get('/diagnostics', (req, res) => {
res.set('Cache-Control', 'no-store');
res.set('Cache-Control', 'no-store');
res.json(require('../diagnostics').snapshot({
res.json(require('../diagnostics').snapshot({
@@ -15,6 +15,7 @@ const express = require('express');
@@ -15,6 +15,7 @@ const express = require('express');
const { http } = require('openvibe-contracts');
const { http } = require('openvibe-contracts');
const { guard } = require('../net/service-guard');
const { guard } = require('../net/service-guard');
const resolver = require('./resolver');
const resolver = require('./resolver');
+const unresolved = require('./unresolved');
const router = express.Router();
const router = express.Router();
router.use(guard('live.lineage.resolve'));
router.use(guard('live.lineage.resolve'));
@@ -23,7 +24,9 @@ async function answer(req, res, raw, flat) {
@@ -23,7 +24,9 @@ async function answer(req, res, raw, flat) {
const { input, error } = resolver.normalizeRequest(raw, { flat });
const { input, error } = resolver.normalizeRequest(raw, { flat });
if (error) return http.sendProblem(res, 400, 'lineage.invalid_request', { detail: error });
if (error) return http.sendProblem(res, 400, 'lineage.invalid_request', { detail: error });
try {
try {
- res.set('Cache-Control', 'no-store').json(await resolver.resolve(input));
+ const out = await resolver.resolve(input);
+ unresolved.record(input, out, req.principal && req.principal.sub); // the operator view (D20)
+ res.set('Cache-Control', 'no-store').json(out);
} catch (err) {
} catch (err) {
console.warn('[Lineage] resolve failed:', err.message);
console.warn('[Lineage] resolve failed:', err.message);
http.sendProblem(res, 500, 'lineage.failed', { detail: 'the resolver failed' });
http.sendProblem(res, 500, 'lineage.failed', { detail: 'the resolver failed' });
new file mode 100644
new file mode 100644
@@ -0,0 +1,85 @@
@@ -0,0 +1,85 @@
+'use strict';
+/**
+ * Unresolved references, for operators (roadmap D20 remaining 2: "an operator view of unresolved
+ * references"). Every unresolved answer of the canonical resolver — to another service through
+ * /internal/lineage/resolve (./routes.js) or to Live's own clip-owner check (media-proxy/clips.js) — is
+ * counted here by its inputs and reason: a conflict, a source that could not be read, a record whose
+ * owner is gone, a display name offered alone. `no_input` is not a reference and is not kept.
+ *
+ * lineage_unresolved (ref, reason) → detail, last caller, count, first and last seen
+ * ref the inputs, canonical and sorted: "clip_id=123 slug=alice" (ids and slugs only;
+ * a display name is recorded as display_name=… because that is what was offered)
+ *
+ * Rows not seen for 30 days are dropped (checked at most once an hour, on a write). Staff read them at
+ * GET /api/admin/lineage/unresolved (admin/routes.js); a failure to record never changes an answer.
+ */
+const db = require('../db/database');
+
+const KEEP_DAYS = 30;
+const PRUNE_EVERY_MS = 3600 * 1000;
+let ensured = false;
+let lastPrune = 0;
+
+function ensureTable() {
+ if (ensured) return;
+ const d = db.getDb();
+ d.exec(`CREATE TABLE IF NOT EXISTS lineage_unresolved (
+ ref TEXT NOT NULL,
+ reason TEXT NOT NULL,
+ detail TEXT,
+ last_caller TEXT,
+ count INTEGER NOT NULL DEFAULT 1,
+ first_at TEXT NOT NULL,
+ last_at TEXT NOT NULL,
+ PRIMARY KEY (ref, reason)
+ )`);
+ d.exec('CREATE INDEX IF NOT EXISTS idx_lineage_unresolved_last ON lineage_unresolved(last_at DESC)');
+ ensured = true;
+}
+
+/** A normalized request (resolver.normalizeRequest) → "k=v k=v", sorted; legacy_ids flattened. */
+function refOf(input) {
+ const parts = [];
+ for (const [k, v] of Object.entries(input || {})) {
+ if (v == null) continue;
+ if (k === 'legacy_ids' && typeof v === 'object') { for (const [lk, lv] of Object.entries(v)) if (lv != null) parts.push(`${lk}=${lv}`); continue; }
+ parts.push(`${k}=${String(v).replace(/\s+/g, ' ').slice(0, 80)}`);
+ }
+ return parts.sort().join(' ').slice(0, 500);
+}
+
+/** Count one unresolved answer. result: { status, reason, detail }; caller: 'svc:media', 'live:clip-owner'… */
+function record(input, result, caller, { now = Date.now() } = {}) {
+ try {
+ if (!result || result.status === 'resolved' || !result.reason || result.reason === 'no_input') return false;
+ const ref = refOf(input);
+ if (!ref) return false;
+ ensureTable();
+ const at = new Date(now).toISOString();
+ db.run(`INSERT INTO lineage_unresolved (ref, reason, detail, last_caller, count, first_at, last_at) VALUES (?, ?, ?, ?, 1, ?, ?)
+ ON CONFLICT (ref, reason) DO UPDATE SET detail = excluded.detail, last_caller = excluded.last_caller, count = count + 1, last_at = excluded.last_at`,
+ [ref, String(result.reason).slice(0, 60), result.detail ? String(result.detail).slice(0, 500) : null, caller ? String(caller).slice(0, 60) : null, at, at]);
+ if (now - lastPrune > PRUNE_EVERY_MS) {
+ lastPrune = now;
+ db.run('DELETE FROM lineage_unresolved WHERE last_at < ?', [new Date(now - KEEP_DAYS * 86400000).toISOString()]);
+ }
+ return true;
+ } catch (err) {
+ console.warn('[Lineage] could not record an unresolved reference:', err.message);
+ return false;
+ }
+}
+
+/** For staff: the newest first, optionally one reason; plus counts per reason. */
+function list({ reason = null, limit = 100 } = {}) {
+ ensureTable();
+ const n = Math.min(Math.max(parseInt(limit, 10) || 100, 1), 500);
+ const rows = reason
+ ? db.all('SELECT * FROM lineage_unresolved WHERE reason = ? ORDER BY last_at DESC LIMIT ?', [reason, n])
+ : db.all('SELECT * FROM lineage_unresolved ORDER BY last_at DESC LIMIT ?', [n]);
+ const byReason = {};
+ for (const r of db.all('SELECT reason, COUNT(*) AS refs, SUM(count) AS answers FROM lineage_unresolved GROUP BY reason')) byReason[r.reason] = { refs: r.refs, answers: r.answers };
+ return { unresolved: rows, by_reason: byReason, keep_days: KEEP_DAYS };
+}
+
+module.exports = { record, list, refOf, KEEP_DAYS };
@@ -88,6 +88,7 @@ function withUserFields(clip) {
@@ -88,6 +88,7 @@ function withUserFields(clip) {
async function clipChannelOwnerId(clip) {
async function clipChannelOwnerId(clip) {
if (!clip || clip.id == null) return null;
if (!clip || clip.id == null) return null;
const r = await lineage.resolveOwner({ clip_id: String(clip.id) }, { records: { clip }, rules: ['stream_lookup', 'vod_parent'] });
const r = await lineage.resolveOwner({ clip_id: String(clip.id) }, { records: { clip }, rules: ['stream_lookup', 'vod_parent'] });
+ if (r.status !== 'resolved') require('../lineage/unresolved').record({ clip_id: String(clip.id) }, r, 'live:clip-owner');
return r.status === 'resolved' ? r.userId : null;
return r.status === 'resolved' ? r.userId : null;
}
}
@@ -290,6 +290,17 @@ function checkContract(out, label) {
@@ -290,6 +290,17 @@ function checkContract(out, label) {
assert.deepStrictEqual([res.status, res.json.status, res.json.reason], [200, 'unresolved', 'display_name_only']);
assert.deepStrictEqual([res.status, res.json.status, res.json.reason], [200, 'unresolved', 'display_name_only']);
res = await call('POST', '', { slug: 'mallory', vod_id: '42' });
res = await call('POST', '', { slug: 'mallory', vod_id: '42' });
assert.deepStrictEqual([res.status, res.json.reason], [200, 'conflict']);
assert.deepStrictEqual([res.status, res.json.reason], [200, 'conflict']);
+ await call('GET', 'slug=mallory&vod_id=42');
+ // The operator view (D20 remaining 2): unresolved answers are counted by inputs and reason; resolved ones are not.
+ const ops = require('../server/lineage/unresolved').list();
+ const byRef = Object.fromEntries(ops.unresolved.map((u) => [`${u.ref}|${u.reason}`, u]));
+ assert.deepStrictEqual([byRef['slug=mallory vod_id=42|conflict'].count, byRef['slug=mallory vod_id=42|conflict'].last_caller], [2, 'svc:community'], 'counted per ref and reason, GET and POST alike');
+ assert.ok(byRef['display_name=Alice Wonder|display_name_only'], 'a display name offered alone is kept as offered');
+ assert.ok(!Object.keys(byRef).some((k) => k.startsWith('clip_id=7 network_user_id=57') || k.startsWith('live_user_id=17 slug=alice')), 'resolved answers are not recorded');
+ assert.ok(['conflict', 'display_name_only'].every((r) => ops.by_reason[r] && ops.by_reason[r].refs >= 1), JSON.stringify(ops.by_reason));
+ assert.ok(ops.unresolved.some((u) => u.last_caller === 'live:clip-owner' && u.reason === 'source_unavailable'), "Live's own clip-owner check records too (Media down earlier in this test)");
+ assert.ok(require('../server/lineage/unresolved').list({ reason: 'conflict' }).unresolved.every((u) => u.reason === 'conflict'));
+ assert.strictEqual(require('../server/lineage/unresolved').record({}, { status: 'unresolved', reason: 'no_input' }, 'x'), false, 'no_input is not a reference');
for (const [q, body] of [['username=alice', null], ['slug=Alice%20Wonder', null], ['', { owner_subject: '17' }], ['', { vod_id: '42', extra: 1 }], ['slug=a&slug=b', null]]) {
for (const [q, body] of [['username=alice', null], ['slug=Alice%20Wonder', null], ['', { owner_subject: '17' }], ['', { vod_id: '42', extra: 1 }], ['slug=a&slug=b', null]]) {
res = await call(body ? 'POST' : 'GET', q, body);
res = await call(body ? 'POST' : 'GET', q, body);
assert.deepStrictEqual([res.status, res.json.code], [400, 'lineage.invalid_request'], `${q || JSON.stringify(body)} is refused`);
assert.deepStrictEqual([res.status, res.json.code], [400, 'lineage.invalid_request'], `${q || JSON.stringify(body)} is refused`);