import { afterEach, describe, expect, it, vi } from "vitest";
import type { Backend, Position, Summary, Trade } from "../../src/api";
import { DeskStream, type FrameScheduler } from "../../src/live/controller";
import type { StreamEvent, StreamFactory } from "../../src/live/types";

const address = `0x${"a1".repeat(20)}`,
  now = 1800000000000;
const summary: Summary = {
  address,
  equity: 100,
  pnl: 0,
  paperMinted: 0,
  stakingRewards: 0,
  asOf: now,
};
const position: Position = {
  positionId: "p",
  symbol: "ETH",
  isLong: true,
  entry: 100,
  mark: 100,
  bustPrice: 99.05,
  marginUsd: 100,
  leverage: 100,
  sizeUsd: 10000,
  pnlUsd: 0,
  pnlPct: 0,
  openedAt: now,
};
const closedTrade: Trade = {
  ...position,
  kind: "Closed",
  exit: 101,
  pnlUsd: 100,
  feeUsd: 0,
  paperMinted: 0,
  closedAt: now + 700,
  txHash: null,
};
function fixture() {
  vi.useFakeTimers();
  vi.setSystemTime(now);
  const tradesStore: { items: Trade[] } = { items: [] };
  const backend = {
    mock: false,
    summary: vi.fn().mockResolvedValue(summary),
    positions: vi.fn().mockResolvedValue([position]),
    trades: vi.fn(async () => ({ items: tradesStore.items, nextCursor: null })),
    cashflows: vi.fn().mockResolvedValue({ items: [], nextCursor: null }),
    markets: vi.fn().mockResolvedValue([{ symbol: "ETH", mark: 100 }]),
  } as unknown as Backend;
  let receive: (event: StreamEvent) => void = () => {},
    error = () => {};
  const factory = vi.fn<StreamFactory>((_a, _s, r, e) => {
    receive = r;
    error = e;
    return { close: vi.fn() };
  });
  let frame: (() => void) | null = null;
  const scheduler: FrameScheduler = {
    schedule: (fn) => {
      frame = fn;
      return () => {
        frame = null;
      };
    },
  };
  const store = new DeskStream(backend, address, "ETH", scheduler, factory);
  const send = (e: Omit<StreamEvent, "receivedAt">) =>
    receive({ ...e, receivedAt: Date.now() } as StreamEvent);
  const flush = () => {
    const fn = frame;
    frame = null;
    fn?.();
  };
  return { store, backend, tradesStore, send, flush, error: () => error() };
}
afterEach(() => vi.useRealTimers());

describe("reconnect resync", () => {
  it("a trade that closes between the fallback poll and the resumed stream is never shown", async () => {
    const f = fixture();
    await f.store.start();
    f.send({ type: "mark", data: { symbol: "ETH", price: 100, t: now } });
    f.send({ type: "positions", data: { address, positions: [position] } });
    f.flush();
    expect(f.store.getSnapshot().live.status).toBe("LIVE");
    // Transport drops at T0: the controller polls once immediately, then reconnects after 500ms.
    f.error();
    await vi.advanceTimersByTimeAsync(100);
    // The position closes on Papertrade at T0+200ms, after the fallback GET /trades answered.
    f.tradesStore.items = [closedTrade];
    await vi.advanceTimersByTimeAsync(400);
    // The resumed stream sends the hub's latest snapshot (no closed trade: the server only
    // stores positions/summary/mark in hub.latest) and fresh marks: the stream is healthy.
    f.send({
      type: "positions",
      data: { address, positions: [] },
    });
    f.send({ type: "mark", data: { symbol: "ETH", price: 101, t: Date.now() } });
    f.flush();
    const live = f.store.getSnapshot().live;
    expect(live.status).toBe("LIVE");
    expect(live.positions).toEqual([]);
    // Fallback polling has stopped; the healthy stream keeps delivering marks for 30s.
    for (let i = 0; i < 150; i++) {
      await vi.advanceTimersByTimeAsync(200);
      f.send({ type: "mark", data: { symbol: "ETH", price: 101, t: Date.now() } });
    }
    f.flush();
    const trades = f.store.getSnapshot().live.trades ?? [];
    // Expected: the closed trade is in history (and its exit marker on the chart).
    expect(trades.map((t) => t.positionId)).toEqual(["p"]);
    f.store.stop();
  });
});

import { afterEach, describe, expect, it, vi } from "vitest";
import type { Backend, Summary } from "../../src/api";
import { DeskStream, type FrameScheduler } from "../../src/live/controller";
import { decodeEvent } from "../../src/live/transport";
import type { StreamEvent, StreamFactory } from "../../src/live/types";

const address = `0x${"a1".repeat(20)}`,
  now = 1800000000000;
const summary: Summary = {
  address,
  equity: 100,
  pnl: 0,
  paperMinted: 0,
  stakingRewards: 0,
  asOf: now,
};
function fixture() {
  vi.useFakeTimers();
  vi.setSystemTime(now);
  const backend = {
    mock: false,
    summary: vi.fn().mockResolvedValue(summary),
    positions: vi.fn().mockResolvedValue([]),
    trades: vi.fn().mockResolvedValue({ items: [], nextCursor: null }),
    cashflows: vi.fn().mockResolvedValue({ items: [], nextCursor: null }),
    markets: vi.fn().mockResolvedValue([{ symbol: "ETH", mark: 100 }]),
  } as unknown as Backend;
  let receive: (event: StreamEvent) => void = () => {};
  const factory = vi.fn<StreamFactory>((_a, _s, r) => {
    receive = r;
    return { close: vi.fn() };
  });
  let frame: (() => void) | null = null;
  const scheduler: FrameScheduler = {
    schedule: (fn) => {
      frame = fn;
      return () => {
        frame = null;
      };
    },
  };
  const store = new DeskStream(backend, address, "ETH", scheduler, factory);
  const send = (e: Omit<StreamEvent, "receivedAt">) =>
    receive({ ...e, receivedAt: Date.now() } as StreamEvent);
  const flush = () => {
    const fn = frame;
    frame = null;
    fn?.();
  };
  return { store, backend, send, flush };
}
afterEach(() => vi.useRealTimers());

describe("client clock skew", () => {
  it("client 2s ahead of the mark source: stream never reaches LIVE and never falls back", async () => {
    const f = fixture();
    await f.store.start();
    // Marks arrive every 200ms with the server's own timestamp, which is 2s behind this client.
    for (let i = 0; i < 100; i++) {
      f.send({
        type: "mark",
        data: { symbol: "ETH", price: 100 + i, t: Date.now() - 2000 },
      });
      f.send({ type: "positions", data: { address, positions: [] } });
      await vi.advanceTimersByTimeAsync(200);
    }
    f.flush();
    const live = f.store.getSnapshot().live;
    expect(live.mark?.price).toBe(199); // marks are flowing and folded
    expect(live.fallback).toBe(false); // no fallback polling either
    expect(live.status).toBe("LIVE"); // actual: "CONNECTING" forever (UI shows LOADING)
    f.store.stop();
  });
  it("client 6s behind the mark source: every mark is discarded at the trust boundary", () => {
    const raw = JSON.stringify({ symbol: "ETH", price: 100, t: now + 6000 });
    expect(decodeEvent("mark", raw, "1", now)).not.toBeNull();
  });
});