How to Build a Real-Time Dashboard with Reactive State

Combine multiple data sources into derived metrics with diamond-safe updates and atomic batching.

The Problem

Real-time dashboards need:

• Multiple data sources updating independently
• Derived metrics that stay consistent (no partial updates)
• Efficient updates — recalculate only what changed
• Atomic writes — update multiple sources without intermediate renders

The Solution

callbag-recharge's derived() handles the hard part: when multiple sources update, derived metrics recompute exactly once with consistent values. batch() makes multi-source updates atomic.

/**
 * Real-time Dashboard
 *
 * Demonstrates: state + derived + effect for a reactive dashboard
 * that combines multiple data sources into derived metrics.
 */

import { batch, derived, effect, state } from "callbag-recharge";

// ── Data sources ─────────────────────────────────────────────

const activeUsers = state(0, { name: "activeUsers" });
const totalRequests = state(0, { name: "totalRequests" });
const errorCount = state(0, { name: "errors" });
const responseTimeMs = state(0, { name: "responseTime" });

// ── Derived metrics ──────────────────────────────────────────

const errorRate = derived(
	[errorCount, totalRequests],
	() => {
		const total = totalRequests.get();
		return total > 0 ? (errorCount.get() / total) * 100 : 0;
	},
	{ name: "errorRate" },
);

const healthStatus = derived(
	[errorRate, responseTimeMs],
	() => {
		const rate = errorRate.get();
		const latency = responseTimeMs.get();
		if (rate > 5 || latency > 1000) return "critical";
		if (rate > 1 || latency > 500) return "warning";
		return "healthy";
	},
	{ name: "healthStatus" },
);

const dashboardSummary = derived(
	[activeUsers, totalRequests, errorRate, healthStatus, responseTimeMs],
	() => ({
		users: activeUsers.get(),
		requests: totalRequests.get(),
		errorRate: `${errorRate.get().toFixed(2)}%`,
		latency: `${responseTimeMs.get()}ms`,
		status: healthStatus.get(),
	}),
	{ name: "summary" },
);

// ── Side effect: log dashboard updates ───────────────────────

const dispose = effect([dashboardSummary], () => {
	console.log("Dashboard:", dashboardSummary.get());
});

// ── Simulate data arriving ───────────────────────────────────

// batch() ensures derived stores recompute once, not per-set
batch(() => {
	activeUsers.set(142);
	totalRequests.set(1500);
	errorCount.set(3);
	responseTimeMs.set(45);
});
// Dashboard: { users: 142, requests: 1500, errorRate: '0.20%', latency: '45ms', status: 'healthy' }

batch(() => {
	errorCount.set(120);
	responseTimeMs.set(1200);
});
// Dashboard: { users: 142, requests: 1500, errorRate: '8.00%', latency: '1200ms', status: 'critical' }

// ── Cleanup ──────────────────────────────────────────────────

dispose();

Why This Works

1. Diamond resolution — healthStatus depends on errorRate which depends on errorCount and requestsPerSec. When both update in a batch(), healthStatus computes once, not twice, with correct values.

2. batch() — groups set() calls. DIRTY signals propagate immediately, but values flow only when the batch ends. No intermediate states.

3. Layered derivations — errorRate derives from raw counts; healthStatus derives from errorRate + responseTimeMs; dashboardSummary aggregates everything. Each layer is cached and recomputes only when its specific deps change.

4. effect() — runs side effects (logging, WebSocket push, DOM update) only when the derived metric actually changes.

Framework Integration

React

function useStore<T>(store: Store<T>): T {
  const [value, setValue] = useState(store.get())
  useEffect(() => subscribe(store, setValue), [store])
  return value
}

function Dashboard() {
  const summary = useStore(dashboardSummary)
  return <div className={`status-${summary.status}`}>
    <span>{summary.users} users</span>
    <span>{summary.rps} rps</span>
    <span>{summary.errorRate} errors</span>
  </div>
}

Streaming to Browser via SSE

import { toSSE } from 'callbag-recharge/adapters'

app.get('/dashboard/stream', (req, res) => {
  toSSE(dashboardSummary, { response: res })
})

See Also

• Data Pipeline — ETL with composable operators
• Cron Pipeline — scheduled data aggregation