openvibers/openvibe.network Commits
The notification badge over Events (roadmap WS-E task 3, WS-F task 1; ADR-005 amendment 2, Contracts 0.61.0): every notification NotificationService.create() stores is announced as network.notification.created, queued in network_event_outbox in the same transaction (both or neither) and relayed to Events: subject the recipient, visibility subject, actor system:network, payload id/type/category/priority/service/created_at/unread_count, never the title, message, link or sender; none for guests or accounts without a usr_ subject. POST /api/v1/realtime/ticket (Bearer or ov_token, 60/min per IP) mints a two-minute RS256 realtime ticket for the signed-in person (iss <issuer>/realtime, sub usr_, aud openvibe.events, typ and purpose realtime, jti rtk_; never a session, never stored or logged); 403 realtime.guest, 503 realtime.disabled with REALTIME_TICKETS=off; OV_EVENTS_PUBLIC_URL overrides the stream origin. The events consumer logs each live.stream.started outcome (counts only), the evidence C-85 waits for. test/notification-events.test.js: the outbox event and its atomicity, ticket mint and refusals, and a restart during a go-live (a fan-out killed mid-transaction leaves nothing; the redelivery after the restart notifies each follower once; a redelivery after the commit is a duplicate). docs/notification-digest.md: the digest design for a later release
Changes
@@@ -165,7 +165,7 @@ sudo apt install certbot python3-certbot-dns-cloudflare1. **Any service** pushes notifications to openvibe.network via `POST /internal/notifications/push`2. **openvibe.network** stores them in SQLite with priority, category, and optional rich content--3. **Clients** poll `GET /api/notifications` every 15 seconds, rendering toasts and updating the bell badge++3. **Clients** poll `GET /api/notifications` every 15 seconds, rendering toasts and updating the bell badge (or, where a site turned realtime on, hear each new notification over OpenVibe.Events and poll every 2 minutes as a safety net: see *Realtime badge* below)4. **Critical notifications** are queued for email delivery via the built-in email service (Resend by default).### Priorities@@@ -250,6 +250,37 @@ The route is inert until the operator does both of these:subscription body, the way Deals, News and Tips do with theirs. It prints the subscription ids,never the secret, and reports an existing identical subscription instead of duplicating it.++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`.++++### Realtime badge (network.notification.created)++++Every notification `NotificationService.create()` stores is announced as **`network.notification.created`** (Contracts 0.61.0; ADR-005 amendment 2). The envelope goes into `network_event_outbox` in the same transaction as the notification: both exist or neither. The relay then publishes it to Events.++- **Envelope:** subject `{ type: user, id: <recipient usr_> }`, visibility `subject` (Events streams it to that person only), actor `system:network`.++- **Payload:** `notification_id`, `type`, `category`, `priority`, `service`, `created_at` and the recipient's `unread_count`. Never the title, message, link or sender.++- **Nothing announced:** a muted category, a blocked sender or a go-live dedupe stores nothing, so it announces nothing. Guests and accounts without a `usr_` subject get no event.++++The badge on other sites cannot use a cookie of events.openvibe.network, so it asks here for a **realtime ticket**:++++```++POST /api/v1/realtime/ticket (Bearer Network JWT, or the ov_token cookie here; 60/min per IP)++→ 200 { ticket, expires_at, expires_in: 120, stream_url, topics: ["network.notification.*"], subject }++```++++It then opens `${stream_url}?topics=network.notification.*&ticket=…[&last_event_id=…]`.++- **The ticket** (`server/auth/realtime-ticket.js`, `identity.realtime-ticket-claims@1`) is an RS256 JWT: `iss <issuer>/realtime`, `sub <usr_>`, `aud [openvibe.events]`, `typ` and `purpose` `realtime`, 120 s, `jti rtk_…`. Events accepts each one once.++- **Never a session:** its issuer, `typ` and audience each rule that out, and this session guard refuses it too.++- **Not stored, logged or audited:** it grants the person only their own stream.++- **Refusals:** a guest gets 403 `realtime.guest`. `REALTIME_TICKETS=off` in `/etc/openvibe/network.env` answers 503 `realtime.disabled`, and every badge stays on polling. `OV_EVENTS_PUBLIC_URL` overrides the stream origin (default `https://events.openvibe.network`).++- **Client:** openvibe-shared `notification-live.js`, on where a site sets `notificationsRealtime: true`.++++A digest (a daily or weekly summary instead of one alert per notification) is designed in [docs/notification-digest.md](docs/notification-digest.md) for a later release.++---## Email Setupnnew file mode 100644@@@ -0,0 +1,42 @@++# Notification digests: design (later release)++++Roadmap WS-E task 3 asks for a digest design now and the digest itself in a later release. Network owns digests (ADR-020: Network keeps addressing, preferences, digests, read state and the store). Nothing here is built yet.++++## Problem++++A person who follows many channels, or watches many deals and alerts, gets one inbox item, toast, push and possibly email per event. The per-streamer window caps go-lives at one an hour and eight a day per streamer, but not per person. Email already has per-person and global daily caps (`email_user_daily_cap`, `email_daily_cap`), and those drop mail silently once reached.++++## Decision (proposed)++++- **A per-category choice.** `notification_preferences` gets a `delivery` column: `instant` (NULL, today's behaviour), `daily` or `weekly`. A person also gets a digest hour and a time zone, kept as a user preference module (`network.notifications`). The default is `instant` everywhere, so nothing changes for anyone who does not choose.++- **What never waits.** Categories `moderation`, `system` and `admin`, and anything `critical`, are always instant: security, bans and account notices. Go-lives may be digested, but the digest says "streamed while you were away", never "is live".++- **Items still land in the inbox.** A notification in a digested category is stored as today, so the badge, the realtime event and the inbox all see it. It is stored with `digest_pending = 1` and without its push, toast, sound or email. `create()` decides this from the preference, next to the mute and block checks.++- **One digest per person per period.** A job (`server/utils/jobs.js` style, every 15 min, off under drill mode) picks people whose digest is due. For each, it claims a row in `notification_digests (user_id, period_end, …)`, unique on `(user_id, period_end)`, in one transaction with the selection of the pending items. It then creates one `DIGEST` notification (category `system`, priority `low`), which is announced as `network.notification.created` like any other. Finally it sends one email through `EmailService` when the person has email on for any digested category.++ - The email's idempotency key is the digest id, so a restart mid-send never sends twice.++ - The claimed items are marked `digest_pending = 0` in the same transaction as the claim.++- **Content at build time.** Items are grouped by category, then service, then sender, newest first. There are at most 50, then "and N more", with a link to the inbox. Items already read or dismissed are left out, and so is anything from a person the recipient has blocked since. When nothing is left, there is no digest.++- **Caps.** The digest email counts once against `email_user_daily_cap`. The instant emails it replaces never count.++++## Events++++No new event type is needed. The digest is a notification, so the badge hears it through `network.notification.created` (type `DIGEST`). A later `network.notification.read` (reads across sites) would join the same `network.notification.*` pattern (ADR-005 amendment 2).++++## Rollout++++1. The schema: an idempotent ADD COLUMN plus the new table. The job stays behind `NOTIFICATION_DIGESTS=on`.++2. The preferences UI on my.openvibe.network, in the Notifications tab: a delivery choice per category, the hour and the time zone.++3. The Shared notification panel shows a `DIGEST` item as a collapsible group.++++To roll back, turn the flag off. Pending items stay in the inbox as ordinary read or unread notifications, and nothing is lost.++++## Tests to write with it++++- A restart during a digest run sends one email and creates one `DIGEST`.++- A muted category, a blocked sender and a read item are left out.++- `critical` and `moderation` items are never delayed.++- The caps hold, and the digest hour falls correctly across a DST change.++++## Open questions++++- Whether the digest should also be a Web Push notification, or email and inbox only.++- Whether per-service digests are wanted, for example Deals weekly and Live daily, beyond the per-category choice.@@@ -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/profile-events.test.js && node test/admin-role-events.test.js && node test/grants-admin.test.js && node test/events-ops.test.js && node test/deploy-drift.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/developer-export-tokens.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 && node test/blocks.test.js && node test/site-config.test.js && node test/perf-budget.test.js && node test/shutdown.test.js && node test/loyalty-policy.test.js && node test/themes-review.test.js && node test/page-checks.test.js && node test/operator-checklist.test.js && node test/release-health.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/admin-role-events.test.js && node test/grants-admin.test.js && node test/events-ops.test.js && node test/deploy-drift.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/developer-export-tokens.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/notification-events.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 && node test/blocks.test.js && node test/site-config.test.js && node test/perf-budget.test.js && node test/shutdown.test.js && node test/loyalty-policy.test.js && node test/themes-review.test.js && node test/page-checks.test.js && node test/operator-checklist.test.js && node test/release-health.test.js","reset-db": "node server/reset-db.js","grant-admin": "node server/grant-admin.js"},nnew file mode 100644@@@ -0,0 +1,81 @@++'use strict';++/**++ * Realtime tickets (Contracts 0.61.0 identity.realtime-ticket-claims@1 and network.realtime-ticket-result@1;++ * ADR-005 amendment 2; roadmap WS-E task 3, WS-F task 1).++ *++ * POST /api/v1/realtime/ticket → 200 { ticket, expires_at, expires_in, stream_url, topics, subject }++ *++ * The notification badge on every OpenVibe site listens for the signed-in person's++ * network.notification.created events on OpenVibe.Events' /realtime/stream. An EventSource cannot send++ * a header, and events.openvibe.network's cookies are third-party on every other site, so the badge asks++ * here (Bearer Network JWT, or the ov_token cookie on openvibe.network) for a ticket and opens++ * `${stream_url}?topics=network.notification.*&ticket=…`. A ticket opens one stream: every reconnect++ * asks for a new one and resumes with last_event_id.++ *++ * A ticket is an RS256 JWT signed with Network's key: iss <issuer>/realtime, sub the person's usr_++ * subject, aud [openvibe.events], typ and purpose realtime, 120 s, jti rtk_<24 hex>. It is never a++ * session anywhere: other services check the session issuer, refuse a token with typ, and are not its++ * audience (Network's own session guard refuses typ too). Events accepts each jti once. The ticket is++ * never stored or logged, so minting one is not audited (it grants the person nothing beyond their own++ * stream); it is rate-limited instead (server/index.js).++ *++ * 401 no session (requireAuth) 403 realtime.guest anonymous (guest) sessions++ * 503 realtime.disabled REALTIME_TICKETS=off: every badge stays on polling++ * 503 realtime.no_subject the account has no usr_ subject yet (retry later)++ */++const crypto = require('crypto');++const express = require('express');++const jwt = require('jsonwebtoken');++++const TTL_S = 120;++const TOPICS = Object.freeze(['network.notification.*']);++const DEFAULT_STREAM_URL = 'https://events.openvibe.network/realtime/stream';++++/** The stream URL browsers use: OV_EVENTS_PUBLIC_URL (an origin, or the stream URL itself) or events.openvibe.network. */++function streamUrlFrom(value) {++ const v = String(value || '').trim().replace(/\/+$/, '');++ if (!v) return DEFAULT_STREAM_URL;++ let u;++ try { u = new URL(v); } catch { return DEFAULT_STREAM_URL; }++ if (!/^https?:$/.test(u.protocol) || u.search || u.hash || u.username || u.password) return DEFAULT_STREAM_URL;++ return /\/realtime\/stream$/.test(u.pathname) ? `${u.origin}${u.pathname}` : `${u.origin}/realtime/stream`;++}++++/** Sign a ticket for `subject` (usr_). → { ticket, claims } */++function mintTicket({ subject, privateKey, issuer, now = Date.now(), ttlS = TTL_S }) {++ if (!/^usr_[0-9A-HJKMNP-TV-Z]{26}$/.test(String(subject))) throw new Error('realtime ticket: subject must be a usr_ id');++ if (!String(privateKey || '').includes('BEGIN')) throw new Error('realtime ticket: an RS256 private key is required');++ const iat = Math.floor(now / 1000);++ const claims = {++ iss: `${String(issuer).replace(/\/+$/, '')}/realtime`, sub: subject, aud: ['openvibe.events'],++ typ: 'realtime', purpose: 'realtime', iat, exp: iat + ttlS, jti: `rtk_${crypto.randomBytes(12).toString('hex')}`,++ };++ return { ticket: jwt.sign(claims, privateKey, { algorithm: 'RS256' }), claims };++}++++/**++ * The router mounted at /api/v1/realtime. `enabled` (REALTIME_TICKETS !== 'off') is read per request++ * from `isEnabled()`, so tests and an operator's env change behave the same.++ */++function router({ db, requireAuth, privateKey, issuer, streamUrl, isEnabled = () => process.env.REALTIME_TICKETS !== 'off', now = () => Date.now() }) {++ const r = express.Router();++ const url = streamUrlFrom(streamUrl);++ r.post('/ticket', requireAuth, (req, res) => {++ res.set('Cache-Control', 'no-store');++ if (!isEnabled()) return res.status(503).json({ code: 'realtime.disabled', error: 'browser realtime is off; poll instead' });++ const user = req.user;++ if (!user || user.is_anon) return res.status(403).json({ code: 'realtime.guest', error: 'sign in with an account to get a realtime ticket' });++ let subject = user.subject_id;++ try { subject = require('../identity/subjects').ensureUserSubject(db, user) || subject; } catch { /* keep what the row has */ }++ if (!/^usr_[0-9A-HJKMNP-TV-Z]{26}$/.test(String(subject || ''))) return res.status(503).json({ code: 'realtime.no_subject', error: 'this account has no subject id yet' });++ if (!String(privateKey || '').includes('BEGIN')) return res.status(503).json({ code: 'realtime.disabled', error: 'no RS256 signing key' });++ const { ticket, claims } = mintTicket({ subject, privateKey, issuer, now: now() });++ res.json({++ ticket, expires_at: new Date(claims.exp * 1000).toISOString(), expires_in: claims.exp - claims.iat,++ stream_url: url, topics: [...TOPICS], subject,++ });++ });++ return r;++}++++module.exports = { router, mintTicket, streamUrlFrom, TTL_S, TOPICS, DEFAULT_STREAM_URL };@@@ -406,6 +406,12 @@ app.get('/internal/integrations/github-token', require('./identity/principals').// (network.blocks.read, service token only). Every change is network.block.changed (server/identity/blocks.js).app.use('/api/v1/me/blocks', rateLimit({ windowMs: 60_000, max: 60 }), require('./identity/blocks').userRouter(requireAuth));app.get('/internal/blocks', require('./identity/principals').guard('network.blocks.read', { legacy: false }), require('./identity/blocks').internalHandler(db));++// Realtime tickets (WS-E task 3, WS-F task 1; ADR-005 amendment 2): the notification badge on any site opens++// OpenVibe.Events' /realtime/stream as the signed-in person with a two-minute, single-use ticket++// (server/auth/realtime-ticket.js). REALTIME_TICKETS=off answers 503 and every badge stays on polling.++app.use('/api/v1/realtime', rateLimit({ windowMs: 60_000, max: 60 }), require('./auth/realtime-ticket').router({++ db, requireAuth, privateKey, issuer: config.jwt.issuer, streamUrl: process.env.OV_EVENTS_PUBLIC_URL,++}));// ── Routes ───────────────────────────────────────────────────// Public key endpoint (services fetch this to verify JWTs).@@@ -276,6 +276,11 @@ function createEventsConsumer({ db, notifications, secrets, liveFollowers = nulllog.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' });}++ // A go-live's outcome is logged (counts only): how an operator sees that the consumer, not Live's++ // direct call, notified a real go-live (compatibility register C-85).++ if (event.event_type === 'live.stream.started' && !out.duplicate && typeof log.log === 'function') {++ log.log(`[Events consumer] live.stream.started ${event.event_id}: ${out.outcome}${out.detail ? ` ${JSON.stringify(out.detail)}` : ''}`);++ }res.status(200).json({ event_id: event.event_id, duplicate: out.duplicate, outcome: out.outcome, ...(out.detail ? { detail: out.detail } : {}) });});@@@ -8,10 +8,27 @@const { v4: uuidv4 } = require('uuid');const { TYPES, PRIORITY, EMAIL_ELIGIBLE_CATEGORIES, EMAIL_DEFAULT_TYPES } = require('openvibe-shared/notifications');++const contracts = require('openvibe-contracts');++const eventRelay = require('../developer/event-relay');// Staff and platform notices: never hidden by a block.const BLOCK_EXEMPT_CATEGORIES = new Set(['moderation', 'system', 'admin']);++/**++ * network.notification.created (Contracts 0.61.0; ADR-005 amendment 2, roadmap WS-E task 3): every stored++ * notification is announced to its person over OpenVibe.Events, so the notification badge on any site++ * updates without polling. The envelope goes into network_event_outbox in the transaction that stores the++ * notification (relayed by server/developer/event-relay.js): both exist or neither. Subject the recipient,++ * visibility subject (Events streams it to that person only), actor system:network (never the sender: a++ * subject event reaches its actor too). The payload is what a badge needs, never what it shows: no title,++ * message, link or sender. Guests and accounts without a usr_ subject get no event.++ */++const NOTIFICATION_EVENT = 'network.notification.created';++const SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;++const PRIORITIES = new Set(['low', 'normal', 'high', 'critical']);++// Checked against its contract once the installed openvibe-contracts knows it (0.61.0+).++const knowsNotificationEvent = (() => { try { return !!contracts.resolve(`${NOTIFICATION_EVENT}@1`); } catch { return false; } })();++class NotificationService {constructor(db) {this.db = db;@@@ -123,6 +140,8 @@ class NotificationService {WHERE r.id = ? AND s.id = ? AND b.active = 1 LIMIT 1`);++ this._recipient = db.prepare('SELECT subject_id, is_anon FROM users WHERE id = ?');++this._newestForUser = db.prepare(`SELECT * FROM notificationsWHERE user_id = ? AND is_dismissed = 0@@@ -157,14 +176,22 @@ class NotificationService {}const richContent = data.rich_content ? JSON.stringify(data.rich_content) : null;---- this._insertNotif.run(-- id, data.user_id, data.type || 'GENERIC', category, priority,-- title, data.message || null, icon,-- data.sender_id || null, data.sender_name || null, data.sender_avatar || null,-- data.service || null, data.url || null, richContent,-- data.expires_at || null,-- );++ const createdAt = new Date().toISOString();++++ // The notification and its network.notification.created event: one transaction.++ let announced = false;++ this.db.transaction(() => {++ this._insertNotif.run(++ id, data.user_id, data.type || 'GENERIC', category, priority,++ title, data.message || null, icon,++ data.sender_id || null, data.sender_name || null, data.sender_avatar || null,++ data.service || null, data.url || null, richContent,++ data.expires_at || null,++ );++ announced = this._announce({ id, userId: data.user_id, type: data.type, category, priority, service: data.service, createdAt });++ })();++ // Wake the relay (it reads after the outermost transaction commits: better-sqlite3 is synchronous).++ if (announced) { const live = eventRelay.outboxFor(this.db); if (live) live.kick(); }// Fire browser push notification (async, non-blocking)try {@@@ -178,7 +205,41 @@ class NotificationService {}).catch(() => {});} catch (_) { /* push module not available */ }-- return { id, user_id: data.user_id, type: data.type, category, priority, title, message: data.message, icon, service: data.service, url: data.url, rich_content: data.rich_content, is_read: 0, created_at: new Date().toISOString() };++ return { id, user_id: data.user_id, type: data.type, category, priority, title, message: data.message, icon, service: data.service, url: data.url, rich_content: data.rich_content, is_read: 0, created_at: createdAt };++ }++++ /**++ * Queue network.notification.created for a notification just inserted (inside its transaction).++ * Returns whether an event was queued. A recipient without a usr_ subject (a guest, an account not yet++ * given one) gets none; an envelope that would not match its contract is skipped with a warning rather++ * than losing the notification (the badge's polling still finds it).++ */++ _announce({ id, userId, type, category, priority, service, createdAt }) {++ const r = this._recipient.get(userId);++ if (!r || r.is_anon || !SUBJECT_RE.test(String(r.subject_id || ''))) return false;++ const payload = {++ notification_id: String(id),++ type: /^[A-Z][A-Z0-9_]{1,63}$/.test(String(type || '')) ? String(type) : 'GENERIC',++ category: /^[a-z][a-z0-9_]{1,31}$/.test(String(category || '')) ? String(category) : 'system',++ priority: PRIORITIES.has(priority) ? priority : 'normal',++ service: /^[a-z][a-z0-9-]{1,39}$/.test(String(service || '')) ? String(service) : null,++ created_at: createdAt,++ unread_count: this._unreadCount.get(userId)?.count || 0,++ };++ const ms = Date.parse(createdAt) || Date.now();++ const env = {++ event_id: contracts.ids.newId('event', ms), event_type: NOTIFICATION_EVENT, version: 1, source: 'network',++ actor: { type: 'system', id: 'network' }, timestamp: createdAt, visibility: 'subject', priority: 'low',++ subject: { type: 'user', id: r.subject_id }, payload,++ };++ const v = contracts.validate('events.event-envelope@1', env);++ const pv = knowsNotificationEvent ? contracts.validate(`${NOTIFICATION_EVENT}@1`, payload) : { valid: true };++ if (!v.valid || !pv.valid) {++ console.warn(`[Notifications] ${NOTIFICATION_EVENT} not queued for notification ${id}: ${JSON.stringify((v.errors || []).concat(pv.errors || [])).slice(0, 200)}`);++ return false;++ }++ eventRelay.writerFor(this.db).enqueue(env);++ return true;}/**@@@ -418,4 +479,4 @@ class NotificationService {}}--module.exports = { NotificationService };++module.exports = { NotificationService, NOTIFICATION_EVENT };nnew file mode 100644@@@ -0,0 +1,263 @@++'use strict';++// The notification badge over OpenVibe.Events (roadmap WS-E task 3, WS-F task 1; ADR-005 amendment 2):++// 1. every stored notification queues network.notification.created in its own transaction (both or++// neither): subject the recipient, visibility subject, actor system:network, no text or sender;++// none for a muted category, a guest or an account without a usr_ subject; the relay publishes it;++// 2. POST /api/v1/realtime/ticket mints a two-minute realtime ticket for the signed-in person, refuses++// guests and signed-out callers, obeys REALTIME_TICKETS=off, and a ticket is never a session;++// 3. a restart during a go-live (ADR-020 acceptance): a fan-out that dies mid-transaction leaves nothing,++// the redelivery after the restart notifies every follower once, and a redelivery after the commit++// notifies nobody again.++// node test/notification-events.test.js++const assert = require('assert');++const fs = require('fs');++const os = require('os');++const path = require('path');++const http = require('http');++const crypto = require('crypto');++const express = require('express');++const jwt = require('jsonwebtoken');++const contracts = require('openvibe-contracts');++const { ids, serviceAuth } = contracts;++const { signDeliveryHeaders } = require('openvibe-sdk/events');++const { initDb } = require('../server/db/database');++const { NotificationService, NOTIFICATION_EVENT } = require('../server/notifications/notification-service');++const { createEventsConsumer } = require('../server/notifications/events-consumer');++const { createLiveFollowers } = require('../server/notifications/live-followers');++const eventRelay = require('../server/developer/event-relay');++const ticketMod = require('../server/auth/realtime-ticket');++const session = require('../server/auth/session');++const { signToken } = require('../server/auth/routes');++++const ISSUER = 'https://openvibe.network';++const { privateKey, publicKey } = crypto.generateKeyPairSync('rsa', { modulusLength: 2048, privateKeyEncoding: { type: 'pkcs8', format: 'pem' }, publicKeyEncoding: { type: 'spki', format: 'pem' } });++const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-notification-events-'));++const DB_PATH = path.join(dir, 'network.db');++const quietly = (f) => { const l = console.log; console.log = () => {}; try { return f(); } finally { console.log = l; } };++let db = quietly(() => initDb(DB_PATH));++++const ALICE = ids.newId('user'), BOB = ids.newId('user'), CAROL = ids.newId('user'), DAVE = ids.newId('user'), ERIN = ids.newId('user');++db.prepare(`INSERT INTO users (id, username, password_hash, subject_id, is_anon) VALUES++ (1, 'alice', 'x', ?, 0), (2, 'bob', 'x', ?, 0), (3, 'guest_1', 'x', NULL, 1), (4, 'nosubject', 'x', NULL, 0),++ (20, 'carol', 'x', ?, 0), (21, 'dave', 'x', ?, 0), (22, 'erin', 'x', ?, 0)`).run(ALICE, BOB, CAROL, DAVE, ERIN);++++const outbox = (d = db) => d.prepare(`SELECT envelope FROM ${eventRelay.TABLE} ORDER BY id`).all().map((r) => JSON.parse(r.envelope)).filter((e) => e.event_type === NOTIFICATION_EVENT);++const PAYLOAD_KEYS = ['category', 'created_at', 'notification_id', 'priority', 'service', 'type', 'unread_count'];++const checkEnvelope = (env, subject) => {++ const v = contracts.validate('events.event-envelope@1', env);++ assert.ok(v.valid, JSON.stringify(v.errors));++ assert.strictEqual(env.source, 'network');++ assert.strictEqual(env.visibility, 'subject', 'Events streams it to its person only');++ assert.deepStrictEqual(env.subject, { type: 'user', id: subject });++ assert.deepStrictEqual(env.actor, { type: 'system', id: 'network' }, 'never the sender: a subject event reaches its actor too');++ assert.deepStrictEqual(Object.keys(env.payload).sort(), PAYLOAD_KEYS, 'what a badge needs, never what it shows');++ try { const pv = contracts.validate(`${NOTIFICATION_EVENT}@1`, env.payload); assert.ok(pv.valid, JSON.stringify(pv.errors)); } catch (err) { if (!/unknown contract/.test(err.message)) throw err; }++};++++(async () => {++ // ── 1. network.notification.created, in the notification's transaction ────────────────────────++ let notifications = new NotificationService(db);++ const n1 = notifications.create({ user_id: 1, type: 'CONTENT_REPLY', category: 'social', title: 'New reply to your comment', message: 'bob replied: secret words', sender_id: 2, sender_name: 'bob', url: 'https://openvibe.live/vod/5#comments', service: 'live' });++ assert.ok(n1);++ let evs = outbox();++ assert.strictEqual(evs.length, 1);++ checkEnvelope(evs[0], ALICE);++ assert.deepStrictEqual({ ...evs[0].payload, created_at: null }, { notification_id: n1.id, type: 'CONTENT_REPLY', category: 'social', priority: 'normal', service: 'live', created_at: null, unread_count: 1 });++ assert.ok(!JSON.stringify(evs[0]).includes('secret words') && !JSON.stringify(evs[0]).includes('bob') && !JSON.stringify(evs[0]).includes(BOB), 'no text, no sender, anywhere in the envelope');++ notifications.create({ user_id: 1, type: 'STREAM_LIVE', title: 'carol is live!', sender_id: 20, service: 'live' });++ evs = outbox();++ assert.strictEqual(evs.length, 2);++ assert.strictEqual(evs[1].payload.unread_count, 2, 'the unread count after the insert');++ assert.strictEqual(evs[1].payload.category, 'stream');++ assert.notStrictEqual(evs[0].event_id, evs[1].event_id);++++ // Nothing stored, nothing announced: a muted category; a blocked sender is the same rule (create() returns null).++ notifications.setPreference(2, 'social', { enabled: false });++ assert.strictEqual(notifications.create({ user_id: 2, type: 'FOLLOW', category: 'social', title: 'x' }), null);++ // Stored but never announced: a guest, an account without a usr_ subject.++ assert.ok(notifications.create({ user_id: 3, type: 'WELCOME', title: 'hi' }));++ assert.ok(notifications.create({ user_id: 4, type: 'WELCOME', title: 'hi' }));++ assert.strictEqual(outbox().length, 2);++ // Odd stored shapes are normalised in the event, never dropped with the notification.++ notifications.create({ user_id: 2, type: 'arena_hot', category: 'Game Stuff', priority: 'urgent', title: 'x', service: 'Live!' });++ evs = outbox();++ assert.strictEqual(evs.length, 3);++ checkEnvelope(evs[2], BOB);++ assert.deepStrictEqual([evs[2].payload.type, evs[2].payload.category, evs[2].payload.priority, evs[2].payload.service], ['GENERIC', 'system', 'normal', null]);++++ // Both or neither: when the event cannot be queued, the notification is not stored either.++ db.exec(`CREATE TRIGGER outbox_down BEFORE INSERT ON ${eventRelay.TABLE} BEGIN SELECT RAISE(ABORT, 'outbox down'); END`);++ const before = db.prepare('SELECT COUNT(*) AS c FROM notifications WHERE user_id = 1').get().c;++ assert.throws(() => notifications.create({ user_id: 1, type: 'WELCOME', title: 'hi' }), /outbox down/);++ assert.strictEqual(db.prepare('SELECT COUNT(*) AS c FROM notifications WHERE user_id = 1').get().c, before, 'no notification without its event');++ db.exec('DROP TRIGGER outbox_down');++ // A bulk fan-out is one transaction with one event per recipient.++ const bulk = notifications.createBulk([1, 2, 21], { type: 'SERVICE_ANNOUNCEMENT', category: 'admin', title: 'Maintenance tonight' });++ assert.strictEqual(bulk.length, 3);++ assert.deepStrictEqual(outbox().slice(-3).map((e) => e.subject.id), [ALICE, BOB, DAVE]);++++ // The relay publishes it to Events like every other Network event.++ const published = [];++ const relay = eventRelay.startRelay(db, {++ eventsUrl: 'http://events.test', privateKey, issuer: ISSUER, autoStart: false, log: { log() {}, warn() {} },++ fetch: async (url, init) => {++ const body = JSON.parse(init.body);++ const list = Array.isArray(body.events) ? body.events : [body];++ published.push(...list);++ return new Response(JSON.stringify({ results: list.map((e, i) => ({ event_id: e.event_id, seq: i + 1 })) }), { status: 201, headers: { 'content-type': 'application/json' } });++ },++ });++ notifications = new NotificationService(db);++ notifications.create({ user_id: 1, type: 'WELCOME', title: 'hi' });++ await relay.flush();++ assert.ok(published.some((e) => e.event_type === NOTIFICATION_EVENT && e.subject.id === ALICE), 'relayed to Events');++ await eventRelay.stopRelay(db);++++ // ── 2. POST /api/v1/realtime/ticket ───────────────────────────────────────────────────────++ const config = { jwt: { issuer: ISSUER, accessTokenExpiry: '1h' } };++ let enabled = true;++ const app = express();++ app.locals.privateKey = privateKey;++ const requireAuth = session.makeRequireAuth(() => ({ db, publicKey, config }), signToken);++ app.use('/api/v1/realtime', ticketMod.router({ db, requireAuth, privateKey, issuer: ISSUER, streamUrl: '', isEnabled: () => enabled }));++ const srv = http.createServer(app);++ await new Promise((r) => srv.listen(0, '127.0.0.1', r));++ const base = `http://127.0.0.1:${srv.address().port}`;++ const mint = async (token, headers = {}) => {++ const r = await fetch(`${base}/api/v1/realtime/ticket`, { method: 'POST', headers: { ...(token ? { Authorization: `Bearer ${token}` } : {}), ...headers } });++ return { status: r.status, body: await r.json().catch(() => null), headers: r.headers };++ };++ const aliceSession = signToken(db.prepare('SELECT * FROM users WHERE id = 1').get(), privateKey, config);++ let r = await mint(aliceSession);++ assert.strictEqual(r.status, 200, JSON.stringify(r.body));++ assert.strictEqual(r.headers.get('cache-control'), 'no-store');++ assert.deepStrictEqual(Object.keys(r.body).sort(), ['expires_at', 'expires_in', 'stream_url', 'subject', 'ticket', 'topics']);++ assert.deepStrictEqual([r.body.expires_in, r.body.stream_url, r.body.topics, r.body.subject], [120, 'https://events.openvibe.network/realtime/stream', ['network.notification.*'], ALICE]);++ try { const v = contracts.validate('network.realtime-ticket-result@1', r.body); assert.ok(v.valid, JSON.stringify(v.errors)); } catch (err) { if (!/unknown contract/.test(err.message)) throw err; }++ // What Events checks (OpenVibe.Events server/auth.js verifyRealtimeTicket).++ const claims = jwt.verify(r.body.ticket, publicKey, { algorithms: ['RS256'], issuer: `${ISSUER}/realtime`, audience: 'openvibe.events' });++ assert.deepStrictEqual(Object.keys(claims).sort(), ['aud', 'exp', 'iat', 'iss', 'jti', 'purpose', 'sub', 'typ']);++ assert.deepStrictEqual([claims.sub, claims.aud, claims.typ, claims.purpose, claims.exp - claims.iat], [ALICE, ['openvibe.events'], 'realtime', 'realtime', 120]);++ assert.match(claims.jti, /^rtk_[0-9a-f]{24}$/);++ assert.strictEqual(jwt.decode(r.body.ticket, { complete: true }).header.alg, 'RS256');++ try { const v = contracts.validate('identity.realtime-ticket-claims@1', claims); assert.ok(v.valid, JSON.stringify(v.errors)); } catch (err) { if (!/unknown contract/.test(err.message)) throw err; }++ const again = await mint(aliceSession);++ assert.notStrictEqual(jwt.decode(again.body.ticket).jti, claims.jti, 'every ticket is its own');++++ // A ticket is never a session: not here, not with the session issuer anyone checks.++ assert.ok(session.verifySession(r.body.ticket, { db, publicKey, config }).error, 'Network\'s session guard refuses it');++ assert.strictEqual((await mint(r.body.ticket)).status, 401, 'no ticket for a ticket');++ assert.throws(() => jwt.verify(r.body.ticket, publicKey, { algorithms: ['RS256'], issuer: ISSUER }), /issuer/, 'services checking the session issuer refuse it');++ assert.throws(() => jwt.verify(r.body.ticket, publicKey, { algorithms: ['RS256'], issuer: `${ISSUER}/realtime`, audience: 'openvibe.live' }), /audience/);++ // Refused: signed out, a guest, a service token, and everyone while the operator has it off.++ assert.strictEqual((await mint(null)).status, 401);++ r = await mint(signToken(db.prepare('SELECT * FROM users WHERE id = 3').get(), privateKey, config));++ assert.deepStrictEqual([r.status, r.body.code], [403, 'realtime.guest']);++ const svc = serviceAuth.signServiceToken({ iss: ISSUER, sub: 'svc:live', actor_type: 'service', aud: ['openvibe.network'], cap: [], iat: Math.floor(Date.now() / 1000), exp: Math.floor(Date.now() / 1000) + 300, jti: 'tok_x' }, privateKey);++ assert.strictEqual((await mint(svc)).status, 401, 'services get no ticket');++ enabled = false;++ r = await mint(aliceSession);++ assert.deepStrictEqual([r.status, r.body.code], [503, 'realtime.disabled']);++ enabled = true;++ // An account that has no subject yet is given one.++ r = await mint(signToken(db.prepare('SELECT * FROM users WHERE id = 4').get(), privateKey, config));++ assert.strictEqual(r.status, 200);++ assert.match(r.body.subject, /^usr_/);++ assert.strictEqual(db.prepare('SELECT subject_id FROM users WHERE id = 4').get().subject_id, r.body.subject);++ // The stream URL: the default, an operator's origin, or the default for anything odd.++ assert.strictEqual(ticketMod.streamUrlFrom(''), 'https://events.openvibe.network/realtime/stream');++ assert.strictEqual(ticketMod.streamUrlFrom('https://events.example.test/'), 'https://events.example.test/realtime/stream');++ assert.strictEqual(ticketMod.streamUrlFrom('https://events.example.test/realtime/stream'), 'https://events.example.test/realtime/stream');++ for (const bad of ['javascript:alert(1)', 'https://x.test/?a=1', 'not a url', 'https://u:[email protected]']) assert.strictEqual(ticketMod.streamUrlFrom(bad), ticketMod.DEFAULT_STREAM_URL, bad);++ assert.throws(() => ticketMod.mintTicket({ subject: '57', privateKey, issuer: ISSUER }), /usr_/);++ srv.close();++++ // ── 3. A restart during a go-live ───────────────────────────────────────────────────────++ // Live's GET /internal/followers: stream 701 is Carol's; Dave and Erin follow her.++ const liveCalls = [];++ const live = http.createServer((req, res) => {++ const u = new URL(req.url, 'http://x');++ liveCalls.push(u.pathname);++ const v = serviceAuth.verifyServiceToken(String(req.headers.authorization || '').slice(7), { publicKey, issuer: ISSUER, audience: 'openvibe.live' });++ res.setHeader('content-type', 'application/json');++ if (!v.ok) { res.statusCode = 401; return res.end('{}'); }++ res.end(JSON.stringify({ stream_id: 701, channel: { subject: CAROL }, followers: [{ subject: DAVE, network_user_id: 21 }, { subject: ERIN, network_user_id: 22 }], next: null }));++ });++ await new Promise((r2) => live.listen(0, '127.0.0.1', r2));++ const SECRET = 's'.repeat(40);++ const logged = []; const discord = [];++ function boot() {++ // A fresh process: a new connection to the same file, new services, nothing in memory.++ const d = quietly(() => initDb(DB_PATH));++ const n = new NotificationService(d);++ const c = createEventsConsumer({++ db: d, notifications: n, secrets: SECRET,++ liveFollowers: createLiveFollowers({ privateKey, issuer: ISSUER, liveUrl: `http://127.0.0.1:${live.address().port}` }),++ discord: () => ({ sendLiveAlert: async (s) => { discord.push(s.username); return { sent: true }; } }),++ log: { log: (m) => logged.push(m), warn() {}, error() {} },++ });++ const a = express();++ a.use('/internal/events', c.router);++ const s = http.createServer(a);++ return new Promise((r2) => s.listen(0, '127.0.0.1', () => r2({ db: d, notifications: n, server: s, url: `http://127.0.0.1:${s.address().port}/internal/events` })));++ }++ const deliver = async (proc, event) => {++ const raw = JSON.stringify({ event, seq: 1 });++ const res = await fetch(proc.url, { method: 'POST', headers: { 'content-type': 'application/json', ...signDeliveryHeaders(raw, SECRET) }, body: raw });++ return { status: res.status, body: await res.json() };++ };++ const stop = (proc) => new Promise((r2) => { proc.server.close(() => { proc.db.close(); r2(); }); });++ const started = {++ event_id: ids.newId('event'), event_type: 'live.stream.started', version: 1, source: 'live',++ actor: { type: 'user', id: CAROL }, subject: { type: 'stream', id: '701', revision: 1 }, visibility: 'public', priority: 'important',++ occurred_at: new Date().toISOString(),++ payload: { stream_id: 701, channel: { username: 'carol', display_name: 'Carol', url: 'https://openvibe.live/@carol', subject: { type: 'user', id: CAROL } }, title: 'Go', category: null, protocol: 'whip', is_nsfw: false, started_at: new Date().toISOString() },++ };++ db.close();++ const goLives = (d) => d.prepare("SELECT user_id, COUNT(*) AS c FROM notifications WHERE type = 'STREAM_LIVE' AND sender_id = 20 AND user_id IN (21, 22) GROUP BY user_id ORDER BY user_id").all().map((x) => [x.user_id, x.c]);++ const goLiveEvents = (d) => outbox(d).filter((e) => [DAVE, ERIN].includes(e.subject.id) && e.payload.type === 'STREAM_LIVE').length;++++ // The process dies in the middle of the fan-out (Dave's notification written, Erin's not): nothing++ // of it survives, not the notification, its event, the announcement window or the inbox claim.++ let proc = await boot();++ const create = proc.notifications.create.bind(proc.notifications);++ let calls = 0;++ proc.notifications.create = (data) => { if (++calls === 2) throw new Error('SIGKILL'); return create(data); };++ let r3 = await deliver(proc, started);++ assert.deepStrictEqual([r3.status, r3.body.code], [500, 'network.event_failed']);++ assert.deepStrictEqual(goLives(proc.db), [], 'the half-done fan-out rolled back');++ assert.strictEqual(goLiveEvents(proc.db), 0);++ assert.ok(!proc.db.prepare('SELECT 1 FROM network_event_inbox WHERE event_id = ?').get(started.event_id));++ const claimed = proc.db.prepare("SELECT 1 FROM sqlite_master WHERE name = 'stream_live_announcements'").get()++ ? proc.db.prepare("SELECT COUNT(*) AS c FROM stream_live_announcements WHERE stream_id = '701'").get().c : 0;++ assert.strictEqual(claimed, 0, 'the announcement window is not spent');++ await stop(proc);++++ // Restarted: Events redelivers, and every follower is told exactly once.++ proc = await boot();++ r3 = await deliver(proc, started);++ assert.strictEqual(r3.status, 200, JSON.stringify(r3.body));++ assert.deepStrictEqual([r3.body.duplicate, r3.body.outcome], [false, 'notified']);++ assert.deepStrictEqual(goLives(proc.db), [[21, 1], [22, 1]]);++ assert.strictEqual(goLiveEvents(proc.db), 2, 'one network.notification.created per follower');++ await new Promise((r2) => setImmediate(r2));++ assert.deepStrictEqual(discord, ['carol'], 'the Discord alert once, after the commit');++ assert.ok(logged.some((m) => m.includes(`live.stream.started ${started.event_id}: notified`)), 'the outcome is in the log (C-85 evidence)');++++ // Restarted again after the commit but before Events saw the 200: the redelivery is a duplicate.++ await stop(proc);++ proc = await boot();++ liveCalls.length = 0;++ r3 = await deliver(proc, started);++ assert.deepStrictEqual([r3.status, r3.body.duplicate, r3.body.outcome], [200, true, null]);++ assert.deepStrictEqual(goLives(proc.db), [[21, 1], [22, 1]], 'nobody is told twice');++ assert.strictEqual(goLiveEvents(proc.db), 2);++ assert.strictEqual(liveCalls.length, 0, 'Live is not asked again');++ await new Promise((r2) => setImmediate(r2));++ assert.deepStrictEqual(discord, ['carol']);++ await stop(proc);++ live.close();++++ fs.rmSync(dir, { recursive: true, force: true });++ console.log('notification events: all checks passed');++})().catch((err) => { console.error(err); process.exit(1); });
@@ -165,7 +165,7 @@ sudo apt install certbot python3-certbot-dns-cloudflare
@@ -165,7 +165,7 @@ sudo apt install certbot python3-certbot-dns-cloudflare
1. **Any service** pushes notifications to openvibe.network via `POST /internal/notifications/push`
1. **Any service** pushes notifications to openvibe.network via `POST /internal/notifications/push`
2. **openvibe.network** stores them in SQLite with priority, category, and optional rich content
2. **openvibe.network** stores them in SQLite with priority, category, and optional rich content
-3. **Clients** poll `GET /api/notifications` every 15 seconds, rendering toasts and updating the bell badge
+3. **Clients** poll `GET /api/notifications` every 15 seconds, rendering toasts and updating the bell badge (or, where a site turned realtime on, hear each new notification over OpenVibe.Events and poll every 2 minutes as a safety net: see *Realtime badge* below)
4. **Critical notifications** are queued for email delivery via the built-in email service (Resend by default).
4. **Critical notifications** are queued for email delivery via the built-in email service (Resend by default).
### Priorities
### Priorities
@@ -250,6 +250,37 @@ The route is inert until the operator does both of these:
@@ -250,6 +250,37 @@ The route is inert until the operator does both of these:
subscription body, the way Deals, News and Tips do with theirs. It prints the subscription ids,
subscription body, the way Deals, News and Tips do with theirs. It prints the subscription ids,
never the secret, and reports an existing identical subscription instead of duplicating it.
never the secret, and reports an existing identical subscription instead of duplicating it.
+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`.
+
+### Realtime badge (network.notification.created)
+
+Every notification `NotificationService.create()` stores is announced as **`network.notification.created`** (Contracts 0.61.0; ADR-005 amendment 2). The envelope goes into `network_event_outbox` in the same transaction as the notification: both exist or neither. The relay then publishes it to Events.
+- **Envelope:** subject `{ type: user, id: <recipient usr_> }`, visibility `subject` (Events streams it to that person only), actor `system:network`.
+- **Payload:** `notification_id`, `type`, `category`, `priority`, `service`, `created_at` and the recipient's `unread_count`. Never the title, message, link or sender.
+- **Nothing announced:** a muted category, a blocked sender or a go-live dedupe stores nothing, so it announces nothing. Guests and accounts without a `usr_` subject get no event.
+
+The badge on other sites cannot use a cookie of events.openvibe.network, so it asks here for a **realtime ticket**:
+
+```
+POST /api/v1/realtime/ticket (Bearer Network JWT, or the ov_token cookie here; 60/min per IP)
+→ 200 { ticket, expires_at, expires_in: 120, stream_url, topics: ["network.notification.*"], subject }
+```
+
+It then opens `${stream_url}?topics=network.notification.*&ticket=…[&last_event_id=…]`.
+- **The ticket** (`server/auth/realtime-ticket.js`, `identity.realtime-ticket-claims@1`) is an RS256 JWT: `iss <issuer>/realtime`, `sub <usr_>`, `aud [openvibe.events]`, `typ` and `purpose` `realtime`, 120 s, `jti rtk_…`. Events accepts each one once.
+- **Never a session:** its issuer, `typ` and audience each rule that out, and this session guard refuses it too.
+- **Not stored, logged or audited:** it grants the person only their own stream.
+- **Refusals:** a guest gets 403 `realtime.guest`. `REALTIME_TICKETS=off` in `/etc/openvibe/network.env` answers 503 `realtime.disabled`, and every badge stays on polling. `OV_EVENTS_PUBLIC_URL` overrides the stream origin (default `https://events.openvibe.network`).
+- **Client:** openvibe-shared `notification-live.js`, on where a site sets `notificationsRealtime: true`.
+
+A digest (a daily or weekly summary instead of one alert per notification) is designed in [docs/notification-digest.md](docs/notification-digest.md) for a later release.
+
---
---
## Email Setup
## Email Setup
new file mode 100644
new file mode 100644
@@ -0,0 +1,42 @@
@@ -0,0 +1,42 @@
+# Notification digests: design (later release)
+
+Roadmap WS-E task 3 asks for a digest design now and the digest itself in a later release. Network owns digests (ADR-020: Network keeps addressing, preferences, digests, read state and the store). Nothing here is built yet.
+
+## Problem
+
+A person who follows many channels, or watches many deals and alerts, gets one inbox item, toast, push and possibly email per event. The per-streamer window caps go-lives at one an hour and eight a day per streamer, but not per person. Email already has per-person and global daily caps (`email_user_daily_cap`, `email_daily_cap`), and those drop mail silently once reached.
+
+## Decision (proposed)
+
+- **A per-category choice.** `notification_preferences` gets a `delivery` column: `instant` (NULL, today's behaviour), `daily` or `weekly`. A person also gets a digest hour and a time zone, kept as a user preference module (`network.notifications`). The default is `instant` everywhere, so nothing changes for anyone who does not choose.
+- **What never waits.** Categories `moderation`, `system` and `admin`, and anything `critical`, are always instant: security, bans and account notices. Go-lives may be digested, but the digest says "streamed while you were away", never "is live".
+- **Items still land in the inbox.** A notification in a digested category is stored as today, so the badge, the realtime event and the inbox all see it. It is stored with `digest_pending = 1` and without its push, toast, sound or email. `create()` decides this from the preference, next to the mute and block checks.
+- **One digest per person per period.** A job (`server/utils/jobs.js` style, every 15 min, off under drill mode) picks people whose digest is due. For each, it claims a row in `notification_digests (user_id, period_end, …)`, unique on `(user_id, period_end)`, in one transaction with the selection of the pending items. It then creates one `DIGEST` notification (category `system`, priority `low`), which is announced as `network.notification.created` like any other. Finally it sends one email through `EmailService` when the person has email on for any digested category.
+ - The email's idempotency key is the digest id, so a restart mid-send never sends twice.
+ - The claimed items are marked `digest_pending = 0` in the same transaction as the claim.
+- **Content at build time.** Items are grouped by category, then service, then sender, newest first. There are at most 50, then "and N more", with a link to the inbox. Items already read or dismissed are left out, and so is anything from a person the recipient has blocked since. When nothing is left, there is no digest.
+- **Caps.** The digest email counts once against `email_user_daily_cap`. The instant emails it replaces never count.
+
+## Events
+
+No new event type is needed. The digest is a notification, so the badge hears it through `network.notification.created` (type `DIGEST`). A later `network.notification.read` (reads across sites) would join the same `network.notification.*` pattern (ADR-005 amendment 2).
+
+## Rollout
+
+1. The schema: an idempotent ADD COLUMN plus the new table. The job stays behind `NOTIFICATION_DIGESTS=on`.
+2. The preferences UI on my.openvibe.network, in the Notifications tab: a delivery choice per category, the hour and the time zone.
+3. The Shared notification panel shows a `DIGEST` item as a collapsible group.
+
+To roll back, turn the flag off. Pending items stay in the inbox as ordinary read or unread notifications, and nothing is lost.
+
+## Tests to write with it
+
+- A restart during a digest run sends one email and creates one `DIGEST`.
+- A muted category, a blocked sender and a read item are left out.
+- `critical` and `moderation` items are never delayed.
+- The caps hold, and the digest hour falls correctly across a DST change.
+
+## Open questions
+
+- Whether the digest should also be a Web Push notification, or email and inbox only.
+- Whether per-service digests are wanted, for example Deals weekly and Live daily, beyond the per-category choice.
@@ -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/profile-events.test.js && node test/admin-role-events.test.js && node test/grants-admin.test.js && node test/events-ops.test.js && node test/deploy-drift.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/developer-export-tokens.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 && node test/blocks.test.js && node test/site-config.test.js && node test/perf-budget.test.js && node test/shutdown.test.js && node test/loyalty-policy.test.js && node test/themes-review.test.js && node test/page-checks.test.js && node test/operator-checklist.test.js && node test/release-health.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/admin-role-events.test.js && node test/grants-admin.test.js && node test/events-ops.test.js && node test/deploy-drift.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/developer-export-tokens.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/notification-events.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 && node test/blocks.test.js && node test/site-config.test.js && node test/perf-budget.test.js && node test/shutdown.test.js && node test/loyalty-policy.test.js && node test/themes-review.test.js && node test/page-checks.test.js && node test/operator-checklist.test.js && node test/release-health.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,81 @@
@@ -0,0 +1,81 @@
+'use strict';
+/**
+ * Realtime tickets (Contracts 0.61.0 identity.realtime-ticket-claims@1 and network.realtime-ticket-result@1;
+ * ADR-005 amendment 2; roadmap WS-E task 3, WS-F task 1).
+ *
+ * POST /api/v1/realtime/ticket → 200 { ticket, expires_at, expires_in, stream_url, topics, subject }
+ *
+ * The notification badge on every OpenVibe site listens for the signed-in person's
+ * network.notification.created events on OpenVibe.Events' /realtime/stream. An EventSource cannot send
+ * a header, and events.openvibe.network's cookies are third-party on every other site, so the badge asks
+ * here (Bearer Network JWT, or the ov_token cookie on openvibe.network) for a ticket and opens
+ * `${stream_url}?topics=network.notification.*&ticket=…`. A ticket opens one stream: every reconnect
+ * asks for a new one and resumes with last_event_id.
+ *
+ * A ticket is an RS256 JWT signed with Network's key: iss <issuer>/realtime, sub the person's usr_
+ * subject, aud [openvibe.events], typ and purpose realtime, 120 s, jti rtk_<24 hex>. It is never a
+ * session anywhere: other services check the session issuer, refuse a token with typ, and are not its
+ * audience (Network's own session guard refuses typ too). Events accepts each jti once. The ticket is
+ * never stored or logged, so minting one is not audited (it grants the person nothing beyond their own
+ * stream); it is rate-limited instead (server/index.js).
+ *
+ * 401 no session (requireAuth) 403 realtime.guest anonymous (guest) sessions
+ * 503 realtime.disabled REALTIME_TICKETS=off: every badge stays on polling
+ * 503 realtime.no_subject the account has no usr_ subject yet (retry later)
+ */
+const crypto = require('crypto');
+const express = require('express');
+const jwt = require('jsonwebtoken');
+
+const TTL_S = 120;
+const TOPICS = Object.freeze(['network.notification.*']);
+const DEFAULT_STREAM_URL = 'https://events.openvibe.network/realtime/stream';
+
+/** The stream URL browsers use: OV_EVENTS_PUBLIC_URL (an origin, or the stream URL itself) or events.openvibe.network. */
+function streamUrlFrom(value) {
+ const v = String(value || '').trim().replace(/\/+$/, '');
+ if (!v) return DEFAULT_STREAM_URL;
+ let u;
+ try { u = new URL(v); } catch { return DEFAULT_STREAM_URL; }
+ if (!/^https?:$/.test(u.protocol) || u.search || u.hash || u.username || u.password) return DEFAULT_STREAM_URL;
+ return /\/realtime\/stream$/.test(u.pathname) ? `${u.origin}${u.pathname}` : `${u.origin}/realtime/stream`;
+}
+
+/** Sign a ticket for `subject` (usr_). → { ticket, claims } */
+function mintTicket({ subject, privateKey, issuer, now = Date.now(), ttlS = TTL_S }) {
+ if (!/^usr_[0-9A-HJKMNP-TV-Z]{26}$/.test(String(subject))) throw new Error('realtime ticket: subject must be a usr_ id');
+ if (!String(privateKey || '').includes('BEGIN')) throw new Error('realtime ticket: an RS256 private key is required');
+ const iat = Math.floor(now / 1000);
+ const claims = {
+ iss: `${String(issuer).replace(/\/+$/, '')}/realtime`, sub: subject, aud: ['openvibe.events'],
+ typ: 'realtime', purpose: 'realtime', iat, exp: iat + ttlS, jti: `rtk_${crypto.randomBytes(12).toString('hex')}`,
+ };
+ return { ticket: jwt.sign(claims, privateKey, { algorithm: 'RS256' }), claims };
+}
+
+/**
+ * The router mounted at /api/v1/realtime. `enabled` (REALTIME_TICKETS !== 'off') is read per request
+ * from `isEnabled()`, so tests and an operator's env change behave the same.
+ */
+function router({ db, requireAuth, privateKey, issuer, streamUrl, isEnabled = () => process.env.REALTIME_TICKETS !== 'off', now = () => Date.now() }) {
+ const r = express.Router();
+ const url = streamUrlFrom(streamUrl);
+ r.post('/ticket', requireAuth, (req, res) => {
+ res.set('Cache-Control', 'no-store');
+ if (!isEnabled()) return res.status(503).json({ code: 'realtime.disabled', error: 'browser realtime is off; poll instead' });
+ const user = req.user;
+ if (!user || user.is_anon) return res.status(403).json({ code: 'realtime.guest', error: 'sign in with an account to get a realtime ticket' });
+ let subject = user.subject_id;
+ try { subject = require('../identity/subjects').ensureUserSubject(db, user) || subject; } catch { /* keep what the row has */ }
+ if (!/^usr_[0-9A-HJKMNP-TV-Z]{26}$/.test(String(subject || ''))) return res.status(503).json({ code: 'realtime.no_subject', error: 'this account has no subject id yet' });
+ if (!String(privateKey || '').includes('BEGIN')) return res.status(503).json({ code: 'realtime.disabled', error: 'no RS256 signing key' });
+ const { ticket, claims } = mintTicket({ subject, privateKey, issuer, now: now() });
+ res.json({
+ ticket, expires_at: new Date(claims.exp * 1000).toISOString(), expires_in: claims.exp - claims.iat,
+ stream_url: url, topics: [...TOPICS], subject,
+ });
+ });
+ return r;
+}
+
+module.exports = { router, mintTicket, streamUrlFrom, TTL_S, TOPICS, DEFAULT_STREAM_URL };
@@ -406,6 +406,12 @@ app.get('/internal/integrations/github-token', require('./identity/principals').
@@ -406,6 +406,12 @@ app.get('/internal/integrations/github-token', require('./identity/principals').
// (network.blocks.read, service token only). Every change is network.block.changed (server/identity/blocks.js).
// (network.blocks.read, service token only). Every change is network.block.changed (server/identity/blocks.js).
app.use('/api/v1/me/blocks', rateLimit({ windowMs: 60_000, max: 60 }), require('./identity/blocks').userRouter(requireAuth));
app.use('/api/v1/me/blocks', rateLimit({ windowMs: 60_000, max: 60 }), require('./identity/blocks').userRouter(requireAuth));
app.get('/internal/blocks', require('./identity/principals').guard('network.blocks.read', { legacy: false }), require('./identity/blocks').internalHandler(db));
app.get('/internal/blocks', require('./identity/principals').guard('network.blocks.read', { legacy: false }), require('./identity/blocks').internalHandler(db));
+// Realtime tickets (WS-E task 3, WS-F task 1; ADR-005 amendment 2): the notification badge on any site opens
+// OpenVibe.Events' /realtime/stream as the signed-in person with a two-minute, single-use ticket
+// (server/auth/realtime-ticket.js). REALTIME_TICKETS=off answers 503 and every badge stays on polling.
+app.use('/api/v1/realtime', rateLimit({ windowMs: 60_000, max: 60 }), require('./auth/realtime-ticket').router({
+ db, requireAuth, privateKey, issuer: config.jwt.issuer, streamUrl: process.env.OV_EVENTS_PUBLIC_URL,
+}));
// ── Routes ───────────────────────────────────────────────────
// ── Routes ───────────────────────────────────────────────────
// Public key endpoint (services fetch this to verify JWTs).
// Public key endpoint (services fetch this to verify JWTs).
@@ -276,6 +276,11 @@ function createEventsConsumer({ db, notifications, secrets, liveFollowers = null
@@ -276,6 +276,11 @@ function createEventsConsumer({ db, notifications, secrets, liveFollowers = null
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' });
}
}
+ // A go-live's outcome is logged (counts only): how an operator sees that the consumer, not Live's
+ // direct call, notified a real go-live (compatibility register C-85).
+ if (event.event_type === 'live.stream.started' && !out.duplicate && typeof log.log === 'function') {
+ log.log(`[Events consumer] live.stream.started ${event.event_id}: ${out.outcome}${out.detail ? ` ${JSON.stringify(out.detail)}` : ''}`);
+ }
res.status(200).json({ event_id: event.event_id, duplicate: out.duplicate, outcome: out.outcome, ...(out.detail ? { detail: out.detail } : {}) });
res.status(200).json({ event_id: event.event_id, duplicate: out.duplicate, outcome: out.outcome, ...(out.detail ? { detail: out.detail } : {}) });
});
});
@@ -8,10 +8,27 @@
@@ -8,10 +8,27 @@
const { v4: uuidv4 } = require('uuid');
const { v4: uuidv4 } = require('uuid');
const { TYPES, PRIORITY, EMAIL_ELIGIBLE_CATEGORIES, EMAIL_DEFAULT_TYPES } = require('openvibe-shared/notifications');
const { TYPES, PRIORITY, EMAIL_ELIGIBLE_CATEGORIES, EMAIL_DEFAULT_TYPES } = require('openvibe-shared/notifications');
+const contracts = require('openvibe-contracts');
+const eventRelay = require('../developer/event-relay');
// Staff and platform notices: never hidden by a block.
// Staff and platform notices: never hidden by a block.
const BLOCK_EXEMPT_CATEGORIES = new Set(['moderation', 'system', 'admin']);
const BLOCK_EXEMPT_CATEGORIES = new Set(['moderation', 'system', 'admin']);
+/**
+ * network.notification.created (Contracts 0.61.0; ADR-005 amendment 2, roadmap WS-E task 3): every stored
+ * notification is announced to its person over OpenVibe.Events, so the notification badge on any site
+ * updates without polling. The envelope goes into network_event_outbox in the transaction that stores the
+ * notification (relayed by server/developer/event-relay.js): both exist or neither. Subject the recipient,
+ * visibility subject (Events streams it to that person only), actor system:network (never the sender: a
+ * subject event reaches its actor too). The payload is what a badge needs, never what it shows: no title,
+ * message, link or sender. Guests and accounts without a usr_ subject get no event.
+ */
+const NOTIFICATION_EVENT = 'network.notification.created';
+const SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
+const PRIORITIES = new Set(['low', 'normal', 'high', 'critical']);
+// Checked against its contract once the installed openvibe-contracts knows it (0.61.0+).
+const knowsNotificationEvent = (() => { try { return !!contracts.resolve(`${NOTIFICATION_EVENT}@1`); } catch { return false; } })();
+
class NotificationService {
class NotificationService {
constructor(db) {
constructor(db) {
this.db = db;
this.db = db;
@@ -123,6 +140,8 @@ class NotificationService {
@@ -123,6 +140,8 @@ class NotificationService {
WHERE r.id = ? AND s.id = ? AND b.active = 1 LIMIT 1
WHERE r.id = ? AND s.id = ? AND b.active = 1 LIMIT 1
`);
`);
+ this._recipient = db.prepare('SELECT subject_id, is_anon FROM users WHERE id = ?');
+
this._newestForUser = db.prepare(`
this._newestForUser = db.prepare(`
SELECT * FROM notifications
SELECT * FROM notifications
WHERE user_id = ? AND is_dismissed = 0
WHERE user_id = ? AND is_dismissed = 0
@@ -157,14 +176,22 @@ class NotificationService {
@@ -157,14 +176,22 @@ class NotificationService {
}
}
const richContent = data.rich_content ? JSON.stringify(data.rich_content) : null;
const richContent = data.rich_content ? JSON.stringify(data.rich_content) : null;
-
- this._insertNotif.run(
- id, data.user_id, data.type || 'GENERIC', category, priority,
- title, data.message || null, icon,
- data.sender_id || null, data.sender_name || null, data.sender_avatar || null,
- data.service || null, data.url || null, richContent,
- data.expires_at || null,
- );
+ const createdAt = new Date().toISOString();
+
+ // The notification and its network.notification.created event: one transaction.
+ let announced = false;
+ this.db.transaction(() => {
+ this._insertNotif.run(
+ id, data.user_id, data.type || 'GENERIC', category, priority,
+ title, data.message || null, icon,
+ data.sender_id || null, data.sender_name || null, data.sender_avatar || null,
+ data.service || null, data.url || null, richContent,
+ data.expires_at || null,
+ );
+ announced = this._announce({ id, userId: data.user_id, type: data.type, category, priority, service: data.service, createdAt });
+ })();
+ // Wake the relay (it reads after the outermost transaction commits: better-sqlite3 is synchronous).
+ if (announced) { const live = eventRelay.outboxFor(this.db); if (live) live.kick(); }
// Fire browser push notification (async, non-blocking)
// Fire browser push notification (async, non-blocking)
try {
try {
@@ -178,7 +205,41 @@ class NotificationService {
@@ -178,7 +205,41 @@ class NotificationService {
}).catch(() => {});
}).catch(() => {});
} catch (_) { /* push module not available */ }
} catch (_) { /* push module not available */ }
- return { id, user_id: data.user_id, type: data.type, category, priority, title, message: data.message, icon, service: data.service, url: data.url, rich_content: data.rich_content, is_read: 0, created_at: new Date().toISOString() };
+ return { id, user_id: data.user_id, type: data.type, category, priority, title, message: data.message, icon, service: data.service, url: data.url, rich_content: data.rich_content, is_read: 0, created_at: createdAt };
+ }
+
+ /**
+ * Queue network.notification.created for a notification just inserted (inside its transaction).
+ * Returns whether an event was queued. A recipient without a usr_ subject (a guest, an account not yet
+ * given one) gets none; an envelope that would not match its contract is skipped with a warning rather
+ * than losing the notification (the badge's polling still finds it).
+ */
+ _announce({ id, userId, type, category, priority, service, createdAt }) {
+ const r = this._recipient.get(userId);
+ if (!r || r.is_anon || !SUBJECT_RE.test(String(r.subject_id || ''))) return false;
+ const payload = {
+ notification_id: String(id),
+ type: /^[A-Z][A-Z0-9_]{1,63}$/.test(String(type || '')) ? String(type) : 'GENERIC',
+ category: /^[a-z][a-z0-9_]{1,31}$/.test(String(category || '')) ? String(category) : 'system',
+ priority: PRIORITIES.has(priority) ? priority : 'normal',
+ service: /^[a-z][a-z0-9-]{1,39}$/.test(String(service || '')) ? String(service) : null,
+ created_at: createdAt,
+ unread_count: this._unreadCount.get(userId)?.count || 0,
+ };
+ const ms = Date.parse(createdAt) || Date.now();
+ const env = {
+ event_id: contracts.ids.newId('event', ms), event_type: NOTIFICATION_EVENT, version: 1, source: 'network',
+ actor: { type: 'system', id: 'network' }, timestamp: createdAt, visibility: 'subject', priority: 'low',
+ subject: { type: 'user', id: r.subject_id }, payload,
+ };
+ const v = contracts.validate('events.event-envelope@1', env);
+ const pv = knowsNotificationEvent ? contracts.validate(`${NOTIFICATION_EVENT}@1`, payload) : { valid: true };
+ if (!v.valid || !pv.valid) {
+ console.warn(`[Notifications] ${NOTIFICATION_EVENT} not queued for notification ${id}: ${JSON.stringify((v.errors || []).concat(pv.errors || [])).slice(0, 200)}`);
+ return false;
+ }
+ eventRelay.writerFor(this.db).enqueue(env);
+ return true;
}
}
/**
/**
@@ -418,4 +479,4 @@ class NotificationService {
@@ -418,4 +479,4 @@ class NotificationService {
}
}
}
}
-module.exports = { NotificationService };
+module.exports = { NotificationService, NOTIFICATION_EVENT };
new file mode 100644
new file mode 100644
@@ -0,0 +1,263 @@
@@ -0,0 +1,263 @@
+'use strict';
+// The notification badge over OpenVibe.Events (roadmap WS-E task 3, WS-F task 1; ADR-005 amendment 2):
+// 1. every stored notification queues network.notification.created in its own transaction (both or
+// neither): subject the recipient, visibility subject, actor system:network, no text or sender;
+// none for a muted category, a guest or an account without a usr_ subject; the relay publishes it;
+// 2. POST /api/v1/realtime/ticket mints a two-minute realtime ticket for the signed-in person, refuses
+// guests and signed-out callers, obeys REALTIME_TICKETS=off, and a ticket is never a session;
+// 3. a restart during a go-live (ADR-020 acceptance): a fan-out that dies mid-transaction leaves nothing,
+// the redelivery after the restart notifies every follower once, and a redelivery after the commit
+// notifies nobody again.
+// node test/notification-events.test.js
+const assert = require('assert');
+const fs = require('fs');
+const os = require('os');
+const path = require('path');
+const http = require('http');
+const crypto = require('crypto');
+const express = require('express');
+const jwt = require('jsonwebtoken');
+const contracts = require('openvibe-contracts');
+const { ids, serviceAuth } = contracts;
+const { signDeliveryHeaders } = require('openvibe-sdk/events');
+const { initDb } = require('../server/db/database');
+const { NotificationService, NOTIFICATION_EVENT } = require('../server/notifications/notification-service');
+const { createEventsConsumer } = require('../server/notifications/events-consumer');
+const { createLiveFollowers } = require('../server/notifications/live-followers');
+const eventRelay = require('../server/developer/event-relay');
+const ticketMod = require('../server/auth/realtime-ticket');
+const session = require('../server/auth/session');
+const { signToken } = require('../server/auth/routes');
+
+const ISSUER = 'https://openvibe.network';
+const { privateKey, publicKey } = crypto.generateKeyPairSync('rsa', { modulusLength: 2048, privateKeyEncoding: { type: 'pkcs8', format: 'pem' }, publicKeyEncoding: { type: 'spki', format: 'pem' } });
+const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-notification-events-'));
+const DB_PATH = path.join(dir, 'network.db');
+const quietly = (f) => { const l = console.log; console.log = () => {}; try { return f(); } finally { console.log = l; } };
+let db = quietly(() => initDb(DB_PATH));
+
+const ALICE = ids.newId('user'), BOB = ids.newId('user'), CAROL = ids.newId('user'), DAVE = ids.newId('user'), ERIN = ids.newId('user');
+db.prepare(`INSERT INTO users (id, username, password_hash, subject_id, is_anon) VALUES
+ (1, 'alice', 'x', ?, 0), (2, 'bob', 'x', ?, 0), (3, 'guest_1', 'x', NULL, 1), (4, 'nosubject', 'x', NULL, 0),
+ (20, 'carol', 'x', ?, 0), (21, 'dave', 'x', ?, 0), (22, 'erin', 'x', ?, 0)`).run(ALICE, BOB, CAROL, DAVE, ERIN);
+
+const outbox = (d = db) => d.prepare(`SELECT envelope FROM ${eventRelay.TABLE} ORDER BY id`).all().map((r) => JSON.parse(r.envelope)).filter((e) => e.event_type === NOTIFICATION_EVENT);
+const PAYLOAD_KEYS = ['category', 'created_at', 'notification_id', 'priority', 'service', 'type', 'unread_count'];
+const checkEnvelope = (env, subject) => {
+ const v = contracts.validate('events.event-envelope@1', env);
+ assert.ok(v.valid, JSON.stringify(v.errors));
+ assert.strictEqual(env.source, 'network');
+ assert.strictEqual(env.visibility, 'subject', 'Events streams it to its person only');
+ assert.deepStrictEqual(env.subject, { type: 'user', id: subject });
+ assert.deepStrictEqual(env.actor, { type: 'system', id: 'network' }, 'never the sender: a subject event reaches its actor too');
+ assert.deepStrictEqual(Object.keys(env.payload).sort(), PAYLOAD_KEYS, 'what a badge needs, never what it shows');
+ try { const pv = contracts.validate(`${NOTIFICATION_EVENT}@1`, env.payload); assert.ok(pv.valid, JSON.stringify(pv.errors)); } catch (err) { if (!/unknown contract/.test(err.message)) throw err; }
+};
+
+(async () => {
+ // ── 1. network.notification.created, in the notification's transaction ────────────────────────
+ let notifications = new NotificationService(db);
+ const n1 = notifications.create({ user_id: 1, type: 'CONTENT_REPLY', category: 'social', title: 'New reply to your comment', message: 'bob replied: secret words', sender_id: 2, sender_name: 'bob', url: 'https://openvibe.live/vod/5#comments', service: 'live' });
+ assert.ok(n1);
+ let evs = outbox();
+ assert.strictEqual(evs.length, 1);
+ checkEnvelope(evs[0], ALICE);
+ assert.deepStrictEqual({ ...evs[0].payload, created_at: null }, { notification_id: n1.id, type: 'CONTENT_REPLY', category: 'social', priority: 'normal', service: 'live', created_at: null, unread_count: 1 });
+ assert.ok(!JSON.stringify(evs[0]).includes('secret words') && !JSON.stringify(evs[0]).includes('bob') && !JSON.stringify(evs[0]).includes(BOB), 'no text, no sender, anywhere in the envelope');
+ notifications.create({ user_id: 1, type: 'STREAM_LIVE', title: 'carol is live!', sender_id: 20, service: 'live' });
+ evs = outbox();
+ assert.strictEqual(evs.length, 2);
+ assert.strictEqual(evs[1].payload.unread_count, 2, 'the unread count after the insert');
+ assert.strictEqual(evs[1].payload.category, 'stream');
+ assert.notStrictEqual(evs[0].event_id, evs[1].event_id);
+
+ // Nothing stored, nothing announced: a muted category; a blocked sender is the same rule (create() returns null).
+ notifications.setPreference(2, 'social', { enabled: false });
+ assert.strictEqual(notifications.create({ user_id: 2, type: 'FOLLOW', category: 'social', title: 'x' }), null);
+ // Stored but never announced: a guest, an account without a usr_ subject.
+ assert.ok(notifications.create({ user_id: 3, type: 'WELCOME', title: 'hi' }));
+ assert.ok(notifications.create({ user_id: 4, type: 'WELCOME', title: 'hi' }));
+ assert.strictEqual(outbox().length, 2);
+ // Odd stored shapes are normalised in the event, never dropped with the notification.
+ notifications.create({ user_id: 2, type: 'arena_hot', category: 'Game Stuff', priority: 'urgent', title: 'x', service: 'Live!' });
+ evs = outbox();
+ assert.strictEqual(evs.length, 3);
+ checkEnvelope(evs[2], BOB);
+ assert.deepStrictEqual([evs[2].payload.type, evs[2].payload.category, evs[2].payload.priority, evs[2].payload.service], ['GENERIC', 'system', 'normal', null]);
+
+ // Both or neither: when the event cannot be queued, the notification is not stored either.
+ db.exec(`CREATE TRIGGER outbox_down BEFORE INSERT ON ${eventRelay.TABLE} BEGIN SELECT RAISE(ABORT, 'outbox down'); END`);
+ const before = db.prepare('SELECT COUNT(*) AS c FROM notifications WHERE user_id = 1').get().c;
+ assert.throws(() => notifications.create({ user_id: 1, type: 'WELCOME', title: 'hi' }), /outbox down/);
+ assert.strictEqual(db.prepare('SELECT COUNT(*) AS c FROM notifications WHERE user_id = 1').get().c, before, 'no notification without its event');
+ db.exec('DROP TRIGGER outbox_down');
+ // A bulk fan-out is one transaction with one event per recipient.
+ const bulk = notifications.createBulk([1, 2, 21], { type: 'SERVICE_ANNOUNCEMENT', category: 'admin', title: 'Maintenance tonight' });
+ assert.strictEqual(bulk.length, 3);
+ assert.deepStrictEqual(outbox().slice(-3).map((e) => e.subject.id), [ALICE, BOB, DAVE]);
+
+ // The relay publishes it to Events like every other Network event.
+ const published = [];
+ const relay = eventRelay.startRelay(db, {
+ eventsUrl: 'http://events.test', privateKey, issuer: ISSUER, autoStart: false, log: { log() {}, warn() {} },
+ fetch: async (url, init) => {
+ const body = JSON.parse(init.body);
+ const list = Array.isArray(body.events) ? body.events : [body];
+ published.push(...list);
+ return new Response(JSON.stringify({ results: list.map((e, i) => ({ event_id: e.event_id, seq: i + 1 })) }), { status: 201, headers: { 'content-type': 'application/json' } });
+ },
+ });
+ notifications = new NotificationService(db);
+ notifications.create({ user_id: 1, type: 'WELCOME', title: 'hi' });
+ await relay.flush();
+ assert.ok(published.some((e) => e.event_type === NOTIFICATION_EVENT && e.subject.id === ALICE), 'relayed to Events');
+ await eventRelay.stopRelay(db);
+
+ // ── 2. POST /api/v1/realtime/ticket ───────────────────────────────────────────────────────
+ const config = { jwt: { issuer: ISSUER, accessTokenExpiry: '1h' } };
+ let enabled = true;
+ const app = express();
+ app.locals.privateKey = privateKey;
+ const requireAuth = session.makeRequireAuth(() => ({ db, publicKey, config }), signToken);
+ app.use('/api/v1/realtime', ticketMod.router({ db, requireAuth, privateKey, issuer: ISSUER, streamUrl: '', isEnabled: () => enabled }));
+ const srv = http.createServer(app);
+ await new Promise((r) => srv.listen(0, '127.0.0.1', r));
+ const base = `http://127.0.0.1:${srv.address().port}`;
+ const mint = async (token, headers = {}) => {
+ const r = await fetch(`${base}/api/v1/realtime/ticket`, { method: 'POST', headers: { ...(token ? { Authorization: `Bearer ${token}` } : {}), ...headers } });
+ return { status: r.status, body: await r.json().catch(() => null), headers: r.headers };
+ };
+ const aliceSession = signToken(db.prepare('SELECT * FROM users WHERE id = 1').get(), privateKey, config);
+ let r = await mint(aliceSession);
+ assert.strictEqual(r.status, 200, JSON.stringify(r.body));
+ assert.strictEqual(r.headers.get('cache-control'), 'no-store');
+ assert.deepStrictEqual(Object.keys(r.body).sort(), ['expires_at', 'expires_in', 'stream_url', 'subject', 'ticket', 'topics']);
+ assert.deepStrictEqual([r.body.expires_in, r.body.stream_url, r.body.topics, r.body.subject], [120, 'https://events.openvibe.network/realtime/stream', ['network.notification.*'], ALICE]);
+ try { const v = contracts.validate('network.realtime-ticket-result@1', r.body); assert.ok(v.valid, JSON.stringify(v.errors)); } catch (err) { if (!/unknown contract/.test(err.message)) throw err; }
+ // What Events checks (OpenVibe.Events server/auth.js verifyRealtimeTicket).
+ const claims = jwt.verify(r.body.ticket, publicKey, { algorithms: ['RS256'], issuer: `${ISSUER}/realtime`, audience: 'openvibe.events' });
+ assert.deepStrictEqual(Object.keys(claims).sort(), ['aud', 'exp', 'iat', 'iss', 'jti', 'purpose', 'sub', 'typ']);
+ assert.deepStrictEqual([claims.sub, claims.aud, claims.typ, claims.purpose, claims.exp - claims.iat], [ALICE, ['openvibe.events'], 'realtime', 'realtime', 120]);
+ assert.match(claims.jti, /^rtk_[0-9a-f]{24}$/);
+ assert.strictEqual(jwt.decode(r.body.ticket, { complete: true }).header.alg, 'RS256');
+ try { const v = contracts.validate('identity.realtime-ticket-claims@1', claims); assert.ok(v.valid, JSON.stringify(v.errors)); } catch (err) { if (!/unknown contract/.test(err.message)) throw err; }
+ const again = await mint(aliceSession);
+ assert.notStrictEqual(jwt.decode(again.body.ticket).jti, claims.jti, 'every ticket is its own');
+
+ // A ticket is never a session: not here, not with the session issuer anyone checks.
+ assert.ok(session.verifySession(r.body.ticket, { db, publicKey, config }).error, 'Network\'s session guard refuses it');
+ assert.strictEqual((await mint(r.body.ticket)).status, 401, 'no ticket for a ticket');
+ assert.throws(() => jwt.verify(r.body.ticket, publicKey, { algorithms: ['RS256'], issuer: ISSUER }), /issuer/, 'services checking the session issuer refuse it');
+ assert.throws(() => jwt.verify(r.body.ticket, publicKey, { algorithms: ['RS256'], issuer: `${ISSUER}/realtime`, audience: 'openvibe.live' }), /audience/);
+ // Refused: signed out, a guest, a service token, and everyone while the operator has it off.
+ assert.strictEqual((await mint(null)).status, 401);
+ r = await mint(signToken(db.prepare('SELECT * FROM users WHERE id = 3').get(), privateKey, config));
+ assert.deepStrictEqual([r.status, r.body.code], [403, 'realtime.guest']);
+ const svc = serviceAuth.signServiceToken({ iss: ISSUER, sub: 'svc:live', actor_type: 'service', aud: ['openvibe.network'], cap: [], iat: Math.floor(Date.now() / 1000), exp: Math.floor(Date.now() / 1000) + 300, jti: 'tok_x' }, privateKey);
+ assert.strictEqual((await mint(svc)).status, 401, 'services get no ticket');
+ enabled = false;
+ r = await mint(aliceSession);
+ assert.deepStrictEqual([r.status, r.body.code], [503, 'realtime.disabled']);
+ enabled = true;
+ // An account that has no subject yet is given one.
+ r = await mint(signToken(db.prepare('SELECT * FROM users WHERE id = 4').get(), privateKey, config));
+ assert.strictEqual(r.status, 200);
+ assert.match(r.body.subject, /^usr_/);
+ assert.strictEqual(db.prepare('SELECT subject_id FROM users WHERE id = 4').get().subject_id, r.body.subject);
+ // The stream URL: the default, an operator's origin, or the default for anything odd.
+ assert.strictEqual(ticketMod.streamUrlFrom(''), 'https://events.openvibe.network/realtime/stream');
+ assert.strictEqual(ticketMod.streamUrlFrom('https://events.example.test/'), 'https://events.example.test/realtime/stream');
+ assert.strictEqual(ticketMod.streamUrlFrom('https://events.example.test/realtime/stream'), 'https://events.example.test/realtime/stream');
+ for (const bad of ['javascript:alert(1)', 'https://x.test/?a=1', 'not a url', 'https://u:[email protected]']) assert.strictEqual(ticketMod.streamUrlFrom(bad), ticketMod.DEFAULT_STREAM_URL, bad);
+ assert.throws(() => ticketMod.mintTicket({ subject: '57', privateKey, issuer: ISSUER }), /usr_/);
+ srv.close();
+
+ // ── 3. A restart during a go-live ───────────────────────────────────────────────────────
+ // Live's GET /internal/followers: stream 701 is Carol's; Dave and Erin follow her.
+ const liveCalls = [];
+ const live = http.createServer((req, res) => {
+ const u = new URL(req.url, 'http://x');
+ liveCalls.push(u.pathname);
+ const v = serviceAuth.verifyServiceToken(String(req.headers.authorization || '').slice(7), { publicKey, issuer: ISSUER, audience: 'openvibe.live' });
+ res.setHeader('content-type', 'application/json');
+ if (!v.ok) { res.statusCode = 401; return res.end('{}'); }
+ res.end(JSON.stringify({ stream_id: 701, channel: { subject: CAROL }, followers: [{ subject: DAVE, network_user_id: 21 }, { subject: ERIN, network_user_id: 22 }], next: null }));
+ });
+ await new Promise((r2) => live.listen(0, '127.0.0.1', r2));
+ const SECRET = 's'.repeat(40);
+ const logged = []; const discord = [];
+ function boot() {
+ // A fresh process: a new connection to the same file, new services, nothing in memory.
+ const d = quietly(() => initDb(DB_PATH));
+ const n = new NotificationService(d);
+ const c = createEventsConsumer({
+ db: d, notifications: n, secrets: SECRET,
+ liveFollowers: createLiveFollowers({ privateKey, issuer: ISSUER, liveUrl: `http://127.0.0.1:${live.address().port}` }),
+ discord: () => ({ sendLiveAlert: async (s) => { discord.push(s.username); return { sent: true }; } }),
+ log: { log: (m) => logged.push(m), warn() {}, error() {} },
+ });
+ const a = express();
+ a.use('/internal/events', c.router);
+ const s = http.createServer(a);
+ return new Promise((r2) => s.listen(0, '127.0.0.1', () => r2({ db: d, notifications: n, server: s, url: `http://127.0.0.1:${s.address().port}/internal/events` })));
+ }
+ const deliver = async (proc, event) => {
+ const raw = JSON.stringify({ event, seq: 1 });
+ const res = await fetch(proc.url, { method: 'POST', headers: { 'content-type': 'application/json', ...signDeliveryHeaders(raw, SECRET) }, body: raw });
+ return { status: res.status, body: await res.json() };
+ };
+ const stop = (proc) => new Promise((r2) => { proc.server.close(() => { proc.db.close(); r2(); }); });
+ const started = {
+ event_id: ids.newId('event'), event_type: 'live.stream.started', version: 1, source: 'live',
+ actor: { type: 'user', id: CAROL }, subject: { type: 'stream', id: '701', revision: 1 }, visibility: 'public', priority: 'important',
+ occurred_at: new Date().toISOString(),
+ payload: { stream_id: 701, channel: { username: 'carol', display_name: 'Carol', url: 'https://openvibe.live/@carol', subject: { type: 'user', id: CAROL } }, title: 'Go', category: null, protocol: 'whip', is_nsfw: false, started_at: new Date().toISOString() },
+ };
+ db.close();
+ const goLives = (d) => d.prepare("SELECT user_id, COUNT(*) AS c FROM notifications WHERE type = 'STREAM_LIVE' AND sender_id = 20 AND user_id IN (21, 22) GROUP BY user_id ORDER BY user_id").all().map((x) => [x.user_id, x.c]);
+ const goLiveEvents = (d) => outbox(d).filter((e) => [DAVE, ERIN].includes(e.subject.id) && e.payload.type === 'STREAM_LIVE').length;
+
+ // The process dies in the middle of the fan-out (Dave's notification written, Erin's not): nothing
+ // of it survives, not the notification, its event, the announcement window or the inbox claim.
+ let proc = await boot();
+ const create = proc.notifications.create.bind(proc.notifications);
+ let calls = 0;
+ proc.notifications.create = (data) => { if (++calls === 2) throw new Error('SIGKILL'); return create(data); };
+ let r3 = await deliver(proc, started);
+ assert.deepStrictEqual([r3.status, r3.body.code], [500, 'network.event_failed']);
+ assert.deepStrictEqual(goLives(proc.db), [], 'the half-done fan-out rolled back');
+ assert.strictEqual(goLiveEvents(proc.db), 0);
+ assert.ok(!proc.db.prepare('SELECT 1 FROM network_event_inbox WHERE event_id = ?').get(started.event_id));
+ const claimed = proc.db.prepare("SELECT 1 FROM sqlite_master WHERE name = 'stream_live_announcements'").get()
+ ? proc.db.prepare("SELECT COUNT(*) AS c FROM stream_live_announcements WHERE stream_id = '701'").get().c : 0;
+ assert.strictEqual(claimed, 0, 'the announcement window is not spent');
+ await stop(proc);
+
+ // Restarted: Events redelivers, and every follower is told exactly once.
+ proc = await boot();
+ r3 = await deliver(proc, started);
+ assert.strictEqual(r3.status, 200, JSON.stringify(r3.body));
+ assert.deepStrictEqual([r3.body.duplicate, r3.body.outcome], [false, 'notified']);
+ assert.deepStrictEqual(goLives(proc.db), [[21, 1], [22, 1]]);
+ assert.strictEqual(goLiveEvents(proc.db), 2, 'one network.notification.created per follower');
+ await new Promise((r2) => setImmediate(r2));
+ assert.deepStrictEqual(discord, ['carol'], 'the Discord alert once, after the commit');
+ assert.ok(logged.some((m) => m.includes(`live.stream.started ${started.event_id}: notified`)), 'the outcome is in the log (C-85 evidence)');
+
+ // Restarted again after the commit but before Events saw the 200: the redelivery is a duplicate.
+ await stop(proc);
+ proc = await boot();
+ liveCalls.length = 0;
+ r3 = await deliver(proc, started);
+ assert.deepStrictEqual([r3.status, r3.body.duplicate, r3.body.outcome], [200, true, null]);
+ assert.deepStrictEqual(goLives(proc.db), [[21, 1], [22, 1]], 'nobody is told twice');
+ assert.strictEqual(goLiveEvents(proc.db), 2);
+ assert.strictEqual(liveCalls.length, 0, 'Live is not asked again');
+ await new Promise((r2) => setImmediate(r2));
+ assert.deepStrictEqual(discord, ['carol']);
+ await stop(proc);
+ live.close();
+
+ fs.rmSync(dir, { recursive: true, force: true });
+ console.log('notification events: all checks passed');
+})().catch((err) => { console.error(err); process.exit(1); });