openvibers/openvibe.trade Commits
T9 step 1, Trade half — OpenVibe.Trade: make scripts/subscribe.js… (#9)
Changes
@@@ -242,6 +242,7 @@ services need to call Trade, is under [Grants the Network must hold](#grants-theEach grant is `[client, capability, audience]`:- `[trade, events.event.publish, openvibe.events]`++- `[trade, events.subscription.manage, openvibe.events]`- `[trade, sources.item.read, openvibe.sources]`- `[trade, sources.source.read, openvibe.sources]`- For OpenVibe.AI to deliver drafts: `[ai, trade.context.read, openvibe.trade]` and@@@ -373,8 +374,11 @@ First install (done once; kept for a rebuild):Events subscription is created, `TRADE_EVENTS_WEBHOOK_SECRET`.2. **Network:** register (or seed) the OAuth client `trade` with redirect`https://openvibe.trade/auth/callback`, set its secret, add the grants above.--3. **Events (optional):** a subscription for consumer `trade`, topic `sources.*`, endpoint-- `http://127.0.0.1:4860/internal/events`, secret = `TRADE_EVENTS_WEBHOOK_SECRET`.++3. **Events:** `node scripts/subscribe.js` creates the subscription for consumer `trade`, topic++ `sources.*`, endpoint `http://127.0.0.1:4860/internal/events` (loopback: nginx never proxies++ `/internal/`), secret = `TRADE_EVENTS_WEBHOOK_SECRET` (32+ characters). It needs++ `OV_OAUTH_CLIENT_SECRET` and the `[trade, events.subscription.manage, openvibe.events]` grant.++ Pass `--endpoint <url>` when the Events host differs.4. **Search:** add `trade` to `SEARCH_EVENT_OWNERS` if it is not there.5. **systemd:** install `deploy/systemd/openvibe-trade.service` (port 4860, `StateDirectory=openvibe-trade`).6. **nginx:** install `deploy/nginx/openvibe.trade.conf` (`/metrics` and `/internal/` never proxied).@@@ -7,21 +7,34 @@* Usage:* node scripts/subscribe.js # subscribe only* node scripts/subscribe.js --reconcile # subscribe, then refresh all indexing and kick outbox++ * node scripts/subscribe.js --endpoint http://127.0.0.1:4860/internal/events*-- * Posts to {EVENTS_URL}/api/v1/subscriptions with the configured webhook secret.-- * A 409 response (subscription already exists) prints 'exists' and continues.++ * Posts to {EVENTS_URL}/api/v1/subscriptions with a service token (events.subscription.manage for++ * audience openvibe.events) and the configured webhook secret. Events delivers to this service's own++ * loopback webhook (127.0.0.1:PORT/internal/events), never the public origin: nginx does not proxy++ * /internal/. 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 { serviceAuth } = require('openvibe-contracts');const { load } = require('../server/config');const { openStore } = require('../server/db');const RECONCILE = process.argv.includes('--reconcile');--async function subscribe(config, { fetch: fetchImpl = globalThis.fetch } = {}) {++/** Read a `--name <value>` option from argv (undefined when absent). */++function option(name, argv = process.argv) {++ const i = argv.indexOf(`--${name}`);++ return i >= 0 ? argv[i + 1] : undefined;++}++++/** The endpoint Events delivers to: this service's own inbound webhook. */++const defaultEndpoint = (config) => `http://127.0.0.1:${config.port}/internal/events`;++++async function subscribe(config, { fetch: fetchImpl = globalThis.fetch, tokens = null, endpoint } = {}) {const secret = config.events.webhookSecret;if (!secret) {throw new Error('TRADE_EVENTS_WEBHOOK_SECRET is not set; cannot subscribe');@@@ -32,17 +45,28 @@ async function subscribe(config, { fetch: fetchImpl = globalThis.fetch } = {}) {if (!config.events.url) {throw new Error('EVENTS_URL is not set; cannot subscribe');}++ if (!config.oauth.clientSecret) {++ throw new Error('OV_OAUTH_CLIENT_SECRET is not set; cannot subscribe as the trade principal');++ }++++ const auth = tokens || serviceAuth.createTokenClient({++ tokenUrl: `${config.networkInternalUrl}/oauth/token`,++ clientId: config.oauth.clientId,++ clientSecret: config.oauth.clientSecret,++ audience: 'openvibe.events',++ scope: 'events.subscription.manage',++ });const url = `${config.events.url}/api/v1/subscriptions`;const body = {topic_pattern: 'sources.*',-- endpoint: `${config.baseUrl}/internal/events`,++ endpoint: endpoint || defaultEndpoint(config),secret,};const res = await fetchImpl(url, {method: 'POST',-- headers: { 'content-type': 'application/json' },++ headers: { 'content-type': 'application/json', ...(await auth.authHeaders()) },body: JSON.stringify(body),});@@@ -70,9 +94,10 @@ async function main() {const config = load();const log = console;-- log.log(`[Trade subscribe] port ${config.port}, events ${config.events.url || '(unset)'}`);++ const endpoint = option('endpoint');++ log.log(`[Trade subscribe] port ${config.port}, events ${config.events.url || '(unset)'}, endpoint ${endpoint || defaultEndpoint(config)}`);-- const result = await subscribe(config);++ const result = await subscribe(config, { endpoint });log.log(`[Trade subscribe] ${result.created ? 'created' : 'exists'}: ${result.subscription_id || '(no id)'}`);if (RECONCILE) {@@@ -2,19 +2,23 @@/*** 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++ * 1. stub-fetch test for subscription: POST to {EVENTS_URL}/api/v1/subscriptions with a++ * service token and the loopback endpoint; handle 200 (created) and 409 (exists).++ * 2. the service token client is required: a missing client secret throws.++ * 3. 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');++// Stand-in for serviceAuth.createTokenClient: the test never mints a real token.++const testTokens = { authHeaders: async () => ({ authorization: 'Bearer test' }) };++(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 () => {++ await check('subscribe posts to Events with a service token and the loopback endpoint', async () => {let captured = null;const stubFetch = async (url, opts) => {captured = { url, opts };@@@ -27,18 +31,38 @@ const { subscribe, reconcile } = require('../scripts/subscribe');};const config = { ...t.ctx.config, events: { ...t.ctx.config.events, url: 'http://events.test' } };-- const result = await subscribe(config, { fetch: stubFetch });++ const result = await subscribe(config, { fetch: stubFetch, tokens: testTokens });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');++ assert.strictEqual(captured.opts.headers.authorization, 'Bearer test');const body = JSON.parse(captured.opts.body);assert.strictEqual(body.topic_pattern, 'sources.*');-- assert.strictEqual(body.endpoint, `${config.baseUrl}/internal/events`);++ assert.strictEqual(body.endpoint, `http://127.0.0.1:${config.port}/internal/events`);assert.strictEqual(body.secret, config.events.webhookSecret);});++ await check('subscribe honours an explicit --endpoint override', async () => {++ let captured = null;++ const stubFetch = async (url, opts) => {++ captured = { url, opts };++ return {++ ok: true,++ status: 200,++ json: async () => ({ subscription_id: 'sub_override' }),++ text: async () => '',++ };++ };++++ const config = { ...t.ctx.config, events: { ...t.ctx.config.events, url: 'http://events.test' } };++ await subscribe(config, { fetch: stubFetch, tokens: testTokens, endpoint: 'http://127.0.0.1:9999/internal/events' });++++ const body = JSON.parse(captured.opts.body);++ assert.strictEqual(body.endpoint, 'http://127.0.0.1:9999/internal/events');++ });++await check('subscribe prints exists on 409 and returns created=false', async () => {const stubFetch = async () => ({ok: false,@@@ -48,12 +72,26 @@ const { subscribe, reconcile } = require('../scripts/subscribe');});const config = { ...t.ctx.config, events: { ...t.ctx.config.events, url: 'http://events.test' } };-- const result = await subscribe(config, { fetch: stubFetch });++ const result = await subscribe(config, { fetch: stubFetch, tokens: testTokens });assert.strictEqual(result.created, false);assert.strictEqual(result.subscription_id, 'sub_existing456');});++ await check('subscribe rejects when OV_OAUTH_CLIENT_SECRET is not set', async () => {++ const config = {++ ...t.ctx.config,++ events: { ...t.ctx.config.events, url: 'http://events.test' },++ oauth: { ...t.ctx.config.oauth, clientSecret: '' },++ };++ const stubFetch = async () => { throw new Error('should not be called'); };++++ await assert.rejects(++ () => subscribe(config, { fetch: stubFetch }),++ /OV_OAUTH_CLIENT_SECRET is not set/,++ );++ });++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'); };
@@ -242,6 +242,7 @@ services need to call Trade, is under [Grants the Network must hold](#grants-the
@@ -242,6 +242,7 @@ services need to call Trade, is under [Grants the Network must hold](#grants-the
Each grant is `[client, capability, audience]`:
Each grant is `[client, capability, audience]`:
- `[trade, events.event.publish, openvibe.events]`
- `[trade, events.event.publish, openvibe.events]`
+- `[trade, events.subscription.manage, openvibe.events]`
- `[trade, sources.item.read, openvibe.sources]`
- `[trade, sources.item.read, openvibe.sources]`
- `[trade, sources.source.read, openvibe.sources]`
- `[trade, sources.source.read, openvibe.sources]`
- For OpenVibe.AI to deliver drafts: `[ai, trade.context.read, openvibe.trade]` and
- For OpenVibe.AI to deliver drafts: `[ai, trade.context.read, openvibe.trade]` and
@@ -373,8 +374,11 @@ First install (done once; kept for a rebuild):
@@ -373,8 +374,11 @@ First install (done once; kept for a rebuild):
Events subscription is created, `TRADE_EVENTS_WEBHOOK_SECRET`.
Events subscription is created, `TRADE_EVENTS_WEBHOOK_SECRET`.
2. **Network:** register (or seed) the OAuth client `trade` with redirect
2. **Network:** register (or seed) the OAuth client `trade` with redirect
`https://openvibe.trade/auth/callback`, set its secret, add the grants above.
`https://openvibe.trade/auth/callback`, set its secret, add the grants above.
-3. **Events (optional):** a subscription for consumer `trade`, topic `sources.*`, endpoint
- `http://127.0.0.1:4860/internal/events`, secret = `TRADE_EVENTS_WEBHOOK_SECRET`.
+3. **Events:** `node scripts/subscribe.js` creates the subscription for consumer `trade`, topic
+ `sources.*`, endpoint `http://127.0.0.1:4860/internal/events` (loopback: nginx never proxies
+ `/internal/`), secret = `TRADE_EVENTS_WEBHOOK_SECRET` (32+ characters). It needs
+ `OV_OAUTH_CLIENT_SECRET` and the `[trade, events.subscription.manage, openvibe.events]` grant.
+ Pass `--endpoint <url>` when the Events host differs.
4. **Search:** add `trade` to `SEARCH_EVENT_OWNERS` if it is not there.
4. **Search:** add `trade` to `SEARCH_EVENT_OWNERS` if it is not there.
5. **systemd:** install `deploy/systemd/openvibe-trade.service` (port 4860, `StateDirectory=openvibe-trade`).
5. **systemd:** install `deploy/systemd/openvibe-trade.service` (port 4860, `StateDirectory=openvibe-trade`).
6. **nginx:** install `deploy/nginx/openvibe.trade.conf` (`/metrics` and `/internal/` never proxied).
6. **nginx:** install `deploy/nginx/openvibe.trade.conf` (`/metrics` and `/internal/` never proxied).
@@ -7,21 +7,34 @@
@@ -7,21 +7,34 @@
* Usage:
* Usage:
* node scripts/subscribe.js # subscribe only
* node scripts/subscribe.js # subscribe only
* node scripts/subscribe.js --reconcile # subscribe, then refresh all indexing and kick outbox
* node scripts/subscribe.js --reconcile # subscribe, then refresh all indexing and kick outbox
+ * node scripts/subscribe.js --endpoint http://127.0.0.1:4860/internal/events
*
*
- * Posts to {EVENTS_URL}/api/v1/subscriptions with the configured webhook secret.
- * A 409 response (subscription already exists) prints 'exists' and continues.
+ * Posts to {EVENTS_URL}/api/v1/subscriptions with a service token (events.subscription.manage for
+ * audience openvibe.events) and the configured webhook secret. Events delivers to this service's own
+ * loopback webhook (127.0.0.1:PORT/internal/events), never the public origin: nginx does not proxy
+ * /internal/. A 409 response (subscription already exists) prints 'exists' and continues.
* The webhook secret must be at least 32 characters (guard against weak secrets).
* 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(),
* --reconcile runs ctx.store.tx(() => ctx.indexing.refreshAll()) then ctx.outbox.kick(),
* which re-evaluates all instrument pages and wakes the outbox relay.
* which re-evaluates all instrument pages and wakes the outbox relay.
*/
*/
+const { serviceAuth } = require('openvibe-contracts');
const { load } = require('../server/config');
const { load } = require('../server/config');
const { openStore } = require('../server/db');
const { openStore } = require('../server/db');
const RECONCILE = process.argv.includes('--reconcile');
const RECONCILE = process.argv.includes('--reconcile');
-async function subscribe(config, { fetch: fetchImpl = globalThis.fetch } = {}) {
+/** Read a `--name <value>` option from argv (undefined when absent). */
+function option(name, argv = process.argv) {
+ const i = argv.indexOf(`--${name}`);
+ return i >= 0 ? argv[i + 1] : undefined;
+}
+
+/** The endpoint Events delivers to: this service's own inbound webhook. */
+const defaultEndpoint = (config) => `http://127.0.0.1:${config.port}/internal/events`;
+
+async function subscribe(config, { fetch: fetchImpl = globalThis.fetch, tokens = null, endpoint } = {}) {
const secret = config.events.webhookSecret;
const secret = config.events.webhookSecret;
if (!secret) {
if (!secret) {
throw new Error('TRADE_EVENTS_WEBHOOK_SECRET is not set; cannot subscribe');
throw new Error('TRADE_EVENTS_WEBHOOK_SECRET is not set; cannot subscribe');
@@ -32,17 +45,28 @@ async function subscribe(config, { fetch: fetchImpl = globalThis.fetch } = {}) {
@@ -32,17 +45,28 @@ async function subscribe(config, { fetch: fetchImpl = globalThis.fetch } = {}) {
if (!config.events.url) {
if (!config.events.url) {
throw new Error('EVENTS_URL is not set; cannot subscribe');
throw new Error('EVENTS_URL is not set; cannot subscribe');
}
}
+ if (!config.oauth.clientSecret) {
+ throw new Error('OV_OAUTH_CLIENT_SECRET is not set; cannot subscribe as the trade principal');
+ }
+
+ const auth = tokens || serviceAuth.createTokenClient({
+ tokenUrl: `${config.networkInternalUrl}/oauth/token`,
+ clientId: config.oauth.clientId,
+ clientSecret: config.oauth.clientSecret,
+ audience: 'openvibe.events',
+ scope: 'events.subscription.manage',
+ });
const url = `${config.events.url}/api/v1/subscriptions`;
const url = `${config.events.url}/api/v1/subscriptions`;
const body = {
const body = {
topic_pattern: 'sources.*',
topic_pattern: 'sources.*',
- endpoint: `${config.baseUrl}/internal/events`,
+ endpoint: endpoint || defaultEndpoint(config),
secret,
secret,
};
};
const res = await fetchImpl(url, {
const res = await fetchImpl(url, {
method: 'POST',
method: 'POST',
- headers: { 'content-type': 'application/json' },
+ headers: { 'content-type': 'application/json', ...(await auth.authHeaders()) },
body: JSON.stringify(body),
body: JSON.stringify(body),
});
});
@@ -70,9 +94,10 @@ async function main() {
@@ -70,9 +94,10 @@ async function main() {
const config = load();
const config = load();
const log = console;
const log = console;
- log.log(`[Trade subscribe] port ${config.port}, events ${config.events.url || '(unset)'}`);
+ const endpoint = option('endpoint');
+ log.log(`[Trade subscribe] port ${config.port}, events ${config.events.url || '(unset)'}, endpoint ${endpoint || defaultEndpoint(config)}`);
- const result = await subscribe(config);
+ const result = await subscribe(config, { endpoint });
log.log(`[Trade subscribe] ${result.created ? 'created' : 'exists'}: ${result.subscription_id || '(no id)'}`);
log.log(`[Trade subscribe] ${result.created ? 'created' : 'exists'}: ${result.subscription_id || '(no id)'}`);
if (RECONCILE) {
if (RECONCILE) {
@@ -2,19 +2,23 @@
@@ -2,19 +2,23 @@
/**
/**
* T9 J7b/J7b' — scripts/subscribe.js: subscribe to OpenVibe.Events and reconcile.
* 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
+ * 1. stub-fetch test for subscription: POST to {EVENTS_URL}/api/v1/subscriptions with a
+ * service token and the loopback endpoint; handle 200 (created) and 409 (exists).
+ * 2. the service token client is required: a missing client secret throws.
+ * 3. reconcile test: asserts one bounded pass (refreshAll called once) and reports
* {sent, unchanged}.
* {sent, unchanged}.
*/
*/
const assert = require('assert');
const assert = require('assert');
const { boot, check, done } = require('./helpers/boot');
const { boot, check, done } = require('./helpers/boot');
const { subscribe, reconcile } = require('../scripts/subscribe');
const { subscribe, reconcile } = require('../scripts/subscribe');
+// Stand-in for serviceAuth.createTokenClient: the test never mints a real token.
+const testTokens = { authHeaders: async () => ({ authorization: 'Bearer test' }) };
+
(async () => {
(async () => {
const t = await boot({ env: { TRADE_EVENTS_WEBHOOK_SECRET: 'a'.repeat(32) } });
const t = await boot({ env: { TRADE_EVENTS_WEBHOOK_SECRET: 'a'.repeat(32) } });
- await check('subscribe posts to Events and returns created on 200', async () => {
+ await check('subscribe posts to Events with a service token and the loopback endpoint', async () => {
let captured = null;
let captured = null;
const stubFetch = async (url, opts) => {
const stubFetch = async (url, opts) => {
captured = { url, opts };
captured = { url, opts };
@@ -27,18 +31,38 @@ const { subscribe, reconcile } = require('../scripts/subscribe');
@@ -27,18 +31,38 @@ const { subscribe, reconcile } = require('../scripts/subscribe');
};
};
const config = { ...t.ctx.config, events: { ...t.ctx.config.events, url: 'http://events.test' } };
const config = { ...t.ctx.config, events: { ...t.ctx.config.events, url: 'http://events.test' } };
- const result = await subscribe(config, { fetch: stubFetch });
+ const result = await subscribe(config, { fetch: stubFetch, tokens: testTokens });
assert.strictEqual(result.created, true);
assert.strictEqual(result.created, true);
assert.strictEqual(result.subscription_id, 'sub_test123');
assert.strictEqual(result.subscription_id, 'sub_test123');
assert.strictEqual(captured.url, 'http://events.test/api/v1/subscriptions');
assert.strictEqual(captured.url, 'http://events.test/api/v1/subscriptions');
assert.strictEqual(captured.opts.method, 'POST');
assert.strictEqual(captured.opts.method, 'POST');
+ assert.strictEqual(captured.opts.headers.authorization, 'Bearer test');
const body = JSON.parse(captured.opts.body);
const body = JSON.parse(captured.opts.body);
assert.strictEqual(body.topic_pattern, 'sources.*');
assert.strictEqual(body.topic_pattern, 'sources.*');
- assert.strictEqual(body.endpoint, `${config.baseUrl}/internal/events`);
+ assert.strictEqual(body.endpoint, `http://127.0.0.1:${config.port}/internal/events`);
assert.strictEqual(body.secret, config.events.webhookSecret);
assert.strictEqual(body.secret, config.events.webhookSecret);
});
});
+ await check('subscribe honours an explicit --endpoint override', async () => {
+ let captured = null;
+ const stubFetch = async (url, opts) => {
+ captured = { url, opts };
+ return {
+ ok: true,
+ status: 200,
+ json: async () => ({ subscription_id: 'sub_override' }),
+ text: async () => '',
+ };
+ };
+
+ const config = { ...t.ctx.config, events: { ...t.ctx.config.events, url: 'http://events.test' } };
+ await subscribe(config, { fetch: stubFetch, tokens: testTokens, endpoint: 'http://127.0.0.1:9999/internal/events' });
+
+ const body = JSON.parse(captured.opts.body);
+ assert.strictEqual(body.endpoint, 'http://127.0.0.1:9999/internal/events');
+ });
+
await check('subscribe prints exists on 409 and returns created=false', async () => {
await check('subscribe prints exists on 409 and returns created=false', async () => {
const stubFetch = async () => ({
const stubFetch = async () => ({
ok: false,
ok: false,
@@ -48,12 +72,26 @@ const { subscribe, reconcile } = require('../scripts/subscribe');
@@ -48,12 +72,26 @@ const { subscribe, reconcile } = require('../scripts/subscribe');
});
});
const config = { ...t.ctx.config, events: { ...t.ctx.config.events, url: 'http://events.test' } };
const config = { ...t.ctx.config, events: { ...t.ctx.config.events, url: 'http://events.test' } };
- const result = await subscribe(config, { fetch: stubFetch });
+ const result = await subscribe(config, { fetch: stubFetch, tokens: testTokens });
assert.strictEqual(result.created, false);
assert.strictEqual(result.created, false);
assert.strictEqual(result.subscription_id, 'sub_existing456');
assert.strictEqual(result.subscription_id, 'sub_existing456');
});
});
+ await check('subscribe rejects when OV_OAUTH_CLIENT_SECRET is not set', async () => {
+ const config = {
+ ...t.ctx.config,
+ events: { ...t.ctx.config.events, url: 'http://events.test' },
+ oauth: { ...t.ctx.config.oauth, clientSecret: '' },
+ };
+ const stubFetch = async () => { throw new Error('should not be called'); };
+
+ await assert.rejects(
+ () => subscribe(config, { fetch: stubFetch }),
+ /OV_OAUTH_CLIENT_SECRET is not set/,
+ );
+ });
+
await check('subscribe rejects when secret is shorter than 32 characters', async () => {
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 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'); };
const stubFetch = async () => { throw new Error('should not be called'); };