openvibers/openvibe.live Commits
Follows, ADR-030 step 4 in Live. With FOLLOWS_AUTHORITY=network, both follow buttons record the change on Network first (PUT/DELETE /internal/follows/channel/<subject>, Live's service token, network.follows.write) and write Live's row only when Network took it; when Network cannot answer the button fails with 503 instead of drifting. A side without a Network subject stays Live-only. network.follow.created/.deleted (added to the --network subscription) keep Live's follows table as a projection, applied in revision order (follow_projection_revisions). Unset: the buttons write Live only, as before. test/network-follows.test.js
Changes
@@@ -31,7 +31,7 @@ const fs = require('fs');const TOPICS = ['media.vod.*', 'media.clip.*', 'media.storage.*'];const DEFAULT_ENDPOINT = 'http://127.0.0.1:3000/internal/media-events';--const NETWORK_TOPICS = ['network.user.token_valid_after', 'network.user.updated'];++const NETWORK_TOPICS = ['network.user.token_valid_after', 'network.user.updated', 'network.follow.created', 'network.follow.deleted'];const NETWORK_ENDPOINT = 'http://127.0.0.1:3000/internal/network-events';const DEFAULT_LIVE_ENV = '/etc/openvibe/live.env';@@@ -9,6 +9,7 @@** token_valid_after the person's older tokens are refused (./revocations.js)* updated the person's profile, role and ban (./subject-projection.js, WS-B task 2)++ * follow.created/.deleted Live's follows table as a projection of Network's graph (../social/network-follows.js)** Applying is idempotent (a cutoff only ever moves forward; a profile revision only up), so a redelivery* is a no-op without an inbox. Anything signed but not ours to act on is acknowledged (204).@@@ -25,8 +26,11 @@ function secret() { return process.env.LIVE_EVENTS_SECRET || process.env.MEDIA_E/** Apply one envelope; returns 'revoked' | 'unchanged' | 'updated' | 'stale' | 'ignored:<why>'. */function apply(ev) {if (!ev || !EVENT_ID_RE.test(String(ev.event_id || ''))) return 'ignored:envelope';-- if (ev.event_type !== 'network.user.token_valid_after' && ev.event_type !== 'network.user.updated') return 'ignored:type';++ const follow = ev.event_type === 'network.follow.created' || ev.event_type === 'network.follow.deleted';++ if (ev.event_type !== 'network.user.token_valid_after' && ev.event_type !== 'network.user.updated' && !follow) return 'ignored:type';if (ev.source !== 'network') return 'ignored:source';++ // The follow graph is Network's (ADR-030); Live's follows table is its projection.++ if (follow) return require('../social/network-follows').apply(ev);const p = ev.payload && typeof ev.payload === 'object' ? ev.payload : {};if (ev.event_type === 'network.user.updated') return require('./subject-projection').apply(p, { notify: notifyChat });const subject = p.subject && p.subject.id;@@@ -44,7 +48,7 @@ function handler(req, res) {if (!delivery) { stats.refused++; return res.status(401).json({ error: 'bad signature' }); }stats.received++;const out = apply(delivery.event);-- if (out === 'revoked') stats.revoked++; else if (out === 'unchanged' || out === 'stale') stats.unchanged++; else if (out === 'updated') stats.profiles++; else stats.ignored++;++ if (out === 'revoked') stats.revoked++; else if (out === 'unchanged' || out === 'stale') stats.unchanged++; else if (out === 'updated' || out === 'followed' || out === 'unfollowed') stats.profiles++; else stats.ignored++;res.status(204).end();}nnew file mode 100644@@@ -0,0 +1,95 @@++'use strict';++/**++ * Follows on OpenVibe.Network (ADR-030; roadmap WS-E task 4). Network owns the follow graph, keyed by subjects;++ * Live's `follows` table is its projection, kept for Live's own reads (channel pages, counts, sort orders).++ *++ * writeThrough(followerId, streamerId, following) step 4: with FOLLOWS_AUTHORITY=network, a follow button++ * records the change on Network first (PUT/DELETE /internal/follows/channel/<streamer subject>, Live's++ * service token, network.follows.write). The route writes Live's row only when Network took it; if++ * Network cannot answer, the button fails (503) rather than letting the two drift. A side with no++ * Network subject yet (an account Network never told us about) stays Live-only, as before.++ * apply(event) network.follow.created / .deleted from OpenVibe.Events (server/auth/network-events.js):++ * follows made anywhere else (my.openvibe.network, another product, the API) reach Live's table.++ * Idempotent and ordered: the highest revision applied per pair is kept (follow_projection_revisions),++ * and an older one is `stale`. A pair whose side has no Live account is `ignored:unmapped`.++ *++ * Unset FOLLOWS_AUTHORITY: the buttons write Live's table only (the state before step 4); events still apply.++ */++const db = require('../db/database');++++const NETWORK_INTERNAL_URL = (process.env.OV_NETWORK_INTERNAL_URL || 'http://127.0.0.1:4000').replace(/\/+$/, '');++const SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;++const stats = { written: 0, local_only: 0, failed: 0, applied: 0, stale: 0, ignored: 0 };++++const authority = () => (process.env.FOLLOWS_AUTHORITY === 'network' ? 'network' : 'live');++++function subjectOf(liveUserId) {++ const r = db.getDb().prepare("SELECT subject_id FROM linked_accounts WHERE service = 'network' AND user_id = ?").get(liveUserId);++ return r && SUBJECT_RE.test(String(r.subject_id || '')) ? r.subject_id : null;++}++function userOf(subject) {++ const r = db.getDb().prepare("SELECT user_id FROM linked_accounts WHERE service = 'network' AND subject_id = ? ORDER BY id LIMIT 1").get(subject);++ return r ? r.user_id : null;++}++++/**++ * Record a follow change on Network before Live writes its row. → { ok: true, network: true } when Network took it,++ * { ok: true, network: false, reason } when it stays Live-only (authority live, or a side without a subject),++ * { ok: false, error } when Network could not take it (the caller answers 503 and changes nothing).++ */++async function writeThrough(followerId, streamerId, following, { fetchImpl = globalThis.fetch } = {}) {++ if (authority() !== 'network') return { ok: true, network: false, reason: 'authority_live' };++ const follower = subjectOf(followerId);++ const target = subjectOf(streamerId);++ if (!follower || !target) { stats.local_only++; return { ok: true, network: false, reason: 'no_subject' }; }++ try {++ const principal = require('../net/network-principal');++ const headers = { Accept: 'application/json', ...(await principal.serviceHeaders('openvibe.network')) };++ const url = `${NETWORK_INTERNAL_URL}/internal/follows/channel/${target}`;++ const res = following++ ? await fetchImpl(url, { method: 'PUT', headers: { ...headers, 'Content-Type': 'application/json' }, body: JSON.stringify({ follower }), signal: AbortSignal.timeout(5000) })++ : await fetchImpl(`${url}?follower=${follower}`, { method: 'DELETE', headers, signal: AbortSignal.timeout(5000) });++ if (res.status === 401) principal.invalidate('openvibe.network');++ if (res.status < 200 || res.status >= 300) { stats.failed++; return { ok: false, error: `Network answered ${res.status}` }; }++ stats.written++;++ return { ok: true, network: true };++ } catch (err) {++ stats.failed++;++ return { ok: false, error: err.message };++ }++}++++function ensureTable() {++ db.getDb().exec(`CREATE TABLE IF NOT EXISTS follow_projection_revisions (++ follower_subject TEXT NOT NULL,++ target_subject TEXT NOT NULL,++ revision INTEGER NOT NULL,++ PRIMARY KEY (follower_subject, target_subject)++ )`);++}++++/** Apply network.follow.created / .deleted. → 'followed' | 'unfollowed' | 'stale' | 'ignored:<why>' */++function apply(ev) {++ const p = ev && ev.payload && typeof ev.payload === 'object' ? ev.payload : {};++ if (p.target_type !== 'channel' || !SUBJECT_RE.test(String(p.follower || '')) || !SUBJECT_RE.test(String(p.target_id || '')) || !Number.isInteger(p.revision)) { stats.ignored++; return 'ignored:payload'; }++ ensureTable();++ const d = db.getDb();++ return d.transaction(() => {++ const prev = d.prepare('SELECT revision FROM follow_projection_revisions WHERE follower_subject = ? AND target_subject = ?').get(p.follower, p.target_id);++ if (prev && prev.revision >= p.revision) { stats.stale++; return 'stale'; }++ const followerId = userOf(p.follower);++ const streamerId = userOf(p.target_id);++ if (followerId == null || streamerId == null) { stats.ignored++; return 'ignored:unmapped'; }++ d.prepare('INSERT INTO follow_projection_revisions (follower_subject, target_subject, revision) VALUES (?, ?, ?) ON CONFLICT (follower_subject, target_subject) DO UPDATE SET revision = excluded.revision')++ .run(p.follower, p.target_id, p.revision);++ stats.applied++;++ if (ev.event_type === 'network.follow.created') {++ d.prepare('INSERT OR IGNORE INTO follows (follower_id, streamer_id) VALUES (?, ?)').run(followerId, streamerId);++ return 'followed';++ }++ d.prepare('DELETE FROM follows WHERE follower_id = ? AND streamer_id = ?').run(followerId, streamerId);++ return 'unfollowed';++ })();++}++++module.exports = { writeThrough, apply, authority, stats };@@@ -2351,11 +2351,14 @@ router.get('/:id/rtmp-status', requireAuth, (req, res) => {});// ── Follow/Unfollow Streamer ─────────────────────────────────--router.post('/:id/follow', requireAuth, (req, res) => {++router.post('/:id/follow', requireAuth, async (req, res) => {try {const stream = db.getStreamById(req.params.id);if (!stream) return res.status(404).json({ error: 'Stream not found' });++ // ADR-030 step 4: Network first (FOLLOWS_AUTHORITY=network); Live's row only when Network took it.++ const w = await require('../social/network-follows').writeThrough(req.user.id, stream.user_id, !db.isFollowing(req.user.id, stream.user_id));++ if (!w.ok) return res.status(503).json({ error: 'Could not update the follow right now; try again in a moment' });if (db.isFollowing(req.user.id, stream.user_id)) {db.unfollowUser(req.user.id, stream.user_id);res.json({ following: false, count: db.getFollowerCount(stream.user_id) });@@@ -2396,11 +2399,14 @@ router.post('/:id/follow', requireAuth, (req, res) => {});// ── Follow/Unfollow by Username ──────────────────────────────--router.post('/channel/:username/follow', requireAuth, (req, res) => {++router.post('/channel/:username/follow', requireAuth, async (req, res) => {try {const user = db.getUserByUsername(req.params.username);if (!user) return res.status(404).json({ error: 'User not found' });++ // ADR-030 step 4: Network first (FOLLOWS_AUTHORITY=network); Live's row only when Network took it.++ const w = await require('../social/network-follows').writeThrough(req.user.id, user.id, !db.isFollowing(req.user.id, user.id));++ if (!w.ok) return res.status(503).json({ error: 'Could not update the follow right now; try again in a moment' });if (db.isFollowing(req.user.id, user.id)) {db.unfollowUser(req.user.id, user.id);res.json({ following: false, count: db.getFollowerCount(user.id) });nnew file mode 100644@@@ -0,0 +1,78 @@++'use strict';++// Follows on OpenVibe.Network (ADR-030 step 4; server/social/network-follows.js): with FOLLOWS_AUTHORITY=network a++// follow button writes Network first (network.follows.write) and Live's row only when Network took it; a side++// without a subject stays Live-only; network.follow.* events keep Live's table as a projection, in revision order.++const assert = require('assert');++const fs = require('fs');++const os = require('os');++const path = require('path');++++const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-netfollows-'));++process.env.DB_PATH = path.join(tmp, 'live.db');++process.env.OV_NETWORK_INTERNAL_URL = 'http://network.test';++const log = console.log; console.log = () => {}; console.warn = () => {};++const db = require('../server/db/database');++db.initDb();++console.log = log;++const principal = require('../server/net/network-principal');++principal.serviceHeaders = async () => ({ Authorization: 'Bearer svc-token' });++const follows = require('../server/social/network-follows');++const events = require('../server/auth/network-events');++++const sub = (tag) => `usr_${`01J${tag}`.padEnd(26, '0')}`;++const ANN = sub('AA'), BOB = sub('BB'), CAT = sub('CC'), GHOST = sub('DD');++const d = db.getDb();++for (const [id, name] of [[1, 'ann'], [2, 'bob'], [3, 'cat'], [4, 'nolink']]) d.prepare("INSERT INTO users (id, username, password_hash) VALUES (?, ?, 'x')").run(id, name);++const link = d.prepare("INSERT INTO linked_accounts (user_id, service, service_user_id, subject_id) VALUES (?, 'network', ?, ?)");++link.run(1, '101', ANN); link.run(2, '102', BOB); link.run(3, '103', CAT);++++(async () => {++ const calls = [];++ let answer = 200;++ const fetchImpl = async (url, o) => { calls.push({ url, method: o.method, body: o.body ? JSON.parse(o.body) : null, auth: o.headers.Authorization }); if (answer instanceof Error) throw answer; return { status: answer }; };++++ // Authority live (unset): no Network call.++ delete process.env.FOLLOWS_AUTHORITY;++ assert.deepStrictEqual(await follows.writeThrough(1, 2, true, { fetchImpl }), { ok: true, network: false, reason: 'authority_live' });++ assert.strictEqual(calls.length, 0);++++ process.env.FOLLOWS_AUTHORITY = 'network';++ let r = await follows.writeThrough(1, 2, true, { fetchImpl });++ assert.deepStrictEqual(r, { ok: true, network: true });++ assert.deepStrictEqual(calls.pop(), { url: `http://network.test/internal/follows/channel/${BOB}`, method: 'PUT', body: { follower: ANN }, auth: 'Bearer svc-token' });++ r = await follows.writeThrough(1, 2, false, { fetchImpl });++ assert.deepStrictEqual(calls.pop(), { url: `http://network.test/internal/follows/channel/${BOB}?follower=${ANN}`, method: 'DELETE', body: null, auth: 'Bearer svc-token' });++ answer = 503;++ assert.strictEqual((await follows.writeThrough(1, 2, true, { fetchImpl })).ok, false, 'Network refused: the button fails, nothing local');++ answer = new Error('ECONNREFUSED');++ assert.strictEqual((await follows.writeThrough(1, 2, true, { fetchImpl })).ok, false);++ answer = 200;++ const n = calls.length;++ assert.deepStrictEqual(await follows.writeThrough(4, 2, true, { fetchImpl }), { ok: true, network: false, reason: 'no_subject' }, 'no subject: Live-only');++ assert.strictEqual(calls.length, n);++++ // The projection, through the Network events endpoint's apply().++ const ev = (type, follower, target, revision) => ({ event_id: `evt_01JAB2C3D4E5F6G7H8J9K0M${String(revision).padStart(3, '0')}`.slice(0, 30), event_type: type, source: 'network', payload: { follower, target_type: 'channel', target_id: target, revision } });++ assert.strictEqual(events.apply(ev('network.follow.created', CAT, BOB, 1)), 'followed');++ assert.ok(db.isFollowing(3, 2));++ assert.strictEqual(events.apply(ev('network.follow.created', CAT, BOB, 1)), 'stale', 'a redelivery');++ assert.strictEqual(events.apply(ev('network.follow.deleted', CAT, BOB, 3)), 'unfollowed');++ assert.ok(!db.isFollowing(3, 2));++ assert.strictEqual(events.apply(ev('network.follow.created', CAT, BOB, 2)), 'stale', 'an older follow after the unfollow changes nothing');++ assert.ok(!db.isFollowing(3, 2));++ assert.strictEqual(events.apply(ev('network.follow.created', GHOST, BOB, 1)), 'ignored:unmapped', 'no Live account');++ assert.strictEqual(events.apply({ ...ev('network.follow.created', CAT, BOB, 9), source: 'live' }), 'ignored:source');++++ // Both follow routes ask Network before writing Live's row.++ const routes = fs.readFileSync(path.join(__dirname, '..', 'server', 'streaming', 'routes.js'), 'utf8');++ for (const route of ["router.post('/:id/follow'", "router.post('/channel/:username/follow'"]) {++ const body = routes.slice(routes.indexOf(route), routes.indexOf(route) + 1200);++ assert.ok(body.indexOf('writeThrough(') > 0 && body.indexOf('writeThrough(') < body.indexOf('db.followUser('), `${route} writes Network first`);++ assert.match(body, /if \(!w\.ok\) return res\.status\(503\)/);++ }++ // Live subscribes to the follow events.++ assert.match(fs.readFileSync(path.join(__dirname, '..', 'scripts', 'subscribe-media-events.js'), 'utf8'), /'network\.follow\.created', 'network\.follow\.deleted'/);++ fs.rmSync(tmp, { recursive: true, force: true });++ console.log('network follows: all checks passed');++ process.exit(0);++})().catch((e) => { console.error(e); process.exit(1); });
@@ -31,7 +31,7 @@ const fs = require('fs');
@@ -31,7 +31,7 @@ const fs = require('fs');
const TOPICS = ['media.vod.*', 'media.clip.*', 'media.storage.*'];
const TOPICS = ['media.vod.*', 'media.clip.*', 'media.storage.*'];
const DEFAULT_ENDPOINT = 'http://127.0.0.1:3000/internal/media-events';
const DEFAULT_ENDPOINT = 'http://127.0.0.1:3000/internal/media-events';
-const NETWORK_TOPICS = ['network.user.token_valid_after', 'network.user.updated'];
+const NETWORK_TOPICS = ['network.user.token_valid_after', 'network.user.updated', 'network.follow.created', 'network.follow.deleted'];
const NETWORK_ENDPOINT = 'http://127.0.0.1:3000/internal/network-events';
const NETWORK_ENDPOINT = 'http://127.0.0.1:3000/internal/network-events';
const DEFAULT_LIVE_ENV = '/etc/openvibe/live.env';
const DEFAULT_LIVE_ENV = '/etc/openvibe/live.env';
@@ -9,6 +9,7 @@
@@ -9,6 +9,7 @@
*
*
* token_valid_after the person's older tokens are refused (./revocations.js)
* token_valid_after the person's older tokens are refused (./revocations.js)
* updated the person's profile, role and ban (./subject-projection.js, WS-B task 2)
* updated the person's profile, role and ban (./subject-projection.js, WS-B task 2)
+ * follow.created/.deleted Live's follows table as a projection of Network's graph (../social/network-follows.js)
*
*
* Applying is idempotent (a cutoff only ever moves forward; a profile revision only up), so a redelivery
* Applying is idempotent (a cutoff only ever moves forward; a profile revision only up), so a redelivery
* is a no-op without an inbox. Anything signed but not ours to act on is acknowledged (204).
* is a no-op without an inbox. Anything signed but not ours to act on is acknowledged (204).
@@ -25,8 +26,11 @@ function secret() { return process.env.LIVE_EVENTS_SECRET || process.env.MEDIA_E
@@ -25,8 +26,11 @@ function secret() { return process.env.LIVE_EVENTS_SECRET || process.env.MEDIA_E
/** Apply one envelope; returns 'revoked' | 'unchanged' | 'updated' | 'stale' | 'ignored:<why>'. */
/** Apply one envelope; returns 'revoked' | 'unchanged' | 'updated' | 'stale' | 'ignored:<why>'. */
function apply(ev) {
function apply(ev) {
if (!ev || !EVENT_ID_RE.test(String(ev.event_id || ''))) return 'ignored:envelope';
if (!ev || !EVENT_ID_RE.test(String(ev.event_id || ''))) return 'ignored:envelope';
- if (ev.event_type !== 'network.user.token_valid_after' && ev.event_type !== 'network.user.updated') return 'ignored:type';
+ const follow = ev.event_type === 'network.follow.created' || ev.event_type === 'network.follow.deleted';
+ if (ev.event_type !== 'network.user.token_valid_after' && ev.event_type !== 'network.user.updated' && !follow) return 'ignored:type';
if (ev.source !== 'network') return 'ignored:source';
if (ev.source !== 'network') return 'ignored:source';
+ // The follow graph is Network's (ADR-030); Live's follows table is its projection.
+ if (follow) return require('../social/network-follows').apply(ev);
const p = ev.payload && typeof ev.payload === 'object' ? ev.payload : {};
const p = ev.payload && typeof ev.payload === 'object' ? ev.payload : {};
if (ev.event_type === 'network.user.updated') return require('./subject-projection').apply(p, { notify: notifyChat });
if (ev.event_type === 'network.user.updated') return require('./subject-projection').apply(p, { notify: notifyChat });
const subject = p.subject && p.subject.id;
const subject = p.subject && p.subject.id;
@@ -44,7 +48,7 @@ function handler(req, res) {
@@ -44,7 +48,7 @@ function handler(req, res) {
if (!delivery) { stats.refused++; return res.status(401).json({ error: 'bad signature' }); }
if (!delivery) { stats.refused++; return res.status(401).json({ error: 'bad signature' }); }
stats.received++;
stats.received++;
const out = apply(delivery.event);
const out = apply(delivery.event);
- if (out === 'revoked') stats.revoked++; else if (out === 'unchanged' || out === 'stale') stats.unchanged++; else if (out === 'updated') stats.profiles++; else stats.ignored++;
+ if (out === 'revoked') stats.revoked++; else if (out === 'unchanged' || out === 'stale') stats.unchanged++; else if (out === 'updated' || out === 'followed' || out === 'unfollowed') stats.profiles++; else stats.ignored++;
res.status(204).end();
res.status(204).end();
}
}
new file mode 100644
new file mode 100644
@@ -0,0 +1,95 @@
@@ -0,0 +1,95 @@
+'use strict';
+/**
+ * Follows on OpenVibe.Network (ADR-030; roadmap WS-E task 4). Network owns the follow graph, keyed by subjects;
+ * Live's `follows` table is its projection, kept for Live's own reads (channel pages, counts, sort orders).
+ *
+ * writeThrough(followerId, streamerId, following) step 4: with FOLLOWS_AUTHORITY=network, a follow button
+ * records the change on Network first (PUT/DELETE /internal/follows/channel/<streamer subject>, Live's
+ * service token, network.follows.write). The route writes Live's row only when Network took it; if
+ * Network cannot answer, the button fails (503) rather than letting the two drift. A side with no
+ * Network subject yet (an account Network never told us about) stays Live-only, as before.
+ * apply(event) network.follow.created / .deleted from OpenVibe.Events (server/auth/network-events.js):
+ * follows made anywhere else (my.openvibe.network, another product, the API) reach Live's table.
+ * Idempotent and ordered: the highest revision applied per pair is kept (follow_projection_revisions),
+ * and an older one is `stale`. A pair whose side has no Live account is `ignored:unmapped`.
+ *
+ * Unset FOLLOWS_AUTHORITY: the buttons write Live's table only (the state before step 4); events still apply.
+ */
+const db = require('../db/database');
+
+const NETWORK_INTERNAL_URL = (process.env.OV_NETWORK_INTERNAL_URL || 'http://127.0.0.1:4000').replace(/\/+$/, '');
+const SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
+const stats = { written: 0, local_only: 0, failed: 0, applied: 0, stale: 0, ignored: 0 };
+
+const authority = () => (process.env.FOLLOWS_AUTHORITY === 'network' ? 'network' : 'live');
+
+function subjectOf(liveUserId) {
+ const r = db.getDb().prepare("SELECT subject_id FROM linked_accounts WHERE service = 'network' AND user_id = ?").get(liveUserId);
+ return r && SUBJECT_RE.test(String(r.subject_id || '')) ? r.subject_id : null;
+}
+function userOf(subject) {
+ const r = db.getDb().prepare("SELECT user_id FROM linked_accounts WHERE service = 'network' AND subject_id = ? ORDER BY id LIMIT 1").get(subject);
+ return r ? r.user_id : null;
+}
+
+/**
+ * Record a follow change on Network before Live writes its row. → { ok: true, network: true } when Network took it,
+ * { ok: true, network: false, reason } when it stays Live-only (authority live, or a side without a subject),
+ * { ok: false, error } when Network could not take it (the caller answers 503 and changes nothing).
+ */
+async function writeThrough(followerId, streamerId, following, { fetchImpl = globalThis.fetch } = {}) {
+ if (authority() !== 'network') return { ok: true, network: false, reason: 'authority_live' };
+ const follower = subjectOf(followerId);
+ const target = subjectOf(streamerId);
+ if (!follower || !target) { stats.local_only++; return { ok: true, network: false, reason: 'no_subject' }; }
+ try {
+ const principal = require('../net/network-principal');
+ const headers = { Accept: 'application/json', ...(await principal.serviceHeaders('openvibe.network')) };
+ const url = `${NETWORK_INTERNAL_URL}/internal/follows/channel/${target}`;
+ const res = following
+ ? await fetchImpl(url, { method: 'PUT', headers: { ...headers, 'Content-Type': 'application/json' }, body: JSON.stringify({ follower }), signal: AbortSignal.timeout(5000) })
+ : await fetchImpl(`${url}?follower=${follower}`, { method: 'DELETE', headers, signal: AbortSignal.timeout(5000) });
+ if (res.status === 401) principal.invalidate('openvibe.network');
+ if (res.status < 200 || res.status >= 300) { stats.failed++; return { ok: false, error: `Network answered ${res.status}` }; }
+ stats.written++;
+ return { ok: true, network: true };
+ } catch (err) {
+ stats.failed++;
+ return { ok: false, error: err.message };
+ }
+}
+
+function ensureTable() {
+ db.getDb().exec(`CREATE TABLE IF NOT EXISTS follow_projection_revisions (
+ follower_subject TEXT NOT NULL,
+ target_subject TEXT NOT NULL,
+ revision INTEGER NOT NULL,
+ PRIMARY KEY (follower_subject, target_subject)
+ )`);
+}
+
+/** Apply network.follow.created / .deleted. → 'followed' | 'unfollowed' | 'stale' | 'ignored:<why>' */
+function apply(ev) {
+ const p = ev && ev.payload && typeof ev.payload === 'object' ? ev.payload : {};
+ if (p.target_type !== 'channel' || !SUBJECT_RE.test(String(p.follower || '')) || !SUBJECT_RE.test(String(p.target_id || '')) || !Number.isInteger(p.revision)) { stats.ignored++; return 'ignored:payload'; }
+ ensureTable();
+ const d = db.getDb();
+ return d.transaction(() => {
+ const prev = d.prepare('SELECT revision FROM follow_projection_revisions WHERE follower_subject = ? AND target_subject = ?').get(p.follower, p.target_id);
+ if (prev && prev.revision >= p.revision) { stats.stale++; return 'stale'; }
+ const followerId = userOf(p.follower);
+ const streamerId = userOf(p.target_id);
+ if (followerId == null || streamerId == null) { stats.ignored++; return 'ignored:unmapped'; }
+ d.prepare('INSERT INTO follow_projection_revisions (follower_subject, target_subject, revision) VALUES (?, ?, ?) ON CONFLICT (follower_subject, target_subject) DO UPDATE SET revision = excluded.revision')
+ .run(p.follower, p.target_id, p.revision);
+ stats.applied++;
+ if (ev.event_type === 'network.follow.created') {
+ d.prepare('INSERT OR IGNORE INTO follows (follower_id, streamer_id) VALUES (?, ?)').run(followerId, streamerId);
+ return 'followed';
+ }
+ d.prepare('DELETE FROM follows WHERE follower_id = ? AND streamer_id = ?').run(followerId, streamerId);
+ return 'unfollowed';
+ })();
+}
+
+module.exports = { writeThrough, apply, authority, stats };
@@ -2351,11 +2351,14 @@ router.get('/:id/rtmp-status', requireAuth, (req, res) => {
@@ -2351,11 +2351,14 @@ router.get('/:id/rtmp-status', requireAuth, (req, res) => {
});
});
// ── Follow/Unfollow Streamer ─────────────────────────────────
// ── Follow/Unfollow Streamer ─────────────────────────────────
-router.post('/:id/follow', requireAuth, (req, res) => {
+router.post('/:id/follow', requireAuth, async (req, res) => {
try {
try {
const stream = db.getStreamById(req.params.id);
const stream = db.getStreamById(req.params.id);
if (!stream) return res.status(404).json({ error: 'Stream not found' });
if (!stream) return res.status(404).json({ error: 'Stream not found' });
+ // ADR-030 step 4: Network first (FOLLOWS_AUTHORITY=network); Live's row only when Network took it.
+ const w = await require('../social/network-follows').writeThrough(req.user.id, stream.user_id, !db.isFollowing(req.user.id, stream.user_id));
+ if (!w.ok) return res.status(503).json({ error: 'Could not update the follow right now; try again in a moment' });
if (db.isFollowing(req.user.id, stream.user_id)) {
if (db.isFollowing(req.user.id, stream.user_id)) {
db.unfollowUser(req.user.id, stream.user_id);
db.unfollowUser(req.user.id, stream.user_id);
res.json({ following: false, count: db.getFollowerCount(stream.user_id) });
res.json({ following: false, count: db.getFollowerCount(stream.user_id) });
@@ -2396,11 +2399,14 @@ router.post('/:id/follow', requireAuth, (req, res) => {
@@ -2396,11 +2399,14 @@ router.post('/:id/follow', requireAuth, (req, res) => {
});
});
// ── Follow/Unfollow by Username ──────────────────────────────
// ── Follow/Unfollow by Username ──────────────────────────────
-router.post('/channel/:username/follow', requireAuth, (req, res) => {
+router.post('/channel/:username/follow', requireAuth, async (req, res) => {
try {
try {
const user = db.getUserByUsername(req.params.username);
const user = db.getUserByUsername(req.params.username);
if (!user) return res.status(404).json({ error: 'User not found' });
if (!user) return res.status(404).json({ error: 'User not found' });
+ // ADR-030 step 4: Network first (FOLLOWS_AUTHORITY=network); Live's row only when Network took it.
+ const w = await require('../social/network-follows').writeThrough(req.user.id, user.id, !db.isFollowing(req.user.id, user.id));
+ if (!w.ok) return res.status(503).json({ error: 'Could not update the follow right now; try again in a moment' });
if (db.isFollowing(req.user.id, user.id)) {
if (db.isFollowing(req.user.id, user.id)) {
db.unfollowUser(req.user.id, user.id);
db.unfollowUser(req.user.id, user.id);
res.json({ following: false, count: db.getFollowerCount(user.id) });
res.json({ following: false, count: db.getFollowerCount(user.id) });
new file mode 100644
new file mode 100644
@@ -0,0 +1,78 @@
@@ -0,0 +1,78 @@
+'use strict';
+// Follows on OpenVibe.Network (ADR-030 step 4; server/social/network-follows.js): with FOLLOWS_AUTHORITY=network a
+// follow button writes Network first (network.follows.write) and Live's row only when Network took it; a side
+// without a subject stays Live-only; network.follow.* events keep Live's table as a projection, in revision order.
+const assert = require('assert');
+const fs = require('fs');
+const os = require('os');
+const path = require('path');
+
+const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-netfollows-'));
+process.env.DB_PATH = path.join(tmp, 'live.db');
+process.env.OV_NETWORK_INTERNAL_URL = 'http://network.test';
+const log = console.log; console.log = () => {}; console.warn = () => {};
+const db = require('../server/db/database');
+db.initDb();
+console.log = log;
+const principal = require('../server/net/network-principal');
+principal.serviceHeaders = async () => ({ Authorization: 'Bearer svc-token' });
+const follows = require('../server/social/network-follows');
+const events = require('../server/auth/network-events');
+
+const sub = (tag) => `usr_${`01J${tag}`.padEnd(26, '0')}`;
+const ANN = sub('AA'), BOB = sub('BB'), CAT = sub('CC'), GHOST = sub('DD');
+const d = db.getDb();
+for (const [id, name] of [[1, 'ann'], [2, 'bob'], [3, 'cat'], [4, 'nolink']]) d.prepare("INSERT INTO users (id, username, password_hash) VALUES (?, ?, 'x')").run(id, name);
+const link = d.prepare("INSERT INTO linked_accounts (user_id, service, service_user_id, subject_id) VALUES (?, 'network', ?, ?)");
+link.run(1, '101', ANN); link.run(2, '102', BOB); link.run(3, '103', CAT);
+
+(async () => {
+ const calls = [];
+ let answer = 200;
+ const fetchImpl = async (url, o) => { calls.push({ url, method: o.method, body: o.body ? JSON.parse(o.body) : null, auth: o.headers.Authorization }); if (answer instanceof Error) throw answer; return { status: answer }; };
+
+ // Authority live (unset): no Network call.
+ delete process.env.FOLLOWS_AUTHORITY;
+ assert.deepStrictEqual(await follows.writeThrough(1, 2, true, { fetchImpl }), { ok: true, network: false, reason: 'authority_live' });
+ assert.strictEqual(calls.length, 0);
+
+ process.env.FOLLOWS_AUTHORITY = 'network';
+ let r = await follows.writeThrough(1, 2, true, { fetchImpl });
+ assert.deepStrictEqual(r, { ok: true, network: true });
+ assert.deepStrictEqual(calls.pop(), { url: `http://network.test/internal/follows/channel/${BOB}`, method: 'PUT', body: { follower: ANN }, auth: 'Bearer svc-token' });
+ r = await follows.writeThrough(1, 2, false, { fetchImpl });
+ assert.deepStrictEqual(calls.pop(), { url: `http://network.test/internal/follows/channel/${BOB}?follower=${ANN}`, method: 'DELETE', body: null, auth: 'Bearer svc-token' });
+ answer = 503;
+ assert.strictEqual((await follows.writeThrough(1, 2, true, { fetchImpl })).ok, false, 'Network refused: the button fails, nothing local');
+ answer = new Error('ECONNREFUSED');
+ assert.strictEqual((await follows.writeThrough(1, 2, true, { fetchImpl })).ok, false);
+ answer = 200;
+ const n = calls.length;
+ assert.deepStrictEqual(await follows.writeThrough(4, 2, true, { fetchImpl }), { ok: true, network: false, reason: 'no_subject' }, 'no subject: Live-only');
+ assert.strictEqual(calls.length, n);
+
+ // The projection, through the Network events endpoint's apply().
+ const ev = (type, follower, target, revision) => ({ event_id: `evt_01JAB2C3D4E5F6G7H8J9K0M${String(revision).padStart(3, '0')}`.slice(0, 30), event_type: type, source: 'network', payload: { follower, target_type: 'channel', target_id: target, revision } });
+ assert.strictEqual(events.apply(ev('network.follow.created', CAT, BOB, 1)), 'followed');
+ assert.ok(db.isFollowing(3, 2));
+ assert.strictEqual(events.apply(ev('network.follow.created', CAT, BOB, 1)), 'stale', 'a redelivery');
+ assert.strictEqual(events.apply(ev('network.follow.deleted', CAT, BOB, 3)), 'unfollowed');
+ assert.ok(!db.isFollowing(3, 2));
+ assert.strictEqual(events.apply(ev('network.follow.created', CAT, BOB, 2)), 'stale', 'an older follow after the unfollow changes nothing');
+ assert.ok(!db.isFollowing(3, 2));
+ assert.strictEqual(events.apply(ev('network.follow.created', GHOST, BOB, 1)), 'ignored:unmapped', 'no Live account');
+ assert.strictEqual(events.apply({ ...ev('network.follow.created', CAT, BOB, 9), source: 'live' }), 'ignored:source');
+
+ // Both follow routes ask Network before writing Live's row.
+ const routes = fs.readFileSync(path.join(__dirname, '..', 'server', 'streaming', 'routes.js'), 'utf8');
+ for (const route of ["router.post('/:id/follow'", "router.post('/channel/:username/follow'"]) {
+ const body = routes.slice(routes.indexOf(route), routes.indexOf(route) + 1200);
+ assert.ok(body.indexOf('writeThrough(') > 0 && body.indexOf('writeThrough(') < body.indexOf('db.followUser('), `${route} writes Network first`);
+ assert.match(body, /if \(!w\.ok\) return res\.status\(503\)/);
+ }
+ // Live subscribes to the follow events.
+ assert.match(fs.readFileSync(path.join(__dirname, '..', 'scripts', 'subscribe-media-events.js'), 'utf8'), /'network\.follow\.created', 'network\.follow\.deleted'/);
+ fs.rmSync(tmp, { recursive: true, force: true });
+ console.log('network follows: all checks passed');
+ process.exit(0);
+})().catch((e) => { console.error(e); process.exit(1); });