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://api.gateio.ws/ws/v4/";
const SYMBOLS = MAJOR_BASES.filter((b) => b !== "PAXG").map((b) => `${b}_USDT`);

/** Gate.io spot WebSocket v4, `spot.tickers` channel. Ping via `spot.ping`. */
export const gateWs = defineConnector({
  metadata: {
    id: "gate-ws",
    name: "Gate — spot tickers (WS v4)",
    version: "1.0.0",
    sourceId: "gate",
    organization: "Gate Technology Inc.",
    sourceType: "WEBSOCKET",
    jurisdiction: null,
    rightsStatus: "PUBLIC_ATTRIBUTED",
    realtimeStatus: "REALTIME",
    expectedLatencyMs: 1000,
    supportsStreaming: true,
    supportsHistorical: false,
    assetClasses: ["CRYPTO"],
    exchanges: ["gate"],
    homepage: "https://www.gate.io/docs/developers/apiv4/ws/en/#tickers-channel",
    description: "Public Gate.io spot ticker channel (last, best bid/ask, 24h high/low/volume) for USDT majors.",
    rightsNotes: "Public market data displayed with attribution to Gate.",
    termsUrl: "https://www.gate.io/user-agreement",
    sourceFamily: "gate",
    enabled: true,
  },
  seeds: SYMBOLS.map((s) => {
    const [b, q] = s.split("_") as [string, string];
    return { symbol: s, hint: planPair(b, q, "gate").hint };
  }),
  defaultSymbols: SYMBOLS,
  async start(ctx) {
    const ws = ctx.openWebSocket(WS_URL, {
      label: "gate",
      staleAfterMs: 60_000,
      heartbeat: { intervalMs: 20_000, message: JSON.stringify({ time: Math.floor(Date.now() / 1000), channel: "spot.ping" }) },
      onOpen: (sock) => sock.send({ time: Math.floor(Date.now() / 1000), channel: "spot.tickers", event: "subscribe", payload: ctx.watchedSymbols() }),
      onMessage: (data) => {
        let msg: any;
        try {
          msg = JSON.parse(data);
        } catch {
          return;
        }
        if (msg?.channel === "spot.tickers" && msg.event === "update" && msg.result) ctx.emit(raw("gate-ws", "gate", "ticker", msg));
        else if (msg?.error) ctx.reportError(new Error(String(msg.error.message ?? "gate error")));
      },
    });
    ws.connect();
  },
  normalize(r): NormalizedBatch {
    const m = r.payload as { time_ms?: number; result?: Record<string, unknown> };
    const d = m.result;
    if (r.kind !== "ticker" || !d || typeof d.currency_pair !== "string") return { observations: [] };
    const [base, quote] = d.currency_pair.split("_");
    if (!base || !quote) return { observations: [] };
    return {
      observations: pairObservations(d.currency_pair, base, quote, "gate", { last: d.last, bid: d.highest_bid, ask: d.lowest_ask, high: d.high_24h, low: d.low_24h, volume: d.base_volume }, {
        sourceTimestamp: parseTimestamp(m.time_ms),
        timestampTrust: "EXCHANGE",
        rightsStatus: "PUBLIC_ATTRIBUTED",
        realtimeStatus: "REALTIME",
      }),
    };
  },
  fixturesDir: "fixtures",
});