openvibers/openvibe.network Commits
The follow graph on Network (roadmap WS-E task 4, ADR-030 steps 1-3; Contracts v0.65.0, pinned from 0.63.0). user_follows is keyed by subjects (a Live channel is its owner's subject), with active, notify flags and a per-pair revision. Public count at GET /api/v1/follows/:type/:target (plus the caller's own follow); a person's follows at /api/v1/me/follows with idempotent PUT and DELETE; followers for the target's owner or a service with network.follows.read (granted to Live for its projection). Every change writes network.follow.created or .deleted to the outbox in the same transaction. scripts/follows-backfill.js imports Live's follows (dry run by default; --apply only with a backup; no events; pairs without a subject held in follow_import_holds) and reconciles counts and pairs. test/follows.test.js
Changes
@@@ -408,6 +408,24 @@ Contracts 0.49.0; `server/identity/blocks.js`). `user_blocks` is keyed by subjec---++## Follow graph (ADR-030)++++Network owns follows, keyed by subjects (`server/identity/follows.js`; roadmap WS-E task 4, Contracts 0.65.0).++A target is a Live channel today (`channel`, named by its owner's subject), and other kinds join as products need them.++++- **Public:** `GET /api/v1/follows/:type/:target` gives the follower count, and for a signed-in caller whether++ they follow and how they are notified.++- **People:** `GET /api/v1/me/follows` lists what the caller follows. `PUT|DELETE /api/v1/me/follows/:type/:target`++ follow and unfollow (`:target` is a subject or a username). Both are idempotent, and the PUT body can carry++ `{ notify_email, notify_push }`.++- **Who follows a target:** `GET /api/v1/follows/:type/:target/followers` is open to the target's owner, and to++ services with `network.follows.read` (service token only). Lists are never public.++- **Events:** every change writes `network.follow.created` or `network.follow.deleted` (subject visibility,++ growing per-pair revision) to the outbox in the same transaction. Live keeps a projection from them.++- **Migration:** `scripts/follows-backfill.js --live-db <live.db>` does a dry run by default. `--apply --backup <file>`++ imports Live's follows without events: a pair whose side has no subject goes to `follow_import_holds`, and++ nothing is dropped. `--reconcile` compares per-channel counts and pair sets.++## Multi-Account SwitchingGoogle-style account management supporting up to 5 accounts:@@@ -611,6 +629,8 @@ Accessible to users with `role = 'admin'`. All endpoints under `/api/admin/`.| `user_modules`, `user_module_revisions`, `user_module_retirements` | User-module records, the last revision issued per (subject, namespace), retired namespace owners || `network_event_outbox` | Network's events on their way to OpenVibe.Events (developer projects, user modules, blocks) || `user_blocks` | Platform blocks by subject (blocker, blocked, active, per-pair revision) |++| `user_follows` | The follow graph by subject (follower, target type and id, active, notify flags, per-pair revision, source) |++| `follow_import_holds` | Imported follows held because a side had no subject (ADR-030 step 2) || `analytics_visitor_days`, `analytics_day_salts` | The current day's salted visitor hashes and salt, deleted after the day's final rollup |---@@@ -20,7 +20,7 @@"helmet": "^8.1.0","jsonwebtoken": "^9.0.0","multer": "^1.4.5-lts.1",-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.63.0",++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.65.0","openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.11.0","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.22.0","uuid": "^11.1.0",@@@ -1465,9 +1465,9 @@}},"node_modules/openvibe-contracts": {-- "version": "0.63.0",-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.63.0",-- "integrity": "sha512-vmc1b1vWg9M5fm57X50MlZEZHkP2IshHwBXt68+kHDV85nyg6FxaYyUWa4ibi1eeFqzMcECPlILSRTOgg2Xbgw==",++ "version": "0.65.0",++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.65.0",++ "integrity": "sha512-oBWCCxa0iThdh28hVCutTVyf91EDsK9+O54xHAW4XNi0CVGQaynLGckE08GWoMCd4T9Q5JEDDPfCA2TIssf1ew==","license": "AGPL-3.0-only","dependencies": {"ajv": "^8.17.1",@@@ -6,7 +6,7 @@"scripts": {"start": "node server/index.js","dev": "NODE_ENV=development node server/index.js",-- "test": "node test/registry-bootstrap.test.js && node test/nginx-generator-live.test.js && node test/admin-refresh.test.js && node test/history-sso.test.js && node test/sso-check.test.js && node test/fedcm.test.js && node test/domains.test.js && node test/frame.test.js && node test/github-integration.test.js && node test/revocation.test.js && node test/usernames.test.js && node test/avatar.test.js && node test/roadmap-baseline.test.js && node test/identity-subjects.test.js && node test/principals.test.js && node test/operator-alerts.test.js && node test/modules.test.js && node test/module-events.test.js && node test/registry-ecosystem.test.js && node test/registry-exposure.test.js && node test/registry-manifest-fixture.test.js && node test/staff-claims.test.js && node test/profile-events.test.js && node test/admin-role-events.test.js && node test/grants-admin.test.js && node test/events-ops.test.js && node test/deploy-drift.test.js && node test/registry-topics.test.js && node test/service-principal.test.js && node test/oauth-pkce.test.js && node test/developer-projects.test.js && node test/developer-defaults.test.js && node test/developer-export-tokens.test.js && node test/project-usage.test.js && node test/public-cors.test.js && node test/observability.test.js && node test/status.test.js && node test/events-consumer.test.js && node test/notification-events.test.js && node test/security-session.test.js && node test/security-redirects.test.js && node test/analytics-privacy.test.js && node test/data-ownership.test.js && node test/unknown-paths.test.js && node test/oidc-discovery.test.js && node test/refresh-tokens.test.js && node test/provider-secrets.test.js && node test/first-party-origins.test.js && node test/updates.test.js && node test/static-headers.test.js && node test/library-tags.test.js && node test/moderation-audit.test.js && node test/blocks.test.js && node test/site-config.test.js && node test/perf-budget.test.js && node test/shutdown.test.js && node test/loyalty-policy.test.js && node test/themes-review.test.js && node test/page-checks.test.js && node test/operator-checklist.test.js && node test/release-health.test.js",++ "test": "node test/registry-bootstrap.test.js && node test/nginx-generator-live.test.js && node test/admin-refresh.test.js && node test/history-sso.test.js && node test/sso-check.test.js && node test/fedcm.test.js && node test/domains.test.js && node test/frame.test.js && node test/github-integration.test.js && node test/revocation.test.js && node test/usernames.test.js && node test/avatar.test.js && node test/roadmap-baseline.test.js && node test/identity-subjects.test.js && node test/principals.test.js && node test/operator-alerts.test.js && node test/follows.test.js && node test/modules.test.js && node test/module-events.test.js && node test/registry-ecosystem.test.js && node test/registry-exposure.test.js && node test/registry-manifest-fixture.test.js && node test/staff-claims.test.js && node test/profile-events.test.js && node test/admin-role-events.test.js && node test/grants-admin.test.js && node test/events-ops.test.js && node test/deploy-drift.test.js && node test/registry-topics.test.js && node test/service-principal.test.js && node test/oauth-pkce.test.js && node test/developer-projects.test.js && node test/developer-defaults.test.js && node test/developer-export-tokens.test.js && node test/project-usage.test.js && node test/public-cors.test.js && node test/observability.test.js && node test/status.test.js && node test/events-consumer.test.js && node test/notification-events.test.js && node test/security-session.test.js && node test/security-redirects.test.js && node test/analytics-privacy.test.js && node test/data-ownership.test.js && node test/unknown-paths.test.js && node test/oidc-discovery.test.js && node test/refresh-tokens.test.js && node test/provider-secrets.test.js && node test/first-party-origins.test.js && node test/updates.test.js && node test/static-headers.test.js && node test/library-tags.test.js && node test/moderation-audit.test.js && node test/blocks.test.js && node test/site-config.test.js && node test/perf-budget.test.js && node test/shutdown.test.js && node test/loyalty-policy.test.js && node test/themes-review.test.js && node test/page-checks.test.js && node test/operator-checklist.test.js && node test/release-health.test.js","reset-db": "node server/reset-db.js","grant-admin": "node server/grant-admin.js"},@@@ -23,7 +23,7 @@"helmet": "^8.1.0","jsonwebtoken": "^9.0.0","multer": "^1.4.5-lts.1",-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.63.0",++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.65.0","openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.11.0","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.22.0","uuid": "^11.1.0",nnew file mode 100644@@@ -0,0 +1,102 @@++#!/usr/bin/env node++'use strict';++/**++ * ADR-030 steps 2 and 3 (roadmap WS-E task 4): Live's follows into Network's user_follows, then a++ * reconciliation of the two.++ *++ * sudo node scripts/follows-backfill.js --live-db /opt/openvibe.live/data/live.db++ * dry run: how many follows map to subjects, how many would be held (and why), nothing changed++ * sudo node scripts/follows-backfill.js --live-db … --apply --backup <new file>++ * takes a sqlite online backup of network.db first (0600, quick_check), then imports in one++ * transaction: no events (Live already has these follows); a pair whose side has no subject is++ * kept in follow_import_holds, never dropped++ * sudo node scripts/follows-backfill.js --live-db … --reconcile++ * compares every mapped channel's follower count and the pair sets; exit 1 on any difference++ *++ * Live's side is read only (a readonly connection). Live user ids map to subjects through Live's++ * linked_accounts (service 'network', subject_id): the same mapping Live's own reads use.++ * --db <path> network.db (default $DB_PATH, else data/network.db)++ */++const fs = require('fs');++const Database = require('better-sqlite3');++const ops = require('./lib/db-ops');++const follows = require('../server/identity/follows');++++const USAGE = 'usage: node scripts/follows-backfill.js --live-db <live.db> [--apply --backup <new file> | --reconcile] [--db <network.db>]';++const SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;++++/** Live's follows and its user id → subject map (readonly). */++function readLive(file) {++ const live = new Database(file, { readonly: true, fileMustExist: true });++ try {++ const rows = live.prepare('SELECT follower_id, streamer_id, email_notify, push_notify, created_at FROM follows ORDER BY id').all()++ .map((r) => ({ follower_ref: r.follower_id, target_ref: r.streamer_id, notify_email: !!r.email_notify, notify_push: !!r.push_notify, created_at: r.created_at ? `${String(r.created_at).replace(' ', 'T')}${/Z|[+-]\d\d:?\d\d$/.test(String(r.created_at)) ? '' : 'Z'}` : null }));++ const subjects = new Map();++ for (const r of live.prepare("SELECT user_id, subject_id FROM linked_accounts WHERE service = 'network' AND subject_id IS NOT NULL").all()) {++ if (SUBJECT_RE.test(String(r.subject_id))) subjects.set(Number(r.user_id), r.subject_id);++ }++ return { rows, subjectOf: (id) => subjects.get(Number(id)) || null };++ } finally { live.close(); }++}++++/** Counts per channel and pair sets, Live (mapped) against Network. → { ok, channels, missing, extra } */++function reconcile(db, rows, subjectOf) {++ const livePairs = new Set();++ for (const r of rows) {++ const a = subjectOf(r.follower_ref);++ const b = subjectOf(r.target_ref);++ if (a && b && a !== b) livePairs.add(`${a}>${b}`);++ }++ const netRows = db.prepare("SELECT follower_subject, target_id FROM user_follows WHERE target_type = 'channel' AND active = 1").all();++ const netPairs = new Set(netRows.map((r) => `${r.follower_subject}>${r.target_id}`));++ const missing = [...livePairs].filter((p) => !netPairs.has(p));++ const targets = new Set([...livePairs].map((p) => p.split('>')[1]));++ // Network may hold follows Live does not (made on Network after the import); only mapped channels are compared.++ const extra = [...netPairs].filter((p) => targets.has(p.split('>')[1]) && !livePairs.has(p));++ const channels = [...targets].map((t) => ({++ channel: t,++ live: [...livePairs].filter((p) => p.endsWith(`>${t}`)).length,++ network: follows.count(db, 'channel', t),++ }));++ return { ok: !missing.length && !extra.length && channels.every((c) => c.live === c.network), channels, missing, extra };++}++++async function main(argv, log = console.log) {++ let args;++ try { args = ops.parseArgs(argv, { flags: ['apply', 'reconcile'], values: ['db', 'backup', 'live-db'] }); } catch (e) { log(`${e.message}\n${USAGE}`); return 2; }++ if (args.help || !args['live-db']) { log(USAGE); return args.help ? 0 : 2; }++ const file = ops.dbPath(args);++ if (!fs.existsSync(file)) { log(`error: no database at ${file}`); return 2; }++ const { rows, subjectOf } = readLive(args['live-db']);++ ops.dropToOwnerOf(file, log);++ const db = new Database(file);++ db.pragma('busy_timeout = 5000');++ try {++ follows.ensureSchema(db);++ if (args.reconcile) {++ const r = reconcile(db, rows, subjectOf);++ log(`live follows ${rows.length}; channels compared ${r.channels.length}; missing on Network ${r.missing.length}; on Network only ${r.extra.length}`);++ for (const c of r.channels.filter((x) => x.live !== x.network)) log(` count differs: ${c.channel} live ${c.live} network ${c.network}`);++ log(r.ok ? 'RECONCILED: counts and pairs match' : 'NOT RECONCILED');++ return r.ok ? 0 : 1;++ }++ const plan = follows.importFollows(db, 'live', rows, subjectOf, { dryRun: true });++ log(`live follows ${rows.length}: import ${plan.imported}, already on Network ${plan.unchanged}, held ${plan.held.length}`);++ const reasons = {};++ for (const h of plan.held) reasons[h.reason] = (reasons[h.reason] || 0) + 1;++ for (const [k, v] of Object.entries(reasons)) log(` held (${k}): ${v}`);++ if (!args.apply) { log('dry run: nothing changed. Re-run with --apply --backup <new file>.'); return 0; }++ if (!args.backup) { log('refusing --apply without --backup <new file> (sqlite online backup, 0600)'); return 2; }++ const target = await ops.backupTo(db, args.backup, { expect: { table: 'users', rows: db.prepare('SELECT COUNT(*) AS n FROM users').get().n } });++ log(`backup ${target} (0600, quick_check ok)`);++ const out = follows.importFollows(db, 'live', rows, subjectOf);++ log(`imported ${out.imported}; already there ${out.unchanged}; held ${out.held.length} (follow_import_holds)`);++ const r = reconcile(db, rows, subjectOf);++ log(r.ok ? 'RECONCILED: counts and pairs match' : `NOT RECONCILED: missing ${r.missing.length}, extra ${r.extra.length}`);++ return r.ok ? 0 : 1;++ } finally { db.close(); }++}++++if (require.main === module) main(process.argv.slice(2)).then((code) => process.exit(code), (err) => { console.error(`error: ${err.message}`); process.exit(1); });++++module.exports = { main, readLive, reconcile };@@@ -748,6 +748,8 @@ function initDb(dbPath) {require('../developer/usage').ensure(db);// Platform blocks, keyed by subjects (server/identity/blocks.js).require('../identity/blocks').ensureSchema(db);++ // The follow graph, keyed by subjects (server/identity/follows.js; ADR-030).++ require('../identity/follows').ensureSchema(db);console.log('[DB] Central database initialized');return db;nnew file mode 100644@@@ -0,0 +1,318 @@++'use strict';++/**++ * The follow graph (roadmap WS-E task 4, ADR-030; Contracts 0.65.0 network.follow.created / .deleted,++ * network.follow-status-result@1, network.follow-list-result@1, network.follows.read).++ *++ * Network owns follows, keyed by subjects. Today a target is a Live channel, named by its owner's++ * subject (target_type `channel`); other kinds join as products need them. Live keeps a projection it can++ * rebuild from the events, and go-live notifications can read the followers here instead of asking Live.++ *++ * user_follows has one row per (follower, target_type, target_id). An unfollow keeps the row with++ * active = 0, so the pair's revision keeps growing: every change raises it by one, and consumers keep++ * the highest revision they have applied. Each change writes network.follow.created (a follow, or new++ * notify flags) or network.follow.deleted into network_event_outbox in the same transaction (relayed to++ * OpenVibe.Events by server/developer/event-relay.js). Following again with the same flags, or unfollowing++ * what is not followed, changes nothing and announces nothing.++ *++ * GET /api/v1/follows/:type/:target public: { target_type, target_id, followers }, and++ * for a signed-in caller following, notify flags, since++ * GET /api/v1/follows/:type/:target/followers the target's owner, or a service with network.follows.read++ * GET /api/v1/me/follows?type=&cursor=&limit= what the caller follows++ * PUT /api/v1/me/follows/:type/:target follow (201 new, 200 unchanged); body { notify_email?, notify_push? }++ * DELETE /api/v1/me/follows/:type/:target unfollow (200, idempotent)++ *++ * :target is a usr_ subject or a current username. Nobody follows themselves; guests do not follow.++ * Lists are never public (counts are). ADR-030's migration: importFollows() backfills Live's follows++ * (scripts/follows-backfill.js), holding any pair whose side has no subject in follow_import_holds.++ */++const express = require('express');++const { ids, validate, http } = require('openvibe-contracts');++const eventRelay = require('../developer/event-relay');++++const TYPES = ['channel'];++const SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;++const NAME_RE = /^[A-Za-z0-9_]{1,64}$/;++const MAX_FOLLOWING = 5000;++++class FollowError extends Error {++ constructor(status, code, detail) { super(detail); this.status = status; this.code = code; }++}++++function ensureSchema(db) {++ db.exec(`CREATE TABLE IF NOT EXISTS user_follows (++ follower_subject TEXT NOT NULL,++ target_type TEXT NOT NULL CHECK (target_type IN ('channel')),++ target_id TEXT NOT NULL,++ active INTEGER NOT NULL DEFAULT 1 CHECK (active IN (0, 1)),++ notify_email INTEGER NOT NULL DEFAULT 1 CHECK (notify_email IN (0, 1)),++ notify_push INTEGER NOT NULL DEFAULT 1 CHECK (notify_push IN (0, 1)),++ revision INTEGER NOT NULL DEFAULT 1 CHECK (revision >= 1),++ source TEXT NOT NULL DEFAULT 'network',++ created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')),++ updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')),++ PRIMARY KEY (follower_subject, target_type, target_id),++ CHECK (follower_subject <> target_id)++ );++ CREATE INDEX IF NOT EXISTS idx_user_follows_target ON user_follows(target_type, target_id, active, created_at);++ CREATE TABLE IF NOT EXISTS follow_import_holds (++ source TEXT NOT NULL,++ follower_ref TEXT NOT NULL,++ target_ref TEXT NOT NULL,++ reason TEXT NOT NULL,++ seen_at TEXT NOT NULL,++ PRIMARY KEY (source, follower_ref, target_ref)++ );`);++}++++const personRow = (db, where, value) => db.prepare(`SELECT id, subject_id, username, display_name, avatar_url, is_anon FROM users WHERE ${where}`).get(value);++++/** The target a path names: a usr_ subject or a current username (any case) → the person row, or null. */++function findTarget(db, type, key) {++ if (!TYPES.includes(type)) throw new FollowError(404, 'follows.unknown_type', `target type must be one of ${TYPES.join(', ')}`);++ const k = String(key || '').trim();++ if (SUBJECT_RE.test(k)) return personRow(db, 'subject_id = ?', k) || null;++ if (!NAME_RE.test(k)) return null;++ return personRow(db, 'username = ? COLLATE NOCASE', k) || null;++}++++function buildEnvelope({ follower, type, target, active, notifyEmail, notifyPush, revision, at, reason = null, actor = null }) {++ const ms = Date.parse(at) || Date.now();++ const eventType = active ? 'network.follow.created' : 'network.follow.deleted';++ const payload = active++ ? { follower, target_type: type, target_id: target, notify_email: !!notifyEmail, notify_push: !!notifyPush, revision, at: new Date(ms).toISOString() }++ : { follower, target_type: type, target_id: target, reason: reason || 'unfollowed', revision, at: new Date(ms).toISOString() };++ const env = {++ event_id: ids.newId('event', ms), event_type: eventType, version: 1, source: 'network',++ actor: actor || { type: 'user', id: follower }, timestamp: new Date(ms).toISOString(), visibility: 'subject',++ subject: { type: 'user', id: follower }, payload,++ };++ const v = validate('events.event-envelope@1', env);++ const pv = validate(`${eventType}@1`, payload);++ if (!v.valid || !pv.valid) throw new Error(`follows: bad event: ${JSON.stringify((v.errors || []).concat(pv.errors || [])).slice(0, 300)}`);++ return env;++}++++/**++ * Set (follower → type/target) to `active` with the notify flags, in one transaction with its event.++ * → { changed, active, notify_email, notify_push, revision, created_at, at }++ * opts: notifyEmail/notifyPush (undefined keeps the stored value, else true), reason and actor (deletes),++ * emit (false for an import: no event), source, at.++ */++function setFollow(db, follower, type, target, active, { notifyEmail, notifyPush, reason = null, actor = null, emit = true, source = 'network', at: atIn = null } = {}) {++ if (!SUBJECT_RE.test(String(follower)) || !SUBJECT_RE.test(String(target))) throw new FollowError(400, 'follows.bad_subject', 'follows are between usr_ subjects');++ if (!TYPES.includes(type)) throw new FollowError(404, 'follows.unknown_type', `target type must be one of ${TYPES.join(', ')}`);++ if (follower === target) throw new FollowError(400, 'follows.self', 'you cannot follow yourself');++ return db.transaction(() => {++ const prev = db.prepare('SELECT * FROM user_follows WHERE follower_subject = ? AND target_type = ? AND target_id = ?').get(follower, type, target);++ const email = notifyEmail === undefined ? (prev ? !!prev.notify_email : true) : !!notifyEmail;++ const push = notifyPush === undefined ? (prev ? !!prev.notify_push : true) : !!notifyPush;++ const same = prev && !!prev.active === !!active && (!active || (!!prev.notify_email === email && !!prev.notify_push === push));++ if (same) return { changed: false, active: !!active, notify_email: email, notify_push: push, revision: prev.revision, created_at: prev.created_at, at: prev.updated_at };++ if (!prev && !active) return { changed: false, active: false, notify_email: email, notify_push: push, revision: 0, created_at: null, at: null };++ if (active && !(prev && prev.active)) {++ const n = db.prepare('SELECT COUNT(*) AS n FROM user_follows WHERE follower_subject = ? AND active = 1').get(follower).n;++ if (n >= MAX_FOLLOWING) throw new FollowError(409, 'follows.limit', `at most ${MAX_FOLLOWING} follows`);++ }++ const at = atIn || new Date().toISOString();++ const revision = prev ? prev.revision + 1 : 1;++ // A follow that starts again starts its "since" again; flag changes keep it.++ const createdAt = prev && prev.active && active ? prev.created_at : at;++ if (prev) {++ db.prepare(`UPDATE user_follows SET active = ?, notify_email = ?, notify_push = ?, revision = ?, source = ?, created_at = ?, updated_at = ?++ WHERE follower_subject = ? AND target_type = ? AND target_id = ?`)++ .run(active ? 1 : 0, email ? 1 : 0, push ? 1 : 0, revision, source, createdAt, at, follower, type, target);++ } else {++ db.prepare(`INSERT INTO user_follows (follower_subject, target_type, target_id, active, notify_email, notify_push, revision, source, created_at, updated_at)++ VALUES (?, ?, ?, 1, ?, ?, 1, ?, ?, ?)`).run(follower, type, target, email ? 1 : 0, push ? 1 : 0, source, at, at);++ }++ if (emit) eventRelay.writerFor(db).enqueue(buildEnvelope({ follower, type, target, active, notifyEmail: email, notifyPush: push, revision, at, reason, actor }));++ return { changed: true, active: !!active, notify_email: email, notify_push: push, revision, created_at: active ? createdAt : null, at };++ })();++}++++function kick(db) { const o = eventRelay.outboxFor(db); if (o) o.kick(); }++++function count(db, type, target) {++ return db.prepare('SELECT COUNT(*) AS n FROM user_follows WHERE target_type = ? AND target_id = ? AND active = 1').get(type, target).n;++}++++/** network.follow-status-result@1 for (type, target), with the viewer's own follow when given. */++function status(db, type, target, viewer = null) {++ const out = { target_type: type, target_id: target, followers: count(db, type, target) };++ if (viewer && SUBJECT_RE.test(viewer)) {++ const r = db.prepare('SELECT * FROM user_follows WHERE follower_subject = ? AND target_type = ? AND target_id = ? AND active = 1').get(viewer, type, target);++ out.following = !!r;++ if (r) Object.assign(out, { notify_email: !!r.notify_email, notify_push: !!r.notify_push, since: r.created_at });++ }++ return out;++}++++const encodeCursor = (r) => Buffer.from(`${r.created_at}|${r.follower_subject}|${r.target_id}`).toString('base64url');++function decodeCursor(c) {++ if (!c) return null;++ const [at, a, b] = Buffer.from(String(c), 'base64url').toString('utf8').split('|');++ if (!at || !SUBJECT_RE.test(a || '') || !SUBJECT_RE.test(b || '')) throw new FollowError(400, 'follows.bad_cursor', 'cursor is not one this API issued');++ return { at, a, b };++}++const item = (r) => ({ follower: r.follower_subject, target_type: r.target_type, target_id: r.target_id, notify_email: !!r.notify_email, notify_push: !!r.notify_push, created_at: r.created_at });++++/** A page of follows (newest first): by follower (what they follow) or by target (who follows it). */++function list(db, { follower = null, type = null, target = null, cursor = null, limit = 50 } = {}) {++ const n = Math.min(Math.max(parseInt(limit, 10) || 50, 1), 200);++ const c = decodeCursor(cursor);++ const where = ['active = 1'];++ const args = [];++ if (follower) { where.push('follower_subject = ?'); args.push(follower); }++ if (type) { where.push('target_type = ?'); args.push(type); }++ if (target) { where.push('target_id = ?'); args.push(target); }++ if (c) { where.push('(created_at < ? OR (created_at = ? AND (follower_subject, target_id) < (?, ?)))'); args.push(c.at, c.at, c.a, c.b); }++ const rows = db.prepare(`SELECT * FROM user_follows WHERE ${where.join(' AND ')} ORDER BY created_at DESC, follower_subject DESC, target_id DESC LIMIT ?`).all(...args, n + 1);++ const more = rows.length > n;++ if (more) rows.pop();++ return { items: rows.map(item), next_cursor: more ? encodeCursor(rows[rows.length - 1]) : null };++}++++/**++ * An account's follows go with it (both directions), each announced as a delete. Call inside the++ * transaction that removes the account (account deletion, WS-B task 7). → the pairs removed++ */++function onSubjectRemoved(db, subject) {++ const rows = db.prepare('SELECT * FROM user_follows WHERE active = 1 AND (follower_subject = ? OR target_id = ?)').all(subject, subject);++ for (const r of rows) {++ setFollow(db, r.follower_subject, r.target_type, r.target_id, false, {++ reason: r.follower_subject === subject ? 'account_removed' : 'target_removed', actor: { type: 'system', id: 'network' },++ });++ }++ return rows.length;++}++++/**++ * ADR-030 step 2: import follows from another system's rows, mapped to subjects by `subjectOf`.++ * rows: [{ follower_ref, target_ref, notify_email, notify_push, created_at }] (the refs are that system's ids).++ * A pair whose side has no subject (or is the same person) is held in follow_import_holds, never dropped.++ * No events (the source system already has these follows); dryRun changes nothing.++ * → { imported, unchanged, held: [{ follower_ref, target_ref, reason }] }++ */++function importFollows(db, source, rows, subjectOf, { dryRun = false, now = new Date().toISOString() } = {}) {++ const out = { imported: 0, unchanged: 0, held: [] };++ const run = () => {++ for (const r of rows) {++ const follower = subjectOf(r.follower_ref);++ const target = subjectOf(r.target_ref);++ let reason = null;++ if (!follower) reason = 'follower_has_no_subject';++ else if (!target) reason = 'target_has_no_subject';++ else if (follower === target) reason = 'self_follow';++ if (reason) {++ out.held.push({ follower_ref: String(r.follower_ref), target_ref: String(r.target_ref), reason });++ if (!dryRun) db.prepare('INSERT OR REPLACE INTO follow_import_holds (source, follower_ref, target_ref, reason, seen_at) VALUES (?, ?, ?, ?, ?)').run(source, String(r.follower_ref), String(r.target_ref), reason, now);++ continue;++ }++ const prev = db.prepare('SELECT active FROM user_follows WHERE follower_subject = ? AND target_type = ? AND target_id = ?').get(follower, 'channel', target);++ if (prev && prev.active) { out.unchanged++; continue; }++ if (!dryRun) {++ setFollow(db, follower, 'channel', target, true, {++ notifyEmail: r.notify_email == null ? true : !!r.notify_email, notifyPush: r.notify_push == null ? true : !!r.notify_push,++ emit: false, source, at: r.created_at ? new Date(r.created_at).toISOString() : now,++ });++ }++ out.imported++;++ }++ };++ if (dryRun) run(); else db.transaction(run)();++ return out;++}++++/** The signed-in person as a follower, or a FollowError. */++function meOf(db, user) {++ if (!user || user.is_anon) throw new FollowError(403, 'follows.guest', 'sign in with an account to follow');++ const sid = require('./subjects').ensureUserSubject(db, user);++ if (!SUBJECT_RE.test(String(sid || ''))) throw new FollowError(403, 'follows.guest', 'sign in with an account to follow');++ return sid;++}++++/** The session user of a request, or null (never refuses: for public reads). */++function viewerOf(req) {++ const h = req.headers.authorization || '';++ const token = h.startsWith('Bearer ') ? h.slice(7) : (req.cookies && req.cookies.ov_token);++ if (!token) return null;++ try {++ const out = require('../auth/session').verifySession(token, { db: req.app.locals.db, publicKey: req.app.locals.publicKey, config: req.app.locals.config });++ return out && out.user ? out.user : null;++ } catch { return null; }++}++++function routers({ requireAuth, followsGuard }) {++ const send = (req, res, fn) => {++ res.set('Cache-Control', 'private, no-store');++ try { return fn(req.app.locals.db); } catch (err) {++ if (err instanceof FollowError) return http.sendProblem(res, err.status, err.code, { detail: err.message, ctx: req.ov });++ console.error('[Follows]', err.message);++ return http.sendProblem(res, 500, 'follows.failed', { detail: 'the follow could not be changed', ctx: req.ov });++ }++ };++ const targetOrFail = (db, type, key) => {++ const u = findTarget(db, type, key);++ if (!u || u.is_anon || !SUBJECT_RE.test(String(u.subject_id || ''))) throw new FollowError(404, 'follows.unknown_target', 'no such channel');++ return u;++ };++++ const me = express.Router();++ me.use(http.middleware());++ me.get('/', requireAuth, (req, res) => send(req, res, (db) => {++ const sid = meOf(db, req.user);++ const type = req.query.type ? String(req.query.type) : null;++ if (type && !TYPES.includes(type)) throw new FollowError(404, 'follows.unknown_type', `target type must be one of ${TYPES.join(', ')}`);++ res.json(list(db, { follower: sid, type, cursor: req.query.cursor, limit: req.query.limit }));++ }));++ me.put('/:type/:target', requireAuth, express.json({ limit: '4kb' }), (req, res) => send(req, res, (db) => {++ const sid = meOf(db, req.user);++ const t = targetOrFail(db, req.params.type, req.params.target);++ const b = req.body && typeof req.body === 'object' ? req.body : {};++ const out = setFollow(db, sid, req.params.type, t.subject_id, true, {++ notifyEmail: typeof b.notify_email === 'boolean' ? b.notify_email : undefined, notifyPush: typeof b.notify_push === 'boolean' ? b.notify_push : undefined,++ });++ if (out.changed) kick(db);++ res.status(out.changed && out.revision === 1 ? 201 : 200).json(status(db, req.params.type, t.subject_id, sid));++ }));++ me.delete('/:type/:target', requireAuth, (req, res) => send(req, res, (db) => {++ const sid = meOf(db, req.user);++ const key = String(req.params.target || '');++ // A subject can be unfollowed even if its account is gone since.++ const t = findTarget(db, req.params.type, key) || (SUBJECT_RE.test(key) ? { subject_id: key } : null);++ if (!t || !t.subject_id) throw new FollowError(404, 'follows.unknown_target', 'no such channel');++ const out = setFollow(db, sid, req.params.type, t.subject_id, false);++ if (out.changed) kick(db);++ res.json(status(db, req.params.type, t.subject_id, sid));++ }));++++ const pub = express.Router();++ pub.use(http.middleware());++ pub.get('/:type/:target', (req, res) => {++ try {++ const db = req.app.locals.db;++ const t = targetOrFail(db, req.params.type, req.params.target);++ const viewer = viewerOf(req);++ const body = status(db, req.params.type, t.subject_id, viewer && !viewer.is_anon ? viewer.subject_id : null);++ res.set('Cache-Control', viewer ? 'private, no-store' : 'public, max-age=30').set('Vary', 'Authorization, Cookie').json(body);++ } catch (err) {++ if (err instanceof FollowError) return http.sendProblem(res, err.status, err.code, { detail: err.message, ctx: req.ov });++ throw err;++ }++ });++ // Who follows a target: its owner (signed in), or a service holding network.follows.read.++ pub.get('/:type/:target/followers', (req, res, next) => {++ const db = req.app.locals.db;++ let t;++ try { t = targetOrFail(db, req.params.type, req.params.target); } catch (err) { return http.sendProblem(res, err.status, err.code, { detail: err.message, ctx: req.ov }); }++ const answer = () => send(req, res, () => res.json(list(db, { type: req.params.type, target: t.subject_id, cursor: req.query.cursor, limit: req.query.limit })));++ const viewer = viewerOf(req);++ if (viewer && viewer.subject_id === t.subject_id) return answer();++ if (viewer) return http.sendProblem(res, 403, 'follows.not_yours', { detail: 'only the channel owner sees who follows it', ctx: req.ov });++ return followsGuard(req, res, answer);++ });++++ return { me, pub };++}++++module.exports = { TYPES, FollowError, ensureSchema, setFollow, status, list, count, findTarget, buildEnvelope, onSubjectRemoved, importFollows, routers, kick };@@@ -121,6 +121,8 @@ const DEFAULT_GRANTS = [['coupons', 'events.event.publish', 'openvibe.events', []],['coupons', 'sources.item.read', 'openvibe.sources', []],['host', 'events.event.publish', 'openvibe.events', []],++ // ADR-030 (WS-E task 4): Live rebuilds its follows projection from Network's graph.++ ['live', 'network.follows.read', SELF_AUDIENCE, []],// WS-H task 11: Host relays the alerts firing on the production host (ovhost alerts relay).['host', 'network.operator.alert', SELF_AUDIENCE, []],// WS-L task 4: OpenVibe.Host's scheduled Tools job proof (openvibe-toolsjob.timer, OpenVibe.Examples@@@ -408,6 +408,13 @@ app.get('/internal/integrations/github-token', require('./identity/principals').// (network.blocks.read, service token only). Every change is network.block.changed (server/identity/blocks.js).app.use('/api/v1/me/blocks', rateLimit({ windowMs: 60_000, max: 60 }), require('./identity/blocks').userRouter(requireAuth));app.get('/internal/blocks', require('./identity/principals').guard('network.blocks.read', { legacy: false }), require('./identity/blocks').internalHandler(db));++// The follow graph (WS-E task 4, ADR-030): public counts, a person's own follows, and who follows a target++// (its owner, or network.follows.read, service token only). Every change is network.follow.* (server/identity/follows.js).++{++ const followRouters = require('./identity/follows').routers({ requireAuth, followsGuard: require('./identity/principals').guard('network.follows.read', { legacy: false }) });++ app.use('/api/v1/me/follows', rateLimit({ windowMs: 60_000, max: 120 }), followRouters.me);++ app.use('/api/v1/follows', rateLimit({ windowMs: 60_000, max: 300 }), followRouters.pub);++}// Realtime tickets (WS-E task 3, WS-F task 1; ADR-005 amendment 2): the notification badge on any site opens// OpenVibe.Events' /realtime/stream as the signed-in person with a two-minute, single-use ticket// (server/auth/realtime-ticket.js). REALTIME_TICKETS=off answers 503 and every badge stays on polling.nnew file mode 100644@@@ -0,0 +1,167 @@++'use strict';++// The follow graph (roadmap WS-E task 4, ADR-030; Contracts 0.65.0): follow and unfollow through++// /api/v1/me/follows (idempotent, never oneself, never a guest), each change writes network.follow.created /++// .deleted with a growing per-pair revision, counts are public and lists are not (the target's owner, or a++// service with network.follows.read), and scripts/follows-backfill.js imports Live's follows (held when a++// side has no subject), then reconciles.++// node test/follows.test.js++const assert = require('assert');++const crypto = require('crypto');++const fs = require('fs');++const os = require('os');++const path = require('path');++const http = require('http');++const express = require('express');++const Database = require('better-sqlite3');++const { validate } = require('openvibe-contracts');++const { initDb } = require('../server/db/database');++++const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-follows-'));++const dbFile = path.join(dir, 'network.db');++const log = console.log; console.log = () => {};++const db = initDb(dbFile);++console.log = log;++require('../server/identity/principals').ensureSchema(db);++for (const c of ['live', 'tools']) db.prepare('UPDATE oauth_clients SET client_secret = ? WHERE client_id = ?').run(`${c}-secret`, c);++++const ANN = 'usr_01JAB2C3D4E5F6G7H8J9K0MNPA';++const BOB = 'usr_01JAB2C3D4E5F6G7H8J9K0MNPB';++const CAT = 'usr_01JAB2C3D4E5F6G7H8J9K0MNPC';++db.prepare(`INSERT INTO users (id, username, display_name, password_hash, subject_id, role) VALUES++ (1, 'ann', 'Ann', 'x', ?, 'user'), (2, 'bob', 'Bob', 'x', ?, 'streamer'), (3, 'cat', NULL, 'x', ?, 'user')`).run(ANN, BOB, CAT);++db.prepare("INSERT INTO users (id, username, password_hash, is_anon) VALUES (5, 'anon1234', 'x', 1)").run();++++const keys = crypto.generateKeyPairSync('rsa', { modulusLength: 2048, publicKeyEncoding: { type: 'spki', format: 'pem' }, privateKeyEncoding: { type: 'pkcs8', format: 'pem' } });++const config = { internalKey: 'legacy-key', jwt: { issuer: 'https://openvibe.network', accessTokenExpiry: '1h' } };++const { signToken } = require('../server/auth/routes');++const requireAuth = require('../server/auth/session').makeRequireAuth(() => ({ db, publicKey: keys.publicKey, config }), signToken);++const follows = require('../server/identity/follows');++const principals = require('../server/identity/principals');++const app = express();++app.use(require('cookie-parser')());++app.use(express.urlencoded({ extended: true }));++Object.assign(app.locals, { db, config, privateKey: keys.privateKey, publicKey: keys.publicKey });++app.use('/oauth', require('../server/auth/oauth-routes'));++const routers = follows.routers({ requireAuth, followsGuard: principals.guard('network.follows.read', { legacy: false }) });++app.use('/api/v1/me/follows', routers.me);++app.use('/api/v1/follows', routers.pub);++const server = http.createServer(app);++++const events = () => db.prepare('SELECT envelope FROM network_event_outbox ORDER BY id').all().map((r) => JSON.parse(r.envelope)).filter((e) => e.event_type.startsWith('network.follow.'));++const valid = (e) => {++ assert.ok(validate('events.event-envelope@1', e).valid);++ const pv = validate(`${e.event_type}@1`, e.payload);++ assert.ok(pv.valid, JSON.stringify(pv.errors));++ assert.deepStrictEqual([e.source, e.visibility, e.subject], ['network', 'subject', { type: 'user', id: e.payload.follower }]);++};++++(async () => {++ await new Promise((r) => server.listen(0, '127.0.0.1', r));++ const base = `http://127.0.0.1:${server.address().port}`;++ const tok = (id) => signToken(db.prepare('SELECT * FROM users WHERE id = ?').get(id), keys.privateKey, config);++ const call = (method, p, { body, headers = {} } = {}) => fetch(base + p, { method, headers: { 'content-type': 'application/json', ...headers }, body: body ? JSON.stringify(body) : undefined })++ .then(async (r) => ({ status: r.status, cache: r.headers.get('cache-control'), body: await r.json().catch(() => null) }));++ const as = (id) => ({ authorization: `Bearer ${tok(id)}` });++ const svc = async (client) => (await (await fetch(`${base}/oauth/token`, { method: 'POST', body: new URLSearchParams({ grant_type: 'client_credentials', client_id: client, client_secret: `${client}-secret`, audience: 'openvibe.network' }) })).json()).access_token;++ try {++ // ── Follow, idempotently; events ──++ assert.strictEqual((await call('PUT', '/api/v1/me/follows/channel/bob')).status, 401, 'signed in only');++ let r = await call('PUT', '/api/v1/me/follows/channel/bob', { headers: as(1) });++ assert.strictEqual(r.status, 201);++ assert.ok(validate('network.follow-status-result@1', r.body).valid, JSON.stringify(r.body));++ assert.deepStrictEqual([r.body.target_id, r.body.followers, r.body.following, r.body.notify_email, r.body.notify_push], [BOB, 1, true, true, true]);++ r = await call('PUT', '/api/v1/me/follows/channel/Bob', { headers: as(1) });++ assert.deepStrictEqual([r.status, r.body.followers], [200, 1], 'following again changes nothing (and names are case-insensitive)');++ assert.strictEqual(events().length, 1, 'and announces nothing');++ r = await call('PUT', `/api/v1/me/follows/channel/${BOB}`, { headers: as(1), body: { notify_email: false } });++ assert.deepStrictEqual([r.status, r.body.notify_email, r.body.notify_push], [200, false, true], 'new flags: a change');++ let ev = events();++ assert.deepStrictEqual(ev.map((e) => [e.event_type, e.payload.revision, e.payload.notify_email]), [['network.follow.created', 1, true], ['network.follow.created', 2, false]]);++ ev.forEach(valid);++ assert.strictEqual((await call('PUT', '/api/v1/me/follows/channel/ann', { headers: as(1) })).status, 400, 'never oneself');++ assert.strictEqual((await call('PUT', '/api/v1/me/follows/channel/anon1234', { headers: as(1) })).status, 404, 'a guest is not a channel');++ assert.strictEqual((await call('PUT', '/api/v1/me/follows/channel/nobody', { headers: as(1) })).status, 404);++ assert.strictEqual((await call('PUT', '/api/v1/me/follows/space/bob', { headers: as(1) })).status, 404, 'unknown target type');++ assert.strictEqual((await call('PUT', '/api/v1/me/follows/channel/bob', { headers: as(5) })).status, 403, 'guests do not follow');++ await call('PUT', '/api/v1/me/follows/channel/bob', { headers: as(3) });++++ // ── Public count; the viewer's own state; lists are not public ──++ r = await call('GET', '/api/v1/follows/channel/bob');++ assert.deepStrictEqual([r.status, r.body], [200, { target_type: 'channel', target_id: BOB, followers: 2 }]);++ assert.match(r.cache, /public/);++ r = await call('GET', '/api/v1/follows/channel/bob', { headers: as(1) });++ assert.deepStrictEqual([r.body.following, r.body.notify_email], [true, false]);++ assert.match(r.cache, /private/);++ r = await call('GET', '/api/v1/me/follows', { headers: as(1) });++ assert.ok(validate('network.follow-list-result@1', r.body).valid);++ assert.deepStrictEqual(r.body.items.map((i) => i.target_id), [BOB]);++ assert.ok([401, 403].includes((await call('GET', '/api/v1/follows/channel/bob/followers')).status), 'no token: not the owner, not a service');++ assert.strictEqual((await call('GET', '/api/v1/follows/channel/bob/followers', { headers: as(1) })).status, 403, 'a follower is not the owner');++ r = await call('GET', '/api/v1/follows/channel/bob/followers', { headers: as(2) });++ assert.deepStrictEqual([r.status, r.body.items.map((i) => i.follower).sort()], [200, [ANN, CAT].sort()], 'the owner sees who follows');++ r = await call('GET', '/api/v1/follows/channel/bob/followers?limit=1', { headers: { authorization: `Bearer ${await svc('live')}` } });++ assert.deepStrictEqual([r.status, r.body.items.length, typeof r.body.next_cursor], [200, 1, 'string'], 'a service with network.follows.read, paged');++ const page2 = await call('GET', `/api/v1/follows/channel/bob/followers?limit=1&cursor=${r.body.next_cursor}`, { headers: { authorization: `Bearer ${await svc('live')}` } });++ assert.strictEqual(page2.body.items.length, 1);++ assert.notStrictEqual(page2.body.items[0].follower, r.body.items[0].follower, 'the cursor moves on');++ assert.strictEqual(page2.body.next_cursor, null);++ assert.strictEqual((await call('GET', '/api/v1/follows/channel/bob/followers', { headers: { authorization: `Bearer ${await svc('tools')}` } })).status, 403, 'a service without the capability');++++ // ── Unfollow, idempotently ──++ r = await call('DELETE', '/api/v1/me/follows/channel/bob', { headers: as(1) });++ assert.deepStrictEqual([r.status, r.body.following, r.body.followers], [200, false, 1]);++ r = await call('DELETE', '/api/v1/me/follows/channel/bob', { headers: as(1) });++ assert.strictEqual(r.status, 200);++ ev = events().filter((e) => e.event_type === 'network.follow.deleted');++ assert.deepStrictEqual(ev.map((e) => [e.payload.revision, e.payload.reason]), [[3, 'unfollowed']], 'one delete, revision 3');++ ev.forEach(valid);++ r = await call('PUT', '/api/v1/me/follows/channel/bob', { headers: as(1) });++ assert.strictEqual(r.status, 200, 'a follow again (the row exists; revision 4)');++ assert.strictEqual(events().pop().payload.revision, 4);++++ // ── Account removal ──++ assert.strictEqual(db.transaction(() => follows.onSubjectRemoved(db, CAT))(), 1);++ assert.deepStrictEqual(events().pop().payload.reason, 'account_removed');++++ // ── Live backfill and reconciliation (scripts/follows-backfill.js) ──++ const liveFile = path.join(dir, 'live.db');++ const live = new Database(liveFile);++ live.exec(`CREATE TABLE follows (id INTEGER PRIMARY KEY AUTOINCREMENT, follower_id INTEGER NOT NULL, streamer_id INTEGER NOT NULL, email_notify INTEGER DEFAULT 0, push_notify INTEGER DEFAULT 0, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, UNIQUE(follower_id, streamer_id));++ CREATE TABLE linked_accounts (id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER NOT NULL, service TEXT NOT NULL, service_user_id TEXT NOT NULL, subject_id TEXT);`);++ live.prepare("INSERT INTO linked_accounts (user_id, service, service_user_id, subject_id) VALUES (11, 'network', '1', ?), (12, 'network', '2', ?), (13, 'network', '3', ?)").run(ANN, BOB, CAT);++ live.prepare("INSERT INTO follows (follower_id, streamer_id, email_notify, push_notify, created_at) VALUES (11, 12, 1, 0, '2026-09-01 10:00:00'), (13, 12, 0, 1, '2026-09-02 10:00:00'), (14, 12, 0, 0, '2026-09-03 10:00:00'), (13, 11, 0, 0, '2026-09-04 10:00:00')").run();++ live.close();++ const backfill = require('../scripts/follows-backfill');++ const lines = [];++ const out = (s) => lines.push(s);++ db.close();++ let code = await backfill.main(['--live-db', liveFile, '--db', dbFile], out);++ assert.strictEqual(code, 0);++ assert.match(lines.join('\n'), /live follows 4: import 2, already on Network 1, held 1/);++ assert.match(lines.join('\n'), /held \(follower_has_no_subject\): 1/);++ code = await backfill.main(['--live-db', liveFile, '--db', dbFile, '--apply'], out);++ assert.strictEqual(code, 2, 'refuses --apply without --backup');++ code = await backfill.main(['--live-db', liveFile, '--db', dbFile, '--apply', '--backup', path.join(dir, 'pre-follows.db')], out);++ assert.strictEqual(code, 0, lines.join('\n'));++ assert.ok(fs.existsSync(path.join(dir, 'pre-follows.db')), 'backup taken first');++ assert.match(lines.join('\n'), /imported {3}2; already there 1; held 1/);++ assert.match(lines.join('\n'), /RECONCILED: counts and pairs match/);++ code = await backfill.main(['--live-db', liveFile, '--db', dbFile, '--reconcile'], out);++ assert.strictEqual(code, 0);++ const db2 = new Database(dbFile);++ assert.deepStrictEqual(db2.prepare("SELECT reason FROM follow_import_holds WHERE source = 'live'").all().map((x) => x.reason), ['follower_has_no_subject'], 'held, never dropped');++ const cat = db2.prepare('SELECT notify_email, notify_push, source, created_at FROM user_follows WHERE follower_subject = ? AND target_id = ?').get(CAT, BOB);++ assert.deepStrictEqual(cat, { notify_email: 0, notify_push: 1, source: 'live', created_at: '2026-09-02T10:00:00.000Z' }, 'flags and date carried over');++ const before = db2.prepare('SELECT COUNT(*) AS n FROM network_event_outbox').get().n;++ db2.close();++ code = await backfill.main(['--live-db', liveFile, '--db', dbFile, '--apply', '--backup', path.join(dir, 'pre-follows-2.db')], out);++ assert.strictEqual(code, 0, 'a second run is a no-op');++ const db3 = new Database(dbFile, { readonly: true });++ assert.strictEqual(db3.prepare('SELECT COUNT(*) AS n FROM network_event_outbox').get().n, before, 'the import sends no events');++ db3.close();++ } finally {++ server.close();++ fs.rmSync(dir, { recursive: true, force: true });++ }++ console.log('follows: all checks passed');++})().catch((e) => { console.error(e); process.exit(1); });@@@ -50,7 +50,7 @@ const server = http.createServer(app);const claims = serviceAuth.verifyServiceToken(t.body.access_token, { publicKey: keys.publicKey, issuer: ISSUER, audience: 'openvibe.network' });assert.ok(claims.ok, claims.reason);assert.strictEqual(claims.claims.sub, 'svc:live');-- assert.deepStrictEqual(claims.claims.cap, ['identity.subject.resolve', 'network.coins.credit', 'network.coins.debit', 'network.modules.read', 'network.modules.write', 'network.notifications.push']);++ assert.deepStrictEqual(claims.claims.cap, ['identity.subject.resolve', 'network.coins.credit', 'network.coins.debit', 'network.follows.read', 'network.modules.read', 'network.modules.write', 'network.notifications.push']);assert.ok(validate('identity.service-token-claims@1', claims.claims).valid);assert.ok(claims.claims.exp - claims.claims.iat <= 300, 'short-lived');const full = t.body.access_token;
@@ -408,6 +408,24 @@ Contracts 0.49.0; `server/identity/blocks.js`). `user_blocks` is keyed by subjec
@@ -408,6 +408,24 @@ Contracts 0.49.0; `server/identity/blocks.js`). `user_blocks` is keyed by subjec
---
---
+## Follow graph (ADR-030)
+
+Network owns follows, keyed by subjects (`server/identity/follows.js`; roadmap WS-E task 4, Contracts 0.65.0).
+A target is a Live channel today (`channel`, named by its owner's subject), and other kinds join as products need them.
+
+- **Public:** `GET /api/v1/follows/:type/:target` gives the follower count, and for a signed-in caller whether
+ they follow and how they are notified.
+- **People:** `GET /api/v1/me/follows` lists what the caller follows. `PUT|DELETE /api/v1/me/follows/:type/:target`
+ follow and unfollow (`:target` is a subject or a username). Both are idempotent, and the PUT body can carry
+ `{ notify_email, notify_push }`.
+- **Who follows a target:** `GET /api/v1/follows/:type/:target/followers` is open to the target's owner, and to
+ services with `network.follows.read` (service token only). Lists are never public.
+- **Events:** every change writes `network.follow.created` or `network.follow.deleted` (subject visibility,
+ growing per-pair revision) to the outbox in the same transaction. Live keeps a projection from them.
+- **Migration:** `scripts/follows-backfill.js --live-db <live.db>` does a dry run by default. `--apply --backup <file>`
+ imports Live's follows without events: a pair whose side has no subject goes to `follow_import_holds`, and
+ nothing is dropped. `--reconcile` compares per-channel counts and pair sets.
+
## Multi-Account Switching
## Multi-Account Switching
Google-style account management supporting up to 5 accounts:
Google-style account management supporting up to 5 accounts:
@@ -611,6 +629,8 @@ Accessible to users with `role = 'admin'`. All endpoints under `/api/admin/`.
@@ -611,6 +629,8 @@ Accessible to users with `role = 'admin'`. All endpoints under `/api/admin/`.
| `user_modules`, `user_module_revisions`, `user_module_retirements` | User-module records, the last revision issued per (subject, namespace), retired namespace owners |
| `user_modules`, `user_module_revisions`, `user_module_retirements` | User-module records, the last revision issued per (subject, namespace), retired namespace owners |
| `network_event_outbox` | Network's events on their way to OpenVibe.Events (developer projects, user modules, blocks) |
| `network_event_outbox` | Network's events on their way to OpenVibe.Events (developer projects, user modules, blocks) |
| `user_blocks` | Platform blocks by subject (blocker, blocked, active, per-pair revision) |
| `user_blocks` | Platform blocks by subject (blocker, blocked, active, per-pair revision) |
+| `user_follows` | The follow graph by subject (follower, target type and id, active, notify flags, per-pair revision, source) |
+| `follow_import_holds` | Imported follows held because a side had no subject (ADR-030 step 2) |
| `analytics_visitor_days`, `analytics_day_salts` | The current day's salted visitor hashes and salt, deleted after the day's final rollup |
| `analytics_visitor_days`, `analytics_day_salts` | The current day's salted visitor hashes and salt, deleted after the day's final rollup |
---
---
@@ -20,7 +20,7 @@
@@ -20,7 +20,7 @@
"helmet": "^8.1.0",
"helmet": "^8.1.0",
"jsonwebtoken": "^9.0.0",
"jsonwebtoken": "^9.0.0",
"multer": "^1.4.5-lts.1",
"multer": "^1.4.5-lts.1",
- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.63.0",
+ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.65.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.11.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.11.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.22.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.22.0",
"uuid": "^11.1.0",
"uuid": "^11.1.0",
@@ -1465,9 +1465,9 @@
@@ -1465,9 +1465,9 @@
}
}
},
},
"node_modules/openvibe-contracts": {
"node_modules/openvibe-contracts": {
- "version": "0.63.0",
- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.63.0",
- "integrity": "sha512-vmc1b1vWg9M5fm57X50MlZEZHkP2IshHwBXt68+kHDV85nyg6FxaYyUWa4ibi1eeFqzMcECPlILSRTOgg2Xbgw==",
+ "version": "0.65.0",
+ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.65.0",
+ "integrity": "sha512-oBWCCxa0iThdh28hVCutTVyf91EDsK9+O54xHAW4XNi0CVGQaynLGckE08GWoMCd4T9Q5JEDDPfCA2TIssf1ew==",
"license": "AGPL-3.0-only",
"license": "AGPL-3.0-only",
"dependencies": {
"dependencies": {
"ajv": "^8.17.1",
"ajv": "^8.17.1",
@@ -6,7 +6,7 @@
@@ -6,7 +6,7 @@
"scripts": {
"scripts": {
"start": "node server/index.js",
"start": "node server/index.js",
"dev": "NODE_ENV=development node server/index.js",
"dev": "NODE_ENV=development node server/index.js",
- "test": "node test/registry-bootstrap.test.js && node test/nginx-generator-live.test.js && node test/admin-refresh.test.js && node test/history-sso.test.js && node test/sso-check.test.js && node test/fedcm.test.js && node test/domains.test.js && node test/frame.test.js && node test/github-integration.test.js && node test/revocation.test.js && node test/usernames.test.js && node test/avatar.test.js && node test/roadmap-baseline.test.js && node test/identity-subjects.test.js && node test/principals.test.js && node test/operator-alerts.test.js && node test/modules.test.js && node test/module-events.test.js && node test/registry-ecosystem.test.js && node test/registry-exposure.test.js && node test/registry-manifest-fixture.test.js && node test/staff-claims.test.js && node test/profile-events.test.js && node test/admin-role-events.test.js && node test/grants-admin.test.js && node test/events-ops.test.js && node test/deploy-drift.test.js && node test/registry-topics.test.js && node test/service-principal.test.js && node test/oauth-pkce.test.js && node test/developer-projects.test.js && node test/developer-defaults.test.js && node test/developer-export-tokens.test.js && node test/project-usage.test.js && node test/public-cors.test.js && node test/observability.test.js && node test/status.test.js && node test/events-consumer.test.js && node test/notification-events.test.js && node test/security-session.test.js && node test/security-redirects.test.js && node test/analytics-privacy.test.js && node test/data-ownership.test.js && node test/unknown-paths.test.js && node test/oidc-discovery.test.js && node test/refresh-tokens.test.js && node test/provider-secrets.test.js && node test/first-party-origins.test.js && node test/updates.test.js && node test/static-headers.test.js && node test/library-tags.test.js && node test/moderation-audit.test.js && node test/blocks.test.js && node test/site-config.test.js && node test/perf-budget.test.js && node test/shutdown.test.js && node test/loyalty-policy.test.js && node test/themes-review.test.js && node test/page-checks.test.js && node test/operator-checklist.test.js && node test/release-health.test.js",
+ "test": "node test/registry-bootstrap.test.js && node test/nginx-generator-live.test.js && node test/admin-refresh.test.js && node test/history-sso.test.js && node test/sso-check.test.js && node test/fedcm.test.js && node test/domains.test.js && node test/frame.test.js && node test/github-integration.test.js && node test/revocation.test.js && node test/usernames.test.js && node test/avatar.test.js && node test/roadmap-baseline.test.js && node test/identity-subjects.test.js && node test/principals.test.js && node test/operator-alerts.test.js && node test/follows.test.js && node test/modules.test.js && node test/module-events.test.js && node test/registry-ecosystem.test.js && node test/registry-exposure.test.js && node test/registry-manifest-fixture.test.js && node test/staff-claims.test.js && node test/profile-events.test.js && node test/admin-role-events.test.js && node test/grants-admin.test.js && node test/events-ops.test.js && node test/deploy-drift.test.js && node test/registry-topics.test.js && node test/service-principal.test.js && node test/oauth-pkce.test.js && node test/developer-projects.test.js && node test/developer-defaults.test.js && node test/developer-export-tokens.test.js && node test/project-usage.test.js && node test/public-cors.test.js && node test/observability.test.js && node test/status.test.js && node test/events-consumer.test.js && node test/notification-events.test.js && node test/security-session.test.js && node test/security-redirects.test.js && node test/analytics-privacy.test.js && node test/data-ownership.test.js && node test/unknown-paths.test.js && node test/oidc-discovery.test.js && node test/refresh-tokens.test.js && node test/provider-secrets.test.js && node test/first-party-origins.test.js && node test/updates.test.js && node test/static-headers.test.js && node test/library-tags.test.js && node test/moderation-audit.test.js && node test/blocks.test.js && node test/site-config.test.js && node test/perf-budget.test.js && node test/shutdown.test.js && node test/loyalty-policy.test.js && node test/themes-review.test.js && node test/page-checks.test.js && node test/operator-checklist.test.js && node test/release-health.test.js",
"reset-db": "node server/reset-db.js",
"reset-db": "node server/reset-db.js",
"grant-admin": "node server/grant-admin.js"
"grant-admin": "node server/grant-admin.js"
},
},
@@ -23,7 +23,7 @@
@@ -23,7 +23,7 @@
"helmet": "^8.1.0",
"helmet": "^8.1.0",
"jsonwebtoken": "^9.0.0",
"jsonwebtoken": "^9.0.0",
"multer": "^1.4.5-lts.1",
"multer": "^1.4.5-lts.1",
- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.63.0",
+ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.65.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.11.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.11.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.22.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.22.0",
"uuid": "^11.1.0",
"uuid": "^11.1.0",
new file mode 100644
new file mode 100644
@@ -0,0 +1,102 @@
@@ -0,0 +1,102 @@
+#!/usr/bin/env node
+'use strict';
+/**
+ * ADR-030 steps 2 and 3 (roadmap WS-E task 4): Live's follows into Network's user_follows, then a
+ * reconciliation of the two.
+ *
+ * sudo node scripts/follows-backfill.js --live-db /opt/openvibe.live/data/live.db
+ * dry run: how many follows map to subjects, how many would be held (and why), nothing changed
+ * sudo node scripts/follows-backfill.js --live-db … --apply --backup <new file>
+ * takes a sqlite online backup of network.db first (0600, quick_check), then imports in one
+ * transaction: no events (Live already has these follows); a pair whose side has no subject is
+ * kept in follow_import_holds, never dropped
+ * sudo node scripts/follows-backfill.js --live-db … --reconcile
+ * compares every mapped channel's follower count and the pair sets; exit 1 on any difference
+ *
+ * Live's side is read only (a readonly connection). Live user ids map to subjects through Live's
+ * linked_accounts (service 'network', subject_id): the same mapping Live's own reads use.
+ * --db <path> network.db (default $DB_PATH, else data/network.db)
+ */
+const fs = require('fs');
+const Database = require('better-sqlite3');
+const ops = require('./lib/db-ops');
+const follows = require('../server/identity/follows');
+
+const USAGE = 'usage: node scripts/follows-backfill.js --live-db <live.db> [--apply --backup <new file> | --reconcile] [--db <network.db>]';
+const SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
+
+/** Live's follows and its user id → subject map (readonly). */
+function readLive(file) {
+ const live = new Database(file, { readonly: true, fileMustExist: true });
+ try {
+ const rows = live.prepare('SELECT follower_id, streamer_id, email_notify, push_notify, created_at FROM follows ORDER BY id').all()
+ .map((r) => ({ follower_ref: r.follower_id, target_ref: r.streamer_id, notify_email: !!r.email_notify, notify_push: !!r.push_notify, created_at: r.created_at ? `${String(r.created_at).replace(' ', 'T')}${/Z|[+-]\d\d:?\d\d$/.test(String(r.created_at)) ? '' : 'Z'}` : null }));
+ const subjects = new Map();
+ for (const r of live.prepare("SELECT user_id, subject_id FROM linked_accounts WHERE service = 'network' AND subject_id IS NOT NULL").all()) {
+ if (SUBJECT_RE.test(String(r.subject_id))) subjects.set(Number(r.user_id), r.subject_id);
+ }
+ return { rows, subjectOf: (id) => subjects.get(Number(id)) || null };
+ } finally { live.close(); }
+}
+
+/** Counts per channel and pair sets, Live (mapped) against Network. → { ok, channels, missing, extra } */
+function reconcile(db, rows, subjectOf) {
+ const livePairs = new Set();
+ for (const r of rows) {
+ const a = subjectOf(r.follower_ref);
+ const b = subjectOf(r.target_ref);
+ if (a && b && a !== b) livePairs.add(`${a}>${b}`);
+ }
+ const netRows = db.prepare("SELECT follower_subject, target_id FROM user_follows WHERE target_type = 'channel' AND active = 1").all();
+ const netPairs = new Set(netRows.map((r) => `${r.follower_subject}>${r.target_id}`));
+ const missing = [...livePairs].filter((p) => !netPairs.has(p));
+ const targets = new Set([...livePairs].map((p) => p.split('>')[1]));
+ // Network may hold follows Live does not (made on Network after the import); only mapped channels are compared.
+ const extra = [...netPairs].filter((p) => targets.has(p.split('>')[1]) && !livePairs.has(p));
+ const channels = [...targets].map((t) => ({
+ channel: t,
+ live: [...livePairs].filter((p) => p.endsWith(`>${t}`)).length,
+ network: follows.count(db, 'channel', t),
+ }));
+ return { ok: !missing.length && !extra.length && channels.every((c) => c.live === c.network), channels, missing, extra };
+}
+
+async function main(argv, log = console.log) {
+ let args;
+ try { args = ops.parseArgs(argv, { flags: ['apply', 'reconcile'], values: ['db', 'backup', 'live-db'] }); } catch (e) { log(`${e.message}\n${USAGE}`); return 2; }
+ if (args.help || !args['live-db']) { log(USAGE); return args.help ? 0 : 2; }
+ const file = ops.dbPath(args);
+ if (!fs.existsSync(file)) { log(`error: no database at ${file}`); return 2; }
+ const { rows, subjectOf } = readLive(args['live-db']);
+ ops.dropToOwnerOf(file, log);
+ const db = new Database(file);
+ db.pragma('busy_timeout = 5000');
+ try {
+ follows.ensureSchema(db);
+ if (args.reconcile) {
+ const r = reconcile(db, rows, subjectOf);
+ log(`live follows ${rows.length}; channels compared ${r.channels.length}; missing on Network ${r.missing.length}; on Network only ${r.extra.length}`);
+ for (const c of r.channels.filter((x) => x.live !== x.network)) log(` count differs: ${c.channel} live ${c.live} network ${c.network}`);
+ log(r.ok ? 'RECONCILED: counts and pairs match' : 'NOT RECONCILED');
+ return r.ok ? 0 : 1;
+ }
+ const plan = follows.importFollows(db, 'live', rows, subjectOf, { dryRun: true });
+ log(`live follows ${rows.length}: import ${plan.imported}, already on Network ${plan.unchanged}, held ${plan.held.length}`);
+ const reasons = {};
+ for (const h of plan.held) reasons[h.reason] = (reasons[h.reason] || 0) + 1;
+ for (const [k, v] of Object.entries(reasons)) log(` held (${k}): ${v}`);
+ if (!args.apply) { log('dry run: nothing changed. Re-run with --apply --backup <new file>.'); return 0; }
+ if (!args.backup) { log('refusing --apply without --backup <new file> (sqlite online backup, 0600)'); return 2; }
+ const target = await ops.backupTo(db, args.backup, { expect: { table: 'users', rows: db.prepare('SELECT COUNT(*) AS n FROM users').get().n } });
+ log(`backup ${target} (0600, quick_check ok)`);
+ const out = follows.importFollows(db, 'live', rows, subjectOf);
+ log(`imported ${out.imported}; already there ${out.unchanged}; held ${out.held.length} (follow_import_holds)`);
+ const r = reconcile(db, rows, subjectOf);
+ log(r.ok ? 'RECONCILED: counts and pairs match' : `NOT RECONCILED: missing ${r.missing.length}, extra ${r.extra.length}`);
+ return r.ok ? 0 : 1;
+ } finally { db.close(); }
+}
+
+if (require.main === module) main(process.argv.slice(2)).then((code) => process.exit(code), (err) => { console.error(`error: ${err.message}`); process.exit(1); });
+
+module.exports = { main, readLive, reconcile };
@@ -748,6 +748,8 @@ function initDb(dbPath) {
@@ -748,6 +748,8 @@ function initDb(dbPath) {
require('../developer/usage').ensure(db);
require('../developer/usage').ensure(db);
// Platform blocks, keyed by subjects (server/identity/blocks.js).
// Platform blocks, keyed by subjects (server/identity/blocks.js).
require('../identity/blocks').ensureSchema(db);
require('../identity/blocks').ensureSchema(db);
+ // The follow graph, keyed by subjects (server/identity/follows.js; ADR-030).
+ require('../identity/follows').ensureSchema(db);
console.log('[DB] Central database initialized');
console.log('[DB] Central database initialized');
return db;
return db;
new file mode 100644
new file mode 100644
@@ -0,0 +1,318 @@
@@ -0,0 +1,318 @@
+'use strict';
+/**
+ * The follow graph (roadmap WS-E task 4, ADR-030; Contracts 0.65.0 network.follow.created / .deleted,
+ * network.follow-status-result@1, network.follow-list-result@1, network.follows.read).
+ *
+ * Network owns follows, keyed by subjects. Today a target is a Live channel, named by its owner's
+ * subject (target_type `channel`); other kinds join as products need them. Live keeps a projection it can
+ * rebuild from the events, and go-live notifications can read the followers here instead of asking Live.
+ *
+ * user_follows has one row per (follower, target_type, target_id). An unfollow keeps the row with
+ * active = 0, so the pair's revision keeps growing: every change raises it by one, and consumers keep
+ * the highest revision they have applied. Each change writes network.follow.created (a follow, or new
+ * notify flags) or network.follow.deleted into network_event_outbox in the same transaction (relayed to
+ * OpenVibe.Events by server/developer/event-relay.js). Following again with the same flags, or unfollowing
+ * what is not followed, changes nothing and announces nothing.
+ *
+ * GET /api/v1/follows/:type/:target public: { target_type, target_id, followers }, and
+ * for a signed-in caller following, notify flags, since
+ * GET /api/v1/follows/:type/:target/followers the target's owner, or a service with network.follows.read
+ * GET /api/v1/me/follows?type=&cursor=&limit= what the caller follows
+ * PUT /api/v1/me/follows/:type/:target follow (201 new, 200 unchanged); body { notify_email?, notify_push? }
+ * DELETE /api/v1/me/follows/:type/:target unfollow (200, idempotent)
+ *
+ * :target is a usr_ subject or a current username. Nobody follows themselves; guests do not follow.
+ * Lists are never public (counts are). ADR-030's migration: importFollows() backfills Live's follows
+ * (scripts/follows-backfill.js), holding any pair whose side has no subject in follow_import_holds.
+ */
+const express = require('express');
+const { ids, validate, http } = require('openvibe-contracts');
+const eventRelay = require('../developer/event-relay');
+
+const TYPES = ['channel'];
+const SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
+const NAME_RE = /^[A-Za-z0-9_]{1,64}$/;
+const MAX_FOLLOWING = 5000;
+
+class FollowError extends Error {
+ constructor(status, code, detail) { super(detail); this.status = status; this.code = code; }
+}
+
+function ensureSchema(db) {
+ db.exec(`CREATE TABLE IF NOT EXISTS user_follows (
+ follower_subject TEXT NOT NULL,
+ target_type TEXT NOT NULL CHECK (target_type IN ('channel')),
+ target_id TEXT NOT NULL,
+ active INTEGER NOT NULL DEFAULT 1 CHECK (active IN (0, 1)),
+ notify_email INTEGER NOT NULL DEFAULT 1 CHECK (notify_email IN (0, 1)),
+ notify_push INTEGER NOT NULL DEFAULT 1 CHECK (notify_push IN (0, 1)),
+ revision INTEGER NOT NULL DEFAULT 1 CHECK (revision >= 1),
+ source TEXT NOT NULL DEFAULT 'network',
+ created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')),
+ updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')),
+ PRIMARY KEY (follower_subject, target_type, target_id),
+ CHECK (follower_subject <> target_id)
+ );
+ CREATE INDEX IF NOT EXISTS idx_user_follows_target ON user_follows(target_type, target_id, active, created_at);
+ CREATE TABLE IF NOT EXISTS follow_import_holds (
+ source TEXT NOT NULL,
+ follower_ref TEXT NOT NULL,
+ target_ref TEXT NOT NULL,
+ reason TEXT NOT NULL,
+ seen_at TEXT NOT NULL,
+ PRIMARY KEY (source, follower_ref, target_ref)
+ );`);
+}
+
+const personRow = (db, where, value) => db.prepare(`SELECT id, subject_id, username, display_name, avatar_url, is_anon FROM users WHERE ${where}`).get(value);
+
+/** The target a path names: a usr_ subject or a current username (any case) → the person row, or null. */
+function findTarget(db, type, key) {
+ if (!TYPES.includes(type)) throw new FollowError(404, 'follows.unknown_type', `target type must be one of ${TYPES.join(', ')}`);
+ const k = String(key || '').trim();
+ if (SUBJECT_RE.test(k)) return personRow(db, 'subject_id = ?', k) || null;
+ if (!NAME_RE.test(k)) return null;
+ return personRow(db, 'username = ? COLLATE NOCASE', k) || null;
+}
+
+function buildEnvelope({ follower, type, target, active, notifyEmail, notifyPush, revision, at, reason = null, actor = null }) {
+ const ms = Date.parse(at) || Date.now();
+ const eventType = active ? 'network.follow.created' : 'network.follow.deleted';
+ const payload = active
+ ? { follower, target_type: type, target_id: target, notify_email: !!notifyEmail, notify_push: !!notifyPush, revision, at: new Date(ms).toISOString() }
+ : { follower, target_type: type, target_id: target, reason: reason || 'unfollowed', revision, at: new Date(ms).toISOString() };
+ const env = {
+ event_id: ids.newId('event', ms), event_type: eventType, version: 1, source: 'network',
+ actor: actor || { type: 'user', id: follower }, timestamp: new Date(ms).toISOString(), visibility: 'subject',
+ subject: { type: 'user', id: follower }, payload,
+ };
+ const v = validate('events.event-envelope@1', env);
+ const pv = validate(`${eventType}@1`, payload);
+ if (!v.valid || !pv.valid) throw new Error(`follows: bad event: ${JSON.stringify((v.errors || []).concat(pv.errors || [])).slice(0, 300)}`);
+ return env;
+}
+
+/**
+ * Set (follower → type/target) to `active` with the notify flags, in one transaction with its event.
+ * → { changed, active, notify_email, notify_push, revision, created_at, at }
+ * opts: notifyEmail/notifyPush (undefined keeps the stored value, else true), reason and actor (deletes),
+ * emit (false for an import: no event), source, at.
+ */
+function setFollow(db, follower, type, target, active, { notifyEmail, notifyPush, reason = null, actor = null, emit = true, source = 'network', at: atIn = null } = {}) {
+ if (!SUBJECT_RE.test(String(follower)) || !SUBJECT_RE.test(String(target))) throw new FollowError(400, 'follows.bad_subject', 'follows are between usr_ subjects');
+ if (!TYPES.includes(type)) throw new FollowError(404, 'follows.unknown_type', `target type must be one of ${TYPES.join(', ')}`);
+ if (follower === target) throw new FollowError(400, 'follows.self', 'you cannot follow yourself');
+ return db.transaction(() => {
+ const prev = db.prepare('SELECT * FROM user_follows WHERE follower_subject = ? AND target_type = ? AND target_id = ?').get(follower, type, target);
+ const email = notifyEmail === undefined ? (prev ? !!prev.notify_email : true) : !!notifyEmail;
+ const push = notifyPush === undefined ? (prev ? !!prev.notify_push : true) : !!notifyPush;
+ const same = prev && !!prev.active === !!active && (!active || (!!prev.notify_email === email && !!prev.notify_push === push));
+ if (same) return { changed: false, active: !!active, notify_email: email, notify_push: push, revision: prev.revision, created_at: prev.created_at, at: prev.updated_at };
+ if (!prev && !active) return { changed: false, active: false, notify_email: email, notify_push: push, revision: 0, created_at: null, at: null };
+ if (active && !(prev && prev.active)) {
+ const n = db.prepare('SELECT COUNT(*) AS n FROM user_follows WHERE follower_subject = ? AND active = 1').get(follower).n;
+ if (n >= MAX_FOLLOWING) throw new FollowError(409, 'follows.limit', `at most ${MAX_FOLLOWING} follows`);
+ }
+ const at = atIn || new Date().toISOString();
+ const revision = prev ? prev.revision + 1 : 1;
+ // A follow that starts again starts its "since" again; flag changes keep it.
+ const createdAt = prev && prev.active && active ? prev.created_at : at;
+ if (prev) {
+ db.prepare(`UPDATE user_follows SET active = ?, notify_email = ?, notify_push = ?, revision = ?, source = ?, created_at = ?, updated_at = ?
+ WHERE follower_subject = ? AND target_type = ? AND target_id = ?`)
+ .run(active ? 1 : 0, email ? 1 : 0, push ? 1 : 0, revision, source, createdAt, at, follower, type, target);
+ } else {
+ db.prepare(`INSERT INTO user_follows (follower_subject, target_type, target_id, active, notify_email, notify_push, revision, source, created_at, updated_at)
+ VALUES (?, ?, ?, 1, ?, ?, 1, ?, ?, ?)`).run(follower, type, target, email ? 1 : 0, push ? 1 : 0, source, at, at);
+ }
+ if (emit) eventRelay.writerFor(db).enqueue(buildEnvelope({ follower, type, target, active, notifyEmail: email, notifyPush: push, revision, at, reason, actor }));
+ return { changed: true, active: !!active, notify_email: email, notify_push: push, revision, created_at: active ? createdAt : null, at };
+ })();
+}
+
+function kick(db) { const o = eventRelay.outboxFor(db); if (o) o.kick(); }
+
+function count(db, type, target) {
+ return db.prepare('SELECT COUNT(*) AS n FROM user_follows WHERE target_type = ? AND target_id = ? AND active = 1').get(type, target).n;
+}
+
+/** network.follow-status-result@1 for (type, target), with the viewer's own follow when given. */
+function status(db, type, target, viewer = null) {
+ const out = { target_type: type, target_id: target, followers: count(db, type, target) };
+ if (viewer && SUBJECT_RE.test(viewer)) {
+ const r = db.prepare('SELECT * FROM user_follows WHERE follower_subject = ? AND target_type = ? AND target_id = ? AND active = 1').get(viewer, type, target);
+ out.following = !!r;
+ if (r) Object.assign(out, { notify_email: !!r.notify_email, notify_push: !!r.notify_push, since: r.created_at });
+ }
+ return out;
+}
+
+const encodeCursor = (r) => Buffer.from(`${r.created_at}|${r.follower_subject}|${r.target_id}`).toString('base64url');
+function decodeCursor(c) {
+ if (!c) return null;
+ const [at, a, b] = Buffer.from(String(c), 'base64url').toString('utf8').split('|');
+ if (!at || !SUBJECT_RE.test(a || '') || !SUBJECT_RE.test(b || '')) throw new FollowError(400, 'follows.bad_cursor', 'cursor is not one this API issued');
+ return { at, a, b };
+}
+const item = (r) => ({ follower: r.follower_subject, target_type: r.target_type, target_id: r.target_id, notify_email: !!r.notify_email, notify_push: !!r.notify_push, created_at: r.created_at });
+
+/** A page of follows (newest first): by follower (what they follow) or by target (who follows it). */
+function list(db, { follower = null, type = null, target = null, cursor = null, limit = 50 } = {}) {
+ const n = Math.min(Math.max(parseInt(limit, 10) || 50, 1), 200);
+ const c = decodeCursor(cursor);
+ const where = ['active = 1'];
+ const args = [];
+ if (follower) { where.push('follower_subject = ?'); args.push(follower); }
+ if (type) { where.push('target_type = ?'); args.push(type); }
+ if (target) { where.push('target_id = ?'); args.push(target); }
+ if (c) { where.push('(created_at < ? OR (created_at = ? AND (follower_subject, target_id) < (?, ?)))'); args.push(c.at, c.at, c.a, c.b); }
+ const rows = db.prepare(`SELECT * FROM user_follows WHERE ${where.join(' AND ')} ORDER BY created_at DESC, follower_subject DESC, target_id DESC LIMIT ?`).all(...args, n + 1);
+ const more = rows.length > n;
+ if (more) rows.pop();
+ return { items: rows.map(item), next_cursor: more ? encodeCursor(rows[rows.length - 1]) : null };
+}
+
+/**
+ * An account's follows go with it (both directions), each announced as a delete. Call inside the
+ * transaction that removes the account (account deletion, WS-B task 7). → the pairs removed
+ */
+function onSubjectRemoved(db, subject) {
+ const rows = db.prepare('SELECT * FROM user_follows WHERE active = 1 AND (follower_subject = ? OR target_id = ?)').all(subject, subject);
+ for (const r of rows) {
+ setFollow(db, r.follower_subject, r.target_type, r.target_id, false, {
+ reason: r.follower_subject === subject ? 'account_removed' : 'target_removed', actor: { type: 'system', id: 'network' },
+ });
+ }
+ return rows.length;
+}
+
+/**
+ * ADR-030 step 2: import follows from another system's rows, mapped to subjects by `subjectOf`.
+ * rows: [{ follower_ref, target_ref, notify_email, notify_push, created_at }] (the refs are that system's ids).
+ * A pair whose side has no subject (or is the same person) is held in follow_import_holds, never dropped.
+ * No events (the source system already has these follows); dryRun changes nothing.
+ * → { imported, unchanged, held: [{ follower_ref, target_ref, reason }] }
+ */
+function importFollows(db, source, rows, subjectOf, { dryRun = false, now = new Date().toISOString() } = {}) {
+ const out = { imported: 0, unchanged: 0, held: [] };
+ const run = () => {
+ for (const r of rows) {
+ const follower = subjectOf(r.follower_ref);
+ const target = subjectOf(r.target_ref);
+ let reason = null;
+ if (!follower) reason = 'follower_has_no_subject';
+ else if (!target) reason = 'target_has_no_subject';
+ else if (follower === target) reason = 'self_follow';
+ if (reason) {
+ out.held.push({ follower_ref: String(r.follower_ref), target_ref: String(r.target_ref), reason });
+ if (!dryRun) db.prepare('INSERT OR REPLACE INTO follow_import_holds (source, follower_ref, target_ref, reason, seen_at) VALUES (?, ?, ?, ?, ?)').run(source, String(r.follower_ref), String(r.target_ref), reason, now);
+ continue;
+ }
+ const prev = db.prepare('SELECT active FROM user_follows WHERE follower_subject = ? AND target_type = ? AND target_id = ?').get(follower, 'channel', target);
+ if (prev && prev.active) { out.unchanged++; continue; }
+ if (!dryRun) {
+ setFollow(db, follower, 'channel', target, true, {
+ notifyEmail: r.notify_email == null ? true : !!r.notify_email, notifyPush: r.notify_push == null ? true : !!r.notify_push,
+ emit: false, source, at: r.created_at ? new Date(r.created_at).toISOString() : now,
+ });
+ }
+ out.imported++;
+ }
+ };
+ if (dryRun) run(); else db.transaction(run)();
+ return out;
+}
+
+/** The signed-in person as a follower, or a FollowError. */
+function meOf(db, user) {
+ if (!user || user.is_anon) throw new FollowError(403, 'follows.guest', 'sign in with an account to follow');
+ const sid = require('./subjects').ensureUserSubject(db, user);
+ if (!SUBJECT_RE.test(String(sid || ''))) throw new FollowError(403, 'follows.guest', 'sign in with an account to follow');
+ return sid;
+}
+
+/** The session user of a request, or null (never refuses: for public reads). */
+function viewerOf(req) {
+ const h = req.headers.authorization || '';
+ const token = h.startsWith('Bearer ') ? h.slice(7) : (req.cookies && req.cookies.ov_token);
+ if (!token) return null;
+ try {
+ const out = require('../auth/session').verifySession(token, { db: req.app.locals.db, publicKey: req.app.locals.publicKey, config: req.app.locals.config });
+ return out && out.user ? out.user : null;
+ } catch { return null; }
+}
+
+function routers({ requireAuth, followsGuard }) {
+ const send = (req, res, fn) => {
+ res.set('Cache-Control', 'private, no-store');
+ try { return fn(req.app.locals.db); } catch (err) {
+ if (err instanceof FollowError) return http.sendProblem(res, err.status, err.code, { detail: err.message, ctx: req.ov });
+ console.error('[Follows]', err.message);
+ return http.sendProblem(res, 500, 'follows.failed', { detail: 'the follow could not be changed', ctx: req.ov });
+ }
+ };
+ const targetOrFail = (db, type, key) => {
+ const u = findTarget(db, type, key);
+ if (!u || u.is_anon || !SUBJECT_RE.test(String(u.subject_id || ''))) throw new FollowError(404, 'follows.unknown_target', 'no such channel');
+ return u;
+ };
+
+ const me = express.Router();
+ me.use(http.middleware());
+ me.get('/', requireAuth, (req, res) => send(req, res, (db) => {
+ const sid = meOf(db, req.user);
+ const type = req.query.type ? String(req.query.type) : null;
+ if (type && !TYPES.includes(type)) throw new FollowError(404, 'follows.unknown_type', `target type must be one of ${TYPES.join(', ')}`);
+ res.json(list(db, { follower: sid, type, cursor: req.query.cursor, limit: req.query.limit }));
+ }));
+ me.put('/:type/:target', requireAuth, express.json({ limit: '4kb' }), (req, res) => send(req, res, (db) => {
+ const sid = meOf(db, req.user);
+ const t = targetOrFail(db, req.params.type, req.params.target);
+ const b = req.body && typeof req.body === 'object' ? req.body : {};
+ const out = setFollow(db, sid, req.params.type, t.subject_id, true, {
+ notifyEmail: typeof b.notify_email === 'boolean' ? b.notify_email : undefined, notifyPush: typeof b.notify_push === 'boolean' ? b.notify_push : undefined,
+ });
+ if (out.changed) kick(db);
+ res.status(out.changed && out.revision === 1 ? 201 : 200).json(status(db, req.params.type, t.subject_id, sid));
+ }));
+ me.delete('/:type/:target', requireAuth, (req, res) => send(req, res, (db) => {
+ const sid = meOf(db, req.user);
+ const key = String(req.params.target || '');
+ // A subject can be unfollowed even if its account is gone since.
+ const t = findTarget(db, req.params.type, key) || (SUBJECT_RE.test(key) ? { subject_id: key } : null);
+ if (!t || !t.subject_id) throw new FollowError(404, 'follows.unknown_target', 'no such channel');
+ const out = setFollow(db, sid, req.params.type, t.subject_id, false);
+ if (out.changed) kick(db);
+ res.json(status(db, req.params.type, t.subject_id, sid));
+ }));
+
+ const pub = express.Router();
+ pub.use(http.middleware());
+ pub.get('/:type/:target', (req, res) => {
+ try {
+ const db = req.app.locals.db;
+ const t = targetOrFail(db, req.params.type, req.params.target);
+ const viewer = viewerOf(req);
+ const body = status(db, req.params.type, t.subject_id, viewer && !viewer.is_anon ? viewer.subject_id : null);
+ res.set('Cache-Control', viewer ? 'private, no-store' : 'public, max-age=30').set('Vary', 'Authorization, Cookie').json(body);
+ } catch (err) {
+ if (err instanceof FollowError) return http.sendProblem(res, err.status, err.code, { detail: err.message, ctx: req.ov });
+ throw err;
+ }
+ });
+ // Who follows a target: its owner (signed in), or a service holding network.follows.read.
+ pub.get('/:type/:target/followers', (req, res, next) => {
+ const db = req.app.locals.db;
+ let t;
+ try { t = targetOrFail(db, req.params.type, req.params.target); } catch (err) { return http.sendProblem(res, err.status, err.code, { detail: err.message, ctx: req.ov }); }
+ const answer = () => send(req, res, () => res.json(list(db, { type: req.params.type, target: t.subject_id, cursor: req.query.cursor, limit: req.query.limit })));
+ const viewer = viewerOf(req);
+ if (viewer && viewer.subject_id === t.subject_id) return answer();
+ if (viewer) return http.sendProblem(res, 403, 'follows.not_yours', { detail: 'only the channel owner sees who follows it', ctx: req.ov });
+ return followsGuard(req, res, answer);
+ });
+
+ return { me, pub };
+}
+
+module.exports = { TYPES, FollowError, ensureSchema, setFollow, status, list, count, findTarget, buildEnvelope, onSubjectRemoved, importFollows, routers, kick };
@@ -121,6 +121,8 @@ const DEFAULT_GRANTS = [
@@ -121,6 +121,8 @@ const DEFAULT_GRANTS = [
['coupons', 'events.event.publish', 'openvibe.events', []],
['coupons', 'events.event.publish', 'openvibe.events', []],
['coupons', 'sources.item.read', 'openvibe.sources', []],
['coupons', 'sources.item.read', 'openvibe.sources', []],
['host', 'events.event.publish', 'openvibe.events', []],
['host', 'events.event.publish', 'openvibe.events', []],
+ // ADR-030 (WS-E task 4): Live rebuilds its follows projection from Network's graph.
+ ['live', 'network.follows.read', SELF_AUDIENCE, []],
// WS-H task 11: Host relays the alerts firing on the production host (ovhost alerts relay).
// WS-H task 11: Host relays the alerts firing on the production host (ovhost alerts relay).
['host', 'network.operator.alert', SELF_AUDIENCE, []],
['host', 'network.operator.alert', SELF_AUDIENCE, []],
// WS-L task 4: OpenVibe.Host's scheduled Tools job proof (openvibe-toolsjob.timer, OpenVibe.Examples
// WS-L task 4: OpenVibe.Host's scheduled Tools job proof (openvibe-toolsjob.timer, OpenVibe.Examples
@@ -408,6 +408,13 @@ app.get('/internal/integrations/github-token', require('./identity/principals').
@@ -408,6 +408,13 @@ app.get('/internal/integrations/github-token', require('./identity/principals').
// (network.blocks.read, service token only). Every change is network.block.changed (server/identity/blocks.js).
// (network.blocks.read, service token only). Every change is network.block.changed (server/identity/blocks.js).
app.use('/api/v1/me/blocks', rateLimit({ windowMs: 60_000, max: 60 }), require('./identity/blocks').userRouter(requireAuth));
app.use('/api/v1/me/blocks', rateLimit({ windowMs: 60_000, max: 60 }), require('./identity/blocks').userRouter(requireAuth));
app.get('/internal/blocks', require('./identity/principals').guard('network.blocks.read', { legacy: false }), require('./identity/blocks').internalHandler(db));
app.get('/internal/blocks', require('./identity/principals').guard('network.blocks.read', { legacy: false }), require('./identity/blocks').internalHandler(db));
+// The follow graph (WS-E task 4, ADR-030): public counts, a person's own follows, and who follows a target
+// (its owner, or network.follows.read, service token only). Every change is network.follow.* (server/identity/follows.js).
+{
+ const followRouters = require('./identity/follows').routers({ requireAuth, followsGuard: require('./identity/principals').guard('network.follows.read', { legacy: false }) });
+ app.use('/api/v1/me/follows', rateLimit({ windowMs: 60_000, max: 120 }), followRouters.me);
+ app.use('/api/v1/follows', rateLimit({ windowMs: 60_000, max: 300 }), followRouters.pub);
+}
// Realtime tickets (WS-E task 3, WS-F task 1; ADR-005 amendment 2): the notification badge on any site opens
// Realtime tickets (WS-E task 3, WS-F task 1; ADR-005 amendment 2): the notification badge on any site opens
// OpenVibe.Events' /realtime/stream as the signed-in person with a two-minute, single-use ticket
// OpenVibe.Events' /realtime/stream as the signed-in person with a two-minute, single-use ticket
// (server/auth/realtime-ticket.js). REALTIME_TICKETS=off answers 503 and every badge stays on polling.
// (server/auth/realtime-ticket.js). REALTIME_TICKETS=off answers 503 and every badge stays on polling.
new file mode 100644
new file mode 100644
@@ -0,0 +1,167 @@
@@ -0,0 +1,167 @@
+'use strict';
+// The follow graph (roadmap WS-E task 4, ADR-030; Contracts 0.65.0): follow and unfollow through
+// /api/v1/me/follows (idempotent, never oneself, never a guest), each change writes network.follow.created /
+// .deleted with a growing per-pair revision, counts are public and lists are not (the target's owner, or a
+// service with network.follows.read), and scripts/follows-backfill.js imports Live's follows (held when a
+// side has no subject), then reconciles.
+// node test/follows.test.js
+const assert = require('assert');
+const crypto = require('crypto');
+const fs = require('fs');
+const os = require('os');
+const path = require('path');
+const http = require('http');
+const express = require('express');
+const Database = require('better-sqlite3');
+const { validate } = require('openvibe-contracts');
+const { initDb } = require('../server/db/database');
+
+const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-follows-'));
+const dbFile = path.join(dir, 'network.db');
+const log = console.log; console.log = () => {};
+const db = initDb(dbFile);
+console.log = log;
+require('../server/identity/principals').ensureSchema(db);
+for (const c of ['live', 'tools']) db.prepare('UPDATE oauth_clients SET client_secret = ? WHERE client_id = ?').run(`${c}-secret`, c);
+
+const ANN = 'usr_01JAB2C3D4E5F6G7H8J9K0MNPA';
+const BOB = 'usr_01JAB2C3D4E5F6G7H8J9K0MNPB';
+const CAT = 'usr_01JAB2C3D4E5F6G7H8J9K0MNPC';
+db.prepare(`INSERT INTO users (id, username, display_name, password_hash, subject_id, role) VALUES
+ (1, 'ann', 'Ann', 'x', ?, 'user'), (2, 'bob', 'Bob', 'x', ?, 'streamer'), (3, 'cat', NULL, 'x', ?, 'user')`).run(ANN, BOB, CAT);
+db.prepare("INSERT INTO users (id, username, password_hash, is_anon) VALUES (5, 'anon1234', 'x', 1)").run();
+
+const keys = crypto.generateKeyPairSync('rsa', { modulusLength: 2048, publicKeyEncoding: { type: 'spki', format: 'pem' }, privateKeyEncoding: { type: 'pkcs8', format: 'pem' } });
+const config = { internalKey: 'legacy-key', jwt: { issuer: 'https://openvibe.network', accessTokenExpiry: '1h' } };
+const { signToken } = require('../server/auth/routes');
+const requireAuth = require('../server/auth/session').makeRequireAuth(() => ({ db, publicKey: keys.publicKey, config }), signToken);
+const follows = require('../server/identity/follows');
+const principals = require('../server/identity/principals');
+const app = express();
+app.use(require('cookie-parser')());
+app.use(express.urlencoded({ extended: true }));
+Object.assign(app.locals, { db, config, privateKey: keys.privateKey, publicKey: keys.publicKey });
+app.use('/oauth', require('../server/auth/oauth-routes'));
+const routers = follows.routers({ requireAuth, followsGuard: principals.guard('network.follows.read', { legacy: false }) });
+app.use('/api/v1/me/follows', routers.me);
+app.use('/api/v1/follows', routers.pub);
+const server = http.createServer(app);
+
+const events = () => db.prepare('SELECT envelope FROM network_event_outbox ORDER BY id').all().map((r) => JSON.parse(r.envelope)).filter((e) => e.event_type.startsWith('network.follow.'));
+const valid = (e) => {
+ assert.ok(validate('events.event-envelope@1', e).valid);
+ const pv = validate(`${e.event_type}@1`, e.payload);
+ assert.ok(pv.valid, JSON.stringify(pv.errors));
+ assert.deepStrictEqual([e.source, e.visibility, e.subject], ['network', 'subject', { type: 'user', id: e.payload.follower }]);
+};
+
+(async () => {
+ await new Promise((r) => server.listen(0, '127.0.0.1', r));
+ const base = `http://127.0.0.1:${server.address().port}`;
+ const tok = (id) => signToken(db.prepare('SELECT * FROM users WHERE id = ?').get(id), keys.privateKey, config);
+ const call = (method, p, { body, headers = {} } = {}) => fetch(base + p, { method, headers: { 'content-type': 'application/json', ...headers }, body: body ? JSON.stringify(body) : undefined })
+ .then(async (r) => ({ status: r.status, cache: r.headers.get('cache-control'), body: await r.json().catch(() => null) }));
+ const as = (id) => ({ authorization: `Bearer ${tok(id)}` });
+ const svc = async (client) => (await (await fetch(`${base}/oauth/token`, { method: 'POST', body: new URLSearchParams({ grant_type: 'client_credentials', client_id: client, client_secret: `${client}-secret`, audience: 'openvibe.network' }) })).json()).access_token;
+ try {
+ // ── Follow, idempotently; events ──
+ assert.strictEqual((await call('PUT', '/api/v1/me/follows/channel/bob')).status, 401, 'signed in only');
+ let r = await call('PUT', '/api/v1/me/follows/channel/bob', { headers: as(1) });
+ assert.strictEqual(r.status, 201);
+ assert.ok(validate('network.follow-status-result@1', r.body).valid, JSON.stringify(r.body));
+ assert.deepStrictEqual([r.body.target_id, r.body.followers, r.body.following, r.body.notify_email, r.body.notify_push], [BOB, 1, true, true, true]);
+ r = await call('PUT', '/api/v1/me/follows/channel/Bob', { headers: as(1) });
+ assert.deepStrictEqual([r.status, r.body.followers], [200, 1], 'following again changes nothing (and names are case-insensitive)');
+ assert.strictEqual(events().length, 1, 'and announces nothing');
+ r = await call('PUT', `/api/v1/me/follows/channel/${BOB}`, { headers: as(1), body: { notify_email: false } });
+ assert.deepStrictEqual([r.status, r.body.notify_email, r.body.notify_push], [200, false, true], 'new flags: a change');
+ let ev = events();
+ assert.deepStrictEqual(ev.map((e) => [e.event_type, e.payload.revision, e.payload.notify_email]), [['network.follow.created', 1, true], ['network.follow.created', 2, false]]);
+ ev.forEach(valid);
+ assert.strictEqual((await call('PUT', '/api/v1/me/follows/channel/ann', { headers: as(1) })).status, 400, 'never oneself');
+ assert.strictEqual((await call('PUT', '/api/v1/me/follows/channel/anon1234', { headers: as(1) })).status, 404, 'a guest is not a channel');
+ assert.strictEqual((await call('PUT', '/api/v1/me/follows/channel/nobody', { headers: as(1) })).status, 404);
+ assert.strictEqual((await call('PUT', '/api/v1/me/follows/space/bob', { headers: as(1) })).status, 404, 'unknown target type');
+ assert.strictEqual((await call('PUT', '/api/v1/me/follows/channel/bob', { headers: as(5) })).status, 403, 'guests do not follow');
+ await call('PUT', '/api/v1/me/follows/channel/bob', { headers: as(3) });
+
+ // ── Public count; the viewer's own state; lists are not public ──
+ r = await call('GET', '/api/v1/follows/channel/bob');
+ assert.deepStrictEqual([r.status, r.body], [200, { target_type: 'channel', target_id: BOB, followers: 2 }]);
+ assert.match(r.cache, /public/);
+ r = await call('GET', '/api/v1/follows/channel/bob', { headers: as(1) });
+ assert.deepStrictEqual([r.body.following, r.body.notify_email], [true, false]);
+ assert.match(r.cache, /private/);
+ r = await call('GET', '/api/v1/me/follows', { headers: as(1) });
+ assert.ok(validate('network.follow-list-result@1', r.body).valid);
+ assert.deepStrictEqual(r.body.items.map((i) => i.target_id), [BOB]);
+ assert.ok([401, 403].includes((await call('GET', '/api/v1/follows/channel/bob/followers')).status), 'no token: not the owner, not a service');
+ assert.strictEqual((await call('GET', '/api/v1/follows/channel/bob/followers', { headers: as(1) })).status, 403, 'a follower is not the owner');
+ r = await call('GET', '/api/v1/follows/channel/bob/followers', { headers: as(2) });
+ assert.deepStrictEqual([r.status, r.body.items.map((i) => i.follower).sort()], [200, [ANN, CAT].sort()], 'the owner sees who follows');
+ r = await call('GET', '/api/v1/follows/channel/bob/followers?limit=1', { headers: { authorization: `Bearer ${await svc('live')}` } });
+ assert.deepStrictEqual([r.status, r.body.items.length, typeof r.body.next_cursor], [200, 1, 'string'], 'a service with network.follows.read, paged');
+ const page2 = await call('GET', `/api/v1/follows/channel/bob/followers?limit=1&cursor=${r.body.next_cursor}`, { headers: { authorization: `Bearer ${await svc('live')}` } });
+ assert.strictEqual(page2.body.items.length, 1);
+ assert.notStrictEqual(page2.body.items[0].follower, r.body.items[0].follower, 'the cursor moves on');
+ assert.strictEqual(page2.body.next_cursor, null);
+ assert.strictEqual((await call('GET', '/api/v1/follows/channel/bob/followers', { headers: { authorization: `Bearer ${await svc('tools')}` } })).status, 403, 'a service without the capability');
+
+ // ── Unfollow, idempotently ──
+ r = await call('DELETE', '/api/v1/me/follows/channel/bob', { headers: as(1) });
+ assert.deepStrictEqual([r.status, r.body.following, r.body.followers], [200, false, 1]);
+ r = await call('DELETE', '/api/v1/me/follows/channel/bob', { headers: as(1) });
+ assert.strictEqual(r.status, 200);
+ ev = events().filter((e) => e.event_type === 'network.follow.deleted');
+ assert.deepStrictEqual(ev.map((e) => [e.payload.revision, e.payload.reason]), [[3, 'unfollowed']], 'one delete, revision 3');
+ ev.forEach(valid);
+ r = await call('PUT', '/api/v1/me/follows/channel/bob', { headers: as(1) });
+ assert.strictEqual(r.status, 200, 'a follow again (the row exists; revision 4)');
+ assert.strictEqual(events().pop().payload.revision, 4);
+
+ // ── Account removal ──
+ assert.strictEqual(db.transaction(() => follows.onSubjectRemoved(db, CAT))(), 1);
+ assert.deepStrictEqual(events().pop().payload.reason, 'account_removed');
+
+ // ── Live backfill and reconciliation (scripts/follows-backfill.js) ──
+ const liveFile = path.join(dir, 'live.db');
+ const live = new Database(liveFile);
+ live.exec(`CREATE TABLE follows (id INTEGER PRIMARY KEY AUTOINCREMENT, follower_id INTEGER NOT NULL, streamer_id INTEGER NOT NULL, email_notify INTEGER DEFAULT 0, push_notify INTEGER DEFAULT 0, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, UNIQUE(follower_id, streamer_id));
+ CREATE TABLE linked_accounts (id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER NOT NULL, service TEXT NOT NULL, service_user_id TEXT NOT NULL, subject_id TEXT);`);
+ live.prepare("INSERT INTO linked_accounts (user_id, service, service_user_id, subject_id) VALUES (11, 'network', '1', ?), (12, 'network', '2', ?), (13, 'network', '3', ?)").run(ANN, BOB, CAT);
+ live.prepare("INSERT INTO follows (follower_id, streamer_id, email_notify, push_notify, created_at) VALUES (11, 12, 1, 0, '2026-09-01 10:00:00'), (13, 12, 0, 1, '2026-09-02 10:00:00'), (14, 12, 0, 0, '2026-09-03 10:00:00'), (13, 11, 0, 0, '2026-09-04 10:00:00')").run();
+ live.close();
+ const backfill = require('../scripts/follows-backfill');
+ const lines = [];
+ const out = (s) => lines.push(s);
+ db.close();
+ let code = await backfill.main(['--live-db', liveFile, '--db', dbFile], out);
+ assert.strictEqual(code, 0);
+ assert.match(lines.join('\n'), /live follows 4: import 2, already on Network 1, held 1/);
+ assert.match(lines.join('\n'), /held \(follower_has_no_subject\): 1/);
+ code = await backfill.main(['--live-db', liveFile, '--db', dbFile, '--apply'], out);
+ assert.strictEqual(code, 2, 'refuses --apply without --backup');
+ code = await backfill.main(['--live-db', liveFile, '--db', dbFile, '--apply', '--backup', path.join(dir, 'pre-follows.db')], out);
+ assert.strictEqual(code, 0, lines.join('\n'));
+ assert.ok(fs.existsSync(path.join(dir, 'pre-follows.db')), 'backup taken first');
+ assert.match(lines.join('\n'), /imported {3}2; already there 1; held 1/);
+ assert.match(lines.join('\n'), /RECONCILED: counts and pairs match/);
+ code = await backfill.main(['--live-db', liveFile, '--db', dbFile, '--reconcile'], out);
+ assert.strictEqual(code, 0);
+ const db2 = new Database(dbFile);
+ assert.deepStrictEqual(db2.prepare("SELECT reason FROM follow_import_holds WHERE source = 'live'").all().map((x) => x.reason), ['follower_has_no_subject'], 'held, never dropped');
+ const cat = db2.prepare('SELECT notify_email, notify_push, source, created_at FROM user_follows WHERE follower_subject = ? AND target_id = ?').get(CAT, BOB);
+ assert.deepStrictEqual(cat, { notify_email: 0, notify_push: 1, source: 'live', created_at: '2026-09-02T10:00:00.000Z' }, 'flags and date carried over');
+ const before = db2.prepare('SELECT COUNT(*) AS n FROM network_event_outbox').get().n;
+ db2.close();
+ code = await backfill.main(['--live-db', liveFile, '--db', dbFile, '--apply', '--backup', path.join(dir, 'pre-follows-2.db')], out);
+ assert.strictEqual(code, 0, 'a second run is a no-op');
+ const db3 = new Database(dbFile, { readonly: true });
+ assert.strictEqual(db3.prepare('SELECT COUNT(*) AS n FROM network_event_outbox').get().n, before, 'the import sends no events');
+ db3.close();
+ } finally {
+ server.close();
+ fs.rmSync(dir, { recursive: true, force: true });
+ }
+ console.log('follows: all checks passed');
+})().catch((e) => { console.error(e); process.exit(1); });
@@ -50,7 +50,7 @@ const server = http.createServer(app);
@@ -50,7 +50,7 @@ const server = http.createServer(app);
const claims = serviceAuth.verifyServiceToken(t.body.access_token, { publicKey: keys.publicKey, issuer: ISSUER, audience: 'openvibe.network' });
const claims = serviceAuth.verifyServiceToken(t.body.access_token, { publicKey: keys.publicKey, issuer: ISSUER, audience: 'openvibe.network' });
assert.ok(claims.ok, claims.reason);
assert.ok(claims.ok, claims.reason);
assert.strictEqual(claims.claims.sub, 'svc:live');
assert.strictEqual(claims.claims.sub, 'svc:live');
- assert.deepStrictEqual(claims.claims.cap, ['identity.subject.resolve', 'network.coins.credit', 'network.coins.debit', 'network.modules.read', 'network.modules.write', 'network.notifications.push']);
+ assert.deepStrictEqual(claims.claims.cap, ['identity.subject.resolve', 'network.coins.credit', 'network.coins.debit', 'network.follows.read', 'network.modules.read', 'network.modules.write', 'network.notifications.push']);
assert.ok(validate('identity.service-token-claims@1', claims.claims).valid);
assert.ok(validate('identity.service-token-claims@1', claims.claims).valid);
assert.ok(claims.claims.exp - claims.claims.iat <= 300, 'short-lived');
assert.ok(claims.claims.exp - claims.claims.iat <= 300, 'short-lived');
const full = t.body.access_token;
const full = t.body.access_token;