Reviews: request telemetry in its own table, 30-day retention (#61)

ddiff --git a/migrations/0003_telemetry.sql b/migrations/0003_telemetry.sql
nnew file mode 100644
iindex 0000000..14260fa
---- /dev/null
++++ b/migrations/0003_telemetry.sql
@@@ -0,0 +1,28 @@
++-- phase: expand
++-- Request telemetry (openvibe-sdk/telemetry telemetrySchema()): one row per aggregated sample, pruned after 30 days (server/index.js).
++CREATE TABLE IF NOT EXISTS telemetry_samples (
++    id                    bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
++    at                    timestamptz NOT NULL,
++    service               text NOT NULL,
++    project               text,
++    subject               text,
++    resource              text,
++    provider              text,
++    node                  text,
++    cell                  text,
++    region                text,
++    operation             text NOT NULL,
++    latency_ms            double precision,
++    queue_delay_ms        double precision,
++    ttfb_ms               double precision,
++    throughput_per_second double precision,
++    bytes                 bigint,
++    status                text,
++    cache_status          text,
++    cost_estimate         double precision,
++    route_epoch           bigint,
++    trace_id              text,
++    extra                 jsonb NOT NULL DEFAULT '{}'::jsonb
++);
++CREATE INDEX IF NOT EXISTS telemetry_samples_at_idx ON telemetry_samples (at);
++CREATE INDEX IF NOT EXISTS telemetry_samples_service_idx ON telemetry_samples (service, operation, at);
ddiff --git a/server/app.js b/server/app.js
iindex 9c3c4be..21a7262 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -42,7 +42,7 @@ function lagged(read, initial) {
      };
  }
  
--function createApp({ config, svc, viewers, platform, sources, sync, keys, db, log = console, rateLimits = true, fetchImpl = globalThis.fetch, limitsNow = null, indexnow = null }) {
++function createApp({ config, telemetry = null, svc, viewers, platform, sources, sync, keys, db, log = console, rateLimits = true, fetchImpl = globalThis.fetch, limitsNow = null, indexnow = null }) {
      const app = express();
      app.disable('x-powered-by');
      app.set('trust proxy', config.trustProxy);
@@@ -54,6 +54,7 @@ function createApp({ config, svc, viewers, platform, sources, sync, keys, db, lo
      // Valkey (ADR-035): shared, never-authoritative state (per-actor limit counters). Optional.
      const valkey = config.valkey.url ? require('openvibe-sdk/valkey').createValkey({ url: config.valkey.url, prefix: config.valkey.prefix, log }) : null;
      const metrics = require('openvibe-shared/metrics').instrument(app, { service: 'reviews', release: release.release });
++    if (telemetry) app.use(telemetry.middleware());   // request telemetry (server/index.js)
      metrics.registry.gauge({
          name: 'reviews_event_outbox', help: 'Events in the outbox by state', labelNames: ['state'],
          collect: lagged(async () => [{ labels: { state: 'pending' }, value: await platform.outbox.pending() }, { labels: { state: 'rejected' }, value: await platform.outbox.rejected() }], []),
ddiff --git a/server/index.js b/server/index.js
iindex 9678346..c53b5a8 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -18,7 +18,8 @@ const { createKeyStore } = require('./auth/keys');
  const { createViewerResolver } = require('./auth/viewer');
  const { createApp } = require('./app');
  const { createIndexNow } = require('openvibe-shared/indexnow');
--const { gracefulStop } = require('openvibe-sdk/service');
++const { gracefulStop, createHttpTelemetry } = require('openvibe-sdk/service');
++const { createPgTelemetrySink, startTelemetryRetention } = require('openvibe-sdk/telemetry');
  
  /**
   * The process stop (openvibe-sdk/service, plan T1): the job timers and the Events outbox relay stop taking new
@@@ -27,7 +28,7 @@ const { gracefulStop } = require('openvibe-sdk/service');
   * its own database (start's `givenDb`), which stays open. `exit` and `signals` are injectable so a test can watch
   * the stop and no test process installs signal handlers.
   */
--function createLifecycle({ server, db, valkey = null, platform, timers = [], closeDb = true, exit, signals, log } = {}) {
++function createLifecycle({ server, db, valkey = null, platform, timers = [], telemetry = null, retention = null, closeDb = true, exit, signals, log } = {}) {
      return gracefulStop({
          name: 'Reviews', server, log, drainMs: 8000, deadlineMs: 10000, deadlineExitCode: 0, exit, signals,
          stop: [
@@@ -35,13 +36,14 @@ function createLifecycle({ server, db, valkey = null, platform, timers = [], clo
              () => platform.outbox.stop(),
          ],
          close: [
++            async () => { if (retention) await retention.stop(); if (telemetry) await telemetry.stop(); },
              () => { if (closeDb) return db.close(); },
              () => { if (valkey) return valkey.close(); },
          ],
      });
  }
  
--async function start({ config, db: givenDb = null, now = () => Date.now(), fetchImpl = globalThis.fetch, tokens = null, publicKey = null, log = console, listen = true, workers = listen, rateLimits = true, limitsNow = null, indexnow: givenIndexnow = undefined, signals = false, exit = () => {} } = {}) {
++async function start({ config, db: givenDb = null, now = () => Date.now(), fetchImpl = globalThis.fetch, tokens = null, publicKey = null, log = console, telemetrySink = null, listen = true, workers = listen, rateLimits = true, limitsNow = null, indexnow: givenIndexnow = undefined, signals = false, exit = () => {} } = {}) {
      config = config || load();
      // PostgreSQL (ADR-035): opened and migrated here unless the caller (a test) hands in a migrated handle.
      const db = givenDb || await openDb(config, { log });
@@@ -58,13 +60,20 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
      const keys = createKeyStore({ config, fetchImpl, log, publicKey });
      keys.ensure().catch(() => {});
      const viewers = createViewerResolver({ keys, config });
--    const app = createApp({ config, svc, viewers, platform, sources, sync, keys, db, log, rateLimits, fetchImpl, limitsNow, indexnow });
++    // Request telemetry (openvibe-sdk/service, plan T1): one aggregated sample per route template, method and status
++    // class each flush, stored in this service's telemetry_samples table (app.js mounts the middleware). With workers
++    // on, a 15 s tick writes each interval (the collector has no timer of its own) and the table keeps 30 days.
++    const telemetry = createHttpTelemetry({ service: 'reviews', sink: telemetrySink || createPgTelemetrySink({ db, log }), skipPrefixes: ['/shared'] });
++    let retention = null;
++    const app = createApp({ config, telemetry, svc, viewers, platform, sources, sync, keys, db, log, rateLimits, fetchImpl, limitsNow, indexnow });
  
      // Search holds whatever the current rules say (idempotent: unchanged documents are not re-sent).
      try { await svc.reconcileIndex(); } catch (err) { log.warn(`[Reviews] index reconcile: ${err.message}`); }
  
      const timers = [];
      if (workers) {
++        timers.push(setInterval(() => { telemetry.flush().catch(() => {}); }, 15_000));
++        retention = startTelemetryRetention({ db, log });
          if (config.sync.enabled && sources.enabled) {
              const pull = () => sync.pull().then((r) => { if (!r.ok) log.warn(`[Reviews] Sources pull: ${r.reason}`); }).catch((err) => log.warn(`[Reviews] Sources pull: ${err.message}`));
              timers.push(setTimeout(pull, 5000));
@@@ -82,12 +91,12 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
          log.log(`[Reviews] ${config.nodeEnv} on http://${config.host}:${config.port} → ${config.baseUrl}`);
      }
  
--    const lifecycle = createLifecycle({ server, db, platform, timers, closeDb: !givenDb, signals, exit, log });
++    const lifecycle = createLifecycle({ server, db, telemetry, retention, platform, timers, closeDb: !givenDb, signals, exit, log });
  
      async function stop() {
          return lifecycle.stop('stop');
      }
--    return { app, db, svc, sync, stores, platform, keys, viewers, indexnow, server, config, timers, lifecycle, stop };
++    return { app, db, telemetry, svc, sync, stores, platform, keys, viewers, indexnow, server, config, timers, lifecycle, stop };
  }
  
  module.exports = { start, createLifecycle };
ddiff --git a/test/helpers.js b/test/helpers.js
iindex 618979d..cc2a882 100644
---- a/test/helpers.js
++++ b/test/helpers.js
@@@ -147,7 +147,7 @@ function fakeCommunity() {
   * (default false) turns on the per-address and per-actor limits; limitsNow is the per-actor limiter's
   * clock; log is Reviews' logger (default quiet).
   */
--async function boot({ env = {}, sources = fakeSources(), community = null, now, workers = false, rateLimits = false, limitsNow = null, log = quiet, indexnow } = {}) {
++async function boot({ env = {}, sources = fakeSources(), community = null, now, workers = false, rateLimits = false, limitsNow = null, log = quiet, indexnow, telemetrySink = null } = {}) {
      const config = load({
          NODE_ENV: 'test', PORT: '0', HOST: '127.0.0.1', BASE_URL: 'http://reviews.test',
          OV_NETWORK_URL: ISSUER, OV_NETWORK_INTERNAL_URL: '', REVIEWS_GATE_MIN_WORDS: '20',
@@@ -167,7 +167,7 @@ async function boot({ env = {}, sources = fakeSources(), community = null, now,
      // One database per boot (PGlite, or REVIEWS_TEST_STORE=pg: the containers), dropped when the boot stops.
      const testdb = await testDb();
      const h = await start({
--        config, db: testdb.db, publicKey, log, listen: true, workers, rateLimits, limitsNow, now, fetchImpl, indexnow,
++        config, db: testdb.db, publicKey, log, listen: true, workers, rateLimits, limitsNow, now, fetchImpl, indexnow, telemetrySink,
          tokens: { getToken: async () => 'stub-token', authHeaders: async () => ({ Authorization: 'Bearer stub-token' }), invalidate() {} },
      });
      const base = `http://127.0.0.1:${h.server.address().port}`;
ddiff --git a/test/telemetry.test.js b/test/telemetry.test.js
nnew file mode 100644
iindex 0000000..39f39cc
---- /dev/null
++++ b/test/telemetry.test.js
@@@ -0,0 +1,47 @@
++'use strict';
++/**
++ * Request telemetry (openvibe-sdk/service, plan T1 step 10): start() builds one collector for service 'reviews' and
++ * createApp mounts it; a page hit five times flushes as ONE aggregated http.request sample (route template, 2xx, count
++ * 5); /metrics and /shared files make none; the lifecycle flushes before the database closes; with workers on, a 15 s
++ * tick and the 30-day retention run; the retention deletes only rows older than 30 days.
++ */
++const assert = require('assert');
++const fs = require('fs');
++const path = require('path');
++const { pruneTelemetry } = require('openvibe-sdk/telemetry');
++const H = require('./helpers');
++
++(async () => {
++    const index = fs.readFileSync(path.join(__dirname, '..', 'server', 'index.js'), 'utf8');
++    assert.match(index, /timers\.push\(setInterval\(\(\) => \{ telemetry\.flush\(\)/, 'the workers tick flushes the collector');
++    assert.match(index, /retention = startTelemetryRetention\(\{ db, log \}\)/, 'the workers start the retention');
++    assert.match(index, /close: \[\n\s+async \(\) => \{ if \(retention\) await retention\.stop\(\); if \(telemetry\) await telemetry\.stop\(\); \},/, 'the flush runs before the database closes');
++
++    const samples = [];
++    const h = await H.boot({ telemetrySink: async (batch) => { samples.push(...batch); } });
++    const requests = () => samples.filter((s) => s.operation === 'http.request');
++    try {
++        for (let i = 0; i < 5; i++) assert.strictEqual((await H.req(h, 'GET', '/')).status, 200);
++        await h.telemetry.flush();
++        const home = requests().filter((s) => s.resource === '/' && s.status === '2xx');
++        assert.strictEqual(home.length, 1, JSON.stringify(requests()));
++        assert.strictEqual(home[0].service, 'reviews');
++        assert.strictEqual(home[0].extra.count, 5);
++
++        samples.length = 0;
++        await H.req(h, 'GET', '/metrics');
++        await H.req(h, 'GET', '/shared/navbar.js');
++        await h.telemetry.flush();
++        assert.deepStrictEqual(requests(), [], '/metrics and /shared make no sample');
++
++        await h.db.query('DELETE FROM telemetry_samples');
++        const insert = (days) => h.db.query("INSERT INTO telemetry_samples (at, service, operation) VALUES (now() - make_interval(days => $1), 'reviews', 'http.request')", [days]);
++        await insert(45);
++        await insert(1);
++        assert.strictEqual(await pruneTelemetry({ db: h.db }), 1);
++        assert.strictEqual((await h.db.query('SELECT count(*)::int AS n FROM telemetry_samples')).rows[0].n, 1);
++        console.log('telemetry ok');
++    } finally {
++        await h.stop();
++    }
++})().catch((err) => { console.error(err); process.exit(1); });