openvibers/openvibe.events Commits
account-data: a person's events, with the guard on every person column (#163)
Changes
@@@ -122,6 +122,11 @@ EVENTS_RECEIPT_RETENTION_DAYS=90#OV_BILLING_INTERNAL_URL=http://127.0.0.1:4600#OV_OAUTH_CLIENT_ID=events#OV_OAUTH_CLIENT_SECRET=++# The account export and deletion consumer (ADR-033): Events' own subscription to the two account topics, delivered++# back to POST /internal/events. Unset: the route is off and the account events do not reach Events. The secret is 32++# characters or more; it is the one Events signs its own deliveries with.++#EVENTS_EVENTS_URL=https://openvibe.events++#EVENTS_EVENTS_SECRET=#OV_BILLING_AUDIENCE=openvibe.billing#OV_BILLING_TIMEOUT_MS=10000@@@ -120,6 +120,15 @@ Operators: `GET /api/v1/deliveries?status=dead` is the dead-letter queue; `POSTPull consumers: start with `GET /api/v1/events?topic=media.vod.*&limit=100`, then request `GET /api/v1/events?topic=media.vod.*&after=<next_cursor>&limit=100`, using the previous response's opaque `next_cursor` ([ADR-042](https://github.com/OpenVibers/OpenVibe.Contracts/blob/main/docs/adr/ADR-042-events-fabric.md) decision 7). The response carries `{ events: [{ cursor, event }], next_cursor, latest_cursor }` (`latest_cursor` is the head as an opaque cursor, for a consumer that starts at "now" without history), plus `gap: { from_seq, to_seq }` when the position is older than both the hot store and the replay tier (below) or from another retention epoch. Use `next_cursor` even when no events match, since it advances past scanned events. A position is only ever an opaque cursor: `after_seq` is refused with 400 (never read as "from the start"), and there is no per-event sequence number (dedupe on `event.event_id`). `PUT /api/v1/checkpoints { topic, cursor, carrier? }` stores a consumer's cursor here if it has nowhere better (a cursor string; a number is refused); `GET /api/v1/checkpoints?topic=` and the PUT answer `{ consumer, topic, cursor, carrier, updated_at }`, with `cursor` as stored (null when none is).++### Account export and deletion (ADR-033)++++Events is the bus, so two of its columns name a person: the `actor` (the SubjectRef of whoever caused the event) and the `subject` (what the event is about). `server/account-data.js` answers the two account topics for them:++++- **Export** (`network.account.export_requested`): the person's part carries the envelope of the events they caused and the events about them (`id`, `seq`, `event_type`, `source`, `subject_type`, `subject_id`, `visibility`, `project_id`, `occurred_at`, `received_at`) — never the payload, which is the publishing service's record and may name other people. The replay tier's rows are in the part too (`events-replay.json`).++- **Deletion** (`network.account.deleted`): the events the person caused and the events about them become tombstones, exactly as a redaction does: the payload goes and the actor becomes the producer, so no person stays linked to them. The envelope keeps its id, seq, type, subject and timestamps, so the log stays continuous and the replay tier stays in step. Both tiers are answered.++++The route is `POST /internal/events`, signed (v2) and loopback only, mounted by `server/app.js`; `server/index.js` creates the two subscriptions at boot with `startSubscriptions` (`EVENTS_EVENTS_URL`, `EVENTS_EVENTS_SECRET` and the service's own OAuth client). Without the secret the route is off and the account events do not reach Events. The receipts of what it applied are in `account_data_events` (migration 0010), so a redelivered export or deletion changes nothing. `test/account-data.test.js` asserts `uncovered()` is empty, so a new table with a person column fails CI until it is covered or exempted with a reason.++### Retention: three tiers[ADR-042](https://github.com/OpenVibers/OpenVibe.Contracts/blob/main/docs/adr/ADR-042-events-fabric.md) decision 8. An event lives in one tier at a time:@@@ -398,6 +407,6 @@ Part of the [OpenVibe network](https://openvibe.network). Built in the open by [<!-- versions:start -->- openvibe-contracts: v0.222.0--- openvibe-sdk: v0.67.0++- openvibe-sdk: v0.74.0- openvibe-shared: v4.48.0<!-- versions:end -->@@@ -8,7 +8,7 @@"port": 4300,"domain": "openvibe.events","contracts": "openvibe-contracts v0.222.0",-- "sdk": "openvibe-sdk v0.67.0",++ "sdk": "openvibe-sdk v0.74.0","shared": "openvibe-shared v4.48.0","capabilities": ["events.event.publish",nnew file mode 100644@@@ -0,0 +1,13 @@++-- phase: expand++-- OpenVibe.Events: 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++);@@@ -13,7 +13,7 @@"express": "^4.21.2","iovalkey": "^0.4.0","openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.222.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"},@@@ -1119,9 +1119,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"@@@ -25,7 +25,7 @@"express": "^4.21.2","iovalkey": "^0.4.0","openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.222.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,103 @@++'use strict';++/**++ * Account export and deletion → Events (ADR-033; openvibe-sdk/account-data).++ *++ * Events is the bus: it stores the events the services publish, and two of its columns name a person — the actor (the++ * SubjectRef of whoever caused the event) and the subject (what the event is about). The payload is the publishing++ * service's record, and the platform's redaction mechanism (server/redaction.js) is what takes a copy of it back.++ *++ * export the events the person caused and the events about them: the envelope (id, type, source, subject,++ * visibility, timestamps), never the payload — that is the publisher's record and may name other people.++ * erase the events the person caused and the events about them become tombstones, exactly as a redaction does:++ * the payload goes and the actor becomes the producer, so no person stays linked to them. The envelope keeps++ * its id, seq, type, subject and timestamps, so the log stays continuous and the replay tier stays in step.++ * Both tiers are answered: the hot store and the replay archive (its body holds the actor, so its rows are++ * opened to find it).++ *++ * 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 archive = require('./archive');++const redaction = require('./redaction');++++const TABLES = [];++++// The envelope the person gets: the event's place in the log, not what it carried.++const ENVELOPE = ['id', 'seq', 'event_type', 'source', 'subject_type', 'subject_id', 'visibility', 'project_id', 'occurred_at', 'received_at'];++const ROW_LIMIT = 5000;++++/** The person in a SubjectRef JSON column: {type: 'user', id: 'usr_…'}. */++function actorIs(row, subject) {++ if (!row || !row.actor) return false;++ let a = null;++ try { a = typeof row.actor === 'string' ? JSON.parse(row.actor) : row.actor; } catch { return false; }++ return Boolean(a) && a.type === 'user' && a.id === subject;++}++++/** The event is about the person. */++function subjectIs(row, subject) {++ return row && row.subject_type === 'user' && row.subject_id === subject;++}++++/** The events that name the person, in the hot store. */++async function hotRows(db, subject) {++ return await db.many(++ `SELECT ${ENVELOPE.join(', ')} FROM events++ WHERE (actor::jsonb ->> 'type' = 'user' AND actor::jsonb ->> 'id' = $1) OR (subject_type = 'user' AND subject_id = $1)++ ORDER BY received_at DESC LIMIT ${ROW_LIMIT}`, [subject]);++}++++/** The events that name the person, in the replay tier: the actor lives in the packed body, so the rows are opened. */++async function archivedRows(db, subject) {++ const out = [];++ for (const a of await db.many('SELECT * FROM events_archive WHERE redacted_at IS NULL')) {++ const r = archive.unpack(a);++ if (actorIs(r, subject) || subjectIs(r, subject)) out.push(r);++ }++ return out;++}++++/** The person's part of an account export: the envelope of the events they caused and the events about them. */++async function extraExport(db, subject) {++ const hot = await hotRows(db, subject);++ const files = [];++ if (hot.length) files.push({ name: 'events.json', content: hot });++ const archived = await archivedRows(db, subject);++ if (archived.length) {++ files.push({++ name: 'events-replay.json',++ content: archived.slice(0, ROW_LIMIT).map((r) => Object.fromEntries(ENVELOPE.map((c) => [c, r[c] === undefined ? null : r[c]]))),++ });++ }++ return files;++}++++/** Turn the person's events into tombstones, as a redaction does: the payload goes, the actor becomes the producer. */++async function extraErase(t, subjects, counts) {++ const now = Date.now();++ for (const subject of subjects) {++ const by = `account-deletion:${subject}`;++ const hot = await t.many('SELECT * FROM events WHERE (actor::jsonb ->> \'type\' = \'user\' AND actor::jsonb ->> \'id\' = $1) OR (subject_type = \'user\' AND subject_id = $1) FOR UPDATE', [subject]);++ for (const r of hot) {++ const cols = redaction.tombstone(r, { by, at: now });++ await t.exec('UPDATE events SET payload = $1, actor = $2, size_bytes = $3, redacted_at = $4, redacted_by = $5 WHERE id = $6',++ [cols.payload, cols.actor, cols.size_bytes, cols.redacted_at, cols.redacted_by, r.id]);++ }++ counts.add(counts.erased, 'events', hot.length);++ // The replay tier: the body holds the actor, so its rows are opened, tombstoned and packed again.++ const archived = await archivedRows(t, subject);++ for (const r of archived) {++ const cols = redaction.tombstone(r, { by, at: now });++ // The replay tier keeps the redaction's who in the packed body; the table itself has only the marker.++ await t.exec('UPDATE events_archive SET body = $1, redacted_at = $2 WHERE id = $3',++ [archive.pack({ ...r, ...cols }), cols.redacted_at, r.id]);++ }++ counts.add(counts.erased, 'events_replay', archived.length);++ }++}++++function create({ db, log = console } = {}) {++ return createAccountData({ db, service: 'events', tables: TABLES, extraExport, extraErase, log });++}++++module.exports = { create, TABLES, TOPICS, extraExport, extraErase };@@@ -24,7 +24,7 @@ const openapi = require('./openapi');const ovServe = require('openvibe-shared/serve');const pkg = require('../package.json');--function createApp({ config, store, auth, keys, worker, realtime, metrics, dnsLookup, clock, log = console, valkey = null, policies = null, planner = null, carriers = null, indexnow = null }) {++function createApp({ config, store, auth, keys, worker, realtime, metrics, dnsLookup, clock, log = console, valkey = null, policies = null, planner = null, carriers = null, indexnow = null, accountData = null, accountSend = null }) {const app = express();app.disable('x-powered-by');app.set('trust proxy', 'loopback');@@@ -48,6 +48,10 @@ function createApp({ config, store, auth, keys, worker, realtime, metrics, dnsLo// routers and outside the per-actor limiter: it is static and unauthenticated.app.use('/shared', ovServe.handler());++ // The account export/deletion consumer (ADR-033, ./account-data.js): the two account topics, signed and loopback++ // only. No body parser before it: the v2 signature covers the raw body.++ if (accountData) app.post('/internal/events', accountData.consumer({ secrets: [config.accountData.secret].filter((s) => s.length >= 32), send: accountSend }));++// Realtime first: it answers CORS preflights and must not go through the JSON body parser.app.options('/realtime/stream', realtime.cors);app.get('/realtime/stream', realtime.cors, realtime.handler);@@@ -231,6 +231,20 @@ function load(env = process.env) {timeoutMs: Math.max(1000, int(env.OV_BILLING_TIMEOUT_MS, 10000)),},++ // The service's own OAuth client at OpenVibe.Network: the account-data consumer signs its part and its++ // confirmation with it, and its Events subscription is created with it.++ oauth: {++ clientId: env.OV_OAUTH_CLIENT_ID || 'events',++ clientSecret: env.OV_OAUTH_CLIENT_SECRET || '',++ },++++ // The account-data consumer (ADR-033): Events' own subscription to the two account topics, delivered back to++ // POST /internal/events. Unset: the route is off and the account events do not reach Events.++ accountData: {++ url: (env.EVENTS_EVENTS_URL || baseUrl).replace(/\/+$/, ''),++ secret: String(env.EVENTS_EVENTS_SECRET || '').trim(),++ },++// Realtime (SSE) gatewayrealtime: {maxConnections: int(env.REALTIME_MAX_CONNECTIONS, 2000),@@@ -17,6 +17,8 @@ const { createAuth } = require('./auth');const { createWorker } = require('./worker');const { createRealtime } = require('./realtime');const { createApp } = require('./app');++const accountDataLib = require('./account-data');++const { createNetworkSender, startSubscriptions } = require('openvibe-sdk/account-data');const { createMetrics } = require('./metrics');const { createGuardedPost } = require('./egress');const { createBillingReadings } = require('./billing');@@@ -63,7 +65,13 @@ async function start({indexnow = createIndexNow({ host: config.siteUrl, key: '', log });}-- const app = createApp({ config, store, auth, keys, worker, realtime, metrics, dnsLookup, clock, log, valkey, policies, planner, carriers, indexnow });++ // The account export/deletion part (ADR-033, ./account-data.js): the handle over this database and the sender++ // that pushes the part and the confirmation to Network's internal routes.++ const accountData = accountDataLib.create({ db, log });++ const accountSend = config.oauth && config.oauth.clientSecret++ ? createNetworkSender({ networkInternalUrl: config.networkInternalUrl, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, fetch: fetchImpl })++ : null;++ const app = createApp({ config, store, auth, keys, worker, realtime, metrics, dnsLookup, clock, log, valkey, policies, planner, carriers, indexnow, accountData, accountSend });app.locals.indexnow = indexnow;// The pages crawlers see change only with a release, so this boot is the moment to announce them.// A no-op without INDEXNOW_KEY.@@@ -88,6 +96,16 @@ async function start({log.warn(`[events] carriers did not start: ${err.message}`);}++ // Events' own subscription to the two account topics (ADR-033): the deliveries come back to++ // POST /internal/events. Off without EVENTS_EVENTS_SECRET or the OAuth client secret.++ const accountSubscription = (config.accountData.secret.length >= 32 && config.oauth && config.oauth.clientSecret)++ ? startSubscriptions({++ eventsUrl: config.accountData.url, endpoint: `http://127.0.0.1:${config.port}/internal/events`,++ secret: config.accountData.secret, networkInternalUrl: config.networkInternalUrl,++ clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, fetch: fetchImpl, log,++ })++ : null;++const prune = async () => {try {// Tiers (ADR-042 decision 8): hot events past EVENTS_RETENTION_DAYS move to the replay tier, which keeps them@@@ -142,6 +160,7 @@ async function start({}async function close() {++ if (accountSubscription) accountSubscription.stop();clearInterval(pruneTimer);if (usageTimer) clearInterval(usageTimer);await billing.stop();nnew file mode 100644@@@ -0,0 +1,87 @@++'use strict';++/**++ * ADR-033: Events' part of an account export and deletion. The events a person caused and the events about them++ * are theirs: the export carries their envelope (never the payload, which is the publishing service's record) and a++ * deletion turns them into tombstones, so the payload goes and the actor becomes the producer. Both tiers are++ * answered: the hot store and the replay archive.++ *++ * 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('./db');++const { TABLES, create } = require('../server/account-data');++const archive = require('../server/archive');++++const A = 'usr_01JZ0000000000000000000AAA';++const B = 'usr_01JZ0000000000000000000BBB';++++let seq = 0;++const nextSeq = () => ++seq;++++/** One event in the hot store. */++async function insertEvent(db, { id, actor, subjectType, subjectId, payload }) {++ await db.exec(`INSERT INTO events (id, seq, event_type, version, source, actor, subject_type, subject_id, trace_id, priority, visibility, occurred_at, received_at, payload, hops, publisher)++ VALUES ($1, $2, 'chat.message.created', 1, 'chat', $3, $4, $5, 'trace-1', 'important', 'internal', '2026-10-01T00:00:00.000Z', 1, $6, 1, 'chat')`,++ [id, nextSeq(), JSON.stringify(actor), subjectType, subjectId, payload]);++}++++/** One event in the replay tier: the actor and the payload live in the packed body. */++async function insertArchived(db, { id, actor, subjectType, subjectId, payload }) {++ const row = {++ id, seq: nextSeq(), event_type: 'chat.message.created', source: 'chat', subject_type: subjectType, subject_id: subjectId,++ visibility: 'internal', project_id: null, env: 'production', received_at: 1, redacted_at: null, redacts: 0,++ version: 1, actor: JSON.stringify(actor), subject_revision: null, trace_id: 'trace-1', priority: 'important',++ occurred_at: '2026-10-01T00:00:00.000Z', payload, hops: 1, publisher: 'chat', request_id: null, size_bytes: 10, redacted_by: null,++ };++ await db.exec(`INSERT INTO events_archive (id, seq, epoch, event_type, source, subject_type, subject_id, visibility, project_id, env, received_at, redacts, body, archived_at)++ VALUES ($1, $2, 1, $3, $4, $5, $6, $7, null, 'production', 1, 0, $8, 1)`,++ [id, row.seq, row.event_type, row.source, row.subject_type, row.subject_id, row.visibility, archive.pack(row)]);++}++++(async () => {++ const d = await testDb();++ try {++ const accountData = create({ db: d.db });++ const missing = await uncovered(d.db, {++ tables: [...TABLES, 'events'],++ exempt: { account_data_events: 'the ledger of requests itself: which export and deletion Events applied' },++ });++ assert.deepStrictEqual(missing, [], `a table with a person column is neither covered nor exempted: ${missing.join(', ')}`);++++ // Ada caused an event (her chat message) and is the subject of another (a notification for her).++ await insertEvent(d.db, { id: 'evt_01JZ0000000000000000000AAA', actor: { type: 'user', id: A }, subjectType: 'chat_message', subjectId: 'cm_1', payload: '{"message":"hello"}' });++ await insertEvent(d.db, { id: 'evt_01JZ0000000000000000000AAB', actor: { type: 'service', id: 'network' }, subjectType: 'user', subjectId: A, payload: '{"title":"a notification"}' });++ await insertEvent(d.db, { id: 'evt_01JZ0000000000000000000AAC', actor: { type: 'user', id: B }, subjectType: 'chat_message', subjectId: 'cm_2', payload: '{"message":"hi"}' });++ await insertArchived(d.db, { id: 'evt_01JZ0000000000000000000AAD', actor: { type: 'user', id: A }, subjectType: 'chat_message', subjectId: 'cm_3', payload: '{"message":"older"}' });++++ const part = await accountData.exportPart(A);++ const file = part.files.find((f) => f.name === 'events.json');++ assert.ok(file, 'events.json is in the part');++ assert.deepStrictEqual(file.content.map((e) => e.id).sort(), ['evt_01JZ0000000000000000000AAA', 'evt_01JZ0000000000000000000AAB']);++ assert.ok(!JSON.stringify(part).includes('hello'), 'the payload is never exported');++ assert.ok(!JSON.stringify(part).includes('evt_01JZ0000000000000000000AAC'), 'nobody else\'s events');++ const replay = part.files.find((f) => f.name === 'events-replay.json');++ assert.ok(replay, 'the replay tier is in the part');++ assert.deepStrictEqual(replay.content.map((e) => e.id), ['evt_01JZ0000000000000000000AAD']);++++ const counts = await accountData.erase([A]);++ assert.strictEqual(counts.erased.events, 2, 'her two events are tombstoned');++ assert.strictEqual(counts.erased.events_replay, 1, 'her replay event too');++ const hot = await d.db.many('SELECT id, payload, actor, redacted_at FROM events WHERE id IN ($1, $2, $3) ORDER BY id',++ ['evt_01JZ0000000000000000000AAA', 'evt_01JZ0000000000000000000AAB', 'evt_01JZ0000000000000000000AAC']);++ const ada = hot.find((e) => e.id === 'evt_01JZ0000000000000000000AAA');++ assert.ok(ada.redacted_at, 'her event is a tombstone');++ assert.strictEqual(JSON.parse(ada.payload).redacted, true, 'the payload is gone');++ assert.strictEqual(JSON.parse(ada.actor).type, 'service', 'no person stays linked to it');++ assert.ok(!JSON.stringify(hot).includes('hello'), 'her message is not in the log any more');++ const bo = hot.find((e) => e.id === 'evt_01JZ0000000000000000000AAC');++ assert.strictEqual(bo.redacted_at, null, 'Bo\'s event stays');++ assert.ok(JSON.parse(bo.actor).id === B, 'Bo stays in his own event');++ const archived = await d.db.many('SELECT body FROM events_archive WHERE id = $1', ['evt_01JZ0000000000000000000AAD']);++ const unpacked = archive.unpack(archived[0]);++ assert.strictEqual(JSON.parse(unpacked.payload).redacted, true, 'the replay copy is tombstoned too');++ assert.strictEqual(JSON.parse(unpacked.actor).type, 'service', 'no person stays linked to the replay copy');++ console.log(' ✓ export, tombstone and the guard');++ } finally { await d.close(); }++})().catch((err) => { console.error(err); process.exit(1); });@@@ -31,6 +31,9 @@ const PLANTED = {// cloud-v1 (an SQS-compatible queue): signs requests, never echoed.CLOUD_QUEUE_ACCESS_KEY: `ovsecret-events-cloud-access-${crypto.randomBytes(12).toString('hex')}`,CLOUD_QUEUE_SECRET_KEY: `ovsecret-events-cloud-secret-${crypto.randomBytes(12).toString('hex')}`,++ // The account-data consumer (ADR-033): the secret its own loopback subscription signs the two++ // account topics with. It is Events' own signing secret, never echoed.++ EVENTS_EVENTS_SECRET: `ovsecret-events-account-${crypto.randomBytes(12).toString('hex')}`,};t('Events reads no secret from its environment but the planted ones', async () => {
@@ -122,6 +122,11 @@ EVENTS_RECEIPT_RETENTION_DAYS=90
@@ -122,6 +122,11 @@ EVENTS_RECEIPT_RETENTION_DAYS=90
#OV_BILLING_INTERNAL_URL=http://127.0.0.1:4600
#OV_BILLING_INTERNAL_URL=http://127.0.0.1:4600
#OV_OAUTH_CLIENT_ID=events
#OV_OAUTH_CLIENT_ID=events
#OV_OAUTH_CLIENT_SECRET=
#OV_OAUTH_CLIENT_SECRET=
+# The account export and deletion consumer (ADR-033): Events' own subscription to the two account topics, delivered
+# back to POST /internal/events. Unset: the route is off and the account events do not reach Events. The secret is 32
+# characters or more; it is the one Events signs its own deliveries with.
+#EVENTS_EVENTS_URL=https://openvibe.events
+#EVENTS_EVENTS_SECRET=
#OV_BILLING_AUDIENCE=openvibe.billing
#OV_BILLING_AUDIENCE=openvibe.billing
#OV_BILLING_TIMEOUT_MS=10000
#OV_BILLING_TIMEOUT_MS=10000
@@ -120,6 +120,15 @@ Operators: `GET /api/v1/deliveries?status=dead` is the dead-letter queue; `POST
@@ -120,6 +120,15 @@ Operators: `GET /api/v1/deliveries?status=dead` is the dead-letter queue; `POST
Pull consumers: start with `GET /api/v1/events?topic=media.vod.*&limit=100`, then request `GET /api/v1/events?topic=media.vod.*&after=<next_cursor>&limit=100`, using the previous response's opaque `next_cursor` ([ADR-042](https://github.com/OpenVibers/OpenVibe.Contracts/blob/main/docs/adr/ADR-042-events-fabric.md) decision 7). The response carries `{ events: [{ cursor, event }], next_cursor, latest_cursor }` (`latest_cursor` is the head as an opaque cursor, for a consumer that starts at "now" without history), plus `gap: { from_seq, to_seq }` when the position is older than both the hot store and the replay tier (below) or from another retention epoch. Use `next_cursor` even when no events match, since it advances past scanned events. A position is only ever an opaque cursor: `after_seq` is refused with 400 (never read as "from the start"), and there is no per-event sequence number (dedupe on `event.event_id`). `PUT /api/v1/checkpoints { topic, cursor, carrier? }` stores a consumer's cursor here if it has nowhere better (a cursor string; a number is refused); `GET /api/v1/checkpoints?topic=` and the PUT answer `{ consumer, topic, cursor, carrier, updated_at }`, with `cursor` as stored (null when none is).
Pull consumers: start with `GET /api/v1/events?topic=media.vod.*&limit=100`, then request `GET /api/v1/events?topic=media.vod.*&after=<next_cursor>&limit=100`, using the previous response's opaque `next_cursor` ([ADR-042](https://github.com/OpenVibers/OpenVibe.Contracts/blob/main/docs/adr/ADR-042-events-fabric.md) decision 7). The response carries `{ events: [{ cursor, event }], next_cursor, latest_cursor }` (`latest_cursor` is the head as an opaque cursor, for a consumer that starts at "now" without history), plus `gap: { from_seq, to_seq }` when the position is older than both the hot store and the replay tier (below) or from another retention epoch. Use `next_cursor` even when no events match, since it advances past scanned events. A position is only ever an opaque cursor: `after_seq` is refused with 400 (never read as "from the start"), and there is no per-event sequence number (dedupe on `event.event_id`). `PUT /api/v1/checkpoints { topic, cursor, carrier? }` stores a consumer's cursor here if it has nowhere better (a cursor string; a number is refused); `GET /api/v1/checkpoints?topic=` and the PUT answer `{ consumer, topic, cursor, carrier, updated_at }`, with `cursor` as stored (null when none is).
+### Account export and deletion (ADR-033)
+
+Events is the bus, so two of its columns name a person: the `actor` (the SubjectRef of whoever caused the event) and the `subject` (what the event is about). `server/account-data.js` answers the two account topics for them:
+
+- **Export** (`network.account.export_requested`): the person's part carries the envelope of the events they caused and the events about them (`id`, `seq`, `event_type`, `source`, `subject_type`, `subject_id`, `visibility`, `project_id`, `occurred_at`, `received_at`) — never the payload, which is the publishing service's record and may name other people. The replay tier's rows are in the part too (`events-replay.json`).
+- **Deletion** (`network.account.deleted`): the events the person caused and the events about them become tombstones, exactly as a redaction does: the payload goes and the actor becomes the producer, so no person stays linked to them. The envelope keeps its id, seq, type, subject and timestamps, so the log stays continuous and the replay tier stays in step. Both tiers are answered.
+
+The route is `POST /internal/events`, signed (v2) and loopback only, mounted by `server/app.js`; `server/index.js` creates the two subscriptions at boot with `startSubscriptions` (`EVENTS_EVENTS_URL`, `EVENTS_EVENTS_SECRET` and the service's own OAuth client). Without the secret the route is off and the account events do not reach Events. The receipts of what it applied are in `account_data_events` (migration 0010), so a redelivered export or deletion changes nothing. `test/account-data.test.js` asserts `uncovered()` is empty, so a new table with a person column fails CI until it is covered or exempted with a reason.
+
### Retention: three tiers
### Retention: three tiers
[ADR-042](https://github.com/OpenVibers/OpenVibe.Contracts/blob/main/docs/adr/ADR-042-events-fabric.md) decision 8. An event lives in one tier at a time:
[ADR-042](https://github.com/OpenVibers/OpenVibe.Contracts/blob/main/docs/adr/ADR-042-events-fabric.md) decision 8. An event lives in one tier at a time:
@@ -398,6 +407,6 @@ Part of the [OpenVibe network](https://openvibe.network). Built in the open by [
@@ -398,6 +407,6 @@ Part of the [OpenVibe network](https://openvibe.network). Built in the open by [
<!-- versions:start -->
<!-- versions:start -->
- openvibe-contracts: v0.222.0
- openvibe-contracts: v0.222.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,7 +8,7 @@
@@ -8,7 +8,7 @@
"port": 4300,
"port": 4300,
"domain": "openvibe.events",
"domain": "openvibe.events",
"contracts": "openvibe-contracts v0.222.0",
"contracts": "openvibe-contracts v0.222.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",
"capabilities": [
"capabilities": [
"events.event.publish",
"events.event.publish",
new file mode 100644
new file mode 100644
@@ -0,0 +1,13 @@
@@ -0,0 +1,13 @@
+-- phase: expand
+-- OpenVibe.Events: 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
+);
@@ -13,7 +13,7 @@
@@ -13,7 +13,7 @@
"express": "^4.21.2",
"express": "^4.21.2",
"iovalkey": "^0.4.0",
"iovalkey": "^0.4.0",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.222.0",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.222.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"
},
},
@@ -1119,9 +1119,9 @@
@@ -1119,9 +1119,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"
@@ -25,7 +25,7 @@
@@ -25,7 +25,7 @@
"express": "^4.21.2",
"express": "^4.21.2",
"iovalkey": "^0.4.0",
"iovalkey": "^0.4.0",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.222.0",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.222.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,103 @@
@@ -0,0 +1,103 @@
+'use strict';
+/**
+ * Account export and deletion → Events (ADR-033; openvibe-sdk/account-data).
+ *
+ * Events is the bus: it stores the events the services publish, and two of its columns name a person — the actor (the
+ * SubjectRef of whoever caused the event) and the subject (what the event is about). The payload is the publishing
+ * service's record, and the platform's redaction mechanism (server/redaction.js) is what takes a copy of it back.
+ *
+ * export the events the person caused and the events about them: the envelope (id, type, source, subject,
+ * visibility, timestamps), never the payload — that is the publisher's record and may name other people.
+ * erase the events the person caused and the events about them become tombstones, exactly as a redaction does:
+ * the payload goes and the actor becomes the producer, so no person stays linked to them. The envelope keeps
+ * its id, seq, type, subject and timestamps, so the log stays continuous and the replay tier stays in step.
+ * Both tiers are answered: the hot store and the replay archive (its body holds the actor, so its rows are
+ * opened to find it).
+ *
+ * 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 archive = require('./archive');
+const redaction = require('./redaction');
+
+const TABLES = [];
+
+// The envelope the person gets: the event's place in the log, not what it carried.
+const ENVELOPE = ['id', 'seq', 'event_type', 'source', 'subject_type', 'subject_id', 'visibility', 'project_id', 'occurred_at', 'received_at'];
+const ROW_LIMIT = 5000;
+
+/** The person in a SubjectRef JSON column: {type: 'user', id: 'usr_…'}. */
+function actorIs(row, subject) {
+ if (!row || !row.actor) return false;
+ let a = null;
+ try { a = typeof row.actor === 'string' ? JSON.parse(row.actor) : row.actor; } catch { return false; }
+ return Boolean(a) && a.type === 'user' && a.id === subject;
+}
+
+/** The event is about the person. */
+function subjectIs(row, subject) {
+ return row && row.subject_type === 'user' && row.subject_id === subject;
+}
+
+/** The events that name the person, in the hot store. */
+async function hotRows(db, subject) {
+ return await db.many(
+ `SELECT ${ENVELOPE.join(', ')} FROM events
+ WHERE (actor::jsonb ->> 'type' = 'user' AND actor::jsonb ->> 'id' = $1) OR (subject_type = 'user' AND subject_id = $1)
+ ORDER BY received_at DESC LIMIT ${ROW_LIMIT}`, [subject]);
+}
+
+/** The events that name the person, in the replay tier: the actor lives in the packed body, so the rows are opened. */
+async function archivedRows(db, subject) {
+ const out = [];
+ for (const a of await db.many('SELECT * FROM events_archive WHERE redacted_at IS NULL')) {
+ const r = archive.unpack(a);
+ if (actorIs(r, subject) || subjectIs(r, subject)) out.push(r);
+ }
+ return out;
+}
+
+/** The person's part of an account export: the envelope of the events they caused and the events about them. */
+async function extraExport(db, subject) {
+ const hot = await hotRows(db, subject);
+ const files = [];
+ if (hot.length) files.push({ name: 'events.json', content: hot });
+ const archived = await archivedRows(db, subject);
+ if (archived.length) {
+ files.push({
+ name: 'events-replay.json',
+ content: archived.slice(0, ROW_LIMIT).map((r) => Object.fromEntries(ENVELOPE.map((c) => [c, r[c] === undefined ? null : r[c]]))),
+ });
+ }
+ return files;
+}
+
+/** Turn the person's events into tombstones, as a redaction does: the payload goes, the actor becomes the producer. */
+async function extraErase(t, subjects, counts) {
+ const now = Date.now();
+ for (const subject of subjects) {
+ const by = `account-deletion:${subject}`;
+ const hot = await t.many('SELECT * FROM events WHERE (actor::jsonb ->> \'type\' = \'user\' AND actor::jsonb ->> \'id\' = $1) OR (subject_type = \'user\' AND subject_id = $1) FOR UPDATE', [subject]);
+ for (const r of hot) {
+ const cols = redaction.tombstone(r, { by, at: now });
+ await t.exec('UPDATE events SET payload = $1, actor = $2, size_bytes = $3, redacted_at = $4, redacted_by = $5 WHERE id = $6',
+ [cols.payload, cols.actor, cols.size_bytes, cols.redacted_at, cols.redacted_by, r.id]);
+ }
+ counts.add(counts.erased, 'events', hot.length);
+ // The replay tier: the body holds the actor, so its rows are opened, tombstoned and packed again.
+ const archived = await archivedRows(t, subject);
+ for (const r of archived) {
+ const cols = redaction.tombstone(r, { by, at: now });
+ // The replay tier keeps the redaction's who in the packed body; the table itself has only the marker.
+ await t.exec('UPDATE events_archive SET body = $1, redacted_at = $2 WHERE id = $3',
+ [archive.pack({ ...r, ...cols }), cols.redacted_at, r.id]);
+ }
+ counts.add(counts.erased, 'events_replay', archived.length);
+ }
+}
+
+function create({ db, log = console } = {}) {
+ return createAccountData({ db, service: 'events', tables: TABLES, extraExport, extraErase, log });
+}
+
+module.exports = { create, TABLES, TOPICS, extraExport, extraErase };
@@ -24,7 +24,7 @@ const openapi = require('./openapi');
@@ -24,7 +24,7 @@ const openapi = require('./openapi');
const ovServe = require('openvibe-shared/serve');
const ovServe = require('openvibe-shared/serve');
const pkg = require('../package.json');
const pkg = require('../package.json');
-function createApp({ config, store, auth, keys, worker, realtime, metrics, dnsLookup, clock, log = console, valkey = null, policies = null, planner = null, carriers = null, indexnow = null }) {
+function createApp({ config, store, auth, keys, worker, realtime, metrics, dnsLookup, clock, log = console, valkey = null, policies = null, planner = null, carriers = null, indexnow = null, accountData = null, accountSend = null }) {
const app = express();
const app = express();
app.disable('x-powered-by');
app.disable('x-powered-by');
app.set('trust proxy', 'loopback');
app.set('trust proxy', 'loopback');
@@ -48,6 +48,10 @@ function createApp({ config, store, auth, keys, worker, realtime, metrics, dnsLo
@@ -48,6 +48,10 @@ function createApp({ config, store, auth, keys, worker, realtime, metrics, dnsLo
// routers and outside the per-actor limiter: it is static and unauthenticated.
// routers and outside the per-actor limiter: it is static and unauthenticated.
app.use('/shared', ovServe.handler());
app.use('/shared', ovServe.handler());
+ // The account export/deletion consumer (ADR-033, ./account-data.js): the two account topics, signed and loopback
+ // only. No body parser before it: the v2 signature covers the raw body.
+ if (accountData) app.post('/internal/events', accountData.consumer({ secrets: [config.accountData.secret].filter((s) => s.length >= 32), send: accountSend }));
+
// Realtime first: it answers CORS preflights and must not go through the JSON body parser.
// Realtime first: it answers CORS preflights and must not go through the JSON body parser.
app.options('/realtime/stream', realtime.cors);
app.options('/realtime/stream', realtime.cors);
app.get('/realtime/stream', realtime.cors, realtime.handler);
app.get('/realtime/stream', realtime.cors, realtime.handler);
@@ -231,6 +231,20 @@ function load(env = process.env) {
@@ -231,6 +231,20 @@ function load(env = process.env) {
timeoutMs: Math.max(1000, int(env.OV_BILLING_TIMEOUT_MS, 10000)),
timeoutMs: Math.max(1000, int(env.OV_BILLING_TIMEOUT_MS, 10000)),
},
},
+ // The service's own OAuth client at OpenVibe.Network: the account-data consumer signs its part and its
+ // confirmation with it, and its Events subscription is created with it.
+ oauth: {
+ clientId: env.OV_OAUTH_CLIENT_ID || 'events',
+ clientSecret: env.OV_OAUTH_CLIENT_SECRET || '',
+ },
+
+ // The account-data consumer (ADR-033): Events' own subscription to the two account topics, delivered back to
+ // POST /internal/events. Unset: the route is off and the account events do not reach Events.
+ accountData: {
+ url: (env.EVENTS_EVENTS_URL || baseUrl).replace(/\/+$/, ''),
+ secret: String(env.EVENTS_EVENTS_SECRET || '').trim(),
+ },
+
// Realtime (SSE) gateway
// Realtime (SSE) gateway
realtime: {
realtime: {
maxConnections: int(env.REALTIME_MAX_CONNECTIONS, 2000),
maxConnections: int(env.REALTIME_MAX_CONNECTIONS, 2000),
@@ -17,6 +17,8 @@ const { createAuth } = require('./auth');
@@ -17,6 +17,8 @@ const { createAuth } = require('./auth');
const { createWorker } = require('./worker');
const { createWorker } = require('./worker');
const { createRealtime } = require('./realtime');
const { createRealtime } = require('./realtime');
const { createApp } = require('./app');
const { createApp } = require('./app');
+const accountDataLib = require('./account-data');
+const { createNetworkSender, startSubscriptions } = require('openvibe-sdk/account-data');
const { createMetrics } = require('./metrics');
const { createMetrics } = require('./metrics');
const { createGuardedPost } = require('./egress');
const { createGuardedPost } = require('./egress');
const { createBillingReadings } = require('./billing');
const { createBillingReadings } = require('./billing');
@@ -63,7 +65,13 @@ async function start({
@@ -63,7 +65,13 @@ async function start({
indexnow = createIndexNow({ host: config.siteUrl, key: '', log });
indexnow = createIndexNow({ host: config.siteUrl, key: '', log });
}
}
- const app = createApp({ config, store, auth, keys, worker, realtime, metrics, dnsLookup, clock, log, valkey, policies, planner, carriers, indexnow });
+ // The account export/deletion part (ADR-033, ./account-data.js): the handle over this database and the sender
+ // that pushes the part and the confirmation to Network's internal routes.
+ const accountData = accountDataLib.create({ db, log });
+ const accountSend = config.oauth && config.oauth.clientSecret
+ ? createNetworkSender({ networkInternalUrl: config.networkInternalUrl, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, fetch: fetchImpl })
+ : null;
+ const app = createApp({ config, store, auth, keys, worker, realtime, metrics, dnsLookup, clock, log, valkey, policies, planner, carriers, indexnow, accountData, accountSend });
app.locals.indexnow = indexnow;
app.locals.indexnow = indexnow;
// The pages crawlers see change only with a release, so this boot is the moment to announce them.
// The pages crawlers see change only with a release, so this boot is the moment to announce them.
// A no-op without INDEXNOW_KEY.
// A no-op without INDEXNOW_KEY.
@@ -88,6 +96,16 @@ async function start({
@@ -88,6 +96,16 @@ async function start({
log.warn(`[events] carriers did not start: ${err.message}`);
log.warn(`[events] carriers did not start: ${err.message}`);
}
}
+ // Events' own subscription to the two account topics (ADR-033): the deliveries come back to
+ // POST /internal/events. Off without EVENTS_EVENTS_SECRET or the OAuth client secret.
+ const accountSubscription = (config.accountData.secret.length >= 32 && config.oauth && config.oauth.clientSecret)
+ ? startSubscriptions({
+ eventsUrl: config.accountData.url, endpoint: `http://127.0.0.1:${config.port}/internal/events`,
+ secret: config.accountData.secret, networkInternalUrl: config.networkInternalUrl,
+ clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, fetch: fetchImpl, log,
+ })
+ : null;
+
const prune = async () => {
const prune = async () => {
try {
try {
// Tiers (ADR-042 decision 8): hot events past EVENTS_RETENTION_DAYS move to the replay tier, which keeps them
// Tiers (ADR-042 decision 8): hot events past EVENTS_RETENTION_DAYS move to the replay tier, which keeps them
@@ -142,6 +160,7 @@ async function start({
@@ -142,6 +160,7 @@ async function start({
}
}
async function close() {
async function close() {
+ if (accountSubscription) accountSubscription.stop();
clearInterval(pruneTimer);
clearInterval(pruneTimer);
if (usageTimer) clearInterval(usageTimer);
if (usageTimer) clearInterval(usageTimer);
await billing.stop();
await billing.stop();
new file mode 100644
new file mode 100644
@@ -0,0 +1,87 @@
@@ -0,0 +1,87 @@
+'use strict';
+/**
+ * ADR-033: Events' part of an account export and deletion. The events a person caused and the events about them
+ * are theirs: the export carries their envelope (never the payload, which is the publishing service's record) and a
+ * deletion turns them into tombstones, so the payload goes and the actor becomes the producer. Both tiers are
+ * answered: the hot store and the replay archive.
+ *
+ * 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('./db');
+const { TABLES, create } = require('../server/account-data');
+const archive = require('../server/archive');
+
+const A = 'usr_01JZ0000000000000000000AAA';
+const B = 'usr_01JZ0000000000000000000BBB';
+
+let seq = 0;
+const nextSeq = () => ++seq;
+
+/** One event in the hot store. */
+async function insertEvent(db, { id, actor, subjectType, subjectId, payload }) {
+ await db.exec(`INSERT INTO events (id, seq, event_type, version, source, actor, subject_type, subject_id, trace_id, priority, visibility, occurred_at, received_at, payload, hops, publisher)
+ VALUES ($1, $2, 'chat.message.created', 1, 'chat', $3, $4, $5, 'trace-1', 'important', 'internal', '2026-10-01T00:00:00.000Z', 1, $6, 1, 'chat')`,
+ [id, nextSeq(), JSON.stringify(actor), subjectType, subjectId, payload]);
+}
+
+/** One event in the replay tier: the actor and the payload live in the packed body. */
+async function insertArchived(db, { id, actor, subjectType, subjectId, payload }) {
+ const row = {
+ id, seq: nextSeq(), event_type: 'chat.message.created', source: 'chat', subject_type: subjectType, subject_id: subjectId,
+ visibility: 'internal', project_id: null, env: 'production', received_at: 1, redacted_at: null, redacts: 0,
+ version: 1, actor: JSON.stringify(actor), subject_revision: null, trace_id: 'trace-1', priority: 'important',
+ occurred_at: '2026-10-01T00:00:00.000Z', payload, hops: 1, publisher: 'chat', request_id: null, size_bytes: 10, redacted_by: null,
+ };
+ await db.exec(`INSERT INTO events_archive (id, seq, epoch, event_type, source, subject_type, subject_id, visibility, project_id, env, received_at, redacts, body, archived_at)
+ VALUES ($1, $2, 1, $3, $4, $5, $6, $7, null, 'production', 1, 0, $8, 1)`,
+ [id, row.seq, row.event_type, row.source, row.subject_type, row.subject_id, row.visibility, archive.pack(row)]);
+}
+
+(async () => {
+ const d = await testDb();
+ try {
+ const accountData = create({ db: d.db });
+ const missing = await uncovered(d.db, {
+ tables: [...TABLES, 'events'],
+ exempt: { account_data_events: 'the ledger of requests itself: which export and deletion Events applied' },
+ });
+ assert.deepStrictEqual(missing, [], `a table with a person column is neither covered nor exempted: ${missing.join(', ')}`);
+
+ // Ada caused an event (her chat message) and is the subject of another (a notification for her).
+ await insertEvent(d.db, { id: 'evt_01JZ0000000000000000000AAA', actor: { type: 'user', id: A }, subjectType: 'chat_message', subjectId: 'cm_1', payload: '{"message":"hello"}' });
+ await insertEvent(d.db, { id: 'evt_01JZ0000000000000000000AAB', actor: { type: 'service', id: 'network' }, subjectType: 'user', subjectId: A, payload: '{"title":"a notification"}' });
+ await insertEvent(d.db, { id: 'evt_01JZ0000000000000000000AAC', actor: { type: 'user', id: B }, subjectType: 'chat_message', subjectId: 'cm_2', payload: '{"message":"hi"}' });
+ await insertArchived(d.db, { id: 'evt_01JZ0000000000000000000AAD', actor: { type: 'user', id: A }, subjectType: 'chat_message', subjectId: 'cm_3', payload: '{"message":"older"}' });
+
+ const part = await accountData.exportPart(A);
+ const file = part.files.find((f) => f.name === 'events.json');
+ assert.ok(file, 'events.json is in the part');
+ assert.deepStrictEqual(file.content.map((e) => e.id).sort(), ['evt_01JZ0000000000000000000AAA', 'evt_01JZ0000000000000000000AAB']);
+ assert.ok(!JSON.stringify(part).includes('hello'), 'the payload is never exported');
+ assert.ok(!JSON.stringify(part).includes('evt_01JZ0000000000000000000AAC'), 'nobody else\'s events');
+ const replay = part.files.find((f) => f.name === 'events-replay.json');
+ assert.ok(replay, 'the replay tier is in the part');
+ assert.deepStrictEqual(replay.content.map((e) => e.id), ['evt_01JZ0000000000000000000AAD']);
+
+ const counts = await accountData.erase([A]);
+ assert.strictEqual(counts.erased.events, 2, 'her two events are tombstoned');
+ assert.strictEqual(counts.erased.events_replay, 1, 'her replay event too');
+ const hot = await d.db.many('SELECT id, payload, actor, redacted_at FROM events WHERE id IN ($1, $2, $3) ORDER BY id',
+ ['evt_01JZ0000000000000000000AAA', 'evt_01JZ0000000000000000000AAB', 'evt_01JZ0000000000000000000AAC']);
+ const ada = hot.find((e) => e.id === 'evt_01JZ0000000000000000000AAA');
+ assert.ok(ada.redacted_at, 'her event is a tombstone');
+ assert.strictEqual(JSON.parse(ada.payload).redacted, true, 'the payload is gone');
+ assert.strictEqual(JSON.parse(ada.actor).type, 'service', 'no person stays linked to it');
+ assert.ok(!JSON.stringify(hot).includes('hello'), 'her message is not in the log any more');
+ const bo = hot.find((e) => e.id === 'evt_01JZ0000000000000000000AAC');
+ assert.strictEqual(bo.redacted_at, null, 'Bo\'s event stays');
+ assert.ok(JSON.parse(bo.actor).id === B, 'Bo stays in his own event');
+ const archived = await d.db.many('SELECT body FROM events_archive WHERE id = $1', ['evt_01JZ0000000000000000000AAD']);
+ const unpacked = archive.unpack(archived[0]);
+ assert.strictEqual(JSON.parse(unpacked.payload).redacted, true, 'the replay copy is tombstoned too');
+ assert.strictEqual(JSON.parse(unpacked.actor).type, 'service', 'no person stays linked to the replay copy');
+ console.log(' ✓ export, tombstone and the guard');
+ } finally { await d.close(); }
+})().catch((err) => { console.error(err); process.exit(1); });
@@ -31,6 +31,9 @@ const PLANTED = {
@@ -31,6 +31,9 @@ const PLANTED = {
// cloud-v1 (an SQS-compatible queue): signs requests, never echoed.
// cloud-v1 (an SQS-compatible queue): signs requests, never echoed.
CLOUD_QUEUE_ACCESS_KEY: `ovsecret-events-cloud-access-${crypto.randomBytes(12).toString('hex')}`,
CLOUD_QUEUE_ACCESS_KEY: `ovsecret-events-cloud-access-${crypto.randomBytes(12).toString('hex')}`,
CLOUD_QUEUE_SECRET_KEY: `ovsecret-events-cloud-secret-${crypto.randomBytes(12).toString('hex')}`,
CLOUD_QUEUE_SECRET_KEY: `ovsecret-events-cloud-secret-${crypto.randomBytes(12).toString('hex')}`,
+ // The account-data consumer (ADR-033): the secret its own loopback subscription signs the two
+ // account topics with. It is Events' own signing secret, never echoed.
+ EVENTS_EVENTS_SECRET: `ovsecret-events-account-${crypto.randomBytes(12).toString('hex')}`,
};
};
t('Events reads no secret from its environment but the planted ones', async () => {
t('Events reads no secret from its environment but the planted ones', async () => {