livebit

A real-time data layer for Next.js. livebit packages three things that usually get hand-rolled per app into one small, typed library:

1. A Server-Sent Events (SSE) invalidation bus — the server publishes tiny "entity X changed" hints; clients refetch through their own data layer.
2. An optimistic cache — normalized Zustand entity stores (with correct rollback) plus SWR-backed query hooks, both wired to the invalidation bus.
3. Contract-driven API hooks — one contract object defines an endpoint's request/response schema and which entities it touches, and powers both the server route and the client hooks, so backend and frontend cannot drift.

It is bundler-agnostic ESM (works in Next.js App Router, webpack, Vite, and native Node) and ships its own state management (Zustand + SWR) so consumers don't wire it up themselves.

Why

Real-time apps tend to grow an ad-hoc tangle of EventSource plumbing, cache invalidation, optimistic updates with buggy rollback, and request/response types that drift from the server. livebit centralizes that into a single source of truth (the contract) and a single transport (the live bus).

Install

npm install livebit

Peer dependencies (provide these in the host app): react, react-dom, swr, zod, zustand.

Entry points

| Import | Use in | Contents | | --- | --- | --- | | livebit | anywhere (isomorphic) | protocol types, defineContract, defineEntity, url/debounce utils | | livebit/server | route handlers / server | SSE route, transports, notifier, mutation route | | livebit/client | client components | LiveProvider, hooks, cache stores, query/mutation/API hooks |

Concepts

• Entity key: an app-defined string (e.g. 'reservations') that routes invalidation. Define your set once with defineEntities.
• Contract: defineContract({ method, path, body?, query?, params?, response?, error?, invalidates }). invalidates lists the entity keys the endpoint relates to — mutations emit them; queries subscribe to them. params validates :param path segments on both sides; error types non-2xx payloads (throw HttpError(status, data) from a handler to produce one).
• Transport: how events fan out. InMemoryTransport (single instance) or RedisTransport (multi-instance pub/sub; you inject an ioredis-compatible client).

Quick start

1. Define entities and a contract (shared)

// shared/contracts.ts
import { defineContract, defineEntities } from 'livebit';
import { z } from 'zod';

export const entities = defineEntities('things');

export const createThing = defineContract({
  method: 'POST',
  path: '/api/things',
  body: z.object({ name: z.string().min(1) }),
  response: z.object({ id: z.string(), name: z.string() }),
  invalidates: ['things'] as const,
});

export const listThings = defineContract({
  method: 'GET',
  path: '/api/things',
  response: z.object({ data: z.array(z.object({ id: z.string(), name: z.string() })) }),
  invalidates: ['things'] as const,
});

2. Server: the SSE route + a shared notifier

// app/api/live/route.ts
import { createSseRoute, InMemoryTransport } from 'livebit/server';

export const transport = new InMemoryTransport(); // or new RedisTransport(redis)

export const GET = createSseRoute({
  transport,
  resolveTenant: async (request) => {
    const session = await getSession(request); // your auth
    return session ? { tenantId: session.tenantId } : null;
  },
  // On serverless platforms with a connection cap, auto-recycle before it:
  maxConnectionMs: 800_000,
});

// app/api/things/route.ts
import { createMutationRoute, createNotifier } from 'livebit/server';
import { createThing } from '@/shared/contracts';
import { transport } from '../live/route';

const notify = createNotifier(transport);

export const POST = createMutationRoute(
  createThing,
  {
    notify,
    resolveAuth: async (request) => {
      const session = await getSession(request);
      return session ? { tenantId: session.tenantId } : null;
    },
    resolveId: (response) => response.id, // enables self-echo dedup on the client
  },
  async ({ body, auth }) => {
    const thing = await db.thing.create({ data: { ...body, tenantId: auth.tenantId } });
    return thing; // validated body in, typed response out; 'things' invalidation emitted on success
  }
);

3. Client: mount the provider

// app/providers.tsx
'use client';
import { LiveProvider } from 'livebit/client';

export function Providers({ children, isAuthenticated }: { children: React.ReactNode; isAuthenticated: boolean }) {
  return (
    <LiveProvider streamUrl="/api/live" enabled={isAuthenticated}>
      {children}
    </LiveProvider>
  );
}

4. Client: read + mutate via contracts

'use client';
import { useApiQuery, useApiMutation } from 'livebit/client';
import { createThing, listThings } from '@/shared/contracts';

function Things() {
  const { data, isLoading } = useApiQuery(listThings); // auto-revalidates on 'things' events
  const create = useApiMutation(createThing, {
    // Optimistic update against the listThings cache, rolled back on failure.
    optimistic: {
      query: listThings,
      apply: (current, vars) => ({
        data: [...(current?.data ?? []), { id: 'optimistic', name: vars.body!.name }],
      }),
    },
    resolveId: (response) => response.id,
  });

  return (
    <button onClick={() => create.mutate({ body: { name: 'New' } })} disabled={create.isLoading}>
      Add
    </button>
  );
}

After the mutation succeeds the server emits a things invalidation; every connected client (including this one) refetches automatically. No manual cache busting.

Waiting for backend confirmation

mutate/mutateAsync accept a second argument controlling what "done" means:

// Resolves when the HTTP response arrives (default):
await create.mutateAsync({ body: { name: 'New' } });

// Resolves only after the server's live invalidation echo for this record
// arrives back over SSE (or the timeout elapses):
await create.mutateAsync({ body: { name: 'New' } }, { confirm: 'live', confirmTimeoutMs: 10_000 });

create.status; // 'idle' | 'optimistic' | 'submitted' | 'confirmed' | 'error'
create.isConfirmed;

With resolveId configured (mirroring the server route's resolveId), confirm: 'live' matches the exact echo event by record id; without it, any event for the contract's invalidates entities counts.

Cache stores (full entity lists)

For full lists with optimistic CRUD, use a contract-driven cache store instead of a query:

import { createContractCacheStore } from 'livebit/client';
import { listThings, createThing, updateThing, deleteThing } from '@/shared/contracts';

export const useThingStore = createContractCacheStore<Thing, typeof listThings, typeof createThing, typeof updateThing, typeof deleteThing>({
  name: 'things',
  contracts: { list: listThings, create: createThing, update: updateThing, remove: deleteThing },
  selectList: (response) => response.data,
  // Optional: optimistic creates with a temp item, reconciled on success.
  optimisticCreate: (input, tempId) => ({ id: tempId, ...input }),
  ttlMs: 10 * 60 * 1000,
});

(createLiveCacheStore with raw fetchFn/createFn/updateFn/deleteFn closures remains available as an escape hatch.)

import { useAutoLiveCache } from 'livebit/client';

function ThingsList() {
  const { items, actions } = useAutoLiveCache(useThingStore);
  // actions.create / actions.update / actions.remove are optimistic with rollback on failure
}

Optimistic mutations snapshot the affected item before applying the change and roll back per item on error, so concurrent mutations of other items are preserved. The acting client suppresses the echo of its own change (matching event.id within a 5s window) to avoid a redundant refetch; completed mutations are retained for ~10s so the echo can be matched even when it arrives after the response.

Range cache (calendar / date-window data)

For data keyed by date ranges (staff calendars, overnight grids) that should grow as the user pans without refetching the whole list, use a range cache store instead of per-day useApiQuery keys:

import {
  createContractRangeCacheStore,
  useAutoLiveRangeCache,
} from 'livebit/client';
import { listReservations } from '@/shared/contracts';

export const useReservationRangeStore = createContractRangeCacheStore({
  name: 'reservations',
  list: listReservations,
  selectList: (response) => response.data,
  queryArgs: { tenant: 't1' },
  itemInRange: (reservation, range) =>
    reservation.services.some((svc) => {
      const start = new Date(svc.startTime);
      return start >= range.start && start <= range.end;
    }),
});

function Calendar({ visibleRange }: { visibleRange: DateRange }) {
  const store = useReservationRangeStore();
  const { items, coverageRanges, fetchingRanges, getDayLoadState } =
    useAutoLiveRangeCache(store, { visibleRange, filterKey: tenantId });

  // Pan the visible window → ensureCoverage fetches only uncovered 14-day chunks
  // Liveness events → refreshVisibleRanges (only chunks overlapping visibleRange)
}

Isomorphic utilities (mergeDateRanges, getSupplementalFetchRange, retryWithLinearBackoff) are exported from livebit for custom fetch orchestration.

Resilience

• Reconnect with linear backoff (1s step, capped at 30s by default).
• Cache refresh retries use linear backoff (1s base step, 10s cap, 3 retries) for both full-list and range caches.
• Catch-up on reconnect: when the stream reopens after a disruption, the client emits one synthetic invalidation per subscribed entity, so events missed while offline are never silently lost.
• Polling fallback: while disconnected, the client emits synthetic invalidations for subscribed entities so caches still refresh; useLiveQuery also raises its SWR refreshInterval.
• Heartbeats keep proxy connections from idling out; the client also treats a silent connection (no heartbeat within heartbeatTimeoutMs, default 75s) as dead and recycles it.
• Serverless-safe notifications: the mutation route awaits event publication before responding (or hands it to a platform waitUntil when provided), so invalidations are not dropped by a frozen runtime.
• Tenant isolation: events are scoped per tenant by the transport and the route's resolveTenant. RedisTransport shares one ref-counted subscriber connection across all SSE clients on an instance.

Development

npm install
npm test         # vitest (unit + an end-to-end server→SSE→client test)
npm run typecheck
npm run build    # tsc → dist (ESM + .d.ts)
npm run ci       # typecheck + test + build
yarn dev         # builds library (watch) + opens playground (http://localhost:3000, or next free port)

License

MIT