Real-time Data

import { Centrifuge } from "centrifuge";

const RELAYER_URL = "https://api.sx.bet"; // Mainnet — use https://api.toronto.sx.bet for testnet
const WS_URL = "wss://realtime.sx.bet/connection/websocket"; // Mainnet — use wss://realtime.toronto.sx.bet/connection/websocket for testnet

async function fetchToken(apiKey) {
  const res = await fetch(`${RELAYER_URL}/user/realtime-token/api-key`, {
    headers: { "x-api-key": apiKey },
  });
  if (!res.ok) throw new Error(`Token endpoint returned ${res.status}`);
  const { token } = await res.json();
  return token;
}

const client = new Centrifuge(WS_URL, {
  getToken: () => fetchToken(YOUR_API_KEY),
});

client.connect();

const sub = client.newSubscription("markets:global");

sub.on("publication", (ctx) => {
  console.log(ctx.data);
});

sub.subscribe();

// Correct — both flags together
const sub = client.newSubscription("order_book:market_abc123", {
  positioned: true,
  recoverable: true,
});

sub.on("subscribed", (ctx) => {
  if (ctx.wasRecovering && ctx.recovered) {
    // Reconnected and gap was filled via history replay.
    // No need to re-fetch from REST — all missed messages were replayed.
  } else if (ctx.wasRecovering && !ctx.recovered) {
    // Reconnected but history was pruned before recovery could complete.
    // Too much time passed — re-seed your local state from REST.
  } else {
    // Fresh connect (first connection, or after a clean disconnect).
    // Seed initial state from REST, then rely on the subscription for updates.
  }
});

const seen = new Set();
const MAX_SEEN = 10_000;

sub.on("publication", (ctx) => {
  const id = ctx.tags?.messageId;
  if (id !== undefined) {
    if (seen.has(id)) return;
    seen.add(id);
    if (seen.size > MAX_SEEN) {
      seen.delete(seen.values().next().value);
    }
  }
  applyUpdate(ctx.data);
});

const resp = await sub.history({ limit: 50 });
for (const pub of resp.publications) {
  applyUpdate(pub.data);
}

let ready = false;
const buffer = [];

const sub = client.newSubscription(`order_book:market_${marketHash}`, {
  positioned: true,
  recoverable: true,
});

sub.on("publication", (ctx) => {
  if (!ready) {
    buffer.push(ctx.data);
  } else {
    applyUpdate(ctx.data);
  }
});

sub.on("subscribed", async (ctx) => {
  if (ctx.wasRecovering && ctx.recovered) {
    // Centrifugo replayed all missed messages — no REST call needed.
    ready = true;
    return;
  }

  // Fresh connect or failed recovery (history gap too large / expired).
  // Reset and re-seed from REST.
  ready = false;
  buffer.length = 0;

  const res = await fetch(`https://api.sx.bet/orders?marketHashes=${marketHash}`);
  applySnapshot(await res.json());

  // Drain buffered publications on top of the snapshot.
  // applyUpdate deduplicates by entity ID so any
  // overlap between the snapshot and the buffer is handled safely.
  for (const data of buffer) applyUpdate(data);
  buffer.length = 0;
  ready = true;
});

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

import { Centrifuge } from "centrifuge";

const client = new Centrifuge("wss://realtime.sx.bet/connection/websocket", {
  getToken: () => fetchToken(YOUR_API_KEY), // see Getting started above
});

const sub = client.newSubscription("markets:global");

sub.on("publication", (ctx) => {
  for (const market of ctx.data) {
    console.log(`${market.marketHash}: status=${market.status}`);
  }
});

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

async function subscribeOrderBook(client, marketHash) {
  const sub = client.newSubscription(`order_book:market_${marketHash}`, {
    positioned: true,
    recoverable: true,
  });

  sub.on("subscribed", async (ctx) => {
    if (!ctx.wasRecovering || (ctx.wasRecovering && !ctx.recovered)) {
      // Fresh connect or failed recovery — seed state from REST
      const orders = await fetchOrdersFromRest(marketHash);
      applySnapshot(orders);
    }
    // wasRecovering && recovered: history replay handled it, nothing to do
  });

  sub.on("publication", (ctx) => {
    applyUpdate(ctx.data);
  });

  sub.subscribe();
}

import { Centrifuge } from "centrifuge";

const client = new Centrifuge("wss://realtime.sx.bet/connection/websocket", {
  getToken: () => fetchToken(YOUR_API_KEY),
});

const sub = client.newSubscription(`active_orders:${YOUR_ADDRESS}`, {
  positioned: true,
  recoverable: true,
});

sub.on("subscribed", async (ctx) => {
  if (!ctx.wasRecovering || (ctx.wasRecovering && !ctx.recovered)) {
    // Seed current open orders from REST on fresh connect or failed recovery
    const orders = await fetchActiveOrdersFromRest(YOUR_ADDRESS);
    applySnapshot(orders);
  }
});

sub.on("publication", (ctx) => {
  for (const update of ctx.data) {
    console.log(`order ${update.orderHash}: ${update.status}`);
    applyUpdate(update);
  }
});

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

import { Centrifuge, UnauthorizedError } from "centrifuge";

const client = new Centrifuge(WS_URL, {
  getToken: async () => {
    const res = await fetch(`${RELAYER_URL}/user/realtime-token/api-key`, {
      headers: { "x-api-key": apiKey },
    });
    if (res.status === 401 || res.status === 403) {
      throw new UnauthorizedError(); // permanent — stops all reconnect attempts
    }
    if (!res.ok) throw new Error(`Status ${res.status}`); // transient — retries with backoff
    const { token } = await res.json();
    return token;
  },
});

sub.on("error", (ctx) => {
  console.error(ctx.error.code, ctx.error.message);
});