openvibers/openvibe.live Commits
Plan T2 follow-graph completion in OpenVibe.Live (#14)
Changes
@@@ -90,7 +90,7 @@ The media subsystem lives in **OpenVibe.Media**:## Capabilities--Live implements, for other services' tokens (manifest `manifests/services/live.json` in OpenVibe.Contracts): `live.chat_context.read`, `live.chat_effects.write`, `live.chat_mirror.write`, `live.follower.read`, `live.lineage.resolve`, `live.tips_delivery.write`, `live.channel.read`, `live.stream.read`, `live.discovery.read`, `live.owner.resolve`.++Live implements, for other services' tokens (manifest `manifests/services/live.json` in OpenVibe.Contracts): `live.chat_context.read`, `live.chat_effects.write`, `live.chat_mirror.write`, `live.lineage.resolve`, `live.tips_delivery.write`, `live.channel.read`, `live.stream.read`, `live.discovery.read`, `live.owner.resolve`.Live's own principal (`live`) holds grants to call:@@@ -11,7 +11,6 @@"live.chat_context.read","live.chat_effects.write","live.chat_mirror.write",-- "live.follower.read","live.lineage.resolve","live.tips_delivery.write","live.channel.read",@@@ -4786,18 +4786,7 @@ function updateUserAvatar(userId, avatarUrl, pasteId = null) {// resetAvatarsForPaste() removed — the media subsystem (vods/clips/pastes writes) moved to OpenVibe.Media.// ── Follow helpers ───────────────────────────────────────────----function followUser(followerId, streamerId) {-- return run(-- `INSERT OR IGNORE INTO follows (follower_id, streamer_id) VALUES (?, ?)`,-- [followerId, streamerId]-- );--}----function unfollowUser(followerId, streamerId) {-- return run(`DELETE FROM follows WHERE follower_id = ? AND streamer_id = ?`,-- [followerId, streamerId]);--}++// Reads of the projection of Network's follow graph; only server/social/network-follows.js writes it.function getFollowerCount(streamerId) {const row = get('SELECT COUNT(*) as count FROM follows WHERE streamer_id = ?', [streamerId]);@@@ -7067,7 +7056,7 @@ module.exports = {getKickChannelCache, setKickChannelCache,getChannelPoints, addChannelPoints, deductChannelPoints, applyChannelPoints,// Follows-- followUser, unfollowUser, getFollowerCount, isFollowing, getFollowerIds,++ getFollowerCount, isFollowing, getFollowerIds,// Transactions (Vibes)createTransaction, addVibes, deductVibes, addVibesCashout, deductVibesCashout,// OpenCoins@@@ -617,9 +617,6 @@ app.use('/internal/chat-context', chatLiveContext.contextRouter);app.use('/internal/chat-effects', chatLiveContext.effectsRouter);// OpenVibe.Tips announces settled tips in the creator's chat here (service token, live.tips_delivery.write).app.use('/internal/tips', require('./tips/delivery-routes'));--// OpenVibe.Network sends go-live notifications from live.stream.started and asks here who follows--// the channel (service token, live.follower.read).--app.use('/internal/followers', require('./streaming/followers-internal'));// The canonical channel/owner resolver for other services (roadmap D20; service token, live.lineage.resolve).app.use('/internal/lineage', require('./lineage/routes'));app.use('/internal', require('./internal/routes'));@@@ -265,7 +265,7 @@ async function _channelMeta(username, page = 1) {let ov = null;try { ov = db.getStreamerOverview ? db.getStreamerOverview(acct.userId) : null; } catch { /* */ }let followers = 0;-- try { followers = db.getFollowerCount ? (db.getFollowerCount(acct.userId) || 0) : 0; } catch { /* */ }++ try { followers = (await require('../social/network-follows').followerCount(acct.userId)) || 0; } catch { /* */ }let live = [];try { live = acct.banned ? [] : (db.getLiveStreamsByUserId(acct.userId) || []); } catch { live = []; }const bio = clean(acct.bio || '', 300);@@@ -1,27 +1,44 @@'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).++ * Follows on OpenVibe.Network (ADR-030; plan T2). Network owns the follow graph, keyed by subjects. Live's++ * `follows` table is only its projection, written by Network's events alone (in revision order) and kept for++ * Live's own bulk reads (go-live notices, Chat's context route, activity numbers); nothing on Live follows or++ * unfollows on its own, and what a person sees (a follow button, a follower count) comes from Network.*-- * 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.++ * set(followerId, streamerId, following) a follow button: PUT / DELETE /internal/follows/channel/<streamer++ * subject> on Network (Live's service token, network.follows.write). Network's answer++ * (network.follow-status-result@1: following, the follower count) is what the button shows, and Live++ * remembers it (ANSWER_TTL_MS) so a reload or a second click sees it before the event arrives. The answer++ * carries no revision, so it never touches the projection: an older event delivered late cannot undo it.++ * `started` is Network's own word that this request started the follow: 201 (the pair's first follow),++ * or a `since` no older than the request (a follow that starts again starts its since again); an++ * idempotent repeat of a follow Network already had is not started, whatever the projection says.++ * Both sides need a Network subject: a side without one cannot follow (409). Network unreachable,++ * refusing Live's token, or answering a malformed document: 503. A request Network refuses for what it++ * is (following oneself, the follow limit, an unknown channel) keeps Network's status and reason.++ * isFollowing(followerId, streamerId) the button state: Network's recent answer for the pair, else the++ * projection.++ * followerCount(streamerId) GET /api/v1/follows/channel/<subject> on Network (the public count), cached++ * COUNT_TTL_MS and refreshed by every write answer; the projection's count only while Network cannot++ * answer (or the channel has no subject).* 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`.++ * follows made anywhere (the buttons here, 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`. An applied event that agrees with Network's remembered answer retires the answer.*-- * Unset FOLLOWS_AUTHORITY: the buttons write Live's table only (the state before step 4); events still apply.++ * Network tells the followed person (FOLLOW) itself; Live sends no follow notification.*/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');++const ANSWER_TTL_MS = 10 * 60 * 1000;++const COUNT_TTL_MS = 30 * 1000; // Network's own public max-age for the count++const COUNT_DOWN_MS = 10 * 1000; // after a failed count read, the projection answers for this long++const stats = { written: 0, no_subject: 0, refused: 0, failed: 0, applied: 0, stale: 0, ignored: 0, count_read: 0, count_fallback: 0 };++const answers = new Map(); // `${followerId}>${streamerId}` → { following, at }++const counts = new Map(); // streamer subject → { n, at } | { down: until }function subjectOf(liveUserId) {const r = db.getDb().prepare("SELECT subject_id FROM linked_accounts WHERE service = 'network' AND user_id = ?").get(liveUserId);@@@ -32,30 +49,95 @@ function userOf(subject) {return r ? r.user_id : null;}++function project(followerId, streamerId, following) {++ if (following) db.getDb().prepare('INSERT OR IGNORE INTO follows (follower_id, streamer_id) VALUES (?, ?)').run(followerId, streamerId);++ else db.getDb().prepare('DELETE FROM follows WHERE follower_id = ? AND streamer_id = ?').run(followerId, streamerId);++}++++/** A network.follow-status-result@1 for `target` with a valid count (and `following`, when `viewer`). */++function validStatus(body, target, viewer) {++ return !!body && typeof body === 'object' && body.target_id === target && Number.isInteger(body.followers) && body.followers >= 0++ && (!viewer || typeof body.following === 'boolean');++}++/**-- * 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).++ * Follow (following = true) or unfollow on Network.++ * → { ok: true, following, count, started }, or { ok: false, status, error } (the caller answers that status).*/--async function writeThrough(followerId, streamerId, following, { fetchImpl = globalThis.fetch } = {}) {-- if (authority() !== 'network') return { ok: true, network: false, reason: 'authority_live' };++async function set(followerId, streamerId, following, { fetchImpl = globalThis.fetch } = {}) {const follower = subjectOf(followerId);const target = subjectOf(streamerId);-- if (!follower || !target) { stats.local_only++; return { ok: true, network: false, reason: 'no_subject' }; }++ if (!follower || !target) {++ stats.no_subject++;++ return { ok: false, status: 409, error: follower ? 'This channel cannot be followed yet' : 'Sign out and back in to follow channels' };++ }++ let res, body;++ const sentAt = Date.now();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++ 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 };++ body = await res.json().catch(() => null);} catch (err) {stats.failed++;-- return { ok: false, error: err.message };++ return { ok: false, status: 503, error: 'Could not update the follow right now; try again in a moment', detail: err.message };++ }++ if ([400, 404, 409].includes(res.status)) {++ stats.refused++;++ return { ok: false, status: res.status, error: (body && typeof body.detail === 'string' && body.detail) || 'Could not update the follow' };++ }++ if (res.status < 200 || res.status >= 300 || !validStatus(body, target, true)) {++ stats.failed++;++ return { ok: false, status: 503, error: 'Could not update the follow right now; try again in a moment', detail: `Network answered ${res.status} without a valid follow status` };++ }++ stats.written++;++ const now = Date.now();++ if (answers.size > 5000) for (const [k, a] of answers) if (now - a.at >= ANSWER_TTL_MS) answers.delete(k);++ answers.set(`${followerId}>${streamerId}`, { following: body.following, at: now });++ counts.set(target, { n: body.followers, at: now });++ const since = typeof body.since === 'string' ? Date.parse(body.since) : NaN;++ return {++ ok: true,++ following: body.following,++ count: body.followers,++ started: body.following && (res.status === 201 || since >= sentAt),++ };++}++++/** Whether followerId follows streamerId, as the button shows it (sync). */++function isFollowing(followerId, streamerId) {++ if (followerId == null) return false;++ const key = `${followerId}>${streamerId}`;++ const a = answers.get(key);++ if (a && Date.now() - a.at < ANSWER_TTL_MS) return a.following;++ if (a) answers.delete(key);++ return db.isFollowing(followerId, streamerId);++}++++/** streamerId's follower count, from Network (cached), or the projection's while Network cannot answer. */++async function followerCount(streamerId, { fetchImpl = globalThis.fetch } = {}) {++ const target = subjectOf(streamerId);++ const local = () => { stats.count_fallback++; return db.getFollowerCount(streamerId); };++ if (!target) return local();++ const now = Date.now();++ const c = counts.get(target);++ if (c && c.down > now) return local();++ if (c && c.n != null && now - c.at < COUNT_TTL_MS) return c.n;++ try {++ const res = await fetchImpl(`${NETWORK_INTERNAL_URL}/api/v1/follows/channel/${target}`, { headers: { Accept: 'application/json' }, signal: AbortSignal.timeout(1500) });++ const body = res.status === 200 ? await res.json().catch(() => null) : null;++ if (!validStatus(body, target, false)) throw new Error(`Network answered ${res.status} without a valid follow status`);++ stats.count_read++;++ counts.set(target, { n: body.followers, at: Date.now() });++ return body.followers;++ } catch (err) {++ console.warn('[Follows] follower count from the projection:', err.message);++ counts.set(target, { down: Date.now() + COUNT_DOWN_MS });++ return local();}}@@@ -83,13 +165,13 @@ function apply(ev) {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';++ const following = ev.event_type === 'network.follow.created';++ project(followerId, streamerId, following);++ const key = `${followerId}>${streamerId}`;++ if (answers.has(key) && answers.get(key).following === following) answers.delete(key);++ counts.delete(p.target_id);++ return following ? 'followed' : 'unfollowed';})();}--module.exports = { writeThrough, apply, authority, stats };++module.exports = { set, apply, isFollowing, followerCount, stats, _reset: () => { answers.clear(); counts.clear(); } };ddeleted file mode 100644@@@ -1,36 +0,0 @@--'use strict';--// GET /internal/followers?stream_id=<id>&limit=<n>[&after=<cursor>] — the followers of the channel that--// streams <id>, for OpenVibe.Network's live.stream.started consumer (go-live notifications are--// Network's job). Service token only (aud openvibe.live, cap live.follower.read), loopback only.--// A follower is named by their Network subject when Live knows it, else their Network user id.--const express = require('express');--const db = require('../db/database');--const { guard } = require('../net/service-guard');--const { http } = require('openvibe-contracts');----const router = express.Router();--router.get('/', guard('live.follower.read'), (req, res) => {-- const streamId = Number(req.query.stream_id);-- if (!Number.isSafeInteger(streamId) || streamId <= 0) return http.sendProblem(res, 400, 'live.bad_request', { detail: 'stream_id is required' });-- const limit = Math.min(Math.max(parseInt(req.query.limit, 10) || 1000, 1), 5000);-- const after = Math.max(parseInt(req.query.after, 10) || 0, 0);-- const d = db.getDb();-- const stream = d.prepare('SELECT user_id FROM streams WHERE id = ?').get(streamId);-- if (!stream) return http.sendProblem(res, 404, 'live.unknown_stream', { detail: `no stream ${streamId}` });-- const channel = d.prepare("SELECT subject_id FROM linked_accounts WHERE service = 'network' AND user_id = ?").get(stream.user_id);-- const rows = d.prepare(`SELECT f.id, la.subject_id, la.service_user_id FROM follows f-- LEFT JOIN linked_accounts la ON la.service = 'network' AND la.user_id = f.follower_id-- WHERE f.streamer_id = ? AND f.id > ? ORDER BY f.id LIMIT ?`).all(stream.user_id, after, limit);-- const followers = rows.map(r => ({-- subject: r.subject_id || null,-- network_user_id: /^\d+$/.test(String(r.service_user_id || '')) ? Number(r.service_user_id) : null,-- }));-- res.set('Cache-Control', 'no-store').json({-- stream_id: streamId,-- channel: { subject: (channel && channel.subject_id) || null },-- followers,-- next: rows.length === limit ? rows[rows.length - 1].id : null,-- });--});----module.exports = router;@@@ -344,8 +344,10 @@ router.get('/channel/:username', optionalAuth, async (req, res) => {aiShort(clipsOfStreams, 'clip_ai_state', 'clip_id');aiShort(aiClips, 'clip_ai_state', 'clip_id');}-- const followerCount = db.getFollowerCount(channel.user_id);-- const isFollowing = req.user ? db.isFollowing(req.user.id, channel.user_id) : false;++ // Follows are Network's (ADR-030): its count, and the viewer's follow as Network last answered it.++ const networkFollows = require('../social/network-follows');++ const followerCount = await networkFollows.followerCount(channel.user_id);++ const isFollowing = req.user ? networkFollows.isFollowing(req.user.id, channel.user_id) : false;// Managed streams for this channelconst managedStreams = db.getManagedStreamsByUserId(channel.user_id) || [];@@@ -1627,7 +1629,7 @@ router.get('/setup-progress', requireAuth, async (req, res) => {const sounds = safe(() => db.get('SELECT COUNT(*) AS n FROM channel_sounds WHERE channel_owner_id = ?', [uid]).n, 0);const goals = safe(() => db.get('SELECT COUNT(*) AS n FROM donation_goals WHERE user_id = ? AND is_active = 1', [uid]).n, 0);const powerchat = safe(() => !!db.get('SELECT 1 FROM powerchat_connections WHERE user_id = ? LIMIT 1', [uid]), false);-- const followers = safe(() => db.getFollowerCount(uid), 0);++ const followers = await require('../social/network-follows').followerCount(uid).catch(() => 0);let panels = 0; try { const p = channel.panels ? JSON.parse(channel.panels) : []; panels = Array.isArray(p) ? p.length : 0; } catch { panels = 0; }const offline = !!(channel.offline_screen_type && channel.offline_screen_type !== 'none');const methodSet = slots.some(sl => !!sl.streaming_method);@@@ -1958,7 +1960,7 @@ router.post('/managed/:id/regenerate-key', requireAuth, async (req, res) => {});// ── Get Stream Details ───────────────────────────────────────--router.get('/:id', optionalAuth, (req, res) => {++router.get('/:id', optionalAuth, async (req, res) => {try {const stream = db.getStreamById(req.params.id);if (!stream) return res.status(404).json({ error: 'Stream not found' });@@@ -1995,8 +1997,9 @@ router.get('/:id', optionalAuth, (req, res) => {? db.getPipOverlayForManagedStream(stream.managed_stream_id): null;-- if (req.user) stream.isFollowing = db.isFollowing(req.user.id, stream.user_id);-- stream.follower_count = db.getFollowerCount(stream.user_id);++ const networkFollows = require('../social/network-follows');++ if (req.user) stream.isFollowing = networkFollows.isFollowing(req.user.id, stream.user_id);++ stream.follower_count = await networkFollows.followerCount(stream.user_id);res.json({ stream });} catch (err) {@@@ -2389,48 +2392,39 @@ router.get('/:id/rtmp-status', requireAuth, (req, res) => {});// ── Follow/Unfollow Streamer ─────────────────────────────────++// Network owns follows (ADR-030, plan T2): a click follows or unfollows there, and the answer (following, the++// follower count) is Network's. Network sends the FOLLOW notification; Live adds the OpenCoins and the PowerChat++// alert for a follow Network says this click started (never for a repeat of a follow Network already had).++async function toggleFollow(req, res, streamerId) {++ const follows = require('../social/network-follows');++ const w = await follows.set(req.user.id, streamerId, !follows.isFollowing(req.user.id, streamerId));++ if (!w.ok) {++ if (w.detail) console.warn('[Streaming] follow:', w.detail);++ return res.status(w.status).json({ error: w.error });++ }++ if (w.started) {++ // Award OpenCoins for following++ try {++ const openvibeCoins = require('../monetization/opencoins');++ openvibeCoins.awardFollow(req.user.id, streamerId);++ } catch { /* non-critical */ }++ // Fire a PowerChat follow alert for the streamer (follows:write).++ try {++ const follower = db.getUserById(req.user.id);++ require('../integrations/powerchat-platform').forwardFollow(streamerId, {++ followerName: follower?.display_name || follower?.username || 'Someone',++ externalId: 'u' + req.user.id,++ });++ } catch { /* non-critical */ }++ }++ res.json({ following: w.following, count: w.count });++}++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) });-- } else {-- db.followUser(req.user.id, stream.user_id);-- // Award OpenCoins for following-- try {-- const openvibeCoins = require('../monetization/opencoins');-- openvibeCoins.awardFollow(req.user.id, stream.user_id);-- } catch { /* non-critical */ }-- // Fire a PowerChat follow alert for the streamer (follows:write).-- try {-- const follower = db.getUserById(req.user.id);-- require('../integrations/powerchat-platform').forwardFollow(stream.user_id, {-- followerName: follower?.display_name || follower?.username || 'Someone',-- externalId: 'u' + req.user.id,-- });-- } catch { /* non-critical */ }-- // Notify the followed user. A follow Network took notifies there (its follow graph sends FOLLOW);-- // only a Live-only follow (FOLLOWS_AUTHORITY unset, or a side with no subject) is pushed from here.-- if (!w.network) try {-- const { pushNotification, actorInfo } = require('../utils/notify');-- const follower = db.getUserById(req.user.id);-- pushNotification({-- user_id: stream.user_id,-- type: 'FOLLOW',-- title: 'New Follower',-- message: `${follower?.display_name || follower?.username || 'Someone'} followed you`,-- url: follower?.username ? `${config.baseUrl}/@${encodeURIComponent(follower.username)}` : config.baseUrl,-- ...actorInfo(follower),-- });-- } catch { /* non-critical */ }-- res.json({ following: true, count: db.getFollowerCount(stream.user_id) });-- }++ await toggleFollow(req, res, stream.user_id);} catch (err) {console.error('[Streaming]', err.message);res.status(500).json({ error: 'Failed to follow/unfollow' });@@@ -2442,43 +2436,7 @@ 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) });-- } else {-- db.followUser(req.user.id, user.id);-- // Award OpenCoins for following-- try {-- const openvibeCoins = require('../monetization/opencoins');-- openvibeCoins.awardFollow(req.user.id, user.id);-- } catch { /* non-critical */ }-- // Fire a PowerChat follow alert for the streamer (follows:write).-- try {-- const f = db.getUserById(req.user.id);-- require('../integrations/powerchat-platform').forwardFollow(user.id, {-- followerName: f?.display_name || f?.username || 'Someone',-- externalId: 'u' + req.user.id,-- });-- } catch { /* non-critical */ }-- // Notify the followed user, unless Network took the follow and notifies (see the route above).-- if (!w.network) try {-- const { pushNotification, actorInfo } = require('../utils/notify');-- const follower = db.getUserById(req.user.id);-- pushNotification({-- user_id: user.id,-- type: 'FOLLOW',-- title: 'New Follower',-- message: `${follower?.display_name || follower?.username || 'Someone'} followed you`,-- url: follower?.username ? `${config.baseUrl}/@${encodeURIComponent(follower.username)}` : config.baseUrl,-- ...actorInfo(follower),-- });-- } catch { /* non-critical */ }-- res.json({ following: true, count: db.getFollowerCount(user.id) });-- }++ await toggleFollow(req, res, user.id);} catch (err) {console.error('[Streaming]', err.message);res.status(500).json({ error: 'Failed to follow/unfollow' });ddeleted file mode 100644@@@ -1,78 +0,0 @@--'use strict';----// GET /internal/followers — OpenVibe.Network asks who follows the channel behind a stream when it--// consumes live.stream.started. Service token with live.follower.read, loopback only, paged.----const assert = require('assert');--const fs = require('fs');--const os = require('os');--const path = require('path');--const http = require('http');--const crypto = require('crypto');--const express = require('express');--const { serviceAuth } = require('openvibe-contracts');----const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-followers-'));--const keys = crypto.generateKeyPairSync('rsa', { modulusLength: 2048, publicKeyEncoding: { type: 'spki', format: 'pem' }, privateKeyEncoding: { type: 'pkcs8', format: 'pem' } });--fs.writeFileSync(path.join(tmp, 'network.pem'), keys.publicKey);--process.env.DB_PATH = path.join(tmp, 'live.db');--process.env.OV_NETWORK_PUBLIC_KEY = path.join(tmp, 'network.pem');--process.env.OV_NETWORK_URL = 'https://openvibe.network';--console.log = () => {};--console.warn = () => {};----const ISS = 'https://openvibe.network';--const now = () => Math.floor(Date.now() / 1000);--let jti = 0;--const token = (cap) => serviceAuth.signServiceToken({ iss: ISS, sub: 'svc:network', actor_type: 'service', aud: ['openvibe.live'], cap, iat: now(), exp: now() + 300, jti: `tok_net_${++jti}` }, keys.privateKey);----(async () => {-- const db = require('../server/db/database');-- db.initDb();-- const d = db.getDb();-- for (const [id, name] of [[601, 'streamer'], [602, 'fan_a'], [603, 'fan_b'], [604, 'fan_c']]) {-- d.prepare("INSERT INTO users (id, username, display_name, password_hash) VALUES (?, ?, ?, 'x')").run(id, name, name);-- }-- d.prepare("INSERT INTO linked_accounts (user_id, service, service_user_id, subject_id) VALUES (601, 'network', '70', 'usr_01J0000000000000000000000S')").run();-- d.prepare("INSERT INTO linked_accounts (user_id, service, service_user_id, subject_id) VALUES (602, 'network', '71', 'usr_01J0000000000000000000000A')").run();-- d.prepare("INSERT INTO linked_accounts (user_id, service, service_user_id) VALUES (603, 'network', '72')").run();-- for (const f of [602, 603, 604]) d.prepare('INSERT INTO follows (follower_id, streamer_id) VALUES (?, 601)').run(f);-- const streamId = d.prepare("INSERT INTO streams (user_id, title, is_live) VALUES (601, 'hello', 1)").run().lastInsertRowid;---- const app = express();-- app.use('/internal/followers', require('../server/streaming/followers-internal'));-- const server = await new Promise((r) => { const s = http.createServer(app); s.listen(0, '127.0.0.1', () => r(s)); });-- const base = `http://127.0.0.1:${server.address().port}`;-- const get = (q, { cap = ['live.follower.read'], headers = {} } = {}) => fetch(`${base}/internal/followers?${q}`, {-- headers: { Authorization: `Bearer ${token(cap)}`, ...headers },-- }).then(async (r) => ({ status: r.status, json: await r.json().catch(() => null) }));---- assert.strictEqual((await get(`stream_id=${streamId}`, { cap: ['live.chat_effects.write'] })).status, 403, 'needs live.follower.read');-- assert.strictEqual((await get(`stream_id=${streamId}`, { headers: { 'X-Forwarded-For': '1.2.3.4' } })).status, 403, 'loopback only');-- assert.strictEqual((await fetch(`${base}/internal/followers?stream_id=${streamId}`)).status, 401, 'no token');-- assert.strictEqual((await get('stream_id=abc')).status, 400);-- assert.strictEqual((await get('stream_id=999999')).status, 404);---- const all = await get(`stream_id=${streamId}`);-- assert.strictEqual(all.status, 200);-- assert.strictEqual(all.json.channel.subject, 'usr_01J0000000000000000000000S');-- assert.deepStrictEqual(all.json.followers, [-- { subject: 'usr_01J0000000000000000000000A', network_user_id: 71 },-- { subject: null, network_user_id: 72 },-- { subject: null, network_user_id: null },-- ]);-- assert.strictEqual(all.json.next, null);---- // Paging: two per page, then the rest.-- const p1 = await get(`stream_id=${streamId}&limit=2`);-- assert.strictEqual(p1.json.followers.length, 2);-- assert.ok(p1.json.next);-- const p2 = await get(`stream_id=${streamId}&limit=2&after=${p1.json.next}`);-- assert.strictEqual(p2.json.followers.length, 1);-- assert.strictEqual(p2.json.next, null);---- server.close();-- fs.rmSync(tmp, { recursive: true, force: true });-- process.stdout.write('followers-internal: all passed\n');-- process.exit(0);--})().catch((e) => { process.stderr.write(`${e.stack}\n`); process.exit(1); });@@@ -297,7 +297,7 @@ function checkContract(out, label) {return { status: res.status, json: await res.json().catch(() => null), cache: res.headers.get('cache-control') };};assert.strictEqual((await fetch(`${base}?slug=alice`)).status, 401, 'no token');-- assert.strictEqual((await call('GET', 'slug=alice', null, { cap: ['live.follower.read'] })).status, 403, 'needs live.lineage.resolve');++ assert.strictEqual((await call('GET', 'slug=alice', null, { cap: ['live.chat_context.read'] })).status, 403, 'needs live.lineage.resolve');assert.strictEqual((await call('GET', 'slug=alice', null, { headers: { 'X-Forwarded-For': '1.2.3.4' } })).status, 403, 'loopback only');let res = await call('GET', 'slug=alice&vod_id=42&live_user_id=17');assert.strictEqual(res.status, 200);@@@ -46,7 +46,7 @@ const SEED = `db.ensureChannel(star); db.ensureChannel(fan);d.prepare("INSERT INTO streams (user_id, title, category, protocol, is_live, viewer_count, started_at, last_heartbeat) VALUES (?, 'N-1 live stream', 'tech', 'webrtc', 1, 2, datetime('now', '-10 minutes'), datetime('now'))").run(star);d.prepare("INSERT INTO streams (user_id, title, category, protocol, is_live, started_at, ended_at) VALUES (?, 'N-1 past stream', 'irl', 'webrtc', 0, datetime('now', '-2 days'), datetime('now', '-2 days', '+1 hour'))").run(star);-- db.followUser(fan, star);++ d.prepare('INSERT INTO follows (follower_id, streamer_id) VALUES (?, ?)').run(fan, star);d.prepare("INSERT INTO chat_messages (stream_id, user_id, username, message) VALUES (1, ?, 'n1fan', 'hello from N-1')").run(fan);db.close();`;@@@ -1,23 +1,39 @@'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.++// Follows on OpenVibe.Network (ADR-030, plan T2; server/social/network-follows.js): a follow button follows or++// unfollows on Network (network.follows.write) and shows Network's answer; only a follow Network says the click++// started earns the coins and the PowerChat alert; a side without a subject cannot follow; Network down or a++// malformed answer changes nothing; channel and stream responses show Network's count and the viewer's follow++// while the projection lags; network.follow.* events alone write Live's table, in revision order, so a late older++// event cannot undo a click. Nothing else on Live writes follows, and GET /internal/followers is gone.const assert = require('assert');const fs = require('fs');const os = require('os');const path = require('path');++const http = require('http');const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-netfollows-'));process.env.DB_PATH = path.join(tmp, 'live.db');++process.env.NODE_ENV = 'test';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' });++let invalidated = 0;++principal.invalidate = () => { invalidated++; };const follows = require('../server/social/network-follows');const events = require('../server/auth/network-events');++const auth = require('../server/auth/auth');++auth.requireAuth = (req, res, next) => { const u = db.getUserById(Number(req.headers['x-test-user'] || 0)); if (!u) return res.status(401).json({ error: 'Authentication required' }); req.user = u; next(); };++auth.optionalAuth = (req, res, next) => { const u = db.getUserById(Number(req.headers['x-test-user'] || 0)); if (u) req.user = u; next(); };++const coins = require('../server/monetization/opencoins');++const powerchat = require('../server/integrations/powerchat-platform');++const side = [];++coins.awardFollow = (a, b) => side.push(['coins', a, b]);++powerchat.forwardFollow = (s) => side.push(['powerchat', s]);++const notify = require('../server/utils/notify');++notify.pushNotification = (n) => side.push(['notify', n.type]);const sub = (tag) => `usr_${`01J${tag}`.padEnd(26, '0')}`;const ANN = sub('AA'), BOB = sub('BB'), CAT = sub('CC'), GHOST = sub('DD');@@@ -25,58 +41,216 @@ 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);++const streamId = Number(db.createStream({ user_id: 2, title: 't', protocol: 'webrtc' }).lastInsertRowid);--(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 }; };++// Network: /internal/follows answers network.follow-status-result@1 from its own graph (201 for a pair's first++// follow; a follow that starts again starts its `since` again), and /api/v1/follows answers the public count.++const graph = new Map(); // `${follower}>${target}` → { active, revision, since }++const calls = [];++let fail = null;++const countOf = (target) => [...graph].filter(([k, v]) => v.active && k.endsWith(`>${target}`)).length + 40;++globalThis.fetch = async (url, o = {}) => {++ const method = o.method || 'GET';++ calls.push({ url, method, body: o.body ? JSON.parse(o.body) : null, auth: o.headers && o.headers.Authorization });++ if (fail instanceof Error) throw fail;++ if (fail) return { status: fail.status, json: async () => fail.body || {} };++ const u = new URL(url);++ const target = u.pathname.split('/').pop();++ if (method === 'GET') return { status: 200, json: async () => ({ target_type: 'channel', target_id: target, followers: countOf(target) }) };++ const follower = method === 'PUT' ? JSON.parse(o.body).follower : u.searchParams.get('follower');++ const key = `${follower}>${target}`;++ const prev = graph.get(key);++ let status = 200;++ if (method === 'PUT' && !(prev && prev.active)) {++ const revision = prev ? prev.revision + 1 : 1;++ graph.set(key, { active: true, revision, since: new Date().toISOString() });++ if (revision === 1) status = 201;++ } else if (method === 'DELETE' && prev && prev.active) graph.set(key, { ...prev, active: false, revision: prev.revision + 1 });++ const now = graph.get(key);++ const body = { target_type: 'channel', target_id: target, followers: countOf(target), following: !!(now && now.active) };++ if (now && now.active) body.since = now.since;++ return { status, json: async () => body };++};-- // 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);++const express = require('express');++const app = express();++app.use(express.json());++app.use('/api/streams', require('../server/streaming/routes'));++const server = http.createServer(app);++const call = (method, p, user) => new Promise((resolve, reject) => {++ const req = http.request(`http://127.0.0.1:${server.address().port}${p}`, { method, headers: user ? { 'x-test-user': String(user) } : {} }, (res) => {++ let b = ''; res.on('data', (c) => { b += c; }); res.on('end', () => resolve({ status: res.statusCode, body: b ? JSON.parse(b) : null }));++ });++ req.on('error', reject); req.end();++});++const post = (p, user) => call('POST', p, user);++const get = (p, user) => call('GET', p, user);++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 } });-- 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);++server.listen(0, '127.0.0.1', async () => {++ try {++ // set(): Network first; its answer is what the button shows, and the projection waits for the event.++ let r = await follows.set(1, 2, true);++ assert.deepStrictEqual(r, { ok: true, following: true, count: 41, started: true });++ assert.deepStrictEqual(calls.pop(), { url: `http://network.test/internal/follows/channel/${BOB}`, method: 'PUT', body: { follower: ANN }, auth: 'Bearer svc-token' });++ assert.ok(!db.isFollowing(1, 2), 'a write answer never touches the projection');++ assert.ok(follows.isFollowing(1, 2), 'the button shows Network\'s answer before the event');++ assert.strictEqual((await follows.set(1, 2, true)).started, false, 'the same follow again starts nothing');++ r = await follows.set(1, 2, false);++ assert.deepStrictEqual(calls.pop(), { url: `http://network.test/internal/follows/channel/${BOB}?follower=${ANN}`, method: 'DELETE', body: null, auth: 'Bearer svc-token' });++ assert.deepStrictEqual(r, { ok: true, following: false, count: 40, started: false });++ assert.ok(!follows.isFollowing(1, 2));++ assert.strictEqual((await follows.set(1, 2, true)).started, true, 'a follow that starts again (200, a new since) is started');++ await follows.set(1, 2, false);-- // 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');++ // Network cannot answer, or answers a malformed document: 503, nothing changes on Live.++ fail = { status: 503 };++ r = await follows.set(1, 2, true);++ assert.strictEqual(r.ok, false); assert.strictEqual(r.status, 503);++ fail = new Error('ECONNREFUSED');++ assert.strictEqual((await follows.set(1, 2, true)).status, 503);++ fail = { status: 401 };++ assert.strictEqual((await follows.set(1, 2, true)).status, 503);++ assert.strictEqual(invalidated, 1, 'a refused service token is dropped');++ fail = { status: 200, body: { nope: true } };++ assert.strictEqual((await follows.set(1, 2, true)).status, 503, 'an answer without `following` is no answer');++ for (const followers of [undefined, '41', -1, 4.5]) {++ fail = { status: 200, body: { target_type: 'channel', target_id: BOB, following: true, followers } };++ assert.strictEqual((await follows.set(1, 2, true)).status, 503, `an answer with followers ${followers} is no answer`);++ }++ fail = { status: 200, body: { target_type: 'channel', target_id: CAT, following: true, followers: 41 } };++ assert.strictEqual((await follows.set(1, 2, true)).status, 503, 'an answer about another channel is no answer');++ assert.ok(!follows.isFollowing(1, 2) && !db.isFollowing(1, 2), 'nothing remembered while Network did not take it');++ // Network refuses the request itself: its status and reason.++ fail = { status: 400, body: { code: 'follows.self', detail: 'you cannot follow yourself' } };++ assert.deepStrictEqual(await follows.set(2, 2, true), { ok: false, status: 400, error: 'you cannot follow yourself' });++ fail = null;++ // A side without a subject cannot follow; Network is not asked.++ const n = calls.length;++ assert.strictEqual((await follows.set(4, 2, true)).status, 409, 'follower without a subject');++ assert.strictEqual((await follows.set(1, 4, true)).status, 409, 'channel without a subject');++ assert.strictEqual(calls.length, n);++ assert.ok(!follows.isFollowing(4, 2) && !follows.isFollowing(1, 4));-- // 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\)/);-- // A follow Network took is notified by Network (FOLLOW from its follow graph); Live pushes only a Live-only one.-- const whole = routes.slice(routes.indexOf(route), routes.indexOf('\nrouter.', routes.indexOf(route) + 1));-- assert.match(whole, /if \(!w\.network\) try \{\s*const \{ pushNotification/, `${route} leaves a Network follow's notification to Network`);-- assert.strictEqual((whole.match(/type: 'FOLLOW'/g) || []).length, 1);++ // The buttons: toggle on the state the viewer sees, answer Network's state and count.++ assert.strictEqual((await post(`/api/streams/${streamId}/follow`)).status, 401, 'signed in only');++ side.length = 0;++ r = await post(`/api/streams/${streamId}/follow`, 3);++ assert.deepStrictEqual(r, { status: 200, body: { following: true, count: 41 } });++ assert.deepStrictEqual(side, [['coins', 3, 2], ['powerchat', 2]], 'coins and the PowerChat alert; Network sends FOLLOW');++ r = await post('/api/streams/channel/bob/follow', 3);++ assert.deepStrictEqual(r, { status: 200, body: { following: false, count: 40 } }, 'the second click unfollows before any event');++ side.length = 0;++ // Network already has the follow (made elsewhere, its event delayed): the click is an idempotent PUT, and++ // nothing started, so no coins and no false PowerChat alert.++ graph.set(`${CAT}>${BOB}`, { active: true, revision: 3, since: '2026-01-01T00:00:00.000Z' });++ follows._reset();++ r = await post('/api/streams/channel/bob/follow', 3);++ assert.deepStrictEqual(r.body, { following: true, count: 41 });++ assert.deepStrictEqual(side, [], 'a follow Network already had starts nothing');++ fail = { status: 502 };++ r = await post('/api/streams/channel/bob/follow', 3);++ assert.strictEqual(r.status, 503);++ assert.ok(follows.isFollowing(3, 2), 'Network down: the follow stays as it was');++ fail = null;++ assert.strictEqual((await post('/api/streams/channel/bob/follow', 4)).status, 409);++ assert.strictEqual((await post('/api/streams/channel/nobody/follow', 3)).status, 404);++ assert.strictEqual((await post('/api/streams/999999/follow', 3)).status, 404);++ assert.ok(!side.some((s) => s[0] === 'notify'), 'Live sends no follow notification');++++ // Channel and stream responses while the projection lags: Network's count, the viewer's follow as Network++ // answered it (the projection still has no row for cat → bob).++ assert.ok(!db.isFollowing(3, 2));++ follows._reset();++ await follows.set(3, 2, true);++ graph.set(`${ANN}>${BOB}`, { active: true, revision: 7, since: '2026-01-01T00:00:00.000Z' }); // a follow made elsewhere++ follows._reset();++ await follows.set(3, 2, true);++ calls.length = 0;++ r = await get('/api/streams/channel/bob', 3);++ assert.strictEqual(r.status, 200, JSON.stringify(r.body));++ assert.strictEqual(r.body.channel.follower_count, 42, 'the count is Network\'s');++ assert.strictEqual(r.body.channel.is_following, true);++ r = await get(`/api/streams/${streamId}`, 3);++ assert.strictEqual(r.status, 200);++ assert.strictEqual(r.body.stream.follower_count, 42);++ assert.strictEqual(r.body.stream.isFollowing, true);++ assert.strictEqual(calls.filter((c) => c.url.includes('/api/v1/follows/')).length, 0, 'the write answer\'s count is reused');++ follows._reset();++ calls.length = 0;++ r = await get('/api/streams/channel/bob', 1);++ assert.strictEqual(r.body.channel.follower_count, 42);++ assert.strictEqual(r.body.channel.is_following, false, 'no answer for ann: the projection');++ assert.deepStrictEqual(calls.filter((c) => c.url.includes('/api/v1/follows/')), [{ url: `http://network.test/api/v1/follows/channel/${BOB}`, method: 'GET', body: null, auth: undefined }], 'the public count, no credentials');++ r = await get('/api/streams/channel/bob');++ assert.strictEqual(r.body.channel.is_following, false, 'signed out');++ // Network cannot answer the count, or answers a malformed one: the projection's count for a while.++ follows._reset();++ fail = new Error('ECONNREFUSED');++ assert.strictEqual(await follows.followerCount(2), 0);++ fail = null;++ assert.strictEqual(await follows.followerCount(2), 0, 'the projection answers while Network was just down');++ follows._reset();++ fail = { status: 200, body: { target_type: 'channel', target_id: BOB, followers: 'many' } };++ assert.strictEqual(await follows.followerCount(2), 0);++ fail = null;++ follows._reset();++ assert.strictEqual(await follows.followerCount(2), 42);++ assert.strictEqual(await follows.followerCount(4), 0, 'a channel without a subject: the projection');++++ // The projection, through the Network events endpoint's apply().++ d.prepare('DELETE FROM follows').run();++ follows._reset();++ 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');++++ // A click, then its events delivered late and out of order: an older follow event arriving after the++ // unfollow click cannot show the follow again, and the unfollow's own event retires the remembered answer.++ graph.delete(`${ANN}>${CAT}`);++ assert.strictEqual((await follows.set(1, 3, true)).started, true); // Network revision 1++ assert.strictEqual((await follows.set(1, 3, false)).following, false); // Network revision 2++ assert.strictEqual(events.apply(ev('network.follow.created', ANN, CAT, 1)), 'followed', 'the late follow event');++ assert.ok(db.isFollowing(1, 3), 'the projection is at revision 1');++ assert.strictEqual(follows.isFollowing(1, 3), false, 'the button still shows the unfollow');++ r = await get('/api/streams/channel/cat', 1);++ assert.strictEqual(r.body.channel.is_following, false);++ assert.strictEqual(events.apply(ev('network.follow.deleted', ANN, CAT, 2)), 'unfollowed');++ assert.ok(!db.isFollowing(1, 3) && !follows.isFollowing(1, 3));++ assert.strictEqual(events.apply(ev('network.follow.created', ANN, CAT, 1)), 'stale');++ assert.strictEqual(events.apply(ev('network.follow.created', ANN, CAT, 3)), 'followed', 'a later follow made elsewhere');++ assert.strictEqual(follows.isFollowing(1, 3), true, 'the agreeing event retired the answer: the projection shows it');++ // 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'/);++++ // Only the projection writes follows (and the account merge repoints it); no Live-only path is left.++ const root = path.join(__dirname, '..', 'server');++ const writers = [];++ const walk = (dir) => { for (const e of fs.readdirSync(dir, { withFileTypes: true })) {++ const p = path.join(dir, e.name);++ if (e.isDirectory()) walk(p);++ else if (e.name.endsWith('.js') && /\b(INSERT(\s+OR\s+\w+)?\s+INTO|DELETE\s+FROM|UPDATE)\s+follows\b/i.test(fs.readFileSync(p, 'utf8'))) writers.push(path.relative(root, p));++ } };++ walk(root);++ assert.deepStrictEqual(writers.sort(), ['auth/subject-merge.js', 'social/network-follows.js'], 'follows writers');++ assert.strictEqual(typeof db.followUser, 'undefined');++ assert.strictEqual(typeof db.unfollowUser, 'undefined');++ assert.ok(!/FOLLOWS_AUTHORITY/.test(fs.readFileSync(path.join(root, 'social', 'network-follows.js'), 'utf8')), 'no switch back to Live-only follows');++ // GET /internal/followers is retired: Network reads its own graph (Network PR #6).++ assert.ok(!fs.existsSync(path.join(root, 'streaming', 'followers-internal.js')));++ assert.ok(!fs.readFileSync(path.join(root, 'index.js'), 'utf8').includes('/internal/followers'));++ console.log = log;++ console.log('network follows: all checks passed');++ } finally {++ server.close();++ fs.rmSync(tmp, { recursive: true, force: true });}-- // 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); });++});++process.on('unhandledRejection', (e) => { console.log = log; console.error(e); process.exit(1); });
@@ -90,7 +90,7 @@ The media subsystem lives in **OpenVibe.Media**:
@@ -90,7 +90,7 @@ The media subsystem lives in **OpenVibe.Media**:
## Capabilities
## Capabilities
-Live implements, for other services' tokens (manifest `manifests/services/live.json` in OpenVibe.Contracts): `live.chat_context.read`, `live.chat_effects.write`, `live.chat_mirror.write`, `live.follower.read`, `live.lineage.resolve`, `live.tips_delivery.write`, `live.channel.read`, `live.stream.read`, `live.discovery.read`, `live.owner.resolve`.
+Live implements, for other services' tokens (manifest `manifests/services/live.json` in OpenVibe.Contracts): `live.chat_context.read`, `live.chat_effects.write`, `live.chat_mirror.write`, `live.lineage.resolve`, `live.tips_delivery.write`, `live.channel.read`, `live.stream.read`, `live.discovery.read`, `live.owner.resolve`.
Live's own principal (`live`) holds grants to call:
Live's own principal (`live`) holds grants to call:
@@ -11,7 +11,6 @@
@@ -11,7 +11,6 @@
"live.chat_context.read",
"live.chat_context.read",
"live.chat_effects.write",
"live.chat_effects.write",
"live.chat_mirror.write",
"live.chat_mirror.write",
- "live.follower.read",
"live.lineage.resolve",
"live.lineage.resolve",
"live.tips_delivery.write",
"live.tips_delivery.write",
"live.channel.read",
"live.channel.read",
@@ -4786,18 +4786,7 @@ function updateUserAvatar(userId, avatarUrl, pasteId = null) {
@@ -4786,18 +4786,7 @@ function updateUserAvatar(userId, avatarUrl, pasteId = null) {
// resetAvatarsForPaste() removed — the media subsystem (vods/clips/pastes writes) moved to OpenVibe.Media.
// resetAvatarsForPaste() removed — the media subsystem (vods/clips/pastes writes) moved to OpenVibe.Media.
// ── Follow helpers ───────────────────────────────────────────
// ── Follow helpers ───────────────────────────────────────────
-
-function followUser(followerId, streamerId) {
- return run(
- `INSERT OR IGNORE INTO follows (follower_id, streamer_id) VALUES (?, ?)`,
- [followerId, streamerId]
- );
-}
-
-function unfollowUser(followerId, streamerId) {
- return run(`DELETE FROM follows WHERE follower_id = ? AND streamer_id = ?`,
- [followerId, streamerId]);
-}
+// Reads of the projection of Network's follow graph; only server/social/network-follows.js writes it.
function getFollowerCount(streamerId) {
function getFollowerCount(streamerId) {
const row = get('SELECT COUNT(*) as count FROM follows WHERE streamer_id = ?', [streamerId]);
const row = get('SELECT COUNT(*) as count FROM follows WHERE streamer_id = ?', [streamerId]);
@@ -7067,7 +7056,7 @@ module.exports = {
@@ -7067,7 +7056,7 @@ module.exports = {
getKickChannelCache, setKickChannelCache,
getKickChannelCache, setKickChannelCache,
getChannelPoints, addChannelPoints, deductChannelPoints, applyChannelPoints,
getChannelPoints, addChannelPoints, deductChannelPoints, applyChannelPoints,
// Follows
// Follows
- followUser, unfollowUser, getFollowerCount, isFollowing, getFollowerIds,
+ getFollowerCount, isFollowing, getFollowerIds,
// Transactions (Vibes)
// Transactions (Vibes)
createTransaction, addVibes, deductVibes, addVibesCashout, deductVibesCashout,
createTransaction, addVibes, deductVibes, addVibesCashout, deductVibesCashout,
// OpenCoins
// OpenCoins
@@ -617,9 +617,6 @@ app.use('/internal/chat-context', chatLiveContext.contextRouter);
@@ -617,9 +617,6 @@ app.use('/internal/chat-context', chatLiveContext.contextRouter);
app.use('/internal/chat-effects', chatLiveContext.effectsRouter);
app.use('/internal/chat-effects', chatLiveContext.effectsRouter);
// OpenVibe.Tips announces settled tips in the creator's chat here (service token, live.tips_delivery.write).
// OpenVibe.Tips announces settled tips in the creator's chat here (service token, live.tips_delivery.write).
app.use('/internal/tips', require('./tips/delivery-routes'));
app.use('/internal/tips', require('./tips/delivery-routes'));
-// OpenVibe.Network sends go-live notifications from live.stream.started and asks here who follows
-// the channel (service token, live.follower.read).
-app.use('/internal/followers', require('./streaming/followers-internal'));
// The canonical channel/owner resolver for other services (roadmap D20; service token, live.lineage.resolve).
// The canonical channel/owner resolver for other services (roadmap D20; service token, live.lineage.resolve).
app.use('/internal/lineage', require('./lineage/routes'));
app.use('/internal/lineage', require('./lineage/routes'));
app.use('/internal', require('./internal/routes'));
app.use('/internal', require('./internal/routes'));
@@ -265,7 +265,7 @@ async function _channelMeta(username, page = 1) {
@@ -265,7 +265,7 @@ async function _channelMeta(username, page = 1) {
let ov = null;
let ov = null;
try { ov = db.getStreamerOverview ? db.getStreamerOverview(acct.userId) : null; } catch { /* */ }
try { ov = db.getStreamerOverview ? db.getStreamerOverview(acct.userId) : null; } catch { /* */ }
let followers = 0;
let followers = 0;
- try { followers = db.getFollowerCount ? (db.getFollowerCount(acct.userId) || 0) : 0; } catch { /* */ }
+ try { followers = (await require('../social/network-follows').followerCount(acct.userId)) || 0; } catch { /* */ }
let live = [];
let live = [];
try { live = acct.banned ? [] : (db.getLiveStreamsByUserId(acct.userId) || []); } catch { live = []; }
try { live = acct.banned ? [] : (db.getLiveStreamsByUserId(acct.userId) || []); } catch { live = []; }
const bio = clean(acct.bio || '', 300);
const bio = clean(acct.bio || '', 300);
@@ -1,27 +1,44 @@
@@ -1,27 +1,44 @@
'use strict';
'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).
+ * Follows on OpenVibe.Network (ADR-030; plan T2). Network owns the follow graph, keyed by subjects. Live's
+ * `follows` table is only its projection, written by Network's events alone (in revision order) and kept for
+ * Live's own bulk reads (go-live notices, Chat's context route, activity numbers); nothing on Live follows or
+ * unfollows on its own, and what a person sees (a follow button, a follower count) comes from Network.
*
*
- * 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.
+ * set(followerId, streamerId, following) a follow button: PUT / DELETE /internal/follows/channel/<streamer
+ * subject> on Network (Live's service token, network.follows.write). Network's answer
+ * (network.follow-status-result@1: following, the follower count) is what the button shows, and Live
+ * remembers it (ANSWER_TTL_MS) so a reload or a second click sees it before the event arrives. The answer
+ * carries no revision, so it never touches the projection: an older event delivered late cannot undo it.
+ * `started` is Network's own word that this request started the follow: 201 (the pair's first follow),
+ * or a `since` no older than the request (a follow that starts again starts its since again); an
+ * idempotent repeat of a follow Network already had is not started, whatever the projection says.
+ * Both sides need a Network subject: a side without one cannot follow (409). Network unreachable,
+ * refusing Live's token, or answering a malformed document: 503. A request Network refuses for what it
+ * is (following oneself, the follow limit, an unknown channel) keeps Network's status and reason.
+ * isFollowing(followerId, streamerId) the button state: Network's recent answer for the pair, else the
+ * projection.
+ * followerCount(streamerId) GET /api/v1/follows/channel/<subject> on Network (the public count), cached
+ * COUNT_TTL_MS and refreshed by every write answer; the projection's count only while Network cannot
+ * answer (or the channel has no subject).
* apply(event) network.follow.created / .deleted from OpenVibe.Events (server/auth/network-events.js):
* 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`.
+ * follows made anywhere (the buttons here, 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`. An applied event that agrees with Network's remembered answer retires the answer.
*
*
- * Unset FOLLOWS_AUTHORITY: the buttons write Live's table only (the state before step 4); events still apply.
+ * Network tells the followed person (FOLLOW) itself; Live sends no follow notification.
*/
*/
const db = require('../db/database');
const db = require('../db/database');
const NETWORK_INTERNAL_URL = (process.env.OV_NETWORK_INTERNAL_URL || 'http://127.0.0.1:4000').replace(/\/+$/, '');
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 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');
+const ANSWER_TTL_MS = 10 * 60 * 1000;
+const COUNT_TTL_MS = 30 * 1000; // Network's own public max-age for the count
+const COUNT_DOWN_MS = 10 * 1000; // after a failed count read, the projection answers for this long
+const stats = { written: 0, no_subject: 0, refused: 0, failed: 0, applied: 0, stale: 0, ignored: 0, count_read: 0, count_fallback: 0 };
+const answers = new Map(); // `${followerId}>${streamerId}` → { following, at }
+const counts = new Map(); // streamer subject → { n, at } | { down: until }
function subjectOf(liveUserId) {
function subjectOf(liveUserId) {
const r = db.getDb().prepare("SELECT subject_id FROM linked_accounts WHERE service = 'network' AND user_id = ?").get(liveUserId);
const r = db.getDb().prepare("SELECT subject_id FROM linked_accounts WHERE service = 'network' AND user_id = ?").get(liveUserId);
@@ -32,30 +49,95 @@ function userOf(subject) {
@@ -32,30 +49,95 @@ function userOf(subject) {
return r ? r.user_id : null;
return r ? r.user_id : null;
}
}
+function project(followerId, streamerId, following) {
+ if (following) db.getDb().prepare('INSERT OR IGNORE INTO follows (follower_id, streamer_id) VALUES (?, ?)').run(followerId, streamerId);
+ else db.getDb().prepare('DELETE FROM follows WHERE follower_id = ? AND streamer_id = ?').run(followerId, streamerId);
+}
+
+/** A network.follow-status-result@1 for `target` with a valid count (and `following`, when `viewer`). */
+function validStatus(body, target, viewer) {
+ return !!body && typeof body === 'object' && body.target_id === target && Number.isInteger(body.followers) && body.followers >= 0
+ && (!viewer || typeof body.following === 'boolean');
+}
+
/**
/**
- * 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).
+ * Follow (following = true) or unfollow on Network.
+ * → { ok: true, following, count, started }, or { ok: false, status, error } (the caller answers that status).
*/
*/
-async function writeThrough(followerId, streamerId, following, { fetchImpl = globalThis.fetch } = {}) {
- if (authority() !== 'network') return { ok: true, network: false, reason: 'authority_live' };
+async function set(followerId, streamerId, following, { fetchImpl = globalThis.fetch } = {}) {
const follower = subjectOf(followerId);
const follower = subjectOf(followerId);
const target = subjectOf(streamerId);
const target = subjectOf(streamerId);
- if (!follower || !target) { stats.local_only++; return { ok: true, network: false, reason: 'no_subject' }; }
+ if (!follower || !target) {
+ stats.no_subject++;
+ return { ok: false, status: 409, error: follower ? 'This channel cannot be followed yet' : 'Sign out and back in to follow channels' };
+ }
+ let res, body;
+ const sentAt = Date.now();
try {
try {
const principal = require('../net/network-principal');
const principal = require('../net/network-principal');
const headers = { Accept: 'application/json', ...(await principal.serviceHeaders('openvibe.network')) };
const headers = { Accept: 'application/json', ...(await principal.serviceHeaders('openvibe.network')) };
const url = `${NETWORK_INTERNAL_URL}/internal/follows/channel/${target}`;
const url = `${NETWORK_INTERNAL_URL}/internal/follows/channel/${target}`;
- const res = following
+ res = following
? await fetchImpl(url, { method: 'PUT', headers: { ...headers, 'Content-Type': 'application/json' }, body: JSON.stringify({ follower }), signal: AbortSignal.timeout(5000) })
? 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) });
: await fetchImpl(`${url}?follower=${follower}`, { method: 'DELETE', headers, signal: AbortSignal.timeout(5000) });
if (res.status === 401) principal.invalidate('openvibe.network');
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 };
+ body = await res.json().catch(() => null);
} catch (err) {
} catch (err) {
stats.failed++;
stats.failed++;
- return { ok: false, error: err.message };
+ return { ok: false, status: 503, error: 'Could not update the follow right now; try again in a moment', detail: err.message };
+ }
+ if ([400, 404, 409].includes(res.status)) {
+ stats.refused++;
+ return { ok: false, status: res.status, error: (body && typeof body.detail === 'string' && body.detail) || 'Could not update the follow' };
+ }
+ if (res.status < 200 || res.status >= 300 || !validStatus(body, target, true)) {
+ stats.failed++;
+ return { ok: false, status: 503, error: 'Could not update the follow right now; try again in a moment', detail: `Network answered ${res.status} without a valid follow status` };
+ }
+ stats.written++;
+ const now = Date.now();
+ if (answers.size > 5000) for (const [k, a] of answers) if (now - a.at >= ANSWER_TTL_MS) answers.delete(k);
+ answers.set(`${followerId}>${streamerId}`, { following: body.following, at: now });
+ counts.set(target, { n: body.followers, at: now });
+ const since = typeof body.since === 'string' ? Date.parse(body.since) : NaN;
+ return {
+ ok: true,
+ following: body.following,
+ count: body.followers,
+ started: body.following && (res.status === 201 || since >= sentAt),
+ };
+}
+
+/** Whether followerId follows streamerId, as the button shows it (sync). */
+function isFollowing(followerId, streamerId) {
+ if (followerId == null) return false;
+ const key = `${followerId}>${streamerId}`;
+ const a = answers.get(key);
+ if (a && Date.now() - a.at < ANSWER_TTL_MS) return a.following;
+ if (a) answers.delete(key);
+ return db.isFollowing(followerId, streamerId);
+}
+
+/** streamerId's follower count, from Network (cached), or the projection's while Network cannot answer. */
+async function followerCount(streamerId, { fetchImpl = globalThis.fetch } = {}) {
+ const target = subjectOf(streamerId);
+ const local = () => { stats.count_fallback++; return db.getFollowerCount(streamerId); };
+ if (!target) return local();
+ const now = Date.now();
+ const c = counts.get(target);
+ if (c && c.down > now) return local();
+ if (c && c.n != null && now - c.at < COUNT_TTL_MS) return c.n;
+ try {
+ const res = await fetchImpl(`${NETWORK_INTERNAL_URL}/api/v1/follows/channel/${target}`, { headers: { Accept: 'application/json' }, signal: AbortSignal.timeout(1500) });
+ const body = res.status === 200 ? await res.json().catch(() => null) : null;
+ if (!validStatus(body, target, false)) throw new Error(`Network answered ${res.status} without a valid follow status`);
+ stats.count_read++;
+ counts.set(target, { n: body.followers, at: Date.now() });
+ return body.followers;
+ } catch (err) {
+ console.warn('[Follows] follower count from the projection:', err.message);
+ counts.set(target, { down: Date.now() + COUNT_DOWN_MS });
+ return local();
}
}
}
}
@@ -83,13 +165,13 @@ function apply(ev) {
@@ -83,13 +165,13 @@ function apply(ev) {
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')
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);
.run(p.follower, p.target_id, p.revision);
stats.applied++;
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';
+ const following = ev.event_type === 'network.follow.created';
+ project(followerId, streamerId, following);
+ const key = `${followerId}>${streamerId}`;
+ if (answers.has(key) && answers.get(key).following === following) answers.delete(key);
+ counts.delete(p.target_id);
+ return following ? 'followed' : 'unfollowed';
})();
})();
}
}
-module.exports = { writeThrough, apply, authority, stats };
+module.exports = { set, apply, isFollowing, followerCount, stats, _reset: () => { answers.clear(); counts.clear(); } };
deleted file mode 100644
deleted file mode 100644
@@ -1,36 +0,0 @@
@@ -1,36 +0,0 @@
-'use strict';
-// GET /internal/followers?stream_id=<id>&limit=<n>[&after=<cursor>] — the followers of the channel that
-// streams <id>, for OpenVibe.Network's live.stream.started consumer (go-live notifications are
-// Network's job). Service token only (aud openvibe.live, cap live.follower.read), loopback only.
-// A follower is named by their Network subject when Live knows it, else their Network user id.
-const express = require('express');
-const db = require('../db/database');
-const { guard } = require('../net/service-guard');
-const { http } = require('openvibe-contracts');
-
-const router = express.Router();
-router.get('/', guard('live.follower.read'), (req, res) => {
- const streamId = Number(req.query.stream_id);
- if (!Number.isSafeInteger(streamId) || streamId <= 0) return http.sendProblem(res, 400, 'live.bad_request', { detail: 'stream_id is required' });
- const limit = Math.min(Math.max(parseInt(req.query.limit, 10) || 1000, 1), 5000);
- const after = Math.max(parseInt(req.query.after, 10) || 0, 0);
- const d = db.getDb();
- const stream = d.prepare('SELECT user_id FROM streams WHERE id = ?').get(streamId);
- if (!stream) return http.sendProblem(res, 404, 'live.unknown_stream', { detail: `no stream ${streamId}` });
- const channel = d.prepare("SELECT subject_id FROM linked_accounts WHERE service = 'network' AND user_id = ?").get(stream.user_id);
- const rows = d.prepare(`SELECT f.id, la.subject_id, la.service_user_id FROM follows f
- LEFT JOIN linked_accounts la ON la.service = 'network' AND la.user_id = f.follower_id
- WHERE f.streamer_id = ? AND f.id > ? ORDER BY f.id LIMIT ?`).all(stream.user_id, after, limit);
- const followers = rows.map(r => ({
- subject: r.subject_id || null,
- network_user_id: /^\d+$/.test(String(r.service_user_id || '')) ? Number(r.service_user_id) : null,
- }));
- res.set('Cache-Control', 'no-store').json({
- stream_id: streamId,
- channel: { subject: (channel && channel.subject_id) || null },
- followers,
- next: rows.length === limit ? rows[rows.length - 1].id : null,
- });
-});
-
-module.exports = router;
@@ -344,8 +344,10 @@ router.get('/channel/:username', optionalAuth, async (req, res) => {
@@ -344,8 +344,10 @@ router.get('/channel/:username', optionalAuth, async (req, res) => {
aiShort(clipsOfStreams, 'clip_ai_state', 'clip_id');
aiShort(clipsOfStreams, 'clip_ai_state', 'clip_id');
aiShort(aiClips, 'clip_ai_state', 'clip_id');
aiShort(aiClips, 'clip_ai_state', 'clip_id');
}
}
- const followerCount = db.getFollowerCount(channel.user_id);
- const isFollowing = req.user ? db.isFollowing(req.user.id, channel.user_id) : false;
+ // Follows are Network's (ADR-030): its count, and the viewer's follow as Network last answered it.
+ const networkFollows = require('../social/network-follows');
+ const followerCount = await networkFollows.followerCount(channel.user_id);
+ const isFollowing = req.user ? networkFollows.isFollowing(req.user.id, channel.user_id) : false;
// Managed streams for this channel
// Managed streams for this channel
const managedStreams = db.getManagedStreamsByUserId(channel.user_id) || [];
const managedStreams = db.getManagedStreamsByUserId(channel.user_id) || [];
@@ -1627,7 +1629,7 @@ router.get('/setup-progress', requireAuth, async (req, res) => {
@@ -1627,7 +1629,7 @@ router.get('/setup-progress', requireAuth, async (req, res) => {
const sounds = safe(() => db.get('SELECT COUNT(*) AS n FROM channel_sounds WHERE channel_owner_id = ?', [uid]).n, 0);
const sounds = safe(() => db.get('SELECT COUNT(*) AS n FROM channel_sounds WHERE channel_owner_id = ?', [uid]).n, 0);
const goals = safe(() => db.get('SELECT COUNT(*) AS n FROM donation_goals WHERE user_id = ? AND is_active = 1', [uid]).n, 0);
const goals = safe(() => db.get('SELECT COUNT(*) AS n FROM donation_goals WHERE user_id = ? AND is_active = 1', [uid]).n, 0);
const powerchat = safe(() => !!db.get('SELECT 1 FROM powerchat_connections WHERE user_id = ? LIMIT 1', [uid]), false);
const powerchat = safe(() => !!db.get('SELECT 1 FROM powerchat_connections WHERE user_id = ? LIMIT 1', [uid]), false);
- const followers = safe(() => db.getFollowerCount(uid), 0);
+ const followers = await require('../social/network-follows').followerCount(uid).catch(() => 0);
let panels = 0; try { const p = channel.panels ? JSON.parse(channel.panels) : []; panels = Array.isArray(p) ? p.length : 0; } catch { panels = 0; }
let panels = 0; try { const p = channel.panels ? JSON.parse(channel.panels) : []; panels = Array.isArray(p) ? p.length : 0; } catch { panels = 0; }
const offline = !!(channel.offline_screen_type && channel.offline_screen_type !== 'none');
const offline = !!(channel.offline_screen_type && channel.offline_screen_type !== 'none');
const methodSet = slots.some(sl => !!sl.streaming_method);
const methodSet = slots.some(sl => !!sl.streaming_method);
@@ -1958,7 +1960,7 @@ router.post('/managed/:id/regenerate-key', requireAuth, async (req, res) => {
@@ -1958,7 +1960,7 @@ router.post('/managed/:id/regenerate-key', requireAuth, async (req, res) => {
});
});
// ── Get Stream Details ───────────────────────────────────────
// ── Get Stream Details ───────────────────────────────────────
-router.get('/:id', optionalAuth, (req, res) => {
+router.get('/:id', optionalAuth, 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' });
@@ -1995,8 +1997,9 @@ router.get('/:id', optionalAuth, (req, res) => {
@@ -1995,8 +1997,9 @@ router.get('/:id', optionalAuth, (req, res) => {
? db.getPipOverlayForManagedStream(stream.managed_stream_id)
? db.getPipOverlayForManagedStream(stream.managed_stream_id)
: null;
: null;
- if (req.user) stream.isFollowing = db.isFollowing(req.user.id, stream.user_id);
- stream.follower_count = db.getFollowerCount(stream.user_id);
+ const networkFollows = require('../social/network-follows');
+ if (req.user) stream.isFollowing = networkFollows.isFollowing(req.user.id, stream.user_id);
+ stream.follower_count = await networkFollows.followerCount(stream.user_id);
res.json({ stream });
res.json({ stream });
} catch (err) {
} catch (err) {
@@ -2389,48 +2392,39 @@ router.get('/:id/rtmp-status', requireAuth, (req, res) => {
@@ -2389,48 +2392,39 @@ router.get('/:id/rtmp-status', requireAuth, (req, res) => {
});
});
// ── Follow/Unfollow Streamer ─────────────────────────────────
// ── Follow/Unfollow Streamer ─────────────────────────────────
+// Network owns follows (ADR-030, plan T2): a click follows or unfollows there, and the answer (following, the
+// follower count) is Network's. Network sends the FOLLOW notification; Live adds the OpenCoins and the PowerChat
+// alert for a follow Network says this click started (never for a repeat of a follow Network already had).
+async function toggleFollow(req, res, streamerId) {
+ const follows = require('../social/network-follows');
+ const w = await follows.set(req.user.id, streamerId, !follows.isFollowing(req.user.id, streamerId));
+ if (!w.ok) {
+ if (w.detail) console.warn('[Streaming] follow:', w.detail);
+ return res.status(w.status).json({ error: w.error });
+ }
+ if (w.started) {
+ // Award OpenCoins for following
+ try {
+ const openvibeCoins = require('../monetization/opencoins');
+ openvibeCoins.awardFollow(req.user.id, streamerId);
+ } catch { /* non-critical */ }
+ // Fire a PowerChat follow alert for the streamer (follows:write).
+ try {
+ const follower = db.getUserById(req.user.id);
+ require('../integrations/powerchat-platform').forwardFollow(streamerId, {
+ followerName: follower?.display_name || follower?.username || 'Someone',
+ externalId: 'u' + req.user.id,
+ });
+ } catch { /* non-critical */ }
+ }
+ res.json({ following: w.following, count: w.count });
+}
+
router.post('/:id/follow', requireAuth, async (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)) {
- db.unfollowUser(req.user.id, stream.user_id);
- res.json({ following: false, count: db.getFollowerCount(stream.user_id) });
- } else {
- db.followUser(req.user.id, stream.user_id);
- // Award OpenCoins for following
- try {
- const openvibeCoins = require('../monetization/opencoins');
- openvibeCoins.awardFollow(req.user.id, stream.user_id);
- } catch { /* non-critical */ }
- // Fire a PowerChat follow alert for the streamer (follows:write).
- try {
- const follower = db.getUserById(req.user.id);
- require('../integrations/powerchat-platform').forwardFollow(stream.user_id, {
- followerName: follower?.display_name || follower?.username || 'Someone',
- externalId: 'u' + req.user.id,
- });
- } catch { /* non-critical */ }
- // Notify the followed user. A follow Network took notifies there (its follow graph sends FOLLOW);
- // only a Live-only follow (FOLLOWS_AUTHORITY unset, or a side with no subject) is pushed from here.
- if (!w.network) try {
- const { pushNotification, actorInfo } = require('../utils/notify');
- const follower = db.getUserById(req.user.id);
- pushNotification({
- user_id: stream.user_id,
- type: 'FOLLOW',
- title: 'New Follower',
- message: `${follower?.display_name || follower?.username || 'Someone'} followed you`,
- url: follower?.username ? `${config.baseUrl}/@${encodeURIComponent(follower.username)}` : config.baseUrl,
- ...actorInfo(follower),
- });
- } catch { /* non-critical */ }
- res.json({ following: true, count: db.getFollowerCount(stream.user_id) });
- }
+ await toggleFollow(req, res, stream.user_id);
} catch (err) {
} catch (err) {
console.error('[Streaming]', err.message);
console.error('[Streaming]', err.message);
res.status(500).json({ error: 'Failed to follow/unfollow' });
res.status(500).json({ error: 'Failed to follow/unfollow' });
@@ -2442,43 +2436,7 @@ router.post('/channel/:username/follow', requireAuth, async (req, res) => {
@@ -2442,43 +2436,7 @@ 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)) {
- db.unfollowUser(req.user.id, user.id);
- res.json({ following: false, count: db.getFollowerCount(user.id) });
- } else {
- db.followUser(req.user.id, user.id);
- // Award OpenCoins for following
- try {
- const openvibeCoins = require('../monetization/opencoins');
- openvibeCoins.awardFollow(req.user.id, user.id);
- } catch { /* non-critical */ }
- // Fire a PowerChat follow alert for the streamer (follows:write).
- try {
- const f = db.getUserById(req.user.id);
- require('../integrations/powerchat-platform').forwardFollow(user.id, {
- followerName: f?.display_name || f?.username || 'Someone',
- externalId: 'u' + req.user.id,
- });
- } catch { /* non-critical */ }
- // Notify the followed user, unless Network took the follow and notifies (see the route above).
- if (!w.network) try {
- const { pushNotification, actorInfo } = require('../utils/notify');
- const follower = db.getUserById(req.user.id);
- pushNotification({
- user_id: user.id,
- type: 'FOLLOW',
- title: 'New Follower',
- message: `${follower?.display_name || follower?.username || 'Someone'} followed you`,
- url: follower?.username ? `${config.baseUrl}/@${encodeURIComponent(follower.username)}` : config.baseUrl,
- ...actorInfo(follower),
- });
- } catch { /* non-critical */ }
- res.json({ following: true, count: db.getFollowerCount(user.id) });
- }
+ await toggleFollow(req, res, user.id);
} catch (err) {
} catch (err) {
console.error('[Streaming]', err.message);
console.error('[Streaming]', err.message);
res.status(500).json({ error: 'Failed to follow/unfollow' });
res.status(500).json({ error: 'Failed to follow/unfollow' });
deleted file mode 100644
deleted file mode 100644
@@ -1,78 +0,0 @@
@@ -1,78 +0,0 @@
-'use strict';
-
-// GET /internal/followers — OpenVibe.Network asks who follows the channel behind a stream when it
-// consumes live.stream.started. Service token with live.follower.read, loopback only, paged.
-
-const assert = require('assert');
-const fs = require('fs');
-const os = require('os');
-const path = require('path');
-const http = require('http');
-const crypto = require('crypto');
-const express = require('express');
-const { serviceAuth } = require('openvibe-contracts');
-
-const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-followers-'));
-const keys = crypto.generateKeyPairSync('rsa', { modulusLength: 2048, publicKeyEncoding: { type: 'spki', format: 'pem' }, privateKeyEncoding: { type: 'pkcs8', format: 'pem' } });
-fs.writeFileSync(path.join(tmp, 'network.pem'), keys.publicKey);
-process.env.DB_PATH = path.join(tmp, 'live.db');
-process.env.OV_NETWORK_PUBLIC_KEY = path.join(tmp, 'network.pem');
-process.env.OV_NETWORK_URL = 'https://openvibe.network';
-console.log = () => {};
-console.warn = () => {};
-
-const ISS = 'https://openvibe.network';
-const now = () => Math.floor(Date.now() / 1000);
-let jti = 0;
-const token = (cap) => serviceAuth.signServiceToken({ iss: ISS, sub: 'svc:network', actor_type: 'service', aud: ['openvibe.live'], cap, iat: now(), exp: now() + 300, jti: `tok_net_${++jti}` }, keys.privateKey);
-
-(async () => {
- const db = require('../server/db/database');
- db.initDb();
- const d = db.getDb();
- for (const [id, name] of [[601, 'streamer'], [602, 'fan_a'], [603, 'fan_b'], [604, 'fan_c']]) {
- d.prepare("INSERT INTO users (id, username, display_name, password_hash) VALUES (?, ?, ?, 'x')").run(id, name, name);
- }
- d.prepare("INSERT INTO linked_accounts (user_id, service, service_user_id, subject_id) VALUES (601, 'network', '70', 'usr_01J0000000000000000000000S')").run();
- d.prepare("INSERT INTO linked_accounts (user_id, service, service_user_id, subject_id) VALUES (602, 'network', '71', 'usr_01J0000000000000000000000A')").run();
- d.prepare("INSERT INTO linked_accounts (user_id, service, service_user_id) VALUES (603, 'network', '72')").run();
- for (const f of [602, 603, 604]) d.prepare('INSERT INTO follows (follower_id, streamer_id) VALUES (?, 601)').run(f);
- const streamId = d.prepare("INSERT INTO streams (user_id, title, is_live) VALUES (601, 'hello', 1)").run().lastInsertRowid;
-
- const app = express();
- app.use('/internal/followers', require('../server/streaming/followers-internal'));
- const server = await new Promise((r) => { const s = http.createServer(app); s.listen(0, '127.0.0.1', () => r(s)); });
- const base = `http://127.0.0.1:${server.address().port}`;
- const get = (q, { cap = ['live.follower.read'], headers = {} } = {}) => fetch(`${base}/internal/followers?${q}`, {
- headers: { Authorization: `Bearer ${token(cap)}`, ...headers },
- }).then(async (r) => ({ status: r.status, json: await r.json().catch(() => null) }));
-
- assert.strictEqual((await get(`stream_id=${streamId}`, { cap: ['live.chat_effects.write'] })).status, 403, 'needs live.follower.read');
- assert.strictEqual((await get(`stream_id=${streamId}`, { headers: { 'X-Forwarded-For': '1.2.3.4' } })).status, 403, 'loopback only');
- assert.strictEqual((await fetch(`${base}/internal/followers?stream_id=${streamId}`)).status, 401, 'no token');
- assert.strictEqual((await get('stream_id=abc')).status, 400);
- assert.strictEqual((await get('stream_id=999999')).status, 404);
-
- const all = await get(`stream_id=${streamId}`);
- assert.strictEqual(all.status, 200);
- assert.strictEqual(all.json.channel.subject, 'usr_01J0000000000000000000000S');
- assert.deepStrictEqual(all.json.followers, [
- { subject: 'usr_01J0000000000000000000000A', network_user_id: 71 },
- { subject: null, network_user_id: 72 },
- { subject: null, network_user_id: null },
- ]);
- assert.strictEqual(all.json.next, null);
-
- // Paging: two per page, then the rest.
- const p1 = await get(`stream_id=${streamId}&limit=2`);
- assert.strictEqual(p1.json.followers.length, 2);
- assert.ok(p1.json.next);
- const p2 = await get(`stream_id=${streamId}&limit=2&after=${p1.json.next}`);
- assert.strictEqual(p2.json.followers.length, 1);
- assert.strictEqual(p2.json.next, null);
-
- server.close();
- fs.rmSync(tmp, { recursive: true, force: true });
- process.stdout.write('followers-internal: all passed\n');
- process.exit(0);
-})().catch((e) => { process.stderr.write(`${e.stack}\n`); process.exit(1); });
@@ -297,7 +297,7 @@ function checkContract(out, label) {
@@ -297,7 +297,7 @@ function checkContract(out, label) {
return { status: res.status, json: await res.json().catch(() => null), cache: res.headers.get('cache-control') };
return { status: res.status, json: await res.json().catch(() => null), cache: res.headers.get('cache-control') };
};
};
assert.strictEqual((await fetch(`${base}?slug=alice`)).status, 401, 'no token');
assert.strictEqual((await fetch(`${base}?slug=alice`)).status, 401, 'no token');
- assert.strictEqual((await call('GET', 'slug=alice', null, { cap: ['live.follower.read'] })).status, 403, 'needs live.lineage.resolve');
+ assert.strictEqual((await call('GET', 'slug=alice', null, { cap: ['live.chat_context.read'] })).status, 403, 'needs live.lineage.resolve');
assert.strictEqual((await call('GET', 'slug=alice', null, { headers: { 'X-Forwarded-For': '1.2.3.4' } })).status, 403, 'loopback only');
assert.strictEqual((await call('GET', 'slug=alice', null, { headers: { 'X-Forwarded-For': '1.2.3.4' } })).status, 403, 'loopback only');
let res = await call('GET', 'slug=alice&vod_id=42&live_user_id=17');
let res = await call('GET', 'slug=alice&vod_id=42&live_user_id=17');
assert.strictEqual(res.status, 200);
assert.strictEqual(res.status, 200);
@@ -46,7 +46,7 @@ const SEED = `
@@ -46,7 +46,7 @@ const SEED = `
db.ensureChannel(star); db.ensureChannel(fan);
db.ensureChannel(star); db.ensureChannel(fan);
d.prepare("INSERT INTO streams (user_id, title, category, protocol, is_live, viewer_count, started_at, last_heartbeat) VALUES (?, 'N-1 live stream', 'tech', 'webrtc', 1, 2, datetime('now', '-10 minutes'), datetime('now'))").run(star);
d.prepare("INSERT INTO streams (user_id, title, category, protocol, is_live, viewer_count, started_at, last_heartbeat) VALUES (?, 'N-1 live stream', 'tech', 'webrtc', 1, 2, datetime('now', '-10 minutes'), datetime('now'))").run(star);
d.prepare("INSERT INTO streams (user_id, title, category, protocol, is_live, started_at, ended_at) VALUES (?, 'N-1 past stream', 'irl', 'webrtc', 0, datetime('now', '-2 days'), datetime('now', '-2 days', '+1 hour'))").run(star);
d.prepare("INSERT INTO streams (user_id, title, category, protocol, is_live, started_at, ended_at) VALUES (?, 'N-1 past stream', 'irl', 'webrtc', 0, datetime('now', '-2 days'), datetime('now', '-2 days', '+1 hour'))").run(star);
- db.followUser(fan, star);
+ d.prepare('INSERT INTO follows (follower_id, streamer_id) VALUES (?, ?)').run(fan, star);
d.prepare("INSERT INTO chat_messages (stream_id, user_id, username, message) VALUES (1, ?, 'n1fan', 'hello from N-1')").run(fan);
d.prepare("INSERT INTO chat_messages (stream_id, user_id, username, message) VALUES (1, ?, 'n1fan', 'hello from N-1')").run(fan);
db.close();
db.close();
`;
`;
@@ -1,23 +1,39 @@
@@ -1,23 +1,39 @@
'use strict';
'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.
+// Follows on OpenVibe.Network (ADR-030, plan T2; server/social/network-follows.js): a follow button follows or
+// unfollows on Network (network.follows.write) and shows Network's answer; only a follow Network says the click
+// started earns the coins and the PowerChat alert; a side without a subject cannot follow; Network down or a
+// malformed answer changes nothing; channel and stream responses show Network's count and the viewer's follow
+// while the projection lags; network.follow.* events alone write Live's table, in revision order, so a late older
+// event cannot undo a click. Nothing else on Live writes follows, and GET /internal/followers is gone.
const assert = require('assert');
const assert = require('assert');
const fs = require('fs');
const fs = require('fs');
const os = require('os');
const os = require('os');
const path = require('path');
const path = require('path');
+const http = require('http');
const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-netfollows-'));
const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-netfollows-'));
process.env.DB_PATH = path.join(tmp, 'live.db');
process.env.DB_PATH = path.join(tmp, 'live.db');
+process.env.NODE_ENV = 'test';
process.env.OV_NETWORK_INTERNAL_URL = 'http://network.test';
process.env.OV_NETWORK_INTERNAL_URL = 'http://network.test';
const log = console.log; console.log = () => {}; console.warn = () => {};
const log = console.log; console.log = () => {}; console.warn = () => {};
const db = require('../server/db/database');
const db = require('../server/db/database');
db.initDb();
db.initDb();
-console.log = log;
const principal = require('../server/net/network-principal');
const principal = require('../server/net/network-principal');
principal.serviceHeaders = async () => ({ Authorization: 'Bearer svc-token' });
principal.serviceHeaders = async () => ({ Authorization: 'Bearer svc-token' });
+let invalidated = 0;
+principal.invalidate = () => { invalidated++; };
const follows = require('../server/social/network-follows');
const follows = require('../server/social/network-follows');
const events = require('../server/auth/network-events');
const events = require('../server/auth/network-events');
+const auth = require('../server/auth/auth');
+auth.requireAuth = (req, res, next) => { const u = db.getUserById(Number(req.headers['x-test-user'] || 0)); if (!u) return res.status(401).json({ error: 'Authentication required' }); req.user = u; next(); };
+auth.optionalAuth = (req, res, next) => { const u = db.getUserById(Number(req.headers['x-test-user'] || 0)); if (u) req.user = u; next(); };
+const coins = require('../server/monetization/opencoins');
+const powerchat = require('../server/integrations/powerchat-platform');
+const side = [];
+coins.awardFollow = (a, b) => side.push(['coins', a, b]);
+powerchat.forwardFollow = (s) => side.push(['powerchat', s]);
+const notify = require('../server/utils/notify');
+notify.pushNotification = (n) => side.push(['notify', n.type]);
const sub = (tag) => `usr_${`01J${tag}`.padEnd(26, '0')}`;
const sub = (tag) => `usr_${`01J${tag}`.padEnd(26, '0')}`;
const ANN = sub('AA'), BOB = sub('BB'), CAT = sub('CC'), GHOST = sub('DD');
const ANN = sub('AA'), BOB = sub('BB'), CAT = sub('CC'), GHOST = sub('DD');
@@ -25,58 +41,216 @@ const d = db.getDb();
@@ -25,58 +41,216 @@ 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);
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', ?, ?)");
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);
link.run(1, '101', ANN); link.run(2, '102', BOB); link.run(3, '103', CAT);
+const streamId = Number(db.createStream({ user_id: 2, title: 't', protocol: 'webrtc' }).lastInsertRowid);
-(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 }; };
+// Network: /internal/follows answers network.follow-status-result@1 from its own graph (201 for a pair's first
+// follow; a follow that starts again starts its `since` again), and /api/v1/follows answers the public count.
+const graph = new Map(); // `${follower}>${target}` → { active, revision, since }
+const calls = [];
+let fail = null;
+const countOf = (target) => [...graph].filter(([k, v]) => v.active && k.endsWith(`>${target}`)).length + 40;
+globalThis.fetch = async (url, o = {}) => {
+ const method = o.method || 'GET';
+ calls.push({ url, method, body: o.body ? JSON.parse(o.body) : null, auth: o.headers && o.headers.Authorization });
+ if (fail instanceof Error) throw fail;
+ if (fail) return { status: fail.status, json: async () => fail.body || {} };
+ const u = new URL(url);
+ const target = u.pathname.split('/').pop();
+ if (method === 'GET') return { status: 200, json: async () => ({ target_type: 'channel', target_id: target, followers: countOf(target) }) };
+ const follower = method === 'PUT' ? JSON.parse(o.body).follower : u.searchParams.get('follower');
+ const key = `${follower}>${target}`;
+ const prev = graph.get(key);
+ let status = 200;
+ if (method === 'PUT' && !(prev && prev.active)) {
+ const revision = prev ? prev.revision + 1 : 1;
+ graph.set(key, { active: true, revision, since: new Date().toISOString() });
+ if (revision === 1) status = 201;
+ } else if (method === 'DELETE' && prev && prev.active) graph.set(key, { ...prev, active: false, revision: prev.revision + 1 });
+ const now = graph.get(key);
+ const body = { target_type: 'channel', target_id: target, followers: countOf(target), following: !!(now && now.active) };
+ if (now && now.active) body.since = now.since;
+ return { status, json: async () => body };
+};
- // 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);
+const express = require('express');
+const app = express();
+app.use(express.json());
+app.use('/api/streams', require('../server/streaming/routes'));
+const server = http.createServer(app);
+const call = (method, p, user) => new Promise((resolve, reject) => {
+ const req = http.request(`http://127.0.0.1:${server.address().port}${p}`, { method, headers: user ? { 'x-test-user': String(user) } : {} }, (res) => {
+ let b = ''; res.on('data', (c) => { b += c; }); res.on('end', () => resolve({ status: res.statusCode, body: b ? JSON.parse(b) : null }));
+ });
+ req.on('error', reject); req.end();
+});
+const post = (p, user) => call('POST', p, user);
+const get = (p, user) => call('GET', p, user);
+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 } });
- 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);
+server.listen(0, '127.0.0.1', async () => {
+ try {
+ // set(): Network first; its answer is what the button shows, and the projection waits for the event.
+ let r = await follows.set(1, 2, true);
+ assert.deepStrictEqual(r, { ok: true, following: true, count: 41, started: true });
+ assert.deepStrictEqual(calls.pop(), { url: `http://network.test/internal/follows/channel/${BOB}`, method: 'PUT', body: { follower: ANN }, auth: 'Bearer svc-token' });
+ assert.ok(!db.isFollowing(1, 2), 'a write answer never touches the projection');
+ assert.ok(follows.isFollowing(1, 2), 'the button shows Network\'s answer before the event');
+ assert.strictEqual((await follows.set(1, 2, true)).started, false, 'the same follow again starts nothing');
+ r = await follows.set(1, 2, false);
+ assert.deepStrictEqual(calls.pop(), { url: `http://network.test/internal/follows/channel/${BOB}?follower=${ANN}`, method: 'DELETE', body: null, auth: 'Bearer svc-token' });
+ assert.deepStrictEqual(r, { ok: true, following: false, count: 40, started: false });
+ assert.ok(!follows.isFollowing(1, 2));
+ assert.strictEqual((await follows.set(1, 2, true)).started, true, 'a follow that starts again (200, a new since) is started');
+ await follows.set(1, 2, false);
- // 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');
+ // Network cannot answer, or answers a malformed document: 503, nothing changes on Live.
+ fail = { status: 503 };
+ r = await follows.set(1, 2, true);
+ assert.strictEqual(r.ok, false); assert.strictEqual(r.status, 503);
+ fail = new Error('ECONNREFUSED');
+ assert.strictEqual((await follows.set(1, 2, true)).status, 503);
+ fail = { status: 401 };
+ assert.strictEqual((await follows.set(1, 2, true)).status, 503);
+ assert.strictEqual(invalidated, 1, 'a refused service token is dropped');
+ fail = { status: 200, body: { nope: true } };
+ assert.strictEqual((await follows.set(1, 2, true)).status, 503, 'an answer without `following` is no answer');
+ for (const followers of [undefined, '41', -1, 4.5]) {
+ fail = { status: 200, body: { target_type: 'channel', target_id: BOB, following: true, followers } };
+ assert.strictEqual((await follows.set(1, 2, true)).status, 503, `an answer with followers ${followers} is no answer`);
+ }
+ fail = { status: 200, body: { target_type: 'channel', target_id: CAT, following: true, followers: 41 } };
+ assert.strictEqual((await follows.set(1, 2, true)).status, 503, 'an answer about another channel is no answer');
+ assert.ok(!follows.isFollowing(1, 2) && !db.isFollowing(1, 2), 'nothing remembered while Network did not take it');
+ // Network refuses the request itself: its status and reason.
+ fail = { status: 400, body: { code: 'follows.self', detail: 'you cannot follow yourself' } };
+ assert.deepStrictEqual(await follows.set(2, 2, true), { ok: false, status: 400, error: 'you cannot follow yourself' });
+ fail = null;
+ // A side without a subject cannot follow; Network is not asked.
+ const n = calls.length;
+ assert.strictEqual((await follows.set(4, 2, true)).status, 409, 'follower without a subject');
+ assert.strictEqual((await follows.set(1, 4, true)).status, 409, 'channel without a subject');
+ assert.strictEqual(calls.length, n);
+ assert.ok(!follows.isFollowing(4, 2) && !follows.isFollowing(1, 4));
- // 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\)/);
- // A follow Network took is notified by Network (FOLLOW from its follow graph); Live pushes only a Live-only one.
- const whole = routes.slice(routes.indexOf(route), routes.indexOf('\nrouter.', routes.indexOf(route) + 1));
- assert.match(whole, /if \(!w\.network\) try \{\s*const \{ pushNotification/, `${route} leaves a Network follow's notification to Network`);
- assert.strictEqual((whole.match(/type: 'FOLLOW'/g) || []).length, 1);
+ // The buttons: toggle on the state the viewer sees, answer Network's state and count.
+ assert.strictEqual((await post(`/api/streams/${streamId}/follow`)).status, 401, 'signed in only');
+ side.length = 0;
+ r = await post(`/api/streams/${streamId}/follow`, 3);
+ assert.deepStrictEqual(r, { status: 200, body: { following: true, count: 41 } });
+ assert.deepStrictEqual(side, [['coins', 3, 2], ['powerchat', 2]], 'coins and the PowerChat alert; Network sends FOLLOW');
+ r = await post('/api/streams/channel/bob/follow', 3);
+ assert.deepStrictEqual(r, { status: 200, body: { following: false, count: 40 } }, 'the second click unfollows before any event');
+ side.length = 0;
+ // Network already has the follow (made elsewhere, its event delayed): the click is an idempotent PUT, and
+ // nothing started, so no coins and no false PowerChat alert.
+ graph.set(`${CAT}>${BOB}`, { active: true, revision: 3, since: '2026-01-01T00:00:00.000Z' });
+ follows._reset();
+ r = await post('/api/streams/channel/bob/follow', 3);
+ assert.deepStrictEqual(r.body, { following: true, count: 41 });
+ assert.deepStrictEqual(side, [], 'a follow Network already had starts nothing');
+ fail = { status: 502 };
+ r = await post('/api/streams/channel/bob/follow', 3);
+ assert.strictEqual(r.status, 503);
+ assert.ok(follows.isFollowing(3, 2), 'Network down: the follow stays as it was');
+ fail = null;
+ assert.strictEqual((await post('/api/streams/channel/bob/follow', 4)).status, 409);
+ assert.strictEqual((await post('/api/streams/channel/nobody/follow', 3)).status, 404);
+ assert.strictEqual((await post('/api/streams/999999/follow', 3)).status, 404);
+ assert.ok(!side.some((s) => s[0] === 'notify'), 'Live sends no follow notification');
+
+ // Channel and stream responses while the projection lags: Network's count, the viewer's follow as Network
+ // answered it (the projection still has no row for cat → bob).
+ assert.ok(!db.isFollowing(3, 2));
+ follows._reset();
+ await follows.set(3, 2, true);
+ graph.set(`${ANN}>${BOB}`, { active: true, revision: 7, since: '2026-01-01T00:00:00.000Z' }); // a follow made elsewhere
+ follows._reset();
+ await follows.set(3, 2, true);
+ calls.length = 0;
+ r = await get('/api/streams/channel/bob', 3);
+ assert.strictEqual(r.status, 200, JSON.stringify(r.body));
+ assert.strictEqual(r.body.channel.follower_count, 42, 'the count is Network\'s');
+ assert.strictEqual(r.body.channel.is_following, true);
+ r = await get(`/api/streams/${streamId}`, 3);
+ assert.strictEqual(r.status, 200);
+ assert.strictEqual(r.body.stream.follower_count, 42);
+ assert.strictEqual(r.body.stream.isFollowing, true);
+ assert.strictEqual(calls.filter((c) => c.url.includes('/api/v1/follows/')).length, 0, 'the write answer\'s count is reused');
+ follows._reset();
+ calls.length = 0;
+ r = await get('/api/streams/channel/bob', 1);
+ assert.strictEqual(r.body.channel.follower_count, 42);
+ assert.strictEqual(r.body.channel.is_following, false, 'no answer for ann: the projection');
+ assert.deepStrictEqual(calls.filter((c) => c.url.includes('/api/v1/follows/')), [{ url: `http://network.test/api/v1/follows/channel/${BOB}`, method: 'GET', body: null, auth: undefined }], 'the public count, no credentials');
+ r = await get('/api/streams/channel/bob');
+ assert.strictEqual(r.body.channel.is_following, false, 'signed out');
+ // Network cannot answer the count, or answers a malformed one: the projection's count for a while.
+ follows._reset();
+ fail = new Error('ECONNREFUSED');
+ assert.strictEqual(await follows.followerCount(2), 0);
+ fail = null;
+ assert.strictEqual(await follows.followerCount(2), 0, 'the projection answers while Network was just down');
+ follows._reset();
+ fail = { status: 200, body: { target_type: 'channel', target_id: BOB, followers: 'many' } };
+ assert.strictEqual(await follows.followerCount(2), 0);
+ fail = null;
+ follows._reset();
+ assert.strictEqual(await follows.followerCount(2), 42);
+ assert.strictEqual(await follows.followerCount(4), 0, 'a channel without a subject: the projection');
+
+ // The projection, through the Network events endpoint's apply().
+ d.prepare('DELETE FROM follows').run();
+ follows._reset();
+ 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');
+
+ // A click, then its events delivered late and out of order: an older follow event arriving after the
+ // unfollow click cannot show the follow again, and the unfollow's own event retires the remembered answer.
+ graph.delete(`${ANN}>${CAT}`);
+ assert.strictEqual((await follows.set(1, 3, true)).started, true); // Network revision 1
+ assert.strictEqual((await follows.set(1, 3, false)).following, false); // Network revision 2
+ assert.strictEqual(events.apply(ev('network.follow.created', ANN, CAT, 1)), 'followed', 'the late follow event');
+ assert.ok(db.isFollowing(1, 3), 'the projection is at revision 1');
+ assert.strictEqual(follows.isFollowing(1, 3), false, 'the button still shows the unfollow');
+ r = await get('/api/streams/channel/cat', 1);
+ assert.strictEqual(r.body.channel.is_following, false);
+ assert.strictEqual(events.apply(ev('network.follow.deleted', ANN, CAT, 2)), 'unfollowed');
+ assert.ok(!db.isFollowing(1, 3) && !follows.isFollowing(1, 3));
+ assert.strictEqual(events.apply(ev('network.follow.created', ANN, CAT, 1)), 'stale');
+ assert.strictEqual(events.apply(ev('network.follow.created', ANN, CAT, 3)), 'followed', 'a later follow made elsewhere');
+ assert.strictEqual(follows.isFollowing(1, 3), true, 'the agreeing event retired the answer: the projection shows it');
+ // 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'/);
+
+ // Only the projection writes follows (and the account merge repoints it); no Live-only path is left.
+ const root = path.join(__dirname, '..', 'server');
+ const writers = [];
+ const walk = (dir) => { for (const e of fs.readdirSync(dir, { withFileTypes: true })) {
+ const p = path.join(dir, e.name);
+ if (e.isDirectory()) walk(p);
+ else if (e.name.endsWith('.js') && /\b(INSERT(\s+OR\s+\w+)?\s+INTO|DELETE\s+FROM|UPDATE)\s+follows\b/i.test(fs.readFileSync(p, 'utf8'))) writers.push(path.relative(root, p));
+ } };
+ walk(root);
+ assert.deepStrictEqual(writers.sort(), ['auth/subject-merge.js', 'social/network-follows.js'], 'follows writers');
+ assert.strictEqual(typeof db.followUser, 'undefined');
+ assert.strictEqual(typeof db.unfollowUser, 'undefined');
+ assert.ok(!/FOLLOWS_AUTHORITY/.test(fs.readFileSync(path.join(root, 'social', 'network-follows.js'), 'utf8')), 'no switch back to Live-only follows');
+ // GET /internal/followers is retired: Network reads its own graph (Network PR #6).
+ assert.ok(!fs.existsSync(path.join(root, 'streaming', 'followers-internal.js')));
+ assert.ok(!fs.readFileSync(path.join(root, 'index.js'), 'utf8').includes('/internal/followers'));
+ console.log = log;
+ console.log('network follows: all checks passed');
+ } finally {
+ server.close();
+ fs.rmSync(tmp, { recursive: true, force: true });
}
}
- // 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);
process.exit(0);
-})().catch((e) => { console.error(e); process.exit(1); });
+});
+process.on('unhandledRejection', (e) => { console.log = log; console.error(e); process.exit(1); });