// src/realtime/wsSymbolStats.js
const WebSocket = require("ws");
const { WebSocketServer } = WebSocket;

function normSymbols(arr) {
  return (Array.isArray(arr) ? arr : [])
    .map((s) => String(s).trim().toUpperCase())
    .filter(Boolean);
}

function round(n, dp = 4) {
  const x = Number(n);
  if (!Number.isFinite(x)) return 0;
  const m = 10 ** dp;
  return Math.round(x * m) / m;
}

function safeSend(ws, obj) {
  if (ws.readyState !== WebSocket.OPEN) return;
  ws.send(JSON.stringify(obj));
}

function toPayload(st, ts) {
  const open = Number(st.open ?? 0);
  const price = Number(st.price ?? 0);
  const change24h = round(price - open, 2);
  const changePct24h = open ? round((change24h / open) * 100, 4) : 0;

  return {
    symbol: st.symbol,
    ts,
    data: {
      price: round(price, 8),
      change24h,
      changePct24h,
      high24h: round(st.high24h ?? price, 8),
      low24h: round(st.low24h ?? price, 8),
      volume24h: Number(st.volume24h ?? 0),
    },
  };
}

/**
 * attachSymbolStatsWS(httpServer, opts)
 * Returns { publishSymbolStats, publishBulkSymbolStats, close }
 */
function attachSymbolStatsWS(httpServer, opts = {}) {
  const {
    allowedOrigins = [],
    path = "/ws/market",
    heartbeatMs = 20000,
    // throttle per client (optional)
    minPushIntervalMs = 0, // e.g. 100 = max 10 pushes/sec/client
  } = opts;

  // ---- In-memory market snapshot ----
  // symbol -> {symbol, open, price, high24h, low24h, volume24h, _lastSentPrice? ...}
  const market = new Map();

  // ---- Subscription index ----
  const subsBySymbol = new Map(); // symbol -> Set<ws>
  const allClients = new Set();   // ws subscribed to "*"

  // ---- Per-client batching (so we can send 1 message even if many symbols update fast) ----
  const pendingByClient = new Map(); // ws -> Map<symbol, payload>
  const lastPushAt = new Map();      // ws -> timestamp
  let flushScheduled = false;

  function initClient(ws) {
    ws.isAlive = true;
    ws.subs = { all: false, symbols: new Set() };
  }

  function ensureMarket(symbol) {
    if (!market.has(symbol)) {
      market.set(symbol, {
        symbol,
        open: 0,
        price: 0,
        high24h: 0,
        low24h: 0,
        volume24h: 0,
        _lastSentPrice: null,
      });
    }
  }

  function addSub(ws, symbol) {
    ensureMarket(symbol);
    let set = subsBySymbol.get(symbol);
    if (!set) subsBySymbol.set(symbol, (set = new Set()));
    set.add(ws);
    ws.subs.symbols.add(symbol);
  }

  function removeSub(ws, symbol) {
    ws.subs.symbols.delete(symbol);
    const set = subsBySymbol.get(symbol);
    if (!set) return;
    set.delete(ws);
    if (set.size === 0) subsBySymbol.delete(symbol);
  }

  function clearAllSubs(ws) {
    for (const sym of ws.subs.symbols) removeSub(ws, sym);
    ws.subs.symbols.clear();
    allClients.delete(ws);
    ws.subs.all = false;
  }

  function queue(ws, payload) {
    if (ws.readyState !== WebSocket.OPEN) return;

    let map = pendingByClient.get(ws);
    if (!map) pendingByClient.set(ws, (map = new Map()));
    // de-dupe: latest payload per symbol wins
    map.set(payload.symbol, payload);

    if (!flushScheduled) {
      flushScheduled = true;
      setImmediate(flush);
    }
  }

  function flush() {
    flushScheduled = false;
    const now = Date.now();

    for (const [ws, map] of pendingByClient.entries()) {
      if (ws.readyState !== WebSocket.OPEN) {
        pendingByClient.delete(ws);
        continue;
      }

      // optional throttle
      if (minPushIntervalMs > 0) {
        const last = lastPushAt.get(ws) || 0;
        if (now - last < minPushIntervalMs) continue;
        lastPushAt.set(ws, now);
      }

      if (map.size === 0) continue;

      const updates = Array.from(map.values());
      map.clear();

      safeSend(ws, { type: "symbol_stats_batch", ts: now, updates });
    }
  }

  // ---- Publishing API (called by your REST/price API) ----
  function publishSymbolStats(symbol, nextPartial) {
    const sym = String(symbol).trim().toUpperCase();
    if (!sym) return;

    ensureMarket(sym);
    const st = market.get(sym);

    // Merge update
    if (nextPartial && typeof nextPartial === "object") {
      if (nextPartial.open != null) st.open = Number(nextPartial.open);
      if (nextPartial.price != null) st.price = Number(nextPartial.price);
      if (nextPartial.high24h != null) st.high24h = Number(nextPartial.high24h);
      if (nextPartial.low24h != null) st.low24h = Number(nextPartial.low24h);
      if (nextPartial.volume24h != null) st.volume24h = Number(nextPartial.volume24h);
    }

    // Optional diffing (skip if no price change)
    if (st._lastSentPrice === st.price) return;
    st._lastSentPrice = st.price;

    const payload = toPayload(st, Date.now());

    // All-clients
    for (const ws of allClients) queue(ws, payload);

    // Per-symbol clients (skip clients that are already in all)
    const set = subsBySymbol.get(sym);
    if (set) {
      for (const ws of set) {
        if (ws.subs?.all) continue;
        queue(ws, payload);
      }
    }
  }

  function publishBulkSymbolStats(items) {
    if (!Array.isArray(items)) return;
    for (const item of items) {
      const sym = String(item?.symbol || "").trim().toUpperCase();
      if (!sym) continue;
      publishSymbolStats(sym, item);
    }
  }

  // ---- WS server ----
  const wss = new WebSocketServer({ server: httpServer, path });

  wss.on("connection", (ws, req) => {
    const origin = req.headers.origin;

    if (origin && allowedOrigins.length && !allowedOrigins.includes(origin)) {
      ws.close(1008, "Origin not allowed");
      return;
    }

    initClient(ws);

    ws.on("pong", () => (ws.isAlive = true));
    safeSend(ws, { type: "welcome", serverTime: Date.now(), path });

    ws.on("message", (buf) => {
      let msg;
      try {
        msg = JSON.parse(buf.toString());
      } catch {
        return safeSend(ws, { type: "error", message: "Invalid JSON" });
      }

      const type = String(msg?.type || "");
      if (type === "ping") return safeSend(ws, { type: "pong", ts: msg?.ts ?? Date.now() });

      const channel = String(msg?.channel || "").toLowerCase();
      const symbols = normSymbols(msg?.symbols);

      if (!["subscribe", "unsubscribe"].includes(type)) {
        return safeSend(ws, { type: "error", message: `Unknown type: ${type}` });
      }
      if (channel !== "symbol_stats") {
        return safeSend(ws, { type: "error", message: `Unsupported channel: ${channel}` });
      }
      if (!symbols.length) {
        return safeSend(ws, { type: "error", message: "symbols is required" });
      }

      const wantsAll = symbols.includes("*");

      if (type === "subscribe") {
        if (wantsAll) {
          clearAllSubs(ws);
          ws.subs.all = true;
          allClients.add(ws);

          safeSend(ws, { type: "subscribed", channel: "symbol_stats", symbols: ["*"] });

          // send snapshot of what we currently have in-memory
          const ts = Date.now();
          const updates = Array.from(market.values()).map((st) => toPayload(st, ts));
          return safeSend(ws, { type: "symbol_stats_batch", ts, updates });
        }

        // specific symbols
        if (ws.subs.all) {
          allClients.delete(ws);
          ws.subs.all = false;
        }

        for (const s of symbols) addSub(ws, s);

        safeSend(ws, {
          type: "subscribed",
          channel: "symbol_stats",
          symbols: [...ws.subs.symbols],
        });

        // snapshot for these symbols
        const ts = Date.now();
        const updates = [...ws.subs.symbols].map((sym) => toPayload(market.get(sym), ts));
        return safeSend(ws, { type: "symbol_stats_batch", ts, updates });
      }

      // unsubscribe
      if (wantsAll) {
        clearAllSubs(ws);
        return safeSend(ws, { type: "unsubscribed", channel: "symbol_stats", symbols: ["*"] });
      }

      for (const s of symbols) removeSub(ws, s);
      safeSend(ws, { type: "unsubscribed", channel: "symbol_stats", symbols });
    });

    ws.on("close", () => {
      clearAllSubs(ws);
      pendingByClient.delete(ws);
      lastPushAt.delete(ws);
    });
  });

  // Heartbeat
  const heartbeat = setInterval(() => {
    wss.clients.forEach((ws) => {
      if (ws.isAlive === false) return ws.terminate();
      ws.isAlive = false;
      ws.ping();
    });
  }, heartbeatMs);

  return {
    wss,
    market, // if you want to inspect
    publishSymbolStats,
    publishBulkSymbolStats,
    close: () => {
      clearInterval(heartbeat);
      wss.close();
    },
  };
}

module.exports = { attachSymbolStatsWS };