openvibers/openvibe.network Commits

Notifications from Events: POST /internal/events turns deals.watch.matched (payload.recipient) and trade.alert.triggered (subject user) into inbox notifications (DEAL_WATCH_MATCH / TRADE_ALERT, category service, so the person's mute and email choices apply). openvibe-sdk v0.4.0 parseDelivery with requireV2 (v1-only, stale or forged deliveries are refused), exactly once per event_id through the SDK inbox (network_event_inbox), unknown types acknowledged and ignored, links only to sites whose domain serves the service. Inert until NETWORK_EVENTS_SECRET is set and scripts/subscribe-events.js has created the two subscriptions (consumer network, endpoint http://127.0.0.1:4000/internal/events). Coupons publishes no watch event yet, so there is nothing of it to consume

OP OpenVibers committed 23 Sept 2026 · 07cd98c7d3 · 8 files, +479 −6





Changes
ddiff --git a/README.md b/README.md
iindex 64b5c719..139f238d 100644
---- a/README.md
++++ b/README.md
@@@ -191,6 +191,46 @@ curl -X POST http://127.0.0.1:4000/internal/notifications/push \
    -d '{"userId": 42, "type": "new_follower", "data": {"actorName": "someone"}}'
  ```
  
++### Notifications from Events
++
++`POST /internal/events` (`server/notifications/events-consumer.js`) turns OpenVibe.Events deliveries
++into inbox notifications for the person they name, through the same `NotificationService.create()`
++as everything else, so a person who turned the `service` category off gets nothing and email follows
++their per-category email choice.
++
++| Event type | From | Person | Notification |
++|---|---|---|---|
++| `deals.watch.matched` | Deals (`source: deals`) | `payload.recipient` (`usr_…`) | `DEAL_WATCH_MATCH`, `service`, normal |
++| `trade.alert.triggered` | Trade (`source: trade`) | `subject.id` (`subject.type: user`) | `TRADE_ALERT`, `service`, high |
++
++Coupons publishes no watch event yet: its merchant watches never leave Coupons, and its `coupons.*`
++events name no person. Every other type is acknowledged and ignored. Deliveries are v2-signed only
++(`openvibe-sdk` `parseDelivery` with `requireV2`, ±300 s), and each `event_id` is handled once
++(openvibe-sdk inbox, table `network_event_inbox`, consumer `network-notifications`). Links point at a
++site only when its domain serves the service (the exposure overlay above); until then the URL is kept
++in `rich_content.context.planned_url`.
++
++The route is inert until the operator does both of these:
++
++1. Generate the signing secret and give it to Network: add
++   `NETWORK_EVENTS_SECRET=<openssl rand -hex 32>` to `/etc/openvibe/network.env` (comma-separate a
++   second value to rotate), and make sure `OV_EVENTS_INTERNAL_URL=http://127.0.0.1:4300` is set.
++   Restart `openvibe-network`; the log says `[Events consumer] on`.
++2. Create the subscriptions in Events (consumer `network`, which Events takes from the token's
++   `svc:network`; endpoint `http://127.0.0.1:4000/internal/events`; topic patterns
++   `deals.watch.matched` and `trade.alert.triggered`; secret = the first `NETWORK_EVENTS_SECRET`):
++
++   ```bash
++   cd /opt/openvibe.network
++   sudo node --env-file=/etc/openvibe/network.env scripts/subscribe-events.js --dry-run
++   sudo node --env-file=/etc/openvibe/network.env scripts/subscribe-events.js
++   ```
++
++   The script signs its own 5-minute service token (Network is the issuer: `sub svc:network`,
++   `aud openvibe.events`, `cap events.subscription.manage`) and hands the secret to Events in the
++   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.
++
  ---
  
  ## Email Setup
ddiff --git a/package-lock.json b/package-lock.json
iindex 4b2c87a2..72b545dd 100644
---- a/package-lock.json
++++ b/package-lock.json
@@@ -21,7 +21,7 @@
          "jsonwebtoken": "^9.0.0",
          "multer": "^1.4.5-lts.1",
          "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.28.0",
--        "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.2.2",
++        "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.4.0",
          "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.3.0",
          "uuid": "^11.1.0",
          "web-push": "^3.6.7"
@@@ -1481,9 +1481,9 @@
        }
      },
      "node_modules/openvibe-sdk": {
--      "version": "0.2.2",
--      "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.2.2",
--      "integrity": "sha512-kDifUp3EcmnTWzElAmaCZnX659WErcnm4+qVEoAiYQWX/w/1p4TWvoHbUPsrX65TSROVR3tiyVkl801n/0yyCw==",
++      "version": "0.4.0",
++      "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.4.0",
++      "integrity": "sha512-Hyz55YOMIs0xwCQ1NXsN0sUYhlbKXAX3uWY2fPCdEEL18FXUYVhK8wlG/5JMMypFHnjDiSd9VHzBNG12hWsmkQ==",
        "license": "MIT",
        "engines": {
          "node": ">=20"
ddiff --git a/package.json b/package.json
iindex 75fd6a29..ca201cf8 100644
---- a/package.json
++++ b/package.json
@@@ -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/chrome.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/registry-ecosystem.test.js && node test/registry-exposure.test.js && node test/service-principal.test.js && node test/oauth-pkce.test.js && node test/developer-projects.test.js && node test/developer-defaults.test.js && node test/public-cors.test.js && node test/observability.test.js && node test/status.test.js && node test/security-session.test.js && node test/security-redirects.test.js && node test/analytics-privacy.test.js && node test/data-ownership.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/chrome.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/registry-ecosystem.test.js && node test/registry-exposure.test.js && node test/service-principal.test.js && node test/oauth-pkce.test.js && node test/developer-projects.test.js && node test/developer-defaults.test.js && node test/public-cors.test.js && node test/observability.test.js && node test/status.test.js && node test/events-consumer.test.js && node test/security-session.test.js && node test/security-redirects.test.js && node test/analytics-privacy.test.js && node test/data-ownership.test.js",
      "reset-db": "node server/reset-db.js",
      "grant-admin": "node server/grant-admin.js"
    },
@@@ -24,7 +24,7 @@
      "jsonwebtoken": "^9.0.0",
      "multer": "^1.4.5-lts.1",
      "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.28.0",
--    "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.2.2",
++    "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.4.0",
      "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.3.0",
      "uuid": "^11.1.0",
      "web-push": "^3.6.7"
ddiff --git a/scripts/subscribe-events.js b/scripts/subscribe-events.js
nnew file mode 100644
iindex 00000000..d04005ca
---- /dev/null
++++ b/scripts/subscribe-events.js
@@@ -0,0 +1,67 @@
++#!/usr/bin/env node
++'use strict';
++/**
++ * Create Network's OpenVibe.Events subscriptions for the notification consumer
++ * (server/notifications/events-consumer.js): one per topic in TOPICS (deals.watch.matched,
++ * trade.alert.triggered), all delivering to POST /internal/events. Events names the consumer after
++ * the calling service: `network`.
++ *
++ *   sudo node --env-file=/etc/openvibe/network.env scripts/subscribe-events.js \
++ *        [--endpoint http://127.0.0.1:4000/internal/events] [--dry-run]
++ *
++ * Reads the environment: OV_EVENTS_INTERNAL_URL (Events, e.g. http://127.0.0.1:4300), OV_NETWORK_URL
++ * (the token issuer), JWT_PRIVATE_KEY (default data/keys/private.pem) and NETWORK_EVENTS_SECRET — the
++ * delivery signing secret; its FIRST value is handed to Events, so generate it and restart Network
++ * with it first (`openssl rand -hex 32`). Network is the token issuer, so it signs its own 5-minute
++ * service token (sub svc:network, aud openvibe.events, cap events.subscription.manage) instead of
++ * asking itself for one. Nothing secret is printed. An existing identical subscription is reported,
++ * not duplicated.
++ */
++require('dotenv').config();
++const fs = require('fs');
++const path = require('path');
++const crypto = require('crypto');
++const { serviceAuth, assertValid } = require('openvibe-contracts');
++const config = require('../server/config');
++const { TOPICS, secretsFrom } = require('../server/notifications/events-consumer');
++
++const args = process.argv.slice(2);
++const opt = (name, d) => { const i = args.indexOf(`--${name}`); return i >= 0 ? args[i + 1] : d; };
++
++function selfToken(privateKey, issuer) {
++    const now = Math.floor(Date.now() / 1000);
++    const claims = {
++        iss: issuer, sub: 'svc:network', actor_type: 'service', aud: ['openvibe.events'], cap: ['events.subscription.manage'],
++        iat: now, exp: now + 300, jti: `tok_${crypto.randomBytes(12).toString('hex')}`,
++    };
++    assertValid('identity.service-token-claims@1', claims);
++    return serviceAuth.signServiceToken(claims, privateKey);
++}
++
++(async () => {
++    const eventsUrl = String(config.eventsInternalUrl || '').replace(/\/+$/, '');
++    const endpoint = opt('endpoint', `http://127.0.0.1:${config.port}/internal/events`);
++    const secret = secretsFrom(config.eventsWebhookSecrets)[0];
++    if (!eventsUrl) throw new Error('OV_EVENTS_INTERNAL_URL is not set');
++    if (!secret) throw new Error('NETWORK_EVENTS_SECRET must be set (32+ characters) before subscribing');
++    const privateKey = fs.readFileSync(path.resolve(config.jwt.privateKeyPath), 'utf8');
++    if (!privateKey.includes('BEGIN')) throw new Error(`${config.jwt.privateKeyPath} is not a PEM private key`);
++    if (args.includes('--dry-run')) {
++        for (const t of TOPICS) console.log(`would subscribe: ${t} → ${endpoint} (consumer network, Events ${eventsUrl})`);
++        return;
++    }
++    const token = selfToken(privateKey, config.jwt.issuer);
++    for (const pattern of TOPICS) {
++        const res = await fetch(`${eventsUrl}/api/v1/subscriptions`, {
++            method: 'POST',
++            headers: { 'Content-Type': 'application/json', Authorization: `Bearer ${token}` },
++            body: JSON.stringify({ topic_pattern: pattern, endpoint, secret }),
++            signal: AbortSignal.timeout(15000),
++        });
++        const body = await res.json().catch(() => ({}));
++        if (res.status === 409 && body.subscription_id) { console.log(`subscription exists: ${body.subscription_id} (${pattern} → ${endpoint})`); continue; }
++        if (!res.ok) throw new Error(`Events answered ${res.status} for ${pattern}: ${body.code || ''} ${body.detail || ''}`.trim());
++        console.log(`subscribed: ${body.id} (${pattern} → ${endpoint})`);
++    }
++    console.log('Replay history if wanted with POST /api/v1/deliveries/replay { subscription_id, from_seq } (events.delivery.admin).');
++})().catch((err) => { console.error(`subscribe failed: ${err.message}`); process.exit(1); });
ddiff --git a/server/config.js b/server/config.js
iindex 1e8c809b..8d4599ee 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -49,6 +49,11 @@ module.exports = {
      // OpenVibe.Events base URL for Network's own events (developer projects). Unset = no relay:
      // events stay in dev_audit and are backfilled when it is set. Production: http://127.0.0.1:4300
      eventsInternalUrl: process.env.OV_EVENTS_INTERNAL_URL || '',
++    // Signing secret(s) of Network's Events subscriptions (POST /internal/events → notifications;
++    // server/notifications/events-consumer.js). Comma-separated for rotation, each 32+ characters.
++    // Unset = the consumer answers 503 and nothing is delivered. Handed to Events by
++    // scripts/subscribe-events.js.
++    eventsWebhookSecrets: process.env.NETWORK_EVENTS_SECRET || '',
  
      // Database
      db: {
ddiff --git a/server/index.js b/server/index.js
iindex 38c856a4..1a689fcb 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -150,6 +150,11 @@ app.use(cookieParser());
  // Provider webhooks (Resend bounces/complaints) — mounted BEFORE the JSON body parser so
  // the route sees the raw bytes it must verify the Svix signature over.
  app.use('/api/webhooks', require('./notifications/resend-webhook')());
++// OpenVibe.Events deliveries → notifications (server/notifications/events-consumer.js), also before the
++// JSON parser (the v2 signature covers the raw body). Built once the database and the notification
++// service exist (below); inert (503) until NETWORK_EVENTS_SECRET is set.
++let eventsConsumer = null;
++app.use('/internal/events', (req, res, next) => (eventsConsumer ? eventsConsumer.router(req, res, next) : res.status(503).json({ error: 'starting' })));
  app.use(express.json({ limit: '1mb' }));
  app.use(express.urlencoded({ extended: true }));
  
@@@ -353,6 +358,8 @@ const notificationService = new NotificationService(db);
  const emailService = new EmailService(db);
  app.locals.notificationService = notificationService;
  app.locals.emailService = emailService;
++eventsConsumer = require('./notifications/events-consumer').createEventsConsumer({ db, notifications: notificationService, secrets: config.eventsWebhookSecrets });
++console.log(`[Events consumer] ${eventsConsumer.enabled ? 'on' : 'off (NETWORK_EVENTS_SECRET unset)'}: POST /internal/events`);
  
  // Discord bot service
  const discordService = new DiscordService(db);
ddiff --git a/server/notifications/events-consumer.js b/server/notifications/events-consumer.js
nnew file mode 100644
iindex 00000000..87ed6989
---- /dev/null
++++ b/server/notifications/events-consumer.js
@@@ -0,0 +1,168 @@
++'use strict';
++/**
++ * OpenVibe.Events → notifications: POST /internal/events, the endpoint of Network's Events
++ * subscriptions (consumer `network`, created by scripts/subscribe-events.js).
++ *
++ *   deals.watch.matched     a Deals watch (keyword, product, price below) matched a fresh observation;
++ *                           payload.recipient is the watcher's usr_ subject
++ *   trade.alert.triggered   a Trade alert rule fired (threshold crossed, filing/document seen);
++ *                           event.subject { type: 'user', id } is the rule's owner
++ *
++ * Coupons publishes no watch event: its merchant watches never leave Coupons (server/domain/watches.js
++ * there: "Delivery is not built yet"), and its coupons.* events name no person, so there is nothing
++ * of Coupons' to consume yet. README "Notifications from Events" has the operator steps.
++ *
++ * Each event becomes one inbox notification for its person through NotificationService.create(),
++ * which applies their preferences: a muted category ('service') creates nothing, and email follows
++ * the same per-category choice (shouldEmail). Exactly once: the openvibe-sdk inbox claims
++ * (consumer, event_id) in the same SQLite transaction as the notification, so a redelivery does
++ * nothing and a failure rolls both back (Events retries).
++ *
++ * Signature: v2 only (openvibe-sdk parseDelivery with requireV2) — HMAC over "<t>.<raw body>" with t
++ * within ±300 s, under NETWORK_EVENTS_SECRET (comma-separated for rotation, each 32+ characters).
++ * A v1-only, stale or unsigned delivery is refused. Unset secret = 503: the route is inert until the
++ * operator sets it and creates the subscriptions.
++ *
++ * Links: a notification links to the event's URL only when that host is one of our sites whose
++ * public domain serves the service (server/chrome/sites.js, via the exposure overlay). Today Deals
++ * and Trade are loopback-only and their domains show placeholder pages, so the link is kept in
++ * rich_content.context.planned_url and the notification has no click-through yet.
++ */
++const express = require('express');
++const { http, ids } = require('openvibe-contracts');
++const { parseDelivery, createInbox } = require('openvibe-sdk/events');
++const { siteForHost } = require('../chrome/sites');
++
++const CONSUMER = 'network-notifications';
++const INBOX_TABLE = 'network_event_inbox';
++const TOPICS = Object.freeze(['deals.watch.matched', 'trade.alert.triggered']);
++const EVENT_ID_RE = /^evt_[0-9A-HJKMNP-TV-Z]{26}$/;
++
++/** Plain, single-line, length-capped text: no markup or control characters reach the inbox. */
++function clean(v, n) {
++    return String(v == null ? '' : v).replace(/[\u0000-\u001f\u007f<>]/g, ' ').replace(/\s+/g, ' ').trim().slice(0, n);
++}
++
++/** An https URL on one of our sites (any, or only those whose domain serves the service), or null. */
++function ownLink(u, { publicOnly = false } = {}) {
++    let url;
++    try { url = new URL(String(u || '')); } catch { return null; }
++    if (url.protocol !== 'https:' || url.username || url.password || url.port) return null;
++    const site = siteForHost(url.hostname);
++    if (!site || (publicOnly && site.status !== 'open')) return null;
++    return url.toString().slice(0, 500);
++}
++const publicLink = (u) => ownLink(u, { publicOnly: true });
++
++function secretsFrom(value) {
++    return String(value || '').split(',').map((s) => s.trim()).filter((s) => s.length >= 32);
++}
++
++/** What each event type turns into. Returns { subject, notification } or an 'ignored:*' outcome. */
++const HANDLERS = {
++    'deals.watch.matched'(event) {
++        if (event.source !== 'deals') return 'ignored:source';
++        const p = event.payload && typeof event.payload === 'object' ? event.payload : {};
++        const title = clean(p.title, 160);
++        if (!title || !p.watch_id) return 'ignored:payload';
++        const obs = p.observation && typeof p.observation === 'object' ? p.observation : {};
++        const price = obs.price ? `${clean(obs.price, 20)}${obs.currency ? ' ' + clean(obs.currency, 3) : ''}` : null;
++        let why = '';
++        if (p.kind === 'price_below' && p.max_price) why = ` (below your ${clean(p.max_price, 20)} ${clean(p.currency, 3)})`;
++        else if (p.kind === 'keyword' && p.query) why = ` (matches "${clean(p.query, 80)}")`;
++        return {
++            subject: p.recipient,
++            notification: {
++                type: 'DEAL_WATCH_MATCH', category: 'service', priority: 'normal', icon: '🏷️', title: 'Deal watch matched',
++                message: clean(`${title}${price ? ' at ' + price : ''}${why}`, 300),
++                service: 'deals', url: publicLink(p.offer_url),
++                rich_content: { service: 'deals', context: { event_id: event.event_id, watch_id: clean(p.watch_id, 64), offer_id: clean(p.offer_id, 64), kind: clean(p.kind, 20), planned_url: ownLink(p.offer_url) } },
++            },
++        };
++    },
++    'trade.alert.triggered'(event) {
++        if (event.source !== 'trade') return 'ignored:source';
++        const s = event.subject && typeof event.subject === 'object' ? event.subject : {};
++        if (s.type !== 'user') return 'ignored:payload';
++        const p = event.payload && typeof event.payload === 'object' ? event.payload : {};
++        const ins = p.instrument && typeof p.instrument === 'object' ? p.instrument : {};
++        const rule = p.rule && typeof p.rule === 'object' ? p.rule : {};
++        const t = p.trigger && typeof p.trigger === 'object' ? p.trigger : {};
++        const symbol = clean(ins.symbol, 20);
++        if (!symbol) return 'ignored:payload';
++        let what;
++        if (t.kind === 'observation') {
++            what = `${clean(t.metric || rule.metric, 60)} is ${clean(t.value, 30)}${t.unit ? ' ' + clean(t.unit, 20) : ''}${rule.operator && rule.threshold ? ` (${clean(rule.operator, 5)} ${clean(rule.threshold, 30)})` : ''}`;
++        } else if (t.kind === 'document') {
++            what = `New ${clean(t.form_type, 20) || 'document'}${t.title ? ': ' + clean(t.title, 160) : ''}`;
++        } else {
++            return 'ignored:payload';
++        }
++        const disclaimer = clean(p.disclaimer, 120);
++        return {
++            subject: s.id,
++            notification: {
++                type: 'TRADE_ALERT', category: 'service', priority: 'high', icon: '📈', title: clean(`Trade alert: ${symbol}`, 60),
++                message: clean(`${ins.name ? clean(ins.name, 80) + ' — ' : ''}${what}.${disclaimer ? ' ' + disclaimer : ''}`, 400),
++                service: 'trade', url: publicLink(ins.url),
++                rich_content: { service: 'trade', context: { event_id: event.event_id, rule_id: clean(rule.id, 64), delivery_id: clean(p.delivery_id, 64), trigger: clean(t.kind, 20), planned_url: ownLink(ins.url) } },
++            },
++        };
++    },
++};
++
++/**
++ * @param {object} o
++ * @param {import('better-sqlite3').Database} o.db
++ * @param {{ create(data: object): object|null }} o.notifications  NotificationService
++ * @param {string|string[]} o.secrets  NETWORK_EVENTS_SECRET (comma list) or an array
++ */
++function createEventsConsumer({ db, notifications, secrets, now = () => Date.now(), log = console }) {
++    const keys = Array.isArray(secrets) ? secrets.filter((s) => typeof s === 'string' && s.length >= 32) : secretsFrom(secrets);
++    const inbox = createInbox(db, { table: INBOX_TABLE, now });
++    inbox.ensureSchema();
++    const userBySubject = db.prepare('SELECT id FROM users WHERE subject_id = ?');
++
++    /** Apply one envelope. Returns { duplicate, outcome }. Throws only on a storage failure. */
++    function apply(event) {
++        const r = inbox.once(CONSUMER, event.event_id, () => {
++            const handler = Object.prototype.hasOwnProperty.call(HANDLERS, event.event_type) ? HANDLERS[event.event_type] : null;
++            if (!handler) return 'ignored:type';
++            const out = handler(event);
++            if (typeof out === 'string') return out;
++            if (!ids.isSubjectId('user', out.subject)) return 'ignored:recipient';
++            const user = userBySubject.get(out.subject);
++            if (!user) return 'ignored:recipient';
++            // create() returns null when the person turned the category off.
++            return notifications.create({ ...out.notification, user_id: user.id }) ? 'notified' : 'suppressed:preference';
++        });
++        return r.duplicate ? { duplicate: true, outcome: null } : { duplicate: false, outcome: r.result };
++    }
++
++    const router = express.Router();
++    router.post('/', express.raw({ type: () => true, limit: '256kb' }), (req, res) => {
++        if (!keys.length) return http.sendProblem(res, 503, 'network.webhook_disabled', { detail: 'NETWORK_EVENTS_SECRET is not set' });
++        const raw = Buffer.isBuffer(req.body) ? req.body : Buffer.alloc(0);
++        let delivery = null;
++        // Signature v2 only: a v1-only (v2 stripped), stale or forged delivery is refused.
++        for (const s of keys) { delivery = parseDelivery(raw, req.headers, s, { requireV2: true, now: now() }); if (delivery) break; }
++        if (!delivery) return http.sendProblem(res, 401, 'network.bad_signature', { detail: 'X-OpenVibe-Signature-V2 does not verify or is outside the replay window' });
++        const event = delivery.event;
++        if (!event || typeof event !== 'object' || typeof event.event_id !== 'string' || !EVENT_ID_RE.test(event.event_id) || typeof event.event_type !== 'string') {
++            return http.sendProblem(res, 400, 'network.bad_delivery', { detail: 'body must be { event: <envelope>, seq }' });
++        }
++        let out;
++        try {
++            out = apply(event);
++        } catch (err) {
++            // Not acknowledged: the inbox claim rolled back with the notification, and Events retries.
++            log.error(`[Events consumer] ${event.event_id} (${event.event_type}) failed:`, err.message);
++            return http.sendProblem(res, 500, 'network.event_failed', { detail: 'processing failed; it will be retried' });
++        }
++        res.status(200).json({ event_id: event.event_id, duplicate: out.duplicate, outcome: out.outcome });
++    });
++
++    return { router, apply, enabled: keys.length > 0 };
++}
++
++module.exports = { createEventsConsumer, CONSUMER, TOPICS, INBOX_TABLE, secretsFrom };
ddiff --git a/test/events-consumer.test.js b/test/events-consumer.test.js
nnew file mode 100644
iindex 00000000..f7ea5ad1
---- /dev/null
++++ b/test/events-consumer.test.js
@@@ -0,0 +1,186 @@
++'use strict';
++// Events → notifications (server/notifications/events-consumer.js): POST /internal/events turns
++// deals.watch.matched and trade.alert.triggered into inbox notifications for the subscribed person,
++// exactly once, v2 signatures only, respecting their preferences.
++//   node test/events-consumer.test.js
++const assert = require('assert');
++const fs = require('fs');
++const os = require('os');
++const path = require('path');
++const http = require('http');
++const express = require('express');
++const { ids } = require('openvibe-contracts');
++const { signDeliveryHeaders } = require('openvibe-sdk/events');
++const { initDb } = require('../server/db/database');
++const { NotificationService } = require('../server/notifications/notification-service');
++const { createEventsConsumer, TOPICS } = require('../server/notifications/events-consumer');
++
++const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-events-consumer-'));
++const log = console.log; console.log = () => {};
++const db = initDb(path.join(dir, 'network.db'));
++console.log = log;
++
++const ALICE = ids.newId('user'), BOB = ids.newId('user'), NOBODY = ids.newId('user');
++db.prepare("INSERT INTO users (id, username, password_hash, subject_id, email, email_verified) VALUES (7, 'alice', 'x', ?, '[email protected]', 1), (8, 'bob', 'x', ?, NULL, 0)").run(ALICE, BOB);
++const notifications = new NotificationService(db);
++const SECRET = 'a'.repeat(40), NEXT = 'b'.repeat(40);
++
++const dealsEvent = (over = {}) => ({
++    event_id: ids.newId('event'), event_type: 'deals.watch.matched', version: 1, source: 'deals',
++    actor: { type: 'service', id: 'deals' }, subject: { type: 'watch', id: 'dwt_1' }, visibility: 'internal',
++    occurred_at: new Date().toISOString(),
++    payload: {
++        watch_id: 'dwt_1', recipient: ALICE, kind: 'price_below', query: null, product_id: 'prd_1', max_price: '250.00', currency: 'USD',
++        offer_id: 'dof_1', offer_url: 'https://openvibe.deals/d/dof_1-cheap-headphones', title: 'Cheap <b>headphones</b>',
++        observation: { id: 'dob_1', observed_at: new Date().toISOString(), price: '199.00', currency: 'USD', availability: 'in_stock' },
++    },
++    ...over,
++});
++const tradeEvent = (over = {}) => ({
++    event_id: ids.newId('event'), event_type: 'trade.alert.triggered', version: 1, source: 'trade',
++    actor: { type: 'service', id: 'trade' }, subject: { type: 'user', id: BOB }, visibility: 'subject', priority: 'important',
++    occurred_at: new Date().toISOString(),
++    payload: {
++        delivery_id: 'ald_1',
++        rule: { id: 'alr_1', kind: 'threshold', metric: 'close', operator: 'above', threshold: '200', unit: 'USD' },
++        instrument: { id: 'ins_1', symbol: 'AAPL', name: 'Apple Inc.', url: 'https://openvibe.trade/i/AAPL' },
++        trigger: { kind: 'observation', id: 'obs_1', metric: 'close', value: '201.50', unit: 'USD' },
++        disclaimer: 'Information only — not investment advice; no trading here.',
++    },
++    ...over,
++});
++const rows = (userId) => db.prepare('SELECT * FROM notifications WHERE user_id = ? ORDER BY created_at, rowid').all(userId);
++
++(async () => {
++    // The subscriptions this consumer is meant for, and nothing else.
++    assert.deepStrictEqual(TOPICS, ['deals.watch.matched', 'trade.alert.triggered']);
++
++    let consumer = createEventsConsumer({ db, notifications, secrets: '' });
++    const app = express();
++    app.use('/internal/events', (req, res, next) => consumer.router(req, res, next));
++    app.use(express.json());   // as in server/index.js: the consumer runs before the JSON parser
++    const srv = http.createServer(app);
++    await new Promise(r => srv.listen(0, '127.0.0.1', r));
++    const url = `http://127.0.0.1:${srv.address().port}/internal/events`;
++    const post = async (event, { secret = SECRET, headers = null, strip = [], now } = {}) => {
++        const raw = JSON.stringify({ event, seq: 1 });
++        const h = { 'content-type': 'application/json', ...(headers || signDeliveryHeaders(raw, secret, now ? { now } : {})) };
++        for (const k of strip) delete h[k];
++        const r = await fetch(url, { method: 'POST', headers: h, body: raw });
++        return { status: r.status, body: await r.json() };
++    };
++
++    // Inert until the secret is set: nothing is accepted.
++    let r = await post(dealsEvent());
++    assert.strictEqual(r.status, 503);
++    assert.strictEqual(r.body.code, 'network.webhook_disabled');
++    assert.strictEqual(rows(7).length, 0);
++
++    consumer = createEventsConsumer({ db, notifications, secrets: `${SECRET},${NEXT}` });
++    assert.strictEqual(consumer.enabled, true);
++
++    // A signed v2 Deals watch match becomes one notification for the watcher.
++    const e1 = dealsEvent();
++    r = await post(e1);
++    assert.strictEqual(r.status, 200);
++    assert.deepStrictEqual(r.body, { event_id: e1.event_id, duplicate: false, outcome: 'notified' });
++    let mine = rows(7);
++    assert.strictEqual(mine.length, 1);
++    assert.strictEqual(mine[0].type, 'DEAL_WATCH_MATCH');
++    assert.strictEqual(mine[0].category, 'service');
++    assert.strictEqual(mine[0].service, 'deals');
++    assert.strictEqual(mine[0].title, 'Deal watch matched');
++    assert.strictEqual(mine[0].message, 'Cheap b headphones /b at 199.00 USD (below your 250.00 USD)', 'no markup reaches the inbox');
++    assert.strictEqual(mine[0].url, null, 'openvibe.deals serves a placeholder: no click-through yet');
++    assert.strictEqual(JSON.parse(mine[0].rich_content).context.planned_url, 'https://openvibe.deals/d/dof_1-cheap-headphones');
++
++    // The same event again (Events retries, or replays): a duplicate, nothing new.
++    r = await post(e1);
++    assert.strictEqual(r.status, 200);
++    assert.strictEqual(r.body.duplicate, true);
++    assert.strictEqual(r.body.outcome, null);
++    assert.strictEqual(rows(7).length, 1);
++
++    // v1-only (the v2 headers stripped), stale v2, wrong secret, unsigned: refused, nothing stored.
++    for (const [label, opts] of [
++        ['v1 only', { strip: ['X-OpenVibe-Signature-V2', 'X-OpenVibe-Timestamp'] }],
++        ['stale v2', { now: Date.now() - 10 * 60 * 1000 }],
++        ['wrong secret', { secret: 'c'.repeat(40) }],
++        ['unsigned', { headers: {} }],
++    ]) {
++        const e = dealsEvent();
++        r = await post(e, opts);
++        assert.strictEqual(r.status, 401, label);
++        assert.strictEqual(r.body.code, 'network.bad_signature', label);
++        assert.ok(!db.prepare('SELECT 1 FROM network_event_inbox WHERE event_id = ?').get(e.event_id), `${label}: not even recorded`);
++    }
++    assert.strictEqual(rows(7).length, 1);
++
++    // The rotated-in secret verifies too.
++    r = await post(dealsEvent({ payload: { ...dealsEvent().payload, kind: 'keyword', query: 'headphones', offer_id: 'dof_2' } }), { secret: NEXT });
++    assert.strictEqual(r.body.outcome, 'notified');
++    assert.match(rows(7)[1].message, /\(matches "headphones"\)$/);
++
++    // Unknown types are acknowledged and ignored (and recorded, so a redelivery is a duplicate).
++    const other = dealsEvent({ event_type: 'deals.offer.created' });
++    r = await post(other);
++    assert.deepStrictEqual([r.status, r.body.outcome], [200, 'ignored:type']);
++    r = await post(other);
++    assert.strictEqual(r.body.duplicate, true);
++    r = await post(dealsEvent({ event_type: 'constructor' }));
++    assert.strictEqual(r.body.outcome, 'ignored:type', 'no prototype lookups');
++    assert.strictEqual(rows(7).length, 2);
++
++    // A publisher may not speak for another service; an unknown person gets nothing.
++    r = await post(dealsEvent({ source: 'trade' }));
++    assert.strictEqual(r.body.outcome, 'ignored:source');
++    r = await post(dealsEvent({ payload: { ...dealsEvent().payload, recipient: NOBODY } }));
++    assert.strictEqual(r.body.outcome, 'ignored:recipient');
++    r = await post(dealsEvent({ payload: { ...dealsEvent().payload, recipient: '7' } }));
++    assert.strictEqual(r.body.outcome, 'ignored:recipient', 'recipients are usr_ subjects, never raw ids');
++
++    // A Trade alert reaches the rule's owner.
++    r = await post(tradeEvent());
++    assert.strictEqual(r.body.outcome, 'notified');
++    let bobs = rows(8);
++    assert.strictEqual(bobs.length, 1);
++    assert.strictEqual(bobs[0].type, 'TRADE_ALERT');
++    assert.strictEqual(bobs[0].title, 'Trade alert: AAPL');
++    assert.strictEqual(bobs[0].priority, 'high');
++    assert.strictEqual(bobs[0].message, 'Apple Inc. — close is 201.50 USD (above 200). Information only — not investment advice; no trading here.');
++    assert.strictEqual(bobs[0].url, null);
++    r = await post(tradeEvent({ payload: { ...tradeEvent().payload, trigger: { kind: 'document', id: 'doc_1', form_type: '8-K', title: 'Current report' } } }));
++    assert.strictEqual(r.body.outcome, 'notified');
++    assert.match(rows(8)[1].message, /New 8-K: Current report\./);
++    r = await post(tradeEvent({ subject: { type: 'watch', id: 'x' } }));
++    assert.strictEqual(r.body.outcome, 'ignored:payload');
++
++    // A link to a site whose domain serves the service is kept; a foreign host never is.
++    r = await post(tradeEvent({ payload: { ...tradeEvent().payload, instrument: { symbol: 'X', url: 'https://openvibe.live/x' } } }));
++    assert.strictEqual(rows(8).pop().url, 'https://openvibe.live/x');
++    r = await post(tradeEvent({ payload: { ...tradeEvent().payload, instrument: { symbol: 'Y', url: 'https://evil.example/y' } } }));
++    const last = rows(8).pop();
++    assert.strictEqual(last.url, null);
++    assert.strictEqual(JSON.parse(last.rich_content).context.planned_url, null);
++
++    // Preferences: with the 'service' category turned off, nothing is created (and it stays handled).
++    const before = rows(8).length;
++    notifications.setPreference(8, 'service', { enabled: false });
++    const muted = tradeEvent();
++    r = await post(muted);
++    assert.strictEqual(r.body.outcome, 'suppressed:preference');
++    assert.strictEqual(rows(8).length, before);
++    assert.strictEqual((await post(muted)).body.duplicate, true);
++    // Email follows the same per-category choice: none by default, yes once opted in (verified address).
++    const n = { ...rows(7)[0], email_verified: 1, email_bounced_at: null };
++    assert.strictEqual(notifications.shouldEmail(n), false);
++    notifications.setPreference(7, 'service', { email: true });
++    assert.strictEqual(notifications.shouldEmail(n), true);
++
++    // Malformed envelopes are refused.
++    r = await post({ event_id: 'nope', event_type: 'deals.watch.matched' });
++    assert.strictEqual(r.status, 400);
++
++    srv.close();
++    console.log('events consumer: all checks passed');
++})().catch((err) => { console.error(err); process.exit(1); });