openvibers/openvibe.network Commits
Network: the go-live consumer always delivers (the cutover flag held every go-live since 10-10) (#142)
Changes
@@@ -79,8 +79,6 @@ JWT_PUBLIC_KEY=data/keys/public.pem# Connected services (internal, not proxied through Cloudflare)OV_LIVE_INTERNAL_URL=http://127.0.0.1:3000--# Enable only after Live writes follows to Network and the final Live snapshot is reconciled.--#NETWORK_GO_LIVE_FOLLOWS_READY=1OV_TOOLS_INTERNAL_URL=http://127.0.0.1:4001OV_GAMES_INTERNAL_URL=http://127.0.0.1:8000OV_MEDIA_INTERNAL_URL=http://127.0.0.1:4100@@@ -287,11 +287,10 @@ notifications. Live is never asked. A channel whose subject has no Network accoua follower whose subject has no account is counted as `unresolved`; the streamer never hears aboutthemself; the `stream` category's mute and email choice apply. A start morethan 30 minutes old (a replay, or Events catching up) announces nothing. The per-streamer window (one--announcement an hour, eight a day; `stream_live_cooldown_min`, `stream_live_daily_cap`) is shared with--Live's direct `POST /internal/events/stream-live` (`server/notifications/stream-live.js`), so while Live--still makes that call, whichever arrives first announces and the other is skipped.--Until `NETWORK_GO_LIVE_FOLLOWS_READY=1` is set on Network, Events go-lives are acknowledged as--`ignored:follows_cutover_pending` and Live's direct POST remains the delivery path.++announcement an hour, eight a day; `stream_live_cooldown_min`, `stream_live_daily_cap`) is kept by++`server/notifications/stream-live.js`. This consumer is the only go-live path (compatibility register C-85): Live++publishes `live.stream.started` and makes no direct call. Follows have been Network's since 2026-09-26, and the++cutover flag that held the consumer back is gone.**Project usage.** `tools.usage.recorded` and `events.usage.recorded` (hourly rollups of a developerproject's use, `common.usage-recorded@1`) are kept per project and day (`server/developer/usage.js`,@@@ -160,7 +160,7 @@ function liveStarted(event, { now, maxAgeMs }) {* @param {() => ({ sendLiveAlert(streamer: object, stream: object): Promise<object> }|null)} [o.discord]* @param {{ record(event: object): string }} [o.projectUsage] ../developer/usage.js createProjectUsage()*/--async function createEventsConsumer({ db, notifications, secrets, discord = () => null, moderationAudit = null, projectUsage = null, liveStartedMaxAgeMs = LIVE_STARTED_MAX_AGE_MS, goLiveFollowersReady = process.env.NETWORK_GO_LIVE_FOLLOWS_READY === '1', now = () => Date.now(), log = console }) {++async function createEventsConsumer({ db, notifications, secrets, discord = () => null, moderationAudit = null, projectUsage = 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 = createPgInbox(db, { table: INBOX_TABLE, now });const userBySubject = db.prepare('SELECT id FROM users WHERE subject_id = ?');@@@ -169,8 +169,6 @@ async function createEventsConsumer({ db, notifications, secrets, discord = () =/** live.stream.started inside the inbox transaction: the announcement window, then one notification per person. */async function liveStreamStarted(event) {-- // Live's direct POST remains the delivery path until its Network writes and the final reconciliation are complete.-- if (!goLiveFollowersReady) return 'ignored:follows_cutover_pending';const v = liveStarted(event, { now: now(), maxAgeMs: liveStartedMaxAgeMs });if (typeof v === 'string') return v;const streamer = await streamerBySubject.get(v.subject);@@@ -206,7 +206,7 @@ const rows = async (userId) => await db.prepare('SELECT * FROM notifications WHEconst discordCalls = [];let clock = Date.now();consumer = await createEventsConsumer({-- db, notifications, secrets: SECRET, now: () => clock, goLiveFollowersReady: true,++ db, notifications, secrets: SECRET, now: () => clock,discord: () => ({ sendLiveAlert: async (streamer, stream) => { discordCalls.push([streamer, stream]); return { sent: true }; } }),});const liveEvent = (over = {}, payload = {}) => ({@@@ -221,9 +221,6 @@ const rows = async (userId) => await db.prepare('SELECT * FROM notifications WHE});const goLives = async (uid) => await db.prepare("SELECT * FROM notifications WHERE user_id = ? AND type = 'STREAM_LIVE' ORDER BY seq").all(uid);-- const pending = await createEventsConsumer({ db, notifications, secrets: SECRET, goLiveFollowersReady: false });-- assert.strictEqual((await pending.apply(liveEvent())).outcome, 'ignored:follows_cutover_pending');-- assert.strictEqual((await goLives(21)).length, 0, 'Network does not announce before Live follow writes are cut over');// One delivery: Dave, Erin and Gina (all go-lives) are notified once each; Frank muted the category, Carol// is the streamer, the stranger has no account, Hank unfollowed.@@@ -36,7 +36,7 @@ const follows = require('../server/identity/follows');const principals = require('../server/identity/principals');const SECRET = 's'.repeat(40);const notifications = new NotificationService(db);--const consumer = await createEventsConsumer({ db, notifications, secrets: SECRET, goLiveFollowersReady: true, log: { log() {}, warn() {}, error() {} } });++const consumer = await createEventsConsumer({ db, notifications, secrets: SECRET, log: { log() {}, warn() {}, error() {} } });const app = express();Object.assign(app.locals, { db, config, privateKey: keys.privateKey, publicKey: keys.publicKey });app.use(express.urlencoded({ extended: true }));@@@ -178,7 +178,7 @@ const checkEnvelope = (env, subject) => {const d = getDb();const n = new NotificationService(d);const c = await createEventsConsumer({-- db: d, notifications: n, secrets: SECRET, goLiveFollowersReady: true,++ db: d, notifications: n, secrets: SECRET,discord: () => ({ sendLiveAlert: async (s) => { discord.push(s.username); return { sent: true }; } }),log: { log: (m) => logged.push(m), warn() {}, error() {} },});
@@ -79,8 +79,6 @@ JWT_PUBLIC_KEY=data/keys/public.pem
@@ -79,8 +79,6 @@ JWT_PUBLIC_KEY=data/keys/public.pem
# Connected services (internal, not proxied through Cloudflare)
# Connected services (internal, not proxied through Cloudflare)
OV_LIVE_INTERNAL_URL=http://127.0.0.1:3000
OV_LIVE_INTERNAL_URL=http://127.0.0.1:3000
-# Enable only after Live writes follows to Network and the final Live snapshot is reconciled.
-#NETWORK_GO_LIVE_FOLLOWS_READY=1
OV_TOOLS_INTERNAL_URL=http://127.0.0.1:4001
OV_TOOLS_INTERNAL_URL=http://127.0.0.1:4001
OV_GAMES_INTERNAL_URL=http://127.0.0.1:8000
OV_GAMES_INTERNAL_URL=http://127.0.0.1:8000
OV_MEDIA_INTERNAL_URL=http://127.0.0.1:4100
OV_MEDIA_INTERNAL_URL=http://127.0.0.1:4100
@@ -287,11 +287,10 @@ notifications. Live is never asked. A channel whose subject has no Network accou
@@ -287,11 +287,10 @@ notifications. Live is never asked. A channel whose subject has no Network accou
a follower whose subject has no account is counted as `unresolved`; the streamer never hears about
a follower whose subject has no account is counted as `unresolved`; the streamer never hears about
themself; the `stream` category's mute and email choice apply. A start more
themself; the `stream` category's mute and email choice apply. A start more
than 30 minutes old (a replay, or Events catching up) announces nothing. The per-streamer window (one
than 30 minutes old (a replay, or Events catching up) announces nothing. The per-streamer window (one
-announcement an hour, eight a day; `stream_live_cooldown_min`, `stream_live_daily_cap`) is shared with
-Live's direct `POST /internal/events/stream-live` (`server/notifications/stream-live.js`), so while Live
-still makes that call, whichever arrives first announces and the other is skipped.
-Until `NETWORK_GO_LIVE_FOLLOWS_READY=1` is set on Network, Events go-lives are acknowledged as
-`ignored:follows_cutover_pending` and Live's direct POST remains the delivery path.
+announcement an hour, eight a day; `stream_live_cooldown_min`, `stream_live_daily_cap`) is kept by
+`server/notifications/stream-live.js`. This consumer is the only go-live path (compatibility register C-85): Live
+publishes `live.stream.started` and makes no direct call. Follows have been Network's since 2026-09-26, and the
+cutover flag that held the consumer back is gone.
**Project usage.** `tools.usage.recorded` and `events.usage.recorded` (hourly rollups of a developer
**Project usage.** `tools.usage.recorded` and `events.usage.recorded` (hourly rollups of a developer
project's use, `common.usage-recorded@1`) are kept per project and day (`server/developer/usage.js`,
project's use, `common.usage-recorded@1`) are kept per project and day (`server/developer/usage.js`,
@@ -160,7 +160,7 @@ function liveStarted(event, { now, maxAgeMs }) {
@@ -160,7 +160,7 @@ function liveStarted(event, { now, maxAgeMs }) {
* @param {() => ({ sendLiveAlert(streamer: object, stream: object): Promise<object> }|null)} [o.discord]
* @param {() => ({ sendLiveAlert(streamer: object, stream: object): Promise<object> }|null)} [o.discord]
* @param {{ record(event: object): string }} [o.projectUsage] ../developer/usage.js createProjectUsage()
* @param {{ record(event: object): string }} [o.projectUsage] ../developer/usage.js createProjectUsage()
*/
*/
-async function createEventsConsumer({ db, notifications, secrets, discord = () => null, moderationAudit = null, projectUsage = null, liveStartedMaxAgeMs = LIVE_STARTED_MAX_AGE_MS, goLiveFollowersReady = process.env.NETWORK_GO_LIVE_FOLLOWS_READY === '1', now = () => Date.now(), log = console }) {
+async function createEventsConsumer({ db, notifications, secrets, discord = () => null, moderationAudit = null, projectUsage = 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 = createPgInbox(db, { table: INBOX_TABLE, now });
const inbox = createPgInbox(db, { table: INBOX_TABLE, now });
const userBySubject = db.prepare('SELECT id FROM users WHERE subject_id = ?');
const userBySubject = db.prepare('SELECT id FROM users WHERE subject_id = ?');
@@ -169,8 +169,6 @@ async function createEventsConsumer({ db, notifications, secrets, discord = () =
@@ -169,8 +169,6 @@ async function createEventsConsumer({ db, notifications, secrets, discord = () =
/** live.stream.started inside the inbox transaction: the announcement window, then one notification per person. */
/** live.stream.started inside the inbox transaction: the announcement window, then one notification per person. */
async function liveStreamStarted(event) {
async function liveStreamStarted(event) {
- // Live's direct POST remains the delivery path until its Network writes and the final reconciliation are complete.
- if (!goLiveFollowersReady) return 'ignored:follows_cutover_pending';
const v = liveStarted(event, { now: now(), maxAgeMs: liveStartedMaxAgeMs });
const v = liveStarted(event, { now: now(), maxAgeMs: liveStartedMaxAgeMs });
if (typeof v === 'string') return v;
if (typeof v === 'string') return v;
const streamer = await streamerBySubject.get(v.subject);
const streamer = await streamerBySubject.get(v.subject);
@@ -206,7 +206,7 @@ const rows = async (userId) => await db.prepare('SELECT * FROM notifications WHE
@@ -206,7 +206,7 @@ const rows = async (userId) => await db.prepare('SELECT * FROM notifications WHE
const discordCalls = [];
const discordCalls = [];
let clock = Date.now();
let clock = Date.now();
consumer = await createEventsConsumer({
consumer = await createEventsConsumer({
- db, notifications, secrets: SECRET, now: () => clock, goLiveFollowersReady: true,
+ db, notifications, secrets: SECRET, now: () => clock,
discord: () => ({ sendLiveAlert: async (streamer, stream) => { discordCalls.push([streamer, stream]); return { sent: true }; } }),
discord: () => ({ sendLiveAlert: async (streamer, stream) => { discordCalls.push([streamer, stream]); return { sent: true }; } }),
});
});
const liveEvent = (over = {}, payload = {}) => ({
const liveEvent = (over = {}, payload = {}) => ({
@@ -221,9 +221,6 @@ const rows = async (userId) => await db.prepare('SELECT * FROM notifications WHE
@@ -221,9 +221,6 @@ const rows = async (userId) => await db.prepare('SELECT * FROM notifications WHE
});
});
const goLives = async (uid) => await db.prepare("SELECT * FROM notifications WHERE user_id = ? AND type = 'STREAM_LIVE' ORDER BY seq").all(uid);
const goLives = async (uid) => await db.prepare("SELECT * FROM notifications WHERE user_id = ? AND type = 'STREAM_LIVE' ORDER BY seq").all(uid);
- const pending = await createEventsConsumer({ db, notifications, secrets: SECRET, goLiveFollowersReady: false });
- assert.strictEqual((await pending.apply(liveEvent())).outcome, 'ignored:follows_cutover_pending');
- assert.strictEqual((await goLives(21)).length, 0, 'Network does not announce before Live follow writes are cut over');
// One delivery: Dave, Erin and Gina (all go-lives) are notified once each; Frank muted the category, Carol
// One delivery: Dave, Erin and Gina (all go-lives) are notified once each; Frank muted the category, Carol
// is the streamer, the stranger has no account, Hank unfollowed.
// is the streamer, the stranger has no account, Hank unfollowed.
@@ -36,7 +36,7 @@ const follows = require('../server/identity/follows');
@@ -36,7 +36,7 @@ const follows = require('../server/identity/follows');
const principals = require('../server/identity/principals');
const principals = require('../server/identity/principals');
const SECRET = 's'.repeat(40);
const SECRET = 's'.repeat(40);
const notifications = new NotificationService(db);
const notifications = new NotificationService(db);
-const consumer = await createEventsConsumer({ db, notifications, secrets: SECRET, goLiveFollowersReady: true, log: { log() {}, warn() {}, error() {} } });
+const consumer = await createEventsConsumer({ db, notifications, secrets: SECRET, log: { log() {}, warn() {}, error() {} } });
const app = express();
const app = express();
Object.assign(app.locals, { db, config, privateKey: keys.privateKey, publicKey: keys.publicKey });
Object.assign(app.locals, { db, config, privateKey: keys.privateKey, publicKey: keys.publicKey });
app.use(express.urlencoded({ extended: true }));
app.use(express.urlencoded({ extended: true }));
@@ -178,7 +178,7 @@ const checkEnvelope = (env, subject) => {
@@ -178,7 +178,7 @@ const checkEnvelope = (env, subject) => {
const d = getDb();
const d = getDb();
const n = new NotificationService(d);
const n = new NotificationService(d);
const c = await createEventsConsumer({
const c = await createEventsConsumer({
- db: d, notifications: n, secrets: SECRET, goLiveFollowersReady: true,
+ db: d, notifications: n, secrets: SECRET,
discord: () => ({ sendLiveAlert: async (s) => { discord.push(s.username); return { sent: true }; } }),
discord: () => ({ sendLiveAlert: async (s) => { discord.push(s.username); return { sent: true }; } }),
log: { log: (m) => logged.push(m), warn() {}, error() {} },
log: { log: (m) => logged.push(m), warn() {}, error() {} },
});
});