openvibers/openvibe.trade Commits

Trade: map the rates its real sources state (Frankfurter, ECB) and recover the skipped items (#43)

OP OpenVibers committed 11 Oct 2026 · c76485c3ff · 6 files, +320 −19

**Why openvibe.trade showed nothing.** Production's three sources hold 1241 items (SEC 1208, Frankfurter 28, ECB 5), all `healthy`, yet Trade stored nothing. The mapper understood only the explicit `fields.metric/value/unit/observed_at` contract (`server/domain/mapping.js`). The real Sources adapters emit something else:
- `frankfurter-eur-rates`: kind `rate`, with `fields.rate`, `base` and `date`, and the quote as the item identity;
- `ecb-euro-reference-rates`: an RSS 1.0 article whose title states the rate ("1.1206 USD = 1 EUR 2026-10-09 ECB Reference rate").

Every rate was skipped as `no_instrument_key`. Skips still advance the chassis cursor, so the items were read, dropped and left behind.

**Changes**
- `mapping.js`: an observation comes from the explicit contract, a structured `fields.rate` with base and quote, or a reference-rate title, always read exactly as stated. A bare date the source states is that day at 00:00 UTC.
- `ingest.js`: the one instrument the sync creates is a currency that a rate names: kind `currency`, the ISO code as symbol, its ISO 4217 English name ("US Dollar"), and no author. An ISO code needs no editorial naming. Equities and everything else still need an editor-created instrument, so the SEC filings keep waiting for editors to add instruments with each filer's CIK, as designed.
- `scripts/rewind-sync-cursor.js`: one-time recovery, dry run by default; `--apply` moves the cursor. It is safe to re-run, because re-reading is idempotent per item revision.
- README documents the rate shapes and the rewind.

**Test:** new `test/sources.test.js` mirrors the real adapter shapes of all three sources. It asserts the filing document, both rate observations, the USD instrument (named "US Dollar") and its aliases, that the instrument page shows the latest rate, that a cursor-reset replay creates nothing, and the rewind script's dry run and re-run. `sources` and `sync` pass.

**After the deploy (operator):** `node scripts/rewind-sync-cursor.js --apply` with Trade's env.

Investigated with a DeepSeek agent; reviewed, and the currency name was added.

Changes

ddiff --git a/.gitignore b/.gitignore
iindex 4c6243c..6313d87 100644
---- a/.gitignore
++++ b/.gitignore
@@@ -1,4 +1,4 @@
--node_modules/
++node_modules
data/
dist/
.env
ddiff --git a/README.md b/README.md
iindex f7927d6..517685c 100644
---- a/README.md
++++ b/README.md
@@@ -74,9 +74,14 @@ Other tables in the same database:
words normalised). Several instruments with the same name → `ambiguous`, candidates listed, none
chosen. No fuzzy or prefix matching. `/i/aapl` and a ticker alias answer 301 to `/i/AAPL`.
- **Observations.** Recorded by a first-party feed (`POST /api/v1/observations`,
-- `trade.observation.write`) or from Sources items that state a value (`fields.metric`, `value`,
-- `unit`, `observed_at`, a symbol or CIK). `observed_at` and `retrieved_at` are required and never
-- defaulted; a replay is `200 created:false`; the same reference with a different value is `409`.
++ `trade.observation.write`) or from Sources items that state a value: the explicit `fields.metric`,
++ `value`, `unit`, `observed_at` (a symbol or CIK), or a rate the real trade sources state — the api
++ adapter's `fields.rate` with `base`/`quote` (`frankfurter-eur-rates`), or a `<rate> <QUOTE> = 1
++ <BASE>` reference-rate title (`ecb-euro-reference-rates`, the value and currencies read as stated).
++ A bare date is that day at 00:00 UTC. `observed_at` and `retrieved_at` are required and never
++ defaulted; a replay is `200 created:false`; the same reference with a different value is `409`. A
++ rate's quote currency with no instrument yet registers one (kind `currency`, symbol/name the stated
++ ISO code); every other item must match an editor-created instrument or it is skipped and counted.
**No observation → no number:** the page says "No observations have been recorded for X, so no
number is shown."
- **Freshness.** Per source: fresh only while the last successful fetch is younger than the source's
@@@ -90,7 +95,10 @@ Other tables in the same database:
URL: CIK and accession come from the URL, the form type from the item summary (the feed's
`<description>`) when it looks like one, the filer from the title. A filing without a date says
"not stated by the source". Items that match no instrument are skipped and counted, never stored.
-- A Sources removal hides the document (the row stays as the record).
++ A Sources removal hides the document (the row stays as the record). If the sync ever advances the
++ cursor past items that were skipped and later fixed, rewind it with
++ `node scripts/rewind-sync-cursor.js` (dry run; `--apply` to move it) — re-reading is idempotent per
++ item revision.
- **Watchlists.** Private per person; plain forms and the API. Anyone else gets 404.
- **Alerts.** `threshold` rules fire on a crossing (armed → fired → re-armed when a newer
observation no longer meets the condition; late observations never fire; unit and currency must
ddiff --git a/scripts/rewind-sync-cursor.js b/scripts/rewind-sync-cursor.js
nnew file mode 100644
iindex 0000000..c0f42fe
---- /dev/null
++++ b/scripts/rewind-sync-cursor.js
@@@ -0,0 +1,81 @@
++#!/usr/bin/env node
++'use strict';
++
++/**
++ * scripts/rewind-sync-cursor.js — move Trade's Sources change cursor back so the sync re-reads items
++ * that were advanced past without being stored.
++ *
++ * The bug this recovers from: the mapper understood only the explicit `fields.metric/value/unit/
++ * observed_at` contract, so every item of Trade's real sources (SEC filings, frankfurter and ECB
++ * rates) was skipped. Skips still advance the cursor (openvibe-publishing/ingest `pullChanges` moves
++ * it past every committed page), so after the fix the cursor sits past the whole feed and the sync
++ * would read nothing. Rewinding makes the sync pull it again.
++ *
++ * Re-reading is safe: apply() is idempotent per (Sources item id, revision) — a document is
++ * upserted on its item id, an observation is keyed by (source_key, source_ref) and a replay answers
++ * `created:false` — so a rewind through already-stored items changes nothing and emits no duplicate
++ * events. The same script is therefore safe to re-run.
++ *
++ * DRY-RUN BY DEFAULT — it prints the current and target cursor and writes nothing. Pass --apply to
++ * move the cursor. The running service's next sync tick (TRADE_SYNC_INTERVAL_MS, default 60 s)
++ * picks up the rewound cursor; no restart is needed.
++ *
++ * Usage:
++ * node scripts/rewind-sync-cursor.js # dry run: show what would change
++ * node scripts/rewind-sync-cursor.js --apply # set the cursor to 0 (the start of the feed)
++ * node scripts/rewind-sync-cursor.js --to 1200 --apply # rewind to a specific change_seq
++ */
++
++const { load } = require('../server/config');
++const { openStore } = require('../server/db');
++
++const CURSOR = 'sources.trade'; // domain/ingest.js CURSOR — the Sources feed's row in trade_ingest_cursor
++
++/** Read a `--name <value>` option from argv (undefined when absent). */
++function option(name, argv = process.argv) {
++ const i = argv.indexOf(`--${name}`);
++ return i >= 0 ? argv[i + 1] : undefined;
++}
++
++async function rewind(store, { to, apply, log = console } = {}) {
++ const current = (await store.db.prepare('SELECT cursor FROM trade_ingest_cursor WHERE name = ?').get(CURSOR) || {}).cursor;
++ const now = store.now();
++ log.log(`[Trade rewind] cursor ${CURSOR}: ${current == null ? '(no row — the feed starts at 0)' : current} → ${to}`);
++ if (current === to) {
++ log.log('[Trade rewind] already at the target; nothing to do');
++ return { current: current == null ? null : current, to, changed: false };
++ }
++ if (!apply) {
++ log.log('[Trade rewind] dry run — not writing. Re-run with --apply to move the cursor.');
++ return { current: current == null ? null : current, to, changed: false, dryRun: true };
++ }
++ await store.db.prepare(`INSERT INTO trade_ingest_cursor (name, cursor, updated_at) VALUES (?, ?, ?)
++ ON CONFLICT (name) DO UPDATE SET cursor = excluded.cursor, updated_at = excluded.updated_at`).run(CURSOR, to, now);
++ log.log(`[Trade rewind] cursor moved to ${to}; the next sync tick re-reads from there (idempotent)`);
++ return { current: current == null ? null : current, to, changed: true };
++}
++
++async function main() {
++ const config = load();
++ const log = console;
++ const apply = process.argv.includes('--apply');
++ const toRaw = option('to');
++ const to = toRaw === undefined ? 0 : Math.floor(Number(toRaw));
++ if (!Number.isInteger(to) || to < 0) throw new Error('--to must be a non-negative integer');
++
++ const store = await openStore(config, { log });
++ try {
++ await rewind(store, { to, apply, log });
++ } finally {
++ await store.close();
++ }
++}
++
++if (require.main === module) {
++ main().catch((err) => {
++ console.error('[Trade rewind] failed:', err.message);
++ process.exit(1);
++ });
++}
++
++module.exports = { rewind, CURSOR };
ddiff --git a/server/domain/ingest.js b/server/domain/ingest.js
iindex 28f5221..6dfe7ae 100644
---- a/server/domain/ingest.js
++++ b/server/domain/ingest.js
@@@ -16,6 +16,10 @@
* filing / keyed item → a document (domain/documents.js; new → document alerts)
* no instrument matches → skipped (counted by reason; nothing is created for it)
*
++ * The one instrument the sync ever creates is a currency a rate states (frankfurter/ECB name the
++ * quote, e.g. USD): an ISO code needs no editorial naming, and a rate with no instrument to hang on
++ * would otherwise never appear. Every other item still requires an editor-created instrument.
++ *
* A failed call is recorded in trade_sync_state (last_error) and changes nothing else; freshness
* then decays on Trade's own clock, so pages show the data as stale instead of pretending.
*/
@@@ -66,12 +70,36 @@ function createIngest({ store, config, ctx, sources, log = console }) {
return null;
}
++ const CURRENCY_RE = /^[A-Z]{3}$/;
++ // The English name of an ISO 4217 code ("US Dollar"), from the runtime's ICU data; the code itself when unknown.
++ const currencyNames = new Intl.DisplayNames(['en'], { type: 'currency', fallback: 'code' });
++ const str1 = (v) => (v == null || v === '' ? null : (String(v).trim() || null));
++
++ /**
++ * The instrument a mapped item belongs to. A rate that states a currency the site does not track
++ * yet registers it (kind currency, symbol the stated ISO code, name its ISO 4217 English name, no author) — a code needs
++ * no editorial judgement and a rate with nowhere to hang never shows. Everything else must match
++ * an instrument an editor created.
++ */
++ async function instrumentFor(m) {
++ const found = await resolveKeys(m.keys);
++ if (found) return found;
++ const code = m.type === 'observation' && m.observation.metric === 'rate' ? str1(m.observation.currency) : null;
++ if (!code || !CURRENCY_RE.test(code) || code !== m.keys.symbol) return null;
++ try {
++ return await ctx.instruments.create({ symbol: code, name: currencyNames.of(code), kind: 'currency' }, null);
++ } catch (err) {
++ if (err && err.status === 409) return await resolveKeys(m.keys); // raced or alias holder
++ throw err;
++ }
++ }
++
/** Apply one item. → a counter name. Inside the page transaction. */
async function applyItem(item) {
if (item.removed) return await ctx.documents.remove(item.id, { at: parseTime(item.removed.at), reason: item.removed.reason }) ? 'documents_removed' : 'removed_unknown';
const m = mapItem(item);
if (m.type === 'skip') return `skipped_${m.reason}`;
-- const instrument = await resolveKeys(m.keys);
++ const instrument = await instrumentFor(m);
if (!instrument) return 'skipped_no_matching_instrument';
if (instrument.status !== 'active') return 'skipped_archived_instrument';
if (m.type === 'observation') {
ddiff --git a/server/domain/mapping.js b/server/domain/mapping.js
iindex 2572af2..4e0245a 100644
---- a/server/domain/mapping.js
++++ b/server/domain/mapping.js
@@@ -4,9 +4,16 @@
* From an OpenVibe.Sources item (sources.item@1, category trade) to what Trade records. Pure and
* deterministic; it reads only what the item states and never fills a gap.
*
-- * observation the item states a value: fields { metric, value, unit, observed_at } plus a
-- * symbol/ticker or CIK (currency, period, max_age_sec optional). An item without its
-- * own observation time is skipped — the retrieval time is NOT an observation time.
++ * observation the item states a value, in one of three shapes:
++ * - the explicit contract: fields { metric, value, unit, observed_at } plus a
++ * symbol/ticker or CIK (currency, period, max_age_sec optional);
++ * - a structured rate (the Sources api adapter, e.g. frankfurter-eur-rates):
++ * fields.rate with fields.base and a quote currency (fields.quote/target_currency,
++ * else the item's identity, which the adapter sets to the quote);
++ * - a reference-rate article (the ECB feed): a "<rate> <QUOTE> = 1 <BASE> …" title,
++ * the value and currencies read from that stated title.
++ * An item without its own observation time is skipped — the retrieval time is NOT an
++ * observation time. A bare date the source states (YYYY-MM-DD) is that day, 00:00 UTC.
* document an SEC EDGAR filing: the canonical URL is under
* sec.gov/Archives/edgar/data/<CIK>/<accession>/ (the xbrlrss feed's index links),
* so the CIK and accession number come from the URL; the form type is the item's
@@@ -16,10 +23,13 @@
* (kind 'record' for record items, else 'article').
* skip everything else, with a reason (counted in the sync state, never guessed at).
*/
--const { parseTime, parseDecimal } = require('./helpers');
++const { parseTime, parseDecimal, iso } = require('./helpers');
const SEC_PATH_RE = /^https?:\/\/(?:www\.)?sec\.gov\/Archives\/edgar\/data\/(\d{1,10})\/(\d{18})(?:\/|$)/i;
const FORM_RE = /^[A-Z0-9][A-Z0-9 \-/.]{0,19}$/;
++// "1.1206 USD = 1 EUR 2026-10-09 ECB Reference rate" → rate, quote, base, read as stated.
++const RATE_TITLE_RE = /^\s*(\d{1,20}(?:\.\d{1,12})?)\s+([A-Z]{3})\s*=\s*1\s+([A-Z]{3})(?:\s|$)/;
++const BARE_DATE_RE = /^\d{4}-\d{2}-\d{2}$/;
function formType(v) {
if (typeof v !== 'string') return null;
@@@ -37,6 +47,50 @@ function filerName(title) {
return s || null;
}
++/** A scalar as a trimmed string, or null. */
++function str1(v) { return v == null || v === '' ? null : (String(v).trim() || null); }
++
++/** A ticker-like identity ("USD"): upper-case it, else null (a URL or guid is not an instrument key). */
++function tickerish(v) {
++ return typeof v === 'string' && /^[A-Za-z][A-Za-z0-9.-]{0,15}$/.test(v.trim()) ? v.trim().toUpperCase() : null;
++}
++
++/** A time the source states: a full date-time with a zone, or a bare date (→ that day, 00:00 UTC). */
++function statedTime(v) {
++ const t = parseTime(v);
++ if (t != null) return t;
++ if (typeof v === 'string' && BARE_DATE_RE.test(v.trim())) {
++ const d = Date.parse(`${v.trim()}T00:00:00Z`);
++ return Number.isFinite(d) ? d : null;
++ }
++ return null;
++}
++
++/**
++ * The observation an item states, or null: the explicit contract, a structured `fields.rate`, or a
++ * reference-rate title. `symbol` is the instrument key the item itself names (the quote currency for
++ * a rate). Nothing here is invented — a missing field means this shape does not apply.
++ */
++function statedObservation(item, f) {
++ if (item.kind === 'observation' || (f.metric != null && f.value != null)) {
++ return { metric: f.metric, value: f.value, unit: f.unit, currency: f.currency == null ? null : f.currency,
++ period: f.period, observed_at: f.observed_at, max_age_sec: f.max_age_sec, symbol: null };
++ }
++ if (f.rate != null) {
++ const quote = str1(f.quote) || str1(f.target_currency) || str1(f.symbol) || str1(f.ticker) || tickerish(item.identity);
++ const base = str1(f.base) || str1(f.base_currency);
++ if (!quote || !base) return null;
++ return { metric: 'rate', value: f.rate, unit: quote, currency: quote, period: null,
++ observed_at: f.observed_at != null ? f.observed_at : f.date, max_age_sec: f.max_age_sec, symbol: quote };
++ }
++ const m = RATE_TITLE_RE.exec(String(item.title || ''));
++ if (m) {
++ return { metric: 'rate', value: m[1], unit: m[2], currency: m[2], period: null,
++ observed_at: f.observed_at != null ? f.observed_at : item.published_at, max_age_sec: f.max_age_sec, symbol: m[2] };
++ }
++ return null;
++}
++
/** → { type: 'observation', keys, observation } | { type: 'document', keys, document } | { type: 'skip', reason } */
function mapItem(item) {
if (!item || typeof item !== 'object' || !item.id) return { type: 'skip', reason: 'malformed_item' };
@@@ -49,17 +103,20 @@ function mapItem(item) {
const retrievedAt = parseTime(prov.retrieved_at);
if (retrievedAt == null) return { type: 'skip', reason: 'no_retrieval_time' };
-- if (item.kind === 'observation' || (f.metric != null && f.value != null)) {
-- if (!keys.cik && !keys.symbol) return { type: 'skip', reason: 'no_instrument_key' };
-- if (f.metric == null || f.unit == null || !parseDecimal(f.value)) return { type: 'skip', reason: 'incomplete_observation' };
-- if (parseTime(f.observed_at) == null) return { type: 'skip', reason: 'no_observation_time' };
++ const obs = statedObservation(item, f);
++ if (obs) {
++ const symbol = obs.symbol || keys.symbol;
++ if (!keys.cik && !symbol) return { type: 'skip', reason: 'no_instrument_key' };
++ if (obs.metric == null || obs.unit == null || !parseDecimal(obs.value)) return { type: 'skip', reason: 'incomplete_observation' };
++ const observedAt = statedTime(obs.observed_at);
++ if (observedAt == null) return { type: 'skip', reason: 'no_observation_time' };
return {
-- type: 'observation', keys,
++ type: 'observation', keys: { cik: keys.cik, symbol },
observation: {
-- metric: f.metric, value: f.value, unit: f.unit, currency: f.currency == null ? null : f.currency,
-- period: f.period == null ? null : String(f.period), observed_at: f.observed_at, retrieved_at: prov.retrieved_at,
++ metric: obs.metric, value: obs.value, unit: obs.unit, currency: obs.currency == null ? null : obs.currency,
++ period: obs.period == null ? null : String(obs.period), observed_at: iso(observedAt), retrieved_at: prov.retrieved_at,
source_key: item.source_key, source_item_id: item.id, source_url: item.canonical_url || null,
-- source_ref: `${item.id}#r${item.revision}:${f.metric}`, max_age_sec: f.max_age_sec == null ? null : f.max_age_sec,
++ source_ref: `${item.id}#r${item.revision}:${obs.metric}`, max_age_sec: obs.max_age_sec == null ? null : obs.max_age_sec,
},
};
}
@@@ -88,4 +145,4 @@ function mapItem(item) {
return { type: 'skip', reason: 'no_instrument_key' };
}
--module.exports = { mapItem, formType, filerName, accessionOf };
++module.exports = { mapItem, formType, filerName, accessionOf, statedTime };
ddiff --git a/test/sources.test.js b/test/sources.test.js
nnew file mode 100644
iindex 0000000..759be97
---- /dev/null
++++ b/test/sources.test.js
@@@ -0,0 +1,127 @@
++'use strict';
++/**
++ * The three sources OpenVibe.Sources actually holds for Trade, in the shape its adapters emit, become
++ * rows:
++ * sec-xbrl-filings (rss2) → a document for the instrument with the filing's CIK
++ * frankfurter-eur-rates (api) → an observation of the quote currency (kind 'rate', fields
++ * { rate, base, date }, identity = quote)
++ * ecb-euro-reference-rates (rss1)→ an observation, the rate read from the feed's stated title
++ *
++ * Nothing is dropped silently: a run stores all three and reports the counters. A replay (cursor
++ * reset) changes nothing.
++ */
++const assert = require('assert');
++const { boot, check, done } = require('./helpers/boot');
++const { rewind } = require('../scripts/rewind-sync-cursor');
++
++(async () => {
++ const t = await boot();
++ const ret = t.iso(t.T0 - 5 * 60e3);
++ const apple = await t.instrument({ symbol: 'AAPL', name: 'Apple Inc.', cik: '320193', exchange: 'NASDAQ' });
++
++ for (const [key, name] of [
++ ['sec-xbrl-filings', 'SEC EDGAR: latest XBRL filings'],
++ ['frankfurter-eur-rates', 'Frankfurter euro reference rates'],
++ ['ecb-euro-reference-rates', 'ECB euro reference rates: US dollar'],
++ ]) t.sources.setSource(key, { name, lastSuccessAt: ret, staleAfterSec: 86400 });
++
++ // sec-xbrl-filings: the feed adapter keeps the index link as canonical_url and the guid as identity.
++ const filing = t.sources.putItem({
++ source_key: 'sec-xbrl-filings', kind: 'article',
++ canonical_url: 'https://www.sec.gov/Archives/edgar/data/320193/000032019326000105/0000320193-26-000105-index.htm',
++ identity: 'https://www.sec.gov/Archives/edgar/data/320193/000032019326000105/0000320193-26-000105-xbrl.zip',
++ title: 'APPLE INC (0000320193) (Filer)', summary: '10-Q', published_at: '2026-09-22T11:30:00.000Z', retrieved_at: ret,
++ fields: { categories: [], enclosure: { url: 'https://www.sec.gov/Archives/edgar/data/320193/000032019326000105/0000320193-26-000105-xbrl.zip', type: 'application/zip', length: 18342 }, comments_url: null },
++ });
++ // frankfurter-eur-rates: the api adapter maps identity=quote and extra { rate, base, date }.
++ const frankfurter = t.sources.putItem({
++ source_key: 'frankfurter-eur-rates', kind: 'rate', identity: 'USD',
++ canonical_url: null, title: null, summary: null, published_at: null, retrieved_at: ret,
++ fields: { rate: 1.0831, base: 'EUR', date: '2026-09-21' },
++ });
++ // ecb-euro-reference-rates: an rss1 item whose title states the rate and dc:date the observation time.
++ const ecb = t.sources.putItem({
++ source_key: 'ecb-euro-reference-rates', kind: 'article',
++ identity: 'https://www.ecb.europa.eu/stats/exchange/eurofxref/html/eurofxref-graph-usd.en.html?date=2026-09-21&rate=1.1188',
++ canonical_url: 'https://www.ecb.europa.eu/stats/exchange/eurofxref/html/eurofxref-graph-usd.en.html?date=2026-09-21&rate=1.1188',
++ title: '1.1188 USD = 1 EUR 2026-09-21 ECB Reference rate',
++ summary: '1 EUR buys 1.1188 US dollar (USD) - The reference exchange rates are published shortly after the concertation procedure has been completed.',
++ published_at: '2026-09-21T14:15:00+01:00', retrieved_at: ret,
++ fields: { categories: ['EUR'], enclosure: null, comments_url: null },
++ });
++
++ await check('a run stores the filing, both rates and the currency they name; nothing is skipped', async () => {
++ const r = await t.ctx.sync.run();
++ assert.strictEqual(r.ok, true, JSON.stringify(r));
++ assert.strictEqual(r.counts.documents_created, 1, 'the filing');
++ assert.strictEqual(r.counts.observations_created, 2, 'frankfurter and ECB');
++ assert.ok(!Object.keys(r.counts).some((k) => k.startsWith('skipped_')), `no skips: ${JSON.stringify(r.counts)}`);
++
++ const doc = await t.ctx.store.db.prepare('SELECT * FROM trade_source_documents WHERE source_item_id = ?').get(filing.id);
++ assert.strictEqual(doc.instrument_id, apple.id);
++ assert.strictEqual(doc.kind, 'filing');
++ assert.strictEqual(doc.form_type, '10-Q');
++ assert.strictEqual(doc.cik, '0000320193');
++ assert.strictEqual(doc.accession, '0000320193-26-000105');
++
++ const usd = await t.ctx.instruments.bySymbol('USD');
++ assert.ok(usd, 'the quote currency the rates name is registered');
++ assert.strictEqual(usd.name, 'US Dollar', 'named by its ISO 4217 English name');
++ assert.strictEqual(usd.kind, 'currency');
++
++ const obs = (await t.ctx.store.db.prepare('SELECT * FROM trade_market_observations ORDER BY source_key').all())
++ .map((o) => ({ source_key: o.source_key, metric: o.metric, value: o.value, unit: o.unit, currency: o.currency, observed_at: t.iso(o.observed_at), item: o.source_item_id, instrument: o.instrument_id }));
++ assert.deepStrictEqual(obs, [
++ { source_key: 'ecb-euro-reference-rates', metric: 'rate', value: '1.1188', unit: 'USD', currency: 'USD', observed_at: '2026-09-21T13:15:00.000Z', item: ecb.id, instrument: usd.id },
++ { source_key: 'frankfurter-eur-rates', metric: 'rate', value: '1.0831', unit: 'USD', currency: 'USD', observed_at: '2026-09-21T00:00:00.000Z', item: frankfurter.id, instrument: usd.id },
++ ]);
++
++ const n = (await t.ctx.store.db.prepare('SELECT COUNT(*) AS n FROM trade_instrument_aliases').get()).n;
++ assert.ok(n >= 5, `AAPL ticker/cik/name and USD ticker/name are aliases: ${n}`);
++ });
++
++ await check('the instrument page shows the latest rate of the series, with its time and source', async () => {
++ // Both rates are the same (metric, unit, currency) series, so the page shows the newer one;
++ // both rows stay in the table (asserted above).
++ const json = (await t.get('/i/USD.json')).json();
++ assert.strictEqual(json.observations.length, 1);
++ assert.deepStrictEqual([json.observations[0].value, json.observations[0].unit, json.observations[0].observed_at, json.observations[0].source.key],
++ ['1.1188', 'USD', '2026-09-21T13:15:00.000Z', 'ecb-euro-reference-rates']);
++ });
++
++ await check('replaying the same Sources pages (cursor reset) creates nothing new', async () => {
++ const docs = (await t.ctx.store.db.prepare('SELECT COUNT(*) AS n FROM trade_source_documents').get()).n;
++ const obs = (await t.ctx.store.db.prepare('SELECT COUNT(*) AS n FROM trade_market_observations').get()).n;
++ const instruments = (await t.ctx.store.db.prepare('SELECT COUNT(*) AS n FROM trade_instruments').get()).n;
++ const events = (await t.events()).length;
++ await t.ctx.store.db.prepare("UPDATE trade_ingest_cursor SET cursor = 0 WHERE name = 'sources.trade'").run();
++ const r = await t.ctx.sync.run();
++ assert.strictEqual(r.ok, true);
++ assert.strictEqual(r.counts.observations_created, undefined, 'no new observations');
++ assert.strictEqual((await t.ctx.store.db.prepare('SELECT COUNT(*) AS n FROM trade_source_documents').get()).n, docs);
++ assert.strictEqual((await t.ctx.store.db.prepare('SELECT COUNT(*) AS n FROM trade_market_observations').get()).n, obs);
++ assert.strictEqual((await t.ctx.store.db.prepare('SELECT COUNT(*) AS n FROM trade_instruments').get()).n, instruments, 'no duplicate currency instrument');
++ assert.strictEqual((await t.events()).length, events, 'no new events');
++ });
++
++ await check('the rewind script is a dry run unless --apply, and re-running it is a no-op', async () => {
++ const cursor = async () => (await t.ctx.store.db.prepare("SELECT cursor FROM trade_ingest_cursor WHERE name = 'sources.trade'").get()).cursor;
++ const quiet = { log() {}, warn() {} };
++ await t.ctx.store.db.prepare("UPDATE trade_ingest_cursor SET cursor = 1241 WHERE name = 'sources.trade'").run();
++ const before = await cursor();
++ assert.strictEqual(before, 1241);
++ const dry = await rewind(t.ctx.store, { to: 0, apply: false, log: quiet });
++ assert.strictEqual(dry.changed, false);
++ assert.strictEqual(dry.dryRun, true);
++ assert.strictEqual(await cursor(), before, 'dry run changes nothing');
++ const applied = await rewind(t.ctx.store, { to: 0, apply: true, log: quiet });
++ assert.strictEqual(applied.changed, true);
++ assert.strictEqual(await cursor(), 0);
++ const again = await rewind(t.ctx.store, { to: 0, apply: true, log: quiet });
++ assert.strictEqual(again.changed, false, 'already at the target');
++ assert.strictEqual(await cursor(), 0);
++ });
++
++ await t.close();
++ done();
++})();