openvibers/openvibe.community Commits
Pulse: a Live recording names its channel's person (#394)
Changes
@@@ -2,6 +2,11 @@## Unreleased++- A Live recording, clip or "went live" card names its channel's person: the one subject Community knows by that++ username, none when two share it. 70 of 112 Live recordings and clips since 11-01 had no subject, so the Following++ feed never showed them and one streamer's evening folded as two runs. An update fills a card that had no person; a++ card's person never changes once known. Test: `test/pulse-sources.test.js`.++- The Feed folds a person's run of one kind on one site while each card is within two hours of the run (it was 15minutes from the run's first card): 11-11's feed showed one streamer's evening of restarts as nine rows and another'smorning of uploads as five "added 2 videos" rows. A trending card is skipped when the page already shows its link or@@@ -16,6 +16,7 @@* server/pulse/consumer.js; no route or page is slowed by this job.*/const store = require('./store');++const { subjectsByUsername } = require('../pastes/store');const { sqlTime } = require('../http/v1');const MINUTE = 60_000;@@@ -47,6 +48,24 @@ function createPulseSources({ db, config = {}, fetchImpl = globalThis.fetch, nowlet timer = null;let running = null;++ /**++ * A channel's person, when a source names only the channel: the one subject Community knows by that username (its++ * subject_projection cache), or none when two people share it. 70 of 112 Live recordings and clips since 11-01++ * had no subject, so the Following feed never showed them and one person's evening folded as two runs.++ */++ function channelSubjects() {++ const cache = new Map();++ return async (username) => {++ const key = clean(username, 64).toLowerCase();++ if (!key) return null;++ if (!cache.has(key)) {++ const ids = (await subjectsByUsername(db, key)).filter((id) => subject(id));++ cache.set(key, ids.length === 1 ? ids[0] : null);++ }++ return cache.get(key);++ };++ }++async function getJson(url, { timeoutMs = 8000 } = {}) {const res = await fetchImpl(url, { headers: { accept: 'application/json' }, signal: AbortSignal.timeout(timeoutMs) });if (!res.ok) throw new Error(`${url} answered ${res.status}`);@@@ -286,6 +305,7 @@ function createPulseSources({ db, config = {}, fetchImpl = globalThis.fetch, nowconst seen = new Set();let cursor = null, created = 0;const live = await getJson(`${liveApi}/api/streams`);++ const personOf = channelSubjects();const onAir = new Map();const liveIds = new Set();for (const s of (Array.isArray(live.streams) ? live.streams : [])) {@@@ -298,7 +318,7 @@ function createPulseSources({ db, config = {}, fetchImpl = globalThis.fetch, nowconst name = clean(s.display_name || channel, 80);const r = await store.upsertItem(db, { source_service: 'live', source_type: 'stream', source_id: String(id),title: `${name} went live: ${clean(s.title, 160) || 'a stream'}`, url: `${liveApi}/@${encodeURIComponent(channel)}`,-- actor_name: name, origin: 'user', occurred_at: started });++ actor_name: name, actor_subject: await personOf(channel), origin: 'user', occurred_at: started });if (r.created) created++;}for (let page = 0; page < 100; page++) {@@@ -331,7 +351,7 @@ function createPulseSources({ db, config = {}, fetchImpl = globalThis.fetch, nowconst r = await store.upsertItem(db, { source_service: 'live', source_type: type, source_id: id,title: `${type === 'recording' ? 'Recording: ' : ''}${clean(v.title, 150) || (type === 'recording' ? 'A stream' : 'A clip')}`,url: https(v.href), picture_url: https(v.thumbnail_url),-- actor_subject: prior && prior.actor_subject || null, actor_name: clean(v.channel && v.channel.display_name, 80),++ actor_subject: (prior && prior.actor_subject) || await personOf(channel), actor_name: clean(v.channel && v.channel.display_name, 80),origin: 'user', occurred_at: at, text: clean(v.excerpt, 280), media_length: Number(v.duration_seconds) > 0 ? Number(v.duration_seconds) : null });if (r.created) created++;if (prior) await store.removeItem(db, 'live', 'stream', prior.source_id);@@@ -84,12 +84,15 @@ async function upsertItem(db, it) {// time forward (a Wiki page edited again, a busy chat hour still filling): provenance is otherwise// what the first record said. A source that sends no text (undefined) keeps the text already there.if (it.text === undefined) it = { ...it, text: had.text };++ // A person found later fills a card that had none (Live recordings named only their channel until++ // 2026-11-12); a card's person, once known, never changes.++ const person = it.actor_subject || null;if (it.touch) {-- await db.prepare('UPDATE pulse_items SET title = ?, url = ?, picture_url = ?, preview_url = ?, media_ratio = ?, media_length = ?, app_name = ?, text = ?, starts_at = ?, actor_name = ?, visibility = ?, occurred_at = ? WHERE source_service = ? AND source_type = ? AND source_id = ?')-- .run(it.title, it.url, it.picture_url || null, it.preview_url || null, it.media_ratio || null, it.media_length || null, it.app_name || null, it.text == null ? null : String(it.text).slice(0, 300), pulseTime(it.starts_at) || null, it.actor_name || null, it.visibility || 'public', it.occurred_at, ...key);++ await db.prepare('UPDATE pulse_items SET title = ?, url = ?, picture_url = ?, preview_url = ?, media_ratio = ?, media_length = ?, app_name = ?, text = ?, starts_at = ?, actor_name = ?, actor_subject = COALESCE(actor_subject, ?), visibility = ?, occurred_at = ? WHERE source_service = ? AND source_type = ? AND source_id = ?')++ .run(it.title, it.url, it.picture_url || null, it.preview_url || null, it.media_ratio || null, it.media_length || null, it.app_name || null, it.text == null ? null : String(it.text).slice(0, 300), pulseTime(it.starts_at) || null, it.actor_name || null, person, it.visibility || 'public', it.occurred_at, ...key);} else {-- await db.prepare('UPDATE pulse_items SET title = ?, url = ?, picture_url = ?, preview_url = ?, media_ratio = ?, media_length = ?, app_name = ?, text = ?, starts_at = ?, actor_name = ?, visibility = ? WHERE source_service = ? AND source_type = ? AND source_id = ?')-- .run(it.title, it.url, it.picture_url || null, it.preview_url || null, it.media_ratio || null, it.media_length || null, it.app_name || null, it.text == null ? null : String(it.text).slice(0, 300), pulseTime(it.starts_at) || null, it.actor_name || null, it.visibility || 'public', ...key);++ await db.prepare('UPDATE pulse_items SET title = ?, url = ?, picture_url = ?, preview_url = ?, media_ratio = ?, media_length = ?, app_name = ?, text = ?, starts_at = ?, actor_name = ?, actor_subject = COALESCE(actor_subject, ?), visibility = ? WHERE source_service = ? AND source_type = ? AND source_id = ?')++ .run(it.title, it.url, it.picture_url || null, it.preview_url || null, it.media_ratio || null, it.media_length || null, it.app_name || null, it.text == null ? null : String(it.text).slice(0, 300), pulseTime(it.starts_at) || null, it.actor_name || null, person, it.visibility || 'public', ...key);}return { item: await find.get(...key), created: false };}@@@ -171,5 +171,33 @@ const day = (ms) => new Date(ms).toISOString().slice(0, 10);assert.ok(await item('recording', 123));});++ await check('a Live recording names its channel\'s person (the one subject known by that username), and fills an older card', async () => {++ const db = t.app.locals.db;++ const { upsertProjection } = require('../server/pastes/store');++ const store = require('../server/pulse/store');++ const subjectOf = async (type, id) => (await db.prepare('SELECT actor_subject FROM pulse_items WHERE source_service = ? AND source_type = ? AND source_id = ?')++ .get('live', type, String(id)) || {}).actor_subject || null;++ await upsertProjection(db, { subject_id: alex.subject_id, username: 'alex', display_name: 'Alex' });++ // An older card that the sync wrote before it knew the person (11-11's 70 of 112), and two people who share++ // the username "twin": nobody is guessed for them.++ await store.upsertItem(db, { source_service: 'live', source_type: 'recording', source_id: '126', title: 'Recording: Before', url: 'https://openvibe.video/watch?v=before',++ actor_name: 'Alex', origin: 'user', occurred_at: badgeAt });++ await upsertProjection(db, { subject_id: 'usr_01JAB2C3D4E5F6G7H8J9K0MNPA', username: 'twin', display_name: 'Twin A' });++ await upsertProjection(db, { subject_id: 'usr_01JAB2C3D4E5F6G7H8J9K0MNPB', username: 'Twin', display_name: 'Twin B' });++ extraContent = [++ { id: 126, kind: 'vod', title: 'Before', href: 'https://openvibe.video/watch?v=before', created_at: badgeAt, channel: { username: 'alex', display_name: 'Alex' } },++ { id: 127, kind: 'vod', title: 'Twin stream', href: 'https://openvibe.video/watch?v=twin', created_at: badgeAt, channel: { username: 'twin', display_name: 'Twin' } },++ ];++ await t.app.locals.pulseSources.runOnce({ force: true });++ assert.strictEqual(await subjectOf('recording', 123), alex.subject_id, 'a new recording carries its channel\'s person');++ assert.strictEqual(await subjectOf('clip', 124), alex.subject_id, 'a clip too');++ assert.strictEqual(await subjectOf('recording', 126), alex.subject_id, 'an older card without a person is filled');++ assert.strictEqual(await subjectOf('recording', 127), null, 'a username two people share names nobody');++ const followed = await store.upsertItem(db, { source_service: 'live', source_type: 'recording', source_id: '126', title: 'Recording: Before', url: 'https://openvibe.video/watch?v=before',++ actor_subject: 'usr_01JAB2C3D4E5F6G7H8J9K0MNPA', actor_name: 'Alex', origin: 'user', occurred_at: badgeAt });++ assert.strictEqual(followed.item.actor_subject, alex.subject_id, 'a card\'s person, once known, never changes');++ extraContent = [];++ });++await done(t);})();
@@ -2,6 +2,11 @@
@@ -2,6 +2,11 @@
## Unreleased
## Unreleased
+- A Live recording, clip or "went live" card names its channel's person: the one subject Community knows by that
+ username, none when two share it. 70 of 112 Live recordings and clips since 11-01 had no subject, so the Following
+ feed never showed them and one streamer's evening folded as two runs. An update fills a card that had no person; a
+ card's person never changes once known. Test: `test/pulse-sources.test.js`.
+
- The Feed folds a person's run of one kind on one site while each card is within two hours of the run (it was 15
- The Feed folds a person's run of one kind on one site while each card is within two hours of the run (it was 15
minutes from the run's first card): 11-11's feed showed one streamer's evening of restarts as nine rows and another's
minutes from the run's first card): 11-11's feed showed one streamer's evening of restarts as nine rows and another's
morning of uploads as five "added 2 videos" rows. A trending card is skipped when the page already shows its link or
morning of uploads as five "added 2 videos" rows. A trending card is skipped when the page already shows its link or
@@ -16,6 +16,7 @@
@@ -16,6 +16,7 @@
* server/pulse/consumer.js; no route or page is slowed by this job.
* server/pulse/consumer.js; no route or page is slowed by this job.
*/
*/
const store = require('./store');
const store = require('./store');
+const { subjectsByUsername } = require('../pastes/store');
const { sqlTime } = require('../http/v1');
const { sqlTime } = require('../http/v1');
const MINUTE = 60_000;
const MINUTE = 60_000;
@@ -47,6 +48,24 @@ function createPulseSources({ db, config = {}, fetchImpl = globalThis.fetch, now
@@ -47,6 +48,24 @@ function createPulseSources({ db, config = {}, fetchImpl = globalThis.fetch, now
let timer = null;
let timer = null;
let running = null;
let running = null;
+ /**
+ * A channel's person, when a source names only the channel: the one subject Community knows by that username (its
+ * subject_projection cache), or none when two people share it. 70 of 112 Live recordings and clips since 11-01
+ * had no subject, so the Following feed never showed them and one person's evening folded as two runs.
+ */
+ function channelSubjects() {
+ const cache = new Map();
+ return async (username) => {
+ const key = clean(username, 64).toLowerCase();
+ if (!key) return null;
+ if (!cache.has(key)) {
+ const ids = (await subjectsByUsername(db, key)).filter((id) => subject(id));
+ cache.set(key, ids.length === 1 ? ids[0] : null);
+ }
+ return cache.get(key);
+ };
+ }
+
async function getJson(url, { timeoutMs = 8000 } = {}) {
async function getJson(url, { timeoutMs = 8000 } = {}) {
const res = await fetchImpl(url, { headers: { accept: 'application/json' }, signal: AbortSignal.timeout(timeoutMs) });
const res = await fetchImpl(url, { headers: { accept: 'application/json' }, signal: AbortSignal.timeout(timeoutMs) });
if (!res.ok) throw new Error(`${url} answered ${res.status}`);
if (!res.ok) throw new Error(`${url} answered ${res.status}`);
@@ -286,6 +305,7 @@ function createPulseSources({ db, config = {}, fetchImpl = globalThis.fetch, now
@@ -286,6 +305,7 @@ function createPulseSources({ db, config = {}, fetchImpl = globalThis.fetch, now
const seen = new Set();
const seen = new Set();
let cursor = null, created = 0;
let cursor = null, created = 0;
const live = await getJson(`${liveApi}/api/streams`);
const live = await getJson(`${liveApi}/api/streams`);
+ const personOf = channelSubjects();
const onAir = new Map();
const onAir = new Map();
const liveIds = new Set();
const liveIds = new Set();
for (const s of (Array.isArray(live.streams) ? live.streams : [])) {
for (const s of (Array.isArray(live.streams) ? live.streams : [])) {
@@ -298,7 +318,7 @@ function createPulseSources({ db, config = {}, fetchImpl = globalThis.fetch, now
@@ -298,7 +318,7 @@ function createPulseSources({ db, config = {}, fetchImpl = globalThis.fetch, now
const name = clean(s.display_name || channel, 80);
const name = clean(s.display_name || channel, 80);
const r = await store.upsertItem(db, { source_service: 'live', source_type: 'stream', source_id: String(id),
const r = await store.upsertItem(db, { source_service: 'live', source_type: 'stream', source_id: String(id),
title: `${name} went live: ${clean(s.title, 160) || 'a stream'}`, url: `${liveApi}/@${encodeURIComponent(channel)}`,
title: `${name} went live: ${clean(s.title, 160) || 'a stream'}`, url: `${liveApi}/@${encodeURIComponent(channel)}`,
- actor_name: name, origin: 'user', occurred_at: started });
+ actor_name: name, actor_subject: await personOf(channel), origin: 'user', occurred_at: started });
if (r.created) created++;
if (r.created) created++;
}
}
for (let page = 0; page < 100; page++) {
for (let page = 0; page < 100; page++) {
@@ -331,7 +351,7 @@ function createPulseSources({ db, config = {}, fetchImpl = globalThis.fetch, now
@@ -331,7 +351,7 @@ function createPulseSources({ db, config = {}, fetchImpl = globalThis.fetch, now
const r = await store.upsertItem(db, { source_service: 'live', source_type: type, source_id: id,
const r = await store.upsertItem(db, { source_service: 'live', source_type: type, source_id: id,
title: `${type === 'recording' ? 'Recording: ' : ''}${clean(v.title, 150) || (type === 'recording' ? 'A stream' : 'A clip')}`,
title: `${type === 'recording' ? 'Recording: ' : ''}${clean(v.title, 150) || (type === 'recording' ? 'A stream' : 'A clip')}`,
url: https(v.href), picture_url: https(v.thumbnail_url),
url: https(v.href), picture_url: https(v.thumbnail_url),
- actor_subject: prior && prior.actor_subject || null, actor_name: clean(v.channel && v.channel.display_name, 80),
+ actor_subject: (prior && prior.actor_subject) || await personOf(channel), actor_name: clean(v.channel && v.channel.display_name, 80),
origin: 'user', occurred_at: at, text: clean(v.excerpt, 280), media_length: Number(v.duration_seconds) > 0 ? Number(v.duration_seconds) : null });
origin: 'user', occurred_at: at, text: clean(v.excerpt, 280), media_length: Number(v.duration_seconds) > 0 ? Number(v.duration_seconds) : null });
if (r.created) created++;
if (r.created) created++;
if (prior) await store.removeItem(db, 'live', 'stream', prior.source_id);
if (prior) await store.removeItem(db, 'live', 'stream', prior.source_id);
@@ -84,12 +84,15 @@ async function upsertItem(db, it) {
@@ -84,12 +84,15 @@ async function upsertItem(db, it) {
// time forward (a Wiki page edited again, a busy chat hour still filling): provenance is otherwise
// time forward (a Wiki page edited again, a busy chat hour still filling): provenance is otherwise
// what the first record said. A source that sends no text (undefined) keeps the text already there.
// what the first record said. A source that sends no text (undefined) keeps the text already there.
if (it.text === undefined) it = { ...it, text: had.text };
if (it.text === undefined) it = { ...it, text: had.text };
+ // A person found later fills a card that had none (Live recordings named only their channel until
+ // 2026-11-12); a card's person, once known, never changes.
+ const person = it.actor_subject || null;
if (it.touch) {
if (it.touch) {
- await db.prepare('UPDATE pulse_items SET title = ?, url = ?, picture_url = ?, preview_url = ?, media_ratio = ?, media_length = ?, app_name = ?, text = ?, starts_at = ?, actor_name = ?, visibility = ?, occurred_at = ? WHERE source_service = ? AND source_type = ? AND source_id = ?')
- .run(it.title, it.url, it.picture_url || null, it.preview_url || null, it.media_ratio || null, it.media_length || null, it.app_name || null, it.text == null ? null : String(it.text).slice(0, 300), pulseTime(it.starts_at) || null, it.actor_name || null, it.visibility || 'public', it.occurred_at, ...key);
+ await db.prepare('UPDATE pulse_items SET title = ?, url = ?, picture_url = ?, preview_url = ?, media_ratio = ?, media_length = ?, app_name = ?, text = ?, starts_at = ?, actor_name = ?, actor_subject = COALESCE(actor_subject, ?), visibility = ?, occurred_at = ? WHERE source_service = ? AND source_type = ? AND source_id = ?')
+ .run(it.title, it.url, it.picture_url || null, it.preview_url || null, it.media_ratio || null, it.media_length || null, it.app_name || null, it.text == null ? null : String(it.text).slice(0, 300), pulseTime(it.starts_at) || null, it.actor_name || null, person, it.visibility || 'public', it.occurred_at, ...key);
} else {
} else {
- await db.prepare('UPDATE pulse_items SET title = ?, url = ?, picture_url = ?, preview_url = ?, media_ratio = ?, media_length = ?, app_name = ?, text = ?, starts_at = ?, actor_name = ?, visibility = ? WHERE source_service = ? AND source_type = ? AND source_id = ?')
- .run(it.title, it.url, it.picture_url || null, it.preview_url || null, it.media_ratio || null, it.media_length || null, it.app_name || null, it.text == null ? null : String(it.text).slice(0, 300), pulseTime(it.starts_at) || null, it.actor_name || null, it.visibility || 'public', ...key);
+ await db.prepare('UPDATE pulse_items SET title = ?, url = ?, picture_url = ?, preview_url = ?, media_ratio = ?, media_length = ?, app_name = ?, text = ?, starts_at = ?, actor_name = ?, actor_subject = COALESCE(actor_subject, ?), visibility = ? WHERE source_service = ? AND source_type = ? AND source_id = ?')
+ .run(it.title, it.url, it.picture_url || null, it.preview_url || null, it.media_ratio || null, it.media_length || null, it.app_name || null, it.text == null ? null : String(it.text).slice(0, 300), pulseTime(it.starts_at) || null, it.actor_name || null, person, it.visibility || 'public', ...key);
}
}
return { item: await find.get(...key), created: false };
return { item: await find.get(...key), created: false };
}
}
@@ -171,5 +171,33 @@ const day = (ms) => new Date(ms).toISOString().slice(0, 10);
@@ -171,5 +171,33 @@ const day = (ms) => new Date(ms).toISOString().slice(0, 10);
assert.ok(await item('recording', 123));
assert.ok(await item('recording', 123));
});
});
+ await check('a Live recording names its channel\'s person (the one subject known by that username), and fills an older card', async () => {
+ const db = t.app.locals.db;
+ const { upsertProjection } = require('../server/pastes/store');
+ const store = require('../server/pulse/store');
+ const subjectOf = async (type, id) => (await db.prepare('SELECT actor_subject FROM pulse_items WHERE source_service = ? AND source_type = ? AND source_id = ?')
+ .get('live', type, String(id)) || {}).actor_subject || null;
+ await upsertProjection(db, { subject_id: alex.subject_id, username: 'alex', display_name: 'Alex' });
+ // An older card that the sync wrote before it knew the person (11-11's 70 of 112), and two people who share
+ // the username "twin": nobody is guessed for them.
+ await store.upsertItem(db, { source_service: 'live', source_type: 'recording', source_id: '126', title: 'Recording: Before', url: 'https://openvibe.video/watch?v=before',
+ actor_name: 'Alex', origin: 'user', occurred_at: badgeAt });
+ await upsertProjection(db, { subject_id: 'usr_01JAB2C3D4E5F6G7H8J9K0MNPA', username: 'twin', display_name: 'Twin A' });
+ await upsertProjection(db, { subject_id: 'usr_01JAB2C3D4E5F6G7H8J9K0MNPB', username: 'Twin', display_name: 'Twin B' });
+ extraContent = [
+ { id: 126, kind: 'vod', title: 'Before', href: 'https://openvibe.video/watch?v=before', created_at: badgeAt, channel: { username: 'alex', display_name: 'Alex' } },
+ { id: 127, kind: 'vod', title: 'Twin stream', href: 'https://openvibe.video/watch?v=twin', created_at: badgeAt, channel: { username: 'twin', display_name: 'Twin' } },
+ ];
+ await t.app.locals.pulseSources.runOnce({ force: true });
+ assert.strictEqual(await subjectOf('recording', 123), alex.subject_id, 'a new recording carries its channel\'s person');
+ assert.strictEqual(await subjectOf('clip', 124), alex.subject_id, 'a clip too');
+ assert.strictEqual(await subjectOf('recording', 126), alex.subject_id, 'an older card without a person is filled');
+ assert.strictEqual(await subjectOf('recording', 127), null, 'a username two people share names nobody');
+ const followed = await store.upsertItem(db, { source_service: 'live', source_type: 'recording', source_id: '126', title: 'Recording: Before', url: 'https://openvibe.video/watch?v=before',
+ actor_subject: 'usr_01JAB2C3D4E5F6G7H8J9K0MNPA', actor_name: 'Alex', origin: 'user', occurred_at: badgeAt });
+ assert.strictEqual(followed.item.actor_subject, alex.subject_id, 'a card\'s person, once known, never changes');
+ extraContent = [];
+ });
+
await done(t);
await done(t);
})();
})();