openvibers/openvibe.tools Commits
Tools in Search and in people's recent tools (platform S9).
Changes
nnew file mode 100644@@@ -0,0 +1,71 @@++'use strict';++/** tools.usage: page views and runs by signed-in people become their recent tools in the Network module, merged, deduplicated and capped. */++const assert = require('assert');++const { createRecorder, usagePages } = require('../usage');++++const USR = 'usr_01JAB2C3D4E5F6G7H8J9K0MNPQ';++const modules = new Map(); // subject -> { data, revision }++let conflictOnce = false;++const calls = [];++const fetchImpl = async (url, opts) => {++ const u = new URL(url);++ const reply = (status, body) => ({ status, ok: status < 300, json: async () => body });++ if (u.pathname === '/oauth/token') { calls.push('token'); return reply(200, { access_token: 'svc', expires_in: 300 }); }++ assert.strictEqual(opts.headers.Authorization, 'Bearer svc');++ const m = u.pathname.match(/^\/internal\/modules\/tools\.usage\/(usr_[0-9A-Z]+)$/);++ assert.ok(m, u.pathname);++ calls.push(`${opts.method}`);++ const rec = modules.get(m[1]);++ if (opts.method === 'GET') return rec ? reply(200, { namespace: 'tools.usage', ...rec }) : reply(404, { code: 'modules.not_found' });++ const want = opts.headers['If-Match'];++ if (conflictOnce) { conflictOnce = false; modules.set(m[1], { data: { recent: [{ tool: 'png', at: '2026-09-24T00:00:00.000Z' }] }, revision: 7 }); return reply(412, { code: 'modules.revision_mismatch' }); }++ if (rec && want !== `"${rec.revision}"`) return reply(412, { code: 'modules.revision_mismatch' });++ const data = JSON.parse(opts.body).data;++ modules.set(m[1], { data, revision: rec ? rec.revision + 1 : 1 });++ return reply(rec ? 200 : 201, { data });++};++++(async () => {++ let t = Date.parse('2026-09-24T01:00:00Z');++ const r = createRecorder({ env: { OV_OAUTH_CLIENT_SECRET: 'x'.repeat(40), OV_NETWORK_INTERNAL_URL: 'http://network.test' }, fetchImpl, now: () => t, flushMs: 3_600_000 });++ assert.ok(r.record(USR, 'yaml'));++ assert.ok(!r.record(USR, 'yaml'), 'the same tool again within 10 minutes is not news');++ assert.ok(!r.record('42', 'yaml'), 'only a Network subject counts');++ assert.ok(!r.record(USR, '../etc'), 'only a tool id');++ t += 1000; r.record(USR, 'uuid');++ await r.flush();++ assert.deepStrictEqual(modules.get(USR).data.recent.map((e) => e.tool), ['uuid', 'yaml'], 'newest first');++ assert.strictEqual(calls.filter((c) => c === 'token').length, 1);++++ t += 11 * 60 * 1000; r.record(USR, 'yaml');++ conflictOnce = true;++ await r.flush();++ assert.strictEqual(r.stats().conflicts, 1);++ assert.strictEqual(r.stats().pending, 1, 'a conflicting write stays pending');++ await r.flush();++ assert.deepStrictEqual(modules.get(USR).data.recent.map((e) => e.tool), ['yaml', 'png'], 'merged with what another writer put there');++++ for (let i = 0; i < 40; i++) { t += 1000; r.record(USR, `tool${i}`); }++ await r.flush();++ assert.strictEqual(modules.get(USR).data.recent.length, 30, 'capped at the namespace limit');++ assert.strictEqual(modules.get(USR).data.recent[0].tool, 'tool39');++++ // The page middleware: a signed-in document request for a tool host counts; assets, APIs and unknown hosts do not.++ const got = [];++ const mw = usagePages({ snapshot: () => ({ tools: [{ id: 'jsonminify', hosts: ['json-minifier.openvibe.tools', 'jsonminify.openvibe.tools'] }] }), rec: { enabled: true, record: (s, id) => got.push(id) } });++ const req = (host, p, headers = {}, user = { subject_id: USR }) => ({ method: 'GET', hostname: host, path: p, user, get: (h) => headers[h.toLowerCase()] });++ const run = (q) => mw(q, {}, () => {});++ run(req('json-minifier.openvibe.tools', '/', { 'sec-fetch-dest': 'document' }));++ run(req('jsonminify.openvibe.tools', '/', { accept: 'text/html' }));++ run(req('json-minifier.openvibe.tools', '/app.js', { 'sec-fetch-dest': 'script' }));++ run(req('json-minifier.openvibe.tools', '/api/x', { accept: 'text/html' }));++ run(req('openvibe.tools', '/', { 'sec-fetch-dest': 'document', 'x-ov-tool': 'evil' }));++ run(req('json-minifier.openvibe.tools', '/', { 'sec-fetch-dest': 'document' }, null));++ assert.deepStrictEqual(got, ['jsonminify', 'jsonminify']);++++ const off = createRecorder({ env: {}, fetchImpl });++ assert.strictEqual(off.enabled, false);++ assert.strictEqual(off.record(USR, 'yaml'), false, 'off without a service secret');++ r.stop();++ console.log('tools usage: all checks passed');++})().catch((e) => { console.error(e); process.exit(1); });@@@ -248,6 +248,8 @@ function createRunApi(o) {guard.record(req, c, { reason: 'session', tool: d.id, enforced: true });throw new Refusal(401, 'tools.session_required', `${d.id} needs a browser session (open its page first), a sign-in or an API token.`);}++ // A person running a tool through the API counts in their recent tools (tools.usage).++ if (c.kind === 'user' && c.claims && c.claims.subject_id) { try { require('../usage').recorder().record(c.claims.subject_id, d.id); } catch { /* best-effort */ } }return c;}nnew file mode 100644@@@ -0,0 +1,139 @@++'use strict';++// ═══════════════════════════════════════════════════════════════++// tools.usage (platform S9): a signed-in person's recently used tools, kept in their Network user++// module `tools.usage` ({ recent: [{ tool, at }] }, newest first, 30 at most; openvibe-contracts++// manifests/namespaces/tools.usage.json) for the launchers across the network.++//++// app.use(usagePages({ snapshot: toolRegistry.snapshot })) after guard.identify: a page view of++// a tool by a signed-in person counts++// recorder().record(subjectId, toolId) e.g. a run API call by a person++//++// Cheap by design: a person counts once per tool per 10 minutes, and a process flushes what it++// gathered every minute: per person one read and one conditional write (If-Match), merged with what++// the module already holds (other satellites write it too; a 412 keeps the entry for the next flush).++// Tools' service token (network.modules.read/write for tools.usage) is fetched here without++// dependencies, so every satellite can use it. Off unless OV_OAUTH_CLIENT_SECRET is set++// (TOOLS_USAGE=off disables it). No dependencies: required by relative path.++// ═══════════════════════════════════════════════════════════════++++const NS = 'tools.usage';++const MAX = 30;++const SAME_TOOL_MS = 10 * 60 * 1000;++const FLUSH_MS = 60 * 1000;++const MAX_PENDING = 2000;++const USR_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;++const TOOL_RE = /^[a-z][a-z0-9-]{0,39}$/;++++function createRecorder({ env = process.env, fetchImpl = globalThis.fetch, now = () => Date.now(), log = console, flushMs = FLUSH_MS } = {}) {++ const network = String(env.OV_NETWORK_INTERNAL_URL || 'http://127.0.0.1:4000').replace(/\/+$/, '');++ const secret = env.OV_OAUTH_CLIENT_SECRET || '';++ const enabled = !!secret && String(env.TOOLS_USAGE || '').toLowerCase() !== 'off';++ const pending = new Map(); // subject -> Map(tool -> at ISO)++ const seen = new Map(); // `${subject}|${tool}` -> ms++ const stats = { recorded: 0, written: 0, conflicts: 0, failed: 0, last_error: null };++ let token = null, tokenExp = 0, timer = null;++++ async function bearer() {++ if (token && now() < tokenExp) return token;++ const body = new URLSearchParams({ grant_type: 'client_credentials', client_id: env.OV_OAUTH_CLIENT_ID || 'tools', client_secret: secret, audience: 'openvibe.network' });++ const r = await fetchImpl(`${network}/oauth/token`, { method: 'POST', headers: { 'Content-Type': 'application/x-www-form-urlencoded', Accept: 'application/json' }, body, signal: AbortSignal.timeout(10000) });++ const j = await r.json().catch(() => ({}));++ if (!r.ok || !j.access_token) throw new Error(`token: ${r.status} ${j.error || ''}`);++ token = j.access_token; tokenExp = now() + Math.max(30, (Number(j.expires_in) || 300) - 60) * 1000;++ return token;++ }++ async function call(method, subject, { body, revision } = {}) {++ const headers = { Accept: 'application/json', Authorization: `Bearer ${await bearer()}` };++ if (body) headers['Content-Type'] = 'application/json';++ if (revision != null) headers['If-Match'] = `"${revision}"`;++ const r = await fetchImpl(`${network}/internal/modules/${NS}/${subject}`, { method, headers, body: body ? JSON.stringify(body) : undefined, signal: AbortSignal.timeout(10000) });++ if (r.status === 401) token = null;++ return { status: r.status, body: await r.json().catch(() => ({})) };++ }++++ function record(subject, tool) {++ if (!enabled || !USR_RE.test(String(subject || '')) || !TOOL_RE.test(String(tool || ''))) return false;++ const k = `${subject}|${tool}`, t = now();++ if (seen.has(k) && t - seen.get(k) < SAME_TOOL_MS) return false;++ if (seen.size > 20000) seen.clear();++ seen.set(k, t);++ if (!pending.has(subject) && pending.size >= MAX_PENDING) return false;++ if (!pending.has(subject)) pending.set(subject, new Map());++ const p = pending.get(subject);++ p.delete(tool); p.set(tool, new Date(t).toISOString());++ stats.recorded++;++ start();++ return true;++ }++++ async function flushOne(subject, fresh) {++ const cur = await call('GET', subject);++ if (cur.status !== 200 && cur.status !== 404) throw new Error(`read: ${cur.status}`);++ const had = cur.status === 200 && cur.body && cur.body.data && Array.isArray(cur.body.data.recent) ? cur.body.data.recent : [];++ const add = [...fresh].reverse().map(([tool, at]) => ({ tool, at }));++ const recent = [...add, ...had.filter((e) => e && !fresh.has(e.tool))].slice(0, MAX);++ const w = await call('PUT', subject, { body: { data: { recent } }, revision: cur.status === 200 ? cur.body.revision : undefined });++ if (w.status === 412 || w.status === 409) { stats.conflicts++; return false; }++ if (w.status >= 300) throw new Error(`write: ${w.status} ${(w.body && w.body.code) || ''}`);++ stats.written++;++ return true;++ }++++ async function flush() {++ const batch = [...pending];++ pending.clear();++ for (const [subject, fresh] of batch) {++ let ok = false;++ try { ok = await flushOne(subject, fresh); } catch (err) { stats.failed++; stats.last_error = err.message; }++ if (!ok && pending.size < MAX_PENDING) {++ const back = pending.get(subject) || new Map();++ for (const [tool, at] of fresh) if (!back.has(tool)) back.set(tool, at);++ pending.set(subject, back);++ }++ }++ if (!pending.size) stop();++ }++ function start() { if (timer || !enabled) return; timer = setInterval(() => { flush().catch(() => {}); }, flushMs); if (timer.unref) timer.unref(); }++ function stop() { if (timer) clearInterval(timer); timer = null; }++++ return { enabled, record, flush, stop, stats: () => ({ enabled, pending: pending.size, ...stats }) };++}++++let _recorder = null;++function recorder() { if (!_recorder) _recorder = createRecorder(); return _recorder; }++++/**++ * Middleware: a GET of a tool's page by a signed-in person (req.user.subject_id from guard.identify)++ * records that tool. The host names the tool (the registry's hosts, or the X-OV-Tool the gateway sends).++ */++function usagePages({ snapshot, rec = null }) {++ let map = null, mapFor = null;++ const hostMap = () => {++ const s = snapshot();++ if (map && mapFor === s) return map;++ map = new Map();++ for (const t of s.tools || []) for (const h of t.hosts || []) map.set(String(h).toLowerCase(), t.id);++ map.ids = new Set((s.tools || []).map((t) => t.id));++ mapFor = s;++ return map;++ };++ return function toolUsage(req, _res, next) {++ try {++ const sid = req.user && req.user.subject_id;++ const r = rec || recorder();++ if (sid && r.enabled && req.method === 'GET' && !req.path.startsWith('/api/') && !/\.[a-z0-9]{1,8}$/i.test(req.path)) {++ const dest = req.get('sec-fetch-dest');++ if (dest ? dest === 'document' : /text\/html/.test(req.get('accept') || '')) {++ const host = String(req.hostname || '').toLowerCase();++ const m = hostMap();++ const named = String(req.get('x-ov-tool') || '');++ const tool = m.get(host) || (m.ids.has(named) ? named : null);++ if (tool) r.record(sid, tool);++ }++ }++ } catch { /* usage is best-effort */ }++ next();++ };++}++++module.exports = { createRecorder, recorder, usagePages, NS };@@@ -130,6 +130,8 @@ app.use('/api/', apiLimiter, guard.apiQuota);// Public (Access-Control-Allow-Origin *), cacheable (ETag), counted by the limiter above. The// gateway (openvibe.tools) answers the same routes for every tool and reads this list for status.const toolRegistry = createLocalRegistry({ specs: require('./descriptors').SPECS, statusOf: requiresStatus((p) => (p === 'ffmpeg' ? !!which(process.env.FFMPEG_PATH || 'ffmpeg') : true)) });++// A signed-in person's page view of a tool goes into their tools.usage module (the launchers' recent tools).++app.use(require('../../_shared/usage').usagePages({ snapshot: toolRegistry.snapshot }));app.use(createToolsApi({ snapshot: toolRegistry.snapshot }));// ── Analytics Middleware ─────────────────────────────────────@@@ -147,6 +147,8 @@ app.use('/api/', apiLimiter, guard.apiQuota);// gateway (openvibe.tools) answers the same routes for every tool and reads this list for status.const pdfPrograms = require('./tools/pdf');const toolRegistry = createLocalRegistry({ specs: require('./descriptors').SPECS, statusOf: requiresStatus((p) => (pdfPrograms[p] ? pdfPrograms[p].available() : true)) });++// A signed-in person's page view of a tool goes into their tools.usage module (the launchers' recent tools).++app.use(require('../../_shared/usage').usagePages({ snapshot: toolRegistry.snapshot }));app.use(createToolsApi({ snapshot: toolRegistry.snapshot }));// ── Analytics Middleware ─────────────────────────────────────@@@ -181,6 +181,8 @@ app.use(exceptRegistry(cors({// ── Who is asking, then rate limits (sign-in first, so a signed-in tier applies) ──app.use(guard.identify);++// A signed-in person's page view of a tool goes into their tools.usage module (the launchers' recent tools).++app.use(require('../../_shared/usage').usagePages({ snapshot: toolRegistry.snapshot }));app.use('/api/', guard.legacyLimiter(rateLimit, { windowMs: 60_000, anonymous: 120, signedIn: 240, message: 'Too many requests. Please try again later.' }), guard.apiQuota);app.use('/auth/', rateLimit({ windowMs: 15 * 60_000, max: 60, keyGenerator: (req) => guard.caller(req).ipKey }));@@@ -253,7 +255,7 @@ async function requireAuth(req, res, next) {// ── Basic API ────────────────────────────────────────────────app.get('/api/health', (_req, res) => {-- res.json({ status: 'ok', service: 'openvibe-tools-gateway', version: '2.0.0', registry: registry.services.status(), run: runApi.stats(), jobs: jobsFacade.stats() });++ res.json({ status: 'ok', service: 'openvibe-tools-gateway', version: '2.0.0', registry: registry.services.status(), run: runApi.stats(), jobs: jobsFacade.stats(), search_index: searchIndex.indexer ? searchIndex.indexer.stats() : { enabled: false, reason: searchIndex.reason } });});app.get('/api/brand', (_req, res) => res.json(BRAND));@@@ -433,6 +435,10 @@ app.get('*', (req, res) => {// What each satellite says about its own tools (a program missing on the host → unavailable here too).toolRegistry.startSatellitePolling(SATELLITES);++// Every tool as an OpenVibe.Search document (owner tools), kept in step with the registry (S9).++const searchIndex = require('./search-index').searchIndexerFromEnv({ snapshot: toolRegistry.snapshot, dataDir: require('path').resolve(__dirname, '..', process.env.DATA_DIR || 'data') });++if (searchIndex.indexer) searchIndex.indexer.start(); else console.log(`[SearchIndex] off: ${searchIndex.reason}`);++app.listen(config.port, config.host, () => {console.log(`\n╔════════════════════════════════════════════╗`);console.log(`║ 🔧 openvibe.tools — Tools Gateway ║`);nnew file mode 100644@@@ -0,0 +1,114 @@++'use strict';++/**++ * Tools → OpenVibe.Search (platform S9): every tool in the registry is one Search document++ * (search.index-document@1, owner tools, type tool, id = the tool id), written through Search's owner++ * API with the tools service token (search.document.write on openvibe.search). A tool that leaves the++ * registry, or becomes unavailable on this host, is tombstoned.++ *++ * Runs a minute after boot and every 6 hours. Search's own inventory (GET /api/v1/owners/tools/documents)++ * gives the stored revisions; a document is re-sent, one revision up, only when its content hash++ * differs from the last one sent (data/search-index.json), so a quiet registry costs one listing.++ *++ * Off unless SEARCH_URL and OV_OAUTH_CLIENT_SECRET are set (TOOLS_SEARCH_INDEX=off disables it).++ */++const fs = require('fs');++const path = require('path');++const crypto = require('crypto');++++const OWNER = 'tools';++const TYPE = 'tool';++const EVERY_MS = 6 * 60 * 60 * 1000;++++/** A descriptor → the document body (identity and revision added by the caller). */++function documentFor(t, updatedAt) {++ const host = Array.isArray(t.hosts) && t.hosts[0];++ return {++ visibility: 'public',++ canonical_url: host ? `https://${host}` : t.docs || `https://openvibe.tools/tool/${t.id}`,++ title: String(t.name || t.id).slice(0, 500),++ summary: String(t.summary || '').slice(0, 4000),++ body: [t.name, t.summary, ...(Array.isArray(t.keywords) ? t.keywords : [])].filter(Boolean).join('\n').slice(0, 48000),++ facets: { family: String(t.family || 'other'), execution: String(t.execution || 'client'), api: Boolean(t.api) },++ language: 'en',++ authorship: 'human',++ publication_state: 'published',++ ...(updatedAt ? { updated_at: updatedAt } : {}),++ indexability: { decision: 'index', reasons: [] },++ };++}++const hashOf = (doc) => crypto.createHash('sha256').update(JSON.stringify(doc)).digest('hex');++++function createSearchIndexer({ snapshot, searchUrl, tokens, stateFile, fetchImpl = globalThis.fetch, log = console, now = () => Date.now() }) {++ const base = String(searchUrl).replace(/\/+$/, '');++ const stats = { runs: 0, upserted: 0, deleted: 0, unchanged: 0, failed: 0, last_run_at: null, last_error: null };++ let state = {};++ try { state = JSON.parse(fs.readFileSync(stateFile, 'utf8')).docs || {}; } catch { state = {}; }++ const save = () => {++ try { fs.mkdirSync(path.dirname(stateFile), { recursive: true }); fs.writeFileSync(`${stateFile}.tmp`, JSON.stringify({ docs: state })); fs.renameSync(`${stateFile}.tmp`, stateFile); } catch (err) { log.warn('[SearchIndex] state not saved:', err.message); }++ };++ async function call(method, p, body) {++ const res = await fetchImpl(`${base}${p}`, { method, headers: { Accept: 'application/json', ...(body ? { 'Content-Type': 'application/json' } : {}), ...(await tokens.authHeaders()) }, body: body ? JSON.stringify(body) : undefined, signal: AbortSignal.timeout(15000) });++ const json = await res.json().catch(() => ({}));++ if (res.status === 401 && tokens.invalidate) tokens.invalidate();++ return { status: res.status, ok: res.ok, body: json };++ }++ async function remoteDocs() {++ const out = new Map();++ let after = 0;++ for (let i = 0; i < 50; i++) {++ const r = await call('GET', `/api/v1/owners/${OWNER}/documents?type=${TYPE}&limit=1000&after=${after}`);++ if (!r.ok) throw new Error(`listing: ${r.status} ${r.body.code || ''}`);++ for (const d of r.body.documents || []) out.set(d.id, d);++ if (!r.body.next_after) break;++ after = r.body.next_after;++ }++ return out;++ }++++ async function run() {++ stats.runs++; stats.last_run_at = new Date(now()).toISOString();++ const snap = snapshot();++ const tools = (snap.tools || []).filter((t) => t.status !== 'unavailable' && /^[a-z][a-z0-9-]{0,39}$/.test(t.id));++ const remote = await remoteDocs();++ const keep = new Set();++ for (const t of tools) {++ keep.add(t.id);++ const doc = documentFor(t, snap.updated_at || null);++ const h = hashOf(doc);++ const r = remote.get(t.id);++ if (r && !r.deleted && state[t.id] && state[t.id].hash === h) { stats.unchanged++; continue; }++ const revision = r ? r.revision + 1 : 1;++ const res = await call('PUT', `/api/v1/documents/${OWNER}/${TYPE}/${t.id}`, { ...doc, revision });++ if (res.ok) { state[t.id] = { hash: h, revision }; stats.upserted++; } else { stats.failed++; stats.last_error = `${t.id}: ${res.status} ${res.body.code || ''}`; }++ }++ for (const [id, r] of remote) {++ if (keep.has(id) || r.deleted) continue;++ const res = await call('DELETE', `/api/v1/documents/${OWNER}/${TYPE}/${encodeURIComponent(id)}?revision=${r.revision + 1}`);++ if (res.ok) { delete state[id]; stats.deleted++; } else { stats.failed++; stats.last_error = `${id}: ${res.status} ${res.body.code || ''}`; }++ }++ save();++ return { ...stats };++ }++++ let timer = null;++ function start({ firstDelayMs = 60_000 } = {}) {++ const tick = () => run().catch((err) => { stats.failed++; stats.last_error = err.message; log.warn('[SearchIndex] run failed:', err.message); });++ const first = setTimeout(tick, firstDelayMs); if (first.unref) first.unref();++ timer = setInterval(tick, EVERY_MS); if (timer.unref) timer.unref();++ }++ function stop() { if (timer) clearInterval(timer); timer = null; }++ return { run, start, stop, stats: () => ({ ...stats }) };++}++++/** The gateway's indexer from its environment, or null (with why) when it is off. */++function searchIndexerFromEnv({ snapshot, dataDir, env = process.env, contracts = require('openvibe-contracts'), log = console }) {++ const url = String(env.SEARCH_URL || '').trim();++ if (!url) return { indexer: null, reason: 'SEARCH_URL is not set' };++ if (String(env.TOOLS_SEARCH_INDEX || '').toLowerCase() === 'off') return { indexer: null, reason: 'TOOLS_SEARCH_INDEX=off' };++ if (!env.OV_OAUTH_CLIENT_SECRET) return { indexer: null, reason: 'OV_OAUTH_CLIENT_SECRET is not set' };++ const network = String(env.OV_NETWORK_INTERNAL_URL || 'http://127.0.0.1:4000').replace(/\/+$/, '');++ const tokens = contracts.serviceAuth.createTokenClient({ tokenUrl: `${network}/oauth/token`, clientId: env.OV_OAUTH_CLIENT_ID || 'tools', clientSecret: env.OV_OAUTH_CLIENT_SECRET, audience: 'openvibe.search', scope: 'search.document.write' });++ return { indexer: createSearchIndexer({ snapshot, searchUrl: url, tokens, stateFile: path.join(dataDir, 'search-index.json'), log }) };++}++++module.exports = { createSearchIndexer, searchIndexerFromEnv, documentFor };nnew file mode 100644@@@ -0,0 +1,65 @@++'use strict';++/** Tools → Search: one search.index-document@1 per available tool, re-sent only when it changes, tombstoned when it leaves. */++const assert = require('assert');++const fs = require('fs');++const os = require('os');++const path = require('path');++const contracts = require('openvibe-contracts');++const { createSearchIndexer } = require('../server/search-index');++++const stored = new Map(); // id -> { revision, deleted, doc }++const calls = [];++const fetchImpl = async (url, opts) => {++ const u = new URL(url);++ calls.push(`${opts.method} ${u.pathname}`);++ assert.strictEqual(opts.headers.Authorization, 'Bearer svc');++ const reply = (status, body) => ({ status, ok: status < 300, json: async () => body });++ if (opts.method === 'GET') return reply(200, { owner: 'tools', documents: [...stored].map(([id, d]) => ({ type: 'tool', id, revision: d.revision, deleted: d.deleted })), next_after: null });++ const id = u.pathname.split('/').pop();++ const prev = stored.get(id);++ if (opts.method === 'PUT') {++ const doc = { ...JSON.parse(opts.body), owner: 'tools', type: 'tool', id };++ const v = contracts.validate('search.index-document@1', doc);++ assert.ok(v.valid, JSON.stringify(v.errors));++ if (prev && doc.revision <= prev.revision) return reply(409, { code: 'search.stale_revision' });++ stored.set(id, { revision: doc.revision, deleted: false, doc });++ return reply(200, { outcome: 'applied' });++ }++ const revision = Number(u.searchParams.get('revision'));++ stored.set(id, { revision, deleted: true });++ return reply(200, { outcome: 'applied' });++};++++const tool = (id, over = {}) => ({ id, family: 'dev', name: `${id} tool`, summary: `Does ${id}`, status: 'stable', execution: 'client', api: true, hosts: [`${id}.openvibe.tools`], keywords: [id, 'online'], ...over });++let tools = [tool('yaml'), tool('uuid'), tool('ping', { status: 'unavailable' })];++const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'tools-search-'));++const make = () => createSearchIndexer({ snapshot: () => ({ tools, updated_at: '2026-09-24T00:00:00.000Z' }), searchUrl: 'http://search.test', tokens: { authHeaders: async () => ({ Authorization: 'Bearer svc' }) }, stateFile: path.join(dir, 'search-index.json'), fetchImpl, log: { warn() {} } });++++(async () => {++ let ix = make();++ let s = await ix.run();++ assert.deepStrictEqual([s.upserted, s.unchanged, s.failed], [2, 0, 0]);++ assert.deepStrictEqual([...stored.keys()].sort(), ['uuid', 'yaml'], 'an unavailable tool is not indexed');++ const y = stored.get('yaml').doc;++ assert.strictEqual(y.canonical_url, 'https://yaml.openvibe.tools');++ assert.deepStrictEqual(y.facets, { family: 'dev', execution: 'client', api: true });++ assert.strictEqual(y.publication_state, 'published');++++ calls.length = 0;++ s = await ix.run();++ assert.deepStrictEqual(calls, ['GET /api/v1/owners/tools/documents'], 'a quiet registry costs one listing');++++ tools = [tool('yaml', { summary: 'Now faster' })];++ s = await ix.run();++ assert.strictEqual(stored.get('yaml').revision, 2, 'a change goes one revision up');++ assert.strictEqual(stored.get('yaml').doc.summary, 'Now faster');++ assert.deepStrictEqual(stored.get('uuid'), { revision: 2, deleted: true }, 'a removed tool is tombstoned above its revision');++++ fs.rmSync(path.join(dir, 'search-index.json'));++ ix = make();++ s = await ix.run();++ assert.strictEqual(stored.get('yaml').revision, 3, 'lost state: re-sent above the stored revision, never refused as stale');++ assert.strictEqual(s.failed, 0);++ fs.rmSync(dir, { recursive: true, force: true });++ console.log('search index: all checks passed');++})().catch((e) => { console.error(e); process.exit(1); });@@@ -153,6 +153,8 @@ app.use('/api/', apiLimiter, guard.apiQuota);// Public (Access-Control-Allow-Origin *), cacheable (ETag), counted by the limiter above. The// gateway (openvibe.tools) answers the same routes for every tool and reads this list for status.const toolRegistry = createLocalRegistry({ specs: require('./descriptors').SPECS, statusOf: requiresStatus((p) => (p === 'heif-dec' ? codec.heif.available() : true)) });++// A signed-in person's page view of a tool goes into their tools.usage module (the launchers' recent tools).++app.use(require('../../_shared/usage').usagePages({ snapshot: toolRegistry.snapshot }));app.use(createToolsApi({ snapshot: toolRegistry.snapshot }));// ── Analytics Middleware ─────────────────────────────────────@@@ -100,6 +100,8 @@ app.use(['/api/geocode', '/api/search', '/api/weather', '/api/terrain', '/api/fo// Public (Access-Control-Allow-Origin *), cacheable (ETag), counted by the limiter above. The// gateway (openvibe.tools) answers the same routes for every tool and reads this list for status.const toolRegistry = createLocalRegistry({ specs: require('./descriptors').SPECS.filter(s => s.id === 'maps') });++// A signed-in person's page view of a tool goes into their tools.usage module (the launchers' recent tools).++app.use(require('../../_shared/usage').usagePages({ snapshot: toolRegistry.snapshot }));app.use(createToolsApi({ snapshot: toolRegistry.snapshot }));// ── Analytics Middleware ─────────────────────────────────────@@@ -96,6 +96,8 @@ app.use('/api/', guard.apiQuota);// Public (Access-Control-Allow-Origin *), cacheable (ETag), counted by the limiter above. The// gateway (openvibe.tools) answers the same routes for every tool and reads this list for status.const toolRegistry = createLocalRegistry({ specs: require('./descriptors').SPECS });++// A signed-in person's page view of a tool goes into their tools.usage module (the launchers' recent tools).++app.use(require('../../_shared/usage').usagePages({ snapshot: toolRegistry.snapshot }));app.use(createToolsApi({ snapshot: toolRegistry.snapshot }));// ── Analytics Middleware ─────────────────────────────────────@@@ -117,6 +117,8 @@ app.use('/api/', (_req, res, next) => { res.setHeader('Cache-Control', 'no-store// Public (Access-Control-Allow-Origin *), cacheable (ETag), counted by the limiter above. The// gateway (openvibe.tools) answers the same routes for every tool and reads this list for status.const toolRegistry = createLocalRegistry({ specs: require('./descriptors').SPECS, statusOf: requiresStatus((p) => (p === 'yt-dlp' ? !!which(config.ytdlpPath) : true)) });++// A signed-in person's page view of a tool goes into their tools.usage module (the launchers' recent tools).++app.use(require('../../_shared/usage').usagePages({ snapshot: toolRegistry.snapshot }));app.use(createToolsApi({ snapshot: toolRegistry.snapshot }));// ── Analytics Middleware ─────────────────────────────────────
new file mode 100644
new file mode 100644
@@ -0,0 +1,71 @@
@@ -0,0 +1,71 @@
+'use strict';
+/** tools.usage: page views and runs by signed-in people become their recent tools in the Network module, merged, deduplicated and capped. */
+const assert = require('assert');
+const { createRecorder, usagePages } = require('../usage');
+
+const USR = 'usr_01JAB2C3D4E5F6G7H8J9K0MNPQ';
+const modules = new Map(); // subject -> { data, revision }
+let conflictOnce = false;
+const calls = [];
+const fetchImpl = async (url, opts) => {
+ const u = new URL(url);
+ const reply = (status, body) => ({ status, ok: status < 300, json: async () => body });
+ if (u.pathname === '/oauth/token') { calls.push('token'); return reply(200, { access_token: 'svc', expires_in: 300 }); }
+ assert.strictEqual(opts.headers.Authorization, 'Bearer svc');
+ const m = u.pathname.match(/^\/internal\/modules\/tools\.usage\/(usr_[0-9A-Z]+)$/);
+ assert.ok(m, u.pathname);
+ calls.push(`${opts.method}`);
+ const rec = modules.get(m[1]);
+ if (opts.method === 'GET') return rec ? reply(200, { namespace: 'tools.usage', ...rec }) : reply(404, { code: 'modules.not_found' });
+ const want = opts.headers['If-Match'];
+ if (conflictOnce) { conflictOnce = false; modules.set(m[1], { data: { recent: [{ tool: 'png', at: '2026-09-24T00:00:00.000Z' }] }, revision: 7 }); return reply(412, { code: 'modules.revision_mismatch' }); }
+ if (rec && want !== `"${rec.revision}"`) return reply(412, { code: 'modules.revision_mismatch' });
+ const data = JSON.parse(opts.body).data;
+ modules.set(m[1], { data, revision: rec ? rec.revision + 1 : 1 });
+ return reply(rec ? 200 : 201, { data });
+};
+
+(async () => {
+ let t = Date.parse('2026-09-24T01:00:00Z');
+ const r = createRecorder({ env: { OV_OAUTH_CLIENT_SECRET: 'x'.repeat(40), OV_NETWORK_INTERNAL_URL: 'http://network.test' }, fetchImpl, now: () => t, flushMs: 3_600_000 });
+ assert.ok(r.record(USR, 'yaml'));
+ assert.ok(!r.record(USR, 'yaml'), 'the same tool again within 10 minutes is not news');
+ assert.ok(!r.record('42', 'yaml'), 'only a Network subject counts');
+ assert.ok(!r.record(USR, '../etc'), 'only a tool id');
+ t += 1000; r.record(USR, 'uuid');
+ await r.flush();
+ assert.deepStrictEqual(modules.get(USR).data.recent.map((e) => e.tool), ['uuid', 'yaml'], 'newest first');
+ assert.strictEqual(calls.filter((c) => c === 'token').length, 1);
+
+ t += 11 * 60 * 1000; r.record(USR, 'yaml');
+ conflictOnce = true;
+ await r.flush();
+ assert.strictEqual(r.stats().conflicts, 1);
+ assert.strictEqual(r.stats().pending, 1, 'a conflicting write stays pending');
+ await r.flush();
+ assert.deepStrictEqual(modules.get(USR).data.recent.map((e) => e.tool), ['yaml', 'png'], 'merged with what another writer put there');
+
+ for (let i = 0; i < 40; i++) { t += 1000; r.record(USR, `tool${i}`); }
+ await r.flush();
+ assert.strictEqual(modules.get(USR).data.recent.length, 30, 'capped at the namespace limit');
+ assert.strictEqual(modules.get(USR).data.recent[0].tool, 'tool39');
+
+ // The page middleware: a signed-in document request for a tool host counts; assets, APIs and unknown hosts do not.
+ const got = [];
+ const mw = usagePages({ snapshot: () => ({ tools: [{ id: 'jsonminify', hosts: ['json-minifier.openvibe.tools', 'jsonminify.openvibe.tools'] }] }), rec: { enabled: true, record: (s, id) => got.push(id) } });
+ const req = (host, p, headers = {}, user = { subject_id: USR }) => ({ method: 'GET', hostname: host, path: p, user, get: (h) => headers[h.toLowerCase()] });
+ const run = (q) => mw(q, {}, () => {});
+ run(req('json-minifier.openvibe.tools', '/', { 'sec-fetch-dest': 'document' }));
+ run(req('jsonminify.openvibe.tools', '/', { accept: 'text/html' }));
+ run(req('json-minifier.openvibe.tools', '/app.js', { 'sec-fetch-dest': 'script' }));
+ run(req('json-minifier.openvibe.tools', '/api/x', { accept: 'text/html' }));
+ run(req('openvibe.tools', '/', { 'sec-fetch-dest': 'document', 'x-ov-tool': 'evil' }));
+ run(req('json-minifier.openvibe.tools', '/', { 'sec-fetch-dest': 'document' }, null));
+ assert.deepStrictEqual(got, ['jsonminify', 'jsonminify']);
+
+ const off = createRecorder({ env: {}, fetchImpl });
+ assert.strictEqual(off.enabled, false);
+ assert.strictEqual(off.record(USR, 'yaml'), false, 'off without a service secret');
+ r.stop();
+ console.log('tools usage: all checks passed');
+})().catch((e) => { console.error(e); process.exit(1); });
@@ -248,6 +248,8 @@ function createRunApi(o) {
@@ -248,6 +248,8 @@ function createRunApi(o) {
guard.record(req, c, { reason: 'session', tool: d.id, enforced: true });
guard.record(req, c, { reason: 'session', tool: d.id, enforced: true });
throw new Refusal(401, 'tools.session_required', `${d.id} needs a browser session (open its page first), a sign-in or an API token.`);
throw new Refusal(401, 'tools.session_required', `${d.id} needs a browser session (open its page first), a sign-in or an API token.`);
}
}
+ // A person running a tool through the API counts in their recent tools (tools.usage).
+ if (c.kind === 'user' && c.claims && c.claims.subject_id) { try { require('../usage').recorder().record(c.claims.subject_id, d.id); } catch { /* best-effort */ } }
return c;
return c;
}
}
new file mode 100644
new file mode 100644
@@ -0,0 +1,139 @@
@@ -0,0 +1,139 @@
+'use strict';
+// ═══════════════════════════════════════════════════════════════
+// tools.usage (platform S9): a signed-in person's recently used tools, kept in their Network user
+// module `tools.usage` ({ recent: [{ tool, at }] }, newest first, 30 at most; openvibe-contracts
+// manifests/namespaces/tools.usage.json) for the launchers across the network.
+//
+// app.use(usagePages({ snapshot: toolRegistry.snapshot })) after guard.identify: a page view of
+// a tool by a signed-in person counts
+// recorder().record(subjectId, toolId) e.g. a run API call by a person
+//
+// Cheap by design: a person counts once per tool per 10 minutes, and a process flushes what it
+// gathered every minute: per person one read and one conditional write (If-Match), merged with what
+// the module already holds (other satellites write it too; a 412 keeps the entry for the next flush).
+// Tools' service token (network.modules.read/write for tools.usage) is fetched here without
+// dependencies, so every satellite can use it. Off unless OV_OAUTH_CLIENT_SECRET is set
+// (TOOLS_USAGE=off disables it). No dependencies: required by relative path.
+// ═══════════════════════════════════════════════════════════════
+
+const NS = 'tools.usage';
+const MAX = 30;
+const SAME_TOOL_MS = 10 * 60 * 1000;
+const FLUSH_MS = 60 * 1000;
+const MAX_PENDING = 2000;
+const USR_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
+const TOOL_RE = /^[a-z][a-z0-9-]{0,39}$/;
+
+function createRecorder({ env = process.env, fetchImpl = globalThis.fetch, now = () => Date.now(), log = console, flushMs = FLUSH_MS } = {}) {
+ const network = String(env.OV_NETWORK_INTERNAL_URL || 'http://127.0.0.1:4000').replace(/\/+$/, '');
+ const secret = env.OV_OAUTH_CLIENT_SECRET || '';
+ const enabled = !!secret && String(env.TOOLS_USAGE || '').toLowerCase() !== 'off';
+ const pending = new Map(); // subject -> Map(tool -> at ISO)
+ const seen = new Map(); // `${subject}|${tool}` -> ms
+ const stats = { recorded: 0, written: 0, conflicts: 0, failed: 0, last_error: null };
+ let token = null, tokenExp = 0, timer = null;
+
+ async function bearer() {
+ if (token && now() < tokenExp) return token;
+ const body = new URLSearchParams({ grant_type: 'client_credentials', client_id: env.OV_OAUTH_CLIENT_ID || 'tools', client_secret: secret, audience: 'openvibe.network' });
+ const r = await fetchImpl(`${network}/oauth/token`, { method: 'POST', headers: { 'Content-Type': 'application/x-www-form-urlencoded', Accept: 'application/json' }, body, signal: AbortSignal.timeout(10000) });
+ const j = await r.json().catch(() => ({}));
+ if (!r.ok || !j.access_token) throw new Error(`token: ${r.status} ${j.error || ''}`);
+ token = j.access_token; tokenExp = now() + Math.max(30, (Number(j.expires_in) || 300) - 60) * 1000;
+ return token;
+ }
+ async function call(method, subject, { body, revision } = {}) {
+ const headers = { Accept: 'application/json', Authorization: `Bearer ${await bearer()}` };
+ if (body) headers['Content-Type'] = 'application/json';
+ if (revision != null) headers['If-Match'] = `"${revision}"`;
+ const r = await fetchImpl(`${network}/internal/modules/${NS}/${subject}`, { method, headers, body: body ? JSON.stringify(body) : undefined, signal: AbortSignal.timeout(10000) });
+ if (r.status === 401) token = null;
+ return { status: r.status, body: await r.json().catch(() => ({})) };
+ }
+
+ function record(subject, tool) {
+ if (!enabled || !USR_RE.test(String(subject || '')) || !TOOL_RE.test(String(tool || ''))) return false;
+ const k = `${subject}|${tool}`, t = now();
+ if (seen.has(k) && t - seen.get(k) < SAME_TOOL_MS) return false;
+ if (seen.size > 20000) seen.clear();
+ seen.set(k, t);
+ if (!pending.has(subject) && pending.size >= MAX_PENDING) return false;
+ if (!pending.has(subject)) pending.set(subject, new Map());
+ const p = pending.get(subject);
+ p.delete(tool); p.set(tool, new Date(t).toISOString());
+ stats.recorded++;
+ start();
+ return true;
+ }
+
+ async function flushOne(subject, fresh) {
+ const cur = await call('GET', subject);
+ if (cur.status !== 200 && cur.status !== 404) throw new Error(`read: ${cur.status}`);
+ const had = cur.status === 200 && cur.body && cur.body.data && Array.isArray(cur.body.data.recent) ? cur.body.data.recent : [];
+ const add = [...fresh].reverse().map(([tool, at]) => ({ tool, at }));
+ const recent = [...add, ...had.filter((e) => e && !fresh.has(e.tool))].slice(0, MAX);
+ const w = await call('PUT', subject, { body: { data: { recent } }, revision: cur.status === 200 ? cur.body.revision : undefined });
+ if (w.status === 412 || w.status === 409) { stats.conflicts++; return false; }
+ if (w.status >= 300) throw new Error(`write: ${w.status} ${(w.body && w.body.code) || ''}`);
+ stats.written++;
+ return true;
+ }
+
+ async function flush() {
+ const batch = [...pending];
+ pending.clear();
+ for (const [subject, fresh] of batch) {
+ let ok = false;
+ try { ok = await flushOne(subject, fresh); } catch (err) { stats.failed++; stats.last_error = err.message; }
+ if (!ok && pending.size < MAX_PENDING) {
+ const back = pending.get(subject) || new Map();
+ for (const [tool, at] of fresh) if (!back.has(tool)) back.set(tool, at);
+ pending.set(subject, back);
+ }
+ }
+ if (!pending.size) stop();
+ }
+ function start() { if (timer || !enabled) return; timer = setInterval(() => { flush().catch(() => {}); }, flushMs); if (timer.unref) timer.unref(); }
+ function stop() { if (timer) clearInterval(timer); timer = null; }
+
+ return { enabled, record, flush, stop, stats: () => ({ enabled, pending: pending.size, ...stats }) };
+}
+
+let _recorder = null;
+function recorder() { if (!_recorder) _recorder = createRecorder(); return _recorder; }
+
+/**
+ * Middleware: a GET of a tool's page by a signed-in person (req.user.subject_id from guard.identify)
+ * records that tool. The host names the tool (the registry's hosts, or the X-OV-Tool the gateway sends).
+ */
+function usagePages({ snapshot, rec = null }) {
+ let map = null, mapFor = null;
+ const hostMap = () => {
+ const s = snapshot();
+ if (map && mapFor === s) return map;
+ map = new Map();
+ for (const t of s.tools || []) for (const h of t.hosts || []) map.set(String(h).toLowerCase(), t.id);
+ map.ids = new Set((s.tools || []).map((t) => t.id));
+ mapFor = s;
+ return map;
+ };
+ return function toolUsage(req, _res, next) {
+ try {
+ const sid = req.user && req.user.subject_id;
+ const r = rec || recorder();
+ if (sid && r.enabled && req.method === 'GET' && !req.path.startsWith('/api/') && !/\.[a-z0-9]{1,8}$/i.test(req.path)) {
+ const dest = req.get('sec-fetch-dest');
+ if (dest ? dest === 'document' : /text\/html/.test(req.get('accept') || '')) {
+ const host = String(req.hostname || '').toLowerCase();
+ const m = hostMap();
+ const named = String(req.get('x-ov-tool') || '');
+ const tool = m.get(host) || (m.ids.has(named) ? named : null);
+ if (tool) r.record(sid, tool);
+ }
+ }
+ } catch { /* usage is best-effort */ }
+ next();
+ };
+}
+
+module.exports = { createRecorder, recorder, usagePages, NS };
@@ -130,6 +130,8 @@ app.use('/api/', apiLimiter, guard.apiQuota);
@@ -130,6 +130,8 @@ app.use('/api/', apiLimiter, guard.apiQuota);
// Public (Access-Control-Allow-Origin *), cacheable (ETag), counted by the limiter above. The
// Public (Access-Control-Allow-Origin *), cacheable (ETag), counted by the limiter above. The
// gateway (openvibe.tools) answers the same routes for every tool and reads this list for status.
// gateway (openvibe.tools) answers the same routes for every tool and reads this list for status.
const toolRegistry = createLocalRegistry({ specs: require('./descriptors').SPECS, statusOf: requiresStatus((p) => (p === 'ffmpeg' ? !!which(process.env.FFMPEG_PATH || 'ffmpeg') : true)) });
const toolRegistry = createLocalRegistry({ specs: require('./descriptors').SPECS, statusOf: requiresStatus((p) => (p === 'ffmpeg' ? !!which(process.env.FFMPEG_PATH || 'ffmpeg') : true)) });
+// A signed-in person's page view of a tool goes into their tools.usage module (the launchers' recent tools).
+app.use(require('../../_shared/usage').usagePages({ snapshot: toolRegistry.snapshot }));
app.use(createToolsApi({ snapshot: toolRegistry.snapshot }));
app.use(createToolsApi({ snapshot: toolRegistry.snapshot }));
// ── Analytics Middleware ─────────────────────────────────────
// ── Analytics Middleware ─────────────────────────────────────
@@ -147,6 +147,8 @@ app.use('/api/', apiLimiter, guard.apiQuota);
@@ -147,6 +147,8 @@ app.use('/api/', apiLimiter, guard.apiQuota);
// gateway (openvibe.tools) answers the same routes for every tool and reads this list for status.
// gateway (openvibe.tools) answers the same routes for every tool and reads this list for status.
const pdfPrograms = require('./tools/pdf');
const pdfPrograms = require('./tools/pdf');
const toolRegistry = createLocalRegistry({ specs: require('./descriptors').SPECS, statusOf: requiresStatus((p) => (pdfPrograms[p] ? pdfPrograms[p].available() : true)) });
const toolRegistry = createLocalRegistry({ specs: require('./descriptors').SPECS, statusOf: requiresStatus((p) => (pdfPrograms[p] ? pdfPrograms[p].available() : true)) });
+// A signed-in person's page view of a tool goes into their tools.usage module (the launchers' recent tools).
+app.use(require('../../_shared/usage').usagePages({ snapshot: toolRegistry.snapshot }));
app.use(createToolsApi({ snapshot: toolRegistry.snapshot }));
app.use(createToolsApi({ snapshot: toolRegistry.snapshot }));
// ── Analytics Middleware ─────────────────────────────────────
// ── Analytics Middleware ─────────────────────────────────────
@@ -181,6 +181,8 @@ app.use(exceptRegistry(cors({
@@ -181,6 +181,8 @@ app.use(exceptRegistry(cors({
// ── Who is asking, then rate limits (sign-in first, so a signed-in tier applies) ──
// ── Who is asking, then rate limits (sign-in first, so a signed-in tier applies) ──
app.use(guard.identify);
app.use(guard.identify);
+// A signed-in person's page view of a tool goes into their tools.usage module (the launchers' recent tools).
+app.use(require('../../_shared/usage').usagePages({ snapshot: toolRegistry.snapshot }));
app.use('/api/', guard.legacyLimiter(rateLimit, { windowMs: 60_000, anonymous: 120, signedIn: 240, message: 'Too many requests. Please try again later.' }), guard.apiQuota);
app.use('/api/', guard.legacyLimiter(rateLimit, { windowMs: 60_000, anonymous: 120, signedIn: 240, message: 'Too many requests. Please try again later.' }), guard.apiQuota);
app.use('/auth/', rateLimit({ windowMs: 15 * 60_000, max: 60, keyGenerator: (req) => guard.caller(req).ipKey }));
app.use('/auth/', rateLimit({ windowMs: 15 * 60_000, max: 60, keyGenerator: (req) => guard.caller(req).ipKey }));
@@ -253,7 +255,7 @@ async function requireAuth(req, res, next) {
@@ -253,7 +255,7 @@ async function requireAuth(req, res, next) {
// ── Basic API ────────────────────────────────────────────────
// ── Basic API ────────────────────────────────────────────────
app.get('/api/health', (_req, res) => {
app.get('/api/health', (_req, res) => {
- res.json({ status: 'ok', service: 'openvibe-tools-gateway', version: '2.0.0', registry: registry.services.status(), run: runApi.stats(), jobs: jobsFacade.stats() });
+ res.json({ status: 'ok', service: 'openvibe-tools-gateway', version: '2.0.0', registry: registry.services.status(), run: runApi.stats(), jobs: jobsFacade.stats(), search_index: searchIndex.indexer ? searchIndex.indexer.stats() : { enabled: false, reason: searchIndex.reason } });
});
});
app.get('/api/brand', (_req, res) => res.json(BRAND));
app.get('/api/brand', (_req, res) => res.json(BRAND));
@@ -433,6 +435,10 @@ app.get('*', (req, res) => {
@@ -433,6 +435,10 @@ app.get('*', (req, res) => {
// What each satellite says about its own tools (a program missing on the host → unavailable here too).
// What each satellite says about its own tools (a program missing on the host → unavailable here too).
toolRegistry.startSatellitePolling(SATELLITES);
toolRegistry.startSatellitePolling(SATELLITES);
+// Every tool as an OpenVibe.Search document (owner tools), kept in step with the registry (S9).
+const searchIndex = require('./search-index').searchIndexerFromEnv({ snapshot: toolRegistry.snapshot, dataDir: require('path').resolve(__dirname, '..', process.env.DATA_DIR || 'data') });
+if (searchIndex.indexer) searchIndex.indexer.start(); else console.log(`[SearchIndex] off: ${searchIndex.reason}`);
+
app.listen(config.port, config.host, () => {
app.listen(config.port, config.host, () => {
console.log(`\n╔════════════════════════════════════════════╗`);
console.log(`\n╔════════════════════════════════════════════╗`);
console.log(`║ 🔧 openvibe.tools — Tools Gateway ║`);
console.log(`║ 🔧 openvibe.tools — Tools Gateway ║`);
new file mode 100644
new file mode 100644
@@ -0,0 +1,114 @@
@@ -0,0 +1,114 @@
+'use strict';
+/**
+ * Tools → OpenVibe.Search (platform S9): every tool in the registry is one Search document
+ * (search.index-document@1, owner tools, type tool, id = the tool id), written through Search's owner
+ * API with the tools service token (search.document.write on openvibe.search). A tool that leaves the
+ * registry, or becomes unavailable on this host, is tombstoned.
+ *
+ * Runs a minute after boot and every 6 hours. Search's own inventory (GET /api/v1/owners/tools/documents)
+ * gives the stored revisions; a document is re-sent, one revision up, only when its content hash
+ * differs from the last one sent (data/search-index.json), so a quiet registry costs one listing.
+ *
+ * Off unless SEARCH_URL and OV_OAUTH_CLIENT_SECRET are set (TOOLS_SEARCH_INDEX=off disables it).
+ */
+const fs = require('fs');
+const path = require('path');
+const crypto = require('crypto');
+
+const OWNER = 'tools';
+const TYPE = 'tool';
+const EVERY_MS = 6 * 60 * 60 * 1000;
+
+/** A descriptor → the document body (identity and revision added by the caller). */
+function documentFor(t, updatedAt) {
+ const host = Array.isArray(t.hosts) && t.hosts[0];
+ return {
+ visibility: 'public',
+ canonical_url: host ? `https://${host}` : t.docs || `https://openvibe.tools/tool/${t.id}`,
+ title: String(t.name || t.id).slice(0, 500),
+ summary: String(t.summary || '').slice(0, 4000),
+ body: [t.name, t.summary, ...(Array.isArray(t.keywords) ? t.keywords : [])].filter(Boolean).join('\n').slice(0, 48000),
+ facets: { family: String(t.family || 'other'), execution: String(t.execution || 'client'), api: Boolean(t.api) },
+ language: 'en',
+ authorship: 'human',
+ publication_state: 'published',
+ ...(updatedAt ? { updated_at: updatedAt } : {}),
+ indexability: { decision: 'index', reasons: [] },
+ };
+}
+const hashOf = (doc) => crypto.createHash('sha256').update(JSON.stringify(doc)).digest('hex');
+
+function createSearchIndexer({ snapshot, searchUrl, tokens, stateFile, fetchImpl = globalThis.fetch, log = console, now = () => Date.now() }) {
+ const base = String(searchUrl).replace(/\/+$/, '');
+ const stats = { runs: 0, upserted: 0, deleted: 0, unchanged: 0, failed: 0, last_run_at: null, last_error: null };
+ let state = {};
+ try { state = JSON.parse(fs.readFileSync(stateFile, 'utf8')).docs || {}; } catch { state = {}; }
+ const save = () => {
+ try { fs.mkdirSync(path.dirname(stateFile), { recursive: true }); fs.writeFileSync(`${stateFile}.tmp`, JSON.stringify({ docs: state })); fs.renameSync(`${stateFile}.tmp`, stateFile); } catch (err) { log.warn('[SearchIndex] state not saved:', err.message); }
+ };
+ async function call(method, p, body) {
+ const res = await fetchImpl(`${base}${p}`, { method, headers: { Accept: 'application/json', ...(body ? { 'Content-Type': 'application/json' } : {}), ...(await tokens.authHeaders()) }, body: body ? JSON.stringify(body) : undefined, signal: AbortSignal.timeout(15000) });
+ const json = await res.json().catch(() => ({}));
+ if (res.status === 401 && tokens.invalidate) tokens.invalidate();
+ return { status: res.status, ok: res.ok, body: json };
+ }
+ async function remoteDocs() {
+ const out = new Map();
+ let after = 0;
+ for (let i = 0; i < 50; i++) {
+ const r = await call('GET', `/api/v1/owners/${OWNER}/documents?type=${TYPE}&limit=1000&after=${after}`);
+ if (!r.ok) throw new Error(`listing: ${r.status} ${r.body.code || ''}`);
+ for (const d of r.body.documents || []) out.set(d.id, d);
+ if (!r.body.next_after) break;
+ after = r.body.next_after;
+ }
+ return out;
+ }
+
+ async function run() {
+ stats.runs++; stats.last_run_at = new Date(now()).toISOString();
+ const snap = snapshot();
+ const tools = (snap.tools || []).filter((t) => t.status !== 'unavailable' && /^[a-z][a-z0-9-]{0,39}$/.test(t.id));
+ const remote = await remoteDocs();
+ const keep = new Set();
+ for (const t of tools) {
+ keep.add(t.id);
+ const doc = documentFor(t, snap.updated_at || null);
+ const h = hashOf(doc);
+ const r = remote.get(t.id);
+ if (r && !r.deleted && state[t.id] && state[t.id].hash === h) { stats.unchanged++; continue; }
+ const revision = r ? r.revision + 1 : 1;
+ const res = await call('PUT', `/api/v1/documents/${OWNER}/${TYPE}/${t.id}`, { ...doc, revision });
+ if (res.ok) { state[t.id] = { hash: h, revision }; stats.upserted++; } else { stats.failed++; stats.last_error = `${t.id}: ${res.status} ${res.body.code || ''}`; }
+ }
+ for (const [id, r] of remote) {
+ if (keep.has(id) || r.deleted) continue;
+ const res = await call('DELETE', `/api/v1/documents/${OWNER}/${TYPE}/${encodeURIComponent(id)}?revision=${r.revision + 1}`);
+ if (res.ok) { delete state[id]; stats.deleted++; } else { stats.failed++; stats.last_error = `${id}: ${res.status} ${res.body.code || ''}`; }
+ }
+ save();
+ return { ...stats };
+ }
+
+ let timer = null;
+ function start({ firstDelayMs = 60_000 } = {}) {
+ const tick = () => run().catch((err) => { stats.failed++; stats.last_error = err.message; log.warn('[SearchIndex] run failed:', err.message); });
+ const first = setTimeout(tick, firstDelayMs); if (first.unref) first.unref();
+ timer = setInterval(tick, EVERY_MS); if (timer.unref) timer.unref();
+ }
+ function stop() { if (timer) clearInterval(timer); timer = null; }
+ return { run, start, stop, stats: () => ({ ...stats }) };
+}
+
+/** The gateway's indexer from its environment, or null (with why) when it is off. */
+function searchIndexerFromEnv({ snapshot, dataDir, env = process.env, contracts = require('openvibe-contracts'), log = console }) {
+ const url = String(env.SEARCH_URL || '').trim();
+ if (!url) return { indexer: null, reason: 'SEARCH_URL is not set' };
+ if (String(env.TOOLS_SEARCH_INDEX || '').toLowerCase() === 'off') return { indexer: null, reason: 'TOOLS_SEARCH_INDEX=off' };
+ if (!env.OV_OAUTH_CLIENT_SECRET) return { indexer: null, reason: 'OV_OAUTH_CLIENT_SECRET is not set' };
+ const network = String(env.OV_NETWORK_INTERNAL_URL || 'http://127.0.0.1:4000').replace(/\/+$/, '');
+ const tokens = contracts.serviceAuth.createTokenClient({ tokenUrl: `${network}/oauth/token`, clientId: env.OV_OAUTH_CLIENT_ID || 'tools', clientSecret: env.OV_OAUTH_CLIENT_SECRET, audience: 'openvibe.search', scope: 'search.document.write' });
+ return { indexer: createSearchIndexer({ snapshot, searchUrl: url, tokens, stateFile: path.join(dataDir, 'search-index.json'), log }) };
+}
+
+module.exports = { createSearchIndexer, searchIndexerFromEnv, documentFor };
new file mode 100644
new file mode 100644
@@ -0,0 +1,65 @@
@@ -0,0 +1,65 @@
+'use strict';
+/** Tools → Search: one search.index-document@1 per available tool, re-sent only when it changes, tombstoned when it leaves. */
+const assert = require('assert');
+const fs = require('fs');
+const os = require('os');
+const path = require('path');
+const contracts = require('openvibe-contracts');
+const { createSearchIndexer } = require('../server/search-index');
+
+const stored = new Map(); // id -> { revision, deleted, doc }
+const calls = [];
+const fetchImpl = async (url, opts) => {
+ const u = new URL(url);
+ calls.push(`${opts.method} ${u.pathname}`);
+ assert.strictEqual(opts.headers.Authorization, 'Bearer svc');
+ const reply = (status, body) => ({ status, ok: status < 300, json: async () => body });
+ if (opts.method === 'GET') return reply(200, { owner: 'tools', documents: [...stored].map(([id, d]) => ({ type: 'tool', id, revision: d.revision, deleted: d.deleted })), next_after: null });
+ const id = u.pathname.split('/').pop();
+ const prev = stored.get(id);
+ if (opts.method === 'PUT') {
+ const doc = { ...JSON.parse(opts.body), owner: 'tools', type: 'tool', id };
+ const v = contracts.validate('search.index-document@1', doc);
+ assert.ok(v.valid, JSON.stringify(v.errors));
+ if (prev && doc.revision <= prev.revision) return reply(409, { code: 'search.stale_revision' });
+ stored.set(id, { revision: doc.revision, deleted: false, doc });
+ return reply(200, { outcome: 'applied' });
+ }
+ const revision = Number(u.searchParams.get('revision'));
+ stored.set(id, { revision, deleted: true });
+ return reply(200, { outcome: 'applied' });
+};
+
+const tool = (id, over = {}) => ({ id, family: 'dev', name: `${id} tool`, summary: `Does ${id}`, status: 'stable', execution: 'client', api: true, hosts: [`${id}.openvibe.tools`], keywords: [id, 'online'], ...over });
+let tools = [tool('yaml'), tool('uuid'), tool('ping', { status: 'unavailable' })];
+const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'tools-search-'));
+const make = () => createSearchIndexer({ snapshot: () => ({ tools, updated_at: '2026-09-24T00:00:00.000Z' }), searchUrl: 'http://search.test', tokens: { authHeaders: async () => ({ Authorization: 'Bearer svc' }) }, stateFile: path.join(dir, 'search-index.json'), fetchImpl, log: { warn() {} } });
+
+(async () => {
+ let ix = make();
+ let s = await ix.run();
+ assert.deepStrictEqual([s.upserted, s.unchanged, s.failed], [2, 0, 0]);
+ assert.deepStrictEqual([...stored.keys()].sort(), ['uuid', 'yaml'], 'an unavailable tool is not indexed');
+ const y = stored.get('yaml').doc;
+ assert.strictEqual(y.canonical_url, 'https://yaml.openvibe.tools');
+ assert.deepStrictEqual(y.facets, { family: 'dev', execution: 'client', api: true });
+ assert.strictEqual(y.publication_state, 'published');
+
+ calls.length = 0;
+ s = await ix.run();
+ assert.deepStrictEqual(calls, ['GET /api/v1/owners/tools/documents'], 'a quiet registry costs one listing');
+
+ tools = [tool('yaml', { summary: 'Now faster' })];
+ s = await ix.run();
+ assert.strictEqual(stored.get('yaml').revision, 2, 'a change goes one revision up');
+ assert.strictEqual(stored.get('yaml').doc.summary, 'Now faster');
+ assert.deepStrictEqual(stored.get('uuid'), { revision: 2, deleted: true }, 'a removed tool is tombstoned above its revision');
+
+ fs.rmSync(path.join(dir, 'search-index.json'));
+ ix = make();
+ s = await ix.run();
+ assert.strictEqual(stored.get('yaml').revision, 3, 'lost state: re-sent above the stored revision, never refused as stale');
+ assert.strictEqual(s.failed, 0);
+ fs.rmSync(dir, { recursive: true, force: true });
+ console.log('search index: all checks passed');
+})().catch((e) => { console.error(e); process.exit(1); });
@@ -153,6 +153,8 @@ app.use('/api/', apiLimiter, guard.apiQuota);
@@ -153,6 +153,8 @@ app.use('/api/', apiLimiter, guard.apiQuota);
// Public (Access-Control-Allow-Origin *), cacheable (ETag), counted by the limiter above. The
// Public (Access-Control-Allow-Origin *), cacheable (ETag), counted by the limiter above. The
// gateway (openvibe.tools) answers the same routes for every tool and reads this list for status.
// gateway (openvibe.tools) answers the same routes for every tool and reads this list for status.
const toolRegistry = createLocalRegistry({ specs: require('./descriptors').SPECS, statusOf: requiresStatus((p) => (p === 'heif-dec' ? codec.heif.available() : true)) });
const toolRegistry = createLocalRegistry({ specs: require('./descriptors').SPECS, statusOf: requiresStatus((p) => (p === 'heif-dec' ? codec.heif.available() : true)) });
+// A signed-in person's page view of a tool goes into their tools.usage module (the launchers' recent tools).
+app.use(require('../../_shared/usage').usagePages({ snapshot: toolRegistry.snapshot }));
app.use(createToolsApi({ snapshot: toolRegistry.snapshot }));
app.use(createToolsApi({ snapshot: toolRegistry.snapshot }));
// ── Analytics Middleware ─────────────────────────────────────
// ── Analytics Middleware ─────────────────────────────────────
@@ -100,6 +100,8 @@ app.use(['/api/geocode', '/api/search', '/api/weather', '/api/terrain', '/api/fo
@@ -100,6 +100,8 @@ app.use(['/api/geocode', '/api/search', '/api/weather', '/api/terrain', '/api/fo
// Public (Access-Control-Allow-Origin *), cacheable (ETag), counted by the limiter above. The
// Public (Access-Control-Allow-Origin *), cacheable (ETag), counted by the limiter above. The
// gateway (openvibe.tools) answers the same routes for every tool and reads this list for status.
// gateway (openvibe.tools) answers the same routes for every tool and reads this list for status.
const toolRegistry = createLocalRegistry({ specs: require('./descriptors').SPECS.filter(s => s.id === 'maps') });
const toolRegistry = createLocalRegistry({ specs: require('./descriptors').SPECS.filter(s => s.id === 'maps') });
+// A signed-in person's page view of a tool goes into their tools.usage module (the launchers' recent tools).
+app.use(require('../../_shared/usage').usagePages({ snapshot: toolRegistry.snapshot }));
app.use(createToolsApi({ snapshot: toolRegistry.snapshot }));
app.use(createToolsApi({ snapshot: toolRegistry.snapshot }));
// ── Analytics Middleware ─────────────────────────────────────
// ── Analytics Middleware ─────────────────────────────────────
@@ -96,6 +96,8 @@ app.use('/api/', guard.apiQuota);
@@ -96,6 +96,8 @@ app.use('/api/', guard.apiQuota);
// Public (Access-Control-Allow-Origin *), cacheable (ETag), counted by the limiter above. The
// Public (Access-Control-Allow-Origin *), cacheable (ETag), counted by the limiter above. The
// gateway (openvibe.tools) answers the same routes for every tool and reads this list for status.
// gateway (openvibe.tools) answers the same routes for every tool and reads this list for status.
const toolRegistry = createLocalRegistry({ specs: require('./descriptors').SPECS });
const toolRegistry = createLocalRegistry({ specs: require('./descriptors').SPECS });
+// A signed-in person's page view of a tool goes into their tools.usage module (the launchers' recent tools).
+app.use(require('../../_shared/usage').usagePages({ snapshot: toolRegistry.snapshot }));
app.use(createToolsApi({ snapshot: toolRegistry.snapshot }));
app.use(createToolsApi({ snapshot: toolRegistry.snapshot }));
// ── Analytics Middleware ─────────────────────────────────────
// ── Analytics Middleware ─────────────────────────────────────
@@ -117,6 +117,8 @@ app.use('/api/', (_req, res, next) => { res.setHeader('Cache-Control', 'no-store
@@ -117,6 +117,8 @@ app.use('/api/', (_req, res, next) => { res.setHeader('Cache-Control', 'no-store
// Public (Access-Control-Allow-Origin *), cacheable (ETag), counted by the limiter above. The
// Public (Access-Control-Allow-Origin *), cacheable (ETag), counted by the limiter above. The
// gateway (openvibe.tools) answers the same routes for every tool and reads this list for status.
// gateway (openvibe.tools) answers the same routes for every tool and reads this list for status.
const toolRegistry = createLocalRegistry({ specs: require('./descriptors').SPECS, statusOf: requiresStatus((p) => (p === 'yt-dlp' ? !!which(config.ytdlpPath) : true)) });
const toolRegistry = createLocalRegistry({ specs: require('./descriptors').SPECS, statusOf: requiresStatus((p) => (p === 'yt-dlp' ? !!which(config.ytdlpPath) : true)) });
+// A signed-in person's page view of a tool goes into their tools.usage module (the launchers' recent tools).
+app.use(require('../../_shared/usage').usagePages({ snapshot: toolRegistry.snapshot }));
app.use(createToolsApi({ snapshot: toolRegistry.snapshot }));
app.use(createToolsApi({ snapshot: toolRegistry.snapshot }));
// ── Analytics Middleware ─────────────────────────────────────
// ── Analytics Middleware ─────────────────────────────────────