Account export and deletion (ADR-033) for what people write here: the export carries their reviews, additions and corrections; the deletion removes their reviews (pages re-sent to Search), leaves additions and corrections authorless, keeps the audit log (#77)

ddiff --git a/migrations/0005_account_data.sql b/migrations/0005_account_data.sql
nnew file mode 100644
iindex 0000000..e2c0f75
---- /dev/null
++++ b/migrations/0005_account_data.sql
@@@ -0,0 +1,12 @@
++-- phase: expand
++-- Account export and deletion (ADR-033) now that Reviews holds what people write (members' reviews, 0004):
++-- account_data_events keeps one receipt per export or deletion delivery Reviews applied, so a redelivered deletion
++-- never erases twice and a confirmation that failed is sent again (openvibe-sdk/account-data, its ACCOUNT_DATA_SCHEMA).
++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
++);
ddiff --git a/server/app.js b/server/app.js
iindex 21a7262..6bcfe09 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -42,7 +42,7 @@ function lagged(read, initial) {
      };
  }
  
--function createApp({ config, telemetry = null, svc, viewers, platform, sources, sync, keys, db, log = console, rateLimits = true, fetchImpl = globalThis.fetch, limitsNow = null, indexnow = null }) {
++function createApp({ config, telemetry = null, svc, viewers, platform, sources, sync, keys, db, log = console, rateLimits = true, fetchImpl = globalThis.fetch, limitsNow = null, indexnow = null, accountData = null, accountSend = null }) {
      const app = express();
      app.disable('x-powered-by');
      app.set('trust proxy', config.trustProxy);
@@@ -102,6 +102,13 @@ function createApp({ config, telemetry = null, svc, viewers, platform, sources,
      // GET /<key>.txt — the IndexNow key file (mounted only when a key is configured; it serves itself).
      if (indexnow && indexnow.enabled) app.use(indexnow.keyFile);
  
++    // Account export and deletion (ADR-033): network.account.export_requested and network.account.deleted, signed by
++    // Events (v2) and answered by openvibe-sdk/account-data over reviews/account-data.js. Its own path: /internal/events
++    // is the Sources consumer.
++    if (accountData) {
++        const send = accountSend || (async () => { throw new Error('OV_OAUTH_CLIENT_SECRET is not set: Reviews cannot answer account events'); });
++        app.post('/internal/account-events', accountData.consumer({ secrets: config.eventsWebhookSecrets || [], send, log }));
++    }
      // Signed deliveries from OpenVibe.Events. Loopback/internal only in nginx (not proxied publicly).
      const consumer = createEvents({ db, svc, sync, config, log });
      app.use('/internal', http.middleware(), consumer.router);
ddiff --git a/server/index.js b/server/index.js
iindex f8f5c56..9ac82d4 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -19,6 +19,8 @@ const { createViewerResolver } = require('./auth/viewer');
  const { createApp } = require('./app');
  const { createIndexNow } = require('openvibe-shared/indexnow');
  const { gracefulStop, createHttpTelemetry } = require('openvibe-sdk/service');
++const { createNetworkSender, startSubscriptions } = require('openvibe-sdk/account-data');
++const accountDataLib = require('./reviews/account-data');
  const { createPgTelemetrySink, startTelemetryRetention } = require('openvibe-sdk/telemetry');
  
  /**
@@@ -28,12 +30,13 @@ const { createPgTelemetrySink, startTelemetryRetention } = require('openvibe-sdk
   * its own database (start's `givenDb`), which stays open. `exit` and `signals` are injectable so a test can watch
   * the stop and no test process installs signal handlers.
   */
--function createLifecycle({ server, db, valkey = null, platform, timers = [], telemetry = null, retention = null, closeDb = true, exit, signals, log } = {}) {
++function createLifecycle({ server, db, valkey = null, platform, timers = [], telemetry = null, retention = null, closeDb = true, exit, signals, log, subscriptions = null } = {}) {
      return gracefulStop({
          name: 'Reviews', server, log, drainMs: 8000, deadlineMs: 10000, deadlineExitCode: 0, exit, signals,
          stop: [
              () => timers.forEach((t) => { clearInterval(t); clearTimeout(t); }),
              () => platform.outbox.stop(),
++            () => { if (subscriptions) subscriptions.stop(); },
          ],
          close: [
              async () => { if (retention) await retention.stop(); if (telemetry) await telemetry.stop(); },
@@@ -43,7 +46,7 @@ function createLifecycle({ server, db, valkey = null, platform, timers = [], tel
      });
  }
  
--async function start({ config, db: givenDb = null, now = () => Date.now(), fetchImpl = globalThis.fetch, tokens = null, publicKey = null, log = console, telemetrySink = null, listen = true, workers = listen, rateLimits = true, limitsNow = null, indexnow: givenIndexnow = undefined, signals = false, exit = () => {} } = {}) {
++async function start({ config, db: givenDb = null, now = () => Date.now(), fetchImpl = globalThis.fetch, tokens = null, publicKey = null, log = console, telemetrySink = null, listen = true, workers = listen, rateLimits = true, limitsNow = null, indexnow: givenIndexnow = undefined, signals = false, exit = () => {}, send: givenSend = null } = {}) {
      config = config || load();
      // PostgreSQL (ADR-035): opened and migrated here unless the caller (a test) hands in a migrated handle.
      const db = givenDb || await openDb(config, { log });
@@@ -65,7 +68,13 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
      // on, a one-minute tick writes each interval (the collector has no timer of its own) and the table keeps 14 days.
      const telemetry = createHttpTelemetry({ service: 'reviews', sink: telemetrySink || createPgTelemetrySink({ db, log }), skipPrefixes: ['/shared'], intervalMs: 60_000 });
      let retention = null;
--    const app = createApp({ config, telemetry, svc, viewers, platform, sources, sync, keys, db, log, rateLimits, fetchImpl, limitsNow, indexnow });
++    // Account export and deletion (ADR-033): the table map over the service, and the sender to Network's internal routes
++    // with Reviews' own client-credentials token (a test hands in a stand-in).
++    const accountData = accountDataLib.create({ db, svc, log });
++    const accountSend = givenSend || (config.oauth.clientSecret
++        ? createNetworkSender({ networkInternalUrl: config.networkInternalUrl, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, fetch: fetchImpl })
++        : null);
++    const app = createApp({ config, telemetry, svc, viewers, platform, sources, sync, keys, db, log, rateLimits, fetchImpl, limitsNow, indexnow, accountData, accountSend });
  
      // Search holds whatever the current rules say (idempotent: unchanged documents are not re-sent).
      try { await svc.reconcileIndex(); } catch (err) { log.warn(`[Reviews] index reconcile: ${err.message}`); }
@@@ -90,8 +99,17 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
          server.keepAliveTimeout = 65000;
          log.log(`[Reviews] ${config.nodeEnv} on http://${config.host}:${config.port} → ${config.baseUrl}`);
      }
++    // The two account subscriptions at OpenVibe.Events, created when missing; off without workers, EVENTS_URL,
++    // REVIEWS_EVENTS_SECRET or the client secret.
++    let subscriptions = null;
++    if (workers && listen) {
++        subscriptions = startSubscriptions({
++            eventsUrl: config.eventsUrl, endpoint: `http://127.0.0.1:${config.port}/internal/account-events`, secret: (config.eventsWebhookSecrets || [])[0],
++            networkInternalUrl: config.networkInternalUrl, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, fetch: fetchImpl, log,
++        });
++    }
  
--    const lifecycle = createLifecycle({ server, db, telemetry, retention, platform, timers, closeDb: !givenDb, signals, exit, log });
++    const lifecycle = createLifecycle({ server, db, telemetry, retention, platform, timers, closeDb: !givenDb, signals, exit, log, subscriptions });
  
      async function stop() {
          return lifecycle.stop('stop');
ddiff --git a/server/reviews/account-data.js b/server/reviews/account-data.js
nnew file mode 100644
iindex 0000000..00838eb
---- /dev/null
++++ b/server/reviews/account-data.js
@@@ -0,0 +1,46 @@
++'use strict';
++/**
++ * Account export and deletion → Reviews (ADR-033; openvibe-sdk/account-data). What Reviews holds about a person:
++ *
++ *   their own, deleted       the reviews they wrote here (review_member_reviews); the pages they were on are re-sent to
++ *                            Search without them, in the same transaction
++ *   their additions, kept    what they added to review (review_entities.created_by) and the corrections they sent
++ *                            (review_corrections.submitted_by) stay, made authorless ('deleted'): other people's
++ *                            reviews and the editors' answers hang under them
++ *   kept                     the editorial audit log (review_audit): who decided what is the record editors answer for
++ */
++const { createAccountData } = require('openvibe-sdk/account-data');
++
++const TABLES = [
++    { table: 'review_audit', subject: 'actor', file: null, erase: { keep: 'the editorial record editors answer for' } },
++];
++
++async function extraExport(db, subject) {
++    const files = [];
++    const reviews = await db.many(`SELECT id, entity_id, rating, title, body, relation, status, created_at, updated_at FROM review_member_reviews
++        WHERE author = $1 ORDER BY created_at DESC LIMIT 5000`, [subject]);
++    if (reviews.length) files.push({ name: 'reviews.json', content: reviews });
++    const added = await db.many('SELECT id, slug, name, kind, description, created_at FROM review_entities WHERE created_by = $1 ORDER BY created_at DESC LIMIT 5000', [subject]);
++    if (added.length) files.push({ name: 'added-to-review.json', content: added });
++    const corrections = await db.many(`SELECT id, entity_id, target_type, target_id, body, evidence_url, status, resolution_note, created_at FROM review_corrections
++        WHERE submitted_by = $1 ORDER BY created_at DESC LIMIT 5000`, [subject]);
++    if (corrections.length) files.push({ name: 'corrections.json', content: corrections });
++    return files;
++}
++
++/** The account-data handle over Reviews' database; `svc.syncIndex` re-sends a page (it joins the erase transaction). */
++function create({ db, svc, log = console } = {}) {
++    async function extraErase(t, subjects, counts) {
++        const pages = (await t.many('SELECT DISTINCT entity_id FROM review_member_reviews WHERE author = ANY($1::text[])', [subjects])).map((r) => r.entity_id);
++        counts.add(counts.erased, 'review_member_reviews', await t.exec('DELETE FROM review_member_reviews WHERE author = ANY($1::text[])', [subjects]));
++        counts.add(counts.retained, 'tombstones', await t.exec("UPDATE review_entities SET created_by = 'deleted' WHERE created_by = ANY($1::text[])", [subjects]));
++        counts.add(counts.retained, 'tombstones', await t.exec("UPDATE review_corrections SET submitted_by = 'deleted' WHERE submitted_by = ANY($1::text[])", [subjects]));
++        for (const id of pages) await svc.syncIndex(id);
++    }
++    return createAccountData({
++        db, service: 'reviews', tables: TABLES, extraExport, extraErase, log,
++        note: 'Your reviews are deleted. What you added to review and the corrections you sent stay without your name.',
++    });
++}
++
++module.exports = { create, TABLES };
ddiff --git a/test/account-data.test.js b/test/account-data.test.js
nnew file mode 100644
iindex 0000000..328589c
---- /dev/null
++++ b/test/account-data.test.js
@@@ -0,0 +1,63 @@
++'use strict';
++/**
++ * Account export and deletion (ADR-033) now that Reviews holds what people write: the export carries a person's reviews,
++ * what they added to review and their corrections; the deletion removes their reviews (the pages are re-sent to Search
++ * without them), leaves what they added and their corrections authorless, keeps the editorial audit log, and confirms
++ * once with the counts. Deliveries are Events' v2-signed webhooks at /internal/account-events.
++ */
++const assert = require('assert');
++const { signDeliveryHeaders } = require('openvibe-sdk/events');
++const { boot, req, suite, userToken, cookieFor, subject, outbox, WEBHOOK_SECRET } = require('./helpers');
++
++const t = suite('account-data');
++const BODY = 'Managers listened, the pay was on time, and the on-call rota was fair. Promotions were slow though.';
++
++t('export, then deletion: reviews gone and re-indexed, additions and corrections authorless, confirmed once', async () => {
++    const calls = [];
++    const send = async (path, body) => { calls.push({ path, body }); return { ok: true, status: 201 }; };
++    const h = await boot({ send });
++    try {
++        const ana = subject();
++        const ben = subject();
++        const anaC = cookieFor(userToken({ subject: ana, username: 'ana' }));
++        const benC = cookieFor(userToken({ subject: ben, username: 'ben' }));
++        await req(h, 'POST', '/review', { cookie: anaC, form: { kind: 'employer', name: 'Acme Corp' } });
++        await req(h, 'POST', '/e/acme-corp/reviews', { cookie: anaC, form: { rating: '4', title: 'Solid place to work', body: BODY } });
++        await req(h, 'POST', '/e/acme-corp/reviews', { cookie: benC, form: { rating: '2', title: 'Burned out', body: 'Constant overtime, and the on-call rota was anything but fair to the night team.' } });
++        const deliver = async (event, { secret = WEBHOOK_SECRET, headers = {} } = {}) => {
++            const body = JSON.stringify({ event, seq: 1 });
++            const r = await fetch(`${h.base}/internal/account-events`, { method: 'POST', body, headers: { 'Content-Type': 'application/json', ...signDeliveryHeaders(body, secret), ...headers } });
++            return { status: r.status, json: await r.json().catch(() => null) };
++        };
++        const ev = (id, type, payload) => ({ event_id: id, event_type: type, source: 'network', version: 1, timestamp: new Date().toISOString(), payload });
++
++        // Export: her review, what she added, nothing of Ben's.
++        const ex = await deliver(ev('evt_01JZ0000000000000000000E01', 'network.account.export_requested', { export_id: 'exp_01JZ0000000000000000000EX1', subject: ana }));
++        assert.deepStrictEqual([ex.status, ex.json && ex.json.outcome], [200, 'exported'], JSON.stringify(ex.json));
++        const part = calls.find((c) => c.path === '/internal/account-exports/exp_01JZ0000000000000000000EX1/parts');
++        const files = JSON.stringify(part.body);
++        assert.ok(files.includes('reviews.json') && files.includes('Solid place to work') && files.includes('added-to-review.json'), files.slice(0, 300));
++        assert.ok(!files.includes('Burned out') && !files.includes(ben), 'nothing of Ben\'s');
++
++        // Deletion: her review gone, the page re-sent to Search without it; Acme stays, made authorless.
++        const before = (await outbox(h)).length;
++        const del = ev('evt_01JZ0000000000000000000D01', 'network.account.deleted', { deletion_id: 'del_01JZ0000000000000000000DE1', subject: ana });
++        const r = await deliver(del);
++        assert.deepStrictEqual([r.status, r.json && r.json.outcome], [200, 'erased'], JSON.stringify(r.json));
++        assert.strictEqual(Number((await h.db.prepare('SELECT COUNT(*) AS n FROM review_member_reviews WHERE author = ?').get(ana)).n), 0);
++        assert.strictEqual(Number((await h.db.prepare('SELECT COUNT(*) AS n FROM review_member_reviews WHERE author = ?').get(ben)).n), 1, 'Ben\'s review stays');
++        assert.strictEqual((await h.db.prepare("SELECT created_by FROM review_entities WHERE slug = 'acme-corp'").get()).created_by, 'deleted');
++        const page = await req(h, 'GET', '/e/acme-corp');
++        assert.ok(page.text.includes('Reviews from OpenVibe members (1)') && !page.text.includes('Solid place to work'));
++        const docs = (await outbox(h)).slice(before).filter((x) => x.event_type === 'reviews.index_document.upserted' && x.payload.title === 'Acme Corp');
++        assert.ok(docs.some((d) => /OpenVibe members: 2 out of 5 from 1 review/.test(d.payload.body || '')), 'Search gets the page without her review');
++        const conf = calls.filter((c) => c.path === '/internal/account-deletions/del_01JZ0000000000000000000DE1/confirmations');
++        assert.strictEqual(conf.length, 1);
++        assert.strictEqual(conf[0].body.erased.review_member_reviews, 1);
++        // A redelivery erases nothing twice.
++        assert.strictEqual((await deliver(del)).json.outcome, 'unchanged');
++        // Signed wrong or relayed from outside: refused.
++        assert.strictEqual((await deliver(del, { secret: `x${WEBHOOK_SECRET}` })).status, 401);
++        assert.strictEqual((await deliver(del, { headers: { 'X-Forwarded-For': '203.0.113.9' } })).status, 403);
++    } finally { await h.stop(); }
++});
ddiff --git a/test/helpers.js b/test/helpers.js
iindex cc2a882..5f9766b 100644
---- a/test/helpers.js
++++ b/test/helpers.js
@@@ -147,7 +147,7 @@ function fakeCommunity() {
   * (default false) turns on the per-address and per-actor limits; limitsNow is the per-actor limiter's
   * clock; log is Reviews' logger (default quiet).
   */
--async function boot({ env = {}, sources = fakeSources(), community = null, now, workers = false, rateLimits = false, limitsNow = null, log = quiet, indexnow, telemetrySink = null } = {}) {
++async function boot({ env = {}, sources = fakeSources(), community = null, now, workers = false, rateLimits = false, limitsNow = null, log = quiet, indexnow, telemetrySink = null, send = null } = {}) {
      const config = load({
          NODE_ENV: 'test', PORT: '0', HOST: '127.0.0.1', BASE_URL: 'http://reviews.test',
          OV_NETWORK_URL: ISSUER, OV_NETWORK_INTERNAL_URL: '', REVIEWS_GATE_MIN_WORDS: '20',
@@@ -167,7 +167,7 @@ async function boot({ env = {}, sources = fakeSources(), community = null, now,
      // One database per boot (PGlite, or REVIEWS_TEST_STORE=pg: the containers), dropped when the boot stops.
      const testdb = await testDb();
      const h = await start({
--        config, db: testdb.db, publicKey, log, listen: true, workers, rateLimits, limitsNow, now, fetchImpl, indexnow, telemetrySink,
++        config, db: testdb.db, publicKey, log, listen: true, workers, rateLimits, limitsNow, now, fetchImpl, indexnow, telemetrySink, send,
          tokens: { getToken: async () => 'stub-token', authHeaders: async () => ({ Authorization: 'Bearer stub-token' }), invalidate() {} },
      });
      const base = `http://127.0.0.1:${h.server.address().port}`;