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

const WS_URL = "wss://stream.crypto.com/exchange/v1/market";
const SYMBOLS = [...MAJOR_BASES.filter((b) => b !== "PAXG").map((b) => `${b}_USD`), "BTC_USDT", "ETH_USDT"];

/** Crypto.com Exchange public market stream, `ticker.<instrument>`. The server sends `public/heartbeat` that must be answered. */
export const cryptocomWs = defineConnector({
  metadata: {
    id: "cryptocom-ws",
    name: "Crypto.com Exchange — ticker channel",
    version: "1.0.0",
    sourceId: "cryptocom",
    organization: "Crypto.com",
    sourceType: "WEBSOCKET",
    jurisdiction: "SG",
    rightsStatus: "PUBLIC_ATTRIBUTED",
    realtimeStatus: "REALTIME",
    expectedLatencyMs: 500,
    supportsStreaming: true,
    supportsHistorical: false,
    assetClasses: ["CRYPTO"],
    exchanges: ["cryptocom"],
    homepage: "https://exchange-docs.crypto.com/exchange/v1/rest-ws/index.html#ticker-instrument_name",
    description: "Public Crypto.com Exchange ticker stream (last, best bid/ask with sizes, 24h high/low/volume) for USD-quoted majors and BTC/ETH in USDT.",
    rightsNotes: "Public market data displayed with attribution to Crypto.com Exchange.",
    termsUrl: "https://crypto.com/exchange/document/terms-of-service",
    sourceFamily: "cryptocom",
    enabled: true,
  },
  seeds: SYMBOLS.map((s) => {
    const [b, q] = s.split("_") as [string, string];
    return { symbol: s, hint: planPair(b, q, "cryptocom").hint };
  }),
  defaultSymbols: SYMBOLS,
  async start(ctx) {
    const ws = ctx.openWebSocket(WS_URL, {
      label: "cryptocom",
      staleAfterMs: 60_000,
      heartbeat: null, // server-initiated heartbeats answered below
      onOpen: (sock) => setTimeout(() => sock.send({ id: 1, method: "subscribe", params: { channels: ctx.watchedSymbols().map((s) => `ticker.${s}`) } }), 1000), // docs: wait 1 s after connect
      onMessage: (data, sock) => {
        let msg: any;
        try {
          msg = JSON.parse(data);
        } catch {
          return;
        }
        if (msg?.method === "public/heartbeat") return void sock.send({ id: msg.id, method: "public/respond-heartbeat" });
        if (msg?.method === "subscribe" && msg.result?.channel === "ticker" && Array.isArray(msg.result.data)) ctx.emit(raw("cryptocom-ws", "cryptocom", "ticker", msg.result));
        else if (msg?.code && msg.code !== 0) ctx.reportError(new Error(String(msg.message ?? `cryptocom code ${msg.code}`)));
      },
    });
    ws.connect();
  },
  normalize(r): NormalizedBatch {
    const m = r.payload as { instrument_name?: string; data?: Array<Record<string, unknown>> };
    if (r.kind !== "ticker" || typeof m.instrument_name !== "string" || !Array.isArray(m.data)) return { observations: [] };
    const [base, quote] = m.instrument_name.split("_");
    if (!base || !quote) return { observations: [] };
    const observations = [];
    for (const t of m.data) {
      observations.push(
        ...pairObservations(m.instrument_name, base, quote, "cryptocom", { last: t.a, bid: t.b, bidSize: t.bs, ask: t.k, askSize: t.ks, high: t.h, low: t.l, volume: t.v }, {
          sourceTimestamp: parseTimestamp(t.t),
          timestampTrust: "EXCHANGE",
          rightsStatus: "PUBLIC_ATTRIBUTED",
          realtimeStatus: "REALTIME",
        }),
      );
    }
    return { observations };
  },
  fixturesDir: "fixtures",
});