openvibers/openvibe.events Commits

Post hourly queue-operation readings per project to Billing (#13)

OP OpenVibers committed 5 Oct 2026 · 12d2aae9b6 · 14 files, +666 −13

Asked: T5 step 14 (govern universal), Events slice — OpenVibe.Events emits its queue-operation usage to Billing as platform.usage-sample@1 readings through billing.usage.record, following the Tools recipe. Dossier: /home/workstation/openvibe/agents/ds/prep/T5.md row 14. ONE pull request from current origi…
Harness job events-t5-step-14-govern-universal-234558 (plan T5).

Changes

ddiff --git a/.env.example b/.env.example
iindex 36eeac0..79feb7f 100644
---- a/.env.example
++++ b/.env.example
@@@ -94,6 +94,17 @@ EVENTS_RECEIPT_RETENTION_DAYS=90
#EVENTS_USAGE=on
#EVENTS_USAGE_FLUSH_MS=300000
++# Billing readings (plan T5 step 14, docs/cutover-events-billing-readings.md): each closed hour's queue operations per
++# production project (metric queue-operation-64kb) become one platform.usage-sample@1 reading posted to Billing
++# (billing.usage.record; the Network client `events` needs that grant). 0 = off. Without the URL or the secret the
++# readings are stored and stay queued.
++#EVENTS_BILLING_INTERVAL_MS=0
++#OV_BILLING_INTERNAL_URL=http://127.0.0.1:4600
++#OV_OAUTH_CLIENT_ID=events
++#OV_OAUTH_CLIENT_SECRET=
++#OV_BILLING_AUDIENCE=openvibe.billing
++#OV_BILLING_TIMEOUT_MS=10000
++
# Developer apps (events.app.publish|read|subscribe, ADR-014). App subscriptions must use public
# https endpoints (no allow-list; checked at subscribe and delivery time). Quotas are per project
# and environment; 0 turns a limit off. EVENTS_APPS=off disables the app paths.
ddiff --git a/README.md b/README.md
iindex 4f2a21d..83342f9 100644
---- a/README.md
++++ b/README.md
@@@ -203,6 +203,8 @@ Quotas recorded in Network (`dev_quotas`) are not read yet: that needs `network.
**Usage** ([server/usage.js](server/usage.js), roadmap WS-N task 4). Each project's use is counted per environment and UTC hour in the transaction that does the work: `events.app.publish` in `events` (events stored; a repeated `event_id` is not counted) with every refused publish request as an error (its problem code, status, trace id and, when one event was refused, its `event_id`), and `events.app.subscribe` in `deliveries` (webhook attempts to the project's subscriptions; one without a 2xx is an error, `events.delivery.http_<status>`, `events.delivery.timeout` or `events.delivery.failed`, with the event id and its trace id). A minute after an hour closes, each rollup is stored once as an `events.usage.recorded` event (source `events`, subject the project, visibility `internal`, payload `common.usage-recorded@1` from openvibe-contracts 0.63.0) in the transaction that marks it sent, so Events is its own outbox; first-party subscribers get it like any event, apps never see it. OpenVibe.Network subscribes to it for the project dashboard on openvibe.codes. `EVENTS_USAGE=off` counts nothing; `EVENTS_USAGE_FLUSH_MS` (300000) is how often closed hours are looked for. Sent rollups are kept 7 days in `app_usage`.
++**Billing readings** ([server/billing.js](server/billing.js), plan T5 step 14; [docs/cutover-events-billing-readings.md](docs/cutover-events-billing-readings.md)). Each closed UTC hour, one aggregate query turns a production project's queue operations (each publish and each delivered webhook counts `ceil(bytes / 65536)`, at least 1) into one `platform.usage-sample@1` reading, metric `queue-operation-64kb`, `idempotency_key` `events:queue-operation-64kb:<project>:<hour>`, stored once in `billing_readings` and never edited, then posted to Billing (`billing.usage.record`, `POST /api/v1/usage`) with Events' Network service token. Off by default: `EVENTS_BILLING_INTERVAL_MS` (0) turns the loop on, `OV_BILLING_INTERNAL_URL` and `OV_OAUTH_CLIENT_SECRET` let it send (unset: readings stay queued). Sandbox and first-party traffic are not billed.
++
## Client library
`require('openvibe-events')` (package `main` is [lib/client.js](lib/client.js); services can depend on this repo by tarball):
@@@ -283,7 +285,7 @@ void connect();
## Owns
--- `events`, `events_archive` (the replay tier), `subscriptions`, `deliveries`, `consumer_checkpoints`, `idempotency_receipts`, `app_revocations`, `app_usage` (PostgreSQL `ov_events`, ADR-035; the realtime fan-out is in-process, one Events process)
++- `events`, `events_archive` (the replay tier), `subscriptions`, `deliveries`, `consumer_checkpoints`, `idempotency_receipts`, `app_revocations`, `app_usage`, `billing_readings`, `billing_periods` (PostgreSQL `ov_events`, ADR-035; the realtime fan-out is in-process, one Events process)
- developer-app event scope, sandbox separation and per-project Events quotas (ADR-014)
- canonical event envelope (event_id, trace_id, type, version, source, actor, subject + revision, payload)
- priority classes `critical|important|low`, loop guards, backpressure, DLQ and replay
@@@ -308,6 +310,7 @@ void connect();
- a guessed private topic yields no data (`test/realtime.test.js`)
- a developer app cannot publish, read or subscribe outside its project, sandbox never meets production, app webhooks reach public addresses only, quotas hold (`test/apps.test.js`)
- a project's publishing and deliveries are counted per hour, refusals and failed attempts as errors, and each closed hour is sent once as `events.usage.recorded`, never visible to apps (`test/usage.test.js`)
++- each closed hour's queue operations per production project become one valid `platform.usage-sample@1` reading, rounded per 64 KiB, created once and never edited; the sender marks it sent on a 2xx, refuses it on a 4xx about the reading, keeps it pending on a 5xx, and two senders never post it twice; off by default (`test/billing.test.js`)
## Bootstrap / extraction source
ddiff --git a/STATUS.json b/STATUS.json
iindex 17edfad..3700ce1 100644
---- a/STATUS.json
++++ b/STATUS.json
@@@ -7,7 +7,7 @@
"runtime": "node 22, express, PostgreSQL 18 (openvibe-sdk/db, async) + Valkey 9",
"port": 4300,
"domain": "events.openvibe.network",
-- "contracts": "openvibe-contracts v0.81.0",
++ "contracts": "openvibe-contracts v0.94.2",
"capabilities": [
"events.event.publish",
"events.subscription.manage",
@@@ -30,6 +30,7 @@
"realtime SSE with cursor resume, gap reports and Network tickets",
"developer-app publish, read and subscribe with sandbox separation, SSRF-guarded webhooks and project quotas",
"hourly project usage rollups (events.usage.recorded)",
++ "queue-operation-64kb billing readings (platform.usage-sample@1) per production project and closed hour, posted to Billing through billing.usage.record (EVENTS_BILLING_INTERVAL_MS, off by default; docs/cutover-events-billing-readings.md)",
"redaction tombstones",
"per-actor limits (openvibe-sdk/limits; shared on Valkey)",
"carrier fabric (ADR-042): the pg-v1 poll beside a valkey-v1 push carrier for QUEUE (streams) and TOPIC (SSE fan-out) an optional nats-v1 NATS Core TOPIC carrier (NATS_URL) and an optional nats-js-v1 JetStream STREAM carrier (NATS_URL, NATS_JETSTREAM; deploy/nats), with a placement planner, per-key pinning, delivery_placements history and a GET /api/v1/placement explain API; EVENTS_CARRIERS[_DISABLED] switches it",
ddiff --git a/docs/cutover-events-billing-readings-seed.sql b/docs/cutover-events-billing-readings-seed.sql
nnew file mode 100644
iindex 0000000..004bc86
---- /dev/null
++++ b/docs/cutover-events-billing-readings-seed.sql
@@@ -0,0 +1,30 @@
++-- PostgreSQL seed for the 0007_billing_readings cutover rehearsal (docs/cutover-events-billing-readings.md). Loaded after
++-- migrations 0001-0006, before 0007: an app project's closed hour as production stores it. All rows sit one minute into the
++-- UTC hour that started two hours ago (closed, past the 5-minute grace):
++-- seq 1 prj_01JAB2C3D4E5F6G7H8J9K0MNPQ production, 100 bytes -> 1 operation (publish)
++-- seq 2 prj_01JAB2C3D4E5F6G7H8J9K0MNPQ production, 70000 bytes -> 2 operations (publish) + 2 (its delivered webhook)
++-- seq 3 the same project in sandbox -> not billed
++-- seq 4 first-party (live), no project, delivered to Search -> not billed
++-- Expected: one reading for the project, quantity 5.
++INSERT INTO events (id, seq, event_type, version, source, actor, subject_type, subject_id, subject_revision, trace_id, priority,
++ visibility, occurred_at, received_at, payload, hops, publisher, request_id, project_id, env, size_bytes, redacted_at, redacted_by, redacts)
++SELECT v.id, v.seq, v.event_type, 1, v.source, '{"type":"app","id":"app_rehearsal"}', 'order', 'o_' || v.seq, NULL,
++ 'trc_billing_' || v.seq, 'normal', 'public', '2026-10-04T00:00:00.000Z',
++ (floor(extract(epoch FROM now()) * 1000 / 3600000)::bigint - 2) * 3600000 + 60000, '{"n":1}', 1, v.publisher, NULL,
++ v.project_id, v.env, v.size_bytes, NULL, NULL, 0
++FROM (VALUES
++ ('evt_billing_0001', 1, 'app.rehearsal.order.created', 'app', 'app:app_rehearsal', 'prj_01JAB2C3D4E5F6G7H8J9K0MNPQ', 'production', 100),
++ ('evt_billing_0002', 2, 'app.rehearsal.order.created', 'app', 'app:app_rehearsal', 'prj_01JAB2C3D4E5F6G7H8J9K0MNPQ', 'production', 70000),
++ ('evt_billing_0003', 3, 'app.rehearsal.order.created', 'app', 'app:app_rehearsal', 'prj_01JAB2C3D4E5F6G7H8J9K0MNPQ', 'sandbox', 100),
++ ('evt_billing_0004', 4, 'live.stream.started', 'live', 'svc:live', NULL, 'production', 100)
++) AS v(id, seq, event_type, source, publisher, project_id, env, size_bytes);
++
++UPDATE sequences SET value = 4 WHERE name = 'events';
++
++INSERT INTO subscriptions (id, consumer, topic_pattern, endpoint, secret, enabled, created_at, updated_at, project_id, env) VALUES
++ ('sub_billing_app', 'app:app_rehearsal', 'app.rehearsal.*', 'https://hooks.example.com/h', 'rehearsal-secret-rehearsal-secret', 1, 0, 0, 'prj_01JAB2C3D4E5F6G7H8J9K0MNPQ', 'production'),
++ ('sub_billing_search', 'search', 'live.stream.*', 'http://127.0.0.1:9/hook', 'rehearsal-secret-rehearsal-secret', 1, 0, 0, NULL, 'production');
++INSERT INTO deliveries (event_id, subscription_id, seq, priority, status, delivered_at, created_at, updated_at)
++SELECT d.event_id, d.subscription_id, d.seq, 1, 'delivered', e.received_at + 1000, e.received_at, e.received_at
++FROM (VALUES ('evt_billing_0002', 'sub_billing_app', 2), ('evt_billing_0004', 'sub_billing_search', 4)) AS d(event_id, subscription_id, seq)
++JOIN events e ON e.id = d.event_id;
ddiff --git a/docs/cutover-events-billing-readings.md b/docs/cutover-events-billing-readings.md
nnew file mode 100644
iindex 0000000..f5a0204
---- /dev/null
++++ b/docs/cutover-events-billing-readings.md
@@@ -0,0 +1,90 @@
++# Cutover: queue-operation readings to OpenVibe.Billing (`0007_billing_readings`, plan T5 step 14)
++
++One additive migration (`-- phase: expand`): two new tables, `billing_readings` (the outbox of `platform.usage-sample@1`
++readings) and `billing_periods` (the closed hours already aggregated), plus a trigger that refuses any edit of a stored
++reading. Nothing in `events`, `deliveries` or any other table changes; no backfill runs. The code that fills them is off
++by default (`EVENTS_BILLING_INTERVAL_MS=0`): after the deploy the tables stay empty until an operator turns it on. No money
++moves here: Billing stores the readings and rates them in its own sweep. It also bumps `openvibe-contracts` from v0.81.0
++to v0.94.2 (`platform.usage-sample@1`).
++
++## What is metered
++
++| | |
++|---|---|
++| metric | `queue-operation-64kb` (the rate card's `metric`; the reading's `resource` and `unit`) |
++| who | a developer project in **production**: the event's project for a publish, the app subscription's project for a delivery. First-party traffic (no project) and sandbox are not billed. |
++| period | one UTC hour `[H, H+1h)`: publishes by `received_at`, webhook deliveries by `delivered_at` (status `delivered`; failed and dead deliveries are not billed). Closed at `H + 1h + 5 min`. |
++| quantity | each publish and each delivered webhook counts `ceil(size_bytes / 65536)` operations, at least 1 |
++| reading | `id` = `idempotency_key` = `events:queue-operation-64kb:<prj_…>:<H as ISO>`, `service` `events`, `project` the project, `provider` `local`, `operation` `events.queue`, `at` = `H`, `source` `openvibe.events` |
++
++One aggregate query over the closed hour (server/billing.js) inserts one row per project with `ON CONFLICT DO NOTHING`,
++in the transaction that marks the hour in `billing_periods`; re-running an hour inserts nothing, and a stored reading is
++never edited, so every retry posts the same body and Billing's dedupe on `idempotency_key` answers a replay with 200.
++Nothing is written on the publish or delivery path.
++
++## Environment (`/etc/openvibe/events.env`)
++
++| variable | default | meaning |
++|---|---|---|
++| `EVENTS_BILLING_INTERVAL_MS` | `0` | how often the loop aggregates closed hours and posts pending readings; `0` = off (nothing aggregated or sent) |
++| `OV_BILLING_INTERNAL_URL` | unset | Billing's base URL (`http://127.0.0.1:4600`); unset = readings are aggregated and stay queued, nothing is sent |
++| `OV_OAUTH_CLIENT_ID` | `events` | Events' Network client (client_credentials) |
++| `OV_OAUTH_CLIENT_SECRET` | unset | its secret; unset = nothing is sent (as without the URL) |
++| `OV_BILLING_AUDIENCE` | `openvibe.billing` | the token audience |
++| `OV_BILLING_TIMEOUT_MS` | `10000` | per request (token and post) |
++| `OV_NETWORK_INTERNAL_URL` | `http://127.0.0.1:4000` | already set: where the token comes from |
++
++Answers: 2xx (201 stored, 200 replay) → `sent`. Any other 4xx (400, 409 `billing.usage_key_reused`, 422
++`billing.invalid_input`, …) → `refused` with `last_error`, never retried. 401, 403 (the grant), 404, 408, 425, 429, 5xx,
++a network error or a token failure → stays `pending` with backoff (30 s doubling, at most 1 h) and the rest of the batch
++waits for the next tick. One loop per process; two processes are safe (rows are claimed `FOR UPDATE SKIP LOCKED` with a
++2-minute lease). After an outage the loop catches up at most 7 days of hours (well inside `EVENTS_RETENTION_DAYS`).
++
++## Network grant (OpenVibe.Network, its own PR)
++
++Events' service client needs `billing.usage.record` on audience `openvibe.billing`. In
++`server/identity/principals.js` `DEFAULT_GRANTS`, next to the Tools row:
++
++```js
++ // Plan T5 step 14: Events posts one queue-operation reading per production project and closed hour to Billing (POST /api/v1/usage).
++ ['events', 'billing.usage.record', 'openvibe.billing', []],
++```
++
++The `events` client must also exist as a confidential client with a secret (the value of `OV_OAUTH_CLIENT_SECRET`).
++Without the grant every token request for the audience fails; each tick logs
++`[billing] reading events:… not sent, retried in … s: token: …` and the readings wait, pending, in `billing_readings`.
++
++## Order
++
++1. Health: `ov access run openvibe-ovh health events` answers ready. Backup: today's `ov_events` dump exists (OpenVibe.Host
++ docs/backups.md); the migration only adds tables.
++2. Deploy the release (pipeline, `ovhost deploy events`). On start it applies `0007_billing_readings.sql` as the owner role
++ (`DATABASE_DIRECT_URL`). `billing_readings` and `billing_periods` stay empty: the loop is off.
++3. Merge and deploy the Network grant above; give the `events` client its secret.
++4. Turn it on: add `EVENTS_BILLING_INTERVAL_MS=300000`, `OV_BILLING_INTERNAL_URL=http://127.0.0.1:4600`,
++ `OV_OAUTH_CLIENT_SECRET=…` to `/etc/openvibe/events.env`, restart (`ovhost deploy events --restart`). The first tick
++ aggregates the newest closed hour only (no back-billing), then every hour as it closes.
++
++## Verification
++
++- `SELECT id, name, phase FROM ov_migrations WHERE id = '0007'` → `0007 | billing_readings | expand`; both tables empty
++ while the loop is off.
++- After step 4: `SELECT state, COUNT(*) FROM billing_readings GROUP BY state` shows `sent` growing and `pending` back to 0
++ after a tick; `refused` rows carry Billing's answer in `last_error`; `pending` rows with `last_error` `token: …` mean the
++ grant or the secret is missing. `SELECT MAX(period_start) FROM billing_periods` is the last closed hour.
++- In Billing: `GET /api/v1/usage?service=events` (`billing.ledger.admin`) lists one reading per project and hour.
++
++```rehearse
++seed: docs/cutover-events-billing-readings-seed.sql
++node -e "const a=require('assert');const {load}=require('./server/config');const {openDb}=require('./server/store');const {createBillingReadings,HOUR_MS}=require('./server/billing');const quiet={log(){},warn(){},error(){}};(async()=>{const c=load();const db=await openDb(c,{log:quiet});const b=createBillingReadings({db,config:c.billing,log:quiet});a.strictEqual(b.enabled,false);a.strictEqual(b.sending,false);const n=async(t)=>(await db.prepare('SELECT COUNT(*)::int AS n FROM '+t).get()).n;a.strictEqual(await n('billing_readings'),0);a.strictEqual(await n('events'),4);const at=(await db.prepare('SELECT received_at AS t FROM events WHERE seq = 1').get()).t;const H=Math.floor(Number(at)/HOUR_MS)*HOUR_MS;const r=await b.aggregate(H);a.deepStrictEqual([r.readings,r.created],[1,1]);const [row]=await db.prepare('SELECT * FROM billing_readings').all();a.deepStrictEqual([row.project_id,row.reading.quantity,row.state],['prj_01JAB2C3D4E5F6G7H8J9K0MNPQ',5,'pending']);a.strictEqual((await b.aggregate(H)).created,0);a.strictEqual(await n('billing_readings'),1);a.strictEqual((await b.send()).sent,0);await db.close();console.log('rehearsal ok '+row.id)})().catch(e=>{console.error(e);process.exit(1)})"
++```
++
++## Rollback
++
++- **Stop sending, no rollback:** set `EVENTS_BILLING_INTERVAL_MS=0` (or remove `OV_BILLING_INTERNAL_URL`) and restart.
++ Pending readings stay in the table and are posted when it is turned on again; Billing dedupes on `idempotency_key`.
++- **Code only:** roll back to the previous release. It does not know the two tables and never reads them; the migration
++ stays applied (an applied id the old runner does not have is not an error).
++- **Remove the tables** (a contract step, by hand as the owner, after a code rollback, only if they must go):
++ `DROP TABLE billing_readings, billing_periods; DROP FUNCTION billing_readings_frozen(); DELETE FROM ov_migrations WHERE id = '0007';`.
++ Readings not yet posted are lost; Billing keeps the ones it stored.
ddiff --git a/migrations/0007_billing_readings.sql b/migrations/0007_billing_readings.sql
nnew file mode 100644
iindex 0000000..8b9c0f4
---- /dev/null
++++ b/migrations/0007_billing_readings.sql
@@@ -0,0 +1,43 @@
++-- phase: expand
++-- Plan T5 step 14 (govern universal): the outbox of platform.usage-sample@1 readings Events posts to OpenVibe.Billing
++-- (POST /api/v1/usage, capability billing.usage.record; server/billing.js). One row per (billable project, metric,
++-- closed UTC hour), made once by one aggregate over `events` and `deliveries` after the hour closed; nothing is
++-- written on the publish or delivery path. `id` is the reading's id and idempotency_key (deterministic: re-running a
++-- period inserts nothing, ON CONFLICT DO NOTHING). A reading is never edited after it is created (the trigger below
++-- refuses it); only the sending columns move: pending -> sent (2xx, or Billing's replay of the key) or refused (Billing
++-- refused the reading itself; never retried). lease_until keeps a row claimed by one sender while it is posted.
++-- Additive only: two new tables, nothing existing changes.
++CREATE TABLE billing_readings (
++ id text COLLATE "C" PRIMARY KEY,
++ project_id text COLLATE "C" NOT NULL,
++ metric text COLLATE "C" NOT NULL,
++ period_start bigint NOT NULL,
++ reading jsonb NOT NULL,
++ state text COLLATE "C" NOT NULL DEFAULT 'pending' CHECK (state IN ('pending', 'sent', 'refused')),
++ attempts bigint NOT NULL DEFAULT 0,
++ last_error text COLLATE "C",
++ next_attempt_at bigint NOT NULL,
++ lease_until bigint,
++ created_at bigint NOT NULL,
++ sent_at bigint,
++ UNIQUE (project_id, metric, period_start)
++);
++CREATE INDEX idx_billing_readings_due ON billing_readings(next_attempt_at) WHERE state = 'pending';
++CREATE FUNCTION billing_readings_frozen() RETURNS trigger LANGUAGE plpgsql AS $$
++BEGIN
++ IF NEW.id IS DISTINCT FROM OLD.id OR NEW.project_id IS DISTINCT FROM OLD.project_id OR NEW.metric IS DISTINCT FROM OLD.metric
++ OR NEW.period_start IS DISTINCT FROM OLD.period_start OR NEW.reading IS DISTINCT FROM OLD.reading
++ OR NEW.created_at IS DISTINCT FROM OLD.created_at THEN
++ RAISE EXCEPTION 'billing_readings: reading % is never edited', OLD.id;
++ END IF;
++ RETURN NEW;
++END $$;
++CREATE TRIGGER billing_readings_frozen BEFORE UPDATE ON billing_readings FOR EACH ROW EXECUTE FUNCTION billing_readings_frozen();
++-- The closed hours already aggregated, per metric (an hour with no billable use leaves no reading, but is marked here).
++CREATE TABLE billing_periods (
++ metric text COLLATE "C" NOT NULL,
++ period_start bigint NOT NULL,
++ readings bigint NOT NULL,
++ aggregated_at bigint NOT NULL,
++ PRIMARY KEY (metric, period_start)
++);
ddiff --git a/package-lock.json b/package-lock.json
iindex cca5bda..e0f14ab 100644
---- a/package-lock.json
++++ b/package-lock.json
@@@ -12,7 +12,7 @@
"dotenv": "^16.4.7",
"express": "^4.21.2",
"iovalkey": "^0.4.0",
-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.81.0",
++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.94.2",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.26.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.5.0",
"pg": "^8.23.0"
@@@ -1102,9 +1102,9 @@
}
},
"node_modules/openvibe-contracts": {
-- "version": "0.81.0",
-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.81.0",
-- "integrity": "sha512-nOM8WXdxHOHFxt777ST1PUbAwYCp86LNzIRcBHR5qtCRDaRvJnBvkstnZpgZ1eJ9buU2NgifatIFt8E0Y4Ua0g==",
++ "version": "0.94.2",
++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.94.2",
++ "integrity": "sha512-taCtyQMTtm/KnyDVxkWICGe/5kmKJVXBQDvBASqBxO0CDs88kPse5raWAtFNG5HPJWyS7B48CvWuHGIHnEBxYQ==",
"license": "AGPL-3.0-only",
"dependencies": {
"ajv": "^8.17.1",
ddiff --git a/package.json b/package.json
iindex fedff40..055f2ce 100644
---- a/package.json
++++ b/package.json
@@@ -24,7 +24,7 @@
"dotenv": "^16.4.7",
"express": "^4.21.2",
"iovalkey": "^0.4.0",
-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.81.0",
++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.94.2",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.26.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.5.0",
"pg": "^8.23.0"
ddiff --git a/server/billing.js b/server/billing.js
nnew file mode 100644
iindex 0000000..76c221e
---- /dev/null
++++ b/server/billing.js
@@@ -0,0 +1,249 @@
++'use strict';
++/**
++ * Billing readings (plan T5 step 14, the Tools recipe of step 7): what each developer project's queue use here costs,
++ * sent to OpenVibe.Billing as platform.usage-sample@1 readings through billing.usage.record (POST /api/v1/usage).
++ *
++ * metric queue-operation-64kb (the platform.rate-card@1 metric: the reading's `resource` and `unit`)
++ * subject the production project that owns the work: an app event's project (`events.project_id`) for a publish,
++ * the app subscription's project (`subscriptions.project_id`) for a delivery. First-party traffic
++ * (no project) and sandbox are never billed.
++ * period one UTC hour [H, H+1h): publishes by `received_at`, webhook deliveries by `delivered_at` (status
++ * delivered; a failed or dead delivery is not billed). Closed once the clock passes H+1h+GRACE_MS.
++ * quantity each publish and each delivery counts ceil(size_bytes / 65536) operations, at least one.
++ *
++ * aggregate(H) runs ONE aggregate query over the closed hour and stores one reading per project in billing_readings
++ * (migrations/0007_billing_readings.sql), in the transaction that marks the hour aggregated in billing_periods.
++ * Nothing is written on the publish or delivery path. The reading id is its idempotency_key,
++ * `events:queue-operation-64kb:<project>:<hour ISO>`: re-running an hour inserts nothing (ON CONFLICT DO NOTHING),
++ * and a stored reading is never edited (a trigger refuses it), so every retry posts the same body.
++ *
++ * send() claims due pending readings (FOR UPDATE SKIP LOCKED plus a lease, so two processes never post one row at
++ * once) and posts each with Events' Network service token (openvibe-sdk createServiceTokenClient, audience
++ * openvibe.billing; the Events client needs the grant billing.usage.record):
++ * 2xx (201 stored, 200 Billing's replay of the key) sent
++ * any other 4xx (the reading itself: 400, 409, 422 …) refused, last_error kept, never retried
++ * 401, 403, 404, 408, 425, 429, 5xx, network, token stays pending with backoff (credentials, grant, address or
++ * pressure, not the reading); the rest of the batch waits
++ *
++ * Off by default (EVENTS_BILLING_INTERVAL_MS=0): no timer, nothing aggregated or sent. With an interval but without
++ * OV_BILLING_INTERNAL_URL (or OV_OAUTH_CLIENT_SECRET) readings are aggregated and stay queued; nothing is sent.
++ */
++const { validate } = require('openvibe-contracts');
++const { createServiceTokenClient } = require('openvibe-sdk/auth');
++
++const HOUR_MS = 60 * 60 * 1000;
++const GRACE_MS = 5 * 60 * 1000;
++const METRIC = 'queue-operation-64kb';
++const OP_BYTES = 65536;
++const PATH = '/api/v1/usage';
++const BATCH = 100;
++const LEASE_MS = 2 * 60 * 1000;
++const BACKOFF_MS = 30 * 1000;
++const MAX_BACKOFF_MS = HOUR_MS;
++const MAX_CATCHUP_HOURS = 7 * 24;
++const PROJECT_RE = /^prj_[0-9A-HJKMNP-TV-Z]{26}$/;
++// 4xx answers that are about the credentials, the grant, the address or load, not the reading: retried.
++const RETRY_4XX = new Set([401, 403, 404, 408, 425, 429]);
++
++const hourOf = (ms) => Math.floor(ms / HOUR_MS) * HOUR_MS;
++const keyOf = (projectId, periodStart) => `events:${METRIC}:${projectId}:${new Date(periodStart).toISOString()}`;
++
++/** The platform.usage-sample@1 reading of one project's closed hour. */
++function readingOf({ projectId, periodStart, quantity }) {
++ const key = keyOf(projectId, periodStart);
++ return {
++ id: key,
++ idempotency_key: key,
++ service: 'events', // the Contracts manifest id: Billing keys budgets on it
++ project: projectId,
++ resource: METRIC, // Billing matches the rate card's metric on `resource`
++ provider: 'local', // self-hosted queue (PostgreSQL), as Media's local storage
++ operation: 'events.queue',
++ quantity,
++ unit: METRIC,
++ at: new Date(periodStart).toISOString(),
++ source: 'openvibe.events',
++ };
++}
++
++/** POST one reading to Billing. { ok, status, final?, error? }; never throws. */
++function createBillingClient({ config, fetchImpl = globalThis.fetch, now = () => Date.now(), networkUrl }) {
++ if (!config.url || !config.clientSecret) return null;
++ const tokens = createServiceTokenClient({ network: networkUrl, clientId: config.clientId, clientSecret: config.clientSecret,
++ audience: config.audience, fetch: fetchImpl, timeoutMs: config.timeoutMs, now });
++ return {
++ async post(reading) {
++ let headers;
++ try { headers = await tokens.authHeaders({ audience: config.audience }); } catch (err) {
++ return { ok: false, status: 0, error: `token: ${String(err && err.message || err).slice(0, 300)}` };
++ }
++ let r;
++ try {
++ r = await fetchImpl(`${config.url}${PATH}`, {
++ method: 'POST', headers: { 'Content-Type': 'application/json', Accept: 'application/json', ...headers },
++ body: JSON.stringify(reading), signal: AbortSignal.timeout(config.timeoutMs),
++ });
++ } catch (err) {
++ return { ok: false, status: 0, error: String(err && err.message || err).slice(0, 300) };
++ }
++ if (r.ok) return { ok: true, status: r.status };
++ if (r.status === 401) tokens.invalidate({ audience: config.audience });
++ const text = (await r.text().catch(() => '')).replace(/\s+/g, ' ').slice(0, 300);
++ return { ok: false, status: r.status, error: `${r.status} ${text}`.trim(), final: r.status >= 400 && r.status < 500 && !RETRY_4XX.has(r.status) };
++ },
++ };
++}
++
++/**
++ * @param {object} o
++ * @param {object} o.db the Events database (openvibe-sdk/db)
++ * @param {object} o.config config.billing (server/config.js)
++ * @param {string} [o.networkUrl] where the service token comes from (OV_NETWORK_INTERNAL_URL)
++ * @param {{ now(): number }} [o.clock]
++ * @param {Function} [o.fetchImpl]
++ * @param {object} [o.client] a ready client ({ post(reading) }); default createBillingClient from the config
++ */
++function createBillingReadings({ db, config, networkUrl, clock = { now: () => Date.now() }, fetchImpl = globalThis.fetch, client, log = console }) {
++ const sender = client !== undefined ? client : createBillingClient({ config, fetchImpl, now: () => clock.now(), networkUrl });
++ const q = {
++ // The one aggregate: every billable operation of the hour, per project.
++ aggregate: db.prepare(`SELECT project_id, SUM(ops)::bigint AS quantity FROM (
++ SELECT e.project_id, GREATEST(1, CEIL(e.size_bytes / ${OP_BYTES}.0))::bigint AS ops
++ FROM events e
++ WHERE e.project_id IS NOT NULL AND e.env = 'production' AND e.received_at >= ? AND e.received_at < ?
++ UNION ALL
++ SELECT s.project_id, GREATEST(1, CEIL(e.size_bytes / ${OP_BYTES}.0))::bigint
++ FROM deliveries d JOIN subscriptions s ON s.id = d.subscription_id JOIN events e ON e.id = d.event_id
++ WHERE s.project_id IS NOT NULL AND s.env = 'production' AND d.status = 'delivered'
++ AND d.delivered_at >= ? AND d.delivered_at < ?
++ ) ops GROUP BY project_id ORDER BY project_id`),
++ add: db.prepare(`INSERT INTO billing_readings (id, project_id, metric, period_start, reading, next_attempt_at, created_at)
++ VALUES (?, ?, ?, ?, ?, ?, ?) ON CONFLICT DO NOTHING RETURNING id`),
++ mark: db.prepare('INSERT INTO billing_periods (metric, period_start, readings, aggregated_at) VALUES (?, ?, ?, ?) ON CONFLICT DO NOTHING'),
++ last: db.prepare('SELECT MAX(period_start) AS p FROM billing_periods WHERE metric = ?'),
++ claim: db.prepare(`UPDATE billing_readings SET lease_until = ? WHERE id IN (
++ SELECT id FROM billing_readings WHERE state = 'pending' AND next_attempt_at <= ? AND (lease_until IS NULL OR lease_until <= ?)
++ ORDER BY next_attempt_at, id LIMIT ? FOR UPDATE SKIP LOCKED)
++ RETURNING id, reading, attempts, next_attempt_at`),
++ sent: db.prepare(`UPDATE billing_readings SET state = 'sent', sent_at = ?, attempts = attempts + 1, last_error = NULL, lease_until = NULL
++ WHERE id = ? AND state = 'pending'`),
++ refused: db.prepare(`UPDATE billing_readings SET state = 'refused', attempts = attempts + 1, last_error = ?, lease_until = NULL
++ WHERE id = ? AND state = 'pending'`),
++ retry: db.prepare(`UPDATE billing_readings SET attempts = attempts + 1, last_error = ?, next_attempt_at = ?, lease_until = NULL
++ WHERE id = ? AND state = 'pending'`),
++ release: db.prepare('UPDATE billing_readings SET lease_until = NULL WHERE id = ? AND state = \'pending\''),
++ counts: db.prepare('SELECT state, COUNT(*) AS n FROM billing_readings GROUP BY state'),
++ };
++ const stats = { invalid: 0, lastError: null };
++
++ /** Store the readings of the closed hour starting at `periodStart`; { created, readings, invalid }. */
++ async function aggregate(periodStart, { now = clock.now() } = {}) {
++ if (periodStart !== hourOf(periodStart)) throw new RangeError(`billing: ${periodStart} is not the start of an hour`);
++ if (periodStart + HOUR_MS + GRACE_MS > now) throw new RangeError(`billing: the hour ${new Date(periodStart).toISOString()} has not closed`);
++ return await db.tx(async () => {
++ const rows = await q.aggregate.all(periodStart, periodStart + HOUR_MS, periodStart, periodStart + HOUR_MS);
++ let created = 0, invalid = 0;
++ for (const row of rows) {
++ const quantity = Number(row.quantity);
++ if (!PROJECT_RE.test(String(row.project_id)) || !(quantity > 0)) continue;
++ const reading = readingOf({ projectId: row.project_id, periodStart, quantity });
++ const v = validate('platform.usage-sample@1', reading);
++ if (!v.valid) {
++ invalid++;
++ stats.invalid++;
++ stats.lastError = `${reading.id}: ${v.errors.map(e => `${e.path || e.instancePath || ''} ${e.message}`).join('; ')}`;
++ log.error(`[billing] reading not stored (does not match platform.usage-sample@1): ${stats.lastError}`);
++ continue;
++ }
++ if (await q.add.get(reading.id, row.project_id, METRIC, periodStart, JSON.stringify(reading), now, now)) created++;
++ }
++ await q.mark.run(METRIC, periodStart, rows.length, now);
++ return { period: periodStart, readings: rows.length, created, invalid };
++ });
++ }
++
++ /**
++ * Aggregate every closed hour not yet aggregated: from the hour after the last one marked (at most
++ * MAX_CATCHUP_HOURS back) to the newest closed hour; the first run starts at the newest closed hour.
++ */
++ async function aggregateClosed({ now = clock.now() } = {}) {
++ const newest = hourOf(now - GRACE_MS) - HOUR_MS;
++ const last = (await q.last.get(METRIC)).p;
++ let from = last == null ? newest : Math.max(Number(last) + HOUR_MS, newest - (MAX_CATCHUP_HOURS - 1) * HOUR_MS);
++ if (last != null && Number(last) + HOUR_MS < from) log.warn(`[billing] hours from ${new Date(Number(last) + HOUR_MS).toISOString()} to ${new Date(from).toISOString()} were never aggregated and are past the catch-up window`);
++ let created = 0;
++ for (; from <= newest; from += HOUR_MS) created += (await aggregate(from, { now })).created;
++ return { created };
++ }
++
++ /** Post due pending readings to Billing; { sent, refused, retried }. No client: nothing (readings stay queued). */
++ async function send({ now = clock.now(), limit = BATCH } = {}) {
++ const out = { sent: 0, refused: 0, retried: 0 };
++ if (!sender) return out;
++ const rows = (await q.claim.all(now + LEASE_MS, now, now, limit))
++ .sort((a, b) => Number(a.next_attempt_at) - Number(b.next_attempt_at) || (a.id < b.id ? -1 : 1));
++ for (let i = 0; i < rows.length; i++) {
++ const row = rows[i];
++ const reading = typeof row.reading === 'string' ? JSON.parse(row.reading) : row.reading;
++ const res = await sender.post(reading);
++ if (res.ok) { await q.sent.run(clock.now(), row.id); out.sent++; continue; }
++ stats.lastError = `${row.id}: ${res.error || res.status}`;
++ if (res.final) {
++ await q.refused.run(String(res.error || res.status).slice(0, 500), row.id);
++ out.refused++;
++ log.error(`[billing] reading ${row.id} refused for good: ${res.error}`);
++ continue;
++ }
++ const backoff = Math.min(MAX_BACKOFF_MS, BACKOFF_MS * 2 ** Math.min(Number(row.attempts), 10));
++ await q.retry.run(String(res.error || res.status).slice(0, 500), clock.now() + backoff, row.id);
++ out.retried++;
++ log.warn(`[billing] reading ${row.id} not sent, retried in ${Math.round(backoff / 1000)} s: ${res.error}`);
++ // Billing, the token or the grant is down: the rest of the batch waits for the next tick.
++ for (const rest of rows.slice(i + 1)) await q.release.run(rest.id);
++ break;
++ }
++ return out;
++ }
++
++ let timer = null, running = null;
++ /** One pass: aggregate the closed hours, then send. Never overlaps itself; never throws. */
++ function tick() {
++ if (!running) running = (async () => {
++ try {
++ const a = await aggregateClosed();
++ const s = await send();
++ return { ...a, ...s };
++ } catch (err) {
++ stats.lastError = err.message;
++ log.error(`[billing] tick failed: ${err.message}`);
++ return null;
++ }
++ })().finally(() => { running = null; });
++ return running;
++ }
++
++ return {
++ enabled: config.intervalMs > 0,
++ sending: !!sender,
++ aggregate, aggregateClosed, send, tick,
++ start() {
++ if (timer || !(config.intervalMs > 0)) return false;
++ if (!sender) log.warn('[billing] OV_BILLING_INTERNAL_URL or OV_OAUTH_CLIENT_SECRET is not set: readings are stored and stay queued');
++ timer = setInterval(tick, config.intervalMs);
++ timer.unref?.();
++ return true;
++ },
++ async stop() {
++ if (timer) clearInterval(timer);
++ timer = null;
++ if (running) await running;
++ },
++ get running() { return !!timer; },
++ async status() {
++ const by = Object.fromEntries((await q.counts.all()).map(r => [r.state, Number(r.n)]));
++ return { enabled: config.intervalMs > 0, sending: !!sender, pending: by.pending || 0, sent: by.sent || 0, refused: by.refused || 0, invalid: stats.invalid, last_error: stats.lastError };
++ },
++ };
++}
++
++module.exports = { createBillingReadings, createBillingClient, readingOf, keyOf, METRIC, HOUR_MS, GRACE_MS };
ddiff --git a/server/config.js b/server/config.js
iindex 373a45a..9213acb 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -179,6 +179,18 @@ function load(env = process.env) {
flushIntervalMs: Math.max(1000, int(env.EVENTS_USAGE_FLUSH_MS, 5 * 60 * 1000)),
},
++ // Billing readings (server/billing.js, plan T5 step 14): each closed hour's queue operations per production
++ // project become one platform.usage-sample@1 reading, posted to Billing (billing.usage.record). intervalMs 0
++ // (the default): off, nothing aggregated or sent. Without OV_BILLING_INTERNAL_URL readings are stored, not sent.
++ billing: {
++ intervalMs: Math.max(0, int(env.EVENTS_BILLING_INTERVAL_MS, 0)),
++ url: (env.OV_BILLING_INTERNAL_URL || '').trim().replace(/\/+$/, ''),
++ audience: env.OV_BILLING_AUDIENCE || 'openvibe.billing',
++ clientId: env.OV_OAUTH_CLIENT_ID || 'events',
++ clientSecret: env.OV_OAUTH_CLIENT_SECRET || '',
++ timeoutMs: Math.max(1000, int(env.OV_BILLING_TIMEOUT_MS, 10000)),
++ },
++
// Realtime (SSE) gateway
realtime: {
maxConnections: int(env.REALTIME_MAX_CONNECTIONS, 2000),
ddiff --git a/server/index.js b/server/index.js
iindex 38761a2..94abada 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -18,11 +18,12 @@ const { createRealtime } = require('./realtime');
const { createApp } = require('./app');
const { createMetrics } = require('./metrics');
const { createGuardedPost } = require('./egress');
++const { createBillingReadings } = require('./billing');
const { gracefulStop } = require('openvibe-sdk/service');
async function start({
config, db: givenDb = null, clock = { now: () => Date.now() }, fetchImpl = globalThis.fetch, deliveryFetch = fetchImpl,
-- appPost = undefined, dnsLookup = undefined, log = console, listen = true,
++ appPost = undefined, dnsLookup = undefined, billingFetch = fetchImpl, log = console, listen = true,
} = {}) {
config = config || load();
// PostgreSQL (ADR-035): opened and migrated here unless the caller (a test) hands in a migrated handle.
@@@ -101,6 +102,11 @@ async function start({
const usageTimer = config.usage && config.usage.enabled ? setInterval(flushUsage, config.usage.flushIntervalMs) : null;
usageTimer?.unref?.();
++ // Billing readings (server/billing.js, plan T5 step 14): one loop per process, off unless EVENTS_BILLING_INTERVAL_MS
++ // is set; it aggregates closed hours into billing_readings and posts them to Billing (OV_BILLING_INTERNAL_URL).
++ const billing = createBillingReadings({ db, config: config.billing, networkUrl: config.networkInternalUrl, clock, fetchImpl: billingFetch, log });
++ billing.start();
++
let server = null;
if (listen) {
server = await new Promise((resolve, reject) => {
@@@ -116,6 +122,7 @@ async function start({
async function close() {
clearInterval(pruneTimer);
if (usageTimer) clearInterval(usageTimer);
++ await billing.stop();
policies.stop();
await carriers.stop().catch(() => {});
realtime.stop();
@@@ -129,7 +136,7 @@ async function start({
if (!givenDb) await db.close();
}
-- return { config, db, store, keys, keyLoaded, auth, worker, realtime, metrics, app, server, policies, carriers, planner, valkey, close, prune, flushUsage };
++ return { config, db, store, keys, keyLoaded, auth, worker, realtime, metrics, app, server, policies, carriers, planner, valkey, close, prune, flushUsage, billing };
}
if (require.main === module) {
ddiff --git a/test/billing.test.js b/test/billing.test.js
nnew file mode 100644
iindex 0000000..a793400
---- /dev/null
++++ b/test/billing.test.js
@@@ -0,0 +1,201 @@
++'use strict';
++// Billing readings (plan T5 step 14, server/billing.js): a closed hour's queue operations per production project
++// (each publish and each delivered webhook counts ceil(bytes / 64 KiB), at least one) become one
++// platform.usage-sample@1 reading, stored once under a deterministic id and never edited; the sender posts it to
++// Billing with the service token, marks it sent on a 2xx (Billing's replay too), refuses it for good on a 4xx about
++// the reading, and keeps it pending with backoff on a 5xx. Off by default.
++const assert = require('assert');
++const { ids, validate } = require('openvibe-contracts');
++const apps = require('../server/apps');
++const { createBillingReadings, keyOf, HOUR_MS, GRACE_MS } = require('../server/billing');
++const { boot, request, appToken, envelope, serviceToken, suite } = require('./helpers');
++
++const t = suite('billing');
++const project = () => {
++ const prj = `prj_${ids.ulid()}`, appId = ids.newId('app');
++ const key = apps.projectKey(prj), source = apps.appSource(`app:${appId}`);
++ return {
++ prj, key, source,
++ token: (env = 'production') => appToken({ appId, projectId: prj, env, cap: ['events.app.publish', 'events.app.subscribe'] }),
++ event: (payload = { n: 1 }) => ({
++ event_id: ids.newId('event'), event_type: `app.${key}.order.created`, version: 1, source,
++ actor: { type: 'app', id: appId }, timestamp: new Date().toISOString(), subject: { type: 'order', id: 'o1' }, payload,
++ }),
++ };
++};
++const P1 = project(), P2 = project(), P3 = project(), SANDBOX = project();
++const BILLING = 'http://billing.test';
++const CONFIG = { intervalMs: 60000, url: BILLING, audience: 'openvibe.billing', clientId: 'events', clientSecret: 's'.repeat(32), timeoutMs: 5000 };
++
++let h, H0, statuses = [], ev = {};
++const readings = async () => await h.db.prepare('SELECT * FROM billing_readings ORDER BY id').all();
++const publish = async (p, body, env = 'production') => {
++ const r = await request(h.base, 'POST', '/api/v1/events', { token: p.token(env), body });
++ assert.strictEqual(r.status, 201, r.text);
++};
++
++/** A Network token endpoint and a Billing that answers from `answers` (a status per call, 201 when empty). */
++function stubBilling(answers = []) {
++ const calls = { token: [], usage: [] };
++ const fetchImpl = async (url, init) => {
++ if (String(url).endsWith('/oauth/token')) {
++ calls.token.push(Object.fromEntries(new URLSearchParams(String(init.body))));
++ return new Response(JSON.stringify({ access_token: 'tok-billing', token_type: 'Bearer', expires_in: 300 }), { status: 200, headers: { 'Content-Type': 'application/json' } });
++ }
++ assert.strictEqual(url, `${BILLING}/api/v1/usage`);
++ calls.usage.push({ auth: init.headers.Authorization, body: JSON.parse(init.body) });
++ const status = answers.length ? answers.shift() : 201;
++ if (status === 'throw') throw new Error('connect ECONNREFUSED');
++ return new Response(JSON.stringify(status < 300 ? { record: {} } : { code: 'billing.x', status }), { status, headers: { 'Content-Type': 'application/json' } });
++ };
++ return { calls, fetchImpl };
++}
++const sender = (fetchImpl, config = CONFIG) => createBillingReadings({ db: h.db, config, networkUrl: 'http://network.test', clock: h.clock, fetchImpl, log: { log() {}, warn() {}, error() {} } });
++
++t('boot: off by default, nothing aggregated or sent', async () => {
++ H0 = Math.floor(Date.now() / HOUR_MS) * HOUR_MS;
++ h = await boot({
++ env: { EVENTS_MAX_PAYLOAD_BYTES: String(200 * 1024) },
++ appPost: async () => ({ status: statuses.length ? statuses.shift() : 204 }),
++ });
++ h.clock.t = H0 + 5 * 60 * 1000;
++ assert.strictEqual(h.config.billing.intervalMs, 0, 'EVENTS_BILLING_INTERVAL_MS defaults to 0');
++ assert.strictEqual(h.config.billing.url, '', 'no OV_BILLING_INTERNAL_URL');
++ assert.strictEqual(h.billing.enabled, false);
++ assert.strictEqual(h.billing.running, false, 'no timer');
++ assert.strictEqual(h.billing.sending, false, 'no Billing client');
++ assert.strictEqual(h.billing.start(), false, 'start() stays off');
++ await h.store.createSubscription({ id: `sub_${ids.ulid()}`, consumer: P1.source, topicPattern: `app.${P1.key}.*`, endpoint: 'https://hooks.example.com/h', secret: 'a'.repeat(40), projectId: P1.prj, env: 'production' });
++});
++
++t('the hour\'s publishes and delivered webhooks are counted per 64 KiB; sandbox, first-party and failed deliveries are not', async () => {
++ ev = { small: P1.event(), big: P1.event({ blob: 'x'.repeat(70000) }), at64: P1.event(), over64: P1.event() };
++ await publish(P1, { events: [ev.small, ev.big, ev.at64, ev.over64] });
++ // The exact boundary: 65536 bytes is one operation, 65537 two.
++ await h.db.prepare('UPDATE events SET size_bytes = ? WHERE id = ?').run(65536, ev.at64.event_id);
++ await h.db.prepare('UPDATE events SET size_bytes = ? WHERE id = ?').run(65537, ev.over64.event_id);
++ const big = await h.db.prepare('SELECT size_bytes FROM events WHERE id = ?').get(ev.big.event_id);
++ assert.ok(big.size_bytes > 65536 && big.size_bytes <= 2 * 65536, `two operations: ${big.size_bytes}`);
++ await publish(P2, P2.event());
++ await publish(P3, { events: [P3.event(), P3.event()] });
++ await publish(SANDBOX, SANDBOX.event(), 'sandbox');
++ const r = await request(h.base, 'POST', '/api/v1/events', { token: serviceToken('live', ['events.event.publish']), body: envelope('live') });
++ assert.strictEqual(r.status, 201, r.text);
++ statuses = [204, 204, 204, 500]; // small, big and at64 delivered; over64 fails
++ await h.worker.drain();
++ const d = await h.db.prepare("SELECT status FROM deliveries WHERE event_id = ?").get(ev.over64.event_id);
++ assert.notStrictEqual(d.status, 'delivered');
++});
++
++t('an open hour cannot be aggregated', async () => {
++ await assert.rejects(h.billing.aggregate(H0), /has not closed/);
++ h.clock.advance(HOUR_MS); // H0 + 65 min: closed once past H0 + 1 h + GRACE_MS
++ assert.ok(H0 + HOUR_MS + GRACE_MS <= h.clock.now());
++ assert.strictEqual(await h.db.prepare('SELECT COUNT(*) AS n FROM billing_readings').get().then(r => r.n), 0, 'nothing on the hot path');
++});
++
++t('a closed hour: one valid reading per production project, quantities in 64 KiB operations', async () => {
++ const out = await h.billing.aggregateClosed();
++ assert.strictEqual(out.created, 3);
++ const rows = await readings();
++ assert.deepStrictEqual(rows.map(r => r.project_id).sort(), [P1.prj, P2.prj, P3.prj].sort(), 'no sandbox, no first-party');
++ const q = Object.fromEntries(rows.map(r => [r.project_id, r.reading.quantity]));
++ // P1 publishes 1 + 2 + 1 + 2, delivers small 1 + big 2 + at64 1 (over64 failed).
++ assert.deepStrictEqual([q[P1.prj], q[P2.prj], q[P3.prj]], [10, 1, 2]);
++ for (const row of rows) {
++ const v = validate('platform.usage-sample@1', row.reading);
++ assert.ok(v.valid, JSON.stringify(v.errors));
++ assert.deepStrictEqual(row.reading, {
++ id: keyOf(row.project_id, H0), idempotency_key: keyOf(row.project_id, H0), service: 'events', project: row.project_id,
++ resource: 'queue-operation-64kb', provider: 'local', operation: 'events.queue', quantity: row.reading.quantity,
++ unit: 'queue-operation-64kb', at: new Date(H0).toISOString(), source: 'openvibe.events',
++ });
++ assert.strictEqual(row.id, `events:queue-operation-64kb:${row.project_id}:${new Date(H0).toISOString()}`);
++ assert.deepStrictEqual([row.state, Number(row.attempts), row.period_start, row.metric], ['pending', 0, H0, 'queue-operation-64kb']);
++ }
++});
++
++t('re-running a closed hour creates nothing; a reading is never edited', async () => {
++ const before = await readings();
++ assert.strictEqual((await h.billing.aggregate(H0)).created, 0);
++ assert.strictEqual((await h.billing.aggregateClosed()).created, 0, 'the hour is marked aggregated');
++ assert.deepStrictEqual(await readings(), before);
++ await assert.rejects(h.db.prepare('UPDATE billing_readings SET reading = ? WHERE id = ?').run(JSON.stringify({ quantity: 1 }), before[0].id), /never edited/);
++});
++
++t('without OV_BILLING_INTERNAL_URL the readings stay queued', async () => {
++ const out = await sender(globalThis.fetch, { ...CONFIG, url: '' }).send();
++ assert.deepStrictEqual(out, { sent: 0, refused: 0, retried: 0 });
++ assert.ok((await readings()).every(r => r.state === 'pending' && Number(r.attempts) === 0));
++});
++
++t('a 5xx keeps the reading pending with backoff and stops the batch', async () => {
++ const b = stubBilling([503]);
++ const out = await sender(b.fetchImpl).send();
++ assert.deepStrictEqual(out, { sent: 0, refused: 0, retried: 1 });
++ assert.strictEqual(b.calls.usage.length, 1, 'the rest waits');
++ assert.deepStrictEqual([b.calls.token[0].client_id, b.calls.token[0].audience, b.calls.token[0].grant_type], ['events', 'openvibe.billing', 'client_credentials']);
++ assert.strictEqual(b.calls.usage[0].auth, 'Bearer tok-billing');
++ const rows = await readings();
++ const tried = rows.find(r => r.id === b.calls.usage[0].body.idempotency_key);
++ assert.deepStrictEqual([tried.state, Number(tried.attempts)], ['pending', 1]);
++ assert.ok(/^503/.test(tried.last_error) && tried.next_attempt_at > h.clock.now());
++ assert.ok(rows.every(r => r.lease_until === null), 'claims released');
++ assert.strictEqual(rows.filter(r => Number(r.attempts) === 0).length, 2);
++ // A network failure is the same: pending, next time.
++ h.clock.advance(31 * 1000);
++ const c = stubBilling(['throw']);
++ assert.deepStrictEqual(await sender(c.fetchImpl).send(), { sent: 0, refused: 0, retried: 1 });
++});
++
++t('2xx (and Billing\'s replay of the key) marks sent; a 4xx about the reading refuses it for good', async () => {
++ h.clock.advance(10 * 60 * 1000);
++ const b = stubBilling([201, 200, 422]);
++ assert.deepStrictEqual(await sender(b.fetchImpl).send(), { sent: 2, refused: 1, retried: 0 });
++ const rows = await readings();
++ const refused = rows.filter(r => r.state === 'refused');
++ assert.strictEqual(refused.length, 1);
++ assert.ok(/^422/.test(refused[0].last_error));
++ for (const r of rows.filter(x => x.state === 'sent')) assert.ok(r.sent_at && r.last_error === null);
++ for (const c of b.calls.usage) assert.deepStrictEqual(c.body, rows.find(r => r.id === c.body.id).reading, 'posted as stored');
++ h.clock.advance(2 * HOUR_MS);
++ const again = stubBilling();
++ assert.deepStrictEqual(await sender(again.fetchImpl).send(), { sent: 0, refused: 0, retried: 0 });
++ assert.strictEqual(again.calls.usage.length, 0, 'sent and refused readings are never posted again');
++ const s = await h.billing.status();
++ assert.deepStrictEqual([s.pending, s.sent, s.refused], [0, 2, 1]);
++});
++
++t('two senders never post one reading twice', async () => {
++ const H1 = Math.floor(h.clock.now() / HOUR_MS) * HOUR_MS;
++ await publish(P2, P2.event());
++ h.clock.t = H1 + HOUR_MS + GRACE_MS + 1000;
++ // The hours between H0 and H1 are aggregated too (nothing billable in them); H1 has P2's one publish.
++ assert.strictEqual((await h.billing.aggregateClosed()).created, 1);
++ let gate;
++ const wait = new Promise(r => { gate = r; });
++ const slow = stubBilling();
++ const slowFetch = async (url, init) => { if (!String(url).endsWith('/oauth/token')) await wait; return slow.fetchImpl(url, init); };
++ const a = sender(slowFetch).send(), b = sender(slowFetch).send();
++ await new Promise(r => setTimeout(r, 50));
++ gate();
++ const [ra, rb] = await Promise.all([a, b]);
++ assert.strictEqual(ra.sent + rb.sent, 1);
++ assert.strictEqual(slow.calls.usage.length, 1);
++ assert.strictEqual(slow.calls.usage[0].body.id, keyOf(P2.prj, H1));
++});
++
++t('the loop runs when EVENTS_BILLING_INTERVAL_MS is set', async () => {
++ const h2 = await boot({ env: { EVENTS_BILLING_INTERVAL_MS: '60000' } });
++ try {
++ assert.strictEqual(h2.billing.enabled, true);
++ assert.strictEqual(h2.billing.running, true);
++ assert.strictEqual(h2.billing.sending, false, 'no OV_BILLING_INTERNAL_URL: aggregated, not sent');
++ assert.ok(await h2.billing.tick());
++ } finally { await h2.stop(); }
++ assert.strictEqual(h2.billing.running, false, 'stopped with the process');
++});
++
++t('teardown', async () => { await h.stop(); });
++
++t.run();
ddiff --git a/test/security-secrets.test.js b/test/security-secrets.test.js
iindex e423811..c5698a7 100644
---- a/test/security-secrets.test.js
++++ b/test/security-secrets.test.js
@@@ -25,8 +25,10 @@ const crawler = require('./security-crawl');
const t = suite('security-secrets');
let w;
const needles = {};
++// Secrets Events does read from its environment, each planted in the world below as a sentinel the crawl looks for.
++const PLANTED = { OV_OAUTH_CLIENT_SECRET: `ovsecret-events-client-${crypto.randomBytes(12).toString('hex')}` };
--t('Events reads no secret from its environment (so there is none to plant a sentinel in)', async () => {
++t('Events reads no secret from its environment but the planted ones', async () => {
const names = new Set();
const walk = (d) => {
for (const f of fs.readdirSync(d)) {
@@@ -37,13 +39,16 @@ t('Events reads no secret from its environment (so there is none to plant a sent
};
walk(path.join(__dirname, '..', 'server'));
assert.ok(names.has('DATABASE_URL') && names.has('OV_NETWORK_PUBLIC_KEY'), 'the scan sees the config');
-- const secretish = [...names].filter((n) => /SECRET|PASSWORD|TOKEN|PRIVATE|CREDENTIAL|_KEY$/.test(n) && n !== 'OV_NETWORK_PUBLIC_KEY');
-- assert.deepStrictEqual(secretish, [], 'a new secret-looking variable: give it a sentinel in this test');
++ const secretish = [...names].filter((n) => /SECRET|PASSWORD|TOKEN|PRIVATE|CREDENTIAL|_KEY$/.test(n) && n !== 'OV_NETWORK_PUBLIC_KEY' && !PLANTED[n]);
++ assert.deepStrictEqual(secretish, [], 'a new secret-looking variable: give it a sentinel in this test (PLANTED)');
});
t('world', async () => {
-- w = await buildWorld();
++ // The Billing client is built (URL and secret set; the loop stays off), so its token client holds the secret.
++ w = await buildWorld({ env: { ...PLANTED, OV_BILLING_INTERNAL_URL: 'http://127.0.0.1:9' } });
++ assert.strictEqual(w.h.billing.sending, true);
Object.assign(needles, w.secrets);
++ Object.entries(PLANTED).forEach(([name, value]) => { needles[`${name} (planted)`] = value; });
// The Network private key (Events only ever has the public half).
const pem = (k) => k.split('\n').filter((l) => l && !l.startsWith('-----'));
const pub = pem(crypto.createPublicKey(w.privateKey).export({ type: 'spki', format: 'pem' })).join('');
ddiff --git a/test/security-ssrf.test.js b/test/security-ssrf.test.js
iindex b0510a8..7085034 100644
---- a/test/security-ssrf.test.js
++++ b/test/security-ssrf.test.js
@@@ -220,6 +220,7 @@ const REVIEWED = {
'server/worker.js appPost': [1, 'developer-app deliveries: the guarded poster (server/egress.js), covered above'],
'server/egress.js http.request': [1, 'the guarded poster\'s https.request (default requestImpl): its own lookup pins the checked address'],
'server/egress.js requestImpl': [1, 'the guarded poster\'s request, after parseAppEndpoint and with the guarded lookup'],
++ 'server/billing.js fetch': [1, 'Billing readings: OV_BILLING_INTERNAL_URL /api/v1/usage (operator config, a fixed path); the reading is built from counts, never a request input'],
'server/fabric/nats-core.js net.connect': [1, 'the nats-v1 broker: NATS_URL (operator config, nats:// on loopback or the private network); never a request input'],
};
/** JS source with comments blanked (strings, template literals and regex literals kept). */