Building a Live Market Dashboard with SSE and Next.js

Everything I have written about my little ticker system so far has been about understanding Kafka and Redis Streams from the inside: partitioning, consumer groups, retries, exactly-once, co-partitioning, stateful processing. All of it invisible unless you go digging for it with a CLI script or an admin tool. After a few weeks of learning about my own system by reading log lines, I wanted to watch it instead, live, in a browser tab, fed by the same two brokers in Docker I had been killing and restarting all along.

So this one is less about Kafka versus Redis and more about the plumbing that gets ticks from a broker into a React component: a process-global engine, a Server-Sent Events route handler, and a couple of Next.js config decisions that keep the whole thing from falling over.

One Engine, Not One Per Request

A naive version of this feature would connect to Kafka and Redis inside the route handler, on every request. In a Next.js app that goes wrong quickly: dev mode hot-reloads modules constantly, and every reload would spin up a fresh producer and fresh consumers without ever closing the old ones. Open the dashboard in three browser tabs and you would have three independent consumer groups fighting over the same partitions.

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;
}

Every route handler calls getLiveEngine() and gets back the exact same instance. Startup is guarded separately with an ensureStarted() idempotency flag, so the first request to hit the dashboard boots the engine (topics, producer, two consumers), and every request after that is a no-op:

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

Once started, the engine does three things on a loop: it produces synthetic ticks to both brokers, it runs two consumer pods that evaluate alerts and join them against local price state, and it broadcasts a snapshot of everything to whoever is subscribed.

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.

Separately, setInterval timers sample throughput deltas every second, poll consumer lag every 1.5 seconds, and broadcast a snapshot to subscribers every 400ms. That last number is deliberate. The dashboard does not need every individual tick pushed to it, it needs a reasonably fresh picture of current state. Coalescing dozens of ticks into one snapshot every 400ms keeps the SSE payload small and the browser doing far less work than if it had to reconcile a stream of deltas.

Two Pods, One Local Join

The two consumers are both in the same group, live-evaluators, both configured with the custom co-partition assigner I wrote a few posts back, and both subscribed to market-updates and notifications at once:

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

Because the assigner guarantees partition p of both topics lands on the same pod, each pod's eachMessage handler can do a real join without a network hop. When a market tick arrives it updates two maps: the engine's global price table (for the dashboard's ticker) and the pod's own localPrice map, which is the co-located state:

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++;
}

pod.evaluator is the stateful AlertEvaluator from the stream processing post: it holds the last price per symbol so it can detect an edge crossing rather than a level match. When a notification the pod itself produced comes back around on the notifications topic, the localJoins versus remoteNeeded counters become live proof that co-partitioning is doing its job, which beats a diagram in a doc.

Why Server-Sent Events, Not WebSockets

The data only flows one direction: engine to browser. Pause and resume are handled through an ordinary POST to a separate endpoint, not a message sent back over the stream. Given that, WebSockets buy nothing here except more protocol to manage, a separate upgrade handshake, and reconnection logic I would have to write myself.

SSE is plain HTTP. EventSource in the browser opens a normal GET request, handles reconnection automatically if the connection drops, and speaks a text format simple enough to hand-encode with a TextEncoder. On the server side, a Next.js App Router route handler can return a ReadableStream directly as the Response body, no extra library, no separate WebSocket server process to run alongside Next.js. For a one-way feed of periodic snapshots, that is the whole feature set I need.

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",
    },
  });
}

A few details matter more than they look:

• 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. That is the only place unsubscribe() gets called on the happy path, and it matters: the engine's subscribers set has no other cleanup mechanism, so a route that forgot this would leak one callback per tab ever opened.
• The response headers (no-cache, no-transform, keep-alive) are what stop intermediate proxies from buffering or compressing the stream into uselessness.

Laid out as a sequence, the same lifecycle makes it clearer how the priming message and the abort handler relate to the engine's own broadcast loop:

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. It does not tear down the producer or the two consumers, 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"],
};

pino and pino-pretty ride along because they are KafkaJS's logging dependencies and hit the same dynamic-require problem. All of this works only because the route handlers run in the Node.js runtime by default, not the Edge runtime, which has no raw socket access at all. Nothing in the route declares export const runtime = "edge", and nothing can: a TCP connection to a broker is a hard requirement here.

What Ends Up On Screen

The dashboard is a plain client component: one useEffect opens an EventSource against /api/live, and every message replaces the component's state wholesale with the latest snapshot. No delta reconciliation, no separate store, one setSnap(JSON.parse(e.data)).

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: the two pods side by side, each showing which partitions of market-updates and notifications it currently owns, a co-located badge (the two partition lists compared with a simple equality check), and the running local-versus-remote join counters described above.
• Consumer lag: the same high - committed calculation from the consumer groups post, polled every 1.5 seconds and colored red past a threshold.

None of it is a canned demo. Pause the engine, and produced/consumed both flatline. Kill a broker or force a rebalance, and the partition badges on the pod cards visibly reshuffle. Same mechanics as every earlier post, rendered somewhere I can watch them happen instead of reading them off a log line.

Watching it move answers one question: whether the thing works. It says nothing about how fast. Pinning that down turned into its own rabbit hole, because benchmarking two brokers fairly is mostly an exercise in not measuring your own laptop by accident.