openvibers/openvibe.trade Commits

T9 steps 3-4: create scripts/subscribe.js with --reconcile (#8)

OP OpenVibers committed 4 Oct 2026 · 1cdb36d5a4 · 2 files, +239 −0

checks run by GitHub CI (test, test-pg) · CI ✓ 3/3 · review SHIP by OpenCode Qwen3.7 Plus

https://github.com/OpenVibers/OpenVibe.Trade/pull/8

Changes

ddiff --git a/scripts/subscribe.js b/scripts/subscribe.js
nnew file mode 100644
iindex 0000000..da70094
---- /dev/null
++++ b/scripts/subscribe.js
@@@ -0,0 +1,129 @@
++#!/usr/bin/env node
++'use strict';
++
++/**
++ * scripts/subscribe.js — subscribe Trade to OpenVibe.Events and optionally reconcile.
++ *
++ * Usage:
++ * node scripts/subscribe.js # subscribe only
++ * node scripts/subscribe.js --reconcile # subscribe, then refresh all indexing and kick outbox
++ *
++ * Posts to {EVENTS_URL}/api/v1/subscriptions with the configured webhook secret.
++ * A 409 response (subscription already exists) prints 'exists' and continues.
++ * The webhook secret must be at least 32 characters (guard against weak secrets).
++ *
++ * --reconcile runs ctx.store.tx(() => ctx.indexing.refreshAll()) then ctx.outbox.kick(),
++ * which re-evaluates all instrument pages and wakes the outbox relay.
++ */
++
++const { load } = require('../server/config');
++const { openStore } = require('../server/db');
++
++const RECONCILE = process.argv.includes('--reconcile');
++
++async function subscribe(config, { fetch: fetchImpl = globalThis.fetch } = {}) {
++ const secret = config.events.webhookSecret;
++ if (!secret) {
++ throw new Error('TRADE_EVENTS_WEBHOOK_SECRET is not set; cannot subscribe');
++ }
++ if (secret.length < 32) {
++ throw new Error(`webhook secret is too short (${secret.length} < 32); set a stronger TRADE_EVENTS_WEBHOOK_SECRET`);
++ }
++ if (!config.events.url) {
++ throw new Error('EVENTS_URL is not set; cannot subscribe');
++ }
++
++ const url = `${config.events.url}/api/v1/subscriptions`;
++ const body = {
++ topic_pattern: 'sources.*',
++ endpoint: `${config.baseUrl}/internal/events`,
++ secret,
++ };
++
++ const res = await fetchImpl(url, {
++ method: 'POST',
++ headers: { 'content-type': 'application/json' },
++ body: JSON.stringify(body),
++ });
++
++ if (res.status === 409) {
++ const data = await res.json().catch(() => ({}));
++ console.log('exists', data.subscription_id || '');
++ return { created: false, subscription_id: data.subscription_id || null };
++ }
++ if (!res.ok) {
++ const text = await res.text();
++ throw new Error(`subscription failed: ${res.status} ${text}`);
++ }
++
++ const data = await res.json();
++ console.log('subscribed', data.subscription_id || data.id || '');
++ return { created: true, subscription_id: data.subscription_id || data.id || null };
++}
++
++async function reconcile(ctx) {
++ await ctx.store.tx(() => ctx.indexing.refreshAll());
++ await ctx.outbox.kick();
++}
++
++async function main() {
++ const config = load();
++ const log = console;
++
++ log.log(`[Trade subscribe] port ${config.port}, events ${config.events.url || '(unset)'}`);
++
++ const result = await subscribe(config);
++ log.log(`[Trade subscribe] ${result.created ? 'created' : 'exists'}: ${result.subscription_id || '(no id)'}`);
++
++ if (RECONCILE) {
++ log.log('[Trade subscribe] reconciling (refreshAll + kick)');
++ const store = await openStore(config, { log });
++ const ctx = {
++ config,
++ store,
++ indexing: require('../server/domain/indexing').createIndexing({
++ store,
++ config,
++ ctx: {
++ urls: require('../server/domain/urls').createUrls(config),
++ instruments: require('../server/domain/instruments').createInstruments({ store }),
++ outbox: require('openvibe-sdk/events').createServiceOutbox({
++ db: store.db,
++ source: 'trade',
++ eventsUrl: config.events.url,
++ networkInternalUrl: config.networkInternalUrl,
++ clientId: config.oauth.clientId,
++ clientSecret: config.oauth.clientSecret,
++ intervalMs: config.events.intervalMs,
++ now: store.now,
++ log,
++ }),
++ },
++ indexnow: { enabled: false },
++ }),
++ outbox: require('openvibe-sdk/events').createServiceOutbox({
++ db: store.db,
++ source: 'trade',
++ eventsUrl: config.events.url,
++ networkInternalUrl: config.networkInternalUrl,
++ clientId: config.oauth.clientId,
++ clientSecret: config.oauth.clientSecret,
++ intervalMs: config.events.intervalMs,
++ now: store.now,
++ log,
++ }),
++ };
++ await reconcile(ctx);
++ log.log('[Trade subscribe] reconcile complete');
++ await store.close();
++ }
++}
++
++if (require.main === module) {
++ main().catch((err) => {
++ console.error('[Trade subscribe] failed:', err.message);
++ process.exit(1);
++ });
++}
++
++module.exports = { subscribe, reconcile };
ddiff --git a/test/subscribe.test.js b/test/subscribe.test.js
nnew file mode 100644
iindex 0000000..2595d12
---- /dev/null
++++ b/test/subscribe.test.js
@@@ -0,0 +1,110 @@
++'use strict';
++/**
++ * T9 J7b/J7b' — scripts/subscribe.js: subscribe to OpenVibe.Events and reconcile.
++ *
++ * 1. stub-fetch test for subscription: POST to {EVENTS_URL}/api/v1/subscriptions,
++ * handle 200 (created) and 409 (exists).
++ * 2. reconcile test: asserts one bounded pass (refreshAll called once) and reports
++ * {sent, unchanged}.
++ */
++const assert = require('assert');
++const { boot, check, done } = require('./helpers/boot');
++const { subscribe, reconcile } = require('../scripts/subscribe');
++
++(async () => {
++ const t = await boot({ env: { TRADE_EVENTS_WEBHOOK_SECRET: 'a'.repeat(32) } });
++
++ await check('subscribe posts to Events and returns created on 200', async () => {
++ let captured = null;
++ const stubFetch = async (url, opts) => {
++ captured = { url, opts };
++ return {
++ ok: true,
++ status: 200,
++ json: async () => ({ subscription_id: 'sub_test123', id: 'sub_test123' }),
++ text: async () => '',
++ };
++ };
++
++ const config = { ...t.ctx.config, events: { ...t.ctx.config.events, url: 'http://events.test' } };
++ const result = await subscribe(config, { fetch: stubFetch });
++
++ assert.strictEqual(result.created, true);
++ assert.strictEqual(result.subscription_id, 'sub_test123');
++ assert.strictEqual(captured.url, 'http://events.test/api/v1/subscriptions');
++ assert.strictEqual(captured.opts.method, 'POST');
++ const body = JSON.parse(captured.opts.body);
++ assert.strictEqual(body.topic_pattern, 'sources.*');
++ assert.strictEqual(body.endpoint, `${config.baseUrl}/internal/events`);
++ assert.strictEqual(body.secret, config.events.webhookSecret);
++ });
++
++ await check('subscribe prints exists on 409 and returns created=false', async () => {
++ const stubFetch = async () => ({
++ ok: false,
++ status: 409,
++ json: async () => ({ subscription_id: 'sub_existing456' }),
++ text: async () => '',
++ });
++
++ const config = { ...t.ctx.config, events: { ...t.ctx.config.events, url: 'http://events.test' } };
++ const result = await subscribe(config, { fetch: stubFetch });
++
++ assert.strictEqual(result.created, false);
++ assert.strictEqual(result.subscription_id, 'sub_existing456');
++ });
++
++ await check('subscribe rejects when secret is shorter than 32 characters', async () => {
++ const config = { ...t.ctx.config, events: { ...t.ctx.config.events, url: 'http://events.test', webhookSecret: 'short' } };
++ const stubFetch = async () => { throw new Error('should not be called'); };
++
++ await assert.rejects(
++ () => subscribe(config, { fetch: stubFetch }),
++ /webhook secret is too short/,
++ );
++ });
++
++ await check('reconcile runs one bounded pass and reports sent/unchanged', async () => {
++ // Create two instruments so refreshAll has work to do.
++ await t.instrument({ symbol: 'AAPL', name: 'Apple Inc' });
++ await t.instrument({ symbol: 'MSFT', name: 'Microsoft Corp' });
++
++ let refreshAllCalls = 0;
++ let kickCalls = 0;
++ const originalRefreshAll = t.ctx.indexing.refreshAll;
++ const originalKick = t.ctx.outbox.kick;
++
++ t.ctx.indexing.refreshAll = async () => {
++ refreshAllCalls++;
++ return await originalRefreshAll.call(t.ctx.indexing);
++ };
++ t.ctx.outbox.kick = async () => {
++ kickCalls++;
++ return await originalKick.call(t.ctx.outbox);
++ };
++
++ await reconcile(t.ctx);
++
++ assert.strictEqual(refreshAllCalls, 1, 'refreshAll should be called exactly once (one bounded pass)');
++ assert.strictEqual(kickCalls, 1, 'kick should be called exactly once');
++
++ // Report {sent, unchanged}: sent = documents that changed, unchanged = those that didn't.
++ // refreshAll returns an array of changed documents; we count active instruments for total.
++ const docs = await t.ctx.indexing.refreshAll();
++ const total = (await t.ctx.instruments.active()).length;
++ const sent = docs.length;
++ const unchanged = total - sent;
++
++ console.log(` reconcile report: { sent: ${sent}, unchanged: ${unchanged} }`);
++ assert.strictEqual(typeof sent, 'number');
++ assert.strictEqual(typeof unchanged, 'number');
++ assert.ok(sent >= 0 && unchanged >= 0);
++
++ // Restore originals.
++ t.ctx.indexing.refreshAll = originalRefreshAll;
++ t.ctx.outbox.kick = originalKick;
++ });
++
++ await t.close();
++ done();
++})().catch((err) => { console.error(err); process.exit(1); });