openvibers/openvibe.shared Commits

Shared 1.29.0: openvibe-shared/analytics/pg, request analytics on PostgreSQL (AnalyticsTrackerPg with the same tables and dashboard shapes, analyticsSchema() for migrations, batched 30-day retention). The request path never waits for the database; the day's salt is shared across processes so uniques stay right; no address reaches the database. npm test 42/42.

OP OpenVibers committed 28 Sept 2026 · 1e4bbc3d28 · 7 files, +504 −4


Changes

ddiff --git a/CHANGELOG.md b/CHANGELOG.md
iindex c8f40bd..8a2ab56 100644
---- a/CHANGELOG.md
++++ b/CHANGELOG.md
@@@ -4,6 +4,10 @@ All notable changes to `openvibe-shared`. Versions follow [semver](https://semve
breaking change to any exported module, browser global or served file name is a new major.
A release is the git tag `vX.Y.Z`; consumers pin the tag's tarball (see README).
++## 1.29.0 — 2026-09-28
++
++**`openvibe-shared/analytics/pg`: request analytics on PostgreSQL** (ADR-035; ADR-021's privacy bounds unchanged). `AnalyticsTrackerPg` is the tracker on an openvibe-sdk/db handle, with the same tables and `getStats`, `getOverview` and `getBotAnalysis` shapes; `analyticsSchema()` is the DDL for the service's migrations; `pruneRawEventsPg` is the 30-day retention in batches. The request path never waits for the database: rows buffer in memory and `flush()` writes them in one transaction every 5 s (or at 100). Unique visitors stay right across processes: the day's salt is shared through `analytics_day_salts`, so the visitor hash is computed at flush, while (address, user agent) wait in memory until then, as the rate counters always did; no address reaches the database. Session ids stay per process, as they were. Every SUM is cast to bigint (numeric comes back as text). `test/analytics-pg.test.js` on PGlite: two trackers count one visitor once, DNT records nothing, rollups, dashboards and retention. Needed by Network's and Live's PostgreSQL passes.
++
## 1.28.1 — 2026-09-28
Fixes the owner reported from phones.
ddiff --git a/STATUS.json b/STATUS.json
iindex 9f95d4e..f51fef4 100644
---- a/STATUS.json
++++ b/STATUS.json
@@@ -4,7 +4,7 @@
"code": true,
"wave": "W2",
"package": "openvibe-shared",
-- "version": "1.28.1",
++ "version": "1.29.0",
"plan": "OpenVibe End-to-End Realignment & Implementation Plan, revision 3 (20 Sep 2026)",
"deployed": false,
"deployment": "none of its own: a package released as git tags; OpenVibe.Network serves /shared/* from its pinned tag, and a site can serve its own pinned copy (openvibe-shared/serve)",
ddiff --git a/analytics/pg.js b/analytics/pg.js
nnew file mode 100644
iindex 0000000..9ddc046
---- /dev/null
++++ b/analytics/pg.js
@@@ -0,0 +1,437 @@
++'use strict';
++/**
++ * Request analytics on PostgreSQL (ADR-035; ADR-021's privacy bounds unchanged). The same tracker, tables and
++ * getStats/getOverview/getBotAnalysis shapes as ./tracker.js, on an openvibe-sdk/db handle:
++ *
++ * const { AnalyticsTrackerPg, analyticsSchema } = require('openvibe-shared/analytics/pg');
++ * // migrations/NNNN_analytics.sql holds analyticsSchema()
++ * const tracker = new AnalyticsTrackerPg(db, 'network', { paramPrefixes: [...] });
++ * app.use(tracker.middleware());
++ * await tracker.flush(); await tracker.aggregate(); await tracker.getStats({ days: 7 });
++ *
++ * What differs from the SQLite tracker:
++ * - the request path never waits for the database: record() buffers the row in memory, and flush() (every
++ * 5 s, and at 100 rows) writes the buffer in one transaction;
++ * - uniques across processes: the visitor hash needs the day's salt that every process of the service shares
++ * (analytics_day_salts), which is async now. So record() keeps (ip, user agent) for the visitor only until the
++ * next flush, in memory, like the rate counters always did; flush() reads or creates the day's salt, hashes,
++ * and writes the hash. The IP never reaches the database;
++ * - session ids are per process (as they always were: the session map was in memory), from a process-local
++ * daily key, so record() stays synchronous;
++ * - reads and rollups are async; SUM() over bigint comes back as numeric (exact), so every sum is cast.
++ */
++const crypto = require('crypto');
++const { AnalyticsTracker, sqlTime, dayOf, hourOf } = require('./tracker');
++const privacy = require('./privacy');
++
++const MAX_BUFFER = 5000;
++const DAY = 86400000;
++
++/** The DDL for a service's migrations (the runtime role creates no tables). */
++function analyticsSchema() {
++ return `
++CREATE TABLE IF NOT EXISTS analytics_events (
++ id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
++ service text NOT NULL,
++ event_type text NOT NULL DEFAULT 'pageview',
++ path text,
++ method text DEFAULT 'GET',
++ status_code integer,
++ response_time_ms integer,
++ user_id bigint,
++ session_id text,
++ ip text,
++ country text,
++ city text,
++ user_agent text,
++ referer text,
++ is_bot integer DEFAULT 0,
++ bot_type text,
++ device_type text,
++ browser text,
++ os text,
++ authenticated integer NOT NULL DEFAULT 0,
++ created_at text NOT NULL
++);
++CREATE TABLE IF NOT EXISTS analytics_hourly (
++ id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
++ service text NOT NULL, hour text NOT NULL,
++ pageviews bigint DEFAULT 0, api_calls bigint DEFAULT 0, unique_visitors bigint DEFAULT 0, unique_users bigint DEFAULT 0,
++ bot_hits bigint DEFAULT 0, avg_response_ms integer DEFAULT 0, error_count bigint DEFAULT 0, bandwidth_bytes bigint DEFAULT 0,
++ top_paths text, top_referers text,
++ UNIQUE (service, hour)
++);
++CREATE TABLE IF NOT EXISTS analytics_daily (
++ id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
++ service text NOT NULL, date text NOT NULL,
++ pageviews bigint DEFAULT 0, api_calls bigint DEFAULT 0, unique_visitors bigint DEFAULT 0, unique_users bigint DEFAULT 0,
++ new_users bigint DEFAULT 0, bot_hits bigint DEFAULT 0, avg_response_ms integer DEFAULT 0, error_count bigint DEFAULT 0,
++ top_paths text, top_referers text, top_countries text, device_breakdown text, browser_breakdown text,
++ UNIQUE (service, date)
++);
++CREATE TABLE IF NOT EXISTS analytics_visitor_days (
++ service text NOT NULL, day text NOT NULL, hour text NOT NULL, vhash text NOT NULL,
++ authenticated integer NOT NULL DEFAULT 0,
++ PRIMARY KEY (service, hour, vhash)
++);
++CREATE TABLE IF NOT EXISTS analytics_day_salts (
++ service text NOT NULL, day text NOT NULL, salt text NOT NULL,
++ PRIMARY KEY (service, day)
++);
++CREATE INDEX IF NOT EXISTS idx_analytics_events_created ON analytics_events (created_at);
++CREATE INDEX IF NOT EXISTS idx_analytics_events_service ON analytics_events (service, created_at);
++CREATE INDEX IF NOT EXISTS idx_analytics_events_path ON analytics_events (service, path, created_at);
++CREATE INDEX IF NOT EXISTS idx_analytics_events_bot ON analytics_events (is_bot, created_at);
++CREATE INDEX IF NOT EXISTS idx_analytics_visitor_days_day ON analytics_visitor_days (service, day);
++`.trim();
++}
++
++class AnalyticsTrackerPg extends AnalyticsTracker {
++ /**
++ * @param {object} db openvibe-sdk/db handle (the service's own database, analyticsSchema() migrated)
++ * @param {string} service
++ * @param {object} [opts] as AnalyticsTracker: paramPrefixes, pathRules, retention ({ days = 30, intervalMs,
++ * initialDelayMs } or false), timers, now
++ */
++ constructor(db, service, opts = {}) {
++ super(db, service, { ...opts, timers: false, __pg: true });
++ this._pending = []; // visitors waiting for the shared salt: { day, hour, ip, ua, authenticated }
++ this._localKeys = new Map(); // day → process-local session key
++ this._salts = new Map(); // day → shared salt (cached after the first read)
++ this._flushing = null;
++ if (opts.timers !== false) {
++ const every = (fn, ms) => { const t = setInterval(fn, ms); if (t.unref) t.unref(); this._timers.push(t); };
++ const once = (fn, ms) => { const t = setTimeout(fn, ms); if (t.unref) t.unref(); this._timers.push(t); };
++ every(() => { this.flush().catch(() => {}); this._sweep(); }, 5000);
++ every(() => { this.aggregate().catch(() => {}); }, 3600e3);
++ once(() => { this.aggregate().catch(() => {}); }, 10_000);
++ if (opts.retention !== false) {
++ const r = opts.retention && typeof opts.retention === 'object' ? opts.retention : {};
++ const days = r.days || 30;
++ const run = () => { pruneRawEventsPg(db, { days, now: this._now }).catch((err) => console.error(`[Analytics:${service}] prune error:`, err.message)); };
++ once(run, r.initialDelayMs == null ? 5 * 60e3 : r.initialDelayMs);
++ every(run, r.intervalMs || DAY);
++ }
++ }
++ }
++
++ /** Process-local session key per day: session ids stay per process, as the in-memory session map always was. */
++ _visitorHash(ip, ua, day) {
++ let k = this._localKeys.get(day);
++ if (!k) { this._localKeys.clear(); k = crypto.randomBytes(32); this._localKeys.set(day, k); this._sessions.clear(); }
++ return crypto.createHmac('sha256', k).update(`${ip}\n${ua}`).digest('hex').slice(0, 16);
++ }
++
++ record(req, res, entryPath, responseTimeMs) {
++ const recorded = super.record(req, res, entryPath, responseTimeMs);
++ if (!recorded) return false;
++ // super._push queued a visitor keyed by the local hash; replace it with what the shared hash needs.
++ const headers = req.headers || {};
++ const nowMs = this._now();
++ this._pending.push({
++ day: dayOf(nowMs), hour: hourOf(nowMs),
++ ip: String(req.ip || (req.socket && req.socket.remoteAddress) || ''), ua: String(headers['user-agent'] || ''),
++ authenticated: !!(req.user && (req.user.id != null || req.user.sub != null)),
++ });
++ if (this._pending.length > MAX_BUFFER) this._pending.splice(0, this._pending.length - MAX_BUFFER);
++ return true;
++ }
++
++ _push(event) {
++ if (this._buffer.length >= MAX_BUFFER) this._buffer.shift();
++ this._buffer.push(event);
++ if (this._buffer.length >= 100) this.flush().catch(() => {});
++ }
++
++ /** The day's salt every process of the service shares: read, or created by the first process to need it. */
++ async _sharedSalt(day) {
++ if (this._salts.has(day)) return this._salts.get(day);
++ await this.db.query('INSERT INTO analytics_day_salts (service, day, salt) VALUES ($1, $2, $3) ON CONFLICT DO NOTHING',
++ [this.service, day, crypto.randomBytes(32).toString('hex')]);
++ const salt = await this.db.value('SELECT salt FROM analytics_day_salts WHERE service = $1 AND day = $2', [this.service, day]);
++ this._salts.clear();
++ this._salts.set(day, salt);
++ return salt;
++ }
++
++ /** Write buffered rows and visitor hashes in one transaction; on failure they wait for the next flush. */
++ flush() {
++ if (this._flushing) return this._flushing;
++ this._flushing = this._flushNow().finally(() => { this._flushing = null; });
++ return this._flushing;
++ }
++
++ async _flushNow() {
++ if (!this._buffer.length && !this._pending.length) return;
++ const events = this._buffer.splice(0);
++ const pending = this._pending.splice(0);
++ try {
++ const visitors = new Map();
++ for (const p of pending) {
++ const vhash = crypto.createHmac('sha256', await this._sharedSalt(p.day)).update(`${p.ip}\n${p.ua}`).digest('hex').slice(0, 16);
++ const key = `${p.hour}|${vhash}`;
++ const prev = visitors.get(key);
++ if (!prev) visitors.set(key, { day: p.day, hour: p.hour, vhash, authenticated: p.authenticated });
++ else if (p.authenticated) prev.authenticated = true;
++ }
++ await this.db.tx(async () => {
++ for (let i = 0; i < events.length; i += 200) {
++ const chunk = events.slice(i, i + 200);
++ const vals = []; const rows = [];
++ for (const e of chunk) {
++ const b = vals.length;
++ rows.push(`(${Array.from({ length: 17 }, (_, k) => `$${b + k + 1}`).join(', ')})`);
++ vals.push(e.service, e.event_type, e.path, e.method, e.status_code, e.response_time_ms, e.session_id, e.country, e.user_agent,
++ e.referer, e.is_bot ? 1 : 0, e.bot_type, e.device_type, e.browser, e.os, e.authenticated ? 1 : 0, e.created_at);
++ }
++ if (rows.length) {
++ await this.db.query(`INSERT INTO analytics_events (service, event_type, path, method, status_code, response_time_ms, session_id, country,
++ user_agent, referer, is_bot, bot_type, device_type, browser, os, authenticated, created_at) VALUES ${rows.join(', ')}`, vals);
++ }
++ }
++ for (const v of visitors.values()) {
++ await this.db.query(`INSERT INTO analytics_visitor_days (service, day, hour, vhash, authenticated) VALUES ($1, $2, $3, $4, $5)
++ ON CONFLICT (service, hour, vhash) DO UPDATE SET authenticated = GREATEST(analytics_visitor_days.authenticated, excluded.authenticated)`,
++ [this.service, v.day, v.hour, v.vhash, v.authenticated ? 1 : 0]);
++ }
++ });
++ } catch (err) {
++ console.error(`[Analytics:${this.service}] Flush error:`, err.message);
++ this._buffer = events.concat(this._buffer).slice(-MAX_BUFFER);
++ this._pending = pending.concat(this._pending).slice(-MAX_BUFFER);
++ }
++ }
++
++ // ── Rollups ──────────────────────────────────────────────
++
++ async aggregate() {
++ try {
++ await this.flush();
++ const nowMs = this._now();
++ const today = dayOf(nowMs);
++ const hours = new Set([hourOf(nowMs), hourOf(nowMs - 3600000)]);
++ const days = new Set([today, dayOf(nowMs - DAY)]);
++ for (const p of await this.db.many('SELECT DISTINCT day, hour FROM analytics_visitor_days WHERE service = $1 AND day < $2', [this.service, today])) {
++ hours.add(p.hour); days.add(p.day);
++ }
++ for (const h of hours) await this._aggregateHour(h);
++ for (const d of days) await this._aggregateDay(d);
++ await this.db.query('DELETE FROM analytics_visitor_days WHERE service = $1 AND day < $2', [this.service, today]);
++ await this.db.query('DELETE FROM analytics_day_salts WHERE service = $1 AND day < $2', [this.service, today]);
++ } catch (err) {
++ console.error(`[Analytics:${this.service}] Aggregation error:`, err.message);
++ }
++ }
++
++ async _aggregateHour(hourStart) {
++ const nextHour = sqlTime(Date.parse(hourStart.replace(' ', 'T') + 'Z') + 3600000);
++ const svc = this.service;
++ const range = [svc, hourStart, nextHour];
++ const stats = await this.db.maybe(`
++ SELECT COUNT(*) FILTER (WHERE event_type = 'pageview') AS pageviews, COUNT(*) FILTER (WHERE event_type = 'api_call') AS api_calls,
++ COUNT(*) FILTER (WHERE is_bot = 1) AS bot_hits, CAST(AVG(response_time_ms) AS integer) AS avg_response_ms,
++ COUNT(*) FILTER (WHERE status_code >= 400) AS error_count
++ FROM analytics_events WHERE service = $1 AND created_at >= $2 AND created_at < $3`, range);
++ if (!stats || (Number(stats.pageviews) === 0 && Number(stats.api_calls) === 0)) return;
++ const v = await this.db.maybe('SELECT COUNT(*) AS visitors, COALESCE(SUM(authenticated), 0)::bigint AS users FROM analytics_visitor_days WHERE service = $1 AND hour = $2', [svc, hourStart]);
++ const prev = (await this.db.maybe('SELECT unique_visitors, unique_users FROM analytics_hourly WHERE service = $1 AND hour = $2', [svc, hourStart])) || {};
++ const topPaths = await this.db.many(`SELECT path, COUNT(*) AS cnt FROM analytics_events WHERE service = $1 AND created_at >= $2 AND created_at < $3 AND is_bot = 0
++ GROUP BY path ORDER BY cnt DESC LIMIT 10`, range);
++ const topReferers = await this.db.many(`SELECT referer, COUNT(*) AS cnt FROM analytics_events WHERE service = $1 AND created_at >= $2 AND created_at < $3
++ AND referer != '' AND referer IS NOT NULL AND is_bot = 0 GROUP BY referer ORDER BY cnt DESC LIMIT 10`, range);
++ await this.db.query(`
++ INSERT INTO analytics_hourly (service, hour, pageviews, api_calls, unique_visitors, unique_users, bot_hits, avg_response_ms, error_count, top_paths, top_referers)
++ VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
++ ON CONFLICT (service, hour) DO UPDATE SET pageviews = excluded.pageviews, api_calls = excluded.api_calls,
++ unique_visitors = excluded.unique_visitors, unique_users = excluded.unique_users, bot_hits = excluded.bot_hits,
++ avg_response_ms = excluded.avg_response_ms, error_count = excluded.error_count, top_paths = excluded.top_paths, top_referers = excluded.top_referers`,
++ [svc, hourStart, stats.pageviews, stats.api_calls, Math.max(Number(prev.unique_visitors || 0), Number(v.visitors || 0)),
++ Math.max(Number(prev.unique_users || 0), Number(v.users || 0)), stats.bot_hits, stats.avg_response_ms || 0, stats.error_count,
++ JSON.stringify(topPaths), JSON.stringify(topReferers)]);
++ }
++
++ async _aggregateDay(date) {
++ const range = [this.service, `${date} 00:00:00`, `${date} 23:59:59`];
++ const svc = this.service;
++ const stats = await this.db.maybe(`
++ SELECT COUNT(*) FILTER (WHERE event_type = 'pageview') AS pageviews, COUNT(*) FILTER (WHERE event_type = 'api_call') AS api_calls,
++ COUNT(*) FILTER (WHERE is_bot = 1) AS bot_hits, CAST(AVG(response_time_ms) AS integer) AS avg_response_ms,
++ COUNT(*) FILTER (WHERE status_code >= 400) AS error_count
++ FROM analytics_events WHERE service = $1 AND created_at >= $2 AND created_at <= $3`, range);
++ if (!stats || (Number(stats.pageviews) === 0 && Number(stats.api_calls) === 0)) return;
++ const v = await this.db.maybe(`SELECT COUNT(DISTINCT vhash) AS visitors, COUNT(DISTINCT CASE WHEN authenticated = 1 THEN vhash END) AS users
++ FROM analytics_visitor_days WHERE service = $1 AND day = $2`, [svc, date]);
++ const prev = (await this.db.maybe('SELECT unique_visitors, unique_users, new_users FROM analytics_daily WHERE service = $1 AND date = $2', [svc, date])) || {};
++ const top = (col, limit, extra = '') => this.db.many(`SELECT ${col}, COUNT(*) AS cnt FROM analytics_events WHERE service = $1 AND created_at >= $2 AND created_at <= $3
++ AND is_bot = 0 ${extra} GROUP BY ${col} ORDER BY cnt DESC${limit ? ` LIMIT ${limit}` : ''}`, range);
++ const [topPaths, topReferers, topCountries, devices, browsers] = [await top('path', 15), await top('referer', 10, "AND referer != '' AND referer IS NOT NULL"),
++ await top('country', 10, 'AND country IS NOT NULL'), await top('device_type', 0), await top('browser', 0)];
++ await this.db.query(`
++ INSERT INTO analytics_daily (service, date, pageviews, api_calls, unique_visitors, unique_users, new_users, bot_hits,
++ avg_response_ms, error_count, top_paths, top_referers, top_countries, device_breakdown, browser_breakdown)
++ VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15)
++ ON CONFLICT (service, date) DO UPDATE SET pageviews = excluded.pageviews, api_calls = excluded.api_calls,
++ unique_visitors = excluded.unique_visitors, unique_users = excluded.unique_users, new_users = excluded.new_users,
++ bot_hits = excluded.bot_hits, avg_response_ms = excluded.avg_response_ms, error_count = excluded.error_count,
++ top_paths = excluded.top_paths, top_referers = excluded.top_referers, top_countries = excluded.top_countries,
++ device_breakdown = excluded.device_breakdown, browser_breakdown = excluded.browser_breakdown`,
++ [svc, date, stats.pageviews, stats.api_calls, Math.max(Number(prev.unique_visitors || 0), Number(v.visitors || 0)),
++ Math.max(Number(prev.unique_users || 0), Number(v.users || 0)), prev.new_users != null ? prev.new_users : null,
++ stats.bot_hits, stats.avg_response_ms || 0, stats.error_count, JSON.stringify(topPaths), JSON.stringify(topReferers),
++ JSON.stringify(topCountries), JSON.stringify(devices), JSON.stringify(browsers)]);
++ }
++
++ // ── Reads (admin dashboards): the SQLite tracker's shapes ──
++
++ async getStats(options = {}) {
++ const { days = 30, hours, service } = options;
++ const svc = service || this.service;
++ const nowMs = this._now();
++ const rangeMs = hours ? hours * 3600000 : days * DAY;
++ const rangeCutoff = sqlTime(nowMs - rangeMs);
++ const isSubDay = hours && hours < 24;
++ const bucketMin = hours <= 1 ? 5 : hours <= 12 ? 15 : 30;
++ const bucketExpr = `substr(created_at, 1, 14) || ((substr(created_at, 15, 2)::int / ${bucketMin}) * ${bucketMin})::text`;
++ const all = (sql, p) => this.db.many(sql, p);
++ const get = (sql, p) => this.db.maybe(sql, p);
++ const safe = async (fn, fallback) => { try { return (await fn()) || fallback; } catch { return fallback; } };
++ const human = [svc, rangeCutoff];
++ const hourlyLimit = hours ? Math.min(hours, 168) : 24;
++ const hourlyCutoff = sqlTime(nowMs - hourlyLimit * 3600000);
++
++ const daily = isSubDay ? [] : await all('SELECT * FROM analytics_daily WHERE service = $1 AND date >= $2 ORDER BY date ASC', [svc, dayOf(nowMs - rangeMs)]);
++ const hourly = await all('SELECT * FROM analytics_hourly WHERE service = $1 AND hour >= $2 ORDER BY hour ASC', [svc, hourlyCutoff]);
++ const timeBuckets = isSubDay ? await all(`
++ SELECT ${bucketExpr} AS bucket,
++ COUNT(*) FILTER (WHERE event_type = 'pageview' AND is_bot = 0) AS pageviews, COUNT(*) FILTER (WHERE event_type = 'api_call' AND is_bot = 0) AS api_calls,
++ COUNT(DISTINCT session_id) FILTER (WHERE is_bot = 0) AS unique_visitors, COUNT(*) FILTER (WHERE is_bot = 1) AS bot_hits,
++ COUNT(*) FILTER (WHERE status_code >= 400) AS errors, CAST(AVG(response_time_ms) AS integer) AS avg_response_ms
++ FROM analytics_events WHERE service = $1 AND created_at >= $2 GROUP BY bucket ORDER BY bucket ASC`, human) : [];
++ const summary = hours && hours < 48 ? await get(`
++ SELECT COUNT(*) FILTER (WHERE event_type = 'pageview' AND is_bot = 0) AS total_pageviews, COUNT(*) FILTER (WHERE event_type = 'api_call' AND is_bot = 0) AS total_api_calls,
++ COUNT(DISTINCT session_id) FILTER (WHERE is_bot = 0) AS total_unique_visitors, COUNT(DISTINCT session_id) FILTER (WHERE is_bot = 0 AND authenticated = 1) AS total_unique_users,
++ COUNT(*) FILTER (WHERE is_bot = 1) AS total_bot_hits, COUNT(*) FILTER (WHERE status_code >= 400) AS total_errors, CAST(AVG(response_time_ms) AS integer) AS avg_response_ms
++ FROM analytics_events WHERE service = $1 AND created_at >= $2`, human)
++ : await get(`
++ SELECT SUM(pageviews)::bigint AS total_pageviews, SUM(api_calls)::bigint AS total_api_calls, SUM(unique_visitors)::bigint AS total_unique_visitors,
++ SUM(unique_users)::bigint AS total_unique_users, SUM(bot_hits)::bigint AS total_bot_hits, SUM(error_count)::bigint AS total_errors,
++ CAST(AVG(avg_response_ms) AS integer) AS avg_response_ms
++ FROM analytics_daily WHERE service = $1 AND date >= $2`, [svc, dayOf(nowMs - rangeMs)]);
++ const realtime = await get(`SELECT COUNT(*) AS requests, COUNT(DISTINCT session_id) AS visitors, COUNT(*) FILTER (WHERE is_bot = 1) AS bots
++ FROM analytics_events WHERE service = $1 AND created_at >= $2`, [svc, sqlTime(nowMs - 300000)]);
++ const topPages = await all(`SELECT path, COUNT(*) AS hits, COUNT(DISTINCT session_id) AS visitors FROM analytics_events
++ WHERE service = $1 AND created_at >= $2 AND is_bot = 0 GROUP BY path ORDER BY hits DESC LIMIT 20`, human);
++ const topReferers = await all(`SELECT referer, COUNT(*) AS hits FROM analytics_events WHERE service = $1 AND created_at >= $2
++ AND referer IS NOT NULL AND referer != '' AND is_bot = 0 GROUP BY referer ORDER BY hits DESC LIMIT 15`, human);
++ const botBreakdown = await all(`SELECT bot_type, COUNT(*) AS hits FROM analytics_events WHERE service = $1 AND created_at >= $2 AND is_bot = 1
++ GROUP BY bot_type ORDER BY hits DESC LIMIT 15`, human);
++ const statusCodes = await all(`SELECT CASE WHEN status_code >= 200 AND status_code < 300 THEN '2xx' WHEN status_code >= 300 AND status_code < 400 THEN '3xx'
++ WHEN status_code >= 400 AND status_code < 500 THEN '4xx' WHEN status_code >= 500 THEN '5xx' ELSE 'other' END AS group_code, COUNT(*) AS cnt
++ FROM analytics_events WHERE service = $1 AND created_at >= $2 GROUP BY group_code ORDER BY cnt DESC`, human);
++ const breakdown = (col) => all(`SELECT ${col}, COUNT(*) AS cnt FROM analytics_events WHERE service = $1 AND created_at >= $2 AND is_bot = 0
++ GROUP BY ${col} ORDER BY cnt DESC`, human);
++ const deviceBreakdown = await breakdown('device_type');
++ const browserBreakdown = await breakdown('browser');
++ const osBreakdown = await breakdown('os');
++ const countryBreakdown = await all(`SELECT country, COUNT(*) AS cnt FROM analytics_events WHERE service = $1 AND created_at >= $2
++ AND country IS NOT NULL AND is_bot = 0 GROUP BY country ORDER BY cnt DESC LIMIT 20`, human);
++ const responsePercentiles = await safe(async () => {
++ const r = await get(`SELECT percentile_disc(0.5) WITHIN GROUP (ORDER BY response_time_ms) AS p50, percentile_disc(0.9) WITHIN GROUP (ORDER BY response_time_ms) AS p90,
++ percentile_disc(0.95) WITHIN GROUP (ORDER BY response_time_ms) AS p95, percentile_disc(0.99) WITHIN GROUP (ORDER BY response_time_ms) AS p99,
++ MAX(response_time_ms) AS max
++ FROM analytics_events WHERE service = $1 AND created_at >= $2 AND response_time_ms IS NOT NULL AND is_bot = 0`, human);
++ return r && r.max != null ? r : {};
++ }, {});
++ const bandwidth = await safe(async () => {
++ const bw = await get(`SELECT COUNT(*) AS total_requests, COUNT(*) FILTER (WHERE event_type = 'pageview') AS page_requests,
++ COUNT(*) FILTER (WHERE event_type = 'api_call') AS api_requests, SUM(CASE WHEN event_type = 'pageview' THEN 50000 ELSE 2000 END)::bigint AS estimated_bytes
++ FROM analytics_events WHERE service = $1 AND created_at >= $2 AND is_bot = 0`, human);
++ return bw && { total_requests: bw.total_requests || 0, page_requests: bw.page_requests || 0, api_requests: bw.api_requests || 0, estimated_bytes: bw.estimated_bytes || 0 };
++ }, {});
++ const authBreakdown = await safe(() => get(`SELECT COUNT(*) FILTER (WHERE authenticated = 1) AS authenticated, COUNT(*) FILTER (WHERE authenticated = 0) AS anonymous,
++ COUNT(DISTINCT session_id) FILTER (WHERE authenticated = 1) AS unique_authenticated_users, COUNT(DISTINCT session_id) FILTER (WHERE session_id IS NOT NULL) AS total_sessions
++ FROM analytics_events WHERE service = $1 AND created_at >= $2 AND is_bot = 0`, human), {});
++ const errorTrend = await safe(() => (isSubDay
++ ? all(`SELECT ${bucketExpr} AS bucket, COUNT(*) AS total, COUNT(*) FILTER (WHERE status_code >= 400) AS errors,
++ COUNT(*) FILTER (WHERE status_code >= 500) AS server_errors
++ FROM analytics_events WHERE service = $1 AND created_at >= $2 AND is_bot = 0 GROUP BY bucket ORDER BY bucket ASC`, human)
++ : all('SELECT hour AS bucket, (pageviews + api_calls) AS total, error_count AS errors FROM analytics_hourly WHERE service = $1 AND hour >= $2 ORDER BY hour ASC', [svc, hourlyCutoff])), []);
++ const peakHours = await safe(() => all(`SELECT substr(created_at, 12, 2)::int AS hour_of_day, COUNT(*) AS hits, COUNT(DISTINCT session_id) AS visitors,
++ CAST(AVG(response_time_ms) AS integer) AS avg_ms
++ FROM analytics_events WHERE service = $1 AND created_at >= $2 AND is_bot = 0 GROUP BY hour_of_day ORDER BY hour_of_day ASC`, human), []);
++ const topApiEndpoints = await safe(() => all(`SELECT path, method, COUNT(*) AS hits, CAST(AVG(response_time_ms) AS integer) AS avg_ms,
++ COUNT(*) FILTER (WHERE status_code >= 400) AS errors
++ FROM analytics_events WHERE service = $1 AND created_at >= $2 AND event_type = 'api_call' AND is_bot = 0 GROUP BY path, method ORDER BY hits DESC LIMIT 20`, human), []);
++ const sessionCount = await safe(() => this.db.value(`SELECT COUNT(DISTINCT session_id) FROM analytics_events WHERE service = $1 AND created_at >= $2
++ AND session_id IS NOT NULL AND is_bot = 0`, human), 0);
++ const visitorTypes = { new_visitors: null, returning_visitors: null, not_collected: 'ADR-021' };
++ const slowestEndpoints = await safe(() => all(`SELECT path, method, COUNT(*) AS hits, CAST(AVG(response_time_ms) AS integer) AS avg_ms, MAX(response_time_ms) AS max_ms
++ FROM analytics_events WHERE service = $1 AND created_at >= $2 AND response_time_ms IS NOT NULL AND is_bot = 0
++ GROUP BY path, method HAVING COUNT(*) >= 3 ORDER BY avg_ms DESC LIMIT 10`, human), []);
++ const authTrend = await safe(() => all(`SELECT ${isSubDay ? bucketExpr : 'substr(created_at, 1, 10)'} AS bucket,
++ COUNT(DISTINCT session_id) FILTER (WHERE authenticated = 1) AS auth_visitors, COUNT(DISTINCT session_id) FILTER (WHERE authenticated = 0) AS anon_visitors,
++ COUNT(*) FILTER (WHERE authenticated = 1) AS auth_hits, COUNT(*) FILTER (WHERE authenticated = 0) AS anon_hits
++ FROM analytics_events WHERE service = $1 AND created_at >= $2 AND is_bot = 0 GROUP BY bucket ORDER BY bucket ASC`, human), []);
++ return {
++ service: svc, period_days: hours ? hours / 24 : days, period_hours: hours || null, summary: summary || {}, realtime: realtime || {},
++ daily, hourly, timeBuckets, topPages, topReferers, botBreakdown, statusCodes, deviceBreakdown, browserBreakdown, osBreakdown,
++ countryBreakdown, responsePercentiles, bandwidth, authBreakdown, errorTrend, peakHours, topApiEndpoints, sessionCount, visitorTypes,
++ slowestEndpoints, authTrend,
++ };
++ }
++
++ async getOverview(days = 30) {
++ const cutoff = dayOf(this._now() - days * DAY);
++ const services = await this.db.many(`SELECT service, SUM(pageviews)::bigint AS pageviews, SUM(api_calls)::bigint AS api_calls, SUM(unique_visitors)::bigint AS unique_visitors,
++ SUM(unique_users)::bigint AS unique_users, SUM(bot_hits)::bigint AS bot_hits, SUM(error_count)::bigint AS errors, CAST(AVG(avg_response_ms) AS integer) AS avg_response_ms
++ FROM analytics_daily WHERE date >= $1 GROUP BY service ORDER BY pageviews DESC`, [cutoff]);
++ const totals = await this.db.maybe(`SELECT SUM(pageviews)::bigint AS pageviews, SUM(api_calls)::bigint AS api_calls, SUM(unique_visitors)::bigint AS unique_visitors,
++ SUM(bot_hits)::bigint AS bot_hits, SUM(error_count)::bigint AS errors FROM analytics_daily WHERE date >= $1`, [cutoff]);
++ const dailyTrend = await this.db.many(`SELECT date, SUM(pageviews)::bigint AS pageviews, SUM(api_calls)::bigint AS api_calls, SUM(unique_visitors)::bigint AS unique_visitors,
++ SUM(bot_hits)::bigint AS bot_hits FROM analytics_daily WHERE date >= $1 GROUP BY date ORDER BY date ASC`, [cutoff]);
++ const realtime = await this.db.many(`SELECT service, COUNT(*) AS requests, COUNT(DISTINCT session_id) AS visitors, COUNT(*) FILTER (WHERE is_bot = 1) AS bots
++ FROM analytics_events WHERE created_at >= $1 GROUP BY service`, [sqlTime(this._now() - 300000)]);
++ return { period_days: days, totals: totals || {}, services, dailyTrend, realtime };
++ }
++
++ async getBotAnalysis(days = 30) {
++ const since = sqlTime(this._now() - Math.min(Math.max(1, days), 365) * DAY);
++ const topBotIPs = await this.db.many(`SELECT NULL AS ip, user_agent AS ua_class, bot_type, COUNT(*) AS hits, COUNT(DISTINCT session_id) AS sessions,
++ MIN(created_at) AS first_seen, MAX(created_at) AS last_seen
++ FROM analytics_events WHERE is_bot = 1 AND created_at >= $1 GROUP BY user_agent, bot_type ORDER BY hits DESC LIMIT 25`, [since]);
++ const botTrend = await this.db.many(`SELECT date, (SUM(pageviews) + SUM(api_calls))::bigint AS human_hits, SUM(bot_hits)::bigint AS bot_hits
++ FROM analytics_daily WHERE date >= $1 GROUP BY date ORDER BY date ASC`, [since.slice(0, 10)]);
++ const botTypes = await this.db.many(`SELECT bot_type, COUNT(*) AS hits, NULL AS unique_ips, COUNT(DISTINCT session_id) AS unique_sessions
++ FROM analytics_events WHERE is_bot = 1 AND created_at >= $1 GROUP BY bot_type ORDER BY hits DESC`, [since]);
++ const suspiciousIPs = await this.db.many(`SELECT NULL AS ip, session_id, COUNT(*) AS total_hits, COUNT(*) FILTER (WHERE status_code >= 400) AS error_hits,
++ COUNT(DISTINCT path) AS unique_paths, MIN(created_at) AS first_seen, MAX(created_at) AS last_seen
++ FROM analytics_events WHERE created_at >= $1 AND is_bot = 0 AND session_id IS NOT NULL
++ GROUP BY session_id HAVING COUNT(*) > 500 ORDER BY total_hits DESC LIMIT 20`, [sqlTime(this._now() - 7 * DAY)]);
++ return { topBotIPs, botTrend, botTypes, suspiciousIPs };
++ }
++
++ async destroy() {
++ await this.flush().catch(() => {});
++ for (const t of this._timers) { clearInterval(t); clearTimeout(t); }
++ this._timers = [];
++ this._rates.clear();
++ this._sessions.clear();
++ }
++}
++
++/** ADR-021's raw-event retention on PostgreSQL: rows older than `days` go, in batches; the rollups stay. */
++async function pruneRawEventsPg(db, { days = 30, now = Date.now, batch = 5000 } = {}) {
++ const cutoff = sqlTime(now() - days * DAY);
++ let removed = 0;
++ for (;;) {
++ const n = await db.exec(`DELETE FROM analytics_events WHERE id IN (SELECT id FROM analytics_events WHERE created_at < $1 LIMIT ${batch})`, [cutoff]);
++ removed += n;
++ if (n < batch) break;
++ }
++ const today = dayOf(now());
++ await db.exec('DELETE FROM analytics_visitor_days WHERE day < $1', [dayOf(now() - DAY)]);
++ await db.exec('DELETE FROM analytics_day_salts WHERE day < $1', [today]);
++ return { removed, cutoff };
++}
++
++module.exports = { AnalyticsTrackerPg, analyticsSchema, pruneRawEventsPg, privacy };
ddiff --git a/analytics/tracker.js b/analytics/tracker.js
iindex b71d7b5..4cc1492 100644
---- a/analytics/tracker.js
++++ b/analytics/tracker.js
@@@ -84,6 +84,7 @@ class AnalyticsTracker {
this._salt = null;
this._timers = [];
++ if (opts.__pg) return; // analytics/pg.js: an openvibe-sdk/db handle, its own writes and timers
ensureSchema(db);
try { db.pragma('secure_delete = ON'); } catch { /* not fatal */ }
try { db.pragma('busy_timeout = 250'); } catch { /* not fatal */ }
ddiff --git a/package-lock.json b/package-lock.json
iindex fc8b7c0..b206806 100644
---- a/package-lock.json
++++ b/package-lock.json
@@@ -1,12 +1,12 @@
{
"name": "openvibe-shared",
-- "version": "1.28.1",
++ "version": "1.29.0",
"lockfileVersion": 3,
"requires": true,
"packages": {
"": {
"name": "openvibe-shared",
-- "version": "1.28.1",
++ "version": "1.29.0",
"license": "MIT",
"dependencies": {
"jsonwebtoken": "^9.0.0"
ddiff --git a/package.json b/package.json
iindex 20e4022..5d0313d 100644
---- a/package.json
++++ b/package.json
@@@ -1,6 +1,6 @@
{
"name": "openvibe-shared",
-- "version": "1.28.1",
++ "version": "1.29.0",
"description": "Shared navbar, footer, themes, icons, legal pages, SEO, auth and analytics for every OpenVibe site",
"main": "index.js",
"exports": {
@@@ -10,6 +10,7 @@
"./analytics/event": "./analytics/event.js",
"./analytics/event.v1.json": "./docs/schemas/analytics-event.v1.json",
"./analytics/privacy": "./analytics/privacy.js",
++ "./analytics/pg": "./analytics/pg.js",
"./analytics/prune-cli": "./analytics/prune-cli.js",
"./analytics/retention": "./analytics/retention.js",
"./analytics/schema": "./analytics/schema.js",
ddiff --git a/test/analytics-pg.test.js b/test/analytics-pg.test.js
nnew file mode 100644
iindex 0000000..21f6e2f
---- /dev/null
++++ b/test/analytics-pg.test.js
@@@ -0,0 +1,57 @@
++'use strict';
++/**
++ * openvibe-shared/analytics/pg on PGlite: request rows reach the database only at flush; the day's salt is shared
++ * across trackers (two processes count one visitor once); rollups and dashboards keep the SQLite tracker's shapes;
++ * opted-out requests record nothing; retention prunes old rows. No IP ever reaches the database.
++ */
++const assert = require('assert');
++let createDb;
++try { ({ createDb } = require('openvibe-sdk/db')); } catch { console.log('analytics-pg: skipped (openvibe-sdk not installed)'); process.exit(0); }
++const { AnalyticsTrackerPg, analyticsSchema, pruneRawEventsPg } = require('../analytics/pg');
++
++const T0 = Date.parse('2026-09-28T12:10:00Z');
++const req = (path, ip, extra = {}) => ({ path, method: 'GET', ip, headers: { 'user-agent': 'Mozilla/5.0 (X11; Linux x86_64) Chrome/140.0 Safari/537.36', ...(extra.headers || {}) }, user: extra.user || null });
++const res = (status = 200) => ({ statusCode: status });
++
++(async () => {
++ const db = createDb({ pglite: true, service: 'analytics-test', log: { log() {}, warn() {}, error() {}, info() {} } });
++ await db.query(analyticsSchema());
++ let now = T0;
++ const a = new AnalyticsTrackerPg(db, 'network', { timers: false, now: () => now });
++ const b = new AnalyticsTrackerPg(db, 'network', { timers: false, now: () => now });
++
++ assert.strictEqual(a.record(req('/about', '203.0.113.7'), res(), '/about', 12), true);
++ assert.strictEqual(Number(await db.value('SELECT COUNT(*) FROM analytics_events')), 0, 'nothing written before the flush');
++ b.record(req('/about', '203.0.113.7'), res(), '/about', 20); // the same visitor, another process
++ a.record(req('/api/v1/x', '198.51.100.2', { user: { id: 1 } }), res(500), '/api/v1/x', 40);
++ assert.strictEqual(a.record(req('/about', '192.0.2.9', { headers: { dnt: '1' } }), res(), '/about', 5), false, 'DNT records nothing');
++ await a.flush(); await b.flush();
++ assert.strictEqual(Number(await db.value('SELECT COUNT(*) FROM analytics_events')), 3);
++ assert.strictEqual(Number(await db.value("SELECT COUNT(*) FROM analytics_events WHERE ip IS NOT NULL OR user_id IS NOT NULL OR city IS NOT NULL")), 0, 'no personal columns');
++ assert.strictEqual(Number(await db.value('SELECT COUNT(*) FROM analytics_visitor_days')), 2, 'one visitor across both processes, plus the signed-in one');
++ assert.strictEqual(Number(await db.value('SELECT COUNT(*) FROM analytics_day_salts')), 1, 'one shared salt for the day');
++
++ await a.aggregate();
++ const h = await db.maybe("SELECT pageviews, api_calls, unique_visitors, unique_users, error_count FROM analytics_hourly WHERE service = 'network'");
++ assert.deepStrictEqual([h.pageviews, h.api_calls, h.unique_visitors, h.unique_users, h.error_count].map(Number), [2, 1, 2, 1, 1]);
++
++ const stats = await a.getStats({ hours: 6 });
++ assert.strictEqual(Number(stats.summary.total_pageviews), 2);
++ assert.ok(stats.timeBuckets.length >= 1 && /^2026-09-28 12:\d+$/.test(stats.timeBuckets[0].bucket), JSON.stringify(stats.timeBuckets[0]));
++ assert.strictEqual(stats.responsePercentiles.max, 40);
++ assert.ok(Array.isArray(stats.peakHours) && stats.peakHours[0].hour_of_day === 12);
++ const overview = await a.getOverview(7);
++ assert.ok(Array.isArray(overview.services));
++ const bots = await a.getBotAnalysis(7);
++ assert.ok(Array.isArray(bots.botTypes));
++
++ // Next day: yesterday's hashes and salt go at aggregation; retention prunes raw rows past the window.
++ now = T0 + 86400000;
++ await a.aggregate();
++ assert.strictEqual(Number(await db.value('SELECT COUNT(*) FROM analytics_visitor_days')), 0);
++ now = T0 + 31 * 86400000;
++ const r = await pruneRawEventsPg(db, { days: 30, now: () => now });
++ assert.strictEqual(r.removed, 3);
++ await a.destroy(); await b.destroy(); await db.close();
++ console.log('analytics-pg: all checks passed');
++})().catch((err) => { console.error(err); process.exit(1); });