Importer: deals.offer.updated names every field an observation moved (expires_at, product_id, title, status), not just 'observation', so consumers never miss a changed field

OP OpenVibers committed 24 Sept 2026 · 24d52a2674 · 1 file, +6 −1

Changes

ddiff --git a/server/domain/importer.js b/server/domain/importer.js
iindex 91c001e..581e5b1 100644
---- a/server/domain/importer.js
++++ b/server/domain/importer.js
@@@ -145,10 +145,15 @@ function createImporter({ config, store, reads, catalog, offers, indexing, sourc
                  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 = offers.recordObservation(existing.offer_id, existing.id, c.obs, { origin, observedAt: retrievedAt, sourceRevision: item.revision });
                  const offer = 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) {
                      db.prepare('UPDATE deal_offers SET expires_at = ?, updated_at = ? WHERE id = ?').run(c.expires_at, store.now(), offer.id);
++                    changed.push('expires_at');
                  }
--                indexing.emitOffer('deals.offer.updated', reads.get(offer.id), { actor: 'svc:deals', extra: { changed: ['observation'], observation_id: obs.id } });
++                // Anything else the observation moved on the offer (price, product) is named too, so consumers never miss a field.
++                const after = 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);
++                indexing.emitOffer('deals.offer.updated', after, { actor: 'svc:deals', extra: { changed, observation_id: obs.id } });
                  indexing.reindex(reads.get(offer.id));
                  outcomes.push(origin === 'import' ? 'updated' : 'refreshed');
                  continue;

diff --git a/server/domain/importer.js b/server/domain/importer.js

index 91c001e..581e5b1 100644

--- a/server/domain/importer.js

+++ b/server/domain/importer.js

@@ -145,10 +145,15 @@ function createImporter({ config, store, reads, catalog, offers, indexing, sourc

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 = offers.recordObservation(existing.offer_id, existing.id, c.obs, { origin, observedAt: retrievedAt, sourceRevision: item.revision });

const offer = 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) {

db.prepare('UPDATE deal_offers SET expires_at = ?, updated_at = ? WHERE id = ?').run(c.expires_at, store.now(), offer.id);

+ changed.push('expires_at');

}

- indexing.emitOffer('deals.offer.updated', reads.get(offer.id), { actor: 'svc:deals', extra: { changed: ['observation'], observation_id: obs.id } });

+ // Anything else the observation moved on the offer (price, product) is named too, so consumers never miss a field.

+ const after = 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);

+ indexing.emitOffer('deals.offer.updated', after, { actor: 'svc:deals', extra: { changed, observation_id: obs.id } });

indexing.reindex(reads.get(offer.id));

outcomes.push(origin === 'import' ? 'updated' : 'refreshed');

continue;