Outbox: openvibe-sdk/events (no money logic changes) (#40)

ddiff --git a/README.md b/README.md
iindex 4668d68..63aaf1e 100644
---- a/README.md
++++ b/README.md
@@@ -61,7 +61,7 @@ promo write changes both in one transaction.
  ## Depends on
  
  - OpenVibe.Network (JWKS for service tokens; SSO with PKCE for the staff console; `identity.subject.resolve` for the Live import)
--- OpenVibe.Events (the outbox relay, only when `EVENTS_URL` is set)
++- OpenVibe.Events (the SDK outbox relay, when `EVENTS_URL` and the OAuth client secret are set)
  - the payment providers whose secrets are configured (none in production today)
  - `openvibe-contracts` v0.129.0, `openvibe-sdk` v0.38.0 (service tokens, per-person limits, `openvibe-sdk/db`, `openvibe-sdk/service`), `openvibe-shared` v3.0.0, pinned by release tarball
  - **PostgreSQL 18 and Valkey 9** (OpenVibe.Host `roles/data/`, ADR-035): every read and write is async through
@@@ -119,10 +119,10 @@ Production: `/opt/openvibe.billing`, env `/etc/openvibe/billing.env`, unit
  - **Freeze** (ADR-012 rule 11): `settings.freeze`; while on, every mutating endpoint answers
    `503 billing.frozen`, reads work, webhooks are stored (202) and processed in arrival order after
    unfreeze. Jobs pause.
--- **Outbox**: `billing.transaction.settled|reversed`, `billing.entitlement.changed`,
++- **Outbox**: openvibe-sdk/events stores pending envelopes in `service_outbox` and relays them with Billing's service token. The legacy `outbox` table stays through the N-1 rollback window. Events include `billing.transaction.settled|reversed`, `billing.entitlement.changed`,
    `billing.subscription.canceled`, `billing.cashout.requested|paid|denied`, `billing.staff.action`,
    `billing.receipt.external`, written in the same transaction as the effect (events.event-envelope@1,
--  source `billing`, actor `service:billing`). Relayed to OpenVibe.Events only when `EVENTS_URL` is set.
++  source `billing`, actor `service:billing`). Relayed to OpenVibe.Events when `EVENTS_URL` and the OAuth client secret are set; otherwise rows wait.
  - **Authority** (`BILLING_AUTHORITY`, `live` by default = shadow; `billing` after the cutover): decides
    whether EXTERNAL tips are announced (see below). Anything else stops Billing at boot.
  - **EXTERNAL receipts** ([server/ops/external.js](server/ops/external.js)): a PowerChat tip paid to the
ddiff --git a/STATUS.json b/STATUS.json
iindex 8e40323..6c890b6 100644
---- a/STATUS.json
++++ b/STATUS.json
@@@ -25,7 +25,7 @@
      "staff console: cashouts, review queue, reconciliation, freeze, audit log",
      "public billing policy page",
      "Live shadow import",
--    "outbox events billing.*",
++    "openvibe-sdk/events outbox in service_outbox for billing.* events (legacy outbox retained for rollback)",
      "per-person limits (openvibe-sdk/limits)",
      "usage readings (billing.usage.record)",
      "usage rating: a rate card -> the subject's promo allowance -> one usage transaction per reading, under hard budgets (BILLING_RATING_INTERVAL_MS off by default; nothing charged without a loaded card, key usage:<reading key>)",
ddiff --git a/docs/sdk-outbox.md b/docs/sdk-outbox.md
nnew file mode 100644
iindex 0000000..610ed52
---- /dev/null
++++ b/docs/sdk-outbox.md
@@@ -0,0 +1,5 @@
++# Billing events outbox
++
++Migration `0006_sdk_outbox.sql` expands the database with the openvibe-sdk/events `service_outbox` schema and copies every unsent row from the legacy `outbox`. Billing writes new envelopes to `service_outbox` in the transaction that changes the ledger or operational state. The SDK relay publishes them to OpenVibe.Events with Billing's service token when `EVENTS_URL` and the OAuth client secret are set. Its `status()` reports whether delivery is enabled and the pending and rejected counts.
++
++The legacy table stays for rollback during the N-1 window. A previous release can still read and relay it; Events deduplicates repeated `event_id` values. A later contract migration can drop `outbox` after that window. Check `billing_outbox_pending`, `billing_outbox_failing`, and `billing_outbox_rejected` at `/metrics` when reviewing delivery.
ddiff --git a/migrations/0006_sdk_outbox.sql b/migrations/0006_sdk_outbox.sql
nnew file mode 100644
iindex 0000000..c5b63fe
---- /dev/null
++++ b/migrations/0006_sdk_outbox.sql
@@@ -0,0 +1,23 @@
++-- phase: expand
++-- openvibe-sdk/events owns service_outbox. Copy pending legacy events before the new relay starts.
++-- A later contract migration drops outbox after the N-1 rollback window.
++
++CREATE TABLE IF NOT EXISTS service_outbox (
++    id              bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
++    event_id        text NOT NULL UNIQUE,
++    envelope        jsonb NOT NULL,
++    traceparent     text,
++    created_at      bigint NOT NULL,
++    attempts        integer NOT NULL DEFAULT 0,
++    next_attempt_at bigint NOT NULL DEFAULT 0,
++    sent_at         bigint,
++    seq             bigint,
++    rejected_at     bigint,
++    last_error      text
++);
++CREATE INDEX IF NOT EXISTS service_outbox_due ON service_outbox (next_attempt_at, id) WHERE sent_at IS NULL AND rejected_at IS NULL;
++CREATE INDEX IF NOT EXISTS service_outbox_sent ON service_outbox (sent_at) WHERE sent_at IS NOT NULL;
++
++INSERT INTO service_outbox (event_id, envelope, created_at, attempts)
++SELECT event_id, event::jsonb, (EXTRACT(EPOCH FROM created_at::timestamptz) * 1000)::bigint, attempts
++FROM outbox WHERE sent_at IS NULL ON CONFLICT (event_id) DO NOTHING;
ddiff --git a/server/app.js b/server/app.js
iindex 55591bb..de57696 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -30,6 +30,9 @@ const { consoleRouter } = require('./console');
  const providers = require('./providers');
  const { isFrozen } = require('./ops/common');
  const { createMetrics } = require('./metrics');
++const { createServiceOutbox } = require('openvibe-sdk/events');
++const contracts = require('openvibe-contracts');
++const { eventsProduced } = require('../docs/service-manifest-proposal.json');
  
  const VERSION = require('../package.json').version;
  
@@@ -40,7 +43,14 @@ async function createApp(opts = {}) {
      const log = opts.log || console;
      const keys = opts.keys || createKeyProvider(config, { fetchImpl, log });
      const identity = opts.identity || createIdentity(config, { fetchImpl });
--    const ctx = { db, config, rates: createRates(config.rates), now: opts.now || (() => Date.now()), network: identity, log };
++    const outbox = createServiceOutbox({
++        db, source: 'billing', eventsUrl: config.events.url, networkInternalUrl: config.network.internalUrl,
++        clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, intervalMs: config.events.intervalMs,
++        log, table: 'service_outbox', eventTypes: eventsProduced,
++        validate: (env) => contracts.validate('events.event-envelope@1', env),
++        ...(opts.fetchImpl ? { fetch: fetchImpl } : {}),
++    });
++    const ctx = { db, config, rates: createRates(config.rates), now: opts.now || (() => Date.now()), network: identity, log, outbox };
      const adapters = opts.adapters || providers.createAdapters(config, { fetchImpl, network: identity });
      const auth = createAuth({ config, keys });
  
@@@ -53,7 +63,7 @@ async function createApp(opts = {}) {
      // HTTP golden signals by route template, process metrics, release_info and Billing's gauges;
      // GET /metrics before any other route (loopback only — relayed requests get 404). After the
      // routing settings above: the first app.use() fixes the router's case sensitivity.
--    const metrics = createMetrics({ db, config, now: ctx.now });
++    const metrics = createMetrics({ db, config, now: ctx.now, outbox });
      const release = createRelease({ service: 'billing', root: path.join(__dirname, '..') });
      instrument(app, { service: 'billing', release: release.release, registry: metrics.registry });
      // GET /release.json (ADR-016, D43) and POST /release-metrics: what this deployment is, for tabs and deploy checks.
@@@ -66,7 +76,7 @@ async function createApp(opts = {}) {
      app.get('/api/health', async (req, res) => res.json({
          ok: true, service: 'billing', version: VERSION, frozen: await isFrozen(db), authority: config.authority,
          providers: Object.fromEntries(Object.values(adapters).map((a) => [a.name, a.enabled])),
--        events_relay: !!config.events.url,
++        events_relay: (await outbox.status()).enabled,
      }));
      // Readiness in the openvibe-shared/ready shape (status, named checks, release): 503 when the ledger
      // database or the Network key fails; the money freeze and the authority show without failing it.
@@@ -81,6 +91,7 @@ async function createApp(opts = {}) {
              { name: 'db', required: true, check: async () => (await db.prepare('SELECT 1 AS ok FROM settings WHERE id = 1').get()).ok === 1 || 'settings row missing' },
              { name: 'network_jwks', required: true, check: () => Boolean(keys.get()) || 'Network public key not loaded' },
              { name: 'money_writes', required: false, check: async () => (await isFrozen(db) ? 'frozen' : true) },
++            { name: 'events_outbox', required: false, check: async () => await outbox.status() },
              { name: 'valkey', required: false, check: async () => (valkey ? await valkey.ready() : { skipped: 'VALKEY_URL not set: per-person limits count in this process only' }) },
          ],
      });
ddiff --git a/server/config.js b/server/config.js
iindex 9fc9876..f788485 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -150,7 +150,7 @@ function loadConfig(env = process.env) {
              },
          },
  
--        // Durable events: the outbox relay runs only when EVENTS_URL is set.
++        // Durable events: the SDK relay runs when EVENTS_URL and the OAuth client secret are set.
          events: {
              url: trim(env.EVENTS_URL || ''),
              intervalMs: int(env.EVENTS_RELAY_INTERVAL_MS, 2000),
ddiff --git a/server/index.js b/server/index.js
iindex 47384fc..caa2277 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -17,7 +17,6 @@ const { createApp } = require('./app');
  const providers = require('./providers');
  const subscriptions = require('./ops/subscriptions');
  const rating = require('./ops/rating');
--const { createRelay } = require('./outbox');
  const { isFrozen } = require('./ops/common');
  const { runScheduled } = require('./reconcile');
  const { gracefulStop } = require('openvibe-sdk/service');
@@@ -29,7 +28,7 @@ const { gracefulStop } = require('openvibe-sdk/service');
      keys.start();
  
      const timers = [];
--    const relay = createRelay({ db: ctx.db, config });
++    const relay = ctx.outbox;
      if (config.jobs.enabled) {
          const every = (ms, fn) => { const t = setInterval(() => { Promise.resolve().then(fn).catch((e) => console.warn('[Billing] job:', e.message)); }, ms); t.unref(); timers.push(t); };
          every(config.jobs.webhookRetryMs, async () => await providers.processPending(ctx, adapters));
@@@ -58,7 +57,7 @@ const { gracefulStop } = require('openvibe-sdk/service');
      const server = app.listen(config.port, config.host, async () => {
          const enabled = Object.values(adapters).filter((a) => a.enabled).map((a) => a.name);
          console.log(`[Billing] ${config.nodeEnv} on http://${config.host}:${config.port} → ${config.baseUrl} (PostgreSQL)`);
--        console.log(`[Billing] providers enabled: ${enabled.length ? enabled.join(', ') : 'none'}; economy ${await isFrozen(ctx.db) ? 'FROZEN' : 'open'}; events relay ${config.events.url ? `→ ${config.events.url}` : 'off (outbox accumulates)'}`);
++        console.log(`[Billing] providers enabled: ${enabled.length ? enabled.join(', ') : 'none'}; economy ${await isFrozen(ctx.db) ? 'FROZEN' : 'open'}; events relay ${(await relay.status()).enabled ? `→ ${config.events.url}` : 'off (outbox accumulates)'}`);
          console.log(`[Billing] money authority: ${config.authority}${config.authority === 'live' ? ' (shadow: EXTERNAL tips are recorded, not announced)' : ' (EXTERNAL tips are announced as billing.receipt.external)'}`);
      });
      server.keepAliveTimeout = 65_000;
ddiff --git a/server/metrics.js b/server/metrics.js
iindex 9241af3..ceb69a9 100644
---- a/server/metrics.js
++++ b/server/metrics.js
@@@ -24,7 +24,7 @@ const { createRegistry } = require('openvibe-shared/metrics');
  const { isFrozen } = require('./ops/common');
  const { latest } = require('./reconcile');
  
--function createMetrics({ db, config, now = () => Date.now() }) {
++function createMetrics({ db, config, now = () => Date.now(), outbox }) {
      const registry = createRegistry();
      const one = async (sql, ...a) => (await db.prepare(sql).get(...a)).n;
  
@@@ -54,8 +54,9 @@ function createMetrics({ db, config, now = () => Date.now() }) {
          name: 'billing_external_receipts', help: 'EXTERNAL receipts (tips on a streamer\'s own provider account) by announcement status', labelNames: ['status'],
          collect: async () => (await Promise.all(['announced', 'not_announced'].map(async (status) => ({ labels: { status }, value: await one('SELECT COUNT(*) AS n FROM external_receipts WHERE status = ?', status) })))),
      });
--    registry.gauge({ name: 'billing_outbox_pending', help: 'Outbox events not yet relayed to OpenVibe.Events', collect: async () => await one('SELECT COUNT(*) AS n FROM outbox WHERE sent_at IS NULL') });
--    registry.gauge({ name: 'billing_outbox_failing', help: 'Unsent outbox events whose last relay attempt failed', collect: async () => await one('SELECT COUNT(*) AS n FROM outbox WHERE sent_at IS NULL AND last_error IS NOT NULL') });
++    registry.gauge({ name: 'billing_outbox_pending', help: 'Outbox events not yet relayed to OpenVibe.Events', collect: async () => (await outbox.status()).pending });
++    registry.gauge({ name: 'billing_outbox_failing', help: 'Pending outbox events whose last relay attempt failed', collect: async () => await one('SELECT COUNT(*) AS n FROM service_outbox WHERE sent_at IS NULL AND rejected_at IS NULL AND last_error IS NOT NULL') });
++    registry.gauge({ name: 'billing_outbox_rejected', help: 'Events permanently rejected by OpenVibe.Events', collect: async () => (await outbox.status()).rejected });
      registry.gauge({
          name: 'billing_cashouts', help: 'Cashouts by status', labelNames: ['status'],
          collect: async () => (await Promise.all(['requested', 'paid', 'denied'].map(async (status) => ({ labels: { status }, value: await one('SELECT COUNT(*) AS n FROM cashouts WHERE status = ?', status) })))),
ddiff --git a/server/outbox.js b/server/outbox.js
iindex 93bca5a..9301730 100644
---- a/server/outbox.js
++++ b/server/outbox.js
@@@ -3,12 +3,9 @@
  /**
   * Transactional outbox (events.event-envelope@1). enqueue() is called inside the same
   * transaction as the money movement it announces, so an event exists exactly when its effect
-- * does. The relay publishes unsent rows to OpenVibe.Events (POST /api/v1/events, service token
-- * for audience openvibe.events with events.event.publish) and runs only when EVENTS_URL is set;
-- * without it rows accumulate and can be relayed later. Events deduplicates on event_id, so a
-- * relay retry after a lost response never publishes twice.
++ * does. openvibe-sdk/events stores and relays the envelope from service_outbox.
   */
--const { ids, validate, serviceAuth } = require('openvibe-contracts');
++const { ids, validate } = require('openvibe-contracts');
  
  const ACTOR = { type: 'service', id: 'billing' };
  
@@@ -30,74 +27,8 @@ async function enqueue(ctx, { event_type, subject, payload, priority = 'importan
      if (traceId && /^[0-9a-f]{32}$/.test(traceId)) env.trace_id = traceId;
      const v = validate('events.event-envelope@1', env);
      if (!v.valid) throw new Error(`outbox: invalid envelope for ${event_type}: ${v.errors.map((e) => `${e.path} ${e.message}`).join('; ')}`);
--    await ctx.db.prepare('INSERT INTO outbox (event_id, event, created_at) VALUES (?, ?, ?)').run(env.event_id, JSON.stringify(env), env.timestamp);
++    await ctx.outbox.emit(env);
      return env;
  }
  
--/** Relay loop. Returns { stop, flush }. */
--function createRelay({ db, config, fetchImpl = globalThis.fetch, tokenClient, log = console }) {
--    const tokens = tokenClient || serviceAuth.createTokenClient({
--        tokenUrl: `${config.network.internalUrl}/oauth/token`,
--        clientId: config.oauth.clientId,
--        clientSecret: config.oauth.clientSecret,
--        audience: 'openvibe.events',
--        scope: 'events.event.publish',
--        fetchImpl,
--    });
--    let timer = null;
--    let busy = false;
--
--    async function flush() {
--        if (busy) return { sent: 0, busy: true };
--        busy = true;
--        let sent = 0;
--        try {
--            for (;;) {
--                const rows = await db.prepare('SELECT seq, event_id, event FROM outbox WHERE sent_at IS NULL ORDER BY seq LIMIT 100').all();
--                if (!rows.length) break;
--                const body = { events: rows.map((r) => JSON.parse(r.event)) };
--                let res;
--                try {
--                    res = await fetchImpl(`${config.events.url}/api/v1/events`, {
--                        method: 'POST',
--                        headers: { 'Content-Type': 'application/json', ...(await tokens.authHeaders()) },
--                        body: JSON.stringify(body),
--                        signal: AbortSignal.timeout(10000),
--                    });
--                } catch (e) {
--                    await markFailed(rows, e.message);
--                    break;
--                }
--                if (res.status === 401) tokens.invalidate();
--                if (!res.ok) {
--                    const text = await res.text().catch(() => '');
--                    await markFailed(rows, `${res.status} ${text.slice(0, 300)}`);
--                    break;
--                }
--                const at = new Date().toISOString();
--                const mark = db.prepare('UPDATE outbox SET sent_at = ?, attempts = attempts + 1, last_error = NULL WHERE seq = ?');
--                await db.tx(async () => { for (const r of rows) await mark.run(at, r.seq); });
--                sent += rows.length;
--            }
--        } finally {
--            busy = false;
--        }
--        return { sent };
--    }
--
--    async function markFailed(rows, message) {
--        const mark = db.prepare('UPDATE outbox SET attempts = attempts + 1, last_error = ? WHERE seq = ?');
--        await db.tx(async () => { for (const r of rows) await mark.run(String(message).slice(0, 500), r.seq); });
--        log.warn(`[Billing] outbox relay: ${rows.length} event(s) not published: ${message}`);
--    }
--
--    function start() {
--        if (timer || !config.events.url) return;
--        timer = setInterval(() => { flush().catch((e) => log.warn('[Billing] outbox relay:', e.message)); }, config.events.intervalMs);
--        if (timer.unref) timer.unref();
--    }
--    function stop() { if (timer) clearInterval(timer); timer = null; }
--    return { start, stop, flush };
--}
--
--module.exports = { enqueue, createRelay };
++module.exports = { enqueue };
ddiff --git a/server/reconcile.js b/server/reconcile.js
iindex a03b17b..0e85555 100644
---- a/server/reconcile.js
++++ b/server/reconcile.js
@@@ -100,9 +100,10 @@ async function reconcile(ctx, { store = true, trigger = 'manual' } = {}) {
  
      const extBooked = await db.prepare('SELECT x.receipt_ref, t.id AS txn_id FROM external_receipts x JOIN transactions t ON t.receipt_ref = x.receipt_ref').all();
      const extUnsent = await db.prepare(`SELECT x.receipt_ref, x.event_id FROM external_receipts x
--        WHERE x.status = 'announced' AND (x.event_id IS NULL OR NOT EXISTS (SELECT 1 FROM outbox o WHERE o.event_id = x.event_id))`).all();
--    const extTwice = await db.prepare(`SELECT json_extract(event, '$.subject.id') AS receipt_ref, COUNT(*) AS events FROM outbox
--        WHERE json_extract(event, '$.event_type') = 'billing.receipt.external' GROUP BY 1 HAVING COUNT(*) > 1`).all();
++        WHERE x.status = 'announced' AND (x.event_id IS NULL OR NOT EXISTS (SELECT 1 FROM service_outbox o WHERE o.event_id = x.event_id)
++            AND NOT EXISTS (SELECT 1 FROM outbox legacy WHERE legacy.event_id = x.event_id AND legacy.sent_at IS NOT NULL))`).all();
++    const extTwice = await db.prepare(`SELECT (envelope #>> '{subject,id}') AS receipt_ref, COUNT(*) AS events FROM service_outbox
++        WHERE (envelope #>> '{event_type}') = 'billing.receipt.external' GROUP BY 1 HAVING COUNT(*) > 1`).all();
      const extOffenders = [...extBooked.map((r) => ({ ...r, problem: 'booked in the journal' })), ...extUnsent.map((r) => ({ ...r, problem: 'announced without an outbox event' })),
          ...extTwice.map((r) => ({ ...r, problem: 'announced more than once' }))];
      check('receipts.external', !extOffenders.length, extOffenders.length ? { offenders: extOffenders }
ddiff --git a/test/auth-outbox.test.js b/test/auth-outbox.test.js
iindex e15144d..1740222 100644
---- a/test/auth-outbox.test.js
++++ b/test/auth-outbox.test.js
@@@ -9,7 +9,8 @@ const { boot, fund, check, done } = require('./helpers/app');
  const { startEvents } = require('./helpers/stubs');
  
  (async () => {
--    const t = await boot();
++    const events = await startEvents();
++    const t = await boot({ env: { EVENTS_URL: events.url } });
      const a = t.user(51);
      const b = t.user(52);
      console.log('auth, outbox, reconciliation');
@@@ -42,7 +43,7 @@ const { startEvents } = require('./helpers/stubs');
          assert.strictEqual((await t.call('POST', '/api/v1/transfers', { body: { from: a, to: b, amount: 600 } })).status, 201);
          await t.call('POST', '/api/v1/cashouts', { body: { subject: b, amount: 500, payout_method: { type: 'paypal', address: '[email protected]' } } });
          assert.strictEqual((await t.call('POST', '/api/v1/subscriptions', { body: { subscriber: a, streamer: b, source: 'credit', auto_renew: false } })).status, 201);
--        const rows = (await t.db.prepare('SELECT event FROM outbox ORDER BY seq').all()).map((r) => JSON.parse(r.event));
++        const rows = (await t.db.prepare('SELECT envelope AS event FROM service_outbox ORDER BY id').all()).map((r) => typeof r.event === 'string' ? JSON.parse(r.event) : r.event);
          const types = rows.map((e) => e.event_type);
          for (const type of ['billing.transaction.settled', 'billing.cashout.requested', 'billing.entitlement.changed']) assert.ok(types.includes(type), `${type} in ${types}`);
          for (const e of rows) {
@@@ -54,25 +55,23 @@ const { startEvents } = require('./helpers/stubs');
          const txnCount = (await t.db.prepare('SELECT COUNT(*) AS n FROM transactions').get()).n;
          const refused = await t.call('POST', '/api/v1/transfers', { body: { from: a, to: b, amount: 999999 } });
          assert.strictEqual(refused.status, 409);
--        assert.strictEqual((await t.db.prepare('SELECT COUNT(*) AS n FROM outbox').get()).n, rows.length, 'a refused operation writes no event');
++        assert.strictEqual((await t.db.prepare('SELECT COUNT(*) AS n FROM service_outbox').get()).n, rows.length, 'a refused operation writes no event');
          assert.strictEqual((await t.db.prepare('SELECT COUNT(*) AS n FROM transactions').get()).n, txnCount);
      });
  
      await check('the relay publishes unsent events to OpenVibe.Events with an events.event.publish token', async () => {
--        const events = await startEvents();
--        const { createRelay } = require('../server/outbox');
--        const relay = createRelay({ db: t.db, config: { ...t.config, events: { url: events.url, intervalMs: 1000 } }, log: { warn() {} } });
--        const n = (await t.db.prepare('SELECT COUNT(*) AS n FROM outbox WHERE sent_at IS NULL').get()).n;
--        const r = await relay.flush();
++        const relay = t.ctx.outbox;
++        assert.strictEqual((await relay.status()).enabled, true);
++        const n = (await t.db.prepare('SELECT COUNT(*) AS n FROM service_outbox WHERE sent_at IS NULL AND rejected_at IS NULL').get()).n;
++        const r = await relay.outbox.flush();
          assert.strictEqual(r.sent, n);
--        assert.strictEqual((await t.db.prepare('SELECT COUNT(*) AS n FROM outbox WHERE sent_at IS NULL').get()).n, 0);
++        assert.strictEqual((await t.db.prepare('SELECT COUNT(*) AS n FROM service_outbox WHERE sent_at IS NULL AND rejected_at IS NULL').get()).n, 0);
          const token = events.tokens[0].slice(7);
          const v = serviceAuth.verifyServiceToken(token, { publicKey: t.network.publicPem, audience: 'openvibe.events' });
          assert.ok(v.ok, v.reason);
          assert.deepStrictEqual(v.claims.cap, ['events.event.publish']);
          assert.strictEqual(v.claims.sub, 'svc:billing');
--        assert.strictEqual((await relay.flush()).sent, 0);
--        await events.close();
++        assert.strictEqual((await relay.outbox.flush()).sent, 0);
      });
  
      await check('GET /admin/reconcile stores a run and detects a deliberately corrupted cached balance', async () => {
@@@ -89,5 +88,6 @@ const { startEvents } = require('./helpers/stubs');
      });
  
      await t.close();
++    await events.close();
      done();
  })();
ddiff --git a/test/console.test.js b/test/console.test.js
iindex c597876..06c3905 100644
---- a/test/console.test.js
++++ b/test/console.test.js
@@@ -54,7 +54,7 @@ const DAY = 86_400_000;
      const csrfOf = (page) => (page.text.match(/name="_csrf" value="([^"]+)"/) || [])[1];
      const keyOf = (page, action) => (page.text.match(new RegExp(`action="${action}"><input type="hidden" name="_csrf" value="[^"]+"><input type="hidden" name="action_key" value="([^"]+)"`)) || [])[1];
      const auditRows = async (action, outcome = 'done') => await t.db.prepare('SELECT * FROM staff_audit WHERE action = ? AND outcome = ?').all(action, outcome);
--    const staffEvents = async (action) => (await t.db.prepare("SELECT event FROM outbox WHERE json_extract(event, '$.event_type') = 'billing.staff.action' AND json_extract(event, '$.payload.action') = ?").all(action)).map((r) => JSON.parse(r.event));
++    const staffEvents = async (action) => (await t.db.prepare("SELECT envelope AS event FROM service_outbox WHERE (envelope #>> '{event_type}') = 'billing.staff.action' AND (envelope #>> '{payload,action}') = ?").all(action)).map((r) => (typeof r.event === 'string' ? JSON.parse(r.event) : r.event));
      async function newCashout(amount = 600) {
          const r = await t.call('POST', '/api/v1/cashouts', { body: { subject: creator, amount, payout_method: { type: 'paypal', address: '[email protected]' } } });
          assert.strictEqual(r.status, 201, r.text);
@@@ -231,7 +231,7 @@ const DAY = 86_400_000;
          assert.deepStrictEqual(ev[0].actor, { type: 'user', id: staff });
          assert.strictEqual(ev[0].visibility, 'internal');
          assert.deepStrictEqual(ev[0].payload.target, { type: 'cashout', id: co1 });
--        assert.strictEqual((await t.db.prepare("SELECT COUNT(*) AS n FROM outbox WHERE json_extract(event, '$.event_type') = 'billing.cashout.paid'").get()).n, 1);
++        assert.strictEqual((await t.db.prepare("SELECT COUNT(*) AS n FROM service_outbox WHERE (envelope #>> '{event_type}') = 'billing.cashout.paid'").get()).n, 1);
          const shown = await get(`/cashouts/${co1}?done=approved`, s);
          assert.ok(shown.text.includes('Payout recorded') && shown.text.includes('PAYOUT-7Q2'));
          await t.assertReconciled('after a console approval');
ddiff --git a/test/external.test.js b/test/external.test.js
iindex 26bcd9b..51bd4df 100644
---- a/test/external.test.js
++++ b/test/external.test.js
@@@ -14,7 +14,7 @@ const { boot, check, done } = require('./helpers/app');
  const { startEvents } = require('./helpers/stubs');
  
  const donation = (data, streamer = { id: 'pc-77', username: 'StreamerPC' }) => ({ type: 'donation.completed', streamer, data: { eventId: `don-${Math.random().toString(36).slice(2)}`, ...data } });
--const externalEvents = async (db) => (await db.prepare("SELECT event FROM outbox WHERE json_extract(event, '$.event_type') = 'billing.receipt.external' ORDER BY seq").all()).map((r) => JSON.parse(r.event));
++const externalEvents = async (db) => (await db.prepare("SELECT envelope AS event FROM service_outbox WHERE (envelope #>> '{event_type}') = 'billing.receipt.external' ORDER BY id").all()).map((r) => typeof r.event === 'string' ? JSON.parse(r.event) : r.event);
  
  (async () => {
      console.log('external receipts');
@@@ -155,9 +155,10 @@ const externalEvents = async (db) => (await db.prepare("SELECT event FROM outbox
  
      await check('the outbox relay publishes billing.receipt.external to OpenVibe.Events', async () => {
          const events = await startEvents();
--        const { createRelay } = require('../server/outbox');
--        const relay = createRelay({ db: t.db, config: { ...t.config, events: { url: events.url, intervalMs: 1000 } }, log: { warn() {} } });
--        await relay.flush();
++        const { createServiceOutbox } = require('openvibe-sdk/events');
++        const relay = createServiceOutbox({ db: t.db, source: 'billing', table: 'service_outbox', eventsUrl: events.url,
++            networkInternalUrl: t.config.network.internalUrl, clientId: t.config.oauth.clientId, clientSecret: t.config.oauth.clientSecret, log: { warn() {} } });
++        await relay.outbox.flush();
          const sent = events.batches.flatMap((b) => b.events).filter((e) => e.event_type === 'billing.receipt.external');
          assert.strictEqual(sent.length, (await externalEvents(t.db)).length);
          assert.strictEqual(sent[0].event_id, first.event_id);
ddiff --git a/test/helpers/stubs.js b/test/helpers/stubs.js
iindex 886e5db..a817819 100644
---- a/test/helpers/stubs.js
++++ b/test/helpers/stubs.js
@@@ -123,8 +123,10 @@ async function startEvents() {
          if (req.url === '/api/v1/events' && req.method === 'POST') {
              tokens.push(req.headers.authorization);
              const body = JSON.parse(raw);
--            batches.push(body);
--            return send(res, 201, { results: (body.events || []).map((e, i) => ({ event_id: e.event_id, seq: i + 1, duplicate: false })) });
++            batches.push(body.events ? body : { events: [body] });
++            return send(res, 201, body.events
++                ? { results: body.events.map((e, i) => ({ event_id: e.event_id, seq: i + 1, duplicate: false })) }
++                : { event_id: body.event_id, seq: 1, duplicate: false });
          }
          send(res, 404, {});
      });
ddiff --git a/test/hygiene.test.js b/test/hygiene.test.js
iindex 84059e2..eae2759 100644
---- a/test/hygiene.test.js
++++ b/test/hygiene.test.js
@@@ -199,11 +199,11 @@ const donation = (data, streamer = 'openvibe') => ({ type: 'donation.completed',
          // Marking the intent settled comes after the journal posting, in the same transaction: failing
          // it must take the posting, the outbox event and the processed mark down with it.
          intents.markSettled = async (...a) => { if (fails-- > 0) throw new Error('disk full'); return await real(...a); };
--        const events = (await t.db.prepare('SELECT COUNT(*) AS n FROM outbox').get()).n;
++        const events = (await t.db.prepare('SELECT COUNT(*) AS n FROM service_outbox').get()).n;
          let r;
          try { r = await t.powerchat(donation({ eventId: 'don-crash', amountUsdCents: 150, appExternalRef: i.checkout_ref })); }
          finally { intents.markSettled = real; }
--        assert.strictEqual((await t.db.prepare('SELECT COUNT(*) AS n FROM outbox').get()).n, events, 'no event for a settlement that did not happen');
++        assert.strictEqual((await t.db.prepare('SELECT COUNT(*) AS n FROM service_outbox').get()).n, events, 'no event for a settlement that did not happen');
          assert.strictEqual(r.json.processed, false);
          assert.strictEqual(await txnCount(), before);
          assert.strictEqual((await t.balances(buyer.id)).credit, credit);
@@@ -216,12 +216,13 @@ const donation = (data, streamer = 'openvibe') => ({ type: 'donation.completed',
      });
  
      await check('OpenVibe.Events down: settlement is unaffected, events wait in the outbox and go out once it is back', async () => {
--        const { createRelay } = require('../server/outbox');
++        const { createServiceOutbox } = require('openvibe-sdk/events');
          const log = { warn() {} };
--        const unsent = async () => (await t.db.prepare('SELECT COUNT(*) AS n FROM outbox WHERE sent_at IS NULL').get()).n;
--        const down = createRelay({ db: t.db, config: { ...t.config, events: { url: 'http://127.0.0.1:9', intervalMs: 1000 } }, log });
++        const unsent = async () => (await t.db.prepare('SELECT COUNT(*) AS n FROM service_outbox WHERE sent_at IS NULL AND rejected_at IS NULL').get()).n;
++        const down = createServiceOutbox({ db: t.db, source: 'billing', table: 'service_outbox', eventsUrl: 'http://127.0.0.1:9',
++            networkInternalUrl: t.config.network.internalUrl, clientId: t.config.oauth.clientId, clientSecret: t.config.oauth.clientSecret, log });
          await fund(t, buyer, 200);
--        const r = await down.flush();
++        const r = await down.outbox.flush();
          assert.strictEqual(r.sent, 0);
          const waiting = await unsent();
          assert.ok(waiting > 0);
@@@ -229,8 +230,10 @@ const donation = (data, streamer = 'openvibe') => ({ type: 'donation.completed',
          assert.match(m, new RegExp(`billing_outbox_pending ${waiting}\\b`));
          assert.match(m, /billing_outbox_failing [1-9]/);
          const events = await startEvents();
--        const up = createRelay({ db: t.db, config: { ...t.config, events: { url: events.url, intervalMs: 1000 } }, log });
--        await up.flush();
++        const up = createServiceOutbox({ db: t.db, source: 'billing', table: 'service_outbox', eventsUrl: events.url,
++            networkInternalUrl: t.config.network.internalUrl, clientId: t.config.oauth.clientId, clientSecret: t.config.oauth.clientSecret, log });
++        await t.db.prepare('UPDATE service_outbox SET next_attempt_at = 0 WHERE sent_at IS NULL').run();
++        await up.outbox.flush();
          assert.strictEqual(await unsent(), 0);
          const ids = events.batches.flatMap((b) => b.events).map((e) => e.event_id);
          assert.strictEqual(new Set(ids).size, ids.length, 'each event published once');
ddiff --git a/test/operations.test.js b/test/operations.test.js
iindex ba34f1c..1fa4f62 100644
---- a/test/operations.test.js
++++ b/test/operations.test.js
@@@ -142,7 +142,7 @@ const { boot, fund, check, done } = require('./helpers/app');
          assert.deepStrictEqual([b.payable, b.pending_payouts], [250, 1000]);
          await t.assertReconciled('after cashout request');
          // The event stream gets the payout TYPE only, never the creator's address (a PayPal email).
--        const outboxText = JSON.stringify(await t.db.prepare("SELECT event FROM outbox WHERE event ILIKE '%billing.cashout.%'").all());
++        const outboxText = JSON.stringify(await t.db.prepare("SELECT envelope AS event FROM service_outbox WHERE envelope::text ILIKE '%billing.cashout.%'").all());
          assert.ok(outboxText.includes('billing.cashout.requested'), 'the requested event is queued');
          assert.ok(!outboxText.includes('[email protected]'), 'no payout address in any cashout event');
          assert.strictEqual(cashout.payout_method.address, '[email protected]', 'the API itself still returns it to the caller');
ddiff --git a/test/renewal-grace.test.js b/test/renewal-grace.test.js
iindex a1d1823..e286fdb 100644
---- a/test/renewal-grace.test.js
++++ b/test/renewal-grace.test.js
@@@ -30,7 +30,7 @@ const DAY = 86_400_000;
      };
      const row = async (id) => await subs.find(t.db, id);
      const renewals = async (id) => (await t.db.prepare("SELECT idempotency_key FROM transactions WHERE type = 'subscription' AND idempotency_key LIKE ? ORDER BY created_at").all(`renew:${id}:%`)).map((r) => r.idempotency_key);
--    const changed = async (u) => (await t.db.prepare('SELECT event FROM outbox ORDER BY seq').all()).map((r) => JSON.parse(r.event))
++    const changed = async (u) => (await t.db.prepare('SELECT envelope AS event FROM service_outbox ORDER BY id').all()).map((r) => (typeof r.event === 'string' ? JSON.parse(r.event) : r.event))
          .filter((e) => e.event_type === 'billing.entitlement.changed' && e.subject.id.startsWith(`${u.id}:`)).map((e) => e.payload);
      const ent = async (u) => (await t.call('GET', `/api/v1/entitlements/${u.id}?streamer=${streamer.id}`)).json;
      const payable = async () => (await t.balances(streamer.id)).payable;
ddiff --git a/test/sdk-outbox-migration.test.js b/test/sdk-outbox-migration.test.js
nnew file mode 100644
iindex 0000000..022d3dd
---- /dev/null
++++ b/test/sdk-outbox-migration.test.js
@@@ -0,0 +1,85 @@
++'use strict';
++/** The expand migration carries pending legacy events into the SDK relay without touching the old table. */
++const assert = require('assert');
++const fs = require('fs');
++const os = require('os');
++const path = require('path');
++const { createDb } = require('openvibe-sdk/db');
++const { createServiceOutbox, outboxSchema } = require('openvibe-sdk/events');
++
++(async () => {
++    const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'billing-outbox-migrate-'));
++    const migrations = path.join(__dirname, '..', 'migrations');
++    const db = createDb({ pglite: true, service: 'billing-outbox-test' });
++    try {
++        for (const file of fs.readdirSync(migrations).filter((f) => /^000[1-5]_/.test(f))) {
++            fs.copyFileSync(path.join(migrations, file), path.join(dir, file));
++        }
++        await db.migrate({ dir, log: { log() {} } });
++        const event = { event_id: 'evt_legacy_pending', event_type: 'billing.transaction.settled', source: 'billing',
++            timestamp: '2026-10-09T12:34:56.000Z', version: 1, payload: {} };
++        await db.query('INSERT INTO outbox (event_id, event, created_at, attempts) VALUES ($1, $2, $3, $4)',
++            [event.event_id, JSON.stringify(event), event.timestamp, 2]);
++        const sent = { ...event, event_id: 'evt_legacy_sent' };
++        await db.query('INSERT INTO outbox (event_id, event, created_at, sent_at) VALUES ($1, $2, $3, $4)',
++            [sent.event_id, JSON.stringify(sent), sent.timestamp, event.timestamp]);
++        const ddl = fs.readFileSync(path.join(migrations, '0006_sdk_outbox.sql'), 'utf8');
++        assert.ok(ddl.includes(outboxSchema('service_outbox')));
++        fs.copyFileSync(path.join(migrations, '0006_sdk_outbox.sql'), path.join(dir, '0006_sdk_outbox.sql'));
++        await db.migrate({ dir, log: { log() {} } });
++        const rows = await db.many('SELECT event_id, envelope, created_at, attempts FROM service_outbox');
++        assert.strictEqual(rows.length, 1);
++        assert.strictEqual(rows[0].event_id, event.event_id);
++        assert.strictEqual(rows[0].attempts, 2);
++        assert.strictEqual(Number(rows[0].created_at), Date.parse(event.timestamp));
++
++        const off = createServiceOutbox({ db, source: 'billing', table: 'service_outbox' });
++        off.start();
++        assert.deepStrictEqual(await off.status(), { enabled: false, pending: 1, rejected: 0, last_error: null });
++        await off.stop();
++        assert.strictEqual(await db.value('SELECT count(*) FROM service_outbox WHERE sent_at IS NULL'), 1);
++
++        const posted = [];
++        let rejectBad = false;
++        const fetch = async (url, options) => {
++            if (url.endsWith('/oauth/token')) return new Response(JSON.stringify({ access_token: 'service-token', expires_in: 300 }), { status: 200 });
++            const body = JSON.parse(options.body);
++            posted.push({ url, authorization: options.headers.Authorization || options.headers.authorization, body });
++            const events = body.events || [body];
++            if (rejectBad && events.some((e) => e.event_id === 'evt_bad')) {
++                return new Response(JSON.stringify({ code: 'events.invalid_envelope' }), { status: 422 });
++            }
++            return new Response(JSON.stringify(body.events
++                ? { results: events.map((e) => ({ event_id: e.event_id, seq: 7, duplicate: false })) }
++                : { event_id: body.event_id, seq: 7, duplicate: false }), { status: 201 });
++        };
++        const outbox = createServiceOutbox({ db, source: 'billing', table: 'service_outbox', eventsUrl: 'http://events.test',
++            networkInternalUrl: 'http://network.test', clientId: 'billing', clientSecret: 'secret', fetch, autoDiscover: false });
++        assert.strictEqual((await outbox.status()).pending, 1);
++        assert.deepStrictEqual(await outbox.outbox.flush(), { sent: 1, failed: 0, rejected: 0 });
++        assert.strictEqual(posted.length, 1);
++        assert.strictEqual(posted[0].body.event_id, event.event_id);
++        assert.strictEqual((await outbox.status()).pending, 0);
++        assert.strictEqual(await db.value('SELECT count(*) FROM outbox'), 2);
++        const later = { ...event, event_id: 'evt_new_committed' };
++        await db.tx(async () => { await outbox.emit(later); });
++        assert.strictEqual(await db.value("SELECT count(*) FROM service_outbox WHERE event_id = 'evt_new_committed'"), 1);
++        await assert.rejects(db.tx(async () => {
++            await outbox.emit({ ...event, event_id: 'evt_new_rolled_back' });
++            throw new Error('roll back the change');
++        }), /roll back the change/);
++        assert.strictEqual(await db.value("SELECT count(*) FROM service_outbox WHERE event_id = 'evt_new_rolled_back'"), 0);
++        rejectBad = true;
++        await db.tx(async () => {
++            await outbox.emit({ ...event, event_id: 'evt_bad' });
++            await outbox.emit({ ...event, event_id: 'evt_after_bad' });
++        });
++        assert.deepStrictEqual(await outbox.outbox.flush(), { sent: 2, failed: 0, rejected: 1 });
++        assert.strictEqual((await outbox.status()).rejected, 1);
++        assert.strictEqual((await outbox.status()).pending, 0);
++        console.log('sdk outbox migration: all passed');
++    } finally {
++        await db.close();
++        fs.rmSync(dir, { recursive: true, force: true });
++    }
++})().catch((err) => { console.error(err); process.exitCode = 1; });
ddiff --git a/test/subscription-refund.test.js b/test/subscription-refund.test.js
iindex 5ebaed4..c82c386 100644
---- a/test/subscription-refund.test.js
++++ b/test/subscription-refund.test.js
@@@ -29,7 +29,7 @@ const DAY = 86_400_000;
      const ent = async (u) => (await t.call('GET', `/api/v1/entitlements/${u.id}?streamer=${streamer.id}`)).json;
      const revenueBits = async () => (await t.db.prepare(`SELECT COALESCE(SUM(e.amount), 0)::bigint AS n FROM ledger_entries e JOIN accounts a ON a.id = e.account_id
          WHERE a.kind = 'platform_revenue' AND a.currency = 'vibes-bits'`).get()).n;
--    const reversedEvents = async () => (await t.db.prepare('SELECT event FROM outbox ORDER BY seq').all()).map((r) => JSON.parse(r.event))
++    const reversedEvents = async () => (await t.db.prepare('SELECT envelope AS event FROM service_outbox ORDER BY id').all()).map((r) => (typeof r.event === 'string' ? JSON.parse(r.event) : r.event))
          .filter((e) => e.event_type === 'billing.transaction.reversed');
      const periodCheck = async () => (await reconcile(t.ctx, { store: false })).checks.find((c) => c.id === 'subscriptions.period_charged');