const sub = client.newSubscription("l2:perps:BTC_USDC-PERPETUAL", { since: {} });
sub.on("publication", (ctx) => {
  // First publication = last known snapshot (delivered on subscribe)
  // Subsequent publications = live updates
  const { timestamp, bids, asks } = ctx.data;
});
sub.subscribe();

ws.onmessage = (e) => {
  const msg = JSON.parse(e.data);
  if (Object.keys(msg).length === 0) {
    ws.send("{}"); // pong — reply within 8 seconds
    return;
  }
  // ... handle other messages
};

const startTime = performance.now();
client.rpc("ping", {}).then(() => {
  const rtt = (performance.now() - startTime).toFixed(2);
  console.log(`RTT: ${rtt}ms`);
});

import { Centrifuge } from "centrifuge";

const client = new Centrifuge("wss://staging.kyan.sh/stream/websocket");
const sub = client.newSubscription("trades_perp:ETH_USDC-PERPETUAL");
let ready = false, buf = [], maxOffset = 0;

sub.on("subscribed", async () => {
  const h = await sub.history({ limit: -1, reverse: false });
  const pubs = h.publications || [];
  for (const pub of pubs) {
    const [price, signedSize, timestamp] = pub.data;
    maxOffset = Math.max(maxOffset, pub.offset ?? 0);
  }
  // Flush live trades that arrived during history fetch, skip duplicates
  for (const pub of buf) {
    if (pub.offset > maxOffset) {
      const [price, signedSize, timestamp] = pub.data;
    }
  }
  buf = [];
  ready = true;
});

sub.on("publication", (ctx) => {
  // price: "2251.50" (string, always positive)
  // signedSize: "1.5" (positive = buy) or "-1.5" (negative = sell)
  // timestamp: 1704067200000 (ms)
  if (!ready) { buf.push(ctx); return; }
  const [price, signedSize, timestamp] = ctx.data;
});

sub.subscribe();
client.connect();

import { Centrifuge } from "centrifuge/build/protobuf";

const client = new Centrifuge("wss://staging.kyan.sh/stream/websocket?format=protobuf");
const sub = client.newSubscription("trades_perp:ETH_USDC-PERPETUAL");

sub.on("publication", (ctx) => {
  const data = ctx.data instanceof Uint8Array
    ? JSON.parse(new TextDecoder().decode(ctx.data))
    : ctx.data;
  const [price, signedSize, timestamp] = data;
});

sub.subscribe();
client.connect();

const ws = new WebSocket("wss://staging.kyan.sh/stream/websocket");
let id = 0;
let historyLoaded = false;
let historyMaxOffset = 0;
const buf = [];

ws.onopen = () => ws.send(JSON.stringify({ id: ++id, connect: {} }));

ws.onmessage = (e) => {
  const msg = JSON.parse(e.data);

  // Ping — reply immediately
  if (Object.keys(msg).length === 0) { ws.send("{}"); return; }

  // Connect reply — subscribe first
  if (msg.id === 1) {
    ws.send(JSON.stringify({ id: ++id, subscribe: { channel: "trades_perp:ETH_USDC-PERPETUAL" } }));
    return;
  }

  // Subscribe reply — fetch history
  if (msg.id === 2) {
    ws.send(JSON.stringify({ id: ++id, history: { channel: "trades_perp:ETH_USDC-PERPETUAL", limit: -1 } }));
    return;
  }

  // History reply — last 100 trades, then flush buffered live trades
  if (msg.id === 3) {
    const pubs = (msg.history || msg.result).publications || [];
    for (const pub of pubs) {
      const [price, signedSize, timestamp] = pub.data;
      historyMaxOffset = Math.max(historyMaxOffset, pub.offset ?? 0);
    }
    // Flush live trades that arrived during history fetch, skip duplicates
    for (const t of buf) {
      if (t.offset > historyMaxOffset) {
        const [price, signedSize, timestamp] = t.data;
      }
    }
    buf.length = 0;
    historyLoaded = true;
    return;
  }

  // Live trades — buffer until history is loaded
  if (msg.push?.pub) {
    if (!historyLoaded) { buf.push(msg.push.pub); return; }
    const [price, signedSize, timestamp] = msg.push.pub.data;
  }
};

const response = await fetch("https://staging.kyan.sh/stream/sse", {
  method: "POST",
  body: JSON.stringify({ subs: { "trades_perp:ETH_USDC-PERPETUAL": {} } }),
});

const reader = response.body!.getReader();
const decoder = new TextDecoder();
let buffer = "";

while (true) {
  const { done, value } = await reader.read();
  if (done) break;
  buffer += decoder.decode(value, { stream: true });
  const lines = buffer.split("\n");
  buffer = lines.pop() ?? "";

  for (const line of lines) {
    if (!line.startsWith("data: ")) continue;
    const msg = JSON.parse(line.slice(6));
    if (msg.push?.pub) {
      const [price, signedSize, timestamp] = msg.push.pub.data;
    }
  }
}

const params = JSON.stringify({ subs: { "trades_perp:ETH_USDC-PERPETUAL": {} } });
const url = new URL("https://staging.kyan.sh/stream/sse");
url.searchParams.append("cf_connect", params);

const es = new EventSource(url);
es.onmessage = (e) => {
  const msg = JSON.parse(e.data);
  if (msg.push?.pub) {
    const [price, signedSize, timestamp] = msg.push.pub.data;
  }
};

const response = await fetch("https://staging.kyan.sh/stream/http", {
  method: "POST",
  body: JSON.stringify({ subs: { "trades_perp:ETH_USDC-PERPETUAL": {} } }),
});

const reader = response.body!.getReader();
const decoder = new TextDecoder();
let buffer = "";

while (true) {
  const { done, value } = await reader.read();
  if (done) break;
  buffer += decoder.decode(value, { stream: true });
  const lines = buffer.split("\n");
  buffer = lines.pop() ?? "";

  for (const line of lines) {
    if (!line.trim()) continue;
    const msg = JSON.parse(line);
    if (msg.push?.pub) {
      const [price, signedSize, timestamp] = msg.push.pub.data;
    }
  }
}