spb/market-atlas

1import type { CanonicalQuote } from "@market-atlas/market-model";2import { pool, tsToMs } from "../db/pool.js";3import { logger } from "../logger.js";4import { bus } from "./bus.js";5import { telemetry } from "./telemetry.js";67/**8 * Current-state cache of canonical quotes with debounced persistence (one upsert per instrument9 * per second at most). The bus receives every recomputation; the DB receives the settled value.10 */11export class QuoteStore {12  private quotes = new Map<string, CanonicalQuote>();13  private dirty = new Map<string, CanonicalQuote>();14  private timer: NodeJS.Timeout | null = null;15  private flushing = false;1617  async load(): Promise<void> {18    const r = await pool.query("select * from canonical_quotes");19    for (const row of r.rows) this.quotes.set(row.instrument_id, rowToQuote(row));20    logger.info({ quotes: this.quotes.size }, "canonical quotes loaded");21  }2223  get(id: string): CanonicalQuote | undefined {24    return this.quotes.get(id);25  }2627  all(): CanonicalQuote[] {28    return [...this.quotes.values()];29  }3031  size() {32    return this.quotes.size;33  }3435  update(q: CanonicalQuote): void {36    this.quotes.set(q.instrumentId, q);37    this.dirty.set(q.instrumentId, q);38    bus.publish("canonical.quote", q);39    telemetry.inc("canonical_quotes_total");40    if (!this.timer) this.timer = setTimeout(() => void this.flush(), 1000);41  }4243  async flush(): Promise<void> {44    this.timer = null;45    if (this.flushing) {46      this.timer = setTimeout(() => void this.flush(), 500);47      return;48    }49    if (!this.dirty.size) return;50    this.flushing = true;51    const batch = [...this.dirty.values()].sort((a, b) => a.instrumentId.localeCompare(b.instrumentId));52    this.dirty.clear();53    const cols = 29;54    const tuples: string[] = [];55    const values: unknown[] = [];56    batch.forEach((q, i) => {57      const b = i * cols;58      tuples.push(`(${Array.from({ length: cols }, (_, j) => `$${b + j + 1}`).join(",")})`);59      values.push(60        q.instrumentId, q.symbol, q.price, q.open, q.high, q.low, q.previousClose, q.change, q.changePercent, q.volume, q.bid, q.ask, q.currency,61        q.sourceCount, q.dispersionBps, q.confidence, q.freshnessMs, q.realtimeStatus, q.rightsStatus, new Date(q.updatedAt).toISOString(),62        q.sourceTimestamp == null ? null : new Date(q.sourceTimestamp).toISOString(), q.sessionHigh, q.sessionLow,63        JSON.stringify(q.contributions.slice(0, 20)), q.consensusVersion,64        q.observationCount, q.proxyCount, q.validatorCount, q.comparability,65      );66    });67    try {68      await pool.query(69        `insert into canonical_quotes (instrument_id, symbol, price, open, high, low, previous_close, change, change_percent, volume, bid, ask, currency,70           source_count, dispersion_bps, confidence, freshness_ms, realtime_status, rights_status, updated_at, source_ts, session_high, session_low, contributions, consensus_version, observation_count, proxy_count, validator_count, comparability)71         values ${tuples.join(",")}72         on conflict (instrument_id) do update set symbol = excluded.symbol, price = excluded.price, open = excluded.open, high = excluded.high, low = excluded.low,73           previous_close = excluded.previous_close, change = excluded.change, change_percent = excluded.change_percent, volume = excluded.volume, bid = excluded.bid, ask = excluded.ask,74           currency = excluded.currency, source_count = excluded.source_count, dispersion_bps = excluded.dispersion_bps, confidence = excluded.confidence, freshness_ms = excluded.freshness_ms,75           realtime_status = excluded.realtime_status, rights_status = excluded.rights_status, updated_at = excluded.updated_at, source_ts = excluded.source_ts,76           session_high = excluded.session_high, session_low = excluded.session_low, contributions = excluded.contributions, consensus_version = excluded.consensus_version,77           observation_count = excluded.observation_count, proxy_count = excluded.proxy_count, validator_count = excluded.validator_count, comparability = excluded.comparability`,78        values,79      );80    } catch (err) {81      logger.error({ err, size: batch.length }, "canonical quote upsert failed");82      for (const q of batch) if (!this.dirty.has(q.instrumentId)) this.dirty.set(q.instrumentId, q);83      this.timer = setTimeout(() => void this.flush(), 5000);84    } finally {85      this.flushing = false;86    }87  }88}8990export function rowToQuote(row: any): CanonicalQuote {91  return {92    instrumentId: row.instrument_id,93    symbol: row.symbol,94    price: row.price,95    open: row.open,96    high: row.high,97    low: row.low,98    previousClose: row.previous_close,99    change: row.change,100    changePercent: row.change_percent,101    volume: row.volume,102    bid: row.bid,103    ask: row.ask,104    currency: row.currency,105    sourceCount: row.source_count,106    observationCount: row.observation_count ?? 0,107    proxyCount: row.proxy_count ?? 0,108    validatorCount: row.validator_count ?? 0,109    comparability: row.comparability ?? null,110    dispersionBps: row.dispersion_bps,111    confidence: row.confidence,112    freshnessMs: row.freshness_ms,113    realtimeStatus: row.realtime_status,114    rightsStatus: row.rights_status,115    updatedAt: tsToMs(row.updated_at) ?? 0,116    sourceTimestamp: tsToMs(row.source_ts),117    sessionHigh: row.session_high,118    sessionLow: row.session_low,119    contributions: row.contributions ?? [],120    consensusVersion: row.consensus_version,121  };122}123124export const quoteStore = new QuoteStore();125