openvibers/openvibe.deals Commits

Deals on the Publishing 1.1.0 chassis (plan T9 J4): the Sources client, change cursor and pull loop, signed event consumer, normalisers and publication glue come from openvibe-publishing/ingest and /publication; Deals keeps its domain half (server/domain/source-items.js, values.js, http/events.js). Migration 0002 seeds deals_ingest_cursor once from import_state (continuity, never a reset to zero). The replaced importer.js, util.js, http/internal.js and clients/sources.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 green (verified).

OP OpenVibers committed 29 Sept 2026 · c56acd8b35 · 33 files, +826 −567


Changes

ddiff --git a/README.md b/README.md
iindex 459b318..38f4d23 100644
---- a/README.md
++++ b/README.md
@@@ -252,8 +252,10 @@ render time, so the cache stays short); signed-in views, forms, API responses, e
- **PostgreSQL 18 and Valkey 9** (OpenVibe.Host `roles/data/`, ADR-035): every read and write is async through
`openvibe-sdk/db`; Valkey holds the per-actor limit counters (optional: without `VALKEY_URL` they count per process).
--- **Packages** (pinned release tarballs): `openvibe-publishing` v1.0.0 (seo gate, JSON-LD, feeds,
-- sitemaps, index-hooks, discussion, ssr), `openvibe-contracts` v0.79.0, `openvibe-shared` v2.2.0
++- **Packages** (pinned release tarballs): `openvibe-publishing` v1.1.0 (seo gate, JSON-LD, feeds,
++ sitemaps, index-hooks, discussion, ssr, and the shared ingest/publication chassis — the Sources
++ client, change cursor and pull loop, signed event consumer, common normalisers, and the publication
++ glue Deals' indexing uses), `openvibe-contracts` v0.79.0, `openvibe-shared` v2.2.0
(Frame, app icon, footer, legal, release, metrics, ready), `openvibe-sdk` v0.25.1 (sso, outbox,
inbox, webhook signatures v2, service tokens, per-actor limits).
- **OpenVibe.Network:** SSO (OAuth client `deals`, registered in production), JWKS,
ddiff --git a/STATUS.json b/STATUS.json
iindex b4d6d73..d24d596 100644
---- a/STATUS.json
++++ b/STATUS.json
@@@ -9,7 +9,7 @@
"domain": "openvibe.deals",
"contracts": "openvibe-contracts v0.79.0",
"packages": [
-- "openvibe-publishing v1.0.0",
++ "openvibe-publishing v1.1.0",
"openvibe-contracts v0.79.0",
"openvibe-shared v2.2.0",
"openvibe-sdk v0.25.1"
ddiff --git a/migrations/0002_ingest_cursor.sql b/migrations/0002_ingest_cursor.sql
nnew file mode 100644
iindex 0000000..2204d3a
---- /dev/null
++++ b/migrations/0002_ingest_cursor.sql
@@@ -0,0 +1,18 @@
++-- phase: expand
++-- The ingest cursor moves from import_state ('sources.deals.after') to the chassis' own table
++-- (openvibe-publishing/ingest, prefix deals → deals_ingest_cursor). Additive: the old key is read
++-- once here so a deployment continues from the cursor it had, never from zero.
++-- The DDL is openvibe-publishing lib/ingest.js schema('deals').
++
++CREATE TABLE IF NOT EXISTS deals_ingest_cursor (
++ name text COLLATE "C" NOT NULL,
++ cursor bigint NOT NULL,
++ updated_at bigint NOT NULL,
++ PRIMARY KEY (name)
++);
++
++INSERT INTO deals_ingest_cursor (name, cursor, updated_at)
++SELECT 'sources.deals.after', value::bigint, (EXTRACT(EPOCH FROM now()) * 1000)::bigint
++FROM import_state
++WHERE key = 'sources.deals.after' AND value ~ '^[0-9]+$'
++ON CONFLICT (name) DO NOTHING;
ddiff --git a/package-lock.json b/package-lock.json
iindex 47ce91b..3047225 100644
---- a/package-lock.json
++++ b/package-lock.json
@@@ -16,7 +16,7 @@
"helmet": "^8.1.0",
"iovalkey": "^0.4.0",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.79.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.25.1",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.2.0",
"pg": "^8.23.0"
@@@ -1163,9 +1163,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 09d4a7b..9481a49 100644
---- a/package.json
++++ b/package.json
@@@ -27,7 +27,7 @@
"helmet": "^8.1.0",
"iovalkey": "^0.4.0",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.79.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.25.1",
"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 58d7181..66125af 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -21,7 +21,7 @@ const { createSsoClient } = require('openvibe-sdk/sso');
const { createViewerResolver } = require('./auth/viewer');
const { createPeople } = require('./clients/network');
const { createCommunity } = require('./clients/community');
--const { createSources } = require('./clients/sources');
++const { createSourcesClient } = require('openvibe-publishing/ingest');
const { createServiceOutbox } = require('openvibe-sdk/events');
const { createIndexNow } = require('openvibe-shared/indexnow');
const { createReads } = require('./domain/reads');
@@@ -36,11 +36,11 @@ const { createOffers } = require('./domain/offers');
const { createFlags } = require('./domain/flags');
const { createVotes } = require('./domain/votes');
const { createListings } = require('./domain/listings');
--const { createImporter } = require('./domain/importer');
++const { createImporter } = require('./domain/source-items');
const { createPages } = require('./http/pages');
const { createApi } = require('./http/api');
const { createDiscoveryRoutes } = require('./http/discovery');
--const { createInternalRoutes } = require('./http/internal');
++const { createEvents } = require('./http/events');
const { createDealsReadiness } = require('./observability');
const { createActorLimits } = require('./http/actor-limits');
const { createWorker } = require('./worker');
@@@ -65,7 +65,7 @@ async function createApp(opts = {}) {
});
const people = createPeople({ store, config, fetchImpl });
const community = createCommunity({ store, config, fetchImpl });
-- const sources = createSources({ config, fetchImpl });
++ const sources = createSourcesClient({ config, fetchImpl });
const reads = createReads({ store });
const catalog = createCatalog({ store });
const publication = createPublication({ config, store, reads, catalog });
@@@ -75,7 +75,7 @@ async function createApp(opts = {}) {
const indexnow = opts.indexnow !== undefined ? opts.indexnow : createIndexNow({
host: config.baseUrl, key: config.indexnowKey, fetch: fetchImpl, log,
});
-- const indexing = createIndexing({ store, publication, outbox, catalog, indexnow });
++ const indexing = createIndexing({ config, store, publication, outbox, catalog, indexnow });
const hotness = createHotness({ store, reads });
const watches = createWatches({ config, store, reads, catalog, publication, outbox });
const limits = createLimits({ config, store });
@@@ -168,7 +168,7 @@ async function createApp(opts = {}) {
// ── Events consumer, API ────────────────────────────────
// Never per-actor limited: Events pushes at its own pace, and a 429 would only make it retry and fall behind.
-- app.use(createInternalRoutes({ config, store, importer }));
++ app.use(createEvents({ config, store, importer }));
app.use('/api/v1', rateLimit({ windowMs: 60_000, max: 240, standardHeaders: true, legacyHeaders: false }), createApi(ctx));
// ── Discovery, public pages ─────────────────────────────
ddiff --git a/server/clients/sources.js b/server/clients/sources.js
ddeleted file mode 100644
iindex 8cb4a4c..0000000
---- a/server/clients/sources.js
++++ /dev/null
@@@ -1,47 +0,0 @@
--'use strict';
--
--/**
-- * OpenVibe.Sources — the deals-category items (capability sources.item.read, audience
-- * openvibe.sources, Deals' client-credentials token).
-- *
-- * items({ after, limit }) GET /api/v1/items?category=deals&after=<change_seq>&include_removed=1
-- * → { items, next_after, more, sources: { key: { status, stale, last_success_at } } }
-- * item(id) GET /api/v1/items/:id → { item, source }
-- *
-- * A failed call throws; the importer keeps its cursor and tries again later. Nothing is invented
-- * for an item Deals could not read.
-- */
--const { serviceAuth, http } = require('openvibe-contracts');
--
--function createSources({ config, fetchImpl = globalThis.fetch }) {
-- const base = config.sources.internalUrl;
-- const enabled = Boolean(config.sources.enabled && config.oauth.clientSecret && base);
-- const tokens = enabled ? serviceAuth.createTokenClient({
-- tokenUrl: `${config.networkInternalUrl}/oauth/token`, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret,
-- audience: 'openvibe.sources', scope: 'sources.item.read', fetchImpl,
-- }) : null;
--
-- async function call(path, ctx) {
-- if (!tokens) throw new Error('Sources import is not configured (OV_OAUTH_CLIENT_SECRET / OV_SOURCES_INTERNAL_URL / DEALS_IMPORT)');
-- const res = await fetchImpl(`${base}${path}`, {
-- headers: { Accept: 'application/json', ...(ctx ? http.outboundHeaders(ctx) : {}), ...(await tokens.authHeaders()) },
-- signal: AbortSignal.timeout(10000),
-- });
-- if (res.status === 401 && tokens.invalidate) tokens.invalidate();
-- const data = await res.json().catch(() => null);
-- if (!res.ok || !data) {
-- const err = new Error(`Sources answered ${res.status}${data && data.code ? ` (${data.code})` : ''}`);
-- err.status = res.status;
-- throw err;
-- }
-- return data;
-- }
--
-- return {
-- enabled,
-- items: async ({ after = 0, limit = 100 } = {}, ctx) => await call(`/api/v1/items?category=deals&after=${encodeURIComponent(after)}&limit=${encodeURIComponent(limit)}&include_removed=1`, ctx),
-- item: async (id, ctx) => await call(`/api/v1/items/${encodeURIComponent(id)}`, ctx),
-- };
--}
--
--module.exports = { createSources };
ddiff --git a/server/config.js b/server/config.js
iindex 90cc117..80b922a 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -89,6 +89,7 @@ function load(env = process.env) {
// OpenVibe.Sources: deals-category items (pull by change cursor; events only wake the pull).
sources: {
internalUrl: trim(env.OV_SOURCES_INTERNAL_URL || 'http://127.0.0.1:4720'),
++ category: 'deals',
enabled: env.DEALS_IMPORT !== 'off',
intervalMs: int(env.DEALS_IMPORT_INTERVAL_MS, 5 * 60 * 1000),
pageSize: int(env.DEALS_IMPORT_PAGE_SIZE, 100),
ddiff --git a/server/domain/catalog.js b/server/domain/catalog.js
iindex f1bea9e..29e1644 100644
---- a/server/domain/catalog.js
++++ b/server/domain/catalog.js
@@@ -9,7 +9,7 @@
* product an alias names, or creates a product when allowed. Two products that turn out to be the
* same are left to moderators (Deals does not auto-merge products).
*/
--const { ApiError, newId, slugify, text, longText } = require('./util');
++const { ApiError, newId, slugify, text, longText } = require('./values');
const ALIAS_KINDS = ['gtin', 'mpn', 'sku', 'url', 'name'];
const normAlias = (kind, v) => {
ddiff --git a/server/domain/flags.js b/server/domain/flags.js
iindex 3773149..43cbc6b 100644
---- a/server/domain/flags.js
++++ b/server/domain/flags.js
@@@ -7,7 +7,7 @@
* noise. Flags never change an offer by themselves; a moderator decides (expire, disable, merge,
* dismiss).
*/
--const { ApiError, newId, text, iso } = require('./util');
++const { ApiError, newId, text, iso } = require('./values');
const USER_KINDS = ['expired', 'price_wrong', 'spam', 'duplicate', 'other'];
ddiff --git a/server/domain/importer.js b/server/domain/importer.js
ddeleted file mode 100644
iindex b48d4f2..0000000
---- a/server/domain/importer.js
++++ /dev/null
@@@ -1,258 +0,0 @@
--'use strict';
--
--/**
-- * Imports deals-category items from OpenVibe.Sources.
-- *
-- * The truth is Sources' change feed (GET /api/v1/items?category=deals&after=<cursor>): the importer
-- * pulls pages in change order and keeps its cursor in import_state. sources.item.* events delivered
-- * to /internal/events only wake the pull early (they carry no prices), so a lost or repeated event
-- * changes nothing.
-- *
-- * Item → offer mapping (values exactly as the source stated them; missing = null):
-- * kind offer one offer: fields.price + fields.currency, availability / condition (schema.org
-- * terms mapped to Deals' enums, unknown terms → null), valid_until → stated expiry,
-- * seller → store name
-- * kind product a product (resolved by gtin / mpn / sku / name) and one offer per stated offer
-- * kind article (RSS/Atom) one offer with the title and link; the price is NOT parsed out of the
-- * headline — it stays "not stated" until a structured source or a person states it
-- * other kinds skipped
-- *
-- * observed_at = the item's provenance.retrieved_at (when Sources saw it). An item Deals already
-- * imported produces a new observation only when its revision or its retrieved_at advanced, so
-- * replays are no-ops. Two items (or a submission and an item) with the same link attach to the same
-- * offer as separate sources. A removed item (takedown, licence) removes its source; an imported
-- * offer left with no source is disabled.
-- *
-- * Imported text is third-party: review_state = pending, so the page is noindex until a person
-- * reviews it.
-- */
--const {
-- normalizeUrl, urlKey, parseAmount, SCHEMA_AVAILABILITY, SCHEMA_CONDITION, iso,
--} = require('./util');
--
--const CURSOR_KEY = 'sources.deals.after';
--
--function createImporter({ config, store, reads, catalog, offers, indexing, sources, log = console }) {
-- const { db } = store;
-- const q = {
-- getState: db.prepare('SELECT value FROM import_state WHERE key = ?'),
-- setState: db.prepare(`INSERT INTO import_state (key, value, updated_at) VALUES (?, ?, ?)
-- ON CONFLICT (key) DO UPDATE SET value = excluded.value, updated_at = excluded.updated_at`),
-- sourceRow: db.prepare("SELECT * FROM deal_offer_sources WHERE kind = 'sources_item' AND ref_service = 'sources' AND ref_id = ? AND ref_part = ?"),
-- itemRows: db.prepare("SELECT * FROM deal_offer_sources WHERE kind = 'sources_item' AND ref_service = 'sources' AND ref_id = ? AND removed_at IS NULL"),
-- bumpSource: db.prepare('UPDATE deal_offer_sources SET ref_revision = ?, retrieved_at = ?, label = COALESCE(?, label), updated_at = ? WHERE id = ?'),
-- removeSource: db.prepare('UPDATE deal_offer_sources SET removed_at = ?, removed_reason = ?, updated_at = ? WHERE id = ?'),
-- liveSources: db.prepare('SELECT COUNT(*) AS n FROM deal_offer_sources WHERE offer_id = ? AND removed_at IS NULL'),
-- refreshDue: db.prepare(`SELECT s.ref_id, MAX(o.observed_at) AS last FROM deal_offer_sources s
-- JOIN deal_offers f ON f.id = s.offer_id
-- JOIN deal_price_observations o ON o.source_id = s.id
-- WHERE s.kind = 'sources_item' AND s.removed_at IS NULL AND f.status = 'active'
-- GROUP BY s.ref_id HAVING MAX(o.observed_at) < ? ORDER BY last LIMIT ?`),
-- };
-- let lastError = null;
-- let lastRunAt = null;
-- let running = null;
-- let kickTimer = null;
--
-- const cursor = async () => Number((await q.getState.get(CURSOR_KEY) || {}).value || 0);
--
-- function amount(v) { try { return parseAmount(v, 'price'); } catch { return null; } }
-- function currencyOf(v) { return typeof v === 'string' && /^[A-Z]{3}$/.test(v) ? v : null; }
--
-- /** Stated fields → observation fields. A malformed value is null, never repaired or guessed. */
-- function obsFields(f = {}) {
-- const price = amount(f.price);
-- return {
-- price: price ? price.text : null, price_num: price ? price.num : null, currency: currencyOf(f.currency),
-- shipping: null, shipping_num: null, shipping_note: null,
-- condition: SCHEMA_CONDITION[f.condition] || null,
-- availability: SCHEMA_AVAILABILITY[f.availability] || null,
-- };
-- }
--
-- function candidates(item) {
-- const f = item.fields || {};
-- const stated = (t) => { const ms = t ? Date.parse(t) : NaN; return Number.isFinite(ms) ? ms : null; };
-- if (item.kind === 'offer') {
-- const url = normalizeUrl(f.url || item.canonical_url);
-- if (!url || !item.title) return [];
-- return [{ part: '', url, title: item.title, description: item.summary, obs: obsFields(f), expires_at: stated(f.valid_until), store_name: f.seller || null }];
-- }
-- if (item.kind === 'product') {
-- if (!item.title) return [];
-- const seen = new Set();
-- const out = [];
-- for (const o of (Array.isArray(f.offers) ? f.offers : []).slice(0, 20)) {
-- const url = normalizeUrl(o.url || item.canonical_url);
-- if (!url || seen.has(urlKey(url))) continue;
-- seen.add(urlKey(url));
-- out.push({
-- part: urlKey(url), url, title: o.seller ? `${item.title} — ${o.seller}` : item.title, description: item.summary,
-- obs: obsFields(o), expires_at: stated(o.valid_until), store_name: o.seller || null,
-- product: { name: item.title, brand: f.brand, gtin: f.gtin, mpn: f.mpn, sku: f.sku },
-- });
-- }
-- return out;
-- }
-- if (item.kind === 'article' || item.kind === 'record') {
-- const url = normalizeUrl(item.canonical_url);
-- if (!url || !item.title) return [];
-- return [{ part: '', url, title: item.title, description: item.summary, obs: obsFields({}), expires_at: null, store_name: null }];
-- }
-- return [];
-- }
--
-- async function removeItem(item) {
-- const reason = (item.removed && item.removed.reason) || 'removed at the source';
-- const rows = await q.itemRows.all(item.id);
-- for (const s of rows) await q.removeSource.run(store.now(), reason, store.now(), s.id);
-- const touched = [...new Set(rows.map((s) => s.offer_id))];
-- for (const id of touched) {
-- const offer = await reads.get(id);
-- if (offer.origin === 'import' && offer.status !== 'disabled' && (await q.liveSources.get(id)).n === 0) {
-- const now = store.now();
-- await db.prepare("UPDATE deal_offers SET status = 'disabled', disabled_at = ?, disabled_reason = ?, disabled_by = 'svc:deals', updated_at = ? WHERE id = ?")
-- .run(now, `source item removed: ${reason}`, now, id);
-- await offers.logAction('disable', { offerId: id, actor: 'svc:deals', reason: `source item removed: ${reason}`, before: { status: offer.status }, after: { status: 'disabled' } });
-- }
-- const after = await reads.get(id);
-- await indexing.emitOffer('deals.offer.updated', after, { actor: 'svc:deals', extra: { changed: ['sources'], source_removed: item.id } });
-- await indexing.reindex(after);
-- }
-- return touched.length ? 'removed' : 'removed:unknown';
-- }
--
-- /** One item, inside a transaction. Returns what happened (for logs and tests). */
-- async function importItem(item) {
-- if (!item || item.category !== 'deals') return 'skipped:category';
-- if (item.removed) return await removeItem(item);
-- const retrievedAt = Date.parse(item.provenance && item.provenance.retrieved_at);
-- if (!Number.isFinite(retrievedAt)) return 'skipped:no_observation_time';
-- const list = candidates(item);
-- if (!list.length) return `skipped:${item.kind}`;
-- const outcomes = [];
-- for (const c of list) {
-- const src = {
-- kind: 'sources_item', ref_service: 'sources', ref_type: 'item', ref_id: item.id, ref_part: c.part, ref_revision: item.revision,
-- source_key: item.source_key, url: item.canonical_url, label: item.title,
-- license_note: item.provenance.license_note || null, retrieved_at: retrievedAt,
-- };
-- const existing = await q.sourceRow.get(item.id, c.part);
-- if (existing) {
-- const advanced = item.revision > (existing.ref_revision || 0) || retrievedAt > (existing.retrieved_at || 0);
-- if (!advanced || existing.removed_at) { outcomes.push('unchanged'); continue; }
-- const origin = item.revision > (existing.ref_revision || 0) ? 'import' : 'import_refresh';
-- await q.bumpSource.run(Math.max(item.revision, existing.ref_revision || 0), Math.max(retrievedAt, existing.retrieved_at || 0), item.title, store.now(), existing.id);
-- const obs = await offers.recordObservation(existing.offer_id, existing.id, c.obs, { origin, observedAt: retrievedAt, sourceRevision: item.revision });
-- const offer = await reads.get(existing.offer_id);
-- const changed = ['observation'];
-- if (origin === 'import' && c.expires_at && offer.status === 'active' && offer.origin === 'import' && c.expires_at !== offer.expires_at) {
-- await db.prepare('UPDATE deal_offers SET expires_at = ?, updated_at = ? WHERE id = ?').run(c.expires_at, store.now(), offer.id);
-- changed.push('expires_at');
-- }
-- // Anything else the observation moved on the offer (price, product) is named too, so consumers never miss a field.
-- const after = await reads.get(offer.id);
-- for (const k of ['product_id', 'title', 'status']) if (after && offer && after[k] !== offer[k] && !changed.includes(k)) changed.push(k);
-- await indexing.emitOffer('deals.offer.updated', after, { actor: 'svc:deals', extra: { changed, observation_id: obs.id } });
-- await indexing.reindex(await reads.get(offer.id));
-- outcomes.push(origin === 'import' ? 'updated' : 'refreshed');
-- continue;
-- }
-- let productId = null;
-- if (c.product) {
-- try { productId = (await catalog.resolve(c.product, { source: 'import', actor: 'svc:deals' })).product.id; } catch { productId = null; }
-- }
-- const dup = await offers.findByUrl(c.url);
-- if (dup) {
-- const s = await offers.ensureSource(dup.id, src);
-- const obs = await offers.recordObservation(dup.id, s.id, c.obs, { origin: 'import', observedAt: retrievedAt, sourceRevision: item.revision });
-- if (productId && !dup.product_id) await db.prepare('UPDATE deal_offers SET product_id = ?, updated_at = ? WHERE id = ?').run(productId, store.now(), dup.id);
-- await indexing.emitOffer('deals.offer.updated', await reads.get(dup.id), { actor: 'svc:deals', extra: { changed: ['sources'], observation_id: obs.id } });
-- await indexing.reindex(await reads.get(dup.id));
-- outcomes.push('attached');
-- continue;
-- }
-- await offers.createImported({
-- url: c.url, title: String(c.title).slice(0, 200), description: c.description, product_id: productId,
-- store_name: c.store_name, expires_at: c.expires_at != null && c.expires_at > store.now() ? c.expires_at : null,
-- }, src, c.obs, { observedAt: retrievedAt, sourceRevision: item.revision });
-- outcomes.push('created');
-- }
-- return outcomes.join(',');
-- }
--
-- const importOne = async (item) => await db.tx(async () => await importItem(item));
--
-- /** Pull the change feed from the cursor. Never throws; the cursor only moves past applied items. */
-- async function pull() {
-- if (!sources.enabled) return { skipped: 'import off' };
-- if (running) return running;
-- running = (async () => {
-- const summary = { pages: 0, items: 0, outcomes: {} };
-- try {
-- for (let p = 0; p < config.sources.maxPages; p++) {
-- const after = await cursor();
-- const page = await sources.items({ after, limit: config.sources.pageSize });
-- summary.pages++;
-- await store.tx(async () => {
-- for (const item of page.items || []) {
-- let out;
-- // Each item in its own savepoint: one unreadable item is reported, not a stuck cursor.
-- try { out = await importOne(item); } catch (err) { out = 'failed'; log.warn(`[Deals] import of ${item && item.id} failed: ${err.message}`); }
-- summary.items++;
-- for (const o of out.split(',')) summary.outcomes[o] = (summary.outcomes[o] || 0) + 1;
-- }
-- if (Number.isFinite(page.next_after) && page.next_after > after) await q.setState.run(CURSOR_KEY, String(page.next_after), store.now());
-- });
-- if (!page.more) break;
-- }
-- lastError = null;
-- } catch (err) {
-- lastError = err.message;
-- log.warn('[Deals] Sources import failed (will retry):', err.message);
-- } finally {
-- lastRunAt = store.now();
-- running = null;
-- }
-- return summary;
-- })();
-- return running;
-- }
--
-- /**
-- * Re-confirm imported offers whose latest observation is getting old: when Sources has fetched
-- * the item again (retrieved_at advanced) the unchanged item is a new observation at that time.
-- * A source that has not been fetched successfully simply lets the offer go stale.
-- */
-- async function refresh() {
-- if (!sources.enabled) return { skipped: 'import off' };
-- const due = await q.refreshDue.all(store.now() - config.freshnessMs / 2, config.sources.refreshBatch);
-- const out = { checked: 0, outcomes: {} };
-- for (const { ref_id: id } of due) {
-- try {
-- const data = await sources.item(id);
-- out.checked++;
-- const o = await store.tx(async () => await importItem(data.item));
-- for (const x of o.split(',')) out.outcomes[x] = (out.outcomes[x] || 0) + 1;
-- } catch (err) {
-- if (err.status === 404) continue;
-- lastError = err.message;
-- break;
-- }
-- }
-- return out;
-- }
--
-- function kick(delayMs = 1000) {
-- if (kickTimer) return;
-- kickTimer = setTimeout(() => { kickTimer = null; pull().catch(() => {}); }, delayMs);
-- kickTimer.unref();
-- }
--
-- return {
-- pull, refresh, kick, importItem, cursor,
-- status: async () => ({ enabled: sources.enabled, cursor: await cursor(), last_error: lastError, last_run_at: iso(lastRunAt) }),
-- };
--}
--
--module.exports = { createImporter };
ddiff --git a/server/domain/indexing.js b/server/domain/indexing.js
iindex 72fb25e..b9e134d 100644
---- a/server/domain/indexing.js
++++ b/server/domain/indexing.js
@@@ -6,17 +6,20 @@
*
* emitOffer('deals.offer.created'|'deals.offer.updated'|'deals.offer.expired', offer, …)
* indexOffer(offer) deals.index_document.upserted|deleted for the canonical offer, or a
-- * tombstone for a merged one. The openvibe-publishing sequencer gives the
-- * same document the same revision, so re-sending an unchanged document emits
-- * nothing; a document that changed (text, status, freshness, indexability)
-- * gets revision + 1 and one event.
++ * tombstone for a merged one. The chassis (openvibe-publishing/publication)
++ * owns the mechanics — the sequencer revision, the search.index-document@1
++ * event, the outbox write and the IndexNow ping — on the caller's transaction
++ * handle, so the same document gets the same revision and re-sending an
++ * unchanged document emits nothing. A document that changed (text, status,
++ * freshness, indexability) gets revision + 1 and one event.
* indexProduct(id) the same for product pages.
*
-- * When an indexable page appears, changes or disappears it also tells IndexNow (openvibe-shared/indexnow,
-- * injected at boot): the page's path and the sitemap. Off without a key.
++ * When an indexable page appears, changes or disappears the chassis also tells IndexNow
++ * (openvibe-shared/indexnow, injected at boot): the page's path and the sitemap. Off without a key.
*/
const hooks = require('openvibe-publishing/index-hooks');
--const { iso } = require('./util');
++const { createPublication: createPublicationGlue } = require('openvibe-publishing/publication');
++const { iso } = require('./values');
function actorRef(actor) {
const s = String(actor || '');
@@@ -25,17 +28,10 @@ function actorRef(actor) {
return { type: 'service', id: 'deals' };
}
--function createIndexing({ store, publication, outbox, catalog, indexnow = null }) {
++function createIndexing({ config, store, publication, outbox, catalog, indexnow = null }) {
const { sequencer, db } = store;
--
-- /**
-- * IndexNow: a public, indexable page appeared, changed or went away — tell the engines its path
-- * (and the sitemap, which lists it). pingSoon never throws and is a no-op without a key, so it can
-- * never take a write down. A draft, private or noindex page is never announced.
-- */
-- function announce(path) {
-- if (indexnow && indexnow.enabled) indexnow.pingSoon([publication.abs(path), publication.abs('/sitemap.xml')]);
-- }
++ // The chassis owns the mechanics; Deals' gate, document builder and canonical paths stay here.
++ const glue = createPublicationGlue({ owner: 'deals', sequencer, outbox, baseUrl: config.baseUrl, indexnow, db, now: store.now });
function latestPayload(v) {
return v.latest ? { price: v.latest.price, currency: v.latest.currency, availability: v.latest.availability, observed_at: iso(v.latest.observed_at) } : null;
@@@ -62,11 +58,6 @@ function createIndexing({ store, publication, outbox, catalog, indexnow = null }
}, { traceparent });
}
-- async function emitIfNew(doc, before) {
-- if (before != null && doc.revision === before) return null;
-- return await outbox.emit(hooks.indexEvent({ document: doc, now: store.now() }));
-- }
--
function offerDocument(v) {
const r = v.root;
const provenance = v.sources.filter((s) => s.ref_service === 'sources' && !s.removed_at).slice(0, 40).map((s) => ({
@@@ -96,30 +87,16 @@ function createIndexing({ store, publication, outbox, catalog, indexnow = null }
async function indexOffer(offer) {
if (!offer) return null;
if (offer.merged_into) {
-- const before = await sequencer.current('deals', 'offer', offer.id);
-- const sent = await emitIfNew(await sequencer.stamp(db, hooks.tombstone({ owner: 'deals', type: 'offer', id: offer.id, revision: 0 })), before);
-- // A merge is a delete: the duplicate's page goes away. Announce it only if Search had it.
-- if (sent && before != null) announce(publication.offerPath(offer));
-- return sent;
++ // A merge is a delete: the duplicate's page goes away (announced only if Search had it).
++ return await glue.tombstone(db, { type: 'offer', id: offer.id, page: publication.offerPath(offer) });
}
const v = await publication.offerView(offer);
-- const before = await sequencer.current('deals', 'offer', v.root.id);
-- const sent = await emitIfNew(await sequencer.stamp(db, offerDocument(v)), before);
-- if (sent) {
-- // An indexable page appears or changes; a page Search already had that is now explicitly
-- // removed (a takedown) disappeared. An expired offer stays published-but-noindex, so it is
-- // not announced — a noindex page is never pinged.
-- const indexable = v.root.status === 'active' && v.decision.indexable;
-- const removed = v.root.status === 'disabled' && before != null;
-- if (indexable || removed) announce(publication.offerPath(v.root));
-- }
-- return sent;
++ return await glue.index(db, { document: offerDocument(v), page: publication.offerPath(v.root) });
}
async function indexProduct(product) {
if (!product) return null;
const pv = await publication.productView(product);
-- const before = await sequencer.current('deals', 'product', product.id);
const lines = pv.offers.filter((v) => v.root.status === 'active').slice(0, 20).map((v) => `${v.root.title}${v.latest && v.latest.price != null ? ` — ${v.latest.price} ${v.latest.currency || ''} as of ${iso(v.latest.observed_at)}` : ''}`);
const doc = hooks.buildIndexDocument({
owner: 'deals', type: 'product', id: product.id, revision: 0, state: 'published', visibility: 'public',
@@@ -132,11 +109,7 @@ function createIndexing({ store, publication, outbox, catalog, indexnow = null }
publishedAt: product.created_at,
updatedAt: pv.freshestObservedAt || product.updated_at,
});
-- const sent = await emitIfNew(await sequencer.stamp(db, doc), before);
-- // A product page appears or changes while indexable; one Search had that is now no longer
-- // indexable (its last fresh offer went) disappeared.
-- if (sent && (pv.decision.indexable || before != null)) announce(publication.productPath(product));
-- return sent;
++ return await glue.index(db, { document: doc, page: publication.productPath(product) });
}
/** After any change to an offer: its index document and its product's. */
ddiff --git a/server/domain/limits.js b/server/domain/limits.js
iindex a9729c2..f6828bc 100644
---- a/server/domain/limits.js
++++ b/server/domain/limits.js
@@@ -10,7 +10,7 @@
* IP addresses are never stored: ipHash() is an HMAC under DEALS_IP_HASH_SECRET, truncated.
*/
const crypto = require('crypto');
--const { ApiError } = require('./util');
++const { ApiError } = require('./values');
const fallbackKey = crypto.randomBytes(32);
ddiff --git a/server/domain/listings.js b/server/domain/listings.js
iindex bf3f8fd..7637a63 100644
---- a/server/domain/listings.js
++++ b/server/domain/listings.js
@@@ -8,7 +8,7 @@
* search active (or all non-disabled, with expired) offers whose title/description contain every word
* store a store's offers, expired ones included and marked
*/
--const { tokens } = require('./util');
++const { tokens } = require('./values');
function createListings({ store }) {
const { db } = store;
ddiff --git a/server/domain/offers.js b/server/domain/offers.js
iindex a161404..8441ee9 100644
---- a/server/domain/offers.js
++++ b/server/domain/offers.js
@@@ -17,7 +17,7 @@
const {
ApiError, newId, normalizeUrl, urlKey, domainOf, slugify, parseAmount, parseCurrency, parseEnum, parseInstant,
text, longText, CONDITIONS, AVAILABILITY,
--} = require('./util');
++} = require('./values');
const FUTURE_SKEW_MS = 5 * 60 * 1000;
ddiff --git a/server/domain/publication.js b/server/domain/publication.js
iindex 0aef193..e0715d3 100644
---- a/server/domain/publication.js
++++ b/server/domain/publication.js
@@@ -19,7 +19,7 @@
* drops the price, a missing availability is omitted (never an assumed InStock).
*/
const seo = require('openvibe-publishing/seo');
--const { iso } = require('./util');
++const { iso } = require('./values');
function createPublication({ config, store, reads, catalog }) {
const abs = (p) => seo.canonicalUrl(config.baseUrl, p);
ddiff --git a/server/domain/reads.js b/server/domain/reads.js
iindex a257223..f8f6057 100644
---- a/server/domain/reads.js
++++ b/server/domain/reads.js
@@@ -8,7 +8,7 @@
* votes — is read across the group, and unmerging (clearing the pointer) gives B back exactly what
* it had. A vote a person cast on both A and B counts once: their most recent row in the group.
*/
--const { ApiError } = require('./util');
++const { ApiError } = require('./values');
function createReads({ store }) {
const { db } = store;
ddiff --git a/server/domain/source-items.js b/server/domain/source-items.js
nnew file mode 100644
iindex 0000000..7be9acd
---- /dev/null
++++ b/server/domain/source-items.js
@@@ -0,0 +1,276 @@
++'use strict';
++
++/**
++ * Imports deals-category items from OpenVibe.Sources.
++ *
++ * The truth is Sources' change feed (the chassis' pullChanges: GET /api/v1/items in change order from
++ * the cursor in deals_ingest_cursor): the importer keeps its cursor there and pages through it.
++ * sources.item.* events delivered to /internal/events only wake the pull early (they carry no prices),
++ * so a lost or repeated event changes nothing.
++ *
++ * The chassis (openvibe-publishing/ingest) owns the Sources client, the change cursor and the pull
++ * loop (one transaction per page, one savepoint per item); this module keeps Deals' domain half: what
++ * a Sources item becomes in Deals' tables (the mapping below, its stated-value parsing and its
++ * observations).
++ *
++ * Item → offer mapping (values exactly as the source stated them; missing = null):
++ * kind offer one offer: fields.price + fields.currency, availability / condition (schema.org
++ * terms mapped to Deals' enums, unknown terms → null), valid_until → stated expiry,
++ * seller → store name
++ * kind product a product (resolved by gtin / mpn / sku / name) and one offer per stated offer
++ * kind article (RSS/Atom) one offer with the title and link; the price is NOT parsed out of the
++ * headline — it stays "not stated" until a structured source or a person states it
++ * other kinds skipped
++ *
++ * observed_at = the item's provenance.retrieved_at (when Sources saw it). An item Deals already
++ * imported produces a new observation only when its revision or its retrieved_at advanced, so
++ * replays are no-ops. Two items (or a submission and an item) with the same link attach to the same
++ * offer as separate sources. A removed item (takedown, licence) removes its source; an imported
++ * offer left with no source is disabled.
++ *
++ * Imported text is third-party: review_state = pending, so the page is noindex until a person
++ * reviews it.
++ */
++const { createChangeCursor, pullChanges } = require('openvibe-publishing/ingest');
++const {
++ normalizeUrl, urlKey, parseAmount, SCHEMA_AVAILABILITY, SCHEMA_CONDITION, iso,
++} = require('./values');
++
++const CURSOR = 'sources.deals.after';
++
++/** importItem's tag(s) → the chassis' three pull outcomes ('applied' | 'hold' | 'removed'). */
++function pullOutcomeFor(out) {
++ const tags = String(out).split(',');
++ if (tags.includes('removed') || tags.includes('removed:unknown')) return 'removed';
++ if (tags.every((t) => t.startsWith('skipped'))) return 'hold';
++ return 'applied';
++}
++
++function createImporter({ config, store, reads, catalog, offers, indexing, sources, log = console }) {
++ const { db } = store;
++ const cursor = createChangeCursor(db, { prefix: 'deals', now: store.now });
++ const q = {
++ sourceRow: db.prepare("SELECT * FROM deal_offer_sources WHERE kind = 'sources_item' AND ref_service = 'sources' AND ref_id = ? AND ref_part = ?"),
++ itemRows: db.prepare("SELECT * FROM deal_offer_sources WHERE kind = 'sources_item' AND ref_service = 'sources' AND ref_id = ? AND removed_at IS NULL"),
++ bumpSource: db.prepare('UPDATE deal_offer_sources SET ref_revision = ?, retrieved_at = ?, label = COALESCE(?, label), updated_at = ? WHERE id = ?'),
++ removeSource: db.prepare('UPDATE deal_offer_sources SET removed_at = ?, removed_reason = ?, updated_at = ? WHERE id = ?'),
++ liveSources: db.prepare('SELECT COUNT(*) AS n FROM deal_offer_sources WHERE offer_id = ? AND removed_at IS NULL'),
++ refreshDue: db.prepare(`SELECT s.ref_id, MAX(o.observed_at) AS last FROM deal_offer_sources s
++ JOIN deal_offers f ON f.id = s.offer_id
++ JOIN deal_price_observations o ON o.source_id = s.id
++ WHERE s.kind = 'sources_item' AND s.removed_at IS NULL AND f.status = 'active'
++ GROUP BY s.ref_id HAVING MAX(o.observed_at) < ? ORDER BY last LIMIT ?`),
++ };
++ let lastError = null;
++ let lastRunAt = null;
++ let running = null;
++ let kickTimer = null;
++
++ // The chassis client knows it is configured from the secret + URL; Deals' own switch (DEALS_IMPORT)
++ // is beside it.
++ const enabled = () => Boolean(sources.enabled && config.sources.enabled);
++ const cursorNow = async () => await cursor.get(CURSOR);
++
++ function amount(v) { try { return parseAmount(v, 'price'); } catch { return null; } }
++ function currencyOf(v) { return typeof v === 'string' && /^[A-Z]{3}$/.test(v) ? v : null; }
++
++ /** Stated fields → observation fields. A malformed value is null, never repaired or guessed. */
++ function obsFields(f = {}) {
++ const price = amount(f.price);
++ return {
++ price: price ? price.text : null, price_num: price ? price.num : null, currency: currencyOf(f.currency),
++ shipping: null, shipping_num: null, shipping_note: null,
++ condition: SCHEMA_CONDITION[f.condition] || null,
++ availability: SCHEMA_AVAILABILITY[f.availability] || null,
++ };
++ }
++
++ function candidates(item) {
++ const f = item.fields || {};
++ const stated = (t) => { const ms = t ? Date.parse(t) : NaN; return Number.isFinite(ms) ? ms : null; };
++ if (item.kind === 'offer') {
++ const url = normalizeUrl(f.url || item.canonical_url);
++ if (!url || !item.title) return [];
++ return [{ part: '', url, title: item.title, description: item.summary, obs: obsFields(f), expires_at: stated(f.valid_until), store_name: f.seller || null }];
++ }
++ if (item.kind === 'product') {
++ if (!item.title) return [];
++ const seen = new Set();
++ const out = [];
++ for (const o of (Array.isArray(f.offers) ? f.offers : []).slice(0, 20)) {
++ const url = normalizeUrl(o.url || item.canonical_url);
++ if (!url || seen.has(urlKey(url))) continue;
++ seen.add(urlKey(url));
++ out.push({
++ part: urlKey(url), url, title: o.seller ? `${item.title} — ${o.seller}` : item.title, description: item.summary,
++ obs: obsFields(o), expires_at: stated(o.valid_until), store_name: o.seller || null,
++ product: { name: item.title, brand: f.brand, gtin: f.gtin, mpn: f.mpn, sku: f.sku },
++ });
++ }
++ return out;
++ }
++ if (item.kind === 'article' || item.kind === 'record') {
++ const url = normalizeUrl(item.canonical_url);
++ if (!url || !item.title) return [];
++ return [{ part: '', url, title: item.title, description: item.summary, obs: obsFields({}), expires_at: null, store_name: null }];
++ }
++ return [];
++ }
++
++ async function removeItem(item) {
++ const reason = (item.removed && item.removed.reason) || 'removed at the source';
++ const rows = await q.itemRows.all(item.id);
++ for (const s of rows) await q.removeSource.run(store.now(), reason, store.now(), s.id);
++ const touched = [...new Set(rows.map((s) => s.offer_id))];
++ for (const id of touched) {
++ const offer = await reads.get(id);
++ if (offer.origin === 'import' && offer.status !== 'disabled' && (await q.liveSources.get(id)).n === 0) {
++ const now = store.now();
++ await db.prepare("UPDATE deal_offers SET status = 'disabled', disabled_at = ?, disabled_reason = ?, disabled_by = 'svc:deals', updated_at = ? WHERE id = ?")
++ .run(now, `source item removed: ${reason}`, now, id);
++ await offers.logAction('disable', { offerId: id, actor: 'svc:deals', reason: `source item removed: ${reason}`, before: { status: offer.status }, after: { status: 'disabled' } });
++ }
++ const after = await reads.get(id);
++ await indexing.emitOffer('deals.offer.updated', after, { actor: 'svc:deals', extra: { changed: ['sources'], source_removed: item.id } });
++ await indexing.reindex(after);
++ }
++ return touched.length ? 'removed' : 'removed:unknown';
++ }
++
++ /** One item, inside the caller's transaction (the chassis' per-item savepoint). */
++ async function importItem(item) {
++ if (!item || item.category !== 'deals') return 'skipped:category';
++ if (item.removed) return await removeItem(item);
++ const retrievedAt = Date.parse(item.provenance && item.provenance.retrieved_at);
++ if (!Number.isFinite(retrievedAt)) return 'skipped:no_observation_time';
++ const list = candidates(item);
++ if (!list.length) return `skipped:${item.kind}`;
++ const outcomes = [];
++ for (const c of list) {
++ const src = {
++ kind: 'sources_item', ref_service: 'sources', ref_type: 'item', ref_id: item.id, ref_part: c.part, ref_revision: item.revision,
++ source_key: item.source_key, url: item.canonical_url, label: item.title,
++ license_note: item.provenance.license_note || null, retrieved_at: retrievedAt,
++ };
++ const existing = await q.sourceRow.get(item.id, c.part);
++ if (existing) {
++ const advanced = item.revision > (existing.ref_revision || 0) || retrievedAt > (existing.retrieved_at || 0);
++ if (!advanced || existing.removed_at) { outcomes.push('unchanged'); continue; }
++ const origin = item.revision > (existing.ref_revision || 0) ? 'import' : 'import_refresh';
++ await q.bumpSource.run(Math.max(item.revision, existing.ref_revision || 0), Math.max(retrievedAt, existing.retrieved_at || 0), item.title, store.now(), existing.id);
++ const obs = await offers.recordObservation(existing.offer_id, existing.id, c.obs, { origin, observedAt: retrievedAt, sourceRevision: item.revision });
++ const offer = await reads.get(existing.offer_id);
++ const changed = ['observation'];
++ if (origin === 'import' && c.expires_at && offer.status === 'active' && offer.origin === 'import' && c.expires_at !== offer.expires_at) {
++ await db.prepare('UPDATE deal_offers SET expires_at = ?, updated_at = ? WHERE id = ?').run(c.expires_at, store.now(), offer.id);
++ changed.push('expires_at');
++ }
++ // Anything else the observation moved on the offer (price, product) is named too, so consumers never miss a field.
++ const after = await reads.get(offer.id);
++ for (const k of ['product_id', 'title', 'status']) if (after && offer && after[k] !== offer[k] && !changed.includes(k)) changed.push(k);
++ await indexing.emitOffer('deals.offer.updated', after, { actor: 'svc:deals', extra: { changed, observation_id: obs.id } });
++ await indexing.reindex(await reads.get(offer.id));
++ outcomes.push(origin === 'import' ? 'updated' : 'refreshed');
++ continue;
++ }
++ let productId = null;
++ if (c.product) {
++ try { productId = (await catalog.resolve(c.product, { source: 'import', actor: 'svc:deals' })).product.id; } catch { productId = null; }
++ }
++ const dup = await offers.findByUrl(c.url);
++ if (dup) {
++ const s = await offers.ensureSource(dup.id, src);
++ const obs = await offers.recordObservation(dup.id, s.id, c.obs, { origin: 'import', observedAt: retrievedAt, sourceRevision: item.revision });
++ if (productId && !dup.product_id) await db.prepare('UPDATE deal_offers SET product_id = ?, updated_at = ? WHERE id = ?').run(productId, store.now(), dup.id);
++ await indexing.emitOffer('deals.offer.updated', await reads.get(dup.id), { actor: 'svc:deals', extra: { changed: ['sources'], observation_id: obs.id } });
++ await indexing.reindex(await reads.get(dup.id));
++ outcomes.push('attached');
++ continue;
++ }
++ await offers.createImported({
++ url: c.url, title: String(c.title).slice(0, 200), description: c.description, product_id: productId,
++ store_name: c.store_name, expires_at: c.expires_at != null && c.expires_at > store.now() ? c.expires_at : null,
++ }, src, c.obs, { observedAt: retrievedAt, sourceRevision: item.revision });
++ outcomes.push('created');
++ }
++ return outcomes.join(',');
++ }
++
++ /**
++ * Pull the change feed from the cursor. The chassis reads pages, runs each item in its own
++ * savepoint and moves the cursor only past applied pages; the cursor is never reset to zero. This
++ * module tallies its own domain outcome tags and never throws (the cursor stays put on a failure).
++ */
++ async function pull() {
++ if (!enabled()) return { skipped: 'import off' };
++ if (running) return running;
++ running = (async () => {
++ const summary = { pages: 0, items: 0, outcomes: {} };
++ try {
++ const res = await pullChanges({
++ db, cursor, source: sources, name: CURSOR,
++ maxPages: config.sources.maxPages, pageSize: config.sources.pageSize,
++ apply: async (item) => {
++ const out = await importItem(item);
++ for (const o of String(out).split(',')) summary.outcomes[o] = (summary.outcomes[o] || 0) + 1;
++ return pullOutcomeFor(out);
++ },
++ onItem: (item, err, outcome) => {
++ summary.items++;
++ if (outcome === 'failed') {
++ summary.outcomes.failed = (summary.outcomes.failed || 0) + 1;
++ log.warn(`[Deals] import of ${item && item.id} failed: ${err && err.message ? err.message : 'error'}`);
++ }
++ },
++ });
++ summary.pages = res.pages;
++ lastError = null;
++ } catch (err) {
++ lastError = err.message;
++ log.warn('[Deals] Sources import failed (will retry):', err.message);
++ } finally {
++ lastRunAt = store.now();
++ running = null;
++ }
++ return summary;
++ })();
++ return running;
++ }
++
++ /**
++ * Re-confirm imported offers whose latest observation is getting old: when Sources has fetched
++ * the item again (retrieved_at advanced) the unchanged item is a new observation at that time.
++ * A source that has not been fetched successfully simply lets the offer go stale.
++ */
++ async function refresh() {
++ if (!enabled()) return { skipped: 'import off' };
++ const due = await q.refreshDue.all(store.now() - config.freshnessMs / 2, config.sources.refreshBatch);
++ const out = { checked: 0, outcomes: {} };
++ for (const { ref_id: id } of due) {
++ try {
++ const data = await sources.getItem(id);
++ out.checked++;
++ const o = await store.tx(async () => await importItem(data.item));
++ for (const x of String(o).split(',')) out.outcomes[x] = (out.outcomes[x] || 0) + 1;
++ } catch (err) {
++ if (err.status === 404) continue;
++ lastError = err.message;
++ break;
++ }
++ }
++ return out;
++ }
++
++ function kick(delayMs = 1000) {
++ if (kickTimer) return;
++ kickTimer = setTimeout(() => { kickTimer = null; pull().catch(() => {}); }, delayMs);
++ kickTimer.unref();
++ }
++
++ return {
++ pull, refresh, kick, importItem, cursor: cursorNow,
++ status: async () => ({ enabled: enabled(), cursor: await cursorNow(), last_error: lastError, last_run_at: iso(lastRunAt) }),
++ };
++}
++
++module.exports = { createImporter, CURSOR };
ddiff --git a/server/domain/util.js b/server/domain/util.js
ddeleted file mode 100644
iindex 9755161..0000000
---- a/server/domain/util.js
++++ /dev/null
@@@ -1,138 +0,0 @@
--'use strict';
--
--/**
-- * Small pure helpers: ids, slugs, URL normalisation, stated-value parsing. Nothing here guesses:
-- * a value that is missing or malformed is null (or a 422), never a default.
-- */
--const { ids } = require('openvibe-contracts');
--
--/** A refusal with a stable problem code (e.g. 404 'offer.not_found'). */
--class ApiError extends Error {
-- constructor(status, code, detail, extra) {
-- super(detail || code);
-- this.name = 'ApiError';
-- this.status = status;
-- this.code = code;
-- this.extra = extra || null;
-- }
--}
--
--const newId = (prefix) => `${prefix}_${ids.ulid()}`;
--
--const CONDITIONS = ['new', 'used', 'refurbished', 'damaged'];
--const AVAILABILITY = ['in_stock', 'out_of_stock', 'preorder', 'discontinued', 'limited', 'sold_out', 'online_only', 'in_store_only'];
--const AVAILABILITY_LABEL = {
-- in_stock: 'In stock', out_of_stock: 'Out of stock', preorder: 'Pre-order', discontinued: 'Discontinued',
-- limited: 'Limited availability', sold_out: 'Sold out', online_only: 'Online only', in_store_only: 'In store only',
--};
--/** schema.org terms (as OpenVibe.Sources passes them through) → our enums; anything else → null. */
--const SCHEMA_AVAILABILITY = {
-- InStock: 'in_stock', OutOfStock: 'out_of_stock', PreOrder: 'preorder', PreSale: 'preorder', Discontinued: 'discontinued',
-- LimitedAvailability: 'limited', SoldOut: 'sold_out', OnlineOnly: 'online_only', InStoreOnly: 'in_store_only',
--};
--const SCHEMA_CONDITION = { NewCondition: 'new', UsedCondition: 'used', RefurbishedCondition: 'refurbished', DamagedCondition: 'damaged' };
--
--// Tracking and affiliate parameters are not part of an offer's identity.
--const TRACKING = /^(utm_[a-z0-9_]+|fbclid|gclid|dclid|msclkid|mc_cid|mc_eid|yclid|_hsenc|_hsmi|ref|ref_|tag|aff|affid|affiliate|affiliate_id|irclickid|clickid)$/i;
--
--/** http(s) URL with the fragment and tracking parameters removed, host lower-cased, params sorted. */
--function normalizeUrl(input) {
-- if (input == null || input === '') return null;
-- let u;
-- try { u = new URL(String(input).trim()); } catch { return null; }
-- if (u.protocol !== 'http:' && u.protocol !== 'https:') return null;
-- if (u.username || u.password) return null;
-- u.hash = '';
-- u.hostname = u.hostname.toLowerCase();
-- const keep = [...u.searchParams.entries()].filter(([k]) => !TRACKING.test(k)).sort(([a], [b]) => (a < b ? -1 : a > b ? 1 : 0));
-- u.search = '';
-- for (const [k, v] of keep) u.searchParams.append(k, v);
-- let s = u.toString();
-- if (u.pathname.length > 1 && s.endsWith('/') && !u.search) s = s.slice(0, -1);
-- return s.length <= 2048 ? s : null;
--}
--
--/** The identity used to find duplicates: normalised URL without scheme and without "www.". */
--function urlKey(normalized) {
-- if (!normalized) return null;
-- return normalized.replace(/^https?:\/\//, '').replace(/^www\./, '');
--}
--
--function domainOf(normalized) {
-- try { return new URL(normalized).hostname.replace(/^www\./, ''); } catch { return null; }
--}
--
--function slugify(text, suffixFrom) {
-- const base = String(text || '').normalize('NFKD').replace(/[̀-ͯ]/g, '').toLowerCase()
-- .replace(/[^a-z0-9]+/g, '-').replace(/^-+|-+$/g, '').slice(0, 60).replace(/-+$/g, '') || 'deal';
-- const suffix = String(suffixFrom || '').replace(/^[a-z]+_/, '').slice(-6).toLowerCase();
-- return suffix ? `${base}-${suffix}` : base;
--}
--
--/**
-- * A decimal amount as stated: '12', '12.5', '1,299.99'. The digits stay as the person or source
-- * gave them (no rounding, no added decimals); only thousands commas and surrounding space go.
-- * → { text, num } | null (blank) ; throws ApiError 422 on anything else (negative, words, symbols).
-- */
--function parseAmount(v, field) {
-- if (v == null || v === '') return null;
-- if (typeof v === 'number') {
-- if (!Number.isFinite(v) || v < 0) throw new ApiError(422, 'request.invalid', `${field} must be a non-negative number`);
-- return { text: String(v), num: v };
-- }
-- const s = String(v).trim().replace(/,(?=\d{3}(\D|$))/g, '');
-- if (s === '') return null;
-- if (!/^\d{1,9}(\.\d{1,4})?$/.test(s)) throw new ApiError(422, 'request.invalid', `${field} must be a plain amount like 19.99 (no currency symbols)`);
-- return { text: s.replace(/^0+(?=\d)/, ''), num: Number(s) };
--}
--
--function parseCurrency(v, field = 'currency') {
-- if (v == null || v === '') return null;
-- const s = String(v).trim().toUpperCase();
-- if (!/^[A-Z]{3}$/.test(s)) throw new ApiError(422, 'request.invalid', `${field} must be an ISO 4217 code such as USD or EUR`);
-- return s;
--}
--
--function parseEnum(v, allowed, field) {
-- if (v == null || v === '') return null;
-- const s = String(v).trim().toLowerCase();
-- if (!allowed.includes(s)) throw new ApiError(422, 'request.invalid', `${field} must be one of ${allowed.join(', ')}`);
-- return s;
--}
--
--/** An instant stated by a person or source (ISO string or epoch ms) → ms; blank → null. */
--function parseInstant(v, field) {
-- if (v == null || v === '') return null;
-- const t = typeof v === 'number' ? v : Date.parse(String(v));
-- if (!Number.isFinite(t)) throw new ApiError(422, 'request.invalid', `${field} must be a date`);
-- return t;
--}
--
--function text(v, { field, min = 0, max = 1000, required = false } = {}) {
-- const s = v == null ? '' : String(v).replace(/\s+/g, ' ').trim();
-- if (!s) {
-- if (required) throw new ApiError(422, 'request.invalid', `${field} is required`);
-- return null;
-- }
-- if (s.length < min) throw new ApiError(422, 'request.invalid', `${field} must be at least ${min} characters`);
-- return s.slice(0, max);
--}
--
--/** Multi-line text (descriptions): keeps line breaks, trims, caps. */
--function longText(v, max = 4000) {
-- if (v == null) return null;
-- const s = String(v).replace(/\r\n?/g, '\n').replace(/[ \t]+\n/g, '\n').trim();
-- return s ? s.slice(0, max) : null;
--}
--
--/** Lower-case word tokens (letters and digits of any script). */
--function tokens(s) {
-- return String(s || '').toLowerCase().normalize('NFKC').split(/[^\p{L}\p{N}]+/u).filter((t) => t.length > 0);
--}
--
--const iso = (ms) => (ms == null ? null : new Date(ms).toISOString());
--
--module.exports = {
-- ApiError, newId, CONDITIONS, AVAILABILITY, AVAILABILITY_LABEL, SCHEMA_AVAILABILITY, SCHEMA_CONDITION,
-- normalizeUrl, urlKey, domainOf, slugify, parseAmount, parseCurrency, parseEnum, parseInstant, text, longText, tokens, iso,
--};
ddiff --git a/server/domain/values.js b/server/domain/values.js
nnew file mode 100644
iindex 0000000..678447f
---- /dev/null
++++ b/server/domain/values.js
@@@ -0,0 +1,87 @@
++'use strict';
++
++/**
++ * Deals' own values and refusals, plus the shared normalisers.
++ *
++ * The generic normalisers (hosts, slugs, amounts, currencies, instants, enum/text parsing, tokens,
++ * iso) come from the chassis (openvibe-publishing/ingest `normalize`), which ports them byte-for-byte
++ * from the five products — away is the local copy (server/domain/util.js, now unrequired). They throw
++ * openvibe-publishing's PublishingError; the wrappers below rethrow Deals' ApiError so the HTTP layer
++ * keeps answering the same 422 problems.
++ *
++ * What stays Deals' own: the refusal type, id minting, the condition/availability enums and the
++ * schema.org maps (policy values), and the URL identity helpers — the chassis `normalizeUrl` strips
++ * a narrower set of tracking parameters than Deals does, so Deals keeps its own (offer identity must
++ * not change).
++ */
++const { ids } = require('openvibe-contracts');
++const { normalize } = require('openvibe-publishing/ingest');
++
++/** A refusal with a stable problem code (e.g. 404 'offer.not_found'). */
++class ApiError extends Error {
++ constructor(status, code, detail, extra) {
++ super(detail || code);
++ this.name = 'ApiError';
++ this.status = status;
++ this.code = code;
++ this.extra = extra || null;
++ }
++}
++
++/** The chassis throws PublishingError; Deals' callers catch ApiError by class. */
++const asApiError = (fn) => (...args) => {
++ try { return fn(...args); } catch (err) {
++ if (err && err.name === 'PublishingError') throw new ApiError(err.status, err.code, err.message, err.extra);
++ throw err;
++ }
++};
++
++const newId = (prefix) => `${prefix}_${ids.ulid()}`;
++
++const CONDITIONS = ['new', 'used', 'refurbished', 'damaged'];
++const AVAILABILITY = ['in_stock', 'out_of_stock', 'preorder', 'discontinued', 'limited', 'sold_out', 'online_only', 'in_store_only'];
++const AVAILABILITY_LABEL = {
++ in_stock: 'In stock', out_of_stock: 'Out of stock', preorder: 'Pre-order', discontinued: 'Discontinued',
++ limited: 'Limited availability', sold_out: 'Sold out', online_only: 'Online only', in_store_only: 'In store only',
++};
++/** schema.org terms (as OpenVibe.Sources passes them through) → our enums; anything else → null. */
++const SCHEMA_AVAILABILITY = {
++ InStock: 'in_stock', OutOfStock: 'out_of_stock', PreOrder: 'preorder', PreSale: 'preorder', Discontinued: 'discontinued',
++ LimitedAvailability: 'limited', SoldOut: 'sold_out', OnlineOnly: 'online_only', InStoreOnly: 'in_store_only',
++};
++const SCHEMA_CONDITION = { NewCondition: 'new', UsedCondition: 'used', RefurbishedCondition: 'refurbished', DamagedCondition: 'damaged' };
++
++// Tracking and affiliate parameters are not part of an offer's identity.
++const TRACKING = /^(utm_[a-z0-9_]+|fbclid|gclid|dclid|msclkid|mc_cid|mc_eid|yclid|_hsenc|_hsmi|ref|ref_|tag|aff|affid|affiliate|affiliate_id|irclickid|clickid)$/i;
++
++/** http(s) URL with the fragment and tracking parameters removed, host lower-cased, params sorted. */
++function normalizeUrl(input) {
++ if (input == null || input === '') return null;
++ let u;
++ try { u = new URL(String(input).trim()); } catch { return null; }
++ if (u.protocol !== 'http:' && u.protocol !== 'https:') return null;
++ if (u.username || u.password) return null;
++ u.hash = '';
++ u.hostname = u.hostname.toLowerCase();
++ const keep = [...u.searchParams.entries()].filter(([k]) => !TRACKING.test(k)).sort(([a], [b]) => (a < b ? -1 : a > b ? 1 : 0));
++ u.search = '';
++ for (const [k, v] of keep) u.searchParams.append(k, v);
++ let s = u.toString();
++ if (u.pathname.length > 1 && s.endsWith('/') && !u.search) s = s.slice(0, -1);
++ return s.length <= 2048 ? s : null;
++}
++
++/** The identity used to find duplicates: normalised URL without scheme and without "www.". */
++function urlKey(normalized) {
++ if (!normalized) return null;
++ return normalized.replace(/^https?:\/\//, '').replace(/^www\./, '');
++}
++
++module.exports = {
++ ApiError, newId, CONDITIONS, AVAILABILITY, AVAILABILITY_LABEL, SCHEMA_AVAILABILITY, SCHEMA_CONDITION,
++ normalizeUrl, urlKey,
++ domainOf: normalize.domainOf, slugify: normalize.slugify,
++ parseAmount: asApiError(normalize.parseAmount), parseCurrency: asApiError(normalize.parseCurrency),
++ parseEnum: asApiError(normalize.parseEnum), parseInstant: asApiError(normalize.parseInstant),
++ text: asApiError(normalize.text), longText: normalize.longText, tokens: normalize.tokens, iso: normalize.iso,
++};
ddiff --git a/server/domain/votes.js b/server/domain/votes.js
iindex 4ed17fc..194fa70 100644
---- a/server/domain/votes.js
++++ b/server/domain/votes.js
@@@ -19,7 +19,7 @@
* within DEALS_RING_WINDOW_MIN minutes of each other each time
*/
const crypto = require('crypto');
--const { ApiError } = require('./util');
++const { ApiError } = require('./values');
const DAY = 24 * 3600 * 1000;
const CROCKFORD = '0123456789ABCDEFGHJKMNPQRSTVWXYZ';
ddiff --git a/server/domain/watches.js b/server/domain/watches.js
iindex 3a5d34b..91bc3aa 100644
---- a/server/domain/watches.js
++++ b/server/domain/watches.js
@@@ -21,7 +21,7 @@
* Deals never emails: deals.watch.matched is an internal event for OpenVibe.Network's notification
* consumer (future work).
*/
--const { ApiError, newId, text, tokens, parseAmount, parseCurrency, iso } = require('./util');
++const { ApiError, newId, text, tokens, parseAmount, parseCurrency, iso } = require('./values');
const KINDS = ['keyword', 'product', 'price_below', 'search'];
ddiff --git a/server/http/api.js b/server/http/api.js
iindex 01029e7..40c511a 100644
---- a/server/http/api.js
++++ b/server/http/api.js
@@@ -30,7 +30,7 @@
const express = require('express');
const { run, jsonBody, privateNoStore, ApiError } = require('./errors');
const { guard } = require('../auth/viewer');
--const { iso } = require('../domain/util');
++const { iso } = require('../domain/values');
function cors(origins) {
const allowed = new Set(origins);
ddiff --git a/server/http/errors.js b/server/http/errors.js
iindex 87b501f..b4d4e01 100644
---- a/server/http/errors.js
++++ b/server/http/errors.js
@@@ -6,7 +6,7 @@
*/
const express = require('express');
const contracts = require('openvibe-contracts');
--const { ApiError } = require('../domain/util');
++const { ApiError } = require('../domain/values');
/** Wrap a JSON handler: its return value is the body; errors become problems. */
function run(fn, status = 200) {
ddiff --git a/server/http/events.js b/server/http/events.js
nnew file mode 100644
iindex 0000000..757c678
---- /dev/null
++++ b/server/http/events.js
@@@ -0,0 +1,45 @@
++'use strict';
++
++/**
++ * Events consumer: POST /internal/events, the endpoint of Deals' OpenVibe.Events subscription to
++ * `sources.item.*` (filter: payload.category = deals). The signature (X-OpenVibe-Signature v2,
++ * ±300 s; several comma-separated secrets allow rotation) and the exactly-once inbox come from the
++ * chassis (openvibe-publishing/ingest.createEventConsumer): the receipt (consumer, event_id) and the
++ * change commit in one PostgreSQL transaction, so a redelivery changes nothing.
++ *
++ * Source item events carry no prices, so a delivery only wakes the importer, which reads the change
++ * feed from its cursor: a lost event delays an import, a repeated one changes nothing.
++ */
++const express = require('express');
++const { http } = require('openvibe-contracts');
++const { createEventConsumer } = require('openvibe-publishing/ingest');
++
++const CONSUMER = 'deals-sources';
++const TYPES = new Set(['sources.item.created', 'sources.item.updated', 'sources.item.removed']);
++
++function createEvents({ config, store, importer }) {
++ const router = express.Router();
++ const consumer = createEventConsumer({ db: store.db, secrets: config.events.webhookSecrets, consumer: CONSUMER, now: store.now });
++
++ router.post('/internal/events', express.raw({ type: () => true, limit: '256kb' }), async (req, res, next) => {
++ const raw = Buffer.isBuffer(req.body) ? req.body : Buffer.alloc(0);
++ let r;
++ try {
++ r = await consumer.apply(raw, req.headers, async (event) => {
++ if (!TYPES.has(event.event_type) || event.source !== 'sources') return 'ignored';
++ const category = event.payload && event.payload.category;
++ if (category !== 'deals') return 'ignored';
++ return 'import_scheduled';
++ });
++ } catch (err) { return next(err); }
++ if (r.status === 503) return http.sendProblem(res, 503, 'deals.webhook_disabled', { detail: 'DEALS_EVENTS_SECRET is not set', ctx: req.ov });
++ if (r.status === 401) return http.sendProblem(res, 401, 'deals.bad_signature', { detail: 'X-OpenVibe-Signature-V2 does not verify or is outside the replay window', ctx: req.ov });
++ if (r.status === 400) return http.sendProblem(res, 400, 'deals.bad_delivery', { detail: 'body must be { event: <envelope>, seq }', ctx: req.ov });
++ if (!r.duplicate && r.outcome === 'import_scheduled') importer.kick();
++ res.status(200).json({ event_id: r.event_id, duplicate: Boolean(r.duplicate), outcome: r.duplicate ? null : r.outcome });
++ });
++
++ return router;
++}
++
++module.exports = { createEvents, CONSUMER };
ddiff --git a/server/http/internal.js b/server/http/internal.js
ddeleted file mode 100644
iindex 75ab80c..0000000
---- a/server/http/internal.js
++++ /dev/null
@@@ -1,51 +0,0 @@
--'use strict';
--
--/**
-- * Events consumer: POST /internal/events, the endpoint of Deals' OpenVibe.Events subscription to
-- * `sources.item.*` (filter: payload.category = deals). Signed with X-OpenVibe-Signature
-- * (HMAC-SHA256 of the raw body under DEALS_EVENTS_SECRET; several comma-separated secrets allow
-- * rotation), recorded exactly once per event_id in the inbox (idempotency_receipts).
-- *
-- * Source item events carry no prices, so a delivery only wakes the importer, which reads the change
-- * feed from its cursor: a lost event delays an import, a repeated one changes nothing.
-- */
--const express = require('express');
--const { http } = require('openvibe-contracts');
--const { verifyDeliveryV2, createPgInbox } = require('openvibe-sdk/events');
--
--const CONSUMER = 'deals-sources';
--const TYPES = new Set(['sources.item.created', 'sources.item.updated', 'sources.item.removed']);
--
--function createInternalRoutes({ config, store, importer }) {
-- const router = express.Router();
-- const inbox = createPgInbox(store.db, { now: store.now }); // idempotency_receipts: migrations/0001_initial.sql
--
-- router.post('/internal/events', express.raw({ type: () => true, limit: '256kb' }), async (req, res, next) => {
-- const secrets = config.events.webhookSecrets;
-- if (!secrets.length) return http.sendProblem(res, 503, 'deals.webhook_disabled', { detail: 'DEALS_EVENTS_SECRET is not set', ctx: req.ov });
-- const raw = Buffer.isBuffer(req.body) ? req.body : Buffer.alloc(0);
-- // v2 only: signature over "<t>.<raw body>" and t within ±300 s (a replayed or v1-only delivery fails).
-- if (!secrets.some((s) => verifyDeliveryV2(raw, req.headers, s))) return http.sendProblem(res, 401, 'deals.bad_signature', { detail: 'X-OpenVibe-Signature-V2 does not verify or is outside the replay window', ctx: req.ov });
-- let body;
-- try { body = JSON.parse(raw.toString('utf8')); } catch { body = null; }
-- const event = body && body.event;
-- if (!event || typeof event.event_id !== 'string' || !/^evt_[0-9A-HJKMNP-TV-Z]{26}$/.test(event.event_id)) {
-- return http.sendProblem(res, 400, 'deals.bad_delivery', { detail: 'body must be { event: <envelope>, seq }', ctx: req.ov });
-- }
-- let r;
-- try {
-- r = await inbox.once(CONSUMER, event.event_id, async () => {
-- if (!TYPES.has(event.event_type) || event.source !== 'sources') return 'ignored';
-- const category = event.payload && event.payload.category;
-- if (category !== 'deals') return 'ignored';
-- return 'import_scheduled';
-- });
-- } catch (err) { return next(err); }
-- if (!r.duplicate && r.result === 'import_scheduled') importer.kick();
-- res.status(200).json({ event_id: event.event_id, duplicate: Boolean(r.duplicate), outcome: r.duplicate ? null : r.result });
-- });
--
-- return router;
--}
--
--module.exports = { createInternalRoutes, CONSUMER };
ddiff --git a/server/http/pages.js b/server/http/pages.js
iindex 683ef45..f311664 100644
---- a/server/http/pages.js
++++ b/server/http/pages.js
@@@ -28,7 +28,7 @@ const ssr = require('openvibe-publishing/ssr');
const { renderPage } = require('../render/layout');
const views = require('../render/views');
const { csrfToken, checkCsrf } = require('../auth/forms');
--const { ApiError, iso } = require('../domain/util');
++const { ApiError, iso } = require('../domain/values');
const PER_PAGE = 25;
ddiff --git a/server/render/views.js b/server/render/views.js
iindex 44dfdc0..61651c3 100644
---- a/server/render/views.js
++++ b/server/render/views.js
@@@ -8,7 +8,7 @@
* an unknown one says "not stated". Nothing here calls a price "current".
*/
const { html, raw, renderMarkdown, paginationHtml, breadcrumbsHtml } = require('openvibe-publishing/ssr');
--const { AVAILABILITY_LABEL, CONDITIONS, AVAILABILITY } = require('../domain/util');
++const { AVAILABILITY_LABEL, CONDITIONS, AVAILABILITY } = require('../domain/values');
const pad = (n) => String(n).padStart(2, '0');
function when(ms) {
ddiff --git a/test/chassis.test.js b/test/chassis.test.js
nnew file mode 100644
iindex 0000000..5d21966
---- /dev/null
++++ b/test/chassis.test.js
@@@ -0,0 +1,96 @@
++'use strict';
++/**
++ * T9 J4: Deals runs on the openvibe-publishing 1.1.0 chassis. This proves the conversion:
++ * - the local copies the brief's §7 lists are gone (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, check, done } = require('./helpers/boot');
++const { feedDeals, reviseDeals, 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 Deals (plus the Sources client, which the chassis' ingest entry
++// point replaces): the chassis (openvibe-publishing/ingest and /publication) owns their code now.
++// The harness denies rm, so the files may remain on disk; they must not be required by any live
++// server file.
++const DELETED = ['server/domain/importer.js', 'server/domain/util.js', 'server/http/internal.js', 'server/clients/sources.js']
++ .map((f) => path.join(ROOT, f));
++
++function jsFiles(dir, out = []) {
++ for (const e of fs.readdirSync(dir, { withFileTypes: true })) {
++ const p = path.join(dir, e.name);
++ if (e.isDirectory()) jsFiles(p, out);
++ else if (e.name.endsWith('.js')) out.push(p);
++ }
++ return out;
++}
++function relativeRequires(file) {
++ const src = fs.readFileSync(file, 'utf8');
++ return [...src.matchAll(/require\(\s*['"]([^'"]+)['"]\s*\)/g)].map((m) => m[1]).filter((s) => s.startsWith('.'));
++}
++
++(async () => {
++ await check('the local chassis copies are gone: no live server file requires them', () => {
++ 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.
++ assert.match(fs.readFileSync(path.join(ROOT, 'server', 'app.js'), 'utf8'), /openvibe-publishing\/ingest/);
++ assert.match(fs.readFileSync(path.join(ROOT, 'server', 'app.js'), 'utf8'), /\.\/domain\/source-items/);
++ assert.match(fs.readFileSync(path.join(ROOT, 'server', 'domain', 'indexing.js'), 'utf8'), /openvibe-publishing\/publication/);
++ });
++
++ const t = await boot();
++ feedDeals(t);
++ await t.ctx.importer.pull();
++ reviseDeals(t);
++ await t.ctx.importer.pull();
++
++ await check('a dry-run ingest from the captured Sources fixture produces the same domain rows as before', async () => {
++ assert.deepStrictEqual(await project(t), GOLDEN);
++ });
++
++ await check('the Search document validates against search.index-document@1, with its provenance and visibility', async () => {
++ const espresso = await t.ctx.store.db.prepare('SELECT * FROM deal_offers WHERE url = ?').get('https://coffee.example/espresso');
++ const r = await t.get(`/mod/offers/${espresso.slug}/review`, { as: t.mod, form: {} });
++ assert.strictEqual(r.status, 303, r.text.slice(0, 200));
++ const doc = (await t.events('deals.index_document.upserted')).filter((e) => e.payload.id === espresso.id).pop().payload;
++ const v = contracts.validate('search.index-document@1', doc);
++ assert.ok(v.valid, JSON.stringify(v.errors));
++ assert.strictEqual(doc.visibility, 'public');
++ assert.strictEqual(doc.indexability.decision, 'index');
++ assert.deepStrictEqual(doc.provenance.filter((p) => p.service === 'sources').map((p) => p.id), ['itm_01K5GOLDEN00000000000A1']);
++ });
++
++ await check('the outbox row and the state change share the caller\'s transaction (a rollback drops both)', async () => {
++ const offer = await t.ctx.store.db.prepare("SELECT * FROM deal_offers WHERE url = 'https://coffee.example/espresso'").get();
++ const row = async () => await t.ctx.store.db.prepare('SELECT title FROM deal_offers WHERE id = ?').get(offer.id);
++ const outboxCount = async () => (await t.ctx.store.db.prepare('SELECT COUNT(*) AS n FROM event_outbox').get()).n;
++ const before = await outboxCount();
++ await assert.rejects(t.ctx.store.db.tx(async () => {
++ await t.ctx.store.db.prepare('UPDATE deal_offers SET title = ? WHERE id = ?').run('Rolled back title', offer.id);
++ await t.ctx.indexing.indexOffer(await t.ctx.reads.get(offer.id));
++ throw new Error('rollback: the outbox row and the state change must not survive together');
++ }));
++ assert.strictEqual(await outboxCount(), before, 'no outbox row survives the rollback');
++ assert.strictEqual((await row()).title, offer.title, 'the state change rolled back with it');
++ });
++
++ await t.close();
++ done();
++})().catch((err) => { console.error(err); process.exit(1); });
ddiff --git a/test/fixtures/ingest-golden.json b/test/fixtures/ingest-golden.json
nnew file mode 100644
iindex 0000000..07f63aa
---- /dev/null
++++ b/test/fixtures/ingest-golden.json
@@@ -0,0 +1,178 @@
++{
++ "offers": [
++ {
++ "url": "https://coffee.example/espresso",
++ "title": "Espresso machine",
++ "description": null,
++ "status": "active",
++ "origin": "import",
++ "review_state": "pending",
++ "text_origin": "imported",
++ "category": null,
++ "expires_at": 1790812800000,
++ "disabled_reason": null,
++ "product": null,
++ "observations": [
++ {
++ "price": "249.00",
++ "currency": "EUR",
++ "shipping": null,
++ "condition": "refurbished",
++ "availability": "in_stock",
++ "origin": "import",
++ "observed_at": 1790074800000,
++ "source_revision": 1
++ },
++ {
++ "price": "229.00",
++ "currency": "EUR",
++ "shipping": null,
++ "condition": null,
++ "availability": "limited",
++ "origin": "import",
++ "observed_at": 1790085600000,
++ "source_revision": 2
++ }
++ ],
++ "sources": [
++ {
++ "kind": "sources_item",
++ "ref_id": "itm_01K5GOLDEN00000000000A1",
++ "ref_part": "",
++ "ref_revision": 2,
++ "source_key": "shop-jsonld",
++ "url": "https://coffee.example/espresso",
++ "label": "Espresso machine",
++ "retrieved_at": 1790085600000,
++ "removed_at": null,
++ "removed_reason": null
++ }
++ ]
++ },
++ {
++ "url": "https://kettles.example/acme-2000",
++ "title": "Acme Kettle 2000 — Kettles",
++ "description": null,
++ "status": "active",
++ "origin": "import",
++ "review_state": "pending",
++ "text_origin": "imported",
++ "category": null,
++ "expires_at": null,
++ "disabled_reason": null,
++ "product": {
++ "name": "Acme Kettle 2000",
++ "brand": "Acme",
++ "category": null
++ },
++ "observations": [
++ {
++ "price": "39.99",
++ "currency": "USD",
++ "shipping": null,
++ "condition": null,
++ "availability": "in_stock",
++ "origin": "import",
++ "observed_at": 1790078400000,
++ "source_revision": 1
++ }
++ ],
++ "sources": [
++ {
++ "kind": "sources_item",
++ "ref_id": "itm_01K5GOLDEN00000000000A2",
++ "ref_part": "kettles.example/acme-2000",
++ "ref_revision": 1,
++ "source_key": "shop-jsonld",
++ "url": "https://kettles.example/acme-2000",
++ "label": "Acme Kettle 2000",
++ "retrieved_at": 1790078400000,
++ "removed_at": null,
++ "removed_reason": null
++ }
++ ]
++ },
++ {
++ "url": "https://market.example/acme-2000",
++ "title": "Acme Kettle 2000 — Market",
++ "description": null,
++ "status": "active",
++ "origin": "import",
++ "review_state": "pending",
++ "text_origin": "imported",
++ "category": null,
++ "expires_at": null,
++ "disabled_reason": null,
++ "product": {
++ "name": "Acme Kettle 2000",
++ "brand": "Acme",
++ "category": null
++ },
++ "observations": [
++ {
++ "price": null,
++ "currency": null,
++ "shipping": null,
++ "condition": null,
++ "availability": "out_of_stock",
++ "origin": "import",
++ "observed_at": 1790078400000,
++ "source_revision": 1
++ }
++ ],
++ "sources": [
++ {
++ "kind": "sources_item",
++ "ref_id": "itm_01K5GOLDEN00000000000A2",
++ "ref_part": "market.example/acme-2000",
++ "ref_revision": 1,
++ "source_key": "shop-jsonld",
++ "url": "https://kettles.example/acme-2000",
++ "label": "Acme Kettle 2000",
++ "retrieved_at": 1790078400000,
++ "removed_at": null,
++ "removed_reason": null
++ }
++ ]
++ },
++ {
++ "url": "https://news.example/garden",
++ "title": "Weekend sale on garden tools",
++ "description": null,
++ "status": "disabled",
++ "origin": "import",
++ "review_state": "pending",
++ "text_origin": "imported",
++ "category": null,
++ "expires_at": null,
++ "disabled_reason": "source item removed: licence withdrawn",
++ "product": null,
++ "observations": [
++ {
++ "price": null,
++ "currency": null,
++ "shipping": null,
++ "condition": null,
++ "availability": null,
++ "origin": "import",
++ "observed_at": 1790078400000,
++ "source_revision": 1
++ }
++ ],
++ "sources": [
++ {
++ "kind": "sources_item",
++ "ref_id": "itm_01K5GOLDEN00000000000A3",
++ "ref_part": "",
++ "ref_revision": 1,
++ "source_key": "shop-jsonld",
++ "url": "https://news.example/garden",
++ "label": "Weekend sale on garden tools",
++ "retrieved_at": 1790078400000,
++ "removed_at": 1790085600000,
++ "removed_reason": "licence withdrawn"
++ }
++ ]
++ }
++ ]
++}
ddiff --git a/test/helpers/golden.js b/test/helpers/golden.js
nnew file mode 100644
iindex 0000000..f00dd8e
---- /dev/null
++++ b/test/helpers/golden.js
@@@ -0,0 +1,77 @@
++'use strict';
++/**
++ * The ingest golden fixture (T9 J4): a deterministic OpenVibe.Sources deals feed and the canonical
++ * projection of the domain rows it produces (deal_offers + deal_price_observations +
++ * deal_offer_sources, and the product a product item resolves to). The same fixture and the same
++ * 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 (ids, slugs, timestamps) are excluded or mapped to stable source keys, so only domain
++ * meaning is compared.
++ */
++const { sourceItem } = require('./mocks');
++
++const HOUR = 3600 * 1000;
++
++/** Add the fixture's items to the Sources mock (same order and fields as the capture run). → reports */
++function feedDeals(t) {
++ const T0 = t.clock.now();
++ t.sources.put(sourceItem({
++ id: 'itm_01K5GOLDEN00000000000A1', title: 'Espresso machine', url: 'https://coffee.example/espresso', retrievedAt: T0 - HOUR,
++ fields: { price: '249.00', currency: 'EUR', availability: 'InStock', condition: 'RefurbishedCondition', valid_until: '2026-10-01', seller: 'Coffee Shop' },
++ }));
++ t.sources.put(sourceItem({
++ id: 'itm_01K5GOLDEN00000000000A2', kind: 'product', title: 'Acme Kettle 2000', url: 'https://kettles.example/acme-2000', retrievedAt: T0,
++ fields: { brand: 'Acme', gtin: '4006381333931', offers: [
++ { price: '39.99', currency: 'USD', availability: 'InStock', url: 'https://kettles.example/acme-2000?utm_source=feed', seller: 'Kettles' },
++ { price: null, currency: null, availability: 'OutOfStock', url: 'https://market.example/acme-2000', seller: 'Market' },
++ ] },
++ }));
++ t.sources.put(sourceItem({
++ id: 'itm_01K5GOLDEN00000000000A3', kind: 'article', title: 'Weekend sale on garden tools', url: 'https://news.example/garden', retrievedAt: T0,
++ }));
++ return { espresso: 'itm_01K5GOLDEN00000000000A1', product: 'itm_01K5GOLDEN00000000000A2', article: 'itm_01K5GOLDEN00000000000A3' };
++}
++
++/** A revision upstream (new price) and a removal, applied by a second pull. */
++function reviseDeals(t) {
++ const T1 = t.clock.now() + 2 * HOUR;
++ t.clock.set(T1);
++ t.sources.put(sourceItem({
++ id: 'itm_01K5GOLDEN00000000000A1', title: 'Espresso machine', url: 'https://coffee.example/espresso', retrievedAt: T1, revision: 2,
++ fields: { price: '229.00', currency: 'EUR', availability: 'LimitedAvailability', seller: 'Coffee Shop' },
++ }));
++ const garden = t.sources.items.find((i) => i.id === 'itm_01K5GOLDEN00000000000A3');
++ t.sources.put({ ...garden, removed: { at: new Date(T1).toISOString(), reason: 'licence withdrawn' } });
++}
++
++const parse = (s, d) => { try { return s ? JSON.parse(s) : d; } catch { return d; } };
++
++/** The canonical projection of the domain rows. Deterministic in everything but ids/time. → { offers } */
++async function project(t) {
++ const db = t.ctx.store.db;
++ const offers = await db.prepare('SELECT * FROM deal_offers ORDER BY url').all();
++ const out = [];
++ for (const o of offers) {
++ const product = o.product_id ? await db.prepare('SELECT * FROM deal_products WHERE id = ?').get(o.product_id) : null;
++ const obs = await db.prepare('SELECT * FROM deal_price_observations WHERE offer_id = ? ORDER BY observed_at, seq').all(o.id);
++ const srcs = await db.prepare('SELECT * FROM deal_offer_sources WHERE offer_id = ? ORDER BY ref_id, ref_part, id').all(o.id);
++ out.push({
++ url: o.url, title: o.title, description: o.description, status: o.status, origin: o.origin,
++ review_state: o.review_state, text_origin: o.text_origin, category: o.category,
++ expires_at: o.expires_at, disabled_reason: o.disabled_reason || null,
++ product: product ? { name: product.name, brand: product.brand, category: product.category } : null,
++ observations: obs.map((x) => ({
++ price: x.price, currency: x.currency, shipping: x.shipping, condition: x.condition,
++ availability: x.availability, origin: x.origin, observed_at: x.observed_at, source_revision: x.source_revision,
++ })),
++ sources: srcs.map((s) => ({
++ kind: s.kind, ref_id: s.ref_id, ref_part: s.ref_part, ref_revision: s.ref_revision,
++ source_key: s.source_key, url: s.url, label: s.label, retrieved_at: s.retrieved_at,
++ removed_at: s.removed_at, removed_reason: s.removed_reason,
++ })),
++ });
++ }
++ return { offers: out };
++}
++
++module.exports = { feedDeals, reviseDeals, project, parse };
ddiff --git a/test/import.test.js b/test/import.test.js
iindex 4982cb7..1ac8105 100644
---- a/test/import.test.js
++++ b/test/import.test.js
@@@ -70,7 +70,7 @@ const { signDelivery, signDeliveryHeaders } = require('openvibe-sdk/events');
await check('replaying the feed changes nothing (same revision, same retrieval)', async () => {
const count = (await t.ctx.store.db.prepare('SELECT COUNT(*) AS n FROM deal_price_observations').get()).n;
const events = (await t.events()).length;
-- await t.ctx.store.db.prepare("UPDATE import_state SET value = '0'").run();
++ await t.ctx.store.db.prepare("UPDATE deals_ingest_cursor SET cursor = 0 WHERE name = 'sources.deals.after'").run();
const s = await t.ctx.importer.pull();
assert.deepStrictEqual(s.outcomes, { unchanged: 4 });
assert.strictEqual((await t.ctx.store.db.prepare('SELECT COUNT(*) AS n FROM deal_price_observations').get()).n, count);
ddiff --git a/test/watches.test.js b/test/watches.test.js
iindex 6448461..0d43feb 100644
---- a/test/watches.test.js
++++ b/test/watches.test.js
@@@ -88,7 +88,7 @@ const { sourceItem } = require('./helpers/mocks');
await t.ctx.importer.pull();
assert.strictEqual((await matched(w.price)).length, before + 1);
// Replay the whole feed from the start: same items, same revision, same retrieval → nothing.
-- await t.ctx.store.db.prepare("UPDATE import_state SET value = '0'").run();
++ await t.ctx.store.db.prepare("UPDATE deals_ingest_cursor SET cursor = 0 WHERE name = 'sources.deals.after'").run();
await t.ctx.importer.pull();
assert.strictEqual((await matched(w.price)).length, before + 1);
const obsCount = (await t.ctx.store.db.prepare("SELECT COUNT(*) AS n FROM deal_price_observations o JOIN deal_offers f ON f.id = o.offer_id WHERE f.url = 'https://audio.example/wh'").get()).n;