openvibers/openvibe.media Commits
Emit Media storage and delivery usage readings to Billing through openvibe-sdk/usage (off by default); pin openvibe-sdk v0.32.0 (#43)
Changes
@@@ -45,7 +45,7 @@ without `DATABASE_URL`, development uses an embedded PGlite database in `data/pgexport and deletion), OpenVibe.Live (`live.lineage.resolve`), OpenVibe.Events (the outbox relay andthe account and revocation subscriptions)- Backblaze B2 and Cloudflare R2 (S3 API) when configured; `ffmpeg`/`ffprobe` on the host--- `openvibe-contracts` v0.79.0, `openvibe-sdk` v0.28.0 (tokens, events outbox, per-actor limits),++- `openvibe-contracts` v0.79.0, `openvibe-sdk` v0.32.0 (tokens, events outbox, usage readings, per-actor limits),`openvibe-shared` v2.10.0, pinned by release tarball## Capabilities@@@ -117,7 +117,7 @@ server/storage.orphans.scan (monthly storage orphan report, service-wide, report only),storage.move.cleanup (the alert for placement moves whose delete keeps failing, service-wide)client-ip.js trust proxy = loopback; req.ip is the only client address--(openvibe-shared v2.10.0, openvibe-contracts v0.79.0, openvibe-sdk v0.28.0: pinned release tarballs, installed by npm)++(openvibe-shared v2.10.0, openvibe-contracts v0.79.0, openvibe-sdk v0.32.0: pinned release tarballs, installed by npm)scripts/smoke-test.sh end-to-end smoke test (boots a temp instance)scripts/reconcile-objects.js / object-invariant.js / object-drift-report.js / no-good-copy-report.js object-model operator toolsscripts/media-jobs.js list jobs, run the size-invariant scan (dry run by default), approve/cancel proposals@@@ -440,6 +440,40 @@ before the route does any work, and one `[Limits]` log line and `media_rate_limiNever limited: `/healthz`, `/api/ready`, `/release.json`, `/metrics`, the signed `/internal/events` deliveries.Counters are per process (a restart forgets them). `test/actor-limits.test.js`.++## Usage readings → Billing++++Media's storage usage goes to OpenVibe.Billing as one `platform.usage-sample@1` reading per (project, subject,++tier) where a project's work is billable (`server/billing.js`, plan T5 step 14):++++| metric (`resource`) | what | period | unit | provider |++|---|---|---|---|---|++| `gb-month` | bytes stored: every present copy in `media_locations`, summed by tier, counted once when more than one object names a location and billed to the oldest owner, as that day's share of a GB-month | closed UTC day | `GB` | the tier (`local`/`b2`/`r2`) |++++**Media emits `gb-month` only.** `gib-delivered` is deliberately not emitted: `demand.record()` counts `route()`++calls, not bytes, so reads × `size_bytes` would bill aborted downloads and range reads in full, would bill++presigned B2/R2 redirects Media never serves, and would miss HLS segment reads. It comes back once responses are++metered by the bytes actually written (`sendSlice`/`streamFileWithRange`) and redirected reads are metered by CDN++logs. Nothing here bills delivery.++++Only the newest closed UTC day is aggregated, because the query reads `media_locations`' current state: a day++missed while the service was down is not billed (a reading never claims today's state for an older day).++++First-party traffic (a tenant with no project) and sandbox tenants are never recorded; an object without an++`owner_subject` is still stored with its project (Billing leaves it unrated until it names a user subject).++The SDK reporter (`openvibe-sdk/usage createUsageReporter`) queues each reading idempotency-keyed in++`billing_readings` inside the aggregation's transaction and relays it to Billing's `POST /api/v1/usage`++(`billing.usage.record`) with Media's Network service token (`clientId` `media`, audience `openvibe.billing`;++OpenVibe.Network grants exactly that). A relay that cannot reach Billing (no grant, a missing route, a timeout,++backpressure, a 401/403/404/408/425/429 or a 5xx) stays pending with backoff across restarts; only Billing++refusing the reading itself (any other 4xx) is marked rejected and never retried, so nothing is billed twice.++A stored reading is never edited (a trigger). Re-aggregating a day inserts nothing++(`billing_periods`).++++**Off by default.** `MEDIA_BILLING_INTERVAL_MS=0` (the default) starts no timer and aggregates nothing; a++restore drill starts nothing at all. Turn it on with `MEDIA_BILLING_INTERVAL_MS` plus `OV_BILLING_URL` and,++only if it differs, `OV_BILLING_AUDIENCE`; without a billing URL or `OV_OAUTH_CLIENT_SECRET` readings are++aggregated and stay queued. The runbook is `docs/cutover-media-billing-readings.md`; `test/billing.test.js`.++## Health, readiness and metrics| path | behavior |nnew file mode 100644@@@ -0,0 +1,90 @@++# Cutover runbook — Media storage readings to OpenVibe.Billing (plan T5 step 14)++++This change makes Media a producer of `platform.usage-sample@1` readings for its own storage:++`migrations/0004_billing_readings.sql` adds the SDK usage reporter's outbox plus Media's aggregation state, and++`server/billing.js` aggregates each closed UTC day into one `gb-month` reading per (project, subject, tier) and++relays it to OpenVibe.Billing's `POST /api/v1/usage`. Nothing is renamed, dropped, altered or backfilled, and the++reads and writes that serve media are untouched. No money moves here: Billing stores the readings and rates them in++its own sweep (its `BILLING_RATING_INTERVAL_MS`, off by default, and no rate card loaded = no charge).++++Media emits `gb-month` only. `gib-delivered` is deliberately not emitted: `demand.record()` counts `route()` calls,++not bytes, so reads × `size_bytes` would bill aborted downloads and range reads in full, would bill presigned B2/R2++redirects Media never serves, and would miss HLS segment reads. It comes back once responses are metered by the++bytes actually written (`sendSlice`/`streamFileWithRange`) and redirected reads are metered by CDN logs. Nothing++here bills delivery.++++## What the migration does++++`migrations/0004_billing_readings.sql` (`phase: expand`) creates two tables and a trigger:++++- `billing_readings` — the `openvibe-sdk/usage` `createUsageReporter` outbox: `event_id` (the reading's++ idempotency key, unique), `envelope` (the reading JSON), and the relay's columns (`attempts`,++ `next_attempt_at`, `sent_at`, `seq`, `rejected_at`, `last_error`). The reporter only ever moves the delivery++ columns; the trigger `billing_readings_frozen` refuses every change to `event_id`/`envelope`/`created_at`, so a++ stored reading is immutable and a retry always posts the same body. A generated `idempotency_key` column names++ `event_id` for readers.++- `billing_periods` — one row per `(metric, period_start)` already aggregated (`gb-month`: the UTC day), so a++ re-run inserts nothing.++++The runner is `openvibe-sdk/db`'s `migrate()`, started by `server/db/database.js` as the owner++(`DATABASE_DIRECT_URL`) when the process boots: it locks `ov_migrations`, runs the file in one transaction,++records id `0004` with its checksum, and refuses a later edit of it.++++## What is recorded++++| metric (`resource`) | source | period | unit | provider |++|---|---|---|---|---|++| `gb-month` | `media_locations` present copies, summed by provider, each location counted once when more than one object names it and billed to the oldest owner, as that day's share of a GB-month (`bytes / GB / days in that UTC month`) | closed UTC day | `GB` | the tier (`local`/`b2`/`r2`) |++++One reading per (project, subject, tier), so a month of daily readings sums to one GB-month per tier. Only++`apps.env = 'production'` tenants with a `project_id` are recorded (first-party and sandbox are dropped before++`record()`); an object without an `owner_subject` is stored with its project alone and Billing leaves it unrated++until it names a user subject.++++Only the newest closed UTC day is aggregated: the storage query reads `media_locations`' current state, so a day++missed while the service was down is simply not billed — a reading never claims today's state for an older day.++Nothing here reads Valkey.++++## Order++++1. **Backup.** Take the logical dump of the `media` database before the deploy (the rule for any data change,++ even additive): `sudo ovhost backup --all --logical`. Note the stamp it prints; the way back uses it.++2. **Merge and deploy** through the pipeline (`ovhost deploy media`; `--wait-idle` is not needed: no recording++ state changes shape). The first process to boot applies `0004`.++3. **Leave `MEDIA_BILLING_INTERVAL_MS` unset (0) at first.** Nothing is aggregated, no reading is queued and++ nothing is posted.++4. **Network grant** (OpenVibe.Network, its own change): the `media` client needs `billing.usage.record` on++ audience `openvibe.billing`. Network's `server/identity/principals.js` already grants it; confirm it is++ deployed before step 5. Without it every token request for that audience is refused, each post is logged as++ `[usage] media reading not delivered (will retry)`, and the readings wait in the table (no media read is++ affected).++5. **Turn it on**: add `MEDIA_BILLING_INTERVAL_MS` (for example `300000`, every 5 minutes) and `OV_BILLING_URL`++ (Billing's internal URL) to `/etc/openvibe/media.env`, and `OV_BILLING_AUDIENCE` only if it differs from++ `openvibe.billing`; then restart through `ovhost deploy media --restart`.++++## Verification++++- `ov access run openvibe-ovh health media` (or `/api/ready`) answers 200 after the deploy.++- The migration is recorded: `SELECT id, name, phase FROM ov_migrations WHERE id = '0004'` returns++ `0004 | billing_readings | expand`, and with the interval still 0 `SELECT COUNT(*) FROM billing_readings` is 0.++- After step 5: `SELECT metric, COUNT(*) AS n, MAX(period_start) AS last FROM billing_periods GROUP BY metric`++ shows the newest closed day appearing; `SELECT COUNT(*) FROM billing_readings WHERE sent_at IS NULL++ AND rejected_at IS NULL` stays small. `GET /api/v1/usage?service=media` in Billing (its `billing.ledger.admin`++ capability) lists the readings with `idempotency_key` `media:store:<project>:<subject>:<tier>:<day>`.++- A reading is never edited: `UPDATE billing_readings SET envelope = '{}'::jsonb WHERE …` fails with++ `billing_readings: reading … is never edited`.++- A relay that cannot reach Billing leaves rows pending; a reading Billing refuses (any other 4xx) is marked++ `rejected_at` with `last_error` and never retried.++++## Rollback++++- **Code only** (the usual way back): `ovhost rollback media` (or `deploy/scripts/deploy.sh --rollback`). The++ previous release does not know the tables and ignores them; the applied migration is accepted (an applied id++ the release does not have is not an error). Pending readings stay and are posted when the release comes back++ (Billing dedupes on the reading's idempotency key).++- **Stop posting without a rollback:** remove `MEDIA_BILLING_INTERVAL_MS` (or set it to 0) and restart. Readings++ already queued stay in `billing_readings` and wait.++- **Remove the tables** (only if they must go; a contract step, by hand as the owner, after a code rollback):++ `DROP TABLE billing_readings; DROP TABLE billing_periods; DELETE FROM ov_migrations WHERE id = '0004';`.++- **Restore:** the dump from step 1 restores the database as it was before the deploy (OpenVibe.Host++ `docs/backups.md`); not needed for this additive change unless the database itself is damaged.@@@ -378,6 +378,14 @@ request/playback → telemetry → Events → {Hotness, Health, Cost} engines(GraphQL), B2 usage reports, Bunny statistics API are imported daily, compared with the estimate, and the gap is ametric. Budgets per class and provider with a forecast; `media.provider.cost_threshold` at 90 % shifts non-critical traffic.++**Media's own usage readings (plan T5 step 14):** Media emits `gb-month` — one `platform.usage-sample@1` reading per++(project, subject, tier) per closed UTC day, that day's share of a GB-month from the present `media_locations` copies++(`server/billing.js`, off by default with `MEDIA_BILLING_INTERVAL_MS`). `gib-delivered` is deliberately not emitted:++`demand.record()` counts `route()` calls, not bytes, so reads × `size_bytes` would bill aborted downloads and range++reads in full, would bill presigned B2/R2 redirects Media never serves, and would miss HLS segment reads. It comes++back once responses are metered by the bytes actually written (`sendSlice`/`streamFileWithRange`) and redirected++reads are metered by CDN logs. Nothing here bills delivery.++## 10. Build order- **F1 foundation (shipped):** provider registry + capability probes; `media.cost_tiers` + per-class budgets in a revisionednnew file mode 100644@@@ -0,0 +1,52 @@++-- phase: expand++-- Plan T5 step 14 (govern universal), Media slice: the outbox of platform.usage-sample@1 readings Media posts to++-- OpenVibe.Billing (POST /api/v1/usage, capability billing.usage.record; server/billing.js) plus the aggregation++-- state that decides which closed UTC periods were already aggregated.++--++-- billing_readings is the openvibe-sdk/usage createUsageReporter outbox (outboxSchema('billing_readings')): one row per++-- reading, `event_id` is the reading's idempotency_key (unique: re-queueing a period inserts nothing, ON CONFLICT DO++-- NOTHING) and `envelope` is the reading JSON. The reporter never invents a reading: it 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 stored reading is immutable: a retry always posts the same body, and Billing's idempotency++-- key sees the same reading.++--++-- billing_periods: one row per (metric, closed UTC period start) already aggregated, so a re-run of an hour or day++-- inserts nothing even when it produced no reading (a period with no billable use leaves no reading).++-- Additive only: two new tables and their indexes/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();++++-- The closed periods already aggregated, per metric (gb-month: the UTC day).++CREATE TABLE billing_periods (++ metric text COLLATE "C" NOT NULL,++ period_start bigint NOT NULL,++ readings bigint NOT NULL DEFAULT 0,++ aggregated_at bigint NOT NULL,++ PRIMARY KEY (metric, period_start)++);@@@ -18,7 +18,7 @@"iovalkey": "^0.4.0","multer": "^1.4.5-lts.1","openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.79.0",-- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.28.0",++ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.32.0","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.10.0","pg": "^8.23.0","sharp": "^0.33.4"@@@ -2026,9 +2026,9 @@}},"node_modules/openvibe-sdk": {-- "version": "0.28.0",-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.28.0",-- "integrity": "sha512-jjLisFDWBN+xDVsixa5BJRr3355kRijmvOh1/9rz4s1yVb3wZWCbpTLNcOvXm57ksnRoqK57393W38G0Ey3PFw==",++ "version": "0.32.0",++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.32.0",++ "integrity": "sha512-O/54OWx6cxR00O9D7BMLkTyWXtS+Jq06sbszrYOcSMy6D6VhSqwFMMh23+Mew01W1HgxorGcUcxqhnt3BrdTUQ==","license": "MIT","engines": {"node": ">=20"@@@ -22,7 +22,7 @@"iovalkey": "^0.4.0","multer": "^1.4.5-lts.1","openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.79.0",-- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.28.0",++ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.32.0","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.10.0","pg": "^8.23.0","sharp": "^0.33.4"nnew file mode 100644@@@ -0,0 +1,218 @@++'use strict';++/**++ * Media's storage usage readings → OpenVibe.Billing (plan T5 step 14, Media slice).++ *++ * Media emits one platform.usage-sample@1 metric, gb-month, the storage metric the platform.rate-card@1 contract++ * names for it:++ *++ * gb-month the bytes stored for one closed UTC day, per (project, subject, tier): every present copy in++ * media_locations, once a day ("daily partition"), in GB. resource 'gb-month', provider the tier++ * (local/b2/r2), unit 'GB', operation 'store'. Each daily reading carries that day's share of a++ * GB-month — bytes / GB / (days in that UTC month) — so a month of daily readings sums to one++ * GB-month. A location more than one object names (F3.4) is counted once, billed to the oldest++ * object naming it.++ *++ * gib-delivered is deliberately not emitted. demand.record() counts route() calls, not bytes: reads × size_bytes++ * would bill aborted downloads and range reads in full, would bill presigned B2/R2 redirects Media never serves,++ * and never sees HLS segment reads. It comes back once responses are metered by the bytes actually written++ * (sendSlice/streamFileWithRange) and redirected reads are metered by CDN logs. Nothing here bills delivery.++ *++ * First-party traffic (a tenant with no project) and sandbox tenants are never recorded: the reporter bills whatever++ * reading it is given, so the aggregate query joins apps and keeps only `env = 'production'` rows with a project id.++ * A reading without the object's owner_subject is still stored (Billing leaves it unrated: rating needs a user++ * subject).++ *++ * The reporter is the SDK's own (openvibe-sdk/usage createUsageReporter): it derives the sample, queues it++ * idempotency-keyed in the outbox inside the aggregation's transaction, and relays it to Billing's++ * billing.usage.record (POST /api/v1/usage) with Media's Network service token (createServiceTokenClient, clientId++ * 'media', audience openvibe.billing; Network grants exactly billing.usage.record). 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, so nothing is billed twice. A reading is never edited (migrations/0004_billing_readings.sql).++ *++ * Off by default: MEDIA_BILLING_INTERVAL_MS=0 starts no timer and aggregates nothing; a restore drill starts++ * nothing at all. With an interval but no OV_BILLING_URL (or OV_OAUTH_CLIENT_SECRET) readings are aggregated and++ * stay queued.++ *++ * const billing = createMediaBilling({ db, config: config.billing, networkUrl: config.network.internalUrl });++ * if (billing.enabled) billing.start();++ */++const { createUsageReporter } = require('openvibe-sdk/usage');++const { createServiceTokenClient } = require('openvibe-sdk/auth');++++const DAY_MS = 24 * 60 * 60 * 1000;++const GRACE_MS = 5 * 60 * 1000; // a day is aggregated only once this much of the next one passed++const GB = 1000 * 1000 * 1000; // a GB in bytes: the unit gb-month is metered in++const TABLE = 'billing_readings';++const STORAGE = 'gb-month';++const UNKNOWN_CONTRACT = /unknown contract/;++++const dayStart = (ms) => Math.floor(ms / DAY_MS) * DAY_MS;++const iso = (ms) => new Date(ms).toISOString();++/** Days in the UTC month containing `ms`: the divisor that turns a day's bytes into its share of a GB-month. */++function daysInMonth(ms) { const d = new Date(ms); return new Date(Date.UTC(d.getUTCFullYear(), d.getUTCMonth() + 1, 0)).getUTCDate(); }++++/**++ * @param {object} o++ * @param {object} o.db the Media database (openvibe-sdk/db)++ * @param {object} o.config config.billing (server/config.js)++ * @param {string} [o.networkUrl] where Media'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 createMediaBilling({ 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: 'media', source: 'media.billing', table: TABLE,++ billingUrl: config && config.url, tokenClient, audience: config && config.audience,++ fetchImpl, timeoutMs: (config && config.timeoutMs) || 5000, now, log,++ });++ let lastError = null;++++ /**++ * Queue one reading in the caller's transaction. The SDK reporter validates it against platform.usage-sample@1++ * when the pinned openvibe-contracts knows the schema; a contracts that predates it (v0.79.0 here) makes++ * validateUsageSample throw "unknown contract", which must not block the queue — the SDK's own rule is that a++ * contracts without the schema leaves the reading unchecked, not refused. Only a real validation refusal throws.++ */++ async function queue(t, reading) {++ try {++ return await reporter.record(t, reading);++ } catch (err) {++ if (!UNKNOWN_CONTRACT.test(String((err && err.message) || ''))) throw err;++ if (typeof reading.idempotency_key !== 'string' || !reading.idempotency_key) {++ throw new TypeError('billing reading: idempotency_key is required (it is the outbox key)');++ }++ const n = await t.exec(`INSERT INTO ${TABLE} (event_id, envelope, created_at) VALUES ($1, $2, $3) ON CONFLICT (event_id) DO NOTHING`,++ [reading.idempotency_key, JSON.stringify(reading), now()]);++ return n > 0;++ }++ }++++ /** The closed period's row, if it was aggregated already. */++ const periodDone = (metric, periodStart) =>++ db.prepare(`SELECT readings FROM billing_periods WHERE metric = ? AND period_start = ?`).get(metric, periodStart);++++ /**++ * Aggregate the closed UTC day [periodStart, +1d) into one gb-month reading per (project, subject, tier), in the++ * transaction that marks the day. A location is counted once even when several objects name it (F3.4): the++ * subquery keeps, per storage identity (provider + key), the oldest non-deleted object naming it and that++ * location's bytes, so the sum bills the bytes once to that object's owner. First-party and sandbox tenants are++ * dropped. Each reading carries the day's share of a GB-month (bytes / GB / days in the UTC month), so a month++ * of daily readings sums to one GB-month. A re-run of the day inserts nothing.++ */++ async function aggregateStorageDay(periodStart, { now: at = now() } = {}) {++ if (periodStart !== dayStart(periodStart)) throw new RangeError(`billing: ${periodStart} is not the start of a day`);++ if (periodStart + DAY_MS + GRACE_MS > at) throw new RangeError(`billing: the day ${iso(periodStart)} has not closed`);++ const rows = await db.prepare(`WITH named AS (++ SELECT DISTINCT ON (l.provider, l.key)++ l.provider AS tier, COALESCE(l.size_bytes, o.size_bytes) AS bytes, o.owner_subject, o.app_id++ FROM media_locations l JOIN media_objects o ON o.id = l.object_id++ WHERE l.state = 'present' AND o.lifecycle_status <> 'deleted'++ ORDER BY l.provider, l.key, o.created_at, o.id++ )++ SELECT a.project_id, n.owner_subject, n.tier, SUM(n.bytes) AS bytes++ FROM named n JOIN apps a ON a.app_id = n.app_id++ WHERE a.env = 'production' AND a.project_id IS NOT NULL++ GROUP BY a.project_id, n.owner_subject, n.tier`).all();++ const days = daysInMonth(periodStart);++ return await db.tx(async (t) => {++ if (await periodDone(STORAGE, periodStart)) return { period: periodStart, readings: 0, created: 0, skipped: true };++ const day = iso(periodStart);++ let created = 0, count = 0;++ for (const row of rows) {++ const bytes = Number(row.bytes || 0);++ if (!(bytes > 0)) continue;++ const project = String(row.project_id), subject = row.owner_subject || undefined, tier = String(row.tier);++ const key = reporter.key('store', project, subject || '-', tier, day);++ const reading = reporter.sample({ idempotency_key: key, project, subject,++ resource: STORAGE, provider: tier, operation: 'store', quantity: (bytes / GB) / days, unit: 'GB', at: day });++ if (await queue(t, reading)) created++;++ count++;++ }++ await t.exec(`INSERT INTO billing_periods (metric, period_start, readings, aggregated_at) VALUES ($1, $2, $3, $4) ON CONFLICT DO NOTHING`,++ [STORAGE, periodStart, count, at]);++ return { period: periodStart, readings: count, created };++ });++ }++++ /**++ * Aggregate the newest closed UTC day. The storage query reads media_locations' current state, so a past day can++ * never be reconstructed later: only this day is aggregated, and a day missed while the service was down is++ * simply not billed (a reading must not claim today's state for an older day).++ */++ async function aggregateClosed({ now: at = now() } = {}) {++ const newestDay = dayStart(at - GRACE_MS) - DAY_MS;++ const r = await aggregateStorageDay(newestDay, { now: at });++ return { created: r.created };++ }++++ /** One pass: aggregate the newest closed day, then relay what is due. Never overlaps itself; never throws. */++ let running = null;++ function tick() {++ if (!running) running = (async () => {++ try {++ const a = await aggregateClosed();++ const s = await reporter.flush();++ return { ...a, ...s };++ } catch (err) {++ lastError = err.message;++ log.error(`[Billing] tick failed: ${err.message}`);++ return null;++ }++ })().finally(() => { running = null; });++ return running;++ }++++ let timer = null;++ return {++ enabled: intervalMs > 0,++ reporting: !!tokenClient,++ reporter,++ aggregateStorageDay, aggregateClosed, tick,++ /** Queue one reading in its own transaction (tests and ad-hoc re-queues). */++ async record(reading) { return await db.tx(async (t) => queue(t, reading)); },++ /** Relay due readings now (alias of the reporter's flush). */++ send: () => reporter.flush(),++ start() {++ if (timer || !(intervalMs > 0)) return false;++ if (!tokenClient) log.warn('[Billing] OV_BILLING_URL or OV_OAUTH_CLIENT_SECRET is not set: readings are stored and stay queued');++ timer = setInterval(tick, intervalMs);++ if (timer.unref) timer.unref();++ return true;++ },++ async stop() {++ if (timer) clearInterval(timer);++ timer = null;++ if (running) await running;++ await reporter.stop();++ },++ get running() { return !!timer; },++ async status() {++ return {++ enabled: intervalMs > 0,++ reporting: !!tokenClient,++ pending: await reporter.pending(),++ rejected: await reporter.rejected(),++ aggregation: Object.fromEntries((await db.prepare(`SELECT metric, COUNT(*) AS n, MAX(period_start) AS last FROM billing_periods GROUP BY metric`).all())++ .map((r) => [r.metric, { periods: Number(r.n), last_period_start: r.last == null ? null : Number(r.last) }])),++ last_error: lastError,++ };++ },++ };++}++++module.exports = {++ createMediaBilling,++ DAY_MS, GRACE_MS, GB, TABLE, STORAGE, dayStart,++};@@@ -145,6 +145,18 @@ const config = {internalUrl: (process.env.OV_NETWORK_INTERNAL_URL || 'http://127.0.0.1:4000').replace(/\/$/, ''),},++ // Usage readings → OpenVibe.Billing (plan T5 step 14; server/billing.js). Off by default:++ // MEDIA_BILLING_INTERVAL_MS=0 starts no timer and aggregates/sends nothing. With an interval but no++ // OV_BILLING_URL (or no OV_OAUTH_CLIENT_SECRET) readings are aggregated and stay queued in billing_readings.++ billing: {++ intervalMs: Math.max(0, intEnv('MEDIA_BILLING_INTERVAL_MS', 0)),++ url: (process.env.OV_BILLING_URL || process.env.OV_BILLING_INTERNAL_URL || '').replace(/\/$/, ''),++ audience: process.env.OV_BILLING_AUDIENCE || 'openvibe.billing',++ clientId: process.env.OV_OAUTH_CLIENT_ID || 'media',++ clientSecret: process.env.OV_OAUTH_CLIENT_SECRET || '',++ timeoutMs: intEnv('MEDIA_BILLING_TIMEOUT_MS', 10000),++ },++// IndexNow (openvibe-shared/indexnow): a key makes search engines recrawl a public page when it// appears or changes (the key file is served at /<key>.txt). Unset: off, no key file, nothing sent.indexnow: { key: String(process.env.INDEXNOW_KEY || '').trim() },@@@ -246,6 +246,7 @@ const ready = (async () => {const thumbService = require('./thumbnails/thumbnail-service');const timers = [];++ let billing = null; // usage readings → Billing (started below only when MEDIA_BILLING_INTERVAL_MS > 0)function every(ms, fn) {const t = setInterval(fn, ms);if (t.unref) t.unref();@@@ -264,6 +265,13 @@ const ready = (async () => {require('./objects/verify-job').start(); // scheduled copy verification (bounded batches; never deletes)require('./jobs/worker').start().catch((err) => console.warn('[Jobs] worker did not start:', err.message)); // media_jobs worker (light + heavy lanes; proposals wait for their owner)require('./objects/owner-subject-job').start(); // owner_subject for objects that name only an app-local owner (asks Network)++ // Usage readings → Billing (plan T5 step 14): off by default (MEDIA_BILLING_INTERVAL_MS=0 starts nothing).++ // With an interval it aggregates each closed UTC day into one platform.usage-sample@1 gb-month reading and++ // relays it to billing.usage.record; without OV_BILLING_URL or OV_OAUTH_CLIENT_SECRET it stays queued.++ if (config.billing.intervalMs > 0) {++ billing = require('./billing').createMediaBilling({ db: db.getDb(), config: config.billing, networkUrl: config.network.internalUrl });++ billing.start();++ }// Descriptor watchdog: a leak here once pinned 80 GB of deleted recordings to the disk.every(5 * 60 * 1000, () => {try {@@@ -365,6 +373,7 @@ const ready = (async () => {() => require('./objects/verify-job').stop(),() => require('./jobs/worker').stop(),() => require('./objects/owner-subject-job').stop(),++ () => (billing ? billing.stop() : undefined),() => auth.stopJwksRefresh(),() => require('./events')._reset(),() => { const v = require('./actor-limits').valkey(); if (v) v.close().catch(() => {}); },nnew file mode 100644@@@ -0,0 +1,188 @@++'use strict';++// Usage readings → Billing (plan T5 step 14, Media slice; server/billing.js, migrations/0004_billing_readings.sql).++// One closed UTC day aggregates into one gb-month reading per (project, subject, tier): that day's share of a++// GB-month (bytes / GB / days in the UTC month), every present location counted once (F3.4) and billed to the++// oldest object naming it. Only the newest closed day is aggregated — the query reads media_locations' current++// state, so a day missed during downtime is not billed. First-party and sandbox traffic is never recorded;++// re-running a period inserts nothing; a stored reading is never edited; and the relay's refused-vs-pending++// classification holds (only Billing refusing the reading itself is permanent). At MEDIA_BILLING_INTERVAL_MS=0++// nothing runs. Media emits gb-month only; gib-delivered is not emitted (route() calls are not bytes), so nothing++// here bills delivery.++const assert = require('assert');++const db = require('../server/db/database');++const { createMediaBilling, DAY_MS, GB, dayStart } = require('../server/billing');++++const PRJ = 'prj_01JAB2C3D4E5F6G7H8J9K0MNPQ';++const USR = 'usr_01JAB2C3D4E5F6G7H8J9K0SEED';++const USR_OLD = 'usr_01JAB2C3D4E5F6G7H8J9K0OLD1';++const USR_NEW = 'usr_01JAB2C3D4E5F6G7H8J9K0NEW2';++const USR_A = 'usr_01JAB2C3D4E5F6G7H8J9K0AAA1';++const USR_B = 'usr_01JAB2C3D4E5F6G7H8J9K0BBB2';++const FIXED = Date.UTC(2026, 9, 4, 12, 0, 0); // a fixed clock: nothing here depends on wall time++const settled = { intervalMs: 1000, url: '', audience: 'openvibe.billing', clientId: 'media', clientSecret: '', timeoutMs: 2000 };++const iso = (ms) => new Date(ms).toISOString();++++(async () => {++ await db.initDb();++ const json = (v) => (typeof v === 'string' ? JSON.parse(v) : v);++ const count = async (where = '1=1', params = []) => Number(await db.getDb().value(`SELECT COUNT(*) FROM billing_readings WHERE ${where}`, params));++ const rowOf = (key) => db.getDb().prepare('SELECT * FROM billing_readings WHERE event_id = ?').get(key);++ const periodStarts = async (metric) => (await db.getDb().prepare(`SELECT period_start FROM billing_periods WHERE metric = ? ORDER BY period_start`).all(metric)).map((r) => Number(r.period_start));++ // Only the locations need clearing between cases: the storage query joins locations to objects, so an object++ // with no location never bills (and objects are not deletable while a hold exists).++ const clearLocations = () => db.run('DELETE FROM media_locations');++ const addObject = (id, app, owner, size, created) => db.run(++ `INSERT INTO media_objects (id, app_id, namespace, kind, owner_subject, lifecycle_status, size_bytes, created_at)++ VALUES (?, ?, 'ns', 'file', ?, 'ready', ?, ?)`, [id, app, owner, size, created]);++ const addLocation = (objectId, provider, key, size) => db.run(++ `INSERT INTO media_locations (object_id, provider, key, state, size_bytes) VALUES (?, ?, ?, 'present', ?)`, [objectId, provider, key, size]);++++ // Two tenant classes plus a first-party app: only the production project tenant is billable.++ await db.run(`INSERT INTO apps (app_id, name, api_key_hash, project_id, env) VALUES ('live', 'live', 'k', NULL, NULL)`);++ await db.run(`INSERT INTO apps (app_id, name, api_key_hash, project_id, env) VALUES ('app.prod', 'prod', 'k', ?, 'production')`, [PRJ]);++ await db.run(`INSERT INTO apps (app_id, name, api_key_hash, project_id, env) VALUES ('app.sbx', 'sandbox', 'k', ?, 'sandbox')`, [PRJ]);++++ const billing = createMediaBilling({ db: db.getDb(), config: settled, clock: { now: () => FIXED } });++++ // 1. One closed UTC day → one gb-month reading per billable (project, subject, tier); first-party/sandbox out.++ await addObject('obj-prod', 'app.prod', USR, 2 * GB, '2026-01-01 00:00:00');++ await addObject('obj-sbx', 'app.sbx', USR, 5 * GB, '2026-01-01 00:00:00');++ await addObject('obj-first', 'live', USR, 9 * GB, '2026-01-01 00:00:00');++ await addLocation('obj-prod', 'local', 'k1', 2 * GB);++ await addLocation('obj-prod', 'r2', 'k2', 2 * GB);++ await addLocation('obj-sbx', 'local', 'k3', 5 * GB);++ await addLocation('obj-first', 'local', 'k4', 9 * GB);++++ const day = dayStart(FIXED) - 2 * DAY_MS; // 2026-10-02++ const first = await billing.aggregateStorageDay(day, { now: FIXED });++ assert.strictEqual(first.readings, 2, 'local and r2, not sandbox or first-party');++ assert.strictEqual(first.created, 2);++ const localKey = `media:store:${PRJ}:${USR}:local:${iso(day)}`;++ const r2Key = `media:store:${PRJ}:${USR}:r2:${iso(day)}`;++ const local = json((await rowOf(localKey)).envelope);++ assert.strictEqual(local.service, 'media');++ assert.strictEqual(local.project, PRJ);++ assert.strictEqual(local.subject, USR);++ assert.strictEqual(local.resource, 'gb-month');++ assert.strictEqual(local.operation, 'store');++ assert.strictEqual(local.unit, 'GB');++ assert.strictEqual(local.provider, 'local');++ assert.strictEqual(local.quantity, (2 * GB) / GB / 31, '2026-10 has 31 days: the day is 1/31 of a GB-month');++ assert.strictEqual(local.at, iso(day));++ assert.strictEqual(json((await rowOf(r2Key)).envelope).provider, 'r2');++++ // 2. Re-aggregating the day inserts nothing (the period is marked; the key is idempotent).++ const before = await count();++ const again = await billing.aggregateStorageDay(day, { now: FIXED });++ assert.strictEqual(again.created, 0);++ assert.strictEqual(again.skipped, true);++ assert.strictEqual(await count(), before, 'a re-run changes no row');++++ // 3. A stored reading is never edited: only the delivery columns move.++ await assert.rejects(++ db.run(`UPDATE billing_readings SET envelope = '{}'::jsonb WHERE event_id = ?`, [localKey]),++ /never edited/, 'the trigger refuses editing the reading');++ await db.run(`UPDATE billing_readings SET attempts = attempts + 1, next_attempt_at = ? WHERE event_id = ?`, [FIXED + 10, localKey]);++++ // 4. A daily reading is that day's share of a GB-month: 1 GB reads 1/30 in a 30-day month, 1/31 in a 31-day one.++ await clearLocations();++ await addObject('obj-month', 'app.prod', USR, GB, '2026-01-01 00:00:00');++ await addLocation('obj-month', 'local', 'month-key', GB);++ const sep = Date.UTC(2026, 8, 10); // 2026-09-10, September (30 days)++ const oct = Date.UTC(2026, 9, 10); // 2026-10-10, October (31 days)++ await billing.aggregateStorageDay(sep, { now: Date.UTC(2026, 9, 15) });++ await billing.aggregateStorageDay(oct, { now: Date.UTC(2026, 9, 15) });++ assert.strictEqual(json((await rowOf(`media:store:${PRJ}:${USR}:local:${iso(sep)}`)).envelope).quantity, 1 / 30);++ assert.strictEqual(json((await rowOf(`media:store:${PRJ}:${USR}:local:${iso(oct)}`)).envelope).quantity, 1 / 31);++++ // 5. A location more than one object names (F3.4) bills once, to the oldest object naming it (created_at, then id).++ await clearLocations();++ await addObject('obj-old', 'app.prod', USR_OLD, GB, '2026-01-01 00:00:00'); // older++ await addObject('obj-new', 'app.prod', USR_NEW, GB, '2026-02-01 00:00:00'); // newer, names the same location++ await addLocation('obj-old', 'local', 'shared', 3 * GB);++ await addLocation('obj-new', 'local', 'shared', 3 * GB);++ await addObject('obj-a1', 'app.prod', USR_A, GB, '2026-03-01 00:00:00'); // same created_at as obj-a2++ await addObject('obj-a2', 'app.prod', USR_B, GB, '2026-03-01 00:00:00');++ await addLocation('obj-a1', 'local', 'tie', 2 * GB);++ await addLocation('obj-a2', 'local', 'tie', 2 * GB);++++ const sharedDay = Date.UTC(2026, 10, 15); // 2026-11-15++ const dedup = await billing.aggregateStorageDay(sharedDay, { now: Date.UTC(2026, 11, 1) });++ assert.strictEqual(dedup.readings, 2, 'one reading per distinct owner: each shared location billed once');++ assert.strictEqual(json((await rowOf(`media:store:${PRJ}:${USR_OLD}:local:${iso(sharedDay)}`)).envelope).quantity, (3 * GB) / GB / 30,++ 'the shared location bills once, to the older object naming it');++ assert.strictEqual(await rowOf(`media:store:${PRJ}:${USR_NEW}:local:${iso(sharedDay)}`), undefined, 'the newer naming object is not billed');++ assert.strictEqual(json((await rowOf(`media:store:${PRJ}:${USR_A}:local:${iso(sharedDay)}`)).envelope).quantity, (2 * GB) / GB / 30,++ 'an equal created_at breaks to the lowest object id');++ assert.strictEqual(await rowOf(`media:store:${PRJ}:${USR_B}:local:${iso(sharedDay)}`), undefined, 'the higher object id is not billed on the tie');++++ // 6. Only the newest closed day is aggregated: after a 3-day gap the two missed days are not billed.++ await clearLocations();++ await addObject('obj-gap', 'app.prod', USR, GB, '2026-01-01 00:00:00');++ await addLocation('obj-gap', 'local', 'gap-key', GB);++ const t0 = Date.UTC(2026, 11, 20, 12, 0, 0); // 2026-12-20 12:00 → newest closed day 12-19++ const t3 = t0 + 3 * DAY_MS; // 2026-12-23 12:00 → newest closed day 12-22++ const gap1 = await billing.aggregateClosed({ now: t0 });++ const gap2 = await billing.aggregateClosed({ now: t3 });++ assert.strictEqual(gap1.created, 1);++ assert.strictEqual(gap2.created, 1, 'the gap run bills only the newest closed day');++ const dec = (await periodStarts('gb-month')).filter((ms) => ms >= Date.UTC(2026, 11, 19) && ms <= Date.UTC(2026, 11, 22));++ assert.deepStrictEqual(dec, [Date.UTC(2026, 11, 19), Date.UTC(2026, 11, 22)], 'the missed days 12-20 and 12-21 are not billed');++ assert.ok(await rowOf(`media:store:${PRJ}:${USR}:local:${iso(Date.UTC(2026, 11, 22))}`), 'the newest closed day has its reading');++ assert.strictEqual(await rowOf(`media:store:${PRJ}:${USR}:local:${iso(Date.UTC(2026, 11, 21))}`), undefined, 'a missed day has none');++++ // 7. Refused vs pending: only Billing refusing the reading itself (any other 4xx) is permanent.++ const statuses = {++ 'k-refuse-422': 422, 'k-refuse-400': 400, 'k-pending-503': 503, 'k-pending-500': 500,++ 'k-pending-401': 401, 'k-pending-403': 403, 'k-pending-404': 404, 'k-pending-408': 408,++ 'k-pending-425': 425, 'k-pending-429': 429, 'k-pending-net': 0,++ };++ const calls = [];++ const fetchImpl = async (url, opts) => {++ if (String(url).endsWith('/oauth/token')) return { ok: true, status: 200, json: async () => ({ access_token: 'tok', expires_in: 300 }) };++ const sent = JSON.parse(opts.body);++ const status = statuses[sent.idempotency_key];++ calls.push(sent.idempotency_key);++ if (status === 0) throw new Error('connection refused');++ return { ok: status < 400, status, text: async () => `status ${status}`, json: async () => ({}) };++ };++ const talking = createMediaBilling({++ db: db.getDb(), clock: { now: () => FIXED }, fetchImpl, networkUrl: 'http://billing.test',++ config: { ...settled, url: 'http://billing.test', clientSecret: 's' },++ });++ // Push the aggregation readings out of the relay's way: this part posts one reading at a time (sequentially).++ await db.run(`UPDATE billing_readings SET next_attempt_at = ? WHERE event_id NOT LIKE 'k-%'`, [FIXED + 10 * DAY_MS]);++ await Object.entries(statuses).reduce(async (previous, [k, status]) => {++ await previous;++ await talking.record(talking.reporter.sample({ idempotency_key: k, project: PRJ, subject: USR,++ resource: 'gb-month', provider: 'local', operation: 'store', quantity: 1, unit: 'GB', at: iso(day) }));++ await talking.send(); // a pending row's backoff keeps it out of the next pass++ const r = await rowOf(k);++ if (status && status >= 400 && status < 500 && ![401, 403, 404, 408, 425, 429].includes(status)) {++ assert.ok(r.rejected_at, `${status} refuses the reading permanently`);++ assert.ok(!r.sent_at, `${status} is never sent`);++ } else {++ assert.ok(!r.rejected_at, `${status || 'network'} stays pending`);++ assert.ok(!r.sent_at, `${status || 'network'} is not marked sent`);++ assert.ok(Number(r.next_attempt_at) > FIXED || status === 0, `${status || 'network'} backs off`);++ }++ }, Promise.resolve());++ assert.ok(calls.length >= 1, 'the relay posted to Billing');++++ // 8. MEDIA_BILLING_INTERVAL_MS=0 runs nothing: no timer, no aggregation, no reading.++ const periods = Number(await db.getDb().value('SELECT COUNT(*) FROM billing_periods'));++ const rows = await count();++ const off = createMediaBilling({ db: db.getDb(), config: { ...settled, intervalMs: 0, url: 'http://billing.test' }, clock: { now: () => FIXED } });++ assert.strictEqual(off.enabled, false);++ assert.strictEqual(off.start(), false);++ assert.strictEqual(off.running, false);++ await new Promise((r) => setTimeout(r, 80));++ assert.strictEqual(Number(await db.getDb().value('SELECT COUNT(*) FROM billing_periods')), periods, 'nothing aggregates at interval 0');++ assert.strictEqual(await count(), rows, 'nothing is queued at interval 0');++ await off.stop();++++ await talking.stop();++ await billing.stop();++ console.log('✅ media billing: daily gb-month readings per project/subject/tier (a month sums to one GB-month), shared locations billed once to the oldest owner, newest closed day only, first-party and sandbox excluded, idempotent and immutable, refused-vs-pending relay, off at interval 0');++ process.exit(0);++})().catch((err) => { console.error(err); process.exit(1); });@@@ -9,17 +9,21 @@ import { fileURLToPath } from 'node:url';const require = createRequire(import.meta.url);const root = path.join(path.dirname(fileURLToPath(import.meta.url)), '..', '..');const { createTestDb } = require('openvibe-sdk/testing');--// openvibe-sdk >= 0.26 createTestDb holds a lease pg.Client (application_name *-test-lease) whose socket is not unref'd,--// so a test file that ends without process.exit() never exits. Unref it here; the lease still holds while the process lives.--// Drop this once Media pins openvibe-sdk >= 0.31.1, which unrefs the lease itself.--const pg = require(require.resolve('pg', { paths: [require.resolve('openvibe-sdk/testing')] }));--const connect = pg.Client.prototype.connect;--pg.Client.prototype.connect = function (...args) {-- const r = connect.apply(this, args);-- if (String(this.connectionParameters && this.connectionParameters.application_name).endsWith('-test-lease')) this.connection.stream.unref();-- return r;--};++// openvibe-sdk >= 0.31.1 unrefs createTestDb's lease pg.Client itself (MEDIA_TEST_STORE=pg), so a test file that++// ends without process.exit() exits. On PGlite, openvibe-sdk >= 0.29 reuses a migrated snapshot (OV_TEST_SNAPSHOT=0++// opts out) and loadDataDir starts PGlite's Emscripten alarm — a ref'd setTimeout the wasm runtime keeps++// rescheduling — which holds the event loop open the same way (a pure test, or a harness that exits only on++// failure, would never exit and the per-file runner would time it out). This is Media-side and only for the timers++// PGlite's own wasm schedules: drop once openvibe-sdk unrefs PGlite's alarm itself (the pin, v0.32.0, does not++// yet), then every timer a test sets stays ref'd.++const realSetTimeout = globalThis.setTimeout, realSetInterval = globalThis.setInterval;++const fromPglite = () => { const s = new Error().stack; return !!s && s.includes('@electric-sql/pglite'); };++globalThis.setTimeout = function (...a) { const h = realSetTimeout.apply(this, a); if (fromPglite()) { try { h.unref?.(); } catch { /* already gone */ } } return h; };++globalThis.setInterval = function (...a) { const h = realSetInterval.apply(this, a); if (fromPglite()) { try { h.unref?.(); } catch { /* already gone */ } } return h; };++// A ref'd keep-alive while the snapshot loads/builds: an unref'd setup timer must still fire.++const keepAlive = realSetInterval(() => {}, 1000);const t = await createTestDb({ migrations: path.join(root, 'migrations'), store: process.env.MEDIA_TEST_STORE || 'pglite', service: 'media', max: 4 });++clearInterval(keepAlive);// Tests insert rows with small explicit ids. A PostgreSQL identity would continue after them, so generated ids// are set to start at 100000 here, clear of every id a test names.for (const r of await t.db.prepare(`SELECT pg_get_serial_sequence(quote_ident(table_name), column_name) AS seq FROM information_schema.columns
@@ -45,7 +45,7 @@ without `DATABASE_URL`, development uses an embedded PGlite database in `data/pg
@@ -45,7 +45,7 @@ without `DATABASE_URL`, development uses an embedded PGlite database in `data/pg
export and deletion), OpenVibe.Live (`live.lineage.resolve`), OpenVibe.Events (the outbox relay and
export and deletion), OpenVibe.Live (`live.lineage.resolve`), OpenVibe.Events (the outbox relay and
the account and revocation subscriptions)
the account and revocation subscriptions)
- Backblaze B2 and Cloudflare R2 (S3 API) when configured; `ffmpeg`/`ffprobe` on the host
- Backblaze B2 and Cloudflare R2 (S3 API) when configured; `ffmpeg`/`ffprobe` on the host
-- `openvibe-contracts` v0.79.0, `openvibe-sdk` v0.28.0 (tokens, events outbox, per-actor limits),
+- `openvibe-contracts` v0.79.0, `openvibe-sdk` v0.32.0 (tokens, events outbox, usage readings, per-actor limits),
`openvibe-shared` v2.10.0, pinned by release tarball
`openvibe-shared` v2.10.0, pinned by release tarball
## Capabilities
## Capabilities
@@ -117,7 +117,7 @@ server/
@@ -117,7 +117,7 @@ server/
storage.orphans.scan (monthly storage orphan report, service-wide, report only),
storage.orphans.scan (monthly storage orphan report, service-wide, report only),
storage.move.cleanup (the alert for placement moves whose delete keeps failing, service-wide)
storage.move.cleanup (the alert for placement moves whose delete keeps failing, service-wide)
client-ip.js trust proxy = loopback; req.ip is the only client address
client-ip.js trust proxy = loopback; req.ip is the only client address
-(openvibe-shared v2.10.0, openvibe-contracts v0.79.0, openvibe-sdk v0.28.0: pinned release tarballs, installed by npm)
+(openvibe-shared v2.10.0, openvibe-contracts v0.79.0, openvibe-sdk v0.32.0: pinned release tarballs, installed by npm)
scripts/smoke-test.sh end-to-end smoke test (boots a temp instance)
scripts/smoke-test.sh end-to-end smoke test (boots a temp instance)
scripts/reconcile-objects.js / object-invariant.js / object-drift-report.js / no-good-copy-report.js object-model operator tools
scripts/reconcile-objects.js / object-invariant.js / object-drift-report.js / no-good-copy-report.js object-model operator tools
scripts/media-jobs.js list jobs, run the size-invariant scan (dry run by default), approve/cancel proposals
scripts/media-jobs.js list jobs, run the size-invariant scan (dry run by default), approve/cancel proposals
@@ -440,6 +440,40 @@ before the route does any work, and one `[Limits]` log line and `media_rate_limi
@@ -440,6 +440,40 @@ before the route does any work, and one `[Limits]` log line and `media_rate_limi
Never limited: `/healthz`, `/api/ready`, `/release.json`, `/metrics`, the signed `/internal/events` deliveries.
Never limited: `/healthz`, `/api/ready`, `/release.json`, `/metrics`, the signed `/internal/events` deliveries.
Counters are per process (a restart forgets them). `test/actor-limits.test.js`.
Counters are per process (a restart forgets them). `test/actor-limits.test.js`.
+## Usage readings → Billing
+
+Media's storage usage goes to OpenVibe.Billing as one `platform.usage-sample@1` reading per (project, subject,
+tier) where a project's work is billable (`server/billing.js`, plan T5 step 14):
+
+| metric (`resource`) | what | period | unit | provider |
+|---|---|---|---|---|
+| `gb-month` | bytes stored: every present copy in `media_locations`, summed by tier, counted once when more than one object names a location and billed to the oldest owner, as that day's share of a GB-month | closed UTC day | `GB` | the tier (`local`/`b2`/`r2`) |
+
+**Media emits `gb-month` only.** `gib-delivered` is deliberately not emitted: `demand.record()` counts `route()`
+calls, not bytes, so reads × `size_bytes` would bill aborted downloads and range reads in full, would bill
+presigned B2/R2 redirects Media never serves, and would miss HLS segment reads. It comes back once responses are
+metered by the bytes actually written (`sendSlice`/`streamFileWithRange`) and redirected reads are metered by CDN
+logs. Nothing here bills delivery.
+
+Only the newest closed UTC day is aggregated, because the query reads `media_locations`' current state: a day
+missed while the service was down is not billed (a reading never claims today's state for an older day).
+
+First-party traffic (a tenant with no project) and sandbox tenants are never recorded; an object without an
+`owner_subject` is still stored with its project (Billing leaves it unrated until it names a user subject).
+The SDK reporter (`openvibe-sdk/usage createUsageReporter`) queues each reading idempotency-keyed in
+`billing_readings` inside the aggregation's transaction and relays it to Billing's `POST /api/v1/usage`
+(`billing.usage.record`) with Media's Network service token (`clientId` `media`, audience `openvibe.billing`;
+OpenVibe.Network grants exactly that). A relay that cannot reach Billing (no grant, a missing route, a timeout,
+backpressure, a 401/403/404/408/425/429 or a 5xx) stays pending with backoff across restarts; only Billing
+refusing the reading itself (any other 4xx) is marked rejected and never retried, so nothing is billed twice.
+A stored reading is never edited (a trigger). Re-aggregating a day inserts nothing
+(`billing_periods`).
+
+**Off by default.** `MEDIA_BILLING_INTERVAL_MS=0` (the default) starts no timer and aggregates nothing; a
+restore drill starts nothing at all. Turn it on with `MEDIA_BILLING_INTERVAL_MS` plus `OV_BILLING_URL` and,
+only if it differs, `OV_BILLING_AUDIENCE`; without a billing URL or `OV_OAUTH_CLIENT_SECRET` readings are
+aggregated and stay queued. The runbook is `docs/cutover-media-billing-readings.md`; `test/billing.test.js`.
+
## Health, readiness and metrics
## Health, readiness and metrics
| path | behavior |
| path | behavior |
new file mode 100644
new file mode 100644
@@ -0,0 +1,90 @@
@@ -0,0 +1,90 @@
+# Cutover runbook — Media storage readings to OpenVibe.Billing (plan T5 step 14)
+
+This change makes Media a producer of `platform.usage-sample@1` readings for its own storage:
+`migrations/0004_billing_readings.sql` adds the SDK usage reporter's outbox plus Media's aggregation state, and
+`server/billing.js` aggregates each closed UTC day into one `gb-month` reading per (project, subject, tier) and
+relays it to OpenVibe.Billing's `POST /api/v1/usage`. Nothing is renamed, dropped, altered or backfilled, and the
+reads and writes that serve media are untouched. No money moves here: Billing stores the readings and rates them in
+its own sweep (its `BILLING_RATING_INTERVAL_MS`, off by default, and no rate card loaded = no charge).
+
+Media emits `gb-month` only. `gib-delivered` is deliberately not emitted: `demand.record()` counts `route()` calls,
+not bytes, so reads × `size_bytes` would bill aborted downloads and range reads in full, would bill presigned B2/R2
+redirects Media never serves, and would miss HLS segment reads. It comes back once responses are metered by the
+bytes actually written (`sendSlice`/`streamFileWithRange`) and redirected reads are metered by CDN logs. Nothing
+here bills delivery.
+
+## What the migration does
+
+`migrations/0004_billing_readings.sql` (`phase: expand`) creates two tables and a trigger:
+
+- `billing_readings` — the `openvibe-sdk/usage` `createUsageReporter` outbox: `event_id` (the reading's
+ idempotency key, unique), `envelope` (the reading JSON), and the relay's columns (`attempts`,
+ `next_attempt_at`, `sent_at`, `seq`, `rejected_at`, `last_error`). The reporter only ever moves the delivery
+ columns; the trigger `billing_readings_frozen` refuses every change to `event_id`/`envelope`/`created_at`, so a
+ stored reading is immutable and a retry always posts the same body. A generated `idempotency_key` column names
+ `event_id` for readers.
+- `billing_periods` — one row per `(metric, period_start)` already aggregated (`gb-month`: the UTC day), so a
+ re-run inserts nothing.
+
+The runner is `openvibe-sdk/db`'s `migrate()`, started by `server/db/database.js` as the owner
+(`DATABASE_DIRECT_URL`) when the process boots: it locks `ov_migrations`, runs the file in one transaction,
+records id `0004` with its checksum, and refuses a later edit of it.
+
+## What is recorded
+
+| metric (`resource`) | source | period | unit | provider |
+|---|---|---|---|---|
+| `gb-month` | `media_locations` present copies, summed by provider, each location counted once when more than one object names it and billed to the oldest owner, as that day's share of a GB-month (`bytes / GB / days in that UTC month`) | closed UTC day | `GB` | the tier (`local`/`b2`/`r2`) |
+
+One reading per (project, subject, tier), so a month of daily readings sums to one GB-month per tier. Only
+`apps.env = 'production'` tenants with a `project_id` are recorded (first-party and sandbox are dropped before
+`record()`); an object without an `owner_subject` is stored with its project alone and Billing leaves it unrated
+until it names a user subject.
+
+Only the newest closed UTC day is aggregated: the storage query reads `media_locations`' current state, so a day
+missed while the service was down is simply not billed — a reading never claims today's state for an older day.
+Nothing here reads Valkey.
+
+## Order
+
+1. **Backup.** Take the logical dump of the `media` database before the deploy (the rule for any data change,
+ even additive): `sudo ovhost backup --all --logical`. Note the stamp it prints; the way back uses it.
+2. **Merge and deploy** through the pipeline (`ovhost deploy media`; `--wait-idle` is not needed: no recording
+ state changes shape). The first process to boot applies `0004`.
+3. **Leave `MEDIA_BILLING_INTERVAL_MS` unset (0) at first.** Nothing is aggregated, no reading is queued and
+ nothing is posted.
+4. **Network grant** (OpenVibe.Network, its own change): the `media` client needs `billing.usage.record` on
+ audience `openvibe.billing`. Network's `server/identity/principals.js` already grants it; confirm it is
+ deployed before step 5. Without it every token request for that audience is refused, each post is logged as
+ `[usage] media reading not delivered (will retry)`, and the readings wait in the table (no media read is
+ affected).
+5. **Turn it on**: add `MEDIA_BILLING_INTERVAL_MS` (for example `300000`, every 5 minutes) and `OV_BILLING_URL`
+ (Billing's internal URL) to `/etc/openvibe/media.env`, and `OV_BILLING_AUDIENCE` only if it differs from
+ `openvibe.billing`; then restart through `ovhost deploy media --restart`.
+
+## Verification
+
+- `ov access run openvibe-ovh health media` (or `/api/ready`) answers 200 after the deploy.
+- The migration is recorded: `SELECT id, name, phase FROM ov_migrations WHERE id = '0004'` returns
+ `0004 | billing_readings | expand`, and with the interval still 0 `SELECT COUNT(*) FROM billing_readings` is 0.
+- After step 5: `SELECT metric, COUNT(*) AS n, MAX(period_start) AS last FROM billing_periods GROUP BY metric`
+ shows the newest closed day appearing; `SELECT COUNT(*) FROM billing_readings WHERE sent_at IS NULL
+ AND rejected_at IS NULL` stays small. `GET /api/v1/usage?service=media` in Billing (its `billing.ledger.admin`
+ capability) lists the readings with `idempotency_key` `media:store:<project>:<subject>:<tier>:<day>`.
+- A reading is never edited: `UPDATE billing_readings SET envelope = '{}'::jsonb WHERE …` fails with
+ `billing_readings: reading … is never edited`.
+- A relay that cannot reach Billing leaves rows pending; a reading Billing refuses (any other 4xx) is marked
+ `rejected_at` with `last_error` and never retried.
+
+## Rollback
+
+- **Code only** (the usual way back): `ovhost rollback media` (or `deploy/scripts/deploy.sh --rollback`). The
+ previous release does not know the tables and ignores them; the applied migration is accepted (an applied id
+ the release does not have is not an error). Pending readings stay and are posted when the release comes back
+ (Billing dedupes on the reading's idempotency key).
+- **Stop posting without a rollback:** remove `MEDIA_BILLING_INTERVAL_MS` (or set it to 0) and restart. Readings
+ already queued stay in `billing_readings` and wait.
+- **Remove the tables** (only if they must go; a contract step, by hand as the owner, after a code rollback):
+ `DROP TABLE billing_readings; DROP TABLE billing_periods; DELETE FROM ov_migrations WHERE id = '0004';`.
+- **Restore:** the dump from step 1 restores the database as it was before the deploy (OpenVibe.Host
+ `docs/backups.md`); not needed for this additive change unless the database itself is damaged.
@@ -378,6 +378,14 @@ request/playback → telemetry → Events → {Hotness, Health, Cost} engines
@@ -378,6 +378,14 @@ request/playback → telemetry → Events → {Hotness, Health, Cost} engines
(GraphQL), B2 usage reports, Bunny statistics API are imported daily, compared with the estimate, and the gap is a
(GraphQL), B2 usage reports, Bunny statistics API are imported daily, compared with the estimate, and the gap is a
metric. Budgets per class and provider with a forecast; `media.provider.cost_threshold` at 90 % shifts non-critical traffic.
metric. Budgets per class and provider with a forecast; `media.provider.cost_threshold` at 90 % shifts non-critical traffic.
+**Media's own usage readings (plan T5 step 14):** Media emits `gb-month` — one `platform.usage-sample@1` reading per
+(project, subject, tier) per closed UTC day, that day's share of a GB-month from the present `media_locations` copies
+(`server/billing.js`, off by default with `MEDIA_BILLING_INTERVAL_MS`). `gib-delivered` is deliberately not emitted:
+`demand.record()` counts `route()` calls, not bytes, so reads × `size_bytes` would bill aborted downloads and range
+reads in full, would bill presigned B2/R2 redirects Media never serves, and would miss HLS segment reads. It comes
+back once responses are metered by the bytes actually written (`sendSlice`/`streamFileWithRange`) and redirected
+reads are metered by CDN logs. Nothing here bills delivery.
+
## 10. Build order
## 10. Build order
- **F1 foundation (shipped):** provider registry + capability probes; `media.cost_tiers` + per-class budgets in a revisioned
- **F1 foundation (shipped):** provider registry + capability probes; `media.cost_tiers` + per-class budgets in a revisioned
new file mode 100644
new file mode 100644
@@ -0,0 +1,52 @@
@@ -0,0 +1,52 @@
+-- phase: expand
+-- Plan T5 step 14 (govern universal), Media slice: the outbox of platform.usage-sample@1 readings Media posts to
+-- OpenVibe.Billing (POST /api/v1/usage, capability billing.usage.record; server/billing.js) plus the aggregation
+-- state that decides which closed UTC periods were already aggregated.
+--
+-- billing_readings is the openvibe-sdk/usage createUsageReporter outbox (outboxSchema('billing_readings')): one row per
+-- reading, `event_id` is the reading's idempotency_key (unique: re-queueing a period inserts nothing, ON CONFLICT DO
+-- NOTHING) and `envelope` is the reading JSON. The reporter never invents a reading: it 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 stored reading is immutable: a retry always posts the same body, and Billing's idempotency
+-- key sees the same reading.
+--
+-- billing_periods: one row per (metric, closed UTC period start) already aggregated, so a re-run of an hour or day
+-- inserts nothing even when it produced no reading (a period with no billable use leaves no reading).
+-- Additive only: two new tables and their indexes/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();
+
+-- The closed periods already aggregated, per metric (gb-month: the UTC day).
+CREATE TABLE billing_periods (
+ metric text COLLATE "C" NOT NULL,
+ period_start bigint NOT NULL,
+ readings bigint NOT NULL DEFAULT 0,
+ aggregated_at bigint NOT NULL,
+ PRIMARY KEY (metric, period_start)
+);
@@ -18,7 +18,7 @@
@@ -18,7 +18,7 @@
"iovalkey": "^0.4.0",
"iovalkey": "^0.4.0",
"multer": "^1.4.5-lts.1",
"multer": "^1.4.5-lts.1",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.79.0",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.79.0",
- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.28.0",
+ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.32.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.10.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.10.0",
"pg": "^8.23.0",
"pg": "^8.23.0",
"sharp": "^0.33.4"
"sharp": "^0.33.4"
@@ -2026,9 +2026,9 @@
@@ -2026,9 +2026,9 @@
}
}
},
},
"node_modules/openvibe-sdk": {
"node_modules/openvibe-sdk": {
- "version": "0.28.0",
- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.28.0",
- "integrity": "sha512-jjLisFDWBN+xDVsixa5BJRr3355kRijmvOh1/9rz4s1yVb3wZWCbpTLNcOvXm57ksnRoqK57393W38G0Ey3PFw==",
+ "version": "0.32.0",
+ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.32.0",
+ "integrity": "sha512-O/54OWx6cxR00O9D7BMLkTyWXtS+Jq06sbszrYOcSMy6D6VhSqwFMMh23+Mew01W1HgxorGcUcxqhnt3BrdTUQ==",
"license": "MIT",
"license": "MIT",
"engines": {
"engines": {
"node": ">=20"
"node": ">=20"
@@ -22,7 +22,7 @@
@@ -22,7 +22,7 @@
"iovalkey": "^0.4.0",
"iovalkey": "^0.4.0",
"multer": "^1.4.5-lts.1",
"multer": "^1.4.5-lts.1",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.79.0",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.79.0",
- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.28.0",
+ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.32.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.10.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.10.0",
"pg": "^8.23.0",
"pg": "^8.23.0",
"sharp": "^0.33.4"
"sharp": "^0.33.4"
new file mode 100644
new file mode 100644
@@ -0,0 +1,218 @@
@@ -0,0 +1,218 @@
+'use strict';
+/**
+ * Media's storage usage readings → OpenVibe.Billing (plan T5 step 14, Media slice).
+ *
+ * Media emits one platform.usage-sample@1 metric, gb-month, the storage metric the platform.rate-card@1 contract
+ * names for it:
+ *
+ * gb-month the bytes stored for one closed UTC day, per (project, subject, tier): every present copy in
+ * media_locations, once a day ("daily partition"), in GB. resource 'gb-month', provider the tier
+ * (local/b2/r2), unit 'GB', operation 'store'. Each daily reading carries that day's share of a
+ * GB-month — bytes / GB / (days in that UTC month) — so a month of daily readings sums to one
+ * GB-month. A location more than one object names (F3.4) is counted once, billed to the oldest
+ * object naming it.
+ *
+ * gib-delivered is deliberately not emitted. demand.record() counts route() calls, not bytes: reads × size_bytes
+ * would bill aborted downloads and range reads in full, would bill presigned B2/R2 redirects Media never serves,
+ * and never sees HLS segment reads. It comes back once responses are metered by the bytes actually written
+ * (sendSlice/streamFileWithRange) and redirected reads are metered by CDN logs. Nothing here bills delivery.
+ *
+ * First-party traffic (a tenant with no project) and sandbox tenants are never recorded: the reporter bills whatever
+ * reading it is given, so the aggregate query joins apps and keeps only `env = 'production'` rows with a project id.
+ * A reading without the object's owner_subject is still stored (Billing leaves it unrated: rating needs a user
+ * subject).
+ *
+ * The reporter is the SDK's own (openvibe-sdk/usage createUsageReporter): it derives the sample, queues it
+ * idempotency-keyed in the outbox inside the aggregation's transaction, and relays it to Billing's
+ * billing.usage.record (POST /api/v1/usage) with Media's Network service token (createServiceTokenClient, clientId
+ * 'media', audience openvibe.billing; Network grants exactly billing.usage.record). 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, so nothing is billed twice. A reading is never edited (migrations/0004_billing_readings.sql).
+ *
+ * Off by default: MEDIA_BILLING_INTERVAL_MS=0 starts no timer and aggregates nothing; a restore drill starts
+ * nothing at all. With an interval but no OV_BILLING_URL (or OV_OAUTH_CLIENT_SECRET) readings are aggregated and
+ * stay queued.
+ *
+ * const billing = createMediaBilling({ db, config: config.billing, networkUrl: config.network.internalUrl });
+ * if (billing.enabled) billing.start();
+ */
+const { createUsageReporter } = require('openvibe-sdk/usage');
+const { createServiceTokenClient } = require('openvibe-sdk/auth');
+
+const DAY_MS = 24 * 60 * 60 * 1000;
+const GRACE_MS = 5 * 60 * 1000; // a day is aggregated only once this much of the next one passed
+const GB = 1000 * 1000 * 1000; // a GB in bytes: the unit gb-month is metered in
+const TABLE = 'billing_readings';
+const STORAGE = 'gb-month';
+const UNKNOWN_CONTRACT = /unknown contract/;
+
+const dayStart = (ms) => Math.floor(ms / DAY_MS) * DAY_MS;
+const iso = (ms) => new Date(ms).toISOString();
+/** Days in the UTC month containing `ms`: the divisor that turns a day's bytes into its share of a GB-month. */
+function daysInMonth(ms) { const d = new Date(ms); return new Date(Date.UTC(d.getUTCFullYear(), d.getUTCMonth() + 1, 0)).getUTCDate(); }
+
+/**
+ * @param {object} o
+ * @param {object} o.db the Media database (openvibe-sdk/db)
+ * @param {object} o.config config.billing (server/config.js)
+ * @param {string} [o.networkUrl] where Media'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 createMediaBilling({ 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: 'media', source: 'media.billing', table: TABLE,
+ billingUrl: config && config.url, tokenClient, audience: config && config.audience,
+ fetchImpl, timeoutMs: (config && config.timeoutMs) || 5000, now, log,
+ });
+ let lastError = null;
+
+ /**
+ * Queue one reading in the caller's transaction. The SDK reporter validates it against platform.usage-sample@1
+ * when the pinned openvibe-contracts knows the schema; a contracts that predates it (v0.79.0 here) makes
+ * validateUsageSample throw "unknown contract", which must not block the queue — the SDK's own rule is that a
+ * contracts without the schema leaves the reading unchecked, not refused. Only a real validation refusal throws.
+ */
+ async function queue(t, reading) {
+ try {
+ return await reporter.record(t, reading);
+ } catch (err) {
+ if (!UNKNOWN_CONTRACT.test(String((err && err.message) || ''))) throw err;
+ if (typeof reading.idempotency_key !== 'string' || !reading.idempotency_key) {
+ throw new TypeError('billing reading: idempotency_key is required (it is the outbox key)');
+ }
+ const n = await t.exec(`INSERT INTO ${TABLE} (event_id, envelope, created_at) VALUES ($1, $2, $3) ON CONFLICT (event_id) DO NOTHING`,
+ [reading.idempotency_key, JSON.stringify(reading), now()]);
+ return n > 0;
+ }
+ }
+
+ /** The closed period's row, if it was aggregated already. */
+ const periodDone = (metric, periodStart) =>
+ db.prepare(`SELECT readings FROM billing_periods WHERE metric = ? AND period_start = ?`).get(metric, periodStart);
+
+ /**
+ * Aggregate the closed UTC day [periodStart, +1d) into one gb-month reading per (project, subject, tier), in the
+ * transaction that marks the day. A location is counted once even when several objects name it (F3.4): the
+ * subquery keeps, per storage identity (provider + key), the oldest non-deleted object naming it and that
+ * location's bytes, so the sum bills the bytes once to that object's owner. First-party and sandbox tenants are
+ * dropped. Each reading carries the day's share of a GB-month (bytes / GB / days in the UTC month), so a month
+ * of daily readings sums to one GB-month. A re-run of the day inserts nothing.
+ */
+ async function aggregateStorageDay(periodStart, { now: at = now() } = {}) {
+ if (periodStart !== dayStart(periodStart)) throw new RangeError(`billing: ${periodStart} is not the start of a day`);
+ if (periodStart + DAY_MS + GRACE_MS > at) throw new RangeError(`billing: the day ${iso(periodStart)} has not closed`);
+ const rows = await db.prepare(`WITH named AS (
+ SELECT DISTINCT ON (l.provider, l.key)
+ l.provider AS tier, COALESCE(l.size_bytes, o.size_bytes) AS bytes, o.owner_subject, o.app_id
+ FROM media_locations l JOIN media_objects o ON o.id = l.object_id
+ WHERE l.state = 'present' AND o.lifecycle_status <> 'deleted'
+ ORDER BY l.provider, l.key, o.created_at, o.id
+ )
+ SELECT a.project_id, n.owner_subject, n.tier, SUM(n.bytes) AS bytes
+ FROM named n JOIN apps a ON a.app_id = n.app_id
+ WHERE a.env = 'production' AND a.project_id IS NOT NULL
+ GROUP BY a.project_id, n.owner_subject, n.tier`).all();
+ const days = daysInMonth(periodStart);
+ return await db.tx(async (t) => {
+ if (await periodDone(STORAGE, periodStart)) return { period: periodStart, readings: 0, created: 0, skipped: true };
+ const day = iso(periodStart);
+ let created = 0, count = 0;
+ for (const row of rows) {
+ const bytes = Number(row.bytes || 0);
+ if (!(bytes > 0)) continue;
+ const project = String(row.project_id), subject = row.owner_subject || undefined, tier = String(row.tier);
+ const key = reporter.key('store', project, subject || '-', tier, day);
+ const reading = reporter.sample({ idempotency_key: key, project, subject,
+ resource: STORAGE, provider: tier, operation: 'store', quantity: (bytes / GB) / days, unit: 'GB', at: day });
+ if (await queue(t, reading)) created++;
+ count++;
+ }
+ await t.exec(`INSERT INTO billing_periods (metric, period_start, readings, aggregated_at) VALUES ($1, $2, $3, $4) ON CONFLICT DO NOTHING`,
+ [STORAGE, periodStart, count, at]);
+ return { period: periodStart, readings: count, created };
+ });
+ }
+
+ /**
+ * Aggregate the newest closed UTC day. The storage query reads media_locations' current state, so a past day can
+ * never be reconstructed later: only this day is aggregated, and a day missed while the service was down is
+ * simply not billed (a reading must not claim today's state for an older day).
+ */
+ async function aggregateClosed({ now: at = now() } = {}) {
+ const newestDay = dayStart(at - GRACE_MS) - DAY_MS;
+ const r = await aggregateStorageDay(newestDay, { now: at });
+ return { created: r.created };
+ }
+
+ /** One pass: aggregate the newest closed day, then relay what is due. Never overlaps itself; never throws. */
+ let running = null;
+ function tick() {
+ if (!running) running = (async () => {
+ try {
+ const a = await aggregateClosed();
+ const s = await reporter.flush();
+ return { ...a, ...s };
+ } catch (err) {
+ lastError = err.message;
+ log.error(`[Billing] tick failed: ${err.message}`);
+ return null;
+ }
+ })().finally(() => { running = null; });
+ return running;
+ }
+
+ let timer = null;
+ return {
+ enabled: intervalMs > 0,
+ reporting: !!tokenClient,
+ reporter,
+ aggregateStorageDay, aggregateClosed, tick,
+ /** Queue one reading in its own transaction (tests and ad-hoc re-queues). */
+ async record(reading) { return await db.tx(async (t) => queue(t, reading)); },
+ /** Relay due readings now (alias of the reporter's flush). */
+ send: () => reporter.flush(),
+ start() {
+ if (timer || !(intervalMs > 0)) return false;
+ if (!tokenClient) log.warn('[Billing] OV_BILLING_URL or OV_OAUTH_CLIENT_SECRET is not set: readings are stored and stay queued');
+ timer = setInterval(tick, intervalMs);
+ if (timer.unref) timer.unref();
+ return true;
+ },
+ async stop() {
+ if (timer) clearInterval(timer);
+ timer = null;
+ if (running) await running;
+ await reporter.stop();
+ },
+ get running() { return !!timer; },
+ async status() {
+ return {
+ enabled: intervalMs > 0,
+ reporting: !!tokenClient,
+ pending: await reporter.pending(),
+ rejected: await reporter.rejected(),
+ aggregation: Object.fromEntries((await db.prepare(`SELECT metric, COUNT(*) AS n, MAX(period_start) AS last FROM billing_periods GROUP BY metric`).all())
+ .map((r) => [r.metric, { periods: Number(r.n), last_period_start: r.last == null ? null : Number(r.last) }])),
+ last_error: lastError,
+ };
+ },
+ };
+}
+
+module.exports = {
+ createMediaBilling,
+ DAY_MS, GRACE_MS, GB, TABLE, STORAGE, dayStart,
+};
@@ -145,6 +145,18 @@ const config = {
@@ -145,6 +145,18 @@ const config = {
internalUrl: (process.env.OV_NETWORK_INTERNAL_URL || 'http://127.0.0.1:4000').replace(/\/$/, ''),
internalUrl: (process.env.OV_NETWORK_INTERNAL_URL || 'http://127.0.0.1:4000').replace(/\/$/, ''),
},
},
+ // Usage readings → OpenVibe.Billing (plan T5 step 14; server/billing.js). Off by default:
+ // MEDIA_BILLING_INTERVAL_MS=0 starts no timer and aggregates/sends nothing. With an interval but no
+ // OV_BILLING_URL (or no OV_OAUTH_CLIENT_SECRET) readings are aggregated and stay queued in billing_readings.
+ billing: {
+ intervalMs: Math.max(0, intEnv('MEDIA_BILLING_INTERVAL_MS', 0)),
+ url: (process.env.OV_BILLING_URL || process.env.OV_BILLING_INTERNAL_URL || '').replace(/\/$/, ''),
+ audience: process.env.OV_BILLING_AUDIENCE || 'openvibe.billing',
+ clientId: process.env.OV_OAUTH_CLIENT_ID || 'media',
+ clientSecret: process.env.OV_OAUTH_CLIENT_SECRET || '',
+ timeoutMs: intEnv('MEDIA_BILLING_TIMEOUT_MS', 10000),
+ },
+
// IndexNow (openvibe-shared/indexnow): a key makes search engines recrawl a public page when it
// IndexNow (openvibe-shared/indexnow): a key makes search engines recrawl a public page when it
// appears or changes (the key file is served at /<key>.txt). Unset: off, no key file, nothing sent.
// appears or changes (the key file is served at /<key>.txt). Unset: off, no key file, nothing sent.
indexnow: { key: String(process.env.INDEXNOW_KEY || '').trim() },
indexnow: { key: String(process.env.INDEXNOW_KEY || '').trim() },
@@ -246,6 +246,7 @@ const ready = (async () => {
@@ -246,6 +246,7 @@ const ready = (async () => {
const thumbService = require('./thumbnails/thumbnail-service');
const thumbService = require('./thumbnails/thumbnail-service');
const timers = [];
const timers = [];
+ let billing = null; // usage readings → Billing (started below only when MEDIA_BILLING_INTERVAL_MS > 0)
function every(ms, fn) {
function every(ms, fn) {
const t = setInterval(fn, ms);
const t = setInterval(fn, ms);
if (t.unref) t.unref();
if (t.unref) t.unref();
@@ -264,6 +265,13 @@ const ready = (async () => {
@@ -264,6 +265,13 @@ const ready = (async () => {
require('./objects/verify-job').start(); // scheduled copy verification (bounded batches; never deletes)
require('./objects/verify-job').start(); // scheduled copy verification (bounded batches; never deletes)
require('./jobs/worker').start().catch((err) => console.warn('[Jobs] worker did not start:', err.message)); // media_jobs worker (light + heavy lanes; proposals wait for their owner)
require('./jobs/worker').start().catch((err) => console.warn('[Jobs] worker did not start:', err.message)); // media_jobs worker (light + heavy lanes; proposals wait for their owner)
require('./objects/owner-subject-job').start(); // owner_subject for objects that name only an app-local owner (asks Network)
require('./objects/owner-subject-job').start(); // owner_subject for objects that name only an app-local owner (asks Network)
+ // Usage readings → Billing (plan T5 step 14): off by default (MEDIA_BILLING_INTERVAL_MS=0 starts nothing).
+ // With an interval it aggregates each closed UTC day into one platform.usage-sample@1 gb-month reading and
+ // relays it to billing.usage.record; without OV_BILLING_URL or OV_OAUTH_CLIENT_SECRET it stays queued.
+ if (config.billing.intervalMs > 0) {
+ billing = require('./billing').createMediaBilling({ db: db.getDb(), config: config.billing, networkUrl: config.network.internalUrl });
+ billing.start();
+ }
// Descriptor watchdog: a leak here once pinned 80 GB of deleted recordings to the disk.
// Descriptor watchdog: a leak here once pinned 80 GB of deleted recordings to the disk.
every(5 * 60 * 1000, () => {
every(5 * 60 * 1000, () => {
try {
try {
@@ -365,6 +373,7 @@ const ready = (async () => {
@@ -365,6 +373,7 @@ const ready = (async () => {
() => require('./objects/verify-job').stop(),
() => require('./objects/verify-job').stop(),
() => require('./jobs/worker').stop(),
() => require('./jobs/worker').stop(),
() => require('./objects/owner-subject-job').stop(),
() => require('./objects/owner-subject-job').stop(),
+ () => (billing ? billing.stop() : undefined),
() => auth.stopJwksRefresh(),
() => auth.stopJwksRefresh(),
() => require('./events')._reset(),
() => require('./events')._reset(),
() => { const v = require('./actor-limits').valkey(); if (v) v.close().catch(() => {}); },
() => { const v = require('./actor-limits').valkey(); if (v) v.close().catch(() => {}); },
new file mode 100644
new file mode 100644
@@ -0,0 +1,188 @@
@@ -0,0 +1,188 @@
+'use strict';
+// Usage readings → Billing (plan T5 step 14, Media slice; server/billing.js, migrations/0004_billing_readings.sql).
+// One closed UTC day aggregates into one gb-month reading per (project, subject, tier): that day's share of a
+// GB-month (bytes / GB / days in the UTC month), every present location counted once (F3.4) and billed to the
+// oldest object naming it. Only the newest closed day is aggregated — the query reads media_locations' current
+// state, so a day missed during downtime is not billed. First-party and sandbox traffic is never recorded;
+// re-running a period inserts nothing; a stored reading is never edited; and the relay's refused-vs-pending
+// classification holds (only Billing refusing the reading itself is permanent). At MEDIA_BILLING_INTERVAL_MS=0
+// nothing runs. Media emits gb-month only; gib-delivered is not emitted (route() calls are not bytes), so nothing
+// here bills delivery.
+const assert = require('assert');
+const db = require('../server/db/database');
+const { createMediaBilling, DAY_MS, GB, dayStart } = require('../server/billing');
+
+const PRJ = 'prj_01JAB2C3D4E5F6G7H8J9K0MNPQ';
+const USR = 'usr_01JAB2C3D4E5F6G7H8J9K0SEED';
+const USR_OLD = 'usr_01JAB2C3D4E5F6G7H8J9K0OLD1';
+const USR_NEW = 'usr_01JAB2C3D4E5F6G7H8J9K0NEW2';
+const USR_A = 'usr_01JAB2C3D4E5F6G7H8J9K0AAA1';
+const USR_B = 'usr_01JAB2C3D4E5F6G7H8J9K0BBB2';
+const FIXED = Date.UTC(2026, 9, 4, 12, 0, 0); // a fixed clock: nothing here depends on wall time
+const settled = { intervalMs: 1000, url: '', audience: 'openvibe.billing', clientId: 'media', clientSecret: '', timeoutMs: 2000 };
+const iso = (ms) => new Date(ms).toISOString();
+
+(async () => {
+ await db.initDb();
+ const json = (v) => (typeof v === 'string' ? JSON.parse(v) : v);
+ const count = async (where = '1=1', params = []) => Number(await db.getDb().value(`SELECT COUNT(*) FROM billing_readings WHERE ${where}`, params));
+ const rowOf = (key) => db.getDb().prepare('SELECT * FROM billing_readings WHERE event_id = ?').get(key);
+ const periodStarts = async (metric) => (await db.getDb().prepare(`SELECT period_start FROM billing_periods WHERE metric = ? ORDER BY period_start`).all(metric)).map((r) => Number(r.period_start));
+ // Only the locations need clearing between cases: the storage query joins locations to objects, so an object
+ // with no location never bills (and objects are not deletable while a hold exists).
+ const clearLocations = () => db.run('DELETE FROM media_locations');
+ const addObject = (id, app, owner, size, created) => db.run(
+ `INSERT INTO media_objects (id, app_id, namespace, kind, owner_subject, lifecycle_status, size_bytes, created_at)
+ VALUES (?, ?, 'ns', 'file', ?, 'ready', ?, ?)`, [id, app, owner, size, created]);
+ const addLocation = (objectId, provider, key, size) => db.run(
+ `INSERT INTO media_locations (object_id, provider, key, state, size_bytes) VALUES (?, ?, ?, 'present', ?)`, [objectId, provider, key, size]);
+
+ // Two tenant classes plus a first-party app: only the production project tenant is billable.
+ await db.run(`INSERT INTO apps (app_id, name, api_key_hash, project_id, env) VALUES ('live', 'live', 'k', NULL, NULL)`);
+ await db.run(`INSERT INTO apps (app_id, name, api_key_hash, project_id, env) VALUES ('app.prod', 'prod', 'k', ?, 'production')`, [PRJ]);
+ await db.run(`INSERT INTO apps (app_id, name, api_key_hash, project_id, env) VALUES ('app.sbx', 'sandbox', 'k', ?, 'sandbox')`, [PRJ]);
+
+ const billing = createMediaBilling({ db: db.getDb(), config: settled, clock: { now: () => FIXED } });
+
+ // 1. One closed UTC day → one gb-month reading per billable (project, subject, tier); first-party/sandbox out.
+ await addObject('obj-prod', 'app.prod', USR, 2 * GB, '2026-01-01 00:00:00');
+ await addObject('obj-sbx', 'app.sbx', USR, 5 * GB, '2026-01-01 00:00:00');
+ await addObject('obj-first', 'live', USR, 9 * GB, '2026-01-01 00:00:00');
+ await addLocation('obj-prod', 'local', 'k1', 2 * GB);
+ await addLocation('obj-prod', 'r2', 'k2', 2 * GB);
+ await addLocation('obj-sbx', 'local', 'k3', 5 * GB);
+ await addLocation('obj-first', 'local', 'k4', 9 * GB);
+
+ const day = dayStart(FIXED) - 2 * DAY_MS; // 2026-10-02
+ const first = await billing.aggregateStorageDay(day, { now: FIXED });
+ assert.strictEqual(first.readings, 2, 'local and r2, not sandbox or first-party');
+ assert.strictEqual(first.created, 2);
+ const localKey = `media:store:${PRJ}:${USR}:local:${iso(day)}`;
+ const r2Key = `media:store:${PRJ}:${USR}:r2:${iso(day)}`;
+ const local = json((await rowOf(localKey)).envelope);
+ assert.strictEqual(local.service, 'media');
+ assert.strictEqual(local.project, PRJ);
+ assert.strictEqual(local.subject, USR);
+ assert.strictEqual(local.resource, 'gb-month');
+ assert.strictEqual(local.operation, 'store');
+ assert.strictEqual(local.unit, 'GB');
+ assert.strictEqual(local.provider, 'local');
+ assert.strictEqual(local.quantity, (2 * GB) / GB / 31, '2026-10 has 31 days: the day is 1/31 of a GB-month');
+ assert.strictEqual(local.at, iso(day));
+ assert.strictEqual(json((await rowOf(r2Key)).envelope).provider, 'r2');
+
+ // 2. Re-aggregating the day inserts nothing (the period is marked; the key is idempotent).
+ const before = await count();
+ const again = await billing.aggregateStorageDay(day, { now: FIXED });
+ assert.strictEqual(again.created, 0);
+ assert.strictEqual(again.skipped, true);
+ assert.strictEqual(await count(), before, 'a re-run changes no row');
+
+ // 3. A stored reading is never edited: only the delivery columns move.
+ await assert.rejects(
+ db.run(`UPDATE billing_readings SET envelope = '{}'::jsonb WHERE event_id = ?`, [localKey]),
+ /never edited/, 'the trigger refuses editing the reading');
+ await db.run(`UPDATE billing_readings SET attempts = attempts + 1, next_attempt_at = ? WHERE event_id = ?`, [FIXED + 10, localKey]);
+
+ // 4. A daily reading is that day's share of a GB-month: 1 GB reads 1/30 in a 30-day month, 1/31 in a 31-day one.
+ await clearLocations();
+ await addObject('obj-month', 'app.prod', USR, GB, '2026-01-01 00:00:00');
+ await addLocation('obj-month', 'local', 'month-key', GB);
+ const sep = Date.UTC(2026, 8, 10); // 2026-09-10, September (30 days)
+ const oct = Date.UTC(2026, 9, 10); // 2026-10-10, October (31 days)
+ await billing.aggregateStorageDay(sep, { now: Date.UTC(2026, 9, 15) });
+ await billing.aggregateStorageDay(oct, { now: Date.UTC(2026, 9, 15) });
+ assert.strictEqual(json((await rowOf(`media:store:${PRJ}:${USR}:local:${iso(sep)}`)).envelope).quantity, 1 / 30);
+ assert.strictEqual(json((await rowOf(`media:store:${PRJ}:${USR}:local:${iso(oct)}`)).envelope).quantity, 1 / 31);
+
+ // 5. A location more than one object names (F3.4) bills once, to the oldest object naming it (created_at, then id).
+ await clearLocations();
+ await addObject('obj-old', 'app.prod', USR_OLD, GB, '2026-01-01 00:00:00'); // older
+ await addObject('obj-new', 'app.prod', USR_NEW, GB, '2026-02-01 00:00:00'); // newer, names the same location
+ await addLocation('obj-old', 'local', 'shared', 3 * GB);
+ await addLocation('obj-new', 'local', 'shared', 3 * GB);
+ await addObject('obj-a1', 'app.prod', USR_A, GB, '2026-03-01 00:00:00'); // same created_at as obj-a2
+ await addObject('obj-a2', 'app.prod', USR_B, GB, '2026-03-01 00:00:00');
+ await addLocation('obj-a1', 'local', 'tie', 2 * GB);
+ await addLocation('obj-a2', 'local', 'tie', 2 * GB);
+
+ const sharedDay = Date.UTC(2026, 10, 15); // 2026-11-15
+ const dedup = await billing.aggregateStorageDay(sharedDay, { now: Date.UTC(2026, 11, 1) });
+ assert.strictEqual(dedup.readings, 2, 'one reading per distinct owner: each shared location billed once');
+ assert.strictEqual(json((await rowOf(`media:store:${PRJ}:${USR_OLD}:local:${iso(sharedDay)}`)).envelope).quantity, (3 * GB) / GB / 30,
+ 'the shared location bills once, to the older object naming it');
+ assert.strictEqual(await rowOf(`media:store:${PRJ}:${USR_NEW}:local:${iso(sharedDay)}`), undefined, 'the newer naming object is not billed');
+ assert.strictEqual(json((await rowOf(`media:store:${PRJ}:${USR_A}:local:${iso(sharedDay)}`)).envelope).quantity, (2 * GB) / GB / 30,
+ 'an equal created_at breaks to the lowest object id');
+ assert.strictEqual(await rowOf(`media:store:${PRJ}:${USR_B}:local:${iso(sharedDay)}`), undefined, 'the higher object id is not billed on the tie');
+
+ // 6. Only the newest closed day is aggregated: after a 3-day gap the two missed days are not billed.
+ await clearLocations();
+ await addObject('obj-gap', 'app.prod', USR, GB, '2026-01-01 00:00:00');
+ await addLocation('obj-gap', 'local', 'gap-key', GB);
+ const t0 = Date.UTC(2026, 11, 20, 12, 0, 0); // 2026-12-20 12:00 → newest closed day 12-19
+ const t3 = t0 + 3 * DAY_MS; // 2026-12-23 12:00 → newest closed day 12-22
+ const gap1 = await billing.aggregateClosed({ now: t0 });
+ const gap2 = await billing.aggregateClosed({ now: t3 });
+ assert.strictEqual(gap1.created, 1);
+ assert.strictEqual(gap2.created, 1, 'the gap run bills only the newest closed day');
+ const dec = (await periodStarts('gb-month')).filter((ms) => ms >= Date.UTC(2026, 11, 19) && ms <= Date.UTC(2026, 11, 22));
+ assert.deepStrictEqual(dec, [Date.UTC(2026, 11, 19), Date.UTC(2026, 11, 22)], 'the missed days 12-20 and 12-21 are not billed');
+ assert.ok(await rowOf(`media:store:${PRJ}:${USR}:local:${iso(Date.UTC(2026, 11, 22))}`), 'the newest closed day has its reading');
+ assert.strictEqual(await rowOf(`media:store:${PRJ}:${USR}:local:${iso(Date.UTC(2026, 11, 21))}`), undefined, 'a missed day has none');
+
+ // 7. Refused vs pending: only Billing refusing the reading itself (any other 4xx) is permanent.
+ const statuses = {
+ 'k-refuse-422': 422, 'k-refuse-400': 400, 'k-pending-503': 503, 'k-pending-500': 500,
+ 'k-pending-401': 401, 'k-pending-403': 403, 'k-pending-404': 404, 'k-pending-408': 408,
+ 'k-pending-425': 425, 'k-pending-429': 429, 'k-pending-net': 0,
+ };
+ const calls = [];
+ const fetchImpl = async (url, opts) => {
+ if (String(url).endsWith('/oauth/token')) return { ok: true, status: 200, json: async () => ({ access_token: 'tok', expires_in: 300 }) };
+ const sent = JSON.parse(opts.body);
+ const status = statuses[sent.idempotency_key];
+ calls.push(sent.idempotency_key);
+ if (status === 0) throw new Error('connection refused');
+ return { ok: status < 400, status, text: async () => `status ${status}`, json: async () => ({}) };
+ };
+ const talking = createMediaBilling({
+ db: db.getDb(), clock: { now: () => FIXED }, fetchImpl, networkUrl: 'http://billing.test',
+ config: { ...settled, url: 'http://billing.test', clientSecret: 's' },
+ });
+ // Push the aggregation readings out of the relay's way: this part posts one reading at a time (sequentially).
+ await db.run(`UPDATE billing_readings SET next_attempt_at = ? WHERE event_id NOT LIKE 'k-%'`, [FIXED + 10 * DAY_MS]);
+ await Object.entries(statuses).reduce(async (previous, [k, status]) => {
+ await previous;
+ await talking.record(talking.reporter.sample({ idempotency_key: k, project: PRJ, subject: USR,
+ resource: 'gb-month', provider: 'local', operation: 'store', quantity: 1, unit: 'GB', at: iso(day) }));
+ await talking.send(); // a pending row's backoff keeps it out of the next pass
+ const r = await rowOf(k);
+ if (status && status >= 400 && status < 500 && ![401, 403, 404, 408, 425, 429].includes(status)) {
+ assert.ok(r.rejected_at, `${status} refuses the reading permanently`);
+ assert.ok(!r.sent_at, `${status} is never sent`);
+ } else {
+ assert.ok(!r.rejected_at, `${status || 'network'} stays pending`);
+ assert.ok(!r.sent_at, `${status || 'network'} is not marked sent`);
+ assert.ok(Number(r.next_attempt_at) > FIXED || status === 0, `${status || 'network'} backs off`);
+ }
+ }, Promise.resolve());
+ assert.ok(calls.length >= 1, 'the relay posted to Billing');
+
+ // 8. MEDIA_BILLING_INTERVAL_MS=0 runs nothing: no timer, no aggregation, no reading.
+ const periods = Number(await db.getDb().value('SELECT COUNT(*) FROM billing_periods'));
+ const rows = await count();
+ const off = createMediaBilling({ db: db.getDb(), config: { ...settled, intervalMs: 0, url: 'http://billing.test' }, clock: { now: () => FIXED } });
+ assert.strictEqual(off.enabled, false);
+ assert.strictEqual(off.start(), false);
+ assert.strictEqual(off.running, false);
+ await new Promise((r) => setTimeout(r, 80));
+ assert.strictEqual(Number(await db.getDb().value('SELECT COUNT(*) FROM billing_periods')), periods, 'nothing aggregates at interval 0');
+ assert.strictEqual(await count(), rows, 'nothing is queued at interval 0');
+ await off.stop();
+
+ await talking.stop();
+ await billing.stop();
+ console.log('✅ media billing: daily gb-month readings per project/subject/tier (a month sums to one GB-month), shared locations billed once to the oldest owner, newest closed day only, first-party and sandbox excluded, idempotent and immutable, refused-vs-pending relay, off at interval 0');
+ process.exit(0);
+})().catch((err) => { console.error(err); process.exit(1); });
@@ -9,17 +9,21 @@ import { fileURLToPath } from 'node:url';
@@ -9,17 +9,21 @@ import { fileURLToPath } from 'node:url';
const require = createRequire(import.meta.url);
const require = createRequire(import.meta.url);
const root = path.join(path.dirname(fileURLToPath(import.meta.url)), '..', '..');
const root = path.join(path.dirname(fileURLToPath(import.meta.url)), '..', '..');
const { createTestDb } = require('openvibe-sdk/testing');
const { createTestDb } = require('openvibe-sdk/testing');
-// openvibe-sdk >= 0.26 createTestDb holds a lease pg.Client (application_name *-test-lease) whose socket is not unref'd,
-// so a test file that ends without process.exit() never exits. Unref it here; the lease still holds while the process lives.
-// Drop this once Media pins openvibe-sdk >= 0.31.1, which unrefs the lease itself.
-const pg = require(require.resolve('pg', { paths: [require.resolve('openvibe-sdk/testing')] }));
-const connect = pg.Client.prototype.connect;
-pg.Client.prototype.connect = function (...args) {
- const r = connect.apply(this, args);
- if (String(this.connectionParameters && this.connectionParameters.application_name).endsWith('-test-lease')) this.connection.stream.unref();
- return r;
-};
+// openvibe-sdk >= 0.31.1 unrefs createTestDb's lease pg.Client itself (MEDIA_TEST_STORE=pg), so a test file that
+// ends without process.exit() exits. On PGlite, openvibe-sdk >= 0.29 reuses a migrated snapshot (OV_TEST_SNAPSHOT=0
+// opts out) and loadDataDir starts PGlite's Emscripten alarm — a ref'd setTimeout the wasm runtime keeps
+// rescheduling — which holds the event loop open the same way (a pure test, or a harness that exits only on
+// failure, would never exit and the per-file runner would time it out). This is Media-side and only for the timers
+// PGlite's own wasm schedules: drop once openvibe-sdk unrefs PGlite's alarm itself (the pin, v0.32.0, does not
+// yet), then every timer a test sets stays ref'd.
+const realSetTimeout = globalThis.setTimeout, realSetInterval = globalThis.setInterval;
+const fromPglite = () => { const s = new Error().stack; return !!s && s.includes('@electric-sql/pglite'); };
+globalThis.setTimeout = function (...a) { const h = realSetTimeout.apply(this, a); if (fromPglite()) { try { h.unref?.(); } catch { /* already gone */ } } return h; };
+globalThis.setInterval = function (...a) { const h = realSetInterval.apply(this, a); if (fromPglite()) { try { h.unref?.(); } catch { /* already gone */ } } return h; };
+// A ref'd keep-alive while the snapshot loads/builds: an unref'd setup timer must still fire.
+const keepAlive = realSetInterval(() => {}, 1000);
const t = await createTestDb({ migrations: path.join(root, 'migrations'), store: process.env.MEDIA_TEST_STORE || 'pglite', service: 'media', max: 4 });
const t = await createTestDb({ migrations: path.join(root, 'migrations'), store: process.env.MEDIA_TEST_STORE || 'pglite', service: 'media', max: 4 });
+clearInterval(keepAlive);
// Tests insert rows with small explicit ids. A PostgreSQL identity would continue after them, so generated ids
// Tests insert rows with small explicit ids. A PostgreSQL identity would continue after them, so generated ids
// are set to start at 100000 here, clear of every id a test names.
// are set to start at 100000 here, clear of every id a test names.
for (const r of await t.db.prepare(`SELECT pg_get_serial_sequence(quote_ident(table_name), column_name) AS seq FROM information_schema.columns
for (const r of await t.db.prepare(`SELECT pg_get_serial_sequence(quote_ident(table_name), column_name) AS seq FROM information_schema.columns