openvibers/openvibe.watch Commits
Queue one check-operation usage reading per completed check for Billing (#54)
Changes
@@@ -105,3 +105,12 @@ VALKEY_PREFIX=# nobody can use it. A credential needs owners and the origins it may be sent to; a webhook key, owners.# Owners are usr_ or agt_ subjects; origins are scheme://host[:port].#WATCH_SECRET_GRANTS={"WATCH_CRED_EXAMPLE_API_KEY":{"owners":["usr_..."],"origins":["https://api.example.com"]},"WATCH_HOOK_EXAMPLE_SECRET":{"owners":["usr_..."]}}++++# Usage readings → OpenVibe.Billing (server/billing.js): one check-operation reading per completed check, relayed++# every WATCH_BILLING_INTERVAL_MS (0, the default, is off: nothing is queued) with this service's client credentials++# (OV_OAUTH_CLIENT_ID/OV_OAUTH_CLIENT_SECRET above; Network grants billing.usage.record). Without the URL or the++# secret, readings are queued and stay local.++#OV_BILLING_INTERNAL_URL=++#OV_BILLING_AUDIENCE=++#WATCH_BILLING_TIMEOUT_MS=++#WATCH_BILLING_INTERVAL_MS=@@@ -3,6 +3,15 @@What shipped on OpenVibe.Watch, newest first. The plan is T18 ("Watch complete"); each entry namesthe step it landed.++## 2026-10-13 — check-operation usage readings (plan T5 step 14, Watch slice)++++- A completed check queues one `platform.usage-sample@1` reading (metric `check-operation`, quantity 1, unit `check`,++ operation and provider the carrier, subject the owner, project only when the watch names one, cost the run's++ `cost_usd`) in the transaction that records its run, keyed `watch:check:<run id>`; a failed check is not billed.++- `server/billing.js` relays them to OpenVibe.Billing (`POST /api/v1/usage`, `billing.usage.record`) with the++ `openvibe-sdk/usage` reporter; 400/422 rejects a reading, 429/5xx keeps it pending. `/api/ready` reports its counts.++- Off by default (`WATCH_BILLING_INTERVAL_MS=0`). Migration `0008_billing_readings.sql`, `test/billing.test.js`.++## 2026-10-08 — account export and deletion (ADR-033)- `POST /internal/events` (loopback, `WATCH_EVENTS_SECRET`) answers network.account.export_requested and@@@ -418,6 +418,7 @@ Reporting: [SECURITY.md](SECURITY.md). The rules the code keeps:- OpenVibe.Network (signing key; service principal `watch`; grants `network.account.export.contribute` and`network.account.deletion.confirm` for account export and deletion, given last, once this release is live)- OpenVibe.Events (outbound events; the two account subscriptions, created at boot with `events.subscription.manage`)++- OpenVibe.Billing (one `check-operation` usage reading per completed check, `POST /api/v1/usage` with `billing.usage.record`; off by default, `WATCH_BILLING_INTERVAL_MS`)## Capabilities@@@ -79,7 +79,8 @@"GET /api/v1/resources and GET /api/v1/resources/:ovrn (watch.resource.read)","watch.* events through openvibe-sdk's service_outbox in the same transaction as the state change","per-actor limits (openvibe-sdk/limits)",++ "usage readings (check operations): one platform.usage-sample@1 check-operation reading per completed check run, relayed to OpenVibe.Billing (billing.usage.record) by openvibe-sdk/usage; off by default (WATCH_BILLING_INTERVAL_MS)","the public site: Network sign-in (PKCE S256, web_sessions with only the token hash), server-rendered pages complete without JavaScript, the four create templates, watch management, observation history and check runs, how-it-works and the changelog, robots/sitemap/llms.txt"],-- "updated": "2026-10-12"++ "updated": "2026-10-13"}nnew file mode 100644@@@ -0,0 +1,40 @@++-- phase: expand++-- Plan T5 step 14 (govern universal), Watch slice: the outbox of platform.usage-sample@1 readings Watch posts to++-- OpenVibe.Billing (POST /api/v1/usage, capability billing.usage.record; server/billing.js), one per completed++-- check run (metric check-operation).++--++-- billing_readings is the openvibe-sdk/usage createUsageReporter outbox (outboxSchema('billing_readings')): one row per++-- reading, `event_id` is the reading's idempotency_key (watch:check:<check run id>; unique, so a retried or replayed++-- run inserts nothing, ON CONFLICT DO NOTHING) and `envelope` is the reading JSON. The reporter only ever moves the++-- sending columns (attempts, next_attempt_at, sent_at, seq, rejected_at, last_error); the trigger below refuses any++-- change to the reading itself, so a retry always posts the same body and Billing's idempotency key sees the same++-- reading. A check is counted, not aggregated: no period table is needed.++-- Additive only: one new table, its indexes and its trigger; nothing existing changes.++CREATE TABLE billing_readings (++ id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,++ event_id text NOT NULL UNIQUE,++ envelope jsonb NOT NULL,++ traceparent text,++ created_at bigint NOT NULL,++ attempts integer NOT NULL DEFAULT 0,++ next_attempt_at bigint NOT NULL DEFAULT 0,++ sent_at bigint,++ seq bigint,++ rejected_at bigint,++ last_error text,++ -- `event_id` IS the reading's idempotency_key; this generated column names it so a reader can say so.++ idempotency_key text GENERATED ALWAYS AS (event_id) STORED++);++CREATE INDEX IF NOT EXISTS billing_readings_due ON billing_readings (next_attempt_at, id) WHERE sent_at IS NULL AND rejected_at IS NULL;++CREATE INDEX IF NOT EXISTS billing_readings_sent ON billing_readings (sent_at) WHERE sent_at IS NOT NULL;++++-- A reading is never edited: only its delivery columns move. event_id/envelope/created_at are the reading.++CREATE FUNCTION billing_readings_frozen() RETURNS trigger LANGUAGE plpgsql AS $$++BEGIN++ IF NEW.event_id IS DISTINCT FROM OLD.event_id OR NEW.envelope IS DISTINCT FROM OLD.envelope++ OR NEW.created_at IS DISTINCT FROM OLD.created_at THEN++ RAISE EXCEPTION 'billing_readings: reading % is never edited', OLD.event_id;++ END IF;++ RETURN NEW;++END $$;++CREATE TRIGGER billing_readings_frozen BEFORE UPDATE ON billing_readings FOR EACH ROW EXECUTE FUNCTION billing_readings_frozen();@@@ -58,7 +58,7 @@ const CSP = "default-src 'self'; script-src 'self' 'unsafe-inline' https://openv+ "connect-src 'self' https://openvibe.network https://openvibe.events https://cloudflareinsights.com; "+ "frame-src 'self' https://openvibe.network; frame-ancestors 'none'; object-src 'none'; base-uri 'self'; form-action 'self' https://openvibe.network";--function createApp({ config, db, registry, check, observations, scheduler, auth, outbox, actions = null, now, log = console, limitsNow = null, sessions, keyStore, siteLimits, fetchImpl = globalThis.fetch, accountData = null, accountSend = null, eventSources = null, webhookSources = null }) {++function createApp({ config, db, registry, check, observations, scheduler, auth, outbox, actions = null, now, log = console, limitsNow = null, sessions, keyStore, siteLimits, fetchImpl = globalThis.fetch, accountData = null, accountSend = null, eventSources = null, webhookSources = null, billing = null }) {const app = express();app.disable('x-powered-by');app.set('trust proxy', config.trustProxy != null ? config.trustProxy : 'loopback');@@@ -98,7 +98,7 @@ function createApp({ config, db, registry, check, observations, scheduler, auth,// Readiness (openvibe-shared/ready): 503 only when the database fails; the Network key and the// check worker (running, queue keeping up) are optional and degrade it (see observability.js).-- const readiness = createWatchReadiness({ db, config, registry, check, observations, scheduler, outbox, actions, now, release: release.release });++ const readiness = createWatchReadiness({ db, config, registry, check, observations, scheduler, outbox, actions, billing, now, release: release.release });app.get('/api/ready', readiness.handler);// GET /release.json (ADR-016) and POST /release-metrics (open tabs' update reports into /metrics).release.mount(app, { registry: metrics.registry });nnew file mode 100644@@@ -0,0 +1,121 @@++'use strict';++/**++ * Watch's check-operation usage readings → OpenVibe.Billing (plan T5 step 14, Watch slice).++ *++ * Watch emits one platform.usage-sample@1 metric, check-operation:++ *++ * check-operation one reading per completed check run (a check_runs row whose state is not a failure, see++ * check.js FAILURE_STATES): quantity 1, unit 'check', operation and provider the carrier++ * (http/feed/api), subject the watch's owner, project the watch's project_id only when it++ * names one, cost_estimate the run's recorded cost_usd (0 when it records none), at the run's++ * finish. Idempotency key watch:check:<check run id>, so a retried or replayed run is never++ * billed twice. A failed check is not billed.++ *++ * The reporter is the SDK's own (openvibe-sdk/usage createUsageReporter): the check engine queues the reading++ * idempotency-keyed in the outbox inside the transaction that records the run (check.js settle), and the reporter++ * relays it to Billing's billing.usage.record (POST /api/v1/usage) with Watch's Network service token++ * (createServiceTokenClient, clientId 'watch', audience openvibe.billing). A relay that cannot reach Billing (no grant++ * yet, a missing route, a timeout, backpressure, a 401/403/404/408/425/429 or a 5xx) stays pending with backoff; only++ * Billing refusing the reading itself (any other 4xx) is marked rejected and never retried. A reading is never edited++ * (migrations/0008_billing_readings.sql).++ *++ * Off by default: WATCH_BILLING_INTERVAL_MS=0 starts no timer and queues nothing. With an interval but no++ * OV_BILLING_INTERNAL_URL (or OV_OAUTH_CLIENT_SECRET) readings are queued and stay local.++ *++ * const billing = createWatchBilling({ db, config: config.billing, networkUrl: config.networkInternalUrl });++ * if (billing.enabled) billing.start();++ */++const { createUsageReporter } = require('openvibe-sdk/usage');++const { createServiceTokenClient } = require('openvibe-sdk/auth');++++const TABLE = 'billing_readings';++const METRIC = 'check-operation';++++/**++ * @param {object} o++ * @param {object} o.db the Watch database (openvibe-sdk/db)++ * @param {object} o.config config.billing (server/config.js)++ * @param {string} [o.networkUrl] where Watch's service token comes from (OV_NETWORK_INTERNAL_URL)++ * @param {{ now(): number }} [o.clock]++ * @param {Function} [o.fetchImpl]++ * @param {object} [o.log]++ * @param {object} [o.reporter] a ready createUsageReporter (tests); default one over `db`/`config`++ */++function createWatchBilling({ db, config, networkUrl, clock = { now: () => Date.now() }, fetchImpl = globalThis.fetch, log = console, reporter: injected } = {}) {++ const intervalMs = Number(config && config.intervalMs) || 0;++ const now = () => clock.now();++ let tokenClient = null;++ if (config && config.url && config.clientSecret) {++ tokenClient = createServiceTokenClient({++ network: networkUrl, clientId: config.clientId, clientSecret: config.clientSecret,++ audience: config.audience, fetch: fetchImpl, timeoutMs: config.timeoutMs, now,++ });++ }++ const reporter = injected || createUsageReporter({++ db, service: 'watch', source: 'watch.check', table: TABLE,++ billingUrl: config && config.url, tokenClient, audience: config && config.audience,++ fetchImpl, timeoutMs: (config && config.timeoutMs) || 5000, now, log,++ });++ let lastError = null;++++ /** The reading of one completed check run (`run` as check.js records it, `watch` as the registry reads it). */++ function readingOf({ watch, run }) {++ const cost = Number(run.cost_usd);++ return reporter.sample({++ idempotency_key: reporter.key('check', run.id),++ project: watch.project_id || undefined,++ subject: watch.owner || undefined,++ resource: METRIC, provider: run.carrier, operation: run.carrier,++ quantity: 1, unit: 'check',++ cost_estimate: cost > 0 ? cost : 0,++ at: new Date(Number(run.finished_at)).toISOString(),++ });++ }++++ /** INSIDE the caller's transaction: queue the run's reading. -> true when inserted, false when already queued. */++ async function queue(t, { watch, run }) {++ return await reporter.record(t, readingOf({ watch, run }));++ }++++ /** One relay pass. Never overlaps itself; never throws. */++ let flushing = null;++ function flush() {++ if (!flushing) flushing = reporter.flush()++ .catch((err) => { lastError = err.message; log.error(`[billing] flush failed: ${err.message}`); return null; })++ .finally(() => { flushing = null; });++ return flushing;++ }++++ let timer = null;++ return {++ enabled: intervalMs > 0,++ reporting: !!tokenClient,++ reporter,++ readingOf, queue, flush,++ start() {++ if (timer || !(intervalMs > 0)) return false;++ if (!tokenClient) log.warn('[billing] OV_BILLING_INTERNAL_URL or OV_OAUTH_CLIENT_SECRET is not set: readings are stored and stay queued');++ timer = setInterval(flush, intervalMs);++ if (timer.unref) timer.unref();++ return true;++ },++ async stop() {++ if (timer) clearInterval(timer);++ timer = null;++ if (flushing) await flushing;++ await reporter.stop();++ },++ get running() { return !!timer; },++ async status() {++ return {++ enabled: intervalMs > 0,++ reporting: !!tokenClient,++ pending: await reporter.pending(),++ rejected: await reporter.rejected(),++ last_error: lastError,++ };++ },++ };++}++++module.exports = { createWatchBilling, TABLE, METRIC };@@@ -16,6 +16,8 @@* NOT dispatched here: a met condition records the trigger and emits the event, nothing else* 7 arm next_due_at (+ jitter, or exponential backoff after a failure, capped) and not_before* (Retry-After)++ * 8 a completed (not failed) check queues one check-operation usage reading for Billing with its run++ * (server/billing.js; off unless WATCH_BILLING_INTERVAL_MS is set)** The run row, the observation and every event commit in ONE transaction: an event exists exactly* when its effect does. Nothing polls a source whose condition is 'changed' twice for the same@@@ -34,7 +36,7 @@ const FAILURE_STATES = new Set(['http_error', 'timeout', 'parse_error', 'rate_liconst PERSON_OWNER = /^(usr|agt)_[0-9A-HJKMNP-TV-Z]{26}$/;const USER_OWNER = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;--function createCheck({ db, registry, config, carriers, extract, condition, observations, outbox, actions = null, now = () => Date.now(), log = console, ids = contractIds }) {++function createCheck({ db, registry, config, carriers, extract, condition, observations, outbox, actions = null, billing = null, now = () => Date.now(), log = console, ids = contractIds }) {const inflight = new Set();const st = {@@@ -163,9 +165,12 @@ function createCheck({ db, registry, config, carriers, extract, condition, obser*/async function settle({ watch, run, failures, successAt = null, notBefore = 0, events = [], observation = null, endpoint = null, fired = null }) {const armed = nextDue(watch, run.finished_at, { failures, maxBackoffMs: config.worker.maxBackoffMs });-- await db.tx(async () => {++ await db.tx(async (t) => {if (observation) await observations.insert(observation);await st.insertRun.run(run);++ // A completed check is one check-operation reading for Billing, queued with its run (server/billing.js);++ // a failed one is not billed, and the run id keys it, so a replayed run inserts nothing.++ if (billing && billing.enabled && run.carrier && !FAILURE_STATES.has(run.state)) await billing.queue(t, { watch, run });if (endpoint) await st.saveEndpoint.run(endpoint);await st.finish.run({id: watch.id, at: run.finished_at, success_at: successAt, state: run.state,@@@ -158,6 +158,19 @@ function load(env = process.env) {},},++ // Usage readings → OpenVibe.Billing (server/billing.js, plan T5 step 14): one check-operation reading per++ // completed check, relayed every WATCH_BILLING_INTERVAL_MS with Watch's own client-credentials token. 0 (the++ // default) is off: nothing is queued and no timer starts. Without OV_BILLING_INTERNAL_URL or the client secret,++ // readings are queued and stay local.++ billing: {++ url: strip(env.OV_BILLING_INTERNAL_URL || ''),++ clientId: env.OV_OAUTH_CLIENT_ID || 'watch',++ clientSecret: env.OV_OAUTH_CLIENT_SECRET || '',++ audience: env.OV_BILLING_AUDIENCE || 'openvibe.billing',++ timeoutMs: int(env.WATCH_BILLING_TIMEOUT_MS, 5000),++ intervalMs: int(env.WATCH_BILLING_INTERVAL_MS, 0),++ },++secrets: env,};}@@@ -24,6 +24,7 @@ const { createCarriers } = require('./carriers');const { extract } = require('./extract');const condition = require('./condition'); // { evaluate, hasChanged, holds, … }const { createCheck } = require('./check');++const { createWatchBilling } = require('./billing');const { createScheduler } = require('./scheduler');const { createEventSources } = require('./event-sources');const { createWebhookSources } = require('./webhook-sources');@@@ -82,7 +83,10 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetchconst runJobs = client ? createRunJobs(client, { baseUrl: config.actions.runUrl }) : null;const codesTasks = client ? createCodesTasks(client, { baseUrl: config.actions.codesUrl }) : null;const actions = createActions({ db, config, notifications: givenNotifications || notifications, post, actor: givenActor || actorTasks, run: givenRun || runJobs, codes: givenCodes || codesTasks, now, log });-- const check = createCheck({ db, registry, config, carriers, extract, condition, observations, outbox, actions, now, log });++ // Usage readings (plan T5 step 14): a completed check queues one check-operation reading for Billing in its own++ // commit; off unless WATCH_BILLING_INTERVAL_MS is set (server/billing.js).++ const billing = createWatchBilling({ db, config: config.billing, networkUrl: config.networkInternalUrl, clock: { now }, fetchImpl, log });++ const check = createCheck({ db, registry, config, carriers, extract, condition, observations, outbox, actions, billing, now, log });const scheduler = createScheduler({ db, check, config, now, log });// Event sources (plan T18 step 5): watches woken by OpenVibe.Events events, with their subscriptions kept in step.const eventSources = createEventSources({ db, check, config, fetchImpl, log, now });@@@ -101,7 +105,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetchconst accountSend = givenSend || (config.oauth.clientSecret? createNetworkSender({ networkInternalUrl: config.networkInternalUrl, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, fetch: fetchImpl }): null);-- const app = createApp({ config, db, registry, check, observations, scheduler, auth, outbox, actions, now, log, limitsNow, sessions, keyStore, siteLimits, accountData, accountSend, eventSources, webhookSources });++ const app = createApp({ config, db, registry, check, observations, scheduler, auth, outbox, actions, now, log, limitsNow, sessions, keyStore, siteLimits, accountData, accountSend, eventSources, webhookSources, billing });// One JWKS client for the process (the SDK shares it with verifyUserToken): refresh in the background// on an unref'd timer, keeping the last good keys through a Network outage.keys.start();@@@ -109,6 +113,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetchoutbox.start();if (config.worker.enabled) scheduler.start();actions.start();++ if (billing.enabled) billing.start();const pruneTimer = setInterval(async () => {try { await observations.prune(); } catch (err) { log.error(`[observations] prune: ${err.message}`); }try { await eventSources.prune(); } catch (err) { log.error(`[event-sources] prune: ${err.message}`); }@@@ -142,6 +147,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetchkeys.stop();await scheduler.stop();await actions.stop();++ await billing.stop();await outbox.stop();if (server) {server.closeAllConnections?.();@@@ -150,7 +156,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetchif (!givenDb) await db.close();}-- return { config, db, registry, eventSources, webhookSources, observations, guard, fetcher, spacer, carriers, extract, condition, check, scheduler, outbox, actions, keys, keyLoaded, auth, sessions, keyStore, siteLimits, accountData, app, server, close };++ return { config, db, registry, eventSources, webhookSources, observations, guard, fetcher, spacer, carriers, extract, condition, check, scheduler, outbox, actions, billing, keys, keyLoaded, auth, sessions, keyStore, siteLimits, accountData, app, server, close };}/**@@@ -48,7 +48,7 @@ function readers(db, { startedAt }) {};}--function createWatchReadiness({ db, config, registry, check, observations, scheduler, outbox, actions = null, now, release = null }) {++function createWatchReadiness({ db, config, registry, check, observations, scheduler, outbox, actions = null, billing = null, now, release = null }) {const read = readers(db, { startedAt: now() });const iso = (v) => (v == null ? null : new Date(v).toISOString());return createReadiness({@@@ -99,6 +99,7 @@ function createWatchReadiness({ db, config, registry, check, observations, schedobservations: dbOk ? await read.observations() : null,outbox: dbOk ? await outbox.status() : null,actions: dbOk && actions ? await actions.status() : null,++ billing: dbOk && billing ? await billing.status() : null,};},});nnew file mode 100644@@@ -0,0 +1,164 @@++'use strict';++/**++ * Usage readings → Billing (plan T5 step 14, Watch slice; server/billing.js, migrations/0008_billing_readings.sql):++ * a completed check queues exactly one check-operation reading with its run, keyed by the run id, so a replay inserts++ * nothing; a failed check queues nothing; a watch without a project sends no project; WATCH_BILLING_INTERVAL_MS=0++ * starts no timer; only Billing refusing the reading itself (400/422) is permanent, a 429 or 5xx stays pending; and no++ * secret or token reaches a log line.++ */++const assert = require('assert');++const { validate } = require('openvibe-contracts');++const { boot, site, suite, newUser } = require('./helpers');++const { createWatchBilling, METRIC } = require('../server/billing');++const { load } = require('../server/config');++++const t = suite('billing');++let svc;++let web;++++const P = 'prj_01JAB2C3D4E5F6G7H8J9K0MNPS';++const person = { sub: newUser() };++const TOKEN = 'not-a-real-token-' + 'x'.repeat(12);++const CLIENT_SECRET = 'billing-client-secret-456';++const lines = [];++const capture = (...a) => lines.push(a.map(String).join(' '));++const log = { log: capture, warn: capture, error: capture };++const projects = { async check(project, subject) { return { ok: project === P && subject === person.sub }; } };++++const def = (overrides = {}) => ({++ name: 'price watch',++ source: { kind: 'http', url: `${web.origin}/price`, format: null },++ cadence: { every_sec: 900 },++ extraction: { kind: 'text' },++ condition: { op: 'changed' },++ action: [{ kind: 'notification' }],++ ...overrides,++});++const json = (v) => (typeof v === 'string' ? JSON.parse(v) : v);++const readings = async () => (await svc.db.many('SELECT * FROM billing_readings ORDER BY id')).map((r) => ({ ...r, envelope: json(r.envelope) }));++const rowOf = (key) => svc.db.prepare('SELECT * FROM billing_readings WHERE event_id = ?').get(key);++++t('boot', async () => {++ web = await site({++ '/price': () => ({ headers: { 'content-type': 'text/plain' }, body: '19.99' }),++ '/down': () => ({ status: 500, body: 'oops' }),++ });++ svc = await boot({ log, projects, env: { WATCH_BILLING_INTERVAL_MS: '3600000' } });++ assert.strictEqual(svc.billing.enabled, true);++ assert.strictEqual(svc.billing.reporting, false, 'no OV_BILLING_INTERNAL_URL: readings stay queued');++});++++t('a completed check queues exactly one check-operation reading with its run', async () => {++ const w = await svc.registry.create(def(), person);++ const out = await svc.check.run(w.id, { trigger: 'manual' });++ assert.ok(!['http_error', 'timeout'].includes(out.run.state), out.run.state);++ const rows = (await readings()).filter((r) => r.envelope.idempotency_key === `watch:check:${out.run.id}`);++ assert.strictEqual(rows.length, 1);++ const r = rows[0].envelope;++ assert.deepStrictEqual(r, {++ id: `watch:check:${out.run.id}`, idempotency_key: `watch:check:${out.run.id}`, service: 'watch', subject: person.sub,++ resource: METRIC, provider: 'http', operation: 'http', quantity: 1, unit: 'check',++ at: new Date(Number(out.run.finished_at)).toISOString(), cost_estimate: 0, source: 'watch.check',++ });++ assert.ok(!('project' in r), 'a watch without a project sends no project');++ const v = validate('platform.usage-sample@1', r);++ assert.ok(v.valid, JSON.stringify(v.errors));++ assert.strictEqual((await svc.billing.status()).pending, (await readings()).length);++ // Readiness (loopback) reports the relay: counts only, never the URL or a credential.++ const ready = await (await fetch(`${svc.base}/api/ready`)).json();++ assert.deepStrictEqual(ready.billing, { enabled: true, reporting: false, pending: (await readings()).length, rejected: 0, last_error: null });++});++++t('a replay of the same run inserts nothing', async () => {++ const w = await svc.registry.create(def(), person);++ const out = await svc.check.run(w.id, { trigger: 'manual' });++ const before = (await readings()).length;++ const watch = svc.registry.fromRow(await svc.db.prepare('SELECT * FROM watches WHERE id = ?').get(w.id));++ const run = await svc.db.prepare('SELECT * FROM check_runs WHERE id = ?').get(out.run.id);++ const inserted = await svc.db.tx((tx) => svc.billing.queue(tx, { watch, run }));++ assert.strictEqual(inserted, false);++ assert.strictEqual((await readings()).length, before);++});++++t('a watch with a project names it; the run cost becomes cost_estimate', async () => {++ const w = await svc.registry.create(def({ project_id: P }), person);++ const out = await svc.check.run(w.id, { trigger: 'manual' });++ const row = await rowOf(`watch:check:${out.run.id}`);++ assert.strictEqual(json(row.envelope).project, P);++ const costed = svc.billing.readingOf({ watch: { project_id: P, owner: person.sub }, run: { ...out.run, cost_usd: '0.25' } });++ assert.strictEqual(costed.cost_estimate, 0.25);++});++++t('a failed check queues nothing', async () => {++ const w = await svc.registry.create(def({ source: { kind: 'http', url: `${web.origin}/down`, format: null } }), person);++ const before = (await readings()).length;++ const out = await svc.check.run(w.id, { trigger: 'manual' });++ assert.strictEqual(out.run.state, 'http_error');++ assert.strictEqual((await readings()).length, before);++ assert.strictEqual(await rowOf(`watch:check:${out.run.id}`), undefined);++});++++t('WATCH_BILLING_INTERVAL_MS=0 (the default) starts no timer; an interval starts one and stop() clears it', async () => {++ const config = load({ NODE_ENV: 'test' }).billing;++ assert.strictEqual(config.intervalMs, 0);++ assert.strictEqual(config.clientId, 'watch');++ assert.strictEqual(config.audience, 'openvibe.billing');++ assert.strictEqual(config.timeoutMs, 5000);++ const off = createWatchBilling({ db: svc.db, config, log });++ assert.strictEqual(off.enabled, false);++ assert.strictEqual(off.start(), false);++ assert.strictEqual(off.running, false);++ await off.stop();++ const on = createWatchBilling({ db: svc.db, config: { ...config, intervalMs: 60000 }, log });++ assert.strictEqual(on.start(), true);++ assert.strictEqual(on.running, true);++ assert.strictEqual(on.start(), false, 'one timer');++ await on.stop();++ assert.strictEqual(on.running, false);++});++++t('Billing 5xx or 429 leaves a reading pending; 400/422 rejects it; no secret or token is logged', async () => {++ const FIXED = Date.UTC(2026, 9, 13, 12, 0, 0);++ // Push the check readings out of the relay's way: this part posts one reading at a time.++ await svc.db.exec('UPDATE billing_readings SET next_attempt_at = $1', [FIXED + 10 * 86400000]);++ const statuses = { 'k-500': 500, 'k-503': 503, 'k-429': 429, 'k-400': 400, 'k-422': 422 };++ const posted = [];++ const fetchImpl = async (url, opts) => {++ if (!String(url).startsWith('http://billing.test/')) {++ return { ok: true, status: 200, headers: new Headers({ 'content-type': 'application/json' }), json: async () => ({ access_token: TOKEN, token_type: 'Bearer', expires_in: 300 }), text: async () => '' };++ }++ assert.strictEqual(String(url), 'http://billing.test/api/v1/usage');++ assert.strictEqual(opts.headers.Authorization, `Bearer ${TOKEN}`);++ const sent = JSON.parse(opts.body);++ posted.push(sent.idempotency_key);++ const status = statuses[sent.idempotency_key];++ return { ok: false, status, headers: new Headers(), text: async () => `status ${status}`, json: async () => ({}) };++ };++ const talking = createWatchBilling({++ db: svc.db, clock: { now: () => FIXED }, fetchImpl, log, networkUrl: 'http://network.test',++ config: { url: 'http://billing.test', clientId: 'watch', clientSecret: CLIENT_SECRET, audience: 'openvibe.billing', timeoutMs: 2000, intervalMs: 1000 },++ });++ assert.strictEqual(talking.reporting, true);++ for (const [k, status] of Object.entries(statuses)) {++ const reading = talking.reporter.sample({ idempotency_key: k, subject: person.sub, resource: METRIC, provider: 'http', operation: 'http', quantity: 1, unit: 'check', at: new Date(FIXED).toISOString() });++ await svc.db.tx((tx) => talking.reporter.record(tx, reading));++ await talking.flush();++ const r = await rowOf(k);++ assert.ok(!r.sent_at, `${status} is not sent`);++ if (status === 400 || status === 422) assert.ok(r.rejected_at, `${status} refuses the reading`);++ else {++ assert.ok(!r.rejected_at, `${status} stays pending`);++ assert.ok(Number(r.next_attempt_at) > FIXED, `${status} backs off`);++ }++ }++ assert.deepStrictEqual(posted, Object.keys(statuses));++ const status = await talking.status();++ assert.strictEqual(status.rejected, 2);++ await talking.stop();++ const leaked = lines.filter((l) => l.includes(TOKEN) || l.includes(CLIENT_SECRET) || l.includes('watch-secret'));++ assert.deepStrictEqual(leaked, [], 'no secret or token in any log line');++});++++t('done', async () => { await web.close(); await svc.stop(); });++++t.run();
@@ -105,3 +105,12 @@ VALKEY_PREFIX=
@@ -105,3 +105,12 @@ VALKEY_PREFIX=
# nobody can use it. A credential needs owners and the origins it may be sent to; a webhook key, owners.
# nobody can use it. A credential needs owners and the origins it may be sent to; a webhook key, owners.
# Owners are usr_ or agt_ subjects; origins are scheme://host[:port].
# Owners are usr_ or agt_ subjects; origins are scheme://host[:port].
#WATCH_SECRET_GRANTS={"WATCH_CRED_EXAMPLE_API_KEY":{"owners":["usr_..."],"origins":["https://api.example.com"]},"WATCH_HOOK_EXAMPLE_SECRET":{"owners":["usr_..."]}}
#WATCH_SECRET_GRANTS={"WATCH_CRED_EXAMPLE_API_KEY":{"owners":["usr_..."],"origins":["https://api.example.com"]},"WATCH_HOOK_EXAMPLE_SECRET":{"owners":["usr_..."]}}
+
+# Usage readings → OpenVibe.Billing (server/billing.js): one check-operation reading per completed check, relayed
+# every WATCH_BILLING_INTERVAL_MS (0, the default, is off: nothing is queued) with this service's client credentials
+# (OV_OAUTH_CLIENT_ID/OV_OAUTH_CLIENT_SECRET above; Network grants billing.usage.record). Without the URL or the
+# secret, readings are queued and stay local.
+#OV_BILLING_INTERNAL_URL=
+#OV_BILLING_AUDIENCE=
+#WATCH_BILLING_TIMEOUT_MS=
+#WATCH_BILLING_INTERVAL_MS=
@@ -3,6 +3,15 @@
@@ -3,6 +3,15 @@
What shipped on OpenVibe.Watch, newest first. The plan is T18 ("Watch complete"); each entry names
What shipped on OpenVibe.Watch, newest first. The plan is T18 ("Watch complete"); each entry names
the step it landed.
the step it landed.
+## 2026-10-13 — check-operation usage readings (plan T5 step 14, Watch slice)
+
+- A completed check queues one `platform.usage-sample@1` reading (metric `check-operation`, quantity 1, unit `check`,
+ operation and provider the carrier, subject the owner, project only when the watch names one, cost the run's
+ `cost_usd`) in the transaction that records its run, keyed `watch:check:<run id>`; a failed check is not billed.
+- `server/billing.js` relays them to OpenVibe.Billing (`POST /api/v1/usage`, `billing.usage.record`) with the
+ `openvibe-sdk/usage` reporter; 400/422 rejects a reading, 429/5xx keeps it pending. `/api/ready` reports its counts.
+- Off by default (`WATCH_BILLING_INTERVAL_MS=0`). Migration `0008_billing_readings.sql`, `test/billing.test.js`.
+
## 2026-10-08 — account export and deletion (ADR-033)
## 2026-10-08 — account export and deletion (ADR-033)
- `POST /internal/events` (loopback, `WATCH_EVENTS_SECRET`) answers network.account.export_requested and
- `POST /internal/events` (loopback, `WATCH_EVENTS_SECRET`) answers network.account.export_requested and
@@ -418,6 +418,7 @@ Reporting: [SECURITY.md](SECURITY.md). The rules the code keeps:
@@ -418,6 +418,7 @@ Reporting: [SECURITY.md](SECURITY.md). The rules the code keeps:
- OpenVibe.Network (signing key; service principal `watch`; grants `network.account.export.contribute` and
- OpenVibe.Network (signing key; service principal `watch`; grants `network.account.export.contribute` and
`network.account.deletion.confirm` for account export and deletion, given last, once this release is live)
`network.account.deletion.confirm` for account export and deletion, given last, once this release is live)
- OpenVibe.Events (outbound events; the two account subscriptions, created at boot with `events.subscription.manage`)
- OpenVibe.Events (outbound events; the two account subscriptions, created at boot with `events.subscription.manage`)
+- OpenVibe.Billing (one `check-operation` usage reading per completed check, `POST /api/v1/usage` with `billing.usage.record`; off by default, `WATCH_BILLING_INTERVAL_MS`)
## Capabilities
## Capabilities
@@ -79,7 +79,8 @@
@@ -79,7 +79,8 @@
"GET /api/v1/resources and GET /api/v1/resources/:ovrn (watch.resource.read)",
"GET /api/v1/resources and GET /api/v1/resources/:ovrn (watch.resource.read)",
"watch.* events through openvibe-sdk's service_outbox in the same transaction as the state change",
"watch.* events through openvibe-sdk's service_outbox in the same transaction as the state change",
"per-actor limits (openvibe-sdk/limits)",
"per-actor limits (openvibe-sdk/limits)",
+ "usage readings (check operations): one platform.usage-sample@1 check-operation reading per completed check run, relayed to OpenVibe.Billing (billing.usage.record) by openvibe-sdk/usage; off by default (WATCH_BILLING_INTERVAL_MS)",
"the public site: Network sign-in (PKCE S256, web_sessions with only the token hash), server-rendered pages complete without JavaScript, the four create templates, watch management, observation history and check runs, how-it-works and the changelog, robots/sitemap/llms.txt"
"the public site: Network sign-in (PKCE S256, web_sessions with only the token hash), server-rendered pages complete without JavaScript, the four create templates, watch management, observation history and check runs, how-it-works and the changelog, robots/sitemap/llms.txt"
],
],
- "updated": "2026-10-12"
+ "updated": "2026-10-13"
}
}
new file mode 100644
new file mode 100644
@@ -0,0 +1,40 @@
@@ -0,0 +1,40 @@
+-- phase: expand
+-- Plan T5 step 14 (govern universal), Watch slice: the outbox of platform.usage-sample@1 readings Watch posts to
+-- OpenVibe.Billing (POST /api/v1/usage, capability billing.usage.record; server/billing.js), one per completed
+-- check run (metric check-operation).
+--
+-- billing_readings is the openvibe-sdk/usage createUsageReporter outbox (outboxSchema('billing_readings')): one row per
+-- reading, `event_id` is the reading's idempotency_key (watch:check:<check run id>; unique, so a retried or replayed
+-- run inserts nothing, ON CONFLICT DO NOTHING) and `envelope` is the reading JSON. The reporter only ever moves the
+-- sending columns (attempts, next_attempt_at, sent_at, seq, rejected_at, last_error); the trigger below refuses any
+-- change to the reading itself, so a retry always posts the same body and Billing's idempotency key sees the same
+-- reading. A check is counted, not aggregated: no period table is needed.
+-- Additive only: one new table, its indexes and its trigger; nothing existing changes.
+CREATE TABLE billing_readings (
+ id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
+ event_id text NOT NULL UNIQUE,
+ envelope jsonb NOT NULL,
+ traceparent text,
+ created_at bigint NOT NULL,
+ attempts integer NOT NULL DEFAULT 0,
+ next_attempt_at bigint NOT NULL DEFAULT 0,
+ sent_at bigint,
+ seq bigint,
+ rejected_at bigint,
+ last_error text,
+ -- `event_id` IS the reading's idempotency_key; this generated column names it so a reader can say so.
+ idempotency_key text GENERATED ALWAYS AS (event_id) STORED
+);
+CREATE INDEX IF NOT EXISTS billing_readings_due ON billing_readings (next_attempt_at, id) WHERE sent_at IS NULL AND rejected_at IS NULL;
+CREATE INDEX IF NOT EXISTS billing_readings_sent ON billing_readings (sent_at) WHERE sent_at IS NOT NULL;
+
+-- A reading is never edited: only its delivery columns move. event_id/envelope/created_at are the reading.
+CREATE FUNCTION billing_readings_frozen() RETURNS trigger LANGUAGE plpgsql AS $$
+BEGIN
+ IF NEW.event_id IS DISTINCT FROM OLD.event_id OR NEW.envelope IS DISTINCT FROM OLD.envelope
+ OR NEW.created_at IS DISTINCT FROM OLD.created_at THEN
+ RAISE EXCEPTION 'billing_readings: reading % is never edited', OLD.event_id;
+ END IF;
+ RETURN NEW;
+END $$;
+CREATE TRIGGER billing_readings_frozen BEFORE UPDATE ON billing_readings FOR EACH ROW EXECUTE FUNCTION billing_readings_frozen();
@@ -58,7 +58,7 @@ const CSP = "default-src 'self'; script-src 'self' 'unsafe-inline' https://openv
@@ -58,7 +58,7 @@ const CSP = "default-src 'self'; script-src 'self' 'unsafe-inline' https://openv
+ "connect-src 'self' https://openvibe.network https://openvibe.events https://cloudflareinsights.com; "
+ "connect-src 'self' https://openvibe.network https://openvibe.events https://cloudflareinsights.com; "
+ "frame-src 'self' https://openvibe.network; frame-ancestors 'none'; object-src 'none'; base-uri 'self'; form-action 'self' https://openvibe.network";
+ "frame-src 'self' https://openvibe.network; frame-ancestors 'none'; object-src 'none'; base-uri 'self'; form-action 'self' https://openvibe.network";
-function createApp({ config, db, registry, check, observations, scheduler, auth, outbox, actions = null, now, log = console, limitsNow = null, sessions, keyStore, siteLimits, fetchImpl = globalThis.fetch, accountData = null, accountSend = null, eventSources = null, webhookSources = null }) {
+function createApp({ config, db, registry, check, observations, scheduler, auth, outbox, actions = null, now, log = console, limitsNow = null, sessions, keyStore, siteLimits, fetchImpl = globalThis.fetch, accountData = null, accountSend = null, eventSources = null, webhookSources = null, billing = null }) {
const app = express();
const app = express();
app.disable('x-powered-by');
app.disable('x-powered-by');
app.set('trust proxy', config.trustProxy != null ? config.trustProxy : 'loopback');
app.set('trust proxy', config.trustProxy != null ? config.trustProxy : 'loopback');
@@ -98,7 +98,7 @@ function createApp({ config, db, registry, check, observations, scheduler, auth,
@@ -98,7 +98,7 @@ function createApp({ config, db, registry, check, observations, scheduler, auth,
// Readiness (openvibe-shared/ready): 503 only when the database fails; the Network key and the
// Readiness (openvibe-shared/ready): 503 only when the database fails; the Network key and the
// check worker (running, queue keeping up) are optional and degrade it (see observability.js).
// check worker (running, queue keeping up) are optional and degrade it (see observability.js).
- const readiness = createWatchReadiness({ db, config, registry, check, observations, scheduler, outbox, actions, now, release: release.release });
+ const readiness = createWatchReadiness({ db, config, registry, check, observations, scheduler, outbox, actions, billing, now, release: release.release });
app.get('/api/ready', readiness.handler);
app.get('/api/ready', readiness.handler);
// GET /release.json (ADR-016) and POST /release-metrics (open tabs' update reports into /metrics).
// GET /release.json (ADR-016) and POST /release-metrics (open tabs' update reports into /metrics).
release.mount(app, { registry: metrics.registry });
release.mount(app, { registry: metrics.registry });
new file mode 100644
new file mode 100644
@@ -0,0 +1,121 @@
@@ -0,0 +1,121 @@
+'use strict';
+/**
+ * Watch's check-operation usage readings → OpenVibe.Billing (plan T5 step 14, Watch slice).
+ *
+ * Watch emits one platform.usage-sample@1 metric, check-operation:
+ *
+ * check-operation one reading per completed check run (a check_runs row whose state is not a failure, see
+ * check.js FAILURE_STATES): quantity 1, unit 'check', operation and provider the carrier
+ * (http/feed/api), subject the watch's owner, project the watch's project_id only when it
+ * names one, cost_estimate the run's recorded cost_usd (0 when it records none), at the run's
+ * finish. Idempotency key watch:check:<check run id>, so a retried or replayed run is never
+ * billed twice. A failed check is not billed.
+ *
+ * The reporter is the SDK's own (openvibe-sdk/usage createUsageReporter): the check engine queues the reading
+ * idempotency-keyed in the outbox inside the transaction that records the run (check.js settle), and the reporter
+ * relays it to Billing's billing.usage.record (POST /api/v1/usage) with Watch's Network service token
+ * (createServiceTokenClient, clientId 'watch', audience openvibe.billing). A relay that cannot reach Billing (no grant
+ * yet, a missing route, a timeout, backpressure, a 401/403/404/408/425/429 or a 5xx) stays pending with backoff; only
+ * Billing refusing the reading itself (any other 4xx) is marked rejected and never retried. A reading is never edited
+ * (migrations/0008_billing_readings.sql).
+ *
+ * Off by default: WATCH_BILLING_INTERVAL_MS=0 starts no timer and queues nothing. With an interval but no
+ * OV_BILLING_INTERNAL_URL (or OV_OAUTH_CLIENT_SECRET) readings are queued and stay local.
+ *
+ * const billing = createWatchBilling({ db, config: config.billing, networkUrl: config.networkInternalUrl });
+ * if (billing.enabled) billing.start();
+ */
+const { createUsageReporter } = require('openvibe-sdk/usage');
+const { createServiceTokenClient } = require('openvibe-sdk/auth');
+
+const TABLE = 'billing_readings';
+const METRIC = 'check-operation';
+
+/**
+ * @param {object} o
+ * @param {object} o.db the Watch database (openvibe-sdk/db)
+ * @param {object} o.config config.billing (server/config.js)
+ * @param {string} [o.networkUrl] where Watch's service token comes from (OV_NETWORK_INTERNAL_URL)
+ * @param {{ now(): number }} [o.clock]
+ * @param {Function} [o.fetchImpl]
+ * @param {object} [o.log]
+ * @param {object} [o.reporter] a ready createUsageReporter (tests); default one over `db`/`config`
+ */
+function createWatchBilling({ db, config, networkUrl, clock = { now: () => Date.now() }, fetchImpl = globalThis.fetch, log = console, reporter: injected } = {}) {
+ const intervalMs = Number(config && config.intervalMs) || 0;
+ const now = () => clock.now();
+ let tokenClient = null;
+ if (config && config.url && config.clientSecret) {
+ tokenClient = createServiceTokenClient({
+ network: networkUrl, clientId: config.clientId, clientSecret: config.clientSecret,
+ audience: config.audience, fetch: fetchImpl, timeoutMs: config.timeoutMs, now,
+ });
+ }
+ const reporter = injected || createUsageReporter({
+ db, service: 'watch', source: 'watch.check', table: TABLE,
+ billingUrl: config && config.url, tokenClient, audience: config && config.audience,
+ fetchImpl, timeoutMs: (config && config.timeoutMs) || 5000, now, log,
+ });
+ let lastError = null;
+
+ /** The reading of one completed check run (`run` as check.js records it, `watch` as the registry reads it). */
+ function readingOf({ watch, run }) {
+ const cost = Number(run.cost_usd);
+ return reporter.sample({
+ idempotency_key: reporter.key('check', run.id),
+ project: watch.project_id || undefined,
+ subject: watch.owner || undefined,
+ resource: METRIC, provider: run.carrier, operation: run.carrier,
+ quantity: 1, unit: 'check',
+ cost_estimate: cost > 0 ? cost : 0,
+ at: new Date(Number(run.finished_at)).toISOString(),
+ });
+ }
+
+ /** INSIDE the caller's transaction: queue the run's reading. -> true when inserted, false when already queued. */
+ async function queue(t, { watch, run }) {
+ return await reporter.record(t, readingOf({ watch, run }));
+ }
+
+ /** One relay pass. Never overlaps itself; never throws. */
+ let flushing = null;
+ function flush() {
+ if (!flushing) flushing = reporter.flush()
+ .catch((err) => { lastError = err.message; log.error(`[billing] flush failed: ${err.message}`); return null; })
+ .finally(() => { flushing = null; });
+ return flushing;
+ }
+
+ let timer = null;
+ return {
+ enabled: intervalMs > 0,
+ reporting: !!tokenClient,
+ reporter,
+ readingOf, queue, flush,
+ start() {
+ if (timer || !(intervalMs > 0)) return false;
+ if (!tokenClient) log.warn('[billing] OV_BILLING_INTERNAL_URL or OV_OAUTH_CLIENT_SECRET is not set: readings are stored and stay queued');
+ timer = setInterval(flush, intervalMs);
+ if (timer.unref) timer.unref();
+ return true;
+ },
+ async stop() {
+ if (timer) clearInterval(timer);
+ timer = null;
+ if (flushing) await flushing;
+ await reporter.stop();
+ },
+ get running() { return !!timer; },
+ async status() {
+ return {
+ enabled: intervalMs > 0,
+ reporting: !!tokenClient,
+ pending: await reporter.pending(),
+ rejected: await reporter.rejected(),
+ last_error: lastError,
+ };
+ },
+ };
+}
+
+module.exports = { createWatchBilling, TABLE, METRIC };
@@ -16,6 +16,8 @@
@@ -16,6 +16,8 @@
* NOT dispatched here: a met condition records the trigger and emits the event, nothing else
* NOT dispatched here: a met condition records the trigger and emits the event, nothing else
* 7 arm next_due_at (+ jitter, or exponential backoff after a failure, capped) and not_before
* 7 arm next_due_at (+ jitter, or exponential backoff after a failure, capped) and not_before
* (Retry-After)
* (Retry-After)
+ * 8 a completed (not failed) check queues one check-operation usage reading for Billing with its run
+ * (server/billing.js; off unless WATCH_BILLING_INTERVAL_MS is set)
*
*
* The run row, the observation and every event commit in ONE transaction: an event exists exactly
* The run row, the observation and every event commit in ONE transaction: an event exists exactly
* when its effect does. Nothing polls a source whose condition is 'changed' twice for the same
* when its effect does. Nothing polls a source whose condition is 'changed' twice for the same
@@ -34,7 +36,7 @@ const FAILURE_STATES = new Set(['http_error', 'timeout', 'parse_error', 'rate_li
@@ -34,7 +36,7 @@ const FAILURE_STATES = new Set(['http_error', 'timeout', 'parse_error', 'rate_li
const PERSON_OWNER = /^(usr|agt)_[0-9A-HJKMNP-TV-Z]{26}$/;
const PERSON_OWNER = /^(usr|agt)_[0-9A-HJKMNP-TV-Z]{26}$/;
const USER_OWNER = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
const USER_OWNER = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
-function createCheck({ db, registry, config, carriers, extract, condition, observations, outbox, actions = null, now = () => Date.now(), log = console, ids = contractIds }) {
+function createCheck({ db, registry, config, carriers, extract, condition, observations, outbox, actions = null, billing = null, now = () => Date.now(), log = console, ids = contractIds }) {
const inflight = new Set();
const inflight = new Set();
const st = {
const st = {
@@ -163,9 +165,12 @@ function createCheck({ db, registry, config, carriers, extract, condition, obser
@@ -163,9 +165,12 @@ function createCheck({ db, registry, config, carriers, extract, condition, obser
*/
*/
async function settle({ watch, run, failures, successAt = null, notBefore = 0, events = [], observation = null, endpoint = null, fired = null }) {
async function settle({ watch, run, failures, successAt = null, notBefore = 0, events = [], observation = null, endpoint = null, fired = null }) {
const armed = nextDue(watch, run.finished_at, { failures, maxBackoffMs: config.worker.maxBackoffMs });
const armed = nextDue(watch, run.finished_at, { failures, maxBackoffMs: config.worker.maxBackoffMs });
- await db.tx(async () => {
+ await db.tx(async (t) => {
if (observation) await observations.insert(observation);
if (observation) await observations.insert(observation);
await st.insertRun.run(run);
await st.insertRun.run(run);
+ // A completed check is one check-operation reading for Billing, queued with its run (server/billing.js);
+ // a failed one is not billed, and the run id keys it, so a replayed run inserts nothing.
+ if (billing && billing.enabled && run.carrier && !FAILURE_STATES.has(run.state)) await billing.queue(t, { watch, run });
if (endpoint) await st.saveEndpoint.run(endpoint);
if (endpoint) await st.saveEndpoint.run(endpoint);
await st.finish.run({
await st.finish.run({
id: watch.id, at: run.finished_at, success_at: successAt, state: run.state,
id: watch.id, at: run.finished_at, success_at: successAt, state: run.state,
@@ -158,6 +158,19 @@ function load(env = process.env) {
@@ -158,6 +158,19 @@ function load(env = process.env) {
},
},
},
},
+ // Usage readings → OpenVibe.Billing (server/billing.js, plan T5 step 14): one check-operation reading per
+ // completed check, relayed every WATCH_BILLING_INTERVAL_MS with Watch's own client-credentials token. 0 (the
+ // default) is off: nothing is queued and no timer starts. Without OV_BILLING_INTERNAL_URL or the client secret,
+ // readings are queued and stay local.
+ billing: {
+ url: strip(env.OV_BILLING_INTERNAL_URL || ''),
+ clientId: env.OV_OAUTH_CLIENT_ID || 'watch',
+ clientSecret: env.OV_OAUTH_CLIENT_SECRET || '',
+ audience: env.OV_BILLING_AUDIENCE || 'openvibe.billing',
+ timeoutMs: int(env.WATCH_BILLING_TIMEOUT_MS, 5000),
+ intervalMs: int(env.WATCH_BILLING_INTERVAL_MS, 0),
+ },
+
secrets: env,
secrets: env,
};
};
}
}
@@ -24,6 +24,7 @@ const { createCarriers } = require('./carriers');
@@ -24,6 +24,7 @@ const { createCarriers } = require('./carriers');
const { extract } = require('./extract');
const { extract } = require('./extract');
const condition = require('./condition'); // { evaluate, hasChanged, holds, … }
const condition = require('./condition'); // { evaluate, hasChanged, holds, … }
const { createCheck } = require('./check');
const { createCheck } = require('./check');
+const { createWatchBilling } = require('./billing');
const { createScheduler } = require('./scheduler');
const { createScheduler } = require('./scheduler');
const { createEventSources } = require('./event-sources');
const { createEventSources } = require('./event-sources');
const { createWebhookSources } = require('./webhook-sources');
const { createWebhookSources } = require('./webhook-sources');
@@ -82,7 +83,10 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
@@ -82,7 +83,10 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
const runJobs = client ? createRunJobs(client, { baseUrl: config.actions.runUrl }) : null;
const runJobs = client ? createRunJobs(client, { baseUrl: config.actions.runUrl }) : null;
const codesTasks = client ? createCodesTasks(client, { baseUrl: config.actions.codesUrl }) : null;
const codesTasks = client ? createCodesTasks(client, { baseUrl: config.actions.codesUrl }) : null;
const actions = createActions({ db, config, notifications: givenNotifications || notifications, post, actor: givenActor || actorTasks, run: givenRun || runJobs, codes: givenCodes || codesTasks, now, log });
const actions = createActions({ db, config, notifications: givenNotifications || notifications, post, actor: givenActor || actorTasks, run: givenRun || runJobs, codes: givenCodes || codesTasks, now, log });
- const check = createCheck({ db, registry, config, carriers, extract, condition, observations, outbox, actions, now, log });
+ // Usage readings (plan T5 step 14): a completed check queues one check-operation reading for Billing in its own
+ // commit; off unless WATCH_BILLING_INTERVAL_MS is set (server/billing.js).
+ const billing = createWatchBilling({ db, config: config.billing, networkUrl: config.networkInternalUrl, clock: { now }, fetchImpl, log });
+ const check = createCheck({ db, registry, config, carriers, extract, condition, observations, outbox, actions, billing, now, log });
const scheduler = createScheduler({ db, check, config, now, log });
const scheduler = createScheduler({ db, check, config, now, log });
// Event sources (plan T18 step 5): watches woken by OpenVibe.Events events, with their subscriptions kept in step.
// Event sources (plan T18 step 5): watches woken by OpenVibe.Events events, with their subscriptions kept in step.
const eventSources = createEventSources({ db, check, config, fetchImpl, log, now });
const eventSources = createEventSources({ db, check, config, fetchImpl, log, now });
@@ -101,7 +105,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
@@ -101,7 +105,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
const accountSend = givenSend || (config.oauth.clientSecret
const accountSend = givenSend || (config.oauth.clientSecret
? createNetworkSender({ networkInternalUrl: config.networkInternalUrl, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, fetch: fetchImpl })
? createNetworkSender({ networkInternalUrl: config.networkInternalUrl, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, fetch: fetchImpl })
: null);
: null);
- const app = createApp({ config, db, registry, check, observations, scheduler, auth, outbox, actions, now, log, limitsNow, sessions, keyStore, siteLimits, accountData, accountSend, eventSources, webhookSources });
+ const app = createApp({ config, db, registry, check, observations, scheduler, auth, outbox, actions, now, log, limitsNow, sessions, keyStore, siteLimits, accountData, accountSend, eventSources, webhookSources, billing });
// One JWKS client for the process (the SDK shares it with verifyUserToken): refresh in the background
// One JWKS client for the process (the SDK shares it with verifyUserToken): refresh in the background
// on an unref'd timer, keeping the last good keys through a Network outage.
// on an unref'd timer, keeping the last good keys through a Network outage.
keys.start();
keys.start();
@@ -109,6 +113,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
@@ -109,6 +113,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
outbox.start();
outbox.start();
if (config.worker.enabled) scheduler.start();
if (config.worker.enabled) scheduler.start();
actions.start();
actions.start();
+ if (billing.enabled) billing.start();
const pruneTimer = setInterval(async () => {
const pruneTimer = setInterval(async () => {
try { await observations.prune(); } catch (err) { log.error(`[observations] prune: ${err.message}`); }
try { await observations.prune(); } catch (err) { log.error(`[observations] prune: ${err.message}`); }
try { await eventSources.prune(); } catch (err) { log.error(`[event-sources] prune: ${err.message}`); }
try { await eventSources.prune(); } catch (err) { log.error(`[event-sources] prune: ${err.message}`); }
@@ -142,6 +147,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
@@ -142,6 +147,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
keys.stop();
keys.stop();
await scheduler.stop();
await scheduler.stop();
await actions.stop();
await actions.stop();
+ await billing.stop();
await outbox.stop();
await outbox.stop();
if (server) {
if (server) {
server.closeAllConnections?.();
server.closeAllConnections?.();
@@ -150,7 +156,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
@@ -150,7 +156,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
if (!givenDb) await db.close();
if (!givenDb) await db.close();
}
}
- return { config, db, registry, eventSources, webhookSources, observations, guard, fetcher, spacer, carriers, extract, condition, check, scheduler, outbox, actions, keys, keyLoaded, auth, sessions, keyStore, siteLimits, accountData, app, server, close };
+ return { config, db, registry, eventSources, webhookSources, observations, guard, fetcher, spacer, carriers, extract, condition, check, scheduler, outbox, actions, billing, keys, keyLoaded, auth, sessions, keyStore, siteLimits, accountData, app, server, close };
}
}
/**
/**
@@ -48,7 +48,7 @@ function readers(db, { startedAt }) {
@@ -48,7 +48,7 @@ function readers(db, { startedAt }) {
};
};
}
}
-function createWatchReadiness({ db, config, registry, check, observations, scheduler, outbox, actions = null, now, release = null }) {
+function createWatchReadiness({ db, config, registry, check, observations, scheduler, outbox, actions = null, billing = null, now, release = null }) {
const read = readers(db, { startedAt: now() });
const read = readers(db, { startedAt: now() });
const iso = (v) => (v == null ? null : new Date(v).toISOString());
const iso = (v) => (v == null ? null : new Date(v).toISOString());
return createReadiness({
return createReadiness({
@@ -99,6 +99,7 @@ function createWatchReadiness({ db, config, registry, check, observations, sched
@@ -99,6 +99,7 @@ function createWatchReadiness({ db, config, registry, check, observations, sched
observations: dbOk ? await read.observations() : null,
observations: dbOk ? await read.observations() : null,
outbox: dbOk ? await outbox.status() : null,
outbox: dbOk ? await outbox.status() : null,
actions: dbOk && actions ? await actions.status() : null,
actions: dbOk && actions ? await actions.status() : null,
+ billing: dbOk && billing ? await billing.status() : null,
};
};
},
},
});
});
new file mode 100644
new file mode 100644
@@ -0,0 +1,164 @@
@@ -0,0 +1,164 @@
+'use strict';
+/**
+ * Usage readings → Billing (plan T5 step 14, Watch slice; server/billing.js, migrations/0008_billing_readings.sql):
+ * a completed check queues exactly one check-operation reading with its run, keyed by the run id, so a replay inserts
+ * nothing; a failed check queues nothing; a watch without a project sends no project; WATCH_BILLING_INTERVAL_MS=0
+ * starts no timer; only Billing refusing the reading itself (400/422) is permanent, a 429 or 5xx stays pending; and no
+ * secret or token reaches a log line.
+ */
+const assert = require('assert');
+const { validate } = require('openvibe-contracts');
+const { boot, site, suite, newUser } = require('./helpers');
+const { createWatchBilling, METRIC } = require('../server/billing');
+const { load } = require('../server/config');
+
+const t = suite('billing');
+let svc;
+let web;
+
+const P = 'prj_01JAB2C3D4E5F6G7H8J9K0MNPS';
+const person = { sub: newUser() };
+const TOKEN = 'not-a-real-token-' + 'x'.repeat(12);
+const CLIENT_SECRET = 'billing-client-secret-456';
+const lines = [];
+const capture = (...a) => lines.push(a.map(String).join(' '));
+const log = { log: capture, warn: capture, error: capture };
+const projects = { async check(project, subject) { return { ok: project === P && subject === person.sub }; } };
+
+const def = (overrides = {}) => ({
+ name: 'price watch',
+ source: { kind: 'http', url: `${web.origin}/price`, format: null },
+ cadence: { every_sec: 900 },
+ extraction: { kind: 'text' },
+ condition: { op: 'changed' },
+ action: [{ kind: 'notification' }],
+ ...overrides,
+});
+const json = (v) => (typeof v === 'string' ? JSON.parse(v) : v);
+const readings = async () => (await svc.db.many('SELECT * FROM billing_readings ORDER BY id')).map((r) => ({ ...r, envelope: json(r.envelope) }));
+const rowOf = (key) => svc.db.prepare('SELECT * FROM billing_readings WHERE event_id = ?').get(key);
+
+t('boot', async () => {
+ web = await site({
+ '/price': () => ({ headers: { 'content-type': 'text/plain' }, body: '19.99' }),
+ '/down': () => ({ status: 500, body: 'oops' }),
+ });
+ svc = await boot({ log, projects, env: { WATCH_BILLING_INTERVAL_MS: '3600000' } });
+ assert.strictEqual(svc.billing.enabled, true);
+ assert.strictEqual(svc.billing.reporting, false, 'no OV_BILLING_INTERNAL_URL: readings stay queued');
+});
+
+t('a completed check queues exactly one check-operation reading with its run', async () => {
+ const w = await svc.registry.create(def(), person);
+ const out = await svc.check.run(w.id, { trigger: 'manual' });
+ assert.ok(!['http_error', 'timeout'].includes(out.run.state), out.run.state);
+ const rows = (await readings()).filter((r) => r.envelope.idempotency_key === `watch:check:${out.run.id}`);
+ assert.strictEqual(rows.length, 1);
+ const r = rows[0].envelope;
+ assert.deepStrictEqual(r, {
+ id: `watch:check:${out.run.id}`, idempotency_key: `watch:check:${out.run.id}`, service: 'watch', subject: person.sub,
+ resource: METRIC, provider: 'http', operation: 'http', quantity: 1, unit: 'check',
+ at: new Date(Number(out.run.finished_at)).toISOString(), cost_estimate: 0, source: 'watch.check',
+ });
+ assert.ok(!('project' in r), 'a watch without a project sends no project');
+ const v = validate('platform.usage-sample@1', r);
+ assert.ok(v.valid, JSON.stringify(v.errors));
+ assert.strictEqual((await svc.billing.status()).pending, (await readings()).length);
+ // Readiness (loopback) reports the relay: counts only, never the URL or a credential.
+ const ready = await (await fetch(`${svc.base}/api/ready`)).json();
+ assert.deepStrictEqual(ready.billing, { enabled: true, reporting: false, pending: (await readings()).length, rejected: 0, last_error: null });
+});
+
+t('a replay of the same run inserts nothing', async () => {
+ const w = await svc.registry.create(def(), person);
+ const out = await svc.check.run(w.id, { trigger: 'manual' });
+ const before = (await readings()).length;
+ const watch = svc.registry.fromRow(await svc.db.prepare('SELECT * FROM watches WHERE id = ?').get(w.id));
+ const run = await svc.db.prepare('SELECT * FROM check_runs WHERE id = ?').get(out.run.id);
+ const inserted = await svc.db.tx((tx) => svc.billing.queue(tx, { watch, run }));
+ assert.strictEqual(inserted, false);
+ assert.strictEqual((await readings()).length, before);
+});
+
+t('a watch with a project names it; the run cost becomes cost_estimate', async () => {
+ const w = await svc.registry.create(def({ project_id: P }), person);
+ const out = await svc.check.run(w.id, { trigger: 'manual' });
+ const row = await rowOf(`watch:check:${out.run.id}`);
+ assert.strictEqual(json(row.envelope).project, P);
+ const costed = svc.billing.readingOf({ watch: { project_id: P, owner: person.sub }, run: { ...out.run, cost_usd: '0.25' } });
+ assert.strictEqual(costed.cost_estimate, 0.25);
+});
+
+t('a failed check queues nothing', async () => {
+ const w = await svc.registry.create(def({ source: { kind: 'http', url: `${web.origin}/down`, format: null } }), person);
+ const before = (await readings()).length;
+ const out = await svc.check.run(w.id, { trigger: 'manual' });
+ assert.strictEqual(out.run.state, 'http_error');
+ assert.strictEqual((await readings()).length, before);
+ assert.strictEqual(await rowOf(`watch:check:${out.run.id}`), undefined);
+});
+
+t('WATCH_BILLING_INTERVAL_MS=0 (the default) starts no timer; an interval starts one and stop() clears it', async () => {
+ const config = load({ NODE_ENV: 'test' }).billing;
+ assert.strictEqual(config.intervalMs, 0);
+ assert.strictEqual(config.clientId, 'watch');
+ assert.strictEqual(config.audience, 'openvibe.billing');
+ assert.strictEqual(config.timeoutMs, 5000);
+ const off = createWatchBilling({ db: svc.db, config, log });
+ assert.strictEqual(off.enabled, false);
+ assert.strictEqual(off.start(), false);
+ assert.strictEqual(off.running, false);
+ await off.stop();
+ const on = createWatchBilling({ db: svc.db, config: { ...config, intervalMs: 60000 }, log });
+ assert.strictEqual(on.start(), true);
+ assert.strictEqual(on.running, true);
+ assert.strictEqual(on.start(), false, 'one timer');
+ await on.stop();
+ assert.strictEqual(on.running, false);
+});
+
+t('Billing 5xx or 429 leaves a reading pending; 400/422 rejects it; no secret or token is logged', async () => {
+ const FIXED = Date.UTC(2026, 9, 13, 12, 0, 0);
+ // Push the check readings out of the relay's way: this part posts one reading at a time.
+ await svc.db.exec('UPDATE billing_readings SET next_attempt_at = $1', [FIXED + 10 * 86400000]);
+ const statuses = { 'k-500': 500, 'k-503': 503, 'k-429': 429, 'k-400': 400, 'k-422': 422 };
+ const posted = [];
+ const fetchImpl = async (url, opts) => {
+ if (!String(url).startsWith('http://billing.test/')) {
+ return { ok: true, status: 200, headers: new Headers({ 'content-type': 'application/json' }), json: async () => ({ access_token: TOKEN, token_type: 'Bearer', expires_in: 300 }), text: async () => '' };
+ }
+ assert.strictEqual(String(url), 'http://billing.test/api/v1/usage');
+ assert.strictEqual(opts.headers.Authorization, `Bearer ${TOKEN}`);
+ const sent = JSON.parse(opts.body);
+ posted.push(sent.idempotency_key);
+ const status = statuses[sent.idempotency_key];
+ return { ok: false, status, headers: new Headers(), text: async () => `status ${status}`, json: async () => ({}) };
+ };
+ const talking = createWatchBilling({
+ db: svc.db, clock: { now: () => FIXED }, fetchImpl, log, networkUrl: 'http://network.test',
+ config: { url: 'http://billing.test', clientId: 'watch', clientSecret: CLIENT_SECRET, audience: 'openvibe.billing', timeoutMs: 2000, intervalMs: 1000 },
+ });
+ assert.strictEqual(talking.reporting, true);
+ for (const [k, status] of Object.entries(statuses)) {
+ const reading = talking.reporter.sample({ idempotency_key: k, subject: person.sub, resource: METRIC, provider: 'http', operation: 'http', quantity: 1, unit: 'check', at: new Date(FIXED).toISOString() });
+ await svc.db.tx((tx) => talking.reporter.record(tx, reading));
+ await talking.flush();
+ const r = await rowOf(k);
+ assert.ok(!r.sent_at, `${status} is not sent`);
+ if (status === 400 || status === 422) assert.ok(r.rejected_at, `${status} refuses the reading`);
+ else {
+ assert.ok(!r.rejected_at, `${status} stays pending`);
+ assert.ok(Number(r.next_attempt_at) > FIXED, `${status} backs off`);
+ }
+ }
+ assert.deepStrictEqual(posted, Object.keys(statuses));
+ const status = await talking.status();
+ assert.strictEqual(status.rejected, 2);
+ await talking.stop();
+ const leaked = lines.filter((l) => l.includes(TOKEN) || l.includes(CLIENT_SECRET) || l.includes('watch-secret'));
+ assert.deepStrictEqual(leaked, [], 'no secret or token in any log line');
+});
+
+t('done', async () => { await web.close(); await svc.stop(); });
+
+t.run();