openvibers/openvibe.coupons Commits
Add shared event-consumer wiring for inbound OpenVibe.Events webhook (#7)
Changes
@@@ -39,6 +39,7 @@ const { createCouponsReadiness } = require('./observability');const { createActorLimits } = require('./http/actor-limits');const { createWorker } = require('./worker');const { assetVersion } = require('./render/layout');++const { createWebhookRouter } = require('./events/consumer');const PUBLIC_DIR = path.join(__dirname, '..', 'public');const VERSION = require('../package.json').version;@@@ -150,6 +151,10 @@ async function createApp(opts = {}) {// GET /<key>.txt — the IndexNow key file, mounted only when a key is configured (nothing else).if (indexnow.enabled) app.get(`/${config.indexnowKey}.txt`, indexnow.keyFile);++ // ── Inbound events (OpenVibe.Events webhook, signed) ─────++ // Mounted before sign-in and body-parser: reads its own raw body for the HMAC.++ app.use(createWebhookRouter(express, { config, store, importer, log }));++// ── Sign-in (OAuth2 client of OpenVibe.Network) ─────────app.use('/auth/', rateLimit({ windowMs: 15 * 60_000, limit: 60, standardHeaders: true, legacyHeaders: false }));app.use('/auth', auth.router(express));@@@ -97,6 +97,8 @@ function load(env = process.env) {events: {url: trim(env.EVENTS_URL || ''),intervalMs: int(env.EVENTS_RELAY_INTERVAL_MS, 2000),++ // Webhook secrets for inbound event deliveries (signed by OpenVibe.Events).++ webhookSecrets: list(env.COUPONS_EVENTS_SECRET),},// Expiry and confidence-decay sweep.nnew file mode 100644@@@ -0,0 +1,55 @@++'use strict';++++/**++ * Inbound event webhook (OpenVibe.Events → Coupons): a signed POST /internal/events endpoint that++ * receives deliveries from OpenVibe.Events and acts on them. The shared half — signature verification,++ * exactly-once through the inbox — is openvibe-publishing/ingest createEventConsumer. What stays here++ * is the coupons policy: which event types trigger what.++ *++ * sources.item.created a new Sources item was published; run the importer to pick it up++ * other sources.* acknowledged (the cursor will catch it on the next scheduled run)++ *++ * Mounted before sign-in and body-parser (server/app.js): the route reads its own raw body for the++ * HMAC and never touches cookies, sessions or the JSON parser.++ */++const { createEventConsumer } = require('openvibe-publishing/ingest');++++function createWebhookRouter(express, { config, store, importer, log = console }) {++ const consumer = createEventConsumer({++ db: store.db,++ secrets: config.events.webhookSecrets,++ consumer: 'coupons-sources',++ });++ // Ensure the idempotency_receipts table exists (the inbox needs it for exactly-once).++ consumer.inbox.ensureSchema().catch((err) => log.warn('[Coupons] inbox schema:', err.message));++++ const router = express.Router();++++ // Raw body for signature verification; never parsed by the shared JSON body-parser.++ router.post('/internal/events', express.raw({ type: 'application/json', limit: '1mb' }), async (req, res) => {++ const r = await consumer.apply(req.body, req.headers, async (event) => {++ const type = event && event.event_type;++ if (type === 'sources.item.created') {++ try {++ const result = await importer.run();++ return result.error ? `error:${result.error}` : 'imported';++ } catch (err) {++ return `error:${err.message}`;++ }++ }++ if (type && typeof type === 'string' && type.startsWith('sources.')) {++ return 'ack';++ }++ return null;++ });++++ if (r.status === 503) return res.status(404).end();++ if (r.status === 401) return res.status(401).end();++ if (r.status === 400) return res.status(400).end();++ return res.status(200).json({ accepted: true, duplicate: !!r.duplicate, outcome: r.outcome || null });++ });++++ return router;++}++++module.exports = { createWebhookRouter };nnew file mode 100644@@@ -0,0 +1,116 @@++'use strict';++/**++ * Inbound event webhook (OpenVibe.Events → Coupons): signed POST /internal/events.++ * - no secret configured → 404 (route not served)++ * - bad signature → 401++ * - valid sources.item.created → 200, importer.run once, duplicate false++ * - replay of the same delivery → 200, duplicate true, importer.run not called again++ * - other sources.* events → 200, importer not called++ */++const assert = require('assert');++const crypto = require('crypto');++const { boot, check, done } = require('./helpers/boot');++const { signDeliveryHeaders } = require('openvibe-sdk/events');++++const EVENT_ID = 'evt_01J8Z6Q3KX0000000000000000';++const ITEM_ID = 'itm_01J8Z6Q3KX0000000000000000';++++function makeDelivery(event, secret, now = Date.now()) {++ const raw = Buffer.from(JSON.stringify({ event, seq: 1 }));++ return { raw, headers: signDeliveryHeaders(raw, secret, { now }) };++}++++function envelope(type, id = EVENT_ID) {++ return {++ event_id: id,++ event_type: type,++ version: 1,++ source: 'sources',++ actor: { type: 'service', id: 'sources' },++ timestamp: '2026-09-22T12:00:00.000Z',++ visibility: 'internal',++ subject: { type: 'item', id: ITEM_ID },++ payload: { category: 'coupons', item_id: ITEM_ID },++ };++}++++(async () => {++ // ── No secret configured → 404 ──────────────────────────────────────────++ await check('no COUPONS_EVENTS_SECRET: POST /internal/events is 404', async () => {++ const t = await boot();++ const res = await t.get('/internal/events', { method: 'POST', json: { event: envelope('sources.item.created') } });++ assert.strictEqual(res.status, 404);++ await t.close();++ });++++ // ── With secret configured ──────────────────────────────────────────────++ const secret = crypto.randomBytes(32).toString('hex');++ const t = await boot({ env: { COUPONS_EVENTS_SECRET: secret } });++++ await check('bad signature: 401', async () => {++ const wrong = crypto.randomBytes(32).toString('hex');++ const d = makeDelivery(envelope('sources.item.created'), wrong);++ const res = await t.get('/internal/events', {++ method: 'POST',++ headers: { 'content-type': 'application/json', ...d.headers },++ body: d.raw,++ });++ assert.strictEqual(res.status, 401);++ });++++ await check('valid sources.item.created: 200, duplicate false, importer.run called once', async () => {++ const d = makeDelivery(envelope('sources.item.created'), secret);++ let runs = 0;++ const origRun = t.ctx.importer.run.bind(t.ctx.importer);++ t.ctx.importer.run = async (...args) => { runs++; return origRun(...args); };++ const res = await t.get('/internal/events', {++ method: 'POST',++ headers: { 'content-type': 'application/json', ...d.headers },++ body: d.raw,++ });++ assert.strictEqual(res.status, 200);++ const body = res.json();++ assert.strictEqual(body.accepted, true);++ assert.strictEqual(body.duplicate, false);++ assert.strictEqual(runs, 1);++ t.ctx.importer.run = origRun;++ });++++ await check('replay of the same delivery: 200, duplicate true, importer not called again', async () => {++ const d = makeDelivery(envelope('sources.item.created'), secret);++ let runs = 0;++ const origRun = t.ctx.importer.run.bind(t.ctx.importer);++ t.ctx.importer.run = async (...args) => { runs++; return origRun(...args); };++ const res = await t.get('/internal/events', {++ method: 'POST',++ headers: { 'content-type': 'application/json', ...d.headers },++ body: d.raw,++ });++ assert.strictEqual(res.status, 200);++ const body = res.json();++ assert.strictEqual(body.duplicate, true);++ assert.strictEqual(runs, 0);++ t.ctx.importer.run = origRun;++ });++++ await check('other sources.* event: 200, importer not called', async () => {++ const d = makeDelivery(envelope('sources.item.updated', 'evt_01J8Z6Q3KX0000000000000001'), secret);++ let runs = 0;++ const origRun = t.ctx.importer.run.bind(t.ctx.importer);++ t.ctx.importer.run = async (...args) => { runs++; return origRun(...args); };++ const res = await t.get('/internal/events', {++ method: 'POST',++ headers: { 'content-type': 'application/json', ...d.headers },++ body: d.raw,++ });++ assert.strictEqual(res.status, 200);++ const body = res.json();++ assert.strictEqual(body.accepted, true);++ assert.strictEqual(body.duplicate, false);++ assert.strictEqual(runs, 0);++ t.ctx.importer.run = origRun;++ });++++ await t.close();++ done();++})();
@@ -39,6 +39,7 @@ const { createCouponsReadiness } = require('./observability');
@@ -39,6 +39,7 @@ const { createCouponsReadiness } = require('./observability');
const { createActorLimits } = require('./http/actor-limits');
const { createActorLimits } = require('./http/actor-limits');
const { createWorker } = require('./worker');
const { createWorker } = require('./worker');
const { assetVersion } = require('./render/layout');
const { assetVersion } = require('./render/layout');
+const { createWebhookRouter } = require('./events/consumer');
const PUBLIC_DIR = path.join(__dirname, '..', 'public');
const PUBLIC_DIR = path.join(__dirname, '..', 'public');
const VERSION = require('../package.json').version;
const VERSION = require('../package.json').version;
@@ -150,6 +151,10 @@ async function createApp(opts = {}) {
@@ -150,6 +151,10 @@ async function createApp(opts = {}) {
// GET /<key>.txt — the IndexNow key file, mounted only when a key is configured (nothing else).
// GET /<key>.txt — the IndexNow key file, mounted only when a key is configured (nothing else).
if (indexnow.enabled) app.get(`/${config.indexnowKey}.txt`, indexnow.keyFile);
if (indexnow.enabled) app.get(`/${config.indexnowKey}.txt`, indexnow.keyFile);
+ // ── Inbound events (OpenVibe.Events webhook, signed) ─────
+ // Mounted before sign-in and body-parser: reads its own raw body for the HMAC.
+ app.use(createWebhookRouter(express, { config, store, importer, log }));
+
// ── Sign-in (OAuth2 client of OpenVibe.Network) ─────────
// ── Sign-in (OAuth2 client of OpenVibe.Network) ─────────
app.use('/auth/', rateLimit({ windowMs: 15 * 60_000, limit: 60, standardHeaders: true, legacyHeaders: false }));
app.use('/auth/', rateLimit({ windowMs: 15 * 60_000, limit: 60, standardHeaders: true, legacyHeaders: false }));
app.use('/auth', auth.router(express));
app.use('/auth', auth.router(express));
@@ -97,6 +97,8 @@ function load(env = process.env) {
@@ -97,6 +97,8 @@ function load(env = process.env) {
events: {
events: {
url: trim(env.EVENTS_URL || ''),
url: trim(env.EVENTS_URL || ''),
intervalMs: int(env.EVENTS_RELAY_INTERVAL_MS, 2000),
intervalMs: int(env.EVENTS_RELAY_INTERVAL_MS, 2000),
+ // Webhook secrets for inbound event deliveries (signed by OpenVibe.Events).
+ webhookSecrets: list(env.COUPONS_EVENTS_SECRET),
},
},
// Expiry and confidence-decay sweep.
// Expiry and confidence-decay sweep.
new file mode 100644
new file mode 100644
@@ -0,0 +1,55 @@
@@ -0,0 +1,55 @@
+'use strict';
+
+/**
+ * Inbound event webhook (OpenVibe.Events → Coupons): a signed POST /internal/events endpoint that
+ * receives deliveries from OpenVibe.Events and acts on them. The shared half — signature verification,
+ * exactly-once through the inbox — is openvibe-publishing/ingest createEventConsumer. What stays here
+ * is the coupons policy: which event types trigger what.
+ *
+ * sources.item.created a new Sources item was published; run the importer to pick it up
+ * other sources.* acknowledged (the cursor will catch it on the next scheduled run)
+ *
+ * Mounted before sign-in and body-parser (server/app.js): the route reads its own raw body for the
+ * HMAC and never touches cookies, sessions or the JSON parser.
+ */
+const { createEventConsumer } = require('openvibe-publishing/ingest');
+
+function createWebhookRouter(express, { config, store, importer, log = console }) {
+ const consumer = createEventConsumer({
+ db: store.db,
+ secrets: config.events.webhookSecrets,
+ consumer: 'coupons-sources',
+ });
+ // Ensure the idempotency_receipts table exists (the inbox needs it for exactly-once).
+ consumer.inbox.ensureSchema().catch((err) => log.warn('[Coupons] inbox schema:', err.message));
+
+ const router = express.Router();
+
+ // Raw body for signature verification; never parsed by the shared JSON body-parser.
+ router.post('/internal/events', express.raw({ type: 'application/json', limit: '1mb' }), async (req, res) => {
+ const r = await consumer.apply(req.body, req.headers, async (event) => {
+ const type = event && event.event_type;
+ if (type === 'sources.item.created') {
+ try {
+ const result = await importer.run();
+ return result.error ? `error:${result.error}` : 'imported';
+ } catch (err) {
+ return `error:${err.message}`;
+ }
+ }
+ if (type && typeof type === 'string' && type.startsWith('sources.')) {
+ return 'ack';
+ }
+ return null;
+ });
+
+ if (r.status === 503) return res.status(404).end();
+ if (r.status === 401) return res.status(401).end();
+ if (r.status === 400) return res.status(400).end();
+ return res.status(200).json({ accepted: true, duplicate: !!r.duplicate, outcome: r.outcome || null });
+ });
+
+ return router;
+}
+
+module.exports = { createWebhookRouter };
new file mode 100644
new file mode 100644
@@ -0,0 +1,116 @@
@@ -0,0 +1,116 @@
+'use strict';
+/**
+ * Inbound event webhook (OpenVibe.Events → Coupons): signed POST /internal/events.
+ * - no secret configured → 404 (route not served)
+ * - bad signature → 401
+ * - valid sources.item.created → 200, importer.run once, duplicate false
+ * - replay of the same delivery → 200, duplicate true, importer.run not called again
+ * - other sources.* events → 200, importer not called
+ */
+const assert = require('assert');
+const crypto = require('crypto');
+const { boot, check, done } = require('./helpers/boot');
+const { signDeliveryHeaders } = require('openvibe-sdk/events');
+
+const EVENT_ID = 'evt_01J8Z6Q3KX0000000000000000';
+const ITEM_ID = 'itm_01J8Z6Q3KX0000000000000000';
+
+function makeDelivery(event, secret, now = Date.now()) {
+ const raw = Buffer.from(JSON.stringify({ event, seq: 1 }));
+ return { raw, headers: signDeliveryHeaders(raw, secret, { now }) };
+}
+
+function envelope(type, id = EVENT_ID) {
+ return {
+ event_id: id,
+ event_type: type,
+ version: 1,
+ source: 'sources',
+ actor: { type: 'service', id: 'sources' },
+ timestamp: '2026-09-22T12:00:00.000Z',
+ visibility: 'internal',
+ subject: { type: 'item', id: ITEM_ID },
+ payload: { category: 'coupons', item_id: ITEM_ID },
+ };
+}
+
+(async () => {
+ // ── No secret configured → 404 ──────────────────────────────────────────
+ await check('no COUPONS_EVENTS_SECRET: POST /internal/events is 404', async () => {
+ const t = await boot();
+ const res = await t.get('/internal/events', { method: 'POST', json: { event: envelope('sources.item.created') } });
+ assert.strictEqual(res.status, 404);
+ await t.close();
+ });
+
+ // ── With secret configured ──────────────────────────────────────────────
+ const secret = crypto.randomBytes(32).toString('hex');
+ const t = await boot({ env: { COUPONS_EVENTS_SECRET: secret } });
+
+ await check('bad signature: 401', async () => {
+ const wrong = crypto.randomBytes(32).toString('hex');
+ const d = makeDelivery(envelope('sources.item.created'), wrong);
+ const res = await t.get('/internal/events', {
+ method: 'POST',
+ headers: { 'content-type': 'application/json', ...d.headers },
+ body: d.raw,
+ });
+ assert.strictEqual(res.status, 401);
+ });
+
+ await check('valid sources.item.created: 200, duplicate false, importer.run called once', async () => {
+ const d = makeDelivery(envelope('sources.item.created'), secret);
+ let runs = 0;
+ const origRun = t.ctx.importer.run.bind(t.ctx.importer);
+ t.ctx.importer.run = async (...args) => { runs++; return origRun(...args); };
+ const res = await t.get('/internal/events', {
+ method: 'POST',
+ headers: { 'content-type': 'application/json', ...d.headers },
+ body: d.raw,
+ });
+ assert.strictEqual(res.status, 200);
+ const body = res.json();
+ assert.strictEqual(body.accepted, true);
+ assert.strictEqual(body.duplicate, false);
+ assert.strictEqual(runs, 1);
+ t.ctx.importer.run = origRun;
+ });
+
+ await check('replay of the same delivery: 200, duplicate true, importer not called again', async () => {
+ const d = makeDelivery(envelope('sources.item.created'), secret);
+ let runs = 0;
+ const origRun = t.ctx.importer.run.bind(t.ctx.importer);
+ t.ctx.importer.run = async (...args) => { runs++; return origRun(...args); };
+ const res = await t.get('/internal/events', {
+ method: 'POST',
+ headers: { 'content-type': 'application/json', ...d.headers },
+ body: d.raw,
+ });
+ assert.strictEqual(res.status, 200);
+ const body = res.json();
+ assert.strictEqual(body.duplicate, true);
+ assert.strictEqual(runs, 0);
+ t.ctx.importer.run = origRun;
+ });
+
+ await check('other sources.* event: 200, importer not called', async () => {
+ const d = makeDelivery(envelope('sources.item.updated', 'evt_01J8Z6Q3KX0000000000000001'), secret);
+ let runs = 0;
+ const origRun = t.ctx.importer.run.bind(t.ctx.importer);
+ t.ctx.importer.run = async (...args) => { runs++; return origRun(...args); };
+ const res = await t.get('/internal/events', {
+ method: 'POST',
+ headers: { 'content-type': 'application/json', ...d.headers },
+ body: d.raw,
+ });
+ assert.strictEqual(res.status, 200);
+ const body = res.json();
+ assert.strictEqual(body.accepted, true);
+ assert.strictEqual(body.duplicate, false);
+ assert.strictEqual(runs, 0);
+ t.ctx.importer.run = origRun;
+ });
+
+ await t.close();
+ done();
+})();