openvibers/openvibe.bot Commits
Bot: the job-usage relay mints its own Billing token (audience openvibe.billing); operators set only BOT_BILLING_URL. (#36)
Changes
@@@ -91,8 +91,10 @@ OV_OAUTH_CLIENT_SECRET=# ── OpenVibe.Billing: job usage (plan T14 L1) ───────────────# Where the per-second readings of jobs run on paired Nodes go: billing.usage.record, POST <url>/api/v1/usage.--# The token is a Network service token for audience openvibe.billing holding billing.usage.record. Either--# unset: readings wait in run_usage_outbox (never dropped) and a later relay with both set delivers each once.++# Bot mints its own Network service token (audience openvibe.billing, capability billing.usage.record) from++# OV_OAUTH_CLIENT_ID/SECRET and OV_NETWORK_INTERNAL_URL; BOT_BILLING_TOKEN, an operator-minted service token of++# the same grant, overrides it. BOT_BILLING_URL unset: readings wait in run_usage_outbox (never dropped) and a++# later relay delivers each once.#BOT_BILLING_URL=http://127.0.0.1:4600#BOT_BILLING_TOKEN=#BOT_BILLING_INTERVAL_MS=2000@@@ -40,7 +40,7 @@ OpenVibe.Bot pairs a robot's machines, keeps their state and gates every operato- Valkey (`VALKEY_URL`, optional): the shared per-actor limit counters; unset, limits are counted per process.- OpenVibe.Network: user and service tokens, node tokens, pairing when the authority is `network`, and identity resolution (a `@username` to a subject, and subjects to names, for the panel's People card).- OpenRe.Stream: the WHIP ingest base (`BOT_WHIP_BASE`) devices publish to.--- OpenVibe.Billing: job usage readings go to `billing.usage.record` (`BOT_BILLING_URL`, `BOT_BILLING_TOKEN`); unset, they wait in `run_usage_outbox`.++- OpenVibe.Billing: job usage readings go to `billing.usage.record` (`BOT_BILLING_URL`; Bot mints its own Network service token for audience `openvibe.billing` with `billing.usage.record`, and `BOT_BILLING_TOKEN` overrides it); unset, they wait in `run_usage_outbox`.- `openvibe-contracts` v0.106.0, `openvibe-sdk` v0.26.0 and `openvibe-shared` v2.11.0 (package.json).## Capabilities@@@ -84,6 +84,8 @@ Set `BOT_WHIP_BASE` in `/etc/openvibe/bot.env` to the WHIP ingest base each deviThe publish key is always an ingest key OpenRe.Stream issued for the robot's stream (OpenRe's WHIP ingest admits no other): set `BOT_OPENRE_URL` and Bot mints its own Network service token for audience `openvibe.openre` from its OAuth client (`OV_OAUTH_CLIENT_ID`/`OV_OAUTH_CLIENT_SECRET` against `OV_NETWORK_INTERNAL_URL`), holding `openre.stream.read`, `openre.stream.write`, `openre.key.rotate` and `openre.session.read`/`openre.output.read`/`openre.output.write` — Network's grant to the `bot` client; the session and output capabilities are for later panel video and restreaming. `BOT_OPENRE_TOKEN` (a Network service token with the same grant, or `openre.stream.*` with `openre.key.rotate`) is an optional operator override; `BOT_OPENRE_TIMEOUT_MS`, default 8000, bounds each call. The same token covers the streaming toggles: `GET /api/v1/robots/:id/streaming` reads the stream with `openre.stream.read`, `POST` PATCHes `recording_mode`/`mirror_to_live` with `openre.stream.write`. Pairing creates the robot's OpenRe stream (or rotates the one it has), a credential rotation rotates its key, revoking a device or removing a robot revokes the key and ends the live session. Bot stores the stream id and the key's hint, never the key. With `BOT_OPENRE_URL` unset, or neither `BOT_OPENRE_TOKEN` nor the Network client credentials available, devices still pair but get no publish key and the answer says `"video": "not_configured"`; Bot mints no key of its own.++Set `BOT_BILLING_URL` in `/etc/openvibe/bot.env` to OpenVibe.Billing's base URL and Bot mints its own Network service token for audience `openvibe.billing` (`billing.usage.record`, Network's grant to the `bot` client) from the same OAuth client credentials, posting each job usage reading to `<BOT_BILLING_URL>/api/v1/usage`; `BOT_BILLING_TOKEN` is an optional operator-minted override, and `BOT_BILLING_TIMEOUT_MS`, default 5000, bounds each post. With `BOT_BILLING_URL` unset the relay stays off and the readings wait in `run_usage_outbox`, retried across restarts and never dropped.++Set `BOT_ONVIF_CAMERAS` (JSON, unset by default) to the server-side cameras Bot may connect to: a map from a robot id (each key names exactly one robot; there is no wildcard) to `{ "url": "http://<host>/onvif/ptz_service", "username_ref": "CAM_USER", "password_ref": "CAM_PASSWORD" }`. The refs are the *names* of the secrets (env vars, or keys a deployment resolves through a secret store), read per request; the URL must be http(s), on one host and carry no credentials, and `BOT_ONVIF_TIMEOUT_MS` (default 5000) bounds each request. Anything else — including a key that is not a robot id — refuses to boot. A robot whose profile is `camera.onvif` and that has a configured camera attaches server-side at boot and when its panel is opened.`/install` is proxied to the app, which answers a 302 to OpenVibe.Node's installer script (`BOT_INSTALLER_SOURCE_URL`, by default the `install.sh` asset on OpenVibe.Node's [latest release](https://github.com/OpenVibers/OpenVibe.Node/releases/latest/download/) — a released, checksummed installer that never drifts with Node's `main`, whose CI keeps `dist/install.sh` byte-identical to `install/install.sh`; https on an allow-listed GitHub or openvibe.bot host, checked at boot, and no query parameter changes the target). The pairing's installer command adds `--driver adeept|adeept-mecanum|cozmo` for those profiles and nothing for the others (the dry-run `none`).@@@ -295,10 +295,11 @@ No field depends on when or how often a frame arrived, and no rating field is seNOTHING`). `run_jobs.usage_read` (seconds queued) only moves forward. Every reason is billed for the seconds theprocess held, never beyond the job's own `limits.wall_ms` (a later second is not metered and `wall_ms` is cappedthere). The relay posts each reading to OpenVibe.Billing's `billing.usage.record` (`POST <BOT_BILLING_URL>/api/v1/usage`,--`Authorization: Bearer <BOT_BILLING_TOKEN>`), one per request; 201 (written) and 200 (identical replay) mark it--sent. 400/409/413/422 mark it rejected (kept, never resent). Anything else, and `BOT_BILLING_URL` or--`BOT_BILLING_TOKEN` unset, leaves it queued and retried with backoff: a reading is never dropped and never billed--twice.++`Authorization: Bearer <token>` with the operator's `BOT_BILLING_TOKEN` or a Network service token Bot mints for++audience `openvibe.billing`), one per request; 201 (written) and 200 (identical replay) mark it++sent. 400/409/413/422 mark it rejected (kept, never resent). Anything else, and `BOT_BILLING_URL` unset (or no++token and no Network client to mint one), leaves it queued and retried with backoff: a reading is never dropped++and never billed twice.## 2. `/control` — the operator WebSocket@@@ -227,8 +227,11 @@ function loadConfig(env = process.env) {timeoutMs: Math.max(100, int(env.BOT_OPENRE_TIMEOUT_MS, 8000)),},// OpenVibe.Billing (plan T14 L1): job usage readings go to billing.usage.record (POST <url>/api/v1/usage).-- // token is a Network service token for audience openvibe.billing holding billing.usage.record; it is never-- // logged or returned. Either unset: readings wait in run_usage_outbox (never dropped) until both are set.++ // By default Bot mints its own Network service token (audience openvibe.billing, capability++ // billing.usage.record) from its OAuth client, exactly as the OpenRe client does; BOT_BILLING_TOKEN, an++ // operator-minted token of the same grant, overrides it. The token is never logged or returned. With++ // BOT_BILLING_URL unset (or no token and no Network client to mint one): readings wait in++ // run_usage_outbox (never dropped).billing: {url: trim(env.BOT_BILLING_URL || ''),token: String(env.BOT_BILLING_TOKEN || '').trim(),@@@ -12,13 +12,17 @@* resend, a job_usage and the job_exit backfill of one second are one key: the outbox keeps the first* (ON CONFLICT DO NOTHING) and Billing replays an identical reading (200) and refuses a different one (409).*-- * The relay (openvibe-sdk createPgOutbox) runs while BOT_BILLING_URL and BOT_BILLING_TOKEN are set. If either is-- * unset, or Billing is down, has no grant for Bot yet or answers 401/403/404/429/5xx, readings wait and are-- * retried with backoff, across restarts: never dropped. Only Billing refusing the reading itself (400, 409, 413,-- * 422) marks a row rejected; it is kept with its error and never sent again, so nothing is billed twice.++ * The relay (openvibe-sdk createPgOutbox) runs while BOT_BILLING_URL is set — with BOT_BILLING_TOKEN when the++ * operator minted one, otherwise with a Network service token Bot mints for audience openvibe.billing++ * (capability billing.usage.record) from its OAuth client, the same helper and cache the OpenRe client uses. If++ * the URL is unset (or Bot has no token and no Network client to mint one), or Billing is down, has no grant for++ * Bot yet or answers 401/403/404/429/5xx, readings wait and are retried with backoff, across restarts: never++ * dropped. Only Billing refusing the reading itself (400, 409, 413, 422) marks a row rejected; it is kept with++ * its error and never sent again, so nothing is billed twice.*/--const { validate } = require('openvibe-contracts');++const { validate, serviceAuth } = require('openvibe-contracts');const { createPgOutbox } = require('openvibe-sdk/events');++const { BotError } = require('../util');const TABLE = 'run_usage_outbox'; // migrations/0004_run_jobs.sqlconst SCHEMA = 'platform.usage-sample@1';@@@ -28,6 +32,11 @@ const UNIT = 's';const SOURCE = 'openvibe-node.worker';const REFUSED = new Set([400, 409, 413, 422]);++// Network's grant to the `bot` client on audience openvibe.billing (Network server/identity/principals.js++// DEFAULT_GRANTS): billing.usage.record is the one capability a reading is posted with.++const BILLING_AUDIENCE = 'openvibe.billing';++const BILLING_SCOPE = 'billing.usage.record';++/** The reading for second `n` of a job; `quantity` is 1 except for the partial last second. */function secondReading({ jobId, n, quantity = 1, startedMs, nodeId, project = null, subject = null, provider = null }) {if (!Number.isInteger(n) || n < 0) throw new TypeError(`second must be an integer >= 0, not ${n}`);@@@ -66,15 +75,18 @@ async function enqueue(t, reading, at) {}// createPgOutbox's `events` seam pointed at Billing. isPermanent() reads err.status: 422 rejects the row, 503 retries.--function billingSink({ url, token, fetchImpl, timeoutMs }) {++function billingSink({ url, authorization, onUnauthorized, fetchImpl, timeoutMs }) {return {prepare: (reading) => reading,async publish(reading) {const res = await fetchImpl(`${url}/api/v1/usage`, {method: 'POST',-- headers: { Authorization: `Bearer ${token}`, 'Content-Type': 'application/json', Accept: 'application/json' },++ headers: { Authorization: await authorization(), 'Content-Type': 'application/json', Accept: 'application/json' },body: JSON.stringify(reading), signal: AbortSignal.timeout(timeoutMs),});++ // A minted token Billing no longer accepts (rotated signing key, clock): drop it so the next sweep++ // mints a fresh one; the reading stays queued, never rejected for it.++ if (res.status === 401 && onUnauthorized) onUnauthorized();// 201 written, 200 an identical reading replayed: either way Billing holds it.if (!res.ok) { const err = new Error(`billing.usage.record answered ${res.status}`); err.status = REFUSED.has(res.status) ? 422 : 503; throw err; }return { event_id: reading.id, seq: null };@@@ -83,20 +95,39 @@ function billingSink({ url, token, fetchImpl, timeoutMs }) {}/**-- * The relay from run_usage_outbox to Billing. enabled only with config.billing.url and .token; otherwise start(),-- * kick() and flush() do nothing and readings accumulate until a process with both set sweeps them.++ * The relay from run_usage_outbox to Billing. enabled with config.billing.url and either an operator++ * BOT_BILLING_TOKEN or the Network client credentials to mint one (exactly as the OpenRe client does); with++ * neither, or with the URL unset, start(), kick() and flush() do nothing and readings accumulate until a++ * process with the URL and a token sweeps them.*/function createUsageRelay({ db, config, fetchImpl = globalThis.fetch, now = () => Date.now(), log = console }) {const { url, token, intervalMs, timeoutMs } = config.billing;const count = (where) => db.value(`SELECT count(*) FROM ${TABLE} WHERE ${where}`).then(Number);const counts = { pending: () => count('sent_at IS NULL AND rejected_at IS NULL'), rejected: () => count('rejected_at IS NOT NULL') };-- if (!url || !token) {++ // An operator-minted BOT_BILLING_TOKEN wins; otherwise Bot mints its own from its Network client (the same++ // openvibe-contracts token client, and its cache, that server/openre/client.js uses for openvibe.openre).++ const { clientId, clientSecret } = config.oauth || {};++ const internalUrl = (config.network || {}).internalUrl;++ const tokens = !token && clientSecret && internalUrl++ ? serviceAuth.createTokenClient({++ tokenUrl: `${internalUrl}/oauth/token`, clientId, clientSecret, audience: BILLING_AUDIENCE, scope: BILLING_SCOPE, fetchImpl,++ })++ : null;++ if (!url || (!token && !tokens)) {const idle = async () => ({ sent: 0, failed: 0, rejected: 0 });return { enabled: false, start() {}, stop: async () => {}, kick() {}, flush: idle, ...counts };}++ /** The Authorization header of this post: the operator's token, or a freshly cached minted one. */++ const authorization = tokens++ ? async () => {++ try { return (await tokens.authHeaders()).Authorization; }++ catch (e) { throw new BotError(503, 'bot.billing_unavailable', 'Bot could not mint its Billing token from Network'); }++ }++ : async () => `Bearer ${token}`;let lastError = null;const relay = createPgOutbox(db, {-- events: billingSink({ url, token, fetchImpl, timeoutMs }), table: TABLE, batchSize: 1, intervalMs, now,++ events: billingSink({ url, authorization, onUnauthorized: tokens ? () => tokens.invalidate() : null, fetchImpl, timeoutMs }),++ table: TABLE, batchSize: 1, intervalMs, now,onError: (err) => { const m = err && err.message; if (m !== lastError) log.warn(`[Bot] usage reading not delivered: ${m}`); lastError = m; },});return { enabled: true, start: relay.start, stop: relay.stop, kick: relay.kick, flush: relay.flush, ...counts };nnew file mode 100644@@@ -0,0 +1,181 @@++'use strict';++// The usage relay authenticates to OpenVibe.Billing (plan T14 L1) the same way Bot talks to OpenRe: with an++// operator BOT_BILLING_TOKEN when one is set, otherwise a Network service token it mints for audience++// openvibe.billing (capability billing.usage.record) from its OAuth client, cached by the shared token client++// until 60 s before expiry. Here the relay runs against a fake Network token endpoint and a fake Billing: with++// only BOT_BILLING_URL set it mints one token for the sweep, posts each queued reading once with it, re-mints++// after expiry; a 401 from Billing drops the cached token and a 401/5xx keeps every reading queued (never++// dropped, never rejected); an operator token overrides minting; and with the URL unset the relay makes no call++// at all.++const assert = require('assert');++const { check, done } = require('./helpers/app');++const { testDb } = require('./helpers/db');++const { secondReading, enqueue, createUsageRelay, TABLE } = require('../server/jobs/metering');++++// Network's grant to the `bot` client on audience openvibe.billing (OpenVibe.Network DEFAULT_GRANTS).++const MINTED_AUDIENCE = 'openvibe.billing';++const MINTED_SCOPE = 'billing.usage.record';++const BILLING_URL = 'http://billing.test';++const NETWORK_URL = 'http://network.test';++const STARTED_MS = 1790935200000;++++const quiet = { warn() {} };++const sleep = (ms) => new Promise((r) => setTimeout(r, ms));++const config = (billing = {}) => ({++ oauth: { clientId: 'bot', clientSecret: 'shh' },++ network: { internalUrl: NETWORK_URL },++ billing: { url: BILLING_URL, token: '', intervalMs: 100, timeoutMs: 2000, ...billing },++});++++/**++ * A fake Network token endpoint (POST /oauth/token → mint_1, mint_2, …) and a fake Billing (POST++ * /api/v1/usage), behind one fetch. tokenCalls records every client-credentials request; posts records every++ * usage post ({ auth, status, body }); billingStatus() decides each post's status.++ */++function fakeServices({ expiresIn = 300, billingStatus = () => 201 } = {}) {++ const tokenCalls = [];++ const posts = [];++ let minted = 0;++ const fetchImpl = async (url, init) => {++ if (url === `${NETWORK_URL}/oauth/token`) {++ const body = Object.fromEntries(new URLSearchParams(init.body));++ tokenCalls.push({ client_id: body.client_id, client_secret: body.client_secret, audience: body.audience, scope: body.scope, grant_type: body.grant_type });++ if (body.client_secret !== 'shh') return new Response(JSON.stringify({ error: 'invalid_client' }), { status: 401 });++ return new Response(JSON.stringify({ access_token: `mint_${++minted}`, token_type: 'Bearer', expires_in: expiresIn }), { status: 200, headers: { 'Content-Type': 'application/json' } });++ }++ if (url === `${BILLING_URL}/api/v1/usage`) {++ const status = billingStatus();++ posts.push({ auth: init.headers.Authorization, status, body: JSON.parse(init.body) });++ return new Response('{}', { status });++ }++ throw new Error(`unexpected fetch ${url}`);++ };++ return { fetchImpl, tokenCalls, posts };++}++++(async () => {++ const store = await testDb();++ const db = store.db;++ // One relay sweep at a time: readings queued inside a transaction, exactly as dispatch/job_exit do.++ const queue = (prefix, seconds) => db.tx(async (t) => {++ for (const n of seconds) {++ await enqueue(t, secondReading({ jobId: `job_${prefix}`, n, startedMs: STARTED_MS, nodeId: 'dev_test' }), Date.now());++ }++ });++ const countLike = (prefix, where) => db.value(`SELECT count(*) FROM ${TABLE} WHERE event_id LIKE $1 AND ${where}`, [`run:job_${prefix}:%`]).then(Number);++ const pending = (prefix) => countLike(prefix, 'sent_at IS NULL AND rejected_at IS NULL');++ const rejected = (prefix) => countLike(prefix, 'rejected_at IS NOT NULL');++ const sent = (prefix) => countLike(prefix, 'sent_at IS NOT NULL');++++ await check('with only BOT_BILLING_URL set the relay mints one token and posts each reading once with it', async () => {++ const svc = fakeServices({ expiresIn: 61 });++ const clock = { t: Date.now() };++ const relay = createUsageRelay({ db, config: config(), fetchImpl: svc.fetchImpl, now: () => clock.t, log: quiet });++ assert.strictEqual(relay.enabled, true);++++ await queue('mint', [0, 1, 2]);++ const before = svc.tokenCalls.length;++ await relay.flush();++ assert.deepStrictEqual(svc.tokenCalls.slice(before),++ [{ client_id: 'bot', client_secret: 'shh', audience: MINTED_AUDIENCE, scope: MINTED_SCOPE, grant_type: 'client_credentials' }]);++ assert.strictEqual(svc.posts.length, 3);++ assert.deepStrictEqual(svc.posts.map((p) => p.auth), ['Bearer mint_1', 'Bearer mint_1', 'Bearer mint_1']);++ assert.deepStrictEqual(svc.posts.map((p) => p.body.idempotency_key), ['run:job_mint:0', 'run:job_mint:1', 'run:job_mint:2']);++ assert.strictEqual(await pending('mint'), 0);++ assert.strictEqual(await sent('mint'), 3);++ assert.strictEqual(await rejected('mint'), 0);++++ // A second read is served by the same cached token, not a second mint.++ await queue('mint2', [0]);++ await relay.flush();++ assert.strictEqual(svc.tokenCalls.length, before + 1, 'the cached token was reused');++ assert.strictEqual(svc.posts.at(-1).auth, 'Bearer mint_1');++++ // Past its expiry (expires_in 61 ⇒ cached only ~1 s) the relay mints again.++ await sleep(1100);++ await queue('mint3', [0]);++ await relay.flush();++ assert.strictEqual(svc.tokenCalls.length, before + 2, 'the expired token was replaced');++ assert.strictEqual(svc.posts.at(-1).auth, 'Bearer mint_2');++ assert.strictEqual(await pending('mint3'), 0);++ await relay.stop();++ });++++ await check('a 401 or 5xx from Billing keeps every reading queued, and the 401 drops the minted token', async () => {++ const svc = fakeServices();++ const clock = { t: Date.now() };++ let mode = 503;++ const relay = createUsageRelay({ db, config: config(), fetchImpl: fakeServices200(svc, () => mode), now: () => clock.t, log: quiet });++ assert.strictEqual(relay.enabled, true);++++ await queue('down', [0, 1]);++ await relay.flush(); // 5xx++ assert.strictEqual(await pending('down'), 2, 'a 5xx leaves both readings queued');++ assert.strictEqual(await rejected('down'), 0);++ clock.t += 2000;++ await relay.flush(); // 5xx again, after the retry backoff++ assert.strictEqual(await pending('down'), 2);++ assert.strictEqual(await rejected('down'), 0);++++ const mintedBefore = svc.tokenCalls.length;++ const auth401 = svc.posts.at(-1).auth;++ mode = 401;++ clock.t += 10000;++ await relay.flush();++ assert.strictEqual(await pending('down'), 2, 'a 401 leaves both readings queued');++ assert.strictEqual(await rejected('down'), 0);++++ mode = 201;++ clock.t += 60000;++ await relay.flush();++ assert.strictEqual(await pending('down'), 0);++ assert.strictEqual(await rejected('down'), 0);++ assert.strictEqual(svc.tokenCalls.length, mintedBefore + 1, 'the 401 dropped the cached token, so one fresh mint served the retry');++ assert.notStrictEqual(svc.posts.at(-1).auth, auth401, 'the retry carried a freshly minted token');++ assert.deepStrictEqual(svc.posts.filter((p) => p.status === 201).map((p) => p.auth), ['Bearer mint_2', 'Bearer mint_2']);++ await relay.stop();++ });++++ await check('BOT_BILLING_TOKEN overrides minting: it is sent verbatim and Network is never asked', async () => {++ const svc = fakeServices();++ const relay = createUsageRelay({ db, config: config({ token: 'operator-minted' }), fetchImpl: svc.fetchImpl, now: () => Date.now(), log: quiet });++ assert.strictEqual(relay.enabled, true);++ await queue('static', [0]);++ await relay.flush();++ assert.strictEqual(svc.tokenCalls.length, 0, 'BOT_BILLING_TOKEN was set: nothing was minted');++ assert.deepStrictEqual(svc.posts.map((p) => p.auth), ['Bearer operator-minted']);++ assert.strictEqual(await pending('static'), 0);++ await relay.stop();++ });++++ await check('with BOT_BILLING_URL unset the relay is off and makes no call; so is a URL with no way to authenticate', async () => {++ const svc = fakeServices();++ const noCall = async (url) => { throw new Error(`the relay called ${url} while off`); };++ const off = createUsageRelay({ db, config: config({ url: '' }), fetchImpl: noCall, now: () => Date.now(), log: quiet });++ assert.strictEqual(off.enabled, false);++ assert.deepStrictEqual(await off.flush(), { sent: 0, failed: 0, rejected: 0 });++ off.start(); off.kick(); await off.stop();++ await queue('off', [0]);++ assert.strictEqual(await pending('off'), 1, 'readings wait while the URL is unset');++ // A URL but neither an operator token nor the Network client credentials: still off, still nothing sent.++ const noCreds = createUsageRelay({ db, config: { oauth: {}, network: {}, billing: { url: BILLING_URL, token: '', intervalMs: 100, timeoutMs: 2000 } }, fetchImpl: noCall, now: () => Date.now(), log: quiet });++ assert.strictEqual(noCreds.enabled, false);++ assert.deepStrictEqual(await noCreds.flush(), { sent: 0, failed: 0, rejected: 0 });++ assert.ok(svc.tokenCalls.length === 0 && svc.posts.length === 0);++ });++++ await store.close();++ done();++})().catch((e) => { console.error(e); process.exit(1); });++++/** The fake Network endpoint with Billing's status coming from `mode()` (so a test can flip it mid-run). */++function fakeServices200(svc, mode) {++ return async (url, init) => {++ if (url === `${BILLING_URL}/api/v1/usage`) {++ const status = mode();++ svc.posts.push({ auth: init.headers.Authorization, status, body: JSON.parse(init.body) });++ return new Response('{}', { status });++ }++ return svc.fetchImpl(url, init);++ };++}
@@ -91,8 +91,10 @@ OV_OAUTH_CLIENT_SECRET=
@@ -91,8 +91,10 @@ OV_OAUTH_CLIENT_SECRET=
# ── OpenVibe.Billing: job usage (plan T14 L1) ───────────────
# ── OpenVibe.Billing: job usage (plan T14 L1) ───────────────
# Where the per-second readings of jobs run on paired Nodes go: billing.usage.record, POST <url>/api/v1/usage.
# Where the per-second readings of jobs run on paired Nodes go: billing.usage.record, POST <url>/api/v1/usage.
-# The token is a Network service token for audience openvibe.billing holding billing.usage.record. Either
-# unset: readings wait in run_usage_outbox (never dropped) and a later relay with both set delivers each once.
+# Bot mints its own Network service token (audience openvibe.billing, capability billing.usage.record) from
+# OV_OAUTH_CLIENT_ID/SECRET and OV_NETWORK_INTERNAL_URL; BOT_BILLING_TOKEN, an operator-minted service token of
+# the same grant, overrides it. BOT_BILLING_URL unset: readings wait in run_usage_outbox (never dropped) and a
+# later relay delivers each once.
#BOT_BILLING_URL=http://127.0.0.1:4600
#BOT_BILLING_URL=http://127.0.0.1:4600
#BOT_BILLING_TOKEN=
#BOT_BILLING_TOKEN=
#BOT_BILLING_INTERVAL_MS=2000
#BOT_BILLING_INTERVAL_MS=2000
@@ -40,7 +40,7 @@ OpenVibe.Bot pairs a robot's machines, keeps their state and gates every operato
@@ -40,7 +40,7 @@ OpenVibe.Bot pairs a robot's machines, keeps their state and gates every operato
- Valkey (`VALKEY_URL`, optional): the shared per-actor limit counters; unset, limits are counted per process.
- Valkey (`VALKEY_URL`, optional): the shared per-actor limit counters; unset, limits are counted per process.
- OpenVibe.Network: user and service tokens, node tokens, pairing when the authority is `network`, and identity resolution (a `@username` to a subject, and subjects to names, for the panel's People card).
- OpenVibe.Network: user and service tokens, node tokens, pairing when the authority is `network`, and identity resolution (a `@username` to a subject, and subjects to names, for the panel's People card).
- OpenRe.Stream: the WHIP ingest base (`BOT_WHIP_BASE`) devices publish to.
- OpenRe.Stream: the WHIP ingest base (`BOT_WHIP_BASE`) devices publish to.
-- OpenVibe.Billing: job usage readings go to `billing.usage.record` (`BOT_BILLING_URL`, `BOT_BILLING_TOKEN`); unset, they wait in `run_usage_outbox`.
+- OpenVibe.Billing: job usage readings go to `billing.usage.record` (`BOT_BILLING_URL`; Bot mints its own Network service token for audience `openvibe.billing` with `billing.usage.record`, and `BOT_BILLING_TOKEN` overrides it); unset, they wait in `run_usage_outbox`.
- `openvibe-contracts` v0.106.0, `openvibe-sdk` v0.26.0 and `openvibe-shared` v2.11.0 (package.json).
- `openvibe-contracts` v0.106.0, `openvibe-sdk` v0.26.0 and `openvibe-shared` v2.11.0 (package.json).
## Capabilities
## Capabilities
@@ -84,6 +84,8 @@ Set `BOT_WHIP_BASE` in `/etc/openvibe/bot.env` to the WHIP ingest base each devi
@@ -84,6 +84,8 @@ Set `BOT_WHIP_BASE` in `/etc/openvibe/bot.env` to the WHIP ingest base each devi
The publish key is always an ingest key OpenRe.Stream issued for the robot's stream (OpenRe's WHIP ingest admits no other): set `BOT_OPENRE_URL` and Bot mints its own Network service token for audience `openvibe.openre` from its OAuth client (`OV_OAUTH_CLIENT_ID`/`OV_OAUTH_CLIENT_SECRET` against `OV_NETWORK_INTERNAL_URL`), holding `openre.stream.read`, `openre.stream.write`, `openre.key.rotate` and `openre.session.read`/`openre.output.read`/`openre.output.write` — Network's grant to the `bot` client; the session and output capabilities are for later panel video and restreaming. `BOT_OPENRE_TOKEN` (a Network service token with the same grant, or `openre.stream.*` with `openre.key.rotate`) is an optional operator override; `BOT_OPENRE_TIMEOUT_MS`, default 8000, bounds each call. The same token covers the streaming toggles: `GET /api/v1/robots/:id/streaming` reads the stream with `openre.stream.read`, `POST` PATCHes `recording_mode`/`mirror_to_live` with `openre.stream.write`. Pairing creates the robot's OpenRe stream (or rotates the one it has), a credential rotation rotates its key, revoking a device or removing a robot revokes the key and ends the live session. Bot stores the stream id and the key's hint, never the key. With `BOT_OPENRE_URL` unset, or neither `BOT_OPENRE_TOKEN` nor the Network client credentials available, devices still pair but get no publish key and the answer says `"video": "not_configured"`; Bot mints no key of its own.
The publish key is always an ingest key OpenRe.Stream issued for the robot's stream (OpenRe's WHIP ingest admits no other): set `BOT_OPENRE_URL` and Bot mints its own Network service token for audience `openvibe.openre` from its OAuth client (`OV_OAUTH_CLIENT_ID`/`OV_OAUTH_CLIENT_SECRET` against `OV_NETWORK_INTERNAL_URL`), holding `openre.stream.read`, `openre.stream.write`, `openre.key.rotate` and `openre.session.read`/`openre.output.read`/`openre.output.write` — Network's grant to the `bot` client; the session and output capabilities are for later panel video and restreaming. `BOT_OPENRE_TOKEN` (a Network service token with the same grant, or `openre.stream.*` with `openre.key.rotate`) is an optional operator override; `BOT_OPENRE_TIMEOUT_MS`, default 8000, bounds each call. The same token covers the streaming toggles: `GET /api/v1/robots/:id/streaming` reads the stream with `openre.stream.read`, `POST` PATCHes `recording_mode`/`mirror_to_live` with `openre.stream.write`. Pairing creates the robot's OpenRe stream (or rotates the one it has), a credential rotation rotates its key, revoking a device or removing a robot revokes the key and ends the live session. Bot stores the stream id and the key's hint, never the key. With `BOT_OPENRE_URL` unset, or neither `BOT_OPENRE_TOKEN` nor the Network client credentials available, devices still pair but get no publish key and the answer says `"video": "not_configured"`; Bot mints no key of its own.
+Set `BOT_BILLING_URL` in `/etc/openvibe/bot.env` to OpenVibe.Billing's base URL and Bot mints its own Network service token for audience `openvibe.billing` (`billing.usage.record`, Network's grant to the `bot` client) from the same OAuth client credentials, posting each job usage reading to `<BOT_BILLING_URL>/api/v1/usage`; `BOT_BILLING_TOKEN` is an optional operator-minted override, and `BOT_BILLING_TIMEOUT_MS`, default 5000, bounds each post. With `BOT_BILLING_URL` unset the relay stays off and the readings wait in `run_usage_outbox`, retried across restarts and never dropped.
+
Set `BOT_ONVIF_CAMERAS` (JSON, unset by default) to the server-side cameras Bot may connect to: a map from a robot id (each key names exactly one robot; there is no wildcard) to `{ "url": "http://<host>/onvif/ptz_service", "username_ref": "CAM_USER", "password_ref": "CAM_PASSWORD" }`. The refs are the *names* of the secrets (env vars, or keys a deployment resolves through a secret store), read per request; the URL must be http(s), on one host and carry no credentials, and `BOT_ONVIF_TIMEOUT_MS` (default 5000) bounds each request. Anything else — including a key that is not a robot id — refuses to boot. A robot whose profile is `camera.onvif` and that has a configured camera attaches server-side at boot and when its panel is opened.
Set `BOT_ONVIF_CAMERAS` (JSON, unset by default) to the server-side cameras Bot may connect to: a map from a robot id (each key names exactly one robot; there is no wildcard) to `{ "url": "http://<host>/onvif/ptz_service", "username_ref": "CAM_USER", "password_ref": "CAM_PASSWORD" }`. The refs are the *names* of the secrets (env vars, or keys a deployment resolves through a secret store), read per request; the URL must be http(s), on one host and carry no credentials, and `BOT_ONVIF_TIMEOUT_MS` (default 5000) bounds each request. Anything else — including a key that is not a robot id — refuses to boot. A robot whose profile is `camera.onvif` and that has a configured camera attaches server-side at boot and when its panel is opened.
`/install` is proxied to the app, which answers a 302 to OpenVibe.Node's installer script (`BOT_INSTALLER_SOURCE_URL`, by default the `install.sh` asset on OpenVibe.Node's [latest release](https://github.com/OpenVibers/OpenVibe.Node/releases/latest/download/) — a released, checksummed installer that never drifts with Node's `main`, whose CI keeps `dist/install.sh` byte-identical to `install/install.sh`; https on an allow-listed GitHub or openvibe.bot host, checked at boot, and no query parameter changes the target). The pairing's installer command adds `--driver adeept|adeept-mecanum|cozmo` for those profiles and nothing for the others (the dry-run `none`).
`/install` is proxied to the app, which answers a 302 to OpenVibe.Node's installer script (`BOT_INSTALLER_SOURCE_URL`, by default the `install.sh` asset on OpenVibe.Node's [latest release](https://github.com/OpenVibers/OpenVibe.Node/releases/latest/download/) — a released, checksummed installer that never drifts with Node's `main`, whose CI keeps `dist/install.sh` byte-identical to `install/install.sh`; https on an allow-listed GitHub or openvibe.bot host, checked at boot, and no query parameter changes the target). The pairing's installer command adds `--driver adeept|adeept-mecanum|cozmo` for those profiles and nothing for the others (the dry-run `none`).
@@ -295,10 +295,11 @@ No field depends on when or how often a frame arrived, and no rating field is se
@@ -295,10 +295,11 @@ No field depends on when or how often a frame arrived, and no rating field is se
NOTHING`). `run_jobs.usage_read` (seconds queued) only moves forward. Every reason is billed for the seconds the
NOTHING`). `run_jobs.usage_read` (seconds queued) only moves forward. Every reason is billed for the seconds the
process held, never beyond the job's own `limits.wall_ms` (a later second is not metered and `wall_ms` is capped
process held, never beyond the job's own `limits.wall_ms` (a later second is not metered and `wall_ms` is capped
there). The relay posts each reading to OpenVibe.Billing's `billing.usage.record` (`POST <BOT_BILLING_URL>/api/v1/usage`,
there). The relay posts each reading to OpenVibe.Billing's `billing.usage.record` (`POST <BOT_BILLING_URL>/api/v1/usage`,
-`Authorization: Bearer <BOT_BILLING_TOKEN>`), one per request; 201 (written) and 200 (identical replay) mark it
-sent. 400/409/413/422 mark it rejected (kept, never resent). Anything else, and `BOT_BILLING_URL` or
-`BOT_BILLING_TOKEN` unset, leaves it queued and retried with backoff: a reading is never dropped and never billed
-twice.
+`Authorization: Bearer <token>` with the operator's `BOT_BILLING_TOKEN` or a Network service token Bot mints for
+audience `openvibe.billing`), one per request; 201 (written) and 200 (identical replay) mark it
+sent. 400/409/413/422 mark it rejected (kept, never resent). Anything else, and `BOT_BILLING_URL` unset (or no
+token and no Network client to mint one), leaves it queued and retried with backoff: a reading is never dropped
+and never billed twice.
## 2. `/control` — the operator WebSocket
## 2. `/control` — the operator WebSocket
@@ -227,8 +227,11 @@ function loadConfig(env = process.env) {
@@ -227,8 +227,11 @@ function loadConfig(env = process.env) {
timeoutMs: Math.max(100, int(env.BOT_OPENRE_TIMEOUT_MS, 8000)),
timeoutMs: Math.max(100, int(env.BOT_OPENRE_TIMEOUT_MS, 8000)),
},
},
// OpenVibe.Billing (plan T14 L1): job usage readings go to billing.usage.record (POST <url>/api/v1/usage).
// OpenVibe.Billing (plan T14 L1): job usage readings go to billing.usage.record (POST <url>/api/v1/usage).
- // token is a Network service token for audience openvibe.billing holding billing.usage.record; it is never
- // logged or returned. Either unset: readings wait in run_usage_outbox (never dropped) until both are set.
+ // By default Bot mints its own Network service token (audience openvibe.billing, capability
+ // billing.usage.record) from its OAuth client, exactly as the OpenRe client does; BOT_BILLING_TOKEN, an
+ // operator-minted token of the same grant, overrides it. The token is never logged or returned. With
+ // BOT_BILLING_URL unset (or no token and no Network client to mint one): readings wait in
+ // run_usage_outbox (never dropped).
billing: {
billing: {
url: trim(env.BOT_BILLING_URL || ''),
url: trim(env.BOT_BILLING_URL || ''),
token: String(env.BOT_BILLING_TOKEN || '').trim(),
token: String(env.BOT_BILLING_TOKEN || '').trim(),
@@ -12,13 +12,17 @@
@@ -12,13 +12,17 @@
* resend, a job_usage and the job_exit backfill of one second are one key: the outbox keeps the first
* resend, a job_usage and the job_exit backfill of one second are one key: the outbox keeps the first
* (ON CONFLICT DO NOTHING) and Billing replays an identical reading (200) and refuses a different one (409).
* (ON CONFLICT DO NOTHING) and Billing replays an identical reading (200) and refuses a different one (409).
*
*
- * The relay (openvibe-sdk createPgOutbox) runs while BOT_BILLING_URL and BOT_BILLING_TOKEN are set. If either is
- * unset, or Billing is down, has no grant for Bot yet or answers 401/403/404/429/5xx, readings wait and are
- * retried with backoff, across restarts: never dropped. Only Billing refusing the reading itself (400, 409, 413,
- * 422) marks a row rejected; it is kept with its error and never sent again, so nothing is billed twice.
+ * The relay (openvibe-sdk createPgOutbox) runs while BOT_BILLING_URL is set — with BOT_BILLING_TOKEN when the
+ * operator minted one, otherwise with a Network service token Bot mints for audience openvibe.billing
+ * (capability billing.usage.record) from its OAuth client, the same helper and cache the OpenRe client uses. If
+ * the URL is unset (or Bot has no token and no Network client to mint one), or Billing is down, has no grant for
+ * Bot yet or answers 401/403/404/429/5xx, readings wait and are retried with backoff, across restarts: never
+ * dropped. Only Billing refusing the reading itself (400, 409, 413, 422) marks a row rejected; it is kept with
+ * its error and never sent again, so nothing is billed twice.
*/
*/
-const { validate } = require('openvibe-contracts');
+const { validate, serviceAuth } = require('openvibe-contracts');
const { createPgOutbox } = require('openvibe-sdk/events');
const { createPgOutbox } = require('openvibe-sdk/events');
+const { BotError } = require('../util');
const TABLE = 'run_usage_outbox'; // migrations/0004_run_jobs.sql
const TABLE = 'run_usage_outbox'; // migrations/0004_run_jobs.sql
const SCHEMA = 'platform.usage-sample@1';
const SCHEMA = 'platform.usage-sample@1';
@@ -28,6 +32,11 @@ const UNIT = 's';
@@ -28,6 +32,11 @@ const UNIT = 's';
const SOURCE = 'openvibe-node.worker';
const SOURCE = 'openvibe-node.worker';
const REFUSED = new Set([400, 409, 413, 422]);
const REFUSED = new Set([400, 409, 413, 422]);
+// Network's grant to the `bot` client on audience openvibe.billing (Network server/identity/principals.js
+// DEFAULT_GRANTS): billing.usage.record is the one capability a reading is posted with.
+const BILLING_AUDIENCE = 'openvibe.billing';
+const BILLING_SCOPE = 'billing.usage.record';
+
/** The reading for second `n` of a job; `quantity` is 1 except for the partial last second. */
/** The reading for second `n` of a job; `quantity` is 1 except for the partial last second. */
function secondReading({ jobId, n, quantity = 1, startedMs, nodeId, project = null, subject = null, provider = null }) {
function secondReading({ jobId, n, quantity = 1, startedMs, nodeId, project = null, subject = null, provider = null }) {
if (!Number.isInteger(n) || n < 0) throw new TypeError(`second must be an integer >= 0, not ${n}`);
if (!Number.isInteger(n) || n < 0) throw new TypeError(`second must be an integer >= 0, not ${n}`);
@@ -66,15 +75,18 @@ async function enqueue(t, reading, at) {
@@ -66,15 +75,18 @@ async function enqueue(t, reading, at) {
}
}
// createPgOutbox's `events` seam pointed at Billing. isPermanent() reads err.status: 422 rejects the row, 503 retries.
// createPgOutbox's `events` seam pointed at Billing. isPermanent() reads err.status: 422 rejects the row, 503 retries.
-function billingSink({ url, token, fetchImpl, timeoutMs }) {
+function billingSink({ url, authorization, onUnauthorized, fetchImpl, timeoutMs }) {
return {
return {
prepare: (reading) => reading,
prepare: (reading) => reading,
async publish(reading) {
async publish(reading) {
const res = await fetchImpl(`${url}/api/v1/usage`, {
const res = await fetchImpl(`${url}/api/v1/usage`, {
method: 'POST',
method: 'POST',
- headers: { Authorization: `Bearer ${token}`, 'Content-Type': 'application/json', Accept: 'application/json' },
+ headers: { Authorization: await authorization(), 'Content-Type': 'application/json', Accept: 'application/json' },
body: JSON.stringify(reading), signal: AbortSignal.timeout(timeoutMs),
body: JSON.stringify(reading), signal: AbortSignal.timeout(timeoutMs),
});
});
+ // A minted token Billing no longer accepts (rotated signing key, clock): drop it so the next sweep
+ // mints a fresh one; the reading stays queued, never rejected for it.
+ if (res.status === 401 && onUnauthorized) onUnauthorized();
// 201 written, 200 an identical reading replayed: either way Billing holds it.
// 201 written, 200 an identical reading replayed: either way Billing holds it.
if (!res.ok) { const err = new Error(`billing.usage.record answered ${res.status}`); err.status = REFUSED.has(res.status) ? 422 : 503; throw err; }
if (!res.ok) { const err = new Error(`billing.usage.record answered ${res.status}`); err.status = REFUSED.has(res.status) ? 422 : 503; throw err; }
return { event_id: reading.id, seq: null };
return { event_id: reading.id, seq: null };
@@ -83,20 +95,39 @@ function billingSink({ url, token, fetchImpl, timeoutMs }) {
@@ -83,20 +95,39 @@ function billingSink({ url, token, fetchImpl, timeoutMs }) {
}
}
/**
/**
- * The relay from run_usage_outbox to Billing. enabled only with config.billing.url and .token; otherwise start(),
- * kick() and flush() do nothing and readings accumulate until a process with both set sweeps them.
+ * The relay from run_usage_outbox to Billing. enabled with config.billing.url and either an operator
+ * BOT_BILLING_TOKEN or the Network client credentials to mint one (exactly as the OpenRe client does); with
+ * neither, or with the URL unset, start(), kick() and flush() do nothing and readings accumulate until a
+ * process with the URL and a token sweeps them.
*/
*/
function createUsageRelay({ db, config, fetchImpl = globalThis.fetch, now = () => Date.now(), log = console }) {
function createUsageRelay({ db, config, fetchImpl = globalThis.fetch, now = () => Date.now(), log = console }) {
const { url, token, intervalMs, timeoutMs } = config.billing;
const { url, token, intervalMs, timeoutMs } = config.billing;
const count = (where) => db.value(`SELECT count(*) FROM ${TABLE} WHERE ${where}`).then(Number);
const count = (where) => db.value(`SELECT count(*) FROM ${TABLE} WHERE ${where}`).then(Number);
const counts = { pending: () => count('sent_at IS NULL AND rejected_at IS NULL'), rejected: () => count('rejected_at IS NOT NULL') };
const counts = { pending: () => count('sent_at IS NULL AND rejected_at IS NULL'), rejected: () => count('rejected_at IS NOT NULL') };
- if (!url || !token) {
+ // An operator-minted BOT_BILLING_TOKEN wins; otherwise Bot mints its own from its Network client (the same
+ // openvibe-contracts token client, and its cache, that server/openre/client.js uses for openvibe.openre).
+ const { clientId, clientSecret } = config.oauth || {};
+ const internalUrl = (config.network || {}).internalUrl;
+ const tokens = !token && clientSecret && internalUrl
+ ? serviceAuth.createTokenClient({
+ tokenUrl: `${internalUrl}/oauth/token`, clientId, clientSecret, audience: BILLING_AUDIENCE, scope: BILLING_SCOPE, fetchImpl,
+ })
+ : null;
+ if (!url || (!token && !tokens)) {
const idle = async () => ({ sent: 0, failed: 0, rejected: 0 });
const idle = async () => ({ sent: 0, failed: 0, rejected: 0 });
return { enabled: false, start() {}, stop: async () => {}, kick() {}, flush: idle, ...counts };
return { enabled: false, start() {}, stop: async () => {}, kick() {}, flush: idle, ...counts };
}
}
+ /** The Authorization header of this post: the operator's token, or a freshly cached minted one. */
+ const authorization = tokens
+ ? async () => {
+ try { return (await tokens.authHeaders()).Authorization; }
+ catch (e) { throw new BotError(503, 'bot.billing_unavailable', 'Bot could not mint its Billing token from Network'); }
+ }
+ : async () => `Bearer ${token}`;
let lastError = null;
let lastError = null;
const relay = createPgOutbox(db, {
const relay = createPgOutbox(db, {
- events: billingSink({ url, token, fetchImpl, timeoutMs }), table: TABLE, batchSize: 1, intervalMs, now,
+ events: billingSink({ url, authorization, onUnauthorized: tokens ? () => tokens.invalidate() : null, fetchImpl, timeoutMs }),
+ table: TABLE, batchSize: 1, intervalMs, now,
onError: (err) => { const m = err && err.message; if (m !== lastError) log.warn(`[Bot] usage reading not delivered: ${m}`); lastError = m; },
onError: (err) => { const m = err && err.message; if (m !== lastError) log.warn(`[Bot] usage reading not delivered: ${m}`); lastError = m; },
});
});
return { enabled: true, start: relay.start, stop: relay.stop, kick: relay.kick, flush: relay.flush, ...counts };
return { enabled: true, start: relay.start, stop: relay.stop, kick: relay.kick, flush: relay.flush, ...counts };
new file mode 100644
new file mode 100644
@@ -0,0 +1,181 @@
@@ -0,0 +1,181 @@
+'use strict';
+// The usage relay authenticates to OpenVibe.Billing (plan T14 L1) the same way Bot talks to OpenRe: with an
+// operator BOT_BILLING_TOKEN when one is set, otherwise a Network service token it mints for audience
+// openvibe.billing (capability billing.usage.record) from its OAuth client, cached by the shared token client
+// until 60 s before expiry. Here the relay runs against a fake Network token endpoint and a fake Billing: with
+// only BOT_BILLING_URL set it mints one token for the sweep, posts each queued reading once with it, re-mints
+// after expiry; a 401 from Billing drops the cached token and a 401/5xx keeps every reading queued (never
+// dropped, never rejected); an operator token overrides minting; and with the URL unset the relay makes no call
+// at all.
+const assert = require('assert');
+const { check, done } = require('./helpers/app');
+const { testDb } = require('./helpers/db');
+const { secondReading, enqueue, createUsageRelay, TABLE } = require('../server/jobs/metering');
+
+// Network's grant to the `bot` client on audience openvibe.billing (OpenVibe.Network DEFAULT_GRANTS).
+const MINTED_AUDIENCE = 'openvibe.billing';
+const MINTED_SCOPE = 'billing.usage.record';
+const BILLING_URL = 'http://billing.test';
+const NETWORK_URL = 'http://network.test';
+const STARTED_MS = 1790935200000;
+
+const quiet = { warn() {} };
+const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
+const config = (billing = {}) => ({
+ oauth: { clientId: 'bot', clientSecret: 'shh' },
+ network: { internalUrl: NETWORK_URL },
+ billing: { url: BILLING_URL, token: '', intervalMs: 100, timeoutMs: 2000, ...billing },
+});
+
+/**
+ * A fake Network token endpoint (POST /oauth/token → mint_1, mint_2, …) and a fake Billing (POST
+ * /api/v1/usage), behind one fetch. tokenCalls records every client-credentials request; posts records every
+ * usage post ({ auth, status, body }); billingStatus() decides each post's status.
+ */
+function fakeServices({ expiresIn = 300, billingStatus = () => 201 } = {}) {
+ const tokenCalls = [];
+ const posts = [];
+ let minted = 0;
+ const fetchImpl = async (url, init) => {
+ if (url === `${NETWORK_URL}/oauth/token`) {
+ const body = Object.fromEntries(new URLSearchParams(init.body));
+ tokenCalls.push({ client_id: body.client_id, client_secret: body.client_secret, audience: body.audience, scope: body.scope, grant_type: body.grant_type });
+ if (body.client_secret !== 'shh') return new Response(JSON.stringify({ error: 'invalid_client' }), { status: 401 });
+ return new Response(JSON.stringify({ access_token: `mint_${++minted}`, token_type: 'Bearer', expires_in: expiresIn }), { status: 200, headers: { 'Content-Type': 'application/json' } });
+ }
+ if (url === `${BILLING_URL}/api/v1/usage`) {
+ const status = billingStatus();
+ posts.push({ auth: init.headers.Authorization, status, body: JSON.parse(init.body) });
+ return new Response('{}', { status });
+ }
+ throw new Error(`unexpected fetch ${url}`);
+ };
+ return { fetchImpl, tokenCalls, posts };
+}
+
+(async () => {
+ const store = await testDb();
+ const db = store.db;
+ // One relay sweep at a time: readings queued inside a transaction, exactly as dispatch/job_exit do.
+ const queue = (prefix, seconds) => db.tx(async (t) => {
+ for (const n of seconds) {
+ await enqueue(t, secondReading({ jobId: `job_${prefix}`, n, startedMs: STARTED_MS, nodeId: 'dev_test' }), Date.now());
+ }
+ });
+ const countLike = (prefix, where) => db.value(`SELECT count(*) FROM ${TABLE} WHERE event_id LIKE $1 AND ${where}`, [`run:job_${prefix}:%`]).then(Number);
+ const pending = (prefix) => countLike(prefix, 'sent_at IS NULL AND rejected_at IS NULL');
+ const rejected = (prefix) => countLike(prefix, 'rejected_at IS NOT NULL');
+ const sent = (prefix) => countLike(prefix, 'sent_at IS NOT NULL');
+
+ await check('with only BOT_BILLING_URL set the relay mints one token and posts each reading once with it', async () => {
+ const svc = fakeServices({ expiresIn: 61 });
+ const clock = { t: Date.now() };
+ const relay = createUsageRelay({ db, config: config(), fetchImpl: svc.fetchImpl, now: () => clock.t, log: quiet });
+ assert.strictEqual(relay.enabled, true);
+
+ await queue('mint', [0, 1, 2]);
+ const before = svc.tokenCalls.length;
+ await relay.flush();
+ assert.deepStrictEqual(svc.tokenCalls.slice(before),
+ [{ client_id: 'bot', client_secret: 'shh', audience: MINTED_AUDIENCE, scope: MINTED_SCOPE, grant_type: 'client_credentials' }]);
+ assert.strictEqual(svc.posts.length, 3);
+ assert.deepStrictEqual(svc.posts.map((p) => p.auth), ['Bearer mint_1', 'Bearer mint_1', 'Bearer mint_1']);
+ assert.deepStrictEqual(svc.posts.map((p) => p.body.idempotency_key), ['run:job_mint:0', 'run:job_mint:1', 'run:job_mint:2']);
+ assert.strictEqual(await pending('mint'), 0);
+ assert.strictEqual(await sent('mint'), 3);
+ assert.strictEqual(await rejected('mint'), 0);
+
+ // A second read is served by the same cached token, not a second mint.
+ await queue('mint2', [0]);
+ await relay.flush();
+ assert.strictEqual(svc.tokenCalls.length, before + 1, 'the cached token was reused');
+ assert.strictEqual(svc.posts.at(-1).auth, 'Bearer mint_1');
+
+ // Past its expiry (expires_in 61 ⇒ cached only ~1 s) the relay mints again.
+ await sleep(1100);
+ await queue('mint3', [0]);
+ await relay.flush();
+ assert.strictEqual(svc.tokenCalls.length, before + 2, 'the expired token was replaced');
+ assert.strictEqual(svc.posts.at(-1).auth, 'Bearer mint_2');
+ assert.strictEqual(await pending('mint3'), 0);
+ await relay.stop();
+ });
+
+ await check('a 401 or 5xx from Billing keeps every reading queued, and the 401 drops the minted token', async () => {
+ const svc = fakeServices();
+ const clock = { t: Date.now() };
+ let mode = 503;
+ const relay = createUsageRelay({ db, config: config(), fetchImpl: fakeServices200(svc, () => mode), now: () => clock.t, log: quiet });
+ assert.strictEqual(relay.enabled, true);
+
+ await queue('down', [0, 1]);
+ await relay.flush(); // 5xx
+ assert.strictEqual(await pending('down'), 2, 'a 5xx leaves both readings queued');
+ assert.strictEqual(await rejected('down'), 0);
+ clock.t += 2000;
+ await relay.flush(); // 5xx again, after the retry backoff
+ assert.strictEqual(await pending('down'), 2);
+ assert.strictEqual(await rejected('down'), 0);
+
+ const mintedBefore = svc.tokenCalls.length;
+ const auth401 = svc.posts.at(-1).auth;
+ mode = 401;
+ clock.t += 10000;
+ await relay.flush();
+ assert.strictEqual(await pending('down'), 2, 'a 401 leaves both readings queued');
+ assert.strictEqual(await rejected('down'), 0);
+
+ mode = 201;
+ clock.t += 60000;
+ await relay.flush();
+ assert.strictEqual(await pending('down'), 0);
+ assert.strictEqual(await rejected('down'), 0);
+ assert.strictEqual(svc.tokenCalls.length, mintedBefore + 1, 'the 401 dropped the cached token, so one fresh mint served the retry');
+ assert.notStrictEqual(svc.posts.at(-1).auth, auth401, 'the retry carried a freshly minted token');
+ assert.deepStrictEqual(svc.posts.filter((p) => p.status === 201).map((p) => p.auth), ['Bearer mint_2', 'Bearer mint_2']);
+ await relay.stop();
+ });
+
+ await check('BOT_BILLING_TOKEN overrides minting: it is sent verbatim and Network is never asked', async () => {
+ const svc = fakeServices();
+ const relay = createUsageRelay({ db, config: config({ token: 'operator-minted' }), fetchImpl: svc.fetchImpl, now: () => Date.now(), log: quiet });
+ assert.strictEqual(relay.enabled, true);
+ await queue('static', [0]);
+ await relay.flush();
+ assert.strictEqual(svc.tokenCalls.length, 0, 'BOT_BILLING_TOKEN was set: nothing was minted');
+ assert.deepStrictEqual(svc.posts.map((p) => p.auth), ['Bearer operator-minted']);
+ assert.strictEqual(await pending('static'), 0);
+ await relay.stop();
+ });
+
+ await check('with BOT_BILLING_URL unset the relay is off and makes no call; so is a URL with no way to authenticate', async () => {
+ const svc = fakeServices();
+ const noCall = async (url) => { throw new Error(`the relay called ${url} while off`); };
+ const off = createUsageRelay({ db, config: config({ url: '' }), fetchImpl: noCall, now: () => Date.now(), log: quiet });
+ assert.strictEqual(off.enabled, false);
+ assert.deepStrictEqual(await off.flush(), { sent: 0, failed: 0, rejected: 0 });
+ off.start(); off.kick(); await off.stop();
+ await queue('off', [0]);
+ assert.strictEqual(await pending('off'), 1, 'readings wait while the URL is unset');
+ // A URL but neither an operator token nor the Network client credentials: still off, still nothing sent.
+ const noCreds = createUsageRelay({ db, config: { oauth: {}, network: {}, billing: { url: BILLING_URL, token: '', intervalMs: 100, timeoutMs: 2000 } }, fetchImpl: noCall, now: () => Date.now(), log: quiet });
+ assert.strictEqual(noCreds.enabled, false);
+ assert.deepStrictEqual(await noCreds.flush(), { sent: 0, failed: 0, rejected: 0 });
+ assert.ok(svc.tokenCalls.length === 0 && svc.posts.length === 0);
+ });
+
+ await store.close();
+ done();
+})().catch((e) => { console.error(e); process.exit(1); });
+
+/** The fake Network endpoint with Billing's status coming from `mode()` (so a test can flip it mid-run). */
+function fakeServices200(svc, mode) {
+ return async (url, init) => {
+ if (url === `${BILLING_URL}/api/v1/usage`) {
+ const status = mode();
+ svc.posts.push({ auth: init.headers.Authorization, status, body: JSON.parse(init.body) });
+ return new Response('{}', { status });
+ }
+ return svc.fetchImpl(url, init);
+ };
+}