openvibers/openvibe.network Commits
network.user.updated (WS-B task 2, Contracts 0.43.0): triggers on users record every username, display name, picture, colour, role and ban change in its own transaction, whatever code made it; a relay turns them into one event per person with the whole current profile, a growing profile_revision and what changed; guests never; a new account once it has its subject; the first boot announces every existing account once
Changes
@@@ -6,7 +6,7 @@"scripts": {"start": "node server/index.js","dev": "NODE_ENV=development node server/index.js",-- "test": "node test/registry-bootstrap.test.js && node test/nginx-generator-live.test.js && node test/admin-refresh.test.js && node test/history-sso.test.js && node test/sso-check.test.js && node test/fedcm.test.js && node test/domains.test.js && node test/frame.test.js && node test/github-integration.test.js && node test/revocation.test.js && node test/usernames.test.js && node test/avatar.test.js && node test/roadmap-baseline.test.js && node test/identity-subjects.test.js && node test/principals.test.js && node test/modules.test.js && node test/module-events.test.js && node test/registry-ecosystem.test.js && node test/registry-exposure.test.js && node test/registry-manifest-fixture.test.js && node test/staff-claims.test.js && node test/registry-topics.test.js && node test/service-principal.test.js && node test/oauth-pkce.test.js && node test/developer-projects.test.js && node test/developer-defaults.test.js && node test/public-cors.test.js && node test/observability.test.js && node test/status.test.js && node test/events-consumer.test.js && node test/security-session.test.js && node test/security-redirects.test.js && node test/analytics-privacy.test.js && node test/data-ownership.test.js && node test/unknown-paths.test.js && node test/oidc-discovery.test.js && node test/refresh-tokens.test.js && node test/provider-secrets.test.js && node test/first-party-origins.test.js && node test/updates.test.js && node test/static-headers.test.js && node test/library-tags.test.js && node test/moderation-audit.test.js",++ "test": "node test/registry-bootstrap.test.js && node test/nginx-generator-live.test.js && node test/admin-refresh.test.js && node test/history-sso.test.js && node test/sso-check.test.js && node test/fedcm.test.js && node test/domains.test.js && node test/frame.test.js && node test/github-integration.test.js && node test/revocation.test.js && node test/usernames.test.js && node test/avatar.test.js && node test/roadmap-baseline.test.js && node test/identity-subjects.test.js && node test/principals.test.js && node test/modules.test.js && node test/module-events.test.js && node test/registry-ecosystem.test.js && node test/registry-exposure.test.js && node test/registry-manifest-fixture.test.js && node test/staff-claims.test.js && node test/profile-events.test.js && node test/registry-topics.test.js && node test/service-principal.test.js && node test/oauth-pkce.test.js && node test/developer-projects.test.js && node test/developer-defaults.test.js && node test/public-cors.test.js && node test/observability.test.js && node test/status.test.js && node test/events-consumer.test.js && node test/security-session.test.js && node test/security-redirects.test.js && node test/analytics-privacy.test.js && node test/data-ownership.test.js && node test/unknown-paths.test.js && node test/oidc-discovery.test.js && node test/refresh-tokens.test.js && node test/provider-secrets.test.js && node test/first-party-origins.test.js && node test/updates.test.js && node test/static-headers.test.js && node test/library-tags.test.js && node test/moderation-audit.test.js","reset-db": "node server/reset-db.js","grant-admin": "node server/grant-admin.js"},nnew file mode 100644@@@ -0,0 +1,109 @@++'use strict';++/**++ * network.user.updated (Contracts 0.43.0; roadmap WS-B task 2): a person's whole current profile after every++ * change other services may know about: username, display name, picture, colour, role or ban.++ *++ * No write path has to remember it. SQLite triggers on `users` record each change in user_profile_changes++ * within the change's own transaction, so a rename, a role grant, a ban or a profile edit from anywhere++ * (admin routes, the API, an import, a hand-run SQL) is never missed. drain() turns those rows into events:++ * per person, it raises users.profile_revision by one, builds the payload from the row as it is now, puts++ * the envelope in network_event_outbox (relayed to OpenVibe.Events by server/developer/event-relay.js) and++ * deletes the rows, all in one transaction. Several quick changes to one person become one event naming++ * everything that changed. A new account ('created') is announced once it has its subject id, and the++ * first boot with this module announces every existing account once.++ *++ * Anonymous (guest) accounts are never announced. Roles outside the staff map's four are sent as 'user'.++ */++const { ids, validate } = require('openvibe-contracts');++const eventRelay = require('../developer/event-relay');++++const EVENT_TYPE = 'network.user.updated';++const ROLES = ['user', 'streamer', 'global_mod', 'admin'];++const FIELDS = ['username', 'display_name', 'avatar_url', 'profile_color', 'role', 'is_banned'];++const NAME = { is_banned: 'banned' };++++function ensureSchema(db) {++ const cols = new Set(db.prepare('PRAGMA table_info(users)').all().map((c) => c.name));++ const first = !cols.has('profile_revision');++ if (first) db.exec('ALTER TABLE users ADD COLUMN profile_revision INTEGER NOT NULL DEFAULT 0');++ const diff = FIELDS.map((f) => `(CASE WHEN OLD.${f} IS NOT NEW.${f} THEN '${NAME[f] || f} ' ELSE '' END)`).join(' || ');++ const any = FIELDS.map((f) => `OLD.${f} IS NOT NEW.${f}`).join(' OR ');++ db.exec(`++ CREATE TABLE IF NOT EXISTS user_profile_changes (++ id INTEGER PRIMARY KEY AUTOINCREMENT,++ user_id INTEGER NOT NULL,++ changed TEXT NOT NULL,++ at INTEGER NOT NULL++ );++ CREATE INDEX IF NOT EXISTS idx_user_profile_changes_user ON user_profile_changes(user_id);++ CREATE TRIGGER IF NOT EXISTS users_profile_changed AFTER UPDATE OF ${FIELDS.join(', ')} ON users++ WHEN COALESCE(NEW.is_anon, 0) = 0 AND (${any})++ BEGIN INSERT INTO user_profile_changes (user_id, changed, at) VALUES (NEW.id, trim(${diff}), CAST(strftime('%s', 'now') AS INTEGER) * 1000); END;++ CREATE TRIGGER IF NOT EXISTS users_profile_created AFTER INSERT ON users++ WHEN COALESCE(NEW.is_anon, 0) = 0++ BEGIN INSERT INTO user_profile_changes (user_id, changed, at) VALUES (NEW.id, 'created', CAST(strftime('%s', 'now') AS INTEGER) * 1000); END;++ `);++ // The first time: every existing account is announced once ('created'), so consumers can build their++ // projection of everyone, not only of people who change something later.++ if (first) db.prepare("INSERT INTO user_profile_changes (user_id, changed, at) SELECT id, 'created', ? FROM users WHERE COALESCE(is_anon, 0) = 0").run(Date.now());++ eventRelay.writerFor(db);++}++++/** The payload for one users row (its current state). */++function payloadOf(u, revision, changed) {++ return {++ subject: { type: 'user', id: u.subject_id }, network_user_id: u.id, revision,++ username: String(u.username).slice(0, 64), display_name: u.display_name ? String(u.display_name).slice(0, 120) : null,++ avatar_url: u.avatar_url ? String(u.avatar_url).slice(0, 500) : null, profile_color: u.profile_color ? String(u.profile_color).slice(0, 32) : null,++ role: ROLES.includes(u.role) ? u.role : 'user', banned: !!u.is_banned, changed,++ };++}++++/** Turn recorded changes into events. → how many events were queued */++function drain(db, { limit = 500 } = {}) {++ const rows = db.prepare('SELECT id, user_id, changed FROM user_profile_changes ORDER BY id LIMIT ?').all(limit);++ if (!rows.length) return 0;++ const byUser = new Map();++ for (const r of rows) { if (!byUser.has(r.user_id)) byUser.set(r.user_id, []); byUser.get(r.user_id).push(r); }++ let queued = 0;++ for (const [userId, list] of byUser) {++ db.transaction(() => {++ const del = db.prepare('DELETE FROM user_profile_changes WHERE id = ?');++ const u = db.prepare('SELECT * FROM users WHERE id = ?').get(userId);++ // A deleted, guest or not-yet-subject account: nothing to announce (a subject assignment is an update++ // of subject_id, which the created event waits for below).++ if (!u || u.is_anon || !/^usr_[0-9A-HJKMNP-TV-Z]{26}$/.test(String(u.subject_id || ''))) {++ if (!u || u.is_anon) for (const r of list) del.run(r.id);++ return;++ }++ const changed = [...new Set(list.flatMap((r) => r.changed.split(' ').filter(Boolean)))];++ const revision = (Number(u.profile_revision) || 0) + 1;++ db.prepare('UPDATE users SET profile_revision = ? WHERE id = ?').run(revision, u.id);++ const ms = Date.now();++ const env = {++ event_id: ids.newId('event', ms), event_type: EVENT_TYPE, version: 1, source: 'network',++ actor: { type: 'system', id: 'network' }, timestamp: new Date(ms).toISOString(), visibility: 'internal',++ subject: { type: 'user', id: u.subject_id, revision }, payload: payloadOf(u, revision, changed),++ };++ const v = validate('events.event-envelope@1', env);++ const pv = validate('network.user.updated@1', env.payload);++ if (!v.valid || !pv.valid) throw new Error(`profile-events: bad event for user ${u.id}: ${JSON.stringify((v.errors || []).concat(pv.errors || [])).slice(0, 300)}`);++ eventRelay.writerFor(db).enqueue(env);++ for (const r of list) del.run(r.id);++ queued++;++ })();++ }++ if (queued) { const live = eventRelay.outboxFor(db); if (live) live.kick(); }++ return queued;++}++++let timer = null;++function start(db, { intervalMs = 3000, log = console } = {}) {++ if (timer) return;++ ensureSchema(db);++ timer = setInterval(() => { try { drain(db); } catch (err) { log.warn && log.warn('[Profile events]', err.message); } }, intervalMs);++ if (timer.unref) timer.unref();++}++function stop() { if (timer) clearInterval(timer); timer = null; }++++module.exports = { EVENT_TYPE, ensureSchema, drain, start, stop, payloadOf };@@@ -470,6 +470,9 @@ app.use('/api/v1/projects', rateLimit({ windowMs: 60_000, max: 60 }), require('.// Their network.app.* / credential / grant events go to OpenVibe.Events through an outbox when// OV_EVENTS_INTERNAL_URL is set (server/developer/event-relay.js); off otherwise.require('./developer/event-relay').startRelay(db, { eventsUrl: config.eventsInternalUrl, privateKey, issuer: config.jwt.issuer });++// network.user.updated (WS-B task 2): profile, role and ban changes, recorded by triggers on users and relayed++// through the same outbox (server/identity/profile-events.js).++require('./identity/profile-events').start(db);// Notification API (authenticated users)app.use('/api/notifications', createNotificationRoutes(db, notificationService, requireAuth));nnew file mode 100644@@@ -0,0 +1,60 @@++'use strict';++// network.user.updated (WS-B task 2, Contracts 0.43.0): triggers on users record every profile, role and ban++// change in its own transaction, whichever code made it; drain() turns them into one event per person with++// the whole current profile, a growing revision and what changed; guests are never announced; a new account++// is announced once it has its subject; the payload and envelope match the contracts.++const assert = require('assert');++const fs = require('fs');++const os = require('os');++const path = require('path');++const { validate } = require('openvibe-contracts');++const { initDb } = require('../server/db/database');++++const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-profile-events-'));++const log = console.log; console.log = () => {};++const db = initDb(path.join(dir, 'network.db'));++console.log = log;++db.prepare("INSERT INTO users (id, username, password_hash, subject_id) VALUES (5, 'early', 'x', 'usr_01JAB2C3D4E5F6G7H8J9K0MNPE')").run();++const profile = require('../server/identity/profile-events');++profile.ensureSchema(db);++assert.strictEqual(profile.drain(db), 1, 'the first boot announces every existing account once');++profile.ensureSchema(db);++assert.strictEqual(profile.drain(db), 0, 'and only the first');++db.prepare('DELETE FROM network_event_outbox').run();++const events = () => db.prepare("SELECT envelope FROM network_event_outbox ORDER BY id").all().map((r) => JSON.parse(r.envelope)).filter((e) => e.event_type === 'network.user.updated');++const ALEX = 'usr_01JAB2C3D4E5F6G7H8J9K0MNPA';++++db.prepare("INSERT INTO users (id, username, password_hash, subject_id, role, display_name) VALUES (7, 'alex', 'x', ?, 'user', 'Alex')").run(ALEX);++db.prepare("INSERT INTO users (id, username, password_hash, is_anon, anon_number) VALUES (8, 'anon-1', 'x', 1, 1)").run();++assert.strictEqual(profile.drain(db), 1, 'the new account; never the guest');++let ev = events();++assert.strictEqual(ev.length, 1);++assert.ok(validate('events.event-envelope@1', ev[0]).valid);++assert.ok(validate('network.user.updated@1', ev[0].payload).valid, JSON.stringify(validate('network.user.updated@1', ev[0].payload).errors));++assert.deepStrictEqual([ev[0].payload.revision, ev[0].payload.changed, ev[0].subject], [1, ['created'], { type: 'user', id: ALEX, revision: 1 }]);++++// Changes from anywhere (here plain SQL, as an admin route or an import would) are recorded.++db.prepare("UPDATE users SET role = 'global_mod' WHERE id = 7").run();++db.prepare("UPDATE users SET avatar_url = 'https://openvibe.media/avatar/alex', display_name = 'Alex!' WHERE id = 7").run();++db.prepare("UPDATE users SET last_seen = CURRENT_TIMESTAMP WHERE id = 7").run();++assert.strictEqual(profile.drain(db), 1, 'several quick changes become one event');++ev = events();++assert.strictEqual(ev[1].payload.revision, 2);++assert.deepStrictEqual(ev[1].payload.changed.sort(), ['avatar_url', 'display_name', 'role']);++assert.strictEqual(ev[1].payload.role, 'global_mod'); assert.strictEqual(ev[1].payload.display_name, 'Alex!');++assert.strictEqual(profile.drain(db), 0, 'last_seen is not a profile change');++++db.prepare("UPDATE users SET is_banned = 1, username = 'alex2' WHERE id = 7").run();++profile.drain(db);++ev = events();++assert.deepStrictEqual([ev[2].payload.revision, ev[2].payload.banned, ev[2].payload.username], [3, true, 'alex2']);++assert.deepStrictEqual(ev[2].payload.changed.sort(), ['banned', 'username']);++++// A new account without a subject yet waits for one.++db.prepare("INSERT INTO users (id, username, password_hash) VALUES (9, 'nosub', 'x')").run();++assert.strictEqual(profile.drain(db), 0, 'no subject yet: waits');++db.prepare("UPDATE users SET subject_id = 'usr_01JAB2C3D4E5F6G7H8J9K0MNPB' WHERE id = 9").run();++assert.strictEqual(profile.drain(db), 1, 'announced once it has one');++db.close();++fs.rmSync(dir, { recursive: true, force: true });++console.log('profile events: all checks passed');
@@ -6,7 +6,7 @@
@@ -6,7 +6,7 @@
"scripts": {
"scripts": {
"start": "node server/index.js",
"start": "node server/index.js",
"dev": "NODE_ENV=development node server/index.js",
"dev": "NODE_ENV=development node server/index.js",
- "test": "node test/registry-bootstrap.test.js && node test/nginx-generator-live.test.js && node test/admin-refresh.test.js && node test/history-sso.test.js && node test/sso-check.test.js && node test/fedcm.test.js && node test/domains.test.js && node test/frame.test.js && node test/github-integration.test.js && node test/revocation.test.js && node test/usernames.test.js && node test/avatar.test.js && node test/roadmap-baseline.test.js && node test/identity-subjects.test.js && node test/principals.test.js && node test/modules.test.js && node test/module-events.test.js && node test/registry-ecosystem.test.js && node test/registry-exposure.test.js && node test/registry-manifest-fixture.test.js && node test/staff-claims.test.js && node test/registry-topics.test.js && node test/service-principal.test.js && node test/oauth-pkce.test.js && node test/developer-projects.test.js && node test/developer-defaults.test.js && node test/public-cors.test.js && node test/observability.test.js && node test/status.test.js && node test/events-consumer.test.js && node test/security-session.test.js && node test/security-redirects.test.js && node test/analytics-privacy.test.js && node test/data-ownership.test.js && node test/unknown-paths.test.js && node test/oidc-discovery.test.js && node test/refresh-tokens.test.js && node test/provider-secrets.test.js && node test/first-party-origins.test.js && node test/updates.test.js && node test/static-headers.test.js && node test/library-tags.test.js && node test/moderation-audit.test.js",
+ "test": "node test/registry-bootstrap.test.js && node test/nginx-generator-live.test.js && node test/admin-refresh.test.js && node test/history-sso.test.js && node test/sso-check.test.js && node test/fedcm.test.js && node test/domains.test.js && node test/frame.test.js && node test/github-integration.test.js && node test/revocation.test.js && node test/usernames.test.js && node test/avatar.test.js && node test/roadmap-baseline.test.js && node test/identity-subjects.test.js && node test/principals.test.js && node test/modules.test.js && node test/module-events.test.js && node test/registry-ecosystem.test.js && node test/registry-exposure.test.js && node test/registry-manifest-fixture.test.js && node test/staff-claims.test.js && node test/profile-events.test.js && node test/registry-topics.test.js && node test/service-principal.test.js && node test/oauth-pkce.test.js && node test/developer-projects.test.js && node test/developer-defaults.test.js && node test/public-cors.test.js && node test/observability.test.js && node test/status.test.js && node test/events-consumer.test.js && node test/security-session.test.js && node test/security-redirects.test.js && node test/analytics-privacy.test.js && node test/data-ownership.test.js && node test/unknown-paths.test.js && node test/oidc-discovery.test.js && node test/refresh-tokens.test.js && node test/provider-secrets.test.js && node test/first-party-origins.test.js && node test/updates.test.js && node test/static-headers.test.js && node test/library-tags.test.js && node test/moderation-audit.test.js",
"reset-db": "node server/reset-db.js",
"reset-db": "node server/reset-db.js",
"grant-admin": "node server/grant-admin.js"
"grant-admin": "node server/grant-admin.js"
},
},
new file mode 100644
new file mode 100644
@@ -0,0 +1,109 @@
@@ -0,0 +1,109 @@
+'use strict';
+/**
+ * network.user.updated (Contracts 0.43.0; roadmap WS-B task 2): a person's whole current profile after every
+ * change other services may know about: username, display name, picture, colour, role or ban.
+ *
+ * No write path has to remember it. SQLite triggers on `users` record each change in user_profile_changes
+ * within the change's own transaction, so a rename, a role grant, a ban or a profile edit from anywhere
+ * (admin routes, the API, an import, a hand-run SQL) is never missed. drain() turns those rows into events:
+ * per person, it raises users.profile_revision by one, builds the payload from the row as it is now, puts
+ * the envelope in network_event_outbox (relayed to OpenVibe.Events by server/developer/event-relay.js) and
+ * deletes the rows, all in one transaction. Several quick changes to one person become one event naming
+ * everything that changed. A new account ('created') is announced once it has its subject id, and the
+ * first boot with this module announces every existing account once.
+ *
+ * Anonymous (guest) accounts are never announced. Roles outside the staff map's four are sent as 'user'.
+ */
+const { ids, validate } = require('openvibe-contracts');
+const eventRelay = require('../developer/event-relay');
+
+const EVENT_TYPE = 'network.user.updated';
+const ROLES = ['user', 'streamer', 'global_mod', 'admin'];
+const FIELDS = ['username', 'display_name', 'avatar_url', 'profile_color', 'role', 'is_banned'];
+const NAME = { is_banned: 'banned' };
+
+function ensureSchema(db) {
+ const cols = new Set(db.prepare('PRAGMA table_info(users)').all().map((c) => c.name));
+ const first = !cols.has('profile_revision');
+ if (first) db.exec('ALTER TABLE users ADD COLUMN profile_revision INTEGER NOT NULL DEFAULT 0');
+ const diff = FIELDS.map((f) => `(CASE WHEN OLD.${f} IS NOT NEW.${f} THEN '${NAME[f] || f} ' ELSE '' END)`).join(' || ');
+ const any = FIELDS.map((f) => `OLD.${f} IS NOT NEW.${f}`).join(' OR ');
+ db.exec(`
+ CREATE TABLE IF NOT EXISTS user_profile_changes (
+ id INTEGER PRIMARY KEY AUTOINCREMENT,
+ user_id INTEGER NOT NULL,
+ changed TEXT NOT NULL,
+ at INTEGER NOT NULL
+ );
+ CREATE INDEX IF NOT EXISTS idx_user_profile_changes_user ON user_profile_changes(user_id);
+ CREATE TRIGGER IF NOT EXISTS users_profile_changed AFTER UPDATE OF ${FIELDS.join(', ')} ON users
+ WHEN COALESCE(NEW.is_anon, 0) = 0 AND (${any})
+ BEGIN INSERT INTO user_profile_changes (user_id, changed, at) VALUES (NEW.id, trim(${diff}), CAST(strftime('%s', 'now') AS INTEGER) * 1000); END;
+ CREATE TRIGGER IF NOT EXISTS users_profile_created AFTER INSERT ON users
+ WHEN COALESCE(NEW.is_anon, 0) = 0
+ BEGIN INSERT INTO user_profile_changes (user_id, changed, at) VALUES (NEW.id, 'created', CAST(strftime('%s', 'now') AS INTEGER) * 1000); END;
+ `);
+ // The first time: every existing account is announced once ('created'), so consumers can build their
+ // projection of everyone, not only of people who change something later.
+ if (first) db.prepare("INSERT INTO user_profile_changes (user_id, changed, at) SELECT id, 'created', ? FROM users WHERE COALESCE(is_anon, 0) = 0").run(Date.now());
+ eventRelay.writerFor(db);
+}
+
+/** The payload for one users row (its current state). */
+function payloadOf(u, revision, changed) {
+ return {
+ subject: { type: 'user', id: u.subject_id }, network_user_id: u.id, revision,
+ username: String(u.username).slice(0, 64), display_name: u.display_name ? String(u.display_name).slice(0, 120) : null,
+ avatar_url: u.avatar_url ? String(u.avatar_url).slice(0, 500) : null, profile_color: u.profile_color ? String(u.profile_color).slice(0, 32) : null,
+ role: ROLES.includes(u.role) ? u.role : 'user', banned: !!u.is_banned, changed,
+ };
+}
+
+/** Turn recorded changes into events. → how many events were queued */
+function drain(db, { limit = 500 } = {}) {
+ const rows = db.prepare('SELECT id, user_id, changed FROM user_profile_changes ORDER BY id LIMIT ?').all(limit);
+ if (!rows.length) return 0;
+ const byUser = new Map();
+ for (const r of rows) { if (!byUser.has(r.user_id)) byUser.set(r.user_id, []); byUser.get(r.user_id).push(r); }
+ let queued = 0;
+ for (const [userId, list] of byUser) {
+ db.transaction(() => {
+ const del = db.prepare('DELETE FROM user_profile_changes WHERE id = ?');
+ const u = db.prepare('SELECT * FROM users WHERE id = ?').get(userId);
+ // A deleted, guest or not-yet-subject account: nothing to announce (a subject assignment is an update
+ // of subject_id, which the created event waits for below).
+ if (!u || u.is_anon || !/^usr_[0-9A-HJKMNP-TV-Z]{26}$/.test(String(u.subject_id || ''))) {
+ if (!u || u.is_anon) for (const r of list) del.run(r.id);
+ return;
+ }
+ const changed = [...new Set(list.flatMap((r) => r.changed.split(' ').filter(Boolean)))];
+ const revision = (Number(u.profile_revision) || 0) + 1;
+ db.prepare('UPDATE users SET profile_revision = ? WHERE id = ?').run(revision, u.id);
+ const ms = Date.now();
+ const env = {
+ event_id: ids.newId('event', ms), event_type: EVENT_TYPE, version: 1, source: 'network',
+ actor: { type: 'system', id: 'network' }, timestamp: new Date(ms).toISOString(), visibility: 'internal',
+ subject: { type: 'user', id: u.subject_id, revision }, payload: payloadOf(u, revision, changed),
+ };
+ const v = validate('events.event-envelope@1', env);
+ const pv = validate('network.user.updated@1', env.payload);
+ if (!v.valid || !pv.valid) throw new Error(`profile-events: bad event for user ${u.id}: ${JSON.stringify((v.errors || []).concat(pv.errors || [])).slice(0, 300)}`);
+ eventRelay.writerFor(db).enqueue(env);
+ for (const r of list) del.run(r.id);
+ queued++;
+ })();
+ }
+ if (queued) { const live = eventRelay.outboxFor(db); if (live) live.kick(); }
+ return queued;
+}
+
+let timer = null;
+function start(db, { intervalMs = 3000, log = console } = {}) {
+ if (timer) return;
+ ensureSchema(db);
+ timer = setInterval(() => { try { drain(db); } catch (err) { log.warn && log.warn('[Profile events]', err.message); } }, intervalMs);
+ if (timer.unref) timer.unref();
+}
+function stop() { if (timer) clearInterval(timer); timer = null; }
+
+module.exports = { EVENT_TYPE, ensureSchema, drain, start, stop, payloadOf };
@@ -470,6 +470,9 @@ app.use('/api/v1/projects', rateLimit({ windowMs: 60_000, max: 60 }), require('.
@@ -470,6 +470,9 @@ app.use('/api/v1/projects', rateLimit({ windowMs: 60_000, max: 60 }), require('.
// Their network.app.* / credential / grant events go to OpenVibe.Events through an outbox when
// Their network.app.* / credential / grant events go to OpenVibe.Events through an outbox when
// OV_EVENTS_INTERNAL_URL is set (server/developer/event-relay.js); off otherwise.
// OV_EVENTS_INTERNAL_URL is set (server/developer/event-relay.js); off otherwise.
require('./developer/event-relay').startRelay(db, { eventsUrl: config.eventsInternalUrl, privateKey, issuer: config.jwt.issuer });
require('./developer/event-relay').startRelay(db, { eventsUrl: config.eventsInternalUrl, privateKey, issuer: config.jwt.issuer });
+// network.user.updated (WS-B task 2): profile, role and ban changes, recorded by triggers on users and relayed
+// through the same outbox (server/identity/profile-events.js).
+require('./identity/profile-events').start(db);
// Notification API (authenticated users)
// Notification API (authenticated users)
app.use('/api/notifications', createNotificationRoutes(db, notificationService, requireAuth));
app.use('/api/notifications', createNotificationRoutes(db, notificationService, requireAuth));
new file mode 100644
new file mode 100644
@@ -0,0 +1,60 @@
@@ -0,0 +1,60 @@
+'use strict';
+// network.user.updated (WS-B task 2, Contracts 0.43.0): triggers on users record every profile, role and ban
+// change in its own transaction, whichever code made it; drain() turns them into one event per person with
+// the whole current profile, a growing revision and what changed; guests are never announced; a new account
+// is announced once it has its subject; the payload and envelope match the contracts.
+const assert = require('assert');
+const fs = require('fs');
+const os = require('os');
+const path = require('path');
+const { validate } = require('openvibe-contracts');
+const { initDb } = require('../server/db/database');
+
+const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-profile-events-'));
+const log = console.log; console.log = () => {};
+const db = initDb(path.join(dir, 'network.db'));
+console.log = log;
+db.prepare("INSERT INTO users (id, username, password_hash, subject_id) VALUES (5, 'early', 'x', 'usr_01JAB2C3D4E5F6G7H8J9K0MNPE')").run();
+const profile = require('../server/identity/profile-events');
+profile.ensureSchema(db);
+assert.strictEqual(profile.drain(db), 1, 'the first boot announces every existing account once');
+profile.ensureSchema(db);
+assert.strictEqual(profile.drain(db), 0, 'and only the first');
+db.prepare('DELETE FROM network_event_outbox').run();
+const events = () => db.prepare("SELECT envelope FROM network_event_outbox ORDER BY id").all().map((r) => JSON.parse(r.envelope)).filter((e) => e.event_type === 'network.user.updated');
+const ALEX = 'usr_01JAB2C3D4E5F6G7H8J9K0MNPA';
+
+db.prepare("INSERT INTO users (id, username, password_hash, subject_id, role, display_name) VALUES (7, 'alex', 'x', ?, 'user', 'Alex')").run(ALEX);
+db.prepare("INSERT INTO users (id, username, password_hash, is_anon, anon_number) VALUES (8, 'anon-1', 'x', 1, 1)").run();
+assert.strictEqual(profile.drain(db), 1, 'the new account; never the guest');
+let ev = events();
+assert.strictEqual(ev.length, 1);
+assert.ok(validate('events.event-envelope@1', ev[0]).valid);
+assert.ok(validate('network.user.updated@1', ev[0].payload).valid, JSON.stringify(validate('network.user.updated@1', ev[0].payload).errors));
+assert.deepStrictEqual([ev[0].payload.revision, ev[0].payload.changed, ev[0].subject], [1, ['created'], { type: 'user', id: ALEX, revision: 1 }]);
+
+// Changes from anywhere (here plain SQL, as an admin route or an import would) are recorded.
+db.prepare("UPDATE users SET role = 'global_mod' WHERE id = 7").run();
+db.prepare("UPDATE users SET avatar_url = 'https://openvibe.media/avatar/alex', display_name = 'Alex!' WHERE id = 7").run();
+db.prepare("UPDATE users SET last_seen = CURRENT_TIMESTAMP WHERE id = 7").run();
+assert.strictEqual(profile.drain(db), 1, 'several quick changes become one event');
+ev = events();
+assert.strictEqual(ev[1].payload.revision, 2);
+assert.deepStrictEqual(ev[1].payload.changed.sort(), ['avatar_url', 'display_name', 'role']);
+assert.strictEqual(ev[1].payload.role, 'global_mod'); assert.strictEqual(ev[1].payload.display_name, 'Alex!');
+assert.strictEqual(profile.drain(db), 0, 'last_seen is not a profile change');
+
+db.prepare("UPDATE users SET is_banned = 1, username = 'alex2' WHERE id = 7").run();
+profile.drain(db);
+ev = events();
+assert.deepStrictEqual([ev[2].payload.revision, ev[2].payload.banned, ev[2].payload.username], [3, true, 'alex2']);
+assert.deepStrictEqual(ev[2].payload.changed.sort(), ['banned', 'username']);
+
+// A new account without a subject yet waits for one.
+db.prepare("INSERT INTO users (id, username, password_hash) VALUES (9, 'nosub', 'x')").run();
+assert.strictEqual(profile.drain(db), 0, 'no subject yet: waits');
+db.prepare("UPDATE users SET subject_id = 'usr_01JAB2C3D4E5F6G7H8J9K0MNPB' WHERE id = 9").run();
+assert.strictEqual(profile.drain(db), 1, 'announced once it has one');
+db.close();
+fs.rmSync(dir, { recursive: true, force: true });
+console.log('profile events: all checks passed');