Building a Live Market Dashboard with SSE and Next.js

I wanted to observe the ticker in a browser after several weeks of tests through logs and CLI tools. The dashboard uses the same Kafka and Redis brokers from the failure experiments. It displays current prices, alerts, consumer assignments, and lag.

A shared engine collects broker state. A Server Sent Events route sends snapshots to a React component. The Next.js configuration keeps broker clients on the server.

One engine per process

A route that creates broker clients for each request can leave duplicate connections after development reloads. Separate browser tabs can also create unnecessary consumers. Independent Kafka groups would each receive their own topic history. Consumers in one shared group would instead trigger assignment changes.

The fix is a plain singleton, stashed on globalThis so it survives both HMR and multiple requests in the same process:

// Process-global singleton (survives Next dev HMR + shared across requests/tabs).
const globalForEngine = globalThis as unknown as { __liveEngine?: LiveEngine };

export function getLiveEngine(): LiveEngine {
  if (!globalForEngine.__liveEngine) globalForEngine.__liveEngine = new LiveEngine();
  return globalForEngine.__liveEngine;
}

Each route calls getLiveEngine() to retrieve the shared instance. The ensureStarted() method guards startup, which creates the topics, producer, and consumers. Later requests reuse that state:

async ensureStarted(): Promise<void> {
  if (this.started) return;
  this.started = true;
  try {
    await this.start();
  } catch (e) {
    this.lastError = msg(e);
    this.started = false; // let a later request retry
  }
}

One engine, one Kafka producer, one Redis connection, two consumers, no matter how many tabs are watching.

What the engine wires together

The engine produces synthetic ticks to both brokers. Two consumer pods evaluate alerts and join them with local price state. The engine then sends a state snapshot to each subscriber.

graph LR
    SIM["Market simulator"] --> KP["Kafka producer<br/>market-updates"]
    SIM --> RX["Redis XADD<br/>stream:market-updates"]
    KP --> KB[("Kafka broker")]
    KB --> C1["Pod 1 consumer<br/>co-partition assigner"]
    KB --> C2["Pod 2 consumer<br/>co-partition assigner"]
    C1 --> ENG["Live engine<br/>snapshot state"]
    C2 --> ENG
    ENG -->|"broadcast every 400ms"| SSE["SSE route handler<br/>ReadableStream"]
    SSE -->|"EventSource"| UI["LiveDashboard"]

The dual write happens on a 150ms interval, three ticks at a time, batched into a single Kafka send and a single Redis pipeline:

private async produceTick(): Promise<void> {
  if (!this.running || !this.producer) return;
  const ticks = [this.sim.next(), this.sim.next(), this.sim.next()];
  await this.producer.send({ topic: TOPICS.marketUpdates, messages: ticks.map(tickToKafka) });
  this.produced += ticks.length;

  const pipe = this.redis.pipeline();
  for (const t of ticks) {
    pipe.xadd(REDIS_KEYS.marketUpdates, "MAXLEN", "~", "20000", "*", ...tickToRedis(t));
  }
  await pipe.exec();
  this.redisAdded += ticks.length;
}

The Redis side caps the stream at roughly 20,000 entries (MAXLEN ~ 20000) so it does not grow unbounded while the dashboard is left running. Kafka does not need that here, since retention is time based at the topic level and not a concern this loop has to manage.

Separate setInterval timers measure throughput each second and poll lag every 1.5 seconds. A third timer sends a snapshot every 400ms. The browser needs current state rather than every intermediate tick. Combining updates into snapshots reduces rendering work.

Two pods, one local join

Both consumers belong to live-evaluators. They use the custom co partition assigner and subscribe to market-updates and notifications:

consumer: kafka.consumer({
  groupId: "live-evaluators",
  partitionAssigners: [coPartitionAssigner],
}),

The assigner puts partition p from both topics on the same pod. Its eachMessage handler can join records when the required local price is available. A market tick updates the global dashboard price table and the pod's localPrice map:

if (topic === TOPICS.marketUpdates) {
  const tick = tickFromKafka(message.value);
  this.prices.set(tick.symbol, tick.price);
  pod.localPrice.set(tick.symbol, tick.price); // local, co-located state
  for (const n of pod.evaluator.process(tick)) {
    await this.producer?.send({ topic: TOPICS.notifications, messages: [notifToKafka(n)] });
  }
} else {
  // Co-partitioned with market-updates, so the symbol's price is already
  // sitting on THIS pod - enrich locally, no remote lookup.
  const n = notifFromKafka(message.value);
  if (pod.localPrice.has(n.symbol)) pod.localJoins++;
  else pod.remoteNeeded++;
}

The pod.evaluator uses the stateful AlertEvaluator to retain the last price per symbol and detect threshold crossings. The localJoins and remoteNeeded counters record whether a notification found local price state. They show observed join results, not a guarantee that every future lookup will succeed.

Why Server Sent Events, not WebSockets

Snapshot data flows from the engine to the browser. Pause and resume use a separate POST endpoint. I chose SSE because this feed needs no messages from the browser over the same connection.

SSE uses HTTP, and the browser EventSource handles reconnection. A Next.js App Router handler can return a ReadableStream inside a Response. TextEncoder converts each event to bytes. The server encodes each snapshot as an SSE message. This is sufficient for the periodic snapshot feed.

The route handler: a response that never ends

The route itself is short. It fetches the singleton engine, opens a stream, and wires the engine's subscriber callback straight into the stream controller:

export const dynamic = "force-dynamic";

export async function GET(req: Request): Promise<Response> {
  const engine = getLiveEngine();
  await engine.ensureStarted();

  const encoder = new TextEncoder();
  let unsubscribe = () => {};

  const stream = new ReadableStream({
    start(controller) {
      const send = (data: unknown) => {
        controller.enqueue(encoder.encode(`data: ${JSON.stringify(data)}\n\n`));
      };

      send(engine.getSnapshot()); // prime immediately
      unsubscribe = engine.subscribe(send);

      req.signal.addEventListener("abort", () => {
        unsubscribe();
        controller.close();
      });
    },
    cancel() {
      unsubscribe();
    },
  });

  return new Response(stream, {
    headers: {
      "Content-Type": "text/event-stream",
      "Cache-Control": "no-cache, no-transform",
      Connection: "keep-alive",
    },
  });
}

The route depends on these details:

• export const dynamic = "force-dynamic" opts the route out of static optimization. A long lived stream is the opposite of something you can cache or pre render.
• The snapshot is sent immediately on connect, before subscribing. Without that, a newly opened tab would sit on a blank screen for up to 400ms waiting for the next scheduled broadcast.
• req.signal fires abort when the client disconnects, whether that is a closed tab or a dropped connection. This calls unsubscribe() and removes the callback from the engine's subscribers set. Without this cleanup, callbacks would remain after their tabs closed.
• The response headers (no-cache, no-transform, keep-alive) describe streaming behavior. Proxy buffering and timeout settings also need verification on the deployment path.

The sequence below shows the initial snapshot, later broadcasts, and cleanup after a disconnect:

sequenceDiagram
    participant B as Browser
    participant H as SSE route handler
    participant E as Live engine

    B->>H: GET /api/live
    H->>E: ensureStarted()
    H->>B: send snapshot (priming)
    H->>E: subscribe(send)
    loop every 400ms
        E->>H: broadcast(snapshot)
        H->>B: send snapshot
    end
    B--xH: tab closes, connection aborts
    H->>E: unsubscribe()
    H->>B: controller.close()

Pause and resume without touching the stream

Control flows through a separate route rather than a message sent back down the SSE connection:

export async function POST(req: Request): Promise<Response> {
  const { action } = (await req.json().catch(() => ({ action: "" }))) as { action?: string };
  const engine = getLiveEngine();

  if (action === "pause") {
    engine.setRunning(false);
  } else if (action === "resume") {
    await engine.ensureStarted();
    engine.setRunning(true);
  }

  return Response.json({ running: engine.isRunning() });
}

Pausing flips a boolean that produceTick checks before doing any work. The producer and two consumers stay connected, so resuming is instant: no reconnect, no rebalance. The dashboard's own running indicator comes straight from the next broadcast snapshot, so the UI never has to track its own optimistic state.

Keeping broker clients out of the bundle

kafkajs and ioredis are Node native: they open raw TCP sockets and lean on dynamic require calls that a bundler cannot statically resolve. Left alone, Next.js will try to bundle them for the route handler and either fail the build or produce something broken at runtime. The Next config opts them out explicitly:

const nextConfig: NextConfig = {
  // kafkajs and ioredis are Node-native (net/tls, dynamic requires). Keep them
  // out of the bundler so route handlers load them as normal Node modules.
  serverExternalPackages: ["kafkajs", "ioredis", "pino", "pino-pretty"],
};

The configuration also externalizes pino and pino-pretty for the application logging setup. The broker clients require the Node.js runtime because they open TCP sockets. The route cannot use export const runtime = "edge" with these clients.

What ends up on screen

The dashboard is a client component. One useEffect opens an EventSource at /api/live. Each message replaces the state with the latest snapshot through setSnap(JSON.parse(e.data)). The component does not merge partial updates or use a separate store.

What lands on screen is everything the engine tracks:

• Throughput tiles: Kafka produced/sec, Kafka consumed/sec, Redis XADD/sec, and a running total of alerts fired, straight from the per second rate sampling in the engine.
• The ticker board: all ten symbols with current price and percent change against their base price, colored up or down.
• The notifications feed: the last 16 alerts, each showing symbol, operator, threshold, and the price that triggered it.
• Partition ownership and joins: each pod shows its partitions for market-updates and notifications. A badge shows whether its partition lists match. Counters show local and remote joins.
• Consumer lag: the same high - committed calculation from the consumer groups post, polled every 1.5 seconds and colored red past a threshold.

The dashboard uses the broker engine from the earlier posts. After a pause, the producer stops and consumers can finish the backlog. A broker failure or rebalance changes the displayed assignments. These observations help inspect behavior while the system runs.

The dashboard shows behavior, but it does not establish performance. The next article describes the benchmark harness and its limits.