openvibers/openvibe.events Commits

Redaction: a producer takes back its own events. Any event may carry payload.redacts ({ event_ids } and/or { subject_type, subject_ids }, up to 1000 each); in the transaction that stores it, every target of the same source (an app: same project and env) becomes a tombstone at its original seq: payload { redacted, redacted_at, redacted_by }, actor the producer itself, envelope otherwise kept. Pull, GET by id, SSE replay (anonymous, signed in, service), queued deliveries and DLQ replays all serve the tombstone; first-party consumers get it and then the deletion event. Naming another source's event is 403 events.redaction_not_allowed (whole batch refused), a malformed directive 422 events.invalid_redaction; no capability beyond events.event.publish. secure_delete on, so old payloads are zeroed on disk. Browsers are replayed public events of the last REALTIME_PUBLIC_REPLAY_SECONDS (300) only, older ones are `event: gap` reason public_window (own subject events and services keep full retention). scripts/redact-backfill.js redacts chat events of messages Chat deleted before it published chat.message.deleted (dry run by default; --apply needs a verified 0600 --backup). Fixes: deleted chat messages (text, anon id) stayed replayable anonymously for 30 days

OP OpenVibers committed 23 Sept 2026 · 18b873f646 · 11 files, +818 −8


Changes

ddiff --git a/.env.example b/.env.example
iindex fe24d6a..e5856b0 100644
---- a/.env.example
++++ b/.env.example
@@@ -74,6 +74,10 @@ EVENTS_RECEIPT_RETENTION_DAYS=90
#REALTIME_MAX_TOPICS=20
#REALTIME_HEARTBEAT_MS=25000
#REALTIME_REPLAY_MAX=1000
++# A browser resuming with Last-Event-ID is replayed public events received this recently only
++# (0: none); older public events come as `event: gap` (reason public_window). Subject events
++# addressed to the viewer, and service viewers, keep the whole retention.
++#REALTIME_PUBLIC_REPLAY_SECONDS=300
# Signed-out browsers may receive public events.
#REALTIME_ALLOW_ANONYMOUS=true
# Exact extra CORS origins besides https://*.openvibe.* (development).
ddiff --git a/README.md b/README.md
iindex be87fda..362b8d9 100644
---- a/README.md
++++ b/README.md
@@@ -53,6 +53,23 @@ Errors are RFC 9457 problem+json (`openvibe-contracts` `http.problem`) with a st
- Loop guard: an event whose trace already carries 8 hops (the cross-service chain depth, or the same source/type/subject repeating in the trace) is refused with `409 events.loop_detected`. The chain depth of a first-party publish counts first-party events only (developer-app events in the same trace never count toward it), so an app cannot poison a trace it has seen.
- Each accepted event gets a global `seq` and is committed with one delivery row per matching subscription before anything is sent.
++### Redaction
++
++A producer takes back what it published (a deleted chat message) with a directive in the payload of any event it publishes, normally its own `*.deleted` event ([server/redaction.js](server/redaction.js)):
++
++```json
++{ "event_type": "chat.message.deleted", "source": "chat", "visibility": "public",
++ "subject": { "type": "chat_message", "id": "123" },
++ "payload": { "message_ids": [123], "redacts": { "subject_type": "chat_message", "subject_ids": ["123"] } } }
++```
++
++- `redacts.event_ids` (up to 1000 `evt_…`) names events directly; `subject_type` + `subject_ids` (up to 1000) names every stored event of the same source about those subjects.
++- In the transaction that stores the directive, each target becomes a **tombstone**: `payload` is replaced by `{ "redacted": true, "redacted_at": "<ISO>", "redacted_by": "<the redacting event_id>" }` and `actor` by the producer itself (`{ "type": "service", "id": "<source>" }`, or the app). `event_id`, `seq`, `event_type`, `subject`, `timestamp` and `visibility` stay, so sequences have no holes. The old payload is overwritten on disk (`secure_delete`), not just hidden.
++- Every read path serves the tombstone from then on: pull, `GET /api/v1/events/:id`, SSE replay (anonymous, signed in or service), queued deliveries and DLQ replays. First-party consumers get the tombstone at the original seq and then the deletion event itself; they should check `payload.redacted`.
++- Authority is the publish rule: only the owning source (for an app, its own project and environment) can redact. Naming another source's event by id refuses the whole publish with `403 events.redaction_not_allowed`; a subject match never reaches another source. A malformed directive is `422 events.invalid_redaction`. Events carrying a directive are never redacted themselves. No capability beyond `events.event.publish` is needed.
++- Redaction applies to what is stored when the directive arrives; it does not block later events about the same subject.
++- Events stored before a producer published deletions: [scripts/redact-backfill.js](scripts/redact-backfill.js) (chat only; dry run by default, `--apply` needs `--backup <new file>`).
++
## Subscriptions and delivery
```http
@@@ -156,6 +173,8 @@ es.addEventListener('gap', (m) => { /* events were missed: refetch state */ });
- Auth: the Network `ov_token` cookie or a Bearer user JWT; a service token with `events.event.read`; or nobody (public events only, `REALTIME_ALLOW_ANONYMOUS`). An expired cookie degrades to anonymous; a bad Bearer is a 401.
- Visibility: `public` events go to anyone subscribed to the topic; `subject` events only to the user whose subject id (`usr_…`) is the event's `actor.id` or its user `subject.id`; `internal` events never reach a browser. A guessed topic yields nothing.
- Resume: the SSE `id` is the seq, so the browser's automatic `Last-Event-ID` (or `?last_event_id=`) replays what was missed. A cursor older than retention first gets `event: gap` (`{ reason, from_seq, to_seq }`), as does a replay that hits `REALTIME_REPLAY_MAX`.
++- Public replay window: browsers (signed out or signed in) are replayed `public` events received in the last `REALTIME_PUBLIC_REPLAY_SECONDS` (300; 0 = none), enough to ride out a reconnect. Older public events are not replayed: the stream opens with `event: gap` (`reason: "public_window"`, `window_seconds`), and the client refetches state from the owning service. `subject` events addressed to the viewer, and service viewers, keep the whole retention. This keeps the stream from paging through a month of public history, chat lines included; nothing in the network needs more (no browser surface replays public events today).
++- Redacted events are replayed as their tombstones ([Redaction](#redaction)).
- Heartbeat comment every 25 s; at most 20 topics per connection and `REALTIME_MAX_CONNECTIONS` (2000) overall; CORS with credentials for `https://*.openvibe.*` only.
## Owns
ddiff --git a/scripts/redact-backfill.js b/scripts/redact-backfill.js
nnew file mode 100644
iindex 0000000..8187c75
---- /dev/null
++++ b/scripts/redact-backfill.js
@@@ -0,0 +1,170 @@
++#!/usr/bin/env node
++/**
++ * One-off backfill: redact the stored chat events of messages OpenVibe.Chat deleted before Chat
++ * published chat.message.deleted (server/redaction.js does it for every deletion since).
++ *
++ * node scripts/redact-backfill.js --chat-db <chat.db> # dry run: counts only
++ * node scripts/redact-backfill.js --chat-db <chat.db> --apply --backup <new file> # backup, then redact
++ *
++ * Options:
++ * --chat-db <file> Chat's database, opened read-only (production: /var/lib/openvibe-chat/chat.db)
++ * --db <file> the Events database (default: $EVENTS_DB_PATH, else data/events.db; relative
++ * paths from the repo root; production: /var/lib/openvibe-events/events.db)
++ * --batch <n> messages per write transaction (default 500, at most 1000)
++ *
++ * A chat event is redacted when it is a first-party `chat` event about a `chat_message` subject
++ * that Chat marks deleted (is_deleted = 1, or an auto-delete time that has passed). It becomes the
++ * same tombstone a live chat.message.deleted produces (payload { redacted, redacted_at,
++ * redacted_by: "backfill" }, actor service:chat), at the same seq. Messages Chat does not have at
++ * all are counted and left alone. Idempotent: redacted events are never selected again, so a
++ * second run changes nothing.
++ *
++ * --apply refuses to run without --backup <file>: a new file, mode 0600, an sqlite online backup of
++ * the whole Events database, verified with quick_check and a row count before anything is changed.
++ * That backup still holds the text being removed: delete it once the run is verified. Safe while
++ * Events is up (short transactions; secure_delete zeroes the old payloads). Run as the service
++ * user, with the Events release that knows redaction deployed (it adds the columns).
++ */
++'use strict';
++const fs = require('fs');
++const path = require('path');
++const Database = require('better-sqlite3');
++const { createStore } = require('../server/store');
++
++const ROOT = path.join(__dirname, '..');
++const HELP_LINES = 26; // header lines 3..26
++const BY = 'backfill';
++
++function parseArgs(argv) {
++ const a = { apply: false, backup: null, chatDb: null, db: null, batch: 500 };
++ for (let i = 0; i < argv.length; i++) {
++ const k = argv[i];
++ const val = () => { if (i + 1 >= argv.length) throw new Error(`${k} needs a value`); return argv[++i]; };
++ if (k === '--apply') a.apply = true;
++ else if (k === '--backup') a.backup = val();
++ else if (k === '--chat-db') a.chatDb = val();
++ else if (k === '--db') a.db = val();
++ else if (k === '--batch') a.batch = Number(val());
++ else if (k === '-h' || k === '--help') a.help = true;
++ else throw new Error(`unknown option ${k}`);
++ }
++ if (!Number.isInteger(a.batch) || a.batch < 1 || a.batch > 1000) throw new Error('--batch must be 1..1000');
++ return a;
++}
++
++/** The database the server uses: --db, else $EVENTS_DB_PATH, else data/events.db (relative to the repo root). */
++function defaultDbPath(args, env = process.env) {
++ return path.resolve(ROOT, args.db || env.EVENTS_DB_PATH || path.join('data', 'events.db'));
++}
++
++const mb = (n) => `${(n / 1048576).toFixed(1)} MB`;
++function fileBytes(f) { try { return fs.statSync(f).size; } catch { return 0; } }
++function freeBytes(dir) { try { const s = fs.statfsSync(dir); return s.bavail * s.bsize; } catch { return null; } }
++const chunks = (list, n) => Array.from({ length: Math.ceil(list.length / n) }, (_, i) => list.slice(i * n, (i + 1) * n));
++
++/** Unredacted first-party chat events about chat messages: [{ id, seq, subject_id }]. */
++function candidates(db) {
++ return db.prepare(`SELECT id, seq, subject_id FROM events
++ WHERE source = 'chat' AND subject_type = 'chat_message' AND project_id IS NULL AND redacted_at IS NULL AND redacts = 0
++ ORDER BY seq`).all();
++}
++
++/** Split message ids into those Chat marks deleted and those it does not have. */
++function classify(chat, messageIds) {
++ const deleted = new Set();
++ const present = new Set();
++ for (const part of chunks(messageIds, 500)) {
++ const marks = part.map(() => '?').join(',');
++ for (const r of chat.prepare(`SELECT id, is_deleted,
++ (auto_delete_at IS NOT NULL AND auto_delete_at <= CURRENT_TIMESTAMP) AS expired
++ FROM chat_messages WHERE id IN (${marks})`).all(...part.map(Number))) {
++ present.add(String(r.id));
++ if (r.is_deleted || r.expired) deleted.add(String(r.id));
++ }
++ }
++ return { deleted: [...deleted], missing: messageIds.filter(id => !present.has(id)) };
++}
++
++async function main(argv, log = console.log) {
++ let args;
++ try { args = parseArgs(argv); } catch (e) { log(`error: ${e.message}`); return 2; }
++ if (args.help) { log(fs.readFileSync(__filename, 'utf8').split('\n').slice(2, HELP_LINES).join('\n')); return 0; }
++ if (!args.chatDb) { log('error: --chat-db <chat.db> is required'); return 2; }
++ const dbPath = defaultDbPath(args);
++ const chatPath = path.resolve(args.chatDb);
++ for (const f of [dbPath, chatPath]) if (!fs.existsSync(f)) { log(`error: ${f} does not exist`); return 2; }
++ if (args.apply && !args.backup) {
++ log('refusing --apply without --backup <new file> (an sqlite online backup of the Events database)');
++ return 2;
++ }
++
++ const db = new Database(dbPath, args.apply ? { fileMustExist: true } : { readonly: true, fileMustExist: true });
++ const chat = new Database(chatPath, { readonly: true, fileMustExist: true });
++ try {
++ db.pragma('busy_timeout = 5000');
++ chat.pragma('busy_timeout = 5000');
++ const cols = new Set(db.prepare('PRAGMA table_info(events)').all().map(c => c.name));
++ if (!cols.has('redacted_at') || !cols.has('redacts')) {
++ log('error: this Events database has no redaction columns yet; deploy and start the Events release with redaction first');
++ return 2;
++ }
++ const rows = candidates(db);
++ const byMessage = new Map();
++ for (const r of rows) {
++ if (!byMessage.has(r.subject_id)) byMessage.set(r.subject_id, []);
++ byMessage.get(r.subject_id).push(r);
++ }
++ const { deleted, missing } = classify(chat, [...byMessage.keys()]);
++ const toRedact = deleted.flatMap(id => byMessage.get(id)).sort((x, y) => x.seq - y.seq);
++ const walBytes = fileBytes(dbPath + '-wal');
++ log(`events db ${dbPath} (${mb(fileBytes(dbPath))} + ${mb(walBytes)} WAL)`);
++ log(`chat db ${chatPath} (read-only)`);
++ log(`chat events about messages, not redacted: ${rows.length} events, ${byMessage.size} messages`);
++ log(`deleted in Chat: ${deleted.length} messages, ${toRedact.length} events to redact${toRedact.length ? ` (seq ${toRedact[0].seq}..${toRedact.at(-1).seq})` : ''}`);
++ log(`not in Chat's database (left alone): ${missing.length} messages`);
++
++ if (!args.apply) {
++ log('\ndry run: nothing changed. Re-run with --apply --backup <new file>.');
++ return 0;
++ }
++
++ const target = path.resolve(args.backup);
++ if (fs.existsSync(target)) { log(`error: backup target ${target} already exists; choose a new file`); return 2; }
++ const need = fileBytes(dbPath) + walBytes;
++ const free = freeBytes(path.dirname(target));
++ if (free != null && free < need * 1.1) { log(`error: ${mb(free)} free at ${path.dirname(target)}, the backup needs about ${mb(need)}`); return 2; }
++ const total = db.prepare('SELECT COUNT(*) FROM events').pluck().get();
++ // A full copy of the event store, deleted text included: owner-only from the first byte.
++ const umask = process.umask(0o077);
++ try { await db.backup(target); } finally { process.umask(umask); }
++ fs.chmodSync(target, 0o600);
++ const b = new Database(target, { readonly: true });
++ try {
++ const ok = b.pragma('quick_check', { simple: true });
++ const n = b.prepare('SELECT COUNT(*) FROM events').pluck().get();
++ if (ok !== 'ok' || n < total) { log(`error: backup check failed (quick_check=${ok}, events=${n} of ${total}); nothing changed`); return 1; }
++ } finally { b.close(); }
++ log(`backup ${target} (${mb(fileBytes(target))}, mode 0600, quick_check ok). It still holds the deleted text: remove it once this run is verified.`);
++
++ db.pragma('secure_delete = ON');
++ const store = createStore(db);
++ let redacted = 0;
++ for (const part of chunks(deleted, args.batch)) {
++ redacted += store.redact('chat', { subject_type: 'chat_message', subject_ids: part }, { by: BY }).length;
++ }
++ log(`redacted ${redacted} events`);
++ try { db.pragma('wal_checkpoint(TRUNCATE)'); } catch { /* the server's next checkpoint does it */ }
++ const left = classify(chat, [...new Set(candidates(db).map(r => r.subject_id))]).deleted.length;
++ log(`now ${left} deleted messages with unredacted events`);
++ return left ? 1 : 0;
++ } finally {
++ chat.close();
++ db.close();
++ }
++}
++
++if (require.main === module) {
++ main(process.argv.slice(2)).then((code) => process.exit(code), (err) => { console.error(err); process.exit(1); });
++}
++
++module.exports = { main, parseArgs, defaultDbPath };
ddiff --git a/server/api/publish.js b/server/api/publish.js
iindex f9dd0f5..da9d109 100644
---- a/server/api/publish.js
++++ b/server/api/publish.js
@@@ -14,11 +14,16 @@
* `event_type` must start with `app.<project_key>.`, and `actor` must be the app itself or the user
* the token acts for. The event is stored with the token's project_id and env, under the project's
* publish-rate and retained-bytes quotas (429 events.quota_exceeded).
++ *
++ * Redaction (server/redaction.js): an event whose payload carries `redacts` turns the named earlier
++ * events of the same source into tombstones when it is stored. A malformed directive is 422
++ * events.invalid_redaction; naming another source's event is 403 events.redaction_not_allowed.
*/
const express = require('express');
const { validate, http } = require('openvibe-contracts');
const { StoreError } = require('../store');
const { CAPS } = require('../auth');
++const { parseDirective } = require('../redaction');
function normalize(input, ctx) {
const env = { ...input };
@@@ -43,6 +48,8 @@ function publishRouter({ config, store, auth, worker, realtime }) {
if (Buffer.byteLength(JSON.stringify(env.payload)) > config.maxPayloadBytes) {
return { status: 413, code: 'events.payload_too_large', detail: `payload is larger than ${config.maxPayloadBytes} bytes` };
}
++ const directive = parseDirective(env.payload);
++ if (directive && directive.error) return { status: 422, code: 'events.invalid_redaction', detail: directive.error };
if (principal.kind === 'app') return checkApp(env, principal);
if (env.source !== principal.service) {
return { status: 403, code: 'events.source_mismatch', detail: `source "${env.source}" is not the calling service "${principal.service}"` };
ddiff --git a/server/config.js b/server/config.js
iindex fc622dd..9199860 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -119,6 +119,8 @@ function load(env = process.env) {
maxTopics: int(env.REALTIME_MAX_TOPICS, 20),
heartbeatMs: int(env.REALTIME_HEARTBEAT_MS, 25000),
replayMax: int(env.REALTIME_REPLAY_MAX, 1000),
++ // Browsers are replayed public events this recent only (0: none); older ones are a gap.
++ publicReplaySeconds: Math.max(0, int(env.REALTIME_PUBLIC_REPLAY_SECONDS, 300)),
allowAnonymous: env.REALTIME_ALLOW_ANONYMOUS !== 'false',
// Extra exact origins allowed besides https://*.openvibe.* (dev: http://localhost:3000).
extraOrigins: list(env.REALTIME_CORS_ORIGINS, []),
ddiff --git a/server/index.js b/server/index.js
iindex 660b026..a63fd89 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -29,7 +29,7 @@ async function start({
// appPost / dnsLookup: developer-app delivery and the subscribe-time DNS check (server/egress.js);
// injectable for tests only.
const worker = createWorker({ store, config, clock, fetchImpl: deliveryFetch, appPost: appPost || createGuardedPost({ lookup: dnsLookup }), log, observe: metrics.observe });
-- const realtime = createRealtime({ store, auth, config, log });
++ const realtime = createRealtime({ store, auth, config, clock, log });
metrics.bind({ realtime });
const app = createApp({ config, store, auth, keys, worker, realtime, metrics, dnsLookup, log });
ddiff --git a/server/realtime.js b/server/realtime.js
iindex 353497f..0528578 100644
---- a/server/realtime.js
++++ b/server/realtime.js
@@@ -15,6 +15,13 @@
* Resume: `Last-Event-ID` (or ?last_event_id=) is a seq. Missed events are replayed first; when the
* cursor is older than retention an `event: gap` message comes first, so the client knows to
* refetch state. Heartbeat comment every 25 s.
++ *
++ * Public replay window: a browser (signed out or signed in) is replayed `public` events received in
++ * the last REALTIME_PUBLIC_REPLAY_SECONDS (300) only, enough to ride out a reconnect. Anything
++ * older comes as `event: gap` with reason `public_window`, so the stream cannot be used to page
++ * through a month of public history (chat lines among it). `subject` events addressed to the viewer,
++ * and service viewers, keep the whole retention. Redacted events (server/redaction.js) are
++ * replayed as their tombstones.
*/
const { http } = require('openvibe-contracts');
const topics = require('./topics');
@@@ -23,7 +30,7 @@ const { rowToEnvelope } = require('./store');
const ORIGIN_RE = /^https:\/\/(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)*openvibe\.[a-z]{2,63}$/;
const MAX_BUFFERED = 1024 * 1024; // a client this far behind is dropped; it resumes with Last-Event-ID
--function createRealtime({ store, auth, config, log = console }) {
++function createRealtime({ store, auth, config, clock = { now: () => Date.now() }, log = console }) {
const opts = config.realtime;
const conns = new Set();
let heartbeat = null;
@@@ -142,8 +149,17 @@ function createRealtime({ store, auth, config, log = console }) {
sendGap(conn, { reason: 'retention', from_seq: lastId + 1, to_seq: oldest - 1, latest_seq: latest });
cursor = oldest - 1;
}
++ // Browsers: public events older than the window are not replayed (a gap says so).
++ const publicFrom = viewer.kind === 'service' ? 0
++ : opts.publicReplaySeconds > 0 ? store.firstSeqSince(clock.now() - opts.publicReplaySeconds * 1000) : latest + 1;
++ if (cursor < publicFrom - 1) {
++ sendGap(conn, { reason: 'public_window', from_seq: cursor + 1, to_seq: publicFrom - 1, latest_seq: latest, window_seconds: opts.publicReplaySeconds });
++ // Signed out, only public events are visible: nothing before the window to scan.
++ if (viewer.kind === 'anonymous') cursor = publicFrom - 1;
++ }
++ const accept = (row) => visibleTo(viewer, row) && !(row.visibility === 'public' && row.seq < publicFrom);
const { rows, cursor: scanned } = store.scan(cursor, {
-- patterns: wanted, limit: opts.replayMax, scanMax: opts.replayMax * 50, accept: row => visibleTo(viewer, row),
++ patterns: wanted, limit: opts.replayMax, scanMax: opts.replayMax * 50, accept,
});
for (const row of rows) sendEvent(conn, row);
if (scanned < latest) {
ddiff --git a/server/redaction.js b/server/redaction.js
nnew file mode 100644
iindex 0000000..13bde16
---- /dev/null
++++ b/server/redaction.js
@@@ -0,0 +1,88 @@
++'use strict';
++/**
++ * Redaction: a producer takes back what it published earlier (a deleted chat message, a removed
++ * post). Any event may carry a directive in its payload:
++ *
++ * "payload": { …, "redacts": { "event_ids": ["evt_…"], "subject_type": "chat_message", "subject_ids": ["123"] } }
++ *
++ * `event_ids` names events directly; `subject_type` + `subject_ids` names every stored event of the
++ * same source about those subjects. Either or both, at most MAX_IDS of each. When the event is
++ * stored, every target is rewritten in the same transaction into a tombstone:
++ *
++ * payload -> { redacted: true, redacted_at, redacted_by: <the redacting event_id> }
++ * actor -> the producer itself (service:<source>, or the app), so no person stays linked to it
++ *
++ * The envelope keeps its event_id, seq, type, subject, timestamp and visibility, so sequences stay
++ * continuous: pull readers and replays get the tombstone at the same seq, and a delivery still
++ * queued (or replayed from the DLQ) sends the tombstone, never the original.
++ *
++ * Authority is the publish rule: only events of the redacting event's own source (and, for a
++ * developer app, its own project and environment) can be redacted. Naming another source's event
++ * by id refuses the whole publish (403 events.redaction_not_allowed); a subject match never reaches
++ * another source. Events that carry a directive themselves are never redacted (they hold only ids),
++ * nor is the redacting event.
++ */
++
++const MAX_IDS = 1000;
++const EVENT_ID_RE = /^evt_[0-9A-HJKMNP-TV-Z]{26}$/;
++const SUBJECT_TYPE_RE = /^[a-z][a-z0-9_]{0,63}$/;
++
++/**
++ * The directive in `payload`, normalized: { eventIds, subjectType, subjectIds }; null when there is
++ * none; { error } when it is malformed.
++ */
++function parseDirective(payload) {
++ if (!payload || typeof payload !== 'object' || !Object.prototype.hasOwnProperty.call(payload, 'redacts')) return null;
++ const r = payload.redacts;
++ if (!r || typeof r !== 'object' || Array.isArray(r)) return { error: 'payload.redacts must be an object' };
++ const extra = Object.keys(r).filter(k => !['event_ids', 'subject_type', 'subject_ids'].includes(k));
++ if (extra.length) return { error: `payload.redacts: unknown field ${extra[0]}` };
++ const eventIds = r.event_ids === undefined ? [] : r.event_ids;
++ if (!Array.isArray(eventIds) || eventIds.length > MAX_IDS || !eventIds.every(id => typeof id === 'string' && EVENT_ID_RE.test(id))) {
++ return { error: `payload.redacts.event_ids must be at most ${MAX_IDS} event ids (evt_…)` };
++ }
++ const hasType = r.subject_type !== undefined;
++ const hasIds = r.subject_ids !== undefined;
++ if (hasType !== hasIds) return { error: 'payload.redacts: subject_type and subject_ids go together' };
++ let subjectType = null;
++ let subjectIds = [];
++ if (hasType) {
++ if (typeof r.subject_type !== 'string' || !SUBJECT_TYPE_RE.test(r.subject_type)) return { error: 'payload.redacts.subject_type must be a subject type ([a-z][a-z0-9_]*)' };
++ if (!Array.isArray(r.subject_ids) || r.subject_ids.length > MAX_IDS
++ || !r.subject_ids.every(id => typeof id === 'string' && id.length > 0 && id.length <= 200)) {
++ return { error: `payload.redacts.subject_ids must be at most ${MAX_IDS} non-empty strings` };
++ }
++ subjectType = r.subject_type;
++ subjectIds = [...new Set(r.subject_ids)];
++ }
++ if (!eventIds.length && !subjectIds.length) return { error: 'payload.redacts names nothing to redact' };
++ return { eventIds: [...new Set(eventIds)], subjectType, subjectIds };
++}
++
++/** May an event stored as `by` (a row: source, project_id, env) redact `target` (a row)? */
++function sameOwner(by, target) {
++ return target.source === by.source
++ && (target.project_id || null) === (by.project_id || null)
++ && (target.env || 'production') === (by.env || 'production');
++}
++
++/** The actor a tombstone carries: the producer itself. */
++function tombstoneActor(row) {
++ if (row.project_id && /^app-[0-9a-hjkmnp-tv-z]{26}$/.test(row.source)) return { type: 'app', id: `app_${row.source.slice(4).toUpperCase()}` };
++ return { type: 'service', id: row.source };
++}
++
++/** Column values that turn `row` into a tombstone. */
++function tombstone(row, { by, at }) {
++ const payload = JSON.stringify({ redacted: true, redacted_at: new Date(at).toISOString(), redacted_by: by });
++ const actor = JSON.stringify(tombstoneActor(row));
++ return {
++ payload,
++ actor,
++ size_bytes: Buffer.byteLength(payload) + Buffer.byteLength(actor) + row.event_type.length + row.subject_id.length,
++ redacted_at: at,
++ redacted_by: by,
++ };
++}
++
++module.exports = { parseDirective, sameOwner, tombstone, tombstoneActor, MAX_IDS };
ddiff --git a/server/store.js b/server/store.js
iindex fdb05c5..f3d87b0 100644
---- a/server/store.js
++++ b/server/store.js
@@@ -10,6 +10,7 @@ const path = require('path');
const Database = require('better-sqlite3');
const topics = require('./topics');
const apps = require('./apps');
++const redaction = require('./redaction');
const PRIORITY_RANK = { critical: 0, important: 1, low: 2 };
const RANK_PRIORITY = ['critical', 'important', 'low'];
@@@ -110,9 +111,13 @@ const PUBLISH_RECEIPT = 'events:publish'; // receipts' consumer column for acce
* with ALTER TABLE; every row stored before is a first-party production row.
* events.project_id / env / size_bytes who owns an app event, its environment, stored bytes
* subscriptions.project_id / env an app subscription's scope (NULL project = first-party)
++ * and redaction (server/redaction.js):
++ * events.redacted_at / redacted_by set when the row became a tombstone (ms, the redacting event_id)
++ * events.redacts 1 when the row carries a payload.redacts directive
*/
const ADDED_COLUMNS = {
-- events: [['project_id', 'TEXT'], ['env', "TEXT NOT NULL DEFAULT 'production'"], ['size_bytes', 'INTEGER NOT NULL DEFAULT 0']],
++ events: [['project_id', 'TEXT'], ['env', "TEXT NOT NULL DEFAULT 'production'"], ['size_bytes', 'INTEGER NOT NULL DEFAULT 0'],
++ ['redacted_at', 'INTEGER'], ['redacted_by', 'TEXT'], ['redacts', 'INTEGER NOT NULL DEFAULT 0']],
subscriptions: [['project_id', 'TEXT'], ['env', "TEXT NOT NULL DEFAULT 'production'"]],
};
@@@ -122,7 +127,8 @@ function migrate(db) {
for (const [name, type] of cols) if (!have.has(name)) db.exec(`ALTER TABLE ${table} ADD COLUMN ${name} ${type}`);
}
db.exec(`CREATE INDEX IF NOT EXISTS idx_events_project ON events(project_id, env, received_at) WHERE project_id IS NOT NULL;
-- CREATE INDEX IF NOT EXISTS idx_subscriptions_project ON subscriptions(project_id, env) WHERE project_id IS NOT NULL;`);
++ CREATE INDEX IF NOT EXISTS idx_subscriptions_project ON subscriptions(project_id, env) WHERE project_id IS NOT NULL;
++ CREATE INDEX IF NOT EXISTS idx_events_subject ON events(source, subject_type, subject_id);`);
}
function openDb(dbPath) {
@@@ -132,6 +138,8 @@ function openDb(dbPath) {
db.pragma('synchronous = NORMAL');
db.pragma('foreign_keys = ON');
db.pragma('busy_timeout = 5000');
++ // A redacted payload (and a pruned row) is overwritten with zeros, not left in free pages.
++ db.pragma('secure_delete = ON');
db.exec(SCHEMA);
migrate(db);
db.prepare("INSERT OR IGNORE INTO sequences (name, value) VALUES ('events', 0)").run();
@@@ -203,10 +211,15 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8 } = {}
AND subject_type = ? AND subject_id = ? AND (request_id IS NULL OR request_id != ?)`),
insertEvent: db.prepare(`INSERT INTO events (id, seq, event_type, version, source, actor, subject_type, subject_id,
subject_revision, trace_id, priority, visibility, occurred_at, received_at, payload, hops, publisher, request_id,
-- project_id, env, size_bytes)
++ project_id, env, size_bytes, redacts)
VALUES (@id, @seq, @event_type, @version, @source, @actor, @subject_type, @subject_id, @subject_revision,
@trace_id, @priority, @visibility, @occurred_at, @received_at, @payload, @hops, @publisher, @request_id,
-- @project_id, @env, @size_bytes)`),
++ @project_id, @env, @size_bytes, @redacts)`),
++ redactBySubject: db.prepare(`SELECT * FROM events WHERE source = ? AND subject_type = ? AND subject_id = ?
++ AND redacted_at IS NULL AND redacts = 0`),
++ applyTombstone: db.prepare(`UPDATE events SET payload = @payload, actor = @actor, size_bytes = @size_bytes,
++ redacted_at = @redacted_at, redacted_by = @redacted_by WHERE id = @id AND redacted_at IS NULL`),
++ firstSeqSince: db.prepare('SELECT MIN(seq) AS s FROM events WHERE received_at >= ?'),
enabledSubs: db.prepare('SELECT id, topic_pattern, project_id, env FROM subscriptions WHERE enabled = 1'),
projectRecent: db.prepare('SELECT COUNT(*) AS n FROM events WHERE project_id = ? AND env = ? AND received_at > ?'),
projectBytes: db.prepare('SELECT COALESCE(SUM(size_bytes), 0) AS b FROM events WHERE project_id = ? AND env = ?'),
@@@ -306,7 +319,11 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8 } = {}
project_id: project ? project.projectId : null,
env: project ? project.env : 'production',
size_bytes: sizeBytes,
++ redacts: 0,
};
++ const directive = redaction.parseDirective(env.payload);
++ if (directive && directive.error) throw new StoreError(422, 'events.invalid_redaction', directive.error, { event_id: env.event_id });
++ if (directive) row.redacts = 1;
q.insertEvent.run(row);
q.addReceipt.run(PUBLISH_RECEIPT, env.event_id, now);
for (const s of subs) {
@@@ -317,10 +334,62 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8 } = {}
revocationHook(row, now);
results.push({ event_id: env.event_id, seq, duplicate: false });
inserted.push(row);
++ if (directive) {
++ const redacted = applyRedaction(row, directive, now);
++ // Rows stored earlier in this batch are fanned out after the commit from memory: they
++ // must go out as the tombstones they now are.
++ for (const t of redacted) {
++ const mine = inserted.find(r => r.id === t.id);
++ if (mine) Object.assign(mine, t.columns);
++ }
++ }
}
return { results, inserted };
});
++ /**
++ * Turn the targets of `directive` (server/redaction.js) into tombstones on behalf of `by` (the
++ * stored row of the redacting event). Throws StoreError 403 when an event id names an event of
++ * another owner. Returns [{ id, seq, columns }] for the rows it rewrote.
++ */
++ function applyRedaction(by, directive, now) {
++ const targets = new Map();
++ for (const id of directive.eventIds) {
++ const r = q.getEvent.get(id);
++ if (!r) continue; // never stored, or pruned already
++ if (!redaction.sameOwner(by, r)) {
++ throw new StoreError(403, 'events.redaction_not_allowed', `event ${id} belongs to ${r.source}; ${by.source} may redact only its own events`,
++ { event_id: by.id, target: id });
++ }
++ if (r.redacted_at == null && !r.redacts) targets.set(r.id, r);
++ }
++ if (directive.subjectType) {
++ for (const subjectId of directive.subjectIds) {
++ for (const r of q.redactBySubject.all(by.source, directive.subjectType, subjectId)) {
++ if (redaction.sameOwner(by, r)) targets.set(r.id, r);
++ }
++ }
++ }
++ const out = [];
++ for (const r of targets.values()) {
++ if (r.id === by.id) continue;
++ const columns = redaction.tombstone(r, { by: by.id, at: now });
++ if (q.applyTombstone.run({ id: r.id, ...columns }).changes) out.push({ id: r.id, seq: r.seq, columns });
++ }
++ return out;
++ }
++
++ /**
++ * Operator redaction (scripts/redact-backfill.js): what an event of first-party `source` carrying
++ * `directive` (the payload.redacts object) would have redacted. The tombstones name `by` as
++ * redacted_by. Same owner rule as a publish. Returns the rewritten [{ id, seq }].
++ */
++ const redact = db.transaction((source, directive, { by = 'operator', now = clock.now() } = {}) => {
++ const parsed = redaction.parseDirective({ redacts: directive });
++ if (!parsed || parsed.error) throw new StoreError(422, 'events.invalid_redaction', parsed ? parsed.error : 'nothing to redact');
++ return applyRedaction({ id: by, source, project_id: null, env: 'production' }, parsed, now).map(({ id, seq }) => ({ id, seq }));
++ });
++
/**
* Network tells Events when an app loses its access (network.app.revoked, also sent for every
* app of an archived project) or its events.app.subscribe grant (network.grant.changed). The
@@@ -356,6 +425,12 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8 } = {}
return q.lastSeq.get().value;
}
++ /** First seq received at or after `ms`; the next seq to be handed out when there is none. */
++ function firstSeqSince(ms) {
++ const m = q.firstSeqSince.get(ms).s;
++ return m == null ? lastSeq() + 1 : m;
++ }
++
/** Oldest seq still stored; when the table is empty, the next seq to be handed out. */
function oldestSeq() {
const m = q.minSeq.get().s;
@@@ -529,7 +604,7 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8 } = {}
}
return {
-- db, insertBatch, getEvent, revokedAt, lastSeq, oldestSeq, scan,
++ db, insertBatch, redact, getEvent, revokedAt, lastSeq, oldestSeq, firstSeqSince, scan,
createSubscription, getSubscription, listSubscriptions, countSubscriptions, countProjectSubscriptions, setSubscriptionEnabled,
dueDeliveries, recordAttempt, getDelivery, listDeliveries, requeue, deliveryCounts,
getCheckpoint, setCheckpoint, prune, ping,
ddiff --git a/test/redact-backfill.test.js b/test/redact-backfill.test.js
nnew file mode 100644
iindex 0000000..0929b5a
---- /dev/null
++++ b/test/redact-backfill.test.js
@@@ -0,0 +1,118 @@
++'use strict';
++/**
++ * scripts/redact-backfill.js: redacts the stored chat events of messages Chat deleted before it
++ * published chat.message.deleted. Dry run by default, --apply only with a verified --backup,
++ * idempotent, first-party chat events only.
++ */
++const assert = require('assert');
++const fs = require('fs');
++const path = require('path');
++const Database = require('better-sqlite3');
++const { ids } = require('openvibe-contracts');
++const { openDb, createStore } = require('../server/store');
++const { main } = require('../scripts/redact-backfill');
++const { tmpDir, envelope, suite } = require('./helpers');
++
++const t = suite('redact-backfill');
++const dir = tmpDir();
++const eventsPath = path.join(dir, 'events.db');
++const chatPath = path.join(dir, 'chat.db');
++const user = ids.newId('user');
++let store, db;
++const ev = {};
++
++function created(messageId, text) {
++ return envelope('chat', {
++ event_type: 'chat.message.created', visibility: 'public', actor: { type: 'user', id: user },
++ subject: { type: 'chat_message', id: String(messageId) }, payload: { message_id: messageId, anon_id: 'anon77', text },
++ });
++}
++async function run(...argv) {
++ const lines = [];
++ const code = await main(['--db', eventsPath, '--chat-db', chatPath, ...argv], (l) => lines.push(l));
++ return { code, out: lines.join('\n') };
++}
++
++t('setup: an Events store with chat events, a Chat database with deleted messages', () => {
++ db = openDb(eventsPath);
++ store = createStore(db);
++ const batch = [created(1, 'deleted one'), created(2, 'still here'), created(3, 'expired one'), created(4, 'never in chat'),
++ envelope('live', { subject: { type: 'chat_message', id: '1' }, payload: { note: 'live about 1' } })];
++ [ev.m1, ev.m2, ev.m3, ev.m4, ev.live] = batch.map(e => e.event_id);
++ store.insertBatch(batch.map(e => ({ ...e, trace_id: 'a'.repeat(32), priority: 'important', visibility: e.visibility || 'internal' })), { publisher: 'svc:test' });
++ const chat = new Database(chatPath);
++ chat.exec(`CREATE TABLE chat_messages (id INTEGER PRIMARY KEY, message TEXT, is_deleted INTEGER DEFAULT 0, auto_delete_at DATETIME);
++ INSERT INTO chat_messages (id, message, is_deleted, auto_delete_at) VALUES
++ (1, 'deleted one', 1, NULL), (2, 'still here', 0, '2999-01-01 00:00:00'), (3, 'expired one', 0, '2000-01-01 00:00:00');`);
++ chat.close();
++});
++
++t('dry run: counts, changes nothing', async () => {
++ const r = await run();
++ assert.strictEqual(r.code, 0, r.out);
++ assert.match(r.out, /not redacted: 4 events, 4 messages/);
++ assert.match(r.out, /deleted in Chat: 2 messages, 2 events to redact/);
++ assert.match(r.out, /not in Chat's database \(left alone\): 1 messages/);
++ assert.match(r.out, /dry run: nothing changed/);
++ for (const id of [ev.m1, ev.m3]) assert.ok(!store.getEvent(id).redacted_at);
++});
++
++t('--apply refuses without --backup, and with an existing backup file', async () => {
++ let r = await run('--apply');
++ assert.strictEqual(r.code, 2);
++ assert.match(r.out, /refusing --apply without --backup/);
++ const existing = path.join(dir, 'exists.db');
++ fs.writeFileSync(existing, 'x');
++ r = await run('--apply', '--backup', existing);
++ assert.strictEqual(r.code, 2);
++ assert.match(r.out, /already exists/);
++ assert.ok(!store.getEvent(ev.m1).redacted_at, 'nothing changed');
++ r = await main(['--db', eventsPath], () => {});
++ assert.strictEqual(r, 2, '--chat-db is required');
++});
++
++t('--apply --backup: verified 0600 backup first, then exactly the deleted messages are redacted', async () => {
++ const backup = path.join(dir, 'backup-1.db');
++ const r = await run('--apply', '--backup', backup);
++ assert.strictEqual(r.code, 0, r.out);
++ assert.match(r.out, /redacted {3}2 events/);
++ assert.strictEqual(fs.statSync(backup).mode & 0o777, 0o600);
++ const b = new Database(backup, { readonly: true });
++ assert.match(b.prepare('SELECT payload FROM events WHERE id = ?').get(ev.m1).payload, /deleted one/, 'the backup is the state before');
++ b.close();
++ for (const id of [ev.m1, ev.m3]) {
++ const row = store.getEvent(id);
++ assert.strictEqual(row.redacted_by, 'backfill');
++ assert.deepStrictEqual(Object.keys(JSON.parse(row.payload)).sort(), ['redacted', 'redacted_at', 'redacted_by']);
++ assert.deepStrictEqual(JSON.parse(row.actor), { type: 'service', id: 'chat' });
++ }
++ assert.match(store.getEvent(ev.m2).payload, /still here/, 'a live message is untouched');
++ assert.match(store.getEvent(ev.m4).payload, /never in chat/, 'unknown messages are left alone');
++ assert.ok(!store.getEvent(ev.live).redacted_at, "another source's event about the same id is untouched");
++ const seqs = db.prepare('SELECT seq FROM events ORDER BY seq').pluck().all();
++ assert.deepStrictEqual(seqs, [1, 2, 3, 4, 5], 'no row removed, sequence intact');
++});
++
++t('idempotent: a second run finds nothing and changes nothing', async () => {
++ const before = db.prepare('SELECT id, payload, redacted_at FROM events ORDER BY seq').all();
++ const r = await run('--apply', '--backup', path.join(dir, 'backup-2.db'));
++ assert.strictEqual(r.code, 0, r.out);
++ assert.match(r.out, /deleted in Chat: 0 messages, 0 events to redact/);
++ assert.match(r.out, /redacted {3}0 events/);
++ assert.deepStrictEqual(db.prepare('SELECT id, payload, redacted_at FROM events ORDER BY seq').all(), before);
++});
++
++t('refuses a database from before redaction', async () => {
++ const old = path.join(dir, 'old.db');
++ const o = new Database(old);
++ o.exec('CREATE TABLE events (id TEXT PRIMARY KEY, seq INTEGER, payload TEXT)');
++ o.close();
++ const lines = [];
++ const code = await main(['--db', old, '--chat-db', chatPath], (l) => lines.push(l));
++ assert.strictEqual(code, 2);
++ assert.match(lines.join('\n'), /no redaction columns/);
++});
++
++t('close', () => { db.close(); });
++
++t.run();
ddiff --git a/test/redaction.test.js b/test/redaction.test.js
nnew file mode 100644
iindex 0000000..1613ee4
---- /dev/null
++++ b/test/redaction.test.js
@@@ -0,0 +1,311 @@
++'use strict';
++/**
++ * Redaction (server/redaction.js) and the public replay window (server/realtime.js).
++ *
++ * The leak this closes: a chat message deleted in OpenVibe.Chat stayed replayable here, text and
++ * anon id included, for the whole retention, to anyone (public visibility, anonymous SSE replay).
++ * Now Chat's chat.message.deleted carries payload.redacts, the stored chat.message.created becomes
++ * a tombstone at the same seq on every read path, and browsers are replayed public events of the
++ * last few minutes only.
++ */
++const assert = require('assert');
++const fs = require('fs');
++const path = require('path');
++const { ids } = require('openvibe-contracts');
++const apps = require('../server/apps');
++const { parseDirective } = require('../server/redaction');
++const { boot, request, serviceToken, userToken, appToken, envelope, subscriber, sse, sleep, suite } = require('./helpers');
++
++const t = suite('redaction');
++let h;
++const chat = serviceToken('chat', ['events.event.publish']);
++const live = serviceToken('live', ['events.event.publish']);
++const reader = serviceToken('search', ['events.event.read']);
++const subscriberSvc = serviceToken('search', ['events.subscription.manage']);
++const alice = ids.newId('user');
++const open = [];
++const SECRET_TEXT = 'probe: my phone number is 555-0199';
++
++async function stream(q, headers) {
++ const c = await sse(h.base, `/realtime/stream?${q}`, { headers });
++ open.push(c);
++ return c;
++}
++async function publish(token, body, status = 201) {
++ const r = await request(h.base, 'POST', '/api/v1/events', { token, body });
++ assert.strictEqual(r.status, status, r.text);
++ return r.body;
++}
++/** chat.message.created as OpenVibe.Chat publishes it (server/db/database.js saveChatMessage). */
++function created(messageId, over = {}) {
++ return envelope('chat', {
++ event_type: 'chat.message.created', visibility: 'public', actor: { type: 'user', id: alice },
++ subject: { type: 'chat_message', id: String(messageId) },
++ payload: {
++ message_id: messageId, room: { type: 'global' }, message_type: 'chat', user_id: null, user_subject: null,
++ anon_id: 'anon4242', username: 'anon4242', text: SECRET_TEXT, source_platform: null, reply_to_id: null,
++ },
++ ...over,
++ });
++}
++/** chat.message.deleted as Chat publishes it: the ids, and the redaction directive. */
++function deleted(messageIds, over = {}) {
++ return envelope('chat', {
++ event_type: 'chat.message.deleted', visibility: 'public', actor: { type: 'service', id: 'chat' },
++ subject: { type: 'chat_message', id: String(messageIds[0]) },
++ payload: { message_ids: messageIds, redacts: { subject_type: 'chat_message', subject_ids: messageIds.map(String) } },
++ ...over,
++ });
++}
++const leaks = (x) => JSON.stringify(x).includes('555-0199') || JSON.stringify(x).includes('anon4242') || JSON.stringify(x).includes(alice);
++
++function assertTombstone(event, byEventId) {
++ assert.strictEqual(event.payload.redacted, true, JSON.stringify(event));
++ assert.strictEqual(event.payload.redacted_by, byEventId);
++ assert.match(event.payload.redacted_at, /^\d{4}-\d\d-\d\dT/);
++ assert.deepStrictEqual(Object.keys(event.payload).sort(), ['redacted', 'redacted_at', 'redacted_by']);
++ assert.deepStrictEqual(event.actor, { type: 'service', id: 'chat' }, 'the author is no longer linked');
++ assert.strictEqual(event.event_type, 'chat.message.created', 'the envelope stays');
++ assert.ok(!leaks(event), 'no text, anon id or author subject');
++}
++
++t('directive parsing', () => {
++ assert.strictEqual(parseDirective({ text: 'x' }), null);
++ assert.deepStrictEqual(parseDirective({ redacts: { subject_type: 'chat_message', subject_ids: ['1', '1', '2'] } }),
++ { eventIds: [], subjectType: 'chat_message', subjectIds: ['1', '2'] });
++ const id = ids.newId('event');
++ assert.deepStrictEqual(parseDirective({ redacts: { event_ids: [id] } }), { eventIds: [id], subjectType: null, subjectIds: [] });
++ for (const bad of [null, [], 'x', {}, { event_ids: ['nope'] }, { subject_type: 'chat_message' }, { subject_ids: ['1'] },
++ { subject_type: 'Chat', subject_ids: ['1'] }, { subject_type: 'chat_message', subject_ids: [1] }, { event_ids: [] },
++ { subject_type: 'chat_message', subject_ids: Array.from({ length: 1001 }, (_, i) => String(i)) }, { event_ids: [id], extra: 1 }]) {
++ assert.ok(parseDirective({ redacts: bad }).error, JSON.stringify(bad));
++ }
++});
++
++t('boot', async () => { h = await boot({ env: { REALTIME_PUBLIC_REPLAY_SECONDS: '300' } }); });
++
++let msgSeq, delSeq, msgEventId, delEventId, sub, stub;
++
++t('a first-party subscriber is queued the original before the deletion', async () => {
++ stub = await subscriber();
++ const r = await request(h.base, 'POST', '/api/v1/subscriptions', { token: subscriberSvc, body: { topic_pattern: 'chat.message.*', endpoint: stub.url } });
++ assert.strictEqual(r.status, 201, r.text);
++ sub = r.body;
++});
++
++t('before deletion the message is replayable (the leak as it was)', async () => {
++ const env = created(101);
++ msgEventId = env.event_id;
++ msgSeq = (await publish(chat, env)).seq;
++ const anon = await stream('topics=chat.message.*', { 'Last-Event-ID': String(msgSeq - 1) });
++ await anon.waitFor(c => c.events().length === 1);
++ assert.strictEqual(anon.events()[0].event.payload.text, SECRET_TEXT);
++ anon.close();
++});
++
++t('chat.message.deleted turns the stored event into a tombstone at the same seq', async () => {
++ const env = deleted([101]);
++ delEventId = env.event_id;
++ const r = await publish(chat, env);
++ delSeq = r.seq;
++ assert.strictEqual(delSeq, msgSeq + 1);
++ const row = h.store.getEvent(msgEventId);
++ assert.strictEqual(row.seq, msgSeq, 'seq unchanged');
++ assert.strictEqual(row.redacted_by, delEventId);
++ assert.ok(row.redacted_at > 0);
++ assert.ok(!row.payload.includes('555-0199') && !row.actor.includes(alice));
++ assert.strictEqual(h.store.getEvent(delEventId).redacts, 1);
++ assert.ok(!h.store.getEvent(delEventId).redacted_at, 'the deletion event itself is kept as is');
++});
++
++t('anonymous SSE replay: the tombstone, never the text', async () => {
++ const anon = await stream('topics=chat.message.*', { 'Last-Event-ID': String(msgSeq - 1) });
++ await anon.waitFor(c => c.events().length === 2);
++ const [a, b] = anon.events();
++ assert.deepStrictEqual([a.seq, b.seq], [msgSeq, delSeq], 'sequence intact');
++ assertTombstone(a.event, delEventId);
++ assert.deepStrictEqual(b.event.payload.message_ids, [101]);
++ assert.ok(!leaks(anon.body));
++ anon.close();
++});
++
++t('signed-in SSE replay (cookie and Bearer) and service replay: the tombstone too', async () => {
++ const byCookie = await stream('topics=chat.message.*', { Cookie: `ov_token=${userToken({ subjectId: alice })}`, 'Last-Event-ID': String(msgSeq - 1) });
++ const byBearer = await stream('topics=chat.*', { Authorization: `Bearer ${userToken()}`, 'Last-Event-ID': String(msgSeq - 1) });
++ const svc = await stream('topics=chat.message.*', { Authorization: `Bearer ${serviceToken('search', ['events.event.read'])}`, 'Last-Event-ID': String(msgSeq - 1) });
++ for (const c of [byCookie, byBearer, svc]) {
++ await c.waitFor(x => x.events().length === 2);
++ assert.deepStrictEqual(c.events().map(e => e.seq), [msgSeq, delSeq]);
++ assertTombstone(c.events()[0].event, delEventId);
++ assert.ok(!leaks(c.body));
++ c.close();
++ }
++});
++
++t('pull (/api/v1/events), by id, and developer-app reads: the tombstone', async () => {
++ let r = await request(h.base, 'GET', `/api/v1/events?topic=chat.message.*&after_seq=${msgSeq - 1}`, { token: reader });
++ assert.strictEqual(r.status, 200);
++ assert.deepStrictEqual(r.body.events.map(e => e.seq), [msgSeq, delSeq], 'no hole in the sequence');
++ assertTombstone(r.body.events[0].event, delEventId);
++ r = await request(h.base, 'GET', `/api/v1/events/${msgEventId}`, { token: reader });
++ assertTombstone(r.body.event, delEventId);
++ assert.strictEqual(r.body.seq, msgSeq);
++ const app = appToken({ env: 'production', cap: ['events.app.read'] });
++ r = await request(h.base, 'GET', `/api/v1/events?topic=chat.*&after_seq=${msgSeq - 1}`, { token: app });
++ assert.strictEqual(r.status, 200, r.text);
++ assert.deepStrictEqual(r.body.events.map(e => e.seq), [msgSeq, delSeq]);
++ assertTombstone(r.body.events[0].event, delEventId);
++ assert.ok(!leaks(r.body));
++});
++
++t('durable delivery: the consumer gets the tombstone and then the deletion, in order', async () => {
++ await h.worker.drain();
++ const got = stub.calls.map(c => c.body);
++ assert.deepStrictEqual(got.map(b => b.seq), [msgSeq, delSeq]);
++ assertTombstone(got[0].event, delEventId);
++ assert.strictEqual(got[1].event.event_type, 'chat.message.deleted');
++ assert.ok(!leaks(stub.calls.map(c => c.rawBody.toString())));
++ // A DLQ replay of the original sends the tombstone as well.
++ const admin = serviceToken('ops', ['events.delivery.admin']);
++ const rq = await request(h.base, 'POST', '/api/v1/deliveries/replay', { token: admin, body: { subscription_id: sub.id, event_ids: [msgEventId] } });
++ assert.strictEqual(rq.body.queued, 1);
++ await h.worker.drain();
++ assertTombstone(stub.calls.at(-1).body.event, delEventId);
++});
++
++t('created and deleted in one batch: live SSE subscribers get the tombstone, not the text', async () => {
++ const anon = await stream('topics=chat.message.*');
++ const svc = await stream('topics=chat.message.*', { Authorization: `Bearer ${serviceToken('search', ['events.event.read'])}` });
++ await sleep(30);
++ const c = created(102);
++ const d = deleted([102]);
++ const r = await publish(chat, { events: [c, d] });
++ assert.deepStrictEqual(r.results.map(x => x.duplicate), [false, false]);
++ for (const s of [anon, svc]) {
++ await s.waitFor(x => x.events().length === 2);
++ assertTombstone(s.events()[0].event, d.event_id);
++ assert.ok(!leaks(s.body));
++ }
++});
++
++t('re-publishing a deletion is a duplicate; a second deletion leaves tombstones and deletions alone', async () => {
++ const again = await request(h.base, 'POST', '/api/v1/events', { token: chat, body: { ...deleted([101]), event_id: delEventId } });
++ assert.strictEqual(again.status, 200);
++ assert.strictEqual(again.body.duplicate, true);
++ const second = deleted([101]);
++ await publish(chat, second);
++ assert.strictEqual(h.store.getEvent(msgEventId).redacted_by, delEventId, 'first redaction stands');
++ assert.ok(!h.store.getEvent(delEventId).redacted_at, 'a deletion event is never redacted by a later one');
++});
++
++t('only the owning service can redact', async () => {
++ const env = created(201);
++ const seq = (await publish(chat, env)).seq;
++ // Live names Chat's event by id: refused, and nothing of the batch is stored.
++ const liveEvent = envelope('live', { payload: { redacts: { event_ids: [env.event_id] } } });
++ let r = await request(h.base, 'POST', '/api/v1/events', { token: live, body: { events: [envelope('live'), liveEvent] } });
++ assert.strictEqual(r.status, 403, r.text);
++ assert.strictEqual(r.body.code, 'events.redaction_not_allowed');
++ assert.strictEqual(h.store.getEvent(liveEvent.event_id), null);
++ // Live by subject: its own namespace only, so Chat's event is untouched.
++ await publish(live, envelope('live', { payload: { redacts: { subject_type: 'chat_message', subject_ids: ['201'] } } }));
++ // Nor can Live publish Chat's deletion (source must be the caller).
++ r = await request(h.base, 'POST', '/api/v1/events', { token: live, body: deleted([201]) });
++ assert.strictEqual(r.status, 403);
++ assert.strictEqual(r.body.code, 'events.source_mismatch');
++ // A developer app cannot reach first-party events.
++ const appId = ids.newId('app');
++ const prj = `prj_${ids.ulid()}`;
++ const appEvent = envelope('x', {
++ source: apps.appSource(appId), event_type: `app.${apps.projectKey(prj)}.thing.deleted`, actor: { type: 'app', id: appId },
++ subject: { type: 'chat_message', id: '201' }, payload: { redacts: { event_ids: [env.event_id] } },
++ });
++ r = await request(h.base, 'POST', '/api/v1/events', { token: appToken({ appId, projectId: prj, env: 'production', cap: ['events.app.publish'] }), body: appEvent });
++ assert.strictEqual(r.status, 403, r.text);
++ assert.strictEqual(r.body.code, 'events.redaction_not_allowed');
++ r = await request(h.base, 'POST', '/api/v1/events', {
++ token: appToken({ appId, projectId: prj, env: 'production', cap: ['events.app.publish'] }),
++ body: { ...appEvent, event_id: ids.newId('event'), payload: { redacts: { subject_type: 'chat_message', subject_ids: ['201'] } } },
++ });
++ assert.strictEqual(r.status, 201, r.text);
++ // A malformed directive is refused before anything is stored.
++ r = await request(h.base, 'POST', '/api/v1/events', { token: chat, body: deleted([201], { payload: { redacts: { subject_type: 'chat_message' } } }) });
++ assert.strictEqual(r.status, 422);
++ assert.strictEqual(r.body.code, 'events.invalid_redaction');
++ const row = h.store.getEvent(env.event_id);
++ assert.ok(!row.redacted_at, 'still intact after every refused or foreign attempt');
++ assert.ok(row.payload.includes('555-0199'));
++ // The owner can.
++ await publish(chat, deleted([201]));
++ assert.ok(h.store.getEvent(env.event_id).redacted_at);
++ assert.strictEqual(h.store.getEvent(env.event_id).seq, seq);
++});
++
++t('the text is gone from the database file, not just hidden', async () => {
++ h.db.pragma('wal_checkpoint(TRUNCATE)');
++ const bytes = fs.readFileSync(path.join(h.dir, 'events.db'));
++ assert.ok(!bytes.includes(Buffer.from('555-0199')), 'secure_delete zeroed the old payloads');
++ assert.ok(!bytes.includes(Buffer.from('anon4242')));
++});
++
++t('operator redaction (backfill): same owner rule, idempotent', async () => {
++ const env = created(301);
++ await publish(chat, env);
++ const other = envelope('live', { subject: { type: 'chat_message', id: '301' } });
++ await publish(live, other);
++ const done = h.store.redact('chat', { subject_type: 'chat_message', subject_ids: ['301'] }, { by: 'backfill' });
++ assert.deepStrictEqual(done.map(d => d.id), [env.event_id]);
++ assert.strictEqual(h.store.getEvent(env.event_id).redacted_by, 'backfill');
++ assert.ok(!h.store.getEvent(other.event_id).redacted_at, "another source's event with the same subject is untouched");
++ assert.deepStrictEqual(h.store.redact('chat', { subject_type: 'chat_message', subject_ids: ['301'] }, { by: 'backfill' }), [], 'second run: nothing');
++ assert.throws(() => h.store.redact('chat', { event_ids: [other.event_id] }), /may redact only its own/);
++});
++
++t('public replay window: browsers get a gap for public events older than 5 minutes', async () => {
++ const w = await boot({ env: { REALTIME_PUBLIC_REPLAY_SECONDS: '300' } });
++ try {
++ const post = async (token, body) => (await request(w.base, 'POST', '/api/v1/events', { token, body })).body.seq;
++ const net = serviceToken('network', ['events.event.publish']);
++ const oldPublic = await post(chat, created(1));
++ const oldMine = await post(net, envelope('network', { event_type: 'network.notification.created', visibility: 'subject', subject: { type: 'user', id: alice } }));
++ w.clock.advance(301 * 1000);
++ const recent = await post(chat, created(2));
++ const topics = 'topics=chat.message.*,network.notification.*';
++
++ const anon = await sse(w.base, `/realtime/stream?${topics}`, { headers: { 'Last-Event-ID': String(oldPublic - 1) } });
++ await anon.waitFor(c => c.events().length === 1);
++ assert.deepStrictEqual(anon.events().map(e => e.seq), [recent]);
++ const gap = anon.gaps()[0];
++ assert.strictEqual(gap.reason, 'public_window');
++ assert.deepStrictEqual([gap.from_seq, gap.to_seq, gap.window_seconds], [oldPublic, recent - 1, 300]);
++ assert.ok(anon.messages.findIndex(m => m.event === 'gap') < anon.messages.findIndex(m => !m.event), 'gap first');
++
++ const user = await sse(w.base, `/realtime/stream?${topics}`, { headers: { Cookie: `ov_token=${userToken({ subjectId: alice })}`, 'Last-Event-ID': String(oldPublic - 1) } });
++ await user.waitFor(c => c.events().length === 2);
++ assert.deepStrictEqual(user.events().map(e => e.seq), [oldMine, recent], 'own subject events keep the whole retention; old public ones do not');
++ assert.strictEqual(user.gaps()[0].reason, 'public_window');
++
++ const svc = await sse(w.base, `/realtime/stream?${topics}`, { headers: { Authorization: `Bearer ${serviceToken('search', ['events.event.read'])}`, 'Last-Event-ID': String(oldPublic - 1) } });
++ await svc.waitFor(c => c.events().length === 3);
++ assert.deepStrictEqual(svc.gaps(), [], 'services are not windowed');
++
++ const fresh = await sse(w.base, `/realtime/stream?${topics}`, { headers: { 'Last-Event-ID': String(recent - 1) } });
++ await fresh.waitFor(c => c.events().length === 1);
++ assert.deepStrictEqual(fresh.gaps(), [], 'a reconnect inside the window is seamless');
++ for (const c of [anon, user, svc, fresh]) c.close();
++
++ const none = await boot({ env: { REALTIME_PUBLIC_REPLAY_SECONDS: '0' } });
++ try {
++ const s = await (async () => (await request(none.base, 'POST', '/api/v1/events', { token: chat, body: created(3) })).body.seq)();
++ const c = await sse(none.base, '/realtime/stream?topics=chat.message.*', { headers: { 'Last-Event-ID': String(s - 1) } });
++ await c.waitFor(x => x.gaps().length === 1);
++ await sleep(50);
++ assert.deepStrictEqual(c.events(), [], 'window 0: no public replay at all');
++ c.close();
++ } finally { await none.stop(); }
++ } finally { await w.stop(); }
++});
++
++t('stop', async () => { for (const c of open) c.close(); if (stub) await stub.close(); await h.stop(); });
++
++t.run();