OpenRestream: request telemetry in its own table, 30-day retention (#81)

ddiff --git a/migrations/0005_telemetry.sql b/migrations/0005_telemetry.sql
nnew file mode 100644
iindex 0000000..14260fa
---- /dev/null
++++ b/migrations/0005_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 908c003..c9113c2 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -22,6 +22,7 @@ function createApp({ rt, auth, keys, log = console, fetchImpl }) {
      // GET /metrics (loopback only, Track O): request rates and latencies, process metrics and live
      // sessions by state, from openvibe-shared/metrics (mounted before every route).
      const metrics = require('openvibe-shared/metrics').instrument(app, { service: 'openre' });
++    app.use(require('openvibe-sdk/service').telemetryMiddleware);   // request telemetry: a no-op until server/index.js calls telemetry.init()
      metrics.registry.gauge({ name: 'openre_sessions', help: 'Ingest sessions by state', labelNames: ['state'],
          collect: async () => (await db.prepare("SELECT state, count(*) AS n FROM ingest_sessions WHERE state IN ('starting','live') GROUP BY state").all()).map((r) => ({ labels: { state: r.state }, value: r.n })) });
      app.use(contracts.http.middleware());
ddiff --git a/server/index.js b/server/index.js
iindex e0bda0e..ed1075f 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -9,7 +9,8 @@
   * start() is what the tests use too: it takes a config plus injectable clock/fetch/log (and a database handle).
   */
  const path = require('path');
--const { gracefulStop } = require('openvibe-sdk/service');
++const { gracefulStop, telemetry } = require('openvibe-sdk/service');
++const { createPgTelemetrySink, startTelemetryRetention } = require('openvibe-sdk/telemetry');
  const { load } = require('./config');
  const { openRuntime } = require('./store');
  const { createNetworkKeys } = require('openvibe-sdk/auth');
@@@ -52,10 +53,14 @@ if (require.main === module) {
          // SIGTERM/SIGINT (openvibe-sdk/service, docs/service.md's OpenRestream entry): the keys poller stops, requests
          // in flight get 8 s, then Valkey and the database close in that order. A database close failure exits 1;
          // past the 10 s deadline the process also exits 1. Workers are separate processes and keep running.
++        // Request telemetry (openvibe-sdk/service, plan T1): one aggregated sample per route template, method and status
++        // class each minute, stored in this service's telemetry_samples table; rows older than 14 days go daily.
++        telemetry.init({ service: 'openre', sink: createPgTelemetrySink({ db: h.rt.db, log: console }), skipPrefixes: ['/shared'], intervalMs: 60_000 });
++        const telemetryRetention = startTelemetryRetention({ db: h.rt.db, log: console, olderThanDays: 14 });
          gracefulStop({
              name: 'openre-api',
              server: h.server,
--            stop: [() => h.keys.stop()],
++            stop: [() => h.keys.stop(), async () => { if (telemetryRetention) await telemetryRetention.stop(); await telemetry.stop(); }],
              close: [() => h.app.locals.valkey && h.app.locals.valkey.close()],
              handles: [() => h.rt.db.close()],
              drainMs: 8000,
ddiff --git a/test/service-kit.test.js b/test/service-kit.test.js
iindex 0cc56b8..871663d 100644
---- a/test/service-kit.test.js
++++ b/test/service-kit.test.js
@@@ -18,7 +18,7 @@ const quiet = { log() {}, warn() {}, error() {} };
  const source = fs.readFileSync(path.join(__dirname, '../server/index.js'), 'utf8');
  
  t('server/index.js imports gracefulStop from openvibe-sdk/service and owns no signal handler', () => {
--    assert.match(source, /const \{ gracefulStop \} = require\('openvibe-sdk\/service'\)/);
++    assert.match(source, /const \{ gracefulStop(, [A-Za-z]+)* \} = require\('openvibe-sdk\/service'\)/);
      assert.doesNotMatch(source, /process\.on\('SIGTERM'/);
      assert.doesNotMatch(source, /process\.on\('SIGINT'/);
      assert.doesNotMatch(source, /setTimeout\(\(\) => process\.exit\(1\), 10000\)/, 'the old 10 s exit-1 timer is gone');
@@@ -29,7 +29,7 @@ t('the one stop names openre-api, passes the server and the store steps, and use
      const call = source.slice(at, at + 500);
      assert.match(call, /name: 'openre-api'/);
      assert.match(call, /server: h\.server/);
--    assert.match(call, /stop: \[\(\) => h\.keys\.stop\(\)\]/, 'the keys poller stops first');
++    assert.match(call, /stop: \[\(\) => h\.keys\.stop\(\), async \(\) => \{ if \(telemetryRetention\) await telemetryRetention\.stop\(\); await telemetry\.stop\(\); \}\]/, 'the keys poller stops first, then request telemetry flushes while the database is open');
      assert.match(call, /close: \[\(\) => h\.app\.locals\.valkey && h\.app\.locals\.valkey\.close\(\)\]/);
      assert.match(call, /handles: \[\(\) => h\.rt\.db\.close\(\)\]/, 'the database closes after Valkey and its failure exits 1');
      assert.match(call, /drainMs: 8000/);
ddiff --git a/test/telemetry.test.js b/test/telemetry.test.js
nnew file mode 100644
iindex 0000000..72b1dc3
---- /dev/null
++++ b/test/telemetry.test.js
@@@ -0,0 +1,50 @@
++'use strict';
++/**
++ * Request telemetry (openvibe-sdk/service, plan T1 step 10): app.js mounts the process-wide telemetryMiddleware (a no-op
++ * until init), server/index.js calls telemetry.init() for service 'openre' with a PostgreSQL sink, starts the 14-day
++ * retention and flushes on stop. Five hits on one route flush as ONE aggregated http.request sample; /metrics and
++ * /shared make none; the retention deletes only rows older than 14 days.
++ */
++const assert = require('assert');
++const fs = require('fs');
++const path = require('path');
++const { telemetry } = require('openvibe-sdk/service');
++const { pruneTelemetry } = require('openvibe-sdk/telemetry');
++const { bootApi, request } = require('./helpers');
++(async () => {
++    const read = (p) => fs.readFileSync(path.join(__dirname, '..', p), 'utf8');
++    assert.match(read('server/index.js'), /telemetry\.init\(\{ service: 'openre', sink: createPgTelemetrySink\(/, 'index.js starts the collector');
++    assert.match(read('server/index.js'), /startTelemetryRetention\(/, 'index.js starts the retention');
++    assert.match(read('server/index.js'), /[Rr]etention\.stop\(\); await telemetry\.stop\(\);/, 'the stop flushes');
++
++    const t = await bootApi();
++    const samples = [];
++    telemetry.init({ service: 'openre', sink: async (batch) => { samples.push(...batch); }, intervalMs: 3600e3 });
++    const requests = () => samples.filter((s) => s.operation === 'http.request');
++    try {
++        for (let i = 0; i < 5; i++) assert.strictEqual((await request(t.base, 'GET', '/')).status, 200);
++        await telemetry.flush();
++        const hit = requests().filter((s) => s.resource === '/' && s.status === '2xx');
++        assert.strictEqual(hit.length, 1, JSON.stringify(requests()));
++        assert.strictEqual(hit[0].service, 'openre');
++        assert.strictEqual(hit[0].extra.count, 5);
++
++        samples.length = 0;
++        await await request(t.base, 'GET', '/metrics');
++        await await request(t.base, 'GET', '/shared/navbar.js');
++        await telemetry.flush();
++        assert.deepStrictEqual(requests(), [], '/metrics and /shared make no sample');
++
++        const db = t.rt.db;
++        await db.query('DELETE FROM telemetry_samples');
++        const insert = (days) => db.query("INSERT INTO telemetry_samples (at, service, operation) VALUES (now() - make_interval(days => $1), 'openre', 'http.request')", [days]);
++        await insert(20);
++        await insert(1);
++        assert.strictEqual(await pruneTelemetry({ db, olderThanDays: 14 }), 1);
++        assert.strictEqual((await db.query('SELECT count(*)::int AS n FROM telemetry_samples')).rows[0].n, 1);
++        console.log('telemetry ok');
++    } finally {
++        await telemetry.stop();
++        await t.close();
++    }
++})().catch((err) => { console.error(err); process.exit(1); });