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

const WS_URL = "wss://ws.kraken.com/v2";
const CRYPTO_PAIRS: Array<[string, string]> = [...MAJOR_BASES.filter((b) => b !== "PAXG").map((b) => [b, "USD"] as [string, string]), ["BTC", "USDT"], ["ETH", "USDT"], ["BTC", "EUR"], ["ETH", "EUR"]];
/** Kraken runs real fiat/fiat spot markets — a live, keyless FX source. */
const FIAT_PAIRS: Array<[string, string]> = [["EUR", "USD"], ["GBP", "USD"], ["USD", "JPY"], ["USD", "CAD"], ["USD", "CHF"], ["AUD", "USD"], ["EUR", "GBP"], ["EUR", "JPY"], ["EUR", "CHF"], ["EUR", "CAD"], ["EUR", "AUD"], ["AUD", "JPY"]];
const PAIRS = [...CRYPTO_PAIRS, ...FIAT_PAIRS];
const SYMBOLS = PAIRS.map(([b, q]) => `${b}/${q}`);
const fiatSeeds: ProposedInstrument[] = FIAT_PAIRS.map(([b, q]) => ({ symbol: `${b}/${q}`, hint: planPair(b, q, "kraken").hint }));

/** Kraken WebSocket API v2, `ticker` channel (event trigger: trades). No per-message timestamp → connector receive time. */
export const krakenWs = defineConnector({
  metadata: {
    id: "kraken-ws",
    name: "Kraken — ticker feed (WS v2)",
    version: "1.0.0",
    sourceId: "kraken",
    organization: "Payward, Inc.",
    sourceType: "WEBSOCKET",
    jurisdiction: "US",
    rightsStatus: "PUBLIC_ATTRIBUTED",
    realtimeStatus: "REALTIME",
    expectedLatencyMs: 400,
    supportsStreaming: true,
    supportsHistorical: false,
    assetClasses: ["CRYPTO", "FOREX"],
    exchanges: ["kraken"],
    homepage: "https://docs.kraken.com/api/docs/websocket-v2/ticker",
    description: "Public Kraken WebSocket v2 ticker channel (best bid/ask, last trade, 24h volume/VWAP/high/low) for the major crypto spot pairs and Kraken's 12 fiat/fiat markets (EUR/USD, USD/JPY, GBP/USD, USD/CAD…) — a live, keyless FX source.",
    rightsNotes: "Public market data displayed with attribution to Kraken.",
    termsUrl: "https://www.kraken.com/legal",
    sourceFamily: "kraken",
    enabled: true,
  },
  seeds: [...cryptoSeeds(CRYPTO_PAIRS, "kraken", (b, q) => `${b}/${q}`), ...fiatSeeds],
  defaultSymbols: SYMBOLS,
  async start(ctx) {
    const ws = ctx.openWebSocket(WS_URL, {
      label: "kraken",
      staleAfterMs: 90_000,
      heartbeat: { intervalMs: 25_000, message: JSON.stringify({ method: "ping" }) },
      onOpen: (sock) => sock.send({ method: "subscribe", params: { channel: "ticker", symbol: ctx.watchedSymbols(), event_trigger: "trades", snapshot: true } }),
      onMessage: (data) => {
        let msg: any;
        try {
          msg = JSON.parse(data);
        } catch {
          return;
        }
        if (msg?.channel === "ticker" && Array.isArray(msg.data)) ctx.emit(raw("kraken-ws", "kraken", "ticker", msg));
        else if (msg?.method === "subscribe" && msg.success === false) ctx.reportError(new Error(String(msg.error ?? "subscribe failed")), { symbol: msg.result?.symbol });
      },
    });
    ws.connect();
  },
  normalize(r): NormalizedBatch {
    const m = r.payload as { channel?: string; data?: Array<Record<string, unknown>> };
    if (r.kind !== "ticker" || m.channel !== "ticker" || !Array.isArray(m.data)) return { observations: [] };
    const observations = [];
    for (const t of m.data) {
      if (typeof t.symbol !== "string") continue;
      const [base, quote] = t.symbol.split("/");
      if (!base || !quote) continue;
      observations.push(
        ...pairObservations(t.symbol, base, quote, "kraken", { last: t.last, high: t.high, low: t.low, bid: t.bid, ask: t.ask, bidSize: t.bid_qty, askSize: t.ask_qty, volume: t.volume, vwap: t.vwap }, {
          sourceTimestamp: null,
          timestampTrust: "CONNECTOR",
          rightsStatus: "PUBLIC_ATTRIBUTED",
          realtimeStatus: "REALTIME",
        }),
      );
    }
    return { observations };
  },
  fixturesDir: "fixtures",
});