Wiki: 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 934fc60..774fb08 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -23,7 +23,7 @@ const { createActorLimits } = require('./http/actor-limits');
  const PUBLIC_DIR = path.join(__dirname, '..', 'public');
  const VERSION = require('../package.json').version;
  
--function createApp({ config, svc, viewers: baseViewers, platform, keys, db, valkey = null, log = console, rateLimits = true, fetchImpl = globalThis.fetch, limitsNow = null, indexnow = null, accountData = null, accountSend = null }) {
++function createApp({ config, telemetry = null, svc, viewers: baseViewers, platform, keys, db, valkey = null, log = console, rateLimits = true, fetchImpl = globalThis.fetch, limitsNow = null, indexnow = null, accountData = null, accountSend = null }) {
      const app = express();
      // Every resolved actor carries its wiki roles (read once per request), so the synchronous access checks work.
      const viewers = { ...baseViewers, resolve: async (req, opts) => svc.loadRoles(await baseViewers.resolve(req, opts)) };
@@@ -35,6 +35,7 @@ function createApp({ config, svc, viewers: baseViewers, platform, keys, db, valk
      const release = require('openvibe-shared/release').createRelease({ service: 'wiki', root: path.join(__dirname, '..') });
      require('./render/layout').setRelease(release.release);
      const metrics = require('openvibe-shared/metrics').instrument(app, { service: 'wiki', release: release.release });
++    if (telemetry) app.use(telemetry.middleware());   // request telemetry (server/index.js)
      metrics.registry.gauge({
          name: 'wiki_event_outbox', help: 'Events in the outbox by state', labelNames: ['state'],
          collect: () => { const c = platform.outbox.counts(); return [{ labels: { state: 'pending' }, value: c.pending }, { labels: { state: 'rejected' }, value: c.rejected }]; },
ddiff --git a/server/index.js b/server/index.js
iindex 3b3949c..2ab4d8b 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -9,7 +9,8 @@
   * Events outbox relay, the Media attachment check) run only when asked (`workers: true`).
   */
  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');
  const { load } = require('./config');
  const { openDb, migrate, createStores } = require('./db');
  const { createPlatform } = require('./integrations/platform');
@@@ -27,7 +28,7 @@ const { createNetworkSender, startSubscriptions } = require('openvibe-sdk/accoun
   * test handed in 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, extra = [] } = {}) {
++function createLifecycle({ server, db, valkey = null, platform, timers = [], telemetry = null, retention = null, closeDb = true, exit, signals, log, extra = [] } = {}) {
      return gracefulStop({
          name: 'Wiki', server, log, drainMs: 8000, deadlineMs: 10000, deadlineExitCode: 0, exit, signals,
          stop: [
@@@ -36,13 +37,14 @@ function createLifecycle({ server, db, valkey = null, platform, timers = [], clo
              ...extra,
          ],
          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 = null, signals = false, exit = () => {}, accountSend: givenSend = null } = {}) {
++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 = null, signals = false, exit = () => {}, accountSend: givenSend = null } = {}) {
      config = config || load();
      // PostgreSQL (ADR-035): migrate on the owner's direct connection, then serve on the pooled runtime role.
      // Tests hand in a migrated database of their own (test/db-helper.js).
@@@ -79,10 +81,17 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
      const accountSend = givenSend || (config.oauth.clientSecret
          ? createNetworkSender({ networkInternalUrl: config.networkInternalUrl, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, fetch: fetchImpl })
          : null);
--    const app = createApp({ config, svc, viewers, platform, keys, db, valkey, log, rateLimits, fetchImpl, limitsNow, indexnow, accountData, accountSend });
++    // 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: 'wiki', sink: telemetrySink || createPgTelemetrySink({ db, log }), skipPrefixes: ['/shared'] });
++    let retention = null;
++    const app = createApp({ config, telemetry, svc, viewers, platform, keys, db, valkey, log, rateLimits, fetchImpl, limitsNow, indexnow, accountData, accountSend });
  
      const timers = [];
      if (workers) {
++        timers.push(setInterval(() => { telemetry.flush().catch(() => {}); }, 15_000));
++        retention = startTelemetryRetention({ db, log });
          let scheduling = false;
          timers.push(setInterval(async () => {
              if (scheduling) return;
@@@ -117,12 +126,12 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
          });
      }
  
--    const lifecycle = createLifecycle({ server, db, valkey, platform, timers, closeDb: !givenDb, signals, exit, log, extra: [() => { if (subscriptions) subscriptions.stop(); }] });
++    const lifecycle = createLifecycle({ server, db, telemetry, retention, valkey, platform, timers, closeDb: !givenDb, signals, exit, log, extra: [() => { if (subscriptions) subscriptions.stop(); }] });
  
      async function stop() {
          return lifecycle.stop('stop');
      }
--    return { app, db, svc, stores, platform, keys, viewers, server, config, indexnow, timers, lifecycle, stop };
++    return { app, db, telemetry, svc, stores, platform, keys, viewers, server, config, indexnow, timers, lifecycle, stop };
  }
  
  module.exports = { start, createLifecycle };
ddiff --git a/test/helpers.js b/test/helpers.js
iindex fbda586..6d2f64d 100644
---- a/test/helpers.js
++++ b/test/helpers.js
@@@ -40,7 +40,7 @@ const quiet = process.env.WIKI_TEST_LOG ? console : { log() {}, warn() {}, error
   * A running Wiki on a database of its own (PGlite by default; WIKI_TEST_STORE=pg: the PostgreSQL + PgBouncer
   * containers). opts.db: boot on an existing handle (a restart keeps the data); h.stop() closes what boot opened.
   */
--async function boot({ env = {}, fetch: fetchImpl, db: givenDb = null, now, workers = false, tokens, rateLimits = false, limitsNow = null, log = quiet, indexnow = undefined, accountSend = null } = {}) {
++async function boot({ env = {}, fetch: fetchImpl, db: givenDb = null, now, workers = false, tokens, rateLimits = false, limitsNow = null, log = quiet, indexnow = undefined, accountSend = null, telemetrySink = null } = {}) {
      const owned = givenDb ? null : await testDb();
      const db = givenDb || owned.db;
      const config = load({
@@@ -52,7 +52,7 @@ async function boot({ env = {}, fetch: fetchImpl, db: givenDb = null, now, worke
          config, db, publicKey, log, listen: true, workers, rateLimits, now, limitsNow,
          fetchImpl: fetchImpl || (async (url) => { throw new Error(`unexpected outbound fetch ${url}`); }),
          tokens: tokens || { getToken: async () => 'stub-token', authHeaders: async () => ({ Authorization: 'Bearer stub-token' }), invalidate() {} },
--        indexnow, accountSend,
++        indexnow, accountSend, telemetrySink,
      });
      const base = `http://127.0.0.1:${h.server.address().port}`;
      const stop = async ({ keepDb = false } = {}) => { await h.stop(); if (owned && !keepDb) await owned.close(); };
ddiff --git a/test/telemetry.test.js b/test/telemetry.test.js
nnew file mode 100644
iindex 0000000..7fa3cc4
---- /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 'wiki' 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, 'wiki');
++        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), 'wiki', '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); });