Key the ordered inbox by an unambiguous, untruncated subject key (#10)

ddiff --git a/README.md b/README.md
iindex 1ebf206..e17f404 100644
---- a/README.md
++++ b/README.md
@@@ -106,7 +106,7 @@ POST /api/v1/subscriptions
  
  Topic patterns are dot-separated segments where `*` stands for one or more whole segments (`media.vod.*`, `*.created`, `media.*.ready`, `*`). Endpoints must be `http(s)` on `127.0.0.1` or `openvibe.<tld>`/its subdomains (`EVENTS_ENDPOINT_HOSTS`); redirects are never followed. A caller may supply its own `secret` (a string of 32..256 characters; the same bound on `POST /api/v1/subscriptions/:id/rotate-secret`), otherwise Events generates a `whsec_…` one; either way it is returned once, on creation, and the consumer needs it to verify the v2 signature. The consumer is the calling service, and re-creating a subscription it already holds for the same `topic_pattern` and `endpoint` answers `409 events.subscription_exists` with the existing `subscription_id`, so an ensure-at-boot script re-runs safely.
  
--Each delivery is `POST <endpoint>` with body `{ "event": <envelope>, "seq": n }` and headers `X-OpenVibe-Event-Id`, `X-OpenVibe-Event-Type`, `X-OpenVibe-Seq`, `X-OpenVibe-Subscription-Id`, `X-OpenVibe-Delivery-Attempt`, `X-OpenVibe-Timestamp: <unix seconds>`, `X-OpenVibe-Signature-V2: t=<that timestamp>,v2=<HMAC-SHA256 of "<t>.<raw body>">` and `traceparent` (the event's trace). During secret rotation, the v2 header also carries a second `v2=` signature made with the previous secret. Any 2xx is delivered. Anything else is retried after 1 s, 5 s, 30 s, 2 min, 10 min, 1 h (then hourly) up to 8 attempts, after which the delivery is `dead`. Priority classes go first (`critical`, `important`, `low`, then seq); one delivery per subscription is in flight at a time and at most 20 overall. Delivery is at least once and not strictly ordered; consumers dedupe with an inbox and order with `subject.revision`.
++Each delivery is `POST <endpoint>` with body `{ "event": <envelope>, "seq": n }` and headers `X-OpenVibe-Event-Id`, `X-OpenVibe-Event-Type`, `X-OpenVibe-Seq`, `X-OpenVibe-Subscription-Id`, `X-OpenVibe-Delivery-Attempt`, `X-OpenVibe-Timestamp: <unix seconds>`, `X-OpenVibe-Signature-V2: t=<that timestamp>,v2=<HMAC-SHA256 of "<t>.<raw body>">` and `traceparent` (the event's trace). During secret rotation, the v2 header also carries a second `v2=` signature made with the previous secret. Any 2xx is delivered. Anything else is retried after 1 s, 5 s, 30 s, 2 min, 10 min, 1 h (then hourly) up to 8 attempts, after which the delivery is `dead`. Priority classes go first (`critical`, `important`, `low`, then seq); one delivery per subscription is in flight at a time and at most 20 overall. Delivery is at least once and not strictly ordered; consumers dedupe with an inbox and order with `subject.revision` (`createPgOrderedInbox` below does both).
  
  **Replay window (signature v2).** Events sends only the v2 signature. It signs the timestamp together with the raw body, and every attempt, retries included, is signed afresh with the time it is sent. Consumers check it with `verifyDeliveryV2(raw, headers, secret, { toleranceSec = 300, now })` (here) or openvibe-sdk ≥ 0.4.0 `parseDelivery(raw, headers, secret, { requireV2: true })` and reject anything more than 300 s from their clock either way. Reject a missing, wrong or stale v2 signature; do not fall back to v1.
  
@@@ -209,6 +209,14 @@ in one transaction, so a crash before commit leaves nothing behind and the redel
  commit makes the redelivery a no-op. The outbox and inbox are openvibe-sdk's PostgreSQL kits (ADR-042): this package
  re-exports them and keeps no SQLite copies.
  
++`createPgOrderedInbox(db)` adds a per-key head (the table from `events.inboxHeadsSchema()`, beside the inbox's): `apply(consumer, event, fn)`
++records the receipt, moves the head of the subject's key (`events.subjectKey(event)`: `["<type>","<id>"]`, or its `sha256:` past 200
++characters, never truncated) to `subject.revision` only if that is higher, and runs
++`fn(t)`, all in one transaction. It answers `{ duplicate: true }`, `{ stale: true, head }` (an equal or older revision, for
++example a dead delivery replayed after its successor: the receipt is kept and `fn` does not run, so answer 2xx) or
++`{ stale: false, result }`. `{ key, revision }` overrides both; an event without them is only deduplicated. Concurrent
++applies on one key serialise on its head row, so effects commit in increasing revisions.
++
  ## Realtime (SSE)
  
  ```js
ddiff --git a/lib/client.js b/lib/client.js
iindex d0f927f..cf1b334 100644
---- a/lib/client.js
++++ b/lib/client.js
@@@ -21,6 +21,10 @@
   *       res.sendStatus(204);
   *   });
   *
++ *   // Consumer that must not apply an older revision of a subject after a newer one (replays, dead predecessors)
++ *   const ordered = events.createPgOrderedInbox(db);                       // tables from inboxSchema() + inboxHeadsSchema()
++ *   const r = await ordered.apply('live', req.body.event, async (t) => { ...writes on t... });   // r.stale: superseded
++ *
   * The outbox and inbox are openvibe-sdk's PostgreSQL kits (ADR-042 decision 9): this package no longer carries its
   * own SQLite copies.
   *
@@@ -189,4 +193,95 @@ function createPublisher({ eventsUrl, tokenClient, fetchImpl = globalThis.fetch,
  
  const { createPgOutbox, createPgInbox, outboxSchema, inboxSchema } = require('openvibe-sdk/events');
  
--module.exports = { sign, verifyDelivery, signV2, verifyDeliveryV2, prepareEnvelope, createPublisher, createPgOutbox, createPgInbox, outboxSchema, inboxSchema, PublishError };
++// ── Ordered inbox: the inbox plus a per-key head ───────────
++//
++// Events delivers at least once and, outside a policy's ordering key, not in order; a replay or a dead
++// delivery's successor can also arrive after a later revision. The ordered inbox keeps, per (consumer,
++// key), the highest subject.revision applied, in the same transaction as the receipt and the effects.
++
++const TABLE_RE = /^[a-z_][a-z0-9_]{0,62}$/;
++const MAX_KEY_LEN = 200;
++
++/** The key-head table's DDL, for a service migration beside events.inboxSchema(). */
++function inboxHeadsSchema(table = 'inbox_key_heads') {
++    if (!TABLE_RE.test(table)) throw new TypeError('bad inbox heads table name');
++    return `CREATE TABLE IF NOT EXISTS ${table} (
++    consumer   text NOT NULL,
++    key        text NOT NULL,
++    revision   bigint NOT NULL,
++    event_id   text NOT NULL,
++    applied_at bigint NOT NULL,
++    PRIMARY KEY (consumer, key)
++);`;
++}
++
++/**
++ * The subject's inbox key, `["<subject.type>","<subject.id>"]`; null without one. Unlike a delivery policy's ordering
++ * key it is never truncated and a `:` cannot join two subjects: one longer than 200 characters is `sha256:<hex>` of that
++ * form, so distinct subjects never share a head.
++ */
++function subjectKey(event) {
++    const s = event && event.subject;
++    if (!s || typeof s.type !== 'string' || !s.type || typeof s.id !== 'string' || !s.id) return null;
++    const k = JSON.stringify([s.type, s.id]);
++    return k.length > MAX_KEY_LEN ? `sha256:${crypto.createHash('sha256').update(k).digest('hex')}` : k;
++}
++
++/**
++ * createPgOrderedInbox(db, { table?, headsTable?, now? }) on an openvibe-sdk/db handle:
++ *   once(consumer, eventId, fn) and seen(consumer, eventId), as createPgInbox;
++ *   apply(consumer, event, fn, { key?, revision? }) -> { duplicate: true }
++ *                                                    | { duplicate: false, stale: true, head: { revision, event_id } }
++ *                                                    | { duplicate: false, stale: false, result }
++ * apply records the receipt, then moves the key's head to `revision` only if it is higher than the head, then runs
++ * fn(t) — all in one transaction. An event at or below the head is stale: its receipt is kept (a redelivery is a
++ * duplicate) and fn does not run, so a consumer answers 2xx either way. `key` defaults to subjectKey(event) and
++ * `revision` to event.subject.revision; without both the event is only deduplicated. Concurrent applies on one key
++ * serialise on its head row.
++ */
++function createPgOrderedInbox(db, { table = 'idempotency_receipts', headsTable = 'inbox_key_heads', now = () => Date.now() } = {}) {
++    if (!TABLE_RE.test(headsTable)) throw new TypeError('bad inbox heads table name');
++    const inbox = createPgInbox(db, { table, now });
++
++    async function apply(consumer, event, fn, { key = subjectKey(event), revision = event && event.subject && event.subject.revision } = {}) {
++        if (!event || typeof event !== 'object' || !event.event_id) throw new TypeError('event with an event_id is required');
++        if (typeof fn !== 'function') throw new TypeError('fn is required');
++        const ordered = key != null && revision != null;
++        if (ordered) {
++            if (typeof key !== 'string' || !key || key.length > MAX_KEY_LEN) throw new TypeError(`key must be a string of 1..${MAX_KEY_LEN} characters`);
++            if (!Number.isSafeInteger(revision) || revision < 0) throw new TypeError('revision must be an integer >= 0');
++        }
++        const r = await inbox.once(consumer, event.event_id, async (t) => {
++            if (ordered) {
++                const moved = await t.exec(
++                    `INSERT INTO ${headsTable} (consumer, key, revision, event_id, applied_at) VALUES ($1, $2, $3, $4, $5)
++                     ON CONFLICT (consumer, key) DO UPDATE SET revision = EXCLUDED.revision, event_id = EXCLUDED.event_id, applied_at = EXCLUDED.applied_at
++                     WHERE ${headsTable}.revision < EXCLUDED.revision`,
++                    [String(consumer), key, revision, String(event.event_id), now()]);
++                if (moved === 0) {
++                    const head = await t.one(`SELECT revision, event_id FROM ${headsTable} WHERE consumer = $1 AND key = $2`, [String(consumer), key]);
++                    return { stale: true, head: { revision: Number(head.revision), event_id: head.event_id } };
++                }
++            }
++            return { stale: false, result: await fn(t) };
++        });
++        return r.duplicate ? r : { duplicate: false, ...r.result };
++    }
++
++    return {
++        schema: () => `${inboxSchema(table)}\n${inboxHeadsSchema(headsTable)}`,
++        ensureSchema: () => db.query(`${inboxSchema(table)}\n${inboxHeadsSchema(headsTable)}`),
++        once: inbox.once,
++        seen: inbox.seen,
++        apply,
++        head: async (consumer, key) => {
++            const row = await db.maybe(`SELECT revision, event_id FROM ${headsTable} WHERE consumer = $1 AND key = $2`, [String(consumer), String(key)]);
++            return row ? { revision: Number(row.revision), event_id: row.event_id } : null;
++        },
++    };
++}
++
++module.exports = {
++    sign, verifyDelivery, signV2, verifyDeliveryV2, prepareEnvelope, createPublisher, createPgOutbox, createPgInbox, outboxSchema, inboxSchema,
++    createPgOrderedInbox, inboxHeadsSchema, subjectKey, PublishError,
++};
ddiff --git a/test/outbox-inbox.test.js b/test/outbox-inbox.test.js
iindex 0b4a9a1..cc7784d 100644
---- a/test/outbox-inbox.test.js
++++ b/test/outbox-inbox.test.js
@@@ -1,7 +1,11 @@
  'use strict';
  const assert = require('assert');
++const fs = require('fs');
++const os = require('os');
++const path = require('path');
  const nodeHttp = require('http');
  const { createDb, sql } = require('openvibe-sdk/db');
++const { createTestDb } = require('openvibe-sdk/testing');
  const { serviceAuth } = require('openvibe-contracts');
  const events = require('..');
  const { boot, request, serviceToken, envelope, subscriber, suite } = require('./helpers');
@@@ -181,6 +185,140 @@ t('consumer crash mid-processing + redelivery + replay -> exactly one effect (cr
      }
  });
  
++// A consumer's database for the ordered inbox, migrated like a service's: PGlite, or with EVENTS_TEST_STORE=pg the
++// containers' PostgreSQL through PgBouncer (a runtime role, a pool of 4: concurrent applies really race).
++async function orderedConsumerDb(extra = '') {
++    const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-inbox-'));
++    fs.writeFileSync(path.join(dir, '0001_consumer.sql'), `-- phase: expand\n${extra}\n${events.inboxSchema()}\n${events.inboxHeadsSchema()}\n`);
++    try {
++        return await createTestDb({ migrations: dir, store: process.env.EVENTS_TEST_STORE || 'pglite', service: 'consumer' });
++    } finally { fs.rmSync(dir, { recursive: true, force: true }); }
++}
++
++function rev(id, revision, extra = {}) {
++    return envelope('live', { event_type: 'live.stream.updated', subject: { type: 'stream', id, revision }, ...extra });
++}
++
++t('ordered inbox: an older or equal revision of a key is stale and never applied; keys are independent', async () => {
++    const { db, close } = await orderedConsumerDb('CREATE TABLE titles (stream text PRIMARY KEY, revision integer);');
++    const inbox = events.createPgOrderedInbox(db);
++    const write = (ev) => async (t) => {
++        await t.query('INSERT INTO titles (stream, revision) VALUES ($1, $2) ON CONFLICT (stream) DO UPDATE SET revision = EXCLUDED.revision', [ev.subject.id, ev.subject.revision]);
++        return ev.subject.revision;
++    };
++    const title = async (id) => Number((await db.one('SELECT revision FROM titles WHERE stream = $1', [id])).revision);
++    try {
++        assert.strictEqual(events.subjectKey(rev('a', 1)), '["stream","a"]');
++        const r2 = rev('a', 2);
++        assert.deepStrictEqual(await inbox.apply('media', r2, write(r2)), { duplicate: false, stale: false, result: 2 });
++        const r1 = rev('a', 1);
++        assert.deepStrictEqual(await inbox.apply('media', r1, write(r1)), { duplicate: false, stale: true, head: { revision: 2, event_id: r2.event_id } });
++        assert.strictEqual(await title('a'), 2, 'the older revision did not overwrite');
++        assert.deepStrictEqual(await inbox.apply('media', r1, write(r1)), { duplicate: true }, 'a stale event keeps its receipt');
++        assert.deepStrictEqual(await inbox.apply('media', r2, write(r2)), { duplicate: true });
++        const again = rev('a', 2);
++        assert.strictEqual((await inbox.apply('media', again, write(again))).stale, true, 'an equal revision is stale');
++        const r3 = rev('a', 3);
++        assert.strictEqual((await inbox.apply('media', r3, write(r3))).result, 3);
++        assert.deepStrictEqual(await inbox.head('media', events.subjectKey(rev('a', 0))), { revision: 3, event_id: r3.event_id });
++
++        const b1 = rev('b', 1);
++        assert.strictEqual((await inbox.apply('media', b1, write(b1))).stale, false, 'another key has its own head');
++        assert.strictEqual((await inbox.apply('network', r1, write(r1))).stale, false, 'another consumer has its own heads');
++        const subj = (type, id, revision) => envelope('live', { event_type: 'live.stream.updated', subject: { type, id, revision } });
++        const long = 'x'.repeat(250);
++        const pairs = [[subj('stream', `${long}1`, 9), subj('stream', `${long}2`, 1)], [subj('a:b', 'c', 9), subj('a', 'b:c', 1)]];
++        for (const [hi, lo] of pairs) {
++            assert.notStrictEqual(events.subjectKey(hi), events.subjectKey(lo), 'distinct subjects have distinct keys');
++            assert.ok(events.subjectKey(hi).length <= 200);
++            assert.strictEqual((await inbox.apply('media', hi, async () => 'hi')).result, 'hi');
++            assert.deepStrictEqual(await inbox.apply('media', lo, async () => 'lo'), { duplicate: false, stale: false, result: 'lo' },
++                'a higher revision of one subject never makes another stale');
++        }
++        const custom = rev('a', 0);
++        assert.strictEqual((await inbox.apply('media', custom, async () => 'k', { key: 'owner:9', revision: 7 })).result, 'k');
++        assert.strictEqual((await inbox.head('media', 'owner:9')).revision, 7);
++
++        const unordered = envelope('live', { subject: undefined });
++        assert.deepStrictEqual(await inbox.apply('media', unordered, async () => 'ran'), { duplicate: false, stale: false, result: 'ran' });
++        assert.deepStrictEqual(await inbox.apply('media', unordered, async () => 'ran'), { duplicate: true }, 'without a revision: deduplicated only');
++        assert.strictEqual(await inbox.seen('media', unordered.event_id), true);
++
++        await assert.rejects(inbox.apply('media', rev('c', -1), async () => {}), /revision/);
++        await assert.rejects(inbox.apply('media', rev('c', 1), async () => {}, { key: 'x'.repeat(201) }), /key/);
++        await assert.rejects(inbox.apply('media', {}, async () => {}), /event_id/);
++        assert.throws(() => events.createPgOrderedInbox(db, { headsTable: 'bad-name' }), /table name/);
++    } finally { await close(); }
++});
++
++t('ordered inbox: a failed handler moves neither the receipt nor the head; concurrent revisions apply in order', async () => {
++    const { db, close } = await orderedConsumerDb('CREATE TABLE effects (id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY, revision integer);');
++    const inbox = events.createPgOrderedInbox(db);
++    const effect = (ev) => (t) => t.query('INSERT INTO effects (revision) VALUES ($1)', [ev.subject.revision]);
++    try {
++        const r5 = rev('a', 5);
++        await assert.rejects(inbox.apply('media', r5, async (t) => {
++            await effect(r5)(t);
++            throw new Error('crashed mid-transaction');
++        }), /crashed/);
++        assert.strictEqual(await inbox.head('media', events.subjectKey(rev('a', 0))), null);
++        assert.strictEqual(await inbox.seen('media', r5.event_id), false);
++        assert.strictEqual(Number(await db.value('SELECT count(*) FROM effects')), 0);
++
++        const evs = [3, 8, 1, 6, 2, 7, 4, 5].map(n => rev('b', n));
++        const rs = await Promise.all(evs.map(ev => inbox.apply('media', ev, effect(ev))));
++        assert.ok(rs.every(r => r.duplicate === false));
++        assert.strictEqual((await inbox.head('media', events.subjectKey(rev('b', 0)))).revision, 8);
++        const applied = (await db.query('SELECT revision FROM effects ORDER BY id')).rows.map(r => Number(r.revision));
++        assert.strictEqual(applied.length, rs.filter(r => !r.stale).length);
++        assert.ok(applied.every((n, i) => i === 0 || n > applied[i - 1]), `applied in increasing revisions: ${applied}`);
++        assert.strictEqual(applied[applied.length - 1], 8);
++    } finally { await close(); }
++});
++
++t('ordered inbox end to end: replaying a dead older revision after a newer one is acknowledged, not applied', async () => {
++    const { db: consumerDb, close } = await orderedConsumerDb('CREATE TABLE titles (stream text PRIMARY KEY, revision integer);');
++    const inbox = events.createPgOrderedInbox(consumerDb);
++    let secret = null;
++    let failNext = true;
++    const outcomes = [];
++    const stub = await subscriber(async (call) => {
++        if (!events.verifyDeliveryV2(call.rawBody, call.headers, secret, { now: h.clock.now() })) return 401;
++        if (failNext) { failNext = false; return 503; }
++        const ev = call.body.event;
++        const r = await inbox.apply('media', ev, (t) => t.query(
++            'INSERT INTO titles (stream, revision) VALUES ($1, $2) ON CONFLICT (stream) DO UPDATE SET revision = EXCLUDED.revision', [ev.subject.id, ev.subject.revision]));
++        outcomes.push(r.duplicate ? 'duplicate' : r.stale ? 'stale' : 'applied');
++        return 204;
++    });
++    try {
++        const media = serviceToken('media', ['events.subscription.manage']);
++        const sub = (await request(h.base, 'POST', '/api/v1/subscriptions', {
++            token: media, body: { topic_pattern: 'live.stream.updated', endpoint: stub.url, retry_policy: { max_attempts: 1 } },
++        })).body;
++        secret = sub.secret;
++        const live = serviceToken('live', ['events.event.publish']);
++        const older = rev('e2e', 1);
++        const newer = rev('e2e', 2);
++        await request(h.base, 'POST', '/api/v1/events', { token: live, body: older });
++        await h.worker.drain();
++        assert.strictEqual((await h.store.getDelivery(older.event_id, sub.id)).status, 'dead');
++        await request(h.base, 'POST', '/api/v1/events', { token: live, body: newer });
++        await h.worker.drain();
++        assert.strictEqual((await h.store.getDelivery(newer.event_id, sub.id)).status, 'delivered');
++
++        const rp = await request(h.base, 'POST', '/api/v1/deliveries/replay', { token: admin, body: { subscription_id: sub.id, event_ids: [older.event_id, newer.event_id] } });
++        assert.strictEqual(rp.body.queued, 2);
++        await h.worker.drain();
++        assert.strictEqual((await h.store.getDelivery(older.event_id, sub.id)).status, 'delivered', 'the stale event is acknowledged');
++        assert.deepStrictEqual(outcomes.sort(), ['applied', 'duplicate', 'stale']);
++        assert.strictEqual(Number((await consumerDb.one('SELECT revision FROM titles WHERE stream = $1', ['e2e'])).revision), 2);
++    } finally {
++        await stub.close();
++        await close();
++    }
++});
++
  t('stop', async () => { await h.stop(); network.close(); });
  
  t.run();