openvibers/openvibe.publishing Commits

search-relay: deliver queued Search documents through the owner API (1.5.0) (#138)

OP OpenVibers committed 11 days ago · ff60fcd2ba · 9 files, +479 −13

Meme, Exchange and Fun queue one search.index-document@1 per change into <product>_outbox and
left the delivery for later: 118 meme entries and 104 Exchange listings never reached Search.
createSearchRelay sends them oldest first (PUT a document, DELETE ?revision= a tombstone),
supersedes a document's older rows, treats a stale revision as done, re-sends a conflict one
revision up, backs off on 5xx/network/401, pauses on 429 until Retry-After, sets other refusals
aside with the reason logged, and claims rows with a lease. At most 100 a tick.

Changes

ddiff --git a/CHANGELOG.md b/CHANGELOG.md
iindex 529893c..2b4a6ee 100644
---- a/CHANGELOG.md
++++ b/CHANGELOG.md
@@@ -4,6 +4,34 @@ All notable changes to `openvibe-publishing`. Versions follow [semver](https://s
  breaking change to any exported function, table layout, reason code or document shape is a new
  major (a minor while 0.x). A release is the git tag `vX.Y.Z`; consumers pin the tag's tarball.
  
++## 1.5.0 — 2026-11-02
++
++**Additive: `openvibe-publishing/search-relay`.** No existing export, table layout, reason code or document shape
++changes; a product on v1.4.x upgrades by pinning v1.5.0.
++
++### Added — `openvibe-publishing/search-relay`
++
++`createSearchRelay({ db, table, owner, searchUrl, auth, … })` delivers the Search documents a product queued in its own
++outbox table (`event_id, topic, payload, created_at, published_at, attempts, last_error, next_attempt_at, claimed_at,
++claimed_by`) through OpenVibe.Search's owner API: PUT `/api/v1/documents/:owner/:type/:id` for a document, DELETE with
++`?revision=` for a tombstone, with the product's service token (`search.document.write` on `openvibe.search`; `auth` is
++`{ authHeaders(), invalidate() }`, e.g. openvibe-contracts `serviceAuth.createTokenClient`).
++
++- Rows go oldest first, at most `maxPerTick` (100) a tick: the owner API allows a service 120 writes a minute. Within
++  a tick only a document's newest row is sent; older ones are marked `superseded`.
++- 2xx sets `published_at`. 409 `search.stale_revision` is done (Search holds a newer revision). 409
++  `search.revision_conflict` is sent once more as the stored revision + 1.
++- 5xx, 408, 401 (which drops the token) and network errors retry with backoff, 15 s doubling to an hour. 429 stops
++  the tick until Retry-After and releases the rest unclaimed, without counting an attempt.
++- Any other refusal (a bad document, not the owner) and a payload naming another owner are set aside for good:
++  `published_at` is set, `last_error` starts `rejected`, and the reason is logged.
++- A claim is a five-minute lease (`FOR UPDATE SKIP LOCKED`), so two processes never send one row at once.
++- `start()` (a tick five seconds after boot, then every minute), `kick()` (a tick in two seconds), `drain()`,
++  `stop()`, `status()` (`pending`, totals, last run, last error, pause). Off without `searchUrl` or `auth`.
++
++Why: Meme, Exchange and Fun queue a document per change into `<product>_outbox` and left the delivery for later;
++118 meme entries and 104 Exchange listings had never reached Search (2026-11-02).
++
  ## 1.4.1 — 2026-10-22
  
  **Fix: `seo.canonicalUrl` keeps a path with several leading slashes on the site.** A request path such as `//feed/` or
ddiff --git a/README.md b/README.md
iindex bd3a653..17d87c6 100644
---- a/README.md
++++ b/README.md
@@@ -3,7 +3,7 @@
  > Shared publishing packages for the OpenVibe publication products: Wiki, Blog, News, Reviews,
  > Deals, Coupons and Trade.
  
--**Status:** alpha, **1.4.1**: the stores run on PostgreSQL through the `openvibe-sdk/db` async data
++**Status:** alpha, **1.5.0**: the stores run on PostgreSQL through the `openvibe-sdk/db` async data
  layer (ADR-035: PostgreSQL 18 behind PgBouncer in production, PGlite in tests). Every store method is
  async, takes the product's `openvibe-sdk/db` handle, and accepts the caller's transaction handle; each
  store gives its DDL as `schema(prefix)` for the product's migrations. v1.1.0 adds the shared ingest
@@@ -12,8 +12,10 @@ additive, no existing export changed. v1.2.0 adds `openvibe-publishing/layout`,
  seven sites render through `openvibe-shared/shell` (additive). v1.3.0 forwards the AI summary, facts,
  `updated` and `url` through `layout.renderDocument` to the shell (additive). v1.4.0 adds `openvibe-publishing/search-feed`,
  which puts a record product's pages in OpenVibe.Search (additive). v1.4.1 keeps a request path that starts with
--several slashes (`//feed/`) on the site in `seo.canonicalUrl`, where it threw (a 500 on News, Deals and Blog). The exit proof (`examples/two-products`) runs in `npm test`.
--Releases v0.1.0 to v1.4.1 are tagged (see [CHANGELOG.md](CHANGELOG.md)). Fifteen products pin a release. The seven
++several slashes (`//feed/`) on the site in `seo.canonicalUrl`, where it threw (a 500 on News, Deals and Blog). v1.5.0 adds
++`openvibe-publishing/search-relay`, which delivers the Search documents a product queued in its own outbox table
++through Search's owner API (Meme, Exchange and Fun queue them; additive). The exit proof (`examples/two-products`) runs in `npm test`.
++Releases v0.1.0 to v1.5.0 are tagged (see [CHANGELOG.md](CHANGELOG.md)). Fifteen products pin a release. The seven
  publication products run on PostgreSQL and are public: openvibe.wiki, openvibe.blog, openvibe.news, openvibe.reviews,
  openvibe.deals, openvibe.coupons and openvibe.trade (on 2026-10-22 Blog, News and Deals pinned v1.4.1, the other four
  v1.3.0). Seven more use `search-feed` to put their pages in OpenVibe.Search: Work, Rent, Help, Inventory, Quest, Host and
@@@ -72,10 +74,11 @@ Every module is its own entry point and can be used alone.
  | `openvibe-publishing/ingest` | The shared ingest chassis: the OpenVibe.Sources client, a named change cursor and the `pullChanges` loop (per-item savepoint isolation, `{applied\|hold\|removed}`), the signed event consumer with an exactly-once inbox, the shared normalisers, and the generic PSL/registrable-host and freshness helpers | `<prefix>_ingest_cursor` |
  | `openvibe-publishing/publication` | The shared publication glue: gate → document → `sequencer.stamp` → events → `outbox.enqueue` on the caller's transaction handle, tombstones on unpublish/merge, and the IndexNow ping; it emits exactly `search.index-document@1` and the product's events | — |
  | `openvibe-publishing/search-feed` | Record pages (listings, job ads, help articles, catalog items) in OpenVibe.Search: `sync`/`remove` on the write's transaction, and a `sweep()` that re-sends only what changed and tombstones vanished rows | — |
++| `openvibe-publishing/search-relay` | Deliver a product's queued Search documents (its `<product>_outbox` rows: one `search.index-document@1` or tombstone each) through Search's owner API with the product's service token: oldest first, older rows of a document superseded, a stale revision done, a conflict sent again one revision up, 5xx/network backoff, 429 pauses until Retry-After, other refusals logged and set aside; at most 100 a tick | — (the product's outbox) |
  | `openvibe-publishing/layout` | `renderDocument(o)`: the whole page through `openvibe-shared/shell` `page()` (≥ 2.6.0), with robots and canonical from the gate's decision (throws without one), article times, prev/next, feeds, the app icon, stylesheets, the boost marker, the body frame and the footer init | — |
  
  `require('openvibe-publishing')` exposes all of them lazily (`.revisions`, `.seo`, `.indexHooks`,
--`.ingest`, `.publication`, `.searchFeed`, `.layout`, …), plus `schema({ … })`, the DDL of several stores at once (below).
++`.ingest`, `.publication`, `.searchFeed`, `.searchRelay`, `.layout`, …), plus `schema({ … })`, the DDL of several stores at once (below).
  
  ## Install
  
ddiff --git a/STATUS.json b/STATUS.json
iindex 28a5936..e2e0143 100644
---- a/STATUS.json
++++ b/STATUS.json
@@@ -5,7 +5,7 @@
    "wave": "W15",
    "kind": "packages",
    "package": "openvibe-publishing",
--  "version": "1.4.1",
++  "version": "1.5.0",
    "plan": "OpenVibe End-to-End Realignment & Implementation Plan, revision 3 (20 Sep 2026)",
    "runtime": null,
    "port": null,
@@@ -26,7 +26,8 @@
      "ingest",
      "publication",
      "layout",
--    "search-feed"
++    "search-feed",
++    "search-relay"
    ],
    "consumers": [
      "OpenVibe.Wiki (https://openvibe.wiki): revisions, schedule, taxonomy, citations, media, discussion, seo, authorship, index-hooks, ssr, layout",
ddiff --git a/index.js b/index.js
iindex abfd2d0..dee3249 100644
---- a/index.js
++++ b/index.js
@@@ -3,7 +3,7 @@
   * openvibe-publishing — every module is also its own entry point and can be used alone:
   *   require('openvibe-publishing/revisions'), …/schedule, …/taxonomy, …/citations, …/media,
   *   …/discussion, …/seo, …/authorship, …/index-hooks, …/ssr, …/ai, …/ingest, …/publication,
-- *   …/search-feed, …/layout
++ *   …/search-feed, …/search-relay, …/layout
   * This root export loads each one lazily, on first access.
   *
   * schema({ revisions: 'wiki_page', citations: 'wiki', … }) is the DDL of every store a product uses,
@@@ -24,6 +24,7 @@ const MODULES = {
      ingest: './lib/ingest',
      publication: './lib/publication',
      searchFeed: './lib/search-feed',
++    searchRelay: './lib/search-relay',
      layout: './lib/layout',
  };
  
ddiff --git a/lib/search-relay.js b/lib/search-relay.js
nnew file mode 100644
iindex 0000000..88baacc
---- /dev/null
++++ b/lib/search-relay.js
@@@ -0,0 +1,233 @@
++'use strict';
++/**
++ * openvibe-publishing/search-relay — deliver the Search documents a product queued in its own outbox table through
++ * OpenVibe.Search's owner API (PUT and DELETE /api/v1/documents/:owner/:type/:id, a service token holding
++ * search.document.write on openvibe.search).
++ *
++ * Meme, Exchange and Fun write one search.index-document@1 (or a tombstone, `deleted: true`) per change into
++ * `<product>_outbox` in the same transaction as the change, and nothing delivered them: 118 meme entries and 104
++ * Exchange listings had waited since their launch (2026-11-02). This is that delivery, for any table of this shape:
++ *
++ *   event_id text PRIMARY KEY, topic text, payload jsonb, created_at timestamptz, published_at timestamptz,
++ *   attempts integer, last_error text, next_attempt_at timestamptz, claimed_at timestamptz, claimed_by text
++ *
++ *   const relay = createSearchRelay({ db, table: 'meme_outbox', owner: 'meme', searchUrl, auth, log });
++ *   relay.start();          // a tick a few seconds after boot, then every minute (never under a drill: the caller decides)
++ *   relay.kick();           // after a change: a tick in two seconds
++ *   await relay.drain();    // one tick now → { sent, stale, superseded, rejected, failed, deferred }
++ *   relay.stop();
++ *   await relay.status();   // { enabled, pending, totals, last_run_at, last_error }
++ *
++ * auth is the product's service-token client for audience openvibe.search, scope search.document.write
++ * (openvibe-contracts serviceAuth.createTokenClient): { authHeaders(), invalidate() }. Without searchUrl or auth the
++ * relay is off and the rows keep waiting.
++ *
++ * What each answer does to a row:
++ *   2xx                         sent: published_at is set
++ *   409 search.stale_revision   Search holds a newer revision: the row is done (last_error says stale)
++ *   409 search.revision_conflict the same revision is indexed with other content (a change that did not move the
++ *                               revision): sent again once as the stored revision + 1
++ *   429                         the tick stops; the row and the rest wait for Retry-After (default a minute)
++ *   401                         the token is dropped; the row is retried with backoff
++ *   other 4xx                   refused for good (bad document, not the owner): published_at is set and last_error
++ *                               starts "rejected", so the row stops blocking the queue and the log says why
++ *   5xx, network                retried with backoff: 15 s doubling to an hour
++ * Rows go oldest first; within one tick only the newest row of a document is sent and the older ones are marked
++ * superseded. A claim is a lease (five minutes), so two processes never send one row at once and a crash frees it.
++ * At most `maxPerTick` rows (default 100) go per tick: Search's owner API allows a service 120 writes a minute.
++ */
++
++const TABLE_RE = /^[a-z_][a-z0-9_]{0,62}$/;
++const OWNER_RE = /^[a-z][a-z0-9-]{1,39}$/;
++const LEASE_MS = 5 * 60_000;
++const BACKOFF_MIN_MS = 15_000;
++const BACKOFF_MAX_MS = 60 * 60_000;
++const RETRY_AFTER_DEFAULT_MS = 60_000;
++const TIMEOUT_MS = 15_000;
++
++const isoOf = (ms) => new Date(ms).toISOString();
++const backoff = (attempts) => Math.min(BACKOFF_MAX_MS, BACKOFF_MIN_MS * 2 ** Math.max(0, attempts - 1));
++
++function retryAfterMs(res) {
++    const h = res && res.headers && typeof res.headers.get === 'function' ? res.headers.get('retry-after') : null;
++    const n = Number(h);
++    return Number.isFinite(n) && n > 0 ? Math.min(BACKOFF_MAX_MS, n * 1000) : RETRY_AFTER_DEFAULT_MS;
++}
++
++function createSearchRelay({
++    db, table, owner, searchUrl, auth, fetchImpl = globalThis.fetch, now = () => Date.now(), log = console,
++    maxPerTick = 100, intervalMs = 60_000, firstDelayMs = 5_000, kickDelayMs = 2_000, workerId = null,
++} = {}) {
++    if (!db) throw new TypeError('db is required');
++    if (!TABLE_RE.test(String(table || ''))) throw new TypeError('table must be a plain table name, e.g. "meme_outbox"');
++    if (!OWNER_RE.test(String(owner || ''))) throw new TypeError('owner must be the service id, e.g. "meme"');
++    const base = String(searchUrl || '').replace(/\/+$/, '');
++    const enabled = Boolean(base && auth && typeof auth.authHeaders === 'function');
++    const me = workerId || `${owner}-relay-${process.pid}-${Math.random().toString(36).slice(2, 8)}`;
++    const totals = { sent: 0, stale: 0, superseded: 0, rejected: 0, failed: 0, deferred: 0 };
++    let lastRunAt = null;
++    let lastError = null;
++    let pausedUntil = 0;
++    let running = null;
++    let timer = null;
++    let first = null;
++    let kicked = null;
++
++    const docUrl = (p) => `${base}/api/v1/documents/${encodeURIComponent(owner)}/${encodeURIComponent(p.type)}/${encodeURIComponent(p.id)}`;
++
++    async function call(method, url, body) {
++        const res = await fetchImpl(url, {
++            method,
++            headers: { Accept: 'application/json', ...(body ? { 'Content-Type': 'application/json' } : {}), ...(await auth.authHeaders()) },
++            body: body ? JSON.stringify(body) : undefined,
++            signal: AbortSignal.timeout(TIMEOUT_MS),
++        });
++        const json = await res.json().catch(() => ({}));
++        return { res, status: res.status, body: json || {} };
++    }
++
++    function send(p, revision) {
++        if (p.deleted) return call('DELETE', `${docUrl(p)}?revision=${Number(revision)}`);
++        return call('PUT', docUrl(p), { ...p, owner, revision });
++    }
++
++    // UPDATE … RETURNING does not keep the subquery's order: the claimed rows are sorted again, oldest first.
++    const created = (r) => new Date(r.created_at).getTime();
++    async function claim(limit) {
++        const t = now();
++        const rows = await db.many(
++            `UPDATE ${table} o SET claimed_at = $1::timestamptz, claimed_by = $2
++              WHERE o.event_id IN (
++                    SELECT o2.event_id FROM ${table} o2
++                     WHERE o2.published_at IS NULL AND o2.next_attempt_at <= $1::timestamptz
++                       AND (o2.claimed_at IS NULL OR o2.claimed_at < $3::timestamptz)
++                     ORDER BY o2.created_at, o2.event_id
++                     LIMIT $4
++                     FOR UPDATE SKIP LOCKED)
++          RETURNING o.event_id, o.payload, o.attempts, o.created_at`,
++            [isoOf(t), me, isoOf(t - LEASE_MS), limit],
++        );
++        return rows.sort((a, b) => created(a) - created(b) || (a.event_id < b.event_id ? -1 : a.event_id > b.event_id ? 1 : 0));
++    }
++
++    const done = (eventId, note = null) => db.exec(
++        `UPDATE ${table} SET published_at = $1::timestamptz, last_error = $2, claimed_at = NULL, claimed_by = NULL WHERE event_id = $3`,
++        [isoOf(now()), note, eventId],
++    );
++    const retry = (row, message, waitMs) => db.exec(
++        `UPDATE ${table} SET attempts = attempts + 1, last_error = $1, next_attempt_at = $2::timestamptz, claimed_at = NULL, claimed_by = NULL
++          WHERE event_id = $3`,
++        [String(message).slice(0, 500), isoOf(now() + waitMs), row.event_id],
++    );
++    const release = (eventIds, untilMs) => (eventIds.length ? db.exec(
++        `UPDATE ${table} SET next_attempt_at = GREATEST(next_attempt_at, $1::timestamptz), claimed_at = NULL, claimed_by = NULL
++          WHERE event_id = ANY($2::text[]) AND claimed_by = $3`,
++        [isoOf(untilMs), eventIds, me],
++    ) : null);
++
++    const problem = (r) => `${r.status} ${(r.body && (r.body.code || r.body.type || r.body.title)) || ''}`.trim();
++
++    /** One row: send it and record what Search said. Returns the outcome name, or 'paused' on a 429. */
++    async function deliver(row) {
++        const p = row.payload || {};
++        if (p.owner !== undefined && p.owner !== owner) { await done(row.event_id, `rejected: payload owner ${p.owner} is not ${owner}`); return 'rejected'; }
++        if (!p.type || !p.id || !Number.isFinite(Number(p.revision))) { await done(row.event_id, 'rejected: payload has no type, id or revision'); return 'rejected'; }
++        let r;
++        try {
++            r = await send(p, Number(p.revision));
++            if (r.status === 409 && r.body && r.body.code === 'search.revision_conflict' && !p.deleted) {
++                const stored = Number(r.body.stored_revision);
++                if (Number.isFinite(stored)) r = await send(p, stored + 1);
++            }
++        } catch (err) {
++            await retry(row, `network: ${(err && err.message) || err}`, backoff(row.attempts + 1));
++            lastError = `network: ${(err && err.message) || err}`;
++            return 'failed';
++        }
++        if (r.status >= 200 && r.status < 300) { await done(row.event_id); return 'sent'; }
++        if (r.status === 409 && r.body && r.body.code === 'search.stale_revision') {
++            await done(row.event_id, `stale: Search holds revision ${r.body.stored_revision}`);
++            return 'stale';
++        }
++        if (r.status === 429) { pausedUntil = now() + retryAfterMs(r.res); lastError = problem(r); return 'paused'; }
++        if (r.status === 401 && typeof auth.invalidate === 'function') auth.invalidate();
++        if (r.status === 401 || r.status === 408 || r.status >= 500) {
++            lastError = problem(r);
++            await retry(row, lastError, backoff(row.attempts + 1));
++            return 'failed';
++        }
++        lastError = `rejected: ${problem(r)}`;
++        log.warn(`[${owner} → Search] ${p.type}/${p.id} r${p.revision} refused: ${problem(r)}${r.body && r.body.detail ? ` (${String(r.body.detail).slice(0, 200)})` : ''}`);
++        await done(row.event_id, lastError.slice(0, 500));
++        return 'rejected';
++    }
++
++    async function tick() {
++        const out = { sent: 0, stale: 0, superseded: 0, rejected: 0, failed: 0, deferred: 0 };
++        if (!enabled) return null;
++        if (now() < pausedUntil) return out;
++        const rows = await claim(maxPerTick);
++        // Within one claim, only the newest row of a document matters: the older ones are superseded.
++        const newest = new Map();
++        for (const row of rows) {
++            const p = row.payload || {};
++            newest.set(`${p.type}\u0000${p.id}`, row.event_id);
++        }
++        const queue = [];
++        for (const row of rows) {
++            const p = row.payload || {};
++            if (newest.get(`${p.type}\u0000${p.id}`) !== row.event_id) { await done(row.event_id, 'superseded'); out.superseded++; } else queue.push(row);
++        }
++        for (let i = 0; i < queue.length; i++) {
++            const o = await deliver(queue[i]);
++            if (o === 'paused') {
++                const rest = queue.slice(i).map((r) => r.event_id);
++                await release(rest, pausedUntil);
++                out.deferred += rest.length;
++                break;
++            }
++            out[o]++;
++        }
++        for (const k of Object.keys(out)) totals[k] += out[k];
++        lastRunAt = isoOf(now());
++        if (out.sent || out.rejected || out.failed || out.deferred) {
++            log.log(`[${owner} → Search] ${out.sent} sent, ${out.stale} stale, ${out.superseded} superseded, ${out.rejected} rejected, ${out.failed} failed, ${out.deferred} deferred`);
++        }
++        return out;
++    }
++
++    /** One tick now (never two at once in this process). */
++    function drain() {
++        if (!running) running = tick().finally(() => { running = null; });
++        return running;
++    }
++    const quietly = () => { drain().catch((err) => { lastError = (err && err.message) || String(err); log.warn(`[${owner} → Search] relay tick failed: ${lastError}`); }); };
++
++    return {
++        enabled,
++        drain,
++        kick() {
++            if (!enabled || !timer || kicked) return;
++            kicked = setTimeout(() => { kicked = null; quietly(); }, kickDelayMs);
++            if (kicked.unref) kicked.unref();
++        },
++        start() {
++            if (!enabled || timer) return false;
++            first = setTimeout(quietly, firstDelayMs);
++            if (first.unref) first.unref();
++            timer = setInterval(quietly, intervalMs);
++            if (timer.unref) timer.unref();
++            return true;
++        },
++        stop() {
++            clearTimeout(first); clearInterval(timer); clearTimeout(kicked);
++            first = null; timer = null; kicked = null;
++        },
++        async status() {
++            const pending = Number(await db.value(`SELECT COUNT(*)::int AS n FROM ${table} WHERE published_at IS NULL`));
++            return { enabled, pending, totals: { ...totals }, last_run_at: lastRunAt, last_error: lastError, paused_until: pausedUntil > now() ? isoOf(pausedUntil) : null };
++        },
++    };
++}
++
++module.exports = { createSearchRelay };
ddiff --git a/package-lock.json b/package-lock.json
iindex 706e04f..e5d55dc 100644
---- a/package-lock.json
++++ b/package-lock.json
@@@ -1,12 +1,12 @@
  {
    "name": "openvibe-publishing",
--  "version": "1.4.1",
++  "version": "1.5.0",
    "lockfileVersion": 3,
    "requires": true,
    "packages": {
      "": {
        "name": "openvibe-publishing",
--      "version": "1.4.1",
++      "version": "1.5.0",
        "license": "MIT",
        "devDependencies": {
          "@electric-sql/pglite": "^0.5.8",
ddiff --git a/package.json b/package.json
iindex 7878f3c..c272931 100644
---- a/package.json
++++ b/package.json
@@@ -1,6 +1,6 @@
  {
    "name": "openvibe-publishing",
--  "version": "1.4.1",
++  "version": "1.5.0",
    "description": "Shared publishing packages for the OpenVibe publication products, on PostgreSQL through openvibe-sdk/db: revisions, scheduling, taxonomy, citations, media attachments, discussion references, the SEO indexability gate, authorship, index hooks, SSR helpers, the ingest chassis (Sources client, change cursor, signed event consumer, normalisers, PSL) and the publication glue. Packages, not an authority.",
    "main": "index.js",
    "exports": {
@@@ -20,7 +20,8 @@
      "./ingest": "./lib/ingest.js",
      "./publication": "./lib/publication.js",
      "./layout": "./lib/layout.js",
--    "./search-feed": "./lib/search-feed.js"
++    "./search-feed": "./lib/search-feed.js",
++    "./search-relay": "./lib/search-relay.js"
    },
    "files": [
      "index.js",
ddiff --git a/test/package.test.js b/test/package.test.js
iindex f18e7b0..c0a9934 100644
---- a/test/package.test.js
++++ b/test/package.test.js
@@@ -12,7 +12,7 @@ const SUBPATHS = Object.keys(pkg.exports).filter((k) => k !== '.' && k !== './pa
  
  test('every subpath export resolves to an existing module', () => {
      assert.deepStrictEqual(SUBPATHS.map((s) => s.slice(2)).sort(),
--        ['ai', 'authorship', 'citations', 'discussion', 'index-hooks', 'ingest', 'layout', 'media', 'publication', 'revisions', 'schedule', 'search-feed', 'seo', 'ssr', 'taxonomy']);
++        ['ai', 'authorship', 'citations', 'discussion', 'index-hooks', 'ingest', 'layout', 'media', 'publication', 'revisions', 'schedule', 'search-feed', 'search-relay', 'seo', 'ssr', 'taxonomy']);
      for (const s of SUBPATHS) assert.ok(require(`openvibe-publishing/${s.slice(2)}`), s);
  });
  
@@@ -37,7 +37,7 @@ test('the root export is lazy and exposes every module', () => {
      assert.strictEqual(out.before, 0);
      assert.ok(out.after > 0);
      assert.strictEqual(out.v, pkg.version);
--    assert.deepStrictEqual(out.names.sort(), ['PublishingError', 'ai', 'authorship', 'citations', 'discussion', 'indexHooks', 'ingest', 'layout', 'media', 'publication', 'revisions', 'schedule', 'schema', 'searchFeed', 'seo', 'ssr', 'taxonomy', 'version']);
++    assert.deepStrictEqual(out.names.sort(), ['PublishingError', 'ai', 'authorship', 'citations', 'discussion', 'indexHooks', 'ingest', 'layout', 'media', 'publication', 'revisions', 'schedule', 'schema', 'searchFeed', 'searchRelay', 'seo', 'ssr', 'taxonomy', 'version']);
  });
  
  test('the package owns no runtime: no listen(), no database of its own, no env config in lib/', () => {
ddiff --git a/test/search-relay.test.js b/test/search-relay.test.js
nnew file mode 100644
iindex 0000000..a19d768
---- /dev/null
++++ b/test/search-relay.test.js
@@@ -0,0 +1,199 @@
++'use strict';
++/**
++ * openvibe-publishing/search-relay (1.5.0): a product's queued Search documents delivered through Search's owner API.
++ *
++ *   - oldest first; a document's older rows in one claim are superseded; upserts PUT the whole document, tombstones
++ *     DELETE with ?revision=; sent rows get published_at and leave the queue;
++ *   - 409 stale is done, 409 conflict is sent once more as the stored revision + 1;
++ *   - 5xx and network errors back off (the row waits, then goes), 401 drops the token, 429 stops the tick until
++ *     Retry-After and leaves the rest unclaimed;
++ *   - any other 4xx and a payload for another owner are rejected for good, and logged;
++ *   - maxPerTick caps a tick; without searchUrl or auth the relay is off and the rows wait.
++ */
++const assert = require('assert');
++const { openDb, fakeClock, suite } = require('./helpers/db');
++const { createSearchRelay } = require('../lib/search-relay');
++
++const { test, run } = suite();
++const silent = { warn() {}, log() {} };
++
++/** A Search owner API that keeps documents and answers like Search's store: applied, unchanged, stale, conflict. */
++function fakeSearch({ owner = 'meme' } = {}) {
++    const docs = new Map();
++    const calls = [];
++    const script = [];
++    const hash = (d) => JSON.stringify({ ...d, revision: undefined });
++    const answer = (status, body, headers = {}) => ({ status, json: async () => body, headers: { get: (k) => headers[k.toLowerCase()] ?? null } });
++    async function fetchImpl(url, init) {
++        const u = new URL(url);
++        const m = /^\/api\/v1\/documents\/([^/]+)\/([^/]+)\/([^/]+)$/.exec(u.pathname);
++        assert.ok(m, `unexpected path ${u.pathname}`);
++        assert.strictEqual(decodeURIComponent(m[1]), owner);
++        assert.match(init.headers.Authorization || '', /^Bearer /);
++        const body = init.body ? JSON.parse(init.body) : null;
++        calls.push({ method: init.method, type: m[2], id: decodeURIComponent(m[3]), revision: body ? body.revision : Number(u.searchParams.get('revision')), body });
++        const next = script.shift();
++        if (next) { if (next instanceof Error) throw next; return answer(next.status, next.body || {}, next.headers); }
++        const key = `${m[2]}/${decodeURIComponent(m[3])}`;
++        const doc = body || { revision: Number(u.searchParams.get('revision')), deleted: true };
++        const cur = docs.get(key);
++        if (cur) {
++            if (doc.revision < cur.revision || (doc.revision === cur.revision && cur.deleted && !doc.deleted)) {
++                return answer(409, { code: 'search.stale_revision', stored_revision: cur.revision });
++            }
++            if (doc.revision === cur.revision && hash(cur) === hash(doc)) return answer(200, { outcome: 'unchanged' });
++            if (doc.revision === cur.revision && !(doc.deleted && !cur.deleted)) return answer(409, { code: 'search.revision_conflict', stored_revision: cur.revision });
++        }
++        docs.set(key, { ...doc, deleted: Boolean(doc.deleted) });
++        return answer(200, { outcome: 'applied', revision: doc.revision });
++    }
++    return { fetchImpl, docs, calls, script };
++}
++
++function tokens() {
++    const t = { invalidated: 0, async authHeaders() { return { Authorization: 'Bearer test' }; }, invalidate() { t.invalidated++; } };
++    return t;
++}
++
++async function setup({ owner = 'meme', maxPerTick = 100, log = silent } = {}) {
++    const db = await openDb();
++    const now = fakeClock();
++    await db.query(`CREATE TABLE ${owner}_outbox (
++        event_id text COLLATE "C" PRIMARY KEY, topic text NOT NULL, payload jsonb NOT NULL,
++        created_at timestamptz NOT NULL DEFAULT now(), published_at timestamptz, attempts integer NOT NULL DEFAULT 0,
++        last_error text, next_attempt_at timestamptz NOT NULL DEFAULT now(), claimed_at timestamptz, claimed_by text)`);
++    const search = fakeSearch({ owner });
++    const auth = tokens();
++    let n = 0;
++    const queue = async (payload, { at = now() } = {}) => {
++        n++;
++        const id = `evt_${String(n).padStart(4, '0')}`;
++        const iso = new Date(at + n).toISOString();
++        await db.exec(`INSERT INTO ${owner}_outbox (event_id, topic, payload, created_at, next_attempt_at) VALUES ($1, $2, $3::jsonb, $4::timestamptz, $4::timestamptz)`,
++            [id, payload.deleted ? `${owner}.index_document.deleted` : `${owner}.index_document.upserted`, JSON.stringify(payload), iso]);
++        return id;
++    };
++    const relay = createSearchRelay({ db, table: `${owner}_outbox`, owner, searchUrl: 'http://search.test/', auth, fetchImpl: search.fetchImpl, now, log, maxPerTick });
++    const row = (id) => db.maybe(`SELECT * FROM ${owner}_outbox WHERE event_id = $1`, [id]);
++    return { db, now, search, auth, queue, relay, row };
++}
++
++const entry = (id, revision, title = `Entry ${id}`) => ({
++    owner: 'meme', type: 'entry', id, revision, deleted: false, visibility: 'public',
++    canonical_url: `https://openvibe.meme/m/${id}`, title, body: title, publication_state: 'published', indexability: { decision: 'index' },
++});
++
++test('oldest first, older rows of one document superseded, upserts PUT and tombstones DELETE', async () => {
++    const { search, queue, relay, row, now } = await setup();
++    const a1 = await queue(entry('a', 1, 'First title'));
++    const b1 = await queue(entry('b', 1));
++    const a2 = await queue(entry('a', 2, 'Second title'));
++    const c1 = await queue({ owner: 'meme', type: 'entry', id: 'c', revision: 3, deleted: true });
++    now.advance(1000);
++    const out = await relay.drain();
++    assert.deepStrictEqual(out, { sent: 3, stale: 0, superseded: 1, rejected: 0, failed: 0, deferred: 0 });
++    assert.deepStrictEqual(search.calls.map((c) => `${c.method} ${c.id} r${c.revision}`), ['PUT b r1', 'PUT a r2', 'DELETE c r3']);
++    assert.strictEqual(search.docs.get('entry/a').title, 'Second title');
++    assert.strictEqual(search.docs.get('entry/c').deleted, true);
++    assert.strictEqual((await row(a1)).last_error, 'superseded');
++    for (const id of [a1, b1, a2, c1]) assert.ok((await row(id)).published_at, `${id} left the queue`);
++    assert.strictEqual((await relay.status()).pending, 0);
++    assert.deepStrictEqual(await relay.drain(), { sent: 0, stale: 0, superseded: 0, rejected: 0, failed: 0, deferred: 0 }, 'nothing left to send');
++});
++
++test('409 stale is done; 409 conflict goes again as the stored revision + 1', async () => {
++    const { search, queue, relay, row, now } = await setup();
++    search.docs.set('entry/a', { ...entry('a', 5), deleted: false });
++    search.docs.set('entry/b', { ...entry('b', 2, 'Old words'), deleted: false });
++    const a = await queue(entry('a', 4));
++    const b = await queue(entry('b', 2, 'New words'));
++    now.advance(1000);
++    const out = await relay.drain();
++    assert.strictEqual(out.stale, 1);
++    assert.strictEqual(out.sent, 1);
++    assert.match((await row(a)).last_error, /^stale: Search holds revision 5/);
++    assert.strictEqual(search.docs.get('entry/b').revision, 3);
++    assert.strictEqual(search.docs.get('entry/b').title, 'New words');
++    assert.strictEqual((await row(b)).last_error, null);
++});
++
++test('5xx and network errors back off, then the row goes; 401 drops the token', async () => {
++    const { search, auth, queue, relay, row, now } = await setup();
++    const a = await queue(entry('a', 1));
++    now.advance(1000);
++    search.script.push({ status: 503, body: { code: 'unavailable' } });
++    assert.strictEqual((await relay.drain()).failed, 1);
++    let r = await row(a);
++    assert.strictEqual(r.attempts, 1);
++    assert.strictEqual(r.published_at, null);
++    assert.strictEqual(r.claimed_at, null, 'the claim is released');
++    assert.strictEqual((await relay.drain()).sent, 0, 'not due yet');
++    now.advance(16_000);
++    search.script.push(new Error('connect ECONNREFUSED'));
++    assert.strictEqual((await relay.drain()).failed, 1);
++    r = await row(a);
++    assert.strictEqual(r.attempts, 2);
++    assert.match(r.last_error, /^network: connect ECONNREFUSED/);
++    now.advance(31_000);
++    search.script.push({ status: 401, body: { code: 'unauthorized' } });
++    await relay.drain();
++    assert.strictEqual(auth.invalidated, 1);
++    now.advance(61_000);
++    assert.strictEqual((await relay.drain()).sent, 1);
++    assert.ok((await row(a)).published_at);
++});
++
++test('429 stops the tick until Retry-After and leaves the rest unclaimed', async () => {
++    const { search, queue, relay, row, now } = await setup();
++    const a = await queue(entry('a', 1));
++    const b = await queue(entry('b', 1));
++    const c = await queue(entry('c', 1));
++    now.advance(1000);
++    search.script.push({ status: 200, body: {} }, { status: 429, body: { code: 'rate_limited' }, headers: { 'retry-after': '30' } });
++    const out = await relay.drain();
++    assert.deepStrictEqual([out.sent, out.deferred], [1, 2]);
++    for (const id of [b, c]) {
++        const r = await row(id);
++        assert.strictEqual(r.claimed_at, null);
++        assert.strictEqual(r.attempts, 0, 'a 429 is not a failed attempt');
++        assert.strictEqual(r.published_at, null);
++    }
++    assert.ok((await row(a)).published_at);
++    assert.ok((await relay.status()).paused_until);
++    assert.strictEqual((await relay.drain()).sent, 0, 'paused');
++    now.advance(31_000);
++    assert.strictEqual((await relay.drain()).sent, 2);
++});
++
++test('other 4xx and a payload for another owner are rejected for good, and logged', async () => {
++    const warned = [];
++    const { search, queue, relay, row, now } = await setup({ log: { log() {}, warn: (m) => warned.push(m) } });
++    const bad = await queue(entry('a', 1));
++    const other = await queue({ ...entry('b', 1), owner: 'rent' });
++    now.advance(1000);
++    search.script.push({ status: 422, body: { code: 'search.bad_document', detail: 'document does not match search.index-document@1' } });
++    const out = await relay.drain();
++    assert.strictEqual(out.rejected, 2);
++    assert.match((await row(bad)).last_error, /^rejected: 422 search\.bad_document/);
++    assert.match((await row(other)).last_error, /payload owner rent is not meme/);
++    assert.ok((await row(bad)).published_at && (await row(other)).published_at, 'neither blocks the queue');
++    assert.strictEqual(search.calls.length, 1, 'the other owner\'s row was never sent');
++    assert.ok(warned.some((m) => /entry\/a r1 refused: 422/.test(m)));
++});
++
++test('maxPerTick caps a tick; without searchUrl or auth the relay is off', async () => {
++    const { db, search, queue, relay, now } = await setup({ maxPerTick: 2 });
++    for (const id of ['a', 'b', 'c', 'd', 'e']) await queue(entry(id, 1));
++    now.advance(1000);
++    assert.strictEqual((await relay.drain()).sent, 2);
++    assert.strictEqual((await relay.drain()).sent, 2);
++    assert.strictEqual((await relay.drain()).sent, 1);
++    assert.strictEqual(search.calls.length, 5);
++    const off = createSearchRelay({ db, table: 'meme_outbox', owner: 'meme', searchUrl: '', auth: tokens(), now });
++    assert.strictEqual(off.enabled, false);
++    assert.strictEqual(off.start(), false);
++    assert.strictEqual(await off.drain(), null);
++    assert.throws(() => createSearchRelay({ db, table: 'meme_outbox; DROP TABLE x', owner: 'meme' }), /plain table name/);
++});
++
++run();