import type { CanonicalQuote } from "@market-atlas/market-model";
import { pool, tsToMs } from "../db/pool.js";
import { logger } from "../logger.js";
import { bus } from "./bus.js";
import { telemetry } from "./telemetry.js";

/**
 * Current-state cache of canonical quotes with debounced persistence (one upsert per instrument
 * per second at most). The bus receives every recomputation; the DB receives the settled value.
 */
export class QuoteStore {
  private quotes = new Map<string, CanonicalQuote>();
  private dirty = new Map<string, CanonicalQuote>();
  private timer: NodeJS.Timeout | null = null;
  private flushing = false;

  async load(): Promise<void> {
    const r = await pool.query("select * from canonical_quotes");
    for (const row of r.rows) this.quotes.set(row.instrument_id, rowToQuote(row));
    logger.info({ quotes: this.quotes.size }, "canonical quotes loaded");
  }

  get(id: string): CanonicalQuote | undefined {
    return this.quotes.get(id);
  }

  all(): CanonicalQuote[] {
    return [...this.quotes.values()];
  }

  size() {
    return this.quotes.size;
  }

  update(q: CanonicalQuote): void {
    this.quotes.set(q.instrumentId, q);
    this.dirty.set(q.instrumentId, q);
    bus.publish("canonical.quote", q);
    telemetry.inc("canonical_quotes_total");
    if (!this.timer) this.timer = setTimeout(() => void this.flush(), 1000);
  }

  async flush(): Promise<void> {
    this.timer = null;
    if (this.flushing) {
      this.timer = setTimeout(() => void this.flush(), 500);
      return;
    }
    if (!this.dirty.size) return;
    this.flushing = true;
    const batch = [...this.dirty.values()].sort((a, b) => a.instrumentId.localeCompare(b.instrumentId));
    this.dirty.clear();
    const cols = 29;
    const tuples: string[] = [];
    const values: unknown[] = [];
    batch.forEach((q, i) => {
      const b = i * cols;
      tuples.push(`(${Array.from({ length: cols }, (_, j) => `$${b + j + 1}`).join(",")})`);
      values.push(
        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,
        q.sourceCount, q.dispersionBps, q.confidence, q.freshnessMs, q.realtimeStatus, q.rightsStatus, new Date(q.updatedAt).toISOString(),
        q.sourceTimestamp == null ? null : new Date(q.sourceTimestamp).toISOString(), q.sessionHigh, q.sessionLow,
        JSON.stringify(q.contributions.slice(0, 20)), q.consensusVersion,
        q.observationCount, q.proxyCount, q.validatorCount, q.comparability,
      );
    });
    try {
      await pool.query(
        `insert into canonical_quotes (instrument_id, symbol, price, open, high, low, previous_close, change, change_percent, volume, bid, ask, currency,
           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)
         values ${tuples.join(",")}
         on conflict (instrument_id) do update set symbol = excluded.symbol, price = excluded.price, open = excluded.open, high = excluded.high, low = excluded.low,
           previous_close = excluded.previous_close, change = excluded.change, change_percent = excluded.change_percent, volume = excluded.volume, bid = excluded.bid, ask = excluded.ask,
           currency = excluded.currency, source_count = excluded.source_count, dispersion_bps = excluded.dispersion_bps, confidence = excluded.confidence, freshness_ms = excluded.freshness_ms,
           realtime_status = excluded.realtime_status, rights_status = excluded.rights_status, updated_at = excluded.updated_at, source_ts = excluded.source_ts,
           session_high = excluded.session_high, session_low = excluded.session_low, contributions = excluded.contributions, consensus_version = excluded.consensus_version,
           observation_count = excluded.observation_count, proxy_count = excluded.proxy_count, validator_count = excluded.validator_count, comparability = excluded.comparability`,
        values,
      );
    } catch (err) {
      logger.error({ err, size: batch.length }, "canonical quote upsert failed");
      for (const q of batch) if (!this.dirty.has(q.instrumentId)) this.dirty.set(q.instrumentId, q);
      this.timer = setTimeout(() => void this.flush(), 5000);
    } finally {
      this.flushing = false;
    }
  }
}

export function rowToQuote(row: any): CanonicalQuote {
  return {
    instrumentId: row.instrument_id,
    symbol: row.symbol,
    price: row.price,
    open: row.open,
    high: row.high,
    low: row.low,
    previousClose: row.previous_close,
    change: row.change,
    changePercent: row.change_percent,
    volume: row.volume,
    bid: row.bid,
    ask: row.ask,
    currency: row.currency,
    sourceCount: row.source_count,
    observationCount: row.observation_count ?? 0,
    proxyCount: row.proxy_count ?? 0,
    validatorCount: row.validator_count ?? 0,
    comparability: row.comparability ?? null,
    dispersionBps: row.dispersion_bps,
    confidence: row.confidence,
    freshnessMs: row.freshness_ms,
    realtimeStatus: row.realtime_status,
    rightsStatus: row.rights_status,
    updatedAt: tsToMs(row.updated_at) ?? 0,
    sourceTimestamp: tsToMs(row.source_ts),
    sessionHigh: row.session_high,
    sessionLow: row.session_low,
    contributions: row.contributions ?? [],
    consensusVersion: row.consensus_version,
  };
}

export const quoteStore = new QuoteStore();