openvibers/openvibe.events Commits
Let any first-party service publish the network-wide provider. namespace (#7)
Changes
@@@ -51,6 +51,9 @@ EVENTS_RECEIPT_RETENTION_DAYS=90# ai games tools codes host sources search wiki blog news reviews deals coupons trade bot. Every prefix# must start with "<source>."; "app" and "app-*" are reserved for developer apps.#EVENTS_SOURCE_PREFIXES={"live":["live."],"media":["media."]}++# Network-wide prefixes any first-party service may publish (default: provider., provider health++# and capacity telemetry). Comma-separated; a prefix's first segment must not be a source.++#EVENTS_SHARED_PREFIXES=provider.# Delivery worker#EVENTS_WORKER=on@@@ -73,7 +73,7 @@ delivery rollup, for Network; never visible to apps).`POST /api/v1/events` takes one `events.event-envelope@1` envelope, or `{ "events": [ … ] }` (up to 100, stored atomically).--- `source` must be the calling service (`svc:live` publishes `source: "live"`), and `event_type` must start with a prefix that source owns (`live.*`, `media.*`, `network.*`, `community.*`, `chat.*`, `openre.*`, `billing.*`, `tips.*`, `vip.*`, `ai.*`, `games.*`, `tools.*`, `codes.*`, `host.*`, `bot.*`; `EVENTS_SOURCE_PREFIXES` overrides).++- `source` must be the calling service (`svc:live` publishes `source: "live"`), and `event_type` must start with a prefix that source owns (`live.*`, `media.*`, `network.*`, `community.*`, `chat.*`, `openre.*`, `billing.*`, `tips.*`, `vip.*`, `ai.*`, `games.*`, `tools.*`, `codes.*`, `host.*`, `bot.*`; `EVENTS_SOURCE_PREFIXES` overrides). A network-wide prefix may be published by any first-party service: `provider.*` carries storage provider health/capacity telemetry (`provider.health.degraded`, `provider.capacity.warning`), so whichever service observes a provider can report it; `EVENTS_SHARED_PREFIXES` overrides.- Idempotent on `event_id`: a repeat answers `200 { event_id, seq, duplicate: true }` and is never stored twice (even after retention, for `EVENTS_RECEIPT_RETENTION_DAYS`).- Missing `trace_id` is taken from the request's `traceparent`; `priority` defaults to `important`, `visibility` to `internal`.- Loop guard: an event whose trace already carries 8 hops (the cross-service chain depth, or the same source/type/subject repeating in the trace) is refused with `409 events.loop_detected`. The chain depth of a first-party publish counts first-party events only (developer-app events in the same trace never count toward it), so an app cannot poison a trace it has seen.@@@ -78,8 +78,11 @@ function publishRouter({ config, store, auth, worker, realtime, limits, fabric =}const prefixes = config.sourcePrefixes[env.source];if (!prefixes) return { status: 403, code: 'events.unknown_source', detail: `source "${env.source}" may not publish events` };-- if (!prefixes.some(p => env.event_type.startsWith(p))) {-- return { status: 403, code: 'events.type_not_allowed', detail: `${env.source} may publish ${prefixes.join(', ')}* only, not ${env.event_type}` };++ // A source owns its own namespace; a network-wide prefix (server/config sharedPrefixes, e.g.++ // provider.* for storage provider telemetry) may be published by any first-party service.++ const allowed = [...prefixes, ...(config.sharedPrefixes || [])];++ if (!allowed.some(p => env.event_type.startsWith(p))) {++ return { status: 403, code: 'events.type_not_allowed', detail: `${env.source} may publish ${allowed.map(p => `${p}*`).join(', ')} only, not ${env.event_type}` };}return { env };}@@@ -23,6 +23,15 @@ const DEFAULT_SOURCES = ['live', 'media', 'network', 'community', 'chat', 'openr// Robot control (OpenVibe.Bot): bot.robot.*, bot.estop.*, bot.command.*'bot'];++/**++ * Namespaces that belong to the network rather than to one service. `provider.*` carries storage++ * provider health/capacity telemetry, which whichever first-party service observes a provider++ * publishes (Media, Sources, Host, ...), so any service token may publish into it. Sources still++ * own their own namespace first; a shared prefix never replaces `<source>.`. Developer apps can++ * never publish a shared prefix. EVENTS_SHARED_PREFIXES (comma-separated) replaces the default.++ */++const DEFAULT_SHARED_PREFIXES = ['provider.'];++function sourcePrefixes(env) {let map = Object.fromEntries(DEFAULT_SOURCES.map(s => [s, [`${s}.`]]));if (env.EVENTS_SOURCE_PREFIXES) {@@@ -43,11 +52,25 @@ function sourcePrefixes(env) {return map;}++function sharedPrefixes(env, map) {++ const out = list(env.EVENTS_SHARED_PREFIXES, DEFAULT_SHARED_PREFIXES);++ for (const p of out) {++ if (!/^[a-z][a-z0-9-]*\.$/.test(p)) throw new Error(`EVENTS_SHARED_PREFIXES: bad prefix "${p}"`);++ const source = p.slice(0, -1);++ if (source === 'app' || source.startsWith('app-')) throw new Error(`EVENTS_SHARED_PREFIXES: "${p}" is reserved for developer apps`);++ if (Object.prototype.hasOwnProperty.call(map, source)) {++ throw new Error(`EVENTS_SHARED_PREFIXES: "${p}" belongs to source "${source}", not to the network`);++ }++ }++ return out;++}++function load(env = process.env) {const nodeEnv = env.NODE_ENV || 'development';const isProduction = nodeEnv === 'production';const port = int(env.PORT, 4300);const maxInflight = int(env.EVENTS_MAX_INFLIGHT, 20);++ const prefixes = sourcePrefixes(env);return {port,host: env.HOST || '127.0.0.1',@@@ -85,7 +108,8 @@ function load(env = process.env) {receiptRetentionDays: int(env.EVENTS_RECEIPT_RETENTION_DAYS, 90),pruneIntervalMs: int(env.EVENTS_PRUNE_INTERVAL_MS, 60 * 60 * 1000),-- sourcePrefixes: sourcePrefixes(env),++ sourcePrefixes: prefixes,++ sharedPrefixes: sharedPrefixes(env, prefixes),maxBatch: 100,maxPayloadBytes: int(env.EVENTS_MAX_PAYLOAD_BYTES, 64 * 1024),maxHops: int(env.EVENTS_MAX_HOPS, 8),@@@ -75,6 +75,25 @@ t('event_type must use a prefix the source owns', async () => {assert.strictEqual(r.body.code, 'events.type_not_allowed');});++t('any first-party service may publish the network-wide provider.* namespace', async () => {++ const media = serviceToken('media', ['events.event.publish']);++ // Media observes a provider's capacity/health and reports it under the network prefix.++ for (const event_type of ['provider.health.degraded', 'provider.capacity.warning']) {++ const r = await request(h.base, 'POST', '/api/v1/events', { token: media, body: envelope('media', { event_type }) });++ assert.strictEqual(r.status, 201, r.text);++ }++ // The review's rename to Media's own namespace also passes, unchanged.++ const r = await request(h.base, 'POST', '/api/v1/events', { token: media, body: envelope('media', { event_type: 'media.provider.health_degraded' }) });++ assert.strictEqual(r.status, 201, r.text);++ // Network-wide means shared: another service may report the same provider stream ...++ const r2 = await request(h.base, 'POST', '/api/v1/events', { token: live, body: envelope('live', { event_type: 'provider.capacity.warning' }) });++ assert.strictEqual(r2.status, 201, r2.text);++ // ... but a shared prefix never hands a source another source's namespace.++ const r3 = await request(h.base, 'POST', '/api/v1/events', { token: live, body: envelope('live', { event_type: 'media.vod.ready' }) });++ assert.strictEqual(r3.status, 403);++ assert.strictEqual(r3.body.code, 'events.type_not_allowed');++});++t('a bot service principal publishes bot.* only', async () => {const bot = serviceToken('bot', ['events.event.publish']);let r = await request(h.base, 'POST', '/api/v1/events', { token: bot, body: envelope('bot', { event_type: 'bot.robot.online' }) });@@@ -42,6 +42,15 @@ t('config: every source prefix must start with the source', () => {assert.strictEqual(load({}).retentionDays, 30);});++t('config: network-wide prefixes are shared, source namespaces stay owned', () => {++ assert.deepStrictEqual(load({}).sharedPrefixes, ['provider.']);++ const c = load({ EVENTS_SHARED_PREFIXES: 'provider.,storage.' });++ assert.deepStrictEqual(c.sharedPrefixes, ['provider.', 'storage.']);++ assert.throws(() => load({ EVENTS_SHARED_PREFIXES: 'provider' }), /bad prefix/);++ assert.throws(() => load({ EVENTS_SHARED_PREFIXES: 'media.' }), /belongs to source "media"/);++ assert.throws(() => load({ EVENTS_SHARED_PREFIXES: 'app.' }), /reserved for developer apps/);++});++t('boot', async () => { h = await boot(); });t('pull: cursor, topic filter, internal events included for services', async () => {
@@ -51,6 +51,9 @@ EVENTS_RECEIPT_RETENTION_DAYS=90
@@ -51,6 +51,9 @@ EVENTS_RECEIPT_RETENTION_DAYS=90
# ai games tools codes host sources search wiki blog news reviews deals coupons trade bot. Every prefix
# ai games tools codes host sources search wiki blog news reviews deals coupons trade bot. Every prefix
# must start with "<source>."; "app" and "app-*" are reserved for developer apps.
# must start with "<source>."; "app" and "app-*" are reserved for developer apps.
#EVENTS_SOURCE_PREFIXES={"live":["live."],"media":["media."]}
#EVENTS_SOURCE_PREFIXES={"live":["live."],"media":["media."]}
+# Network-wide prefixes any first-party service may publish (default: provider., provider health
+# and capacity telemetry). Comma-separated; a prefix's first segment must not be a source.
+#EVENTS_SHARED_PREFIXES=provider.
# Delivery worker
# Delivery worker
#EVENTS_WORKER=on
#EVENTS_WORKER=on
@@ -73,7 +73,7 @@ delivery rollup, for Network; never visible to apps).
@@ -73,7 +73,7 @@ delivery rollup, for Network; never visible to apps).
`POST /api/v1/events` takes one `events.event-envelope@1` envelope, or `{ "events": [ … ] }` (up to 100, stored atomically).
`POST /api/v1/events` takes one `events.event-envelope@1` envelope, or `{ "events": [ … ] }` (up to 100, stored atomically).
-- `source` must be the calling service (`svc:live` publishes `source: "live"`), and `event_type` must start with a prefix that source owns (`live.*`, `media.*`, `network.*`, `community.*`, `chat.*`, `openre.*`, `billing.*`, `tips.*`, `vip.*`, `ai.*`, `games.*`, `tools.*`, `codes.*`, `host.*`, `bot.*`; `EVENTS_SOURCE_PREFIXES` overrides).
+- `source` must be the calling service (`svc:live` publishes `source: "live"`), and `event_type` must start with a prefix that source owns (`live.*`, `media.*`, `network.*`, `community.*`, `chat.*`, `openre.*`, `billing.*`, `tips.*`, `vip.*`, `ai.*`, `games.*`, `tools.*`, `codes.*`, `host.*`, `bot.*`; `EVENTS_SOURCE_PREFIXES` overrides). A network-wide prefix may be published by any first-party service: `provider.*` carries storage provider health/capacity telemetry (`provider.health.degraded`, `provider.capacity.warning`), so whichever service observes a provider can report it; `EVENTS_SHARED_PREFIXES` overrides.
- Idempotent on `event_id`: a repeat answers `200 { event_id, seq, duplicate: true }` and is never stored twice (even after retention, for `EVENTS_RECEIPT_RETENTION_DAYS`).
- Idempotent on `event_id`: a repeat answers `200 { event_id, seq, duplicate: true }` and is never stored twice (even after retention, for `EVENTS_RECEIPT_RETENTION_DAYS`).
- Missing `trace_id` is taken from the request's `traceparent`; `priority` defaults to `important`, `visibility` to `internal`.
- Missing `trace_id` is taken from the request's `traceparent`; `priority` defaults to `important`, `visibility` to `internal`.
- Loop guard: an event whose trace already carries 8 hops (the cross-service chain depth, or the same source/type/subject repeating in the trace) is refused with `409 events.loop_detected`. The chain depth of a first-party publish counts first-party events only (developer-app events in the same trace never count toward it), so an app cannot poison a trace it has seen.
- Loop guard: an event whose trace already carries 8 hops (the cross-service chain depth, or the same source/type/subject repeating in the trace) is refused with `409 events.loop_detected`. The chain depth of a first-party publish counts first-party events only (developer-app events in the same trace never count toward it), so an app cannot poison a trace it has seen.
@@ -78,8 +78,11 @@ function publishRouter({ config, store, auth, worker, realtime, limits, fabric =
@@ -78,8 +78,11 @@ function publishRouter({ config, store, auth, worker, realtime, limits, fabric =
}
}
const prefixes = config.sourcePrefixes[env.source];
const prefixes = config.sourcePrefixes[env.source];
if (!prefixes) return { status: 403, code: 'events.unknown_source', detail: `source "${env.source}" may not publish events` };
if (!prefixes) return { status: 403, code: 'events.unknown_source', detail: `source "${env.source}" may not publish events` };
- if (!prefixes.some(p => env.event_type.startsWith(p))) {
- return { status: 403, code: 'events.type_not_allowed', detail: `${env.source} may publish ${prefixes.join(', ')}* only, not ${env.event_type}` };
+ // A source owns its own namespace; a network-wide prefix (server/config sharedPrefixes, e.g.
+ // provider.* for storage provider telemetry) may be published by any first-party service.
+ const allowed = [...prefixes, ...(config.sharedPrefixes || [])];
+ if (!allowed.some(p => env.event_type.startsWith(p))) {
+ return { status: 403, code: 'events.type_not_allowed', detail: `${env.source} may publish ${allowed.map(p => `${p}*`).join(', ')} only, not ${env.event_type}` };
}
}
return { env };
return { env };
}
}
@@ -23,6 +23,15 @@ const DEFAULT_SOURCES = ['live', 'media', 'network', 'community', 'chat', 'openr
@@ -23,6 +23,15 @@ const DEFAULT_SOURCES = ['live', 'media', 'network', 'community', 'chat', 'openr
// Robot control (OpenVibe.Bot): bot.robot.*, bot.estop.*, bot.command.*
// Robot control (OpenVibe.Bot): bot.robot.*, bot.estop.*, bot.command.*
'bot'];
'bot'];
+/**
+ * Namespaces that belong to the network rather than to one service. `provider.*` carries storage
+ * provider health/capacity telemetry, which whichever first-party service observes a provider
+ * publishes (Media, Sources, Host, ...), so any service token may publish into it. Sources still
+ * own their own namespace first; a shared prefix never replaces `<source>.`. Developer apps can
+ * never publish a shared prefix. EVENTS_SHARED_PREFIXES (comma-separated) replaces the default.
+ */
+const DEFAULT_SHARED_PREFIXES = ['provider.'];
+
function sourcePrefixes(env) {
function sourcePrefixes(env) {
let map = Object.fromEntries(DEFAULT_SOURCES.map(s => [s, [`${s}.`]]));
let map = Object.fromEntries(DEFAULT_SOURCES.map(s => [s, [`${s}.`]]));
if (env.EVENTS_SOURCE_PREFIXES) {
if (env.EVENTS_SOURCE_PREFIXES) {
@@ -43,11 +52,25 @@ function sourcePrefixes(env) {
@@ -43,11 +52,25 @@ function sourcePrefixes(env) {
return map;
return map;
}
}
+function sharedPrefixes(env, map) {
+ const out = list(env.EVENTS_SHARED_PREFIXES, DEFAULT_SHARED_PREFIXES);
+ for (const p of out) {
+ if (!/^[a-z][a-z0-9-]*\.$/.test(p)) throw new Error(`EVENTS_SHARED_PREFIXES: bad prefix "${p}"`);
+ const source = p.slice(0, -1);
+ if (source === 'app' || source.startsWith('app-')) throw new Error(`EVENTS_SHARED_PREFIXES: "${p}" is reserved for developer apps`);
+ if (Object.prototype.hasOwnProperty.call(map, source)) {
+ throw new Error(`EVENTS_SHARED_PREFIXES: "${p}" belongs to source "${source}", not to the network`);
+ }
+ }
+ return out;
+}
+
function load(env = process.env) {
function load(env = process.env) {
const nodeEnv = env.NODE_ENV || 'development';
const nodeEnv = env.NODE_ENV || 'development';
const isProduction = nodeEnv === 'production';
const isProduction = nodeEnv === 'production';
const port = int(env.PORT, 4300);
const port = int(env.PORT, 4300);
const maxInflight = int(env.EVENTS_MAX_INFLIGHT, 20);
const maxInflight = int(env.EVENTS_MAX_INFLIGHT, 20);
+ const prefixes = sourcePrefixes(env);
return {
return {
port,
port,
host: env.HOST || '127.0.0.1',
host: env.HOST || '127.0.0.1',
@@ -85,7 +108,8 @@ function load(env = process.env) {
@@ -85,7 +108,8 @@ function load(env = process.env) {
receiptRetentionDays: int(env.EVENTS_RECEIPT_RETENTION_DAYS, 90),
receiptRetentionDays: int(env.EVENTS_RECEIPT_RETENTION_DAYS, 90),
pruneIntervalMs: int(env.EVENTS_PRUNE_INTERVAL_MS, 60 * 60 * 1000),
pruneIntervalMs: int(env.EVENTS_PRUNE_INTERVAL_MS, 60 * 60 * 1000),
- sourcePrefixes: sourcePrefixes(env),
+ sourcePrefixes: prefixes,
+ sharedPrefixes: sharedPrefixes(env, prefixes),
maxBatch: 100,
maxBatch: 100,
maxPayloadBytes: int(env.EVENTS_MAX_PAYLOAD_BYTES, 64 * 1024),
maxPayloadBytes: int(env.EVENTS_MAX_PAYLOAD_BYTES, 64 * 1024),
maxHops: int(env.EVENTS_MAX_HOPS, 8),
maxHops: int(env.EVENTS_MAX_HOPS, 8),
@@ -75,6 +75,25 @@ t('event_type must use a prefix the source owns', async () => {
@@ -75,6 +75,25 @@ t('event_type must use a prefix the source owns', async () => {
assert.strictEqual(r.body.code, 'events.type_not_allowed');
assert.strictEqual(r.body.code, 'events.type_not_allowed');
});
});
+t('any first-party service may publish the network-wide provider.* namespace', async () => {
+ const media = serviceToken('media', ['events.event.publish']);
+ // Media observes a provider's capacity/health and reports it under the network prefix.
+ for (const event_type of ['provider.health.degraded', 'provider.capacity.warning']) {
+ const r = await request(h.base, 'POST', '/api/v1/events', { token: media, body: envelope('media', { event_type }) });
+ assert.strictEqual(r.status, 201, r.text);
+ }
+ // The review's rename to Media's own namespace also passes, unchanged.
+ const r = await request(h.base, 'POST', '/api/v1/events', { token: media, body: envelope('media', { event_type: 'media.provider.health_degraded' }) });
+ assert.strictEqual(r.status, 201, r.text);
+ // Network-wide means shared: another service may report the same provider stream ...
+ const r2 = await request(h.base, 'POST', '/api/v1/events', { token: live, body: envelope('live', { event_type: 'provider.capacity.warning' }) });
+ assert.strictEqual(r2.status, 201, r2.text);
+ // ... but a shared prefix never hands a source another source's namespace.
+ const r3 = await request(h.base, 'POST', '/api/v1/events', { token: live, body: envelope('live', { event_type: 'media.vod.ready' }) });
+ assert.strictEqual(r3.status, 403);
+ assert.strictEqual(r3.body.code, 'events.type_not_allowed');
+});
+
t('a bot service principal publishes bot.* only', async () => {
t('a bot service principal publishes bot.* only', async () => {
const bot = serviceToken('bot', ['events.event.publish']);
const bot = serviceToken('bot', ['events.event.publish']);
let r = await request(h.base, 'POST', '/api/v1/events', { token: bot, body: envelope('bot', { event_type: 'bot.robot.online' }) });
let r = await request(h.base, 'POST', '/api/v1/events', { token: bot, body: envelope('bot', { event_type: 'bot.robot.online' }) });
@@ -42,6 +42,15 @@ t('config: every source prefix must start with the source', () => {
@@ -42,6 +42,15 @@ t('config: every source prefix must start with the source', () => {
assert.strictEqual(load({}).retentionDays, 30);
assert.strictEqual(load({}).retentionDays, 30);
});
});
+t('config: network-wide prefixes are shared, source namespaces stay owned', () => {
+ assert.deepStrictEqual(load({}).sharedPrefixes, ['provider.']);
+ const c = load({ EVENTS_SHARED_PREFIXES: 'provider.,storage.' });
+ assert.deepStrictEqual(c.sharedPrefixes, ['provider.', 'storage.']);
+ assert.throws(() => load({ EVENTS_SHARED_PREFIXES: 'provider' }), /bad prefix/);
+ assert.throws(() => load({ EVENTS_SHARED_PREFIXES: 'media.' }), /belongs to source "media"/);
+ assert.throws(() => load({ EVENTS_SHARED_PREFIXES: 'app.' }), /reserved for developer apps/);
+});
+
t('boot', async () => { h = await boot(); });
t('boot', async () => { h = await boot(); });
t('pull: cursor, topic filter, internal events included for services', async () => {
t('pull: cursor, topic filter, internal events included for services', async () => {