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, splitVenueSymbol } from "../_shared/pairs.js";

const WS_URL = "wss://stream.bybit.com/v5/public/spot";
const SYMBOLS = [...MAJOR_BASES.filter((b) => b !== "PAXG").map((b) => `${b}USDT`), "USDTEUR", "USDCEUR", "USDTBRL", "USDTTRY", "USDTAED"];

/** Bybit v5 public spot stream, `tickers.<symbol>` (snapshot + deltas with the full last price). Ping `{"op":"ping"}` every 20 s. */
export const bybitWs = defineConnector({
  metadata: {
    id: "bybit-ws",
    name: "Bybit — spot tickers (v5)",
    version: "1.0.0",
    sourceId: "bybit",
    organization: "Bybit",
    sourceType: "WEBSOCKET",
    jurisdiction: null,
    rightsStatus: "PUBLIC_ATTRIBUTED",
    realtimeStatus: "REALTIME",
    expectedLatencyMs: 500,
    supportsStreaming: true,
    supportsHistorical: false,
    assetClasses: ["CRYPTO", "FOREX"],
    exchanges: ["bybit"],
    homepage: "https://bybit-exchange.github.io/docs/v5/websocket/public/ticker",
    description: "Public Bybit v5 spot ticker stream (last price, 24h high/low/volume, previous 24h price) for USDT majors and the USDT/EUR, USDC/EUR, USDT/BRL, USDT/TRY, USDT/AED stablecoin markets (live FX proxies).",
    rightsNotes: "Public market data displayed with attribution to Bybit.",
    termsUrl: "https://www.bybit.com/en/terms-service/",
    sourceFamily: "bybit",
    enabled: true,
  },
  seeds: SYMBOLS.map((s) => {
    const [b, q] = splitVenueSymbol(s, null)!;
    return { symbol: s, hint: planPair(b, q, "bybit").hint };
  }),
  defaultSymbols: SYMBOLS,
  async start(ctx) {
    const ws = ctx.openWebSocket(WS_URL, {
      label: "bybit",
      staleAfterMs: 60_000,
      heartbeat: { intervalMs: 20_000, message: JSON.stringify({ op: "ping" }) },
      onOpen: (sock) => {
        // Bybit accepts at most 10 topics per subscribe request.
        const topics = ctx.watchedSymbols().map((s) => `tickers.${s}`);
        for (let i = 0; i < topics.length; i += 10) sock.send({ op: "subscribe", args: topics.slice(i, i + 10) });
      },
      onMessage: (data) => {
        let msg: any;
        try {
          msg = JSON.parse(data);
        } catch {
          return;
        }
        if (typeof msg?.topic === "string" && msg.topic.startsWith("tickers.") && msg.data) ctx.emit(raw("bybit-ws", "bybit", "ticker", msg));
        else if (msg?.success === false) ctx.reportError(new Error(String(msg.ret_msg ?? "bybit error")));
      },
    });
    ws.connect();
  },
  normalize(r): NormalizedBatch {
    const m = r.payload as { ts?: number; data?: Record<string, unknown>; cs?: number };
    const d = m.data;
    if (r.kind !== "ticker" || !d || typeof d.symbol !== "string") return { observations: [] };
    const split = splitVenueSymbol(d.symbol, null);
    if (!split) return { observations: [] };
    const [base, quote] = split;
    return {
      observations: pairObservations(d.symbol, base, quote, "bybit", { last: d.lastPrice, high: d.highPrice24h, low: d.lowPrice24h, volume: d.volume24h, open: d.prevPrice24h }, {
        sourceTimestamp: parseTimestamp(m.ts),
        timestampTrust: "EXCHANGE",
        sequence: typeof m.cs === "number" ? m.cs : null,
        rightsStatus: "PUBLIC_ATTRIBUTED",
        realtimeStatus: "REALTIME",
      }),
    };
  },
  fixturesDir: "fixtures",
});