openvibers/openvibe.shoes Commits

Shoe models in OpenVibe.Search (search-feed, contracts 0.273.0) (#39)

OP OpenVibers committed 11 days ago · 9e543f0a5f · 9 files, +232 −9

Changes

ddiff --git a/README.md b/README.md
iindex 0b5a8b6..a1d2577 100644
---- a/README.md
++++ b/README.md
@@@ -159,7 +159,8 @@ section says so plainly when no store has cut a price yet.
  - No affiliate links, no sponsored order, no Marketplace scraping.
  
  <!-- versions:start -->
--- openvibe-contracts: v0.265.0
++- openvibe-contracts: v0.273.0
++- openvibe-publishing: v1.4.1
  - openvibe-sdk: v0.81.0
  - openvibe-shared: v4.103.0
  <!-- versions:end -->
ddiff --git a/STATUS.json b/STATUS.json
iindex 8eda1c2..f3a6bf7 100644
---- a/STATUS.json
++++ b/STATUS.json
@@@ -7,11 +7,12 @@
    "runtime": "node 22, express 4, PostgreSQL 18 (openvibe-sdk/db, async) + Valkey 9",
    "port": "5180",
    "domain": "open.shoes",
--  "contracts": "openvibe-contracts v0.265.0",
++  "contracts": "openvibe-contracts v0.273.0",
    "sdk": "openvibe-sdk v0.81.0",
    "shared": "openvibe-shared v4.103.0",
    "packages": [
--    "openvibe-contracts v0.265.0",
++    "openvibe-contracts v0.273.0",
++    "openvibe-publishing v1.4.1",
      "openvibe-sdk v0.81.0",
      "openvibe-shared v4.103.0"
    ],
ddiff --git a/migrations/0002_search_index.sql b/migrations/0002_search_index.sql
nnew file mode 100644
iindex 0000000..bf71f3c
---- /dev/null
++++ b/migrations/0002_search_index.sql
@@@ -0,0 +1,31 @@
++-- phase: expand
++-- OpenVibe.Shoes: shoe model pages in OpenVibe.Search (server/search-index.js, openvibe-publishing/search-feed).
++-- event_outbox holds the shoes.index_document.* events until the relay delivers them to OpenVibe.Events (written in
++-- the same transaction as the revision that produced them); shoes_index_revisions is the index sequencer, the last
++-- revision and content hash Search was sent for each page, so an unchanged page is never sent twice.
++
++CREATE TABLE IF NOT EXISTS event_outbox (
++    id              bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
++    event_id        text NOT NULL UNIQUE,
++    envelope        jsonb NOT NULL,
++    traceparent     text,
++    created_at      bigint NOT NULL,
++    attempts        integer NOT NULL DEFAULT 0,
++    next_attempt_at bigint NOT NULL DEFAULT 0,
++    sent_at         bigint,
++    seq             bigint,
++    rejected_at     bigint,
++    last_error      text
++);
++CREATE INDEX IF NOT EXISTS event_outbox_due ON event_outbox (next_attempt_at, id) WHERE sent_at IS NULL AND rejected_at IS NULL;
++CREATE INDEX IF NOT EXISTS event_outbox_sent ON event_outbox (sent_at) WHERE sent_at IS NOT NULL;
++
++CREATE TABLE IF NOT EXISTS shoes_index_revisions (
++    owner      text COLLATE "C" NOT NULL,
++    type       text COLLATE "C" NOT NULL,
++    id         text COLLATE "C" NOT NULL,
++    revision   integer NOT NULL,
++    hash       text NOT NULL,
++    updated_at bigint NOT NULL,
++    PRIMARY KEY (owner, type, id)
++);
ddiff --git a/package-lock.json b/package-lock.json
iindex d7ba652..2f7f1bf 100644
---- a/package-lock.json
++++ b/package-lock.json
@@@ -15,7 +15,8 @@
          "express-rate-limit": "^7.5.0",
          "helmet": "^8.1.0",
          "iovalkey": "^0.4.0",
--        "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.265.0",
++        "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.273.0",
++        "openvibe-publishing": "https://codeload.github.com/OpenVibers/OpenVibe.Publishing/tar.gz/refs/tags/v1.4.1",
          "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.81.0",
          "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.103.0",
          "pg": "^8.23.1"
@@@ -1150,9 +1151,9 @@
        }
      },
      "node_modules/openvibe-contracts": {
--      "version": "0.265.0",
--      "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.265.0",
--      "integrity": "sha512-0KnKj1Rig6acZcv4U2uwaM2msarGUY36pNqFbqmKdLeOPPozf7haEDTam27A/WzSU9ii2yEZ6SnuusPPQoGZaA==",
++      "version": "0.273.0",
++      "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.273.0",
++      "integrity": "sha512-NgcTWnKG1UwHZOLYlJLo8usPWxeld7Cnh2doFDfjUvL6duKqru7aO9hKjM0fYFROWvSHKaURLYhQLQOpJEhdog==",
        "license": "AGPL-3.0-only",
        "dependencies": {
          "ajv": "^8.17.1",
@@@ -1165,6 +1166,24 @@
          "node": ">=20"
        }
      },
++    "node_modules/openvibe-publishing": {
++      "version": "1.4.1",
++      "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Publishing/tar.gz/refs/tags/v1.4.1",
++      "integrity": "sha512-YjrcxCnXbpDQwyYpgsRcDj0+ckqZBm/ErEmQjvf0mCFMImwErG+lzZKXnXWb6eWOPxXEVj9YYuOgZiWVL41Vfg==",
++      "license": "MIT",
++      "engines": {
++        "node": ">=20"
++      },
++      "peerDependencies": {
++        "openvibe-sdk": ">=0.15.0",
++        "openvibe-shared": ">=1.5.0"
++      },
++      "peerDependenciesMeta": {
++        "openvibe-shared": {
++          "optional": true
++        }
++      }
++    },
      "node_modules/openvibe-sdk": {
        "version": "0.81.0",
        "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.81.0",
ddiff --git a/package.json b/package.json
iindex 88e70ae..9a15d96 100644
---- a/package.json
++++ b/package.json
@@@ -23,7 +23,8 @@
      "express-rate-limit": "^7.5.0",
      "helmet": "^8.1.0",
      "iovalkey": "^0.4.0",
--    "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.265.0",
++    "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.273.0",
++    "openvibe-publishing": "https://codeload.github.com/OpenVibers/OpenVibe.Publishing/tar.gz/refs/tags/v1.4.1",
      "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.81.0",
      "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.103.0",
      "pg": "^8.23.1"
ddiff --git a/server/app.js b/server/app.js
iindex 79024a0..eeedabc 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -35,6 +35,7 @@ const { assetVersion, setRelease } = require('./render/layout');
  const { createCatalogue } = require('./catalogue');
  const { createUpstreams } = require('./upstreams');
  const { createJobs } = require('./jobs');
++const { createSearchIndex } = require('./search-index');
  
  const accountDataLib = require('./identity/account-data');
  const { createNetworkSender } = require('openvibe-sdk/account-data');
@@@ -60,6 +61,8 @@ async function createApp(opts = {}) {
      const catalogue = createCatalogue(s, { staleHours: config.sources.staleHours });
      const upstreams = createUpstreams({ config, fetchImpl, log, urlCheck: opts.urlCheck });
      const jobs = createJobs({ config, catalogue, upstreams, log });
++    // The model pages in OpenVibe.Search (started by server/index.js, never under SHOES_DRILL).
++    const searchIndex = createSearchIndex({ config, s, log });
  
      // Account export and deletion (ADR-033). The sender posts to Network's internal routes with this service's own
      // client-credentials token; a test injects a stand-in through opts.accountSend.
@@@ -68,7 +71,7 @@ async function createApp(opts = {}) {
          ? createNetworkSender({ networkInternalUrl: config.networkInternalUrl, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, fetch: fetchImpl })
          : async () => { throw new Error('OV_OAUTH_CLIENT_SECRET is not set: Shoes cannot answer account events'); });
  
--    const ctx = { config, s, keys, sso, principal, catalogue, upstreams, jobs, accountData, accountSend, log };
++    const ctx = { config, s, keys, sso, principal, catalogue, upstreams, jobs, searchIndex, accountData, accountSend, log };
  
      const app = express();
      app.disable('x-powered-by');
ddiff --git a/server/index.js b/server/index.js
iindex 083ed53..aa8015e 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -20,6 +20,7 @@ function createLifecycle({ server, ctx, exit, signals, timers = [], extra = [] }
              () => { for (const t of timers) clearInterval(t); },
              () => ctx.keys.client.stop(),
              () => ctx.jobs.stop(),
++            () => { if (ctx.searchIndex) ctx.searchIndex.stop(); },
              async () => { clearInterval(ctx.telemetryTick); if (ctx.retention) await ctx.retention.stop(); if (ctx.telemetry) await ctx.telemetry.stop(); },
              () => ctx.s.close(),
              ...extra,
@@@ -45,6 +46,8 @@ async function start() {
  
      // Jobs. Each refuses to start under SHOES_DRILL; tests call their runOnce().
      ctx.jobs.start();
++    // The model pages in OpenVibe.Search: a sweep a minute after boot and every six hours (off under SHOES_DRILL).
++    ctx.searchIndex.start();
  
      // The two account topics (export_requested, deleted) at OpenVibe.Events: idempotent; off without the URL, the
      // delivery secret and the client secret. The consumer itself is mounted in server/app.js.
ddiff --git a/server/search-index.js b/server/search-index.js
nnew file mode 100644
iindex 0000000..1f1f2c5
---- /dev/null
++++ b/server/search-index.js
@@@ -0,0 +1,101 @@
++'use strict';
++
++/**
++ * open.shoes' model pages in OpenVibe.Search (openvibe-publishing/search-feed, contracts 0.273.0): one
++ * search.index-document@1 per shoe model (/shoe/<slug>, the slug as the id), sent as shoes.index_document.upserted|deleted
++ * through this service's events outbox. The document is what the page says: the model's name and brand, the lowest
++ * all-in price seen today (in dollars, the way the page shows it) and how many stores sell it. A network search for a
++ * shoe found deals and coupons but never the one page with every store's price (2026-11-02).
++ *
++ * A model with no live offer is a thin page, kept in Search but noindex; a model that left the catalogue gets a
++ * tombstone (exists()). The catalogue is small (199 models on 2026-11-02) and its prices move through the day, so a
++ * sweep a minute after boot and every six hours is enough: the sequencer (shoes_index_revisions) gives an unchanged
++ * document its old revision, and a quiet sweep sends nothing. Off under SHOES_DRILL.
++ */
++const { createServiceOutbox } = require('openvibe-sdk/events');
++const { createSearchFeed } = require('openvibe-publishing/search-feed');
++
++const EVENT_TYPES = ['shoes.index_document.upserted', 'shoes.index_document.deleted'];
++const START_DELAY_MS = 60_000;
++const INTERVAL_MS = 6 * 3_600_000;
++
++const dollars = (n) => (n == null || !Number.isFinite(Number(n)) ? null : `$${Math.round(Number(n)).toLocaleString('en-US')}`);
++const iso = (t) => (t ? new Date(t).toISOString() : null);
++
++/** One model row (with its live offers' figures) → what its page says, as search-feed's document description. */
++function describe(row) {
++    const stores = Number(row.store_count) || 0;
++    const from = dollars(row.price_min);
++    const where = stores ? `${stores} store${stores === 1 ? '' : 's'}` : null;
++    const price = from && where ? `From ${from} at ${where} today.` : where ? `Sold at ${where} today.` : 'No store lists it right now.';
++    return {
++        listed: true,
++        noindex: !stores,
++        title: row.name,
++        summary: `${row.name} by ${row.brand}. ${price} Every store's price, the price history and the size curve on open.shoes.`,
++        body: [row.name, row.brand, row.model, price].filter(Boolean).join('\n'),
++        facets: { brand: row.brand, stores, ...(row.price_min != null ? { lowest_price: Math.round(Number(row.price_min)) } : {}) },
++        authorship: { mode: 'imported' },
++        publishedAt: iso(row.first_seen_at),
++        updatedAt: iso(row.checked_at || row.updated_at),
++    };
++}
++
++function createSearchIndex({ config, s, log = console, outbox = null }) {
++    const db = s.db;
++    const out = outbox || createServiceOutbox({
++        db, source: 'shoes', eventsUrl: config.events.url || null, networkInternalUrl: config.networkInternalUrl,
++        clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, log, eventTypes: EVENT_TYPES,
++    });
++    // The page is /shoe/<slug>, so the slug is the document's id: rows come back keyed by it, in slug order.
++    const rows = async (after, limit) => db.many(
++        `SELECT m.slug AS id, m.name, m.brand, m.model, m.first_seen_at, m.updated_at,
++                (SELECT MIN(o.price_amount + COALESCE(o.shipping_amount, 0)) FROM shoe_offers o
++                  WHERE o.model_id = m.id AND o.active AND o.price_amount IS NOT NULL AND o.currency = 'USD') AS price_min,
++                (SELECT COUNT(DISTINCT o.store_domain)::int FROM shoe_offers o WHERE o.model_id = m.id AND o.active) AS store_count,
++                (SELECT MAX(o.checked_at) FROM shoe_offers o WHERE o.model_id = m.id AND o.active) AS checked_at
++           FROM shoe_models m WHERE m.slug > $1 ORDER BY m.slug LIMIT $2`,
++        [String(after || ''), limit],
++    );
++    const feed = createSearchFeed({
++        owner: 'shoes', db, outbox: out, baseUrl: config.baseUrl, now: s.now, log,
++        types: {
++            shoe: {
++                page: (row) => `/shoe/${row.id}`,
++                document: describe,
++                rows,
++                exists: async (ids) => (ids.length ? (await db.many('SELECT slug FROM shoe_models WHERE slug = ANY($1::text[])', [ids])).map((r) => r.slug) : []),
++            },
++        },
++    });
++    const timers = [];
++
++    async function sweep() {
++        const res = await feed.sweep();
++        const r = res.shoe;
++        if (r.sent || r.removed || r.failed) log.log(`[Search] shoe models: ${r.sent} sent, ${r.removed} removed, ${r.failed} failed of ${r.seen}`);
++        if (out.kick) out.kick().catch(() => {});
++        return res;
++    }
++    const quietly = () => { sweep().catch((err) => log.warn(`[Search] sweep failed: ${(err && err.message) || err}`)); };
++
++    function start() {
++        if (config.drill || timers.length) return false;
++        out.start();
++        const kick = setTimeout(quietly, START_DELAY_MS);
++        if (kick.unref) kick.unref();
++        const tick = setInterval(quietly, INTERVAL_MS);
++        if (tick.unref) tick.unref();
++        timers.push(kick, tick);
++        return true;
++    }
++
++    function stop() {
++        for (const t of timers.splice(0)) { clearTimeout(t); clearInterval(t); }
++        out.stop();
++    }
++
++    return { feed, outbox: out, describe, sweep, start, stop, status: () => out.status() };
++}
++
++module.exports = { createSearchIndex, describe, EVENT_TYPES };
ddiff --git a/test/search-index.test.js b/test/search-index.test.js
nnew file mode 100644
iindex 0000000..28f3dd6
---- /dev/null
++++ b/test/search-index.test.js
@@@ -0,0 +1,63 @@
++'use strict';
++/**
++ * Shoe models in OpenVibe.Search (server/search-index.js): a sweep puts every model in the outbox as one
++ * shoes.index_document.upserted, exactly the contract (contracts 0.273.0), with its /shoe/<slug> page (which answers),
++ * its brand, store count and lowest price; a second sweep sends nothing; a model that left the catalogue becomes a
++ * tombstone, once.
++ */
++const assert = require('assert');
++const contracts = require('openvibe-contracts');
++const { boot, check, done } = require('./helpers/boot');
++const { describe } = require('../server/search-index');
++
++(async () => {
++    const t = await boot();
++    const { s, searchIndex } = t.ctx;
++    const outbox = async () => (await s.db.many('SELECT envelope FROM event_outbox ORDER BY id')).map((r) => r.envelope);
++    const valid = (env) => {
++        assert.ok(contracts.validate('events.event-envelope@1', env).valid, 'envelope');
++        const p = contracts.validate(env.event_type, env.payload);
++        assert.ok(p.valid, `${env.event_type}: ${JSON.stringify(p.errors)}`);
++        assert.strictEqual(env.source, 'shoes');
++        return env;
++    };
++
++    await t.ctx.jobs.offers.runOnce({ pages: 2 });
++    const models = await s.db.many('SELECT slug FROM shoe_models ORDER BY slug');
++
++    await check('a sweep sends every model once, as the contract says, and the page it names answers', async () => {
++        assert.ok(models.length >= 2, `the offers job made models (${models.length})`);
++        const res = await searchIndex.sweep();
++        assert.deepStrictEqual(res.shoe, { seen: models.length, sent: models.length, removed: 0, failed: 0 });
++        const envs = (await outbox()).map(valid);
++        const doc = envs.find((e) => e.payload.id === models[0].slug).payload;
++        assert.strictEqual(doc.canonical_url, `${t.config.baseUrl}/shoe/${models[0].slug}`);
++        assert.strictEqual(doc.type, 'shoe');
++        assert.ok(doc.title && doc.facets.brand, 'a name and a brand');
++        assert.match(doc.summary, /store/);
++        assert.strictEqual((await t.get(`/shoe/${models[0].slug}`)).status, 200);
++        assert.strictEqual((await searchIndex.sweep()).shoe.sent, 0, 'a second sweep sends nothing');
++    });
++
++    await check('a model with no live offer is noindex; with offers, the lowest dollar price and the store count', async () => {
++        const none = describe({ name: 'Plain Runner', brand: 'Acme', model: 'Plain Runner', store_count: 0, price_min: null });
++        assert.strictEqual(none.noindex, true);
++        assert.match(none.summary, /No store lists it right now/);
++        const some = describe({ name: 'Air Max 90', brand: 'Nike', model: 'Air Max 90', store_count: 3, price_min: '98.40' });
++        assert.strictEqual(some.noindex, false);
++        assert.match(some.summary, /From \$98 at 3 stores today/);
++        assert.deepStrictEqual(some.facets, { brand: 'Nike', stores: 3, lowest_price: 98 });
++    });
++
++    await check('a model that left the catalogue becomes a tombstone, once', async () => {
++        const gone = models[models.length - 1].slug;
++        await s.db.many('DELETE FROM shoe_models WHERE slug = $1 RETURNING slug', [gone]);
++        assert.strictEqual((await searchIndex.sweep()).shoe.removed, 1);
++        const env = valid((await outbox()).pop());
++        assert.strictEqual(env.event_type, 'shoes.index_document.deleted');
++        assert.deepStrictEqual([env.payload.type, env.payload.id], ['shoe', gone]);
++        assert.strictEqual((await searchIndex.sweep()).shoe.removed, 0);
++    });
++
++    done();
++})();