openvibers/openvibe.tools Commits
Post one Billing usage reading per ended tool job
Changes
@@@ -28,9 +28,10 @@ offline against the Network's public key (JWKS).- OpenVibe.Network (SSO as OAuth client `tools`, JWKS, client-credentials tokens, the domains registry),OpenVibe.Media (job results), OpenVibe.Events (job and usage events, revocation cutoffs),-- OpenVibe.Search (tool documents), OpenVibe.Community (the paste hand-over)++ OpenVibe.Search (tool documents), OpenVibe.Community (the paste hand-over), OpenVibe.Billing (job usage++ readings)- host programs some tools need (ffmpeg, yt-dlp, qpdf, poppler, libheif; [Host packages](#host-packages-some-tools-need))--- in each app: `openvibe-contracts` v0.63.0, `openvibe-sdk` v0.12.0 and `openvibe-shared` v1.23.0, pinned++- in each app: `openvibe-contracts` v0.92.0 (img, audio, docs, gateway), `openvibe-sdk` v0.12.0 and `openvibe-shared` v1.23.0, pinnedby release tarball, and `apps/_shared` (openvibe-tools-shared) by relative path## Capabilities@@@ -46,6 +47,7 @@ Called elsewhere, as the service principal `tools` (the OAuth client `tools`):| OpenVibe.Media | `media.object.upload`, `media.object.read` | job results || OpenVibe.Events | `events.event.publish`, `events.subscription.manage` | job and usage events; the `network.user.token_valid_after` subscription || OpenVibe.Search | `search.document.write` | tool documents for the network index |++| OpenVibe.Billing | `billing.usage.record` (audience `openvibe.billing`) | one usage reading per ended job |## Layout@@@ -300,12 +302,27 @@ reload reattaches.`env: sandbox`. No owner, session, address, input, file name or output leaves. OpenVibe.Network adds the rollupsup for the project's usage page on openvibe.codes. Off without `EVENTS_URL`, like the job events;`GET /api/health` shows `jobs.usage` (pending, invalid).++- **Usage readings to OpenVibe.Billing** (`apps/_shared/billing.js`, `apps/_shared/jobs/usage.js`, plan T5 step 7):++ every job that succeeds or fails (anyone's, not only a project's; a cancelled one is not billed) gets one++ `platform.usage-sample@1` reading — id `tools-job-<id>`, `idempotency_key` `tools:job:<id>`, service and source++ `openvibe.tools`, operation `tools.tool.run` (a tool run) or `tools.job.create`, resource the tool or job type,++ quantity 1, unit `jobs`, `at` the job's end, plus `project` and `trace_id` when the job has them and `subject`++ (`user:usr_…`) only for a signed-in person's job, never an app or a browser session. It is stored in++ `tool_job_billing_readings` in the transaction that records the end (no network call there); the pruner's timer++ posts the unsent ones (500 at most a tick) to `POST /api/v1/usage` and marks each sent. A refused post is logged,++ never thrown: the reading keeps its error and attempt count and is posted again on the next tick (Billing answers++ a replay with 200), and an outage, a 401/403 or a 429 stops that tick's batch. The hourly `tools.usage.recorded`++ rollups above are unchanged and stay subject-less. Off without `OV_BILLING_URL` (nothing is stored). The token is++ Network `client_credentials` for the `tools` client on audience `OV_BILLING_AUDIENCE` (default++ `openvibe.billing`); Network must grant that client `billing.usage.record` on it — until the grant exists every++ post is refused (401) and only logged, and the readings wait in the table. `GET /api/health` shows++ `jobs.billing` (pending, posted, refused, invalid, last_error) when it is on.Environment (all in `/etc/openvibe/tools.env`): `TOOLS_WORKERS`, `TOOLS_WORKERS_<APP>`, `TOOLS_WORKER_MEMORY_MB`,`TOOLS_WORKER_MEMORY_MB_<APP>`, `TOOLS_WORKER_QUEUE`, `TOOLS_WORKER_IDLE_MS`, `TOOLS_JOBS_CONCURRENCY`, `TOOLS_JOBS_CONCURRENCY_<APP>`,`TOOLS_JOBS_MAX_ACTIVE`, `TOOLS_JOB_RESULTS`, `TOOLS_MEDIA_NAMESPACE`, `OV_MEDIA_INTERNAL_URL`, `OV_MEDIA_URL`,`OV_NETWORK_INTERNAL_URL`, `OV_OAUTH_CLIENT_ID`, `OV_OAUTH_CLIENT_SECRET`, `EVENTS_URL`, `EVENTS_PUBLISH`,--`EVENTS_RELAY_INTERVAL_MS`.++`EVENTS_RELAY_INTERVAL_MS`, `OV_BILLING_URL`, `OV_BILLING_AUDIENCE`.Not done yet: quotas for external developer apps are the guard's service and sandbox tiers, not a Codes-issued quota.nnew file mode 100644@@@ -0,0 +1,59 @@++'use strict';++// ═══════════════════════════════════════════════════════════════++// Usage readings → OpenVibe.Billing (plan T5 step 7): post(reading) sends one platform.usage-sample@1++// to POST /api/v1/usage with Tools' service token (Network client_credentials, OV_OAUTH_CLIENT_ID/SECRET,++// audience OV_BILLING_AUDIENCE, default openvibe.billing; the Tools client needs the grant++// billing.usage.record). Billing dedupes on the reading's idempotency_key (a replay answers 200), so a++// retry is safe.++//++// Inert unless OV_BILLING_URL is set. post() never throws: a refused post (or no token) is logged and++// answered { ok: false, status, error }, so the caller keeps the reading for its next flush. Never called++// inside a job's transaction (./jobs/usage.js stores the reading there and posts it from the flush).++// No dependencies: required by relative path.++// ═══════════════════════════════════════════════════════════════++++const PATH = '/api/v1/usage';++const TIMEOUT_MS = 10000;++++function createBillingClient({ env = process.env, fetchImpl = globalThis.fetch, now = () => Date.now(), log = console } = {}) {++ const base = String(env.OV_BILLING_URL || '').trim().replace(/\/+$/, '');++ if (!base) return { enabled: false, post: async () => ({ ok: false, skipped: true }) };++ const network = String(env.OV_NETWORK_INTERNAL_URL || 'http://127.0.0.1:4000').replace(/\/+$/, '');++ const audience = env.OV_BILLING_AUDIENCE || 'openvibe.billing';++ let token = null, tokenExp = 0;++++ async function bearer() {++ if (token && now() < tokenExp) return token;++ const body = new URLSearchParams({ grant_type: 'client_credentials', client_id: env.OV_OAUTH_CLIENT_ID || 'tools', client_secret: env.OV_OAUTH_CLIENT_SECRET || '', audience });++ const r = await fetchImpl(`${network}/oauth/token`, { method: 'POST', headers: { 'Content-Type': 'application/x-www-form-urlencoded', Accept: 'application/json' }, body, signal: AbortSignal.timeout(TIMEOUT_MS) });++ const j = await r.json().catch(() => ({}));++ if (!r.ok || !j.access_token) throw Object.assign(new Error(`token: ${r.status} ${j.error || ''}`.trim()), { status: r.status });++ token = j.access_token; tokenExp = now() + Math.max(30, (Number(j.expires_in) || 300) - 60) * 1000;++ return token;++ }++++ /** One reading → Billing. { ok: true, status } when stored (201) or replayed (200); else logged, { ok: false, status, error }. */++ async function post(reading) {++ try {++ const r = await fetchImpl(`${base}${PATH}`, {++ method: 'POST',++ headers: { 'Content-Type': 'application/json', Accept: 'application/json', Authorization: `Bearer ${await bearer()}` },++ body: JSON.stringify(reading), signal: AbortSignal.timeout(TIMEOUT_MS),++ });++ if (r.ok) return { ok: true, status: r.status };++ if (r.status === 401) token = null;++ const text = (await r.text().catch(() => '')).replace(/\s+/g, ' ').slice(0, 300);++ const error = `${r.status} ${text}`.trim();++ log.error(`[Billing] reading ${reading && reading.idempotency_key} refused: ${error}`);++ return { ok: false, status: r.status, error };++ } catch (err) {++ const error = String(err && err.message || err).slice(0, 300);++ log.error(`[Billing] reading ${reading && reading.idempotency_key} not sent: ${error}`);++ return { ok: false, status: err && err.status || 0, error };++ }++ }++++ return { enabled: true, post };++}++++module.exports = { createBillingClient };@@@ -31,6 +31,10 @@// transaction that records its end (./usage.js), and each closed hour goes out as tools.usage.recorded// from the pruner's timer.//++// Billing (plan T5 step 7): with OV_BILLING_URL set (../billing.js), every job that succeeds or fails++// also stores its usage reading in that transaction, and the pruner's timer posts the unsent ones to++// OpenVibe.Billing (./usage.js flushReadings). Without it, nothing is stored or posted.++//// No dependencies of its own: the app passes the shared openvibe-sdk/db handle and openvibe-contracts.// ═══════════════════════════════════════════════════════════════@@@ -41,6 +45,7 @@ const crypto = require('crypto');const { createStore, TERMINAL } = require('./store');const { createJobEvents } = require('./events');const { createJobUsage } = require('./usage');++const { createBillingClient } = require('../billing');const HOUR = 60 * 60 * 1000;@@@ -98,6 +103,7 @@ const scrub = (msg) => String(msg || '').replace(/(?:\/[\w.-]+){2,}/g, '[file]')* @param {number} [o.diskBudgetBytes=0] bytes under <dataDir>/jobs before busy() says so; 0 = no bound* @param {object} [o.media] result store from ./media (null = keep results on local disk)* @param {object} [o.outbox] openvibe-sdk outbox on `db` (./events outboxFromEnv); null = no platform events++ * @param {object} [o.billing] ../billing createBillingClient (default: one from process.env; inert without OV_BILLING_URL)* @param {number} [o.pruneIntervalMs=300000]* @param {number} [o.progressThrottleMs=250]*/@@@ -117,8 +123,9 @@ function createJobSystem(o) {const diskBudget = o.diskBudgetBytes > 0 ? o.diskBudgetBytes : 0;const disk = { bytes: 0, at: 0, scanning: null };const announce = createJobEvents({ contracts, service: o.service, outbox: o.outbox || null, referenceCount: (id) => store.referenceCount(id), log });-- const usage = createJobUsage({ db: o.db, app: o.service, contracts, outbox: o.outbox || null, log });-- /** Inside the transaction `t` that recorded the end: tools.job.succeeded|failed and the project's usage. */++ const billing = o.billing || createBillingClient({ log });++ const usage = createJobUsage({ db: o.db, app: o.service, contracts, outbox: o.outbox || null, billing, log });++ /** Inside the transaction `t` that recorded the end: tools.job.succeeded|failed, the project's usage and the Billing reading. */async function ended(t, id) {const row = await store.get(id);await announce.finished(t, row);@@@ -696,9 +703,11 @@ function createJobSystem(o) {if (r.requeued || r.failed) log.log(`[Jobs] ${o.service}: ${r.requeued} job(s) re-queued, ${r.failed} failed after restart`);await prune().catch(() => {});await flushUsage();++ flushReadings(); // not awaited: posting to Billing never holds up the startpruneTimer = setInterval(() => {prune().catch(err => log.error('[Jobs] prune:', err.message));flushUsage();++ flushReadings();}, o.pruneIntervalMs || 5 * 60 * 1000);if (pruneTimer.unref) pruneTimer.unref();await kick();@@@ -713,6 +722,10 @@ function createJobSystem(o) {async function flushUsage(at) {try { return await usage.flush(at); } catch (err) { log.error('[Jobs] usage flush:', err.message); return { queued: 0, invalid: 0 }; }}++ /** Unsent Billing readings → OpenVibe.Billing (./usage.js); a failure is logged and retried next tick. */++ async function flushReadings(at) {++ try { return await usage.flushReadings(at); } catch (err) { log.error('[Jobs] billing flush:', err.message); return { posted: 0, refused: 0 }; }++ }function stop() {stopped = true;@@@ -753,8 +766,8 @@ function createJobSystem(o) {types: () => [...types.keys()],media,// running (after the spread) is the store's count of rows in 'running'; executing is this process's.-- stats: async () => ({ running: active.size, concurrency, ...(await store.counts()), executing: active.size, results: media ? 'media' : 'local', events: await announce.status(), usage: usage.enabled ? await usage.status() : { enabled: false } }),-- usage, flushUsage,++ stats: async () => ({ running: active.size, concurrency, ...(await store.counts()), executing: active.size, results: media ? 'media' : 'local', events: await announce.status(), usage: usage.enabled ? await usage.status() : { enabled: false }, ...(usage.billed ? { billing: await usage.readings() } : {}) }),++ usage, flushUsage, flushReadings,outbox: o.outbox || null,/** True between start() and stop(): the worker picks up queued jobs. */isRunning: () => started && !stopped,@@@ -19,8 +19,16 @@// outbox relay publishes it. A job that ends in an hour already sent (a clock step back) reopens it as// revision + 1. Only counts leave: no owner, session, address, input, file name or output.//--// On PostgreSQL (plan T8): openvibe-sdk/db, ambient transactions, createPgOutbox. Inert without an--// outbox (EVENTS_URL unset, like ./events.js): nothing is counted.++// Billing readings (plan T5 step 7): with a Billing client (../billing.js, OV_BILLING_URL set), the same++// transaction also stores the job's platform.usage-sample@1 reading (readingOf: anyone's job that++// succeeded or failed, not only a project's; quantity 1, unit jobs, idempotency_key tools:job:<id>) in++// tool_job_billing_readings, never posting it there. flushReadings() posts the unsent ones (500 at most++// a call) and marks each sent; a refused post counts an attempt with its error and stays for the next++// call. The reading names the person (subject user:usr_…) for a person's job, never an app or a browser++// session; no input, file name or output. Billing keeps it; the hourly rollups above stay subject-less.++//++// On PostgreSQL (plan T8): openvibe-sdk/db, ambient transactions, createPgOutbox. The rollups are inert++// without an outbox (EVENTS_URL unset, like ./events.js), the readings without a Billing client.// ═══════════════════════════════════════════════════════════════const HOUR_MS = 60 * 60 * 1000;@@@ -32,6 +40,8 @@ const PROJECT_RE = /^prj_[0-9A-HJKMNP-TV-Z]{26}$/;const CODE_RE = /^[a-z][a-z0-9_]*(\.[a-z0-9_]+)+$/;const TRACE_RE = /^[0-9a-f]{32}$/;const DIMENSION_RE = /^[a-z0-9][a-z0-9_.:-]{0,79}$/;++const PERSON_RE = /^user:usr_[0-9A-HJKMNP-TV-Z]{26}$/;++const READINGS_PER_FLUSH = 500;const hourOf = (ms) => Math.floor(ms / HOUR_MS) * HOUR_MS;const parse = (s, d) => { try { return JSON.parse(s); } catch { return d; } };@@@ -49,17 +59,93 @@ function keyOf(row) {};}++/** The Billing reading (platform.usage-sample@1) of an ended tool_jobs row, or null when it is not billed. */++function readingOf(row) {++ if (!row || !row.id || (row.state !== 'succeeded' && row.state !== 'failed')) return null;++ const resource = String(row.tool || row.type || '');++ const r = {++ id: `tools-job-${row.id}`,++ idempotency_key: `tools:job:${row.id}`,++ service: 'openvibe.tools',++ operation: row.tool ? 'tools.tool.run' : 'tools.job.create',++ resource: DIMENSION_RE.test(resource) ? resource : 'other',++ quantity: 1,++ unit: 'jobs',++ at: new Date(Number(row.finished_at) || Date.now()).toISOString(),++ source: 'openvibe.tools',++ };++ if (PROJECT_RE.test(String(row.project_id || ''))) r.project = row.project_id;++ if (PERSON_RE.test(String(row.owner || ''))) r.subject = row.owner; // a person's job; never an app or a session++ if (TRACE_RE.test(String(row.trace_id || ''))) r.trace_id = row.trace_id;++ return r;++}++/*** @param {object} o* @param {object} o.db openvibe-sdk/db handle (the one tools database), where the outbox lives too* @param {string} o.app the owning app ('img' | …)* @param {object} o.contracts require('openvibe-contracts') (0.63.0+ knows tools.usage.recorded@1)-- * @param {object|null} o.outbox openvibe-sdk createPgOutbox on `db` (./events outboxFromEnv); null = inert++ * @param {object|null} o.outbox openvibe-sdk createPgOutbox on `db` (./events outboxFromEnv); null = no rollups++ * @param {object|null} [o.billing] ../billing createBillingClient; null or not enabled = no readings* @param {() => number} [o.now]* @param {object} [o.log]*/--function createJobUsage({ db, app, contracts, outbox = null, now = () => Date.now(), log = console }) {-- if (!outbox) return { enabled: false, finished: async () => {}, flush: async () => ({ queued: 0, invalid: 0 }), pending: async () => 0, status: async () => ({ enabled: false }) };++function createJobUsage({ db, app, contracts, outbox = null, billing = null, now = () => Date.now(), log = console }) {++ const billed = !!(billing && billing.enabled);++ const r = {++ add: db.prepare('INSERT INTO tool_job_billing_readings (job_id, app, reading, created_at) VALUES (@job_id, @app, @reading, @at) ON CONFLICT (job_id) DO NOTHING'),++ unsent: db.prepare('SELECT job_id, reading FROM tool_job_billing_readings WHERE app = @app AND sent_at IS NULL ORDER BY attempts, created_at LIMIT @limit'),++ sent: db.prepare('UPDATE tool_job_billing_readings SET sent_at = @at, last_error = NULL WHERE job_id = @job_id'),++ failed: db.prepare('UPDATE tool_job_billing_readings SET attempts = attempts + 1, last_error = @error WHERE job_id = @job_id'),++ prune: db.prepare('DELETE FROM tool_job_billing_readings WHERE app = @app AND sent_at IS NOT NULL AND sent_at < @at'),++ pending: db.prepare('SELECT COUNT(*) AS n FROM tool_job_billing_readings WHERE app = @app AND sent_at IS NULL'),++ };++ const readings = { posted: 0, refused: 0, invalid: 0, lastError: null, running: null };++++ /** INSIDE the job-end transaction: store the reading (once per job); nothing leaves here. */++ async function record(row) {++ const reading = billed ? readingOf(row) : null;++ if (!reading) return;++ let v;++ try { v = contracts.validate('platform.usage-sample@1', reading); } catch (err) { v = { valid: false, errors: [{ path: '', message: err.message }] }; }++ if (!v.valid) {++ readings.invalid++;++ readings.lastError = `${reading.idempotency_key}: ${v.errors.map(e => `${e.path} ${e.message}`).join('; ')}`;++ log.error(`[Jobs] billing reading not stored (does not match the contract): ${readings.lastError}`);++ return;++ }++ await r.add.run({ job_id: row.id, app, reading: JSON.stringify(reading), at: now() });++ }++++ /** Post the unsent readings to Billing, oldest and least-tried first; a refused one stays for the next call. */++ function flushReadings(at = now()) {++ if (!billed) return Promise.resolve({ posted: 0, refused: 0 });++ if (!readings.running) readings.running = (async () => {++ let posted = 0, refused = 0;++ for (const row of await r.unsent.all({ app, limit: READINGS_PER_FLUSH })) {++ const reading = typeof row.reading === 'string' ? parse(row.reading, null) : row.reading;++ const res = await billing.post(reading);++ if (res.ok) { await r.sent.run({ job_id: row.job_id, at: now() }); posted++; continue; }++ refused++;++ readings.lastError = res.error || String(res.status);++ await r.failed.run({ job_id: row.job_id, error: readings.lastError.slice(0, 500) });++ // Billing or the token is down (or the grant is missing): stop here, the rest waits for the next call.++ if (!res.status || res.status === 401 || res.status === 403 || res.status === 429 || res.status >= 500) break;++ }++ await r.prune.run({ app, at: at - KEEP_SENT_MS });++ readings.posted += posted; readings.refused += refused;++ return { posted, refused };++ })().finally(() => { readings.running = null; });++ return readings.running;++ }++ const readingsStatus = async () => ({ pending: Number((await r.pending.get({ app })).n), posted: readings.posted, refused: readings.refused, invalid: readings.invalid, last_error: readings.lastError });++++ if (!outbox) {++ return {++ enabled: false, billed, finished: record, flush: async () => ({ queued: 0, invalid: 0 }), flushReadings,++ pending: async () => 0, status: async () => ({ enabled: false }), readings: readingsStatus,++ };++ }const q = {add: db.prepare(`INSERT INTO tool_job_usage (app, project_id, env, capability, dimension, window_start, quantity, errors)VALUES (@app, @project_id, @env, @capability, @dimension, @window_start, 1, @errors)@@@ -77,6 +163,7 @@ function createJobUsage({ db, app, contracts, outbox = null, now = () => Date.no/** INSIDE the transaction that records the job's end (store.finish), with the row read back. */async function finished(row) {++ await record(row);const k = keyOf(row);if (!k) return;const failed = row.state === 'failed';@@@ -142,10 +229,10 @@ function createJobUsage({ db, app, contracts, outbox = null, now = () => Date.no}return {-- enabled: true, finished, flush, payloadOf,++ enabled: true, billed, finished, flush, flushReadings, payloadOf, readings: readingsStatus,pending: async () => (await q.pending.get({ app })).n,status: async () => ({ pending: (await q.pending.get({ app })).n, invalid: stats.invalid, last_invalid: stats.lastInvalid }),};}--module.exports = { createJobUsage, keyOf, HOUR_MS, GRACE_MS };++module.exports = { createJobUsage, keyOf, readingOf, HOUR_MS, GRACE_MS };@@@ -5,6 +5,10 @@// count; each closed hour is written to the outbox once as tools.usage.recorded (valid against// openvibe-contracts, no owner, input or file name); a job ending in an hour already sent re-sends it// as revision 2; without an outbox nothing is counted. On PostgreSQL (plan T8): openvibe-sdk/db.++// Billing readings (plan T5 step 7, a fetch stub as OpenVibe.Billing): every job that succeeded or failed++// stores one platform.usage-sample@1 reading in that transaction and the flush posts it once (subject only++// for a person, idempotency_key tools:job:<id>); a cancelled job posts nothing; a refused post is logged,++// never thrown, and retried; without OV_BILLING_URL nothing is stored and the rollups still work.const assert = require('assert');const fs = require('fs');const os = require('os');@@@ -16,7 +20,8 @@ const contracts = dep('openvibe-contracts');const sdk = dep('openvibe-sdk');const jobs = require('../jobs');const { outboxFromEnv } = require('../jobs/events');--const { keyOf, HOUR_MS } = require('../jobs/usage');++const { keyOf, readingOf, HOUR_MS } = require('../jobs/usage');++const { createBillingClient } = require('../billing');const PRJ = 'prj_01JDDDDDDDDDDDDDDDDDDDDDDD';const APP = 'app_01JCCCCCCCCCCCCCCCCCCCCCCC';@@@ -70,6 +75,9 @@ function define(system) {await until(async () => (await system.get(job.id)).state === 'succeeded', 'inert job');assert.strictEqual(Number((await db.prepare('SELECT COUNT(*) AS n FROM tool_job_usage').get()).n), 0, 'inert: nothing counted');assert.deepStrictEqual((await system.stats()).usage, { enabled: false });++ assert.strictEqual(Number((await db.prepare('SELECT COUNT(*) AS n FROM tool_job_billing_readings').get()).n), 0, 'inert: no reading stored');++ assert.strictEqual(createBillingClient({ env: {}, fetchImpl: () => { throw new Error('no fetch while inert'); } }).enabled, false);++ assert.ok(!('billing' in (await system.stats())), 'no billing stats while inert');system.stop();}@@@ -145,8 +153,115 @@ function define(system) {const p = (typeof resent.envelope === 'string' ? JSON.parse(resent.envelope) : resent.envelope).payload;assert.deepStrictEqual([p.dimension, p.quantity, p.errors, p.revision], ['test.boom', 2, 1, 2]);assert.deepStrictEqual((await system.stats()).usage, { pending: 0, invalid: 0, last_invalid: null });++ assert.strictEqual(Number((await db.prepare('SELECT COUNT(*) AS n FROM tool_job_billing_readings').get()).n), 0, 'rollups without OV_BILLING_URL: no reading stored');system.stop(); await outbox.stop();++++ // ── Billing readings (no events outbox: the readings work without one) ──++ {++ const bdir = path.join(root, 'billing');++ fs.mkdirSync(bdir, { recursive: true });++ const bdb = await testDb(bdir);++ const posts = []; // what Billing stored, by idempotency_key++ const stored = new Map();++ const tokens = [];++ let mode = 'ok';++ const billingFetch = async (url, init) => {++ if (String(url) === 'http://network.test/oauth/token') {++ const form = new URLSearchParams(String(init.body));++ tokens.push(form.get('audience'));++ if (mode === 'no-grant') return new Response(JSON.stringify({ error: 'invalid_scope' }), { status: 401 });++ return new Response(JSON.stringify({ access_token: 'tools-billing-token', expires_in: 3600 }), { status: 200 });++ }++ assert.strictEqual(String(url), 'http://billing.test/api/v1/usage');++ assert.strictEqual(init.headers.Authorization, 'Bearer tools-billing-token');++ if (mode === 'down') return new Response('{"error":{"code":"billing.unavailable"}}', { status: 503 });++ if (mode === 'bad') return new Response('{"error":{"code":"billing.invalid_reading"}}', { status: 400 });++ const reading = JSON.parse(init.body);++ posts.push(reading);++ const replay = stored.has(reading.idempotency_key);++ stored.set(reading.idempotency_key, reading);++ return new Response(JSON.stringify(reading), { status: replay ? 200 : 201 });++ };++ const errors = [];++ const blog = { log() {}, warn() {}, error: (...a) => errors.push(a.join(' ')) };++ const billing = createBillingClient({ env: { OV_BILLING_URL: 'http://billing.test/', OV_NETWORK_INTERNAL_URL: 'http://network.test', OV_OAUTH_CLIENT_SECRET: 'tools-secret' }, fetchImpl: billingFetch, log: blog });++ assert.strictEqual(billing.enabled, true);++ const bsys = jobs.createJobSystem({ db: bdb, contracts, service: 'img', dataDir: bdir, concurrency: 1, billing, progressThrottleMs: 0, pruneIntervalMs: 60 * 60 * 1000, log: blog });++ define(bsys);++ await bsys.start();++ const brun = async (o, state) => {++ const { job } = await bsys.submit({ owner: `app:${APP}`, type: 'test.upper', input: { text: SECRET_INPUT }, ...o });++ await until(async () => { const r = await bsys.get(job.id); return r && r.state === state; }, `${job.id} ${state}`);++ return job.id;++ };++ const readingRows = async () => bdb.prepare('SELECT * FROM tool_job_billing_readings ORDER BY created_at, job_id').all();++++ const personId = await brun({ owner: `user:${USER}`, project: null, tool: 'image-resize', traceId: TRACE }, 'succeeded');++ const appId = await brun({ project: PRJ }, 'succeeded');++ const failId = await brun({ project: PRJ, type: 'test.boom', files: [{ buffer: Buffer.from('x'), name: SECRET_FILE, mime: 'text/plain' }] }, 'failed');++ const { job: held } = await bsys.submit({ owner: `user:${USER}`, type: 'test.upper', input: { hold: true } });++ await until(async () => (await bsys.get(held.id)).state === 'running', 'held job running');++ await bsys.cancel(held.id);++ await until(async () => (await bsys.get(held.id)).state === 'cancelled', 'held job cancelled');++ assert.deepStrictEqual((await readingRows()).map(r => [r.job_id, r.sent_at]), [[personId, null], [appId, null], [failId, null]], 'stored in the job-end transaction, unsent; a cancelled job has none');++ assert.strictEqual(posts.length, 0, 'nothing is posted inside the job\'s transaction');++++ assert.deepStrictEqual(await bsys.flushReadings(), { posted: 3, refused: 0 });++ assert.deepStrictEqual(tokens, ['openvibe.billing'], 'one token for the Billing audience, cached');++ assert.deepStrictEqual(posts.map(p => p.idempotency_key), [personId, appId, failId].map(id => `tools:job:${id}`));++ for (const p of posts) {++ const v = contracts.validate('platform.usage-sample@1', p);++ assert.ok(v.valid, JSON.stringify(v.errors));++ assert.deepStrictEqual([p.service, p.source, p.quantity, p.unit], ['openvibe.tools', 'openvibe.tools', 1, 'jobs']);++ }++ const [person, appJob, failJob] = posts;++ assert.deepStrictEqual([person.id, person.operation, person.resource, person.subject, person.trace_id, person.project], [`tools-job-${personId}`, 'tools.tool.run', 'image-resize', `user:${USER}`, TRACE, undefined]);++ assert.deepStrictEqual([appJob.operation, appJob.resource, appJob.project, 'subject' in appJob], ['tools.job.create', 'test.upper', PRJ, false], 'an app\'s job carries its project, no subject');++ assert.deepStrictEqual([failJob.operation, failJob.resource, failJob.quantity], ['tools.job.create', 'test.boom', 1], 'a failed job is billed');++ const text = JSON.stringify(posts);++ for (const s of [APP, SECRET_INPUT, SECRET_FILE, 'secret-output-name', 'Cannot read', held.id]) assert.ok(!text.includes(s), `no "${s}" in readings`);++ assert.ok((await readingRows()).every(r => r.sent_at != null && Number(r.attempts) === 0));++++ // ── Once per job: a second flush and a re-run of the job's end add nothing ──++ assert.deepStrictEqual(await bsys.flushReadings(), { posted: 0, refused: 0 });++ await bdb.tx(async () => bsys.usage.finished(await bsys.get(personId)));++ assert.strictEqual((await readingRows()).length, 3, 'a re-run stores no second reading');++ assert.deepStrictEqual(await bsys.flushReadings(), { posted: 0, refused: 0 });++ assert.strictEqual(posts.length, 3);++ assert.strictEqual(readingOf(await bsys.get(personId)).idempotency_key, `tools:job:${personId}`, 'the key is stable');++ assert.deepStrictEqual(readingOf(await bsys.get(personId)), person, 'the stored reading is the job\'s reading');++ assert.strictEqual(readingOf(await bsys.get(held.id)), null);++++ // ── Refused: logged, never thrown into the job, kept unsent with the error for the next flush ──++ mode = 'down';++ const downId = await brun({ project: PRJ }, 'succeeded');++ const downId2 = await brun({ project: PRJ }, 'succeeded');++ assert.deepStrictEqual(await bsys.flushReadings(), { posted: 0, refused: 1 }, 'Billing down: stop after the first refusal');++ let rows = await readingRows();++ const down = rows.find(r => r.job_id === downId);++ assert.deepStrictEqual([down.sent_at, Number(down.attempts)], [null, 1]);++ assert.match(down.last_error, /^503 /);++ assert.ok(errors.some(e => e.includes(`tools:job:${downId}`) && e.includes('503')), 'the refusal is logged');++ mode = 'bad';++ assert.deepStrictEqual(await bsys.flushReadings(), { posted: 0, refused: 2 }, 'a 4xx for one reading does not stop the others');++ mode = 'ok';++ assert.deepStrictEqual(await bsys.flushReadings(), { posted: 2, refused: 0 });++ rows = await readingRows();++ assert.ok(rows.every(r => r.sent_at != null), 'sent on the next flush');++ assert.deepStrictEqual(posts.slice(3).map(p => p.idempotency_key).sort(), [downId, downId2].map(id => `tools:job:${id}`).sort());++ assert.strictEqual((await bsys.get(downId)).state, 'succeeded', 'the job is untouched by the refusal');++++ // ── No grant yet (the token is refused): logged, never thrown ──++ const noGrant = createBillingClient({ env: { OV_BILLING_URL: 'http://billing.test', OV_NETWORK_INTERNAL_URL: 'http://network.test' }, fetchImpl: billingFetch, log: blog });++ mode = 'no-grant';++ const res = await noGrant.post(person);++ assert.deepStrictEqual([res.ok, res.status], [false, 401]);++ mode = 'ok';++ const st = (await bsys.stats()).billing;++ assert.deepStrictEqual([st.pending, st.posted, st.refused, st.invalid], [0, 5, 3, 0]);++ bsys.stop();++ }console.log('job usage: all checks passed');} finally {await closeAllTestDbs();@@@ -20,7 +20,7 @@"iovalkey": "^0.4.0","jsonwebtoken": "^9.0.2","multer": "^1.4.5-lts.1",-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.79.0",++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.92.0","openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.25.1","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.3.1","pg": "^8.23.0",@@@ -1290,9 +1290,9 @@}},"node_modules/openvibe-contracts": {-- "version": "0.79.0",-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.79.0",-- "integrity": "sha512-FY5GLbiOq0kDw5lUPPzX6h3ATbzvsr/mX5DUCF4GRfn6vGMP0p89rZZDhgtVDaS+yCBCYMZ8zFaZxuHEkEVwVg==",++ "version": "0.92.0",++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.92.0",++ "integrity": "sha512-2Wm1CdsNqIBou+XYAtIiIcQ6GyXWWh5miG81iFYrPvWGLVF1BUjEN4XxQGK2S2bC6TCKz7XQtp5IJx3Mg0vPIw==","license": "AGPL-3.0-only","dependencies": {"ajv": "^8.17.1",@@@ -23,7 +23,7 @@"helmet": "^8.1.0","jsonwebtoken": "^9.0.2","multer": "^1.4.5-lts.1",-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.79.0",++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.92.0","openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.25.1","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.3.1","uuid": "^11.1.0"@@@ -19,7 +19,7 @@"iovalkey": "^0.4.0","jsonwebtoken": "^9.0.2","multer": "^1.4.5-lts.1",-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.79.0",++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.92.0","openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.25.1","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.3.1","pdf-lib": "^1.17.1",@@@ -1367,9 +1367,9 @@}},"node_modules/openvibe-contracts": {-- "version": "0.79.0",-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.79.0",-- "integrity": "sha512-FY5GLbiOq0kDw5lUPPzX6h3ATbzvsr/mX5DUCF4GRfn6vGMP0p89rZZDhgtVDaS+yCBCYMZ8zFaZxuHEkEVwVg==",++ "version": "0.92.0",++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.92.0",++ "integrity": "sha512-2Wm1CdsNqIBou+XYAtIiIcQ6GyXWWh5miG81iFYrPvWGLVF1BUjEN4XxQGK2S2bC6TCKz7XQtp5IJx3Mg0vPIw==","license": "AGPL-3.0-only","dependencies": {"ajv": "^8.17.1",@@@ -22,7 +22,7 @@"helmet": "^8.1.0","jsonwebtoken": "^9.0.2","multer": "^1.4.5-lts.1",-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.79.0",++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.92.0","openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.25.1","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.3.1","pdf-lib": "^1.17.1",@@@ -60,3 +60,7 @@ MEDIA_PUBLIC_URL=https://openvibe.media#EVENTS_URL=http://127.0.0.1:4300#EVENTS_PUBLISH=off#EVENTS_RELAY_INTERVAL_MS=2000++# One usage reading per ended job → OpenVibe.Billing POST /api/v1/usage (off while unset). Needs OV_OAUTH_CLIENT_SECRET++# and the Network grant [tools, billing.usage.record, openvibe.billing]; until then every post is 401 and only logged.++#OV_BILLING_URL=http://127.0.0.1:4600++#OV_BILLING_AUDIENCE=openvibe.billing@@@ -18,7 +18,7 @@"helmet": "^8.1.0","iovalkey": "^0.4.0","jsonwebtoken": "^9.0.0",-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.79.0",++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.92.0","openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.25.1","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.3.1","pg": "^8.23.0"@@@ -1167,9 +1167,9 @@}},"node_modules/openvibe-contracts": {-- "version": "0.79.0",-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.79.0",-- "integrity": "sha512-FY5GLbiOq0kDw5lUPPzX6h3ATbzvsr/mX5DUCF4GRfn6vGMP0p89rZZDhgtVDaS+yCBCYMZ8zFaZxuHEkEVwVg==",++ "version": "0.92.0",++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.92.0",++ "integrity": "sha512-2Wm1CdsNqIBou+XYAtIiIcQ6GyXWWh5miG81iFYrPvWGLVF1BUjEN4XxQGK2S2bC6TCKz7XQtp5IJx3Mg0vPIw==","license": "AGPL-3.0-only","dependencies": {"ajv": "^8.17.1",@@@ -21,7 +21,7 @@"express-rate-limit": "^7.5.0","helmet": "^8.1.0","jsonwebtoken": "^9.0.0",-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.79.0",++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.92.0","openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.25.1","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.3.1"}@@@ -19,7 +19,7 @@"iovalkey": "^0.4.0","jsonwebtoken": "^9.0.2","multer": "^1.4.5-lts.1",-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.79.0",++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.92.0","openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.25.1","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.3.1","pg": "^8.23.0",@@@ -1858,9 +1858,9 @@}},"node_modules/openvibe-contracts": {-- "version": "0.79.0",-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.79.0",-- "integrity": "sha512-FY5GLbiOq0kDw5lUPPzX6h3ATbzvsr/mX5DUCF4GRfn6vGMP0p89rZZDhgtVDaS+yCBCYMZ8zFaZxuHEkEVwVg==",++ "version": "0.92.0",++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.92.0",++ "integrity": "sha512-2Wm1CdsNqIBou+XYAtIiIcQ6GyXWWh5miG81iFYrPvWGLVF1BUjEN4XxQGK2S2bC6TCKz7XQtp5IJx3Mg0vPIw==","license": "AGPL-3.0-only","dependencies": {"ajv": "^8.17.1",@@@ -22,7 +22,7 @@"helmet": "^8.1.0","jsonwebtoken": "^9.0.2","multer": "^1.4.5-lts.1",-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.79.0",++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.92.0","openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.25.1","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.3.1","sharp": "^0.33.5",nnew file mode 100644@@@ -0,0 +1,74 @@++# Cutover runbook — Tools job usage readings to OpenVibe.Billing (plan T5 step 7)++++This PR adds one empty table, `tool_job_billing_readings`, and the code that fills it: every tool job that++succeeds or fails stores one `platform.usage-sample@1` reading in the transaction that records its end, and++the jobs tick posts the unsent ones to OpenVibe.Billing `POST /api/v1/usage`. It also bumps++`openvibe-contracts` from v0.79.0 to v0.92.0 in the img, audio, docs and gateway apps. Nothing existing is++renamed, dropped, altered or backfilled; the hourly `tools.usage.recorded` rollups (`tool_job_usage`) are++unchanged. No money moves here: Billing stores the readings and rates them in its own sweep.++++## What the migration does++++`migrations/0005_billing_readings.sql` (`phase: expand`) creates the new table and two partial indexes:++++```sql++CREATE TABLE IF NOT EXISTS tool_job_billing_readings (++ job_id text PRIMARY KEY, app text NOT NULL DEFAULT '', reading jsonb NOT NULL,++ created_at bigint NOT NULL, sent_at bigint, attempts integer NOT NULL DEFAULT 0, last_error text++);++CREATE INDEX IF NOT EXISTS tool_job_billing_readings_unsent ON tool_job_billing_readings (app, attempts, created_at) WHERE sent_at IS NULL;++CREATE INDEX IF NOT EXISTS tool_job_billing_readings_sent ON tool_job_billing_readings (sent_at) WHERE sent_at IS NOT NULL;++```++++The runner is `openvibe-sdk/db`'s `migrate()`, started by `apps/_shared/db.js` as the owner++(`DATABASE_DIRECT_URL`) when the first app of the new release boots: it locks `ov_migrations`, runs the file++in one transaction, records id `0005` with its checksum, and refuses a later edit of it. The other apps of the++release wait on the same lock and find it applied.++++## Order++++1. **Backup.** Take the logical dump of the `tools` database before the deploy (the pre-merge rule for any++ data change, even an additive one): `sudo ovhost backup --all --logical`. Note the stamp it prints; the++ way back uses it.++2. **Merge and deploy** through the pipeline (`ovhost deploy tools`, `--wait-idle` is not needed: no job++ state changes shape). The first app to boot applies `0005`.++3. **Leave `OV_BILLING_URL` unset at first.** With it unset the code is inert: nothing is stored in the new++ table, nothing is posted, the rollups and job events behave as before.++4. **Network grant** (OpenVibe.Network, its own PR): the `tools` client needs++ `billing.usage.record` on audience `openvibe.billing`. Without it every token request for that audience is++ refused, each post is logged as `[Billing] reading tools:job:<id> not sent: token: 401 …`, and the readings++ wait in the table (no job is affected).++5. **Turn it on** once the grant is live: add `OV_BILLING_URL` (Billing's internal URL) and, only if it++ differs, `OV_BILLING_AUDIENCE` to `/etc/openvibe/tools.env`, then restart through `ovhost deploy tools++ --restart`.++++## Verification++++- `ov access run openvibe-ovh health tools` (or `/api/ready` on each app) answers 200 after the deploy.++- The migration is recorded: `SELECT id, name, phase FROM ov_migrations WHERE id = '0005'` returns++ `0005 | billing_readings | expand`, and `SELECT COUNT(*) FROM tool_job_billing_readings` is 0 while++ `OV_BILLING_URL` is unset.++- After step 5: `GET /api/health` on img, audio or docs shows `jobs.billing` with `pending` going back to 0++ after a tick (5 minutes) and `posted` growing; `refused` with a `last_error` of `401 …` means the grant is++ missing. `SELECT COUNT(*) FROM tool_job_billing_readings WHERE sent_at IS NULL` stays small.++- In Billing: `GET /api/v1/usage?service=openvibe.tools` (`billing.ledger.admin`) lists the readings, one per++ ended job, `idempotency_key` `tools:job:<job id>`.++++```rehearse++# 0005 is additive (one new table, no backfill): it must exist, empty, with its two indexes, next to the++# untouched tool_jobs, tool_job_usage and event_outbox.++node -e 'const a=require("assert");const {dep}=require("./apps/_shared/test/deps");const db=dep("openvibe-sdk/db").createDb({url:process.env.DATABASE_DIRECT_URL,service:"rehearsal",max:1});(async()=>{const c=await db.prepare("SELECT column_name AS c FROM information_schema.columns WHERE table_name = ? ORDER BY ordinal_position").all("tool_job_billing_readings");a.deepStrictEqual(c.map(r=>r.c),["job_id","app","reading","created_at","sent_at","attempts","last_error"]);a.strictEqual(Number((await db.prepare("SELECT COUNT(*) AS n FROM tool_job_billing_readings").get()).n),0);const i=await db.prepare("SELECT indexname AS i FROM pg_indexes WHERE tablename = ? ORDER BY indexname").all("tool_job_billing_readings");a.deepStrictEqual(i.map(r=>r.i),["tool_job_billing_readings_pkey","tool_job_billing_readings_sent","tool_job_billing_readings_unsent"]);for(const t of ["tool_jobs","tool_job_usage","event_outbox"])a.ok(await db.prepare("SELECT to_regclass(?) AS r").get(t).then(r=>r.r),t);console.log("0005: tool_job_billing_readings created empty with its two indexes; tool_jobs, tool_job_usage and event_outbox untouched")})().finally(()=>db.close()).catch((e)=>{console.error(e);process.exitCode=1})'++```++++## Rollback++++- **Code only** (the usual way back): `ovhost rollback tools` (or `deploy/scripts/deploy.sh --rollback`). The++ previous release does not know the table and ignores it; the migration stays applied, which the previous++ release's runner accepts (an applied id it does not have is not an error). Unsent readings stay in the table++ and are posted when the release comes back (Billing dedupes on `idempotency_key`).++- **Stop posting without a rollback:** remove `OV_BILLING_URL` from `/etc/openvibe/tools.env` and restart.++- **Remove the table** (only if it must go; a contract step, by hand as the owner, after a code rollback):++ `DROP TABLE tool_job_billing_readings; DELETE FROM ov_migrations WHERE id = '0005';`. Readings not yet posted++ are lost with it; Billing keeps the ones it stored.++- **Restore:** the dump from step 1 restores the database as it was before the deploy (`pg_restore` per++ OpenVibe.Host `docs/backups.md`); not needed for this additive change unless the database itself is damaged.nnew file mode 100644@@@ -0,0 +1,15 @@++-- phase: expand++-- tool_job_billing_readings (plan T5 step 7): one platform.usage-sample@1 reading per ended job++-- (succeeded or failed), written in the transaction that records the job's end and posted to++-- OpenVibe.Billing from the jobs tick (apps/_shared/jobs/usage.js). Additive: a new empty table.++CREATE TABLE IF NOT EXISTS tool_job_billing_readings (++ job_id text PRIMARY KEY,++ app text NOT NULL DEFAULT '',++ reading jsonb NOT NULL,++ created_at bigint NOT NULL,++ sent_at bigint,++ attempts integer NOT NULL DEFAULT 0,++ last_error text++);++CREATE INDEX IF NOT EXISTS tool_job_billing_readings_unsent ON tool_job_billing_readings (app, attempts, created_at) WHERE sent_at IS NULL;++CREATE INDEX IF NOT EXISTS tool_job_billing_readings_sent ON tool_job_billing_readings (sent_at) WHERE sent_at IS NOT NULL;
@@ -28,9 +28,10 @@ offline against the Network's public key (JWKS).
@@ -28,9 +28,10 @@ offline against the Network's public key (JWKS).
- OpenVibe.Network (SSO as OAuth client `tools`, JWKS, client-credentials tokens, the domains registry),
- OpenVibe.Network (SSO as OAuth client `tools`, JWKS, client-credentials tokens, the domains registry),
OpenVibe.Media (job results), OpenVibe.Events (job and usage events, revocation cutoffs),
OpenVibe.Media (job results), OpenVibe.Events (job and usage events, revocation cutoffs),
- OpenVibe.Search (tool documents), OpenVibe.Community (the paste hand-over)
+ OpenVibe.Search (tool documents), OpenVibe.Community (the paste hand-over), OpenVibe.Billing (job usage
+ readings)
- host programs some tools need (ffmpeg, yt-dlp, qpdf, poppler, libheif; [Host packages](#host-packages-some-tools-need))
- host programs some tools need (ffmpeg, yt-dlp, qpdf, poppler, libheif; [Host packages](#host-packages-some-tools-need))
-- in each app: `openvibe-contracts` v0.63.0, `openvibe-sdk` v0.12.0 and `openvibe-shared` v1.23.0, pinned
+- in each app: `openvibe-contracts` v0.92.0 (img, audio, docs, gateway), `openvibe-sdk` v0.12.0 and `openvibe-shared` v1.23.0, pinned
by release tarball, and `apps/_shared` (openvibe-tools-shared) by relative path
by release tarball, and `apps/_shared` (openvibe-tools-shared) by relative path
## Capabilities
## Capabilities
@@ -46,6 +47,7 @@ Called elsewhere, as the service principal `tools` (the OAuth client `tools`):
@@ -46,6 +47,7 @@ Called elsewhere, as the service principal `tools` (the OAuth client `tools`):
| OpenVibe.Media | `media.object.upload`, `media.object.read` | job results |
| OpenVibe.Media | `media.object.upload`, `media.object.read` | job results |
| OpenVibe.Events | `events.event.publish`, `events.subscription.manage` | job and usage events; the `network.user.token_valid_after` subscription |
| OpenVibe.Events | `events.event.publish`, `events.subscription.manage` | job and usage events; the `network.user.token_valid_after` subscription |
| OpenVibe.Search | `search.document.write` | tool documents for the network index |
| OpenVibe.Search | `search.document.write` | tool documents for the network index |
+| OpenVibe.Billing | `billing.usage.record` (audience `openvibe.billing`) | one usage reading per ended job |
## Layout
## Layout
@@ -300,12 +302,27 @@ reload reattaches.
@@ -300,12 +302,27 @@ reload reattaches.
`env: sandbox`. No owner, session, address, input, file name or output leaves. OpenVibe.Network adds the rollups
`env: sandbox`. No owner, session, address, input, file name or output leaves. OpenVibe.Network adds the rollups
up for the project's usage page on openvibe.codes. Off without `EVENTS_URL`, like the job events;
up for the project's usage page on openvibe.codes. Off without `EVENTS_URL`, like the job events;
`GET /api/health` shows `jobs.usage` (pending, invalid).
`GET /api/health` shows `jobs.usage` (pending, invalid).
+- **Usage readings to OpenVibe.Billing** (`apps/_shared/billing.js`, `apps/_shared/jobs/usage.js`, plan T5 step 7):
+ every job that succeeds or fails (anyone's, not only a project's; a cancelled one is not billed) gets one
+ `platform.usage-sample@1` reading — id `tools-job-<id>`, `idempotency_key` `tools:job:<id>`, service and source
+ `openvibe.tools`, operation `tools.tool.run` (a tool run) or `tools.job.create`, resource the tool or job type,
+ quantity 1, unit `jobs`, `at` the job's end, plus `project` and `trace_id` when the job has them and `subject`
+ (`user:usr_…`) only for a signed-in person's job, never an app or a browser session. It is stored in
+ `tool_job_billing_readings` in the transaction that records the end (no network call there); the pruner's timer
+ posts the unsent ones (500 at most a tick) to `POST /api/v1/usage` and marks each sent. A refused post is logged,
+ never thrown: the reading keeps its error and attempt count and is posted again on the next tick (Billing answers
+ a replay with 200), and an outage, a 401/403 or a 429 stops that tick's batch. The hourly `tools.usage.recorded`
+ rollups above are unchanged and stay subject-less. Off without `OV_BILLING_URL` (nothing is stored). The token is
+ Network `client_credentials` for the `tools` client on audience `OV_BILLING_AUDIENCE` (default
+ `openvibe.billing`); Network must grant that client `billing.usage.record` on it — until the grant exists every
+ post is refused (401) and only logged, and the readings wait in the table. `GET /api/health` shows
+ `jobs.billing` (pending, posted, refused, invalid, last_error) when it is on.
Environment (all in `/etc/openvibe/tools.env`): `TOOLS_WORKERS`, `TOOLS_WORKERS_<APP>`, `TOOLS_WORKER_MEMORY_MB`,
Environment (all in `/etc/openvibe/tools.env`): `TOOLS_WORKERS`, `TOOLS_WORKERS_<APP>`, `TOOLS_WORKER_MEMORY_MB`,
`TOOLS_WORKER_MEMORY_MB_<APP>`, `TOOLS_WORKER_QUEUE`, `TOOLS_WORKER_IDLE_MS`, `TOOLS_JOBS_CONCURRENCY`, `TOOLS_JOBS_CONCURRENCY_<APP>`,
`TOOLS_WORKER_MEMORY_MB_<APP>`, `TOOLS_WORKER_QUEUE`, `TOOLS_WORKER_IDLE_MS`, `TOOLS_JOBS_CONCURRENCY`, `TOOLS_JOBS_CONCURRENCY_<APP>`,
`TOOLS_JOBS_MAX_ACTIVE`, `TOOLS_JOB_RESULTS`, `TOOLS_MEDIA_NAMESPACE`, `OV_MEDIA_INTERNAL_URL`, `OV_MEDIA_URL`,
`TOOLS_JOBS_MAX_ACTIVE`, `TOOLS_JOB_RESULTS`, `TOOLS_MEDIA_NAMESPACE`, `OV_MEDIA_INTERNAL_URL`, `OV_MEDIA_URL`,
`OV_NETWORK_INTERNAL_URL`, `OV_OAUTH_CLIENT_ID`, `OV_OAUTH_CLIENT_SECRET`, `EVENTS_URL`, `EVENTS_PUBLISH`,
`OV_NETWORK_INTERNAL_URL`, `OV_OAUTH_CLIENT_ID`, `OV_OAUTH_CLIENT_SECRET`, `EVENTS_URL`, `EVENTS_PUBLISH`,
-`EVENTS_RELAY_INTERVAL_MS`.
+`EVENTS_RELAY_INTERVAL_MS`, `OV_BILLING_URL`, `OV_BILLING_AUDIENCE`.
Not done yet: quotas for external developer apps are the guard's service and sandbox tiers, not a Codes-issued quota.
Not done yet: quotas for external developer apps are the guard's service and sandbox tiers, not a Codes-issued quota.
new file mode 100644
new file mode 100644
@@ -0,0 +1,59 @@
@@ -0,0 +1,59 @@
+'use strict';
+// ═══════════════════════════════════════════════════════════════
+// Usage readings → OpenVibe.Billing (plan T5 step 7): post(reading) sends one platform.usage-sample@1
+// to POST /api/v1/usage with Tools' service token (Network client_credentials, OV_OAUTH_CLIENT_ID/SECRET,
+// audience OV_BILLING_AUDIENCE, default openvibe.billing; the Tools client needs the grant
+// billing.usage.record). Billing dedupes on the reading's idempotency_key (a replay answers 200), so a
+// retry is safe.
+//
+// Inert unless OV_BILLING_URL is set. post() never throws: a refused post (or no token) is logged and
+// answered { ok: false, status, error }, so the caller keeps the reading for its next flush. Never called
+// inside a job's transaction (./jobs/usage.js stores the reading there and posts it from the flush).
+// No dependencies: required by relative path.
+// ═══════════════════════════════════════════════════════════════
+
+const PATH = '/api/v1/usage';
+const TIMEOUT_MS = 10000;
+
+function createBillingClient({ env = process.env, fetchImpl = globalThis.fetch, now = () => Date.now(), log = console } = {}) {
+ const base = String(env.OV_BILLING_URL || '').trim().replace(/\/+$/, '');
+ if (!base) return { enabled: false, post: async () => ({ ok: false, skipped: true }) };
+ const network = String(env.OV_NETWORK_INTERNAL_URL || 'http://127.0.0.1:4000').replace(/\/+$/, '');
+ const audience = env.OV_BILLING_AUDIENCE || 'openvibe.billing';
+ let token = null, tokenExp = 0;
+
+ async function bearer() {
+ if (token && now() < tokenExp) return token;
+ const body = new URLSearchParams({ grant_type: 'client_credentials', client_id: env.OV_OAUTH_CLIENT_ID || 'tools', client_secret: env.OV_OAUTH_CLIENT_SECRET || '', audience });
+ const r = await fetchImpl(`${network}/oauth/token`, { method: 'POST', headers: { 'Content-Type': 'application/x-www-form-urlencoded', Accept: 'application/json' }, body, signal: AbortSignal.timeout(TIMEOUT_MS) });
+ const j = await r.json().catch(() => ({}));
+ if (!r.ok || !j.access_token) throw Object.assign(new Error(`token: ${r.status} ${j.error || ''}`.trim()), { status: r.status });
+ token = j.access_token; tokenExp = now() + Math.max(30, (Number(j.expires_in) || 300) - 60) * 1000;
+ return token;
+ }
+
+ /** One reading → Billing. { ok: true, status } when stored (201) or replayed (200); else logged, { ok: false, status, error }. */
+ async function post(reading) {
+ try {
+ const r = await fetchImpl(`${base}${PATH}`, {
+ method: 'POST',
+ headers: { 'Content-Type': 'application/json', Accept: 'application/json', Authorization: `Bearer ${await bearer()}` },
+ body: JSON.stringify(reading), signal: AbortSignal.timeout(TIMEOUT_MS),
+ });
+ if (r.ok) return { ok: true, status: r.status };
+ if (r.status === 401) token = null;
+ const text = (await r.text().catch(() => '')).replace(/\s+/g, ' ').slice(0, 300);
+ const error = `${r.status} ${text}`.trim();
+ log.error(`[Billing] reading ${reading && reading.idempotency_key} refused: ${error}`);
+ return { ok: false, status: r.status, error };
+ } catch (err) {
+ const error = String(err && err.message || err).slice(0, 300);
+ log.error(`[Billing] reading ${reading && reading.idempotency_key} not sent: ${error}`);
+ return { ok: false, status: err && err.status || 0, error };
+ }
+ }
+
+ return { enabled: true, post };
+}
+
+module.exports = { createBillingClient };
@@ -31,6 +31,10 @@
@@ -31,6 +31,10 @@
// transaction that records its end (./usage.js), and each closed hour goes out as tools.usage.recorded
// transaction that records its end (./usage.js), and each closed hour goes out as tools.usage.recorded
// from the pruner's timer.
// from the pruner's timer.
//
//
+// Billing (plan T5 step 7): with OV_BILLING_URL set (../billing.js), every job that succeeds or fails
+// also stores its usage reading in that transaction, and the pruner's timer posts the unsent ones to
+// OpenVibe.Billing (./usage.js flushReadings). Without it, nothing is stored or posted.
+//
// No dependencies of its own: the app passes the shared openvibe-sdk/db handle and openvibe-contracts.
// No dependencies of its own: the app passes the shared openvibe-sdk/db handle and openvibe-contracts.
// ═══════════════════════════════════════════════════════════════
// ═══════════════════════════════════════════════════════════════
@@ -41,6 +45,7 @@ const crypto = require('crypto');
@@ -41,6 +45,7 @@ const crypto = require('crypto');
const { createStore, TERMINAL } = require('./store');
const { createStore, TERMINAL } = require('./store');
const { createJobEvents } = require('./events');
const { createJobEvents } = require('./events');
const { createJobUsage } = require('./usage');
const { createJobUsage } = require('./usage');
+const { createBillingClient } = require('../billing');
const HOUR = 60 * 60 * 1000;
const HOUR = 60 * 60 * 1000;
@@ -98,6 +103,7 @@ const scrub = (msg) => String(msg || '').replace(/(?:\/[\w.-]+){2,}/g, '[file]')
@@ -98,6 +103,7 @@ const scrub = (msg) => String(msg || '').replace(/(?:\/[\w.-]+){2,}/g, '[file]')
* @param {number} [o.diskBudgetBytes=0] bytes under <dataDir>/jobs before busy() says so; 0 = no bound
* @param {number} [o.diskBudgetBytes=0] bytes under <dataDir>/jobs before busy() says so; 0 = no bound
* @param {object} [o.media] result store from ./media (null = keep results on local disk)
* @param {object} [o.media] result store from ./media (null = keep results on local disk)
* @param {object} [o.outbox] openvibe-sdk outbox on `db` (./events outboxFromEnv); null = no platform events
* @param {object} [o.outbox] openvibe-sdk outbox on `db` (./events outboxFromEnv); null = no platform events
+ * @param {object} [o.billing] ../billing createBillingClient (default: one from process.env; inert without OV_BILLING_URL)
* @param {number} [o.pruneIntervalMs=300000]
* @param {number} [o.pruneIntervalMs=300000]
* @param {number} [o.progressThrottleMs=250]
* @param {number} [o.progressThrottleMs=250]
*/
*/
@@ -117,8 +123,9 @@ function createJobSystem(o) {
@@ -117,8 +123,9 @@ function createJobSystem(o) {
const diskBudget = o.diskBudgetBytes > 0 ? o.diskBudgetBytes : 0;
const diskBudget = o.diskBudgetBytes > 0 ? o.diskBudgetBytes : 0;
const disk = { bytes: 0, at: 0, scanning: null };
const disk = { bytes: 0, at: 0, scanning: null };
const announce = createJobEvents({ contracts, service: o.service, outbox: o.outbox || null, referenceCount: (id) => store.referenceCount(id), log });
const announce = createJobEvents({ contracts, service: o.service, outbox: o.outbox || null, referenceCount: (id) => store.referenceCount(id), log });
- const usage = createJobUsage({ db: o.db, app: o.service, contracts, outbox: o.outbox || null, log });
- /** Inside the transaction `t` that recorded the end: tools.job.succeeded|failed and the project's usage. */
+ const billing = o.billing || createBillingClient({ log });
+ const usage = createJobUsage({ db: o.db, app: o.service, contracts, outbox: o.outbox || null, billing, log });
+ /** Inside the transaction `t` that recorded the end: tools.job.succeeded|failed, the project's usage and the Billing reading. */
async function ended(t, id) {
async function ended(t, id) {
const row = await store.get(id);
const row = await store.get(id);
await announce.finished(t, row);
await announce.finished(t, row);
@@ -696,9 +703,11 @@ function createJobSystem(o) {
@@ -696,9 +703,11 @@ function createJobSystem(o) {
if (r.requeued || r.failed) log.log(`[Jobs] ${o.service}: ${r.requeued} job(s) re-queued, ${r.failed} failed after restart`);
if (r.requeued || r.failed) log.log(`[Jobs] ${o.service}: ${r.requeued} job(s) re-queued, ${r.failed} failed after restart`);
await prune().catch(() => {});
await prune().catch(() => {});
await flushUsage();
await flushUsage();
+ flushReadings(); // not awaited: posting to Billing never holds up the start
pruneTimer = setInterval(() => {
pruneTimer = setInterval(() => {
prune().catch(err => log.error('[Jobs] prune:', err.message));
prune().catch(err => log.error('[Jobs] prune:', err.message));
flushUsage();
flushUsage();
+ flushReadings();
}, o.pruneIntervalMs || 5 * 60 * 1000);
}, o.pruneIntervalMs || 5 * 60 * 1000);
if (pruneTimer.unref) pruneTimer.unref();
if (pruneTimer.unref) pruneTimer.unref();
await kick();
await kick();
@@ -713,6 +722,10 @@ function createJobSystem(o) {
@@ -713,6 +722,10 @@ function createJobSystem(o) {
async function flushUsage(at) {
async function flushUsage(at) {
try { return await usage.flush(at); } catch (err) { log.error('[Jobs] usage flush:', err.message); return { queued: 0, invalid: 0 }; }
try { return await usage.flush(at); } catch (err) { log.error('[Jobs] usage flush:', err.message); return { queued: 0, invalid: 0 }; }
}
}
+ /** Unsent Billing readings → OpenVibe.Billing (./usage.js); a failure is logged and retried next tick. */
+ async function flushReadings(at) {
+ try { return await usage.flushReadings(at); } catch (err) { log.error('[Jobs] billing flush:', err.message); return { posted: 0, refused: 0 }; }
+ }
function stop() {
function stop() {
stopped = true;
stopped = true;
@@ -753,8 +766,8 @@ function createJobSystem(o) {
@@ -753,8 +766,8 @@ function createJobSystem(o) {
types: () => [...types.keys()],
types: () => [...types.keys()],
media,
media,
// running (after the spread) is the store's count of rows in 'running'; executing is this process's.
// running (after the spread) is the store's count of rows in 'running'; executing is this process's.
- stats: async () => ({ running: active.size, concurrency, ...(await store.counts()), executing: active.size, results: media ? 'media' : 'local', events: await announce.status(), usage: usage.enabled ? await usage.status() : { enabled: false } }),
- usage, flushUsage,
+ stats: async () => ({ running: active.size, concurrency, ...(await store.counts()), executing: active.size, results: media ? 'media' : 'local', events: await announce.status(), usage: usage.enabled ? await usage.status() : { enabled: false }, ...(usage.billed ? { billing: await usage.readings() } : {}) }),
+ usage, flushUsage, flushReadings,
outbox: o.outbox || null,
outbox: o.outbox || null,
/** True between start() and stop(): the worker picks up queued jobs. */
/** True between start() and stop(): the worker picks up queued jobs. */
isRunning: () => started && !stopped,
isRunning: () => started && !stopped,
@@ -19,8 +19,16 @@
@@ -19,8 +19,16 @@
// outbox relay publishes it. A job that ends in an hour already sent (a clock step back) reopens it as
// outbox relay publishes it. A job that ends in an hour already sent (a clock step back) reopens it as
// revision + 1. Only counts leave: no owner, session, address, input, file name or output.
// revision + 1. Only counts leave: no owner, session, address, input, file name or output.
//
//
-// On PostgreSQL (plan T8): openvibe-sdk/db, ambient transactions, createPgOutbox. Inert without an
-// outbox (EVENTS_URL unset, like ./events.js): nothing is counted.
+// Billing readings (plan T5 step 7): with a Billing client (../billing.js, OV_BILLING_URL set), the same
+// transaction also stores the job's platform.usage-sample@1 reading (readingOf: anyone's job that
+// succeeded or failed, not only a project's; quantity 1, unit jobs, idempotency_key tools:job:<id>) in
+// tool_job_billing_readings, never posting it there. flushReadings() posts the unsent ones (500 at most
+// a call) and marks each sent; a refused post counts an attempt with its error and stays for the next
+// call. The reading names the person (subject user:usr_…) for a person's job, never an app or a browser
+// session; no input, file name or output. Billing keeps it; the hourly rollups above stay subject-less.
+//
+// On PostgreSQL (plan T8): openvibe-sdk/db, ambient transactions, createPgOutbox. The rollups are inert
+// without an outbox (EVENTS_URL unset, like ./events.js), the readings without a Billing client.
// ═══════════════════════════════════════════════════════════════
// ═══════════════════════════════════════════════════════════════
const HOUR_MS = 60 * 60 * 1000;
const HOUR_MS = 60 * 60 * 1000;
@@ -32,6 +40,8 @@ const PROJECT_RE = /^prj_[0-9A-HJKMNP-TV-Z]{26}$/;
@@ -32,6 +40,8 @@ const PROJECT_RE = /^prj_[0-9A-HJKMNP-TV-Z]{26}$/;
const CODE_RE = /^[a-z][a-z0-9_]*(\.[a-z0-9_]+)+$/;
const CODE_RE = /^[a-z][a-z0-9_]*(\.[a-z0-9_]+)+$/;
const TRACE_RE = /^[0-9a-f]{32}$/;
const TRACE_RE = /^[0-9a-f]{32}$/;
const DIMENSION_RE = /^[a-z0-9][a-z0-9_.:-]{0,79}$/;
const DIMENSION_RE = /^[a-z0-9][a-z0-9_.:-]{0,79}$/;
+const PERSON_RE = /^user:usr_[0-9A-HJKMNP-TV-Z]{26}$/;
+const READINGS_PER_FLUSH = 500;
const hourOf = (ms) => Math.floor(ms / HOUR_MS) * HOUR_MS;
const hourOf = (ms) => Math.floor(ms / HOUR_MS) * HOUR_MS;
const parse = (s, d) => { try { return JSON.parse(s); } catch { return d; } };
const parse = (s, d) => { try { return JSON.parse(s); } catch { return d; } };
@@ -49,17 +59,93 @@ function keyOf(row) {
@@ -49,17 +59,93 @@ function keyOf(row) {
};
};
}
}
+/** The Billing reading (platform.usage-sample@1) of an ended tool_jobs row, or null when it is not billed. */
+function readingOf(row) {
+ if (!row || !row.id || (row.state !== 'succeeded' && row.state !== 'failed')) return null;
+ const resource = String(row.tool || row.type || '');
+ const r = {
+ id: `tools-job-${row.id}`,
+ idempotency_key: `tools:job:${row.id}`,
+ service: 'openvibe.tools',
+ operation: row.tool ? 'tools.tool.run' : 'tools.job.create',
+ resource: DIMENSION_RE.test(resource) ? resource : 'other',
+ quantity: 1,
+ unit: 'jobs',
+ at: new Date(Number(row.finished_at) || Date.now()).toISOString(),
+ source: 'openvibe.tools',
+ };
+ if (PROJECT_RE.test(String(row.project_id || ''))) r.project = row.project_id;
+ if (PERSON_RE.test(String(row.owner || ''))) r.subject = row.owner; // a person's job; never an app or a session
+ if (TRACE_RE.test(String(row.trace_id || ''))) r.trace_id = row.trace_id;
+ return r;
+}
+
/**
/**
* @param {object} o
* @param {object} o
* @param {object} o.db openvibe-sdk/db handle (the one tools database), where the outbox lives too
* @param {object} o.db openvibe-sdk/db handle (the one tools database), where the outbox lives too
* @param {string} o.app the owning app ('img' | …)
* @param {string} o.app the owning app ('img' | …)
* @param {object} o.contracts require('openvibe-contracts') (0.63.0+ knows tools.usage.recorded@1)
* @param {object} o.contracts require('openvibe-contracts') (0.63.0+ knows tools.usage.recorded@1)
- * @param {object|null} o.outbox openvibe-sdk createPgOutbox on `db` (./events outboxFromEnv); null = inert
+ * @param {object|null} o.outbox openvibe-sdk createPgOutbox on `db` (./events outboxFromEnv); null = no rollups
+ * @param {object|null} [o.billing] ../billing createBillingClient; null or not enabled = no readings
* @param {() => number} [o.now]
* @param {() => number} [o.now]
* @param {object} [o.log]
* @param {object} [o.log]
*/
*/
-function createJobUsage({ db, app, contracts, outbox = null, now = () => Date.now(), log = console }) {
- if (!outbox) return { enabled: false, finished: async () => {}, flush: async () => ({ queued: 0, invalid: 0 }), pending: async () => 0, status: async () => ({ enabled: false }) };
+function createJobUsage({ db, app, contracts, outbox = null, billing = null, now = () => Date.now(), log = console }) {
+ const billed = !!(billing && billing.enabled);
+ const r = {
+ add: db.prepare('INSERT INTO tool_job_billing_readings (job_id, app, reading, created_at) VALUES (@job_id, @app, @reading, @at) ON CONFLICT (job_id) DO NOTHING'),
+ unsent: db.prepare('SELECT job_id, reading FROM tool_job_billing_readings WHERE app = @app AND sent_at IS NULL ORDER BY attempts, created_at LIMIT @limit'),
+ sent: db.prepare('UPDATE tool_job_billing_readings SET sent_at = @at, last_error = NULL WHERE job_id = @job_id'),
+ failed: db.prepare('UPDATE tool_job_billing_readings SET attempts = attempts + 1, last_error = @error WHERE job_id = @job_id'),
+ prune: db.prepare('DELETE FROM tool_job_billing_readings WHERE app = @app AND sent_at IS NOT NULL AND sent_at < @at'),
+ pending: db.prepare('SELECT COUNT(*) AS n FROM tool_job_billing_readings WHERE app = @app AND sent_at IS NULL'),
+ };
+ const readings = { posted: 0, refused: 0, invalid: 0, lastError: null, running: null };
+
+ /** INSIDE the job-end transaction: store the reading (once per job); nothing leaves here. */
+ async function record(row) {
+ const reading = billed ? readingOf(row) : null;
+ if (!reading) return;
+ let v;
+ try { v = contracts.validate('platform.usage-sample@1', reading); } catch (err) { v = { valid: false, errors: [{ path: '', message: err.message }] }; }
+ if (!v.valid) {
+ readings.invalid++;
+ readings.lastError = `${reading.idempotency_key}: ${v.errors.map(e => `${e.path} ${e.message}`).join('; ')}`;
+ log.error(`[Jobs] billing reading not stored (does not match the contract): ${readings.lastError}`);
+ return;
+ }
+ await r.add.run({ job_id: row.id, app, reading: JSON.stringify(reading), at: now() });
+ }
+
+ /** Post the unsent readings to Billing, oldest and least-tried first; a refused one stays for the next call. */
+ function flushReadings(at = now()) {
+ if (!billed) return Promise.resolve({ posted: 0, refused: 0 });
+ if (!readings.running) readings.running = (async () => {
+ let posted = 0, refused = 0;
+ for (const row of await r.unsent.all({ app, limit: READINGS_PER_FLUSH })) {
+ const reading = typeof row.reading === 'string' ? parse(row.reading, null) : row.reading;
+ const res = await billing.post(reading);
+ if (res.ok) { await r.sent.run({ job_id: row.job_id, at: now() }); posted++; continue; }
+ refused++;
+ readings.lastError = res.error || String(res.status);
+ await r.failed.run({ job_id: row.job_id, error: readings.lastError.slice(0, 500) });
+ // Billing or the token is down (or the grant is missing): stop here, the rest waits for the next call.
+ if (!res.status || res.status === 401 || res.status === 403 || res.status === 429 || res.status >= 500) break;
+ }
+ await r.prune.run({ app, at: at - KEEP_SENT_MS });
+ readings.posted += posted; readings.refused += refused;
+ return { posted, refused };
+ })().finally(() => { readings.running = null; });
+ return readings.running;
+ }
+ const readingsStatus = async () => ({ pending: Number((await r.pending.get({ app })).n), posted: readings.posted, refused: readings.refused, invalid: readings.invalid, last_error: readings.lastError });
+
+ if (!outbox) {
+ return {
+ enabled: false, billed, finished: record, flush: async () => ({ queued: 0, invalid: 0 }), flushReadings,
+ pending: async () => 0, status: async () => ({ enabled: false }), readings: readingsStatus,
+ };
+ }
const q = {
const q = {
add: db.prepare(`INSERT INTO tool_job_usage (app, project_id, env, capability, dimension, window_start, quantity, errors)
add: db.prepare(`INSERT INTO tool_job_usage (app, project_id, env, capability, dimension, window_start, quantity, errors)
VALUES (@app, @project_id, @env, @capability, @dimension, @window_start, 1, @errors)
VALUES (@app, @project_id, @env, @capability, @dimension, @window_start, 1, @errors)
@@ -77,6 +163,7 @@ function createJobUsage({ db, app, contracts, outbox = null, now = () => Date.no
@@ -77,6 +163,7 @@ function createJobUsage({ db, app, contracts, outbox = null, now = () => Date.no
/** INSIDE the transaction that records the job's end (store.finish), with the row read back. */
/** INSIDE the transaction that records the job's end (store.finish), with the row read back. */
async function finished(row) {
async function finished(row) {
+ await record(row);
const k = keyOf(row);
const k = keyOf(row);
if (!k) return;
if (!k) return;
const failed = row.state === 'failed';
const failed = row.state === 'failed';
@@ -142,10 +229,10 @@ function createJobUsage({ db, app, contracts, outbox = null, now = () => Date.no
@@ -142,10 +229,10 @@ function createJobUsage({ db, app, contracts, outbox = null, now = () => Date.no
}
}
return {
return {
- enabled: true, finished, flush, payloadOf,
+ enabled: true, billed, finished, flush, flushReadings, payloadOf, readings: readingsStatus,
pending: async () => (await q.pending.get({ app })).n,
pending: async () => (await q.pending.get({ app })).n,
status: async () => ({ pending: (await q.pending.get({ app })).n, invalid: stats.invalid, last_invalid: stats.lastInvalid }),
status: async () => ({ pending: (await q.pending.get({ app })).n, invalid: stats.invalid, last_invalid: stats.lastInvalid }),
};
};
}
}
-module.exports = { createJobUsage, keyOf, HOUR_MS, GRACE_MS };
+module.exports = { createJobUsage, keyOf, readingOf, HOUR_MS, GRACE_MS };
@@ -5,6 +5,10 @@
@@ -5,6 +5,10 @@
// count; each closed hour is written to the outbox once as tools.usage.recorded (valid against
// count; each closed hour is written to the outbox once as tools.usage.recorded (valid against
// openvibe-contracts, no owner, input or file name); a job ending in an hour already sent re-sends it
// openvibe-contracts, no owner, input or file name); a job ending in an hour already sent re-sends it
// as revision 2; without an outbox nothing is counted. On PostgreSQL (plan T8): openvibe-sdk/db.
// as revision 2; without an outbox nothing is counted. On PostgreSQL (plan T8): openvibe-sdk/db.
+// Billing readings (plan T5 step 7, a fetch stub as OpenVibe.Billing): every job that succeeded or failed
+// stores one platform.usage-sample@1 reading in that transaction and the flush posts it once (subject only
+// for a person, idempotency_key tools:job:<id>); a cancelled job posts nothing; a refused post is logged,
+// never thrown, and retried; without OV_BILLING_URL nothing is stored and the rollups still work.
const assert = require('assert');
const assert = require('assert');
const fs = require('fs');
const fs = require('fs');
const os = require('os');
const os = require('os');
@@ -16,7 +20,8 @@ const contracts = dep('openvibe-contracts');
@@ -16,7 +20,8 @@ const contracts = dep('openvibe-contracts');
const sdk = dep('openvibe-sdk');
const sdk = dep('openvibe-sdk');
const jobs = require('../jobs');
const jobs = require('../jobs');
const { outboxFromEnv } = require('../jobs/events');
const { outboxFromEnv } = require('../jobs/events');
-const { keyOf, HOUR_MS } = require('../jobs/usage');
+const { keyOf, readingOf, HOUR_MS } = require('../jobs/usage');
+const { createBillingClient } = require('../billing');
const PRJ = 'prj_01JDDDDDDDDDDDDDDDDDDDDDDD';
const PRJ = 'prj_01JDDDDDDDDDDDDDDDDDDDDDDD';
const APP = 'app_01JCCCCCCCCCCCCCCCCCCCCCCC';
const APP = 'app_01JCCCCCCCCCCCCCCCCCCCCCCC';
@@ -70,6 +75,9 @@ function define(system) {
@@ -70,6 +75,9 @@ function define(system) {
await until(async () => (await system.get(job.id)).state === 'succeeded', 'inert job');
await until(async () => (await system.get(job.id)).state === 'succeeded', 'inert job');
assert.strictEqual(Number((await db.prepare('SELECT COUNT(*) AS n FROM tool_job_usage').get()).n), 0, 'inert: nothing counted');
assert.strictEqual(Number((await db.prepare('SELECT COUNT(*) AS n FROM tool_job_usage').get()).n), 0, 'inert: nothing counted');
assert.deepStrictEqual((await system.stats()).usage, { enabled: false });
assert.deepStrictEqual((await system.stats()).usage, { enabled: false });
+ assert.strictEqual(Number((await db.prepare('SELECT COUNT(*) AS n FROM tool_job_billing_readings').get()).n), 0, 'inert: no reading stored');
+ assert.strictEqual(createBillingClient({ env: {}, fetchImpl: () => { throw new Error('no fetch while inert'); } }).enabled, false);
+ assert.ok(!('billing' in (await system.stats())), 'no billing stats while inert');
system.stop();
system.stop();
}
}
@@ -145,8 +153,115 @@ function define(system) {
@@ -145,8 +153,115 @@ function define(system) {
const p = (typeof resent.envelope === 'string' ? JSON.parse(resent.envelope) : resent.envelope).payload;
const p = (typeof resent.envelope === 'string' ? JSON.parse(resent.envelope) : resent.envelope).payload;
assert.deepStrictEqual([p.dimension, p.quantity, p.errors, p.revision], ['test.boom', 2, 1, 2]);
assert.deepStrictEqual([p.dimension, p.quantity, p.errors, p.revision], ['test.boom', 2, 1, 2]);
assert.deepStrictEqual((await system.stats()).usage, { pending: 0, invalid: 0, last_invalid: null });
assert.deepStrictEqual((await system.stats()).usage, { pending: 0, invalid: 0, last_invalid: null });
+ assert.strictEqual(Number((await db.prepare('SELECT COUNT(*) AS n FROM tool_job_billing_readings').get()).n), 0, 'rollups without OV_BILLING_URL: no reading stored');
system.stop(); await outbox.stop();
system.stop(); await outbox.stop();
+
+ // ── Billing readings (no events outbox: the readings work without one) ──
+ {
+ const bdir = path.join(root, 'billing');
+ fs.mkdirSync(bdir, { recursive: true });
+ const bdb = await testDb(bdir);
+ const posts = []; // what Billing stored, by idempotency_key
+ const stored = new Map();
+ const tokens = [];
+ let mode = 'ok';
+ const billingFetch = async (url, init) => {
+ if (String(url) === 'http://network.test/oauth/token') {
+ const form = new URLSearchParams(String(init.body));
+ tokens.push(form.get('audience'));
+ if (mode === 'no-grant') return new Response(JSON.stringify({ error: 'invalid_scope' }), { status: 401 });
+ return new Response(JSON.stringify({ access_token: 'tools-billing-token', expires_in: 3600 }), { status: 200 });
+ }
+ assert.strictEqual(String(url), 'http://billing.test/api/v1/usage');
+ assert.strictEqual(init.headers.Authorization, 'Bearer tools-billing-token');
+ if (mode === 'down') return new Response('{"error":{"code":"billing.unavailable"}}', { status: 503 });
+ if (mode === 'bad') return new Response('{"error":{"code":"billing.invalid_reading"}}', { status: 400 });
+ const reading = JSON.parse(init.body);
+ posts.push(reading);
+ const replay = stored.has(reading.idempotency_key);
+ stored.set(reading.idempotency_key, reading);
+ return new Response(JSON.stringify(reading), { status: replay ? 200 : 201 });
+ };
+ const errors = [];
+ const blog = { log() {}, warn() {}, error: (...a) => errors.push(a.join(' ')) };
+ const billing = createBillingClient({ env: { OV_BILLING_URL: 'http://billing.test/', OV_NETWORK_INTERNAL_URL: 'http://network.test', OV_OAUTH_CLIENT_SECRET: 'tools-secret' }, fetchImpl: billingFetch, log: blog });
+ assert.strictEqual(billing.enabled, true);
+ const bsys = jobs.createJobSystem({ db: bdb, contracts, service: 'img', dataDir: bdir, concurrency: 1, billing, progressThrottleMs: 0, pruneIntervalMs: 60 * 60 * 1000, log: blog });
+ define(bsys);
+ await bsys.start();
+ const brun = async (o, state) => {
+ const { job } = await bsys.submit({ owner: `app:${APP}`, type: 'test.upper', input: { text: SECRET_INPUT }, ...o });
+ await until(async () => { const r = await bsys.get(job.id); return r && r.state === state; }, `${job.id} ${state}`);
+ return job.id;
+ };
+ const readingRows = async () => bdb.prepare('SELECT * FROM tool_job_billing_readings ORDER BY created_at, job_id').all();
+
+ const personId = await brun({ owner: `user:${USER}`, project: null, tool: 'image-resize', traceId: TRACE }, 'succeeded');
+ const appId = await brun({ project: PRJ }, 'succeeded');
+ const failId = await brun({ project: PRJ, type: 'test.boom', files: [{ buffer: Buffer.from('x'), name: SECRET_FILE, mime: 'text/plain' }] }, 'failed');
+ const { job: held } = await bsys.submit({ owner: `user:${USER}`, type: 'test.upper', input: { hold: true } });
+ await until(async () => (await bsys.get(held.id)).state === 'running', 'held job running');
+ await bsys.cancel(held.id);
+ await until(async () => (await bsys.get(held.id)).state === 'cancelled', 'held job cancelled');
+ assert.deepStrictEqual((await readingRows()).map(r => [r.job_id, r.sent_at]), [[personId, null], [appId, null], [failId, null]], 'stored in the job-end transaction, unsent; a cancelled job has none');
+ assert.strictEqual(posts.length, 0, 'nothing is posted inside the job\'s transaction');
+
+ assert.deepStrictEqual(await bsys.flushReadings(), { posted: 3, refused: 0 });
+ assert.deepStrictEqual(tokens, ['openvibe.billing'], 'one token for the Billing audience, cached');
+ assert.deepStrictEqual(posts.map(p => p.idempotency_key), [personId, appId, failId].map(id => `tools:job:${id}`));
+ for (const p of posts) {
+ const v = contracts.validate('platform.usage-sample@1', p);
+ assert.ok(v.valid, JSON.stringify(v.errors));
+ assert.deepStrictEqual([p.service, p.source, p.quantity, p.unit], ['openvibe.tools', 'openvibe.tools', 1, 'jobs']);
+ }
+ const [person, appJob, failJob] = posts;
+ assert.deepStrictEqual([person.id, person.operation, person.resource, person.subject, person.trace_id, person.project], [`tools-job-${personId}`, 'tools.tool.run', 'image-resize', `user:${USER}`, TRACE, undefined]);
+ assert.deepStrictEqual([appJob.operation, appJob.resource, appJob.project, 'subject' in appJob], ['tools.job.create', 'test.upper', PRJ, false], 'an app\'s job carries its project, no subject');
+ assert.deepStrictEqual([failJob.operation, failJob.resource, failJob.quantity], ['tools.job.create', 'test.boom', 1], 'a failed job is billed');
+ const text = JSON.stringify(posts);
+ for (const s of [APP, SECRET_INPUT, SECRET_FILE, 'secret-output-name', 'Cannot read', held.id]) assert.ok(!text.includes(s), `no "${s}" in readings`);
+ assert.ok((await readingRows()).every(r => r.sent_at != null && Number(r.attempts) === 0));
+
+ // ── Once per job: a second flush and a re-run of the job's end add nothing ──
+ assert.deepStrictEqual(await bsys.flushReadings(), { posted: 0, refused: 0 });
+ await bdb.tx(async () => bsys.usage.finished(await bsys.get(personId)));
+ assert.strictEqual((await readingRows()).length, 3, 'a re-run stores no second reading');
+ assert.deepStrictEqual(await bsys.flushReadings(), { posted: 0, refused: 0 });
+ assert.strictEqual(posts.length, 3);
+ assert.strictEqual(readingOf(await bsys.get(personId)).idempotency_key, `tools:job:${personId}`, 'the key is stable');
+ assert.deepStrictEqual(readingOf(await bsys.get(personId)), person, 'the stored reading is the job\'s reading');
+ assert.strictEqual(readingOf(await bsys.get(held.id)), null);
+
+ // ── Refused: logged, never thrown into the job, kept unsent with the error for the next flush ──
+ mode = 'down';
+ const downId = await brun({ project: PRJ }, 'succeeded');
+ const downId2 = await brun({ project: PRJ }, 'succeeded');
+ assert.deepStrictEqual(await bsys.flushReadings(), { posted: 0, refused: 1 }, 'Billing down: stop after the first refusal');
+ let rows = await readingRows();
+ const down = rows.find(r => r.job_id === downId);
+ assert.deepStrictEqual([down.sent_at, Number(down.attempts)], [null, 1]);
+ assert.match(down.last_error, /^503 /);
+ assert.ok(errors.some(e => e.includes(`tools:job:${downId}`) && e.includes('503')), 'the refusal is logged');
+ mode = 'bad';
+ assert.deepStrictEqual(await bsys.flushReadings(), { posted: 0, refused: 2 }, 'a 4xx for one reading does not stop the others');
+ mode = 'ok';
+ assert.deepStrictEqual(await bsys.flushReadings(), { posted: 2, refused: 0 });
+ rows = await readingRows();
+ assert.ok(rows.every(r => r.sent_at != null), 'sent on the next flush');
+ assert.deepStrictEqual(posts.slice(3).map(p => p.idempotency_key).sort(), [downId, downId2].map(id => `tools:job:${id}`).sort());
+ assert.strictEqual((await bsys.get(downId)).state, 'succeeded', 'the job is untouched by the refusal');
+
+ // ── No grant yet (the token is refused): logged, never thrown ──
+ const noGrant = createBillingClient({ env: { OV_BILLING_URL: 'http://billing.test', OV_NETWORK_INTERNAL_URL: 'http://network.test' }, fetchImpl: billingFetch, log: blog });
+ mode = 'no-grant';
+ const res = await noGrant.post(person);
+ assert.deepStrictEqual([res.ok, res.status], [false, 401]);
+ mode = 'ok';
+ const st = (await bsys.stats()).billing;
+ assert.deepStrictEqual([st.pending, st.posted, st.refused, st.invalid], [0, 5, 3, 0]);
+ bsys.stop();
+ }
console.log('job usage: all checks passed');
console.log('job usage: all checks passed');
} finally {
} finally {
await closeAllTestDbs();
await closeAllTestDbs();
@@ -20,7 +20,7 @@
@@ -20,7 +20,7 @@
"iovalkey": "^0.4.0",
"iovalkey": "^0.4.0",
"jsonwebtoken": "^9.0.2",
"jsonwebtoken": "^9.0.2",
"multer": "^1.4.5-lts.1",
"multer": "^1.4.5-lts.1",
- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.79.0",
+ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.92.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.25.1",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.25.1",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.3.1",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.3.1",
"pg": "^8.23.0",
"pg": "^8.23.0",
@@ -1290,9 +1290,9 @@
@@ -1290,9 +1290,9 @@
}
}
},
},
"node_modules/openvibe-contracts": {
"node_modules/openvibe-contracts": {
- "version": "0.79.0",
- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.79.0",
- "integrity": "sha512-FY5GLbiOq0kDw5lUPPzX6h3ATbzvsr/mX5DUCF4GRfn6vGMP0p89rZZDhgtVDaS+yCBCYMZ8zFaZxuHEkEVwVg==",
+ "version": "0.92.0",
+ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.92.0",
+ "integrity": "sha512-2Wm1CdsNqIBou+XYAtIiIcQ6GyXWWh5miG81iFYrPvWGLVF1BUjEN4XxQGK2S2bC6TCKz7XQtp5IJx3Mg0vPIw==",
"license": "AGPL-3.0-only",
"license": "AGPL-3.0-only",
"dependencies": {
"dependencies": {
"ajv": "^8.17.1",
"ajv": "^8.17.1",
@@ -23,7 +23,7 @@
@@ -23,7 +23,7 @@
"helmet": "^8.1.0",
"helmet": "^8.1.0",
"jsonwebtoken": "^9.0.2",
"jsonwebtoken": "^9.0.2",
"multer": "^1.4.5-lts.1",
"multer": "^1.4.5-lts.1",
- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.79.0",
+ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.92.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.25.1",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.25.1",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.3.1",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.3.1",
"uuid": "^11.1.0"
"uuid": "^11.1.0"
@@ -19,7 +19,7 @@
@@ -19,7 +19,7 @@
"iovalkey": "^0.4.0",
"iovalkey": "^0.4.0",
"jsonwebtoken": "^9.0.2",
"jsonwebtoken": "^9.0.2",
"multer": "^1.4.5-lts.1",
"multer": "^1.4.5-lts.1",
- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.79.0",
+ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.92.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.25.1",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.25.1",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.3.1",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.3.1",
"pdf-lib": "^1.17.1",
"pdf-lib": "^1.17.1",
@@ -1367,9 +1367,9 @@
@@ -1367,9 +1367,9 @@
}
}
},
},
"node_modules/openvibe-contracts": {
"node_modules/openvibe-contracts": {
- "version": "0.79.0",
- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.79.0",
- "integrity": "sha512-FY5GLbiOq0kDw5lUPPzX6h3ATbzvsr/mX5DUCF4GRfn6vGMP0p89rZZDhgtVDaS+yCBCYMZ8zFaZxuHEkEVwVg==",
+ "version": "0.92.0",
+ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.92.0",
+ "integrity": "sha512-2Wm1CdsNqIBou+XYAtIiIcQ6GyXWWh5miG81iFYrPvWGLVF1BUjEN4XxQGK2S2bC6TCKz7XQtp5IJx3Mg0vPIw==",
"license": "AGPL-3.0-only",
"license": "AGPL-3.0-only",
"dependencies": {
"dependencies": {
"ajv": "^8.17.1",
"ajv": "^8.17.1",
@@ -22,7 +22,7 @@
@@ -22,7 +22,7 @@
"helmet": "^8.1.0",
"helmet": "^8.1.0",
"jsonwebtoken": "^9.0.2",
"jsonwebtoken": "^9.0.2",
"multer": "^1.4.5-lts.1",
"multer": "^1.4.5-lts.1",
- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.79.0",
+ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.92.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.25.1",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.25.1",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.3.1",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.3.1",
"pdf-lib": "^1.17.1",
"pdf-lib": "^1.17.1",
@@ -60,3 +60,7 @@ MEDIA_PUBLIC_URL=https://openvibe.media
@@ -60,3 +60,7 @@ MEDIA_PUBLIC_URL=https://openvibe.media
#EVENTS_URL=http://127.0.0.1:4300
#EVENTS_URL=http://127.0.0.1:4300
#EVENTS_PUBLISH=off
#EVENTS_PUBLISH=off
#EVENTS_RELAY_INTERVAL_MS=2000
#EVENTS_RELAY_INTERVAL_MS=2000
+# One usage reading per ended job → OpenVibe.Billing POST /api/v1/usage (off while unset). Needs OV_OAUTH_CLIENT_SECRET
+# and the Network grant [tools, billing.usage.record, openvibe.billing]; until then every post is 401 and only logged.
+#OV_BILLING_URL=http://127.0.0.1:4600
+#OV_BILLING_AUDIENCE=openvibe.billing
@@ -18,7 +18,7 @@
@@ -18,7 +18,7 @@
"helmet": "^8.1.0",
"helmet": "^8.1.0",
"iovalkey": "^0.4.0",
"iovalkey": "^0.4.0",
"jsonwebtoken": "^9.0.0",
"jsonwebtoken": "^9.0.0",
- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.79.0",
+ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.92.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.25.1",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.25.1",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.3.1",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.3.1",
"pg": "^8.23.0"
"pg": "^8.23.0"
@@ -1167,9 +1167,9 @@
@@ -1167,9 +1167,9 @@
}
}
},
},
"node_modules/openvibe-contracts": {
"node_modules/openvibe-contracts": {
- "version": "0.79.0",
- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.79.0",
- "integrity": "sha512-FY5GLbiOq0kDw5lUPPzX6h3ATbzvsr/mX5DUCF4GRfn6vGMP0p89rZZDhgtVDaS+yCBCYMZ8zFaZxuHEkEVwVg==",
+ "version": "0.92.0",
+ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.92.0",
+ "integrity": "sha512-2Wm1CdsNqIBou+XYAtIiIcQ6GyXWWh5miG81iFYrPvWGLVF1BUjEN4XxQGK2S2bC6TCKz7XQtp5IJx3Mg0vPIw==",
"license": "AGPL-3.0-only",
"license": "AGPL-3.0-only",
"dependencies": {
"dependencies": {
"ajv": "^8.17.1",
"ajv": "^8.17.1",
@@ -21,7 +21,7 @@
@@ -21,7 +21,7 @@
"express-rate-limit": "^7.5.0",
"express-rate-limit": "^7.5.0",
"helmet": "^8.1.0",
"helmet": "^8.1.0",
"jsonwebtoken": "^9.0.0",
"jsonwebtoken": "^9.0.0",
- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.79.0",
+ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.92.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.25.1",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.25.1",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.3.1"
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.3.1"
}
}
@@ -19,7 +19,7 @@
@@ -19,7 +19,7 @@
"iovalkey": "^0.4.0",
"iovalkey": "^0.4.0",
"jsonwebtoken": "^9.0.2",
"jsonwebtoken": "^9.0.2",
"multer": "^1.4.5-lts.1",
"multer": "^1.4.5-lts.1",
- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.79.0",
+ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.92.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.25.1",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.25.1",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.3.1",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.3.1",
"pg": "^8.23.0",
"pg": "^8.23.0",
@@ -1858,9 +1858,9 @@
@@ -1858,9 +1858,9 @@
}
}
},
},
"node_modules/openvibe-contracts": {
"node_modules/openvibe-contracts": {
- "version": "0.79.0",
- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.79.0",
- "integrity": "sha512-FY5GLbiOq0kDw5lUPPzX6h3ATbzvsr/mX5DUCF4GRfn6vGMP0p89rZZDhgtVDaS+yCBCYMZ8zFaZxuHEkEVwVg==",
+ "version": "0.92.0",
+ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.92.0",
+ "integrity": "sha512-2Wm1CdsNqIBou+XYAtIiIcQ6GyXWWh5miG81iFYrPvWGLVF1BUjEN4XxQGK2S2bC6TCKz7XQtp5IJx3Mg0vPIw==",
"license": "AGPL-3.0-only",
"license": "AGPL-3.0-only",
"dependencies": {
"dependencies": {
"ajv": "^8.17.1",
"ajv": "^8.17.1",
@@ -22,7 +22,7 @@
@@ -22,7 +22,7 @@
"helmet": "^8.1.0",
"helmet": "^8.1.0",
"jsonwebtoken": "^9.0.2",
"jsonwebtoken": "^9.0.2",
"multer": "^1.4.5-lts.1",
"multer": "^1.4.5-lts.1",
- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.79.0",
+ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.92.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.25.1",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.25.1",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.3.1",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.3.1",
"sharp": "^0.33.5",
"sharp": "^0.33.5",
new file mode 100644
new file mode 100644
@@ -0,0 +1,74 @@
@@ -0,0 +1,74 @@
+# Cutover runbook — Tools job usage readings to OpenVibe.Billing (plan T5 step 7)
+
+This PR adds one empty table, `tool_job_billing_readings`, and the code that fills it: every tool job that
+succeeds or fails stores one `platform.usage-sample@1` reading in the transaction that records its end, and
+the jobs tick posts the unsent ones to OpenVibe.Billing `POST /api/v1/usage`. It also bumps
+`openvibe-contracts` from v0.79.0 to v0.92.0 in the img, audio, docs and gateway apps. Nothing existing is
+renamed, dropped, altered or backfilled; the hourly `tools.usage.recorded` rollups (`tool_job_usage`) are
+unchanged. No money moves here: Billing stores the readings and rates them in its own sweep.
+
+## What the migration does
+
+`migrations/0005_billing_readings.sql` (`phase: expand`) creates the new table and two partial indexes:
+
+```sql
+CREATE TABLE IF NOT EXISTS tool_job_billing_readings (
+ job_id text PRIMARY KEY, app text NOT NULL DEFAULT '', reading jsonb NOT NULL,
+ created_at bigint NOT NULL, sent_at bigint, attempts integer NOT NULL DEFAULT 0, last_error text
+);
+CREATE INDEX IF NOT EXISTS tool_job_billing_readings_unsent ON tool_job_billing_readings (app, attempts, created_at) WHERE sent_at IS NULL;
+CREATE INDEX IF NOT EXISTS tool_job_billing_readings_sent ON tool_job_billing_readings (sent_at) WHERE sent_at IS NOT NULL;
+```
+
+The runner is `openvibe-sdk/db`'s `migrate()`, started by `apps/_shared/db.js` as the owner
+(`DATABASE_DIRECT_URL`) when the first app of the new release boots: it locks `ov_migrations`, runs the file
+in one transaction, records id `0005` with its checksum, and refuses a later edit of it. The other apps of the
+release wait on the same lock and find it applied.
+
+## Order
+
+1. **Backup.** Take the logical dump of the `tools` database before the deploy (the pre-merge rule for any
+ data change, even an additive one): `sudo ovhost backup --all --logical`. Note the stamp it prints; the
+ way back uses it.
+2. **Merge and deploy** through the pipeline (`ovhost deploy tools`, `--wait-idle` is not needed: no job
+ state changes shape). The first app to boot applies `0005`.
+3. **Leave `OV_BILLING_URL` unset at first.** With it unset the code is inert: nothing is stored in the new
+ table, nothing is posted, the rollups and job events behave as before.
+4. **Network grant** (OpenVibe.Network, its own PR): the `tools` client needs
+ `billing.usage.record` on audience `openvibe.billing`. Without it every token request for that audience is
+ refused, each post is logged as `[Billing] reading tools:job:<id> not sent: token: 401 …`, and the readings
+ wait in the table (no job is affected).
+5. **Turn it on** once the grant is live: add `OV_BILLING_URL` (Billing's internal URL) and, only if it
+ differs, `OV_BILLING_AUDIENCE` to `/etc/openvibe/tools.env`, then restart through `ovhost deploy tools
+ --restart`.
+
+## Verification
+
+- `ov access run openvibe-ovh health tools` (or `/api/ready` on each app) answers 200 after the deploy.
+- The migration is recorded: `SELECT id, name, phase FROM ov_migrations WHERE id = '0005'` returns
+ `0005 | billing_readings | expand`, and `SELECT COUNT(*) FROM tool_job_billing_readings` is 0 while
+ `OV_BILLING_URL` is unset.
+- After step 5: `GET /api/health` on img, audio or docs shows `jobs.billing` with `pending` going back to 0
+ after a tick (5 minutes) and `posted` growing; `refused` with a `last_error` of `401 …` means the grant is
+ missing. `SELECT COUNT(*) FROM tool_job_billing_readings WHERE sent_at IS NULL` stays small.
+- In Billing: `GET /api/v1/usage?service=openvibe.tools` (`billing.ledger.admin`) lists the readings, one per
+ ended job, `idempotency_key` `tools:job:<job id>`.
+
+```rehearse
+# 0005 is additive (one new table, no backfill): it must exist, empty, with its two indexes, next to the
+# untouched tool_jobs, tool_job_usage and event_outbox.
+node -e 'const a=require("assert");const {dep}=require("./apps/_shared/test/deps");const db=dep("openvibe-sdk/db").createDb({url:process.env.DATABASE_DIRECT_URL,service:"rehearsal",max:1});(async()=>{const c=await db.prepare("SELECT column_name AS c FROM information_schema.columns WHERE table_name = ? ORDER BY ordinal_position").all("tool_job_billing_readings");a.deepStrictEqual(c.map(r=>r.c),["job_id","app","reading","created_at","sent_at","attempts","last_error"]);a.strictEqual(Number((await db.prepare("SELECT COUNT(*) AS n FROM tool_job_billing_readings").get()).n),0);const i=await db.prepare("SELECT indexname AS i FROM pg_indexes WHERE tablename = ? ORDER BY indexname").all("tool_job_billing_readings");a.deepStrictEqual(i.map(r=>r.i),["tool_job_billing_readings_pkey","tool_job_billing_readings_sent","tool_job_billing_readings_unsent"]);for(const t of ["tool_jobs","tool_job_usage","event_outbox"])a.ok(await db.prepare("SELECT to_regclass(?) AS r").get(t).then(r=>r.r),t);console.log("0005: tool_job_billing_readings created empty with its two indexes; tool_jobs, tool_job_usage and event_outbox untouched")})().finally(()=>db.close()).catch((e)=>{console.error(e);process.exitCode=1})'
+```
+
+## Rollback
+
+- **Code only** (the usual way back): `ovhost rollback tools` (or `deploy/scripts/deploy.sh --rollback`). The
+ previous release does not know the table and ignores it; the migration stays applied, which the previous
+ release's runner accepts (an applied id it does not have is not an error). Unsent readings stay in the table
+ and are posted when the release comes back (Billing dedupes on `idempotency_key`).
+- **Stop posting without a rollback:** remove `OV_BILLING_URL` from `/etc/openvibe/tools.env` and restart.
+- **Remove the table** (only if it must go; a contract step, by hand as the owner, after a code rollback):
+ `DROP TABLE tool_job_billing_readings; DELETE FROM ov_migrations WHERE id = '0005';`. Readings not yet posted
+ are lost with it; Billing keeps the ones it stored.
+- **Restore:** the dump from step 1 restores the database as it was before the deploy (`pg_restore` per
+ OpenVibe.Host `docs/backups.md`); not needed for this additive change unless the database itself is damaged.
new file mode 100644
new file mode 100644
@@ -0,0 +1,15 @@
@@ -0,0 +1,15 @@
+-- phase: expand
+-- tool_job_billing_readings (plan T5 step 7): one platform.usage-sample@1 reading per ended job
+-- (succeeded or failed), written in the transaction that records the job's end and posted to
+-- OpenVibe.Billing from the jobs tick (apps/_shared/jobs/usage.js). Additive: a new empty table.
+CREATE TABLE IF NOT EXISTS tool_job_billing_readings (
+ job_id text PRIMARY KEY,
+ app text NOT NULL DEFAULT '',
+ reading jsonb NOT NULL,
+ created_at bigint NOT NULL,
+ sent_at bigint,
+ attempts integer NOT NULL DEFAULT 0,
+ last_error text
+);
+CREATE INDEX IF NOT EXISTS tool_job_billing_readings_unsent ON tool_job_billing_readings (app, attempts, created_at) WHERE sent_at IS NULL;
+CREATE INDEX IF NOT EXISTS tool_job_billing_readings_sent ON tool_job_billing_readings (sent_at) WHERE sent_at IS NOT NULL;