"use client";

import { useEffect, useMemo, useRef, useSyncExternalStore } from "react";
import type { MarketEvent, StreamMessage, StreamQuote } from "./types";

/**
 * Market Atlas stream client (singleton per tab). One WebSocket, reference-counted channel
 * subscriptions, a quote store keyed by instrument id, and per-key listeners notified at most
 * every ~100 ms so tables never re-render per tick.
 */
type Listener = () => void;

export type ConnectionState = "idle" | "connecting" | "open" | "reconnecting" | "closed";

class MarketStreamClient {
  private ws: WebSocket | null = null;
  private refs = new Map<string, number>();
  private wanted = new Set<string>();
  private sent = new Set<string>();
  private quotes = new Map<string, StreamQuote>();
  private dirty = new Set<string>();
  private listeners = new Map<string, Set<Listener>>();
  private eventListeners = new Set<Listener>();
  private events: MarketEvent[] = [];
  private eventsVersion = 0;
  private tape: StreamQuote[] = [];
  private tapeVersion = 0;
  private flushTimer: ReturnType<typeof setTimeout> | null = null;
  private attempt = 0;
  private timer: ReturnType<typeof setTimeout> | null = null;
  state: ConnectionState = "idle";
  private stateListeners = new Set<Listener>();
  lastSeq = 0;
  frames = 0;

  private url(): string {
    const env = process.env.NEXT_PUBLIC_WS_URL;
    if (env) return env;
    const proto = window.location.protocol === "https:" ? "wss" : "ws";
    return `${proto}://${window.location.host}/v1/stream`;
  }

  private setState(s: ConnectionState) {
    if (this.state === s) return;
    this.state = s;
    for (const l of this.stateListeners) l();
  }

  private ensure() {
    if (typeof window === "undefined") return;
    if (this.ws || this.timer) return;
    if (!this.wanted.size) return;
    this.setState(this.attempt ? "reconnecting" : "connecting");
    let ws: WebSocket;
    try {
      ws = new WebSocket(this.url());
    } catch {
      this.scheduleReconnect();
      return;
    }
    this.ws = ws;
    ws.onopen = () => {
      this.attempt = 0;
      this.sent.clear();
      this.setState("open");
      this.sync();
    };
    ws.onmessage = (ev) => this.onFrame(String(ev.data));
    ws.onclose = () => {
      this.ws = null;
      this.sent.clear();
      if (this.wanted.size) this.scheduleReconnect();
      else this.setState("closed");
    };
    ws.onerror = () => {
      /* onclose follows */
    };
  }

  private scheduleReconnect() {
    if (this.timer) return;
    this.setState("reconnecting");
    const wait = Math.min(30_000, 1000 * 2 ** Math.min(6, this.attempt)) * (0.5 + Math.random());
    this.attempt++;
    this.timer = setTimeout(() => {
      this.timer = null;
      this.ensure();
    }, wait);
  }

  private sync() {
    if (!this.ws || this.ws.readyState !== WebSocket.OPEN) return;
    const add = [...this.wanted].filter((c) => !this.sent.has(c));
    const del = [...this.sent].filter((c) => !this.wanted.has(c));
    if (add.length) this.ws.send(JSON.stringify({ action: "subscribe", channels: add }));
    if (del.length) this.ws.send(JSON.stringify({ action: "unsubscribe", channels: del }));
    for (const c of add) this.sent.add(c);
    for (const c of del) this.sent.delete(c);
    if (!this.wanted.size) {
      this.ws.close();
    }
  }

  subscribe(channels: string[]): () => void {
    for (const c of channels) {
      this.refs.set(c, (this.refs.get(c) ?? 0) + 1);
      this.wanted.add(c);
    }
    this.ensure();
    this.sync();
    return () => {
      for (const c of channels) {
        const n = (this.refs.get(c) ?? 1) - 1;
        if (n <= 0) {
          this.refs.delete(c);
          this.wanted.delete(c);
        } else this.refs.set(c, n);
      }
      // Debounce unsubscribes so quick navigations reuse the socket.
      setTimeout(() => this.sync(), 500);
    };
  }

  private onFrame(raw: string) {
    let msg: { type?: string; seq?: number; messages?: StreamMessage[] };
    try {
      msg = JSON.parse(raw);
    } catch {
      return;
    }
    if (msg.type !== "batch" || !Array.isArray(msg.messages)) return;
    this.frames++;
    if (typeof msg.seq === "number") this.lastSeq = msg.seq;
    for (const m of msg.messages) {
      if (m.type === "quote") {
        const prev = this.quotes.get(m.instrument_id);
        if (prev && prev.received > m.received) continue;
        this.quotes.set(m.instrument_id, m);
        this.dirty.add(m.instrument_id);
        this.dirty.add(`symbol:${m.symbol.toUpperCase()}`);
        if (!prev || prev.price !== m.price) {
          this.tape.push(m);
          if (this.tape.length > 200) this.tape.splice(0, this.tape.length - 200);
          this.tapeVersion++;
          this.dirty.add("__tape__");
        }
      } else if (m.type === "event") {
        this.events.unshift(m.event);
        if (this.events.length > 500) this.events.length = 500;
        this.eventsVersion++;
        this.dirty.add("__events__");
      } else if (m.type === "market_state") {
        this.dirty.add("__market__");
      }
    }
    this.scheduleFlush();
  }

  private scheduleFlush() {
    if (this.flushTimer) return;
    this.flushTimer = setTimeout(() => {
      this.flushTimer = null;
      const keys = [...this.dirty];
      this.dirty.clear();
      for (const k of keys) {
        const ls = this.listeners.get(k);
        if (ls) for (const l of ls) l();
      }
      this.dirty.clear();
    }, 100);
  }

  listen(key: string, l: Listener): () => void {
    let s = this.listeners.get(key);
    if (!s) {
      s = new Set();
      this.listeners.set(key, s);
    }
    s.add(l);
    return () => {
      s!.delete(l);
      if (!s!.size) this.listeners.delete(key);
    };
  }

  listenState(l: Listener): () => void {
    this.stateListeners.add(l);
    return () => this.stateListeners.delete(l);
  }

  quote(id: string): StreamQuote | undefined {
    return this.quotes.get(id);
  }
  quoteBySymbol(symbol: string): StreamQuote | undefined {
    const u = symbol.toUpperCase();
    for (const q of this.quotes.values()) if (q.symbol.toUpperCase() === u) return q;
    return undefined;
  }
  recentEvents(): MarketEvent[] {
    return this.events;
  }
  eventsSnapshot(): number {
    return this.eventsVersion;
  }
  tapeSnapshot(): number {
    return this.tapeVersion;
  }
  recentTape(): StreamQuote[] {
    return this.tape;
  }
}

let client: MarketStreamClient | null = null;
export function marketStream(): MarketStreamClient {
  if (!client) client = new MarketStreamClient();
  return client;
}

/** Keep the given channels subscribed while the component is mounted. */
export function useMarketStream(channels: string[]): ConnectionState {
  const key = channels.join("|");
  useEffect(() => {
    if (!channels.length) return;
    return marketStream().subscribe(channels);
    // eslint-disable-next-line react-hooks/exhaustive-deps
  }, [key]);
  return useSyncExternalStore(
    (l) => marketStream().listenState(l),
    () => marketStream().state,
    () => "idle" as ConnectionState,
  );
}

/** Latest streamed quote for an instrument id (undefined until the first frame). Re-renders only that consumer. */
export function useLiveQuote(instrumentId: string | null | undefined): StreamQuote | undefined {
  const id = instrumentId ?? "";
  return useSyncExternalStore(
    (l) => (id ? marketStream().listen(id, l) : () => {}),
    () => (id ? marketStream().quote(id) : undefined),
    () => undefined,
  );
}

/** Events received on the stream since mount (newest first), capped. */
export function useLiveEvents(max = 100): MarketEvent[] {
  const version = useSyncExternalStore(
    (l) => marketStream().listen("__events__", l),
    () => marketStream().eventsSnapshot(),
    () => 0,
  );
  const ref = useRef<MarketEvent[]>([]);
  return useMemo(() => {
    void version;
    ref.current = marketStream().recentEvents().slice(0, max);
    return ref.current;
  }, [version, max]);
}

/** Recent price changes (tape). */
export function useTape(max = 60): StreamQuote[] {
  const version = useSyncExternalStore(
    (l) => marketStream().listen("__tape__", l),
    () => marketStream().tapeSnapshot(),
    () => 0,
  );
  return useMemo(() => {
    void version;
    const t = marketStream().recentTape();
    return t.slice(Math.max(0, t.length - max)).reverse();
  }, [version, max]);
}

/** Ticks once per second — for relative "x s ago" labels. */
export function useNow(intervalMs = 1000): number {
  return useSyncExternalStore(
    (l) => {
      const t = setInterval(l, intervalMs);
      return () => clearInterval(t);
    },
    () => Math.floor(Date.now() / intervalMs) * intervalMs,
    () => 0,
  );
}