import { defineConnector, raw, type NormalizedBatch } from "@market-atlas/connector-sdk";
import { parseTimestamp } from "@market-atlas/market-model";
import { MAJOR_BASES, cryptoSeeds } from "../_shared/crypto.js";
import { pairObservations } from "../_shared/pairs.js";

const WS_URL = "wss://ws.okx.com:8443/ws/v5/public";
const CRYPTO_PAIRS: Array<[string, string]> = MAJOR_BASES.filter((b) => b !== "PAXG").map((b) => [b, "USDT"] as [string, string]).concat([["TRX", "USDT"]]);
/** USDT markets against fiat → live FX proxies (STABLECOIN_PROXY). */
const FX_PROXY_PAIRS: Array<[string, string]> = [["USDT", "EUR"], ["USDT", "AUD"], ["USDT", "BRL"], ["USDT", "SGD"], ["USDT", "TRY"], ["USDT", "AED"]];
const PAIRS = [...CRYPTO_PAIRS, ...FX_PROXY_PAIRS];
const SYMBOLS = PAIRS.map(([b, q]) => `${b}-${q}`);

/** OKX public WebSocket v5, `tickers` channel. Requires a text "ping" at least every 30 s. */
export const okxWs = defineConnector({
  metadata: {
    id: "okx-ws",
    name: "OKX — tickers channel",
    version: "1.0.0",
    sourceId: "okx",
    organization: "OKX",
    sourceType: "WEBSOCKET",
    jurisdiction: null,
    rightsStatus: "PUBLIC_ATTRIBUTED",
    realtimeStatus: "REALTIME",
    expectedLatencyMs: 500,
    supportsStreaming: true,
    supportsHistorical: false,
    assetClasses: ["CRYPTO", "FOREX"],
    exchanges: ["okx"],
    homepage: "https://www.okx.com/docs-v5/en/#public-data-websocket-tickers-channel",
    description: "Public OKX v5 WebSocket tickers channel (last, bid/ask with sizes, 24h open/high/low/volume) for USDT spot majors. Exchange timestamp `ts`.",
    rightsNotes: "Public market data displayed with attribution to OKX.",
    termsUrl: "https://www.okx.com/help/terms-of-service",
    sourceFamily: "okx",
    enabled: true,
  },
  seeds: cryptoSeeds(CRYPTO_PAIRS, "okx", (b, q) => `${b}-${q}`),
  defaultSymbols: SYMBOLS,
  async start(ctx) {
    const ws = ctx.openWebSocket(WS_URL, {
      label: "okx",
      staleAfterMs: 60_000,
      heartbeat: { intervalMs: 20_000, message: "ping" },
      onOpen: (sock) => sock.send({ op: "subscribe", args: ctx.watchedSymbols().map((instId) => ({ channel: "tickers", instId })) }),
      onMessage: (data) => {
        if (data === "pong") return;
        let msg: any;
        try {
          msg = JSON.parse(data);
        } catch {
          return;
        }
        if (msg?.arg?.channel === "tickers" && Array.isArray(msg.data)) ctx.emit(raw("okx-ws", "okx", "tickers", msg));
        else if (msg?.event === "error") ctx.reportError(new Error(String(msg.msg ?? "okx error")), { code: msg.code });
      },
    });
    ws.connect();
  },
  normalize(r): NormalizedBatch {
    const m = r.payload as { arg?: { channel?: string }; data?: Array<Record<string, unknown>> };
    if (r.kind !== "tickers" || m.arg?.channel !== "tickers" || !Array.isArray(m.data)) return { observations: [] };
    const observations = [];
    for (const t of m.data) {
      if (typeof t.instId !== "string") continue;
      const [base, quote] = t.instId.split("-");
      if (!base || !quote) continue;
      observations.push(
        ...pairObservations(t.instId, base, quote, "okx", { last: t.last, open: t.open24h, high: t.high24h, low: t.low24h, bid: t.bidPx, ask: t.askPx, bidSize: t.bidSz, askSize: t.askSz, volume: t.vol24h }, {
          sourceTimestamp: parseTimestamp(t.ts),
          timestampTrust: "EXCHANGE",
          rightsStatus: "PUBLIC_ATTRIBUTED",
          realtimeStatus: "REALTIME",
        }),
      );
    }
    return { observations };
  },
  fixturesDir: "fixtures",
});