openvibers/openvibe.network Commits

One feed: Network reads Events and the sites' APIs into feed.item@1 (#869)

OP OpenVibers committed 20 h ago · 4b0ad73f83 · 16 files, +1866 −17

One feed for all of OpenVibe: Network reads public Events and the sites' public APIs into feed.item@1; the menu and Community's Pulse read the same rows. Contracts v0.415.0.

Changes

ddiff --git a/migrations/0080_feed.sql b/migrations/0080_feed.sql
nnew file mode 100644
iindex 00000000..7e096248
---- /dev/null
++++ b/migrations/0080_feed.sql
@@@ -0,0 +1,79 @@
++-- phase: expand
++-- One feed for all of OpenVibe (feed.item@1): the items Network gathers from OpenVibe.Events and the
++-- sites' public APIs, the state of each collector, and the daily online peak the Report card reads.
++--
++-- Expand only: an older release never reads these tables, and nothing here changes an existing one.
++-- The feed keeps only public or follower-visible activity; actor_subject is the person the item is
++-- about (null for AI, system and aggregate rows), never a copy of their profile.
++CREATE TABLE IF NOT EXISTS feed_items (
++ id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
++ source_service text COLLATE "C" NOT NULL,
++ source_type text COLLATE "C" NOT NULL,
++ source_id text COLLATE "C" NOT NULL,
++ kind text COLLATE "C" NOT NULL,
++ title text COLLATE "C" NOT NULL,
++ text text COLLATE "C",
++ url text COLLATE "C" NOT NULL,
++ site text COLLATE "C" NOT NULL,
++ site_name text COLLATE "C" NOT NULL,
++ actor_subject text COLLATE "C",
++ actor_name text COLLATE "C",
++ origin text COLLATE "C" NOT NULL DEFAULT 'user' CHECK (origin IN ('user', 'ai', 'system')),
++ visibility text COLLATE "C" NOT NULL DEFAULT 'public' CHECK (visibility IN ('public', 'followers')),
++ picture_url text COLLATE "C",
++ preview_url text COLLATE "C",
++ media_ratio double precision,
++ media_length bigint,
++ counts jsonb NOT NULL DEFAULT '{}'::jsonb,
++ live_now boolean NOT NULL DEFAULT false,
++ trend jsonb,
++ label text COLLATE "C",
++ starts_at timestamptz,
++ occurred_at timestamptz NOT NULL DEFAULT now(),
++ created_at timestamptz NOT NULL DEFAULT now(),
++ updated_at timestamptz NOT NULL DEFAULT now(),
++ UNIQUE (source_service, source_type, source_id)
++);
++CREATE INDEX IF NOT EXISTS feed_items_recent ON feed_items (occurred_at DESC, id DESC);
++CREATE INDEX IF NOT EXISTS feed_items_actor ON feed_items (actor_subject, occurred_at DESC);
++CREATE INDEX IF NOT EXISTS feed_items_kind ON feed_items (kind, occurred_at DESC);
++
++-- One row per collector: when it ran and when it is due next, so a restart does not skip a source and
++-- two processes never both fetch one (the second finds it not due).
++CREATE TABLE IF NOT EXISTS feed_source_runs (
++ name text COLLATE "C" PRIMARY KEY,
++ next_run_at timestamptz NOT NULL,
++ last_run_at timestamptz NOT NULL,
++ last_error text COLLATE "C"
++);
++
++-- Small key/value state: the Events pull cursor, the last online peak sampled, the last report card.
++CREATE TABLE IF NOT EXISTS feed_state (
++ key text COLLATE "C" PRIMARY KEY,
++ value jsonb NOT NULL,
++ updated_at timestamptz NOT NULL DEFAULT now()
++);
++
++-- The network's online peak per UTC day (members seen in the last thirty minutes plus distinct guests,
++-- server/frame/seen.js). The Report card says the day's peak and the hour it happened.
++CREATE TABLE IF NOT EXISTS feed_peaks (
++ day date PRIMARY KEY,
++ peak integer NOT NULL CHECK (peak >= 0),
++ peak_at timestamptz NOT NULL,
++ updated_at timestamptz NOT NULL DEFAULT now()
++);
++
++-- A source that names a channel (Live) or a forum author by handle: the one person Network knows by it,
++-- so the Following filter and the "watched together" summary can name them. Never more than one match.
++CREATE TABLE IF NOT EXISTS feed_channel_people (
++ handle text COLLATE "C" PRIMARY KEY,
++ subject_id text COLLATE "C" NOT NULL,
++ updated_at timestamptz NOT NULL DEFAULT now()
++);
++
++-- When a person's Quest badges were last checked (the public badges API is the only read Quest offers,
++-- so the feed checks people recently active in it, at most once a day each).
++CREATE TABLE IF NOT EXISTS feed_quest_people (
++ subject text COLLATE "C" PRIMARY KEY,
++ checked_at timestamptz NOT NULL DEFAULT now()
++);
ddiff --git a/package-lock.json b/package-lock.json
iindex 2b17bc90..44cc3f7c 100644
---- a/package-lock.json
++++ b/package-lock.json
@@@ -22,7 +22,7 @@
"iovalkey": "^0.4.0",
"jsonwebtoken": "^9.0.0",
"multer": "^1.4.5-lts.1",
-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.410.0",
++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.415.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.152.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.223.0",
"otpauth": "9.6.0",
@@@ -2679,9 +2679,9 @@
}
},
"node_modules/openvibe-contracts": {
-- "version": "0.410.0",
-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.410.0",
-- "integrity": "sha512-rmmq2mebXim2fv+sip2YQ0PrrZLxGUhEfe1w+1EFfGdoVvLI5tp0/eavY2sW6/VU+rzs7cQMC/A0NZD5th5wrA==",
++ "version": "0.415.0",
++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.415.0",
++ "integrity": "sha512-1fAn8klkUvcR1d5mzrLUD9nt4xBMl6a5B/tD/H+QvpFXzurpNOWwxZ5ayrEFmFLo03IjhdqBqZRyJUAqnZRcfw==",
"license": "AGPL-3.0-only",
"dependencies": {
"ajv": "^8.17.1",
ddiff --git a/package.json b/package.json
iindex b71c58f4..fba88bee 100644
---- a/package.json
++++ b/package.json
@@@ -26,10 +26,9 @@
"iovalkey": "^0.4.0",
"jsonwebtoken": "^9.0.0",
"multer": "^1.4.5-lts.1",
-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.410.0",
++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.415.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.152.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.223.0",
-- "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.223.0",
"otpauth": "9.6.0",
"pg": "^8.23.0",
"qrcode": "1.5.4",
ddiff --git a/server/config.js b/server/config.js
iindex b6bed3e5..7897010c 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -84,6 +84,26 @@ module.exports = {
apiKeys: process.env.NETWORK_DEV_API_KEYS === '1',
},
++ // One feed for all of OpenVibe (server/feed/): the public APIs the collectors read when a source has
++ // no Events topic of its own. Every URL is operator config, never a request value; the env name lets
++ // production point a collector at the service's loopback port without a release.
++ feed: {
++ communityUrl: process.env.OV_COMMUNITY_URL || '',
++ todayUrl: process.env.OV_TODAY_URL || 'https://openvibe.today',
++ questUrl: process.env.OV_QUEST_URL || 'https://openvibe.quest',
++ playUrl: process.env.OV_PLAY_URL || 'https://openvibe.games',
++ memeUrl: process.env.OV_MEME_URL || 'https://openvibe.meme',
++ picsUrl: process.env.OV_PICS_URL || 'https://openvibe.pics',
++ liveUrl: process.env.OV_LIVE_URL || '',
++ spaceUrl: process.env.OV_SPACE_URL || 'https://openvibe.space',
++ workUrl: process.env.OV_WORK_URL || 'https://openvibe.work',
++ dealsUrl: process.env.OV_DEALS_URL || 'https://openvibe.deals',
++ newsUrl: process.env.OV_NEWS_URL || 'https://openvibe.news',
++ blogUrl: process.env.OV_BLOG_URL || 'https://openvibe.blog',
++ wikiUrl: process.env.OV_WIKI_URL || 'https://openvibe.wiki',
++ reportUrl: process.env.OV_REPORT_URL || 'https://openvibe.report',
++ },
++
// OpenVibe.Events base URL for Network's own events (developer projects). Unset = no relay:
// events stay in dev_audit and are backfilled when it is set. Production: http://127.0.0.1:4300
eventsInternalUrl: process.env.OV_EVENTS_INTERNAL_URL || '',
ddiff --git a/server/feed/api.js b/server/feed/api.js
nnew file mode 100644
iindex 00000000..785f3f10
---- /dev/null
++++ b/server/feed/api.js
@@@ -0,0 +1,150 @@
++'use strict';
++
++/**
++ * GET /api/v1/feed — one feed for all of OpenVibe (feed.item@1).
++ *
++ * GET /api/v1/feed?kind=&site=&q=&following=1&cursor=&limit=
++ * GET /api/v1/feed/search?q=&cursor=&limit=
++ * GET /api/v1/feed/examples
++ * GET /api/v1/feed/wall?limit=&cursor= the menu's older wall, the same rows and ids
++ *
++ * Signed out returns public items only; signed in reads the person's follows (Following) and ranks
++ * For you by follows, what they open (Network's cross-site history) and freshness. A private or
++ * follower-only item never reaches a signed-out answer, and never anyone but its author or a follower.
++ * The menu asks the same handler through /api/frame/feed (server/frame/service.js); Community's
++ * /api/v1/feed proxies it, so the menu and Pulse show the same ids.
++ */
++const express = require('express');
++const store = require('./store');
++const { cardOf, foldCards, rankCards, searchMatches, searchExamples, wallItem } = require('./card');
++const { createFeedSources } = require('./sources');
++const { createEventsPull } = require('./events');
++const session = require('../auth/session');
++
++const clean = (v, n) => String(v == null ? '' : v).replace(/[\u0000-\u001f\u007f]+/g, ' ').replace(/\s+/g, ' ').trim().slice(0, n);
++const KINDS = ['post', 'pic', 'clip', 'video', 'live', 'stream', 'rundown', 'quest', 'member', 'chat', 'event', 'wiki', 'update', 'work', 'deal', 'news', 'report', 'poll', 'game', 'meme', 'paste', 'episode', 'space', 'avatar', 'banner', 'follow', 'site'];
++const SITE_RE = /^[a-z0-9][a-z0-9:-]{0,39}$/;
++const encodeCursor = ([at, id]) => Buffer.from(JSON.stringify([new Date(at).toISOString(), Number(id)])).toString('base64url');
++const decodeCursor = (v) => {
++ if (!v || typeof v !== 'string' || v.length > 400) return null;
++ try {
++ const [at, id] = JSON.parse(Buffer.from(v, 'base64url').toString('utf8'));
++ return typeof at === 'string' && Number.isSafeInteger(id) && id > 0 ? [at, id] : null;
++ } catch { return null; }
++};
++const problem = (res, status, code, detail) => res.status(status).json({ type: `https://openvibe.network/problems/${code}`, title: status === 401 ? 'Unauthorized' : 'Bad request', status, code, detail });
++
++/** The person asking, when they carry a live session (Bearer or page cookie); never a service token. */
++async function viewerOf(req, { db, publicKey, config }) {
++ try {
++ const token = session.sessionToken(req);
++ if (!token) return null;
++ const v = await session.verifySession(token, { db, publicKey, config });
++ if (!v || !v.user) return null;
++ return { id: v.user.id, subject: v.user.subject_id || null };
++ } catch { return null; }
++}
++
++async function actorsFor(db, rows) {
++ const subjects = [...new Set(rows.map((r) => r.actor_subject).filter((s) => /^usr_[0-9A-HJKMNP-TV-Z]{26}$/.test(String(s || ''))))];
++ if (!subjects.length) return new Map();
++ const users = await db.prepare(`SELECT subject_id, display_name, username, avatar_url FROM users WHERE subject_id IN (${subjects.map(() => '?').join(', ')})`).all(...subjects);
++ return new Map(users.map((u) => [u.subject_id, { name: u.display_name || u.username, avatar: u.avatar_url || null }]));
++}
++
++function createFeedApi({ db, config = {}, privateKey = null, issuer = null, publicKey = '', fetchImpl = globalThis.fetch, now = () => Date.now(), log = console, presence = null, startCollectors = true } = {}) {
++ const sources = createFeedSources({ db, config, fetchImpl, now, log, presence });
++ const events = createEventsPull({ db, eventsUrl: config.eventsInternalUrl, privateKey, issuer, fetchImpl, log, now });
++ let started = false;
++
++ const headers = (req, res) => {
++ res.set({
++ 'Access-Control-Allow-Origin': '*',
++ 'Access-Control-Allow-Headers': 'Authorization',
++ Vary: 'Authorization',
++ 'Content-Type': 'application/json; charset=utf-8',
++ 'Cache-Control': req.get('authorization') ? 'private, no-store' : 'public, max-age=30',
++ });
++ };
++
++ async function page(req, { wall = false } = {}) {
++ const kind = KINDS.includes(String(req.query.kind || '')) ? String(req.query.kind) : '';
++ const site = SITE_RE.test(String(req.query.site || '')) ? String(req.query.site) : '';
++ const q = clean(req.query.q, 80);
++ const following = /^(1|true|yes|on)$/i.test(String(req.query.following || ''));
++ const want = Math.min(Math.max(parseInt(req.query.limit, 10) || 20, 1), 40);
++ const viewer = await viewerOf(req, { db, publicKey, config });
++ if (following && !viewer) return { error: 'sign_in' };
++ const followed = viewer && viewer.subject ? await store.followedSubjects(db, viewer.subject) : [];
++ const affinity = viewer ? await store.affinityServices(db, viewer.id) : new Map();
++ const before = decodeCursor(String(req.query.cursor || ''));
++
++ const read = await store.listItems(db, {
++ kind, site, q, followed, following, before, viewerSubject: viewer && viewer.subject,
++ limit: Math.min(400, want * 6 + 40), since: new Date(now() - 60 * 86400000).toISOString(),
++ });
++ const actors = await actorsFor(db, read.rows);
++ let cards = read.rows.map((r) => cardOf(r, actors.get(r.actor_subject) || null));
++ if (q) cards = cards.filter((c) => searchMatches(c, q));
++ cards = foldCards(cards, now());
++ cards = rankCards(cards, { followed: new Set(followed), affinity, now: now() });
++ const items = cards.slice(0, want);
++ const last = read.rows[read.rows.length - 1];
++ const next = read.hasMore && last ? encodeCursor([last.occurred_at, last.id]) : null;
++ if (wall) return { items: items.map(wallItem), next_cursor: next };
++ return { items, next_cursor: next };
++ }
++
++ async function apiPage(req, res, opts) {
++ headers(req, res);
++ const out = await page(req, opts);
++ if (out.error === 'sign_in') return problem(res, 401, 'feed.sign_in_required', 'Sign in to see the people you follow');
++ return res.json(out);
++ }
++
++ const handleFeed = (req, res, next) => apiPage(req, res).catch(next);
++ const handleWall = (req, res, next) => apiPage(req, res, { wall: true }).catch(next);
++ const handleExamples = async (req, res, next) => {
++ try {
++ headers(req, res);
++ const read = await store.listItems(db, { limit: 100, since: new Date(now() - 7 * 86400000).toISOString() });
++ const actors = await actorsFor(db, read.rows);
++ res.json({ examples: searchExamples(read.rows.map((r) => cardOf(r, actors.get(r.actor_subject) || null))) });
++ } catch (err) { next(err); }
++ };
++ const handleSearch = async (req, res, next) => {
++ try {
++ headers(req, res);
++ const q = clean(req.query.q, 80);
++ if (!q) return problem(res, 400, 'feed.search_required', 'Enter a search term');
++ const want = Math.min(Math.max(parseInt(req.query.limit, 10) || 20, 1), 40);
++ const viewer = await viewerOf(req, { db, publicKey, config });
++ const followed = viewer && viewer.subject ? await store.followedSubjects(db, viewer.subject) : [];
++ const read = await store.listItems(db, { q, followed, viewerSubject: viewer && viewer.subject, limit: 400, since: new Date(now() - 365 * 86400000).toISOString() });
++ const actors = await actorsFor(db, read.rows);
++ const cards = foldCards(read.rows.map((r) => cardOf(r, actors.get(r.actor_subject) || null)).filter((c) => searchMatches(c, q)), now());
++ res.json({ items: rankCards(cards, { followed: new Set(followed), now: now() }).slice(0, want), next_cursor: null });
++ } catch (err) { next(err); }
++ };
++ const router = express.Router();
++ router.get('/', handleFeed);
++ router.get('/wall', handleWall);
++ router.get('/examples', handleExamples);
++ router.get('/search', handleSearch);
++ router.options('*', (req, res) => {
++ res.set({ 'Access-Control-Allow-Origin': '*', 'Access-Control-Allow-Headers': 'Authorization, Content-Type', 'Access-Control-Allow-Methods': 'GET, OPTIONS', 'Access-Control-Max-Age': '600' });
++ res.status(204).end();
++ });
++
++ function start() {
++ if (started || !startCollectors) return;
++ started = true;
++ sources.start();
++ events.start();
++ }
++ function stop() { sources.stop(); events.stop(); }
++
++ return { router, page, handleFeed, handleWall, handleSearch, handleExamples, start, stop, sources, events, viewerOf };
++}
++
++module.exports = { createFeedApi, viewerOf, actorsFor, encodeCursor, decodeCursor, KINDS };
ddiff --git a/server/feed/card.js b/server/feed/card.js
nnew file mode 100644
iindex 00000000..9e3b46e9
---- /dev/null
++++ b/server/feed/card.js
@@@ -0,0 +1,257 @@
++'use strict';
++
++/**
++ * A feed row as feed.item@1, and the page-level operations the feed API applies after reading it:
++ * kind mapping, folding a run of the same kind, and the For you / Following ranking.
++ *
++ * The ranking is the owner's ask: live first, then what the person follows and opens, then
++ * freshness and engagement, with anything older than seven days below the last week's items (the
++ * same order Community's feed proved, 11-11). No field here is private: a row only reaches this
++ * module after store.listItems() applied the visibility and test-account rules.
++ */
++
++const SITE_NAMES = {
++ live: 'Live', 'media-hub': 'Video', media: 'Media', pics: 'Pics', video: 'Video', community: 'Community', blog: 'Blog',
++ wiki: 'Wiki', quest: 'Quest', chat: 'Chat', tools: 'Tools', news: 'News', report: 'Report', today: 'Today',
++ network: 'Network', space: 'Space', meme: 'Meme', games: 'Games', play: 'Games', work: 'Work', deals: 'Deals',
++ rundown: 'Live', stream: 'Live', exchange: 'Exchange', blog2: 'Blog',
++};
++
++const clean = (v, n) => String(v == null ? '' : v).replace(/\s+/g, ' ').trim().slice(0, n);
++const https = (u) => {
++ try { const x = new URL(String(u)); return x.protocol === 'https:' ? x.href : (x.protocol === 'http:' ? x.href : null); } catch { return null; }
++};
++
++/** What a source row is, as a feed kind. An unknown type is a post, never dropped silently. */
++function kindOf(service, type, row = {}) {
++ const s = String(service || '');
++ const t = String(type || '');
++ if (s === 'live') {
++ if (/^(episode|rundown)$/.test(t)) return 'rundown';
++ if (/^(recording|vod|video)$/.test(t)) return 'video';
++ if (/^(clip|moment)$/.test(t)) return 'clip';
++ if (/^(stream|went_live|broadcast|live)$/.test(t)) return 'live';
++ }
++ if (s === 'media-hub' || s === 'media') return /picture|image|photo/.test(t) ? 'pic' : 'video';
++ if (s === 'pics') return 'pic';
++ if (s === 'games' || s === 'play') return 'game';
++ if (s === 'quest') return 'quest';
++ if (s === 'chat') return 'chat';
++ if (s === 'today') return 'event';
++ if (s === 'community') {
++ if (t === 'poll') return 'poll';
++ if (t === 'thread' || t === 'status' || t === 'post' || t === 'submission') return 'post';
++ if (t === 'paste') return row.picture_url || row.kind === 'pic' || row.media_ratio ? 'pic' : 'paste';
++ return 'post';
++ }
++ if (s === 'blog') return 'update';
++ if (s === 'wiki') return 'wiki';
++ if (s === 'news') return 'news';
++ if (s === 'space') return 'space';
++ if (s === 'work') return 'work';
++ if (s === 'deals') return 'deal';
++ if (s === 'report') return 'report';
++ if (s === 'meme') return 'meme';
++ if (s === 'network') {
++ if (/member|join/.test(t)) return 'member';
++ if (t === 'avatar') return 'avatar';
++ if (t === 'banner') return 'banner';
++ if (t === 'follow') return 'follow';
++ if (t === 'site') return 'site';
++ }
++ if (s === 'rundown') return 'rundown';
++ if (s === 'stream') return 'stream';
++ return 'post';
++}
++
++/** One row as a card. `actor` is the resolved person { name, avatar } when the row names one. */
++function cardOf(row, actor = null) {
++ const counts = row.counts && typeof row.counts === 'object' ? row.counts : {};
++ const safe = {};
++ for (const k of ['views', 'likes', 'comments', 'watchers', 'people']) {
++ const n = Number(counts[k]);
++ if (Number.isSafeInteger(n) && n >= 0) safe[k] = n;
++ }
++ const ratio = Number(row.media_ratio);
++ const length = Number(row.media_length);
++ const card = {
++ id: `f:${row.source_service}:${row.source_type}:${row.source_id}`,
++ kind: row.kind,
++ url: https(row.url) || String(row.url || ''),
++ title: clean(row.title, 300),
++ text: clean(row.text, 280),
++ at: row.occurred_at ? new Date(row.occurred_at).toISOString() : null,
++ site: String(row.site || row.source_service || ''),
++ site_name: String(row.site_name || SITE_NAMES[row.site] || SITE_NAMES[row.source_service] || row.site || ''),
++ actor: actor ? { name: clean(actor.name, 80) || null, avatar: https(actor.avatar) } : (row.actor_name ? { name: clean(row.actor_name, 80) || null, avatar: null } : null),
++ subject: row.actor_subject || null,
++ media: row.picture_url && https(row.picture_url)
++ ? { thumb: https(row.picture_url), ...(row.preview_url && https(row.preview_url) ? { preview: https(row.preview_url) } : {}), ratio: Number.isFinite(ratio) && ratio > 0 ? ratio : 16 / 9, ...(Number.isFinite(length) && length > 0 ? { length } : {}) }
++ : null,
++ counts: safe,
++ };
++ if (row.starts_at) card.starts_at = new Date(row.starts_at).toISOString();
++ if (row.label) card.label = row.label;
++ if (row.visibility === 'followers') card.visibility = 'followers';
++ if (card.kind === 'live' || card.kind === 'stream') {
++ card.live_now = row.kind === 'live' && row.live_now === true;
++ if (!card.live_now) card.kind = 'stream';
++ if (card.live_now) card.label = 'Live';
++ }
++ if (card.kind === 'report' && row.trend && typeof row.trend === 'object') {
++ const direction = ['up', 'down', 'flat'].includes(row.trend.direction) ? row.trend.direction : null;
++ if (direction) {
++ card.trend = { direction };
++ if (Number.isFinite(Number(row.trend.pct)) && Number(row.trend.pct) >= 0) card.trend.pct = Number(row.trend.pct);
++ if (Number.isSafeInteger(Number(row.trend.previous)) && Number(row.trend.previous) >= 0) card.trend.previous = Number(row.trend.previous);
++ }
++ }
++ return card;
++}
++
++const BURST_MS = 15 * 60_000;
++const BURST_WORD = { pic: ['photo', 'photos'], clip: ['clip', 'clips'], video: ['video', 'videos'], game: ['achievement', 'achievements'], quest: ['quest', 'quests'], avatar: ['picture change', 'picture changes'], banner: ['banner change', 'banner changes'], paste: ['paste', 'pastes'], space: ['space', 'spaces'] };
++
++function titleKey(card) {
++ return String(card.title || '').toLowerCase().replace(/\s+/g, ' ').trim();
++}
++
++/**
++ * Many of the same kind in a short time become one card: a person's ten recordings in one evening,
++ * their four pictures in ten minutes. Work is grouped per UTC hour (the owner's ask: new jobs as one
++ * grouped item per hour); Report is one card per day and is already one row.
++ */
++function foldCards(cards, now = Date.now()) {
++ const groups = [];
++ for (const card of cards) {
++ if (card.kind === 'work') {
++ const hour = String(card.at || '').slice(0, 13);
++ const prior = groups.find((g) => g.workHour === hour);
++ if (prior) { prior.cards.push(card); continue; }
++ groups.push({ workHour: hour, card, cards: [card] });
++ continue;
++ }
++ const at = Date.parse(card.at || '');
++ // A picture source may name only its author, not a subject (OpenVibe.Pics): group by that name too.
++ const who = card.subject || (card.actor && card.actor.name) || null;
++ const prior = BURST_WORD[card.kind] && who && Number.isFinite(at) && groups.find((g) => g.subject != null
++ && g.subject === who && g.site === card.site && g.card.kind === card.kind && Math.abs(g.at - at) <= BURST_MS);
++ if (prior) { prior.cards.push(card); continue; }
++ groups.push({ subject: who, at, site: card.site, card, cards: [card] });
++ }
++ return groups.map((g) => {
++ if (g.workHour !== undefined) {
++ if (g.cards.length < 2) return g.card;
++ const sites = new Set(g.cards.map((c) => c.site_name).filter(Boolean));
++ return {
++ ...g.card,
++ id: `g:work:${g.workHour}`,
++ title: `${g.cards.length} new jobs this hour`,
++ text: clean(g.cards.map((c) => c.title).slice(0, 3).join(' · '), 280),
++ group: { count: g.cards.length, items: g.cards.slice(0, 20).map((c) => ({ id: c.id, url: c.url, title: c.title, media: c.media })) },
++ counts: { ...g.card.counts, people: g.cards.length },
++ site: g.card.site,
++ site_name: sites.size === 1 ? [...sites][0] : g.card.site_name,
++ };
++ }
++ if (g.cards.length < 2) return g.card;
++ const [one, many] = BURST_WORD[g.card.kind];
++ const name = (g.card.actor && g.card.actor.name) || 'Someone';
++ return {
++ ...g.card,
++ title: `${name} added ${g.cards.length} ${many}`,
++ text: '',
++ group: { count: g.cards.length, items: g.cards.slice(0, 20).map((c) => ({ id: c.id, url: c.url, title: c.title, media: c.media })) },
++ counts: { ...g.card.counts, people: g.cards.length },
++ ...(one ? {} : {}),
++ };
++ });
++}
++
++const DAY_MS = 24 * 60 * 60_000;
++const engagement = (card) => {
++ const c = card.counts || {};
++ return Math.log1p(Math.max(0, Number(c.likes) || 0) + 2 * Math.max(0, Number(c.comments) || 0)
++ + 0.2 * Math.max(0, Number(c.views) || 0) + 0.1 * Math.max(0, Number(c.watchers) || 0)) * 2;
++};
++
++/** The order the menu and the Pulse page show: live, the last week's best for you, then older. */
++function compareRankedCards(a, b, { followed = new Set(), affinity = new Map(), now = Date.now() } = {}) {
++ const liveOrder = Number(b.kind === 'live' && b.live_now === true) - Number(a.kind === 'live' && a.live_now === true);
++ if (liveOrder) return liveOrder;
++ const ageDays = (card) => {
++ const at = Date.parse(card.at || '');
++ return Number.isFinite(at) ? Math.max(0, (now - at) / DAY_MS) : Infinity;
++ };
++ const aAge = ageDays(a), bAge = ageDays(b);
++ const staleOrder = Number(aAge > 7) - Number(bAge > 7);
++ if (staleOrder) return staleOrder;
++ const score = (card, age) => (Number.isFinite(age) ? Math.max(0, 7 - age) : 0)
++ + engagement(card)
++ + (card.subject && followed.has(card.subject) ? 4 : 0)
++ + (card.site ? Math.min(2, 0.2 * (affinity.get(card.site) || 0)) : 0);
++ const scoreOrder = score(b, bAge) - score(a, aAge);
++ if (scoreOrder) return scoreOrder;
++ const timeOrder = (Date.parse(b.at || '') || 0) - (Date.parse(a.at || '') || 0);
++ return timeOrder || String(a.id).localeCompare(String(b.id));
++}
++
++function rankCards(cards, options) {
++ return cards.slice().sort((a, b) => compareRankedCards(a, b, options));
++}
++
++const ALIASES = {
++ clip: 'clip', clips: 'clip', video: 'video', videos: 'video', recording: 'video', recordings: 'video',
++ photo: 'pic', photos: 'pic', pic: 'pic', pics: 'pic', picture: 'pic', pictures: 'pic', post: 'post', posts: 'post',
++ paste: 'paste', pastes: 'paste', stream: 'stream', streams: 'stream', live: 'live', game: 'game', games: 'game',
++ update: 'update', updates: 'update', wiki: 'wiki', quest: 'quest', quests: 'quest', chat: 'chat', event: 'event',
++ events: 'event', work: 'work', jobs: 'work', deal: 'deal', deals: 'deal', news: 'news', report: 'report',
++ rundown: 'rundown', member: 'member', members: 'member', meme: 'meme', memes: 'meme', poll: 'poll', polls: 'poll',
++ episode: 'episode', episodes: 'episode', space: 'space', spaces: 'space',
++};
++
++/** Search words, kind words and dates ("new clips from today") over one card. */
++function searchMatches(card, query) {
++ const q = clean(query, 80).toLowerCase();
++ if (!q) return true;
++ const kindWord = (q.match(/\b[a-z]+\b/g) || []).find((word) => ALIASES[word]);
++ const kind = kindWord && ALIASES[kindWord];
++ if (kind && card.kind !== kind && !(kind === 'stream' && card.kind === 'live')) return false;
++ const day = String(card.at || '').slice(0, 10);
++ if (/\btoday\b/.test(q) && day !== new Date().toISOString().slice(0, 10)) return false;
++ if (/\byesterday\b/.test(q) && day !== new Date(Date.now() - 86_400_000).toISOString().slice(0, 10)) return false;
++ const date = /\b\d{4}-\d{2}-\d{2}\b/.exec(q);
++ if (date && day !== date[0]) return false;
++ const words = q.replace(/\b(clips?|videos?|recordings?|photos?|pics?|pictures?|posts?|pastes?|streams?|live|games?|updates?|jobs?|deals?|wikis?|quests?|polls?|episodes?|spaces?|from|today|yesterday|new)\b/g, '')
++ .replace(/\b\d{4}-\d{2}-\d{2}\b/g, '').trim();
++ return !words || [card.title, card.text, card.site, card.site_name, card.kind, card.actor && card.actor.name]
++ .some((v) => String(v || '').toLowerCase().includes(words));
++}
++
++/** The "Try" words under the search box. */
++function searchExamples(cards) {
++ const examples = [];
++ const add = (text, reason) => { if (text && !examples.some((x) => x.text === text)) examples.push({ text, reason }); };
++ for (const c of cards) if (c.actor && c.actor.name) add(c.actor.name, 'person');
++ for (const kind of new Set(cards.map((c) => c.kind))) add(`new ${kind === 'pic' ? 'photos' : `${kind}s`}`, 'kind');
++ if (cards.some((c) => String(c.at || '').slice(0, 10) === new Date().toISOString().slice(0, 10))) add('from today', 'date');
++ return examples.slice(0, 8);
++}
++
++/** A feed card as the menu's older wall item (GET /api/v1/pulse shape): the same ids, another drawing. */
++function wallItem(card) {
++ return {
++ id: card.id,
++ source: { service: card.site, type: card.kind, id: card.id },
++ title: card.title,
++ text: card.text,
++ url: card.url,
++ picture_url: card.media ? card.media.thumb : null,
++ origin: card.label === 'AI' ? 'ai' : (card.subject ? 'user' : 'system'),
++ label: card.label || null,
++ actor: card.actor ? { display_name: card.actor.name, avatar_url: card.actor.avatar } : null,
++ occurred_at: card.at,
++ };
++}
++
++module.exports = { SITE_NAMES, kindOf, cardOf, foldCards, compareRankedCards, rankCards, searchMatches, searchExamples, wallItem, titleKey, clean, https };
ddiff --git a/server/feed/events.js b/server/feed/events.js
nnew file mode 100644
iindex 00000000..5dbdb8b8
---- /dev/null
++++ b/server/feed/events.js
@@@ -0,0 +1,261 @@
++'use strict';
++
++/**
++ * OpenVibe.Events → one feed.
++ *
++ * The feed reads events, not each site: a public event that says something happened becomes one row
++ * (migrations/0080_feed.sql), in the same item shape every other source uses. Pull, not a webhook:
++ * Network already signs its own service tokens for Events (server/developer/event-relay.js), so the
++ * feed needs no operator step to start, survives a restart with its cursor in feed_state, and starts
++ * at the head on a fresh database (the public API collectors fill the look-back).
++ *
++ * Only public envelopes enter the feed; a `subject` or `internal` event never does. A topic that
++ * names a private or unlisted thing carries an internal visibility, so filtering here also drops it.
++ * The mapper is pure (eventToItems) so the recorded-input tests cover every kind.
++ */
++const crypto = require('crypto');
++const { serviceAuth, assertValid } = require('openvibe-contracts');
++const store = require('./store');
++
++const CURSOR_KEY = 'events_cursor';
++const TOKEN_TTL_S = 300;
++const PAGE = 200;
++const MAX_PAGES = 20;
++
++const clean = (v, n) => String(v == null ? '' : v).replace(/\s+/g, ' ').trim().slice(0, n);
++const https = (v) => {
++ try {
++ const u = new URL(String(v));
++ return u.protocol === 'https:' ? u.href.slice(0, 2000) : null;
++ } catch { return null; }
++};
++const userOf = (ref) => (ref && ref.type === 'user' && /^usr_[0-9A-HJKMNP-TV-Z]{26}$/.test(String(ref.id)) ? String(ref.id) : null);
++const idOf = (event, payload, ...keys) => {
++ for (const k of keys) {
++ const v = payload && payload[k];
++ if (typeof v === 'string' && /^[A-Za-z0-9_.:-]{1,120}$/.test(v)) return v;
++ if (Number.isSafeInteger(v) && v >= 0) return String(v);
++ }
++ return String((event.subject && event.subject.id) || event.event_id);
++};
++const urlOf = (payload) => https(payload && (payload.canonical_url || payload.url || payload.href || (payload.media && payload.media.page)));
++const mediaOf = (payload) => {
++ const p = payload || {};
++ const m = p.media && typeof p.media === 'object' ? p.media : p;
++ const thumb = https(m.thumb || m.thumbnail_url || m.poster_url || m.picture_url);
++ if (!thumb) return null;
++ const w = Number(p.width || m.width), h = Number(p.height || m.height);
++ return { thumb, ratio: Number.isFinite(w) && Number.isFinite(h) && w > 0 && h > 0 ? w / h : 16 / 9,
++ ...(Number(m.duration_seconds || p.duration_seconds) > 0 ? { length: Number(m.duration_seconds || p.duration_seconds) } : {}) };
++};
++const originOf = (event, subject) => (event.actor && event.actor.type === 'ai' ? 'ai' : (subject ? 'user' : 'system'));
++const countsOf = (payload, extra = {}) => {
++ const c = payload && (payload.counts || payload.stats) || {};
++ const out = { ...extra };
++ for (const k of ['views', 'likes', 'comments', 'watchers', 'people']) {
++ const n = Number(c[k]);
++ if (Number.isSafeInteger(n) && n >= 0) out[k] = n;
++ }
++ return out;
++};
++
++const SITE = { live: 'Live', 'media-hub': 'Video', media: 'Media', games: 'Games', blog: 'Blog', wiki: 'Wiki', news: 'News', chat: 'Chat', deals: 'Deals', space: 'Space', rundown: 'Live', today: 'Today' };
++
++/** One public envelope → { items, remove } (never anything for an internal or subject event). */
++function eventToItems(event) {
++ if (!event || typeof event !== 'object' || event.visibility !== 'public') return { items: [], remove: [] };
++ const p = event.payload && typeof event.payload === 'object' ? event.payload : {};
++ const at = store.iso(event.timestamp) || new Date().toISOString();
++ const actor = event.actor && event.actor.type === 'user' ? userOf(event.actor) : null;
++ const subject = userOf(event.subject) || actor;
++ const source = String(event.source || '');
++ const type = String(event.event_type || '');
++ const item = (over) => ({
++ source_service: over.service || source, source_type: over.type, source_id: over.id, kind: over.kind,
++ title: clean(over.title, 300), text: clean(over.text, 280), url: over.url,
++ site: over.service || source, site_name: SITE[over.service || source] || over.service || source,
++ actor_subject: over.subject === undefined ? subject : over.subject,
++ actor_name: null, origin: over.origin || originOf(event, subject), visibility: 'public',
++ picture_url: over.media && over.media.thumb, preview_url: over.media && over.media.preview,
++ media_ratio: over.media && over.media.ratio, media_length: over.media && over.media.length,
++ counts: over.counts || {}, live_now: over.live_now === true, label: over.label || null,
++ starts_at: over.starts_at || null, occurred_at: over.at || at, touch: over.touch === true,
++ });
++
++ if (type === 'live.stream.started') {
++ const channel = p.channel && typeof p.channel === 'object' ? p.channel : {};
++ const url = https(channel.url) || (channel.username ? `https://openvibe.live/@${encodeURIComponent(channel.username)}` : null);
++ if (!url) return { items: [], remove: [] };
++ const who = userOf(channel.subject) || actor;
++ const name = clean(channel.display_name || channel.username, 80) || 'A channel';
++ return { items: [item({ service: 'live', type: 'stream', id: idOf(event, p, 'stream_id') || String(channel.username || 'live'), kind: 'live',
++ title: p.title ? `${name} went live: ${clean(p.title, 160)}` : `${name} went live`, url,
++ subject: who, at: p.started_at || at, live_now: true })], remove: [] };
++ }
++ if (type === 'mediahub.video.published') {
++ const url = urlOf(p);
++ if (!url) return { items: [], remove: [] };
++ return { items: [item({ service: 'media-hub', type: 'video', id: idOf(event, p, 'video_id'), kind: 'video',
++ title: clean(p.title, 160) || 'A video', text: p.description || p.excerpt, url, media: mediaOf(p), counts: countsOf(p) })], remove: [] };
++ }
++ if (type === 'mediahub.picture.uploaded') {
++ const url = urlOf(p);
++ if (!url) return { items: [], remove: [] };
++ return { items: [item({ service: 'media-hub', type: 'picture', id: idOf(event, p, 'picture_id'), kind: 'pic',
++ title: clean(p.title || p.caption, 160) || 'A picture', url, media: mediaOf(p), counts: countsOf(p) })], remove: [] };
++ }
++ if (type === 'games.world.published') {
++ const url = urlOf(p);
++ if (!url) return { items: [], remove: [] };
++ return { items: [item({ service: 'games', type: 'world', id: idOf(event, p, 'world_id'), kind: 'game',
++ title: clean(p.title, 160) || 'A new world', text: p.description, url, media: mediaOf(p) })], remove: [] };
++ }
++ if (type === 'games.skill.leveled') {
++ const url = https(p.url) || 'https://openvibe.games/';
++ const level = Number(p.level);
++ return { items: [item({ service: 'games', type: 'level', id: `${subject || idOf(event, p, 'game')}:${Number.isSafeInteger(level) ? level : idOf(event, p, 'level')}`, kind: 'game',
++ title: `Reached level ${Number.isSafeInteger(level) ? level : '?'} in ${clean(p.game_name || p.world || 'a game', 80)}`, url, at })], remove: [] };
++ }
++ if (type === 'blog.post.published' || type === 'wiki.page.published' || type === 'wiki.page.updated' || type === 'news.story.published' || type === 'news.story.updated') {
++ const url = urlOf(p);
++ if (!url) return { items: [], remove: [] };
++ const service = type.startsWith('blog') ? 'blog' : type.startsWith('wiki') ? 'wiki' : 'news';
++ const kind = service === 'blog' ? 'update' : service === 'wiki' ? 'wiki' : 'news';
++ const title = service === 'blog'
++ ? (p.blog ? `New post on the ${clean(p.blog.handle || p.blog, 60)} blog` : 'New post on an OpenVibe blog')
++ : service === 'wiki' ? (clean(p.title, 160) || `Wiki: ${clean(p.space, 40)} / ${clean(p.slug, 60)}`)
++ : `News: ${clean(p.topic, 60) || clean(p.title, 120) || 'a new story'}`;
++ return { items: [item({ service, type: service === 'news' ? 'story' : 'post', id: idOf(event, p, `${service}_id`, 'page_id', 'post_id'), kind,
++ title, text: p.description || p.excerpt || p.summary, url, media: mediaOf(p), counts: countsOf(p),
++ ...(type.endsWith('.updated') ? { touch: true } : {}) })], remove: [] };
++ }
++ if (type === 'wiki.page.deleted' || type === 'wiki.page.unpublished') {
++ return { items: [], remove: [{ service: 'wiki', type: 'post', id: idOf(event, p, 'page_id') }] };
++ }
++ if (type === 'news.story.unpublished' || type === 'news.story.retracted') {
++ return { items: [], remove: [{ service: 'news', type: 'story', id: idOf(event, p, 'story_id') }] };
++ }
++ if (type === 'chat.room.message.created') {
++ const room = p.room && typeof p.room === 'object' ? p.room : {};
++ const slug = clean(room.slug, 64);
++ const when = store.iso(p.created_at) || at;
++ if (!slug) return { items: [], remove: [] };
++ const hour = String(when).slice(0, 13);
++ return { items: [item({ service: 'chat', type: 'hour', id: `${slug}:${hour}`, kind: 'chat',
++ title: `A busy hour in #${clean(room.name || slug, 60)}`, text: '1 message', url: `https://openvibe.chat/r/${encodeURIComponent(slug)}`,
++ subject: null, at: `${hour}:00:00.000Z`, counts: { comments: 1 } })], remove: [] };
++ }
++ if (type === 'deals.offer.created' || type === 'deals.offer.updated') {
++ const url = urlOf(p);
++ if (!url) return { items: [], remove: [] };
++ return { items: [item({ service: 'deals', type: 'offer', id: idOf(event, p, 'offer_id'), kind: 'deal',
++ title: clean(p.title, 160) || 'A deal', text: p.description || p.excerpt, url, media: mediaOf(p), counts: countsOf(p),
++ ...(type.endsWith('.updated') ? { touch: true } : {}) })], remove: [] };
++ }
++ if (type === 'deals.offer.expired') return { items: [], remove: [{ service: 'deals', type: 'offer', id: idOf(event, p, 'offer_id') }] };
++ if (type === 'space.deploy.activated') {
++ const url = urlOf(p);
++ if (!url) return { items: [], remove: [] };
++ return { items: [item({ service: 'space', type: 'space', id: idOf(event, p, 'deploy_id'), kind: 'space',
++ title: clean(p.name || p.title, 160) || 'A new space', text: p.description, url })], remove: [] };
++ }
++ return { items: [], remove: [] };
++}
++
++/** Apply mapped rows; a chat hour grows instead of duplicating. */
++async function apply(db, mapped) {
++ let n = 0;
++ for (const it of mapped.items) {
++ if (it.source_service === 'chat' && it.source_type === 'hour') {
++ const had = await db.prepare('SELECT counts, text FROM feed_items WHERE source_service = ? AND source_type = ? AND source_id = ?').get('chat', 'hour', it.source_id);
++ if (had) {
++ const messages = Math.max(0, Number(had.counts && had.counts.comments) || 0) + 1;
++ await db.prepare("UPDATE feed_items SET text = ?, counts = ?, updated_at = now() WHERE source_service = 'chat' AND source_type = 'hour' AND source_id = ?")
++ .run(`${messages} messages`, JSON.stringify({ comments: messages }), it.source_id);
++ continue;
++ }
++ }
++ await store.upsertItem(db, it);
++ n++;
++ }
++ for (const r of mapped.remove) await store.removeItem(db, r.service, r.type, r.id);
++ return n;
++}
++
++function tokenSource({ privateKey, issuer }) {
++ let cached = null;
++ return () => {
++ const now = Math.floor(Date.now() / 1000);
++ if (cached && cached.exp - 60 > now) return cached.token;
++ const claims = { iss: issuer, sub: 'svc:network', actor_type: 'service', aud: ['openvibe.events'],
++ cap: ['events.event.read'], iat: now, exp: now + TOKEN_TTL_S, jti: `tok_${crypto.randomBytes(12).toString('hex')}` };
++ assertValid('identity.service-token-claims@1', claims);
++ cached = { token: serviceAuth.signServiceToken(claims, privateKey), exp: claims.exp };
++ return cached.token;
++ };
++}
++
++function createEventsPull({ db, eventsUrl, privateKey, issuer, fetchImpl = globalThis.fetch, log = console, now = () => Date.now() }) {
++ const base = String(eventsUrl || '').replace(/\/+$/, '');
++ const token = privateKey && issuer ? tokenSource({ privateKey, issuer }) : null;
++ let timer = null;
++ let running = null;
++
++ async function get(params) {
++ const u = new URL('/api/v1/events', base);
++ for (const [k, v] of Object.entries(params)) if (v) u.searchParams.set(k, v);
++ const res = await fetchImpl(u.href, { headers: { Authorization: `Bearer ${token()}`, Accept: 'application/json' }, signal: AbortSignal.timeout(10000) });
++ if (!res.ok) throw new Error(`Events ${res.status}`);
++ return await res.json();
++ }
++ const saved = async () => {
++ const row = await db.prepare('SELECT value FROM feed_state WHERE key = ?').get(CURSOR_KEY);
++ return row && row.value && typeof row.value.cursor === 'string' ? row.value.cursor : null;
++ };
++ const save = async (cursor) => {
++ await db.prepare(`INSERT INTO feed_state (key, value, updated_at) VALUES (?, ?, now())
++ ON CONFLICT (key) DO UPDATE SET value = excluded.value, updated_at = now()`).run(CURSOR_KEY, JSON.stringify({ cursor, at: new Date(now()).toISOString() }));
++ };
++
++ /** Read pages after the saved cursor. A fresh feed starts at the head (the API collectors cover the look-back). */
++ async function pullOnce({ pages = MAX_PAGES } = {}) {
++ if (!base || !token) return { skipped: 'events not configured' };
++ let cursor = await saved();
++ if (!cursor) {
++ const head = await get({ limit: '1' });
++ if (typeof head.latest_cursor === 'string') { await save(head.latest_cursor); return { bootstrapped: true }; }
++ return { bootstrapped: false };
++ }
++ let created = 0, seen = 0;
++ for (let i = 0; i < pages; i++) {
++ const out = await get({ topic: '*', after: cursor, limit: String(PAGE) });
++ const events = Array.isArray(out.events) ? out.events : [];
++ for (const row of events) {
++ seen++;
++ const mapped = eventToItems(row && row.event);
++ created += await apply(db, mapped);
++ }
++ if (typeof out.next_cursor === 'string') { cursor = out.next_cursor; await save(cursor); }
++ if (!events.length || !out.next_cursor) break;
++ }
++ return { seen, created };
++ }
++
++ function start({ every = 20_000 } = {}) {
++ if (timer || !base || !token) return;
++ const tick = () => { if (!running) running = pullOnce().catch((err) => log.warn && log.warn('[Feed events]', err.message)).finally(() => { running = null; }); };
++ timer = setInterval(tick, every);
++ if (timer.unref) timer.unref();
++ const first = setTimeout(tick, 2500);
++ if (first.unref) first.unref();
++ timer.first = first;
++ }
++ function stop() {
++ if (timer) clearInterval(timer);
++ if (timer && timer.first) clearTimeout(timer.first);
++ timer = null;
++ }
++ return { pullOnce, start, stop, eventToItems, apply };
++}
++
++module.exports = { eventToItems, apply, createEventsPull, SITE };
ddiff --git a/server/feed/sources.js b/server/feed/sources.js
nnew file mode 100644
iindex 00000000..acfb54c6
---- /dev/null
++++ b/server/feed/sources.js
@@@ -0,0 +1,660 @@
++'use strict';
++
++/**
++ * The public API collectors: the activity that has no Events topic of its own, read on a due basis.
++ *
++ * The feed never calls a site while a person is reading it (server/feed/api.js reads feed_items only).
++ * These jobs fill that table:
++ * - OpenVibe.Community: GET /api/v1/activity (public, direct query, no Pulse store) → posts, pastes,
++ * accepted submissions and their counts. Community's own writes are not Events topics with words.
++ * - OpenVibe.Today: GET /api/v1/events → one event card per scheduled stream or event.
++ * - OpenVibe.Network: its own users (never HTTP) → one grouped member card per day.
++ * - OpenVibe.Quest: /api/v1/profiles/:subject/badges for people recently active in the feed.
++ * - OpenVibe.Play: /api/v1/achievements/feed → public unlock cards.
++ * - OpenVibe.Meme: /api/v1/entries → public meme entries.
++ * - OpenVibe.Pics: /api/v1/gallery → public pictures.
++ * - OpenVibe.Live: /api/streams (live now), /api/content/feed (recordings and clips),
++ * /api/rundown/streams (rundown episodes); creator watch summaries from Network's own analytics.
++ * - OpenVibe.Space: /api/v1/apps/gallery → public spaces.
++ * - OpenVibe.Work: /api/v1/jobs → new jobs (one grouped card per hour at read time).
++ * - OpenVibe.Deals: /api/v1/offers → new deals.
++ * - OpenVibe.News: /api/v1/stories → published stories.
++ * - OpenVibe.Report: /api/v1/stats → one report card per day, with Network's own online peak.
++ *
++ * Due-ness lives in feed_source_runs, so a restart does not skip a source and two processes do not
++ * both fetch one. A minute timer only checks those rows.
++ */
++const store = require('./store');
++
++const MINUTE = 60_000;
++const DAY = 24 * 60 * MINUTE;
++const QUEST_RECENT_MS = 14 * DAY;
++const QUEST_CHECK_MS = 24 * 60 * MINUTE;
++
++const clean = (v, n) => String(v == null ? '' : v).replace(/\s+/g, ' ').trim().slice(0, n);
++const https = (u) => {
++ try { const x = new URL(String(u)); return x.protocol === 'https:' ? x.href.slice(0, 2000) : null; } catch { return null; }
++};
++const subject = (v) => (/^usr_[0-9A-HJKMNP-TV-Z]{26}$/.test(String(v || '')) ? String(v) : null);
++const sqlFrom = (v) => {
++ const t = Date.parse(String(v == null ? '' : v));
++ return Number.isFinite(t) ? new Date(t).toISOString() : null;
++};
++const siteName = (id) => ({ live: 'Live', today: 'Today', network: 'Network', quest: 'Quest', play: 'Games', games: 'Games', meme: 'Meme', pics: 'Pics', space: 'Space', work: 'Work', deals: 'Deals', news: 'News', report: 'Report', community: 'Community' })[id] || id;
++
++function createFeedSources({ db, config = {}, fetchImpl = globalThis.fetch, now = () => Date.now(), log = console, presence = null } = {}) {
++ const feed = config.feed || {};
++ const urls = {
++ community: String(feed.communityUrl || (config.services && config.services.community && config.services.community.internalUrl) || 'https://openvibe.community').replace(/\/+$/, ''),
++ today: String(feed.todayUrl || 'https://openvibe.today').replace(/\/+$/, ''),
++ quest: String(feed.questUrl || 'https://openvibe.quest').replace(/\/+$/, ''),
++ play: String(feed.playUrl || 'https://openvibe.games').replace(/\/+$/, ''),
++ meme: String(feed.memeUrl || 'https://openvibe.meme').replace(/\/+$/, ''),
++ pics: String(feed.picsUrl || 'https://openvibe.pics').replace(/\/+$/, ''),
++ live: String(feed.liveUrl || (config.services && config.services.live && config.services.live.internalUrl) || 'https://openvibe.live').replace(/\/+$/, ''),
++ space: String(feed.spaceUrl || 'https://openvibe.space').replace(/\/+$/, ''),
++ work: String(feed.workUrl || 'https://openvibe.work').replace(/\/+$/, ''),
++ deals: String(feed.dealsUrl || 'https://openvibe.deals').replace(/\/+$/, ''),
++ news: String(feed.newsUrl || 'https://openvibe.news').replace(/\/+$/, ''),
++ blog: String(feed.blogUrl || 'https://openvibe.blog').replace(/\/+$/, ''),
++ wiki: String(feed.wikiUrl || 'https://openvibe.wiki').replace(/\/+$/, ''),
++ report: String(feed.reportUrl || 'https://openvibe.report').replace(/\/+$/, ''),
++ };
++ let timer = null;
++ let running = null;
++
++ async function getJson(url, { timeoutMs = 10000 } = {}) {
++ const res = await fetchImpl(url, { headers: { accept: 'application/json' }, signal: AbortSignal.timeout(timeoutMs) });
++ if (!res.ok) throw new Error(`${url} answered ${res.status}`);
++ return await res.json();
++ }
++ async function getText(url, { timeoutMs = 10000 } = {}) {
++ const res = await fetchImpl(url, { headers: { accept: 'application/rss+xml, application/xml, text/xml' }, signal: AbortSignal.timeout(timeoutMs) });
++ if (!res.ok) throw new Error(`${url} answered ${res.status}`);
++ return await res.text();
++ }
++
++ /** A Live channel's person: the one Network account with that username, cached. */
++ async function personOf(handle) {
++ const key = clean(handle, 64).toLowerCase();
++ if (!key) return null;
++ const hit = await db.prepare('SELECT subject_id FROM feed_channel_people WHERE handle = ?').get(key);
++ if (hit) return hit.subject_id;
++ const rows = await db.prepare('SELECT subject_id FROM users WHERE lower(username) = ? AND subject_id IS NOT NULL AND COALESCE(is_test, 0) = 0 LIMIT 2').all(key);
++ const one = rows.length === 1 ? subject(rows[0].subject_id) : null;
++ if (one) await db.prepare('INSERT INTO feed_channel_people (handle, subject_id) VALUES (?, ?) ON CONFLICT (handle) DO UPDATE SET subject_id = excluded.subject_id, updated_at = now()').run(key, one);
++ return one;
++ }
++
++ async function upsert(it) { return await store.upsertItem(db, it); }
++
++ // ── Community ─────────────────────────────────────────────
++ /** Community's public activity, read directly from its tables: no Pulse store, no second pipeline. */
++ async function communityActivity() {
++ const since = new Date(now() - 30 * DAY).toISOString();
++ const out = await getJson(`${urls.community}/api/v1/activity?since=${encodeURIComponent(since)}&limit=1000`);
++ let created = 0;
++ for (const a of (Array.isArray(out.items) ? out.items : [])) {
++ const id = clean(a && a.id, 120);
++ const url = https(a && a.url);
++ const title = clean(a && a.title, 300);
++ if (!id || !url || !title) continue;
++ const type = clean(a.type, 40) || 'post';
++ const kind = a.kind || (type === 'paste' ? (a.picture_url ? 'pic' : 'paste') : type === 'thread' || type === 'status' ? 'post' : 'post');
++ const r = await upsert({
++ source_service: 'community', source_type: type, source_id: id, kind,
++ title, text: clean(a.text, 280), url, site: 'community', site_name: 'Community',
++ actor_subject: subject(a.actor), actor_name: clean(a.actor_name, 80) || null,
++ origin: a.origin === 'ai' ? 'ai' : (a.origin === 'system' ? 'system' : 'user'),
++ visibility: a.visibility === 'followers' ? 'followers' : 'public',
++ picture_url: https(a.picture_url), media_ratio: Number(a.media_ratio) > 0 ? Number(a.media_ratio) : null,
++ counts: a.counts && typeof a.counts === 'object' ? a.counts : {}, occurred_at: sqlFrom(a.at) || new Date().toISOString(),
++ });
++ if (r.created) created++;
++ }
++ return created;
++ }
++
++ // ── Today ─────────────────────────────────────────────────
++ async function todayEvents() {
++ const at = now();
++ const out = await getJson(`${urls.today}/api/v1/events?from=${encodeURIComponent(new Date(at).toISOString())}&to=${encodeURIComponent(new Date(at + 60 * DAY).toISOString())}&limit=100`);
++ let created = 0;
++ for (const e of (Array.isArray(out.events) ? out.events : [])) {
++ if (!e || typeof e.id !== 'string' || !e.id || typeof e.title !== 'string' || !e.title.trim()) continue;
++ if (e.visibility !== 'public' || e.status === 'cancelled') continue;
++ const starts = Date.parse(String(e.starts_at || ''));
++ if (!Number.isFinite(starts) || starts <= at) continue;
++ const owner = subject(e.owner_subject);
++ const whenLabel = `Starts ${new Date(starts).toISOString().replace('T', ' ').slice(0, 16)} UTC`;
++ const r = await upsert({
++ source_service: 'today', source_type: 'event', source_id: e.id, kind: 'event',
++ title: `Coming up: ${clean(e.title, 140)}`, url: https(e.url) || `${urls.today}/e/${encodeURIComponent(e.id)}`,
++ site: 'today', site_name: 'Today', actor_subject: owner, origin: owner ? 'user' : 'system',
++ text: [whenLabel, clean(e.description, 220)].filter(Boolean).join(' · '), starts_at: e.starts_at,
++ occurred_at: sqlFrom(e.created_at) || new Date(at).toISOString(),
++ counts: Number.isSafeInteger(Number(e.attending)) && Number(e.attending) > 0 ? { people: Number(e.attending) } : {},
++ });
++ if (r.created) created++;
++ }
++ return created;
++ }
++
++ // ── Network members ───────────────────────────────────────
++ async function networkMembers() {
++ const at = now();
++ const rows = await db.prepare(`SELECT substring(created_at, 1, 10) AS day, COUNT(*) AS n FROM users
++ WHERE COALESCE(is_test, 0) = 0 AND COALESCE(is_anon, 0) = 0 AND created_at >= ?
++ GROUP BY substring(created_at, 1, 10) ORDER BY day DESC LIMIT 7`).all(new Date(at - 7 * DAY).toISOString().slice(0, 10));
++ const total = await db.prepare('SELECT COUNT(*) AS n FROM users WHERE COALESCE(is_test, 0) = 0 AND COALESCE(is_anon, 0) = 0').get();
++ const today = new Date(at).toISOString().slice(0, 10);
++ let created = 0;
++ for (const row of rows) {
++ const count = Number(row.n);
++ const day = String(row.day || '');
++ if (!Number.isInteger(count) || count <= 0 || !/^\d{4}-\d{2}-\d{2}$/.test(day)) continue;
++ const r = await upsert({
++ source_service: 'network', source_type: 'members', source_id: day, kind: 'member',
++ title: day === today ? `${count} ${count === 1 ? 'person' : 'people'} joined today` : `${count} ${count === 1 ? 'person' : 'people'} joined on ${day}`,
++ url: `${config.networkUrl || 'https://openvibe.network'}/stats`, site: 'network', site_name: 'Network',
++ actor_subject: null, origin: 'system', label: 'System',
++ text: Number(total && total.n) > 0 ? `${Number(total.n).toLocaleString('en-US')} people on OpenVibe so far` : '',
++ occurred_at: day === today ? new Date(at).toISOString() : `${day}T12:00:00.000Z`,
++ });
++ if (r.created) created++;
++ }
++ return created;
++ }
++
++ // ── Quest badges ──────────────────────────────────────────
++ async function questBadges() {
++ const at = now();
++ const people = await db.prepare(`SELECT DISTINCT i.actor_subject AS subject FROM feed_items i
++ LEFT JOIN feed_quest_people q ON q.subject = i.actor_subject
++ WHERE i.actor_subject IS NOT NULL AND i.origin = 'user' AND i.occurred_at >= ?
++ AND (q.subject IS NULL OR q.checked_at < ?) ORDER BY i.actor_subject LIMIT 25`)
++ .all(new Date(at - QUEST_RECENT_MS).toISOString(), new Date(at - QUEST_CHECK_MS).toISOString());
++ let created = 0;
++ for (const p of people) {
++ const who = subject(p.subject);
++ if (!who) continue;
++ try {
++ const res = await fetchImpl(`${urls.quest}/api/v1/profiles/${encodeURIComponent(who)}/badges`, { headers: { accept: 'application/json' }, signal: AbortSignal.timeout(8000) });
++ if (res.ok) {
++ const body = await res.json();
++ const recent = (Array.isArray(body.badges) ? body.badges : [])
++ .filter((b) => b && typeof b.id === 'string' && b.id && typeof b.name === 'string' && b.name.trim())
++ .map((b) => ({ ...b, awarded: Date.parse(String(b.awarded_at || '')) }))
++ .filter((b) => Number.isFinite(b.awarded) && at - b.awarded <= QUEST_RECENT_MS)
++ .sort((a, b) => b.awarded - a.awarded).slice(0, 2);
++ for (const b of recent) {
++ const r = await upsert({ source_service: 'quest', source_type: 'badge', source_id: `${who}:${b.id}`, kind: 'quest',
++ title: `Finished a quest: ${clean(b.name, 90)}`, url: `${urls.quest}/`, site: 'quest', site_name: 'Quest',
++ actor_subject: who, origin: 'user', occurred_at: new Date(b.awarded).toISOString(), text: '' });
++ if (r.created) created++;
++ }
++ }
++ } catch (err) { log.warn && log.warn(`[Feed sources] Quest ${who}:`, err.message); }
++ await db.prepare(`INSERT INTO feed_quest_people (subject, checked_at) VALUES (?, now()) ON CONFLICT (subject) DO UPDATE SET checked_at = now()`).run(who);
++ }
++ return created;
++ }
++
++ // ── Play achievements ─────────────────────────────────────
++ async function playAchievements() {
++ let cursor = null, created = 0;
++ const seen = new Set();
++ for (let page = 0; page < 20; page++) {
++ const url = new URL('/api/v1/achievements/feed', urls.play);
++ url.searchParams.set('limit', '100');
++ if (cursor) url.searchParams.set('cursor', cursor);
++ const out = await getJson(url.href);
++ for (const a of (Array.isArray(out.items) ? out.items : [])) {
++ const who = subject(a && a.player);
++ const game = clean(a && a.title, 64);
++ const id = clean(a && a.id, 80);
++ const at = sqlFrom(a && a.unlocked_at);
++ if (!who || !/^[a-z0-9-]{2,40}$/.test(game) || !/^[a-z0-9_-]{1,64}$/.test(id) || !at) continue;
++ const sourceId = `${who}:${game}:${id}`;
++ seen.add(sourceId);
++ const r = await upsert({ source_service: 'games', source_type: 'achievement', source_id: sourceId, kind: 'game',
++ title: `Unlocked ${clean(a.name, 120)} in ${clean(a.title_name || game, 80)}`,
++ url: `https://openvibe.games/play?game=${encodeURIComponent(game)}`, site: 'games', site_name: 'Games',
++ actor_subject: who, origin: 'user', occurred_at: at, text: clean(a.description, 280) });
++ if (r.created) created++;
++ }
++ cursor = typeof out.next === 'string' && out.next ? out.next : null;
++ if (!cursor) break;
++ }
++ const old = await db.prepare("SELECT source_id FROM feed_items WHERE source_service = 'games' AND source_type = 'achievement'").all();
++ for (const row of old) if (!seen.has(String(row.source_id))) await store.removeItem(db, 'games', 'achievement', row.source_id);
++ return created;
++ }
++
++ // ── Meme entries ──────────────────────────────────────────
++ async function memeEntries() {
++ let cursor = null, created = 0;
++ for (let page = 0; page < 4; page++) {
++ const url = new URL('/api/v1/entries', urls.meme);
++ url.searchParams.set('limit', '50');
++ if (cursor) url.searchParams.set('cursor', cursor);
++ const out = await getJson(url.href);
++ for (const e of (Array.isArray(out.entries) ? out.entries : [])) {
++ const slug = clean(e && e.slug, 100);
++ const title = clean(e && e.title, 300);
++ if (!slug || !/^[A-Za-z0-9-]+$/.test(slug) || !title || (e.visibility && e.visibility !== 'public')) continue;
++ const r = await upsert({ source_service: 'meme', source_type: 'meme', source_id: slug, kind: 'meme', title,
++ url: `${urls.meme}/m/${encodeURIComponent(slug)}`, site: 'meme', site_name: 'Meme',
++ actor_subject: null, origin: 'system', label: 'System',
++ occurred_at: sqlFrom(e.published_at || e.created_at) || new Date(now()).toISOString(),
++ text: clean(e.summary || e.description, 280) });
++ if (r.created) created++;
++ }
++ cursor = typeof out.next_cursor === 'string' && out.next_cursor ? out.next_cursor : null;
++ if (!cursor) break;
++ }
++ return created;
++ }
++
++ // ── Pics ──────────────────────────────────────────────────
++ async function publicPictures() {
++ const seen = new Set();
++ let cursor = null, created = 0;
++ for (let page = 0; page < 20; page++) {
++ const url = new URL('/api/v1/gallery', urls.pics);
++ url.searchParams.set('sort', 'new');
++ url.searchParams.set('limit', '30');
++ if (cursor) url.searchParams.set('cursor', cursor);
++ const out = await getJson(url.href);
++ for (const p of (Array.isArray(out.pictures) ? out.pictures : [])) {
++ const id = clean(p && p.id, 80);
++ const links = p && p.links || {};
++ if (!/^pic_[0-9A-HJKMNP-TV-Z]{26}$/.test(id) || !p.listed || !https(links.page) || !https(links.thumb)) continue;
++ seen.add(id);
++ const r = await upsert({ source_service: 'pics', source_type: 'picture', source_id: id, kind: 'pic',
++ title: clean(p.title, 160) || 'Picture on OpenVibe.Pics', url: https(links.page), site: 'pics', site_name: 'Pics',
++ picture_url: https(links.thumb), actor_name: clean(p.author, 80), actor_subject: null, origin: 'user',
++ occurred_at: sqlFrom(p.published_at) || new Date(now()).toISOString(), text: clean(p.description, 280),
++ media_ratio: Number(p.width) > 0 && Number(p.height) > 0 ? Number(p.width) / Number(p.height) : null });
++ if (r.created) created++;
++ }
++ cursor = typeof out.next === 'string' && out.next ? out.next : null;
++ if (!cursor) break;
++ }
++ if (!cursor) await store.removeMissing(db, 'pics', 'picture', seen);
++ return created;
++ }
++
++ // ── Rundown ───────────────────────────────────────────────
++ async function rundownEpisodes() {
++ let before = null, created = 0;
++ const seen = new Set();
++ for (let page = 0; page < 20; page++) {
++ const url = new URL('/api/rundown/streams', urls.live);
++ if (before) url.searchParams.set('before', before);
++ const out = await getJson(url.href);
++ for (const s of (Array.isArray(out.streams) ? out.streams : [])) {
++ if (!s || s.live || !s.episode || s.episode.audience !== 'public' || !Number.isSafeInteger(Number(s.stream_id)) || Number(s.stream_id) < 1) continue;
++ const id = String(s.stream_id);
++ const title = clean(s.episode.title || s.title, 160);
++ if (!title) continue;
++ seen.add(id);
++ const prior = await db.prepare("SELECT actor_subject FROM feed_items WHERE source_service = 'live' AND source_type = 'stream' AND source_id = ?").get(id);
++ const who = (prior && prior.actor_subject) || await personOf(s.user && s.user.username);
++ const r = await upsert({ source_service: 'live', source_type: 'episode', source_id: id, kind: 'rundown',
++ title: `Rundown: ${title}`, url: `${urls.live}/rundown/e/${id}`, site: 'live', site_name: 'Live',
++ actor_subject: who, origin: who ? 'user' : 'system', label: who ? null : 'System',
++ occurred_at: sqlFrom(s.episode.aired_at || s.started_at) || new Date(now()).toISOString(),
++ text: clean(s.episode.logline, 280), picture_url: https(s.thumbnail_url) });
++ if (r.created) created++;
++ }
++ before = typeof out.next === 'string' && out.next ? out.next : null;
++ if (!before) break;
++ }
++ if (!before) await store.removeMissing(db, 'live', 'episode', seen);
++ return created;
++ }
++
++ // ── Live: on air, recordings and clips ────────────────────
++ async function publicContent() {
++ const seen = new Set();
++ let cursor = null, created = 0;
++ const live = await getJson(`${urls.live}/api/streams`);
++ const onAir = new Map();
++ const liveIds = new Set();
++ for (const s of (Array.isArray(live.streams) ? live.streams : [])) {
++ const id = Number(s && s.id), channel = clean(s && s.username, 64);
++ const started = sqlFrom(s && s.started_at);
++ if (!Number.isSafeInteger(id) || id < 1 || !/^[A-Za-z0-9_.-]{1,64}$/.test(channel) || !started) continue;
++ if (!onAir.has(channel.toLowerCase())) onAir.set(channel.toLowerCase(), []);
++ onAir.get(channel.toLowerCase()).push(Date.parse(started));
++ liveIds.add(String(id));
++ const name = clean(s.display_name || channel, 80);
++ const who = await personOf(channel);
++ const r = await store.upsertLiveItem(db, { source_service: 'live', source_type: 'stream', source_id: String(id), kind: 'live',
++ title: `${name} went live: ${clean(s.title, 160) || 'a stream'}`, url: `${urls.live}/@${encodeURIComponent(channel)}`,
++ site: 'live', site_name: 'Live', actor_name: name, actor_subject: who, origin: 'user',
++ occurred_at: started, live_now: true });
++ if (r.created) created++;
++ await db.prepare("UPDATE feed_items SET live_now = true, updated_at = now() WHERE source_service = 'live' AND source_type = 'stream' AND source_id = ?").run(String(r.item.source_id));
++ }
++ // Anything no longer on Live's list is a past broadcast, not "live now".
++ const onAirRows = await db.prepare("SELECT source_id FROM feed_items WHERE source_service = 'live' AND source_type = 'stream' AND live_now = true").all();
++ for (const row of onAirRows) if (!liveIds.has(String(row.source_id))) await db.prepare("UPDATE feed_items SET live_now = false, updated_at = now() WHERE source_service = 'live' AND source_type = 'stream' AND source_id = ?").run(String(row.source_id));
++ for (let page = 0; page < 20; page++) {
++ const url = new URL('/api/content/feed', urls.live);
++ url.searchParams.set('sort', 'new');
++ url.searchParams.set('window', 'all');
++ url.searchParams.set('limit', '30');
++ if (cursor) url.searchParams.set('cursor', cursor);
++ const out = await getJson(url.href);
++ if (out.partial || (out.sources && (out.sources.vods !== 'ok' || out.sources.clips !== 'ok'))) throw new Error('Live Content feed was partial; retaining prior cards');
++ for (const v of (Array.isArray(out.items) ? out.items : [])) {
++ if (!v || !['vod', 'clip'].includes(v.kind) || !https(v.href)) continue;
++ const id = clean(v.id, 80);
++ if (!/^[A-Za-z0-9_-]{1,80}$/.test(id)) continue;
++ const type = v.kind === 'vod' ? 'recording' : 'clip';
++ const at = sqlFrom(v.created_at);
++ if (!at) continue;
++ const channel = clean(v.channel && v.channel.username, 64);
++ const current = onAir.get(channel.toLowerCase()) || [];
++ if (type === 'recording' && current.some((started) => Number.isFinite(started) && Math.abs(Date.parse(at) - started) <= 10 * MINUTE)) continue;
++ seen.add(`${type}:${id}`);
++ let prior = null;
++ if (type === 'recording' && channel) {
++ const candidates = await db.prepare(`SELECT actor_subject, source_id FROM feed_items WHERE source_service = 'live' AND source_type = 'stream'
++ AND lower(url) = lower(?) AND occurred_at BETWEEN ? AND ? ORDER BY occurred_at DESC LIMIT 20`)
++ .all(`${urls.live}/@${encodeURIComponent(channel)}`, new Date(Date.parse(at) - 60 * MINUTE).toISOString(), new Date(Date.parse(at) + 60 * MINUTE).toISOString());
++ prior = candidates.find((row) => !liveIds.has(String(row.source_id))) || null;
++ }
++ const r = await upsert({ source_service: 'live', source_type: type, source_id: id, kind: type === 'recording' ? 'video' : 'clip',
++ title: `${type === 'recording' ? 'Recording: ' : ''}${clean(v.title, 150) || (type === 'recording' ? 'A stream' : 'A clip')}`,
++ url: https(v.href), site: 'live', site_name: 'Live', picture_url: https(v.thumbnail_url),
++ 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,
++ counts: Number(v.view_count) > 0 ? { views: Number(v.view_count) } : {} });
++ if (r.created) created++;
++ if (prior) await store.removeItem(db, 'live', 'stream', prior.source_id);
++ }
++ cursor = typeof out.next === 'string' && out.next ? out.next : null;
++ if (!cursor) break;
++ }
++ if (!cursor) await store.removeMissing(db, 'live', 'recording', new Set([...seen].filter((x) => x.startsWith('recording:')).map((x) => x.split(':')[1])));
++ if (!cursor) await store.removeMissing(db, 'live', 'clip', new Set([...seen].filter((x) => x.startsWith('clip:')).map((x) => x.split(':')[1])));
++ return created;
++ }
++
++ /** "312 people watched Maticus tonight": Network's own creator analytics for the streams that ended. */
++ async function watchedTogether() {
++ let created = 0;
++ const rows = await db.prepare(`SELECT stream_id, creator_subject, title, ended_at, peak_viewers FROM creator_streams
++ WHERE ended_at IS NOT NULL AND peak_viewers > 0 AND ended_at >= ? ORDER BY ended_at DESC LIMIT 40`)
++ .all(new Date(now() - 7 * DAY).toISOString());
++ for (const s of rows) {
++ const who = subject(s.creator_subject);
++ const name = who ? await db.prepare('SELECT display_name, username FROM users WHERE subject_id = ?').get(who) : null;
++ const label = clean(name && (name.display_name || name.username), 60) || 'a creator';
++ const peak = Number(s.peak_viewers);
++ if (!Number.isSafeInteger(peak) || peak <= 0) continue;
++ const ended = sqlFrom(s.ended_at);
++ if (!ended) continue;
++ const r = await upsert({ source_service: 'live', source_type: 'watched', source_id: String(s.stream_id), kind: 'stream',
++ title: `${peak.toLocaleString('en-US')} ${peak === 1 ? 'person' : 'people'} watched ${label} tonight`,
++ text: clean(s.title, 220), url: `https://openvibe.live/@${encodeURIComponent((name && name.username) || label)}`,
++ site: 'live', site_name: 'Live', actor_subject: who, origin: 'user', occurred_at: ended,
++ counts: { watchers: peak } });
++ if (r.created) created++;
++ }
++ return created;
++ }
++
++ // ── Spaces ────────────────────────────────────────────────
++ async function publicSpaces() {
++ const seen = new Set();
++ let cursor = null, created = 0;
++ for (let page = 0; page < 20; page++) {
++ const url = new URL('/api/v1/apps/gallery', urls.space);
++ url.searchParams.set('limit', '100');
++ if (cursor) url.searchParams.set('cursor', cursor);
++ const out = await getJson(url.href);
++ for (const s of (Array.isArray(out.spaces) ? out.spaces : [])) {
++ const id = clean(s && s.id, 80);
++ const at = Number(s && s.published_at);
++ if (!/^spc_[0-9A-HJKMNP-TV-Z]{26}$/.test(id) || !Number.isFinite(at) || at <= 0 || !https(s.url)) continue;
++ seen.add(id);
++ const r = await upsert({ source_service: 'space', source_type: 'space', source_id: id, kind: 'space',
++ title: clean(s.name, 160) || 'New space', url: https(s.url), site: 'space', site_name: 'Space',
++ text: clean(s.description, 280), actor_name: clean(s.owner_name || s.owner_username, 80), actor_subject: null,
++ origin: 'user', occurred_at: new Date(at).toISOString() });
++ if (r.created) created++;
++ }
++ cursor = typeof out.next_cursor === 'string' && out.next_cursor ? out.next_cursor : null;
++ if (!cursor) break;
++ }
++ if (!cursor) await store.removeMissing(db, 'space', 'space', seen);
++ return created;
++ }
++
++ // ── Work ──────────────────────────────────────────────────
++ async function workJobs() {
++ const out = await getJson(`${urls.work}/api/v1/jobs?limit=60`);
++ let created = 0;
++ for (const j of (Array.isArray(out.jobs) ? out.jobs : [])) {
++ const id = clean(j && j.id, 100);
++ const url = https(j && j.url);
++ const title = clean(j && j.title, 180);
++ if (!/^job_[A-Za-z0-9]{8,80}$/.test(id) || !url || !title) continue;
++ const where = [clean(j.company, 80), clean(j.location, 60), j.remote ? 'Remote' : ''].filter(Boolean).join(' · ');
++ const r = await upsert({ source_service: 'work', source_type: 'job', source_id: id, kind: 'work',
++ title, text: clean([where, clean(j.excerpt, 200)].filter(Boolean).join(' — '), 280), url,
++ site: 'work', site_name: 'Work', actor_subject: null, origin: 'system', label: 'System',
++ occurred_at: sqlFrom(j.first_seen_at || j.posted_at) || new Date(now()).toISOString() });
++ if (r.created) created++;
++ }
++ return created;
++ }
++
++ // ── Deals ─────────────────────────────────────────────────
++ async function dealOffers() {
++ const out = await getJson(`${urls.deals}/api/v1/offers?limit=40&sort=new`);
++ let created = 0;
++ for (const o of (Array.isArray(out.offers) ? out.offers : [])) {
++ if (!o || o.status === 'expired' || o.expired_at) continue;
++ const url = https(o.url);
++ const title = clean(o.title, 180);
++ if (!url || !title) continue;
++ const votes = Number(o.votes);
++ const r = await upsert({ source_service: 'deals', source_type: 'offer', source_id: clean(o.id, 100), kind: 'deal',
++ title, text: clean(o.ai_summary || o.description, 280), url, site: 'deals', site_name: 'Deals',
++ actor_subject: null, origin: 'system', label: 'System', media_ratio: null,
++ counts: Number.isSafeInteger(votes) && votes > 0 ? { likes: votes } : {},
++ occurred_at: sqlFrom(o.created_at) || new Date(now()).toISOString() });
++ if (r.created) created++;
++ }
++ return created;
++ }
++
++ // ── News ──────────────────────────────────────────────────
++ async function newsStories() {
++ const out = await getJson(`${urls.news}/api/v1/stories?limit=40`);
++ let created = 0;
++ for (const s of (Array.isArray(out.stories) ? out.stories : [])) {
++ const id = clean(s && s.id, 100);
++ const url = https(s && (s.url || s.canonical_url));
++ const title = clean(s && s.title, 300);
++ if (!id || !url || !title || (s.status && s.status !== 'published')) continue;
++ const r = await upsert({ source_service: 'news', source_type: 'story', source_id: id, kind: 'news', title,
++ text: clean(s.summary || s.excerpt, 280), url, site: 'news', site_name: 'News', actor_subject: null,
++ origin: 'system', label: 'System', occurred_at: sqlFrom(s.published_at || s.created_at) || new Date(now()).toISOString() });
++ if (r.created) created++;
++ }
++ return created;
++ }
++
++ // ── Blog ──────────────────────────────────────────────────
++ /** The blog's public RSS: its written posts, decoded to plain text. */
++ async function blogPosts() {
++ const xml = await getText(`${urls.blog}/feed.xml`);
++ const decode = (s) => String(s == null ? '' : s).replace(/<!\[CDATA\[([\s\S]*?)\]\]>/g, '$1')
++ .replace(/<[^>]*>/g, ' ').replace(/&lt;/g, '<').replace(/&gt;/g, '>').replace(/&quot;/g, '"').replace(/&#39;|&apos;/g, "'").replace(/&amp;/g, '&')
++ .replace(/\s+/g, ' ').trim();
++ const blocks = String(xml).split(/<item[\s>]/i).slice(1);
++ let created = 0;
++ for (const raw of blocks) {
++ const body = raw.split(/<\/item>/i)[0];
++ const title = decode((/<title>([\s\S]*?)<\/title>/i.exec(body) || [])[1]);
++ const link = decode((/<link>([\s\S]*?)<\/link>/i.exec(body) || [])[1]);
++ const date = Date.parse(decode((/<pubDate>([\s\S]*?)<\/pubDate>/i.exec(body) || [])[1]));
++ const desc = decode((/<description>([\s\S]*?)<\/description>/i.exec(body) || [])[1]);
++ const url = https(link);
++ if (!title || !url) continue;
++ const r = await upsert({ source_service: 'blog', source_type: 'post', source_id: url, kind: 'update',
++ title: clean(title, 300), text: clean(desc, 280), url, site: 'blog', site_name: 'Blog',
++ actor_subject: null, origin: 'system', label: 'System',
++ occurred_at: Number.isFinite(date) ? new Date(date).toISOString() : new Date(now()).toISOString() });
++ if (r.created) created++;
++ }
++ return created;
++ }
++
++ // ── Wiki ──────────────────────────────────────────────────
++ /** Wiki's public search for a common word, filtered to pages published in the last 30 days. */
++ async function wikiPages() {
++ const out = await getJson(`${urls.wiki}/api/v1/search?q=the&limit=50`);
++ let created = 0;
++ const since = now() - 30 * DAY;
++ for (const p of (Array.isArray(out.results) ? out.results : [])) {
++ if (!p || p.state !== 'published' || p.visibility !== 'public' || p.noindex) continue;
++ const at = sqlFrom(p.revision_published_at || p.published_at || p.updated_at);
++ if (!at || Date.parse(at) < since) continue;
++ const url = https(p.url);
++ const title = clean(p.title, 300);
++ if (!url || !title) continue;
++ const r = await upsert({ source_service: 'wiki', source_type: 'page', source_id: clean(p.id, 100), kind: 'wiki',
++ title, text: '', url, site: 'wiki', site_name: 'Wiki', actor_subject: null, origin: 'system', label: 'System',
++ occurred_at: at, touch: true });
++ if (r.created) created++;
++ }
++ return created;
++ }
++
++ // ── Report: the online peak and this window's numbers ─────
++ async function samplePeak() {
++ if (!presence || typeof presence.snapshot !== 'function') return 0;
++ const snap = presence.snapshot(now());
++ const online = snap.members.size + snap.guests;
++ if (!Number.isFinite(online) || online <= 0) return 0;
++ const day = new Date(now()).toISOString().slice(0, 10);
++ await db.prepare(`INSERT INTO feed_peaks (day, peak, peak_at) VALUES (?, ?, now())
++ ON CONFLICT (day) DO UPDATE SET peak = GREATEST(feed_peaks.peak, excluded.peak),
++ peak_at = CASE WHEN excluded.peak > feed_peaks.peak THEN excluded.peak_at ELSE feed_peaks.peak_at END,
++ updated_at = now()`).run(day, online);
++ return 1;
++ }
++
++ async function reportCard() {
++ const at = now();
++ let stats = null;
++ try { stats = await getJson(`${urls.report}/api/v1/stats`); } catch (err) { log.warn && log.warn('[Feed sources] Report stats:', err.message); }
++ const day = new Date(at).toISOString().slice(0, 10);
++ const peak = await db.prepare('SELECT peak, peak_at FROM feed_peaks WHERE day = ?').get(day);
++ let created = 0;
++ if (peak && Number(peak.peak) > 0) {
++ const when = new Date(peak.peak_at).toISOString().slice(11, 16);
++ const r = await upsert({ source_service: 'report', source_type: 'peak', source_id: `peak:${day}`, kind: 'report',
++ title: `Peak today: ${Number(peak.peak).toLocaleString('en-US')} people online at ${when} UTC`,
++ text: 'The most people on OpenVibe at once today', url: `${urls.report}/network`, site: 'report', site_name: 'Report',
++ actor_subject: null, origin: 'system', label: 'System', counts: { people: Number(peak.peak) },
++ occurred_at: new Date(at).toISOString(), touch: true });
++ if (r.created) created++;
++ }
++ const totals = stats && stats.totals;
++ if (totals && Number.isSafeInteger(Number(totals.pageviews))) {
++ const pct = Number(totals.change_pct);
++ const views = Number(totals.pageviews);
++ const r = await upsert({ source_service: 'report', source_type: 'views', source_id: `views:${day}`, kind: 'report',
++ title: `${views.toLocaleString('en-US')} page views in the last 30 days`,
++ text: [`${Number(totals.sites_open) || 0} sites open`, `${Number(totals.accounts) || 0} accounts`, Number.isFinite(pct) ? `${pct >= 0 ? '+' : ''}${pct}% on the 30 days before` : ''].filter(Boolean).join(' · '),
++ url: `${urls.report}/`, site: 'report', site_name: 'Report', actor_subject: null, origin: 'system', label: 'System',
++ counts: { people: views }, trend: Number.isFinite(pct) ? { direction: pct > 0 ? 'up' : pct < 0 ? 'down' : 'flat', pct: Math.abs(pct) } : undefined,
++ occurred_at: new Date(at).toISOString(), touch: true });
++ if (r.created) created++;
++ }
++ return created;
++ }
++
++ const SOURCES = [
++ { name: 'community-activity', every: 2 * MINUTE, run: communityActivity },
++ { name: 'today-events', every: 10 * MINUTE, run: todayEvents },
++ { name: 'network-members', every: 60 * MINUTE, run: networkMembers },
++ { name: 'quest-badges', every: 15 * MINUTE, run: questBadges },
++ { name: 'play-achievements', every: 10 * MINUTE, run: playAchievements },
++ { name: 'meme-entries', every: 10 * MINUTE, run: memeEntries },
++ { name: 'public-pictures', every: 5 * MINUTE, run: publicPictures },
++ { name: 'rundown-episodes', every: 10 * MINUTE, run: rundownEpisodes },
++ { name: 'public-content', every: 3 * MINUTE, run: publicContent },
++ { name: 'watched-together', every: 10 * MINUTE, run: watchedTogether },
++ { name: 'public-spaces', every: 10 * MINUTE, run: publicSpaces },
++ { name: 'work-jobs', every: 10 * MINUTE, run: workJobs },
++ { name: 'deal-offers', every: 10 * MINUTE, run: dealOffers },
++ { name: 'news-stories', every: 10 * MINUTE, run: newsStories },
++ { name: 'blog-posts', every: 10 * MINUTE, run: blogPosts },
++ { name: 'wiki-pages', every: 10 * MINUTE, run: wikiPages },
++ { name: 'online-peak', every: 2 * MINUTE, run: samplePeak },
++ { name: 'report-card', every: 30 * MINUTE, run: reportCard },
++ ];
++
++ async function remember(name, { nextRun, lastRun, error = null }) {
++ await db.prepare(`INSERT INTO feed_source_runs (name, next_run_at, last_run_at, last_error) VALUES (?, ?, ?, ?)
++ ON CONFLICT (name) DO UPDATE SET next_run_at = excluded.next_run_at, last_run_at = excluded.last_run_at, last_error = excluded.last_error`)
++ .run(name, nextRun, lastRun, error);
++ }
++
++ /** Run every source due now (all with { force: true }). → [{ name, items } | { name, error }] */
++ async function runOnce({ force = false } = {}) {
++ if (running) return running;
++ running = (async () => {
++ const at = now();
++ const states = new Map((await db.prepare('SELECT * FROM feed_source_runs').all()).map((s) => [s.name, s]));
++ const out = [];
++ for (const source of SOURCES) {
++ const state = states.get(source.name);
++ if (!force && state && state.next_run_at && Date.parse(state.next_run_at) > at) continue;
++ try {
++ const items = await source.run();
++ await remember(source.name, { nextRun: new Date(at + source.every).toISOString(), lastRun: new Date(at).toISOString() });
++ out.push({ name: source.name, items });
++ } catch (err) {
++ log.warn && log.warn(`[Feed sources] ${source.name}:`, err.message);
++ await remember(source.name, { nextRun: new Date(at + source.every).toISOString(), lastRun: new Date(at).toISOString(), error: String(err.message).slice(0, 300) });
++ out.push({ name: source.name, error: err.message });
++ }
++ }
++ return out;
++ })().finally(() => { running = null; });
++ return await running;
++ }
++
++ function start() {
++ if (timer) return;
++ const tick = () => { runOnce().catch((err) => log.warn && log.warn('[Feed sources]:', err.message)); };
++ timer = setInterval(tick, MINUTE);
++ if (timer.unref) timer.unref();
++ const first = setTimeout(tick, 5000);
++ if (first.unref) first.unref();
++ timer.first = first;
++ }
++ function stop() {
++ if (timer) clearInterval(timer);
++ if (timer && timer.first) clearTimeout(timer.first);
++ timer = null;
++ }
++ async function status() {
++ return { sources: SOURCES.map((s) => s.name), runs: await db.prepare('SELECT * FROM feed_source_runs ORDER BY name').all() };
++ }
++
++ return { runOnce, start, stop, status, SOURCES: SOURCES.map((s) => s.name), communityActivity, todayEvents, networkMembers, questBadges, playAchievements, memeEntries, publicPictures, rundownEpisodes, publicContent, watchedTogether, publicSpaces, workJobs, dealOffers, newsStories, blogPosts, wikiPages, samplePeak, reportCard };
++}
++
++module.exports = { createFeedSources };
ddiff --git a/server/feed/store.js b/server/feed/store.js
nnew file mode 100644
iindex 00000000..0d33688b
---- /dev/null
++++ b/server/feed/store.js
@@@ -0,0 +1,174 @@
++'use strict';
++
++/**
++ * The feed store — pure SQL over feed_items (migrations/0080_feed.sql).
++ *
++ * One row per source item, keyed by (service, type, id). Re-recording the same source refreshes its
++ * words, link and picture; provenance (when it happened, who it is about) is what the first record
++ * said unless the source says `touch` (a wiki page edited again, a busy chat hour still filling).
++ *
++ * Live is the exception: a dropped connection starts a new Live session (a new stream id) but it is
++ * the same broadcast. upsertLiveItem keeps one item for the same person (or channel URL) and title
++ * within six hours and moves its time to the newest start.
++ *
++ * Privacy is a property of the row: only 'public' or 'followers' exists here. listItems() never
++ * returns a followers item to anyone but its author or a follower, and never returns a test
++ * account's item (users.is_test), whatever the source said.
++ */
++const LIVE_WINDOW_MS = 6 * 60 * 60 * 1000;
++const COLUMNS = 'id, source_service, source_type, source_id, kind, title, text, url, site, site_name, actor_subject, actor_name, origin, visibility, picture_url, preview_url, media_ratio, media_length, counts, live_now, label, starts_at, occurred_at, trend';
++
++const iso = (v) => {
++ if (v == null || v === '') return null;
++ const t = Date.parse(String(v));
++ return Number.isFinite(t) ? new Date(t).toISOString() : null;
++};
++const ms = (v) => { const t = Date.parse(String(v == null ? '' : v)); return Number.isFinite(t) ? t : NaN; };
++const countsJson = (v) => {
++ if (!v || typeof v !== 'object') return '{}';
++ const out = {};
++ for (const k of ['views', 'likes', 'comments', 'watchers', 'people']) {
++ const n = Number(v[k]);
++ if (Number.isSafeInteger(n) && n >= 0) out[k] = n;
++ }
++ return JSON.stringify(out);
++};
++
++/** → { item, created } */
++async function upsertItem(db, it) {
++ const at = iso(it.occurred_at) || new Date().toISOString();
++ const key = [it.source_service, it.source_type, String(it.source_id)];
++ const find = db.prepare(`SELECT ${COLUMNS} FROM feed_items WHERE source_service = ? AND source_type = ? AND source_id = ?`);
++ return await db.tx(async () => {
++ const had = await find.get(...key);
++ const values = [
++ String(it.kind), String(it.title || '').slice(0, 300), it.text == null ? null : String(it.text).slice(0, 300),
++ String(it.url), String(it.site || it.source_service).slice(0, 40), String(it.site_name || it.site || it.source_service).slice(0, 40),
++ it.actor_subject || null, it.actor_name ? String(it.actor_name).slice(0, 80) : null, it.origin || 'user',
++ it.visibility === 'followers' ? 'followers' : 'public',
++ it.picture_url || null, it.preview_url || null,
++ Number.isFinite(Number(it.media_ratio)) && Number(it.media_ratio) > 0 ? Number(it.media_ratio) : null,
++ Number.isFinite(Number(it.media_length)) && Number(it.media_length) >= 0 ? Math.round(Number(it.media_length)) : null,
++ countsJson(it.counts), it.live_now === true, it.trend && typeof it.trend === 'object' ? JSON.stringify(it.trend) : null, it.label || null, iso(it.starts_at), at,
++ ];
++ if (had) {
++ const moving = it.touch === true;
++ await db.prepare(`UPDATE feed_items SET kind = ?, title = ?, text = ?, url = ?, site = ?, site_name = ?,
++ actor_subject = COALESCE(actor_subject, ?), actor_name = COALESCE(actor_name, ?), origin = ?, visibility = ?,
++ picture_url = ?, preview_url = ?, media_ratio = ?, media_length = ?, counts = ?, live_now = ?, trend = ?, label = ?, starts_at = ?,
++ occurred_at = CASE WHEN ? THEN ? ELSE occurred_at END, updated_at = now()
++ WHERE source_service = ? AND source_type = ? AND source_id = ?`)
++ .run(...values.slice(0, 19), moving ? true : false, at, ...key);
++ return { item: await find.get(...key), created: false };
++ }
++ await db.prepare(`INSERT INTO feed_items (source_service, source_type, source_id, kind, title, text, url, site, site_name,
++ actor_subject, actor_name, origin, visibility, picture_url, preview_url, media_ratio, media_length, counts, live_now, trend, label, starts_at, occurred_at)
++ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`)
++ .run(...key, ...values);
++ return { item: await find.get(...key), created: true };
++ });
++}
++
++/** A live.stream.started item: one row per broadcast, not per reconnect. → { item, created } */
++async function upsertLiveItem(db, it) {
++ const key = [it.source_service, it.source_type, String(it.source_id)];
++ const find = db.prepare(`SELECT ${COLUMNS} FROM feed_items WHERE source_service = ? AND source_type = ? AND source_id = ?`);
++ return await db.tx(async () => {
++ const had = await find.get(...key);
++ if (had) return { item: had, created: false };
++ const at = ms(it.occurred_at);
++ const actor = it.actor_subject || null;
++ if (Number.isFinite(at)) {
++ const candidates = actor
++ ? await db.prepare(`SELECT ${COLUMNS} FROM feed_items WHERE source_service = 'live' AND source_type = 'stream'
++ AND title = ? AND ((actor_subject IS NOT NULL AND actor_subject = ?) OR url = ?)
++ ORDER BY occurred_at DESC, id DESC`).all(it.title, actor, it.url)
++ : await db.prepare(`SELECT ${COLUMNS} FROM feed_items WHERE source_service = 'live' AND source_type = 'stream'
++ AND title = ? AND url = ? ORDER BY occurred_at DESC, id DESC`).all(it.title, it.url);
++ let prior = null; let priorAt = -Infinity;
++ for (const c of candidates) {
++ const t = ms(c.occurred_at);
++ if (Number.isFinite(t) && t <= at && at - t <= LIVE_WINDOW_MS && t > priorAt) { prior = c; priorAt = t; }
++ }
++ if (prior) {
++ if (at > priorAt) await db.prepare('UPDATE feed_items SET occurred_at = ?, updated_at = now() WHERE id = ?').run(iso(it.occurred_at), prior.id);
++ return { item: await db.prepare(`SELECT ${COLUMNS} FROM feed_items WHERE id = ?`).get(prior.id), created: false };
++ }
++ }
++ return await upsertItem(db, it);
++ });
++}
++
++async function removeItem(db, service, type, id) {
++ const r = await db.prepare('DELETE FROM feed_items WHERE source_service = ? AND source_type = ? AND source_id = ?').run(service, type, String(id));
++ return Number(r.changes || 0);
++}
++
++/** Remove every item a source no longer lists (a gallery page, a recordings feed): the kept ids survive. */
++async function removeMissing(db, service, type, keepIds) {
++ const keep = new Set([...keepIds].map(String));
++ const rows = await db.prepare('SELECT source_id FROM feed_items WHERE source_service = ? AND source_type = ?').all(service, type);
++ let removed = 0;
++ for (const row of rows) if (!keep.has(String(row.source_id))) removed += await removeItem(db, service, type, row.source_id);
++ return removed;
++}
++
++/** What a person follows: their active targets (the channel targets Network's follow graph keeps). */
++async function followedSubjects(db, subject) {
++ if (!subject) return [];
++ const rows = await db.prepare("SELECT target_id FROM user_follows WHERE follower_subject = ? AND active = 1 AND target_type = 'channel'").all(subject);
++ return [...new Set(rows.map((r) => String(r.target_id)).filter((x) => /^usr_[0-9A-HJKMNP-TV-Z]{26}$/.test(x)))];
++}
++
++/**
++ * What the person opens: the services in their recent cross-site history (server/history/routes.js),
++ * most used first. A small affinity boost, never a filter.
++ */
++async function affinityServices(db, userId) {
++ if (!userId) return new Map();
++ const rows = await db.prepare('SELECT service, COUNT(*) AS n FROM user_history WHERE user_id = ? AND service IS NOT NULL GROUP BY service ORDER BY n DESC LIMIT 12').all(userId);
++ return new Map(rows.map((r) => [String(r.service), Number(r.n) || 0]));
++}
++
++/**
++ * Newest first, keyset-paginated by (occurred_at, id). Filters:
++ * opts.kind, opts.site, opts.q the chips and the search
++ * opts.followed, opts.following the person's follow graph
++ * opts.before [at, id] of the last row already read
++ * opts.viewerSubject the signed-in reader (their own followers items)
++ * opts.since only this recent (the collectors read the last 7 days)
++ * → { rows, hasMore }
++ */
++async function listItems(db, { kind = '', site = '', q = '', followed = [], following = false, before = null, viewerSubject = null, limit = 40, since = null } = {}) {
++ const where = [];
++ const args = [];
++ if (kind) { where.push('i.kind = ?'); args.push(kind); }
++ if (site) { where.push('i.site = ?'); args.push(site); }
++ if (q) { where.push('(i.title ILIKE ? OR i.text ILIKE ?)'); args.push(`%${q}%`, `%${q}%`); }
++ if (since) { where.push('i.occurred_at >= ?'); args.push(since); }
++ if (following) {
++ if (!followed.length) return { rows: [], hasMore: false };
++ where.push(`i.actor_subject IN (${followed.map(() => '?').join(', ')})`);
++ args.push(...followed);
++ }
++ if (viewerSubject) {
++ const list = followed.length ? followed.map(() => '?').join(', ') : null;
++ where.push(`(i.visibility = 'public' OR (i.visibility = 'followers' AND (i.actor_subject = ?${list ? ` OR i.actor_subject IN (${list})` : ''})))`);
++ args.push(viewerSubject, ...followed);
++ } else {
++ where.push("i.visibility = 'public'");
++ }
++ // A test account's activity never reaches a public feed, signed in or not (users.is_test; the same
++ // rule as Network's stats and home). The actor's own row still shows them their own item.
++ where.push(`(i.actor_subject IS NULL OR i.actor_subject = ? OR NOT EXISTS (
++ SELECT 1 FROM users u WHERE u.subject_id = i.actor_subject AND COALESCE(u.is_test, 0) = 1))`);
++ args.push(viewerSubject);
++ if (before) { where.push('(i.occurred_at < ? OR (i.occurred_at = ? AND i.id < ?))'); args.push(before[0], before[0], Number(before[1])); }
++ const rows = await db.prepare(`SELECT ${COLUMNS} FROM feed_items i WHERE ${where.join(' AND ')}
++ ORDER BY i.occurred_at DESC, i.id DESC LIMIT ?`).all(...args, limit + 1);
++ const hasMore = rows.length > limit;
++ if (hasMore) rows.pop();
++ return { rows, hasMore };
++}
++
++module.exports = { upsertItem, upsertLiveItem, removeItem, removeMissing, followedSubjects, affinityServices, listItems, iso, ms };
ddiff --git a/server/frame/service.js b/server/frame/service.js
iindex 79aa32f9..f79991eb 100644
---- a/server/frame/service.js
++++ b/server/frame/service.js
@@@ -169,7 +169,7 @@ function aiSiteCopy({ privateKey, issuer, aiUrl }) {
};
}
--async function createFrameService(db, config, analytics, { privateKey = null, issuer = null, selfToken = () => null, aiUrl = null, requireAuth = null, person = null, today = null, now = Date.now, sites = SITES, places = LISTED, mentionLimit = 120 } = {}) {
++async function createFrameService(db, config, analytics, { privateKey = null, issuer = null, selfToken = () => null, aiUrl = null, requireAuth = null, person = null, today = null, now = Date.now, sites = SITES, places = LISTED, mentionLimit = 120, feed = null } = {}) {
const services = (config && config.services) || {};
aiUrl = aiUrl || (services.ai && services.ai.internalUrl) || process.env.OV_AI_INTERNAL_URL || 'http://127.0.0.1:4700';
const choices = createFrameChoices({
@@@ -731,20 +731,23 @@ async function createFrameService(db, config, analytics, { privateKey = null, is
});
router.post('/posts', express.json({ limit: '8kb' }), (req, res) => sendOn(req, res, '/api/v1/posts', JSON.stringify(req.body || {}), 'application/json'));
-- router.get('/feed', (req, res) => relay(req, res, '/api/v1/feed', {
++ // The feed is Network's own (server/feed/api.js, feed.item@1) since 11-12: the same handler Community's
++ // /api/v1/feed proxies, so the menu and Pulse return the same ids. The old relay stays only for a
++ // Frame built without it (a unit test), never in the server.
++ router.get('/feed', (req, res, next) => (feed ? feed.handleFeed(req, res, next) : relay(req, res, '/api/v1/feed', {
kind: FEED_KINDS.includes(String(req.query.kind || '')) ? String(req.query.kind) : '',
site: qSite(req.query.site), q: qWords(req.query.q),
following: /^(1|true)$/.test(String(req.query.following || '')) ? '1' : '',
cursor: qCursor(req.query.cursor), limit: qLimit(req.query.limit, 40),
-- }));
-- router.get('/feed/search', (req, res) => relay(req, res, '/api/v1/feed/search', {
++ })));
++ router.get('/feed/search', (req, res, next) => (feed ? feed.handleSearch(req, res, next) : relay(req, res, '/api/v1/feed/search', {
q: qWords(req.query.q),
-- }));
-- router.get('/feed/examples', (req, res) => relay(req, res, '/api/v1/feed/examples', {}));
-- router.get('/wall', (req, res) => relay(req, res, '/api/v1/pulse', {
++ })));
++ router.get('/feed/examples', (req, res, next) => (feed ? feed.handleExamples(req, res, next) : relay(req, res, '/api/v1/feed/examples', {})));
++ router.get('/wall', (req, res, next) => (feed ? feed.handleWall(req, res, next) : relay(req, res, '/api/v1/pulse', {
limit: qLimit(req.query.limit, 50), after: qCursor(req.query.after), cursor: qCursor(req.query.cursor),
for: req.query.for === 'me' ? 'me' : '', origin: ['user', 'ai', 'system', 'all'].includes(String(req.query.origin || '')) ? String(req.query.origin) : '',
-- }));
++ })));
// ── The wallet sheet's money and items (openvibe-shared ov-wallet.js) ──────────────────────────────────────
// Every site's connect-src allows openvibe.network and refuses billing./inventory.openvibe.network, so the
ddiff --git a/server/index.js b/server/index.js
iindex db44e85a..78294ba4 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -899,6 +899,7 @@ require('./developer/confirmations').setNotifier(notificationService);
require('./developer/confirmations').start(db);
// The Action Center (plan 2026-10-28 §3.10, server/actions): the person's agent work from Actor, Codes and AI and their
// pending confirmations, in five buckets, for the ✦ beside the bell on every site.
++let feedApi = null;
{
const actions = require('./actions').createActions({ db, selfToken, urls: {
actor: process.env.OV_ACTOR_INTERNAL_URL || 'http://127.0.0.1:4950',
@@@ -918,6 +919,14 @@ require('./developer/confirmations').start(db);
const menuFeeds = require('./menu-feeds').createMenuFeeds({ db });
app.use('/api/v1/me/menu-feeds', require('./menu-feeds').router({ menuFeeds, requireAuth }));
app.use('/api/v1/me/feed-seen', require('./feed-seen').router({ requireAuth }));
++ // One feed for all of OpenVibe (feed.item@1): Network reads Events and the sites' public APIs into
++ // feed_items, and serves the menu's /api/frame/feed and Community's /api/v1/feed from the same rows.
++ feedApi = require('./feed/api').createFeedApi({
++ db, config, privateKey, issuer: config.jwt.issuer, publicKey, presence: require('./frame/seen'), log: console,
++ });
++ app.locals.feedApi = feedApi;
++ app.use('/api/v1/feed', feedApi.router);
++ feedApi.start();
const homeScreen = require('./home-screen').createHomeScreen({ db });
app.use('/api/v1/me/home-screen', require('./home-screen').router({ homeScreen, requireAuth }));
// People's own controls (design 2026-11-11, piece 4): the active arrangement per surface and device,
@@@ -1047,7 +1056,7 @@ app.get('/@:username', rateLimit({ windowMs: 60_000, max: 120 }), publicProfiles
// /api/chrome is the old name, kept for copies of openvibe-shared older than 1.11.0.
// The menu's day counts (Report's daily activity): the strip's "today" counts reuse the same cached read.
const frameToday = require('./frame/today').createToday({ reportUrl: config.services.report.internalUrl });
--const frameService = await require('./frame/service').createFrameService(db, config, analytics, { privateKey, issuer: config.jwt.issuer, selfToken, requireAuth, today: () => frameToday.get() });
++const frameService = await require('./frame/service').createFrameService(db, config, analytics, { privateKey, issuer: config.jwt.issuer, selfToken, requireAuth, today: () => frameToday.get(), feed: feedApi });
app.use('/api/v1/frame/today', rateLimit({ windowMs: 60_000, max: 240 }), frameToday.router);
const frameLimit = rateLimit({ windowMs: 60_000, max: 240 });
app.use('/api/frame', frameLimit, frameService.router);
ddiff --git a/test/agent-schema.test.js b/test/agent-schema.test.js
iindex 9321ca84..f8b6b565 100644
---- a/test/agent-schema.test.js
++++ b/test/agent-schema.test.js
@@@ -44,7 +44,7 @@ const quiet = { log() {}, warn() {}, error() {} };
// The real directory: 0016 applies on top of 0015, once.
const first = await asOwner((db) => db.migrate({ dir: MIGRATIONS, log: quiet }));
-- assert.deepStrictEqual(first.applied.map((m) => [m.id, m.name, m.phase]), [['0016', 'agents', 'expand'], ['0017', 'agent_grants', 'expand'], ['0018', 'confirmations', 'expand'], ['0019', 'agent_budgets', 'expand'], ['0020', 'user_owned_trust', 'expand'], ['0021', 'services_portal', 'migrate'], ['0022', 'public_profiles', 'expand'], ['0023', 'project_secrets', 'expand'], ['0024', 'retired_theme_campfire', 'migrate'], ['0026', 'theme_history', 'expand'], ['0027', 'notification_quiet_digest', 'expand'], ['0028', 'theme_site_overrides', 'expand'], ['0029', 'site_prefs', 'expand'], ['0030', 'run_tokens', 'expand'], ['0031', 'dock_states', 'expand'], ['0032', 'oauth_public_clients', 'expand'], ['0033', 'consents', 'expand'], ['0034', 'look_dates', 'expand'], ['0035', 'dev_api_keys', 'expand'], ['0036', 'qr_signin', 'expand'], ['0037', 'analytics_bot_api', 'expand'], ['0038', 'sharing', 'expand'], ['0039', 'locker_keys', 'expand'], ['0040', 'ai_offers', 'expand'], ['0041', 'adult_confirmation', 'expand'], ['0042', 'chat_app_links', 'expand'], ['0043', 'connect_apps', 'expand'], ['0044', 'consent_refusals', 'expand'], ['0045', 'machine_pools', 'expand'], ['0046', 'menu_clicks', 'expand'], ['0047', 'presence', 'expand'], ['0048', 'menu_feeds', 'expand'], ['0049', 'oauth_session_public_clients', 'expand'], ['0050', 'oauth_token_session_families', 'expand'], ['0051', 'outbox_rejected_index', 'expand'], ['0052', 'public_person_pages', 'expand'], ['0053', 'link_send_relay', 'expand'], ['0054', 'account_emails', 'expand'], ['0055', 'companion_mask_scope', 'expand'], ['0056', 'companion_locker_scope', 'expand'], ['0057', 'companion_run_fill_scope', 'expand'], ['0058', 'link_egress', 'expand'], ['0060', 'companion_actor_scopes', 'expand'], ['0061', 'home_screen', 'expand'], ['0062', 'link_principal_ids', 'expand'], ['0063', 'retire_forgejo_client', 'migrate'], ['0064', 'history_timeline', 'expand'], ['0065', 'levels', 'expand'], ['0066', 'authenticator', 'expand'], ['0067', 'passkeys', 'expand'], ['0068', 'link_send_resume', 'expand'], ['0069', 'user_card', 'expand'], ['0070', 'coin_balance_reads', 'expand'], ['0071', 'handoff_tickets', 'expand'], ['0072', 'composer_prefs', 'expand'], ['0073', 'card_actions', 'expand'], ['0074', 'channel_points_reads', 'expand'], ['0075', 'test_accounts', 'expand'], ['0076', 'test_accounts_fix', 'expand'], ['0077', 'link_access_agents', 'expand'], ['0078', 'feed_seen', 'expand'], ['0079', 'ui_layouts', 'expand']]);
++ assert.deepStrictEqual(first.applied.map((m) => [m.id, m.name, m.phase]), [['0016', 'agents', 'expand'], ['0017', 'agent_grants', 'expand'], ['0018', 'confirmations', 'expand'], ['0019', 'agent_budgets', 'expand'], ['0020', 'user_owned_trust', 'expand'], ['0021', 'services_portal', 'migrate'], ['0022', 'public_profiles', 'expand'], ['0023', 'project_secrets', 'expand'], ['0024', 'retired_theme_campfire', 'migrate'], ['0026', 'theme_history', 'expand'], ['0027', 'notification_quiet_digest', 'expand'], ['0028', 'theme_site_overrides', 'expand'], ['0029', 'site_prefs', 'expand'], ['0030', 'run_tokens', 'expand'], ['0031', 'dock_states', 'expand'], ['0032', 'oauth_public_clients', 'expand'], ['0033', 'consents', 'expand'], ['0034', 'look_dates', 'expand'], ['0035', 'dev_api_keys', 'expand'], ['0036', 'qr_signin', 'expand'], ['0037', 'analytics_bot_api', 'expand'], ['0038', 'sharing', 'expand'], ['0039', 'locker_keys', 'expand'], ['0040', 'ai_offers', 'expand'], ['0041', 'adult_confirmation', 'expand'], ['0042', 'chat_app_links', 'expand'], ['0043', 'connect_apps', 'expand'], ['0044', 'consent_refusals', 'expand'], ['0045', 'machine_pools', 'expand'], ['0046', 'menu_clicks', 'expand'], ['0047', 'presence', 'expand'], ['0048', 'menu_feeds', 'expand'], ['0049', 'oauth_session_public_clients', 'expand'], ['0050', 'oauth_token_session_families', 'expand'], ['0051', 'outbox_rejected_index', 'expand'], ['0052', 'public_person_pages', 'expand'], ['0053', 'link_send_relay', 'expand'], ['0054', 'account_emails', 'expand'], ['0055', 'companion_mask_scope', 'expand'], ['0056', 'companion_locker_scope', 'expand'], ['0057', 'companion_run_fill_scope', 'expand'], ['0058', 'link_egress', 'expand'], ['0060', 'companion_actor_scopes', 'expand'], ['0061', 'home_screen', 'expand'], ['0062', 'link_principal_ids', 'expand'], ['0063', 'retire_forgejo_client', 'migrate'], ['0064', 'history_timeline', 'expand'], ['0065', 'levels', 'expand'], ['0066', 'authenticator', 'expand'], ['0067', 'passkeys', 'expand'], ['0068', 'link_send_resume', 'expand'], ['0069', 'user_card', 'expand'], ['0070', 'coin_balance_reads', 'expand'], ['0071', 'handoff_tickets', 'expand'], ['0072', 'composer_prefs', 'expand'], ['0073', 'card_actions', 'expand'], ['0074', 'channel_points_reads', 'expand'], ['0075', 'test_accounts', 'expand'], ['0076', 'test_accounts_fix', 'expand'], ['0077', 'link_access_agents', 'expand'], ['0078', 'feed_seen', 'expand'], ['0079', 'ui_layouts', 'expand'], ['0080', 'feed', 'expand']]);
assert.ok(await hasAgents());
assert.deepStrictEqual((await asOwner((db) => db.migrate({ dir: MIGRATIONS, log: quiet }))).applied, [], 'already applied');
// Expand only, IF NOT EXISTS throughout: running the file again changes nothing and fails nothing.
ddiff --git a/test/feed-sources.test.js b/test/feed-sources.test.js
nnew file mode 100644
iindex 00000000..5f242fa3
---- /dev/null
++++ b/test/feed-sources.test.js
@@@ -0,0 +1,131 @@
++'use strict';
++// The public API collectors, each against a recorded input: every one writes feed.item@1 rows with the
++// kind the feed draws, its own link and the source's numbers. No network: fetch is a map of URL → JSON.
++const assert = require('assert');
++const { getDb } = require('../server/db/database');
++const { createFeedSources } = require('../server/feed/sources');
++
++const NOW = Date.parse('2026-11-12T09:00:00.000Z');
++const A = 'usr_01JQ8Z6K6M0V8Y2Q3R4S5T6V7W';
++
++function fakeFetch(routes) {
++ return async (url) => {
++ const u = String(url);
++ for (const [needle, body] of Object.entries(routes)) {
++ if (u.includes(needle)) return { ok: true, status: 200, async json() { return typeof body === 'function' ? body(u) : body; },
++ async text() { return typeof body === 'function' ? body(u) : (typeof body === 'string' ? body : JSON.stringify(body)); } };
++ }
++ return { ok: false, status: 404, async json() { return {}; } };
++ };
++}
++
++(async () => {
++ const db = getDb();
++ const rows = () => db.prepare('SELECT * FROM feed_items ORDER BY id').all();
++ const kinds = async () => (await rows()).map((r) => r.kind);
++ const make = (routes, presence = null) => createFeedSources({
++ db, now: () => NOW, log: { warn() {} },
++ config: { networkUrl: 'https://openvibe.network', feed: {
++ communityUrl: 'https://community.test', todayUrl: 'https://today.test', questUrl: 'https://quest.test',
++ playUrl: 'https://play.test', memeUrl: 'https://meme.test', picsUrl: 'https://pics.test', liveUrl: 'https://live.test',
++ spaceUrl: 'https://space.test', workUrl: 'https://work.test', dealsUrl: 'https://deals.test',
++ newsUrl: 'https://news.test', reportUrl: 'https://report.test', blogUrl: 'https://blog.test', wikiUrl: 'https://wiki.test',
++ } },
++ fetchImpl: fakeFetch(routes), presence,
++ });
++
++ // Community's direct public activity (no Pulse store).
++ let src = make({ '/api/v1/activity': { items: [
++ { id: '2', type: 'thread', title: 'A public thread', text: 'hello', url: 'https://openvibe.community/forum/status/t', actor: A, at: new Date(NOW - 60_000).toISOString(), counts: { likes: 3, comments: 1 } },
++ { id: 'shot', type: 'paste', title: 'A screenshot', url: 'https://openvibe.community/p/shot', picture_url: 'https://openvibe.media/t/shot.jpg', at: new Date(NOW - 120_000).toISOString() },
++ ] } });
++ assert.strictEqual(await src.communityActivity(), 2);
++ let row = (await rows()).find((r) => r.source_id === '2');
++ assert.deepStrictEqual([row.kind, row.actor_subject, row.counts.likes], ['post', A, 3]);
++ assert.strictEqual((await rows()).find((r) => r.source_id === 'shot').kind, 'pic');
++
++ // Today's calendar.
++ src = make({ '/api/v1/events': { events: [{ id: 'tev_1', title: 'A stream', starts_at: new Date(NOW + 86400000).toISOString(), created_at: new Date(NOW).toISOString(), visibility: 'public', url: 'https://openvibe.today/e/tev_1', owner_subject: A }] } });
++ assert.strictEqual(await src.todayEvents(), 1);
++ assert.strictEqual((await rows()).find((r) => r.source_id === 'tev_1').kind, 'event');
++
++ // Quest badges for a recent actor.
++ src = make({ '/badges': { badges: [{ id: 'first-broadcast', name: 'First Broadcast', awarded_at: new Date(NOW - 3600000).toISOString() }] } });
++ await src.questBadges();
++ assert.strictEqual((await rows()).some((r) => r.kind === 'quest'), true);
++
++ // Play achievements.
++ src = make({ '/api/v1/achievements/feed': { items: [{ id: 'a1', player: A, title: 'scraplandia', title_name: 'Scraplandia', name: 'First landing', description: '', unlocked_at: new Date(NOW - 3600000).toISOString() }], next: null } });
++ assert.strictEqual(await src.playAchievements(), 1);
++ assert.strictEqual((await rows()).find((r) => r.source_id === `${A}:scraplandia:a1`).kind, 'game');
++
++ // Meme entries.
++ src = make({ '/api/v1/entries': { entries: [{ id: 'mem_1', slug: 'a-meme', title: 'A meme', summary: 'with receipts', visibility: 'public', published_at: new Date(NOW - 3600000).toISOString() }], next_cursor: null } });
++ assert.strictEqual(await src.memeEntries(), 1);
++ assert.strictEqual((await rows()).find((r) => r.source_id === 'a-meme').kind, 'meme');
++
++ // Pics gallery.
++ src = make({ '/api/v1/gallery': { pictures: [{ id: 'pic_01M6SV8E153FM2T70WAF14TB4W', listed: true, title: 'A picture', author: 'Alex', links: { page: 'https://openvibe.pics/p/1', thumb: 'https://openvibe.media/t/1.jpg' }, published_at: new Date(NOW - 3600000).toISOString(), width: 1200, height: 800 }], next: null } });
++ assert.strictEqual(await src.publicPictures(), 1);
++ assert.strictEqual((await rows()).find((r) => r.kind === 'pic' && r.source_service === 'pics').source_service, 'pics');
++
++ // Live: on air, recordings and clips; the Rundown episodes.
++ src = make({
++ '/api/streams': { streams: [{ id: 9, username: 'alex', display_name: 'Alex', title: 'a stream', started_at: '2026-11-12 08:55:00' }] },
++ '/api/content/feed': { sources: { vods: 'ok', clips: 'ok' }, items: [
++ { kind: 'vod', id: 'r1', href: 'https://openvibe.video/watch?v=r1', title: 'A stream', created_at: '2026-11-12 07:00:00', channel: { username: 'alex', display_name: 'Alex' }, duration_seconds: 3600 },
++ { kind: 'clip', id: 'c1', href: 'https://openvibe.live/clip/c1', title: 'A clip', created_at: '2026-11-12 08:00:00', channel: { username: 'alex' } },
++ ], next: null },
++ '/api/rundown/streams': { streams: [{ stream_id: 3005, user: { username: 'alex', display_name: 'Alex' }, title: 'An episode', started_at: '2026-11-11 20:00:00', live: false, episode: { title: 'The case for a feed', logline: 'A logline', audience: 'public', aired_at: '2026-11-11 21:00:00' }, thumbnail_url: 'https://openvibe.media/t/e.jpg' }], next: null },
++ });
++ assert.ok(await src.publicContent() >= 2);
++ assert.ok((await rows()).some((r) => r.kind === 'live' && r.live_now === true));
++ assert.ok((await rows()).some((r) => r.kind === 'video'));
++ assert.ok((await rows()).some((r) => r.kind === 'clip'));
++ await src.rundownEpisodes();
++ assert.ok((await rows()).some((r) => r.kind === 'rundown'));
++
++ // "312 people watched …": Network's own ended-stream analytics.
++ await db.prepare(`INSERT INTO creator_streams (stream_id, creator_subject, title, started_at, ended_at, duration_seconds, peak_viewers, received_at)
++ VALUES (?, ?, ?, ?, ?, ?, ?, ?)`).run(4242, A, 'The evening stream', new Date(NOW - 7200000).toISOString(), new Date(NOW - 3600000).toISOString(), 3600, 312, new Date(NOW).toISOString());
++ assert.strictEqual(await src.watchedTogether(), 1);
++ const watched = (await rows()).find((r) => r.source_id === '4242');
++ assert.deepStrictEqual([watched.kind, watched.counts.watchers], ['stream', 312]);
++
++ // Spaces.
++ src = make({ '/api/v1/apps/gallery': { spaces: [{ id: 'spc_01M6SV8E153FM2T70WAF14TB4W', name: 'A space', url: 'https://openvibe.space/s/1', published_at: NOW - 3600000, owner_name: 'Alex' }], next_cursor: null } });
++ assert.strictEqual(await src.publicSpaces(), 1);
++ assert.strictEqual((await rows()).find((r) => r.source_service === 'space').kind, 'space');
++
++ // Work jobs, Deals offers, News stories.
++ src = make({ '/api/v1/jobs': { jobs: [{ id: 'job_xnsvvqkfxe1kv3hcnkb0g17wfs', title: 'GIS engineer', company: 'Esri', location: 'Cardiff', remote: false, url: 'https://openvibe.work/j/1', excerpt: 'Maps', first_seen_at: new Date(NOW).toISOString() }] } });
++ assert.strictEqual(await src.workJobs(), 1);
++ assert.strictEqual((await rows()).find((r) => r.kind === 'work').source_service, 'work');
++ src = make({ '/api/v1/offers': { offers: [{ id: 'dof_1', title: 'A deal', description: '20% off', url: 'https://openvibe.deals/d/1', created_at: new Date(NOW).toISOString(), votes: 4 }] } });
++ assert.strictEqual(await src.dealOffers(), 1);
++ assert.strictEqual((await rows()).find((r) => r.kind === 'deal').counts.likes, 4);
++ src = make({ '/api/v1/stories': { stories: [{ id: 'sty_1', title: 'A story', summary: 'Sourced', url: 'https://openvibe.news/s/1', status: 'published', published_at: new Date(NOW).toISOString() }] } });
++ assert.strictEqual(await src.newsStories(), 1);
++ assert.strictEqual((await rows()).find((r) => r.kind === 'news').source_service, 'news');
++
++ // Blog RSS and recent Wiki pages: the sources that only have a public feed/search.
++ src = make({ '/feed.xml': `<rss><channel><item><title>A blog post</title><link>https://openvibe.blog/p/a</link>`
++ + `<pubDate>Wed, 12 Nov 2026 08:00:00 GMT</pubDate><description>What changed this week.</description></item></channel></rss>` });
++ assert.strictEqual(await src.blogPosts(), 1);
++ assert.strictEqual((await rows()).find((r) => r.kind === 'update').source_service, 'blog');
++ src = make({ '/api/v1/search': { results: [{ id: 'pg_01M6SV8E153FM2T70WAF14TB4W', state: 'published', visibility: 'public', title: 'A wiki page',
++ url: 'https://openvibe.wiki/w/openvibe/feed', revision_published_at: new Date(NOW - 3600000).toISOString() }] } });
++ assert.strictEqual(await src.wikiPages(), 1);
++ assert.strictEqual((await rows()).find((r) => r.kind === 'wiki').source_service, 'wiki');
++
++ // Report: the online peak from presence, then the day's report card with its trend.
++ src = make({ '/api/v1/stats': { totals: { pageviews: 363633, change_pct: 44, sites_open: 56, accounts: 506 } } },
++ { snapshot: () => ({ members: new Map([[A, { at: NOW, site: 'live' }]]), guests: 3 }) });
++ assert.strictEqual(await src.samplePeak(), 1);
++ assert.ok(await src.reportCard() >= 1);
++ const report = (await rows()).find((r) => r.kind === 'report');
++ assert.match(report.title, /people online|page views/);
++ assert.ok(report.trend || report.counts.people >= 1);
++
++ console.log('feed sources: community, today, quest, play, meme, pics, live, rundown, watch, space, work, deals, news and report map to feed.item@1');
++})().catch((err) => { console.error(err); process.exit(1); });
ddiff --git a/test/feed.test.js b/test/feed.test.js
nnew file mode 100644
iindex 00000000..cf8ef77a
---- /dev/null
++++ b/test/feed.test.js
@@@ -0,0 +1,104 @@
++'use strict';
++// One feed for all of OpenVibe (feed.item@1): each source mapping, the folding of a run, the Followers
++// privacy rule, the For you / Following ranking, and the menu and the feed returning the same ids.
++const assert = require('assert');
++const express = require('express');
++const contracts = require('openvibe-contracts');
++const { getDb } = require('../server/db/database');
++const store = require('../server/feed/store');
++const { eventToItems } = require('../server/feed/events');
++const { cardOf, foldCards, rankCards, searchMatches } = require('../server/feed/card');
++const { createFeedApi } = require('../server/feed/api');
++const { createFrameService } = require('../server/frame/service');
++
++const envelope = (type, payload, over = {}) => ({
++ event_id: 'evt_01JQ8Z6K6M0V8Y2Q3R4S5T6V7W', event_type: type, version: 1, source: String(type).split('.')[0],
++ actor: { type: 'user', id: 'usr_01JQ8Z6K6M0V8Y2Q3R4S5T6V7W' }, timestamp: '2026-11-12T09:00:00.000Z',
++ visibility: 'public', subject: { type: 'user', id: 'usr_01JQ8Z6K6M0V8Y2Q3R4S5T6V7W' }, payload, ...over,
++});
++
++(async () => {
++ const db = getDb();
++ const A = 'usr_01JQ8Z6K6M0V8Y2Q3R4S5T6V7W';
++ const B = 'usr_01JQ8Z6K6M0V8Y2Q3R4S5T6V7X';
++ const now = Date.now();
++ const iso = (ms) => new Date(ms).toISOString();
++
++ // ── every Events mapping ─────────────────────────────────
++ const kindOf = (type, payload, over = {}) => {
++ const out = eventToItems(envelope(type, payload, over));
++ assert.ok(out.items.length, `${type} maps to an item`);
++ return out.items[0].kind;
++ };
++ assert.strictEqual(kindOf('live.stream.started', { channel: { username: 'alex', display_name: 'Alex', url: 'https://openvibe.live/@alex', subject: { type: 'user', id: A } }, title: 'a stream' }), 'live');
++ assert.strictEqual(kindOf('mediahub.video.published', { video_id: 'v1', title: 'A video', canonical_url: 'https://openvibe.video/watch?v=v1' }), 'video');
++ assert.strictEqual(kindOf('mediahub.picture.uploaded', { picture_id: 'p1', url: 'https://openvibe.media/p/1', thumb: 'https://openvibe.media/t/1.jpg' }), 'pic');
++ assert.strictEqual(kindOf('games.world.published', { world_id: 'w1', title: 'A world', url: 'https://openvibe.games/w/1' }), 'game');
++ assert.strictEqual(kindOf('blog.post.published', { post_id: 'b1', url: 'https://openvibe.blog/p/1', blog: { handle: 'openvibe' } }), 'update');
++ assert.strictEqual(kindOf('wiki.page.published', { page_id: 'w1', url: 'https://openvibe.wiki/space/page', title: 'A page' }), 'wiki');
++ assert.strictEqual(kindOf('news.story.published', { story_id: 's1', url: 'https://openvibe.news/s/1', topic: 'feeds' }), 'news');
++ assert.strictEqual(kindOf('chat.room.message.created', { room: { slug: 'feed', name: 'Feed' }, created_at: '2026-11-12T09:00:00.000Z' }), 'chat');
++ assert.strictEqual(kindOf('deals.offer.created', { offer_id: 'd1', title: 'A deal', url: 'https://openvibe.deals/d/1' }), 'deal');
++ assert.strictEqual(kindOf('space.deploy.activated', { deploy_id: 'd1', name: 'A space', url: 'https://openvibe.space/s/1' }), 'space');
++ assert.deepStrictEqual(eventToItems(envelope('live.stream.started', { channel: { username: 'alex', url: 'https://openvibe.live/@alex' } }, { visibility: 'internal' })).items, [], 'a non-public event never enters the feed');
++ assert.deepStrictEqual(eventToItems(envelope('wiki.page.deleted', { page_id: 'w1' })).remove[0], { service: 'wiki', type: 'post', id: 'w1' });
++
++ // ── grouping: a run becomes one card, work per hour ──────
++ const card = (over) => cardOf({ source_service: 'live', source_type: 'recording', source_id: over.id || 'r', kind: 'video', title: over.title || 'Recording', text: '', url: over.url || 'https://openvibe.video/watch?v=1', site: 'live', site_name: 'Live', actor_subject: over.subject || A, counts: {}, occurred_at: over.at || iso(now), media_ratio: null }, { name: 'Alex', avatar: null });
++ const run = foldCards([card({ at: iso(now - 3 * 60000) }), card({ at: iso(now - 2 * 60000) }), card({ at: iso(now - 60000) })]);
++ assert.strictEqual(run.length, 1);
++ assert.strictEqual(run[0].group.count, 3);
++ assert.match(run[0].title, /Alex added 3 videos/);
++ const one = foldCards([card({ at: iso(now - 3600000) }), card({ at: iso(now) })]);
++ assert.strictEqual(one.length, 2, 'an hour apart is not a run');
++ const jobs = foldCards([1, 2, 3].map((n) => cardOf({ source_service: 'work', source_type: 'job', source_id: `j${n}`, kind: 'work', title: `Job ${n}`, url: 'https://openvibe.work/jobs', site: 'work', site_name: 'Work', counts: {}, occurred_at: '2026-11-12T09:00:00.000Z' })));
++ assert.strictEqual(jobs.length, 1);
++ assert.strictEqual(jobs[0].title, '3 new jobs this hour');
++ assert.strictEqual(jobs[0].group.count, 3);
++
++ // ── ranking: live, followed, fresh, then older than a week ─
++ const live = cardOf({ source_service: 'live', source_type: 'stream', source_id: 'l', kind: 'live', title: 'Live', url: 'https://openvibe.live/@alex', site: 'live', site_name: 'Live', actor_subject: B, live_now: true, occurred_at: iso(now - 86400000) });
++ const followed = { ...card({ at: iso(now - 3600000) }), subject: B, id: 'followed' };
++ const fresh = { ...card({ at: iso(now - 60000) }), subject: A, id: 'fresh' };
++ const stale = { ...card({ at: iso(now - 8 * 86400000) }), subject: A, id: 'stale' };
++ const ranked = rankCards([stale, followed, fresh, live], { followed: new Set([B]), now });
++ assert.strictEqual(ranked[0].id, live.id, 'on air leads');
++ assert.strictEqual(ranked[1].id, followed.id, 'a followed creator comes next');
++ assert.strictEqual(ranked[ranked.length - 1].id, stale.id, 'older than seven days stays below the recent ones');
++ assert.ok(searchMatches({ ...fresh, kind: 'clip', title: 'A clip', text: '' }, 'new clips'));
++
++ // ── the API: signed out is public only; the wall has the same ids ─
++ await store.upsertItem(db, { source_service: 'community', source_type: 'thread', source_id: '1', kind: 'post', title: 'Public post', text: 'hello', url: 'https://openvibe.community/forum/status/public', site: 'community', site_name: 'Community', actor_subject: A, counts: { likes: 2 }, occurred_at: iso(now - 120000) });
++ await store.upsertItem(db, { source_service: 'community', source_type: 'thread', source_id: '2', kind: 'post', title: 'Followers post', text: 'hello', url: 'https://openvibe.community/forum/status/followers', site: 'community', site_name: 'Community', actor_subject: A, visibility: 'followers', occurred_at: iso(now - 60000) });
++ const pub = await store.listItems(db, { limit: 20 });
++ assert.deepStrictEqual(pub.rows.map((r) => r.title), ['Public post'], 'a followers item never reaches a signed-out read');
++ const author = await store.listItems(db, { limit: 20, viewerSubject: A, followed: [] });
++ assert.deepStrictEqual(author.rows.map((r) => r.title).sort(), ['Followers post', 'Public post'], 'its author sees it');
++ const other = await store.listItems(db, { limit: 20, viewerSubject: B, followed: [] });
++ assert.deepStrictEqual(other.rows.map((r) => r.title), ['Public post'], 'another person does not');
++
++ const feed = createFeedApi({ db, config: {}, publicKey: '', startCollectors: false, now: () => now });
++ const app = express();
++ app.use('/api/v1/feed', feed.router);
++ app.use('/api/frame', (await createFrameService(db, { services: {} }, null, { feed })).router);
++ const server = app.listen(0, '127.0.0.1');
++ await new Promise((r) => server.once('listening', r));
++ const base = `http://127.0.0.1:${server.address().port}`;
++ const get = async (p, headers) => { const r = await fetch(`${base}${p}`, { headers }); return { r, d: await r.json() }; };
++
++ let { r, d } = await get('/api/v1/feed?limit=20');
++ assert.strictEqual(r.status, 200);
++ assert.deepStrictEqual(d.items.map((x) => x.title), ['Public post'], 'signed out works, public items only');
++ assert.ok(d.items.every((x) => /^f:/.test(x.id)), 'ids are stable source ids');
++ for (const item of d.items) assert.ok(contracts.validate('feed.item@1', item).valid, `feed.item@1: ${JSON.stringify(contracts.validate('feed.item@1', item).errors)}`);
++ const menu = await get('/api/frame/feed?limit=20');
++ assert.deepStrictEqual(menu.d.items.map((x) => x.id), d.items.map((x) => x.id), 'the menu and the feed return the same ids');
++ const wall = await get('/api/v1/feed/wall?limit=20');
++ assert.deepStrictEqual(wall.d.items.map((x) => x.id), d.items.map((x) => x.id), 'the wall shows the same rows and ids');
++ ({ r } = await get('/api/v1/feed?following=1'));
++ assert.strictEqual(r.status, 401, 'Following needs a session');
++ ({ d } = await get('/api/v1/feed/search?q=public'));
++ assert.strictEqual(d.items[0].title, 'Public post');
++ server.close();
++ console.log('feed: mappings, grouping, privacy, ranking and the menu/feed ids pass');
++})().catch((err) => { console.error(err); process.exit(1); });
ddiff --git a/test/security-ssrf.test.js b/test/security-ssrf.test.js
iindex d2e996e9..de29dd4b 100644
---- a/test/security-ssrf.test.js
++++ b/test/security-ssrf.test.js
@@@ -122,6 +122,8 @@ const REVIEWED = {
'server/frame/ask.js fetch': [2, 'the composer\'s Ask: Search (OV_SEARCH_INTERNAL_URL) fixed /api/v1/query path with checked scopes and the person\'s Bearer, and AI (OV_AI_INTERNAL_URL) fixed /api/v1/runs?stream=1 path with the person\'s Bearer; both operator-configured internal URLs, no request value names a host, redirects refused, 30-150 s timeouts'],
'server/admin/analytics-routes.js fetch': [2, 'staff analytics: the services\' internal URLs from the registry (owner/admin set), and Insights\' Prometheus (PROMETHEUS_URL, loopback by default) with fixed queries; never a request value'],
'server/admin/events-ops.js fetch': [1, 'OV_EVENTS_INTERNAL_URL (operator config) for staff DLQ views'],
++ 'server/feed/events.js fetch': [1, 'OpenVibe.Events at OV_EVENTS_INTERNAL_URL (operator config), the fixed /api/v1/events pull path with an opaque cursor and the service token; no request value names a host'],
++ 'server/feed/sources.js fetch': [3, 'the public APIs and feeds of OpenVibe.Community, Today, Quest, Play, Meme, Pics, Live, Space, Work, Deals, News, Blog and Wiki at operator-configured bases (OV_*_URL; Community and Live default to their loopback internal URL), fixed paths with encoded, validated ids; never a request value'],
'server/admin/routes.js fetch': [1, 'registry refresh: POSTs to the service refresh targets resolved from the registry (admin-set service URLs, staff only)'],
'server/auth/discord-link.js fetch': [2, 'fixed Discord API host (discord.com/api)'],
'server/developer/event-relay.js sdk client': [4, 'OV_EVENTS_INTERNAL_URL (operator config)'],
ddiff --git a/test/user-owned-trust.test.js b/test/user-owned-trust.test.js
iindex ea6aaa3a..5c20f991 100644
---- a/test/user-owned-trust.test.js
++++ b/test/user-owned-trust.test.js
@@@ -56,7 +56,7 @@ const quiet = { log() {}, warn() {}, error() {} };
// ── 0020 applies on top of 0019, once, and rewrites nothing.
const first = await asOwner((o) => o.migrate({ dir: MIGRATIONS, log: quiet }));
-- assert.deepStrictEqual(first.applied.map((m) => [m.id, m.name, m.phase]), [['0020', 'user_owned_trust', 'expand'], ['0021', 'services_portal', 'migrate'], ['0022', 'public_profiles', 'expand'], ['0023', 'project_secrets', 'expand'], ['0024', 'retired_theme_campfire', 'migrate'], ['0026', 'theme_history', 'expand'], ['0027', 'notification_quiet_digest', 'expand'], ['0028', 'theme_site_overrides', 'expand'], ['0029', 'site_prefs', 'expand'], ['0030', 'run_tokens', 'expand'], ['0031', 'dock_states', 'expand'], ['0032', 'oauth_public_clients', 'expand'], ['0033', 'consents', 'expand'], ['0034', 'look_dates', 'expand'], ['0035', 'dev_api_keys', 'expand'], ['0036', 'qr_signin', 'expand'], ['0037', 'analytics_bot_api', 'expand'], ['0038', 'sharing', 'expand'], ['0039', 'locker_keys', 'expand'], ['0040', 'ai_offers', 'expand'], ['0041', 'adult_confirmation', 'expand'], ['0042', 'chat_app_links', 'expand'], ['0043', 'connect_apps', 'expand'], ['0044', 'consent_refusals', 'expand'], ['0045', 'machine_pools', 'expand'], ['0046', 'menu_clicks', 'expand'], ['0047', 'presence', 'expand'], ['0048', 'menu_feeds', 'expand'], ['0049', 'oauth_session_public_clients', 'expand'], ['0050', 'oauth_token_session_families', 'expand'], ['0051', 'outbox_rejected_index', 'expand'], ['0052', 'public_person_pages', 'expand'], ['0053', 'link_send_relay', 'expand'], ['0054', 'account_emails', 'expand'], ['0055', 'companion_mask_scope', 'expand'], ['0056', 'companion_locker_scope', 'expand'], ['0057', 'companion_run_fill_scope', 'expand'], ['0058', 'link_egress', 'expand'], ['0060', 'companion_actor_scopes', 'expand'], ['0061', 'home_screen', 'expand'], ['0062', 'link_principal_ids', 'expand'], ['0063', 'retire_forgejo_client', 'migrate'], ['0064', 'history_timeline', 'expand'], ['0065', 'levels', 'expand'], ['0066', 'authenticator', 'expand'], ['0067', 'passkeys', 'expand'], ['0068', 'link_send_resume', 'expand'], ['0069', 'user_card', 'expand'], ['0070', 'coin_balance_reads', 'expand'], ['0071', 'handoff_tickets', 'expand'], ['0072', 'composer_prefs', 'expand'], ['0073', 'card_actions', 'expand'], ['0074', 'channel_points_reads', 'expand'], ['0075', 'test_accounts', 'expand'], ['0076', 'test_accounts_fix', 'expand'], ['0077', 'link_access_agents', 'expand'], ['0078', 'feed_seen', 'expand'], ['0079', 'ui_layouts', 'expand']]);
++ assert.deepStrictEqual(first.applied.map((m) => [m.id, m.name, m.phase]), [['0020', 'user_owned_trust', 'expand'], ['0021', 'services_portal', 'migrate'], ['0022', 'public_profiles', 'expand'], ['0023', 'project_secrets', 'expand'], ['0024', 'retired_theme_campfire', 'migrate'], ['0026', 'theme_history', 'expand'], ['0027', 'notification_quiet_digest', 'expand'], ['0028', 'theme_site_overrides', 'expand'], ['0029', 'site_prefs', 'expand'], ['0030', 'run_tokens', 'expand'], ['0031', 'dock_states', 'expand'], ['0032', 'oauth_public_clients', 'expand'], ['0033', 'consents', 'expand'], ['0034', 'look_dates', 'expand'], ['0035', 'dev_api_keys', 'expand'], ['0036', 'qr_signin', 'expand'], ['0037', 'analytics_bot_api', 'expand'], ['0038', 'sharing', 'expand'], ['0039', 'locker_keys', 'expand'], ['0040', 'ai_offers', 'expand'], ['0041', 'adult_confirmation', 'expand'], ['0042', 'chat_app_links', 'expand'], ['0043', 'connect_apps', 'expand'], ['0044', 'consent_refusals', 'expand'], ['0045', 'machine_pools', 'expand'], ['0046', 'menu_clicks', 'expand'], ['0047', 'presence', 'expand'], ['0048', 'menu_feeds', 'expand'], ['0049', 'oauth_session_public_clients', 'expand'], ['0050', 'oauth_token_session_families', 'expand'], ['0051', 'outbox_rejected_index', 'expand'], ['0052', 'public_person_pages', 'expand'], ['0053', 'link_send_relay', 'expand'], ['0054', 'account_emails', 'expand'], ['0055', 'companion_mask_scope', 'expand'], ['0056', 'companion_locker_scope', 'expand'], ['0057', 'companion_run_fill_scope', 'expand'], ['0058', 'link_egress', 'expand'], ['0060', 'companion_actor_scopes', 'expand'], ['0061', 'home_screen', 'expand'], ['0062', 'link_principal_ids', 'expand'], ['0063', 'retire_forgejo_client', 'migrate'], ['0064', 'history_timeline', 'expand'], ['0065', 'levels', 'expand'], ['0066', 'authenticator', 'expand'], ['0067', 'passkeys', 'expand'], ['0068', 'link_send_resume', 'expand'], ['0069', 'user_card', 'expand'], ['0070', 'coin_balance_reads', 'expand'], ['0071', 'handoff_tickets', 'expand'], ['0072', 'composer_prefs', 'expand'], ['0073', 'card_actions', 'expand'], ['0074', 'channel_points_reads', 'expand'], ['0075', 'test_accounts', 'expand'], ['0076', 'test_accounts_fix', 'expand'], ['0077', 'link_access_agents', 'expand'], ['0078', 'feed_seen', 'expand'], ['0079', 'ui_layouts', 'expand'], ['0080', 'feed', 'expand']]);
assert.deepStrictEqual((await asOwner((o) => o.migrate({ dir: MIGRATIONS, log: quiet }))).applied, [], 'already applied');
await asOwner((o) => o.exec(fs.readFileSync(path.join(MIGRATIONS, '0020_user_owned_trust.sql'), 'utf8'))); // safe to run twice
// 0020–0061 keep every row; 0062 moves the principal ids to lnk_ in place and references follow.