openvibers/openvibe.mediahub Commits
MediaHub: subscribe to the two account export/deletion topics at boot (MEDIAHUB_EVENTS_URL), like the other skeleton services
Changes
@@@ -27,6 +27,8 @@ OV_OAUTH_CLIENT_SECRET=# OpenVibe.Events: the ADR-033 account export/deletion deliveries (POST /internal/events, v2 signature, loopback# only). Without it the route answers 503 and Events retries.MEDIAHUB_EVENTS_SECRET=++# Where the two account export/deletion subscriptions are created at boot (OpenVibe.Events, loopback on the host).++MEDIAHUB_EVENTS_URL=# ── OpenVibe.Media: where the bytes live ────────────────────# The host-internal address of the Media object API, and the app key of Media's `media-hub` tenant. The key never@@@ -74,7 +74,12 @@ function load(env = process.env) {// OpenVibe.Events: the ADR-033 account export/deletion events (network.account.export_requested and// network.account.deleted). Without the secret the route answers 503 and Network retries.-- events: { secrets: [env.MEDIAHUB_EVENTS_SECRET, env.OV_EVENTS_SECRET].filter(Boolean) },++ events: {++ secrets: [env.MEDIAHUB_EVENTS_SECRET, env.OV_EVENTS_SECRET].filter(Boolean).flatMap((x) => String(x).split(',')).map((x) => x.trim()).filter(Boolean),++ // Where the two ADR-033 subscriptions are created at boot (server/events-consumer.js); off when unset.++ url: String(env.MEDIAHUB_EVENTS_URL || env.EVENTS_URL || '').replace(/\/+$/, ''),++ endpoint: env.MEDIAHUB_EVENTS_ENDPOINT || '',++ },// OpenVibe.Network: SSO (OAuth2 authorization server with PKCE) and its JWKS.networkUrl,nnew file mode 100644@@@ -0,0 +1,52 @@++'use strict';++++/**++ * OpenVibe.Events → Food. Food has no consumer of its own: the two topics it takes are account export and deletion++ * (ADR-033), and POST /internal/events is answered by openvibe-sdk/account-data's own consumer, mounted in++ * server/app.js over server/identity/account-data.js (the table map). What is left here is the boot-time half:++ * startSubscriptions() creates the two subscriptions at Events if they are missing, idempotently, retried with++ * backoff, and off when the URL or the secret is unset (QUEST_EVENTS_SUBSCRIBE=0 style: MEDIAHUB_EVENTS_SUBSCRIBE=0).++ *++ * The endpoint is this service's loopback /internal/events (Events posts to http://127.0.0.1:<port> directly; nginx++ * answers 404 there), and the token is this service's own for audience openvibe.events with++ * scope events.subscription.manage.++ */++const { serviceAuth } = require('openvibe-contracts');++const { TOPICS } = require('openvibe-sdk/account-data');++++/** Create any missing subscription for TOPICS at Events (idempotent; retried in the background at boot). */++function startSubscriptions({ config, port, secret, eventsUrl = config.events.url, fetchImpl = globalThis.fetch, log = console }) {++ if (!eventsUrl || !secret || !config.oauth.clientSecret || process.env.MEDIAHUB_EVENTS_SUBSCRIBE === '0') return null;++ const base = String(eventsUrl).replace(/\/+$/, '');++ const endpoint = config.events.endpoint || `http://127.0.0.1:${port}/internal/events`;++ 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 call = async (method, path, body) => {++ const res = await fetchImpl(`${base}${path}`, { method, headers: { Accept: 'application/json', ...(body ? { 'Content-Type': 'application/json' } : {}), ...(await tokens.authHeaders()) }, body: body ? JSON.stringify(body) : undefined, signal: AbortSignal.timeout(15000) });++ const json = await res.json().catch(() => ({}));++ return { status: res.status, ok: res.ok, body: json };++ };++ const attempt = async () => {++ const listed = await call('GET', '/api/v1/subscriptions');++ if (!listed.ok) throw new Error(`listing subscriptions: ${listed.status}`);++ const mine = (listed.body.subscriptions || []).filter((s) => s.endpoint === endpoint);++ for (const topic of TOPICS) {++ if (mine.some((s) => s.topic_pattern === topic)) continue;++ const r = await call('POST', '/api/v1/subscriptions', { topic_pattern: topic, endpoint, secret });++ if (!r.ok && r.status !== 409) throw new Error(`subscribing to ${topic}: ${r.status} ${r.body.code || ''}`);++ if (r.ok) log.log(`[Events consumer] subscription created: ${r.body.id} (${topic} → ${endpoint})`);++ }++ };++ const delays = [0, 10_000, 60_000, 5 * 60_000, 15 * 60_000];++ let i = 0;++ let timer = null;++ let stopped = false;++ const run = () => { timer = null; if (stopped) return; attempt().catch((err) => {++ if (stopped) return;++ if (++i < delays.length) { timer = setTimeout(run, delays[i]); if (timer.unref) timer.unref(); } else log.warn('[Events consumer] subscriptions not created:', err.message);++ }); };++ timer = setTimeout(run, delays[0]); if (timer.unref) timer.unref();++ // Graceful stop: no further attempts (they run again at the next start).++ return { topics: TOPICS, endpoint, stop() { stopped = true; if (timer) clearTimeout(timer); timer = null; } };++}++++module.exports = { startSubscriptions, TOPICS };@@@ -11,10 +11,10 @@ const { gracefulStop } = require('openvibe-sdk/service');* The process stop (openvibe-sdk/service): the HTTP drain runs, then the JWKS refresher stops and the store closes.* Exported so a test can inject `exit` and `signals: false`.*/--function createLifecycle({ server, ctx, exit, signals, timers = [] }) {++function createLifecycle({ server, ctx, exit, signals, timers = [], extra = [] }) {return gracefulStop({name: 'OpenVibe.MediaHub', server, deadlineExitCode: 0, exit, signals, deadlineMs: 10_000,-- close: [() => { for (const t of timers) clearInterval(t); }, () => ctx.keys.client.stop(), () => ctx.s.close()],++ close: [() => { for (const t of timers) clearInterval(t); }, () => ctx.keys.client.stop(), () => ctx.s.close(), ...extra],});}@@@ -28,7 +28,10 @@ async function start() {server.keepAliveTimeout = 65_000;ctx.keys.client.start();-- createLifecycle({ server, ctx });++ // Subscribe to the two ADR-033 topics at OpenVibe.Events (idempotent; off without MEDIAHUB_EVENTS_URL and++ // MEDIAHUB_EVENTS_SECRET). The consumer itself is mounted in server/app.js.++ const subscriptions = require('./events-consumer').startSubscriptions({ config, port: config.port, secret: config.events.secrets[0] || '' });++ createLifecycle({ server, ctx, extra: [() => { if (subscriptions) subscriptions.stop(); }] });return { server, ctx };}
@@ -27,6 +27,8 @@ OV_OAUTH_CLIENT_SECRET=
@@ -27,6 +27,8 @@ OV_OAUTH_CLIENT_SECRET=
# OpenVibe.Events: the ADR-033 account export/deletion deliveries (POST /internal/events, v2 signature, loopback
# OpenVibe.Events: the ADR-033 account export/deletion deliveries (POST /internal/events, v2 signature, loopback
# only). Without it the route answers 503 and Events retries.
# only). Without it the route answers 503 and Events retries.
MEDIAHUB_EVENTS_SECRET=
MEDIAHUB_EVENTS_SECRET=
+# Where the two account export/deletion subscriptions are created at boot (OpenVibe.Events, loopback on the host).
+MEDIAHUB_EVENTS_URL=
# ── OpenVibe.Media: where the bytes live ────────────────────
# ── OpenVibe.Media: where the bytes live ────────────────────
# The host-internal address of the Media object API, and the app key of Media's `media-hub` tenant. The key never
# The host-internal address of the Media object API, and the app key of Media's `media-hub` tenant. The key never
@@ -74,7 +74,12 @@ function load(env = process.env) {
@@ -74,7 +74,12 @@ function load(env = process.env) {
// OpenVibe.Events: the ADR-033 account export/deletion events (network.account.export_requested and
// OpenVibe.Events: the ADR-033 account export/deletion events (network.account.export_requested and
// network.account.deleted). Without the secret the route answers 503 and Network retries.
// network.account.deleted). Without the secret the route answers 503 and Network retries.
- events: { secrets: [env.MEDIAHUB_EVENTS_SECRET, env.OV_EVENTS_SECRET].filter(Boolean) },
+ events: {
+ secrets: [env.MEDIAHUB_EVENTS_SECRET, env.OV_EVENTS_SECRET].filter(Boolean).flatMap((x) => String(x).split(',')).map((x) => x.trim()).filter(Boolean),
+ // Where the two ADR-033 subscriptions are created at boot (server/events-consumer.js); off when unset.
+ url: String(env.MEDIAHUB_EVENTS_URL || env.EVENTS_URL || '').replace(/\/+$/, ''),
+ endpoint: env.MEDIAHUB_EVENTS_ENDPOINT || '',
+ },
// OpenVibe.Network: SSO (OAuth2 authorization server with PKCE) and its JWKS.
// OpenVibe.Network: SSO (OAuth2 authorization server with PKCE) and its JWKS.
networkUrl,
networkUrl,
new file mode 100644
new file mode 100644
@@ -0,0 +1,52 @@
@@ -0,0 +1,52 @@
+'use strict';
+
+/**
+ * OpenVibe.Events → Food. Food has no consumer of its own: the two topics it takes are account export and deletion
+ * (ADR-033), and POST /internal/events is answered by openvibe-sdk/account-data's own consumer, mounted in
+ * server/app.js over server/identity/account-data.js (the table map). What is left here is the boot-time half:
+ * startSubscriptions() creates the two subscriptions at Events if they are missing, idempotently, retried with
+ * backoff, and off when the URL or the secret is unset (QUEST_EVENTS_SUBSCRIBE=0 style: MEDIAHUB_EVENTS_SUBSCRIBE=0).
+ *
+ * The endpoint is this service's loopback /internal/events (Events posts to http://127.0.0.1:<port> directly; nginx
+ * answers 404 there), and the token is this service's own for audience openvibe.events with
+ * scope events.subscription.manage.
+ */
+const { serviceAuth } = require('openvibe-contracts');
+const { TOPICS } = require('openvibe-sdk/account-data');
+
+/** Create any missing subscription for TOPICS at Events (idempotent; retried in the background at boot). */
+function startSubscriptions({ config, port, secret, eventsUrl = config.events.url, fetchImpl = globalThis.fetch, log = console }) {
+ if (!eventsUrl || !secret || !config.oauth.clientSecret || process.env.MEDIAHUB_EVENTS_SUBSCRIBE === '0') return null;
+ const base = String(eventsUrl).replace(/\/+$/, '');
+ const endpoint = config.events.endpoint || `http://127.0.0.1:${port}/internal/events`;
+ 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 call = async (method, path, body) => {
+ const res = await fetchImpl(`${base}${path}`, { method, headers: { Accept: 'application/json', ...(body ? { 'Content-Type': 'application/json' } : {}), ...(await tokens.authHeaders()) }, body: body ? JSON.stringify(body) : undefined, signal: AbortSignal.timeout(15000) });
+ const json = await res.json().catch(() => ({}));
+ return { status: res.status, ok: res.ok, body: json };
+ };
+ const attempt = async () => {
+ const listed = await call('GET', '/api/v1/subscriptions');
+ if (!listed.ok) throw new Error(`listing subscriptions: ${listed.status}`);
+ const mine = (listed.body.subscriptions || []).filter((s) => s.endpoint === endpoint);
+ for (const topic of TOPICS) {
+ if (mine.some((s) => s.topic_pattern === topic)) continue;
+ const r = await call('POST', '/api/v1/subscriptions', { topic_pattern: topic, endpoint, secret });
+ if (!r.ok && r.status !== 409) throw new Error(`subscribing to ${topic}: ${r.status} ${r.body.code || ''}`);
+ if (r.ok) log.log(`[Events consumer] subscription created: ${r.body.id} (${topic} → ${endpoint})`);
+ }
+ };
+ const delays = [0, 10_000, 60_000, 5 * 60_000, 15 * 60_000];
+ let i = 0;
+ let timer = null;
+ let stopped = false;
+ const run = () => { timer = null; if (stopped) return; attempt().catch((err) => {
+ if (stopped) return;
+ if (++i < delays.length) { timer = setTimeout(run, delays[i]); if (timer.unref) timer.unref(); } else log.warn('[Events consumer] subscriptions not created:', err.message);
+ }); };
+ timer = setTimeout(run, delays[0]); if (timer.unref) timer.unref();
+ // Graceful stop: no further attempts (they run again at the next start).
+ return { topics: TOPICS, endpoint, stop() { stopped = true; if (timer) clearTimeout(timer); timer = null; } };
+}
+
+module.exports = { startSubscriptions, TOPICS };
@@ -11,10 +11,10 @@ const { gracefulStop } = require('openvibe-sdk/service');
@@ -11,10 +11,10 @@ const { gracefulStop } = require('openvibe-sdk/service');
* The process stop (openvibe-sdk/service): the HTTP drain runs, then the JWKS refresher stops and the store closes.
* The process stop (openvibe-sdk/service): the HTTP drain runs, then the JWKS refresher stops and the store closes.
* Exported so a test can inject `exit` and `signals: false`.
* Exported so a test can inject `exit` and `signals: false`.
*/
*/
-function createLifecycle({ server, ctx, exit, signals, timers = [] }) {
+function createLifecycle({ server, ctx, exit, signals, timers = [], extra = [] }) {
return gracefulStop({
return gracefulStop({
name: 'OpenVibe.MediaHub', server, deadlineExitCode: 0, exit, signals, deadlineMs: 10_000,
name: 'OpenVibe.MediaHub', server, deadlineExitCode: 0, exit, signals, deadlineMs: 10_000,
- close: [() => { for (const t of timers) clearInterval(t); }, () => ctx.keys.client.stop(), () => ctx.s.close()],
+ close: [() => { for (const t of timers) clearInterval(t); }, () => ctx.keys.client.stop(), () => ctx.s.close(), ...extra],
});
});
}
}
@@ -28,7 +28,10 @@ async function start() {
@@ -28,7 +28,10 @@ async function start() {
server.keepAliveTimeout = 65_000;
server.keepAliveTimeout = 65_000;
ctx.keys.client.start();
ctx.keys.client.start();
- createLifecycle({ server, ctx });
+ // Subscribe to the two ADR-033 topics at OpenVibe.Events (idempotent; off without MEDIAHUB_EVENTS_URL and
+ // MEDIAHUB_EVENTS_SECRET). The consumer itself is mounted in server/app.js.
+ const subscriptions = require('./events-consumer').startSubscriptions({ config, port: config.port, secret: config.events.secrets[0] || '' });
+ createLifecycle({ server, ctx, extra: [() => { if (subscriptions) subscriptions.stop(); }] });
return { server, ctx };
return { server, ctx };
}
}