openvibers/openvibe.network Commits
Go-live notifications from Events: POST /internal/events consumes live.stream.started (source live) and notifies every follower of payload.channel.subject once (STREAM_LIVE, category stream, so the person's mute and email choice apply), plus everyone with stream_live_all, never the streamer; the Discord live alert runs after the commit. Live owns the follow graph, so before the inbox transaction Network reads it from Live's GET /internal/followers?stream_id=&limit=&after= with a self-signed service token (sub svc:network, aud openvibe.live, cap live.follower.read; server/notifications/live-followers.js), resolving followers by usr_ subject, then Network user id; if Live cannot answer the delivery gets 503 and nothing is recorded, so Events retries. A redelivery is a duplicate before Live is asked. A start more than 30 minutes old announces nothing. The per-streamer window (1/hour, 8/day) moves to server/notifications/stream-live.js and is shared with POST /internal/events/stream-live, so Live's direct call and the event cannot both announce during the switch. scripts/subscribe-events.js subscribes the new topic too (--topic to add it alone). test/events-consumer.test.js: followers notified once, preference applies, duplicate ignored without asking Live, v1-only refused, Live down then retried, stale/foreign/mismatched/unknown streams ignored
Changes
@@@ -202,6 +202,19 @@ their per-category email choice.|---|---|---|---|| `deals.watch.matched` | Deals (`source: deals`) | `payload.recipient` (`usr_…`) | `DEAL_WATCH_MATCH`, `service`, normal || `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++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.Coupons publishes no watch event yet: its merchant watches never leave Coupons, and its `coupons.*`events name no person. Every other type is acknowledged and ignored. Deliveries are v2-signed only@@@ -218,7 +231,10 @@ The route is inert until the operator does both of these:Restart `openvibe-network`; the log says `[Events consumer] on`.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` and `trade.alert.triggered`; secret = the first `NETWORK_EVENTS_SECRET`):++ `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++ `--topic live.stream.started`:```bashcd /opt/openvibe.network@@@ -3,11 +3,16 @@/*** Create Network's OpenVibe.Events subscriptions for the notification consumer* (server/notifications/events-consumer.js): one per topic in TOPICS (deals.watch.matched,-- * trade.alert.triggered), all delivering to POST /internal/events. Events names the consumer after-- * the calling service: `network`.++ * trade.alert.triggered, live.stream.started), all delivering to POST /internal/events. Events names++ * the consumer after the calling service: `network`.** sudo node --env-file=/etc/openvibe/network.env scripts/subscribe-events.js \-- * [--endpoint http://127.0.0.1:4000/internal/events] [--dry-run]++ * [--endpoint http://127.0.0.1:4000/internal/events] [--topic live.stream.started] [--dry-run]++ *++ * --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. 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.** 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@@@ -27,6 +32,11 @@ const { TOPICS, secretsFrom } = require('../server/notifications/events-consumerconst args = process.argv.slice(2);const opt = (name, d) => { const i = args.indexOf(`--${name}`); return i >= 0 ? args[i + 1] : d; };++const topicsFrom = (argv) => {++ const picked = argv.flatMap((a, i) => (a === '--topic' && argv[i + 1] ? [argv[i + 1]] : []));++ for (const t of picked) if (!TOPICS.includes(t)) throw new Error(`--topic ${t} is not one of ${TOPICS.join(', ')}`);++ return picked.length ? picked : [...TOPICS];++};function selfToken(privateKey, issuer) {const now = Math.floor(Date.now() / 1000);@@@ -38,7 +48,10 @@ function selfToken(privateKey, issuer) {return serviceAuth.signServiceToken(claims, privateKey);}++if (require.main !== module) { module.exports = { topicsFrom }; return; }++(async () => {++ const topics = topicsFrom(args);const eventsUrl = String(config.eventsInternalUrl || '').replace(/\/+$/, '');const endpoint = opt('endpoint', `http://127.0.0.1:${config.port}/internal/events`);const secret = secretsFrom(config.eventsWebhookSecrets)[0];@@@ -47,11 +60,11 @@ function selfToken(privateKey, issuer) {const privateKey = fs.readFileSync(path.resolve(config.jwt.privateKeyPath), 'utf8');if (!privateKey.includes('BEGIN')) throw new Error(`${config.jwt.privateKeyPath} is not a PEM private key`);if (args.includes('--dry-run')) {-- for (const t of TOPICS) console.log(`would subscribe: ${t} → ${endpoint} (consumer network, Events ${eventsUrl})`);++ for (const t of topics) console.log(`would subscribe: ${t} → ${endpoint} (consumer network, Events ${eventsUrl})`);return;}const token = selfToken(privateKey, config.jwt.issuer);-- for (const pattern of TOPICS) {++ for (const pattern of topics) {const res = await fetch(`${eventsUrl}/api/v1/subscriptions`, {method: 'POST',headers: { 'Content-Type': 'application/json', Authorization: `Bearer ${token}` },@@@ -358,7 +358,14 @@ const notificationService = new NotificationService(db);const emailService = new EmailService(db);app.locals.notificationService = notificationService;app.locals.emailService = emailService;--eventsConsumer = require('./notifications/events-consumer').createEventsConsumer({ db, notifications: notificationService, secrets: config.eventsWebhookSecrets });++// 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;++eventsConsumer = require('./notifications/events-consumer').createEventsConsumer({++ db, notifications: notificationService, secrets: config.eventsWebhookSecrets,++ liveFollowers, discord: () => app.locals.discordService || null,++});console.log(`[Events consumer] ${eventsConsumer.enabled ? 'on' : 'off (NETWORK_EVENTS_SECRET unset)'}: POST /internal/events`);// Discord bot service@@@ -334,6 +334,7 @@ router.post('/coins/transfer', principals.guard('network.coins.transfer', { ownA// followers + "all streamer" subscribers.// ═══════════════════════════════════════════════════════════════++const streamLive = require('../notifications/stream-live');router.post('/events/stream-live', principals.guard('network.notifications.push'), async (req, res) => {const { streamer, stream, follower_network_ids } = req.body;if (!streamer?.username || !stream?.id) {@@@ -347,31 +348,17 @@ router.post('/events/stream-live', principals.guard('network.notifications.push'// every time. One fan-out per streamer per `stream_live_cooldown_min` (default 60) and// at most `stream_live_daily_cap` (default 8) per rolling 24h — gating inbox, push,// email AND Discord. `force:true` (admin/manual) bypasses the cooldown, not the cap.++ // The same window is claimed by the live.stream.started consumer (../notifications/stream-live.js).{-- const db = getDb(req);-- try {-- db.exec(`CREATE TABLE IF NOT EXISTS stream_live_announcements (-- id INTEGER PRIMARY KEY AUTOINCREMENT, streamer_key TEXT NOT NULL, stream_id TEXT, sent_at DATETIME DEFAULT CURRENT_TIMESTAMP);-- CREATE INDEX IF NOT EXISTS idx_sla_key ON stream_live_announcements(streamer_key, sent_at DESC)`);-- } catch { /* */ }-- const key = String(streamer.network_id || streamer.id || streamer.username).toLowerCase();-- const cooldownMin = Math.max(1, parseInt(db.getSetting('stream_live_cooldown_min'), 10) || 60);-- const dailyCap = Math.max(1, parseInt(db.getSetting('stream_live_daily_cap'), 10) || 8);-- const last = db.prepare('SELECT sent_at FROM stream_live_announcements WHERE streamer_key = ? ORDER BY sent_at DESC LIMIT 1').get(key);-- const today = db.prepare("SELECT COUNT(*) AS c FROM stream_live_announcements WHERE streamer_key = ? AND sent_at > datetime('now','-1 day')").get(key)?.c || 0;-- const lastMs = last ? new Date(String(last.sent_at).replace(' ', 'T') + 'Z').getTime() : 0;-- const sinceMin = last ? (Date.now() - lastMs) / 60000 : Infinity;-- if (today >= dailyCap) {-- console.log(`[StreamLive] ${streamer.username}: daily cap (${dailyCap}) reached — not announcing`);-- return res.json({ ok: true, skipped: true, reason: 'daily-cap', announced_today: today, cap: dailyCap });++ const claim = streamLive.claimAnnouncement(getDb(req), { streamerKey: streamer.network_id || streamer.id || streamer.username, streamId: stream.id, force: !!req.body.force });++ if (claim.skipped && claim.reason === 'daily-cap') {++ console.log(`[StreamLive] ${streamer.username}: daily cap (${claim.cap}) reached — not announcing`);++ return res.json({ ok: true, ...claim });}-- if (!req.body.force && sinceMin < cooldownMin) {-- const nextAt = new Date(lastMs + cooldownMin * 60000).toISOString();-- console.log(`[StreamLive] ${streamer.username}: announced ${Math.round(sinceMin)}m ago — cooling down until ${nextAt}`);-- return res.json({ ok: true, skipped: true, reason: 'cooldown', next_allowed_at: nextAt });++ if (claim.skipped) {++ console.log(`[StreamLive] ${streamer.username}: cooling down until ${claim.next_allowed_at}`);++ return res.json({ ok: true, ...claim });}-- db.prepare('INSERT INTO stream_live_announcements (streamer_key, stream_id) VALUES (?, ?)').run(key, stream.id != null ? String(stream.id) : null);-- db.prepare("DELETE FROM stream_live_announcements WHERE sent_at < datetime('now','-7 days')").run();}// ── Discord Alert ────────────────────────────────────────@@@ -388,27 +375,10 @@ router.post('/events/stream-live', principals.guard('network.notifications.push'const notifService = req.app.locals.notificationService;if (notifService) {const db = getDb(req);-- const displayName = streamer.display_name || streamer.username;-- const notifData = {-- type: 'STREAM_LIVE',-- title: `${displayName} is live!`,-- message: stream.title || 'Started streaming',-- icon: '🔴',-- sender_id: streamer.id || null,-- sender_name: displayName,-- sender_avatar: streamer.avatar_url || null,-- service: 'live',-- url: `https://openvibe.live/${streamer.username}`,-- rich_content: {-- thumbnail: streamer.avatar_url || null,-- context: {-- stream_id: stream.id,-- username: streamer.username,-- title: stream.title || 'Started streaming',-- protocol: stream.protocol || null,-- },-- },-- };++ const notifData = streamLive.streamLiveNotification({++ username: streamer.username, displayName: streamer.display_name, avatarUrl: streamer.avatar_url,++ senderId: streamer.id || null, stream, url: `https://openvibe.live/${streamer.username}`,++ });// The streaming follow graph lives in OpenVibe.Live, keyed by LIVE user ids; Live// translates its followers to NETWORK ids via linked_accounts and sends them here.@@@ -429,10 +399,7 @@ router.post('/events/stream-live', principals.guard('network.notifications.push'} catch { /* keep Live id */ }// Find users who opted into "all live" notifications-- const allLiveRows = db.prepare(-- "SELECT user_id FROM notification_preferences WHERE category = 'stream_live_all' AND enabled = 1"-- ).all();-- const allLiveUserIds = allLiveRows.map(r => r.user_id);++ const allLiveUserIds = streamLive.allLiveSubscribers(db);// Merge and deduplicateconst targetIds = [...new Set([...followerIds, ...allLiveUserIds])];@@@ -7,6 +7,18 @@* payload.recipient is the watcher's usr_ subject* trade.alert.triggered a Trade alert rule fired (threshold crossed, filing/document seen);* event.subject { type: 'user', id } is the rule's owner++ * live.stream.started a Live channel went live (OpenVibe.Live server/events/stream-events.js);++ * payload.channel.subject is the streamer. Every follower is notified++ * (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++ * 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++ * outage) announces nothing: the stream may be long over.** Coupons publishes no watch event: its merchant watches never leave Coupons (server/domain/watches.js* there: "Delivery is not built yet"), and its coupons.* events name no person, so there is nothing@@@ -32,10 +44,13 @@ const express = require('express');const { http, ids } = require('openvibe-contracts');const { parseDelivery, createInbox } = require('openvibe-sdk/events');const { siteForHost } = require('../chrome/sites');++const streamLive = require('./stream-live');++const { LiveFollowersError } = require('./live-followers');const CONSUMER = 'network-notifications';const INBOX_TABLE = 'network_event_inbox';--const TOPICS = Object.freeze(['deals.watch.matched', 'trade.alert.triggered']);++const TOPICS = Object.freeze(['deals.watch.matched', 'trade.alert.triggered', 'live.stream.started']);++const LIVE_STARTED_MAX_AGE_MS = 30 * 60 * 1000;const EVENT_ID_RE = /^evt_[0-9A-HJKMNP-TV-Z]{26}$/;/** Plain, single-line, length-capped text: no markup or control characters reach the inbox. */@@@ -111,21 +126,105 @@ const HANDLERS = {},};++/** The facts of a live.stream.started envelope, or an 'ignored:*' outcome. No I/O. */++function liveStarted(event, { now, maxAgeMs }) {++ if (event.source !== 'live') return 'ignored:source';++ const p = event.payload && typeof event.payload === 'object' ? event.payload : {};++ if (p.redacted === true) return 'ignored:redacted';++ const streamId = typeof p.stream_id === 'number' || /^\d{1,15}$/.test(String(p.stream_id)) ? Number(p.stream_id) : NaN;++ if (!Number.isSafeInteger(streamId) || streamId <= 0) return 'ignored:payload';++ const ch = p.channel && typeof p.channel === 'object' ? p.channel : {};++ const subject = ch.subject && typeof ch.subject === 'object' && ch.subject.type === 'user' ? ch.subject.id : null;++ if (!ids.isSubjectId('user', subject)) return 'ignored:channel';++ const username = clean(ch.username, 40);++ if (!username) return 'ignored:payload';++ const started = Date.parse(p.started_at || event.occurred_at || '');++ if (Number.isFinite(started) && now - started > maxAgeMs) return 'ignored:stale';++ return {++ streamId, subject, username, displayName: clean(ch.display_name, 60) || username,++ title: clean(p.title, 200) || null, protocol: clean(p.protocol, 20) || null,++ url: publicLink(ch.url) || `https://openvibe.live/${encodeURIComponent(username)}`,++ };++}++/*** @param {object} o* @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]*/--function createEventsConsumer({ db, notifications, secrets, now = () => Date.now(), log = console }) {++function createEventsConsumer({ db, notifications, secrets, liveFollowers = null, discord = () => null, liveStartedMaxAgeMs = LIVE_STARTED_MAX_AGE_MS, now = () => Date.now(), log = console }) {const keys = Array.isArray(secrets) ? secrets.filter((s) => typeof s === 'string' && s.length >= 32) : secretsFrom(secrets);const inbox = createInbox(db, { table: INBOX_TABLE, now });inbox.ensureSchema();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 (!streamerBySubject.get(v.subject)) return { skip: 'ignored:channel' }; // not a Network account: nobody to announce++ if (!liveFollowers) throw new LiveFollowersError('no Live followers client (RS256 signing key and OV_LIVE_INTERNAL_URL needed)');++ return liveFollowers.forStream(v.streamId);++ },++ };++++ /** live.stream.started inside the inbox transaction: the announcement window, then one notification per person. */++ function liveStreamStarted(event, prep) {++ if (prep && prep.skip) return prep.skip;++ 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 = streamerBySubject.get(v.subject);++ if (!streamer) return 'ignored:channel';++ const claim = 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 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 = userBySubject.get(f.subject); if (u) uid = u.id; }++ if (uid == null && f.network_user_id) { const u = userById.get(f.network_user_id); if (u) uid = u.id; }++ if (uid == null) { unresolved++; continue; }++ targets.add(uid);++ }++ for (const uid of streamLive.allLiveSubscribers(db)) targets.add(uid);++ targets.delete(streamer.id); // never tell streamers about themselves++ const notification = streamLive.streamLiveNotification({++ username: v.username, displayName: v.displayName, avatarUrl: streamer.avatar_url, senderId: streamer.id,++ stream: { id: v.streamId, title: v.title, protocol: v.protocol, event_id: event.event_id }, url: v.url,++ });++ const created = targets.size ? notifications.createBulk([...targets], notification) : [];++ const detail = { followers: (prep.followers || []).length, unresolved, targets: targets.size, notified: created.length, ...(prep.truncated ? { truncated: true } : {}) };++ const after = () => {++ const d = discord && discord();++ if (!d || typeof d.sendLiveAlert !== 'function') return;++ Promise.resolve()++ .then(() => d.sendLiveAlert({ username: v.username, display_name: v.displayName, avatar_url: streamer.avatar_url || null }, { id: v.streamId, title: v.title, protocol: v.protocol }))++ .catch((err) => log.warn(`[Events consumer] Discord live alert for ${v.username} failed:`, err.message));++ };++ return { outcome: created.length ? 'notified' : targets.size ? 'suppressed:preference' : 'no-recipients', detail, after };++ }-- /** Apply one envelope. Returns { duplicate, outcome }. Throws only on a storage failure. */-- function apply(event) {++ /** Apply one envelope (with what PREPARE fetched for it). Returns { duplicate, outcome, detail }. Throws only on a storage failure. */++ function apply(event, prep) {++ let after = null;const r = inbox.once(CONSUMER, event.event_id, () => {++ if (event.event_type === 'live.stream.started') {++ const out = liveStreamStarted(event, prep);++ if (typeof out === 'string') return out;++ after = out.after || null;++ return { outcome: out.outcome, detail: out.detail };++ }const handler = Object.prototype.hasOwnProperty.call(HANDLERS, event.event_type) ? HANDLERS[event.event_type] : null;if (!handler) return 'ignored:type';const out = handler(event);@@@ -136,11 +235,14 @@ function createEventsConsumer({ db, notifications, secrets, now = () => Date.now// create() returns null when the person turned the category off.return notifications.create({ ...out.notification, user_id: user.id }) ? 'notified' : 'suppressed:preference';});-- return r.duplicate ? { duplicate: true, outcome: null } : { duplicate: false, outcome: r.result };++ if (r.duplicate) return { duplicate: true, outcome: null };++ // Side effects outside the database run only after the commit, once per event.++ if (after) { try { after(); } catch (err) { log.warn('[Events consumer] after-commit step failed:', err.message); } }++ return typeof r.result === 'object' && r.result ? { duplicate: false, outcome: r.result.outcome, detail: r.result.detail } : { duplicate: false, outcome: r.result };}const router = express.Router();-- router.post('/', express.raw({ type: () => true, limit: '256kb' }), (req, res) => {++ router.post('/', express.raw({ type: () => true, limit: '256kb' }), async (req, res) => {if (!keys.length) return http.sendProblem(res, 503, 'network.webhook_disabled', { detail: 'NETWORK_EVENTS_SECRET is not set' });const raw = Buffer.isBuffer(req.body) ? req.body : Buffer.alloc(0);let delivery = null;@@@ -151,18 +253,30 @@ function createEventsConsumer({ db, notifications, secrets, now = () => Date.nowif (!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 (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 = apply(event);++ out = apply(event, prep);} 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);return http.sendProblem(res, 500, 'network.event_failed', { detail: 'processing failed; it will be retried' });}-- res.status(200).json({ event_id: event.event_id, duplicate: out.duplicate, outcome: out.outcome });++ res.status(200).json({ event_id: event.event_id, duplicate: out.duplicate, outcome: out.outcome, ...(out.detail ? { detail: out.detail } : {}) });});return { router, apply, enabled: keys.length > 0 };}--module.exports = { createEventsConsumer, CONSUMER, TOPICS, INBOX_TABLE, secretsFrom };++module.exports = { createEventsConsumer, CONSUMER, TOPICS, INBOX_TABLE, LIVE_STARTED_MAX_AGE_MS, secretsFrom };nnew file mode 100644@@@ -0,0 +1,75 @@++'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 };nnew file mode 100644@@@ -0,0 +1,79 @@++'use strict';++/**++ * "X is live" fan-out, shared by the two ways a go-live reaches Network:++ *++ * 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)++ *++ * 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++ * consumes the event, whichever arrives first announces and the other is skipped as `cooldown`.++ */++++/** One fan-out per streamer per `stream_live_cooldown_min` (60) and at most `stream_live_daily_cap` (8) per 24h. */++function ensureAnnouncements(db) {++ db.exec(`CREATE TABLE IF NOT EXISTS stream_live_announcements (++ id INTEGER PRIMARY KEY AUTOINCREMENT, streamer_key TEXT NOT NULL, stream_id TEXT, sent_at DATETIME DEFAULT CURRENT_TIMESTAMP);++ CREATE INDEX IF NOT EXISTS idx_sla_key ON stream_live_announcements(streamer_key, sent_at DESC)`);++}++++function setting(db, key) {++ if (typeof db.getSetting === 'function') return db.getSetting(key);++ try { const r = db.prepare('SELECT value FROM settings WHERE key = ?').get(key); return r ? r.value : null; } catch { return null; }++}++++/**++ * Claim the announcement for a streamer, or say why not. `force` (admin/manual) bypasses the++ * cooldown, never the daily cap. Returns { ok: true } or { skipped: true, reason, ... }.++ */++function claimAnnouncement(db, { streamerKey, streamId, force = false, now = Date.now() }) {++ try { ensureAnnouncements(db); } catch { /* exists */ }++ const key = String(streamerKey).toLowerCase();++ const cooldownMin = Math.max(1, parseInt(setting(db, 'stream_live_cooldown_min'), 10) || 60);++ const dailyCap = Math.max(1, parseInt(setting(db, 'stream_live_daily_cap'), 10) || 8);++ const last = db.prepare('SELECT sent_at FROM stream_live_announcements WHERE streamer_key = ? ORDER BY sent_at DESC, id DESC LIMIT 1').get(key);++ const today = db.prepare("SELECT COUNT(*) AS c FROM stream_live_announcements WHERE streamer_key = ? AND sent_at > datetime('now','-1 day')").get(key)?.c || 0;++ const lastMs = last ? new Date(String(last.sent_at).replace(' ', 'T') + 'Z').getTime() : 0;++ const sinceMin = last ? (now - lastMs) / 60000 : Infinity;++ if (today >= dailyCap) return { skipped: true, reason: 'daily-cap', announced_today: today, cap: dailyCap };++ if (!force && sinceMin < cooldownMin) return { skipped: true, reason: 'cooldown', next_allowed_at: new Date(lastMs + cooldownMin * 60000).toISOString() };++ db.prepare('INSERT INTO stream_live_announcements (streamer_key, stream_id) VALUES (?, ?)').run(key, streamId != null ? String(streamId) : null);++ db.prepare("DELETE FROM stream_live_announcements WHERE sent_at < datetime('now','-7 days')").run();++ return { ok: true };++}++++/** People who asked for every go-live on the network (preference category stream_live_all). */++function allLiveSubscribers(db) {++ return db.prepare("SELECT user_id FROM notification_preferences WHERE category = 'stream_live_all' AND enabled = 1").all().map(r => r.user_id);++}++++/** The STREAM_LIVE notification (category 'stream' from openvibe-shared, so the person's mute and email choice apply). */++function streamLiveNotification({ username, displayName, avatarUrl, senderId, stream, url }) {++ const name = displayName || username;++ const title = (stream && stream.title) || 'Started streaming';++ return {++ type: 'STREAM_LIVE',++ title: `${name} is live!`,++ message: title,++ icon: '🔴',++ sender_id: senderId || null,++ sender_name: name,++ sender_avatar: avatarUrl || null,++ service: 'live',++ url: url || `https://openvibe.live/${encodeURIComponent(username)}`,++ rich_content: {++ thumbnail: avatarUrl || null,++ context: {++ stream_id: stream ? stream.id : null,++ username,++ title,++ protocol: (stream && stream.protocol) || null,++ ...(stream && stream.event_id ? { event_id: stream.event_id } : {}),++ },++ },++ };++}++++module.exports = { ensureAnnouncements, claimAnnouncement, allLiveSubscribers, streamLiveNotification };@@@ -1,7 +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,--// exactly once, v2 signatures only, respecting their preferences.++// 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.// node test/events-consumer.test.jsconst assert = require('assert');const fs = require('fs');@@@ -9,11 +10,15 @@ const os = require('os');const path = require('path');const http = require('http');const express = require('express');--const { ids } = require('openvibe-contracts');++const crypto = require('crypto');++const { ids, serviceAuth } = require('openvibe-contracts');const { signDeliveryHeaders } = require('openvibe-sdk/events');const { initDb } = 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');const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-events-consumer-'));const log = console.log; console.log = () => {};@@@ -53,7 +58,10 @@ const rows = (userId) => db.prepare('SELECT * FROM notifications WHERE user_id =(async () => {// The subscriptions this consumer is meant for, and nothing else.-- assert.deepStrictEqual(TOPICS, ['deals.watch.matched', 'trade.alert.triggered']);++ assert.deepStrictEqual(TOPICS, ['deals.watch.matched', 'trade.alert.triggered', 'live.stream.started']);++ assert.deepStrictEqual(topicsFrom([]), TOPICS, 'scripts/subscribe-events.js subscribes every topic');++ assert.deepStrictEqual(topicsFrom(['--topic', 'live.stream.started']), ['live.stream.started']);++ assert.throws(() => topicsFrom(['--topic', 'live.*']));let consumer = createEventsConsumer({ db, notifications, secrets: '' });const app = express();@@@ -181,6 +189,140 @@ const rows = (userId) => db.prepare('SELECT * FROM notifications WHERE user_id =r = await post({ event_id: 'nope', event_type: 'deals.watch.matched' });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');++ 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);++ notifications.setPreference(23, 'stream', { enabled: false }); // Frank muted go-lives++ 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));++ const discordCalls = [];++ let clock = Date.now();++ consumer = createEventsConsumer({++ db, notifications, secrets: SECRET, now: () => clock,++ liveFollowers: createLiveFollowers({ privateKey, issuer: 'https://openvibe.network', liveUrl: `http://127.0.0.1:${live.address().port}` }),++ discord: () => ({ sendLiveAlert: async (streamer, stream) => { discordCalls.push([streamer, stream]); return { sent: true }; } }),++ });++ const liveEvent = (over = {}, payload = {}) => ({++ event_id: ids.newId('event'), event_type: 'live.stream.started', version: 1, source: 'live',++ actor: { type: 'user', id: CAROL }, subject: { type: 'stream', id: '501', revision: 1 }, visibility: 'public', priority: 'important',++ occurred_at: new Date(clock).toISOString(),++ payload: {++ stream_id: 501, channel: { username: 'carol', display_name: 'Carol <3', url: 'https://openvibe.live/carol', subject: { type: 'user', id: CAROL } },++ title: 'Building a <b>robot</b>', category: null, protocol: 'rtmp', is_nsfw: false, started_at: new Date(clock).toISOString(), ...payload,++ },++ ...over,++ });++ const goLives = (uid) => db.prepare("SELECT * FROM notifications WHERE user_id = ? AND type = 'STREAM_LIVE' ORDER BY rowid").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 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 });++ for (const uid of [21, 22, 24]) assert.strictEqual(goLives(uid).length, 1, `user ${uid} is told once`);++ assert.strictEqual(goLives(23).length, 0, 'the stream category is muted: preferences apply');++ assert.strictEqual(goLives(20).length, 0, 'the streamer is not told about themself');++ const n1 = 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');++ assert.strictEqual(n1.category, 'stream');++ assert.strictEqual(n1.url, 'https://openvibe.live/carol');++ 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;++ 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(goLives(uid).length, 1);++ assert.strictEqual(discordCalls.length, 1);++++ // A new stream row from the same channel within the hour (a reconnect): the announcement window holds.++ r = await post(liveEvent({}, { stream_id: 501 }));++ assert.strictEqual(r.body.outcome, 'skipped:cooldown');++ assert.strictEqual(goLives(21).length, 1);++ // ...and it is the same window Live's direct POST /internal/events/stream-live claims (keyed by the Network id),++ // so during the switch the two paths never both announce.++ assert.strictEqual(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;++ 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(!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(!db.prepare('SELECT 1 FROM network_event_inbox WHERE event_id = ?').get(v1.event_id));++ assert.strictEqual(goLives(21).length, 1);++ liveMode = 'ok';++ r = await post(v1);++ assert.strictEqual(r.body.outcome, 'notified');++ assert.strictEqual(goLives(21).length, 2, 'Dave follows Hank too');++ assert.strictEqual(goLives(24).length, 2, 'Gina gets every go-live');++++ // Refused or ignored without notifying anyone.++ liveCalls.length = 0;++ const count = () => db.prepare("SELECT COUNT(*) AS c FROM notifications WHERE type = 'STREAM_LIVE'").get().c;++ const total = count();++ r = await post(liveEvent({ source: 'tips' }));++ assert.strictEqual(r.body.outcome, 'ignored:source', 'only Live speaks for Live channels');++ // The signatures are made at the consumer's clock (it moves forward below).++ clock += 45 * 60 * 1000;++ r = await post(liveEvent({ occurred_at: new Date(clock - 45 * 60 * 1000).toISOString() }, { started_at: new Date(clock - 45 * 60 * 1000).toISOString() }), { now: clock });++ assert.strictEqual(r.body.outcome, 'ignored:stale', 'a late or replayed start announces nothing');++ r = await post(liveEvent({}, { channel: { username: 'carol', subject: { type: 'user', id: '20' } } }), { now: clock });++ 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(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); });
@@ -202,6 +202,19 @@ their per-category email choice.
@@ -202,6 +202,19 @@ their per-category email choice.
|---|---|---|---|
|---|---|---|---|
| `deals.watch.matched` | Deals (`source: deals`) | `payload.recipient` (`usr_…`) | `DEAL_WATCH_MATCH`, `service`, normal |
| `deals.watch.matched` | Deals (`source: deals`) | `payload.recipient` (`usr_…`) | `DEAL_WATCH_MATCH`, `service`, normal |
| `trade.alert.triggered` | Trade (`source: trade`) | `subject.id` (`subject.type: user`) | `TRADE_ALERT`, `service`, high |
| `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
+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.
Coupons publishes no watch event yet: its merchant watches never leave Coupons, and its `coupons.*`
Coupons publishes no watch event yet: its merchant watches never leave Coupons, and its `coupons.*`
events name no person. Every other type is acknowledged and ignored. Deliveries are v2-signed only
events name no person. Every other type is acknowledged and ignored. Deliveries are v2-signed only
@@ -218,7 +231,10 @@ The route is inert until the operator does both of these:
@@ -218,7 +231,10 @@ The route is inert until the operator does both of these:
Restart `openvibe-network`; the log says `[Events consumer] on`.
Restart `openvibe-network`; the log says `[Events consumer] on`.
2. Create the subscriptions in Events (consumer `network`, which Events takes from the token's
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
`svc:network`; endpoint `http://127.0.0.1:4000/internal/events`; topic patterns
- `deals.watch.matched` and `trade.alert.triggered`; secret = the first `NETWORK_EVENTS_SECRET`):
+ `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
+ `--topic live.stream.started`:
```bash
```bash
cd /opt/openvibe.network
cd /opt/openvibe.network
@@ -3,11 +3,16 @@
@@ -3,11 +3,16 @@
/**
/**
* Create Network's OpenVibe.Events subscriptions for the notification consumer
* Create Network's OpenVibe.Events subscriptions for the notification consumer
* (server/notifications/events-consumer.js): one per topic in TOPICS (deals.watch.matched,
* (server/notifications/events-consumer.js): one per topic in TOPICS (deals.watch.matched,
- * trade.alert.triggered), all delivering to POST /internal/events. Events names the consumer after
- * the calling service: `network`.
+ * trade.alert.triggered, live.stream.started), all delivering to POST /internal/events. Events names
+ * the consumer after the calling service: `network`.
*
*
* sudo node --env-file=/etc/openvibe/network.env scripts/subscribe-events.js \
* sudo node --env-file=/etc/openvibe/network.env scripts/subscribe-events.js \
- * [--endpoint http://127.0.0.1:4000/internal/events] [--dry-run]
+ * [--endpoint http://127.0.0.1:4000/internal/events] [--topic live.stream.started] [--dry-run]
+ *
+ * --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. 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.
*
*
* Reads the environment: OV_EVENTS_INTERNAL_URL (Events, e.g. http://127.0.0.1:4300), OV_NETWORK_URL
* 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
* (the token issuer), JWT_PRIVATE_KEY (default data/keys/private.pem) and NETWORK_EVENTS_SECRET — the
@@ -27,6 +32,11 @@ const { TOPICS, secretsFrom } = require('../server/notifications/events-consumer
@@ -27,6 +32,11 @@ const { TOPICS, secretsFrom } = require('../server/notifications/events-consumer
const args = process.argv.slice(2);
const args = process.argv.slice(2);
const opt = (name, d) => { const i = args.indexOf(`--${name}`); return i >= 0 ? args[i + 1] : d; };
const opt = (name, d) => { const i = args.indexOf(`--${name}`); return i >= 0 ? args[i + 1] : d; };
+const topicsFrom = (argv) => {
+ const picked = argv.flatMap((a, i) => (a === '--topic' && argv[i + 1] ? [argv[i + 1]] : []));
+ for (const t of picked) if (!TOPICS.includes(t)) throw new Error(`--topic ${t} is not one of ${TOPICS.join(', ')}`);
+ return picked.length ? picked : [...TOPICS];
+};
function selfToken(privateKey, issuer) {
function selfToken(privateKey, issuer) {
const now = Math.floor(Date.now() / 1000);
const now = Math.floor(Date.now() / 1000);
@@ -38,7 +48,10 @@ function selfToken(privateKey, issuer) {
@@ -38,7 +48,10 @@ function selfToken(privateKey, issuer) {
return serviceAuth.signServiceToken(claims, privateKey);
return serviceAuth.signServiceToken(claims, privateKey);
}
}
+if (require.main !== module) { module.exports = { topicsFrom }; return; }
+
(async () => {
(async () => {
+ const topics = topicsFrom(args);
const eventsUrl = String(config.eventsInternalUrl || '').replace(/\/+$/, '');
const eventsUrl = String(config.eventsInternalUrl || '').replace(/\/+$/, '');
const endpoint = opt('endpoint', `http://127.0.0.1:${config.port}/internal/events`);
const endpoint = opt('endpoint', `http://127.0.0.1:${config.port}/internal/events`);
const secret = secretsFrom(config.eventsWebhookSecrets)[0];
const secret = secretsFrom(config.eventsWebhookSecrets)[0];
@@ -47,11 +60,11 @@ function selfToken(privateKey, issuer) {
@@ -47,11 +60,11 @@ function selfToken(privateKey, issuer) {
const privateKey = fs.readFileSync(path.resolve(config.jwt.privateKeyPath), 'utf8');
const privateKey = fs.readFileSync(path.resolve(config.jwt.privateKeyPath), 'utf8');
if (!privateKey.includes('BEGIN')) throw new Error(`${config.jwt.privateKeyPath} is not a PEM private key`);
if (!privateKey.includes('BEGIN')) throw new Error(`${config.jwt.privateKeyPath} is not a PEM private key`);
if (args.includes('--dry-run')) {
if (args.includes('--dry-run')) {
- for (const t of TOPICS) console.log(`would subscribe: ${t} → ${endpoint} (consumer network, Events ${eventsUrl})`);
+ for (const t of topics) console.log(`would subscribe: ${t} → ${endpoint} (consumer network, Events ${eventsUrl})`);
return;
return;
}
}
const token = selfToken(privateKey, config.jwt.issuer);
const token = selfToken(privateKey, config.jwt.issuer);
- for (const pattern of TOPICS) {
+ for (const pattern of topics) {
const res = await fetch(`${eventsUrl}/api/v1/subscriptions`, {
const res = await fetch(`${eventsUrl}/api/v1/subscriptions`, {
method: 'POST',
method: 'POST',
headers: { 'Content-Type': 'application/json', Authorization: `Bearer ${token}` },
headers: { 'Content-Type': 'application/json', Authorization: `Bearer ${token}` },
@@ -358,7 +358,14 @@ const notificationService = new NotificationService(db);
@@ -358,7 +358,14 @@ const notificationService = new NotificationService(db);
const emailService = new EmailService(db);
const emailService = new EmailService(db);
app.locals.notificationService = notificationService;
app.locals.notificationService = notificationService;
app.locals.emailService = emailService;
app.locals.emailService = emailService;
-eventsConsumer = require('./notifications/events-consumer').createEventsConsumer({ db, notifications: notificationService, secrets: config.eventsWebhookSecrets });
+// 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;
+eventsConsumer = require('./notifications/events-consumer').createEventsConsumer({
+ db, notifications: notificationService, secrets: config.eventsWebhookSecrets,
+ liveFollowers, discord: () => app.locals.discordService || null,
+});
console.log(`[Events consumer] ${eventsConsumer.enabled ? 'on' : 'off (NETWORK_EVENTS_SECRET unset)'}: POST /internal/events`);
console.log(`[Events consumer] ${eventsConsumer.enabled ? 'on' : 'off (NETWORK_EVENTS_SECRET unset)'}: POST /internal/events`);
// Discord bot service
// Discord bot service
@@ -334,6 +334,7 @@ router.post('/coins/transfer', principals.guard('network.coins.transfer', { ownA
@@ -334,6 +334,7 @@ router.post('/coins/transfer', principals.guard('network.coins.transfer', { ownA
// followers + "all streamer" subscribers.
// followers + "all streamer" subscribers.
// ═══════════════════════════════════════════════════════════════
// ═══════════════════════════════════════════════════════════════
+const streamLive = require('../notifications/stream-live');
router.post('/events/stream-live', principals.guard('network.notifications.push'), async (req, res) => {
router.post('/events/stream-live', principals.guard('network.notifications.push'), async (req, res) => {
const { streamer, stream, follower_network_ids } = req.body;
const { streamer, stream, follower_network_ids } = req.body;
if (!streamer?.username || !stream?.id) {
if (!streamer?.username || !stream?.id) {
@@ -347,31 +348,17 @@ router.post('/events/stream-live', principals.guard('network.notifications.push'
@@ -347,31 +348,17 @@ router.post('/events/stream-live', principals.guard('network.notifications.push'
// every time. One fan-out per streamer per `stream_live_cooldown_min` (default 60) and
// every time. One fan-out per streamer per `stream_live_cooldown_min` (default 60) and
// at most `stream_live_daily_cap` (default 8) per rolling 24h — gating inbox, push,
// at most `stream_live_daily_cap` (default 8) per rolling 24h — gating inbox, push,
// email AND Discord. `force:true` (admin/manual) bypasses the cooldown, not the cap.
// email AND Discord. `force:true` (admin/manual) bypasses the cooldown, not the cap.
+ // The same window is claimed by the live.stream.started consumer (../notifications/stream-live.js).
{
{
- const db = getDb(req);
- try {
- db.exec(`CREATE TABLE IF NOT EXISTS stream_live_announcements (
- id INTEGER PRIMARY KEY AUTOINCREMENT, streamer_key TEXT NOT NULL, stream_id TEXT, sent_at DATETIME DEFAULT CURRENT_TIMESTAMP);
- CREATE INDEX IF NOT EXISTS idx_sla_key ON stream_live_announcements(streamer_key, sent_at DESC)`);
- } catch { /* */ }
- const key = String(streamer.network_id || streamer.id || streamer.username).toLowerCase();
- const cooldownMin = Math.max(1, parseInt(db.getSetting('stream_live_cooldown_min'), 10) || 60);
- const dailyCap = Math.max(1, parseInt(db.getSetting('stream_live_daily_cap'), 10) || 8);
- const last = db.prepare('SELECT sent_at FROM stream_live_announcements WHERE streamer_key = ? ORDER BY sent_at DESC LIMIT 1').get(key);
- const today = db.prepare("SELECT COUNT(*) AS c FROM stream_live_announcements WHERE streamer_key = ? AND sent_at > datetime('now','-1 day')").get(key)?.c || 0;
- const lastMs = last ? new Date(String(last.sent_at).replace(' ', 'T') + 'Z').getTime() : 0;
- const sinceMin = last ? (Date.now() - lastMs) / 60000 : Infinity;
- if (today >= dailyCap) {
- console.log(`[StreamLive] ${streamer.username}: daily cap (${dailyCap}) reached — not announcing`);
- return res.json({ ok: true, skipped: true, reason: 'daily-cap', announced_today: today, cap: dailyCap });
+ const claim = streamLive.claimAnnouncement(getDb(req), { streamerKey: streamer.network_id || streamer.id || streamer.username, streamId: stream.id, force: !!req.body.force });
+ if (claim.skipped && claim.reason === 'daily-cap') {
+ console.log(`[StreamLive] ${streamer.username}: daily cap (${claim.cap}) reached — not announcing`);
+ return res.json({ ok: true, ...claim });
}
}
- if (!req.body.force && sinceMin < cooldownMin) {
- const nextAt = new Date(lastMs + cooldownMin * 60000).toISOString();
- console.log(`[StreamLive] ${streamer.username}: announced ${Math.round(sinceMin)}m ago — cooling down until ${nextAt}`);
- return res.json({ ok: true, skipped: true, reason: 'cooldown', next_allowed_at: nextAt });
+ if (claim.skipped) {
+ console.log(`[StreamLive] ${streamer.username}: cooling down until ${claim.next_allowed_at}`);
+ return res.json({ ok: true, ...claim });
}
}
- db.prepare('INSERT INTO stream_live_announcements (streamer_key, stream_id) VALUES (?, ?)').run(key, stream.id != null ? String(stream.id) : null);
- db.prepare("DELETE FROM stream_live_announcements WHERE sent_at < datetime('now','-7 days')").run();
}
}
// ── Discord Alert ────────────────────────────────────────
// ── Discord Alert ────────────────────────────────────────
@@ -388,27 +375,10 @@ router.post('/events/stream-live', principals.guard('network.notifications.push'
@@ -388,27 +375,10 @@ router.post('/events/stream-live', principals.guard('network.notifications.push'
const notifService = req.app.locals.notificationService;
const notifService = req.app.locals.notificationService;
if (notifService) {
if (notifService) {
const db = getDb(req);
const db = getDb(req);
- const displayName = streamer.display_name || streamer.username;
- const notifData = {
- type: 'STREAM_LIVE',
- title: `${displayName} is live!`,
- message: stream.title || 'Started streaming',
- icon: '🔴',
- sender_id: streamer.id || null,
- sender_name: displayName,
- sender_avatar: streamer.avatar_url || null,
- service: 'live',
- url: `https://openvibe.live/${streamer.username}`,
- rich_content: {
- thumbnail: streamer.avatar_url || null,
- context: {
- stream_id: stream.id,
- username: streamer.username,
- title: stream.title || 'Started streaming',
- protocol: stream.protocol || null,
- },
- },
- };
+ const notifData = streamLive.streamLiveNotification({
+ username: streamer.username, displayName: streamer.display_name, avatarUrl: streamer.avatar_url,
+ senderId: streamer.id || null, stream, url: `https://openvibe.live/${streamer.username}`,
+ });
// The streaming follow graph lives in OpenVibe.Live, keyed by LIVE user ids; Live
// The streaming follow graph lives in OpenVibe.Live, keyed by LIVE user ids; Live
// translates its followers to NETWORK ids via linked_accounts and sends them here.
// translates its followers to NETWORK ids via linked_accounts and sends them here.
@@ -429,10 +399,7 @@ router.post('/events/stream-live', principals.guard('network.notifications.push'
@@ -429,10 +399,7 @@ router.post('/events/stream-live', principals.guard('network.notifications.push'
} catch { /* keep Live id */ }
} catch { /* keep Live id */ }
// Find users who opted into "all live" notifications
// Find users who opted into "all live" notifications
- const allLiveRows = db.prepare(
- "SELECT user_id FROM notification_preferences WHERE category = 'stream_live_all' AND enabled = 1"
- ).all();
- const allLiveUserIds = allLiveRows.map(r => r.user_id);
+ const allLiveUserIds = streamLive.allLiveSubscribers(db);
// Merge and deduplicate
// Merge and deduplicate
const targetIds = [...new Set([...followerIds, ...allLiveUserIds])];
const targetIds = [...new Set([...followerIds, ...allLiveUserIds])];
@@ -7,6 +7,18 @@
@@ -7,6 +7,18 @@
* payload.recipient is the watcher's usr_ subject
* payload.recipient is the watcher's usr_ subject
* trade.alert.triggered a Trade alert rule fired (threshold crossed, filing/document seen);
* trade.alert.triggered a Trade alert rule fired (threshold crossed, filing/document seen);
* event.subject { type: 'user', id } is the rule's owner
* event.subject { type: 'user', id } is the rule's owner
+ * live.stream.started a Live channel went live (OpenVibe.Live server/events/stream-events.js);
+ * payload.channel.subject is the streamer. Every follower is notified
+ * (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
+ * 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
+ * outage) announces nothing: the stream may be long over.
*
*
* Coupons publishes no watch event: its merchant watches never leave Coupons (server/domain/watches.js
* Coupons publishes no watch event: its merchant watches never leave Coupons (server/domain/watches.js
* there: "Delivery is not built yet"), and its coupons.* events name no person, so there is nothing
* there: "Delivery is not built yet"), and its coupons.* events name no person, so there is nothing
@@ -32,10 +44,13 @@ const express = require('express');
@@ -32,10 +44,13 @@ const express = require('express');
const { http, ids } = require('openvibe-contracts');
const { http, ids } = require('openvibe-contracts');
const { parseDelivery, createInbox } = require('openvibe-sdk/events');
const { parseDelivery, createInbox } = require('openvibe-sdk/events');
const { siteForHost } = require('../chrome/sites');
const { siteForHost } = require('../chrome/sites');
+const streamLive = require('./stream-live');
+const { LiveFollowersError } = require('./live-followers');
const CONSUMER = 'network-notifications';
const CONSUMER = 'network-notifications';
const INBOX_TABLE = 'network_event_inbox';
const INBOX_TABLE = 'network_event_inbox';
-const TOPICS = Object.freeze(['deals.watch.matched', 'trade.alert.triggered']);
+const TOPICS = Object.freeze(['deals.watch.matched', 'trade.alert.triggered', 'live.stream.started']);
+const LIVE_STARTED_MAX_AGE_MS = 30 * 60 * 1000;
const EVENT_ID_RE = /^evt_[0-9A-HJKMNP-TV-Z]{26}$/;
const EVENT_ID_RE = /^evt_[0-9A-HJKMNP-TV-Z]{26}$/;
/** Plain, single-line, length-capped text: no markup or control characters reach the inbox. */
/** Plain, single-line, length-capped text: no markup or control characters reach the inbox. */
@@ -111,21 +126,105 @@ const HANDLERS = {
@@ -111,21 +126,105 @@ const HANDLERS = {
},
},
};
};
+/** The facts of a live.stream.started envelope, or an 'ignored:*' outcome. No I/O. */
+function liveStarted(event, { now, maxAgeMs }) {
+ if (event.source !== 'live') return 'ignored:source';
+ const p = event.payload && typeof event.payload === 'object' ? event.payload : {};
+ if (p.redacted === true) return 'ignored:redacted';
+ const streamId = typeof p.stream_id === 'number' || /^\d{1,15}$/.test(String(p.stream_id)) ? Number(p.stream_id) : NaN;
+ if (!Number.isSafeInteger(streamId) || streamId <= 0) return 'ignored:payload';
+ const ch = p.channel && typeof p.channel === 'object' ? p.channel : {};
+ const subject = ch.subject && typeof ch.subject === 'object' && ch.subject.type === 'user' ? ch.subject.id : null;
+ if (!ids.isSubjectId('user', subject)) return 'ignored:channel';
+ const username = clean(ch.username, 40);
+ if (!username) return 'ignored:payload';
+ const started = Date.parse(p.started_at || event.occurred_at || '');
+ if (Number.isFinite(started) && now - started > maxAgeMs) return 'ignored:stale';
+ return {
+ streamId, subject, username, displayName: clean(ch.display_name, 60) || username,
+ title: clean(p.title, 200) || null, protocol: clean(p.protocol, 20) || null,
+ url: publicLink(ch.url) || `https://openvibe.live/${encodeURIComponent(username)}`,
+ };
+}
+
/**
/**
* @param {object} o
* @param {object} o
* @param {import('better-sqlite3').Database} o.db
* @param {import('better-sqlite3').Database} o.db
* @param {{ create(data: object): object|null }} o.notifications NotificationService
* @param {{ create(data: object): object|null }} o.notifications NotificationService
* @param {string|string[]} o.secrets NETWORK_EVENTS_SECRET (comma list) or an array
* @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]
*/
*/
-function createEventsConsumer({ db, notifications, secrets, now = () => Date.now(), log = console }) {
+function createEventsConsumer({ db, notifications, secrets, liveFollowers = null, discord = () => null, liveStartedMaxAgeMs = LIVE_STARTED_MAX_AGE_MS, now = () => Date.now(), log = console }) {
const keys = Array.isArray(secrets) ? secrets.filter((s) => typeof s === 'string' && s.length >= 32) : secretsFrom(secrets);
const keys = Array.isArray(secrets) ? secrets.filter((s) => typeof s === 'string' && s.length >= 32) : secretsFrom(secrets);
const inbox = createInbox(db, { table: INBOX_TABLE, now });
const inbox = createInbox(db, { table: INBOX_TABLE, now });
inbox.ensureSchema();
inbox.ensureSchema();
const userBySubject = db.prepare('SELECT id FROM users WHERE subject_id = ?');
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 (!streamerBySubject.get(v.subject)) return { skip: 'ignored:channel' }; // not a Network account: nobody to announce
+ if (!liveFollowers) throw new LiveFollowersError('no Live followers client (RS256 signing key and OV_LIVE_INTERNAL_URL needed)');
+ return liveFollowers.forStream(v.streamId);
+ },
+ };
+
+ /** live.stream.started inside the inbox transaction: the announcement window, then one notification per person. */
+ function liveStreamStarted(event, prep) {
+ if (prep && prep.skip) return prep.skip;
+ 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 = streamerBySubject.get(v.subject);
+ if (!streamer) return 'ignored:channel';
+ const claim = 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 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 = userBySubject.get(f.subject); if (u) uid = u.id; }
+ if (uid == null && f.network_user_id) { const u = userById.get(f.network_user_id); if (u) uid = u.id; }
+ if (uid == null) { unresolved++; continue; }
+ targets.add(uid);
+ }
+ for (const uid of streamLive.allLiveSubscribers(db)) targets.add(uid);
+ targets.delete(streamer.id); // never tell streamers about themselves
+ const notification = streamLive.streamLiveNotification({
+ username: v.username, displayName: v.displayName, avatarUrl: streamer.avatar_url, senderId: streamer.id,
+ stream: { id: v.streamId, title: v.title, protocol: v.protocol, event_id: event.event_id }, url: v.url,
+ });
+ const created = targets.size ? notifications.createBulk([...targets], notification) : [];
+ const detail = { followers: (prep.followers || []).length, unresolved, targets: targets.size, notified: created.length, ...(prep.truncated ? { truncated: true } : {}) };
+ const after = () => {
+ const d = discord && discord();
+ if (!d || typeof d.sendLiveAlert !== 'function') return;
+ Promise.resolve()
+ .then(() => d.sendLiveAlert({ username: v.username, display_name: v.displayName, avatar_url: streamer.avatar_url || null }, { id: v.streamId, title: v.title, protocol: v.protocol }))
+ .catch((err) => log.warn(`[Events consumer] Discord live alert for ${v.username} failed:`, err.message));
+ };
+ return { outcome: created.length ? 'notified' : targets.size ? 'suppressed:preference' : 'no-recipients', detail, after };
+ }
- /** Apply one envelope. Returns { duplicate, outcome }. Throws only on a storage failure. */
- function apply(event) {
+ /** Apply one envelope (with what PREPARE fetched for it). Returns { duplicate, outcome, detail }. Throws only on a storage failure. */
+ function apply(event, prep) {
+ let after = null;
const r = inbox.once(CONSUMER, event.event_id, () => {
const r = inbox.once(CONSUMER, event.event_id, () => {
+ if (event.event_type === 'live.stream.started') {
+ const out = liveStreamStarted(event, prep);
+ if (typeof out === 'string') return out;
+ after = out.after || null;
+ return { outcome: out.outcome, detail: out.detail };
+ }
const handler = Object.prototype.hasOwnProperty.call(HANDLERS, event.event_type) ? HANDLERS[event.event_type] : null;
const handler = Object.prototype.hasOwnProperty.call(HANDLERS, event.event_type) ? HANDLERS[event.event_type] : null;
if (!handler) return 'ignored:type';
if (!handler) return 'ignored:type';
const out = handler(event);
const out = handler(event);
@@ -136,11 +235,14 @@ function createEventsConsumer({ db, notifications, secrets, now = () => Date.now
@@ -136,11 +235,14 @@ function createEventsConsumer({ db, notifications, secrets, now = () => Date.now
// create() returns null when the person turned the category off.
// create() returns null when the person turned the category off.
return notifications.create({ ...out.notification, user_id: user.id }) ? 'notified' : 'suppressed:preference';
return notifications.create({ ...out.notification, user_id: user.id }) ? 'notified' : 'suppressed:preference';
});
});
- return r.duplicate ? { duplicate: true, outcome: null } : { duplicate: false, outcome: r.result };
+ if (r.duplicate) return { duplicate: true, outcome: null };
+ // Side effects outside the database run only after the commit, once per event.
+ if (after) { try { after(); } catch (err) { log.warn('[Events consumer] after-commit step failed:', err.message); } }
+ return typeof r.result === 'object' && r.result ? { duplicate: false, outcome: r.result.outcome, detail: r.result.detail } : { duplicate: false, outcome: r.result };
}
}
const router = express.Router();
const router = express.Router();
- router.post('/', express.raw({ type: () => true, limit: '256kb' }), (req, res) => {
+ router.post('/', express.raw({ type: () => true, limit: '256kb' }), async (req, res) => {
if (!keys.length) return http.sendProblem(res, 503, 'network.webhook_disabled', { detail: 'NETWORK_EVENTS_SECRET is not set' });
if (!keys.length) return http.sendProblem(res, 503, 'network.webhook_disabled', { detail: 'NETWORK_EVENTS_SECRET is not set' });
const raw = Buffer.isBuffer(req.body) ? req.body : Buffer.alloc(0);
const raw = Buffer.isBuffer(req.body) ? req.body : Buffer.alloc(0);
let delivery = null;
let delivery = null;
@@ -151,18 +253,30 @@ function createEventsConsumer({ db, notifications, secrets, now = () => Date.now
@@ -151,18 +253,30 @@ function createEventsConsumer({ db, notifications, secrets, now = () => Date.now
if (!event || typeof event !== 'object' || typeof event.event_id !== 'string' || !EVENT_ID_RE.test(event.event_id) || typeof event.event_type !== 'string') {
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 }' });
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 (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;
let out;
try {
try {
- out = apply(event);
+ out = apply(event, prep);
} catch (err) {
} catch (err) {
// Not acknowledged: the inbox claim rolled back with the notification, and Events retries.
// 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);
log.error(`[Events consumer] ${event.event_id} (${event.event_type}) failed:`, err.message);
return http.sendProblem(res, 500, 'network.event_failed', { detail: 'processing failed; it will be retried' });
return http.sendProblem(res, 500, 'network.event_failed', { detail: 'processing failed; it will be retried' });
}
}
- res.status(200).json({ event_id: event.event_id, duplicate: out.duplicate, outcome: out.outcome });
+ res.status(200).json({ event_id: event.event_id, duplicate: out.duplicate, outcome: out.outcome, ...(out.detail ? { detail: out.detail } : {}) });
});
});
return { router, apply, enabled: keys.length > 0 };
return { router, apply, enabled: keys.length > 0 };
}
}
-module.exports = { createEventsConsumer, CONSUMER, TOPICS, INBOX_TABLE, secretsFrom };
+module.exports = { createEventsConsumer, CONSUMER, TOPICS, INBOX_TABLE, LIVE_STARTED_MAX_AGE_MS, secretsFrom };
new file mode 100644
new file mode 100644
@@ -0,0 +1,75 @@
@@ -0,0 +1,75 @@
+'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 };
new file mode 100644
new file mode 100644
@@ -0,0 +1,79 @@
@@ -0,0 +1,79 @@
+'use strict';
+/**
+ * "X is live" fan-out, shared by the two ways a go-live reaches Network:
+ *
+ * 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)
+ *
+ * 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
+ * consumes the event, whichever arrives first announces and the other is skipped as `cooldown`.
+ */
+
+/** One fan-out per streamer per `stream_live_cooldown_min` (60) and at most `stream_live_daily_cap` (8) per 24h. */
+function ensureAnnouncements(db) {
+ db.exec(`CREATE TABLE IF NOT EXISTS stream_live_announcements (
+ id INTEGER PRIMARY KEY AUTOINCREMENT, streamer_key TEXT NOT NULL, stream_id TEXT, sent_at DATETIME DEFAULT CURRENT_TIMESTAMP);
+ CREATE INDEX IF NOT EXISTS idx_sla_key ON stream_live_announcements(streamer_key, sent_at DESC)`);
+}
+
+function setting(db, key) {
+ if (typeof db.getSetting === 'function') return db.getSetting(key);
+ try { const r = db.prepare('SELECT value FROM settings WHERE key = ?').get(key); return r ? r.value : null; } catch { return null; }
+}
+
+/**
+ * Claim the announcement for a streamer, or say why not. `force` (admin/manual) bypasses the
+ * cooldown, never the daily cap. Returns { ok: true } or { skipped: true, reason, ... }.
+ */
+function claimAnnouncement(db, { streamerKey, streamId, force = false, now = Date.now() }) {
+ try { ensureAnnouncements(db); } catch { /* exists */ }
+ const key = String(streamerKey).toLowerCase();
+ const cooldownMin = Math.max(1, parseInt(setting(db, 'stream_live_cooldown_min'), 10) || 60);
+ const dailyCap = Math.max(1, parseInt(setting(db, 'stream_live_daily_cap'), 10) || 8);
+ const last = db.prepare('SELECT sent_at FROM stream_live_announcements WHERE streamer_key = ? ORDER BY sent_at DESC, id DESC LIMIT 1').get(key);
+ const today = db.prepare("SELECT COUNT(*) AS c FROM stream_live_announcements WHERE streamer_key = ? AND sent_at > datetime('now','-1 day')").get(key)?.c || 0;
+ const lastMs = last ? new Date(String(last.sent_at).replace(' ', 'T') + 'Z').getTime() : 0;
+ const sinceMin = last ? (now - lastMs) / 60000 : Infinity;
+ if (today >= dailyCap) return { skipped: true, reason: 'daily-cap', announced_today: today, cap: dailyCap };
+ if (!force && sinceMin < cooldownMin) return { skipped: true, reason: 'cooldown', next_allowed_at: new Date(lastMs + cooldownMin * 60000).toISOString() };
+ db.prepare('INSERT INTO stream_live_announcements (streamer_key, stream_id) VALUES (?, ?)').run(key, streamId != null ? String(streamId) : null);
+ db.prepare("DELETE FROM stream_live_announcements WHERE sent_at < datetime('now','-7 days')").run();
+ return { ok: true };
+}
+
+/** People who asked for every go-live on the network (preference category stream_live_all). */
+function allLiveSubscribers(db) {
+ return db.prepare("SELECT user_id FROM notification_preferences WHERE category = 'stream_live_all' AND enabled = 1").all().map(r => r.user_id);
+}
+
+/** The STREAM_LIVE notification (category 'stream' from openvibe-shared, so the person's mute and email choice apply). */
+function streamLiveNotification({ username, displayName, avatarUrl, senderId, stream, url }) {
+ const name = displayName || username;
+ const title = (stream && stream.title) || 'Started streaming';
+ return {
+ type: 'STREAM_LIVE',
+ title: `${name} is live!`,
+ message: title,
+ icon: '🔴',
+ sender_id: senderId || null,
+ sender_name: name,
+ sender_avatar: avatarUrl || null,
+ service: 'live',
+ url: url || `https://openvibe.live/${encodeURIComponent(username)}`,
+ rich_content: {
+ thumbnail: avatarUrl || null,
+ context: {
+ stream_id: stream ? stream.id : null,
+ username,
+ title,
+ protocol: (stream && stream.protocol) || null,
+ ...(stream && stream.event_id ? { event_id: stream.event_id } : {}),
+ },
+ },
+ };
+}
+
+module.exports = { ensureAnnouncements, claimAnnouncement, allLiveSubscribers, streamLiveNotification };
@@ -1,7 +1,8 @@
@@ -1,7 +1,8 @@
'use strict';
'use strict';
// Events → notifications (server/notifications/events-consumer.js): POST /internal/events turns
// 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,
// deals.watch.matched and trade.alert.triggered into inbox notifications for the subscribed person,
-// exactly once, v2 signatures only, respecting their preferences.
+// 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.
// node test/events-consumer.test.js
// node test/events-consumer.test.js
const assert = require('assert');
const assert = require('assert');
const fs = require('fs');
const fs = require('fs');
@@ -9,11 +10,15 @@ const os = require('os');
@@ -9,11 +10,15 @@ const os = require('os');
const path = require('path');
const path = require('path');
const http = require('http');
const http = require('http');
const express = require('express');
const express = require('express');
-const { ids } = require('openvibe-contracts');
+const crypto = require('crypto');
+const { ids, serviceAuth } = require('openvibe-contracts');
const { signDeliveryHeaders } = require('openvibe-sdk/events');
const { signDeliveryHeaders } = require('openvibe-sdk/events');
const { initDb } = require('../server/db/database');
const { initDb } = require('../server/db/database');
const { NotificationService } = require('../server/notifications/notification-service');
const { NotificationService } = require('../server/notifications/notification-service');
const { createEventsConsumer, TOPICS } = require('../server/notifications/events-consumer');
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');
const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-events-consumer-'));
const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-events-consumer-'));
const log = console.log; console.log = () => {};
const log = console.log; console.log = () => {};
@@ -53,7 +58,10 @@ const rows = (userId) => db.prepare('SELECT * FROM notifications WHERE user_id =
@@ -53,7 +58,10 @@ const rows = (userId) => db.prepare('SELECT * FROM notifications WHERE user_id =
(async () => {
(async () => {
// The subscriptions this consumer is meant for, and nothing else.
// The subscriptions this consumer is meant for, and nothing else.
- assert.deepStrictEqual(TOPICS, ['deals.watch.matched', 'trade.alert.triggered']);
+ assert.deepStrictEqual(TOPICS, ['deals.watch.matched', 'trade.alert.triggered', 'live.stream.started']);
+ assert.deepStrictEqual(topicsFrom([]), TOPICS, 'scripts/subscribe-events.js subscribes every topic');
+ assert.deepStrictEqual(topicsFrom(['--topic', 'live.stream.started']), ['live.stream.started']);
+ assert.throws(() => topicsFrom(['--topic', 'live.*']));
let consumer = createEventsConsumer({ db, notifications, secrets: '' });
let consumer = createEventsConsumer({ db, notifications, secrets: '' });
const app = express();
const app = express();
@@ -181,6 +189,140 @@ const rows = (userId) => db.prepare('SELECT * FROM notifications WHERE user_id =
@@ -181,6 +189,140 @@ const rows = (userId) => db.prepare('SELECT * FROM notifications WHERE user_id =
r = await post({ event_id: 'nope', event_type: 'deals.watch.matched' });
r = await post({ event_id: 'nope', event_type: 'deals.watch.matched' });
assert.strictEqual(r.status, 400);
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');
+ 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);
+ notifications.setPreference(23, 'stream', { enabled: false }); // Frank muted go-lives
+ 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));
+ const discordCalls = [];
+ let clock = Date.now();
+ consumer = createEventsConsumer({
+ db, notifications, secrets: SECRET, now: () => clock,
+ liveFollowers: createLiveFollowers({ privateKey, issuer: 'https://openvibe.network', liveUrl: `http://127.0.0.1:${live.address().port}` }),
+ discord: () => ({ sendLiveAlert: async (streamer, stream) => { discordCalls.push([streamer, stream]); return { sent: true }; } }),
+ });
+ const liveEvent = (over = {}, payload = {}) => ({
+ event_id: ids.newId('event'), event_type: 'live.stream.started', version: 1, source: 'live',
+ actor: { type: 'user', id: CAROL }, subject: { type: 'stream', id: '501', revision: 1 }, visibility: 'public', priority: 'important',
+ occurred_at: new Date(clock).toISOString(),
+ payload: {
+ stream_id: 501, channel: { username: 'carol', display_name: 'Carol <3', url: 'https://openvibe.live/carol', subject: { type: 'user', id: CAROL } },
+ title: 'Building a <b>robot</b>', category: null, protocol: 'rtmp', is_nsfw: false, started_at: new Date(clock).toISOString(), ...payload,
+ },
+ ...over,
+ });
+ const goLives = (uid) => db.prepare("SELECT * FROM notifications WHERE user_id = ? AND type = 'STREAM_LIVE' ORDER BY rowid").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 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 });
+ for (const uid of [21, 22, 24]) assert.strictEqual(goLives(uid).length, 1, `user ${uid} is told once`);
+ assert.strictEqual(goLives(23).length, 0, 'the stream category is muted: preferences apply');
+ assert.strictEqual(goLives(20).length, 0, 'the streamer is not told about themself');
+ const n1 = 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');
+ assert.strictEqual(n1.category, 'stream');
+ assert.strictEqual(n1.url, 'https://openvibe.live/carol');
+ 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;
+ 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(goLives(uid).length, 1);
+ assert.strictEqual(discordCalls.length, 1);
+
+ // A new stream row from the same channel within the hour (a reconnect): the announcement window holds.
+ r = await post(liveEvent({}, { stream_id: 501 }));
+ assert.strictEqual(r.body.outcome, 'skipped:cooldown');
+ assert.strictEqual(goLives(21).length, 1);
+ // ...and it is the same window Live's direct POST /internal/events/stream-live claims (keyed by the Network id),
+ // so during the switch the two paths never both announce.
+ assert.strictEqual(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;
+ 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(!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(!db.prepare('SELECT 1 FROM network_event_inbox WHERE event_id = ?').get(v1.event_id));
+ assert.strictEqual(goLives(21).length, 1);
+ liveMode = 'ok';
+ r = await post(v1);
+ assert.strictEqual(r.body.outcome, 'notified');
+ assert.strictEqual(goLives(21).length, 2, 'Dave follows Hank too');
+ assert.strictEqual(goLives(24).length, 2, 'Gina gets every go-live');
+
+ // Refused or ignored without notifying anyone.
+ liveCalls.length = 0;
+ const count = () => db.prepare("SELECT COUNT(*) AS c FROM notifications WHERE type = 'STREAM_LIVE'").get().c;
+ const total = count();
+ r = await post(liveEvent({ source: 'tips' }));
+ assert.strictEqual(r.body.outcome, 'ignored:source', 'only Live speaks for Live channels');
+ // The signatures are made at the consumer's clock (it moves forward below).
+ clock += 45 * 60 * 1000;
+ r = await post(liveEvent({ occurred_at: new Date(clock - 45 * 60 * 1000).toISOString() }, { started_at: new Date(clock - 45 * 60 * 1000).toISOString() }), { now: clock });
+ assert.strictEqual(r.body.outcome, 'ignored:stale', 'a late or replayed start announces nothing');
+ r = await post(liveEvent({}, { channel: { username: 'carol', subject: { type: 'user', id: '20' } } }), { now: clock });
+ 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(count(), total);
+ assert.strictEqual(discordCalls.length, 2);
+
+ live.close();
srv.close();
srv.close();
console.log('events consumer: all checks passed');
console.log('events consumer: all checks passed');
})().catch((err) => { console.error(err); process.exit(1); });
})().catch((err) => { console.error(err); process.exit(1); });