openvibers/openvibe.events Commits

Project usage rollups (roadmap WS-N task 4; Contracts v0.63.0 events.usage.recorded, common.usage-recorded@1): each developer project's use is counted per environment and UTC hour in the transaction that does the work (server/usage.js, table app_usage): events.app.publish in events (events stored in insertBatch; a repeated event_id is not counted), every refused publish request as an error with its problem code, status, trace id and the refused event id; events.app.subscribe in deliveries (webhook attempts to the project's subscriptions, recorded with the delivery row in recordAttempt), an attempt without a 2xx as events.delivery.http_<status>, .timeout or .failed with the event id and its trace. A minute after an hour closes, each rollup is stored once as an events.usage.recorded event (source events, subject the project, visibility internal, priority low) in the transaction that marks it sent, so Events is its own outbox; first-party subscribers (Network's project usage) get it like any event, apps never see it; a count landing in an hour already sent re-sends it as revision 2. EVENTS_USAGE=off counts nothing; EVENTS_USAGE_FLUSH_MS (300000). Pins: openvibe-contracts 0.61.0 -> 0.63.0. test/usage.test.js

OP OpenVibers committed 26 Sept 2026 · bb78877681 · 11 files, +389 −20





Changes
ddiff --git a/.env.example b/.env.example
iindex e5856b0..2c65ac2 100644
---- a/.env.example
++++ b/.env.example
@@@ -57,6 +57,12 @@ EVENTS_RECEIPT_RETENTION_DAYS=90
  # and its subdomains; IP literals only when listed exactly.
  #EVENTS_ENDPOINT_HOSTS=127.0.0.1,*.openvibe.*
  
++# Project usage (WS-N task 4): each closed hour of a developer project's publishing and webhook
++# deliveries becomes one events.usage.recorded event per capability and environment, delivered to
++# Network (the project dashboard on openvibe.codes). off = nothing counted or sent.
++#EVENTS_USAGE=on
++#EVENTS_USAGE_FLUSH_MS=300000
++
  # Developer apps (events.app.publish|read|subscribe, ADR-014). App subscriptions must use public
  # https endpoints (no allow-list; checked at subscribe and delivery time). Quotas are per project
  # and environment; 0 turns a limit off. EVENTS_APPS=off disables the app paths.
ddiff --git a/README.md b/README.md
iindex f535d04..4567c97 100644
---- a/README.md
++++ b/README.md
@@@ -133,6 +133,8 @@ Quotas recorded in Network (`dev_quotas`) are not read yet: that needs `network.
  
  `EVENTS_APPS=off` turns the developer-app paths off (app tokens are then judged like any other token and refused).
  
++**Usage** ([server/usage.js](server/usage.js), roadmap WS-N task 4). Each project's use is counted per environment and UTC hour in the transaction that does the work: `events.app.publish` in `events` (events stored; a repeated `event_id` is not counted) with every refused publish request as an error (its problem code, status, trace id and, when one event was refused, its `event_id`), and `events.app.subscribe` in `deliveries` (webhook attempts to the project's subscriptions; one without a 2xx is an error, `events.delivery.http_<status>`, `events.delivery.timeout` or `events.delivery.failed`, with the event id and its trace id). A minute after an hour closes, each rollup is stored once as an `events.usage.recorded` event (source `events`, subject the project, visibility `internal`, payload `common.usage-recorded@1` from openvibe-contracts 0.63.0) in the transaction that marks it sent, so Events is its own outbox; first-party subscribers get it like any event, apps never see it. OpenVibe.Network subscribes to it for the project dashboard on openvibe.codes. `EVENTS_USAGE=off` counts nothing; `EVENTS_USAGE_FLUSH_MS` (300000) is how often closed hours are looked for. Sent rollups are kept 7 days in `app_usage`.
++
  ## Client library
  
  `require('openvibe-events')` (package `main` is [lib/client.js](lib/client.js); services can depend on this repo by tarball):
@@@ -187,7 +189,7 @@ es.addEventListener('gap', (m) => { /* events were missed: refetch state */ });
  
  ## Owns
  
--- `events`, `subscriptions`, `deliveries`, `consumer_checkpoints`, `idempotency_receipts`, `app_revocations` (SQLite today; the plan's PostgreSQL + Redis fanout is a later step)
++- `events`, `subscriptions`, `deliveries`, `consumer_checkpoints`, `idempotency_receipts`, `app_revocations`, `app_usage` (SQLite today; the plan's PostgreSQL + Redis fanout is a later step)
  - developer-app event scope, sandbox separation and per-project Events quotas (ADR-014)
  - canonical event envelope (event_id, trace_id, type, version, source, actor, subject + revision, payload)
  - priority classes `critical|important|low`, loop guards, backpressure, DLQ and replay
@@@ -210,6 +212,7 @@ es.addEventListener('gap', (m) => { /* events were missed: refetch state */ });
  - browser reconnect resumes from a cursor or reports a gap (`test/realtime.test.js`)
  - a guessed private topic yields no data (`test/realtime.test.js`)
  - a developer app cannot publish, read or subscribe outside its project, sandbox never meets production, app webhooks reach public addresses only, quotas hold (`test/apps.test.js`)
++- a project's publishing and deliveries are counted per hour, refusals and failed attempts as errors, and each closed hour is sent once as `events.usage.recorded`, never visible to apps (`test/usage.test.js`)
  
  ## Bootstrap / extraction source
  
ddiff --git a/package-lock.json b/package-lock.json
iindex 2ec4301..1ddd563 100644
---- a/package-lock.json
++++ b/package-lock.json
@@@ -12,7 +12,7 @@
          "better-sqlite3": "^11.10.0",
          "dotenv": "^16.4.7",
          "express": "^4.21.2",
--        "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.61.0",
++        "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.63.0",
          "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.22.0"
        },
        "engines": {
@@@ -967,9 +967,9 @@
        }
      },
      "node_modules/openvibe-contracts": {
--      "version": "0.61.0",
--      "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.61.0",
--      "integrity": "sha512-8hpxyckfVwShA7Tku7mb7zHJBJsj9p3mXL7vM1QHmVnzKfuvUoX+I9JfX3uWPbDVAU/TcbPv53PKyCEUUWUK7g==",
++      "version": "0.63.0",
++      "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.63.0",
++      "integrity": "sha512-vmc1b1vWg9M5fm57X50MlZEZHkP2IshHwBXt68+kHDV85nyg6FxaYyUWa4ibi1eeFqzMcECPlILSRTOgg2Xbgw==",
        "license": "AGPL-3.0-only",
        "dependencies": {
          "ajv": "^8.17.1",
ddiff --git a/package.json b/package.json
iindex f185da4..e6f9491 100644
---- a/package.json
++++ b/package.json
@@@ -17,7 +17,7 @@
      "better-sqlite3": "^11.10.0",
      "dotenv": "^16.4.7",
      "express": "^4.21.2",
--    "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.61.0",
++    "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.63.0",
      "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.22.0"
    },
    "engines": {
ddiff --git a/server/api/publish.js b/server/api/publish.js
iindex da9d109..bac2e2c 100644
---- a/server/api/publish.js
++++ b/server/api/publish.js
@@@ -18,6 +18,10 @@
   * Redaction (server/redaction.js): an event whose payload carries `redacts` turns the named earlier
   * events of the same source into tombstones when it is stored. A malformed directive is 422
   * events.invalid_redaction; naming another source's event is 403 events.redaction_not_allowed.
++ *
++ * Usage (server/usage.js): an app's stored events count toward its project's events.app.publish in
++ * insertBatch's transaction; a request refused with a problem counts as one error, sampled with its
++ * code, status and trace id (and the event id, when one event was refused).
   */
  const express = require('express');
  const { validate, http } = require('openvibe-contracts');
@@@ -80,24 +84,35 @@ function publishRouter({ config, store, auth, worker, realtime }) {
      router.post('/api/v1/events', auth.appOrService(CAPS.publish, CAPS.appPublish, { requireService: true }), (req, res) => {
          const ctx = req.ov;
          const body = req.body;
++        const p = req.principal;
++        // Every refusal of an app's publish is one error in its project's usage.
++        const refuse = (status, code, o = {}, ref = null) => {
++            if (p.kind === 'app') {
++                try {
++                    store.usage.record({ projectId: p.projectId, env: p.env, capability: 'events.app.publish', unit: 'events', error: { code, status, traceId: ctx.traceId, ref } });
++                } catch (err) { console.error(`[usage] refusal not counted: ${err.message}`); }
++            }
++            return http.sendProblem(res, status, code, { ...o, ctx });
++        };
          const isBatch = body && typeof body === 'object' && !Array.isArray(body) && Array.isArray(body.events) && body.event_id === undefined;
          const items = isBatch ? body.events : [body];
--        if (isBatch && !items.length) return http.sendProblem(res, 400, 'events.bad_request', { detail: 'events must not be empty', ctx });
++        if (isBatch && !items.length) return refuse(400, 'events.bad_request', { detail: 'events must not be empty' });
          if (items.length > config.maxBatch) {
--            return http.sendProblem(res, 413, 'events.batch_too_large', { detail: `at most ${config.maxBatch} events per request`, ctx });
++            return refuse(413, 'events.batch_too_large', { detail: `at most ${config.maxBatch} events per request` });
          }
  
          const envelopes = [];
          const seen = new Set();
          for (let i = 0; i < items.length; i++) {
              const r = checkOne(items[i], req.principal, ctx);
++            const ref = items[i] && typeof items[i] === 'object' ? items[i].event_id : null;
              if (r.env && seen.has(r.env.event_id)) {
--                return http.sendProblem(res, 422, 'events.invalid_envelope', { detail: `events[${i}]: event_id repeated within the batch`, ctx, extra: { index: i } });
++                return refuse(422, 'events.invalid_envelope', { detail: `events[${i}]: event_id repeated within the batch`, extra: { index: i } }, ref);
              }
              if (!r.env) {
--                return http.sendProblem(res, r.status, r.code, {
--                    detail: isBatch ? `events[${i}]: ${r.detail}` : r.detail, errors: r.errors, ctx, extra: isBatch ? { index: i } : undefined,
--                });
++                return refuse(r.status, r.code, {
++                    detail: isBatch ? `events[${i}]: ${r.detail}` : r.detail, errors: r.errors, extra: isBatch ? { index: i } : undefined,
++                }, ref);
              }
              seen.add(r.env.event_id);
              envelopes.push(r.env);
@@@ -105,13 +120,12 @@ function publishRouter({ config, store, auth, worker, realtime }) {
  
          let out;
          try {
--            const p = req.principal;
              const project = p.kind === 'app' ? { projectId: p.projectId, env: p.env, ...config.apps.quotas[p.env] } : null;
              out = store.insertBatch(envelopes, { publisher: p.sub, requestId: ctx.requestId, project });
          } catch (err) {
              if (err instanceof StoreError) {
                  if (err.status === 429 && err.extra && err.extra.retry_after) res.setHeader('Retry-After', String(err.extra.retry_after));
--                return http.sendProblem(res, err.status, err.code, { detail: err.message, ctx, extra: err.extra });
++                return refuse(err.status, err.code, { detail: err.message, extra: err.extra }, err.extra && err.extra.event_id);
              }
              throw err;
          }
ddiff --git a/server/config.js b/server/config.js
iindex 9199860..80adf8c 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -113,6 +113,13 @@ function load(env = process.env) {
              sandboxRetentionDays: int(env.EVENTS_APP_SANDBOX_RETENTION_DAYS, 7),
          },
  
++        // Project usage rollups (server/usage.js, WS-N task 4): each closed hour of a project's
++        // publishing and webhook deliveries is stored as events.usage.recorded. off: nothing counted.
++        usage: {
++            enabled: env.EVENTS_USAGE !== 'off',
++            flushIntervalMs: Math.max(1000, int(env.EVENTS_USAGE_FLUSH_MS, 5 * 60 * 1000)),
++        },
++
          // Realtime (SSE) gateway
          realtime: {
              maxConnections: int(env.REALTIME_MAX_CONNECTIONS, 2000),
ddiff --git a/server/index.js b/server/index.js
iindex a63fd89..2effa0b 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -22,7 +22,7 @@ async function start({
  } = {}) {
      config = config || load();
      const db = openDb(config.dbPath);
--    const store = createStore(db, { clock, maxHops: config.maxHops });
++    const store = createStore(db, { clock, maxHops: config.maxHops, usage: config.usage });
      const keys = createKeyStore({ urls: [config.networkInternalUrl, config.networkUrl], pem: config.networkPublicKey, fetchImpl, log });
      const auth = createAuth({ config, keys, store });
      const metrics = createMetrics({ store });
@@@ -50,6 +50,25 @@ async function start({
      pruneTimer.unref?.();
      if (config.worker.enabled) prune();
  
++    // Project usage rollups (server/usage.js): each closed hour is stored as events.usage.recorded and
++    // fanned out like a published event.
++    const flushUsage = (opts) => {
++        try {
++            const r = store.flushUsage(opts);
++            if (r.stored.length) {
++                realtime.publish(r.stored);
++                worker.kick();
++            }
++            if (r.invalid) log.error(`[usage] ${r.invalid} rollup(s) do not match events.usage.recorded@1 and stay unsent`);
++            return r;
++        } catch (err) {
++            log.error(`[usage] flush failed: ${err.message}`);
++            return { stored: [], invalid: 0 };
++        }
++    };
++    const usageTimer = config.usage && config.usage.enabled ? setInterval(flushUsage, config.usage.flushIntervalMs) : null;
++    usageTimer?.unref?.();
++
      let server = null;
      if (listen) {
          server = await new Promise((resolve, reject) => {
@@@ -64,6 +83,7 @@ async function start({
  
      async function close() {
          clearInterval(pruneTimer);
++        if (usageTimer) clearInterval(usageTimer);
          realtime.stop();
          keys.stop();
          await worker.stop();
@@@ -74,7 +94,7 @@ async function start({
          db.close();
      }
  
--    return { config, db, store, keys, keyLoaded, auth, worker, realtime, metrics, app, server, close, prune };
++    return { config, db, store, keys, keyLoaded, auth, worker, realtime, metrics, app, server, close, prune, flushUsage };
  }
  
  if (require.main === module) {
ddiff --git a/server/store.js b/server/store.js
iindex b2ed814..37ea403 100644
---- a/server/store.js
++++ b/server/store.js
@@@ -11,6 +11,7 @@ const Database = require('better-sqlite3');
  const topics = require('./topics');
  const apps = require('./apps');
  const redaction = require('./redaction');
++const { createUsage, deliveryCode } = require('./usage');
  
  const PRIORITY_RANK = { critical: 0, important: 1, low: 2 };
  const RANK_PRIORITY = ['critical', 'important', 'low'];
@@@ -193,7 +194,9 @@ function subscriptionView(row, { withSecret = false } = {}) {
      return out;
  }
  
--function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8 } = {}) {
++function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage: usageConfig = { enabled: true } } = {}) {
++    // Project usage rollups (./usage.js): counted in the transactions below, sent as events.usage.recorded.
++    const usage = createUsage(db, { clock, enabled: usageConfig.enabled !== false });
      const q = {
          getEvent: db.prepare('SELECT * FROM events WHERE id = ?'),
          getReceipt: db.prepare('SELECT processed_at FROM idempotency_receipts WHERE consumer = ? AND event_id = ?'),
@@@ -344,6 +347,10 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8 } = {}
                  }
              }
          }
++        // An app publish counts toward its project's usage in the same transaction as the rows.
++        if (project && inserted.length) {
++            usage.record({ projectId: project.projectId, env: project.env, capability: 'events.app.publish', unit: 'events', quantity: inserted.length, at: now });
++        }
          return { results, inserted };
      });
  
@@@ -520,8 +527,19 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8 } = {}
          return rows.filter(r => !skip.has(r.subscription_id)).slice(0, limit);
      }
  
--    function recordAttempt(eventId, subscriptionId, outcome) {
++    /**
++     * Record one delivery attempt. `app` ({ projectId, env, traceId }) marks a developer-app
++     * subscription: the attempt counts toward the project's events.app.subscribe usage in the same
++     * transaction.
++     */
++    const recordAttempt = db.transaction((eventId, subscriptionId, outcome, app = null) => {
          const now = clock.now();
++        if (app) {
++            usage.record({
++                projectId: app.projectId, env: app.env, capability: 'events.app.subscribe', unit: 'deliveries', quantity: 1, at: now,
++                error: outcome.ok ? null : { code: deliveryCode(outcome), status: outcome.status || undefined, traceId: app.traceId, ref: eventId },
++            });
++        }
          if (outcome.ok) {
              db.prepare(`UPDATE deliveries SET status = 'delivered', attempt = ?, delivered_at = ?, last_status = ?, last_error = NULL,
                  next_attempt_at = NULL, updated_at = ? WHERE event_id = ? AND subscription_id = ?`)
@@@ -532,6 +550,11 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8 } = {}
                  .run(outcome.dead ? 'dead' : 'failed', outcome.attempt, outcome.status ?? null, String(outcome.error || '').slice(0, 500),
                      outcome.dead ? null : outcome.nextAttemptAt, now, eventId, subscriptionId);
          }
++    });
++
++    /** Send the closed hours' usage rollups (./usage.js); returns the stored rows. */
++    function flushUsage(opts) {
++        return usage.flush(insertBatch, opts);
      }
  
      function getDelivery(eventId, subscriptionId) {
@@@ -615,7 +638,7 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8 } = {}
          db, insertBatch, redact, getEvent, revokedAt, lastSeq, oldestSeq, firstSeqSince, scan,
          createSubscription, getSubscription, listSubscriptions, countSubscriptions, countProjectSubscriptions, setSubscriptionEnabled, rotateSubscriptionSecret,
          dueDeliveries, recordAttempt, getDelivery, listDeliveries, requeue, deliveryCounts,
--        getCheckpoint, setCheckpoint, prune, ping,
++        getCheckpoint, setCheckpoint, prune, ping, usage, flushUsage,
      };
  }
  
ddiff --git a/server/usage.js b/server/usage.js
nnew file mode 100644
iindex 0000000..4a74faa
---- /dev/null
++++ b/server/usage.js
@@@ -0,0 +1,164 @@
++'use strict';
++/**
++ * Project usage rollups (roadmap WS-N task 4, ADR-014): what each developer project used here, per
++ * environment and UTC hour, reported as `events.usage.recorded` (openvibe-contracts
++ * common.usage-recorded@1) for the project's dashboard on OpenVibe.Codes (Network adds them up).
++ *
++ *   events.app.publish    unit events       quantity: app events stored (a repeated event_id is not counted)
++ *                                            errors: publish requests refused with a problem (429 quota,
++ *                                            403 type/source/actor, 413 size, 422 envelope)
++ *   events.app.subscribe  unit deliveries   quantity: webhook delivery attempts to the project's subscriptions
++ *                                            errors: attempts without a 2xx (events.delivery.http_<status>,
++ *                                            events.delivery.timeout, events.delivery.failed)
++ *
++ * Counted in the transaction that does the accounting: insertBatch (the stored rows and their quota
++ * check) and recordAttempt (the delivery row). A refused publish rolls its transaction back, so the
++ * refusal is counted on its own as it is answered.
++ *
++ * Once an hour has closed (plus GRACE_MS), flush() stores each of its rollups as an event of source
++ * `events` in this same database (Events is its own outbox) in the transaction that marks the row
++ * sent: a rollup is stored once, and delivered like any first-party event (Network's usage consumer).
++ * A count that lands in an hour already sent (a clock step back) reopens it with revision + 1, and the
++ * next flush sends the corrected totals. Samples keep the time, code, status, trace id and event id of
++ * the last ten failures; never a subject, an address or anything of a payload.
++ */
++const crypto = require('crypto');
++const { validate, ids } = require('openvibe-contracts');
++
++const HOUR_MS = 60 * 60 * 1000;
++const DAY_MS = 24 * HOUR_MS;
++const GRACE_MS = 60 * 1000;
++const KEEP_SENT_MS = 7 * DAY_MS;
++const MAX_CODES = 20;
++const MAX_SAMPLES = 10;
++const CODE_RE = /^[a-z][a-z0-9_]*(\.[a-z0-9_]+)+$/;
++const TRACE_RE = /^[0-9a-f]{32}$/;
++const REF_RE = /^evt_[0-9A-HJKMNP-TV-Z]{26}$/;
++const PROJECT_RE = /^prj_[0-9A-HJKMNP-TV-Z]{26}$/;
++
++const SCHEMA = `
++CREATE TABLE IF NOT EXISTS app_usage (
++    project_id   TEXT NOT NULL,
++    env          TEXT NOT NULL,
++    capability   TEXT NOT NULL,
++    unit         TEXT NOT NULL,
++    window_start INTEGER NOT NULL,
++    quantity     INTEGER NOT NULL DEFAULT 0,
++    errors       INTEGER NOT NULL DEFAULT 0,
++    error_codes  TEXT NOT NULL DEFAULT '{}',
++    samples      TEXT NOT NULL DEFAULT '[]',
++    revision     INTEGER NOT NULL DEFAULT 1,
++    event_id     TEXT,
++    emitted_at   INTEGER,
++    PRIMARY KEY (project_id, env, capability, unit, window_start)
++);
++CREATE INDEX IF NOT EXISTS idx_app_usage_open ON app_usage(emitted_at, window_start);
++`;
++
++const hourOf = (ms) => Math.floor(ms / HOUR_MS) * HOUR_MS;
++const parse = (s, d) => { try { return JSON.parse(s); } catch { return d; } };
++
++/** The problem code of a failed delivery attempt (worker.js outcome). */
++function deliveryCode(outcome) {
++    if (outcome.status) return `events.delivery.http_${outcome.status}`;
++    return outcome.error === 'timeout' ? 'events.delivery.timeout' : 'events.delivery.failed';
++}
++
++/**
++ * @param {import('better-sqlite3').Database} db  the Events database
++ * @param {object} [o]
++ * @param {{ now(): number }} [o.clock]
++ * @param {boolean} [o.enabled=true]  false (EVENTS_USAGE=off): nothing is counted
++ */
++function createUsage(db, { clock = { now: () => Date.now() }, enabled = true } = {}) {
++    db.exec(SCHEMA);
++    const q = {
++        add: db.prepare(`INSERT INTO app_usage (project_id, env, capability, unit, window_start, quantity, errors)
++            VALUES (@project_id, @env, @capability, @unit, @window_start, @quantity, @errors)
++            ON CONFLICT(project_id, env, capability, unit, window_start) DO UPDATE SET
++                quantity = quantity + excluded.quantity, errors = errors + excluded.errors,
++                revision = revision + (CASE WHEN emitted_at IS NULL THEN 0 ELSE 1 END), emitted_at = NULL`),
++        get: db.prepare('SELECT error_codes, samples FROM app_usage WHERE project_id = ? AND env = ? AND capability = ? AND unit = ? AND window_start = ?'),
++        setErrors: db.prepare(`UPDATE app_usage SET error_codes = ?, samples = ?
++            WHERE project_id = ? AND env = ? AND capability = ? AND unit = ? AND window_start = ?`),
++        due: db.prepare('SELECT * FROM app_usage WHERE emitted_at IS NULL AND window_start <= ? ORDER BY window_start LIMIT ?'),
++        sent: db.prepare('UPDATE app_usage SET emitted_at = ?, event_id = ? WHERE project_id = ? AND env = ? AND capability = ? AND unit = ? AND window_start = ? AND emitted_at IS NULL'),
++        prune: db.prepare('DELETE FROM app_usage WHERE emitted_at IS NOT NULL AND window_start < ?'),
++        pending: db.prepare('SELECT COUNT(*) AS n FROM app_usage WHERE emitted_at IS NULL'),
++    };
++
++    /**
++     * Count usage (and at most one failure) for a project, inside the caller's transaction when there
++     * is one. `error` = { code, status?, traceId?, ref? } counts one error with a sample.
++     */
++    const record = db.transaction(({ projectId, env, capability, unit, quantity = 0, error = null, at = clock.now() }) => {
++        if (!enabled || !PROJECT_RE.test(String(projectId || '')) || (env !== 'sandbox' && env !== 'production')) return;
++        const key = { project_id: projectId, env, capability, unit, window_start: hourOf(at) };
++        q.add.run({ ...key, quantity: Math.max(0, Math.trunc(Number(quantity) || 0)), errors: error ? 1 : 0 });
++        if (!error) return;
++        const code = CODE_RE.test(String(error.code || '')) ? String(error.code) : 'events.error';
++        const row = q.get.get(projectId, env, capability, unit, key.window_start);
++        const codes = parse(row.error_codes, {});
++        if (codes[code] || Object.keys(codes).length < MAX_CODES) codes[code] = (codes[code] || 0) + 1;
++        const sample = { at: new Date(at).toISOString(), code };
++        if (Number.isInteger(error.status) && error.status >= 100 && error.status <= 599) sample.status = error.status;
++        if (TRACE_RE.test(String(error.traceId || ''))) sample.trace_id = error.traceId;
++        if (REF_RE.test(String(error.ref || ''))) sample.ref = error.ref;
++        const samples = [sample, ...parse(row.samples, [])].slice(0, MAX_SAMPLES);
++        q.setErrors.run(JSON.stringify(codes), JSON.stringify(samples), projectId, env, capability, unit, key.window_start);
++    });
++
++    function payloadOf(row) {
++        const p = {
++            project_id: row.project_id, env: row.env, capability: row.capability, unit: row.unit, window: 'hour',
++            window_start: new Date(row.window_start).toISOString(), window_end: new Date(row.window_start + HOUR_MS).toISOString(),
++            quantity: row.quantity, errors: row.errors,
++        };
++        const codes = parse(row.error_codes, {});
++        if (Object.keys(codes).length) p.error_codes = codes;
++        const samples = parse(row.samples, []);
++        if (samples.length) p.samples = samples;
++        if (row.revision > 1) p.revision = row.revision;
++        return p;
++    }
++
++    /**
++     * Store every closed hour's rollups as events.usage.recorded (at most `limit`). `insertBatch` is
++     * the store's; the returned rows are what it stored, for the realtime fan-out and the worker kick.
++     */
++    function flush(insertBatch, { now = clock.now(), limit = 500 } = {}) {
++        const stored = [];
++        let invalid = 0;
++        const rows = q.due.all(hourOf(now - GRACE_MS) - HOUR_MS, limit);
++        for (const row of rows) {
++            const payload = payloadOf(row);
++            const envelope = {
++                event_id: ids.newId('event'), event_type: 'events.usage.recorded', version: 1, source: 'events',
++                actor: { type: 'service', id: 'events' }, timestamp: new Date(now).toISOString(),
++                subject: { type: 'project', id: row.project_id }, visibility: 'internal', priority: 'low',
++                trace_id: crypto.randomBytes(16).toString('hex'), payload,
++            };
++            const v = validate('events.event-envelope@1', envelope);
++            const p = v.valid ? validate('events.usage.recorded@1', payload) : v;
++            if (!p.valid) {
++                // Never stop the others; the row stays unsent and is reported by status().
++                invalid++;
++                continue;
++            }
++            db.transaction(() => {
++                const out = insertBatch([envelope], { publisher: 'svc:events', requestId: null });
++                q.sent.run(now, envelope.event_id, row.project_id, row.env, row.capability, row.unit, row.window_start);
++                stored.push(...out.inserted);
++            })();
++        }
++        q.prune.run(now - KEEP_SENT_MS);
++        return { stored, invalid };
++    }
++
++    return {
++        record, flush, payloadOf, deliveryCode,
++        pending: () => q.pending.get().n,
++    };
++}
++
++module.exports = { createUsage, deliveryCode, HOUR_MS, GRACE_MS };
ddiff --git a/server/worker.js b/server/worker.js
iindex 792a734..cf2f6ed 100644
---- a/server/worker.js
++++ b/server/worker.js
@@@ -108,7 +108,9 @@ function createWorker({ store, config, clock = { now: () => Date.now() }, fetchI
              if (!outcome.dead) outcome.nextAttemptAt = clock.now() + backoffMs[Math.min(attempt - 1, backoffMs.length - 1)];
              if (outcome.dead) log.warn(`[worker] ${row.id} -> ${sub.id} dead after ${attempt} attempts: ${outcome.error}`);
          }
--        store.recordAttempt(delivery.event_id, delivery.subscription_id, outcome);
++        // A developer app's subscription: the attempt counts toward its project's usage (./usage.js).
++        const app = sub.project_id ? { projectId: sub.project_id, env: sub.env || 'production', traceId: row.trace_id } : null;
++        store.recordAttempt(delivery.event_id, delivery.subscription_id, outcome, app);
          if (observe) {
              try {
                  observe.attempt(outcome.ok ? 'delivered' : outcome.dead ? 'dead' : 'retry');
ddiff --git a/test/usage.test.js b/test/usage.test.js
nnew file mode 100644
iindex 0000000..74f9654
---- /dev/null
++++ b/test/usage.test.js
@@@ -0,0 +1,130 @@
++'use strict';
++// Project usage rollups (WS-N task 4, server/usage.js): an app's stored events and its webhook
++// delivery attempts are counted per project, environment and hour in the transactions that do the
++// work, refusals as errors with their code and trace id; once the hour closes each rollup is stored
++// exactly once as events.usage.recorded (common.usage-recorded@1), delivered to first-party
++// subscribers and never shown to apps; a count in an hour already sent re-sends it as revision 2.
++const assert = require('assert');
++const { ids, validate } = require('openvibe-contracts');
++const apps = require('../server/apps');
++const { boot, request, appToken, subscriber, suite } = require('./helpers');
++
++const t = suite('usage');
++const HOUR = 60 * 60 * 1000;
++const prj = `prj_${ids.ulid()}`;
++const appId = ids.newId('app');
++const key = apps.projectKey(prj);
++const source = apps.appSource(`app:${appId}`);
++const token = appToken({ appId, projectId: prj, env: 'sandbox', cap: ['events.app.publish', 'events.app.read'] });
++const appEvent = (over = {}) => ({
++    event_id: ids.newId('event'), event_type: `app.${key}.order.created`, version: 1, source,
++    actor: { type: 'app', id: appId }, timestamp: new Date().toISOString(), subject: { type: 'order', id: 'o1' }, payload: { n: 1 }, ...over,
++});
++const TRACE = '4bf92f3577b34da6a3ce929d0e0e4736';
++
++let h, network, statuses = [];
++const rows = () => h.db.prepare('SELECT * FROM app_usage ORDER BY capability').all();
++
++t('boot: an app subscription and a first-party usage subscriber', async () => {
++    // The hour starts at a round time so the test controls when it closes.
++    const clockStart = Math.floor(Date.now() / HOUR) * HOUR + 5 * 60 * 1000;
++    h = await boot({
++        env: { EVENTS_APP_SANDBOX_PUBLISH_PER_MINUTE: '5' },
++        appPost: async () => ({ status: statuses.length ? statuses.shift() : 204 }),
++    });
++    h.clock.t = clockStart;
++    network = await subscriber();
++    h.store.createSubscription({ id: `sub_${ids.ulid()}`, consumer: source, topicPattern: `app.${key}.*`, endpoint: 'https://hooks.example.com/h', secret: 'a'.repeat(40), projectId: prj, env: 'sandbox' });
++    h.store.createSubscription({ id: `sub_${ids.ulid()}`, consumer: 'network', topicPattern: 'events.usage.recorded', endpoint: network.url, secret: 'b'.repeat(40) });
++});
++
++t('stored app events count; a repeated event_id does not', async () => {
++    const batch = [appEvent(), appEvent(), appEvent()];
++    let r = await request(h.base, 'POST', '/api/v1/events', { token, body: { events: batch } });
++    assert.strictEqual(r.status, 201);
++    r = await request(h.base, 'POST', '/api/v1/events', { token, body: batch[0] });
++    assert.strictEqual(r.status, 200, 'a duplicate');
++    const [pub] = rows();
++    assert.deepStrictEqual([pub.project_id, pub.env, pub.capability, pub.unit, pub.quantity, pub.errors], [prj, 'sandbox', 'events.app.publish', 'events', 3, 0]);
++    assert.strictEqual(pub.window_start % HOUR, 0, 'an hour window');
++});
++
++t('refusals count as errors with code, status, trace id and event id; nothing of them as quantity', async () => {
++    const wrong = appEvent({ event_type: 'live.stream.started' });
++    let r = await request(h.base, 'POST', '/api/v1/events', { token, body: wrong, headers: { traceparent: `00-${TRACE}-00f067aa0ba902b7-01` } });
++    assert.strictEqual(r.status, 403);
++    r = await request(h.base, 'POST', '/api/v1/events', { token, body: { events: [appEvent(), appEvent(), appEvent()] } });
++    assert.strictEqual(r.status, 429, 'past the per-minute quota: the whole batch is refused');
++    const [pub] = rows();
++    assert.strictEqual(pub.quantity, 3);
++    assert.strictEqual(pub.errors, 2);
++    assert.deepStrictEqual(JSON.parse(pub.error_codes), { 'events.type_not_allowed': 1, 'events.quota_exceeded': 1 });
++    const samples = JSON.parse(pub.samples);
++    assert.deepStrictEqual(samples.map(s => [s.code, s.status]), [['events.quota_exceeded', 429], ['events.type_not_allowed', 403]], 'newest first');
++    assert.strictEqual(samples[1].trace_id, TRACE, 'the refused request\'s trace id');
++    assert.strictEqual(samples[1].ref, wrong.event_id, 'the refused event');
++});
++
++t('webhook delivery attempts count, a non-2xx as an error with the event and its trace', async () => {
++    statuses = [500];
++    await h.worker.drain();
++    const sub = rows().find(r => r.capability === 'events.app.subscribe');
++    assert.deepStrictEqual([sub.unit, sub.quantity, sub.errors], ['deliveries', 3, 1]);
++    assert.deepStrictEqual(JSON.parse(sub.error_codes), { 'events.delivery.http_500': 1 });
++    const [s] = JSON.parse(sub.samples);
++    assert.ok(/^evt_/.test(s.ref) && /^[0-9a-f]{32}$/.test(s.trace_id) && s.status === 500);
++});
++
++t('nothing is sent before the hour closes', async () => {
++    const out = h.flushUsage();
++    assert.strictEqual(out.stored.length, 0);
++    assert.strictEqual(h.store.usage.pending(), 2);
++});
++
++t('after the hour: one events.usage.recorded per rollup, valid, internal, delivered to first-party subscribers', async () => {
++    h.clock.advance(HOUR);
++    const out = h.flushUsage();
++    assert.strictEqual(out.stored.length, 2);
++    for (const row of out.stored) {
++        assert.deepStrictEqual([row.event_type, row.source, row.visibility, row.priority, row.subject_type, row.subject_id, row.project_id], ['events.usage.recorded', 'events', 'internal', 'low', 'project', prj, null]);
++        const payload = JSON.parse(row.payload);
++        assert.ok(validate('events.usage.recorded@1', payload).valid, JSON.stringify(validate('events.usage.recorded@1', payload).errors));
++        const text = JSON.stringify(payload);
++        assert.ok(!text.includes(appId) && !text.includes(source) && !text.includes('order.created'), 'no app, source or event content');
++    }
++    assert.strictEqual(h.flushUsage().stored.length, 0, 'sent once');
++    await h.worker.drain();
++    const got = network.calls.map(c => c.body.event);
++    assert.strictEqual(got.length, 2);
++    const pub = got.find(e => e.payload.capability === 'events.app.publish').payload;
++    assert.deepStrictEqual([pub.quantity, pub.errors, pub.window, pub.env], [3, 2, 'hour', 'sandbox']);
++    assert.strictEqual(Date.parse(pub.window_end) - Date.parse(pub.window_start), HOUR);
++});
++
++t('apps never see the rollups', async () => {
++    const r = await request(h.base, 'GET', '/api/v1/events?topic=events.*&after_seq=0', { token });
++    assert.strictEqual(r.status, 200);
++    assert.strictEqual(r.body.events.length, 0);
++});
++
++t('a count in an hour already sent re-sends it as revision 2', async () => {
++    const sent = rows().find(r => r.capability === 'events.app.publish');
++    h.store.usage.record({ projectId: prj, env: 'sandbox', capability: 'events.app.publish', unit: 'events', quantity: 1, at: sent.window_start + 1000 });
++    const out = h.flushUsage();
++    assert.strictEqual(out.stored.length, 1);
++    const p = JSON.parse(out.stored[0].payload);
++    assert.deepStrictEqual([p.quantity, p.revision], [4, 2], 'the corrected totals');
++});
++
++t('EVENTS_USAGE=off counts nothing', async () => {
++    const h2 = await boot({ env: { EVENTS_USAGE: 'off' } });
++    try {
++        const r = await request(h2.base, 'POST', '/api/v1/events', { token, body: appEvent() });
++        assert.strictEqual(r.status, 201);
++        assert.strictEqual(h2.db.prepare('SELECT COUNT(*) AS n FROM app_usage').get().n, 0);
++    } finally { await h2.stop(); }
++});
++
++t('teardown', async () => { await network.close(); await h.stop(); });
++
++t.run();