openvibers/openvibe.run Commits
account-data: a person's sessions and jobs, with the guard on every person column (#115)
Changes
@@@ -99,7 +99,7 @@ placed from the Fabric's offers and actually runs — the poller in `server/jobs(`bot.job.dispatch`, audience `openvibe.bot`, minted from `OV_OAUTH_CLIENT_ID`/`OV_OAUTH_CLIENT_SECRET` against`OV_NETWORK_INTERNAL_URL` and cached until 60 s before expiry) and mirrors its answer. Unset the client secret andthe bridge is off: `/api/ready` skips the `bot` check and every job stays `queued`.--- `openvibe-contracts` v0.106.0, `openvibe-sdk` v0.67.0 and `openvibe-shared` v4.48.0 (package.json).++- `openvibe-contracts` v0.106.0, `openvibe-sdk` v0.74.0 and `openvibe-shared` v4.48.0 (package.json).## Capabilities and routes@@@ -253,6 +253,6 @@ until Run has a product surface; the vhost below is installed then.<!-- versions:start -->- openvibe-contracts: v0.223.0--- openvibe-sdk: v0.67.0++- openvibe-sdk: v0.74.0- openvibe-shared: v4.48.0<!-- versions:end -->@@@ -8,11 +8,11 @@"port": 4920,"domain": "openvibe.run","contracts": "openvibe-contracts v0.223.0",-- "sdk": "openvibe-sdk v0.67.0",++ "sdk": "openvibe-sdk v0.74.0","shared": "openvibe-shared v4.48.0","packages": ["openvibe-contracts v0.223.0",-- "openvibe-sdk v0.67.0",++ "openvibe-sdk v0.74.0","openvibe-shared v4.48.0"],"capabilities": [nnew file mode 100644@@@ -0,0 +1,13 @@++-- phase: expand++-- OpenVibe.Run: the receipts of the ADR-033 account export and deletion deliveries this service applied++-- (openvibe-sdk/account-data's ACCOUNT_DATA_SCHEMA), so a redelivered export or deletion changes nothing.++-- The row holds an export_id or deletion_id, the subject and the counts sent to Network; it is not a secret.++++CREATE TABLE IF NOT EXISTS account_data_events (++ id TEXT PRIMARY KEY,++ kind TEXT NOT NULL,++ subject TEXT NOT NULL,++ outcome JSONB,++ applied_at TIMESTAMPTZ NOT NULL DEFAULT now(),++ sent_at TIMESTAMPTZ++);@@@ -16,7 +16,7 @@"helmet": "^8.1.0","iovalkey": "^0.4.0","openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.223.0",-- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.67.0",++ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.74.0","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.48.0","pg": "^8.23.0"},@@@ -1163,9 +1163,9 @@}},"node_modules/openvibe-sdk": {-- "version": "0.67.0",-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.67.0",-- "integrity": "sha512-s7NzjaNcowVcOIiup2z0ze+xsq/qnr++3pHSI8JzirqJPQ5JsBSUF5RkjOHyfk9bRKlmhY0fsNS6x4dsYylyqw==",++ "version": "0.74.0",++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.74.0",++ "integrity": "sha512-DO9KVPnUriY/I4lX6ZEJtXaUvRDbJyqnqJxvlpl+zkNGFILpmuBXVJogNOCfK071Qci1gg7C4Iw1Xrseklcajg==","license": "MIT","bin": {"openvibe-download": "bin/openvibe-download.js"@@@ -27,7 +27,7 @@"helmet": "^8.1.0","iovalkey": "^0.4.0","openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.223.0",-- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.67.0",++ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.74.0","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.48.0","pg": "^8.23.0"},nnew file mode 100644@@@ -0,0 +1,43 @@++'use strict';++/**++ * Account export and deletion → Run (ADR-033; openvibe-sdk/account-data).++ *++ * Run keeps two things about a person:++ *++ * - their open browser sessions (run_sessions, server/sessions/), keyed `owner = user:usr_…`. The export carries each++ * session's settings and timing, never its ticket or stream secret; deletion removes them.++ * - the jobs they submitted (run_jobs.requester = `user:usr_…`). The export carries the job, its result and its++ * timings; the row itself stays, because it is the project's record of the work and of what it cost, but the++ * requester becomes `user:deleted`, so the job no longer leads to a person.++ *++ * Telemetry samples that name a person are covered by the SDK itself (telemetry.json, then subject NULL).++ */++const { createAccountData, TOPICS } = require('openvibe-sdk/account-data');++++const TABLES = [++ {++ table: 'run_sessions', subject: 'owner', value: (usr) => `user:${usr}`, file: 'sessions.json', order: 'created_at',++ columns: ['id', 'via', 'kind', 'state', 'width', 'height', 'ttl_s', 'idle_s', 'machine_asked', 'requirements', 'machine', 'created_at', 'started_at', 'last_used_at', 'ended_at', 'end_reason', 'end_detail', 'metered_minutes', 'updated_at'],++ },++ // The jobs are handled by the service itself (the requester is tombstoned, not deleted), so they are not in the++ // table map; they are listed as covered in the test's guard.++];++++/** The jobs the person submitted: their own work, with its result. */++async function extraExport(db, subject) {++ const jobs = await db.many(`SELECT id, project_id, class, state, provider, region, job, result, ttl_ms, created_at, updated_at, started_ms, finished_ms, wall_ms, usage_read, error_code++ FROM run_jobs WHERE requester = $1 ORDER BY created_at DESC LIMIT 5000`, [`user:${subject}`]);++ return jobs.length ? [{ name: 'jobs.json', content: jobs }] : [];++}++++/** The requester of the person's jobs becomes a tombstone: the job stays, the person does not lead to it. */++async function extraErase(t, subjects, counts) {++ const forms = subjects.map((s) => `user:${s}`);++ counts.add(counts.retained, 'run_jobs_tombstoned', await t.exec(`UPDATE run_jobs SET requester = 'user:deleted' WHERE requester = ANY($1::text[])`, [forms]));++}++++function create({ db, log = console } = {}) {++ return createAccountData({ db, service: 'run', tables: TABLES, extraExport, extraErase, log });++}++++module.exports = { create, TABLES, TOPICS, extraExport, extraErase };@@@ -74,6 +74,9 @@ function loadConfig(env = process.env) {events: {url: trim(env.EVENTS_URL || ''),intervalMs: int(env.EVENTS_RELAY_INTERVAL_MS, 2000),++ // The account-data consumer (POST /internal/events): the signing secret Network's deliveries use. Unset:++ // the route is off and Network's account export/deletion does not wait for Run.++ secret: String(env.RUN_EVENTS_SECRET || '').trim(),},// The background jobs (the events relay and the dispatcher poller); RUN_JOBS=off runs the API alone,@@@ -24,6 +24,8 @@ const { createValkey } = require('openvibe-sdk/valkey');const { gracefulStop, telemetry } = require('openvibe-sdk/service');const { createPgTelemetrySink, startTelemetryRetention } = require('openvibe-sdk/telemetry');const { createRegistry } = require('openvibe-shared/metrics');++const accountDataLib = require('./account-data');++const { createNetworkSender, startSubscriptions } = require('openvibe-sdk/account-data');async function main() {const config = loadConfig();@@@ -33,6 +35,8 @@ async function main() {if (m.held.length) console.warn(`[Run] migrations held: ${m.held.map((h) => `${h.id} (${h.reason})`).join('; ')}`);const valkey = createValkey({ url: config.valkey.url, prefix: config.valkey.prefix });++ const accountData = accountDataLib.create({ db, log: console });++ const accountSend = config.oauth.clientSecret ? createNetworkSender({ networkInternalUrl: config.network.internalUrl, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret }) : null;const app = createApp({ config, db, valkey, registry });const { keys, outbox, stream, tickets, dispatch, indexnow, sessions } = app.locals;keys.start();@@@ -48,8 +52,16 @@ async function main() {sessions.start();}++ app.post('/internal/events', accountData.consumer({ secrets: [config.events.secret].filter((s) => s.length >= 32), send: accountSend }));++ let accountSubscription = null;const server = app.listen(config.port, config.host, () => {console.log(`[Run] ${config.nodeEnv} on http://${config.host}:${config.port} → ${config.baseUrl} (store ${db.store}; valkey ${valkey ? 'on' : 'off: this process only'})`);++ if (config.events.secret.length >= 32 && config.events.url) {++ accountSubscription = startSubscriptions({++ eventsUrl: config.events.url, endpoint: `http://127.0.0.1:${config.port}/internal/events`, secret: config.events.secret,++ networkInternalUrl: config.network.internalUrl, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret,++ });++ }console.log(`[Run] jobs ${config.jobs.enabled ? 'on' : 'off (RUN_JOBS=off)'}; events relay ${outbox.enabled ? `→ ${config.events.url}` : 'off (run.job.* events wait in the outbox)'}; stream tickets ${ticketKey.enabled ? `on (${config.stream.keyId}${ticketKey.ephemeral ? ', ephemeral key' : ''})` : 'off (no stream key)'}; dispatcher ${dispatch.enabled ? `on (poll ${config.dispatch.pollMs} ms): ${dispatch.note}` : `off: ${dispatch.note}`}; ${dispatch.status().usage_relay.note}; ${sessions.status().note}`);});server.keepAliveTimeout = 65_000;@@@ -77,6 +89,7 @@ async function main() {() => sessions.stop(),() => stream.close(),() => outbox.stop(),++ () => { if (accountSubscription) accountSubscription.stop(); },],close: [async () => { if (telemetryRetention) await telemetryRetention.stop(); await telemetry.stop(); },nnew file mode 100644@@@ -0,0 +1,46 @@++'use strict';++/**++ * ADR-033: Run's part of an account export and deletion. A person's open browser sessions are exported (their++ * settings and timing, never a ticket) and deleted; someone else's stay. The guard fails when a new table with a++ * person column is neither covered nor exempted with a reason.++ */++const assert = require('assert');++const { uncovered } = require('openvibe-sdk/account-data');++const { testDb } = require('./helpers/db');++const { TABLES, create } = require('../server/account-data');++++const A = 'usr_01JZ0000000000000000000AAA';++const B = 'usr_01JZ0000000000000000000BBB';++++(async () => {++ const d = await testDb();++ try {++ const accountData = create({ db: d.db });++ const missing = await uncovered(d.db, { tables: [...TABLES, 'run_jobs'], exempt: { account_data_events: 'the ledger of requests itself: which export and deletion Run applied' } });++ assert.deepStrictEqual(missing, [], `a table with a person column is neither covered nor exempted: ${missing.join(', ')}`);++++ for (const [subject, name, jobId] of [[A, 'ada', 'job_01JZ0000000000000000000ADA'], [B, 'bo', 'job_01JZ0000000000000000000BQB']]) {++ await d.db.exec(`INSERT INTO run_sessions (id, owner, kind, state, width, height, ttl_s, idle_s, request_digest, created_at, updated_at)++ VALUES ($1, $2, 'browser', 'ended', 1280, 720, 900, 60, 'digest', 1, 1)`, [`ses_${name}`, `user:${subject}`]);++ await d.db.exec(`INSERT INTO run_jobs (id, project_id, requester, idempotency_key, class, job, ttl_ms, state, created_at, updated_at)++ VALUES ($1, 'proj_1', $2, $3, 'browser', '{}', 1000, 'succeeded', 1, 1)`, [jobId, `user:${subject}`, `idem-${name}`]);++ }++ const part = await accountData.exportPart(A);++ const file = part.files.find((f) => f.name === 'sessions.json');++ assert.ok(file, 'sessions.json is in the part');++ assert.strictEqual(file.content.length, 1);++ assert.strictEqual(file.content[0].id, 'ses_ada');++ const jobs = part.files.find((f) => f.name === 'jobs.json');++ assert.ok(jobs, 'jobs.json is in the part');++ assert.deepStrictEqual(jobs.content.map((j) => j.id), ['job_01JZ0000000000000000000ADA'], 'only their own job');++++ const counts = await accountData.erase([A]);++ assert.strictEqual(counts.erased.run_sessions, 1);++ const left = (await d.db.many('SELECT owner FROM run_sessions')).map((r) => r.owner);++ assert.deepStrictEqual(left, [`user:${B}`], 'only Bo\'s session is left');++ assert.strictEqual(await d.db.value('SELECT requester FROM run_jobs WHERE id = $1', ['job_01JZ0000000000000000000ADA']), 'user:deleted', 'the job stays, without its requester');++ assert.strictEqual(await d.db.value('SELECT requester FROM run_jobs WHERE id = $1', ['job_01JZ0000000000000000000BQB']), `user:${B}`, 'Bo\'s job is untouched');++ assert.strictEqual(counts.retained.run_jobs_tombstoned, 1);++ console.log(' ✓ export, erase and the guard');++ } finally { await d.close(); }++})();
@@ -99,7 +99,7 @@ placed from the Fabric's offers and actually runs — the poller in `server/jobs
@@ -99,7 +99,7 @@ placed from the Fabric's offers and actually runs — the poller in `server/jobs
(`bot.job.dispatch`, audience `openvibe.bot`, minted from `OV_OAUTH_CLIENT_ID`/`OV_OAUTH_CLIENT_SECRET` against
(`bot.job.dispatch`, audience `openvibe.bot`, minted from `OV_OAUTH_CLIENT_ID`/`OV_OAUTH_CLIENT_SECRET` against
`OV_NETWORK_INTERNAL_URL` and cached until 60 s before expiry) and mirrors its answer. Unset the client secret and
`OV_NETWORK_INTERNAL_URL` and cached until 60 s before expiry) and mirrors its answer. Unset the client secret and
the bridge is off: `/api/ready` skips the `bot` check and every job stays `queued`.
the bridge is off: `/api/ready` skips the `bot` check and every job stays `queued`.
-- `openvibe-contracts` v0.106.0, `openvibe-sdk` v0.67.0 and `openvibe-shared` v4.48.0 (package.json).
+- `openvibe-contracts` v0.106.0, `openvibe-sdk` v0.74.0 and `openvibe-shared` v4.48.0 (package.json).
## Capabilities and routes
## Capabilities and routes
@@ -253,6 +253,6 @@ until Run has a product surface; the vhost below is installed then.
@@ -253,6 +253,6 @@ until Run has a product surface; the vhost below is installed then.
<!-- versions:start -->
<!-- versions:start -->
- openvibe-contracts: v0.223.0
- openvibe-contracts: v0.223.0
-- openvibe-sdk: v0.67.0
+- openvibe-sdk: v0.74.0
- openvibe-shared: v4.48.0
- openvibe-shared: v4.48.0
<!-- versions:end -->
<!-- versions:end -->
@@ -8,11 +8,11 @@
@@ -8,11 +8,11 @@
"port": 4920,
"port": 4920,
"domain": "openvibe.run",
"domain": "openvibe.run",
"contracts": "openvibe-contracts v0.223.0",
"contracts": "openvibe-contracts v0.223.0",
- "sdk": "openvibe-sdk v0.67.0",
+ "sdk": "openvibe-sdk v0.74.0",
"shared": "openvibe-shared v4.48.0",
"shared": "openvibe-shared v4.48.0",
"packages": [
"packages": [
"openvibe-contracts v0.223.0",
"openvibe-contracts v0.223.0",
- "openvibe-sdk v0.67.0",
+ "openvibe-sdk v0.74.0",
"openvibe-shared v4.48.0"
"openvibe-shared v4.48.0"
],
],
"capabilities": [
"capabilities": [
new file mode 100644
new file mode 100644
@@ -0,0 +1,13 @@
@@ -0,0 +1,13 @@
+-- phase: expand
+-- OpenVibe.Run: the receipts of the ADR-033 account export and deletion deliveries this service applied
+-- (openvibe-sdk/account-data's ACCOUNT_DATA_SCHEMA), so a redelivered export or deletion changes nothing.
+-- The row holds an export_id or deletion_id, the subject and the counts sent to Network; it is not a secret.
+
+CREATE TABLE IF NOT EXISTS account_data_events (
+ id TEXT PRIMARY KEY,
+ kind TEXT NOT NULL,
+ subject TEXT NOT NULL,
+ outcome JSONB,
+ applied_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+ sent_at TIMESTAMPTZ
+);
@@ -16,7 +16,7 @@
@@ -16,7 +16,7 @@
"helmet": "^8.1.0",
"helmet": "^8.1.0",
"iovalkey": "^0.4.0",
"iovalkey": "^0.4.0",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.223.0",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.223.0",
- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.67.0",
+ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.74.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.48.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.48.0",
"pg": "^8.23.0"
"pg": "^8.23.0"
},
},
@@ -1163,9 +1163,9 @@
@@ -1163,9 +1163,9 @@
}
}
},
},
"node_modules/openvibe-sdk": {
"node_modules/openvibe-sdk": {
- "version": "0.67.0",
- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.67.0",
- "integrity": "sha512-s7NzjaNcowVcOIiup2z0ze+xsq/qnr++3pHSI8JzirqJPQ5JsBSUF5RkjOHyfk9bRKlmhY0fsNS6x4dsYylyqw==",
+ "version": "0.74.0",
+ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.74.0",
+ "integrity": "sha512-DO9KVPnUriY/I4lX6ZEJtXaUvRDbJyqnqJxvlpl+zkNGFILpmuBXVJogNOCfK071Qci1gg7C4Iw1Xrseklcajg==",
"license": "MIT",
"license": "MIT",
"bin": {
"bin": {
"openvibe-download": "bin/openvibe-download.js"
"openvibe-download": "bin/openvibe-download.js"
@@ -27,7 +27,7 @@
@@ -27,7 +27,7 @@
"helmet": "^8.1.0",
"helmet": "^8.1.0",
"iovalkey": "^0.4.0",
"iovalkey": "^0.4.0",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.223.0",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.223.0",
- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.67.0",
+ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.74.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.48.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.48.0",
"pg": "^8.23.0"
"pg": "^8.23.0"
},
},
new file mode 100644
new file mode 100644
@@ -0,0 +1,43 @@
@@ -0,0 +1,43 @@
+'use strict';
+/**
+ * Account export and deletion → Run (ADR-033; openvibe-sdk/account-data).
+ *
+ * Run keeps two things about a person:
+ *
+ * - their open browser sessions (run_sessions, server/sessions/), keyed `owner = user:usr_…`. The export carries each
+ * session's settings and timing, never its ticket or stream secret; deletion removes them.
+ * - the jobs they submitted (run_jobs.requester = `user:usr_…`). The export carries the job, its result and its
+ * timings; the row itself stays, because it is the project's record of the work and of what it cost, but the
+ * requester becomes `user:deleted`, so the job no longer leads to a person.
+ *
+ * Telemetry samples that name a person are covered by the SDK itself (telemetry.json, then subject NULL).
+ */
+const { createAccountData, TOPICS } = require('openvibe-sdk/account-data');
+
+const TABLES = [
+ {
+ table: 'run_sessions', subject: 'owner', value: (usr) => `user:${usr}`, file: 'sessions.json', order: 'created_at',
+ columns: ['id', 'via', 'kind', 'state', 'width', 'height', 'ttl_s', 'idle_s', 'machine_asked', 'requirements', 'machine', 'created_at', 'started_at', 'last_used_at', 'ended_at', 'end_reason', 'end_detail', 'metered_minutes', 'updated_at'],
+ },
+ // The jobs are handled by the service itself (the requester is tombstoned, not deleted), so they are not in the
+ // table map; they are listed as covered in the test's guard.
+];
+
+/** The jobs the person submitted: their own work, with its result. */
+async function extraExport(db, subject) {
+ const jobs = await db.many(`SELECT id, project_id, class, state, provider, region, job, result, ttl_ms, created_at, updated_at, started_ms, finished_ms, wall_ms, usage_read, error_code
+ FROM run_jobs WHERE requester = $1 ORDER BY created_at DESC LIMIT 5000`, [`user:${subject}`]);
+ return jobs.length ? [{ name: 'jobs.json', content: jobs }] : [];
+}
+
+/** The requester of the person's jobs becomes a tombstone: the job stays, the person does not lead to it. */
+async function extraErase(t, subjects, counts) {
+ const forms = subjects.map((s) => `user:${s}`);
+ counts.add(counts.retained, 'run_jobs_tombstoned', await t.exec(`UPDATE run_jobs SET requester = 'user:deleted' WHERE requester = ANY($1::text[])`, [forms]));
+}
+
+function create({ db, log = console } = {}) {
+ return createAccountData({ db, service: 'run', tables: TABLES, extraExport, extraErase, log });
+}
+
+module.exports = { create, TABLES, TOPICS, extraExport, extraErase };
@@ -74,6 +74,9 @@ function loadConfig(env = process.env) {
@@ -74,6 +74,9 @@ function loadConfig(env = process.env) {
events: {
events: {
url: trim(env.EVENTS_URL || ''),
url: trim(env.EVENTS_URL || ''),
intervalMs: int(env.EVENTS_RELAY_INTERVAL_MS, 2000),
intervalMs: int(env.EVENTS_RELAY_INTERVAL_MS, 2000),
+ // The account-data consumer (POST /internal/events): the signing secret Network's deliveries use. Unset:
+ // the route is off and Network's account export/deletion does not wait for Run.
+ secret: String(env.RUN_EVENTS_SECRET || '').trim(),
},
},
// The background jobs (the events relay and the dispatcher poller); RUN_JOBS=off runs the API alone,
// The background jobs (the events relay and the dispatcher poller); RUN_JOBS=off runs the API alone,
@@ -24,6 +24,8 @@ const { createValkey } = require('openvibe-sdk/valkey');
@@ -24,6 +24,8 @@ const { createValkey } = require('openvibe-sdk/valkey');
const { gracefulStop, telemetry } = require('openvibe-sdk/service');
const { gracefulStop, telemetry } = require('openvibe-sdk/service');
const { createPgTelemetrySink, startTelemetryRetention } = require('openvibe-sdk/telemetry');
const { createPgTelemetrySink, startTelemetryRetention } = require('openvibe-sdk/telemetry');
const { createRegistry } = require('openvibe-shared/metrics');
const { createRegistry } = require('openvibe-shared/metrics');
+const accountDataLib = require('./account-data');
+const { createNetworkSender, startSubscriptions } = require('openvibe-sdk/account-data');
async function main() {
async function main() {
const config = loadConfig();
const config = loadConfig();
@@ -33,6 +35,8 @@ async function main() {
@@ -33,6 +35,8 @@ async function main() {
if (m.held.length) console.warn(`[Run] migrations held: ${m.held.map((h) => `${h.id} (${h.reason})`).join('; ')}`);
if (m.held.length) console.warn(`[Run] migrations held: ${m.held.map((h) => `${h.id} (${h.reason})`).join('; ')}`);
const valkey = createValkey({ url: config.valkey.url, prefix: config.valkey.prefix });
const valkey = createValkey({ url: config.valkey.url, prefix: config.valkey.prefix });
+ const accountData = accountDataLib.create({ db, log: console });
+ const accountSend = config.oauth.clientSecret ? createNetworkSender({ networkInternalUrl: config.network.internalUrl, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret }) : null;
const app = createApp({ config, db, valkey, registry });
const app = createApp({ config, db, valkey, registry });
const { keys, outbox, stream, tickets, dispatch, indexnow, sessions } = app.locals;
const { keys, outbox, stream, tickets, dispatch, indexnow, sessions } = app.locals;
keys.start();
keys.start();
@@ -48,8 +52,16 @@ async function main() {
@@ -48,8 +52,16 @@ async function main() {
sessions.start();
sessions.start();
}
}
+ app.post('/internal/events', accountData.consumer({ secrets: [config.events.secret].filter((s) => s.length >= 32), send: accountSend }));
+ let accountSubscription = null;
const server = app.listen(config.port, config.host, () => {
const server = app.listen(config.port, config.host, () => {
console.log(`[Run] ${config.nodeEnv} on http://${config.host}:${config.port} → ${config.baseUrl} (store ${db.store}; valkey ${valkey ? 'on' : 'off: this process only'})`);
console.log(`[Run] ${config.nodeEnv} on http://${config.host}:${config.port} → ${config.baseUrl} (store ${db.store}; valkey ${valkey ? 'on' : 'off: this process only'})`);
+ if (config.events.secret.length >= 32 && config.events.url) {
+ accountSubscription = startSubscriptions({
+ eventsUrl: config.events.url, endpoint: `http://127.0.0.1:${config.port}/internal/events`, secret: config.events.secret,
+ networkInternalUrl: config.network.internalUrl, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret,
+ });
+ }
console.log(`[Run] jobs ${config.jobs.enabled ? 'on' : 'off (RUN_JOBS=off)'}; events relay ${outbox.enabled ? `→ ${config.events.url}` : 'off (run.job.* events wait in the outbox)'}; stream tickets ${ticketKey.enabled ? `on (${config.stream.keyId}${ticketKey.ephemeral ? ', ephemeral key' : ''})` : 'off (no stream key)'}; dispatcher ${dispatch.enabled ? `on (poll ${config.dispatch.pollMs} ms): ${dispatch.note}` : `off: ${dispatch.note}`}; ${dispatch.status().usage_relay.note}; ${sessions.status().note}`);
console.log(`[Run] jobs ${config.jobs.enabled ? 'on' : 'off (RUN_JOBS=off)'}; events relay ${outbox.enabled ? `→ ${config.events.url}` : 'off (run.job.* events wait in the outbox)'}; stream tickets ${ticketKey.enabled ? `on (${config.stream.keyId}${ticketKey.ephemeral ? ', ephemeral key' : ''})` : 'off (no stream key)'}; dispatcher ${dispatch.enabled ? `on (poll ${config.dispatch.pollMs} ms): ${dispatch.note}` : `off: ${dispatch.note}`}; ${dispatch.status().usage_relay.note}; ${sessions.status().note}`);
});
});
server.keepAliveTimeout = 65_000;
server.keepAliveTimeout = 65_000;
@@ -77,6 +89,7 @@ async function main() {
@@ -77,6 +89,7 @@ async function main() {
() => sessions.stop(),
() => sessions.stop(),
() => stream.close(),
() => stream.close(),
() => outbox.stop(),
() => outbox.stop(),
+ () => { if (accountSubscription) accountSubscription.stop(); },
],
],
close: [
close: [
async () => { if (telemetryRetention) await telemetryRetention.stop(); await telemetry.stop(); },
async () => { if (telemetryRetention) await telemetryRetention.stop(); await telemetry.stop(); },
new file mode 100644
new file mode 100644
@@ -0,0 +1,46 @@
@@ -0,0 +1,46 @@
+'use strict';
+/**
+ * ADR-033: Run's part of an account export and deletion. A person's open browser sessions are exported (their
+ * settings and timing, never a ticket) and deleted; someone else's stay. The guard fails when a new table with a
+ * person column is neither covered nor exempted with a reason.
+ */
+const assert = require('assert');
+const { uncovered } = require('openvibe-sdk/account-data');
+const { testDb } = require('./helpers/db');
+const { TABLES, create } = require('../server/account-data');
+
+const A = 'usr_01JZ0000000000000000000AAA';
+const B = 'usr_01JZ0000000000000000000BBB';
+
+(async () => {
+ const d = await testDb();
+ try {
+ const accountData = create({ db: d.db });
+ const missing = await uncovered(d.db, { tables: [...TABLES, 'run_jobs'], exempt: { account_data_events: 'the ledger of requests itself: which export and deletion Run applied' } });
+ assert.deepStrictEqual(missing, [], `a table with a person column is neither covered nor exempted: ${missing.join(', ')}`);
+
+ for (const [subject, name, jobId] of [[A, 'ada', 'job_01JZ0000000000000000000ADA'], [B, 'bo', 'job_01JZ0000000000000000000BQB']]) {
+ await d.db.exec(`INSERT INTO run_sessions (id, owner, kind, state, width, height, ttl_s, idle_s, request_digest, created_at, updated_at)
+ VALUES ($1, $2, 'browser', 'ended', 1280, 720, 900, 60, 'digest', 1, 1)`, [`ses_${name}`, `user:${subject}`]);
+ await d.db.exec(`INSERT INTO run_jobs (id, project_id, requester, idempotency_key, class, job, ttl_ms, state, created_at, updated_at)
+ VALUES ($1, 'proj_1', $2, $3, 'browser', '{}', 1000, 'succeeded', 1, 1)`, [jobId, `user:${subject}`, `idem-${name}`]);
+ }
+ const part = await accountData.exportPart(A);
+ const file = part.files.find((f) => f.name === 'sessions.json');
+ assert.ok(file, 'sessions.json is in the part');
+ assert.strictEqual(file.content.length, 1);
+ assert.strictEqual(file.content[0].id, 'ses_ada');
+ const jobs = part.files.find((f) => f.name === 'jobs.json');
+ assert.ok(jobs, 'jobs.json is in the part');
+ assert.deepStrictEqual(jobs.content.map((j) => j.id), ['job_01JZ0000000000000000000ADA'], 'only their own job');
+
+ const counts = await accountData.erase([A]);
+ assert.strictEqual(counts.erased.run_sessions, 1);
+ const left = (await d.db.many('SELECT owner FROM run_sessions')).map((r) => r.owner);
+ assert.deepStrictEqual(left, [`user:${B}`], 'only Bo\'s session is left');
+ assert.strictEqual(await d.db.value('SELECT requester FROM run_jobs WHERE id = $1', ['job_01JZ0000000000000000000ADA']), 'user:deleted', 'the job stays, without its requester');
+ assert.strictEqual(await d.db.value('SELECT requester FROM run_jobs WHERE id = $1', ['job_01JZ0000000000000000000BQB']), `user:${B}`, 'Bo\'s job is untouched');
+ assert.strictEqual(counts.retained.run_jobs_tombstoned, 1);
+ console.log(' ✓ export, erase and the guard');
+ } finally { await d.close(); }
+})();