openvibers/openvibe.deals Commits
reads: one latest-observation and one snapshot query per page, not per offer (#254)
Changes
@@@ -63,12 +63,12 @@ function createPublication({ config, store, reads, catalog }) {}/** Everything a page, the JSON twin, an event or the index needs about one canonical offer. */-- async function offerView(offer, now = store.now()) {-- const root = await reads.root(offer);-- const ids = await reads.groupIds(root.id);-- const latest = await reads.latestObservation(ids);++ async function offerView(offer, now = store.now(), pre = null) {++ const root = pre ? pre.root : await reads.root(offer);++ const ids = pre ? pre.ids : await reads.groupIds(root.id);++ const latest = pre ? pre.latest : await reads.latestObservation(ids);const tally = await reads.tally(ids);-- const snap = await reads.lastSnapshot(root.id);++ const snap = pre ? pre.snap : await reads.lastSnapshot(root.id);const sources = await reads.sources(ids);const observations = await reads.observations(ids);return {@@@ -93,17 +93,52 @@ function createPublication({ config, store, reads, catalog }) {};}++ /**++ * The page's shared reads: every root, group, latest observation and (for offer views) snapshot, once. A page of 25++ * offers used to repeat each of those per offer: 2.1 M snapshot lookups and 3.5 M latest-observation lookups a day++ * on 2026-11-08 (324 s and 251 s of DB time, pg_stat_statements).++ */++ async function prefetch(offers, { snapshots = true } = {}) {++ const roots = await Promise.all(offers.map(async (o) => (o.merged_into ? await reads.root(o) : o)));++ // A group read per offer, at most eight at once: the pool has a handful of connections for everyone.++ const groups = new Array(roots.length);++ for (let i = 0; i < roots.length; i += 8) {++ await Promise.all(roots.slice(i, i + 8).map(async (r, j) => { groups[i + j] = await reads.groupIds(r.id); }));++ }++ const [latest, snaps] = await Promise.all([++ reads.latestObservations(groups),++ snapshots ? reads.lastSnapshots(roots.map((r) => r.id)) : Promise.resolve(new Map()),++ ]);++ return { roots, groups, latest, snaps };++ }++++ /** A whole page of offer views from one read per fact, each view with its own group's latest and its root's snapshot. */++ async function offerViews(offers, now = store.now()) {++ if (!offers.length) return [];++ const pre = await prefetch(offers);++ return await Promise.all(offers.map((o, i) => offerView(o, now, {++ root: pre.roots[i], ids: pre.groups[i], latest: pre.latest[i], snap: pre.snaps.get(pre.roots[i].id) || null,++ })));++ }++/*** The little a card needs (the home page's sections): the group's latest observation, its freshness, the store and* the attribution a source requires. Four reads instead of offerView's ten, so a page of 40 cards stays quick.*/-- async function cardView(offer, now = store.now()) {-- const root = offer.merged_into ? await reads.root(offer) : offer;-- const ids = await reads.groupIds(root.id);-- const latest = await reads.latestObservation(ids);++ async function cardView(offer, now = store.now(), pre = null) {++ const root = pre ? pre.root : (offer.merged_into ? await reads.root(offer) : offer);++ const ids = pre ? pre.ids : await reads.groupIds(root.id);++ const latest = pre ? pre.latest : await reads.latestObservation(ids);return { root, ids, store: await catalog.store(root.store_id), latest, freshness: freshness(latest, now), attribution: attributionOf(await reads.sources(ids)), now };}++ /** A whole home page of cards from one read per fact (no snapshots: a card has no hotness). */++ async function cardViews(offers, now = store.now()) {++ if (!offers.length) return [];++ const pre = await prefetch(offers, { snapshots: false });++ return await Promise.all(offers.map((o, i) => cardView(o, now, { root: pre.roots[i], ids: pre.groups[i], latest: pre.latest[i] })));++ }++function observationDto(o) {return {id: o.id, observed_at: iso(o.observed_at), recorded_at: iso(o.recorded_at),@@@ -206,7 +241,8 @@ function createPublication({ config, store, reads, catalog }) {* an old or unknown price never sorts as if it were the price now, and unknown never sorts as 0.*/async function productView(product, now = store.now()) {-- const rows = await Promise.all((await productOfferIds.all(product.id)).map(async (r) => await offerView(await reads.get(r.id), now)));++ const offers = await Promise.all((await productOfferIds.all(product.id)).map(async (r) => await reads.get(r.id)));++ const rows = await offerViews(offers, now);const rank = (v) => (v.root.status !== 'active' ? 4 : v.freshness.state !== 'fresh' ? 3 : (v.latest && v.latest.price != null && v.latest.currency) ? 1 : 2);rows.sort((a, b) => rank(a) - rank(b)|| (rank(a) === 1 ? (a.latest.currency < b.latest.currency ? -1 : a.latest.currency > b.latest.currency ? 1 : a.latest.price_num - b.latest.price_num) : 0)@@@ -238,10 +274,10 @@ function createPublication({ config, store, reads, catalog }) {* the full views instead cost ~6.7 KB an offer — ~340 MB of a 50,000-offer catalog, held twice over once products* were built (2026-11-07: the Deals process at 667 MB-923 MB, peaking at 2 GB).*/-- async function indexEntry(offer, now = store.now()) {-- const root = offer.merged_into ? await reads.root(offer) : offer;-- const ids = await reads.groupIds(root.id);-- const latest = await reads.latestObservation(ids);++ async function indexEntry(offer, now = store.now(), pre = null) {++ const root = pre ? pre.root : (offer.merged_into ? await reads.root(offer) : offer);++ const ids = pre ? pre.ids : await reads.groupIds(root.id);++ const latest = pre ? pre.latest : await reads.latestObservation(ids);// listings.indexable() joins the store onto each row; any other caller gets it looked up here.const s = root.store_domain !== undefined? (root.store_domain ? { name: root.store_name, domain: root.store_domain } : null)@@@ -256,6 +292,17 @@ function createPublication({ config, store, reads, catalog }) {};}++ /** The sitemap arrays, prefetched in chunks: one latest-observation read per 100 offers instead of per offer. */++ async function indexEntries(offers, now = store.now(), { chunk = 100 } = {}) {++ const out = [];++ for (let i = 0; i < offers.length; i += chunk) {++ const part = offers.slice(i, i + chunk);++ const pre = await prefetch(part, { snapshots: false });++ out.push(...await Promise.all(part.map((o, j) => indexEntry(o, now, { root: pre.roots[j], ids: pre.groups[j], latest: pre.latest[j] }))));++ }++ return out;++ }++// A product's lastmod and freshest active observation in one read: each offer's group latest observation (a merge// group shares one price history), without loading every offer's view. The same facts productView derives below.const productIndex = store.db.prepare(`SELECT MAX(GREATEST(o.updated_at, COALESCE(l.observed_at, 0))) AS lastmod,@@@ -281,7 +328,7 @@ function createPublication({ config, store, reads, catalog }) {};}-- return { abs, offerPath, productPath, storePath, sellerPath, categoryPath, freshness, decideOffer, decideProduct, offerView, cardView, offerDto, observationDto, offerLd, offerJsonLd, productView, productJsonLd, indexEntry, productIndexEntry };++ return { abs, offerPath, productPath, storePath, sellerPath, categoryPath, freshness, decideOffer, decideProduct, offerView, offerViews, cardView, cardViews, offerDto, observationDto, offerLd, offerJsonLd, productView, productJsonLd, indexEntry, indexEntries, productIndexEntry };}module.exports = { createPublication };@@@ -31,6 +31,14 @@ function createReads({ store }) {WHERE offer_id IN (SELECT value FROM jsonb_array_elements_text(?::jsonb)) ORDER BY updated_at, seq`),members: db.prepare('SELECT * FROM deal_offers WHERE merged_into = ? ORDER BY merged_at'),lastSnapshot: db.prepare('SELECT * FROM deal_hotness_snapshots WHERE offer_id = ? ORDER BY id DESC LIMIT 1'),++ // The batched forms: one statement for every offer of a page instead of one per offer.++ latestByOffer: db.prepare(`SELECT DISTINCT ON (o.offer_id) o.*, s.kind AS source_kind, s.ref_service, s.ref_id, s.source_key, s.url AS source_url, s.label AS source_label++ FROM deal_price_observations o JOIN deal_offer_sources s ON s.id = o.source_id++ WHERE o.offer_id IN (SELECT value FROM jsonb_array_elements_text(?::jsonb)) AND s.removed_at IS NULL++ ORDER BY o.offer_id, o.observed_at DESC, o.seq DESC`),++ lastSnapshots: db.prepare(`SELECT DISTINCT ON (offer_id) * FROM deal_hotness_snapshots++ WHERE offer_id IN (SELECT value FROM jsonb_array_elements_text(?::jsonb))++ ORDER BY offer_id, id DESC`),};const get = async (id) => (id ? await q.byId.get(String(id)) : null);@@@ -64,6 +72,32 @@ function createReads({ store }) {return bySubject;}++ /**++ * The latest observation of each offer in one query, then the group's newest of those (the old LIMIT 1 read, one++ * query per offer). `lists` is one merge group per page entry; the answer is aligned with it.++ */++ const pickLatest = (ids, byOffer) => {++ let best = null;++ for (const id of ids) {++ const o = byOffer.get(id);++ if (o && (!best || o.observed_at > best.observed_at || (o.observed_at === best.observed_at && o.seq > best.seq))) best = o;++ }++ return best || null;++ };++ async function latestObservations(lists) {++ const ids = [...new Set(lists.flat())];++ if (!ids.length) return lists.map(() => null);++ const byOffer = new Map((await q.latestByOffer.all(J(ids))).map((o) => [o.offer_id, o]));++ return lists.map((list) => pickLatest(list, byOffer));++ }++++ /** Each root's latest snapshot in one query, keyed by offer id. */++ async function lastSnapshots(ids) {++ const unique = [...new Set(ids)];++ if (!unique.length) return new Map();++ return new Map((await q.lastSnapshots.all(J(unique))).map((s) => [s.offer_id, s]));++ }++async function tally(ids) {let up = 0, down = 0, upW = 0, downW = 0;for (const v of (await effectiveVotes(ids)).values()) {@@@ -80,12 +114,14 @@ function createReads({ store }) {root,groupIds,observations: async (ids) => await q.observations.all(J(ids)),-- latestObservation: async (ids) => await q.latest.get(J(ids)) || null,sources: async (ids) => await q.sources.all(J(ids)),effectiveVotes,tally,mergedMembers: async (id) => await q.members.all(id),++ latestObservation: async (ids) => await q.latest.get(J(ids)) || null,++ latestObservations,lastSnapshot: async (id) => await q.lastSnapshot.get(id) || null,++ lastSnapshots,};}@@@ -83,7 +83,7 @@ function createApi(ctx) {const page = Math.max(1, parseInt(req.query.page, 10) || 1);const per = 25;const rows = req.query.sort === 'new' ? await listings.newest(per, (page - 1) * per) : await listings.hot(per, (page - 1) * per);-- const views = await Promise.all(rows.map(async (o) => publication.offerView(o)));++ const views = await publication.offerViews(rows);const visible = fromExtension(req) ? views.filter((v) => !v.extensionExcluded) : views;return { sort: req.query.sort === 'new' ? 'new' : 'hot', page, total: await listings.countActive(), offers: visible.map((v) => publication.offerDto(v)) };}));@@@ -201,7 +201,7 @@ function createApi(ctx) {router.get('/stores/:domain', run(async (req) => {const st = await catalog.storeByDomain(req.params.domain);if (!st) throw new ApiError(404, 'store.not_found', 'No such store');-- const views = await Promise.all((await listings.byStore(st.id, 50, 0)).map(async (o) => await publication.offerView(o)));++ const views = await publication.offerViews(await listings.byStore(st.id, 50, 0));const offers = fromExtension(req) ? views.filter((v) => !v.extensionExcluded) : views;return { store: { id: st.id, domain: st.domain, name: st.name, url: publication.abs(publication.storePath(st)) }, offers: offers.map((v) => publication.offerDto(v)) };}));@@@ -90,7 +90,7 @@ function createDiscoveryRoutes({ config, store, publication, listings }) {// and the cache is bounded by the listings' own 50,000-offer and 50,000-product caps.const ttl = Number(config.sitemapTtlMs) || 0;const offerEntries = shared(async () => {-- const entries = await mapLimit(await listings.indexable(), VIEWS_PARALLEL, (o) => publication.indexEntry(o));++ const entries = await publication.indexEntries(await listings.indexable());let kept = 0;for (const e of entries) {if (e.decision.indexable && ++kept <= LLMS_SUMMARIES_MAX) continue;@@@ -142,7 +142,7 @@ function createPages(ctx) {const total = await L.count();const pager = ssr.paginate({ page, perPage: PER_PAGE, total, href: (p) => (p === 1 ? path : `${path}?page=${p}`) });if (pager.outOfRange && total) return null;-- const list = await Promise.all((await L.rows(PER_PAGE, pager.offset)).map(async (o) => await publication.offerView(o)));++ const list = await publication.offerViews(await L.rows(PER_PAGE, pager.offset));const canonical = seo.canonicalUrl(config.baseUrl, pager.page === 1 ? path : `${path}?page=${pager.page}`, { query: ['page'] });return {agent: true,@@@ -177,9 +177,11 @@ function createPages(ctx) {listings.picks(HOME_CARDS * 2, 0), listings.upvoted(HOME_CARDS * 2), listings.newest(HOME_CARDS * 3, 0),listings.endingSoon(now, now + 7 * DAY, HOME_CARDS * 2, 0), listings.under(UNDER.amount, UNDER.currency, fresh, HOME_CARDS * 4),]);-- const viewOf = new Map();-- for (const o of [...pickRows, ...upRows, ...newRows, ...endRows, ...underRows]) if (!viewOf.has(o.id)) viewOf.set(o.id, publication.cardView(o, now));-- await Promise.all(viewOf.values());++ const candidates = [];++ const candidateIds = new Set();++ for (const o of [...pickRows, ...upRows, ...newRows, ...endRows, ...underRows]) if (!candidateIds.has(o.id)) { candidateIds.add(o.id); candidates.push(o); }++ const cards = await publication.cardViews(candidates, now);++ const viewOf = new Map(candidates.map((o, i) => [o.id, cards[i]]));const top = new Set();const shown = new Set();const take = async (rows, { skip = shown, keep = () => true, into = [top, shown] } = {}) => {@@@ -449,7 +451,7 @@ function createPages(ctx) {const total = await listings.countStore(st.id);const pager = ssr.paginate({ page, perPage: PER_PAGE, total, href: (p) => (p === 1 ? path : `${path}?page=${p}`) });if (pager.outOfRange && total) return null;-- const list = await Promise.all((await listings.byStore(st.id, PER_PAGE, pager.offset)).map(async (o) => await publication.offerView(o)));++ const list = await publication.offerViews(await listings.byStore(st.id, PER_PAGE, pager.offset));const canonical = seo.canonicalUrl(config.baseUrl, pager.page === 1 ? path : `${path}?page=${pager.page}`, { query: ['page'] });return {agent: true,@@@ -491,7 +493,7 @@ function createPages(ctx) {const total = await listings.countSeller(hit.seller);const pager = ssr.paginate({ page, perPage: PER_PAGE, total, href: (p) => (p === 1 ? path : `${path}?page=${p}`) });if (pager.outOfRange && total) return null;-- const list = await Promise.all((await listings.bySeller(hit.seller, PER_PAGE, pager.offset)).map(async (o) => await publication.offerView(o)));++ const list = await publication.offerViews(await listings.bySeller(hit.seller, PER_PAGE, pager.offset));const canonical = seo.canonicalUrl(config.baseUrl, pager.page === 1 ? path : `${path}?page=${pager.page}`, { query: ['page'] });return {agent: true,@@@ -529,7 +531,7 @@ function createPages(ctx) {const total = await listings.countCategory(hit.category);const pager = ssr.paginate({ page, perPage: PER_PAGE, total, href: (p) => (p === 1 ? path : `${path}?page=${p}`) });if (pager.outOfRange && total) return null;-- const list = await Promise.all((await listings.byCategory(hit.category, PER_PAGE, pager.offset)).map(async (o) => await publication.offerView(o)));++ const list = await publication.offerViews(await listings.byCategory(hit.category, PER_PAGE, pager.offset));const canonical = seo.canonicalUrl(config.baseUrl, pager.page === 1 ? path : `${path}?page=${pager.page}`, { query: ['page'] });return {agent: true,@@@ -556,7 +558,7 @@ function createPages(ctx) {send(req, res, 200, {title: q ? `Search: ${q}` : 'Search deals',decision: pageDecision('/search', { indexable: false }),-- body: views.searchPage({ q, results: await Promise.all(r.rows.map(async (o) => await publication.offerView(o))), pager, urls, csrf: csrf(req), signedIn: signedIn(req) }),++ body: views.searchPage({ q, results: await publication.offerViews(r.rows), pager, urls, csrf: csrf(req), signedIn: signedIn(req) }),});}));@@@ -706,8 +708,10 @@ function createPages(ctx) {// ── feeds (never viewer-dependent) ──────────────────────async function feedItems() {-- return Promise.all((await listings.newest(50, 0)).map(async (o) => {-- const v = await publication.offerView(o);++ const offers = await listings.newest(50, 0);++ const pageViews = await publication.offerViews(offers);++ return offers.map((o, i) => {++ const v = pageViews[i];const price = `${views.priceText(v.latest)}${v.latest ? ` as of ${iso(v.latest.observed_at)}` : ''}`;// A source policy may require attribution: carried as text in the summary and as a tag.const via = v.attribution ? `Deal via ${v.attribution.text}` : null;@@@ -717,7 +721,7 @@ function createPages(ctx) {tags: via ? [via] : undefined,published: o.created_at, updated: Math.max(o.updated_at, v.latest ? v.latest.observed_at : 0), decision: v.decision,};-- }));++ });}const channel = () => ({ title: 'OpenVibe.Deals — new deals', link: publication.abs('/new'), description: 'New deals, each with its source and the time its price was observed.', language: 'en' });nnew file mode 100644@@@ -0,0 +1,68 @@++'use strict';++/**++ * A list of offers reads each offer's latest observation and latest hotness snapshot once, not once per offer.++ *++ * On 2026-11-08 the two per-offer reads were 2.1 M and 3.5 M calls a day (324 s and 251 s of DB time,++ * pg_stat_statements, `reads.lastSnapshot` and `reads.latestObservation`), repeated by every page that renders++ * offers. They are now one statement per turn: a page of six offers is one query of each, and each view still++ * carries its own offer's latest observation and snapshot.++ */++const assert = require('assert');++const { boot, check, done } = require('./helpers/boot');++++(async () => {++ const t = await boot();++ const db = t.ctx.store.db;++ const poster = t.network.addUser('poster');++++ const offers = [];++ for (let i = 0; i < 6; i++) {++ offers.push(await t.submit(poster, { url: `https://batch.example/${i}`, title: `Batch offer ${i}`, price: String(10 + i), currency: 'USD' }));++ }++ // A snapshot of its own for each offer (snapshot() also writes the offer's hot_latest).++ for (const [i, o] of offers.entries()) await t.ctx.hotness.snapshot(o.id, 'tick', t.clock.now() + i);++++ const seen = { latest: 0, snapshots: 0 };++ const query = db.query.bind(db);++ db.query = (text, values) => {++ if (/DISTINCT ON \(o\.offer_id\)/.test(text)) seen.latest++;++ else if (/DISTINCT ON \(offer_id\)/.test(text)) seen.snapshots++;++ return query(text, values);++ };++ let views;++ try {++ views = await t.ctx.publication.offerViews(offers);++ } finally {++ db.query = query;++ }++++ await check('a page of six offers reads latest observations and snapshots once each', async () => {++ assert.strictEqual(seen.latest, 1, 'one latest-observation query for the page');++ assert.strictEqual(seen.snapshots, 1, 'one snapshot query for the page');++ });++++ await check('each view still carries its own latest observation and snapshot', async () => {++ for (const [i, v] of views.entries()) {++ assert.strictEqual(v.root.id, offers[i].id);++ assert.strictEqual(v.latest.offer_id, offers[i].id);++ assert.strictEqual(v.latest.price_num, 10 + i);++ assert.strictEqual(v.hotness.offer_id, offers[i].id);++ }++ });++++ await check('the batched answer is the row the old one-per-offer query returned', async () => {++ const id = offers[2].id;++ const batched = await t.ctx.reads.latestObservation([id]);++ const direct = await db.prepare(`SELECT o.*, s.kind AS source_kind, s.ref_service, s.ref_id, s.source_key, s.url AS source_url, s.label AS source_label++ FROM deal_price_observations o JOIN deal_offer_sources s ON s.id = o.source_id++ WHERE o.offer_id IN (SELECT value FROM jsonb_array_elements_text(?::jsonb)) AND s.removed_at IS NULL++ ORDER BY o.observed_at DESC, o.seq DESC LIMIT 1`).get(JSON.stringify([id]));++ assert.deepStrictEqual(batched, direct);++ const snap = await t.ctx.reads.lastSnapshot(id);++ const directSnap = await db.prepare('SELECT * FROM deal_hotness_snapshots WHERE offer_id = ? ORDER BY id DESC LIMIT 1').get(id);++ assert.deepStrictEqual(snap, directSnap);++ });++++ await t.close();++ done();++})();@@@ -31,6 +31,12 @@ function routes(ttlMs, clock, { offerCount = 20, products = [] } = {}) {seen.now--;return { loc: `https://deals.test/d/${o.slug}`, lastmod: o.updated_at, decision: indexable, title: o.slug, text: '1 USD' };},++ async indexEntries(offers) {++ // The real indexEntries prefetches per chunk and reads the page views at most eight at a time.++ const out = new Array(offers.length);++ for (let i = 0; i < offers.length; i += 8) await Promise.all(offers.slice(i, i + 8).map(async (o, j) => { out[i + j] = await publication.indexEntry(o); }));++ return out;++ },async productIndexEntry(p) { return { loc: `https://deals.test/p/${p.slug}`, lastmod: p.updated_at, decision: indexable }; },};const listings = {
@@ -63,12 +63,12 @@ function createPublication({ config, store, reads, catalog }) {
@@ -63,12 +63,12 @@ function createPublication({ config, store, reads, catalog }) {
}
}
/** Everything a page, the JSON twin, an event or the index needs about one canonical offer. */
/** Everything a page, the JSON twin, an event or the index needs about one canonical offer. */
- async function offerView(offer, now = store.now()) {
- const root = await reads.root(offer);
- const ids = await reads.groupIds(root.id);
- const latest = await reads.latestObservation(ids);
+ async function offerView(offer, now = store.now(), pre = null) {
+ const root = pre ? pre.root : await reads.root(offer);
+ const ids = pre ? pre.ids : await reads.groupIds(root.id);
+ const latest = pre ? pre.latest : await reads.latestObservation(ids);
const tally = await reads.tally(ids);
const tally = await reads.tally(ids);
- const snap = await reads.lastSnapshot(root.id);
+ const snap = pre ? pre.snap : await reads.lastSnapshot(root.id);
const sources = await reads.sources(ids);
const sources = await reads.sources(ids);
const observations = await reads.observations(ids);
const observations = await reads.observations(ids);
return {
return {
@@ -93,17 +93,52 @@ function createPublication({ config, store, reads, catalog }) {
@@ -93,17 +93,52 @@ function createPublication({ config, store, reads, catalog }) {
};
};
}
}
+ /**
+ * The page's shared reads: every root, group, latest observation and (for offer views) snapshot, once. A page of 25
+ * offers used to repeat each of those per offer: 2.1 M snapshot lookups and 3.5 M latest-observation lookups a day
+ * on 2026-11-08 (324 s and 251 s of DB time, pg_stat_statements).
+ */
+ async function prefetch(offers, { snapshots = true } = {}) {
+ const roots = await Promise.all(offers.map(async (o) => (o.merged_into ? await reads.root(o) : o)));
+ // A group read per offer, at most eight at once: the pool has a handful of connections for everyone.
+ const groups = new Array(roots.length);
+ for (let i = 0; i < roots.length; i += 8) {
+ await Promise.all(roots.slice(i, i + 8).map(async (r, j) => { groups[i + j] = await reads.groupIds(r.id); }));
+ }
+ const [latest, snaps] = await Promise.all([
+ reads.latestObservations(groups),
+ snapshots ? reads.lastSnapshots(roots.map((r) => r.id)) : Promise.resolve(new Map()),
+ ]);
+ return { roots, groups, latest, snaps };
+ }
+
+ /** A whole page of offer views from one read per fact, each view with its own group's latest and its root's snapshot. */
+ async function offerViews(offers, now = store.now()) {
+ if (!offers.length) return [];
+ const pre = await prefetch(offers);
+ return await Promise.all(offers.map((o, i) => offerView(o, now, {
+ root: pre.roots[i], ids: pre.groups[i], latest: pre.latest[i], snap: pre.snaps.get(pre.roots[i].id) || null,
+ })));
+ }
+
/**
/**
* The little a card needs (the home page's sections): the group's latest observation, its freshness, the store and
* The little a card needs (the home page's sections): the group's latest observation, its freshness, the store and
* the attribution a source requires. Four reads instead of offerView's ten, so a page of 40 cards stays quick.
* the attribution a source requires. Four reads instead of offerView's ten, so a page of 40 cards stays quick.
*/
*/
- async function cardView(offer, now = store.now()) {
- const root = offer.merged_into ? await reads.root(offer) : offer;
- const ids = await reads.groupIds(root.id);
- const latest = await reads.latestObservation(ids);
+ async function cardView(offer, now = store.now(), pre = null) {
+ const root = pre ? pre.root : (offer.merged_into ? await reads.root(offer) : offer);
+ const ids = pre ? pre.ids : await reads.groupIds(root.id);
+ const latest = pre ? pre.latest : await reads.latestObservation(ids);
return { root, ids, store: await catalog.store(root.store_id), latest, freshness: freshness(latest, now), attribution: attributionOf(await reads.sources(ids)), now };
return { root, ids, store: await catalog.store(root.store_id), latest, freshness: freshness(latest, now), attribution: attributionOf(await reads.sources(ids)), now };
}
}
+ /** A whole home page of cards from one read per fact (no snapshots: a card has no hotness). */
+ async function cardViews(offers, now = store.now()) {
+ if (!offers.length) return [];
+ const pre = await prefetch(offers, { snapshots: false });
+ return await Promise.all(offers.map((o, i) => cardView(o, now, { root: pre.roots[i], ids: pre.groups[i], latest: pre.latest[i] })));
+ }
+
function observationDto(o) {
function observationDto(o) {
return {
return {
id: o.id, observed_at: iso(o.observed_at), recorded_at: iso(o.recorded_at),
id: o.id, observed_at: iso(o.observed_at), recorded_at: iso(o.recorded_at),
@@ -206,7 +241,8 @@ function createPublication({ config, store, reads, catalog }) {
@@ -206,7 +241,8 @@ function createPublication({ config, store, reads, catalog }) {
* an old or unknown price never sorts as if it were the price now, and unknown never sorts as 0.
* an old or unknown price never sorts as if it were the price now, and unknown never sorts as 0.
*/
*/
async function productView(product, now = store.now()) {
async function productView(product, now = store.now()) {
- const rows = await Promise.all((await productOfferIds.all(product.id)).map(async (r) => await offerView(await reads.get(r.id), now)));
+ const offers = await Promise.all((await productOfferIds.all(product.id)).map(async (r) => await reads.get(r.id)));
+ const rows = await offerViews(offers, now);
const rank = (v) => (v.root.status !== 'active' ? 4 : v.freshness.state !== 'fresh' ? 3 : (v.latest && v.latest.price != null && v.latest.currency) ? 1 : 2);
const rank = (v) => (v.root.status !== 'active' ? 4 : v.freshness.state !== 'fresh' ? 3 : (v.latest && v.latest.price != null && v.latest.currency) ? 1 : 2);
rows.sort((a, b) => rank(a) - rank(b)
rows.sort((a, b) => rank(a) - rank(b)
|| (rank(a) === 1 ? (a.latest.currency < b.latest.currency ? -1 : a.latest.currency > b.latest.currency ? 1 : a.latest.price_num - b.latest.price_num) : 0)
|| (rank(a) === 1 ? (a.latest.currency < b.latest.currency ? -1 : a.latest.currency > b.latest.currency ? 1 : a.latest.price_num - b.latest.price_num) : 0)
@@ -238,10 +274,10 @@ function createPublication({ config, store, reads, catalog }) {
@@ -238,10 +274,10 @@ function createPublication({ config, store, reads, catalog }) {
* the full views instead cost ~6.7 KB an offer — ~340 MB of a 50,000-offer catalog, held twice over once products
* the full views instead cost ~6.7 KB an offer — ~340 MB of a 50,000-offer catalog, held twice over once products
* were built (2026-11-07: the Deals process at 667 MB-923 MB, peaking at 2 GB).
* were built (2026-11-07: the Deals process at 667 MB-923 MB, peaking at 2 GB).
*/
*/
- async function indexEntry(offer, now = store.now()) {
- const root = offer.merged_into ? await reads.root(offer) : offer;
- const ids = await reads.groupIds(root.id);
- const latest = await reads.latestObservation(ids);
+ async function indexEntry(offer, now = store.now(), pre = null) {
+ const root = pre ? pre.root : (offer.merged_into ? await reads.root(offer) : offer);
+ const ids = pre ? pre.ids : await reads.groupIds(root.id);
+ const latest = pre ? pre.latest : await reads.latestObservation(ids);
// listings.indexable() joins the store onto each row; any other caller gets it looked up here.
// listings.indexable() joins the store onto each row; any other caller gets it looked up here.
const s = root.store_domain !== undefined
const s = root.store_domain !== undefined
? (root.store_domain ? { name: root.store_name, domain: root.store_domain } : null)
? (root.store_domain ? { name: root.store_name, domain: root.store_domain } : null)
@@ -256,6 +292,17 @@ function createPublication({ config, store, reads, catalog }) {
@@ -256,6 +292,17 @@ function createPublication({ config, store, reads, catalog }) {
};
};
}
}
+ /** The sitemap arrays, prefetched in chunks: one latest-observation read per 100 offers instead of per offer. */
+ async function indexEntries(offers, now = store.now(), { chunk = 100 } = {}) {
+ const out = [];
+ for (let i = 0; i < offers.length; i += chunk) {
+ const part = offers.slice(i, i + chunk);
+ const pre = await prefetch(part, { snapshots: false });
+ out.push(...await Promise.all(part.map((o, j) => indexEntry(o, now, { root: pre.roots[j], ids: pre.groups[j], latest: pre.latest[j] }))));
+ }
+ return out;
+ }
+
// A product's lastmod and freshest active observation in one read: each offer's group latest observation (a merge
// A product's lastmod and freshest active observation in one read: each offer's group latest observation (a merge
// group shares one price history), without loading every offer's view. The same facts productView derives below.
// group shares one price history), without loading every offer's view. The same facts productView derives below.
const productIndex = store.db.prepare(`SELECT MAX(GREATEST(o.updated_at, COALESCE(l.observed_at, 0))) AS lastmod,
const productIndex = store.db.prepare(`SELECT MAX(GREATEST(o.updated_at, COALESCE(l.observed_at, 0))) AS lastmod,
@@ -281,7 +328,7 @@ function createPublication({ config, store, reads, catalog }) {
@@ -281,7 +328,7 @@ function createPublication({ config, store, reads, catalog }) {
};
};
}
}
- return { abs, offerPath, productPath, storePath, sellerPath, categoryPath, freshness, decideOffer, decideProduct, offerView, cardView, offerDto, observationDto, offerLd, offerJsonLd, productView, productJsonLd, indexEntry, productIndexEntry };
+ return { abs, offerPath, productPath, storePath, sellerPath, categoryPath, freshness, decideOffer, decideProduct, offerView, offerViews, cardView, cardViews, offerDto, observationDto, offerLd, offerJsonLd, productView, productJsonLd, indexEntry, indexEntries, productIndexEntry };
}
}
module.exports = { createPublication };
module.exports = { createPublication };
@@ -31,6 +31,14 @@ function createReads({ store }) {
@@ -31,6 +31,14 @@ function createReads({ store }) {
WHERE offer_id IN (SELECT value FROM jsonb_array_elements_text(?::jsonb)) ORDER BY updated_at, seq`),
WHERE offer_id IN (SELECT value FROM jsonb_array_elements_text(?::jsonb)) ORDER BY updated_at, seq`),
members: db.prepare('SELECT * FROM deal_offers WHERE merged_into = ? ORDER BY merged_at'),
members: db.prepare('SELECT * FROM deal_offers WHERE merged_into = ? ORDER BY merged_at'),
lastSnapshot: db.prepare('SELECT * FROM deal_hotness_snapshots WHERE offer_id = ? ORDER BY id DESC LIMIT 1'),
lastSnapshot: db.prepare('SELECT * FROM deal_hotness_snapshots WHERE offer_id = ? ORDER BY id DESC LIMIT 1'),
+ // The batched forms: one statement for every offer of a page instead of one per offer.
+ latestByOffer: db.prepare(`SELECT DISTINCT ON (o.offer_id) o.*, s.kind AS source_kind, s.ref_service, s.ref_id, s.source_key, s.url AS source_url, s.label AS source_label
+ FROM deal_price_observations o JOIN deal_offer_sources s ON s.id = o.source_id
+ WHERE o.offer_id IN (SELECT value FROM jsonb_array_elements_text(?::jsonb)) AND s.removed_at IS NULL
+ ORDER BY o.offer_id, o.observed_at DESC, o.seq DESC`),
+ lastSnapshots: db.prepare(`SELECT DISTINCT ON (offer_id) * FROM deal_hotness_snapshots
+ WHERE offer_id IN (SELECT value FROM jsonb_array_elements_text(?::jsonb))
+ ORDER BY offer_id, id DESC`),
};
};
const get = async (id) => (id ? await q.byId.get(String(id)) : null);
const get = async (id) => (id ? await q.byId.get(String(id)) : null);
@@ -64,6 +72,32 @@ function createReads({ store }) {
@@ -64,6 +72,32 @@ function createReads({ store }) {
return bySubject;
return bySubject;
}
}
+ /**
+ * The latest observation of each offer in one query, then the group's newest of those (the old LIMIT 1 read, one
+ * query per offer). `lists` is one merge group per page entry; the answer is aligned with it.
+ */
+ const pickLatest = (ids, byOffer) => {
+ let best = null;
+ for (const id of ids) {
+ const o = byOffer.get(id);
+ if (o && (!best || o.observed_at > best.observed_at || (o.observed_at === best.observed_at && o.seq > best.seq))) best = o;
+ }
+ return best || null;
+ };
+ async function latestObservations(lists) {
+ const ids = [...new Set(lists.flat())];
+ if (!ids.length) return lists.map(() => null);
+ const byOffer = new Map((await q.latestByOffer.all(J(ids))).map((o) => [o.offer_id, o]));
+ return lists.map((list) => pickLatest(list, byOffer));
+ }
+
+ /** Each root's latest snapshot in one query, keyed by offer id. */
+ async function lastSnapshots(ids) {
+ const unique = [...new Set(ids)];
+ if (!unique.length) return new Map();
+ return new Map((await q.lastSnapshots.all(J(unique))).map((s) => [s.offer_id, s]));
+ }
+
async function tally(ids) {
async function tally(ids) {
let up = 0, down = 0, upW = 0, downW = 0;
let up = 0, down = 0, upW = 0, downW = 0;
for (const v of (await effectiveVotes(ids)).values()) {
for (const v of (await effectiveVotes(ids)).values()) {
@@ -80,12 +114,14 @@ function createReads({ store }) {
@@ -80,12 +114,14 @@ function createReads({ store }) {
root,
root,
groupIds,
groupIds,
observations: async (ids) => await q.observations.all(J(ids)),
observations: async (ids) => await q.observations.all(J(ids)),
- latestObservation: async (ids) => await q.latest.get(J(ids)) || null,
sources: async (ids) => await q.sources.all(J(ids)),
sources: async (ids) => await q.sources.all(J(ids)),
effectiveVotes,
effectiveVotes,
tally,
tally,
mergedMembers: async (id) => await q.members.all(id),
mergedMembers: async (id) => await q.members.all(id),
+ latestObservation: async (ids) => await q.latest.get(J(ids)) || null,
+ latestObservations,
lastSnapshot: async (id) => await q.lastSnapshot.get(id) || null,
lastSnapshot: async (id) => await q.lastSnapshot.get(id) || null,
+ lastSnapshots,
};
};
}
}
@@ -83,7 +83,7 @@ function createApi(ctx) {
@@ -83,7 +83,7 @@ function createApi(ctx) {
const page = Math.max(1, parseInt(req.query.page, 10) || 1);
const page = Math.max(1, parseInt(req.query.page, 10) || 1);
const per = 25;
const per = 25;
const rows = req.query.sort === 'new' ? await listings.newest(per, (page - 1) * per) : await listings.hot(per, (page - 1) * per);
const rows = req.query.sort === 'new' ? await listings.newest(per, (page - 1) * per) : await listings.hot(per, (page - 1) * per);
- const views = await Promise.all(rows.map(async (o) => publication.offerView(o)));
+ const views = await publication.offerViews(rows);
const visible = fromExtension(req) ? views.filter((v) => !v.extensionExcluded) : views;
const visible = fromExtension(req) ? views.filter((v) => !v.extensionExcluded) : views;
return { sort: req.query.sort === 'new' ? 'new' : 'hot', page, total: await listings.countActive(), offers: visible.map((v) => publication.offerDto(v)) };
return { sort: req.query.sort === 'new' ? 'new' : 'hot', page, total: await listings.countActive(), offers: visible.map((v) => publication.offerDto(v)) };
}));
}));
@@ -201,7 +201,7 @@ function createApi(ctx) {
@@ -201,7 +201,7 @@ function createApi(ctx) {
router.get('/stores/:domain', run(async (req) => {
router.get('/stores/:domain', run(async (req) => {
const st = await catalog.storeByDomain(req.params.domain);
const st = await catalog.storeByDomain(req.params.domain);
if (!st) throw new ApiError(404, 'store.not_found', 'No such store');
if (!st) throw new ApiError(404, 'store.not_found', 'No such store');
- const views = await Promise.all((await listings.byStore(st.id, 50, 0)).map(async (o) => await publication.offerView(o)));
+ const views = await publication.offerViews(await listings.byStore(st.id, 50, 0));
const offers = fromExtension(req) ? views.filter((v) => !v.extensionExcluded) : views;
const offers = fromExtension(req) ? views.filter((v) => !v.extensionExcluded) : views;
return { store: { id: st.id, domain: st.domain, name: st.name, url: publication.abs(publication.storePath(st)) }, offers: offers.map((v) => publication.offerDto(v)) };
return { store: { id: st.id, domain: st.domain, name: st.name, url: publication.abs(publication.storePath(st)) }, offers: offers.map((v) => publication.offerDto(v)) };
}));
}));
@@ -90,7 +90,7 @@ function createDiscoveryRoutes({ config, store, publication, listings }) {
@@ -90,7 +90,7 @@ function createDiscoveryRoutes({ config, store, publication, listings }) {
// and the cache is bounded by the listings' own 50,000-offer and 50,000-product caps.
// and the cache is bounded by the listings' own 50,000-offer and 50,000-product caps.
const ttl = Number(config.sitemapTtlMs) || 0;
const ttl = Number(config.sitemapTtlMs) || 0;
const offerEntries = shared(async () => {
const offerEntries = shared(async () => {
- const entries = await mapLimit(await listings.indexable(), VIEWS_PARALLEL, (o) => publication.indexEntry(o));
+ const entries = await publication.indexEntries(await listings.indexable());
let kept = 0;
let kept = 0;
for (const e of entries) {
for (const e of entries) {
if (e.decision.indexable && ++kept <= LLMS_SUMMARIES_MAX) continue;
if (e.decision.indexable && ++kept <= LLMS_SUMMARIES_MAX) continue;
@@ -142,7 +142,7 @@ function createPages(ctx) {
@@ -142,7 +142,7 @@ function createPages(ctx) {
const total = await L.count();
const total = await L.count();
const pager = ssr.paginate({ page, perPage: PER_PAGE, total, href: (p) => (p === 1 ? path : `${path}?page=${p}`) });
const pager = ssr.paginate({ page, perPage: PER_PAGE, total, href: (p) => (p === 1 ? path : `${path}?page=${p}`) });
if (pager.outOfRange && total) return null;
if (pager.outOfRange && total) return null;
- const list = await Promise.all((await L.rows(PER_PAGE, pager.offset)).map(async (o) => await publication.offerView(o)));
+ const list = await publication.offerViews(await L.rows(PER_PAGE, pager.offset));
const canonical = seo.canonicalUrl(config.baseUrl, pager.page === 1 ? path : `${path}?page=${pager.page}`, { query: ['page'] });
const canonical = seo.canonicalUrl(config.baseUrl, pager.page === 1 ? path : `${path}?page=${pager.page}`, { query: ['page'] });
return {
return {
agent: true,
agent: true,
@@ -177,9 +177,11 @@ function createPages(ctx) {
@@ -177,9 +177,11 @@ function createPages(ctx) {
listings.picks(HOME_CARDS * 2, 0), listings.upvoted(HOME_CARDS * 2), listings.newest(HOME_CARDS * 3, 0),
listings.picks(HOME_CARDS * 2, 0), listings.upvoted(HOME_CARDS * 2), listings.newest(HOME_CARDS * 3, 0),
listings.endingSoon(now, now + 7 * DAY, HOME_CARDS * 2, 0), listings.under(UNDER.amount, UNDER.currency, fresh, HOME_CARDS * 4),
listings.endingSoon(now, now + 7 * DAY, HOME_CARDS * 2, 0), listings.under(UNDER.amount, UNDER.currency, fresh, HOME_CARDS * 4),
]);
]);
- const viewOf = new Map();
- for (const o of [...pickRows, ...upRows, ...newRows, ...endRows, ...underRows]) if (!viewOf.has(o.id)) viewOf.set(o.id, publication.cardView(o, now));
- await Promise.all(viewOf.values());
+ const candidates = [];
+ const candidateIds = new Set();
+ for (const o of [...pickRows, ...upRows, ...newRows, ...endRows, ...underRows]) if (!candidateIds.has(o.id)) { candidateIds.add(o.id); candidates.push(o); }
+ const cards = await publication.cardViews(candidates, now);
+ const viewOf = new Map(candidates.map((o, i) => [o.id, cards[i]]));
const top = new Set();
const top = new Set();
const shown = new Set();
const shown = new Set();
const take = async (rows, { skip = shown, keep = () => true, into = [top, shown] } = {}) => {
const take = async (rows, { skip = shown, keep = () => true, into = [top, shown] } = {}) => {
@@ -449,7 +451,7 @@ function createPages(ctx) {
@@ -449,7 +451,7 @@ function createPages(ctx) {
const total = await listings.countStore(st.id);
const total = await listings.countStore(st.id);
const pager = ssr.paginate({ page, perPage: PER_PAGE, total, href: (p) => (p === 1 ? path : `${path}?page=${p}`) });
const pager = ssr.paginate({ page, perPage: PER_PAGE, total, href: (p) => (p === 1 ? path : `${path}?page=${p}`) });
if (pager.outOfRange && total) return null;
if (pager.outOfRange && total) return null;
- const list = await Promise.all((await listings.byStore(st.id, PER_PAGE, pager.offset)).map(async (o) => await publication.offerView(o)));
+ const list = await publication.offerViews(await listings.byStore(st.id, PER_PAGE, pager.offset));
const canonical = seo.canonicalUrl(config.baseUrl, pager.page === 1 ? path : `${path}?page=${pager.page}`, { query: ['page'] });
const canonical = seo.canonicalUrl(config.baseUrl, pager.page === 1 ? path : `${path}?page=${pager.page}`, { query: ['page'] });
return {
return {
agent: true,
agent: true,
@@ -491,7 +493,7 @@ function createPages(ctx) {
@@ -491,7 +493,7 @@ function createPages(ctx) {
const total = await listings.countSeller(hit.seller);
const total = await listings.countSeller(hit.seller);
const pager = ssr.paginate({ page, perPage: PER_PAGE, total, href: (p) => (p === 1 ? path : `${path}?page=${p}`) });
const pager = ssr.paginate({ page, perPage: PER_PAGE, total, href: (p) => (p === 1 ? path : `${path}?page=${p}`) });
if (pager.outOfRange && total) return null;
if (pager.outOfRange && total) return null;
- const list = await Promise.all((await listings.bySeller(hit.seller, PER_PAGE, pager.offset)).map(async (o) => await publication.offerView(o)));
+ const list = await publication.offerViews(await listings.bySeller(hit.seller, PER_PAGE, pager.offset));
const canonical = seo.canonicalUrl(config.baseUrl, pager.page === 1 ? path : `${path}?page=${pager.page}`, { query: ['page'] });
const canonical = seo.canonicalUrl(config.baseUrl, pager.page === 1 ? path : `${path}?page=${pager.page}`, { query: ['page'] });
return {
return {
agent: true,
agent: true,
@@ -529,7 +531,7 @@ function createPages(ctx) {
@@ -529,7 +531,7 @@ function createPages(ctx) {
const total = await listings.countCategory(hit.category);
const total = await listings.countCategory(hit.category);
const pager = ssr.paginate({ page, perPage: PER_PAGE, total, href: (p) => (p === 1 ? path : `${path}?page=${p}`) });
const pager = ssr.paginate({ page, perPage: PER_PAGE, total, href: (p) => (p === 1 ? path : `${path}?page=${p}`) });
if (pager.outOfRange && total) return null;
if (pager.outOfRange && total) return null;
- const list = await Promise.all((await listings.byCategory(hit.category, PER_PAGE, pager.offset)).map(async (o) => await publication.offerView(o)));
+ const list = await publication.offerViews(await listings.byCategory(hit.category, PER_PAGE, pager.offset));
const canonical = seo.canonicalUrl(config.baseUrl, pager.page === 1 ? path : `${path}?page=${pager.page}`, { query: ['page'] });
const canonical = seo.canonicalUrl(config.baseUrl, pager.page === 1 ? path : `${path}?page=${pager.page}`, { query: ['page'] });
return {
return {
agent: true,
agent: true,
@@ -556,7 +558,7 @@ function createPages(ctx) {
@@ -556,7 +558,7 @@ function createPages(ctx) {
send(req, res, 200, {
send(req, res, 200, {
title: q ? `Search: ${q}` : 'Search deals',
title: q ? `Search: ${q}` : 'Search deals',
decision: pageDecision('/search', { indexable: false }),
decision: pageDecision('/search', { indexable: false }),
- body: views.searchPage({ q, results: await Promise.all(r.rows.map(async (o) => await publication.offerView(o))), pager, urls, csrf: csrf(req), signedIn: signedIn(req) }),
+ body: views.searchPage({ q, results: await publication.offerViews(r.rows), pager, urls, csrf: csrf(req), signedIn: signedIn(req) }),
});
});
}));
}));
@@ -706,8 +708,10 @@ function createPages(ctx) {
@@ -706,8 +708,10 @@ function createPages(ctx) {
// ── feeds (never viewer-dependent) ──────────────────────
// ── feeds (never viewer-dependent) ──────────────────────
async function feedItems() {
async function feedItems() {
- return Promise.all((await listings.newest(50, 0)).map(async (o) => {
- const v = await publication.offerView(o);
+ const offers = await listings.newest(50, 0);
+ const pageViews = await publication.offerViews(offers);
+ return offers.map((o, i) => {
+ const v = pageViews[i];
const price = `${views.priceText(v.latest)}${v.latest ? ` as of ${iso(v.latest.observed_at)}` : ''}`;
const price = `${views.priceText(v.latest)}${v.latest ? ` as of ${iso(v.latest.observed_at)}` : ''}`;
// A source policy may require attribution: carried as text in the summary and as a tag.
// A source policy may require attribution: carried as text in the summary and as a tag.
const via = v.attribution ? `Deal via ${v.attribution.text}` : null;
const via = v.attribution ? `Deal via ${v.attribution.text}` : null;
@@ -717,7 +721,7 @@ function createPages(ctx) {
@@ -717,7 +721,7 @@ function createPages(ctx) {
tags: via ? [via] : undefined,
tags: via ? [via] : undefined,
published: o.created_at, updated: Math.max(o.updated_at, v.latest ? v.latest.observed_at : 0), decision: v.decision,
published: o.created_at, updated: Math.max(o.updated_at, v.latest ? v.latest.observed_at : 0), decision: v.decision,
};
};
- }));
+ });
}
}
const channel = () => ({ title: 'OpenVibe.Deals — new deals', link: publication.abs('/new'), description: 'New deals, each with its source and the time its price was observed.', language: 'en' });
const channel = () => ({ title: 'OpenVibe.Deals — new deals', link: publication.abs('/new'), description: 'New deals, each with its source and the time its price was observed.', language: 'en' });
new file mode 100644
new file mode 100644
@@ -0,0 +1,68 @@
@@ -0,0 +1,68 @@
+'use strict';
+/**
+ * A list of offers reads each offer's latest observation and latest hotness snapshot once, not once per offer.
+ *
+ * On 2026-11-08 the two per-offer reads were 2.1 M and 3.5 M calls a day (324 s and 251 s of DB time,
+ * pg_stat_statements, `reads.lastSnapshot` and `reads.latestObservation`), repeated by every page that renders
+ * offers. They are now one statement per turn: a page of six offers is one query of each, and each view still
+ * carries its own offer's latest observation and snapshot.
+ */
+const assert = require('assert');
+const { boot, check, done } = require('./helpers/boot');
+
+(async () => {
+ const t = await boot();
+ const db = t.ctx.store.db;
+ const poster = t.network.addUser('poster');
+
+ const offers = [];
+ for (let i = 0; i < 6; i++) {
+ offers.push(await t.submit(poster, { url: `https://batch.example/${i}`, title: `Batch offer ${i}`, price: String(10 + i), currency: 'USD' }));
+ }
+ // A snapshot of its own for each offer (snapshot() also writes the offer's hot_latest).
+ for (const [i, o] of offers.entries()) await t.ctx.hotness.snapshot(o.id, 'tick', t.clock.now() + i);
+
+ const seen = { latest: 0, snapshots: 0 };
+ const query = db.query.bind(db);
+ db.query = (text, values) => {
+ if (/DISTINCT ON \(o\.offer_id\)/.test(text)) seen.latest++;
+ else if (/DISTINCT ON \(offer_id\)/.test(text)) seen.snapshots++;
+ return query(text, values);
+ };
+ let views;
+ try {
+ views = await t.ctx.publication.offerViews(offers);
+ } finally {
+ db.query = query;
+ }
+
+ await check('a page of six offers reads latest observations and snapshots once each', async () => {
+ assert.strictEqual(seen.latest, 1, 'one latest-observation query for the page');
+ assert.strictEqual(seen.snapshots, 1, 'one snapshot query for the page');
+ });
+
+ await check('each view still carries its own latest observation and snapshot', async () => {
+ for (const [i, v] of views.entries()) {
+ assert.strictEqual(v.root.id, offers[i].id);
+ assert.strictEqual(v.latest.offer_id, offers[i].id);
+ assert.strictEqual(v.latest.price_num, 10 + i);
+ assert.strictEqual(v.hotness.offer_id, offers[i].id);
+ }
+ });
+
+ await check('the batched answer is the row the old one-per-offer query returned', async () => {
+ const id = offers[2].id;
+ const batched = await t.ctx.reads.latestObservation([id]);
+ const direct = await db.prepare(`SELECT o.*, s.kind AS source_kind, s.ref_service, s.ref_id, s.source_key, s.url AS source_url, s.label AS source_label
+ FROM deal_price_observations o JOIN deal_offer_sources s ON s.id = o.source_id
+ WHERE o.offer_id IN (SELECT value FROM jsonb_array_elements_text(?::jsonb)) AND s.removed_at IS NULL
+ ORDER BY o.observed_at DESC, o.seq DESC LIMIT 1`).get(JSON.stringify([id]));
+ assert.deepStrictEqual(batched, direct);
+ const snap = await t.ctx.reads.lastSnapshot(id);
+ const directSnap = await db.prepare('SELECT * FROM deal_hotness_snapshots WHERE offer_id = ? ORDER BY id DESC LIMIT 1').get(id);
+ assert.deepStrictEqual(snap, directSnap);
+ });
+
+ await t.close();
+ done();
+})();
@@ -31,6 +31,12 @@ function routes(ttlMs, clock, { offerCount = 20, products = [] } = {}) {
@@ -31,6 +31,12 @@ function routes(ttlMs, clock, { offerCount = 20, products = [] } = {}) {
seen.now--;
seen.now--;
return { loc: `https://deals.test/d/${o.slug}`, lastmod: o.updated_at, decision: indexable, title: o.slug, text: '1 USD' };
return { loc: `https://deals.test/d/${o.slug}`, lastmod: o.updated_at, decision: indexable, title: o.slug, text: '1 USD' };
},
},
+ async indexEntries(offers) {
+ // The real indexEntries prefetches per chunk and reads the page views at most eight at a time.
+ const out = new Array(offers.length);
+ for (let i = 0; i < offers.length; i += 8) await Promise.all(offers.slice(i, i + 8).map(async (o, j) => { out[i + j] = await publication.indexEntry(o); }));
+ return out;
+ },
async productIndexEntry(p) { return { loc: `https://deals.test/p/${p.slug}`, lastmod: p.updated_at, decision: indexable }; },
async productIndexEntry(p) { return { loc: `https://deals.test/p/${p.slug}`, lastmod: p.updated_at, decision: indexable }; },
};
};
const listings = {
const listings = {