Next.js WebSocket tutorial: build a real-time market dashboard

markets-dashboard/
  .env.local                      TICKERLAYER_API_KEY=... (never committed)
  lib/watchlist.ts                the instruments, shared by server and client
  lib/market-hub.ts               the only code that sees the key
  app/api/quotes/stream/route.ts  Server-Sent Events relay
  app/dashboard.tsx               client component with the useQuotes() hook
  app/page.tsx                    server component that renders the dashboard

// The only instruments the dashboard can show. The browser never decides
// what the server subscribes to with your key.
export const WATCHLIST = [
  { channel: "crypto.quotes", symbol: "BTCUSD", name: "Bitcoin" },
  { channel: "crypto.quotes", symbol: "ETHUSD", name: "Ether" },
  { channel: "forex.quotes", symbol: "EURUSD", name: "Euro / US dollar" },
  { channel: "forex.quotes", symbol: "USDJPY", name: "US dollar / yen" },
  { channel: "commodities.quotes", symbol: "XAUUSD", name: "Gold" },
  { channel: "stocks.quotes", symbol: "US:KO", name: "Coca-Cola" },
] as const;

export type Quote = { symbol: string; bid: number; ask: number; ts: number };

import "server-only";
import { WATCHLIST, type Quote } from "./watchlist";

type Listener = (quote: Quote) => void;

const STREAM_URL = "wss://stream.tickerlayer.com/?apiKey=";
const IDLE_CLOSE_MS = 60_000; // keep the socket a minute after the last viewer leaves

// One subscribe message per channel, built from the watchlist.
function subscribeMessages() {
  const byChannel = new Map<string, string[]>();
  for (const { channel, symbol } of WATCHLIST) {
    byChannel.set(channel, [...(byChannel.get(channel) ?? []), symbol]);
  }
  return [...byChannel].map(([channel, symbols]) => ({
    action: "subscribe",
    channels: [channel],
    symbols,
    snapshot: "always", // show the last known value even while a market is closed
  }));
}

class MarketHub {
  private ws: WebSocket | null = null;
  private listeners = new Set<Listener>();
  private latest = new Map<string, Quote>();
  private attempt = 0;
  private idleTimer: ReturnType<typeof setTimeout> | null = null;

  /** Register a viewer. Returns the function that unregisters it. */
  subscribe(listener: Listener): () => void {
    this.listeners.add(listener);
    if (this.idleTimer) clearTimeout(this.idleTimer);
    this.idleTimer = null;
    if (!this.ws) this.connect();
    return () => {
      this.listeners.delete(listener);
      if (this.listeners.size === 0 && !this.idleTimer) {
        this.idleTimer = setTimeout(() => this.ws?.close(1000), IDLE_CLOSE_MS);
      }
    };
  }

  /** Last known quote per symbol, for viewers who just arrived. */
  snapshot(): Quote[] {
    return [...this.latest.values()];
  }

  private connect() {
    const key = process.env.TICKERLAYER_API_KEY;
    if (!key) throw new Error("TICKERLAYER_API_KEY is not set");
    const ws = new WebSocket(STREAM_URL + encodeURIComponent(key));
    this.ws = ws;
    let openedAt = 0;

    ws.onopen = () => { openedAt = Date.now(); };
    ws.onmessage = (event) => {
      const msg = JSON.parse(String(event.data));
      if (msg.type === "system" && msg.event === "ready") {
        for (const sub of subscribeMessages()) ws.send(JSON.stringify(sub));
      } else if (msg.type === "quote") {
        // Prices arrive as strings on crypto, forex and stocks: normalise once, here.
        const quote: Quote = { symbol: msg.symbol, bid: Number(msg.bid), ask: Number(msg.ask), ts: Number(msg.ts) };
        this.latest.set(quote.symbol, quote);
        for (const listener of this.listeners) listener(quote);
      } else if (msg.type === "error") {
        console.error(`[market-hub] ${msg.code}: ${msg.message}`);
      }
    };
    ws.onerror = () => console.warn("[market-hub] socket error");
    ws.onclose = (event) => {
      this.ws = null;
      if (this.listeners.size === 0) return; // nobody is watching: stay closed
      if (openedAt && Date.now() - openedAt > 30_000) this.attempt = 0;
      const step = Math.min(30_000, 1_000 * 2 ** this.attempt++);
      const delay = event.code === 1012 ? Math.random() * 1_000 : step / 2 + Math.random() * (step / 2);
      console.warn(`[market-hub] closed with ${event.code}; reconnecting in ${Math.round(delay)} ms`);
      setTimeout(() => {
        if (!this.ws && this.listeners.size > 0) this.connect();
      }, delay);
    };
  }
}

// One hub per server process, even when dev mode reloads this module.
const globalForHub = globalThis as unknown as { marketHub?: MarketHub };
export const marketHub = (globalForHub.marketHub ??= new MarketHub());

import { marketHub } from "@/lib/market-hub";
import type { Quote } from "@/lib/watchlist";

export const runtime = "nodejs"; // a long-lived socket needs Node.js, not the edge runtime

const FLUSH_MS = 250;        // at most four browser updates a second
const KEEP_ALIVE_MS = 15_000;

export async function GET(request: Request) {
  const encoder = new TextEncoder();
  let cleanup = () => {};

  const body = new ReadableStream<Uint8Array>({
    start(controller) {
      const send = (chunk: string) => {
        try {
          controller.enqueue(encoder.encode(chunk));
        } catch {
          cleanup(); // the browser went away between two writes
        }
      };

      // Newest quote per symbol since the last flush; the browser never sees a backlog.
      const pending = new Map<string, Quote>();
      for (const quote of marketHub.snapshot()) pending.set(quote.symbol, quote);
      const unsubscribe = marketHub.subscribe((quote) => pending.set(quote.symbol, quote));

      const flush = setInterval(() => {
        if (pending.size === 0) return;
        send(`data: ${JSON.stringify([...pending.values()])}\n\n`);
        pending.clear();
      }, FLUSH_MS);
      const keepAlive = setInterval(() => send(": keep-alive\n\n"), KEEP_ALIVE_MS);

      cleanup = () => {
        clearInterval(flush);
        clearInterval(keepAlive);
        unsubscribe();
        cleanup = () => {};
      };
      request.signal.addEventListener("abort", () => {
        cleanup();
        try { controller.close(); } catch { /* already closed */ }
      });
    },
    cancel() {
      cleanup();
    },
  });

  return new Response(body, {
    headers: {
      "Content-Type": "text/event-stream; charset=utf-8",
      "Cache-Control": "no-cache, no-transform", // no-transform: do not let compression buffer it
      "X-Accel-Buffering": "no",                   // and tell nginx not to buffer it either
    },
  });
}

data: [{"symbol":"BTCUSD","bid":82849.99,"ask":82850,"ts":1790591203836},{"symbol":"EURUSD","bid":1.137107,"ask":1.137157,"ts":1790591203002},{"symbol":"XAUUSD","bid":4157.74375,"ask":4158.21375,"ts":1790591203896},{"symbol":"US:KO","bid":88.11,"ask":88.3,"ts":1790590887491}]

data: [{"symbol":"BTCUSD","bid":82849.99,"ask":82850,"ts":1790591204022}]

: keep-alive

"use client";

import { useEffect, useState } from "react";
import type { Quote } from "@/lib/watchlist";

type Row = { symbol: string; name: string };

function useQuotes() {
  const [quotes, setQuotes] = useState<Record<string, Quote>>({});
  const [live, setLive] = useState(false);

  useEffect(() => {
    const source = new EventSource("/api/quotes/stream");
    source.onopen = () => setLive(true);
    source.onerror = () => setLive(false); // EventSource reconnects on its own
    source.onmessage = (event) => {
      const batch: Quote[] = JSON.parse(event.data);
      setQuotes((previous) => {
        const next = { ...previous };
        for (const quote of batch) next[quote.symbol] = quote;
        return next;
      });
    };
    return () => source.close();
  }, []);

  return { quotes, live };
}

const price = (n: number) => {
  const digits = n >= 1_000 ? 2 : n >= 10 ? 3 : 5;
  return n.toLocaleString("en-US", { minimumFractionDigits: digits, maximumFractionDigits: digits });
};
const utc = (ms: number) => new Date(ms).toISOString().slice(11, 19);
const right = { textAlign: "right" } as const;

export default function Dashboard({ rows }: { rows: readonly Row[] }) {
  const { quotes, live } = useQuotes();

  return (
    <section>
      <p aria-live="polite">{live ? "Live" : "Reconnecting..."}</p>
      <table>
        <thead>
          <tr>
            <th style={{ textAlign: "left" }}>Instrument</th>
            <th style={right}>Bid</th>
            <th style={right}>Ask</th>
            <th style={right}>Spread (bps)</th>
            <th style={right}>As of (UTC)</th>
          </tr>
        </thead>
        <tbody>
          {rows.map(({ symbol, name }) => {
            const q = quotes[symbol];
            const mid = q ? (q.bid + q.ask) / 2 : 0;
            return (
              <tr key={symbol}>
                <td>
                  <strong>{symbol}</strong> {name}
                </td>
                <td style={right}>{q ? price(q.bid) : "..."}</td>
                <td style={right}>{q ? price(q.ask) : "..."}</td>
                <td style={right}>{q ? (((q.ask - q.bid) / mid) * 10_000).toFixed(2) : "..."}</td>
                <td style={right}>{q ? utc(q.ts) : "..."}</td>
              </tr>
            );
          })}
        </tbody>
      </table>
    </section>
  );
}

import Dashboard from "./dashboard";
import { WATCHLIST } from "@/lib/watchlist";

export default function Page() {
  return (
    <main style={{ maxWidth: 760, margin: "48px auto", padding: "0 16px", fontFamily: "system-ui" }}>
      <h1>Markets</h1>
      <Dashboard rows={WATCHLIST.map(({ symbol, name }) => ({ symbol, name }))} />
    </main>
  );
}