Media: request telemetry in its own table, 30-day retention (#129)

ddiff --git a/migrations/0012_telemetry.sql b/migrations/0012_telemetry.sql
nnew file mode 100644
iindex 0000000..14260fa
---- /dev/null
++++ b/migrations/0012_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/index.js b/server/index.js
iindex 1dfb5bf..868805c 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -27,7 +27,8 @@ if (drill.enabled) {
  
  const fs = require('fs');
  const express = require('express');
--const { gracefulStop } = require('openvibe-sdk/service');
++const { gracefulStop, telemetry } = require('openvibe-sdk/service');
++const { createPgTelemetrySink, startTelemetryRetention } = require('openvibe-sdk/telemetry');
  const config = require('./config');
  const db = require('./db/database');
  const auth = require('./auth');
@@@ -77,6 +78,7 @@ const ready = (async () => {
      const release = require('openvibe-shared/release').createRelease({ service: 'media', root: require('path').join(__dirname, '..') });
      const observability = require('./observability');
      const instrumented = observability.instrument(app, { release: release.release });
++    app.use(telemetry.middleware);   // request telemetry: a no-op until telemetry.init() below (never under a drill)
      // Per-actor limit refusals (server/actor-limits.js) are counted there: media_rate_limited_total.
      require('./actor-limits').bindMetrics(instrumented.registry);
      // A restore-drill instance answers reads only: 403 for every other method, on every path (before the
@@@ -382,6 +384,14 @@ const ready = (async () => {
      // (deploy/systemd/openvibe-media.service). Media exits 0 past the deadline, like the 5 s family. A step
      // that throws is logged and the stop goes on: the try/catch the hand-written shutdown carried lives in
      // the kit now.
++    // Request telemetry (openvibe-sdk/service, plan T1): one aggregated sample per route template, method and status
++    // class each minute, stored in Media's telemetry_samples table; rows older than 14 days go daily. A restore drill
++    // starts neither (server/drill.js: no timer, no write).
++    let telemetryRetention = null;
++    if (!drill.enabled) {
++        telemetry.init({ service: 'media', sink: createPgTelemetrySink({ db: db.getDb(), log: console }), skipPrefixes: ['/shared'], intervalMs: 60_000 });
++        telemetryRetention = startTelemetryRetention({ db: db.getDb(), log: console, olderThanDays: 14 });
++    }
      gracefulStop({
          name: 'Media',
          server,
@@@ -400,7 +410,7 @@ const ready = (async () => {
              () => { for (const t of timers) clearInterval(t); },
              () => require('./s3/usage').flush().catch((err) => console.warn('[S3] usage flush at stop:', err.message)),
          ],
--        close: [() => db.close()],
++        close: [async () => { if (telemetryRetention) await telemetryRetention.stop(); await telemetry.stop(); }, () => db.close()],
          deadlineMs: 70000,
          deadlineExitCode: 0,
      });
ddiff --git a/test/service-kit.test.js b/test/service-kit.test.js
iindex 57c4457..6c30ff5 100644
---- a/test/service-kit.test.js
++++ b/test/service-kit.test.js
@@@ -58,7 +58,7 @@ function suite(name) {
  const t = suite('service-kit');
  
  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, /function shutdown\(/, 'the hand-written shutdown() is gone');
@@@ -71,7 +71,7 @@ t("the one stop names Media, stops the recorder first, keeps every handle, and u
      assert.match(call, /drainMs: 60000/, "a 60 s drain (VOD_STOP_GRACE_MS) so in-flight uploads/large downloads are not cut at the kit's 4 s default");
      assert.match(call, /deadlineMs: 70000/, 'the 70 s stop the systemd unit leaves room for (TimeoutStopSec=80)');
      assert.match(call, /deadlineExitCode: 0/, 'Media exits 0 past the deadline, like the 5 s family');
--    assert.match(call, /close: \[\(\) => db\.close\(\)\]/);
++    assert.match(call, /close: \[async \(\) => \{ if \(telemetryRetention\) await telemetryRetention\.stop\(\); await telemetry\.stop\(\); \}, \(\) => db\.close\(\)\]/, 'request telemetry flushes, then the database closes');
      assert.doesNotMatch(call, /beforeDrain/, 'the recorder is a stop step now, not beforeDrain (the kit runs beforeDrain after stop)');
  
      // The recorder stops first, before every other handle — the order the hand-written shutdown ran.
ddiff --git a/test/telemetry.test.js b/test/telemetry.test.js
nnew file mode 100644
iindex 0000000..eaffeac
---- /dev/null
++++ b/test/telemetry.test.js
@@@ -0,0 +1,47 @@
++'use strict';
++// Request telemetry (openvibe-sdk/service, plan T1 step 10): server/index.js mounts the process-wide middleware right after
++// the metrics, calls telemetry.init() for service 'media' with a PostgreSQL sink and starts the 14-day retention, never
++// under a restore drill, and flushes before the database closes. Through express, five hits on one route flush as ONE
++// aggregated http.request sample; the migration's table takes the retention.
++const assert = require('assert');
++const fs = require('fs');
++const path = require('path');
++const express = require('express');
++const { telemetry } = require('openvibe-sdk/service');
++const { pruneTelemetry } = require('openvibe-sdk/telemetry');
++
++(async () => {
++    const src = fs.readFileSync(path.join(__dirname, '..', 'server', 'index.js'), 'utf8');
++    assert.ok(src.indexOf('app.use(telemetry.middleware)') > src.indexOf('observability.instrument(app'), 'the middleware follows the metrics');
++    assert.match(src, /if \(!drill\.enabled\) \{\n\s+telemetry\.init\(\{ service: 'media', sink: createPgTelemetrySink\(/, 'init, never under a drill');
++    assert.match(src, /telemetryRetention = startTelemetryRetention\(\{ db: db\.getDb\(\), log: console, olderThanDays: 14 \}\)/);
++    assert.match(src, /close: \[async \(\) => \{ if \(telemetryRetention\) await telemetryRetention\.stop\(\); await telemetry\.stop\(\); \}, \(\) => db\.close\(\)\]/, 'the flush runs before the database closes');
++
++    const app = express();
++    app.use(telemetry.middleware);
++    app.get('/v/:id', (req, res) => res.send('watch'));
++    const server = await new Promise((r) => { const s = app.listen(0, '127.0.0.1', () => r(s)); });
++    const base = `http://127.0.0.1:${server.address().port}`;
++    const samples = [];
++    telemetry.init({ service: 'media', sink: async (batch) => { samples.push(...batch); }, intervalMs: 3600e3 });
++    try {
++        for (let i = 0; i < 5; i++) assert.strictEqual((await fetch(`${base}/v/${i}`)).status, 200);
++        await telemetry.flush();
++        const hit = samples.filter((s) => s.operation === 'http.request' && s.resource === '/v/:id' && s.status === '2xx');
++        assert.strictEqual(hit.length, 1, JSON.stringify(samples));
++        assert.strictEqual(hit[0].extra.count, 5, 'one sample per route template, never per URL');
++
++        const db = require('../server/db/database').getDb();
++        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), 'media', '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 new Promise((r) => server.close(r));
++    }
++    process.exit(0);
++})().catch((err) => { console.error(err); process.exit(1); });