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

interface Tracker {
  connectorId: string;
  sourceId: string;
  state: ConnectorState;
  startedAt: number | null;
  lastMessageAt: number | null;
  lastSuccessAt: number | null;
  lastErrorAt: number | null;
  lastError: string | null;
  messagesTotal: number;
  errorsTotal: number;
  parseOk: number;
  parseFail: number;
  reconnects: number;
  latency: Reservoir;
  instruments: Set<string>;
  schemaFingerprints: Map<string, Set<string>>;
  nextPollAt: number | null;
  /** Rolling availability samples (1 per minute, 1 = healthy). */
  availability: number[];
}

/**
 * Operational health and reliability scoring per connector/source. The score measures the
 * connector's operational behaviour (availability, latency, parse success, stability) — never
 * financial truth.
 */
export class HealthEngine {
  private trackers = new Map<string, Tracker>();
  private sourceOfConnector = new Map<string, string>();

  register(connectorId: string, sourceId: string): void {
    this.sourceOfConnector.set(connectorId, sourceId);
    if (!this.trackers.has(connectorId)) {
      this.trackers.set(connectorId, {
        connectorId,
        sourceId,
        state: "STARTING",
        startedAt: null,
        lastMessageAt: null,
        lastSuccessAt: null,
        lastErrorAt: null,
        lastError: null,
        messagesTotal: 0,
        errorsTotal: 0,
        parseOk: 0,
        parseFail: 0,
        reconnects: 0,
        latency: new Reservoir(256),
        instruments: new Set(),
        schemaFingerprints: new Map(),
        nextPollAt: null,
        availability: [],
      });
    }
  }

  private t(id: string): Tracker {
    const t = this.trackers.get(id);
    if (!t) throw new Error(`unknown connector ${id}`);
    return t;
  }

  setState(id: string, state: ConnectorState) {
    const t = this.t(id);
    if (t.state === state) return;
    const prev = t.state;
    t.state = state;
    if (state === "HEALTHY" && !t.startedAt) t.startedAt = Date.now();
    telemetry.gauge("connector_state", stateNum(state), { connector: id });
    bus.publish("connector.health", this.snapshot(id));
    logger.info({ connector: id, from: prev, to: state }, "connector state");
  }

  state(id: string): ConnectorState {
    return this.trackers.get(id)?.state ?? "DISABLED";
  }

  message(id: string, latencyMs: number | null, instrumentIds: string[] = []) {
    const t = this.t(id);
    t.messagesTotal++;
    t.lastMessageAt = Date.now();
    t.lastSuccessAt = t.lastMessageAt;
    if (latencyMs != null && latencyMs >= 0 && latencyMs < 3_600_000) t.latency.add(latencyMs);
    for (const i of instrumentIds) t.instruments.add(i);
    telemetry.inc("connector_messages_total", 1, { connector: id });
    telemetry.inc("observations_received_total");
  }

  parse(id: string, ok: boolean) {
    const t = this.t(id);
    if (ok) t.parseOk++;
    else t.parseFail++;
  }

  error(id: string, err: unknown) {
    const t = this.t(id);
    t.errorsTotal++;
    t.lastErrorAt = Date.now();
    t.lastError = (err instanceof Error ? err.message : String(err)).slice(0, 500);
    telemetry.inc("connector_errors_total", 1, { connector: id });
  }

  reconnect(id: string) {
    this.t(id).reconnects++;
  }

  poll(id: string, nextAt: number | null) {
    this.t(id).nextPollAt = nextAt;
  }

  recordFingerprint(id: string, kind: string, fp: string): "known" | "new" | "first" {
    const t = this.t(id);
    let set = t.schemaFingerprints.get(kind);
    if (!set) {
      set = new Set();
      t.schemaFingerprints.set(kind, set);
    }
    if (set.has(fp)) return "known";
    const first = set.size === 0;
    set.add(fp);
    return first ? "first" : "new";
  }

  /** Restore lifetime counters and 24 h instrument coverage from the database so restarts do not zero the directory. */
  async hydrate(): Promise<void> {
    const ids = [...this.trackers.keys()];
    if (!ids.length) return;
    const [rows, cov] = await Promise.all([
      pool.query<{ id: string; messages_total: number; errors_total: number; reconnects: number }>("select id, messages_total, errors_total, reconnects from connectors where id = any($1)", [ids]),
      pool.query<{ connector_id: string; n: number }>("select connector_id, count(distinct instrument_id)::int as n from observations where received_at > now() - interval '24 hours' group by 1"),
    ]);
    for (const r of rows.rows) {
      const t = this.trackers.get(r.id);
      if (!t) continue;
      t.messagesTotal = Math.max(t.messagesTotal, Number(r.messages_total) || 0);
      t.errorsTotal = Math.max(t.errorsTotal, Number(r.errors_total) || 0);
      t.reconnects = Math.max(t.reconnects, Number(r.reconnects) || 0);
    }
    this.coverageFloor.clear();
    for (const c of cov.rows) this.coverageFloor.set(c.connector_id, c.n);
  }

  /** Coverage observed in the DB over the last 24 h (floor for the live set, which starts empty at boot). */
  private coverageFloor = new Map<string, number>();

  loadFingerprints(id: string, fps: Record<string, string[]>) {
    const t = this.t(id);
    for (const [kind, arr] of Object.entries(fps)) t.schemaFingerprints.set(kind, new Set(arr));
  }

  fingerprints(id: string): Record<string, string[]> {
    const t = this.t(id);
    return Object.fromEntries([...t.schemaFingerprints].map(([k, v]) => [k, [...v]]));
  }

  /** 0–100 operational reliability. */
  score(id: string): number | null {
    const t = this.trackers.get(id);
    if (!t || t.messagesTotal === 0) return null;
    const avail = t.availability.length ? t.availability.reduce((a, b) => a + b, 0) / t.availability.length : t.state === "HEALTHY" || t.state === "STARTING" ? 1 : 0.5;
    const parseTotal = t.parseOk + t.parseFail;
    const parse = parseTotal ? t.parseOk / parseTotal : 1;
    const p95 = t.latency.quantile(0.95);
    const latencyScore = p95 == null ? 0.8 : p95 < 1000 ? 1 : p95 < 5000 ? 0.85 : p95 < 60_000 ? 0.7 : 0.5;
    const stability = Math.max(0.4, 1 - t.reconnects / Math.max(50, t.messagesTotal / 100));
    const errorRate = t.errorsTotal / Math.max(1, t.messagesTotal + t.errorsTotal);
    const errScore = Math.max(0, 1 - errorRate * 10);
    return Math.round(100 * (0.4 * avail + 0.2 * parse + 0.15 * latencyScore + 0.15 * stability + 0.1 * errScore));
  }

  /** Reliability 0..1 for the consensus engine, keyed by source id (best connector of the source). */
  sourceReliability(sourceId: string): number {
    const scores: number[] = [];
    for (const t of this.trackers.values()) {
      if (t.sourceId !== sourceId) continue;
      const s = this.score(t.connectorId);
      if (s != null) scores.push(s / 100);
    }
    return scores.length ? Math.max(...scores) : 0.6;
  }

  snapshot(id: string): ConnectorHealth {
    const t = this.t(id);
    const parseTotal = t.parseOk + t.parseFail;
    return {
      connectorId: id,
      state: t.state,
      lastMessageAt: t.lastMessageAt,
      lastSuccessAt: t.lastSuccessAt,
      lastErrorAt: t.lastErrorAt,
      lastError: t.lastError,
      messages1m: telemetry.lastMinute("connector_messages_total", { connector: id }),
      messagesTotal: t.messagesTotal,
      errorsTotal: t.errorsTotal,
      reconnects: t.reconnects,
      medianLatencyMs: t.latency.quantile(0.5),
      p95LatencyMs: t.latency.quantile(0.95),
      parseSuccessRate: parseTotal ? t.parseOk / parseTotal : null,
      instrumentsCovered: Math.max(t.instruments.size, this.coverageFloor.get(id) ?? 0),
      schemaFingerprints: [...t.schemaFingerprints.values()].flatMap((s) => [...s]),
      reliabilityScore: this.score(id),
      startedAt: t.startedAt,
      nextPollAt: t.nextPollAt,
    };
  }

  all(): ConnectorHealth[] {
    return [...this.trackers.keys()].map((id) => this.snapshot(id));
  }

  /** Called every minute: staleness detection, availability sample, DB snapshot. */
  async tick(staleAfterMs: (id: string) => number): Promise<void> {
    const now = Date.now();
    const rows: unknown[][] = [];
    for (const t of this.trackers.values()) {
      if (t.state === "HEALTHY" && t.lastMessageAt && now - t.lastMessageAt > staleAfterMs(t.connectorId)) this.setState(t.connectorId, "STALE");
      const healthy = t.state === "HEALTHY" ? 1 : t.state === "DEGRADED" || t.state === "STALE" ? 0.5 : t.state === "PAUSED" || t.state === "DISABLED" ? NaN : 0;
      if (!Number.isNaN(healthy)) {
        t.availability.push(healthy);
        if (t.availability.length > 24 * 60) t.availability.shift();
      }
      const s = this.snapshot(t.connectorId);
      rows.push([new Date(now).toISOString(), t.connectorId, t.state, s.messages1m, telemetry.lastMinute("connector_errors_total", { connector: t.connectorId }), s.medianLatencyMs, s.p95LatencyMs, s.parseSuccessRate, s.instrumentsCovered, t.reconnects, s.reliabilityScore]);
    }
    if (!rows.length) return;
    const cols = 11;
    const tuples = rows.map((_, i) => `(${Array.from({ length: cols }, (_, j) => `$${i * cols + j + 1}`).join(",")})`);
    try {
      await pool.query(
        `insert into connector_health (ts, connector_id, status, messages_1m, errors_1m, median_latency_ms, p95_latency_ms, parse_success_rate, instruments_covered, reconnects, reliability_score) values ${tuples.join(",")} on conflict do nothing`,
        rows.flat(),
      );
      for (const t of this.trackers.values()) {
        await pool.query(
          `update connectors set status = $2, last_message_at = $3, last_success_at = $4, last_error_at = $5, last_error = $6, messages_total = $7, errors_total = $8, reconnects = $9, reliability_score = $10, schema_fingerprints = $11, updated_at = now() where id = $1`,
          [t.connectorId, t.state, iso(t.lastMessageAt), iso(t.lastSuccessAt), iso(t.lastErrorAt), t.lastError, t.messagesTotal, t.errorsTotal, t.reconnects, this.score(t.connectorId), JSON.stringify(this.fingerprints(t.connectorId))],
        );
      }
    } catch (err) {
      logger.error({ err }, "health snapshot failed");
    }
  }
}

const iso = (ms: number | null) => (ms == null ? null : new Date(ms).toISOString());
const stateNum = (s: ConnectorState) => ({ HEALTHY: 1, DEGRADED: 2, STALE: 3, RECONNECTING: 4, FAILED: 5, PAUSED: 6, DISABLED: 7, STARTING: 0 })[s];

export const health = new HealthEngine();