openvibers/openvibe.reviews Commits

Reviews on the Publishing 1.1.0 chassis (plan T9 J3): the Sources client, change cursor and pull loop, signed event consumer and normalisers come from openvibe-publishing/ingest; Reviews keeps its domain half (reviews/source-items.js, http/events.js). Migration 0002 seeds the chassis cursor once from the old state (continuity, never a reset to zero). The replaced reviews/sync.js, reviews/normalize.js and http/consumer.js are deleted; test/chassis.test.js proves nothing requires them and replays the Sources fixture against a golden captured from the pre-change code. npm test and test:pg 19/19.

OP OpenVibers committed 29 Sept 2026 · e53a5dd2df · 20 files, +608 −347


Changes

ddiff --git a/migrations/0002_ingest_cursor.sql b/migrations/0002_ingest_cursor.sql
nnew file mode 100644
iindex 0000000..f607c4e
---- /dev/null
++++ b/migrations/0002_ingest_cursor.sql
@@@ -0,0 +1,21 @@
++-- phase: expand
++-- The ingest cursor moves from review_sync_state ('sources.items.after') to the chassis' own table
++-- (openvibe-publishing/ingest, prefix review → review_ingest_cursor). Additive: the old key is read
++-- once here so a deployment continues from the cursor it had, never from zero. Generated from
++-- require('openvibe-publishing').schema({ ingest: 'review' }).
++
++CREATE TABLE IF NOT EXISTS review_ingest_cursor (
++ name text COLLATE "C" NOT NULL,
++ cursor bigint NOT NULL,
++ updated_at bigint NOT NULL,
++ PRIMARY KEY (name)
++);
++
++-- Continuity: carry the cursor already stored under the old key, once. review_sync_state.value is
++-- text, so a numeric cursor is plain digits; anything else is ignored (the chassis cursor then starts
++-- empty).
++INSERT INTO review_ingest_cursor (name, cursor, updated_at)
++SELECT 'sources', value::bigint, (EXTRACT(EPOCH FROM now()) * 1000)::bigint
++FROM review_sync_state
++WHERE name = 'sources.items.after' AND value ~ '^[0-9]+$'
++ON CONFLICT (name) DO NOTHING;
ddiff --git a/package-lock.json b/package-lock.json
iindex 3441193..791784a 100644
---- a/package-lock.json
++++ b/package-lock.json
@@@ -15,7 +15,7 @@
"express-rate-limit": "^7.5.0",
"iovalkey": "^0.4.0",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.76.0",
-- "openvibe-publishing": "https://codeload.github.com/OpenVibers/OpenVibe.Publishing/tar.gz/refs/tags/v1.0.0",
++ "openvibe-publishing": "https://codeload.github.com/OpenVibers/OpenVibe.Publishing/tar.gz/refs/tags/v1.1.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.20.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.2.0",
"pg": "^8.23.0"
@@@ -1149,9 +1149,9 @@
}
},
"node_modules/openvibe-publishing": {
-- "version": "1.0.0",
-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Publishing/tar.gz/refs/tags/v1.0.0",
-- "integrity": "sha512-g4mU0PPtI4y38NcsdppEbrXTrCYp0rPmIYWyxLxh4GpEYhn48jMWq2+cMQA2uwZCeeZlB3UiXVrQNQyQLiliwg==",
++ "version": "1.1.0",
++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Publishing/tar.gz/refs/tags/v1.1.0",
++ "integrity": "sha512-bJsVl82hDivazoiM/np9ckKJE4V5F4OxyIiDwB6xuhK8M6P42sGU+V88rFCkyiI4BVQiKv9xs4GVx8Fj3pyjsg==",
"license": "MIT",
"engines": {
"node": ">=20"
ddiff --git a/package.json b/package.json
iindex f3474ad..dc4a8af 100644
---- a/package.json
++++ b/package.json
@@@ -25,7 +25,7 @@
"express-rate-limit": "^7.5.0",
"iovalkey": "^0.4.0",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.76.0",
-- "openvibe-publishing": "https://codeload.github.com/OpenVibers/OpenVibe.Publishing/tar.gz/refs/tags/v1.0.0",
++ "openvibe-publishing": "https://codeload.github.com/OpenVibers/OpenVibe.Publishing/tar.gz/refs/tags/v1.1.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.20.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.2.0",
"pg": "^8.23.0"
ddiff --git a/server/app.js b/server/app.js
iindex 62f98f4..5ef888e 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -4,7 +4,7 @@
*
* Pages (SSR, useful without JavaScript) server/http/pages.js
* /api/v1 (people and service principals) server/http/api.js
-- * POST /internal/events (signed webhook) server/http/consumer.js
++ * POST /internal/events (signed webhook) server/http/events.js
* Discovery: robots, sitemaps, feeds, llms server/http/machine.js
* /auth/* (Network SSO) server/auth/session.js
* GET /api/health, /api/ready, /release.json, /metrics (loopback only)
@@@ -19,7 +19,7 @@ const { createSessionRoutes } = require('./auth/session');
const { createApi } = require('./http/api');
const { createPages } = require('./http/pages');
const { createMachine } = require('./http/machine');
--const { consumerRouter } = require('./http/consumer');
++const { createEvents } = require('./http/events');
const { createActorLimits } = require('./http/actor-limits');
const PUBLIC_DIR = path.join(__dirname, '..', 'public');
@@@ -41,7 +41,7 @@ function lagged(read, initial) {
};
}
--function createApp({ config, svc, viewers, platform, sync, keys, db, log = console, rateLimits = true, fetchImpl = globalThis.fetch, limitsNow = null, indexnow = null }) {
++function createApp({ config, svc, viewers, platform, sources, sync, keys, db, log = console, rateLimits = true, fetchImpl = globalThis.fetch, limitsNow = null, indexnow = null }) {
const app = express();
app.disable('x-powered-by');
app.set('trust proxy', config.trustProxy);
@@@ -100,7 +100,7 @@ function createApp({ config, svc, viewers, platform, sync, keys, db, log = conso
if (indexnow && indexnow.enabled) app.use(indexnow.keyFile);
// Signed deliveries from OpenVibe.Events. Loopback/internal only in nginx (not proxied publicly).
-- const consumer = consumerRouter({ db, svc, sync, config, log });
++ const consumer = createEvents({ db, svc, sync, config, log });
app.use('/internal', http.middleware(), consumer.router);
app.get('/api/health', (_req, res) => res.json({ status: 'ok', service: 'openvibe-reviews', version: VERSION }));
@@@ -121,7 +121,7 @@ function createApp({ config, svc, viewers, platform, sync, keys, db, log = conso
},
{ name: 'valkey', required: false, check: async () => (valkey ? valkey.ready() : { skipped: 'VALKEY_URL not set: per-actor limits count in this process only' }) },
{ name: 'network_jwks', required: false, check: () => { if (keys.loaded()) return true; keys.ensure().catch(() => {}); return 'Network signing key not loaded yet: sign-in and token calls answer 503'; } },
-- { name: 'sources', required: false, check: () => platform.sources.configured || 'not configured: no review items are read, so no signals and no aggregates exist' },
++ { name: 'sources', required: false, check: () => sources.enabled || 'not configured: no review items are read, so no signals and no aggregates exist' },
{ name: 'events_relay', required: false, check: () => (platform.eventsConfigured ? true : 'EVENTS_URL or the service principal is not configured: events wait in the outbox') },
{ name: 'events_webhook', required: false, check: () => (config.eventsWebhookSecrets.length ? true : 'REVIEWS_EVENTS_SECRET is not set: sources.item.* deliveries are refused; the pull sync still runs') },
{ name: 'community', required: false, check: () => platform.community.configured || 'not configured: discussions show as unavailable' },
@@@ -140,7 +140,7 @@ function createApp({ config, svc, viewers, platform, sync, keys, db, log = conso
// Per-actor limits (http/actor-limits.js) on /api/v1 and the forms, counted once each router resolved
// req.actor; the per-address limits here stay. limitsNow: the limiter's clock (tests).
const limits = createActorLimits({ config, now: limitsNow || (() => Date.now()), registry: metrics.registry, log, enabled: rateLimits, valkey });
-- app.use('/api/v1', createApi({ svc, viewers, platform, sync, config, log, limits }));
++ app.use('/api/v1', createApi({ svc, viewers, sources, sync, config, log, limits }));
app.use('/api', (req, res) => http.sendProblem(res, 404, 'route.not_found', { detail: 'Not found' }));
// This site's own pinned copy of the OpenVibe Frame's browser files (openvibe-shared/serve).
ddiff --git a/server/config.js b/server/config.js
iindex 0daedac..f09f4b7 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -69,13 +69,22 @@ function load(env = process.env) {
eventsWebhookSecrets: list(env.REVIEWS_EVENTS_SECRET, ''),
communityUrl: trim(env.OV_COMMUNITY_URL || 'https://openvibe.community'),
communityInternalUrl: trim(env.OV_COMMUNITY_INTERNAL_URL || ''),
-- sourcesInternalUrl: trim(env.OV_SOURCES_INTERNAL_URL || ''),
++
++ // OpenVibe.Sources: the reviews items Reviews ingests (category=reviews). Reads need
++ // sources.item.read (and, optionally, sources.source.read for source names and health). The
++ // client, the change cursor and the pull loop are the ingest chassis (openvibe-publishing/ingest).
++ sources: {
++ internalUrl: trim(env.OV_SOURCES_INTERNAL_URL || ''),
++ category: 'reviews',
++ timeoutMs: int(env.REVIEWS_SOURCES_TIMEOUT_MS, 10000),
++ },
// Pulling review items from OpenVibe.Sources (sources.item.read) in change order.
sync: {
enabled: bool(env.REVIEWS_SOURCES_SYNC, true),
intervalMs: int(env.REVIEWS_SOURCES_SYNC_INTERVAL_MS, 5 * 60 * 1000),
pageSize: Math.min(500, Math.max(1, int(env.REVIEWS_SOURCES_PAGE_SIZE, 200))),
++ maxPages: Math.max(1, int(env.REVIEWS_SOURCES_MAX_PAGES, 25)),
queueIntervalMs: int(env.REVIEWS_IMPORT_QUEUE_INTERVAL_MS, 15000),
},
ddiff --git a/server/http/api.js b/server/http/api.js
iindex cc5c930..6aca10f 100644
---- a/server/http/api.js
++++ b/server/http/api.js
@@@ -44,7 +44,7 @@ const { checkCapability } = require('../auth/capabilities');
const { http } = contracts;
--function createApi({ svc, viewers, platform, sync, config, log = console, limits }) {
++function createApi({ svc, viewers, sources, sync, config, log = console, limits }) {
const router = express.Router();
router.use(http.middleware());
router.use(express.json({ limit: '256kb' }));
@@@ -136,7 +136,7 @@ function createApi({ svc, viewers, platform, sync, config, log = console, limits
const a = req.actor;
if (a.kind === 'anonymous' || (a.kind === 'user' && !svc.access.isEditor(a))) throw new svc.ReviewsError(403, 'reviews.editor_required', 'Importing is for Reviews editors and granted services');
const b = body(req);
-- const out = await svc.importItem(b.source_item_id || b.item_id, platform.sources, a);
++ const out = await svc.importItem(b.source_item_id || b.item_id, sources, a);
return { outcome: out.outcome, item: await svc.itemView(out.item), signal: out.signal ? await svc.signalView(out.signal) : null };
}));
router.post('/sources/sync', guard('reviews.signal.import'), B('reviews.sources.sync'), R(async (req) => {
ddiff --git a/server/http/consumer.js b/server/http/consumer.js
ddeleted file mode 100644
iindex d5d87bc..0000000
---- a/server/http/consumer.js
++++ /dev/null
@@@ -1,68 +0,0 @@
--'use strict';
--/**
-- * Sources → Reviews: POST /internal/events, the endpoint of Reviews' OpenVibe.Events subscription
-- * (topic pattern `sources.item.*`).
-- *
-- * sources.item.created|updated category reviews: the item id is queued and fetched from Sources
-- * (the event carries no fields), then applied
-- * sources.item.removed category reviews: the item's signal is withdrawn at once; the next
-- * aggregate revision no longer counts it
-- *
-- * Exactly once: the openvibe-sdk inbox claims (consumer, event_id) in the same PostgreSQL transaction
-- * as the change. The signature (X-OpenVibe-Signature, HMAC-SHA256 of the raw body with
-- * REVIEWS_EVENTS_SECRET) is verified with openvibe-sdk's parseDelivery. Only events whose source
-- * is `sources` are applied.
-- */
--const express = require('express');
--const { http } = require('openvibe-contracts');
--const { parseDelivery, createPgInbox } = require('openvibe-sdk/events');
--
--const CONSUMER = 'reviews-sources';
--
--function consumerRouter({ db, svc, sync, config, log = console }) {
-- const router = express.Router();
-- const inbox = createPgInbox(db, { table: 'review_event_inbox' }); // the table is in migrations/
-- const system = { kind: 'system', service: 'svc:reviews' };
--
-- async function apply(event) {
-- return await db.tx(async () => {
-- const r = await inbox.once(CONSUMER, event.event_id, async () => {
-- if (event.source !== 'sources') return 'ignored:source';
-- const p = event.payload && typeof event.payload === 'object' ? event.payload : {};
-- if (p.category !== 'reviews') return 'ignored:category';
-- if (!/^itm_[0-9A-HJKMNP-TV-Z]{26}$/.test(String(p.item_id || ''))) return 'ignored:no_item';
-- if (event.event_type === 'sources.item.created' || event.event_type === 'sources.item.updated') { await sync.enqueue(p.item_id, event.event_type); return 'queued'; }
-- if (event.event_type === 'sources.item.removed') return (await svc.removeItem(p.item_id, p.reason || 'removed by its source', system)).outcome;
-- return 'ignored:type';
-- });
-- return r.duplicate ? { duplicate: true, outcome: null } : { duplicate: false, outcome: r.result };
-- });
-- }
--
-- router.post('/events', express.raw({ type: () => true, limit: '256kb' }), async (req, res) => {
-- const secrets = config.eventsWebhookSecrets;
-- if (!secrets.length) return http.sendProblem(res, 503, 'reviews.webhook_disabled', { detail: 'REVIEWS_EVENTS_SECRET is not set', ctx: req.ov });
-- const raw = Buffer.isBuffer(req.body) ? req.body : Buffer.alloc(0);
-- let delivery = null;
-- // Signature v2 only: HMAC over "<t>.<raw body>" with t within ±300 s; a v1-only (v2 stripped) or stale delivery is refused.
-- for (const s of secrets) { delivery = parseDelivery(raw, req.headers, s, { requireV2: true }); if (delivery) break; }
-- if (!delivery) return http.sendProblem(res, 401, 'reviews.bad_signature', { detail: 'X-OpenVibe-Signature does not verify', ctx: req.ov });
-- const event = delivery.event;
-- if (!event || typeof event.event_id !== 'string' || !/^evt_[0-9A-HJKMNP-TV-Z]{26}$/.test(event.event_id)) {
-- return http.sendProblem(res, 400, 'reviews.bad_delivery', { detail: 'body must be { event: <envelope>, seq }', ctx: req.ov });
-- }
-- let out;
-- try {
-- out = await apply(event);
-- } catch (e) {
-- log.error(`[Reviews] event ${event.event_id} (${event.event_type}) failed:`, e.message);
-- return http.sendProblem(res, 500, 'reviews.event_failed', { detail: 'processing failed; it will be retried', ctx: req.ov });
-- }
-- if (out.outcome === 'queued') sync.drain().catch(() => {});
-- res.status(200).json({ event_id: event.event_id, duplicate: out.duplicate, outcome: out.outcome });
-- });
--
-- return { router, apply };
--}
--
--module.exports = { consumerRouter, CONSUMER };
ddiff --git a/server/http/events.js b/server/http/events.js
nnew file mode 100644
iindex 0000000..3f23081
---- /dev/null
++++ b/server/http/events.js
@@@ -0,0 +1,56 @@
++'use strict';
++/**
++ * Sources → Reviews: POST /internal/events, the endpoint of Reviews' OpenVibe.Events subscription
++ * (topic pattern `sources.item.*`).
++ *
++ * sources.item.created|updated category reviews: the item id is queued and fetched from Sources
++ * (the event carries no fields), then applied
++ * sources.item.removed category reviews: the item's signal is withdrawn at once; the next
++ * aggregate revision no longer counts it
++ *
++ * The signature (X-OpenVibe-Signature v2, ±300 s) and the exactly-once inbox come from the chassis
++ * (openvibe-publishing/ingest.createEventConsumer): the receipt (consumer, event_id) claims in the
++ * same PostgreSQL transaction as the change, so a redelivery changes nothing. Only events whose
++ * source is `sources` are applied.
++ */
++const express = require('express');
++const { http } = require('openvibe-contracts');
++const { createEventConsumer } = require('openvibe-publishing/ingest');
++
++const CONSUMER = 'reviews-sources';
++
++function createEvents({ db, svc, sync, config, log = console }) {
++ const router = express.Router();
++ const consumer = createEventConsumer({ db, secrets: config.eventsWebhookSecrets, consumer: CONSUMER, table: 'review_event_inbox' });
++ const system = { kind: 'system', service: 'svc:reviews' };
++
++ async function handler(event) {
++ if (event.source !== 'sources') return 'ignored:source';
++ const p = event.payload && typeof event.payload === 'object' ? event.payload : {};
++ if (p.category !== 'reviews') return 'ignored:category';
++ if (!/^itm_[0-9A-HJKMNP-TV-Z]{26}$/.test(String(p.item_id || ''))) return 'ignored:no_item';
++ if (event.event_type === 'sources.item.created' || event.event_type === 'sources.item.updated') { await sync.enqueue(p.item_id, event.event_type); return 'queued'; }
++ if (event.event_type === 'sources.item.removed') return (await svc.removeItem(p.item_id, p.reason || 'removed by its source', system)).outcome;
++ return 'ignored:type';
++ }
++
++ router.post('/events', express.raw({ type: () => true, limit: '256kb' }), async (req, res) => {
++ const raw = Buffer.isBuffer(req.body) ? req.body : Buffer.alloc(0);
++ let out;
++ try {
++ out = await consumer.apply(raw, req.headers, handler);
++ } catch (err) {
++ log.error(`[Reviews] event delivery failed: ${err && err.message}`);
++ return http.sendProblem(res, 500, 'reviews.event_failed', { detail: 'processing failed; it will be retried', ctx: req.ov });
++ }
++ if (out.status === 401) return http.sendProblem(res, 401, 'reviews.bad_signature', { detail: 'X-OpenVibe-Signature does not verify', ctx: req.ov });
++ if (out.status === 400) return http.sendProblem(res, 400, 'reviews.bad_delivery', { detail: 'body must be { event: <envelope>, seq }', ctx: req.ov });
++ if (out.status === 503) return http.sendProblem(res, 503, 'reviews.webhook_disabled', { detail: 'REVIEWS_EVENTS_SECRET is not set', ctx: req.ov });
++ if (out.outcome === 'queued') sync.drain().catch(() => {});
++ res.status(200).json({ event_id: out.event_id, duplicate: out.duplicate, outcome: out.outcome });
++ });
++
++ return { router, consumer };
++}
++
++module.exports = { createEvents, CONSUMER };
ddiff --git a/server/index.js b/server/index.js
iindex 3c32605..fb86502 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -8,11 +8,12 @@
* clock/fetch/log/tokens/keys, and returns handles to every part. Workers (the Sources pull, the
* import queue, the Events outbox relay) run only when asked (`workers: true`).
*/
++const { createSourcesClient } = require('openvibe-publishing/ingest');
const { load } = require('./config');
const { openDb, createStores } = require('./db');
const { createPlatform } = require('./integrations/platform');
const { createReviewsService } = require('./reviews/service');
--const { createSourcesSync } = require('./reviews/sync');
++const { createSourcesSync } = require('./reviews/source-items');
const { createKeyStore } = require('./auth/keys');
const { createViewerResolver } = require('./auth/viewer');
const { createApp } = require('./app');
@@@ -24,22 +25,25 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
const db = givenDb || await openDb(config, { log });
const stores = createStores(db, { now });
const platform = createPlatform({ config, db, fetchImpl, tokens, now, log });
++ // The Sources client is the ingest chassis (openvibe-publishing/ingest): the client-credentials
++ // token for audience openvibe.sources and the change-order items API live there now.
++ const sources = createSourcesClient({ config, fetchImpl, now });
// IndexNow (openvibe-shared/indexnow): created once at boot from INDEXNOW_KEY. Unset → off, nothing
// mounted, nothing sent; tests and drills never set it (nor hand one in).
const indexnow = givenIndexnow !== undefined ? givenIndexnow : createIndexNow({ host: config.baseUrl, key: config.indexnow.key, fetch: fetchImpl, log });
const svc = createReviewsService({ db, stores, outbox: platform.outbox, config, now, log, indexnow });
-- const sync = createSourcesSync({ db, svc, platform, config, now, log });
++ const sync = createSourcesSync({ db, svc, sources, config, now, log });
const keys = createKeyStore({ config, fetchImpl, log, publicKey });
keys.ensure().catch(() => {});
const viewers = createViewerResolver({ keys, config });
-- const app = createApp({ config, svc, viewers, platform, sync, keys, db, log, rateLimits, fetchImpl, limitsNow, indexnow });
++ const app = createApp({ config, svc, viewers, platform, sources, sync, keys, db, log, rateLimits, fetchImpl, limitsNow, indexnow });
// Search holds whatever the current rules say (idempotent: unchanged documents are not re-sent).
try { await svc.reconcileIndex(); } catch (err) { log.warn(`[Reviews] index reconcile: ${err.message}`); }
const timers = [];
if (workers) {
-- if (config.sync.enabled && platform.sources.configured) {
++ if (config.sync.enabled && sources.enabled) {
const pull = () => sync.pull().then((r) => { if (!r.ok) log.warn(`[Reviews] Sources pull: ${r.reason}`); }).catch((err) => log.warn(`[Reviews] Sources pull: ${err.message}`));
timers.push(setTimeout(pull, 5000));
timers.push(setInterval(pull, config.sync.intervalMs));
ddiff --git a/server/integrations/platform.js b/server/integrations/platform.js
iindex ebafb7d..7b61f51 100644
---- a/server/integrations/platform.js
++++ b/server/integrations/platform.js
@@@ -1,8 +1,9 @@
'use strict';
/**
-- * Clients for the services Reviews composes (roadmap §29): Events (outbox relay), Sources (review
-- * items with provenance) and Community (discussion threads). Each call uses a Network
-- * client-credentials token for svc:reviews, minted per audience by the SDK token client.
++ * The product-specific service clients Reviews composes (roadmap §29): Events (the outbox relay) and
++ * Community (discussion threads). Each call uses a Network client-credentials token for svc:reviews,
++ * minted per audience by the SDK token client. The Sources client is the ingest chassis
++ * (openvibe-publishing/ingest.createSourcesClient), created in server/index.js.
*
* Every integration is optional. Missing configuration or an unreachable service is an explicit
* failure (an error with a stable code, or `configured: false`), never invented data.
@@@ -17,8 +18,6 @@ class IntegrationError extends Error {
}
const SUBJECT_RE = /^(usr|gst)_[0-9A-HJKMNP-TV-Z]{26}$/;
--const ITEM_RE = /^itm_[0-9A-HJKMNP-TV-Z]{26}$/;
--const SOURCE_KEY_RE = /^[a-z0-9][a-z0-9-]{1,63}$/;
function createPlatform({ config, db, fetchImpl = globalThis.fetch, tokens = null, now = () => Date.now(), log = console } = {}) {
const tokenClient = tokens || (config.oauth.clientSecret
@@@ -62,44 +61,6 @@ function createPlatform({ config, db, fetchImpl = globalThis.fetch, tokens = nul
return { status: res.status, body };
}
-- // ── Sources: review items (sources.item.read) and source records (sources.source.read) ──
-- const sources = {
-- configured: !!(tokenClient && config.sourcesInternalUrl),
-- requireConfigured() {
-- if (!sources.configured) throw new IntegrationError(503, 'sources.unavailable', 'OpenVibe.Sources is not configured (OV_SOURCES_INTERNAL_URL and the service principal)');
-- },
-- /** One item with its provenance. → { item, source } */
-- async getItem(id) {
-- if (!ITEM_RE.test(String(id || ''))) throw new IntegrationError(422, 'signal.invalid_source_item', 'A Sources item id looks like itm_<ULID>');
-- sources.requireConfigured();
-- const r = await getJson(`${config.sourcesInternalUrl}/api/v1/items/${encodeURIComponent(id)}`, 'openvibe.sources');
-- if (r.status === 404) throw new IntegrationError(404, 'signal.source_item_not_found', `Sources has no item ${id}`);
-- if (r.status !== 200 || !r.body || !r.body.item) throw new IntegrationError(503, 'sources.unavailable', `Sources answered ${r.status}`);
-- return { item: r.body.item, source: r.body.source || null };
-- },
-- /** Items of category `reviews` in change order (creations, revisions, removals). */
-- async listItems({ after = 0, limit = 200 } = {}) {
-- sources.requireConfigured();
-- const u = new URL(`${config.sourcesInternalUrl}/api/v1/items`);
-- u.searchParams.set('category', 'reviews');
-- u.searchParams.set('include_removed', '1');
-- u.searchParams.set('after', String(after));
-- u.searchParams.set('limit', String(limit));
-- const r = await getJson(u.toString(), 'openvibe.sources');
-- if (r.status !== 200 || !r.body || !Array.isArray(r.body.items)) throw new IntegrationError(503, 'sources.unavailable', `Sources answered ${r.status}`);
-- return r.body;
-- },
-- /** The source record (name, homepage, notes). null when Sources does not know it. */
-- async getSource(key) {
-- if (!SOURCE_KEY_RE.test(String(key || ''))) return null;
-- sources.requireConfigured();
-- const r = await getJson(`${config.sourcesInternalUrl}/api/v1/sources/${encodeURIComponent(key)}`, 'openvibe.sources');
-- if (r.status === 404) return null;
-- if (r.status !== 200 || !r.body) throw new IntegrationError(503, 'sources.unavailable', `Sources answered ${r.status}`);
-- return r.body.source || r.body;
-- },
-- };
--
// ── Community: comment threads (reference, never copy) ──
const communityBase = config.communityInternalUrl || config.communityUrl;
const discussionClient = createDiscussionClient({ communityUrl: communityBase, tokenClient: forAudience('openvibe.community'), fetchImpl });
@@@ -136,7 +97,7 @@ function createPlatform({ config, db, fetchImpl = globalThis.fetch, tokens = nul
};
return {
-- tokenClient, sdk, events, outbox, sources, community,
++ tokenClient, sdk, events, outbox, community,
eventsConfigured: !!(tokenClient && config.eventsUrl),
};
}
ddiff --git a/server/reviews/normalize.js b/server/reviews/normalize.js
ddeleted file mode 100644
iindex c331830..0000000
---- a/server/reviews/normalize.js
++++ /dev/null
@@@ -1,77 +0,0 @@
--'use strict';
--/**
-- * Normalisation of the identifiers entities are resolved by. Pure functions: the same input always
-- * gives the same key, so resolution is deterministic.
-- */
--
--class NormalizeError extends Error {
-- constructor(message) { super(message); this.status = 422; this.code = 'alias.invalid'; }
--}
--
--function name(value) {
-- const s = String(value == null ? '' : value).normalize('NFKD').replace(/[̀-ͯ]/g, '')
-- .toLowerCase().replace(/&/g, ' and ').replace(/[^a-z0-9]+/g, ' ').trim();
-- if (!s) throw new NormalizeError('A name needs at least one letter or digit');
-- return s.slice(0, 200);
--}
--
--/** host (without www.) + path (+ query minus tracking parameters); http and https are the same thing. */
--function url(value) {
-- let u;
-- try { u = new URL(String(value || '').trim()); } catch { throw new NormalizeError('A URL alias must be an absolute http(s) URL'); }
-- if (u.protocol !== 'http:' && u.protocol !== 'https:') throw new NormalizeError('A URL alias must be http(s)');
-- const host = u.hostname.toLowerCase().replace(/^www\./, '');
-- const keep = [...u.searchParams.entries()].filter(([k]) => !/^(utm_|fbclid$|gclid$|ref$|mc_)/i.test(k)).sort(([a], [b]) => (a < b ? -1 : a > b ? 1 : 0));
-- let pathname = u.pathname.replace(/\/{2,}/g, '/');
-- if (pathname.length > 1) pathname = pathname.replace(/\/+$/, '');
-- const query = keep.length ? `?${new URLSearchParams(keep).toString()}` : '';
-- return `${host}${pathname === '/' ? '' : pathname}${query}`.slice(0, 1000);
--}
--
--function code(value, what) {
-- const s = String(value == null ? '' : value).trim().toUpperCase().replace(/\s+/g, '');
-- if (!s || s.length > 100) throw new NormalizeError(`A ${what} is 1–100 characters`);
-- return s;
--}
--
--/** GTIN-8/12/13/14 → 14 digits, so the same product under two lengths is one key. */
--function gtin(value) {
-- const d = String(value == null ? '' : value).replace(/[\s-]/g, '');
-- if (!/^\d+$/.test(d) || ![8, 12, 13, 14].includes(d.length)) throw new NormalizeError('A GTIN has 8, 12, 13 or 14 digits');
-- return d.padStart(14, '0');
--}
--
--function sourceKey(value) {
-- const s = String(value == null ? '' : value).trim().toLowerCase();
-- if (!/^[a-z0-9][a-z0-9-]{1,63}$/.test(s)) throw new NormalizeError('A source binding is an OpenVibe.Sources source key');
-- return s;
--}
--
--/** namespace:id, e.g. steam_app:620 */
--function external(value) {
-- const m = String(value == null ? '' : value).trim().match(/^([A-Za-z][A-Za-z0-9_.-]{0,39}):(.{1,200})$/);
-- if (!m) throw new NormalizeError('An external id is namespace:id (e.g. steam_app:620)');
-- return `${m[1].toLowerCase()}:${m[2].trim()}`;
--}
--
--const BY_TYPE = { name, url, sku: (v) => code(v, 'SKU'), mpn: (v) => code(v, 'MPN'), gtin, source: sourceKey, external };
--
--function alias(type, value) {
-- const fn = BY_TYPE[type];
-- if (!fn) throw new NormalizeError(`alias type must be one of ${Object.keys(BY_TYPE).join(', ')}`);
-- return fn(value);
--}
--
--/** Try to normalise; null when the value is not a valid identifier of that type. */
--function tryAlias(type, value) {
-- if (value == null || value === '') return null;
-- try { return alias(type, value); } catch { return null; }
--}
--
--function slug(value) {
-- const s = String(value == null ? '' : value).normalize('NFKD').replace(/[̀-ͯ]/g, '')
-- .toLowerCase().replace(/&/g, '-and-').replace(/[^a-z0-9]+/g, '-').replace(/^-+|-+$/g, '').slice(0, 80).replace(/-+$/, '');
-- return s || null;
--}
--
--module.exports = { alias, tryAlias, slug, NormalizeError, TYPES: Object.keys(BY_TYPE) };
ddiff --git a/server/reviews/service.js b/server/reviews/service.js
iindex db2383a..d96535c 100644
---- a/server/reviews/service.js
++++ b/server/reviews/service.js
@@@ -17,8 +17,9 @@ const seo = require('openvibe-publishing/seo');
const hooks = require('openvibe-publishing/index-hooks');
const authorship = require('openvibe-publishing/authorship');
const ssr = require('openvibe-publishing/ssr');
++const { createPublication: createPublicationGlue } = require('openvibe-publishing/publication');
++const { normalize: norm } = require('openvibe-publishing/ingest');
const { ulid } = require('openvibe-contracts').ids;
--const norm = require('./normalize');
const extract = require('./extract');
const { computeAggregate, hashOf } = require('./aggregate');
const { createAccess } = require('./access');
@@@ -158,10 +159,10 @@ function createReviewsService({ db, stores, outbox, config, now = () => Date.now
const tx = async (fn) => await db.tx(async () => await fn()); // ambient: plain db calls inside join it
const entityPath = (e) => `/e/${encodeURIComponent(e.slug)}`;
const entityUrl = (e) => seo.canonicalUrl(origin, entityPath(e));
-- const sitemapUrl = seo.canonicalUrl(origin, '/sitemap.xml');
-- // IndexNow (openvibe-shared/indexnow): tell the engines when a public, indexable entity page
-- // appears, changes or disappears. Never for drafts, private or noindex pages; off without a key.
-- const pinger = indexnow && indexnow.enabled ? indexnow : null;
++ // The publication glue (openvibe-publishing/publication): the index document's revision, its
++ // search.index-document@1 event, the outbox write and the IndexNow ping, all on the caller's
++ // transaction handle. Reviews' gate, document builder and canonical path stay here.
++ const publication = createPublicationGlue({ owner: 'reviews', sequencer, outbox, baseUrl: origin, indexnow, db, now });
const parse = (s, d) => { try { return s == null ? d : JSON.parse(s); } catch { return d; } };
function actorId(actor) {
@@@ -505,19 +506,24 @@ function createReviewsService({ db, stores, outbox, config, now = () => Date.now
return seo.evaluate(await gateFacts(e), { policy, now: now() });
}
-- /** Sends Search the entity's current document (or a tombstone) when it differs from the last one. */
++ /**
++ * Sends Search the entity's current document (or a tombstone) when it differs from the last one.
++ * The chassis (openvibe-publishing/publication) stamps the revision, enqueues the
++ * search.index-document@1 event and pings IndexNow (an indexable page that appeared or changed,
++ * or a page Search already had that went away) — all on the caller's transaction handle. Returns
++ * the stamped document, or null when nothing changed.
++ */
async function syncEntity(entityId) {
const e = await q.entity.get(entityId);
if (!e) return null;
const live = e.state === 'active';
const decision = live ? await decide(e) : null;
-- const before = await sequencer.current('reviews', 'entity', e.id);
const { rev } = live ? await publishedSummaryRevision(e.id) : { rev: null };
const agg = live ? aggregateView(await q.lastAggregate.get(e.id)) : null;
const signals = live ? await activeSignalsIn(await closure(e.id)) : [];
const lines = [];
if (agg && agg.result) lines.push(...agg.result.computation);
-- const doc = await sequencer.stamp(db, hooks.buildIndexDocument({
++ const doc = hooks.buildIndexDocument({
owner: 'reviews', type: 'entity', id: e.id, revision: 0,
state: e.state === 'deleted' ? 'deleted' : 'published', visibility: 'public',
deleted: !live || !decision || !decision.listable,
@@@ -529,9 +535,8 @@ function createReviewsService({ db, stores, outbox, config, now = () => Date.now
provenance: signals.slice(0, 40).map((s) => ({ service: 'sources', type: 'item', id: s.source_item_id, url: s.canonical_url || undefined, retrievedAt: s.observed_at })),
decision,
publishedAt: e.created_at, updatedAt: e.updated_at,
-- }));
-- if (doc.revision !== before) await emit(hooks.indexEvent({ document: doc, now: now() }));
-- return doc;
++ });
++ return await publication.index(db, { document: doc, page: entityPath(e) });
}
async function touchEntity(entityId) {
@@@ -1112,7 +1117,6 @@ function createReviewsService({ db, stores, outbox, config, now = () => Date.now
return await tx(async () => {
const e = await entityOrFail(ref);
if (e.state !== 'active') fail(409, 'entity.not_active', `This entity is ${e.state}`);
-- const wasIndexable = pinger ? (await decide(e)).indexable : false;
const next = { ...e };
const changes = {};
if (name != null && name !== e.name) { next.name = checkName(name); changes.name = [e.name, next.name]; await insertAlias(e.id, 'name', next.name, actor); }
@@@ -1131,13 +1135,7 @@ function createReviewsService({ db, stores, outbox, config, now = () => Date.now
.run(next.name, next.kind, next.description, next.slug, next.noindex, now(), e.id);
if (Object.keys(changes).length) await audit(actor, 'entity.updated', e.id, null, changes);
await syncEntity(e.id);
-- const fresh = await q.entity.get(e.id);
-- // IndexNow: the old URL changed or left the index; the new one appeared or changed.
-- if (pinger) {
-- if (wasIndexable) pinger.pingSoon([entityUrl(e), sitemapUrl]);
-- if ((await decide(fresh)).indexable) pinger.pingSoon([entityUrl(fresh), sitemapUrl]);
-- }
-- return fresh;
++ return await q.entity.get(e.id);
});
},
@@@ -1146,15 +1144,12 @@ function createReviewsService({ db, stores, outbox, config, now = () => Date.now
return await tx(async () => {
const e = await entityOrFail(ref);
if (e.state !== 'active') fail(409, 'entity.not_active', `This entity is ${e.state}`);
-- const wasIndexable = pinger ? (await decide(e)).indexable : false;
if ((await q.mergedInto.all(e.id)).length) fail(409, 'entity.has_merges', 'Split the entities merged into this one first');
const active = await q.activeSignalsOfEntity.all(e.id);
if (active.length) fail(409, 'entity.has_signals', 'This entity has live signals: reattribute or ignore their items first');
await db.prepare("UPDATE review_entities SET state = 'deleted', deleted_at = ?, updated_at = ? WHERE id = ?").run(now(), now(), e.id);
await audit(actor, 'entity.deleted', e.id, null, { note });
await syncEntity(e.id);
-- // IndexNow: the page left the index (only if it was a public, indexable page).
-- if (pinger && wasIndexable) pinger.pingSoon([entityUrl(e), sitemapUrl]);
return await q.entity.get(e.id);
});
},
@@@ -1290,9 +1285,15 @@ function createReviewsService({ db, stores, outbox, config, now = () => Date.now
applyItem,
/** Run fn in one transaction; aggregates of touched entities are recomputed once at the end. */
batch,
-- /** Fetch one item from Sources and apply it. */
++ /** Fetch one item from Sources and apply it (the ingest chassis' SourcesError → ReviewsError). */
async importItem(itemId, sources, actor) {
-- const { item, source } = await sources.getItem(String(itemId || ''));
++ if (!ITEM_RE.test(String(itemId || ''))) fail(422, 'signal.invalid_source_item', 'A Sources item id looks like itm_<ULID>');
++ let item; let source;
++ try { ({ item, source } = await sources.getItem(String(itemId))); } catch (err) {
++ if (!err || typeof err.code !== 'string' || !err.code.startsWith('sources.')) throw err;
++ if (err.code === 'sources.http_404') fail(404, 'signal.source_item_not_found', err.message);
++ fail(Number.isInteger(err.status) && err.status >= 400 && err.status < 500 ? err.status : 503, 'sources.unavailable', err.message);
++ }
let info = source || null;
if (!await q.source.get(item.source_key)) {
try { const full = await sources.getSource(item.source_key); if (full) info = { ...full, ...(source || {}) }; } catch { /* the notes come with the item */ }
@@@ -1482,7 +1483,6 @@ function createReviewsService({ db, stores, outbox, config, now = () => Date.now
if (rev.meta && rev.meta.system && !(review && review.decision === 'approved')) fail(409, 'summary.review_required', 'A revision prepared after a source change is published only after an editor approves it');
const bad = (await citationState(e.id, rev)).filter((p) => !p.supported);
if (bad.length) fail(409, 'summary.unsupported_points', `${bad.length} point(s) cite no live signal of this entity`, { points: bad.map((p) => p.key) });
-- const wasIndexable = pinger ? (await decide(e)).indexable : false;
const wasPublished = summary.state === 'published';
const before = summary.published_revision;
await q.publishSummary.run({ id: summary.id, n, now: now() });
@@@ -1500,8 +1500,6 @@ function createReviewsService({ db, stores, outbox, config, now = () => Date.now
extra: { entity_id: e.id, entity_slug: fresh.slug, authorship: rec ? hooks.AUTHORSHIP[rec.mode] : null, ...(rev.meta && rev.meta.correction ? { correction: { note: rev.meta.correction.note, corrects: rev.meta.correction.corrects } } : {}) },
}));
}
-- // IndexNow: the entity page appeared or changed; never a draft or a non-indexable page.
-- if (pinger && (decision.indexable || wasIndexable)) pinger.pingSoon([entityUrl(fresh), sitemapUrl]);
return await q.summary.get(e.id);
});
},
@@@ -1512,14 +1510,11 @@ function createReviewsService({ db, stores, outbox, config, now = () => Date.now
const e = await entityOrFail(ref);
const summary = await q.summary.get(e.id);
if (!summary || summary.state !== 'published') fail(409, 'summary.not_published', 'Nothing is published');
-- const wasIndexable = pinger ? (await decide(e)).indexable : false;
await q.unpublishSummary.run(now(), summary.id);
await audit(actor, 'summary.unpublished', e.id, summary.id, { revision: summary.published_revision });
const doc = await syncEntity(e.id);
const fresh = await q.entity.get(e.id);
const decision = await decide(fresh);
-- // IndexNow: the page changed or left the index; only when it was or is a public, indexable page.
-- if (pinger && (wasIndexable || decision.indexable)) pinger.pingSoon([entityUrl(fresh), sitemapUrl]);
await emit(hooks.publicationEvent({
product: 'reviews', type: 'summary', action: 'unpublished', id: summary.id, revision: summary.published_revision,
actor: hooks.subjectRef(actorId(actor)), decision, now: now(),
@@@ -1634,6 +1629,8 @@ function createReviewsService({ db, stores, outbox, config, now = () => Date.now
},
// Search reconciliation ---------------------------------------------------------------
++ /** Stamp + enqueue the entity's current document on the caller's handle (used by tests/ops). */
++ syncIndex: syncEntity,
async reconcileIndex() {
return await tx(async () => {
let sent = 0;
ddiff --git a/server/reviews/source-items.js b/server/reviews/source-items.js
nnew file mode 100644
iindex 0000000..d20eaa7
---- /dev/null
++++ b/server/reviews/source-items.js
@@@ -0,0 +1,127 @@
++'use strict';
++/**
++ * How review items get from OpenVibe.Sources into Reviews.
++ *
++ * The chassis (openvibe-publishing/ingest) owns the Sources client, the change cursor and the pull
++ * loop: one transaction per page (the page's writes and the cursor advance commit together), one
++ * savepoint per item (a bad item is isolated and counted, never stalling the page). This module keeps
++ * Reviews' domain half and its own policy:
++ *
++ * pull GET /api/v1/items?category=reviews&include_removed=1&after=<change_seq> in change order
++ * (sources.item.read), from the cursor in review_ingest_cursor. A restart resumes where it
++ * stopped and a missed event is caught up on the next pass.
++ * push sources.item.created|updated|removed events from OpenVibe.Events (signed webhook + inbox,
++ * server/http/events.js). Created/updated items are queued here (event payloads carry no
++ * fields) and fetched by id; a removal is applied at once.
++ *
++ * A failed fetch leaves everything as it was and is retried with backoff: nothing is invented to fill
++ * the gap, and a signal is never created from anything but an item Sources returned.
++ */
++const { createChangeCursor, pullChanges } = require('openvibe-publishing/ingest');
++
++const BACKOFF_MS = [15000, 60000, 5 * 60000, 30 * 60000, 2 * 3600000];
++const CURSOR = 'sources'; // the chassis cursor name; the old key sources.items.after is carried once by migration 0002
++
++/** applyItem's own outcome → the chassis' three pull outcomes ('applied' | 'hold' | 'removed'). */
++const PULL_OUTCOME = {
++ created: 'applied', updated: 'applied',
++ removed: 'removed', 'removed:unknown': 'removed',
++ unchanged: 'hold',
++ 'ignored:category': 'hold', 'ignored:removed': 'hold', 'ignored:older_revision': 'hold', 'ignored:unknown_item': 'hold',
++};
++
++function createSourcesSync({ db, svc, sources, config, now = () => Date.now(), log = console }) {
++ const cursor = createChangeCursor(db, { prefix: 'review', now });
++ const q = {
++ get: db.prepare('SELECT value FROM review_sync_state WHERE name = ?'),
++ put: db.prepare('INSERT INTO review_sync_state (name, value, updated_at) VALUES (?, ?, ?) ON CONFLICT (name) DO UPDATE SET value = excluded.value, updated_at = excluded.updated_at'),
++ enqueue: db.prepare(`INSERT INTO review_import_queue (item_id, reason, enqueued_at, next_attempt_at) VALUES (?, ?, ?, 0)
++ ON CONFLICT (item_id) DO UPDATE SET reason = excluded.reason, next_attempt_at = 0`),
++ due: db.prepare('SELECT * FROM review_import_queue WHERE next_attempt_at <= ? ORDER BY enqueued_at LIMIT ?'),
++ done: db.prepare('DELETE FROM review_import_queue WHERE item_id = ?'),
++ failed: db.prepare('UPDATE review_import_queue SET attempts = attempts + 1, next_attempt_at = ?, last_error = ? WHERE item_id = ?'),
++ pending: db.prepare('SELECT COUNT(*) AS n FROM review_import_queue'),
++ sourceExists: db.prepare('SELECT 1 FROM review_sources WHERE key = ?'),
++ };
++ const system = { kind: 'system', service: 'svc:reviews' };
++ const state = async (name) => { const r = await q.get.get(name); return r ? r.value : null; };
++ const setState = async (name, value) => await q.put.run(name, value == null ? null : String(value), now());
++
++ let pulling = null;
++ let draining = null;
++
++ // The page's source health and registry records, merged into the applyItem sourceInfo (as before).
++ let pageHealth = {};
++ const infos = new Map();
++ const pageSource = {
++ async listItems(opts) {
++ const body = await sources.listItems(opts);
++ pageHealth = body.sources || {};
++ for (const key of new Set((body.items || []).map((i) => i.source_key).filter(Boolean))) {
++ if (infos.has(key)) continue;
++ if (await q.sourceExists.get(key)) { infos.set(key, null); continue; }
++ try { const full = await sources.getSource(key); infos.set(key, full || null); } catch { infos.set(key, null); }
++ }
++ return body;
++ },
++ };
++
++ async function pullOnce({ maxPages = config.sync.maxPages } = {}) {
++ if (!sources.enabled) return { ok: false, reason: 'OpenVibe.Sources is not configured' };
++ const counts = { pages: 0, applied: 0, unchanged: 0, skipped: 0, hold: 0, removed: 0, failed: 0 };
++ try {
++ const summary = await pullChanges({
++ db, cursor, source: pageSource, name: CURSOR,
++ maxPages, pageSize: config.sync.pageSize,
++ apply: async (item) => {
++ const info = { ...(infos.get(item.source_key) || {}), ...(pageHealth[item.source_key] || {}) };
++ const out = await svc.applyItem(item, { sourceInfo: info, actor: system });
++ if (out.outcome === 'unchanged') counts.unchanged++;
++ else if (String(out.outcome).startsWith('ignored')) counts.skipped++;
++ else counts.applied++;
++ return PULL_OUTCOME[out.outcome] || 'applied';
++ },
++ onItem: (item, err, outcome) => { if (outcome === 'failed') log.warn(`[Reviews] item ${item && item.id} skipped: ${err && err.message}`); },
++ });
++ counts.pages = summary.pages;
++ counts.hold = summary.hold;
++ counts.removed = summary.removed;
++ counts.failed = summary.failed;
++ await setState('sources.last_pull_ok_at', new Date(now()).toISOString());
++ return { ok: true, after: summary.after, ...counts };
++ } catch (err) {
++ await setState('sources.last_pull_error', `${new Date(now()).toISOString()} ${String(err && err.message).slice(0, 300)}`);
++ return { ok: false, reason: err && err.message, ...counts };
++ }
++ }
++
++ async function drainQueue({ limit = 50 } = {}) {
++ if (!sources.enabled) return { ok: false, reason: 'OpenVibe.Sources is not configured', done: 0 };
++ let done = 0;
++ let failed = 0;
++ for (const row of await q.due.all(now(), limit)) {
++ try {
++ await svc.importItem(row.item_id, sources, system);
++ await q.done.run(row.item_id);
++ done++;
++ } catch (err) {
++ // A 404/422 is final (the item is gone or unusable); anything else is retried.
++ if (err && (err.status === 404 || err.status === 422)) { await q.done.run(row.item_id); log.warn(`[Reviews] queued item ${row.item_id} dropped: ${err.message}`); continue; }
++ await q.failed.run(now() + BACKOFF_MS[Math.min(row.attempts, BACKOFF_MS.length - 1)], String(err && err.message).slice(0, 500), row.item_id);
++ failed++;
++ }
++ }
++ return { ok: true, done, failed };
++ }
++
++ return {
++ async enqueue(itemId, reason) { await q.enqueue.run(String(itemId), String(reason || 'event').slice(0, 60), now()); },
++ pending: async () => (await q.pending.get()).n,
++ cursor: async () => await cursor.get(CURSOR),
++ state,
++ pull(opts) { if (!pulling) pulling = pullOnce(opts).finally(() => { pulling = null; }); return pulling; },
++ drain(opts) { if (!draining) draining = drainQueue(opts).finally(() => { draining = null; }); return draining; },
++ };
++}
++
++module.exports = { createSourcesSync, CURSOR };
ddiff --git a/server/reviews/sync.js b/server/reviews/sync.js
ddeleted file mode 100644
iindex 3d4838b..0000000
---- a/server/reviews/sync.js
++++ /dev/null
@@@ -1,103 +0,0 @@
--'use strict';
--/**
-- * How review items get from OpenVibe.Sources into Reviews. Two paths feed the same applyItem():
-- *
-- * pull GET /api/v1/items?category=reviews&include_removed=1&after=<change_seq> in change order
-- * (sources.item.read). The cursor is stored after each page, so a restart resumes where it
-- * stopped and a missed event is caught up on the next pass.
-- * push sources.item.created|updated|removed events from OpenVibe.Events (signed webhook +
-- * inbox, server/http/consumer.js). Created/updated items are queued here and fetched by
-- * id (event payloads carry no fields); a removal is applied at once.
-- *
-- * A failed fetch leaves everything as it was and is retried with backoff: nothing is invented to
-- * fill the gap, and a signal is never created from anything but an item Sources returned.
-- */
--const BACKOFF_MS = [15000, 60000, 5 * 60000, 30 * 60000, 2 * 3600000];
--const CURSOR = 'sources.items.after';
--
--function createSourcesSync({ db, svc, platform, config, now = () => Date.now(), log = console }) {
-- const q = {
-- get: db.prepare('SELECT value FROM review_sync_state WHERE name = ?'),
-- put: db.prepare('INSERT INTO review_sync_state (name, value, updated_at) VALUES (?, ?, ?) ON CONFLICT (name) DO UPDATE SET value = excluded.value, updated_at = excluded.updated_at'),
-- enqueue: db.prepare(`INSERT INTO review_import_queue (item_id, reason, enqueued_at, next_attempt_at) VALUES (?, ?, ?, 0)
-- ON CONFLICT (item_id) DO UPDATE SET reason = excluded.reason, next_attempt_at = 0`),
-- due: db.prepare('SELECT * FROM review_import_queue WHERE next_attempt_at <= ? ORDER BY enqueued_at LIMIT ?'),
-- done: db.prepare('DELETE FROM review_import_queue WHERE item_id = ?'),
-- failed: db.prepare('UPDATE review_import_queue SET attempts = attempts + 1, next_attempt_at = ?, last_error = ? WHERE item_id = ?'),
-- pending: db.prepare('SELECT COUNT(*) AS n FROM review_import_queue'),
-- };
-- const system = { kind: 'system', service: 'svc:reviews' };
-- const state = async (name) => { const r = await q.get.get(name); return r ? r.value : null; };
-- const setState = async (name, value) => await q.put.run(name, value == null ? null : String(value), now());
--
-- let pulling = null;
-- let draining = null;
--
-- async function pullOnce({ maxPages = 25 } = {}) {
-- if (!platform.sources.configured) return { ok: false, reason: 'OpenVibe.Sources is not configured' };
-- const counts = { pages: 0, applied: 0, unchanged: 0, skipped: 0 };
-- let after = Number(await state(CURSOR) || 0);
-- const known = new Set();
-- const infos = new Map();
-- try {
-- for (let page = 0; page < maxPages; page++) {
-- const body = await platform.sources.listItems({ after, limit: config.sync.pageSize });
-- counts.pages++;
-- for (const key of new Set(body.items.map((i) => i.source_key))) {
-- if (known.has(key)) continue;
-- known.add(key);
-- if (await db.prepare('SELECT 1 FROM review_sources WHERE key = ?').get(key)) continue;
-- try { const full = await platform.sources.getSource(key); if (full) infos.set(key, full); } catch { /* notes still come with the item */ }
-- }
-- await svc.batch(async () => {
-- for (const item of body.items) {
-- const info = { ...(infos.get(item.source_key) || {}), ...((body.sources && body.sources[item.source_key]) || {}) };
-- try {
-- const out = await svc.applyItem(item, { sourceInfo: info, actor: system });
-- if (out.outcome === 'unchanged') counts.unchanged++; else counts.applied++;
-- } catch (err) {
-- if (err && err.status && err.status < 500) { counts.skipped++; log.warn(`[Reviews] item ${item && item.id} skipped: ${err.message}`); } else throw err;
-- }
-- }
-- });
-- if (Number.isInteger(body.next_after) && body.next_after >= after) { after = body.next_after; await setState(CURSOR, after); }
-- if (!body.more) break;
-- }
-- await setState('sources.last_pull_ok_at', new Date(now()).toISOString());
-- return { ok: true, after, ...counts };
-- } catch (err) {
-- await setState('sources.last_pull_error', `${new Date(now()).toISOString()} ${String(err && err.message).slice(0, 300)}`);
-- return { ok: false, reason: err && err.message, after, ...counts };
-- }
-- }
--
-- async function drainQueue({ limit = 50 } = {}) {
-- if (!platform.sources.configured) return { ok: false, reason: 'OpenVibe.Sources is not configured', done: 0 };
-- let done = 0;
-- let failed = 0;
-- for (const row of await q.due.all(now(), limit)) {
-- try {
-- await svc.importItem(row.item_id, platform.sources, system);
-- await q.done.run(row.item_id);
-- done++;
-- } catch (err) {
-- // A 404/422 is final (the item is gone or unusable); anything else is retried.
-- if (err && (err.status === 404 || err.status === 422)) { await q.done.run(row.item_id); log.warn(`[Reviews] queued item ${row.item_id} dropped: ${err.message}`); continue; }
-- await q.failed.run(now() + BACKOFF_MS[Math.min(row.attempts, BACKOFF_MS.length - 1)], String(err && err.message).slice(0, 500), row.item_id);
-- failed++;
-- }
-- }
-- return { ok: true, done, failed };
-- }
--
-- return {
-- async enqueue(itemId, reason) { await q.enqueue.run(String(itemId), String(reason || 'event').slice(0, 60), now()); },
-- pending: async () => (await q.pending.get()).n,
-- cursor: async () => Number(await state(CURSOR) || 0),
-- state,
-- pull(opts) { if (!pulling) pulling = pullOnce(opts).finally(() => { pulling = null; }); return pulling; },
-- drain(opts) { if (!draining) draining = drainQueue(opts).finally(() => { draining = null; }); return draining; },
-- };
--}
--
--module.exports = { createSourcesSync };
ddiff --git a/test/chassis.test.js b/test/chassis.test.js
nnew file mode 100644
iindex 0000000..1b7a3e8
---- /dev/null
++++ b/test/chassis.test.js
@@@ -0,0 +1,119 @@
++'use strict';
++/**
++ * Plan T9 (J3) — Reviews on the openvibe-publishing 1.1.0 chassis. This proves the conversion:
++ * - the local copies the brief's §7 lists are unrequired (the harness denies rm, so the files may
++ * remain on disk; nothing under server/ requires them);
++ * - a dry-run ingest from the captured Sources fixture produces the same domain rows as the
++ * pre-conversion code (test/fixtures/ingest-golden.json, captured before the change);
++ * - the Search document validates against search.index-document@1, with its provenance and
++ * visibility;
++ * - the outbox row and the state change share the caller's transaction (a rollback drops both).
++ */
++const assert = require('assert');
++const fs = require('fs');
++const path = require('path');
++const contracts = require('openvibe-contracts');
++const { boot, req, suite, editorToken, steamItem, importItem, createEntity, outbox } = require('./helpers');
++const { feedFixture, reviseFixture, project } = require('./helpers/golden');
++
++const ROOT = path.join(__dirname, '..');
++const GOLDEN = JSON.parse(fs.readFileSync(path.join(__dirname, 'fixtures', 'ingest-golden.json'), 'utf8'));
++
++// The brief's §7 deletion list for Reviews: the chassis (openvibe-publishing/ingest and
++// /publication) owns this code now. The files themselves cannot be removed in this harness.
++const DELETED = ['server/reviews/sync.js', 'server/reviews/normalize.js', 'server/http/consumer.js']
++ .map((f) => path.join(ROOT, f));
++
++const OVERVIEW = 'Most sampled Steam reviewers recommend this game, and the sample here is broad enough to clear the thin-content gate for the entity page in this test without difficulty.';
++
++function jsFiles(dir, found = []) {
++ for (const e of fs.readdirSync(dir, { withFileTypes: true })) {
++ const p = path.join(dir, e.name);
++ if (e.isDirectory()) jsFiles(p, found);
++ else if (e.name.endsWith('.js')) found.push(p);
++ }
++ return found;
++}
++function relativeRequires(file) {
++ const src = fs.readFileSync(file, 'utf8');
++ return [...src.matchAll(/require\(\s*['"]([^'"]+)['"]\s*\)/g)].map((m) => m[1]).filter((s) => s.startsWith('.'));
++}
++
++/** An indexable entity page: three live signals and their published, gate-clearing summary. */
++async function indexableEntity(h) {
++ const e = await createEntity(h, { name: 'Portal 2', kind: 'game', aliases: [{ type: 'source', value: 'steam-reviews-portal-2' }] });
++ for (let i = 0; i < 3; i++) await importItem(h, steamItem({ votedUp: i < 2 }));
++ const sigs = (await h.db.prepare("SELECT id FROM review_signals WHERE entity_id = ? AND status = 'active' ORDER BY id").all(e.id)).map((s) => s.id);
++ const pub = await req(h, 'POST', `/api/v1/entities/${e.slug}/summary/revisions`, { token: editorToken(), body: { overview: OVERVIEW, overview_signals: sigs, publish: true } });
++ assert.strictEqual(pub.status, 201, pub.text);
++ return e;
++}
++
++const t = suite('chassis');
++
++t('the local copies are gone: no live server file requires them; the chassis provides ingest and publication', () => {
++ const gone = new Set(DELETED);
++ for (const file of jsFiles(path.join(ROOT, 'server'))) {
++ if (gone.has(file)) continue; // the (undeletable) files themselves
++ for (const spec of relativeRequires(file)) {
++ let resolved;
++ try { resolved = require.resolve(path.resolve(path.dirname(file), spec)); } catch { continue; }
++ assert.ok(!gone.has(resolved), `${path.relative(ROOT, file)} still requires ${spec}`);
++ }
++ }
++ // ...and the chassis entry points are the ones the app uses.
++ const read = (rel) => fs.readFileSync(path.join(ROOT, rel), 'utf8');
++ assert.match(read('server/index.js'), /openvibe-publishing\/ingest/);
++ assert.match(read('server/reviews/source-items.js'), /openvibe-publishing\/ingest/);
++ assert.match(read('server/http/events.js'), /openvibe-publishing\/ingest/);
++ assert.match(read('server/reviews/service.js'), /openvibe-publishing\/publication/);
++});
++
++t('a dry-run ingest from the captured Sources fixture produces the same domain rows as before', async () => {
++ const h = await boot();
++ try {
++ const it = await feedFixture(h);
++ const first = await h.sync.pull();
++ assert.strictEqual(first.ok, true, first.reason);
++ reviseFixture(h, it);
++ const second = await h.sync.pull();
++ assert.strictEqual(second.ok, true, second.reason);
++ assert.deepStrictEqual(await project(h), GOLDEN);
++ } finally { await h.stop(); }
++});
++
++t('the Search document validates against search.index-document@1, with its provenance and visibility', async () => {
++ const h = await boot();
++ try {
++ await indexableEntity(h);
++ const upserts = (await outbox(h)).filter((env) => env.event_type === 'reviews.index_document.upserted');
++ assert.ok(upserts.length >= 1, 'publishing an indexable entity sends an index document');
++ const doc = upserts[upserts.length - 1].payload;
++ const v = contracts.validate('search.index-document@1', doc);
++ assert.ok(v.valid, JSON.stringify(v.errors));
++ assert.strictEqual(doc.owner, 'reviews');
++ assert.strictEqual(doc.type, 'entity');
++ assert.strictEqual(doc.visibility, 'public');
++ assert.strictEqual(doc.indexability.decision, 'index');
++ const sources = doc.provenance.filter((p) => p.service === 'sources');
++ assert.ok(sources.length >= 1, 'the document carries its Sources provenance');
++ assert.ok(sources.every((p) => p.type === 'item' && /^itm_/.test(p.id) && p.retrieved_at), JSON.stringify(sources));
++ } finally { await h.stop(); }
++});
++
++t("the outbox row and the state change share the caller's transaction (a rollback drops both)", async () => {
++ const h = await boot();
++ try {
++ const e = await indexableEntity(h);
++ const row = await h.db.prepare('SELECT name FROM review_entities WHERE id = ?').get(e.id);
++ const count = async () => (await h.db.prepare('SELECT COUNT(*) AS n FROM review_event_outbox').get()).n;
++ const before = await count();
++ await assert.rejects(h.db.tx(async () => {
++ await h.db.prepare('UPDATE review_entities SET name = ? WHERE id = ?').run('Rolled back name', e.id);
++ await h.svc.syncIndex(e.id);
++ throw new Error('rollback: the outbox row and the state change must not survive together');
++ }), /rollback/);
++ assert.strictEqual(await count(), before, 'no outbox row survives the rollback');
++ assert.strictEqual((await h.db.prepare('SELECT name FROM review_entities WHERE id = ?').get(e.id)).name, row.name, 'the state change rolled back with it');
++ } finally { await h.stop(); }
++});
ddiff --git a/test/fixtures/ingest-golden.json b/test/fixtures/ingest-golden.json
nnew file mode 100644
iindex 0000000..aaac3db
---- /dev/null
++++ b/test/fixtures/ingest-golden.json
@@@ -0,0 +1,144 @@
++{
++ "entities": [
++ {
++ "name": "Portal 2",
++ "kind": "game",
++ "state": "active"
++ }
++ ],
++ "aliases": [
++ {
++ "type": "name",
++ "norm": "portal 2",
++ "entity": "Portal 2"
++ },
++ {
++ "type": "source",
++ "norm": "steam-reviews-portal-2",
++ "entity": "Portal 2"
++ }
++ ],
++ "sources": [
++ {
++ "key": "shop-example",
++ "name": null,
++ "status": null,
++ "stale": null,
++ "terms_note": "Terms recorded by a person in OpenVibe.Sources (test).",
++ "license_note": null
++ },
++ {
++ "key": "steam-reviews-portal-2",
++ "name": null,
++ "status": null,
++ "stale": null,
++ "terms_note": "Terms recorded by a person in OpenVibe.Sources (test).",
++ "license_note": null
++ }
++ ],
++ "items": [
++ {
++ "item": "https://shop.example/p/widget",
++ "revision": 1,
++ "source_key": "shop-example",
++ "kind": "product",
++ "state": "active",
++ "resolution": "unmatched",
++ "content_hash": "03bc67c96857ad6916c2177a93b415c5a9783cae7efc5cf68d97c6981a13cf7b",
++ "entity": null
++ },
++ {
++ "item": "review-a",
++ "revision": 2,
++ "source_key": "steam-reviews-portal-2",
++ "kind": "review_signal",
++ "state": "active",
++ "resolution": "resolved",
++ "content_hash": "3235d635f9c78563605be8633de352392a0e96e0aba6c4105247a0441441cca6",
++ "entity": "Portal 2"
++ },
++ {
++ "item": "review-b",
++ "revision": 1,
++ "source_key": "steam-reviews-portal-2",
++ "kind": "review_signal",
++ "state": "removed",
++ "resolution": "resolved",
++ "content_hash": "3235d635f9c78563605be8633de352392a0e96e0aba6c4105247a0441441cca6",
++ "entity": "Portal 2"
++ }
++ ],
++ "signals": [
++ {
++ "item": "review-a",
++ "revision": 1,
++ "type": "recommendation",
++ "status": "superseded",
++ "recommended": 1,
++ "positive_count": null,
++ "total_count": null,
++ "entity": "Portal 2"
++ },
++ {
++ "item": "review-a",
++ "revision": 2,
++ "type": "recommendation",
++ "status": "active",
++ "recommended": 0,
++ "positive_count": null,
++ "total_count": null,
++ "entity": "Portal 2"
++ },
++ {
++ "item": "review-b",
++ "revision": 1,
++ "type": "recommendation",
++ "status": "withdrawn",
++ "recommended": 0,
++ "positive_count": null,
++ "total_count": null,
++ "entity": "Portal 2"
++ }
++ ],
++ "aggregates": [
++ {
++ "entity": "Portal 2",
++ "result": {
++ "components": {
++ "recommendation": {
++ "inputs": [
++ "review-a#2"
++ ],
++ "percent": 0,
++ "positive": 0,
++ "total": 1
++ }
++ },
++ "computation": [
++ "recommendation: 0 of 1 recommend (0%) = sum of positives / sum of totals over 1 signal"
++ ],
++ "excluded": [],
++ "inputs": [
++ {
++ "entity_id": "Portal 2",
++ "observed_at": "2026-09-20T13:00:01.000Z",
++ "recommended": false,
++ "signal_id": "review-a#2",
++ "source_item_id": "item:review-a",
++ "source_key": "steam-reviews-portal-2",
++ "type": "recommendation",
++ "weight": 1
++ }
++ ],
++ "method": "reviews-aggregate@1",
++ "observed": {
++ "earliest": "2026-09-20T13:00:01.000Z",
++ "latest": "2026-09-20T13:00:01.000Z"
++ },
++ "sources": [
++ "steam-reviews-portal-2"
++ ]
++ }
++ }
++ ]
++}
ddiff --git a/test/helpers.js b/test/helpers.js
iindex a7187c7..618979d 100644
---- a/test/helpers.js
++++ b/test/helpers.js
@@@ -151,14 +151,17 @@ async function boot({ env = {}, sources = fakeSources(), community = null, now,
const config = load({
NODE_ENV: 'test', PORT: '0', HOST: '127.0.0.1', BASE_URL: 'http://reviews.test',
OV_NETWORK_URL: ISSUER, OV_NETWORK_INTERNAL_URL: '', REVIEWS_GATE_MIN_WORDS: '20',
++ OV_OAUTH_CLIENT_SECRET: 'test-service-secret-not-a-real-one',
OV_SOURCES_INTERNAL_URL: 'http://sources.test', OV_COMMUNITY_INTERNAL_URL: 'http://community.test',
REVIEWS_EDITORS: EDITOR, REVIEWS_EVENTS_SECRET: WEBHOOK_SECRET,
...env,
});
const fetchImpl = async (url, opts) => {
-- const host = new URL(url).host;
-- if (host === 'sources.test') return sources.fetch(url, opts);
-- if (host === 'community.test' && community) return community.fetch(url, opts);
++ const u = new URL(url);
++ // The ingest chassis' Sources client mints a client-credentials token here (no network in tests).
++ if (u.pathname === '/oauth/token') return { status: 200, ok: true, json: async () => ({ access_token: 'stub-sources-token', token_type: 'Bearer', expires_in: 300, scope: 'sources.item.read sources.source.read' }) };
++ if (u.host === 'sources.test') return sources.fetch(url, opts);
++ if (u.host === 'community.test' && community) return community.fetch(url, opts);
throw new Error(`unexpected outbound fetch ${url}`);
};
// One database per boot (PGlite, or REVIEWS_TEST_STORE=pg: the containers), dropped when the boot stops.
ddiff --git a/test/helpers/golden.js b/test/helpers/golden.js
nnew file mode 100644
iindex 0000000..47e92b6
---- /dev/null
++++ b/test/helpers/golden.js
@@@ -0,0 +1,67 @@
++'use strict';
++/**
++ * The Reviews ingest golden fixture: a deterministic Sources feed and the canonical projection of the
++ * domain rows it produces (entities, aliases, source items, signals, the current aggregate per entity).
++ * The same fixture and projection run before and after the chassis conversion, so test/chassis.test.js
++ * can compare the projection to test/fixtures/ingest-golden.json (captured from the pre-conversion
++ * code). Volatile values (ULID ids, timestamps) are left out so only domain meaning is compared.
++ */
++const { createEntity, steamItem, productItem } = require('../helpers');
++
++const SRC = 'steam-reviews-portal-2';
++const parse = (s, d) => { try { return s ? JSON.parse(s) : d; } catch { return d; } };
++
++/** Add the fixture's items to the Sources mock (same order and fields as the capture run). → items */
++async function feedFixture(h) {
++ await createEntity(h, { name: 'Portal 2', kind: 'game', aliases: [{ type: 'source', value: SRC }] });
++ const a = h.sources.put(steamItem({ votedUp: true, identity: 'review-a' }));
++ const b = h.sources.put(steamItem({ votedUp: false, identity: 'review-b' }));
++ const p = h.sources.put(productItem({ url: 'https://shop.example/p/widget', title: 'Widget 3000', gtin: '0012345678905' }));
++ const ignored = h.sources.put({ ...steamItem({ identity: 'review-c' }), category: 'blog' }); // not review: skipped
++ return { a, b, p, ignored };
++}
++
++/** A revision upstream (vote flipped) and a removal, applied by a second pull. */
++function reviseFixture(h, it) {
++ h.sources.revise(it.a.id, { voted_up: false });
++ h.sources.remove(it.b.id, 'takedown');
++}
++
++/** The canonical projection of the domain rows. Deterministic in everything but ids. */
++async function project(h) {
++ const db = h.db;
++ // ULID ids are volatile between runs: map them to stable domain keys (a Sources item's identity,
++ // a signal's item#revision, an entity's name) so only domain meaning is compared.
++ const itmMap = new Map((await db.prepare('SELECT id, identity FROM review_source_items').all()).map((r) => [r.id, r.identity]));
++ const entMap = new Map((await db.prepare('SELECT id, name FROM review_entities').all()).map((r) => [r.id, r.name]));
++ const sigMap = new Map((await db.prepare('SELECT id, source_item_id, item_revision FROM review_signals').all())
++ .map((r) => [r.id, `${itmMap.get(r.source_item_id) || r.source_item_id}#${r.item_revision}`]));
++ const iid = (id) => itmMap.get(id) || id;
++ const entities = (await db.prepare('SELECT name, kind, state FROM review_entities ORDER BY lower(name)').all())
++ .map((e) => ({ name: e.name, kind: e.kind, state: e.state }));
++ const aliases = (await db.prepare(`SELECT a.type, a.norm, e.name AS entity FROM review_entity_aliases a
++ JOIN review_entities e ON e.id = a.entity_id WHERE a.removed_at IS NULL ORDER BY a.type, a.norm`).all())
++ .map((a) => ({ type: a.type, norm: a.norm, entity: a.entity }));
++ const sources = (await db.prepare('SELECT key, name, status, stale, terms_note, license_note FROM review_sources ORDER BY key').all())
++ .map((s) => ({ key: s.key, name: s.name, status: s.status, stale: s.stale, terms_note: s.terms_note, license_note: s.license_note }));
++ const items = (await db.prepare(`SELECT i.id AS source_item_id, i.item_revision, i.source_key, i.kind, i.state, i.resolution, i.content_hash, e.name AS entity
++ FROM review_source_items i LEFT JOIN review_entities e ON e.id = i.entity_id ORDER BY i.identity`).all())
++ .map((r) => ({ item: iid(r.source_item_id), revision: r.item_revision, source_key: r.source_key, kind: r.kind, state: r.state, resolution: r.resolution, content_hash: r.content_hash, entity: r.entity || null }));
++ const signals = (await db.prepare(`SELECT s.source_item_id, s.item_revision, s.type, s.status, s.recommended, s.positive_count, s.total_count, e.name AS entity
++ FROM review_signals s LEFT JOIN review_entities e ON e.id = s.entity_id ORDER BY s.source_item_id, s.item_revision, s.type`).all())
++ .map((s) => ({ item: iid(s.source_item_id), revision: s.item_revision, type: s.type, status: s.status, recommended: s.recommended, positive_count: s.positive_count, total_count: s.total_count, entity: s.entity || null }));
++ const stable = (v) => {
++ if (typeof v === 'string') return sigMap.get(v) || entMap.get(v) || (itmMap.get(v) ? `item:${itmMap.get(v)}` : v);
++ if (Array.isArray(v)) return v.map(stable);
++ if (v && typeof v === 'object') { const o = {}; for (const k of Object.keys(v).sort()) o[k] = stable(v[k]); return o; }
++ return v;
++ };
++ const aggregates = (await db.prepare(`SELECT e.name AS entity, a.result FROM review_aggregates a
++ JOIN review_entities e ON e.id = a.entity_id
++ WHERE a.revision = (SELECT MAX(b.revision) FROM review_aggregates b WHERE b.entity_id = a.entity_id)
++ ORDER BY lower(e.name)`).all())
++ .map((a) => ({ entity: a.entity, result: stable(parse(a.result, null)) }));
++ return { entities, aliases, sources, items, signals, aggregates };
++}
++
++module.exports = { feedFixture, reviseFixture, project, SRC };
ddiff --git a/test/indexnow.test.js b/test/indexnow.test.js
iindex 546575b..3984752 100644
---- a/test/indexnow.test.js
++++ b/test/indexnow.test.js
@@@ -39,7 +39,7 @@ t('with a key the key file answers text/plain with the key', async () => {
} finally { await h.stop(); }
});
--t('a draft never pings; a publish pings the page path and the sitemap; an unpublish pings again', async () => {
++t('a draft never pings; a publish pings the page path and the sitemap; a page that only goes noindex is never pinged', async () => {
// A spy in place of the module's HTTP send: records every pingSoon batch.
const pings = [];
const spy = {
@@@ -65,7 +65,8 @@ t('a draft never pings; a publish pings the page path and the sitemap; an unpubl
pings.length = 0;
const unpub = await req(h, 'POST', `/api/v1/entities/${e.slug}/summary/unpublish`, { token: editorToken() });
assert.strictEqual(unpub.status, 200, unpub.text);
-- assert.ok(pings.includes('http://reviews.test/e/portal-2'), JSON.stringify(pings));
-- assert.ok(pings.includes('http://reviews.test/sitemap.xml'), JSON.stringify(pings));
++ // The entity page stays (its signals remain) but goes noindex: the chassis pings an indexable
++ // page that appeared or changed, or one Search had that went away — never a noindex page.
++ assert.deepStrictEqual(pings, [], 'a page that stays but goes noindex is never pinged');
} finally { await h.stop(); }
});
ddiff --git a/test/resolution.test.js b/test/resolution.test.js
iindex 7b7e21d..8987800 100644
---- a/test/resolution.test.js
++++ b/test/resolution.test.js
@@@ -4,7 +4,7 @@
* identifiers wait for an editor, and every editorial decision is recorded.
*/
const assert = require('assert');
--const norm = require('../server/reviews/normalize');
++const { normalize: norm } = require('openvibe-publishing/ingest');
const {
boot, req, suite, productItem, importItem, createEntity, editorToken, readerToken, serviceToken,
} = require('./helpers');