import type { RawObservation } from "@market-atlas/market-model";
import {
  HttpClient,
  ManagedWebSocket,
  RateLimiter,
  backoffMs,
  schemaFingerprint,
  type ConnectorContext,
  type ConnectorDefinition,
  type ManagedWebSocketOptions,
} from "@market-atlas/connector-sdk";
import { config } from "../config.js";
import { pool } from "../db/pool.js";
import { logger } from "../logger.js";
import { calendar } from "./calendar.js";
import { eventEngine } from "./events.js";
import { health } from "./health.js";
import { pipeline } from "./pipeline.js";
import { rawArchive } from "./raw-archive.js";
import { telemetry } from "./telemetry.js";

interface Runtime {
  def: ConnectorDefinition;
  ctx: ConnectorContext;
  sockets: ManagedWebSocket[];
  pollTimer: NodeJS.Timeout | null;
  running: boolean;
  paused: boolean;
  consecutiveFailures: number;
  driftStrikes: number;
  symbols: string[];
  stateCache: Record<string, unknown>;
  stateDirty: boolean;
}

/**
 * Runs connectors: lifecycle, adaptive polling, per-host rate limits, circuit breaker with
 * jittered backoff, schema-drift detection, L0 archiving and hand-off to the pipeline.
 */
export class ConnectorManager {
  private runtimes = new Map<string, Runtime>();
  readonly limiter = new RateLimiter({ ratePerSec: 2, burst: 4 });
  readonly http = new HttpClient({
    userAgent: config.userAgent,
    limiter: this.limiter,
    onRequest: (i) => {
      telemetry.inc("http_requests_total", 1, { host: i.host, status: String(i.status) });
      telemetry.observe("http_request_ms", i.durationMs, { host: i.host });
    },
  });

  list(): Runtime[] {
    return [...this.runtimes.values()];
  }

  get(id: string): Runtime | undefined {
    return this.runtimes.get(id);
  }

  async register(def: ConnectorDefinition, opts: { paused: boolean; state: Record<string, unknown>; fingerprints: Record<string, string[]>; symbols?: string[] }): Promise<void> {
    const id = def.metadata.id;
    for (const [host, rate] of Object.entries(def.rateLimits ?? {})) this.limiter.configure(host, rate);
    health.register(id, def.metadata.sourceId);
    health.loadFingerprints(id, opts.fingerprints);
    pipeline.registerSource(def.metadata.sourceId, {
      family: def.metadata.sourceFamily,
      isOfficial: def.metadata.rightsStatus === "OFFICIAL_OPEN_DATA",
      realtimeStatus: def.metadata.realtimeStatus,
    });
    const rt: Runtime = {
      def,
      ctx: null as unknown as ConnectorContext,
      sockets: [],
      pollTimer: null,
      running: false,
      paused: opts.paused,
      consecutiveFailures: 0,
      driftStrikes: 0,
      symbols: opts.symbols ?? def.defaultSymbols ?? [],
      stateCache: opts.state ?? {},
      stateDirty: false,
    };
    rt.ctx = this.makeContext(rt);
    this.runtimes.set(id, rt);
  }

  private makeContext(rt: Runtime): ConnectorContext {
    const id = rt.def.metadata.id;
    const log = logger.child({ connector: id });
    return {
      connectorId: id,
      logger: {
        debug: (m, d) => log.debug(d ?? {}, m),
        info: (m, d) => log.info(d ?? {}, m),
        warn: (m, d) => log.warn(d ?? {}, m),
        error: (m, d) => log.error(d ?? {}, m),
      },
      http: this.http,
      state: {
        get: async (k) => rt.stateCache[k] as never,
        set: async (k, v) => {
          rt.stateCache[k] = v;
          rt.stateDirty = true;
        },
      },
      emit: (raw) => {
        for (const r of Array.isArray(raw) ? raw : [raw]) void this.handleRaw(rt, r, false);
      },
      openWebSocket: (url: string, o: ManagedWebSocketOptions) => {
        const ws = new ManagedWebSocket(url, {
          ...o,
          onError: (err) => {
            health.error(id, err);
            o.onError?.(err);
          },
        });
        ws.on((ev, detail) => {
          if (ev === "connected") {
            rt.consecutiveFailures = 0;
            health.setState(id, "HEALTHY");
          } else if (ev === "reconnecting") {
            health.reconnect(id);
            health.setState(id, "RECONNECTING");
            if ((detail?.attempt as number) >= 5) this.markFailed(rt, `reconnect attempt ${detail?.attempt}`);
          } else if (ev === "stale") {
            health.setState(id, "STALE");
          }
        });
        rt.sockets.push(ws);
        return ws;
      },
      isMarketOpen: (exchangeId) => calendar.isOpen(exchangeId),
      watchedSymbols: () => rt.symbols,
      reportError: (err, context) => {
        health.error(id, err);
        log.warn({ err: err instanceof Error ? err.message : String(err), ...context }, "connector reported error");
      },
      secret: (name) => process.env[name],
      now: () => Date.now(),
    };
  }

  async startAll(): Promise<void> {
    for (const rt of this.runtimes.values()) {
      if (rt.paused) {
        health.setState(rt.def.metadata.id, "PAUSED");
        continue;
      }
      await this.start(rt.def.metadata.id);
    }
  }

  async start(id: string): Promise<void> {
    const rt = this.runtimes.get(id);
    if (!rt || rt.running) return;
    rt.running = true;
    rt.paused = false;
    health.setState(id, "STARTING");
    try {
      if (rt.def.start) {
        await rt.def.start(rt.ctx);
        if (!rt.sockets.length) health.setState(id, "HEALTHY");
      }
      if (rt.def.poll) this.schedulePoll(rt, 500 + Math.random() * 3000);
    } catch (err) {
      health.error(id, err);
      this.markFailed(rt, err instanceof Error ? err.message : String(err));
      this.scheduleRestart(rt);
    }
  }

  async stop(id: string, reason = "stop"): Promise<void> {
    const rt = this.runtimes.get(id);
    if (!rt) return;
    rt.running = false;
    if (rt.pollTimer) clearTimeout(rt.pollTimer);
    rt.pollTimer = null;
    for (const s of rt.sockets) s.close();
    rt.sockets = [];
    try {
      await rt.def.stop?.(rt.ctx);
    } catch (err) {
      logger.warn({ connector: id, err }, "connector stop failed");
    }
    await this.persistState(rt);
    logger.info({ connector: id, reason }, "connector stopped");
  }

  async pause(id: string): Promise<void> {
    const rt = this.runtimes.get(id);
    if (!rt) return;
    await this.stop(id, "pause");
    rt.paused = true;
    health.setState(id, "PAUSED");
    await pool.query("update connectors set paused = true, updated_at = now() where id = $1", [id]);
  }

  async resume(id: string): Promise<void> {
    const rt = this.runtimes.get(id);
    if (!rt) return;
    await pool.query("update connectors set paused = false, updated_at = now() where id = $1", [id]);
    rt.consecutiveFailures = 0;
    rt.driftStrikes = 0;
    await this.start(id);
  }

  async restart(id: string): Promise<void> {
    await this.stop(id, "restart");
    const rt = this.runtimes.get(id);
    if (rt) {
      rt.consecutiveFailures = 0;
      await this.start(id);
    }
  }

  /** Run one poll immediately (admin "run test"); returns accepted observation count. */
  async runOnce(id: string): Promise<{ accepted: number; raws: number; durationMs: number }> {
    const rt = this.runtimes.get(id);
    if (!rt) throw new Error("unknown connector");
    const started = Date.now();
    let accepted = 0;
    let raws = 0;
    if (rt.def.poll) {
      const out = await rt.def.poll(rt.ctx);
      raws = out.length;
      for (const r of out) accepted += await this.handleRaw(rt, r, true);
    } else if (rt.def.healthCheck) {
      const res = await rt.def.healthCheck(rt.ctx);
      if (!res.ok) throw new Error(res.detail ?? "health check failed");
    }
    return { accepted, raws, durationMs: Date.now() - started };
  }

  private schedulePoll(rt: Runtime, delayMs: number) {
    if (!rt.running) return;
    if (rt.pollTimer) clearTimeout(rt.pollTimer);
    const at = Date.now() + delayMs;
    health.poll(rt.def.metadata.id, at);
    rt.pollTimer = setTimeout(() => void this.poll(rt), delayMs);
  }

  private nextInterval(rt: Runtime): number {
    const s = rt.def.schedule!;
    if ("intervalMs" in s) return s.intervalMs;
    if (!s.exchangeId) return s.openMs;
    const st = calendar.status(s.exchangeId);
    if (st.state === "OPEN" || st.state === "PRE" || st.state === "POST") return s.openMs;
    const day = new Date().getUTCDay();
    if (day === 0 || day === 6 || st.isHoliday) return s.weekendMs;
    return s.closedMs;
  }

  private async poll(rt: Runtime) {
    const id = rt.def.metadata.id;
    if (!rt.running || !rt.def.poll) return;
    const started = Date.now();
    try {
      const raws = await rt.def.poll(rt.ctx);
      let accepted = 0;
      for (const r of raws) accepted += await this.handleRaw(rt, r, true);
      rt.consecutiveFailures = 0;
      if (health.state(id) !== "HEALTHY") health.setState(id, "HEALTHY");
      telemetry.observe("connector_poll_ms", Date.now() - started, { connector: id });
      logger.debug({ connector: id, raws: raws.length, accepted, ms: Date.now() - started }, "poll done");
      await this.persistState(rt);
      this.schedulePoll(rt, this.nextInterval(rt));
    } catch (err) {
      rt.consecutiveFailures++;
      health.error(id, err);
      // Retry sooner than the regular cadence (a daily connector must not wait 24 h after a transient network error).
      const wait = Math.max(5000, Math.min(this.nextInterval(rt), backoffMs(rt.consecutiveFailures, 15_000, 30 * 60_000)));
      logger.warn({ connector: id, err: err instanceof Error ? err.message : String(err), failures: rt.consecutiveFailures, retryInMs: wait }, "poll failed");
      if (rt.consecutiveFailures >= 3) this.markFailed(rt, err instanceof Error ? err.message : String(err));
      else health.setState(id, "DEGRADED");
      this.schedulePoll(rt, wait);
    }
  }

  private markFailed(rt: Runtime, reason: string) {
    const id = rt.def.metadata.id;
    if (health.state(id) === "FAILED") return;
    health.setState(id, "FAILED");
    eventEngine.system("SOURCE_FAILURE", `${id}:${Math.floor(Date.now() / 3_600_000)}`, Date.now(), "WARNING", `Source failure: ${rt.def.metadata.name}`, { connectorId: id, reason: reason.slice(0, 200) }, [rt.def.metadata.sourceId]);
  }

  private scheduleRestart(rt: Runtime) {
    const wait = backoffMs(Math.min(8, rt.consecutiveFailures + 1), 10_000, 15 * 60_000);
    rt.consecutiveFailures++;
    setTimeout(() => {
      if (!rt.paused) {
        rt.running = false;
        void this.start(rt.def.metadata.id);
      }
    }, wait);
  }

  /** Every raw payload: fingerprint, archive, normalize, hand to the pipeline. Returns accepted count. */
  private async handleRaw(rt: Runtime, raw: RawObservation, forceArchive: boolean): Promise<number> {
    const id = rt.def.metadata.id;
    telemetry.inc("raw_total", 1, { connector: id });
    const fp = schemaFingerprint(raw.payload);
    const status = health.recordFingerprint(id, raw.kind, fp);
    let batch;
    try {
      batch = rt.def.normalize(raw);
      health.parse(id, true);
    } catch (err) {
      health.parse(id, false);
      health.error(id, err);
      rt.driftStrikes++;
      this.recordSchemaChange(rt, raw, fp, status).catch(() => {});
      if (rt.driftStrikes >= 20) this.driftPause(rt, raw.kind);
      return 0;
    }
    const produced = batch.observations.length + (batch.events?.length ?? 0) + (batch.instruments?.length ?? 0) + (batch.filings?.length ?? 0) + (batch.holidays?.length ?? 0);
    if (status === "new") {
      await this.recordSchemaChange(rt, raw, fp, status);
      if (produced === 0) {
        rt.driftStrikes++;
        if (rt.driftStrikes >= 5) this.driftPause(rt, raw.kind);
      }
    }
    if (produced > 0) rt.driftStrikes = Math.max(0, rt.driftStrikes - 1);
    const rawRef = rawArchive.store(raw, forceArchive || rt.def.metadata.sourceType !== "WEBSOCKET");
    const st = health.state(id);
    if (st === "STALE" || st === "DEGRADED" || st === "STARTING" || st === "RECONNECTING") health.setState(id, "HEALTHY");
    return pipeline.process(rt.def, raw, batch, rawRef);
  }

  private async recordSchemaChange(rt: Runtime, raw: RawObservation, fp: string, status: "known" | "new" | "first") {
    if (status === "known") return;
    const id = rt.def.metadata.id;
    const known = health.fingerprints(id)[raw.kind] ?? [];
    await pool.query(
      `insert into connector_schema_changes (connector_id, kind, old_fingerprint, new_fingerprint, sample) values ($1,$2,$3,$4,$5)`,
      [id, raw.kind, known.filter((k) => k !== fp).at(-1) ?? null, fp, JSON.stringify(truncate(raw.payload))],
    );
    if (status === "new") {
      telemetry.inc("schema_changes_total", 1, { connector: id });
      logger.warn({ connector: id, kind: raw.kind, fingerprint: fp }, "schema change detected");
    }
  }

  private driftPause(rt: Runtime, kind: string) {
    const id = rt.def.metadata.id;
    logger.error({ connector: id, kind }, "schema drift: pausing connector to avoid producing wrong values");
    eventEngine.system("SCHEMA_DRIFT", `${id}:${kind}:${Math.floor(Date.now() / 3_600_000)}`, Date.now(), "CRITICAL", `Schema drift: ${rt.def.metadata.name}`, { connectorId: id, kind }, [rt.def.metadata.sourceId]);
    void this.pause(id);
  }

  private async persistState(rt: Runtime) {
    if (!rt.stateDirty) return;
    rt.stateDirty = false;
    try {
      await pool.query("update connectors set state = $2, updated_at = now() where id = $1", [rt.def.metadata.id, JSON.stringify(rt.stateCache)]);
    } catch (err) {
      rt.stateDirty = true;
      logger.warn({ connector: rt.def.metadata.id, err }, "state persist failed");
    }
  }

  async stopAll(): Promise<void> {
    await Promise.all([...this.runtimes.keys()].map((id) => this.stop(id, "shutdown")));
  }
}

function truncate(v: unknown): unknown {
  const s = JSON.stringify(v);
  if (!s) return null;
  return s.length > 4000 ? { truncated: s.slice(0, 4000) } : v;
}

export const connectorManager = new ConnectorManager();