spb/market-atlas

1import type { Observation } from "@market-atlas/market-model";2import { config } from "../config.js";3import { pool } from "../db/pool.js";4import { ensureObservationPartition } from "../db/migrate.js";5import { logger } from "../logger.js";6import { telemetry } from "./telemetry.js";78/**9 * Batched, idempotent observation writer. Bounded memory: when the queue exceeds the hard cap,10 * P2 tick observations are dropped (counted), never events or non-REALTIME values.11 */12export class ObservationWriter {13  private queue: Observation[] = [];14  private timer: NodeJS.Timeout | null = null;15  private flushing = false;16  private partitionsReady = new Set<string>();17  private seen = new Map<string, number>(); // fingerprint -> ts (dedupe replays for 10 min)18  readonly hardCap = 200_000;1920  enqueue(o: Observation): void {21    const fp = o.observationId;22    const now = Date.now();23    const prev = this.seen.get(fp);24    if (prev && now - prev < 600_000) {25      telemetry.inc("observations_duplicate_total");26      return;27    }28    this.seen.set(fp, now);29    if (this.seen.size > 500_000) this.pruneSeen(now);30    if (this.queue.length >= this.hardCap) {31      if (o.realtimeStatus === "REALTIME") {32        telemetry.inc("observations_dropped_total", 1, { reason: "backpressure" });33        return;34      }35    }36    this.queue.push(o);37    if (this.queue.length >= config.writerMaxBatch) void this.flush();38    else if (!this.timer) this.timer = setTimeout(() => void this.flush(), config.writerFlushMs);39  }4041  private pruneSeen(now: number) {42    for (const [k, t] of this.seen) if (now - t > 600_000) this.seen.delete(k);43  }4445  depth(): number {46    return this.queue.length;47  }4849  async flush(): Promise<void> {50    if (this.timer) {51      clearTimeout(this.timer);52      this.timer = null;53    }54    if (this.flushing || !this.queue.length) return;55    this.flushing = true;56    const batch = this.queue.splice(0, config.writerMaxBatch);57    const started = Date.now();58    try {59      const days = new Set(batch.map((o) => new Date(o.receivedAt).toISOString().slice(0, 10)));60      for (const d of days) {61        if (!this.partitionsReady.has(d)) {62          await ensureObservationPartition(Date.parse(d));63          this.partitionsReady.add(d);64        }65      }66      const cols = 18;67      const values: unknown[] = [];68      const tuples: string[] = [];69      batch.forEach((o, i) => {70        const b = i * cols;71        tuples.push(`(${Array.from({ length: cols }, (_, j) => `$${b + j + 1}`).join(",")})`);72        values.push(73          new Date(o.receivedAt).toISOString(),74          o.sourceTimestamp == null ? null : new Date(o.sourceTimestamp).toISOString(),75          o.instrumentId,76          o.field,77          o.value,78          o.currency ?? null,79          o.sourceId,80          o.connectorId,81          o.rightsStatus,82          o.realtimeStatus,83          o.timestampTrust,84          o.confidence ?? null,85          o.latencyMs,86          o.sequence == null ? null : String(o.sequence),87          o.observationId,88          o.rawRef,89          o.normalizerVersion,90          o.observationType ?? null,91        );92      });93      await pool.query(94        `insert into observations (received_at, source_ts, instrument_id, field, value, currency, source_id, connector_id, rights_status, realtime_status, timestamp_trust, confidence, latency_ms, sequence, fingerprint, raw_ref, normalizer_version, observation_type)95         values ${tuples.join(",")}`,96        values,97      );98      await pool.query(99        `insert into daily_stats (day, observations) values (current_date, $1)100         on conflict (day) do update set observations = daily_stats.observations + excluded.observations, updated_at = now()`,101        [batch.length],102      );103      telemetry.inc("observations_written_total", batch.length);104      telemetry.observe("db_insert_latency_ms", Date.now() - started);105    } catch (err) {106      telemetry.inc("observations_write_errors_total");107      logger.error({ err: err instanceof Error ? err.message : String(err), size: batch.length }, "observation batch insert failed");108      // Re-queue once at the front (bounded), so a transient DB hiccup does not lose canonical history.109      if (this.queue.length + batch.length <= this.hardCap) this.queue.unshift(...batch);110    } finally {111      this.flushing = false;112      telemetry.gauge("observation_queue_depth", this.queue.length);113      if (this.queue.length) this.timer = setTimeout(() => void this.flush(), config.writerFlushMs);114    }115  }116}117118export const observationWriter = new ObservationWriter();119