import type { CanonicalQuote, EventType, MarketEvent, Severity } from "@market-atlas/market-model";
import { makeEventId } from "@market-atlas/market-model";
import { shortHash } from "@market-atlas/connector-sdk";
import type { ProposedEvent } from "@market-atlas/connector-sdk";
import { pool } from "../db/pool.js";
import { logger } from "../logger.js";
import { bus } from "./bus.js";
import { telemetry } from "./telemetry.js";

interface InstrumentStats {
  lastPrice: number | null;
  lastEventPrice: number | null;
  sessionHigh: number;
  sessionLow: number;
  /** EWMA of absolute 1-tick returns for volatility baselines. */
  ewmaAbsRet: number;
  ewmaVar: number;
  ticks: number;
  lastEventAt: Record<string, number>;
  lastSourceDivergenceAt: number;
  recentReturns: number[];
}

/**
 * Event engine: turns canonical quotes into meaningful, deduplicated events and canonicalizes
 * connector-proposed events (filings, halts…). Thresholds are relative to per-instrument baselines,
 * so a 0.5% move in a treasury yield is not treated like 0.5% on a crypto pair.
 */
export class EventEngine {
  private stats = new Map<string, InstrumentStats>();
  private recentFingerprints = new Map<string, number>();
  private pending: MarketEvent[] = [];
  private timer: NodeJS_Timeout | null = null;

  private stat(id: string): InstrumentStats {
    let s = this.stats.get(id);
    if (!s) {
      s = { lastPrice: null, lastEventPrice: null, sessionHigh: -Infinity, sessionLow: Infinity, ewmaAbsRet: 0, ewmaVar: 0, ticks: 0, lastEventAt: {}, lastSourceDivergenceAt: 0, recentReturns: [] };
      this.stats.set(id, s);
    }
    return s;
  }

  resetSession(instrumentId: string) {
    const s = this.stat(instrumentId);
    s.sessionHigh = -Infinity;
    s.sessionLow = Infinity;
  }

  onQuote(q: CanonicalQuote, name: string): void {
    if (q.price == null || q.realtimeStatus === "STALE") return;
    const s = this.stat(q.instrumentId);
    const now = q.updatedAt;
    const p = q.price;
    if (s.lastPrice != null && s.lastPrice !== 0) {
      const ret = (p - s.lastPrice) / s.lastPrice;
      const absRet = Math.abs(ret);
      const alpha = 0.05;
      s.ewmaAbsRet = s.ticks < 20 ? (s.ewmaAbsRet * s.ticks + absRet) / (s.ticks + 1) : s.ewmaAbsRet * (1 - alpha) + absRet * alpha;
      s.ewmaVar = s.ticks < 20 ? (s.ewmaVar * s.ticks + ret * ret) / (s.ticks + 1) : s.ewmaVar * (1 - alpha) + ret * ret * alpha;
      s.recentReturns.push(ret);
      if (s.recentReturns.length > 60) s.recentReturns.shift();
      // Volatility spike: realized volatility over the last 30 ticks vs long-run baseline (needs history).
      if (s.ticks > 120 && s.recentReturns.length >= 30) {
        const recent = Math.sqrt(s.recentReturns.slice(-30).reduce((a, r) => a + r * r, 0) / 30);
        const baseline = Math.sqrt(s.ewmaVar);
        if (baseline > 0 && recent > 4 * baseline && this.cooldown(s, "VOLATILITY_SPIKE", now, 10 * 60_000)) {
          this.emitDerived("VOLATILITY_SPIKE", q, name, now, "WARNING", `${q.symbol} volatility ${(recent / baseline).toFixed(1)}× baseline`, {
            ratio: round(recent / baseline, 2),
            realized: recent,
            baseline,
          });
        }
      }
    }
    // Price change events on a per-instrument adaptive threshold (≥ 0.5% and ≥ 8× typical tick move).
    const ref = s.lastEventPrice ?? s.lastPrice;
    if (ref != null && ref !== 0 && s.ticks > 60) {
      const move = (p - ref) / ref;
      const threshold = Math.max(0.01, 12 * s.ewmaAbsRet);
      if (Math.abs(move) >= threshold && this.cooldown(s, "PRICE_CHANGE", now, 5 * 60_000)) {
        this.emitDerived("PRICE_CHANGE", q, name, now, Math.abs(move) >= 0.03 ? "WARNING" : "INFO", `${q.symbol} ${move >= 0 ? "+" : ""}${(move * 100).toFixed(2)}%`, {
          from: ref,
          to: p,
          changePercent: round(move * 100, 3),
        });
        s.lastEventPrice = p;
      }
    }
    if (s.lastEventPrice == null) s.lastEventPrice = p;
    // Session highs/lows: only after meaningful history, when the previous extreme is beaten by ≥ 0.1 %, with a 15-minute cooldown.
    if (s.ticks > 200) {
      if (p > s.sessionHigh * 1.001 && Number.isFinite(s.sessionHigh) && this.cooldown(s, "SESSION_HIGH", now, 15 * 60_000)) {
        this.emitDerived("SESSION_HIGH", q, name, now, "INFO", `${q.symbol} new session high ${fmt(p)}`, { price: p, previousHigh: s.sessionHigh });
      }
      if (p < s.sessionLow * 0.999 && Number.isFinite(s.sessionLow) && this.cooldown(s, "SESSION_LOW", now, 15 * 60_000)) {
        this.emitDerived("SESSION_LOW", q, name, now, "INFO", `${q.symbol} new session low ${fmt(p)}`, { price: p, previousLow: s.sessionLow });
      }
    }
    if (p > s.sessionHigh) s.sessionHigh = p;
    if (p < s.sessionLow) s.sessionLow = p;
    // Source divergence: included sources disagree by more than 50 bps (crypto venues legitimately differ; flag ≥ 50).
    if (q.dispersionBps != null && q.dispersionBps > 100 && q.sourceCount >= 2 && now - s.lastSourceDivergenceAt > 60 * 60_000) {
      s.lastSourceDivergenceAt = now;
      this.emitDerived("SOURCE_DIVERGENCE", q, name, now, "NOTICE", `${q.symbol}: sources diverge by ${q.dispersionBps.toFixed(0)} bps`, {
        dispersionBps: q.dispersionBps,
        contributions: q.contributions.filter((c) => c.included).map((c) => ({ source: c.sourceId, value: c.value, ageMs: c.ageMs })),
      });
    }
    s.lastPrice = p;
    s.ticks++;
  }

  private cooldown(s: InstrumentStats, type: string, now: number, ms: number): boolean {
    const last = s.lastEventAt[type] ?? 0;
    if (now - last < ms) return false;
    s.lastEventAt[type] = now;
    return true;
  }

  private emitDerived(type: EventType, q: CanonicalQuote, name: string, ts: number, severity: Severity, title: string, data: Record<string, unknown>) {
    const minute = Math.floor(ts / 60_000);
    const fingerprint = shortHash(`${type}|${q.instrumentId}|${minute}|${JSON.stringify(data)}`, 24);
    this.publish({
      version: 1,
      id: makeEventId(ts, fingerprint),
      type,
      instrumentIds: [q.instrumentId],
      entityIds: [],
      timestamp: ts,
      severity,
      confidence: q.confidence,
      sourceCount: q.sourceCount,
      sources: q.contributions.filter((c) => c.included).map((c) => c.sourceId),
      title,
      summary: `${name} · ${q.sourceCount} independent source${q.sourceCount === 1 ? "" : "s"} · confidence ${(q.confidence * 100).toFixed(1)}%`,
      data: { ...data, symbol: q.symbol, price: q.price, currency: q.currency },
      fingerprint,
      supportingObservations: [],
    });
  }

  /** Canonicalize an event proposed by a connector or the runtime (dedupe on dedupeKey within a 24 h window). */
  proposed(e: ProposedEvent, sourceId: string, instrumentIds: string[]): MarketEvent | null {
    const fingerprint = shortHash(`${e.type}|${e.dedupeKey}`, 24);
    const ev: MarketEvent = {
      version: 1,
      id: makeEventId(e.timestamp, fingerprint),
      type: e.type,
      instrumentIds,
      entityIds: e.entityIds ?? [],
      timestamp: e.timestamp,
      severity: e.severity ?? "INFO",
      confidence: e.confidence ?? 0.9,
      sourceCount: 1,
      sources: [sourceId],
      title: e.title,
      summary: e.summary ?? null,
      data: e.data ?? {},
      fingerprint,
      supportingObservations: [],
    };
    return this.publish(ev) ? ev : null;
  }

  /** System events (connector failure/recovery, schema drift, market open/close). */
  system(type: EventType, key: string, ts: number, severity: Severity, title: string, data: Record<string, unknown>, sources: string[] = [], instrumentIds: string[] = []) {
    const fingerprint = shortHash(`${type}|${key}`, 24);
    this.publish({
      version: 1,
      id: makeEventId(ts, fingerprint),
      type,
      instrumentIds,
      entityIds: [],
      timestamp: ts,
      severity,
      confidence: 1,
      sourceCount: sources.length,
      sources,
      title,
      summary: null,
      data,
      fingerprint,
      supportingObservations: [],
    });
  }

  private publish(ev: MarketEvent): boolean {
    const now = Date.now();
    const prev = this.recentFingerprints.get(ev.fingerprint);
    if (prev && now - prev < 86_400_000) {
      // Another source confirms an existing event → bump confirmation in DB, no new feed item.
      this.confirm(ev).catch(() => {});
      return false;
    }
    this.recentFingerprints.set(ev.fingerprint, now);
    if (this.recentFingerprints.size > 200_000) for (const [k, t] of this.recentFingerprints) if (now - t > 86_400_000) this.recentFingerprints.delete(k);
    this.pending.push(ev);
    bus.publish("market.event", ev);
    telemetry.inc("events_total", 1, { type: ev.type });
    if (!this.timer) this.timer = setTimeout(() => void this.flush(), 500);
    return true;
  }

  private async confirm(ev: MarketEvent) {
    await pool.query(
      `update market_events set source_count = source_count + 1, sources = array(select distinct unnest(sources || $2::text[])), confirmed_at = coalesce(confirmed_at, now()) where fingerprint = $1`,
      [ev.fingerprint, ev.sources],
    );
  }

  async flush(): Promise<void> {
    this.timer = null;
    const batch = this.pending.splice(0);
    if (!batch.length) return;
    const cols = 14;
    const tuples: string[] = [];
    const values: unknown[] = [];
    batch.forEach((e, i) => {
      const b = i * cols;
      tuples.push(`(${Array.from({ length: cols }, (_, j) => `$${b + j + 1}`).join(",")})`);
      values.push(e.id, e.type, new Date(e.timestamp).toISOString(), e.instrumentIds, e.entityIds, e.severity, e.confidence, e.sourceCount, e.sources, e.title, e.summary, JSON.stringify(e.data), e.fingerprint, e.supportingObservations);
    });
    try {
      await pool.query(
        `insert into market_events (id, type, ts, instrument_ids, entity_ids, severity, confidence, source_count, sources, title, summary, data, fingerprint, supporting_observations)
         values ${tuples.join(",")} on conflict (fingerprint) do update set source_count = market_events.source_count + 1, confirmed_at = coalesce(market_events.confirmed_at, now())`,
        values,
      );
      await pool.query(`insert into daily_stats (day, events) values (current_date, $1) on conflict (day) do update set events = daily_stats.events + excluded.events, updated_at = now()`, [batch.length]);
    } catch (err) {
      logger.error({ err, size: batch.length }, "event insert failed");
    }
  }

  /** Warm the dedupe cache from recent DB events so restarts do not re-emit filings/halts. */
  async warm(): Promise<void> {
    const r = await pool.query<{ fingerprint: string; ts: string }>("select fingerprint, ts from market_events where ts > now() - interval '30 hours'");
    for (const row of r.rows) this.recentFingerprints.set(row.fingerprint, Date.parse(row.ts));
  }
}

type NodeJS_Timeout = ReturnType<typeof setTimeout>;
const round = (n: number, d: number) => Math.round(n * 10 ** d) / 10 ** d;
const fmt = (n: number) => (Math.abs(n) >= 1000 ? n.toLocaleString("en-US", { maximumFractionDigits: 2 }) : n.toPrecision(6).replace(/\.?0+$/, ""));

export const eventEngine = new EventEngine();