import { describe, expect, it } from "vitest";
import { dirname, join } from "node:path";
import { fileURLToPath } from "node:url";
import { loadFixture, makeTestContext, normalizeFixture, fakeFetch } from "@market-atlas/connector-sdk";
import { ConnectorMetadataSchema } from "@market-atlas/market-model";
import { CONNECTORS, SOURCES } from "./index.js";
import { coinbaseWs } from "./coinbase-ws/index.js";
import { krakenWs } from "./kraken-ws/index.js";
import { binanceWs } from "./binance-ws/index.js";
import { okxWs } from "./okx-ws/index.js";
import { cboeDelayedQuotes } from "./cboe-delayed-quotes/index.js";
import { ecbFrankfurter } from "./ecb-frankfurter/index.js";
import { bankOfCanada } from "./bank-of-canada/index.js";
import { usTreasuryYieldCurve } from "./us-treasury-yield-curve/index.js";
import { secEdgarFilings } from "./sec-edgar-filings/index.js";
import { secCompanyTickers, titleCase } from "./sec-company-tickers/index.js";
import { nasdaqSymbolDirectory, cleanName } from "./nasdaq-symbol-directory/index.js";
import { nasdaqTradeHalts } from "./nasdaq-trade-halts/index.js";
import { nasdaqMarketCalendar, parseLongDate } from "./nasdaq-market-calendar/index.js";
import { hfmarketdata } from "./hfmarketdata/index.js";

const here = dirname(fileURLToPath(import.meta.url));
const fx = (dir: string, name: string) => loadFixture(join(here, dir, "fixtures"), name);

describe("registry", () => {
  it("every connector has valid metadata, a known source, docs and fixtures", () => {
    const ids = new Set<string>();
    for (const c of CONNECTORS) {
      expect(ConnectorMetadataSchema.safeParse(c.metadata).success).toBe(true);
      expect(ids.has(c.metadata.id)).toBe(false);
      ids.add(c.metadata.id);
      expect(SOURCES.some((s) => s.id === c.metadata.sourceId)).toBe(true);
      expect(c.metadata.rightsStatus).not.toBe("UNKNOWN");
      expect(c.fixturesDir).toBe("fixtures");
    }
  });
  it("covers the V1 source-type diversity goal", () => {
    const types = new Set(CONNECTORS.map((c) => c.metadata.sourceType));
    for (const t of ["WEBSOCKET", "XHR", "OFFICIAL_API", "HTML", "RSS", "BULK_FILE", "XML"]) expect(types.has(t as never)).toBe(true);
  });
});

describe("crypto WebSocket connectors", () => {
  it("coinbase ticker → observations with exchange timestamp", () => {
    const b = normalizeFixture(coinbaseWs, "ticker", fx("coinbase-ws", "ticker.json"));
    const last = b.observations.find((o) => o.field === "LAST_PRICE")!;
    expect(last.value).toBe(77279.97);
    expect(last.symbol).toBe("BTC-USD");
    expect(last.instrumentHint?.base).toBe("BTC");
    expect(last.sourceTimestamp).toBe(Date.parse("2026-09-12T07:54:03.482921Z"));
    expect(last.timestampTrust).toBe("EXCHANGE");
    expect(b.observations.map((o) => o.field)).toContain("BID");
    expect(b.observations.map((o) => o.field)).toContain("VOLUME");
  });
  it("coinbase malformed payload yields no observations and does not throw", () => {
    const b = normalizeFixture(coinbaseWs, "ticker", fx("coinbase-ws", "malformed.json"));
    expect(b.observations).toHaveLength(0);
  });
  it("kraken snapshot → connector-timed observations", () => {
    const b = normalizeFixture(krakenWs, "ticker", fx("kraken-ws", "ticker.json"));
    const last = b.observations.find((o) => o.field === "LAST_PRICE")!;
    expect(last.value).toBe(77281.3);
    expect(last.sourceTimestamp).toBeNull();
    expect(last.timestampTrust).toBe("CONNECTOR");
    expect(b.observations.find((o) => o.field === "VWAP")?.value).toBe(77900.1);
  });
  it("binance miniTicker → USDT instrument", () => {
    const b = normalizeFixture(binanceWs, "miniTicker", fx("binance-ws", "miniTicker.json"));
    const last = b.observations.find((o) => o.field === "LAST_PRICE")!;
    expect(last.value).toBeCloseTo(77298.87);
    expect(last.instrumentHint?.quote).toBe("USDT");
    expect(last.sourceTimestamp).toBe(1789199657016);
  });
  it("okx tickers → observations", () => {
    const b = normalizeFixture(okxWs, "tickers", fx("okx-ws", "tickers.json"));
    expect(b.observations.find((o) => o.field === "LAST_PRICE")?.value).toBe(77303.9);
    expect(b.observations.find((o) => o.field === "ASK")?.value).toBe(77303.9);
    expect(b.observations[0]?.sourceTimestamp).toBe(1789199660123);
  });
  it("unexpected schema (wrong kind) is ignored", () => {
    expect(normalizeFixture(okxWs, "trades", { arg: { channel: "trades" }, data: [{ px: "1" }] }).observations).toHaveLength(0);
  });
  it("streaming connectors open exactly one socket and subscribe on open", async () => {
    for (const def of [coinbaseWs, krakenWs, binanceWs, okxWs]) {
      const ctx = makeTestContext(def);
      await def.start!(ctx);
      expect(ctx.sockets).toHaveLength(1);
      ctx.sockets[0]!.close();
    }
  });
});

describe("cboe-delayed-quotes", () => {
  it("equity quote → DELAYED observations with Eastern timestamps converted to UTC", () => {
    const b = normalizeFixture(cboeDelayedQuotes, "quote", fx("cboe-delayed-quotes", "quote.json"));
    const last = b.observations.find((o) => o.field === "LAST_PRICE")!;
    expect(last.value).toBe(332.58);
    expect(last.realtimeStatus).toBe("DELAYED");
    expect(last.rightsStatus).toBe("DELAYED");
    // 2026-09-11T15:59:59 America/New_York (EDT, UTC-4) → 19:59:59Z
    expect(new Date(last.sourceTimestamp!).toISOString()).toBe("2026-09-11T19:59:59.000Z");
    expect(b.observations.find((o) => o.field === "PREVIOUS_CLOSE")?.value).toBe(332.27);
    expect(b.observations.find((o) => o.field === "VOLUME")?.value).toBe(50716865);
    expect(b.observations.find((o) => o.field === "IMPLIED_VOLATILITY")?.value).toBe(23.849);
  });
  it("index quote keeps the _SPX source symbol and INDEX hint", () => {
    const b = normalizeFixture(cboeDelayedQuotes, "quote", fx("cboe-delayed-quotes", "index.json"));
    const last = b.observations.find((o) => o.field === "LAST_PRICE")!;
    expect(last.symbol).toBe("_SPX");
    expect(last.instrumentHint?.assetClass).toBe("INDEX");
    expect(b.observations.find((o) => o.field === "VOLUME")?.value).toBe(0);
  });
  it("never-traded symbol (zeros) yields no price observations", () => {
    const b = normalizeFixture(cboeDelayedQuotes, "quote", fx("cboe-delayed-quotes", "closed_market.json"));
    expect(b.observations.filter((o) => o.field === "LAST_PRICE" || o.field === "BID")).toHaveLength(0);
  });
  it("poll tolerates a failing symbol", async () => {
    const ctx = makeTestContext(cboeDelayedQuotes, {
      symbols: ["AAPL", "BROKEN"],
      fetchImpl: fakeFetch({ "/AAPL.json": JSON.stringify(fx("cboe-delayed-quotes", "quote.json")), "/BROKEN.json": { status: 500, body: "boom" } }),
    });
    const raws = await cboeDelayedQuotes.poll!(ctx);
    expect(raws).toHaveLength(1);
    expect(ctx.errors).toHaveLength(1);
  });
});

describe("official sources", () => {
  it("frankfurter USD base → conventional pairs, latest + previous close, EUR pairs skipped", () => {
    const b = normalizeFixture(ecbFrankfurter, "timeseries", fx("ecb-frankfurter", "timeseries.json"));
    const usdcad = b.observations.filter((o) => o.symbol === "USDCAD");
    expect(usdcad.find((o) => o.field === "LAST_PRICE")?.value).toBe(1.3858);
    expect(usdcad.find((o) => o.field === "PREVIOUS_CLOSE")?.value).toBe(1.3871);
    expect(b.observations.some((o) => o.symbol === "EURUSD")).toBe(false); // EUR pairs come from the EUR-base call
    const gbpusd = b.observations.find((o) => o.symbol === "GBPUSD" && o.field === "LAST_PRICE")!;
    expect(gbpusd.value).toBeCloseTo(1 / 0.7403, 6);
    expect(gbpusd.realtimeStatus).toBe("END_OF_DAY");
    expect(gbpusd.rightsStatus).toBe("OFFICIAL_OPEN_DATA");
  });
  it("bank of canada fx + rates", () => {
    const fxb = normalizeFixture(bankOfCanada, "fx", fx("bank-of-canada", "fx.json"));
    expect(fxb.observations.find((o) => o.symbol === "USDCAD" && o.field === "LAST_PRICE")?.value).toBe(1.3855);
    expect(fxb.observations.find((o) => o.symbol === "EURCAD" && o.field === "PREVIOUS_CLOSE")?.value).toBe(1.606);
    const rates = normalizeFixture(bankOfCanada, "rates", fx("bank-of-canada", "rates.json"));
    expect(rates.observations.find((o) => o.symbol === "CA_POLICY_RATE" && o.field === "RATE")?.value).toBe(2.25);
    expect(rates.observations.find((o) => o.symbol === "CA_10Y" && o.field === "YIELD")?.value).toBe(3.24);
    expect(rates.observations.find((o) => o.symbol === "CA_10Y" && o.field === "PREVIOUS_CLOSE")?.value).toBe(3.21);
  });
  it("treasury XML → 13 tenors with yields and previous close", () => {
    const b = normalizeFixture(usTreasuryYieldCurve, "month", fx("us-treasury-yield-curve", "month.xml"));
    const us10 = b.observations.filter((o) => o.symbol === "US10Y");
    expect(us10.find((o) => o.field === "YIELD")?.value).toBe(4.02);
    expect(us10.find((o) => o.field === "PREVIOUS_CLOSE")?.value).toBe(4.05);
    expect(us10.find((o) => o.field === "YIELD")?.instrumentHint?.assetClass).toBe("TREASURY");
    expect(b.observations.some((o) => o.symbol === "US4M")).toBe(false); // absent tenor is skipped
  });
});

describe("regulatory / directory / halts", () => {
  it("edgar atom → filings; Form 4 reporting is not material, 8-K is", () => {
    const b = normalizeFixture(secEdgarFilings, "atom", fx("sec-edgar-filings", "atom.xml"));
    expect(b.filings).toHaveLength(2);
    const f4 = b.filings!.find((f) => f.formType === "4")!;
    expect(f4.cik).toBe("1734770");
    expect(f4.metadata?.material).toBe(false);
    const k8 = b.filings!.find((f) => f.formType === "8-K")!;
    expect(k8.companyName).toBe("Apple Inc.");
    expect(k8.cik).toBe("320193");
    expect(k8.metadata?.material).toBe(true);
    expect(k8.url).toContain("0000320193-26-000099");
  });
  it("sec directory → listed created, OTC enrich-only", () => {
    const b = normalizeFixture(secCompanyTickers, "directory", fx("sec-company-tickers", "directory.json"));
    const brk = b.instruments!.find((i) => i.symbol === "BRK.B")!;
    expect(brk.hint.exchangeId).toBe("xnys");
    expect(brk.hint.cik).toBe("1067983");
    expect(brk.aliases).toContain("BRK-B");
    expect(b.instruments!.find((i) => i.symbol === "SOTC")?.createIfMissing).toBe(false);
    expect(titleCase("MICROSOFT CORP")).toBe("Microsoft Corp");
    expect(titleCase("Apple Inc.")).toBe("Apple Inc.");
  });
  it("nasdaq directories → instruments with venues, ETF flags and no test issues", () => {
    const n = normalizeFixture(nasdaqSymbolDirectory, "nasdaqlisted", fx("nasdaq-symbol-directory", "nasdaqlisted.txt"));
    expect(n.instruments!.map((i) => i.symbol)).toEqual(["AAAP", "AACG", "AAPL", "QQQ"]);
    expect(n.instruments!.find((i) => i.symbol === "AAPL")?.hint.name).toBe("Apple Inc.");
    expect(n.instruments!.find((i) => i.symbol === "QQQ")?.hint.assetClass).toBe("ETF");
    expect(n.instruments!.find((i) => i.symbol === "AACG")?.hint.securityType).toBe("ADR");
    const o = normalizeFixture(nasdaqSymbolDirectory, "otherlisted", fx("nasdaq-symbol-directory", "otherlisted.txt"));
    const brk = o.instruments!.find((i) => i.symbol === "BRK.B")!;
    expect(brk.hint.exchangeId).toBe("xnys");
    expect(brk.aliases).toEqual(expect.arrayContaining(["BRK-B", "BRK B"]));
    expect(o.instruments!.find((i) => i.symbol === "SPY")?.hint.exchangeId).toBe("arcx");
    expect(o.instruments!.find((i) => i.symbol === "BAC-PB")?.hint.securityType).toBe("PREFERRED");
    expect(cleanName("Alcoa Corporation Common Stock ")).toBe("Alcoa Corporation");
  });
  it("halts rss → halt + resume events with Eastern → UTC conversion", () => {
    const b = normalizeFixture(nasdaqTradeHalts, "rss", fx("nasdaq-trade-halts", "rss.xml"));
    const halts = b.events!.filter((e) => e.type === "TRADING_HALT");
    expect(halts).toHaveLength(2);
    const hubc = halts.find((e) => e.symbols?.[0] === "HUBC")!;
    expect(new Date(hubc.timestamp).toISOString()).toBe("2026-09-11T23:50:00.000Z");
    expect(hubc.data?.reason).toBe("News pending");
    const resume = b.events!.find((e) => e.type === "TRADING_RESUME")!;
    expect(resume.symbols).toEqual(["ABCD"]);
    expect(new Date(resume.timestamp).toISOString()).toBe("2026-09-11T14:22:31.000Z");
    expect(b.events!.find((e) => e.symbols?.[0] === "ABCD" && e.type === "TRADING_HALT")?.severity).toBe("NOTICE");
  });
  it("calendar html → holidays for every US venue, early close parsed, change event only when changed", () => {
    const html = fx("nasdaq-market-calendar", "page.html") as string;
    const first = nasdaqMarketCalendar.normalize({ connectorId: "nasdaq-market-calendar", sourceId: "nasdaq-trader", kind: "page", payload: html, receivedAt: Date.now(), meta: { changed: true, first: true } });
    expect(first.events).toHaveLength(0);
    expect(first.holidays!.filter((h) => h.exchangeId === "xnys")).toHaveLength(7);
    const early = first.holidays!.find((h) => h.date === "2026-11-27" && h.exchangeId === "xnas")!;
    expect(early.kind).toBe("EARLY_CLOSE");
    expect(early.closeTime).toBe("13:00");
    expect(first.holidays!.find((h) => h.date === "2026-01-19")?.name).toBe("Martin Luther King, Jr. Day");
    const changed = nasdaqMarketCalendar.normalize({ connectorId: "nasdaq-market-calendar", sourceId: "nasdaq-trader", kind: "page", payload: html, receivedAt: Date.now(), meta: { changed: true, first: false, document_hash: "abc", changed_sections: ["x"] } });
    expect(changed.events).toHaveLength(1);
    expect(changed.events![0]!.type).toBe("DOCUMENT_CHANGED");
    expect(parseLongDate("December 25, 2026")).toBe("2026-12-25");
  });
});

describe("hfmarketdata-daily", () => {
  it("daily bars → 1d bars + END_OF_DAY reference observations", () => {
    const b = hfmarketdata.normalize({ connectorId: "hfmarketdata-daily", sourceId: "hfmarketdata", kind: "bars", payload: fx("hfmarketdata", "bars.json"), receivedAt: Date.now(), meta: { symbol: "AAPL" } });
    expect(b.bars).toHaveLength(3);
    expect(b.bars![0]!.ts).toBe(Date.parse("2026-09-01T00:00:00Z"));
    const last = b.observations.find((o) => o.field === "LAST_PRICE")!;
    expect(last.value).toBe(328.21);
    expect(last.realtimeStatus).toBe("END_OF_DAY");
    expect(b.observations.find((o) => o.field === "PREVIOUS_CLOSE")?.value).toBe(324.96);
    // 2026-09-03 16:00 America/New_York → 20:00Z
    expect(new Date(last.sourceTimestamp!).toISOString()).toBe("2026-09-03T20:00:00.000Z");
  });
  it("continuous futures → COMMODITY instrument with contract metadata", () => {
    const b = hfmarketdata.normalize({ connectorId: "hfmarketdata-daily", sourceId: "hfmarketdata", kind: "continuous", payload: fx("hfmarketdata", "continuous.json"), receivedAt: Date.now(), meta: { symbol: "CL=F" } });
    expect(b.bars).toHaveLength(2);
    const last = b.observations.find((o) => o.field === "LAST_PRICE")!;
    expect(last.value).toBe(91.01);
    expect(last.instrumentHint?.assetClass).toBe("COMMODITY");
    expect(last.meta?.contract).toBe("CLV26");
  });
  it("unknown symbol meta yields nothing", () => {
    expect(hfmarketdata.normalize({ connectorId: "hfmarketdata-daily", sourceId: "hfmarketdata", kind: "bars", payload: { data: [] }, receivedAt: 0, meta: { symbol: "NOPE" } }).observations).toHaveLength(0);
  });
});

describe("source mesh v2 — venues, live FX and proxies", async () => {
  const { bitstampWs } = await import("./bitstamp-ws/index.js");
  const { geminiWs } = await import("./gemini-ws/index.js");
  const { bitfinexWs } = await import("./bitfinex-ws/index.js");
  const { bybitWs } = await import("./bybit-ws/index.js");
  const { gateWs } = await import("./gate-ws/index.js");
  const { cryptocomWs } = await import("./cryptocom-ws/index.js");
  const { kucoinWs } = await import("./kucoin-ws/index.js");
  const { nasdaqQuoteApi, parseNasdaqTime } = await import("./nasdaq-quote-api/index.js");
  const { planPair } = await import("./_shared/pairs.js");

  it("plans venue pairs: crypto, real fiat markets, stablecoin proxies (with inversion)", () => {
    expect(planPair("BTC", "USD", "x").hint.assetClass).toBe("CRYPTO");
    const fx = planPair("EUR", "USD", "kraken");
    expect(fx.hint.assetClass).toBe("FOREX");
    expect(fx.type).toBe("TRADE");
    const inv = planPair("USDT", "EUR", "okx");
    expect(inv.type).toBe("STABLECOIN_PROXY");
    expect(inv.hint.base).toBe("EUR");
    expect(inv.invert).toBe(true);
    const brl = planPair("USDT", "BRL", "binance");
    expect(brl.hint.base).toBe("USD");
    expect(brl.hint.quote).toBe("BRL");
    expect(brl.invert).toBe(false);
  });
  it("kraken fiat ticker → FOREX TRADE observation on the conventional pair", () => {
    const b = normalizeFixture(krakenWs, "ticker", { channel: "ticker", type: "update", data: [{ symbol: "EUR/USD", bid: 1.1724, ask: 1.1726, last: 1.1725, volume: 120000.5, high: 1.175, low: 1.17 }] });
    const last = b.observations.find((o) => o.field === "LAST_PRICE")!;
    expect(last.instrumentHint?.assetClass).toBe("FOREX");
    expect(last.observationType).toBe("TRADE");
    expect(last.value).toBe(1.1725);
  });
  it("okx USDT-EUR ticker → inverted EURUSD stablecoin proxy", () => {
    const b = normalizeFixture(okxWs, "tickers", { arg: { channel: "tickers", instId: "USDT-EUR" }, data: [{ instId: "USDT-EUR", last: "0.8530", bidPx: "0.8529", askPx: "0.8531", ts: "1789199660123" }] });
    const last = b.observations.find((o) => o.field === "LAST_PRICE")!;
    expect(last.instrumentHint?.base).toBe("EUR");
    expect(last.observationType).toBe("STABLECOIN_PROXY");
    expect(last.value).toBeCloseTo(1 / 0.853, 6);
    const bid = b.observations.find((o) => o.field === "BID")!;
    expect(bid.value).toBeCloseTo(1 / 0.8531, 6); // venue ask becomes our bid
    expect(b.observations.some((o) => o.field === "VOLUME")).toBe(false);
  });
  it("bitstamp trade + book (EUR/USD fiat market)", () => {
    const t = normalizeFixture(bitstampWs, "trade", fx("bitstamp-ws", "trade.json"));
    expect(t.observations[0]).toMatchObject({ field: "LAST_PRICE", value: 77203.59, observationType: "TRADE" });
    expect(t.observations[0]!.sourceTimestamp).toBe(1789273894413);
    const bk = normalizeFixture(bitstampWs, "book", fx("bitstamp-ws", "book.json"));
    expect(bk.observations.find((o) => o.field === "BID")?.value).toBe(1.17252);
    expect(bk.observations[0]!.instrumentHint?.assetClass).toBe("FOREX");
  });
  it("gemini, bitfinex, bybit, gate, crypto.com, kucoin tickers", () => {
    expect(normalizeFixture(geminiWs, "trade", fx("gemini-ws", "trade.json")).observations[0]).toMatchObject({ field: "LAST_PRICE", value: 77212.78, sourceTimestamp: 1789274085234 });
    const bf = normalizeFixture(bitfinexWs, "ticker", fx("bitfinex-ws", "ticker.json"));
    expect(bf.observations.find((o) => o.field === "LAST_PRICE")?.value).toBe(77118);
    expect(bf.observations.find((o) => o.field === "BID")?.value).toBe(77123);
    expect(normalizeFixture(bybitWs, "ticker", fx("bybit-ws", "ticker.json")).observations.find((o) => o.field === "LAST_PRICE")?.value).toBe(77229.4);
    expect(normalizeFixture(gateWs, "ticker", fx("gate-ws", "ticker.json")).observations.find((o) => o.field === "ASK")?.value).toBe(77227.8);
    expect(normalizeFixture(cryptocomWs, "ticker", fx("cryptocom-ws", "ticker.json")).observations.find((o) => o.field === "LAST_PRICE")?.value).toBe(77219.2);
    const ku = normalizeFixture(kucoinWs, "ticker", fx("kucoin-ws", "ticker.json"));
    expect(ku.observations.find((o) => o.field === "LAST_PRICE")?.value).toBe(77225.1);
    expect(ku.observations[0]!.sourceTimestamp).toBe(1789274087087);
  });
  it("nasdaq.com quote is a restricted validator with parsed Eastern timestamps", () => {
    const b = normalizeFixture(nasdaqQuoteApi, "info", fx("nasdaq-quote-api", "info.json"));
    const last = b.observations.find((o) => o.field === "LAST_PRICE")!;
    expect(last.value).toBe(332.27);
    expect(last.rightsStatus).toBe("PUBLIC_RESTRICTED_REDISTRIBUTION");
    expect(new Date(last.sourceTimestamp!).toISOString()).toBe("2026-09-10T20:00:00.000Z");
    expect(b.observations.find((o) => o.field === "VOLUME")?.value).toBe(50716997);
    expect(parseNasdaqTime("Sep 12, 2026 9:45 AM ET")).toBe(Date.parse("2026-09-12T13:45:00Z"));
    expect(parseNasdaqTime("garbage")).toBeNull();
  });
  it("every streaming connector still opens exactly one socket", async () => {
    for (const def of [bitstampWs, geminiWs, bitfinexWs, bybitWs, gateWs, cryptocomWs]) {
      const ctx = makeTestContext(def);
      await def.start!(ctx);
      expect(ctx.sockets).toHaveLength(1);
      ctx.sockets[0]!.close();
    }
  });
});