openvibers/openvibe.coupons Commits
Add Events Sources subscriptions and a bounded Search reconcile script (#8)
Changes
@@@ -273,6 +273,11 @@ The charter's `coupons.created|updated|expired|disabled` are these `coupons.couptypes. Every envelope validates as `events.event-envelope@1` (tested), with actor`{type: service, id: coupons}`, except `coupons.moderation.action`, whose actor is the staff member.++Coupons subscribes to `sources.item.*` and `sources.fetch.failed` with `scripts/subscribe.js`++(endpoint `/internal/events`, delivery signed with `COUPONS_EVENTS_SECRET`; the import's cursor pull++works without it). `scripts/subscribe.js --reconcile` re-stamps every merchant and every code (any status, so a code that++left active results gets its tombstone) in one bounded, idempotent pass and reports `{ sent, unchanged }`.++## Capabilities (registered in openvibe-contracts v0.25.0)Service tokens use audience `openvibe.coupons`, with one capability per route:@@@ -317,6 +322,8 @@ What Coupons calls elsewhere, and with which grant, is underEach grant is `[client, capability, audience]`:- `[coupons, events.event.publish, openvibe.events]`++- `[coupons, events.subscription.manage, openvibe.events]` (only to create the Sources++ subscriptions: `scripts/subscribe.js`)- `[coupons, sources.item.read, openvibe.sources]` (only if the Sources import is turned on)- For OpenVibe.AI to deliver `coupons.extract_coupon` drafts:`[ai, coupons.coupon.submit, openvibe.coupons]`nnew file mode 100644@@@ -0,0 +1,146 @@++#!/usr/bin/env node++'use strict';++/**++ * Coupons' OpenVibe.Events side: create the Sources subscriptions the import loop wakes on, or++ * reconcile the OpenVibe.Search index in one bounded pass.++ *++ * node scripts/subscribe.js [--endpoint http://127.0.0.1:4850/internal/events]++ * node scripts/subscribe.js --reconcile [--page-size 200] [--max-pages 200]++ *++ * Subscribe reads the environment (.env; the service itself gets /etc/openvibe/coupons.env from systemd):++ * EVENTS_URL, OV_NETWORK_INTERNAL_URL, OV_OAUTH_CLIENT_ID, OV_OAUTH_CLIENT_SECRET (the coupons++ * principal needs events.subscription.manage for audience openvibe.events)++ * COUPONS_EVENTS_SECRET the delivery signing secret; the first value is handed to Events, so++ * generate it before running this (e.g. `openssl rand -hex 32`). Nothing++ * secret is printed.++ * Patterns: sources.item.* and sources.fetch.failed. Only sources.item.created runs the import; the++ * rest are acknowledged and the cursor pull catches up either way. An existing identical subscription is reported++ * ("exists"), never duplicated.++ *++ * --reconcile pages coupon_merchants and coupons, every status of both (a shop or code that left active++ * results must reach Search as unpublished or as a tombstone), re-stamping each Search document through++ * ctx.publication.sendDocument — an unchanged document costs nothing. The rows land in event_outbox;++ * when the relay is configured the CLI publishes them once before exiting, otherwise the running++ * service relays them. It is idempotent and bounded by --page-size/--max-pages; the report is++ * { sent, unchanged }.++ */++require('dotenv').config();++const { serviceAuth } = require('openvibe-contracts');++const { load } = require('../server/config');++++const PATTERNS = ['sources.item.*', 'sources.fetch.failed'];++const DEFAULT_PAGE_SIZE = 200;++const DEFAULT_MAX_PAGES = 200;++++/** The endpoint Events delivers to: this service's own inbound webhook. */++const defaultEndpoint = (config) => `http://127.0.0.1:${config.port}/internal/events`;++++function parseArgs(argv) {++ const o = { reconcile: false, endpoint: null, pageSize: DEFAULT_PAGE_SIZE, maxPages: DEFAULT_MAX_PAGES };++ for (let i = 0; i < argv.length; i++) {++ const a = argv[i];++ if (a === '--reconcile') o.reconcile = true;++ else if (a === '--endpoint') o.endpoint = argv[++i];++ else if (a === '--page-size') o.pageSize = Number(argv[++i]);++ else if (a === '--max-pages') o.maxPages = Number(argv[++i]);++ else throw new Error(`unknown argument ${a}`);++ }++ if (o.endpoint !== null && !o.endpoint) throw new Error('--endpoint needs a value');++ if (!Number.isInteger(o.pageSize) || o.pageSize < 1) throw new Error('--page-size must be a positive integer');++ if (!Number.isInteger(o.maxPages) || o.maxPages < 1) throw new Error('--max-pages must be a positive integer');++ return o;++}++++/**++ * Create one subscription per pattern, as the coupons principal. The caller's delivery secret is++ * handed to Events here; `fetchImpl` is injectable so the test can stub Network and Events.++ */++async function subscribe({ config, endpoint = defaultEndpoint(config), fetchImpl = globalThis.fetch, log = console.log } = {}) {++ if (!config.events.url) throw new Error('EVENTS_URL is not set');++ const secret = (config.events.webhookSecrets || [])[0];++ if (!secret || secret.length < 32) throw new Error('COUPONS_EVENTS_SECRET must be set (32+ characters) before subscribing');++ if (!config.oauth.clientSecret) throw new Error('OV_OAUTH_CLIENT_SECRET is not set');++ const tokens = serviceAuth.createTokenClient({++ tokenUrl: `${config.networkInternalUrl}/oauth/token`, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret,++ audience: 'openvibe.events', scope: 'events.subscription.manage', fetchImpl,++ });++ const out = [];++ for (const pattern of PATTERNS) {++ const res = await fetchImpl(`${config.events.url}/api/v1/subscriptions`, {++ method: 'POST',++ headers: { 'Content-Type': 'application/json', ...(await tokens.authHeaders()) },++ body: JSON.stringify({ topic_pattern: pattern, endpoint, secret }),++ });++ const body = await res.json().catch(() => ({}));++ if (res.status === 409 && body.subscription_id) {++ log(`subscription exists: ${body.subscription_id} (${pattern} → ${endpoint})`);++ out.push({ pattern, subscription_id: body.subscription_id, existed: true });++ continue;++ }++ if (!res.ok) throw new Error(`Events answered ${res.status} for ${pattern}: ${body.code || ''} ${body.detail || ''}`.trim());++ log(`subscribed: ${body.id} (${pattern} → ${endpoint}). Replay history with POST /api/v1/deliveries/replay { subscription_id, from_seq }.`);++ out.push({ pattern, subscription_id: body.id, existed: false });++ }++ return out;++}++++/**++ * One bounded pass over every merchant and every code, re-stamping each Search document into the++ * outbox. → { sent, unchanged } (sendDocument returns null for an unchanged doc).++ */++async function reconcile({ ctx, pageSize = DEFAULT_PAGE_SIZE, maxPages = DEFAULT_MAX_PAGES, log = console.log } = {}) {++ if (!ctx || !ctx.merchants || !ctx.coupons || !ctx.publication) throw new Error('reconcile needs the running service context (see server/app.js createApp)');++ let sent = 0;++ let unchanged = 0;++ const account = (doc) => { if (doc) sent++; else unchanged++; };++++ for (let page = 0; page < maxPages; page++) {++ const merchants = await ctx.merchants.listAll({ limit: pageSize, offset: page * pageSize });++ for (const m of merchants) {++ account(await ctx.publication.sendDocument(ctx.publication.merchantDocument(m, await ctx.coupons.activeCount(m.id)), { page: ctx.publication.merchantPath(m) }));++ }++ if (merchants.length < pageSize) break;++ }++ for (let page = 0; page < maxPages; page++) {++ const coupons = await ctx.coupons.allPage({ limit: pageSize, offset: page * pageSize });++ for (const c of coupons) {++ const m = await ctx.merchants.byId(c.merchant_id);++ const doc = ctx.publication.couponDocument(c, m, {++ restrictionsText: ctx.coupons.restrictionsText(await ctx.coupons.restrictionsOf(c.id)),++ sourceRefs: await ctx.coupons.sourceRefs(c.id),++ });++ account(await ctx.publication.sendDocument(doc, { page: ctx.publication.couponPath(c) }));++ }++ if (coupons.length < pageSize) break;++ }++ log(`reconcile: ${sent} sent, ${unchanged} unchanged`);++ return { sent, unchanged };++}++++/** CLI. Reconcile boots the app; subscribing needs only the config. */++async function main(argv, { fetchImpl = globalThis.fetch, log = console.log } = {}) {++ const args = parseArgs(argv);++ const config = load();++ if (!args.reconcile) return await subscribe({ config, endpoint: args.endpoint || defaultEndpoint(config), fetchImpl, log });++ const { createApp } = require('../server/app');++ const { ctx } = await createApp({ config });++ try {++ const report = await reconcile({ ctx, pageSize: args.pageSize, maxPages: args.maxPages, log });++ // The relay is not started here, so publish once; without EVENTS_URL the running service relays.++ if (ctx.outbox.enabled) await ctx.outbox.outbox.flush();++ else log('events relay off: the rows wait in event_outbox for the running service');++ return report;++ } finally {++ await ctx.outbox.stop();++ await ctx.store.close();++ }++}++++if (require.main === module) {++ main(process.argv.slice(2)).then(() => process.exit(0), (err) => {++ console.error(`subscribe failed: ${err.message}`);++ process.exit(1);++ });++}++++module.exports = { subscribe, reconcile, parseArgs, PATTERNS, defaultEndpoint, DEFAULT_PAGE_SIZE, DEFAULT_MAX_PAGES };@@@ -97,7 +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).++ // Webhook secrets for inbound event deliveries (signed by OpenVibe.Events) and the value++ // scripts/subscribe.js hands to Events when it creates the Sources subscriptions.webhookSecrets: list(env.COUPONS_EVENTS_SECRET),},@@@ -184,6 +184,8 @@ function createCoupons({ store, merchants, publication, outbox = null }) {recentActive: db.prepare(`SELECT c.* FROM coupons c JOIN coupon_merchants m ON m.id = c.merchant_id WHERE ${ACTIVE_SQL}ORDER BY c.created_at DESC LIMIT @limit`),allActive: db.prepare(`SELECT c.* FROM coupons c JOIN coupon_merchants m ON m.id = c.merchant_id WHERE ${ACTIVE_SQL} ORDER BY c.created_at DESC LIMIT 50000`),++ activePage: db.prepare(`SELECT c.* FROM coupons c JOIN coupon_merchants m ON m.id = c.merchant_id WHERE ${ACTIVE_SQL} ORDER BY c.id LIMIT @limit OFFSET @offset`),++ allPage: db.prepare('SELECT c.* FROM coupons c ORDER BY c.id LIMIT @limit OFFSET @offset'),dueExpiry: db.prepare(`SELECT id FROM coupons WHERE expires_at IS NOT NULL AND expires_at <= ? AND status NOT IN ('expired','disabled') LIMIT 500`),decaying: db.prepare(`SELECT DISTINCT c.id FROM coupons c LEFT JOIN coupon_validation_reports r ON r.coupon_id = c.idWHERE c.status NOT IN ('expired','disabled') AND (c.confidence IS NOT NULL OR c.status <> 'unknown' OR r.updated_at > ?) LIMIT 2000`),@@@ -521,12 +523,16 @@ function createCoupons({ store, merchants, publication, outbox = null }) {return {ACTIVE_SQL, parseSubmission, parseExpiry, parseRestrictions, formatMinor, restrictionsText,get, submit, checkSubmissionRate, importFromSource, withdrawSourceItem, recompute, setStatus, approve, setExpiry, addMerchantHint, sweep, view,-- restrictionsOf, reportCounts, activeCount, activeCounts, syncIndex, syncMerchant,++ restrictionsOf, sourceRefs, reportCounts, activeCount, activeCounts, syncIndex, syncMerchant,history: async (id) => await q.historyFor.all(id),active: async (merchantId) => await q.active.all({ merchant: merchantId, now: store.now() }),recentlyEnded: async (merchantId) => await q.recentlyEnded.all({ merchant: merchantId, now: store.now() }),recentActive: async (limit = 30) => await q.recentActive.all({ now: store.now(), limit }),allActive: async () => await q.allActive.all({ now: store.now() }),++ /** One page of the active-results set (scripts/subscribe.js --reconcile). */++ activePage: async ({ limit = 200, offset = 0 } = {}) => await q.activePage.all({ now: store.now(), limit, offset }),++ /** One page of every code, any status (scripts/subscribe.js --reconcile: a code that left active results gets its tombstone). */++ allPage: async ({ limit = 200, offset = 0 } = {}) => await q.allPage.all({ limit, offset }),pendingReview: async () => await q.pendingReview.all(),pendingOfMerchant: async (merchantId) => (await q.byMerchantPending.all(merchantId)).map((r) => r.id),/** Publish the codes that waited only because their merchant was pending (not AI or source codes). */@@@ -38,6 +38,7 @@ function createMerchants({ store }) {rulesFor: db.prepare(`SELECT d.*, m.status AS merchant_status FROM coupon_merchant_domains d JOIN coupon_merchants m ON m.id = d.merchant_idWHERE d.registrable_domain = ?`),listActive: db.prepare(`SELECT * FROM coupon_merchants WHERE status = 'active' ORDER BY lower(name) LIMIT ? OFFSET ?`),++ listAll: db.prepare('SELECT * FROM coupon_merchants ORDER BY id LIMIT ? OFFSET ?'),countActive: db.prepare("SELECT COUNT(*) AS n FROM coupon_merchants WHERE status = 'active'"),search: db.prepare(`SELECT m.* FROM coupon_merchants mWHERE m.status = 'active' AND (m.name ILIKE @q ESCAPE '\\'@@@ -138,6 +139,7 @@ function createMerchants({ store }) {resolve,ownsUrl,listActive: async ({ limit = 100, offset = 0 } = {}) => await q.listActive.all(limit, offset),++ listAll: async ({ limit = 200, offset = 0 } = {}) => await q.listAll.all(limit, offset),countActive: async () => (await q.countActive.get()).n,async search(text) {const t = String(text || '').trim().toLowerCase().slice(0, 60);nnew file mode 100644@@@ -0,0 +1,143 @@++'use strict';++/**++ * Plan T9 (J7a/J7a') — scripts/subscribe.js.++ *++ * - it creates one OpenVibe.Events subscription per Sources pattern, as the coupons principal,++ * with the configured delivery secret; an existing (409) subscription is reported, not retried,++ * and nothing secret is ever printed;++ * - --reconcile is one idempotent, bounded pass over every merchant and every code (any status, so a++ * code that left active results gets its tombstone), reporting { sent, unchanged }.++ */++const assert = require('assert');++const { load } = require('../server/config');++const { subscribe, reconcile, parseArgs, PATTERNS, defaultEndpoint, DEFAULT_PAGE_SIZE } = require('../scripts/subscribe');++const { boot, check, done } = require('./helpers/boot');++++const SECRET = 's'.repeat(64);++const CLIENT_SECRET = 'c'.repeat(48);++const ENV = {++ EVENTS_URL: 'http://events.test/', OV_NETWORK_INTERNAL_URL: 'http://network.test',++ OV_OAUTH_CLIENT_ID: 'coupons', OV_OAUTH_CLIENT_SECRET: CLIENT_SECRET,++ COUPONS_EVENTS_SECRET: SECRET, PORT: '4850',++};++++/** A stub Network token endpoint and Events subscriptions endpoint; records every call. */++function stub({ tokenStatus = 200, createStatus = 201 } = {}) {++ const calls = [];++ const reply = (status, body) => ({ status, ok: status < 300, json: async () => body });++ const fetchImpl = async (url, opts = {}) => {++ calls.push({ url, method: opts.method, headers: opts.headers || {}, body: opts.body || null });++ if (url === 'http://network.test/oauth/token') {++ const p = new URLSearchParams(opts.body);++ assert.strictEqual(p.get('grant_type'), 'client_credentials');++ assert.strictEqual(p.get('client_id'), 'coupons');++ assert.strictEqual(p.get('audience'), 'openvibe.events');++ assert.strictEqual(p.get('scope'), 'events.subscription.manage');++ return tokenStatus === 200++ ? reply(200, { access_token: 'tok-coupons', token_type: 'Bearer', expires_in: 300 })++ : reply(tokenStatus, { error: 'invalid_client' });++ }++ if (url === 'http://events.test/api/v1/subscriptions' && opts.method === 'POST') {++ const b = JSON.parse(opts.body);++ if (createStatus === 409) return reply(409, { code: 'events.subscription_exists', subscription_id: 'sub_existing' });++ return reply(201, { id: `sub_${b.topic_pattern === PATTERNS[0] ? 1 : 2}` });++ }++ return reply(404, { code: 'not_found' });++ };++ return { fetchImpl, calls };++}++++const logs = () => { const lines = []; const log = (s) => lines.push(String(s)); log.lines = lines; return log; };++++(async () => {++ await check('subscribes both Sources patterns as coupons with the secret; nothing secret is printed', async () => {++ const s = stub();++ const log = logs();++ const out = await subscribe({ config: load(ENV), fetchImpl: s.fetchImpl, log });++ assert.deepStrictEqual(out, [++ { pattern: 'sources.item.*', subscription_id: 'sub_1', existed: false },++ { pattern: 'sources.fetch.failed', subscription_id: 'sub_2', existed: false },++ ]);++ const posts = s.calls.filter((c) => c.method === 'POST' && c.url === 'http://events.test/api/v1/subscriptions');++ assert.strictEqual(posts.length, 2, 'one subscription per pattern');++ assert.deepStrictEqual(JSON.parse(posts[0].body), {++ topic_pattern: 'sources.item.*', endpoint: 'http://127.0.0.1:4850/internal/events', secret: SECRET,++ });++ assert.strictEqual(posts[1].headers.Authorization, 'Bearer tok-coupons');++ assert.match(log.lines.join('\n'), /subscribed: sub_1 \(sources\.item\.\* → http:\/\/127\.0\.0\.1:4850\/internal\/events\)/);++ for (const l of log.lines) for (const v of [SECRET, CLIENT_SECRET, 'tok-coupons']) assert.ok(!l.includes(v), `printed a secret: ${l}`);++ });++++ await check('an existing subscription is reported ("exists"), not duplicated; a 409 race is handled', async () => {++ const s = stub({ createStatus: 409 });++ const log = logs();++ const out = await subscribe({ config: load(ENV), fetchImpl: s.fetchImpl, log });++ assert.deepStrictEqual(out.map((o) => o.subscription_id), ['sub_existing', 'sub_existing']);++ assert.match(log.lines.join('\n'), /subscription exists: sub_existing/);++ assert.strictEqual(s.calls.filter((c) => c.method === 'POST' && c.url === 'http://events.test/api/v1/subscriptions').length, 2, 'no retry loop');++ });++++ await check('refuses before calling anything when the environment is not usable; a refused token is an error', async () => {++ let called = 0;++ const fetchImpl = async () => { called++; return { status: 500, ok: false, json: async () => ({}) }; };++ await assert.rejects(subscribe({ config: load({ ...ENV, EVENTS_URL: '' }), fetchImpl, log: () => {} }), /EVENTS_URL is not set/);++ await assert.rejects(subscribe({ config: load({ ...ENV, COUPONS_EVENTS_SECRET: 'short' }), fetchImpl, log: () => {} }), /COUPONS_EVENTS_SECRET must be set/);++ await assert.rejects(subscribe({ config: load({ ...ENV, OV_OAUTH_CLIENT_SECRET: '' }), fetchImpl, log: () => {} }), /OV_OAUTH_CLIENT_SECRET is not set/);++ assert.strictEqual(called, 0);++ await assert.rejects(subscribe({ config: load(ENV), fetchImpl: stub({ tokenStatus: 401 }).fetchImpl, log: () => {} }), /token endpoint 401/);++ });++++ await check('arguments', () => {++ assert.strictEqual(parseArgs([]).reconcile, false);++ assert.strictEqual(parseArgs(['--reconcile']).reconcile, true);++ assert.strictEqual(parseArgs(['--endpoint', 'http://x/internal/events']).endpoint, 'http://x/internal/events');++ const o = parseArgs(['--page-size', '50', '--max-pages', '3']);++ assert.strictEqual(o.pageSize, 50);++ assert.strictEqual(o.maxPages, 3);++ assert.strictEqual(DEFAULT_PAGE_SIZE, 200);++ assert.throws(() => parseArgs(['--nope']), /unknown argument/);++ assert.throws(() => parseArgs(['--page-size', '0']), /--page-size must be a positive integer/);++ assert.throws(() => parseArgs(['--endpoint']), /--endpoint needs a value/);++ });++++ await check('--reconcile re-stamps every Search document, the next pass is unchanged, and a code that left active results is tombstoned', async () => {++ const t = await boot();++ try {++ const alice = t.network.addUser('alice');++ const a = await t.merchant({ name: 'Reconcile A', host: 'reconcile-a.com' });++ const b = await t.merchant({ name: 'Reconcile B', host: 'reconcile-b.com' });++ await t.submit(alice, { merchant_id: a.id, code: 'RA10', title: 'Ten off A' });++ await t.submit(alice, { merchant_id: b.id, code: 'RB10', title: 'Ten off B' });++ // A Search index that lost its publications: no revisions survive.++ await t.ctx.store.db.query('DELETE FROM coupons_index_revisions');++ assert.deepStrictEqual(await reconcile({ ctx: t.ctx, log: () => {} }), { sent: 4, unchanged: 0 }, '2 merchants + 2 public coupons');++ assert.deepStrictEqual(await reconcile({ ctx: t.ctx, log: () => {} }), { sent: 0, unchanged: 4 }, 'idempotent: nothing left to send');++ // A code that left active results behind the index's back still reaches Search.++ await t.ctx.store.db.query("UPDATE coupons SET status = 'disabled' WHERE code = 'RA10'");++ const before = Number((await t.ctx.store.db.query("SELECT COUNT(*) AS n FROM event_outbox WHERE envelope->>'event_type' = 'coupons.index_document.deleted'")).rows[0].n);++ assert.deepStrictEqual(await reconcile({ ctx: t.ctx, log: () => {} }), { sent: 2, unchanged: 2 }, 'its tombstone + its merchant (one fewer active code)');++ const after = Number((await t.ctx.store.db.query("SELECT COUNT(*) AS n FROM event_outbox WHERE envelope->>'event_type' = 'coupons.index_document.deleted'")).rows[0].n);++ assert.strictEqual(after - before, 1, 'the disabled code is tombstoned');++ } finally { await t.close(); }++ });++++ await check('--page-size and --max-pages bound one pass to a page per table', async () => {++ const t = await boot();++ try {++ const alice = t.network.addUser('alice');++ for (const n of ['C', 'D']) {++ const m = await t.merchant({ name: `Bounded ${n}`, host: `bounded-${n.toLowerCase()}.com` });++ await t.submit(alice, { merchant_id: m.id, code: `B${n}10`, title: `Ten off ${n}` });++ }++ await t.ctx.store.db.query('DELETE FROM coupons_index_revisions');++ assert.deepStrictEqual(await reconcile({ ctx: t.ctx, pageSize: 1, maxPages: 1, log: () => {} }), { sent: 2, unchanged: 0 }, 'one merchant page + one coupon page');++ } finally { await t.close(); }++ });++++ await check('the default endpoint is this service\'s port 4850 webhook', () => {++ assert.strictEqual(defaultEndpoint(load(ENV)), 'http://127.0.0.1:4850/internal/events');++ assert.strictEqual(load(ENV).port, 4850);++ });++++ done();++})();
@@ -273,6 +273,11 @@ The charter's `coupons.created|updated|expired|disabled` are these `coupons.coup
@@ -273,6 +273,11 @@ The charter's `coupons.created|updated|expired|disabled` are these `coupons.coup
types. Every envelope validates as `events.event-envelope@1` (tested), with actor
types. Every envelope validates as `events.event-envelope@1` (tested), with actor
`{type: service, id: coupons}`, except `coupons.moderation.action`, whose actor is the staff member.
`{type: service, id: coupons}`, except `coupons.moderation.action`, whose actor is the staff member.
+Coupons subscribes to `sources.item.*` and `sources.fetch.failed` with `scripts/subscribe.js`
+(endpoint `/internal/events`, delivery signed with `COUPONS_EVENTS_SECRET`; the import's cursor pull
+works without it). `scripts/subscribe.js --reconcile` re-stamps every merchant and every code (any status, so a code that
+left active results gets its tombstone) in one bounded, idempotent pass and reports `{ sent, unchanged }`.
+
## Capabilities (registered in openvibe-contracts v0.25.0)
## Capabilities (registered in openvibe-contracts v0.25.0)
Service tokens use audience `openvibe.coupons`, with one capability per route:
Service tokens use audience `openvibe.coupons`, with one capability per route:
@@ -317,6 +322,8 @@ What Coupons calls elsewhere, and with which grant, is under
@@ -317,6 +322,8 @@ What Coupons calls elsewhere, and with which grant, is under
Each grant is `[client, capability, audience]`:
Each grant is `[client, capability, audience]`:
- `[coupons, events.event.publish, openvibe.events]`
- `[coupons, events.event.publish, openvibe.events]`
+- `[coupons, events.subscription.manage, openvibe.events]` (only to create the Sources
+ subscriptions: `scripts/subscribe.js`)
- `[coupons, sources.item.read, openvibe.sources]` (only if the Sources import is turned on)
- `[coupons, sources.item.read, openvibe.sources]` (only if the Sources import is turned on)
- For OpenVibe.AI to deliver `coupons.extract_coupon` drafts:
- For OpenVibe.AI to deliver `coupons.extract_coupon` drafts:
`[ai, coupons.coupon.submit, openvibe.coupons]`
`[ai, coupons.coupon.submit, openvibe.coupons]`
new file mode 100644
new file mode 100644
@@ -0,0 +1,146 @@
@@ -0,0 +1,146 @@
+#!/usr/bin/env node
+'use strict';
+/**
+ * Coupons' OpenVibe.Events side: create the Sources subscriptions the import loop wakes on, or
+ * reconcile the OpenVibe.Search index in one bounded pass.
+ *
+ * node scripts/subscribe.js [--endpoint http://127.0.0.1:4850/internal/events]
+ * node scripts/subscribe.js --reconcile [--page-size 200] [--max-pages 200]
+ *
+ * Subscribe reads the environment (.env; the service itself gets /etc/openvibe/coupons.env from systemd):
+ * EVENTS_URL, OV_NETWORK_INTERNAL_URL, OV_OAUTH_CLIENT_ID, OV_OAUTH_CLIENT_SECRET (the coupons
+ * principal needs events.subscription.manage for audience openvibe.events)
+ * COUPONS_EVENTS_SECRET the delivery signing secret; the first value is handed to Events, so
+ * generate it before running this (e.g. `openssl rand -hex 32`). Nothing
+ * secret is printed.
+ * Patterns: sources.item.* and sources.fetch.failed. Only sources.item.created runs the import; the
+ * rest are acknowledged and the cursor pull catches up either way. An existing identical subscription is reported
+ * ("exists"), never duplicated.
+ *
+ * --reconcile pages coupon_merchants and coupons, every status of both (a shop or code that left active
+ * results must reach Search as unpublished or as a tombstone), re-stamping each Search document through
+ * ctx.publication.sendDocument — an unchanged document costs nothing. The rows land in event_outbox;
+ * when the relay is configured the CLI publishes them once before exiting, otherwise the running
+ * service relays them. It is idempotent and bounded by --page-size/--max-pages; the report is
+ * { sent, unchanged }.
+ */
+require('dotenv').config();
+const { serviceAuth } = require('openvibe-contracts');
+const { load } = require('../server/config');
+
+const PATTERNS = ['sources.item.*', 'sources.fetch.failed'];
+const DEFAULT_PAGE_SIZE = 200;
+const DEFAULT_MAX_PAGES = 200;
+
+/** The endpoint Events delivers to: this service's own inbound webhook. */
+const defaultEndpoint = (config) => `http://127.0.0.1:${config.port}/internal/events`;
+
+function parseArgs(argv) {
+ const o = { reconcile: false, endpoint: null, pageSize: DEFAULT_PAGE_SIZE, maxPages: DEFAULT_MAX_PAGES };
+ for (let i = 0; i < argv.length; i++) {
+ const a = argv[i];
+ if (a === '--reconcile') o.reconcile = true;
+ else if (a === '--endpoint') o.endpoint = argv[++i];
+ else if (a === '--page-size') o.pageSize = Number(argv[++i]);
+ else if (a === '--max-pages') o.maxPages = Number(argv[++i]);
+ else throw new Error(`unknown argument ${a}`);
+ }
+ if (o.endpoint !== null && !o.endpoint) throw new Error('--endpoint needs a value');
+ if (!Number.isInteger(o.pageSize) || o.pageSize < 1) throw new Error('--page-size must be a positive integer');
+ if (!Number.isInteger(o.maxPages) || o.maxPages < 1) throw new Error('--max-pages must be a positive integer');
+ return o;
+}
+
+/**
+ * Create one subscription per pattern, as the coupons principal. The caller's delivery secret is
+ * handed to Events here; `fetchImpl` is injectable so the test can stub Network and Events.
+ */
+async function subscribe({ config, endpoint = defaultEndpoint(config), fetchImpl = globalThis.fetch, log = console.log } = {}) {
+ if (!config.events.url) throw new Error('EVENTS_URL is not set');
+ const secret = (config.events.webhookSecrets || [])[0];
+ if (!secret || secret.length < 32) throw new Error('COUPONS_EVENTS_SECRET must be set (32+ characters) before subscribing');
+ if (!config.oauth.clientSecret) throw new Error('OV_OAUTH_CLIENT_SECRET is not set');
+ const tokens = serviceAuth.createTokenClient({
+ tokenUrl: `${config.networkInternalUrl}/oauth/token`, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret,
+ audience: 'openvibe.events', scope: 'events.subscription.manage', fetchImpl,
+ });
+ const out = [];
+ for (const pattern of PATTERNS) {
+ const res = await fetchImpl(`${config.events.url}/api/v1/subscriptions`, {
+ method: 'POST',
+ headers: { 'Content-Type': 'application/json', ...(await tokens.authHeaders()) },
+ body: JSON.stringify({ topic_pattern: pattern, endpoint, secret }),
+ });
+ const body = await res.json().catch(() => ({}));
+ if (res.status === 409 && body.subscription_id) {
+ log(`subscription exists: ${body.subscription_id} (${pattern} → ${endpoint})`);
+ out.push({ pattern, subscription_id: body.subscription_id, existed: true });
+ continue;
+ }
+ if (!res.ok) throw new Error(`Events answered ${res.status} for ${pattern}: ${body.code || ''} ${body.detail || ''}`.trim());
+ log(`subscribed: ${body.id} (${pattern} → ${endpoint}). Replay history with POST /api/v1/deliveries/replay { subscription_id, from_seq }.`);
+ out.push({ pattern, subscription_id: body.id, existed: false });
+ }
+ return out;
+}
+
+/**
+ * One bounded pass over every merchant and every code, re-stamping each Search document into the
+ * outbox. → { sent, unchanged } (sendDocument returns null for an unchanged doc).
+ */
+async function reconcile({ ctx, pageSize = DEFAULT_PAGE_SIZE, maxPages = DEFAULT_MAX_PAGES, log = console.log } = {}) {
+ if (!ctx || !ctx.merchants || !ctx.coupons || !ctx.publication) throw new Error('reconcile needs the running service context (see server/app.js createApp)');
+ let sent = 0;
+ let unchanged = 0;
+ const account = (doc) => { if (doc) sent++; else unchanged++; };
+
+ for (let page = 0; page < maxPages; page++) {
+ const merchants = await ctx.merchants.listAll({ limit: pageSize, offset: page * pageSize });
+ for (const m of merchants) {
+ account(await ctx.publication.sendDocument(ctx.publication.merchantDocument(m, await ctx.coupons.activeCount(m.id)), { page: ctx.publication.merchantPath(m) }));
+ }
+ if (merchants.length < pageSize) break;
+ }
+ for (let page = 0; page < maxPages; page++) {
+ const coupons = await ctx.coupons.allPage({ limit: pageSize, offset: page * pageSize });
+ for (const c of coupons) {
+ const m = await ctx.merchants.byId(c.merchant_id);
+ const doc = ctx.publication.couponDocument(c, m, {
+ restrictionsText: ctx.coupons.restrictionsText(await ctx.coupons.restrictionsOf(c.id)),
+ sourceRefs: await ctx.coupons.sourceRefs(c.id),
+ });
+ account(await ctx.publication.sendDocument(doc, { page: ctx.publication.couponPath(c) }));
+ }
+ if (coupons.length < pageSize) break;
+ }
+ log(`reconcile: ${sent} sent, ${unchanged} unchanged`);
+ return { sent, unchanged };
+}
+
+/** CLI. Reconcile boots the app; subscribing needs only the config. */
+async function main(argv, { fetchImpl = globalThis.fetch, log = console.log } = {}) {
+ const args = parseArgs(argv);
+ const config = load();
+ if (!args.reconcile) return await subscribe({ config, endpoint: args.endpoint || defaultEndpoint(config), fetchImpl, log });
+ const { createApp } = require('../server/app');
+ const { ctx } = await createApp({ config });
+ try {
+ const report = await reconcile({ ctx, pageSize: args.pageSize, maxPages: args.maxPages, log });
+ // The relay is not started here, so publish once; without EVENTS_URL the running service relays.
+ if (ctx.outbox.enabled) await ctx.outbox.outbox.flush();
+ else log('events relay off: the rows wait in event_outbox for the running service');
+ return report;
+ } finally {
+ await ctx.outbox.stop();
+ await ctx.store.close();
+ }
+}
+
+if (require.main === module) {
+ main(process.argv.slice(2)).then(() => process.exit(0), (err) => {
+ console.error(`subscribe failed: ${err.message}`);
+ process.exit(1);
+ });
+}
+
+module.exports = { subscribe, reconcile, parseArgs, PATTERNS, defaultEndpoint, DEFAULT_PAGE_SIZE, DEFAULT_MAX_PAGES };
@@ -97,7 +97,8 @@ function load(env = process.env) {
@@ -97,7 +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).
+ // Webhook secrets for inbound event deliveries (signed by OpenVibe.Events) and the value
+ // scripts/subscribe.js hands to Events when it creates the Sources subscriptions.
webhookSecrets: list(env.COUPONS_EVENTS_SECRET),
webhookSecrets: list(env.COUPONS_EVENTS_SECRET),
},
},
@@ -184,6 +184,8 @@ function createCoupons({ store, merchants, publication, outbox = null }) {
@@ -184,6 +184,8 @@ function createCoupons({ store, merchants, publication, outbox = null }) {
recentActive: db.prepare(`SELECT c.* FROM coupons c JOIN coupon_merchants m ON m.id = c.merchant_id WHERE ${ACTIVE_SQL}
recentActive: db.prepare(`SELECT c.* FROM coupons c JOIN coupon_merchants m ON m.id = c.merchant_id WHERE ${ACTIVE_SQL}
ORDER BY c.created_at DESC LIMIT @limit`),
ORDER BY c.created_at DESC LIMIT @limit`),
allActive: db.prepare(`SELECT c.* FROM coupons c JOIN coupon_merchants m ON m.id = c.merchant_id WHERE ${ACTIVE_SQL} ORDER BY c.created_at DESC LIMIT 50000`),
allActive: db.prepare(`SELECT c.* FROM coupons c JOIN coupon_merchants m ON m.id = c.merchant_id WHERE ${ACTIVE_SQL} ORDER BY c.created_at DESC LIMIT 50000`),
+ activePage: db.prepare(`SELECT c.* FROM coupons c JOIN coupon_merchants m ON m.id = c.merchant_id WHERE ${ACTIVE_SQL} ORDER BY c.id LIMIT @limit OFFSET @offset`),
+ allPage: db.prepare('SELECT c.* FROM coupons c ORDER BY c.id LIMIT @limit OFFSET @offset'),
dueExpiry: db.prepare(`SELECT id FROM coupons WHERE expires_at IS NOT NULL AND expires_at <= ? AND status NOT IN ('expired','disabled') LIMIT 500`),
dueExpiry: db.prepare(`SELECT id FROM coupons WHERE expires_at IS NOT NULL AND expires_at <= ? AND status NOT IN ('expired','disabled') LIMIT 500`),
decaying: db.prepare(`SELECT DISTINCT c.id FROM coupons c LEFT JOIN coupon_validation_reports r ON r.coupon_id = c.id
decaying: db.prepare(`SELECT DISTINCT c.id FROM coupons c LEFT JOIN coupon_validation_reports r ON r.coupon_id = c.id
WHERE c.status NOT IN ('expired','disabled') AND (c.confidence IS NOT NULL OR c.status <> 'unknown' OR r.updated_at > ?) LIMIT 2000`),
WHERE c.status NOT IN ('expired','disabled') AND (c.confidence IS NOT NULL OR c.status <> 'unknown' OR r.updated_at > ?) LIMIT 2000`),
@@ -521,12 +523,16 @@ function createCoupons({ store, merchants, publication, outbox = null }) {
@@ -521,12 +523,16 @@ function createCoupons({ store, merchants, publication, outbox = null }) {
return {
return {
ACTIVE_SQL, parseSubmission, parseExpiry, parseRestrictions, formatMinor, restrictionsText,
ACTIVE_SQL, parseSubmission, parseExpiry, parseRestrictions, formatMinor, restrictionsText,
get, submit, checkSubmissionRate, importFromSource, withdrawSourceItem, recompute, setStatus, approve, setExpiry, addMerchantHint, sweep, view,
get, submit, checkSubmissionRate, importFromSource, withdrawSourceItem, recompute, setStatus, approve, setExpiry, addMerchantHint, sweep, view,
- restrictionsOf, reportCounts, activeCount, activeCounts, syncIndex, syncMerchant,
+ restrictionsOf, sourceRefs, reportCounts, activeCount, activeCounts, syncIndex, syncMerchant,
history: async (id) => await q.historyFor.all(id),
history: async (id) => await q.historyFor.all(id),
active: async (merchantId) => await q.active.all({ merchant: merchantId, now: store.now() }),
active: async (merchantId) => await q.active.all({ merchant: merchantId, now: store.now() }),
recentlyEnded: async (merchantId) => await q.recentlyEnded.all({ merchant: merchantId, now: store.now() }),
recentlyEnded: async (merchantId) => await q.recentlyEnded.all({ merchant: merchantId, now: store.now() }),
recentActive: async (limit = 30) => await q.recentActive.all({ now: store.now(), limit }),
recentActive: async (limit = 30) => await q.recentActive.all({ now: store.now(), limit }),
allActive: async () => await q.allActive.all({ now: store.now() }),
allActive: async () => await q.allActive.all({ now: store.now() }),
+ /** One page of the active-results set (scripts/subscribe.js --reconcile). */
+ activePage: async ({ limit = 200, offset = 0 } = {}) => await q.activePage.all({ now: store.now(), limit, offset }),
+ /** One page of every code, any status (scripts/subscribe.js --reconcile: a code that left active results gets its tombstone). */
+ allPage: async ({ limit = 200, offset = 0 } = {}) => await q.allPage.all({ limit, offset }),
pendingReview: async () => await q.pendingReview.all(),
pendingReview: async () => await q.pendingReview.all(),
pendingOfMerchant: async (merchantId) => (await q.byMerchantPending.all(merchantId)).map((r) => r.id),
pendingOfMerchant: async (merchantId) => (await q.byMerchantPending.all(merchantId)).map((r) => r.id),
/** Publish the codes that waited only because their merchant was pending (not AI or source codes). */
/** Publish the codes that waited only because their merchant was pending (not AI or source codes). */
@@ -38,6 +38,7 @@ function createMerchants({ store }) {
@@ -38,6 +38,7 @@ function createMerchants({ store }) {
rulesFor: db.prepare(`SELECT d.*, m.status AS merchant_status FROM coupon_merchant_domains d JOIN coupon_merchants m ON m.id = d.merchant_id
rulesFor: db.prepare(`SELECT d.*, m.status AS merchant_status FROM coupon_merchant_domains d JOIN coupon_merchants m ON m.id = d.merchant_id
WHERE d.registrable_domain = ?`),
WHERE d.registrable_domain = ?`),
listActive: db.prepare(`SELECT * FROM coupon_merchants WHERE status = 'active' ORDER BY lower(name) LIMIT ? OFFSET ?`),
listActive: db.prepare(`SELECT * FROM coupon_merchants WHERE status = 'active' ORDER BY lower(name) LIMIT ? OFFSET ?`),
+ listAll: db.prepare('SELECT * FROM coupon_merchants ORDER BY id LIMIT ? OFFSET ?'),
countActive: db.prepare("SELECT COUNT(*) AS n FROM coupon_merchants WHERE status = 'active'"),
countActive: db.prepare("SELECT COUNT(*) AS n FROM coupon_merchants WHERE status = 'active'"),
search: db.prepare(`SELECT m.* FROM coupon_merchants m
search: db.prepare(`SELECT m.* FROM coupon_merchants m
WHERE m.status = 'active' AND (m.name ILIKE @q ESCAPE '\\'
WHERE m.status = 'active' AND (m.name ILIKE @q ESCAPE '\\'
@@ -138,6 +139,7 @@ function createMerchants({ store }) {
@@ -138,6 +139,7 @@ function createMerchants({ store }) {
resolve,
resolve,
ownsUrl,
ownsUrl,
listActive: async ({ limit = 100, offset = 0 } = {}) => await q.listActive.all(limit, offset),
listActive: async ({ limit = 100, offset = 0 } = {}) => await q.listActive.all(limit, offset),
+ listAll: async ({ limit = 200, offset = 0 } = {}) => await q.listAll.all(limit, offset),
countActive: async () => (await q.countActive.get()).n,
countActive: async () => (await q.countActive.get()).n,
async search(text) {
async search(text) {
const t = String(text || '').trim().toLowerCase().slice(0, 60);
const t = String(text || '').trim().toLowerCase().slice(0, 60);
new file mode 100644
new file mode 100644
@@ -0,0 +1,143 @@
@@ -0,0 +1,143 @@
+'use strict';
+/**
+ * Plan T9 (J7a/J7a') — scripts/subscribe.js.
+ *
+ * - it creates one OpenVibe.Events subscription per Sources pattern, as the coupons principal,
+ * with the configured delivery secret; an existing (409) subscription is reported, not retried,
+ * and nothing secret is ever printed;
+ * - --reconcile is one idempotent, bounded pass over every merchant and every code (any status, so a
+ * code that left active results gets its tombstone), reporting { sent, unchanged }.
+ */
+const assert = require('assert');
+const { load } = require('../server/config');
+const { subscribe, reconcile, parseArgs, PATTERNS, defaultEndpoint, DEFAULT_PAGE_SIZE } = require('../scripts/subscribe');
+const { boot, check, done } = require('./helpers/boot');
+
+const SECRET = 's'.repeat(64);
+const CLIENT_SECRET = 'c'.repeat(48);
+const ENV = {
+ EVENTS_URL: 'http://events.test/', OV_NETWORK_INTERNAL_URL: 'http://network.test',
+ OV_OAUTH_CLIENT_ID: 'coupons', OV_OAUTH_CLIENT_SECRET: CLIENT_SECRET,
+ COUPONS_EVENTS_SECRET: SECRET, PORT: '4850',
+};
+
+/** A stub Network token endpoint and Events subscriptions endpoint; records every call. */
+function stub({ tokenStatus = 200, createStatus = 201 } = {}) {
+ const calls = [];
+ const reply = (status, body) => ({ status, ok: status < 300, json: async () => body });
+ const fetchImpl = async (url, opts = {}) => {
+ calls.push({ url, method: opts.method, headers: opts.headers || {}, body: opts.body || null });
+ if (url === 'http://network.test/oauth/token') {
+ const p = new URLSearchParams(opts.body);
+ assert.strictEqual(p.get('grant_type'), 'client_credentials');
+ assert.strictEqual(p.get('client_id'), 'coupons');
+ assert.strictEqual(p.get('audience'), 'openvibe.events');
+ assert.strictEqual(p.get('scope'), 'events.subscription.manage');
+ return tokenStatus === 200
+ ? reply(200, { access_token: 'tok-coupons', token_type: 'Bearer', expires_in: 300 })
+ : reply(tokenStatus, { error: 'invalid_client' });
+ }
+ if (url === 'http://events.test/api/v1/subscriptions' && opts.method === 'POST') {
+ const b = JSON.parse(opts.body);
+ if (createStatus === 409) return reply(409, { code: 'events.subscription_exists', subscription_id: 'sub_existing' });
+ return reply(201, { id: `sub_${b.topic_pattern === PATTERNS[0] ? 1 : 2}` });
+ }
+ return reply(404, { code: 'not_found' });
+ };
+ return { fetchImpl, calls };
+}
+
+const logs = () => { const lines = []; const log = (s) => lines.push(String(s)); log.lines = lines; return log; };
+
+(async () => {
+ await check('subscribes both Sources patterns as coupons with the secret; nothing secret is printed', async () => {
+ const s = stub();
+ const log = logs();
+ const out = await subscribe({ config: load(ENV), fetchImpl: s.fetchImpl, log });
+ assert.deepStrictEqual(out, [
+ { pattern: 'sources.item.*', subscription_id: 'sub_1', existed: false },
+ { pattern: 'sources.fetch.failed', subscription_id: 'sub_2', existed: false },
+ ]);
+ const posts = s.calls.filter((c) => c.method === 'POST' && c.url === 'http://events.test/api/v1/subscriptions');
+ assert.strictEqual(posts.length, 2, 'one subscription per pattern');
+ assert.deepStrictEqual(JSON.parse(posts[0].body), {
+ topic_pattern: 'sources.item.*', endpoint: 'http://127.0.0.1:4850/internal/events', secret: SECRET,
+ });
+ assert.strictEqual(posts[1].headers.Authorization, 'Bearer tok-coupons');
+ assert.match(log.lines.join('\n'), /subscribed: sub_1 \(sources\.item\.\* → http:\/\/127\.0\.0\.1:4850\/internal\/events\)/);
+ for (const l of log.lines) for (const v of [SECRET, CLIENT_SECRET, 'tok-coupons']) assert.ok(!l.includes(v), `printed a secret: ${l}`);
+ });
+
+ await check('an existing subscription is reported ("exists"), not duplicated; a 409 race is handled', async () => {
+ const s = stub({ createStatus: 409 });
+ const log = logs();
+ const out = await subscribe({ config: load(ENV), fetchImpl: s.fetchImpl, log });
+ assert.deepStrictEqual(out.map((o) => o.subscription_id), ['sub_existing', 'sub_existing']);
+ assert.match(log.lines.join('\n'), /subscription exists: sub_existing/);
+ assert.strictEqual(s.calls.filter((c) => c.method === 'POST' && c.url === 'http://events.test/api/v1/subscriptions').length, 2, 'no retry loop');
+ });
+
+ await check('refuses before calling anything when the environment is not usable; a refused token is an error', async () => {
+ let called = 0;
+ const fetchImpl = async () => { called++; return { status: 500, ok: false, json: async () => ({}) }; };
+ await assert.rejects(subscribe({ config: load({ ...ENV, EVENTS_URL: '' }), fetchImpl, log: () => {} }), /EVENTS_URL is not set/);
+ await assert.rejects(subscribe({ config: load({ ...ENV, COUPONS_EVENTS_SECRET: 'short' }), fetchImpl, log: () => {} }), /COUPONS_EVENTS_SECRET must be set/);
+ await assert.rejects(subscribe({ config: load({ ...ENV, OV_OAUTH_CLIENT_SECRET: '' }), fetchImpl, log: () => {} }), /OV_OAUTH_CLIENT_SECRET is not set/);
+ assert.strictEqual(called, 0);
+ await assert.rejects(subscribe({ config: load(ENV), fetchImpl: stub({ tokenStatus: 401 }).fetchImpl, log: () => {} }), /token endpoint 401/);
+ });
+
+ await check('arguments', () => {
+ assert.strictEqual(parseArgs([]).reconcile, false);
+ assert.strictEqual(parseArgs(['--reconcile']).reconcile, true);
+ assert.strictEqual(parseArgs(['--endpoint', 'http://x/internal/events']).endpoint, 'http://x/internal/events');
+ const o = parseArgs(['--page-size', '50', '--max-pages', '3']);
+ assert.strictEqual(o.pageSize, 50);
+ assert.strictEqual(o.maxPages, 3);
+ assert.strictEqual(DEFAULT_PAGE_SIZE, 200);
+ assert.throws(() => parseArgs(['--nope']), /unknown argument/);
+ assert.throws(() => parseArgs(['--page-size', '0']), /--page-size must be a positive integer/);
+ assert.throws(() => parseArgs(['--endpoint']), /--endpoint needs a value/);
+ });
+
+ await check('--reconcile re-stamps every Search document, the next pass is unchanged, and a code that left active results is tombstoned', async () => {
+ const t = await boot();
+ try {
+ const alice = t.network.addUser('alice');
+ const a = await t.merchant({ name: 'Reconcile A', host: 'reconcile-a.com' });
+ const b = await t.merchant({ name: 'Reconcile B', host: 'reconcile-b.com' });
+ await t.submit(alice, { merchant_id: a.id, code: 'RA10', title: 'Ten off A' });
+ await t.submit(alice, { merchant_id: b.id, code: 'RB10', title: 'Ten off B' });
+ // A Search index that lost its publications: no revisions survive.
+ await t.ctx.store.db.query('DELETE FROM coupons_index_revisions');
+ assert.deepStrictEqual(await reconcile({ ctx: t.ctx, log: () => {} }), { sent: 4, unchanged: 0 }, '2 merchants + 2 public coupons');
+ assert.deepStrictEqual(await reconcile({ ctx: t.ctx, log: () => {} }), { sent: 0, unchanged: 4 }, 'idempotent: nothing left to send');
+ // A code that left active results behind the index's back still reaches Search.
+ await t.ctx.store.db.query("UPDATE coupons SET status = 'disabled' WHERE code = 'RA10'");
+ const before = Number((await t.ctx.store.db.query("SELECT COUNT(*) AS n FROM event_outbox WHERE envelope->>'event_type' = 'coupons.index_document.deleted'")).rows[0].n);
+ assert.deepStrictEqual(await reconcile({ ctx: t.ctx, log: () => {} }), { sent: 2, unchanged: 2 }, 'its tombstone + its merchant (one fewer active code)');
+ const after = Number((await t.ctx.store.db.query("SELECT COUNT(*) AS n FROM event_outbox WHERE envelope->>'event_type' = 'coupons.index_document.deleted'")).rows[0].n);
+ assert.strictEqual(after - before, 1, 'the disabled code is tombstoned');
+ } finally { await t.close(); }
+ });
+
+ await check('--page-size and --max-pages bound one pass to a page per table', async () => {
+ const t = await boot();
+ try {
+ const alice = t.network.addUser('alice');
+ for (const n of ['C', 'D']) {
+ const m = await t.merchant({ name: `Bounded ${n}`, host: `bounded-${n.toLowerCase()}.com` });
+ await t.submit(alice, { merchant_id: m.id, code: `B${n}10`, title: `Ten off ${n}` });
+ }
+ await t.ctx.store.db.query('DELETE FROM coupons_index_revisions');
+ assert.deepStrictEqual(await reconcile({ ctx: t.ctx, pageSize: 1, maxPages: 1, log: () => {} }), { sent: 2, unchanged: 0 }, 'one merchant page + one coupon page');
+ } finally { await t.close(); }
+ });
+
+ await check('the default endpoint is this service\'s port 4850 webhook', () => {
+ assert.strictEqual(defaultEndpoint(load(ENV)), 'http://127.0.0.1:4850/internal/events');
+ assert.strictEqual(load(ENV).port, 4850);
+ });
+
+ done();
+})();