openvibers/openvibe.network Commits

Platform plan T2 "Follows", job 1 of 2: Network resolves go-live… (#6)

OP OpenVibers committed 2 Oct 2026 · e6b426cb99 · 18 files, +488 −330

local checks ✓ 3/3 · CI ✓ 3/3 · review SHIP by DeepSeek harness (QUALITY=high) · 1 fix round

https://github.com/OpenVibers/OpenVibe.Network/pull/6

Changes

ddiff --git a/.env.example b/.env.example
iindex 4f5830af..7994df75 100644
---- a/.env.example
++++ b/.env.example
@@@ -72,6 +72,8 @@ JWT_PUBLIC_KEY=data/keys/public.pem
# Connected services (internal, not proxied through Cloudflare)
OV_LIVE_INTERNAL_URL=http://127.0.0.1:3000
++# Enable only after Live writes follows to Network and the final Live snapshot is reconciled.
++#NETWORK_GO_LIVE_FOLLOWS_READY=1
OV_TOOLS_INTERNAL_URL=http://127.0.0.1:4001
OV_GAMES_INTERNAL_URL=http://127.0.0.1:8000
OV_MEDIA_INTERNAL_URL=http://127.0.0.1:4100
ddiff --git a/README.md b/README.md
iindex 3ebe5f11..de284927 100644
---- a/README.md
++++ b/README.md
@@@ -270,17 +270,17 @@ their per-category email choice.
| `trade.alert.triggered` | Trade (`source: trade`) | `subject.id` (`subject.type: user`) | `TRADE_ALERT`, `service`, high |
| `live.stream.started` | Live (`source: live`) | every follower of `payload.channel.subject`, plus everyone with `stream_live_all` on | `STREAM_LIVE`, `stream`, high; the Discord live alert after commit |
--**Go-lives.** Live owns the follow graph, so Network reads it when the event arrives:
--`GET <OV_LIVE_INTERNAL_URL>/internal/followers?stream_id=<id>&limit=5000[&after=<cursor>]` with a
--self-signed service token (`sub svc:network`, `aud openvibe.live`, `cap live.follower.read`), answered
--with `{ channel: { subject }, followers: [{ subject, network_user_id }], next }`
--(`server/notifications/live-followers.js`). Followers resolve by `usr_` subject first, then by Network
--user id; the streamer never hears about themself; the `stream` category's mute and email choice apply.
--If Live cannot answer, the delivery gets 503 and nothing is recorded, so Events retries. A start more
++**Go-lives.** The followers are Network's own follow graph (see "Follow graph"): the active
++`user_follows` rows for `('channel', payload.channel.subject)`, read in the same transaction as the
++notifications. Live is never asked. A channel whose subject has no Network account is `ignored:channel`;
++a follower whose subject has no account is counted as `unresolved`; the streamer never hears about
++themself; the `stream` category's mute and email choice apply. A start more
than 30 minutes old (a replay, or Events catching up) announces nothing. The per-streamer window (one
announcement an hour, eight a day; `stream_live_cooldown_min`, `stream_live_daily_cap`) is shared with
Live's direct `POST /internal/events/stream-live` (`server/notifications/stream-live.js`), so while Live
still makes that call, whichever arrives first announces and the other is skipped.
++Until `NETWORK_GO_LIVE_FOLLOWS_READY=1` is set on Network, Events go-lives are acknowledged as
++`ignored:follows_cutover_pending` and Live's direct POST remains the delivery path.
**Project usage.** `tools.usage.recorded` and `events.usage.recorded` (hourly rollups of a developer
project's use, `common.usage-recorded@1`) are kept per project and day (`server/developer/usage.js`,
@@@ -304,8 +304,7 @@ The route is inert until the operator does both of these:
2. Create the subscriptions in Events (consumer `network`, which Events takes from the token's
`svc:network`; endpoint `http://127.0.0.1:4000/internal/events`; topic patterns
`deals.watch.matched`, `trade.alert.triggered` and `live.stream.started`; secret = the first
-- `NETWORK_EVENTS_SECRET`). Subscribe `live.stream.started` only once Live serves
-- `GET /internal/followers`; on a host that already has the other two, add it alone with
++ `NETWORK_EVENTS_SECRET`). On a host that already has the other two, add `live.stream.started` alone with
`--topic live.stream.started` (and the project usage rollups with
`--topic tools.usage.recorded --topic events.usage.recorded`):
@@@ -488,8 +487,27 @@ A target is a Live channel today (`channel`, named by its owner's subject), and
- **Products writing for a person** (ADR-030 step 4): `PUT|DELETE /internal/follows/:type/:target` needs
`network.follows.write`, a service token and `network.follow-write-request@1`. It is granted to Live, whose
buttons use it.
--- **Go-live source:** with `FOLLOWS_AUTHORITY=network`, go-live notifications read the followers here
-- instead of asking Live. Unset it to roll back.
++- **Go-live source:** go-live notifications read the followers here, and only here (plan T2); there is no
++ switch back to Live.
++- **Backfill from Live:** follows made on Live while Live's own `FOLLOWS_AUTHORITY` was unset never reached
++ Network. `npm run follows-import -- --live-db /opt/openvibe.live/data/live.db` reads Live's `follows` and
++ `linked_accounts` read-only and prints what it would import (dry run); add `--apply` to import in one
++ transaction (no events, no notifications). A pair whose side has no subject is held in
++ `follow_import_holds`; a pair Network already has, followed or unfollowed, is left alone. Safe to re-run;
++ a later run imports newly mapped pairs and clears their holds. Both scripts use `DATABASE_URL` and never
++ migrate or seed.
++- **Cutover order:** Keep Live's direct `POST /internal/events/stream-live` enabled and leave
++ `NETWORK_GO_LIVE_FOLLOWS_READY` unset. Pause Live follow writes, then run
++ `npm run follows-import -- --live-db /opt/openvibe.live/data/live.db --reconcile` to preview the
++ final Live snapshot and rerun with `--reconcile --apply`. Reconciliation deactivates Live-sourced
++ active Network pairs missing from the snapshot, without emitting follow events or notifications;
++ it preserves Network-sourced pairs and earlier Network unfollows. Check `npm run follows-preflight`
++ and resolve holds before proceeding. Set Live's `FOLLOWS_AUTHORITY=network` and restart Live while
++ writes are paused; verify a follow and unfollow reach Network, then resume Live writes. Only then
++ set `NETWORK_GO_LIVE_FOLLOWS_READY=1` in Network's environment and restart Network. Keep Live's
++ direct call enabled until Network's `live.stream.started` outcome logs show delivery.
++- **Preflight:** `npm run follows-preflight` prints, read-only and counts only, the active follows (with
++ their channels and followers), the unfollowed rows and the unresolved `follow_import_holds` by reason.
## Multi-Account Switching
ddiff --git a/package.json b/package.json
iindex 366fd38f..4ef910d5 100644
---- a/package.json
++++ b/package.json
@@@ -11,7 +11,9 @@
"grant-admin": "node server/grant-admin.js",
"test:pg": "NETWORK_TEST_STORE=pg node test/run.js",
"migrate-to-postgres": "node scripts/migrate-to-postgres.js",
-- "pg-preflight": "node scripts/pg-preflight-collisions.js"
++ "pg-preflight": "node scripts/pg-preflight-collisions.js",
++ "follows-import": "node scripts/follows-import.js",
++ "follows-preflight": "node scripts/follows-preflight.js"
},
"license": "MIT",
"dependencies": {
ddiff --git a/scripts/follows-import.js b/scripts/follows-import.js
nnew file mode 100644
iindex 00000000..17bd5db0
---- /dev/null
++++ b/scripts/follows-import.js
@@@ -0,0 +1,107 @@
++#!/usr/bin/env node
++/**
++ * ADR-030 backfill (plan T2 "Follows"): Live's follows into Network's user_follows, the graph go-live
++ * notifications read (server/notifications/events-consumer.js). Follows made on Live while Live's own
++ * FOLLOWS_AUTHORITY was unset never reached Network. Keep Live's direct go-live delivery active,
++ * pause Live follow writes, then do the final reconciliation before enabling Network delivery.
++ *
++ * npm run follows-import -- --live-db /opt/openvibe.live/data/live.db # dry run: counts only
++ * npm run follows-import -- --live-db /opt/openvibe.live/data/live.db --apply # import, one transaction
++ * npm run follows-import -- --live-db /opt/openvibe.live/data/live.db --reconcile --apply
++ *
++ * Live's side is read only (a readonly SQLite connection). Live user ids map to subjects through Live's
++ * linked_accounts (service 'network', subject_id), the mapping Live's own reads use. The database written is
++ * the service's own PostgreSQL (DATABASE_URL), opened without migrating or seeding. Each pair goes through importFollows()
++ * (server/identity/follows.js): no events and no notifications (Live already has these follows); a pair whose
++ * side has no subject is kept in follow_import_holds, never dropped; a pair Network already has, followed or
++ * unfollowed since, is left alone. --reconcile also deactivates Live-sourced active pairs missing from
++ * Live's snapshot; use it only with Live follow writes paused. Prints counts only.
++ */
++'use strict';
++const fs = require('fs');
++const path = require('path');
++const follows = require('../server/identity/follows');
++
++const USAGE = 'usage: npm run follows-import -- --live-db <live.db> [--reconcile] [--apply]';
++const SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
++
++function parseArgs(argv) {
++ const out = {};
++ for (let i = 0; i < argv.length; i++) {
++ const a = argv[i];
++ if (a === '--apply') out.apply = true;
++ else if (a === '--reconcile') out.reconcile = true;
++ else if (a === '--help' || a === '-h') out.help = true;
++ else if (a === '--live-db') {
++ const v = argv[++i];
++ if (v === undefined || v.startsWith('--')) throw new Error('--live-db needs a value');
++ out.liveDb = v;
++ } else throw new Error(`unknown argument ${a}`);
++ }
++ return out;
++}
++
++/** A SQLite file: better-sqlite3, else Node's built-in node:sqlite (when the native module is not built for this Node). */
++function openSqlite(file, { readonly = true } = {}) {
++ try {
++ const Database = require('better-sqlite3');
++ return new Database(file, { readonly, fileMustExist: readonly });
++ } catch (e) {
++ if (!/better-sqlite3|NODE_MODULE_VERSION|Cannot find module/.test(e.message)) throw e;
++ const { DatabaseSync } = require('node:sqlite');
++ return new DatabaseSync(file, { readOnly: readonly });
++ }
++}
++
++/** Live's follows and its user id → subject map (readonly). → { rows, subjectOf } */
++function readLive(file) {
++ const live = openSqlite(file);
++ 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: liveTime(r.created_at) }));
++ 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(); }
++}
++
++/** SQLite's CURRENT_TIMESTAMP ('2026-01-02 03:04:05', UTC) as ISO 8601; anything unreadable → null (now). */
++function liveTime(v) {
++ if (!v) return null;
++ const s = String(v).replace(' ', 'T');
++ const t = Date.parse(/Z|[+-]\d\d:?\d\d$/.test(s) ? s : `${s}Z`);
++ return Number.isFinite(t) ? new Date(t).toISOString() : null;
++}
++
++/** Open the service database: the injected handle (tests), else DATABASE_URL as the service uses it (no migration, no seeding). */
++async function openDb(injected) {
++ if (injected) return injected;
++ if (!process.env.DATABASE_URL) throw new Error('DATABASE_URL is not set: the follows live in the service database');
++ const { createDb } = require('openvibe-sdk/db');
++ return createDb({ url: process.env.DATABASE_URL, service: 'network-follows-import', max: 1 });
++}
++
++async function main(argv, { db: injected, log = console.log } = {}) {
++ let args;
++ try { args = parseArgs(argv); } catch (e) { log(`${e.message}\n${USAGE}`); return 2; }
++ if (args.help) { log(USAGE); return 0; }
++ if (!args.liveDb) { log(USAGE); return 2; }
++ const file = path.resolve(process.cwd(), args.liveDb);
++ if (!fs.existsSync(file)) { log(`error: ${args.liveDb} does not exist`); return 2; }
++ let live;
++ try { live = readLive(file); } catch (e) { log(`error: cannot read Live's follows: ${e.message}`); return 2; }
++ const db = await openDb(injected);
++ const out = await follows.importFollows(db, 'live', live.rows, live.subjectOf, { dryRun: !args.apply, reconcile: !!args.reconcile });
++ const held = {};
++ for (const h of out.held) held[h.reason] = (held[h.reason] || 0) + 1;
++ const why = Object.entries(held).map(([k, n]) => `${k} ${n}`).join(', ') || 'none';
++ log(`live rows ${live.rows.length}; ${args.apply ? 'imported' : 'to import'} ${out.imported}; already on Network ${out.unchanged}; ${args.apply ? 'deactivated' : 'to deactivate'} ${out.deactivated}; held: ${why}`);
++ if (!args.apply) log(out.imported || out.deactivated || out.held.length ? 'dry run: nothing changed. Re-run with --apply to make these changes.' : 'nothing to import or deactivate.');
++ return 0;
++}
++
++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, liveTime, openSqlite };
ddiff --git a/scripts/follows-preflight.js b/scripts/follows-preflight.js
nnew file mode 100644
iindex 00000000..4ed0020e
---- /dev/null
++++ b/scripts/follows-preflight.js
@@@ -0,0 +1,51 @@
++#!/usr/bin/env node
++/**
++ * Read-only preflight for go-live notifications from Network's own follow graph (plan T2 "Follows"): how many
++ * follows Network holds and how many imported pairs are still waiting for a subject. Counts only: no subjects,
++ * usernames or Live ids are printed. Changes nothing.
++ *
++ * npm run follows-preflight
++ *
++ * The database is the service's own PostgreSQL (DATABASE_URL), opened without migrating or seeding. A pair held in
++ * follow_import_holds is resolved by running npm run follows-import again once both sides have a subject.
++ */
++'use strict';
++
++/** → { active, channels, followers, inactive, holds: { total, by_reason: { reason: n } } } */
++async function counts(db) {
++ const f = await db.prepare(`SELECT
++ COUNT(*) FILTER (WHERE active = 1) AS active,
++ COUNT(DISTINCT target_id) FILTER (WHERE active = 1) AS channels,
++ COUNT(DISTINCT follower_subject) FILTER (WHERE active = 1) AS followers,
++ COUNT(*) FILTER (WHERE active <> 1) AS inactive
++ FROM user_follows WHERE target_type = 'channel'`).get();
++ const holds = await db.prepare('SELECT source, reason, COUNT(*) AS n FROM follow_import_holds GROUP BY source, reason ORDER BY source, reason').all();
++ const byReason = {};
++ for (const h of holds) byReason[`${h.source}:${h.reason}`] = Number(h.n);
++ return {
++ active: Number(f.active), channels: Number(f.channels), followers: Number(f.followers), inactive: Number(f.inactive),
++ holds: { total: Object.values(byReason).reduce((a, n) => a + n, 0), by_reason: byReason },
++ };
++}
++
++/** Open the service database: the injected handle (tests), else DATABASE_URL as the service uses it (no migration, no seeding). */
++async function openDb(injected) {
++ if (injected) return injected;
++ if (!process.env.DATABASE_URL) throw new Error('DATABASE_URL is not set: the follows live in the service database');
++ const { createDb } = require('openvibe-sdk/db');
++ return createDb({ url: process.env.DATABASE_URL, service: 'network-follows-preflight', max: 1 });
++}
++
++async function main(argv, { db: injected, log = console.log } = {}) {
++ if (argv.includes('--help') || argv.includes('-h')) { log('usage: npm run follows-preflight'); return 0; }
++ if (argv.length) { log(`unknown argument ${argv[0]}\nusage: npm run follows-preflight`); return 2; }
++ const db = await openDb(injected);
++ const c = await counts(db);
++ const why = Object.entries(c.holds.by_reason).map(([k, n]) => `${k} ${n}`).join(', ') || 'none';
++ log(`active follows ${c.active} (${c.channels} channels, ${c.followers} followers)\nunfollowed rows ${c.inactive}\nunresolved holds ${c.holds.total}${c.holds.total ? ` (${why})` : ''}`);
++ return 0;
++}
++
++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, counts };
ddiff --git a/scripts/subscribe-events.js b/scripts/subscribe-events.js
iindex a01e7ffd..9b7b2c5b 100644
---- a/scripts/subscribe-events.js
++++ b/scripts/subscribe-events.js
@@@ -12,9 +12,8 @@
*
* --topic limits the run to one of TOPICS (repeatable), e.g. to add live.stream.started to a host that
* already has the other two, or `--topic tools.usage.recorded --topic events.usage.recorded` for the
-- * project dashboards (WS-N task 4). live.stream.started needs Live's GET /internal/followers first (the
-- * consumer answers 503 until Live serves it, and Events retries), and the token Network signs for it
-- * carries live.follower.read for audience openvibe.live.
++ * project dashboards (WS-N task 4). live.stream.started reads the followers from Network's own user_follows
++ * (run npm run follows-import first on a host whose Live follows were never imported).
*
* Reads the environment: OV_EVENTS_INTERNAL_URL (Events, e.g. http://127.0.0.1:4300), OV_NETWORK_URL
* (the token issuer), JWT_PRIVATE_KEY (default data/keys/private.pem) and NETWORK_EVENTS_SECRET — the
ddiff --git a/server/identity/follows.js b/server/identity/follows.js
iindex 4c4c5947..0ca43042 100644
---- a/server/identity/follows.js
++++ b/server/identity/follows.js
@@@ -23,8 +23,8 @@
*
* :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
-- * (the one-time scripts/follows-backfill.js was retired in plan T2), holding any pair whose side has no
-- * subject in follow_import_holds.
++ * (scripts/follows-import.js, npm run follows-import), holding any pair whose side has no subject in
++ * follow_import_holds.
*/
const express = require('express');
const { ids, validate, http } = require('openvibe-contracts');
@@@ -176,12 +176,17 @@ async function onSubjectRemoved(db, subject) {
* 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 }] }
++ * A pair Network already has, followed or unfollowed since, is Network's decision and is left alone, so a
++ * re-run never brings back an unfollow. No events (the source system already has these follows); dryRun
++ * changes nothing.
++ * Reconciliation is an explicit, final cutover step while Live follow writes are paused: deactivate
++ * Live-sourced active pairs absent from its snapshot, leaving Network-sourced pairs and tombstones alone.
++ * → { imported, unchanged, deactivated, held: [{ follower_ref, target_ref, reason }] }
*/
--async function importFollows(db, source, rows, subjectOf, { dryRun = false, now = new Date().toISOString() } = {}) {
-- const out = { imported: 0, unchanged: 0, held: [] };
++async function importFollows(db, source, rows, subjectOf, { dryRun = false, reconcile = false, now = new Date().toISOString() } = {}) {
++ const out = { imported: 0, unchanged: 0, deactivated: 0, held: [] };
const run = async () => {
++ const present = new Set();
for (const r of rows) {
const follower = subjectOf(r.follower_ref);
const target = subjectOf(r.target_ref);
@@@ -194,8 +199,11 @@ async function importFollows(db, source, rows, subjectOf, { dryRun = false, now
if (!dryRun) await db.prepare('INSERT INTO follow_import_holds (source, follower_ref, target_ref, reason, seen_at) VALUES (?, ?, ?, ?, ?) ON CONFLICT (source, follower_ref, target_ref) DO UPDATE SET reason = excluded.reason, seen_at = excluded.seen_at').run(source, String(r.follower_ref), String(r.target_ref), reason, now);
continue;
}
++ present.add(`${follower}:${target}`);
++ // Both sides have subjects now: any earlier hold of this pair is resolved.
++ if (!dryRun) await db.prepare('DELETE FROM follow_import_holds WHERE source = ? AND follower_ref = ? AND target_ref = ?').run(source, String(r.follower_ref), String(r.target_ref));
const prev = await 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 (prev) { out.unchanged++; continue; }
if (!dryRun) {
await setFollow(db, follower, 'channel', target, true, {
notifyEmail: r.notify_email == null ? true : !!r.notify_email, notifyPush: r.notify_push == null ? true : !!r.notify_push,
@@@ -204,6 +212,14 @@ async function importFollows(db, source, rows, subjectOf, { dryRun = false, now
}
out.imported++;
}
++ if (reconcile) {
++ const active = await db.prepare("SELECT follower_subject, target_id FROM user_follows WHERE target_type = 'channel' AND source = ? AND active = 1").all(source);
++ for (const row of active) {
++ if (present.has(`${row.follower_subject}:${row.target_id}`)) continue;
++ if (!dryRun) await setFollow(db, row.follower_subject, 'channel', row.target_id, false, { emit: false, source, at: now });
++ out.deactivated++;
++ }
++ }
};
if (dryRun) await run(); else await db.tx(run);
return out;
ddiff --git a/server/index.js b/server/index.js
iindex 349809ba..df88550f 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -387,18 +387,13 @@ const emailService = new EmailService(db);
await emailService.ready; // its settings are read from the database before the first request
app.locals.notificationService = notificationService;
app.locals.emailService = emailService;
--// live.stream.started: the followers are read from Live (its follow graph) with Network's own service token.
--const liveFollowers = privateKey.includes('BEGIN')
-- ? require('./notifications/live-followers').createLiveFollowers({ privateKey, issuer: config.jwt.issuer, liveUrl: config.services.live.internalUrl })
-- : null;
// The moderation audit log (ADR-022): staff actions every service reports, readable by staff.
const moderationAudit = require('./admin/moderation-audit').createModerationAudit(db);
// Developer projects' usage (WS-N task 4): the services' hourly rollups, per project and day.
const projectUsage = require('./developer/usage').createProjectUsage(db);
eventsConsumer = await require('./notifications/events-consumer').createEventsConsumer({
db, notifications: notificationService, secrets: config.eventsWebhookSecrets,
-- liveFollowers, discord: () => app.locals.discordService || null, moderationAudit, projectUsage,
-- followsAuthority: process.env.FOLLOWS_AUTHORITY === 'network' ? 'network' : 'live', // ADR-030 step 4
++ discord: () => app.locals.discordService || null, moderationAudit, projectUsage,
});
console.log(`[Events consumer] ${eventsConsumer.enabled ? 'on' : 'off (NETWORK_EVENTS_SECRET unset)'}: POST /internal/events`);
ddiff --git a/server/notifications/events-consumer.js b/server/notifications/events-consumer.js
iindex 70f24651..4027be23 100644
---- a/server/notifications/events-consumer.js
++++ b/server/notifications/events-consumer.js
@@@ -12,9 +12,9 @@
* (STREAM_LIVE, category 'stream'), plus everyone who opted into all go-lives
* (stream_live_all), and the Discord live alert is sent once it has committed.
*
-- * Followers of a Live channel live in Live's database. Before the inbox transaction, Network reads
-- * them from Live's GET /internal/followers with its own service token (./live-followers.js); if Live
-- * cannot answer, the delivery is answered 503 and Events retries. The announcement window (one per
++ * The followers are Network's own graph (user_follows, server/identity/follows.js; ADR-030, plan T2):
++ * the active ('channel', <streamer subject>) rows, read in the inbox transaction. Live is never asked.
++ * The announcement window (one per
* streamer per hour, eight a day) is the same one Live's direct POST /internal/events/stream-live
* claims (./stream-live.js), so both paths can run during the switch without double notifications.
* A started event older than LIVE_STARTED_MAX_AGE (30 min; a replay, or Events catching up after an
@@@ -49,7 +49,6 @@ const { http, ids } = require('openvibe-contracts');
const { parseDelivery, createPgInbox } = require('openvibe-sdk/events');
const { siteForHost } = require('../frame/sites');
const streamLive = require('./stream-live');
--const { LiveFollowersError } = require('./live-followers');
const CONSUMER = 'network-notifications';
const INBOX_TABLE = 'network_event_inbox';
@@@ -158,56 +157,32 @@ function liveStarted(event, { now, maxAgeMs }) {
* @param {import('better-sqlite3').Database} o.db
* @param {{ create(data: object): object|null }} o.notifications NotificationService
* @param {string|string[]} o.secrets NETWORK_EVENTS_SECRET (comma list) or an array
-- * @param {{ forStream(id: number): Promise<object> }} [o.liveFollowers] ./live-followers.js
* @param {() => ({ sendLiveAlert(streamer: object, stream: object): Promise<object> }|null)} [o.discord]
* @param {{ record(event: object): string }} [o.projectUsage] ../developer/usage.js createProjectUsage()
*/
--async function createEventsConsumer({ db, notifications, secrets, liveFollowers = null, followsAuthority = 'live', discord = () => null, moderationAudit = null, projectUsage = null, liveStartedMaxAgeMs = LIVE_STARTED_MAX_AGE_MS, now = () => Date.now(), log = console }) {
++async function createEventsConsumer({ db, notifications, secrets, discord = () => null, moderationAudit = null, projectUsage = null, liveStartedMaxAgeMs = LIVE_STARTED_MAX_AGE_MS, goLiveFollowersReady = process.env.NETWORK_GO_LIVE_FOLLOWS_READY === '1', now = () => Date.now(), log = console }) {
const keys = Array.isArray(secrets) ? secrets.filter((s) => typeof s === 'string' && s.length >= 32) : secretsFrom(secrets);
const inbox = createPgInbox(db, { table: INBOX_TABLE, now });
const userBySubject = db.prepare('SELECT id FROM users WHERE subject_id = ?');
const streamerBySubject = db.prepare('SELECT id, username, display_name, avatar_url FROM users WHERE subject_id = ?');
-- const userById = db.prepare('SELECT id FROM users WHERE id = ?');
--
-- /**
-- * Work that needs I/O before the inbox transaction (it must be synchronous). Returns what the
-- * handler needs, or throws LiveFollowersError when the source cannot answer (retried by Events).
-- */
-- const PREPARE = {
-- async 'live.stream.started'(event) {
-- const v = liveStarted(event, { now: now(), maxAgeMs: liveStartedMaxAgeMs });
-- if (typeof v === 'string') return { skip: v };
-- if (!await streamerBySubject.get(v.subject)) return { skip: 'ignored:channel' }; // not a Network account: nobody to announce
-- // ADR-030 step 4: with FOLLOWS_AUTHORITY=network the followers come from Network's own graph
-- // (server/identity/follows.js), not from Live. Rollback: unset it.
-- if (followsAuthority === 'network') {
-- const rows = await db.prepare("SELECT follower_subject FROM user_follows WHERE target_type = 'channel' AND target_id = ? AND active = 1").all(v.subject);
-- return { channelSubject: v.subject, followers: rows.map((r) => ({ subject: r.follower_subject })) };
-- }
-- if (!liveFollowers) throw new LiveFollowersError('no Live followers client (RS256 signing key and OV_LIVE_INTERNAL_URL needed)');
-- return liveFollowers.forStream(v.streamId);
-- },
-- };
++ const followersOf = db.prepare("SELECT follower_subject FROM user_follows WHERE target_type = 'channel' AND target_id = ? AND active = 1");
/** live.stream.started inside the inbox transaction: the announcement window, then one notification per person. */
-- async function liveStreamStarted(event, prep) {
-- if (prep && prep.skip) return prep.skip;
++ async function liveStreamStarted(event) {
++ // Live's direct POST remains the delivery path until its Network writes and the final reconciliation are complete.
++ if (!goLiveFollowersReady) return 'ignored:follows_cutover_pending';
const v = liveStarted(event, { now: now(), maxAgeMs: liveStartedMaxAgeMs });
if (typeof v === 'string') return v;
-- if (!prep || prep.missing) return 'ignored:stream';
-- if (prep.channelSubject !== v.subject) return 'ignored:channel-mismatch'; // that stream is not this channel's
const streamer = await streamerBySubject.get(v.subject);
-- if (!streamer) return 'ignored:channel';
++ if (!streamer) return 'ignored:channel'; // not a Network account: nobody to announce
const claim = await streamLive.claimAnnouncement(db, { streamerKey: streamer.id, streamId: v.streamId, now: now() });
if (claim.skipped) return { outcome: `skipped:${claim.reason}`, detail: { next_allowed_at: claim.next_allowed_at || null } };
++ const followers = await followersOf.all(v.subject);
const targets = new Set();
let unresolved = 0;
-- for (const f of prep.followers || []) {
-- let uid = null;
-- if (f.subject && ids.isSubjectId('user', f.subject)) { const u = await userBySubject.get(f.subject); if (u) uid = u.id; }
-- if (uid == null && f.network_user_id) { const u = await userById.get(f.network_user_id); if (u) uid = u.id; }
-- if (uid == null) { unresolved++; continue; }
-- targets.add(uid);
++ for (const f of followers) {
++ const u = await userBySubject.get(f.follower_subject);
++ if (u) targets.add(u.id); else unresolved++;
}
for (const uid of await streamLive.allLiveSubscribers(db)) targets.add(uid);
targets.delete(streamer.id); // never tell streamers about themselves
@@@ -216,7 +191,7 @@ async function createEventsConsumer({ db, notifications, secrets, liveFollowers
stream: { id: v.streamId, title: v.title, protocol: v.protocol, event_id: event.event_id }, url: v.url,
});
const created = targets.size ? await notifications.createBulk([...targets], notification) : [];
-- const detail = { followers: (prep.followers || []).length, unresolved, targets: targets.size, notified: created.length, ...(prep.truncated ? { truncated: true } : {}) };
++ const detail = { followers: followers.length, unresolved, targets: targets.size, notified: created.length };
const after = () => {
const d = discord && discord();
if (!d || typeof d.sendLiveAlert !== 'function') return;
@@@ -227,8 +202,8 @@ async function createEventsConsumer({ db, notifications, secrets, liveFollowers
return { outcome: created.length ? 'notified' : targets.size ? 'suppressed:preference' : 'no-recipients', detail, after };
}
-- /** Apply one envelope (with what PREPARE fetched for it). Returns { duplicate, outcome, detail }. Throws only on a storage failure. */
-- async function apply(event, prep) {
++ /** Apply one envelope. Returns { duplicate, outcome, detail }. Throws only on a storage failure. */
++ async function apply(event) {
let after = null;
const r = await inbox.once(CONSUMER, event.event_id, async () => {
// Staff actions go to the moderation audit log (ADR-022), never to anyone's inbox.
@@@ -238,7 +213,7 @@ async function createEventsConsumer({ db, notifications, secrets, liveFollowers
// Creator analytics (WS-E task 6): each ended stream's totals, counts only.
if (event.event_type === 'live.stream.ended') return require('../analytics/creators').record(db, event);
if (event.event_type === 'live.stream.started') {
-- const out = await liveStreamStarted(event, prep);
++ const out = await liveStreamStarted(event);
if (typeof out === 'string') return out;
after = out.after || null;
return { outcome: out.outcome, detail: out.detail };
@@@ -271,21 +246,9 @@ async function createEventsConsumer({ db, notifications, secrets, liveFollowers
if (!event || typeof event !== 'object' || typeof event.event_id !== 'string' || !EVENT_ID_RE.test(event.event_id) || typeof event.event_type !== 'string') {
return http.sendProblem(res, 400, 'network.bad_delivery', { detail: 'body must be { event: <envelope>, seq }' });
}
-- let prep;
-- if (Object.prototype.hasOwnProperty.call(PREPARE, event.event_type)) {
-- // Already handled: answer without asking anyone anything.
-- if (await inbox.seen(CONSUMER, event.event_id)) return res.status(200).json({ event_id: event.event_id, duplicate: true, outcome: null });
-- try {
-- prep = await PREPARE[event.event_type](event);
-- } catch (err) {
-- const known = err instanceof LiveFollowersError;
-- log.warn(`[Events consumer] ${event.event_id} (${event.event_type}) not ready: ${err.message}`);
-- return http.sendProblem(res, known ? 503 : 500, known ? 'network.dependency_unavailable' : 'network.event_failed', { detail: known ? 'the followers could not be read from Live; it will be retried' : 'processing failed; it will be retried' });
-- }
-- }
let out;
try {
-- out = await apply(event, prep);
++ out = await apply(event);
} catch (err) {
// Not acknowledged: the inbox claim rolled back with the notification, and Events retries.
log.error(`[Events consumer] ${event.event_id} (${event.event_type}) failed:`, err.message);
ddiff --git a/server/notifications/follow-notify.js b/server/notifications/follow-notify.js
iindex 41e25259..5d24c361 100644
---- a/server/notifications/follow-notify.js
++++ b/server/notifications/follow-notify.js
@@@ -4,7 +4,7 @@
* starts, made on any site, through the API or by Live's write-through (network.follows.write), tells the
* followed person once: setFollow calls this inside its transaction (server/identity/follows.js setNotifier).
* Live stopped pushing its own FOLLOW notification for follows Network took; it still pushes one for a follow
-- * that stays Live-only (FOLLOWS_AUTHORITY unset, or a side with no Network subject).
++ * that stays Live-only (Live's FOLLOWS_AUTHORITY unset, or a side with no Network subject).
*
* The notification service applies the recipient's social preference, their blocks (actor_subject) and the
* one-an-hour dedupe per sender. The link is the follower's channel, as Live's was.
ddiff --git a/server/notifications/live-followers.js b/server/notifications/live-followers.js
ddeleted file mode 100644
iindex 7fc59ef6..00000000
---- a/server/notifications/live-followers.js
++++ /dev/null
@@@ -1,75 +0,0 @@
--'use strict';
--/**
-- * Who follows a Live channel — asked of OpenVibe.Live, which owns the follow graph (its `follows`
-- * table, keyed by Live user ids). Network never keeps a copy: it reads the list when a
-- * live.stream.started event arrives and turns it into notifications.
-- *
-- * GET <live>/internal/followers?stream_id=<id>&limit=<n>[&after=<cursor>]
-- * Authorization: Bearer <service token: sub svc:network, aud openvibe.live, cap live.follower.read>
-- *
-- * 200 { stream_id, channel: { subject }, followers: [{ subject, network_user_id }], next }
-- * 404 { code: 'live.unknown_stream' } the stream row does not exist (nothing to announce)
-- *
-- * Network is the token issuer, so it signs its own 5-minute token (as for OpenVibe.AI and Events).
-- * Live verifies it with the Network key it already verifies user JWTs with (server/net/service-guard.js
-- * there) and serves the route on loopback only.
-- */
--const crypto = require('crypto');
--
--const CAPABILITY = 'live.follower.read';
--const PAGE = 5000;
--const MAX = 50000;
--
--class LiveFollowersError extends Error {}
--
--function createLiveFollowers({ privateKey, issuer, liveUrl, fetchImpl = globalThis.fetch, timeoutMs = 10000 }) {
-- const base = String(liveUrl || '').replace(/\/+$/, '');
-- let cached = null;
-- function token() {
-- const { serviceAuth } = require('openvibe-contracts');
-- const now = Math.floor(Date.now() / 1000);
-- if (cached && cached.exp - 60 > now) return cached.token;
-- const claims = { iss: issuer, sub: 'svc:network', actor_type: 'service', aud: ['openvibe.live'], cap: [CAPABILITY], iat: now, exp: now + 300, jti: `tok_${crypto.randomBytes(12).toString('hex')}` };
-- cached = { token: serviceAuth.signServiceToken(claims, privateKey), exp: claims.exp };
-- return cached.token;
-- }
--
-- /**
-- * Every follower of the channel that streams `streamId`: { channelSubject, followers: [{ subject, network_user_id }], truncated }
-- * or { missing: true } when Live has no such stream. Throws LiveFollowersError when Live cannot
-- * answer (down, route not deployed, token refused): the delivery is then retried by Events.
-- */
-- async function forStream(streamId) {
-- if (!base || !privateKey || !issuer) throw new LiveFollowersError('Live followers client is not configured');
-- const followers = [];
-- let after = null; let channelSubject; let truncated = false;
-- for (;;) {
-- const qs = new URLSearchParams({ stream_id: String(streamId), limit: String(PAGE) });
-- if (after != null) qs.set('after', String(after));
-- let res;
-- try {
-- res = await fetchImpl(`${base}/internal/followers?${qs}`, { headers: { Authorization: `Bearer ${token()}`, accept: 'application/json' }, signal: AbortSignal.timeout(timeoutMs) });
-- } catch (err) {
-- throw new LiveFollowersError(`Live unreachable: ${err.name === 'TimeoutError' ? 'timeout' : err.message}`);
-- }
-- const body = await res.json().catch(() => null);
-- if (res.status === 404 && body && body.code === 'live.unknown_stream') return { missing: true };
-- if (res.status === 401) cached = null;
-- if (!res.ok || !body || !Array.isArray(body.followers)) throw new LiveFollowersError(`Live answered ${res.status}${body && body.code ? ` ${body.code}` : ''}`);
-- const subj = body.channel && typeof body.channel.subject === 'string' ? body.channel.subject : null;
-- if (channelSubject === undefined) channelSubject = subj;
-- for (const f of body.followers) {
-- if (!f || typeof f !== 'object') continue;
-- followers.push({ subject: typeof f.subject === 'string' ? f.subject : null, network_user_id: Number.isInteger(f.network_user_id) && f.network_user_id > 0 ? f.network_user_id : null });
-- }
-- if (followers.length >= MAX) { truncated = body.next != null; followers.length = Math.min(followers.length, MAX); break; }
-- if (body.next == null || body.next === after || !body.followers.length) break;
-- after = body.next;
-- }
-- return { channelSubject, followers, truncated };
-- }
--
-- return { forStream, enabled: !!(base && privateKey && issuer) };
--}
--
--module.exports = { createLiveFollowers, LiveFollowersError, CAPABILITY };
ddiff --git a/server/notifications/stream-live.js b/server/notifications/stream-live.js
iindex ed3e4381..215d8a49 100644
---- a/server/notifications/stream-live.js
++++ b/server/notifications/stream-live.js
@@@ -5,7 +5,7 @@
* POST /internal/events/stream-live Live's direct call (server/internal/routes.js), with the
* follower list in the body (follower_network_ids)
* live.stream.started via Events server/notifications/events-consumer.js; the followers are
-- * read from Live with a service token (./live-followers.js)
++ * read from Network's active user_follows rows
*
* Both claim the same persisted per-streamer announcement window (stream_live_announcements, keyed
* by the streamer's NETWORK user id), so while Live still makes the direct call and Network also
ddiff --git a/test/events-consumer.test.js b/test/events-consumer.test.js
iindex 7513cebb..8ecebb58 100644
---- a/test/events-consumer.test.js
++++ b/test/events-consumer.test.js
@@@ -1,8 +1,8 @@
'use strict';
// Events → notifications (server/notifications/events-consumer.js): POST /internal/events turns
// deals.watch.matched and trade.alert.triggered into inbox notifications for the subscribed person,
--// and live.stream.started into one go-live notification per follower (read from Live with a service
--// token), exactly once, v2 signatures only, respecting their preferences.
++// and live.stream.started into one go-live notification per follower (Network's own user_follows; Live is
++// never asked), exactly once, v2 signatures only, respecting their preferences.
// node test/events-consumer.test.js
const assert = require('assert');
const fs = require('fs');
@@@ -10,13 +10,11 @@ const os = require('os');
const path = require('path');
const http = require('http');
const express = require('express');
--const crypto = require('crypto');
--const { ids, serviceAuth } = require('openvibe-contracts');
++const { ids } = require('openvibe-contracts');
const { signDeliveryHeaders } = require('openvibe-sdk/events');
const { getDb } = require('../server/db/database');
const { NotificationService } = require('../server/notifications/notification-service');
const { createEventsConsumer, TOPICS } = require('../server/notifications/events-consumer');
--const { createLiveFollowers } = require('../server/notifications/live-followers');
const streamLive = require('../server/notifications/stream-live');
const { topicsFrom } = require('../scripts/subscribe-events');
@@@ -193,41 +191,22 @@ const rows = async (userId) => await db.prepare('SELECT * FROM notifications WHE
assert.strictEqual(r.status, 400);
// ── live.stream.started: every follower of the channel, once ──────────────────────────────
-- const { privateKey, publicKey } = crypto.generateKeyPairSync('rsa', { modulusLength: 2048, privateKeyEncoding: { type: 'pkcs8', format: 'pem' }, publicKeyEncoding: { type: 'spki', format: 'pem' } });
-- const CAROL = ids.newId('user'), DAVE = ids.newId('user'), FRANK = ids.newId('user'), GINA = ids.newId('user'), HANK = ids.newId('user'), STRANGER = ids.newId('user');
++ const CAROL = ids.newId('user'), DAVE = ids.newId('user'), ERIN = ids.newId('user'), FRANK = ids.newId('user'), GINA = ids.newId('user'), HANK = ids.newId('user'), STRANGER = ids.newId('user');
await db.prepare(`INSERT INTO users (id, username, password_hash, subject_id, avatar_url) VALUES
-- (20, 'carol', 'x', ?, 'https://openvibe.media/avatar/carol'), (21, 'dave', 'x', ?, NULL), (22, 'erin', 'x', NULL, NULL),
-- (23, 'frank', 'x', ?, NULL), (24, 'gina', 'x', ?, NULL), (25, 'hank', 'x', ?, NULL)`).run(CAROL, DAVE, FRANK, GINA, HANK);
++ (20, 'carol', 'x', ?, 'https://openvibe.media/avatar/carol'), (21, 'dave', 'x', ?, NULL), (22, 'erin', 'x', ?, NULL),
++ (23, 'frank', 'x', ?, NULL), (24, 'gina', 'x', ?, NULL), (25, 'hank', 'x', ?, NULL)`).run(CAROL, DAVE, ERIN, FRANK, GINA, HANK);
await notifications.setPreference(23, 'stream', { enabled: false }); // Frank muted go-lives
await notifications.setPreference(24, 'stream_live_all', { enabled: true }); // Gina wants every go-live
-- // Live's GET /internal/followers, as the Live-side change specifies. Streams: 501 carol, 502 hank.
-- let liveMode = 'ok'; const liveCalls = []; const liveAuth = [];
-- const FOLLOWERS = {
-- 501: { channel: CAROL, list: [{ subject: DAVE, network_user_id: 21 }, { subject: null, network_user_id: 22 }, { subject: FRANK, network_user_id: 23 }, { subject: STRANGER, network_user_id: null }, { subject: CAROL, network_user_id: 20 }] },
-- 502: { channel: HANK, list: [{ subject: DAVE, network_user_id: 21 }] },
-- };
-- const live = http.createServer((req, res) => {
-- const u = new URL(req.url, 'http://x');
-- liveCalls.push(u.pathname + u.search);
-- res.setHeader('content-type', 'application/json');
-- if (liveMode === 'down') { res.statusCode = 502; return res.end('{}'); }
-- const v = serviceAuth.verifyServiceToken(String(req.headers.authorization || '').slice(7), { publicKey, issuer: 'https://openvibe.network', audience: 'openvibe.live' });
-- liveAuth.push(v.ok ? v.claims : null);
-- if (!v.ok || !v.claims.cap.includes('live.follower.read')) { res.statusCode = 401; return res.end('{"code":"token.invalid"}'); }
-- if (u.pathname !== '/internal/followers') { res.statusCode = 404; return res.end('{}'); }
-- const f = FOLLOWERS[u.searchParams.get('stream_id')];
-- if (!f) { res.statusCode = 404; return res.end('{"code":"live.unknown_stream"}'); }
-- const after = Number(u.searchParams.get('after') || 0);
-- const page = f.list.slice(after, after + 2); // two per page: exercises the cursor
-- const next = after + 2 < f.list.length ? after + 2 : null;
-- res.end(JSON.stringify({ stream_id: Number(u.searchParams.get('stream_id')), channel: { subject: f.channel }, followers: page, next }));
-- });
-- await new Promise(r => live.listen(0, '127.0.0.1', r));
++ // Network's follow graph: Dave, Erin, Frank and a subject with no account follow Carol; Hank unfollowed her;
++ // Dave follows Hank.
++ const follow = (follower, target, active = 1) => db.prepare("INSERT INTO user_follows (follower_subject, target_type, target_id, active) VALUES (?, 'channel', ?, ?)").run(follower, target, active);
++ for (const f of [DAVE, ERIN, FRANK, STRANGER]) await follow(f, CAROL);
++ await follow(HANK, CAROL, 0);
++ await follow(DAVE, HANK);
const discordCalls = [];
let clock = Date.now();
consumer = await createEventsConsumer({
-- db, notifications, secrets: SECRET, now: () => clock,
-- liveFollowers: createLiveFollowers({ privateKey, issuer: 'https://openvibe.network', liveUrl: `http://127.0.0.1:${live.address().port}` }),
++ db, notifications, secrets: SECRET, now: () => clock, goLiveFollowersReady: true,
discord: () => ({ sendLiveAlert: async (streamer, stream) => { discordCalls.push([streamer, stream]); return { sent: true }; } }),
});
const liveEvent = (over = {}, payload = {}) => ({
@@@ -242,16 +221,21 @@ const rows = async (userId) => await db.prepare('SELECT * FROM notifications WHE
});
const goLives = async (uid) => await db.prepare("SELECT * FROM notifications WHERE user_id = ? AND type = 'STREAM_LIVE' ORDER BY seq").all(uid);
-- // One delivery: Dave (by subject), Erin (by network id, no subject yet) and Gina (all go-lives) are
-- // notified once each; Frank muted the category, Carol is the streamer, the stranger has no account.
++ const pending = await createEventsConsumer({ db, notifications, secrets: SECRET, goLiveFollowersReady: false });
++ assert.strictEqual((await pending.apply(liveEvent())).outcome, 'ignored:follows_cutover_pending');
++ assert.strictEqual((await goLives(21)).length, 0, 'Network does not announce before Live follow writes are cut over');
++
++ // One delivery: Dave, Erin and Gina (all go-lives) are notified once each; Frank muted the category, Carol
++ // is the streamer, the stranger has no account, Hank unfollowed.
const s1 = liveEvent();
r = await post(s1);
assert.strictEqual(r.status, 200, JSON.stringify(r.body));
assert.strictEqual(r.body.outcome, 'notified');
-- assert.deepStrictEqual(r.body.detail, { followers: 5, unresolved: 1, targets: 4, notified: 3 });
++ assert.deepStrictEqual(r.body.detail, { followers: 4, unresolved: 1, targets: 4, notified: 3 });
for (const uid of [21, 22, 24]) assert.strictEqual((await goLives(uid)).length, 1, `user ${uid} is told once`);
assert.strictEqual((await goLives(23)).length, 0, 'the stream category is muted: preferences apply');
assert.strictEqual((await goLives(20)).length, 0, 'the streamer is not told about themself');
++ assert.strictEqual((await goLives(25)).length, 0, 'an unfollow is not a follower');
const n1 = (await goLives(21))[0];
assert.strictEqual(n1.title, 'Carol 3 is live!');
assert.strictEqual(n1.message, 'Building a b robot /b', 'no markup reaches the inbox');
@@@ -269,17 +253,13 @@ const rows = async (userId) => await db.prepare('SELECT * FROM notifications WHE
assert.strictEqual(n1.sender_id, 20, 'sender is the streamer\'s Network account (dedupe key)');
assert.strictEqual(n1.sender_avatar, 'https://openvibe.media/avatar/carol');
assert.strictEqual(JSON.parse(n1.rich_content).context.event_id, s1.event_id);
-- assert.deepStrictEqual(liveCalls.map(c => c.replace(/limit=\d+/, 'limit=N')), ['/internal/followers?stream_id=501&limit=N', '/internal/followers?stream_id=501&limit=N&after=2', '/internal/followers?stream_id=501&limit=N&after=4']);
-- assert.ok(liveAuth.every(c => c && c.sub === 'svc:network' && c.aud.includes('openvibe.live') && c.cap.includes('live.follower.read')), 'a Network service token for Live');
await new Promise(r2 => setImmediate(r2));
assert.strictEqual(discordCalls.length, 1, 'the Discord live alert, after the commit');
assert.deepStrictEqual(discordCalls[0][1], { id: 501, title: 'Building a b robot /b', protocol: 'rtmp' });
-- // The same delivery again: a duplicate. Live is not asked again, nobody is told twice, no second Discord post.
-- liveCalls.length = 0;
++ // The same delivery again: a duplicate. Nobody is told twice, no second Discord post.
r = await post(s1);
assert.deepStrictEqual([r.status, r.body.duplicate, r.body.outcome], [200, true, null]);
-- assert.strictEqual(liveCalls.length, 0);
for (const uid of [21, 22, 24]) assert.strictEqual((await goLives(uid)).length, 1);
assert.strictEqual(discordCalls.length, 1);
@@@ -291,28 +271,18 @@ const rows = async (userId) => await db.prepare('SELECT * FROM notifications WHE
// so during the switch the two paths never both announce.
assert.strictEqual((await streamLive.claimAnnouncement(db, { streamerKey: 20, streamId: 501 })).reason, 'cooldown');
-- // v1-only (and any unverifiable) live delivery: refused, not recorded, Live never asked.
-- liveCalls.length = 0;
++ // v1-only (and any unverifiable) live delivery: refused, not recorded.
const v1 = liveEvent({ subject: { type: 'stream', id: '502', revision: 1 }, actor: { type: 'user', id: HANK } }, { stream_id: 502, channel: { username: 'hank', subject: { type: 'user', id: HANK } } });
r = await post(v1, { strip: ['X-OpenVibe-Signature-V2', 'X-OpenVibe-Timestamp'] });
assert.deepStrictEqual([r.status, r.body.code], [401, 'network.bad_signature']);
assert.ok(!await db.prepare('SELECT 1 FROM network_event_inbox WHERE event_id = ?').get(v1.event_id));
-- assert.strictEqual(liveCalls.length, 0);
--
-- // Live down: 503, nothing recorded, so Events' retry does the work once Live answers.
-- liveMode = 'down';
-- r = await post(v1);
-- assert.deepStrictEqual([r.status, r.body.code], [503, 'network.dependency_unavailable']);
-- assert.ok(!await db.prepare('SELECT 1 FROM network_event_inbox WHERE event_id = ?').get(v1.event_id));
assert.strictEqual((await goLives(21)).length, 1);
-- liveMode = 'ok';
r = await post(v1);
assert.strictEqual(r.body.outcome, 'notified');
assert.strictEqual((await goLives(21)).length, 2, 'Dave follows Hank too');
assert.strictEqual((await goLives(24)).length, 2, 'Gina gets every go-live');
// Refused or ignored without notifying anyone.
-- liveCalls.length = 0;
const count = async () => (await db.prepare("SELECT COUNT(*) AS c FROM notifications WHERE type = 'STREAM_LIVE'").get()).c;
const total = await count();
r = await post(liveEvent({ source: 'tips' }));
@@@ -325,16 +295,9 @@ const rows = async (userId) => await db.prepare('SELECT * FROM notifications WHE
assert.strictEqual(r.body.outcome, 'ignored:channel');
r = await post(liveEvent({}, { channel: { username: 'ghost', subject: { type: 'user', id: STRANGER } } }), { now: clock });
assert.strictEqual(r.body.outcome, 'ignored:channel', 'a channel with no Network account');
-- assert.strictEqual(liveCalls.length, 0, 'none of those asked Live');
-- clock += 2 * 60 * 60 * 1000; // past Carol's cooldown
-- r = await post(liveEvent({}, { stream_id: 502 }), { now: clock });
-- assert.strictEqual(r.body.outcome, 'ignored:channel-mismatch', 'stream 502 is Hank\'s, not Carol\'s');
-- r = await post(liveEvent({}, { stream_id: 999 }), { now: clock });
-- assert.strictEqual(r.body.outcome, 'ignored:stream');
assert.strictEqual(await count(), total);
assert.strictEqual(discordCalls.length, 2);
-- live.close();
srv.close();
console.log('events consumer: all checks passed');
})().catch((err) => { console.error(err); process.exit(1); });
ddiff --git a/test/follows-authority.test.js b/test/follows-authority.test.js
ddeleted file mode 100644
iindex 4f1768ae..00000000
---- a/test/follows-authority.test.js
++++ /dev/null
@@@ -1,92 +0,0 @@
--'use strict';
--// ADR-030 step 4: products write follows through /internal/follows (network.follows.write, service token only),
--// and with FOLLOWS_AUTHORITY=network the go-live notifications take the followers from Network's own graph
--// (no call to Live).
--// node test/follows-authority.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 { ids } = require('openvibe-contracts');
--const { signDeliveryHeaders } = require('openvibe-sdk/events');
--const { getDb } = require('../server/db/database');
--const { NotificationService } = require('../server/notifications/notification-service');
--const { createEventsConsumer } = require('../server/notifications/events-consumer');
--
--(async () => {
--const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-follow-auth-'));
--const log = console.log; console.log = () => {};
--const db = getDb();
--console.log = log;
--await require('../server/identity/principals').ensureSchema(db);
--for (const c of ['live', 'tools']) await db.prepare('UPDATE oauth_clients SET client_secret = ? WHERE client_id = ?').run(`${c}-secret`, c);
--const CAROL = ids.newId('user'), DAVE = ids.newId('user'), ERIN = ids.newId('user');
--await db.prepare('INSERT INTO users (id, username, password_hash, subject_id, role) VALUES (20, ?, ?, ?, ?), (21, ?, ?, ?, ?), (22, ?, ?, ?, ?)')
-- .run('carol', 'x', CAROL, 'streamer', 'dave', 'x', DAVE, 'user', 'erin', 'x', ERIN, 'user');
--
--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 SECRET = 's'.repeat(40);
--const notifications = new NotificationService(db);
--const consumer = await createEventsConsumer({ db, notifications, secrets: SECRET, followsAuthority: 'network', log: { log() {}, warn() {}, error() {} } });
--const app = express();
--Object.assign(app.locals, { db, config, privateKey: keys.privateKey, publicKey: keys.publicKey });
--app.use(express.urlencoded({ extended: true }));
--app.use('/oauth', require('../server/auth/oauth-routes'));
--const routers = follows.routers({ requireAuth, followsGuard: principals.guard('network.follows.read', { legacy: false }) });
--app.use('/internal/follows', principals.guard('network.follows.write', { legacy: false }), routers.internal);
--app.use('/internal/events', consumer.router);
--const server = http.createServer(app);
--
--(async () => {
-- await new Promise((r) => server.listen(0, '127.0.0.1', r));
-- const base = `http://127.0.0.1:${server.address().port}`;
-- 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;
-- const call = (method, p, { body, token } = {}) => fetch(base + p, { method, headers: { 'content-type': 'application/json', ...(token ? { authorization: `Bearer ${token}` } : {}) }, body: body ? JSON.stringify(body) : undefined })
-- .then(async (r) => ({ status: r.status, body: await r.json().catch(() => null) }));
-- try {
-- const live = await svc('live');
-- // ── Writes on a person's behalf ──
-- assert.ok([401, 403].includes((await call('PUT', '/internal/follows/channel/carol', { body: { follower: DAVE } })).status), 'no token');
-- assert.strictEqual((await call('PUT', '/internal/follows/channel/carol', { body: { follower: DAVE }, token: await svc('tools') })).status, 403, 'without network.follows.write');
-- let r = await call('PUT', '/internal/follows/channel/carol', { body: { follower: DAVE, notify_push: false }, token: live });
-- assert.deepStrictEqual([r.status, r.body.followers, r.body.following, r.body.notify_push], [201, 1, true, false]);
-- r = await call('PUT', `/internal/follows/channel/${CAROL}`, { body: { follower: ERIN }, token: live });
-- assert.strictEqual(r.body.followers, 2);
-- assert.strictEqual((await db.prepare('SELECT source FROM user_follows WHERE follower_subject = ?').get(ERIN)).source, 'live', 'the writing service is recorded');
-- assert.strictEqual((await call('PUT', '/internal/follows/channel/carol', { body: { follower: 'usr_01JAB2C3D4E5F6G7H8J9K0MNPZ' }, token: live })).status, 404, 'the follower must be a Network account');
-- assert.strictEqual((await call('PUT', '/internal/follows/channel/carol', { body: { follower: DAVE, extra: 1 }, token: live })).status, 400);
-- r = await call('DELETE', `/internal/follows/channel/carol?follower=${ERIN}`, { token: live });
-- assert.deepStrictEqual([r.status, r.body.following, r.body.followers], [200, false, 1]);
-- await call('PUT', '/internal/follows/channel/carol', { body: { follower: ERIN }, token: live });
-- const evs = (await db.prepare('SELECT envelope FROM network_event_outbox ORDER BY id').all()).map((x) => x.envelope).filter((e) => e.event_type.startsWith('network.follow.'));
-- assert.deepStrictEqual(evs.map((e) => e.event_type), ['network.follow.created', 'network.follow.created', 'network.follow.deleted', 'network.follow.created']);
--
-- // ── Go-live with FOLLOWS_AUTHORITY=network: Network's followers, no Live call ──
-- const started = {
-- event_id: ids.newId('event'), event_type: 'live.stream.started', version: 1, source: 'live',
-- actor: { type: 'user', id: CAROL }, subject: { type: 'stream', id: '801', revision: 1 }, visibility: 'public', priority: 'important',
-- occurred_at: new Date().toISOString(),
-- payload: { stream_id: 801, channel: { username: 'carol', display_name: 'Carol', url: 'https://openvibe.live/@carol', subject: { type: 'user', id: CAROL } }, title: 'Go', category: null, protocol: 'whip', is_nsfw: false, started_at: new Date().toISOString() },
-- };
-- const raw = JSON.stringify({ event: started, seq: 1 });
-- const res = await fetch(`${base}/internal/events`, { method: 'POST', headers: { 'content-type': 'application/json', ...signDeliveryHeaders(raw, SECRET) }, body: raw });
-- const body = await res.json();
-- assert.strictEqual(res.status, 200, JSON.stringify(body));
-- assert.strictEqual(body.outcome, 'notified', JSON.stringify(body));
-- const told = (await db.prepare("SELECT user_id FROM notifications WHERE type = 'STREAM_LIVE' AND sender_id = 20 ORDER BY user_id").all()).map((x) => x.user_id);
-- assert.deepStrictEqual(told, [21, 22], 'both followers, from Network\'s graph');
-- } finally {
-- server.close();
-- fs.rmSync(dir, { recursive: true, force: true });
-- }
-- console.log('follows authority: all checks passed');
--})().catch((e) => { console.error(e); process.exit(1); });
--})().catch(err => { console.error(err); process.exit(1); });
ddiff --git a/test/follows-go-live.test.js b/test/follows-go-live.test.js
nnew file mode 100644
iindex 00000000..94720d28
---- /dev/null
++++ b/test/follows-go-live.test.js
@@@ -0,0 +1,107 @@
++'use strict';
++// ADR-030 step 4, plan T2 "Follows": products write follows through /internal/follows (network.follows.write,
++// service token only), and go-live notifications take the followers from Network's own graph only (no call to
++// Live, whatever the environment says); a channel with no Network account announces nothing.
++// node test/follows-go-live.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 { ids } = require('openvibe-contracts');
++const { signDeliveryHeaders } = require('openvibe-sdk/events');
++const { getDb } = require('../server/db/database');
++const { NotificationService } = require('../server/notifications/notification-service');
++const { createEventsConsumer } = require('../server/notifications/events-consumer');
++
++(async () => {
++const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-follow-auth-'));
++const log = console.log; console.log = () => {};
++const db = getDb();
++console.log = log;
++await require('../server/identity/principals').ensureSchema(db);
++for (const c of ['live', 'tools']) await db.prepare('UPDATE oauth_clients SET client_secret = ? WHERE client_id = ?').run(`${c}-secret`, c);
++const CAROL = ids.newId('user'), DAVE = ids.newId('user'), ERIN = ids.newId('user');
++await db.prepare('INSERT INTO users (id, username, password_hash, subject_id, role) VALUES (20, ?, ?, ?, ?), (21, ?, ?, ?, ?), (22, ?, ?, ?, ?)')
++ .run('carol', 'x', CAROL, 'streamer', 'dave', 'x', DAVE, 'user', 'erin', 'x', ERIN, 'user');
++
++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 SECRET = 's'.repeat(40);
++const notifications = new NotificationService(db);
++const consumer = await createEventsConsumer({ db, notifications, secrets: SECRET, goLiveFollowersReady: true, log: { log() {}, warn() {}, error() {} } });
++const app = express();
++Object.assign(app.locals, { db, config, privateKey: keys.privateKey, publicKey: keys.publicKey });
++app.use(express.urlencoded({ extended: true }));
++app.use('/oauth', require('../server/auth/oauth-routes'));
++const routers = follows.routers({ requireAuth, followsGuard: principals.guard('network.follows.read', { legacy: false }) });
++app.use('/internal/follows', principals.guard('network.follows.write', { legacy: false }), routers.internal);
++app.use('/internal/events', consumer.router);
++const server = http.createServer(app);
++
++(async () => {
++ await new Promise((r) => server.listen(0, '127.0.0.1', r));
++ const base = `http://127.0.0.1:${server.address().port}`;
++ // Every request this process makes, to prove nothing but this test's own server is called.
++ const fetched = [];
++ const realFetch = globalThis.fetch;
++ globalThis.fetch = (url, ...rest) => { fetched.push(String(url)); return realFetch(url, ...rest); };
++ 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;
++ const call = (method, p, { body, token } = {}) => fetch(base + p, { method, headers: { 'content-type': 'application/json', ...(token ? { authorization: `Bearer ${token}` } : {}) }, body: body ? JSON.stringify(body) : undefined })
++ .then(async (r) => ({ status: r.status, body: await r.json().catch(() => null) }));
++ try {
++ const live = await svc('live');
++ // ── Writes on a person's behalf ──
++ assert.ok([401, 403].includes((await call('PUT', '/internal/follows/channel/carol', { body: { follower: DAVE } })).status), 'no token');
++ assert.strictEqual((await call('PUT', '/internal/follows/channel/carol', { body: { follower: DAVE }, token: await svc('tools') })).status, 403, 'without network.follows.write');
++ let r = await call('PUT', '/internal/follows/channel/carol', { body: { follower: DAVE, notify_push: false }, token: live });
++ assert.deepStrictEqual([r.status, r.body.followers, r.body.following, r.body.notify_push], [201, 1, true, false]);
++ r = await call('PUT', `/internal/follows/channel/${CAROL}`, { body: { follower: ERIN }, token: live });
++ assert.strictEqual(r.body.followers, 2);
++ assert.strictEqual((await db.prepare('SELECT source FROM user_follows WHERE follower_subject = ?').get(ERIN)).source, 'live', 'the writing service is recorded');
++ assert.strictEqual((await call('PUT', '/internal/follows/channel/carol', { body: { follower: 'usr_01JAB2C3D4E5F6G7H8J9K0MNPZ' }, token: live })).status, 404, 'the follower must be a Network account');
++ assert.strictEqual((await call('PUT', '/internal/follows/channel/carol', { body: { follower: DAVE, extra: 1 }, token: live })).status, 400);
++ r = await call('DELETE', `/internal/follows/channel/carol?follower=${ERIN}`, { token: live });
++ assert.deepStrictEqual([r.status, r.body.following, r.body.followers], [200, false, 1]);
++ await call('PUT', '/internal/follows/channel/carol', { body: { follower: ERIN }, token: live });
++ const evs = (await db.prepare('SELECT envelope FROM network_event_outbox ORDER BY id').all()).map((x) => x.envelope).filter((e) => e.event_type.startsWith('network.follow.'));
++ assert.deepStrictEqual(evs.map((e) => e.event_type), ['network.follow.created', 'network.follow.created', 'network.follow.deleted', 'network.follow.created']);
++
++ // ── Go-live: Network's followers, no Live call (even with Live's old switch set) ──
++ process.env.FOLLOWS_AUTHORITY = 'live';
++ const started = {
++ event_id: ids.newId('event'), event_type: 'live.stream.started', version: 1, source: 'live',
++ actor: { type: 'user', id: CAROL }, subject: { type: 'stream', id: '801', revision: 1 }, visibility: 'public', priority: 'important',
++ occurred_at: new Date().toISOString(),
++ payload: { stream_id: 801, channel: { username: 'carol', display_name: 'Carol', url: 'https://openvibe.live/@carol', subject: { type: 'user', id: CAROL } }, title: 'Go', category: null, protocol: 'whip', is_nsfw: false, started_at: new Date().toISOString() },
++ };
++ const raw = JSON.stringify({ event: started, seq: 1 });
++ const res = await fetch(`${base}/internal/events`, { method: 'POST', headers: { 'content-type': 'application/json', ...signDeliveryHeaders(raw, SECRET) }, body: raw });
++ const body = await res.json();
++ assert.strictEqual(res.status, 200, JSON.stringify(body));
++ assert.strictEqual(body.outcome, 'notified', JSON.stringify(body));
++ const told = (await db.prepare("SELECT user_id FROM notifications WHERE type = 'STREAM_LIVE' AND sender_id = 20 ORDER BY user_id").all()).map((x) => x.user_id);
++ assert.deepStrictEqual(told, [21, 22], 'both followers, from Network\'s graph');
++ assert.deepStrictEqual(body.detail, { followers: 2, unresolved: 0, targets: 2, notified: 2 });
++
++ // A channel whose subject Network does not know: ignored, nobody told.
++ const ghost = { ...started, event_id: ids.newId('event'), payload: { ...started.payload, stream_id: 802, channel: { ...started.payload.channel, username: 'ghost', subject: { type: 'user', id: ids.newId('user') } } } };
++ const raw2 = JSON.stringify({ event: ghost, seq: 2 });
++ const res2 = await fetch(`${base}/internal/events`, { method: 'POST', headers: { 'content-type': 'application/json', ...signDeliveryHeaders(raw2, SECRET) }, body: raw2 });
++ assert.deepStrictEqual([res2.status, (await res2.json()).outcome], [200, 'ignored:channel']);
++ assert.strictEqual((await db.prepare("SELECT COUNT(*) AS c FROM notifications WHERE type = 'STREAM_LIVE'").get()).c, 2);
++ assert.ok(fetched.length && fetched.every((u) => u.startsWith(base)), `only this test's server was called: ${fetched.filter((u) => !u.startsWith(base))}`);
++ globalThis.fetch = realFetch;
++ } finally {
++ server.close();
++ fs.rmSync(dir, { recursive: true, force: true });
++ }
++ console.log('follows go-live: all checks passed');
++})().catch((e) => { console.error(e); process.exit(1); });
++})().catch(err => { console.error(err); process.exit(1); });
ddiff --git a/test/follows-import.test.js b/test/follows-import.test.js
nnew file mode 100644
iindex 00000000..23f418d0
---- /dev/null
++++ b/test/follows-import.test.js
@@@ -0,0 +1,117 @@
++'use strict';
++// The follows backfill and its preflight (plan T2 "Follows"): npm run follows-import reads Live's follows from a
++// read-only SQLite file and imports them into user_follows (dry run by default, no events, pairs without a
++// subject held), a re-run imports nothing and never brings back an unfollow, and a hold is cleared once its pair
++// maps; npm run follows-preflight prints counts only.
++// node test/follows-import.test.js
++const assert = require('assert');
++const fs = require('fs');
++const os = require('os');
++const path = require('path');
++const { ids } = require('openvibe-contracts');
++const { getDb } = require('../server/db/database');
++const importer = require('../scripts/follows-import');
++const preflight = require('../scripts/follows-preflight');
++
++(async () => {
++const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-follows-import-'));
++const log = console.log; console.log = () => {};
++const db = getDb();
++console.log = log;
++try {
++ const ANN = ids.newId('user'), BOB = ids.newId('user'), CAT = ids.newId('user'), DAN = ids.newId('user');
++ await db.prepare("INSERT INTO users (id, username, password_hash, subject_id) VALUES (1, 'ann', 'x', ?), (2, 'bob', 'x', ?), (3, 'cat', 'x', ?), (4, 'dan', 'x', ?)").run(ANN, BOB, CAT, DAN);
++
++ // Live's database: users 11..14 are ann, bob, cat and dan; 15 has no Network subject yet.
++ const liveFile = path.join(dir, 'live.db');
++ const live = importer.openSqlite(liveFile, { readonly: false });
++ 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);`);
++ const link = live.prepare("INSERT INTO linked_accounts (user_id, service, service_user_id, subject_id) VALUES (?, 'network', ?, ?)");
++ [[11, ANN], [12, BOB], [13, CAT], [14, DAN]].forEach(([u, s], i) => link.run(u, String(i + 1), s));
++ const liveFollow = live.prepare("INSERT INTO follows (follower_id, streamer_id, email_notify, push_notify, created_at) VALUES (?, ?, ?, 1, '2026-01-02 03:04:05')");
++ liveFollow.run(12, 11, 1); // bob → ann
++ liveFollow.run(13, 11, 0); // cat → ann
++ liveFollow.run(14, 11, 0); // dan → ann: unfollowed on Network since
++ liveFollow.run(15, 11, 0); // no subject: held
++ liveFollow.run(13, 12, 0); // cat → bob: already on Network
++ live.close();
++ await db.prepare("INSERT INTO user_follows (follower_subject, target_type, target_id, active) VALUES (?, 'channel', ?, 0), (?, 'channel', ?, 1)").run(DAN, ANN, CAT, BOB);
++
++ const lines = [];
++ const run = async (fn, argv) => { lines.length = 0; const code = await fn.main(argv, { db, log: (m) => lines.push(m) }); return { code, out: lines.join('\n') }; };
++ const active = async () => (await db.prepare("SELECT follower_subject AS f, target_id AS t, notify_email AS e, source FROM user_follows WHERE active = 1 ORDER BY follower_subject, target_id").all());
++
++ assert.strictEqual((await run(importer, [])).code, 2, 'the Live database is required');
++ assert.strictEqual((await run(importer, ['--live-db', path.join(dir, 'nope.db')])).code, 2);
++
++ // Dry run: counts, nothing changed.
++ let r = await run(importer, ['--live-db', liveFile]);
++ assert.strictEqual(r.code, 0, r.out);
++ assert.match(r.out, /live rows {2}5; to import 2; already on Network 2; to deactivate 0; held: follower_has_no_subject 1/);
++ assert.match(r.out, /dry run: nothing changed/);
++ assert.strictEqual((await active()).length, 1);
++ assert.strictEqual((await db.prepare('SELECT COUNT(*) AS c FROM follow_import_holds').get()).c, 0);
++
++ // Preflight before: counts only.
++ r = await run(preflight, []);
++ assert.strictEqual(r.code, 0, r.out);
++ assert.match(r.out, /active follows {5}1 \(1 channels, 1 followers\)\nunfollowed rows {4}1\nunresolved holds {3}0$/);
++
++ // Applied: bob and cat now follow ann (Live's flags and time kept, source live), dan's unfollow stands.
++ r = await run(importer, ['--live-db', liveFile, '--apply']);
++ assert.match(r.out, /imported 2; already on Network 2; deactivated 0; held: follower_has_no_subject 1/);
++ const rows = await active();
++ assert.deepStrictEqual(rows.filter((x) => x.t === ANN).map((x) => [x.f, x.e, x.source]).sort(), [[BOB, 1, 'live'], [CAT, 0, 'live']].sort());
++ assert.ok(!rows.some((x) => x.f === DAN), 'an unfollow made on Network is never brought back');
++ assert.strictEqual((await db.prepare('SELECT created_at FROM user_follows WHERE follower_subject = ? AND target_id = ?').get(BOB, ANN)).created_at, '2026-01-02T03:04:05.000Z');
++ assert.strictEqual((await db.prepare("SELECT COUNT(*) AS c FROM network_event_outbox WHERE envelope->>'event_type' LIKE 'network.follow.%'").get()).c, 0, 'no events');
++ assert.strictEqual((await db.prepare("SELECT COUNT(*) AS c FROM notifications WHERE type = 'FOLLOW'").get()).c, 0, 'nobody is notified');
++
++ // Preflight: the held pair is counted, never named.
++ r = await run(preflight, []);
++ assert.match(r.out, /active follows {5}3 \(2 channels, 2 followers\)\nunfollowed rows {4}1\nunresolved holds {3}1 \(live:follower_has_no_subject 1\)$/);
++ for (const s of [ANN, BOB, CAT, DAN]) assert.ok(!r.out.includes(s), `no personal data: ${s}`);
++ assert.doesNotMatch(r.out, /\b(ann|bob|cat|dan)\b|usr_/, 'no usernames or subjects');
++
++ // Idempotent: a second run imports nothing.
++ r = await run(importer, ['--live-db', liveFile, '--apply']);
++ assert.match(r.out, /imported 0; already on Network 4; deactivated 0; held: follower_has_no_subject 1/);
++ assert.strictEqual((await active()).length, 3);
++
++ // Live user 15 gets a Network account: the next run imports the pair and clears its hold.
++ const EVE = ids.newId('user');
++ await db.prepare("INSERT INTO users (id, username, password_hash, subject_id) VALUES (5, 'eve', 'x', ?)").run(EVE);
++ const live2 = importer.openSqlite(liveFile, { readonly: false });
++ live2.prepare("INSERT INTO linked_accounts (user_id, service, service_user_id, subject_id) VALUES (15, 'network', '5', ?)").run(EVE);
++ live2.close();
++ r = await run(importer, ['--live-db', liveFile, '--apply']);
++ assert.match(r.out, /imported 1; already on Network 4; deactivated 0; held: none/);
++ r = await run(preflight, []);
++ assert.match(r.out, /unresolved holds {3}0$/);
++ // Live unfollowed cat after the import. Ordinary import preserves Network state;
++ // final reconciliation removes the stale Live-sourced follow but keeps Network's own pair.
++ const live3 = importer.openSqlite(liveFile, { readonly: false });
++ live3.prepare('DELETE FROM follows WHERE follower_id = 13 AND streamer_id = 11').run();
++ live3.close();
++ await db.prepare("INSERT INTO user_follows (follower_subject, target_type, target_id, active, source) VALUES (?, 'channel', ?, 1, 'network')").run(BOB, DAN);
++ r = await run(importer, ['--live-db', liveFile, '--apply']);
++ assert.match(r.out, /deactivated 0/);
++ assert.strictEqual((await db.prepare('SELECT active FROM user_follows WHERE follower_subject = ? AND target_id = ?').get(CAT, ANN)).active, 1);
++ r = await run(importer, ['--live-db', liveFile, '--reconcile']);
++ assert.match(r.out, /to deactivate 1/);
++ assert.strictEqual((await db.prepare('SELECT active FROM user_follows WHERE follower_subject = ? AND target_id = ?').get(CAT, ANN)).active, 1, 'preview changes nothing');
++ r = await run(importer, ['--live-db', liveFile, '--reconcile', '--apply']);
++ assert.match(r.out, /deactivated 1/);
++ assert.strictEqual((await db.prepare('SELECT active FROM user_follows WHERE follower_subject = ? AND target_id = ?').get(CAT, ANN)).active, 0);
++ assert.strictEqual((await db.prepare('SELECT active FROM user_follows WHERE follower_subject = ? AND target_id = ?').get(BOB, DAN)).active, 1);
++ assert.strictEqual((await db.prepare("SELECT COUNT(*) AS c FROM network_event_outbox WHERE envelope->>'event_type' LIKE 'network.follow.%'").get()).c, 0, 'reconciliation emits no follow events');
++ r = await run(importer, ['--live-db', liveFile, '--reconcile', '--apply']);
++ assert.match(r.out, /deactivated 0/, 'reconciliation is idempotent');
++ assert.strictEqual((await run(preflight, ['--apply'])).code, 2, 'the preflight takes no options');
++} finally {
++ fs.rmSync(dir, { recursive: true, force: true });
++}
++console.log('follows import: all checks passed');
++})().catch((e) => { console.error(e); process.exit(1); });
ddiff --git a/test/notification-events.test.js b/test/notification-events.test.js
iindex fd475bd4..9e3d54f2 100644
---- a/test/notification-events.test.js
++++ b/test/notification-events.test.js
@@@ -24,7 +24,6 @@ const { signDeliveryHeaders } = require('openvibe-sdk/events');
const { getDb } = require('../server/db/database');
const { NotificationService, NOTIFICATION_EVENT } = require('../server/notifications/notification-service');
const { createEventsConsumer } = require('../server/notifications/events-consumer');
--const { createLiveFollowers } = require('../server/notifications/live-followers');
const eventRelay = require('../server/developer/event-relay');
const ticketMod = require('../server/auth/realtime-ticket');
const session = require('../server/auth/session');
@@@ -175,17 +174,8 @@ const checkEnvelope = (env, subject) => {
srv.close();
// ── 3. A restart during a go-live ───────────────────────────────────────────────────────
-- // Live's GET /internal/followers: stream 701 is Carol's; Dave and Erin follow her.
-- const liveCalls = [];
-- const live = http.createServer((req, res) => {
-- const u = new URL(req.url, 'http://x');
-- liveCalls.push(u.pathname);
-- const v = serviceAuth.verifyServiceToken(String(req.headers.authorization || '').slice(7), { publicKey, issuer: ISSUER, audience: 'openvibe.live' });
-- res.setHeader('content-type', 'application/json');
-- if (!v.ok) { res.statusCode = 401; return res.end('{}'); }
-- res.end(JSON.stringify({ stream_id: 701, channel: { subject: CAROL }, followers: [{ subject: DAVE, network_user_id: 21 }, { subject: ERIN, network_user_id: 22 }], next: null }));
-- });
-- await new Promise((r2) => live.listen(0, '127.0.0.1', r2));
++ // Network's follow graph: Dave and Erin follow Carol.
++ for (const f of [DAVE, ERIN]) await db.prepare("INSERT INTO user_follows (follower_subject, target_type, target_id) VALUES (?, 'channel', ?)").run(f, CAROL);
const SECRET = 's'.repeat(40);
const logged = []; const discord = [];
async function boot() {
@@@ -193,8 +183,7 @@ const checkEnvelope = (env, subject) => {
const d = getDb();
const n = new NotificationService(d);
const c = await createEventsConsumer({
-- db: d, notifications: n, secrets: SECRET,
-- liveFollowers: createLiveFollowers({ privateKey, issuer: ISSUER, liveUrl: `http://127.0.0.1:${live.address().port}` }),
++ db: d, notifications: n, secrets: SECRET, goLiveFollowersReady: true,
discord: () => ({ sendLiveAlert: async (s) => { discord.push(s.username); return { sent: true }; } }),
log: { log: (m) => logged.push(m), warn() {}, error() {} },
});
@@@ -250,17 +239,14 @@ const checkEnvelope = (env, subject) => {
// Restarted again after the commit but before Events saw the 200: the redelivery is a duplicate.
await stop(proc);
proc = await boot();
-- liveCalls.length = 0;
r3 = await deliver(proc, started);
assert.deepStrictEqual([r3.status, r3.body.duplicate, r3.body.outcome], [200, true, null]);
assert.deepStrictEqual(await goLives(proc.db), [[21, 1], [22, 1]], 'nobody is told twice');
assert.strictEqual(await goLiveEvents(proc.db), 2);
-- assert.strictEqual(liveCalls.length, 0, 'Live is not asked again');
await new Promise((r2) => setImmediate(r2));
assert.deepStrictEqual(discord, ['carol']);
await stop(proc);
-- live.close();
fs.rmSync(dir, { recursive: true, force: true });
console.log('notification events: all checks passed');
})().catch((err) => { console.error(err); process.exit(1); });
ddiff --git a/test/security-ssrf.test.js b/test/security-ssrf.test.js
iindex f6a6cc6a..ee40db3d 100644
---- a/test/security-ssrf.test.js
++++ b/test/security-ssrf.test.js
@@@ -123,7 +123,6 @@ const REVIEWED = {
'server/index.js fetch': [1, 'the admin streamer proxy: Live\'s internal URL from config, fixed path prefixes'],
'server/integrations/github.js fetch': [1, 'fixed host api.github.com (asserted below)'],
'server/notifications/email-service.js http.request': [1, 'fixed host api.resend.com'],
-- 'server/notifications/live-followers.js fetch': [1, 'Live\'s internal URL from config'],
'server/profile/avatar.js fetch': [3, 'Live and Media internal URLs from config; verifyImage only for https://openvibe.media/ addresses (normalizeAvatar, asserted below)'],
'server/push/push-service.js web-push': [1, 'a person\'s push endpoint: endpointAllowed() at subscribe and send, connections through safeLookup (asserted below)'],
'server/registry/deploy-drift.js fetch': [1, 'fixed host api.github.com'],