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 CRYPTO_PAIRS: Array<[string, string]> = MAJOR_BASES.filter((b) => b !== "PAXG").map((b) => [b, "USDT"] as [string, string]).concat([["BNB", "USDT"], ["TRX", "USDT"], ["PAXG", "USDT"]]);
/** Fiat/stablecoin markets → live FX proxies (USDT ≈ USD), labelled STABLECOIN_PROXY. */
const FX_PROXY_PAIRS: Array<[string, string]> = [["EUR", "USDT"], ["USDT", "BRL"], ["USDT", "MXN"], ["USDT", "TRY"], ["USDT", "ZAR"], ["USDT", "ARS"]];
const PAIRS = [...CRYPTO_PAIRS, ...FX_PROXY_PAIRS];
const SYMBOLS = PAIRS.map(([b, q]) => `${b}${q}`);
const streamUrl = (symbols: string[]) => `wss://stream.binance.com:9443/stream?streams=${symbols.map((s) => `${s.toLowerCase()}@miniTicker`).join("/")}`;

/** Binance combined stream, `<symbol>@miniTicker` (1 s cadence, exchange event time). USDT-quoted pairs. */
export const binanceWs = defineConnector({
  metadata: {
    id: "binance-ws",
    name: "Binance — miniTicker combined stream",
    version: "1.0.0",
    sourceId: "binance",
    organization: "Binance",
    sourceType: "WEBSOCKET",
    jurisdiction: null,
    rightsStatus: "PUBLIC_ATTRIBUTED",
    realtimeStatus: "REALTIME",
    expectedLatencyMs: 1000,
    supportsStreaming: true,
    supportsHistorical: false,
    assetClasses: ["CRYPTO", "FOREX"],
    exchanges: ["binance"],
    homepage: "https://developers.binance.com/docs/binance-spot-api-docs/web-socket-streams",
    description: "Public Binance spot combined WebSocket stream (miniTicker per symbol, pushed every second) for USDT-quoted majors. Exchange event time (E) is used as source timestamp.",
    rightsNotes: "Public market data displayed with attribution to Binance. Availability may vary by jurisdiction.",
    termsUrl: "https://www.binance.com/en/terms",
    sourceFamily: "binance",
    enabled: true,
  },
  seeds: cryptoSeeds(CRYPTO_PAIRS, "binance", (b, q) => `${b}${q}`),
  defaultSymbols: SYMBOLS,
  async start(ctx) {
    const ws = ctx.openWebSocket(streamUrl(ctx.watchedSymbols()), {
      label: "binance",
      staleAfterMs: 60_000,
      heartbeat: null, // Binance pings the client; ws library answers pongs automatically.
      onMessage: (data) => {
        let msg: any;
        try {
          msg = JSON.parse(data);
        } catch {
          return;
        }
        if (msg?.data?.e === "24hrMiniTicker") ctx.emit(raw("binance-ws", "binance", "miniTicker", msg.data));
      },
    });
    ws.connect();
  },
  normalize(r): NormalizedBatch {
    const m = r.payload as Record<string, unknown>;
    if (r.kind !== "miniTicker" || typeof m.s !== "string") return { observations: [] };
    const pair = PAIRS.find(([b, q]) => `${b}${q}` === m.s);
    const base = pair?.[0] ?? (m.s.endsWith("USDT") ? m.s.slice(0, -4) : null);
    const quote = pair?.[1] ?? (m.s.endsWith("USDT") ? "USDT" : null);
    if (!base || !quote) return { observations: [] };
    return {
      observations: pairObservations(m.s, base, quote, "binance", { last: m.c, open: m.o, high: m.h, low: m.l, volume: m.v }, {
        sourceTimestamp: parseTimestamp(m.E),
        timestampTrust: "EXCHANGE",
        rightsStatus: "PUBLIC_ATTRIBUTED",
        realtimeStatus: "REALTIME",
      }),
    };
  },
  fixturesDir: "fixtures",
});