openvibers/openvibe.network Commits
Network: POST /internal/events/stream-live is gone (go-lives arrive as live.stream.started only, C-85) (#143)
Changes
@@@ -335,7 +335,7 @@ Each go-live's outcome is logged, with counts only:[Events consumer] live.stream.started evt_…: notified {"followers":3,"unresolved":0,"targets":3,"notified":3}```--This is the evidence compatibility register row C-85 waits for. A `notified` line for a real go-live means the consumer, not Live's direct call, told the followers. While Live still makes the direct `POST /internal/events/stream-live`, that call usually claims the per-streamer window first, and the consumer logs `skipped:cooldown`.++The consumer is the only go-live path (compatibility register C-85, closed 2026-10-11): Live publishes `live.stream.started` and makes no direct call, and `POST /internal/events/stream-live` is gone. A reconnect inside the per-streamer window logs `skipped:cooldown`.### Realtime badge (network.notification.created)@@@ -218,100 +218,6 @@ router.post('/coins/transfer', principals.guard('network.coins.transfer', { ownA}});--// ═══════════════════════════════════════════════════════--// Stream-Live Event Handler--// OpenVibe.Live calls this when a stream goes live.--// Handles: Discord alerts (via bot), push notifications to--// 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) {-- return res.status(400).json({ error: 'streamer and stream objects required' });-- }---- const results = { discord: null, notifications: null };---- // ── Per-streamer rate limit (persisted — survives deploys) ───────────────────-- // A streamer flapping off/on (reconnects, restarts, OBS hiccups) must not re-announce-- // 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 claim = await streamLive.claimAnnouncement(await 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 (claim.skipped) {-- console.log(`[StreamLive] ${streamer.username}: cooling down until ${claim.next_allowed_at}`);-- return res.json({ ok: true, ...claim });-- }-- }---- // ── Discord Alert ────────────────────────────────────────-- const discordService = req.app.locals.discordService;-- if (discordService) {-- try {-- results.discord = await discordService.sendLiveAlert(streamer, stream);-- } catch (err) {-- results.discord = { sent: false, reason: 'error', error: err.message };-- }-- }---- // ── Push Notifications to Followers ──────────────────────-- const notifService = req.app.locals.notificationService;-- if (notifService) {-- const db = await getDb(req);-- const notifData = streamLive.streamLiveNotification({-- username: streamer.username, displayName: streamer.display_name, avatarUrl: streamer.avatar_url,-- senderId: streamer.id || null, stream, url: streamLive.channelUrl(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.-- // (Network's own `follows` table is a separate, tiny social graph and `streamer.id`-- // is a Live id — the old lookup below silently addressed the wrong accounts.)-- let followerIds = Array.isArray(follower_network_ids)-- ? follower_network_ids.map(Number).filter(n => Number.isInteger(n) && n > 0).slice(0, 20000)-- : [];-- if (!Array.isArray(follower_network_ids)) {-- // Legacy caller: fall back to Network's graph, resolving the streamer's NETWORK id first.-- const link = streamer.id ? await db.prepare("SELECT user_id FROM linked_accounts WHERE service = 'live' AND service_user_id = ?").get(String(streamer.id)) : null;-- if (link) followerIds = (await db.prepare('SELECT follower_id FROM follows WHERE followed_id = ?').all(link.user_id)).map(r => r.follower_id);-- }-- // sender_id must be the streamer's NETWORK id for dedupe + "who is this" lookups.-- try {-- const link = streamer.id ? await db.prepare("SELECT user_id FROM linked_accounts WHERE service = 'live' AND service_user_id = ?").get(String(streamer.id)) : null;-- if (link) notifData.sender_id = link.user_id;-- // A Live id is not a Network person: no block check against whoever has that Network id.-- else notifData.actor_subject = null;-- } catch { notifData.actor_subject = null; /* keep Live id */ }---- // Find users who opted into "all live" notifications-- const allLiveUserIds = await streamLive.allLiveSubscribers(db);---- // Merge and deduplicate-- const targetIds = [...new Set([...followerIds, ...allLiveUserIds])];---- if (targetIds.length > 0) {-- try {-- const created = await notifService.createBulk(targetIds, notifData);-- results.notifications = { sent: created.length, total: targetIds.length };-- } catch (err) {-- results.notifications = { error: err.message };-- }-- } else {-- results.notifications = { sent: 0, total: 0 };-- }-- }---- res.json({ ok: true, ...results });--});--// ═══════════════════════════════════════════════════════════════// Cross-Service Notification Push// Services call these to create notifications for users without@@@ -15,8 +15,8 @@* The followers are Network's own graph (user_follows, server/identity/follows.js; ADR-030, plan T2):* the active ('channel', <streamer subject>) rows, read in the inbox transaction. Live is never asked.* The announcement window (one per-- * streamer per hour, eight a day) is the same one Live's direct POST /internal/events/stream-live-- * claims (./stream-live.js), so both paths can run during the switch without double notifications.++ * streamer per hour, eight a day) is kept by ./stream-live.js. This consumer is the only go-live path (compatibility++ * register C-85): Live makes no direct call, and POST /internal/events/stream-live is gone.* 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.*@@@ -1,15 +1,9 @@'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 Network's own active user_follows rows-- *-- * Both claim the same persisted per-streamer announcement window (stream_live_announcements, keyed-- * by the streamer's NETWORK user id), so while Live still makes the direct call and Network also-- * consumes the event, whichever arrives first announces and the other is skipped as `cooldown`.++ * "X is live" fan-out for live.stream.started (server/notifications/events-consumer.js), the only way a go-live++ * reaches Network (compatibility register C-85: Live's direct POST /internal/events/stream-live is gone). The++ * followers are Network's own active user_follows rows; the persisted per-streamer announcement window++ * (stream_live_announcements, keyed by the streamer's NETWORK user id) keeps reconnects from re-announcing.*//**@@@ -264,8 +264,7 @@ const rows = async (userId) => await db.prepare('SELECT * FROM notifications WHEr = await post(liveEvent({}, { stream_id: 501 }));assert.strictEqual(r.body.outcome, 'skipped:cooldown');assert.strictEqual((await 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.++ // ...and the window is the per-streamer claim itself (keyed by the Network id).assert.strictEqual((await streamLive.claimAnnouncement(db, { streamerKey: 20, streamId: 501 })).reason, 'cooldown');// v1-only (and any unverifiable) live delivery: refused, not recorded.@@@ -239,11 +239,9 @@ const server = http.createServer(app);assert.strictEqual(tipsClaims.claims.sub, 'svc:tips');assert.deepStrictEqual(tipsClaims.claims.cap, ['chat.event.publish', 'chat.message.send']);-- // Go-live fan-out accepts Live's token (network.notifications.push); a narrower token is refused.-- r = await post('/internal/events/stream-live', {}, { authorization: `Bearer ${full}` });-- assert.strictEqual(r.status, 400, 'guard passed, handler validated the body');-- r = await post('/internal/events/stream-live', {}, { authorization: `Bearer ${creditOnly}` });-- assert.strictEqual(r.status, 403);++ // The direct go-live fan-out is gone (C-85): go-lives arrive as live.stream.started through Events.++ const gone = await fetch(base + '/internal/events/stream-live', { method: 'POST', headers: { 'content-type': 'application/json', authorization: `Bearer ${full}` }, body: '{}' });++ assert.strictEqual(gone.status, 404);// ── The routes Live called with the key (register C-50/C-52) take its token, and only that ──const get = (p, headers) => fetch(base + p, { headers }).then(async x => ({ status: x.status, body: await x.json() }));
@@ -335,7 +335,7 @@ Each go-live's outcome is logged, with counts only:
@@ -335,7 +335,7 @@ Each go-live's outcome is logged, with counts only:
[Events consumer] live.stream.started evt_…: notified {"followers":3,"unresolved":0,"targets":3,"notified":3}
[Events consumer] live.stream.started evt_…: notified {"followers":3,"unresolved":0,"targets":3,"notified":3}
```
```
-This is the evidence compatibility register row C-85 waits for. A `notified` line for a real go-live means the consumer, not Live's direct call, told the followers. While Live still makes the direct `POST /internal/events/stream-live`, that call usually claims the per-streamer window first, and the consumer logs `skipped:cooldown`.
+The consumer is the only go-live path (compatibility register C-85, closed 2026-10-11): Live publishes `live.stream.started` and makes no direct call, and `POST /internal/events/stream-live` is gone. A reconnect inside the per-streamer window logs `skipped:cooldown`.
### Realtime badge (network.notification.created)
### Realtime badge (network.notification.created)
@@ -218,100 +218,6 @@ router.post('/coins/transfer', principals.guard('network.coins.transfer', { ownA
@@ -218,100 +218,6 @@ router.post('/coins/transfer', principals.guard('network.coins.transfer', { ownA
}
}
});
});
-// ═══════════════════════════════════════════════════════
-// Stream-Live Event Handler
-// OpenVibe.Live calls this when a stream goes live.
-// Handles: Discord alerts (via bot), push notifications to
-// 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) {
- return res.status(400).json({ error: 'streamer and stream objects required' });
- }
-
- const results = { discord: null, notifications: null };
-
- // ── Per-streamer rate limit (persisted — survives deploys) ───────────────────
- // A streamer flapping off/on (reconnects, restarts, OBS hiccups) must not re-announce
- // 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 claim = await streamLive.claimAnnouncement(await 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 (claim.skipped) {
- console.log(`[StreamLive] ${streamer.username}: cooling down until ${claim.next_allowed_at}`);
- return res.json({ ok: true, ...claim });
- }
- }
-
- // ── Discord Alert ────────────────────────────────────────
- const discordService = req.app.locals.discordService;
- if (discordService) {
- try {
- results.discord = await discordService.sendLiveAlert(streamer, stream);
- } catch (err) {
- results.discord = { sent: false, reason: 'error', error: err.message };
- }
- }
-
- // ── Push Notifications to Followers ──────────────────────
- const notifService = req.app.locals.notificationService;
- if (notifService) {
- const db = await getDb(req);
- const notifData = streamLive.streamLiveNotification({
- username: streamer.username, displayName: streamer.display_name, avatarUrl: streamer.avatar_url,
- senderId: streamer.id || null, stream, url: streamLive.channelUrl(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.
- // (Network's own `follows` table is a separate, tiny social graph and `streamer.id`
- // is a Live id — the old lookup below silently addressed the wrong accounts.)
- let followerIds = Array.isArray(follower_network_ids)
- ? follower_network_ids.map(Number).filter(n => Number.isInteger(n) && n > 0).slice(0, 20000)
- : [];
- if (!Array.isArray(follower_network_ids)) {
- // Legacy caller: fall back to Network's graph, resolving the streamer's NETWORK id first.
- const link = streamer.id ? await db.prepare("SELECT user_id FROM linked_accounts WHERE service = 'live' AND service_user_id = ?").get(String(streamer.id)) : null;
- if (link) followerIds = (await db.prepare('SELECT follower_id FROM follows WHERE followed_id = ?').all(link.user_id)).map(r => r.follower_id);
- }
- // sender_id must be the streamer's NETWORK id for dedupe + "who is this" lookups.
- try {
- const link = streamer.id ? await db.prepare("SELECT user_id FROM linked_accounts WHERE service = 'live' AND service_user_id = ?").get(String(streamer.id)) : null;
- if (link) notifData.sender_id = link.user_id;
- // A Live id is not a Network person: no block check against whoever has that Network id.
- else notifData.actor_subject = null;
- } catch { notifData.actor_subject = null; /* keep Live id */ }
-
- // Find users who opted into "all live" notifications
- const allLiveUserIds = await streamLive.allLiveSubscribers(db);
-
- // Merge and deduplicate
- const targetIds = [...new Set([...followerIds, ...allLiveUserIds])];
-
- if (targetIds.length > 0) {
- try {
- const created = await notifService.createBulk(targetIds, notifData);
- results.notifications = { sent: created.length, total: targetIds.length };
- } catch (err) {
- results.notifications = { error: err.message };
- }
- } else {
- results.notifications = { sent: 0, total: 0 };
- }
- }
-
- res.json({ ok: true, ...results });
-});
-
// ═══════════════════════════════════════════════════════════════
// ═══════════════════════════════════════════════════════════════
// Cross-Service Notification Push
// Cross-Service Notification Push
// Services call these to create notifications for users without
// Services call these to create notifications for users without
@@ -15,8 +15,8 @@
@@ -15,8 +15,8 @@
* The followers are Network's own graph (user_follows, server/identity/follows.js; ADR-030, plan T2):
* The followers are Network's own graph (user_follows, server/identity/follows.js; ADR-030, plan T2):
* the active ('channel', <streamer subject>) rows, read in the inbox transaction. Live is never asked.
* the active ('channel', <streamer subject>) rows, read in the inbox transaction. Live is never asked.
* The announcement window (one per
* 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.
+ * streamer per hour, eight a day) is kept by ./stream-live.js. This consumer is the only go-live path (compatibility
+ * register C-85): Live makes no direct call, and POST /internal/events/stream-live is gone.
* A started event older than LIVE_STARTED_MAX_AGE (30 min; a replay, or Events catching up after an
* 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.
* outage) announces nothing: the stream may be long over.
*
*
@@ -1,15 +1,9 @@
@@ -1,15 +1,9 @@
'use strict';
'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 Network's own active user_follows rows
- *
- * Both claim the same persisted per-streamer announcement window (stream_live_announcements, keyed
- * by the streamer's NETWORK user id), so while Live still makes the direct call and Network also
- * consumes the event, whichever arrives first announces and the other is skipped as `cooldown`.
+ * "X is live" fan-out for live.stream.started (server/notifications/events-consumer.js), the only way a go-live
+ * reaches Network (compatibility register C-85: Live's direct POST /internal/events/stream-live is gone). The
+ * followers are Network's own active user_follows rows; the persisted per-streamer announcement window
+ * (stream_live_announcements, keyed by the streamer's NETWORK user id) keeps reconnects from re-announcing.
*/
*/
/**
/**
@@ -264,8 +264,7 @@ const rows = async (userId) => await db.prepare('SELECT * FROM notifications WHE
@@ -264,8 +264,7 @@ const rows = async (userId) => await db.prepare('SELECT * FROM notifications WHE
r = await post(liveEvent({}, { stream_id: 501 }));
r = await post(liveEvent({}, { stream_id: 501 }));
assert.strictEqual(r.body.outcome, 'skipped:cooldown');
assert.strictEqual(r.body.outcome, 'skipped:cooldown');
assert.strictEqual((await goLives(21)).length, 1);
assert.strictEqual((await 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.
+ // ...and the window is the per-streamer claim itself (keyed by the Network id).
assert.strictEqual((await streamLive.claimAnnouncement(db, { streamerKey: 20, streamId: 501 })).reason, 'cooldown');
assert.strictEqual((await streamLive.claimAnnouncement(db, { streamerKey: 20, streamId: 501 })).reason, 'cooldown');
// v1-only (and any unverifiable) live delivery: refused, not recorded.
// v1-only (and any unverifiable) live delivery: refused, not recorded.
@@ -239,11 +239,9 @@ const server = http.createServer(app);
@@ -239,11 +239,9 @@ const server = http.createServer(app);
assert.strictEqual(tipsClaims.claims.sub, 'svc:tips');
assert.strictEqual(tipsClaims.claims.sub, 'svc:tips');
assert.deepStrictEqual(tipsClaims.claims.cap, ['chat.event.publish', 'chat.message.send']);
assert.deepStrictEqual(tipsClaims.claims.cap, ['chat.event.publish', 'chat.message.send']);
- // Go-live fan-out accepts Live's token (network.notifications.push); a narrower token is refused.
- r = await post('/internal/events/stream-live', {}, { authorization: `Bearer ${full}` });
- assert.strictEqual(r.status, 400, 'guard passed, handler validated the body');
- r = await post('/internal/events/stream-live', {}, { authorization: `Bearer ${creditOnly}` });
- assert.strictEqual(r.status, 403);
+ // The direct go-live fan-out is gone (C-85): go-lives arrive as live.stream.started through Events.
+ const gone = await fetch(base + '/internal/events/stream-live', { method: 'POST', headers: { 'content-type': 'application/json', authorization: `Bearer ${full}` }, body: '{}' });
+ assert.strictEqual(gone.status, 404);
// ── The routes Live called with the key (register C-50/C-52) take its token, and only that ──
// ── The routes Live called with the key (register C-50/C-52) take its token, and only that ──
const get = (p, headers) => fetch(base + p, { headers }).then(async x => ({ status: x.status, body: await x.json() }));
const get = (p, headers) => fetch(base + p, { headers }).then(async x => ({ status: x.status, body: await x.json() }));