svelte-realtime

v0.5.10

Published

10 days ago

Realtime RPC and reactive subscriptions for SvelteKit, built on svelte-adapter-uws

Downloads

4,321

0High
0Medium
0Low

meredithrodneymckay

svelte sveltekit websocket rpc realtime pubsub uwebsockets streams live

Write server functions. Import them in components. Call them over WebSocket. No boilerplate, no manual pub/sub wiring, no protocol design.

Upgrading from 0.4.x? See the migration guide for every breaking change between 0.4.x and 0.5.x.

Quick start

npx svelte-realtime my-app
cd my-app
npm run dev

This creates a SvelteKit project with svelte-realtime fully wired: adapter, vite plugins, WebSocket hooks, and a working counter example you can open in your browser right away.

Version compatibility

The three ecosystem packages move together. Bump them as a group:

| svelte-adapter-uws | svelte-realtime | svelte-adapter-uws-extensions | Notes | |---|---|---|---| | ^0.4.x | ^0.4.x | ^0.4.x | Legacy stable | | ^0.5.0 | ^0.5.0 | ^0.5.0 | Current. Node 22+ required. See MIGRATION.md if upgrading from 0.4. |

Mixed-version installs are rejected at install time with a peer-dep warning.

Manual setup

Starting from a SvelteKit project. If you do not have one yet, run npx sv create my-app && cd my-app && npm install first.

Step 1: Install everything

npm install svelte-adapter-uws svelte-realtime
npm install uNetworking/uWebSockets.js#v20.60.0
npm install -D ws

What each package does:

svelte-adapter-uws (>=0.4.0) - the SvelteKit adapter that runs your app on uWebSockets.js with built-in WebSocket support
svelte-realtime - this library (RPC + streams on top of the adapter)
uWebSockets.js - the native C++ HTTP/WebSocket server (installed from GitHub, not npm)
ws - dev dependency used by the adapter during npm run dev (not needed in production)

Step 2: Configure the adapter

Open svelte.config.js and replace the default adapter:

// svelte.config.js
import adapter from 'svelte-adapter-uws';

export default {
  kit: {
    adapter: adapter({ websocket: true })
  }
};

Step 3: Add the Vite plugins

Open vite.config.js and add the adapter and realtime plugins:

// vite.config.js
import { sveltekit } from '@sveltejs/kit/vite';
import uws from 'svelte-adapter-uws/vite';
import realtime from 'svelte-realtime/vite';

export default {
  plugins: [sveltekit(), uws(), realtime()]
};

All three plugins are required. Order does not matter.

Step 4: Create the WebSocket hooks file

Create src/hooks.ws.js in your project root. This file tells the adapter how to handle WebSocket connections and messages.

// src/hooks.ws.js
export { message } from 'svelte-realtime/server';

export function upgrade({ cookies }) {
  // Return user data to attach to the connection, or false to reject.
  // This runs on every new WebSocket connection.
  const session = validateSession(cookies.session_id);
  if (!session) return false;
  return { id: session.userId, name: session.name };
}

message is a ready-made hook that routes incoming WebSocket messages to your live functions. upgrade decides who can connect and attaches user data to the connection.

This file is required. The Vite plugin will warn at startup if it finds live modules in src/live/ but no src/hooks.ws.js (or .ts). Without it, WebSocket messages have nothing on the server side to route them, and all RPC calls will silently time out.

Step 5: Write a server function

Create the src/live/ directory. Every .js file in this directory becomes a module of callable server functions.

// src/live/chat.js
import { live, LiveError } from 'svelte-realtime/server';
import { db } from '$lib/server/db';

// A plain RPC function - clients can call this like a regular async function
export const sendMessage = live(async (ctx, text) => {
  if (!ctx.user) throw new LiveError('UNAUTHORIZED', 'Login required');

  const msg = await db.messages.insert({ userId: ctx.user.id, text });
  ctx.publish('messages', 'created', msg);
  return msg;
});

// A stream - clients get a Svelte store with initial data + live updates
export const messages = live.stream('messages', async (ctx) => {
  return db.messages.latest(50);
}, { merge: 'crud', key: 'id', prepend: true });

live() marks a function as callable over WebSocket. The first argument is always ctx (context), which contains the user data from upgrade(), the WebSocket connection, and a publish function for sending events to all subscribers.

live.stream() creates a reactive subscription. When a client subscribes, it gets the initial data from the function, then receives live updates whenever someone publishes to that topic.

Step 6: Use it in a component

<!-- src/routes/chat/+page.svelte -->
<script>
  import { sendMessage, messages } from '$live/chat';
  let text = $state('');

  async function send() {
    await sendMessage(text);
    text = '';
  }
</script>

{#if $messages === undefined}
  <p>Loading...</p>
{:else}
  {#each $messages as msg (msg.id)}
    <p><b>{msg.userId}:</b> {msg.text}</p>
  {/each}
{/if}

<input bind:value={text} />
<button onclick={send}>Send</button>

$live/chat is a virtual import. The Vite plugin reads your src/live/chat.js file, sees which functions are wrapped in live() and live.stream(), and generates lightweight client stubs that call them over WebSocket.

sendMessage becomes a regular async function on the client.
messages becomes a Svelte store. It starts as undefined (loading), then populates with the initial data, then merges live updates as they arrive.

That is the entire setup. Run npm run dev and it works.

How it works

You write:     src/live/chat.js       (server functions)
You import:    $live/chat             (auto-generated client stubs)

live()         -> RPC call over WebSocket (async function)
live.stream()  -> Svelte store with initial data + live updates

The ctx object passed to every server function contains:

| Field | Description | |---|---| | ctx.user | Whatever upgrade() returned (your user data) | | ctx.ws | The raw WebSocket connection | | ctx.platform | The adapter platform API | | ctx.publish | Shorthand for platform.publish(). Rejects __-prefixed topics with LiveError('INVALID_TOPIC') (those are framework-internal channels; use ctx.platform.publish() directly if you genuinely need to broadcast on one) | | ctx.cursor | Cursor from a loadMore() call, or null | | ctx.requestId | Correlation id from platform.requestId (per WS connection or per HTTP request); honors X-Request-ID | | ctx.throttle | (topic, event, data, ms) - publish at most once per ms ms | | ctx.debounce | (topic, event, data, ms) - publish after ms ms of silence | | ctx.signal | (userId, event, data) - point-to-point message | | ctx.batch | (messages) - publish multiple messages in one call via platform.batch() |

Note: ctx.user may contain adapter-injected properties (__subscriptions, remoteAddress) in addition to whatever your upgrade() function returned. These are stripped automatically by the adapter before broadcasting to other clients.

Core

Client features

Server features

Deployment

Reference

Merge strategies

The merge option on live.stream() controls how live pub/sub events are applied to the store.

crud (default)

Handles created, updated, deleted events. The store maintains an array, keyed by id (configurable with key). Set max to cap the buffer size and drop the oldest items when exceeded (useful for live feeds with prepend: true).

// Server
export const todos = live.stream('todos', async (ctx) => {
  return db.todos.all();
}, { merge: 'crud', key: 'id' });

export const addTodo = live(async (ctx, text) => {
  const todo = await db.todos.insert({ text });
  ctx.publish('todos', 'created', todo);
  return todo;
});

<!-- Client -->
<script>
  import { todos, addTodo } from '$live/todos';
</script>

{#each $todos as todo (todo.id)}
  <p>{todo.text}</p>
{/each}

latest

Ring buffer of the last N events. Good for activity feeds and logs.

export const activity = live.stream('activity', async (ctx) => {
  return db.activity.recent(100);
}, { merge: 'latest', max: 100 });

set

Replaces the entire value. Good for counters, status indicators, and aggregated data.

export const stats = live.stream('stats', async (ctx) => {
  return { users: 42, messages: 1337 };
}, { merge: 'set' });

presence

Tracks connected users with join and leave events. Items are keyed by .key.

export const presence = live.stream(
  (ctx, roomId) => 'presence:' + roomId,
  async (ctx, roomId) => [],
  { merge: 'presence' }
);

export const join = live(async (ctx, roomId) => {
  ctx.publish('presence:' + roomId, 'join', { key: ctx.user.id, name: ctx.user.name });
});

<script>
  import { presence } from '$live/room';
  const users = presence(data.roomId);
</script>

{#each $users as user (user.key)}
  <span>{user.name}</span>
{/each}

Events: join (add/update by key), leave (remove by key), set (replace all).

cursor

Tracks cursor positions with update and remove events. Items are keyed by .key.

export const cursors = live.stream(
  (ctx, docId) => 'cursors:' + docId,
  async (ctx, docId) => [],
  { merge: 'cursor' }
);

export const moveCursor = live(async (ctx, docId, x, y) => {
  ctx.publish('cursors:' + docId, 'update', { key: ctx.user.id, x, y, color: ctx.user.color });
});

Events: update (add/update by key), remove (remove by key), set (replace all).

Stream options reference

| Option | Default | Description | |---|---|---| | merge | 'crud' | Merge strategy: 'crud', 'latest', 'set', 'presence', 'cursor' | | key | 'id' | Key field for crud mode | | prepend | false | Prepend new items instead of appending (crud mode) | | max | 50 / 0 | Max items to keep. Defaults to 50 for latest, 0 (unlimited) for crud. Oldest items are dropped when exceeded | | replay | false | Enable seq-based replay for gap-free reconnection | | args | - | Standard Schema (Zod / ArkType / Valibot) for stream arguments. Validated before topic resolution - prevents topic injection via malformed dynamic-topic args | | transform | - | (data) => projection applied to BOTH initial-load data (per-item for arrays) AND every live publish for this topic. Ship a wide row from the database, emit a narrow shape on the wire | | coalesceBy | - | (data) => key extractor. Publishes fan out via per-socket sendCoalesced; the latest value for each (topic, key) pair wins. For high-frequency latest-value streams (prices, cursors, presence). Cannot combine with volatile | | volatile | false | Mark messages fire-and-forget. Disables seq stamping for this topic so reconnects with lastSeenSeq won't try to backfill. Wire-level drop-on-backpressure is the adapter's default. For typing indicators, telemetry pings, cursors | | staleAfterMs | - | Per-topic staleness watchdog. If no events arrive for N ms, the loader re-runs and the result broadcasts as a refreshed event. Useful for streams whose source can quietly stop emitting. See Stream lifecycle hooks | | invalidateOn | - | String or array of glob-style topic patterns (e.g. 'todos:*'). When ctx.publish hits a matching topic, the stream's loader reruns and the result broadcasts as a refreshed event. See Stream lifecycle hooks | | onError | - | (err, ctx, topic) per-stream observer. Fires on loader throws (subscribe / stale-reload / .load() SSR). Errors thrown inside are silently swallowed | | classOfService | - | Names a class registered via live.admission(). New subscribes are shed under matching pressure. See Load shedding | | onSubscribe | - | Callback (ctx, topic) fired when a client subscribes | | onUnsubscribe | - | Callback (ctx, topic, remainingSubscribers) fired when a client disconnects. remainingSubscribers counts OTHER WebSockets still on the topic - use it to tear down upstream feeds at zero | | filter / access | - | Per-connection publish filter (see Access control) | | delta | - | Delta sync config (see Delta sync and replay) | | version | - | Schema version (see Schema evolution) | | migrate | - | Migration functions (see Schema evolution) |

Reconnection

When the WebSocket reconnects, streams automatically refetch initial data and resubscribe. The store keeps showing stale data during the refetch - it does not reset to undefined.

Mid-flight RPCs reject with DISCONNECTED. If the WS drops while an RPC or mutate is awaiting a response, the corresponding promise rejects with RpcError('DISCONNECTED'). Callers that wrap the call in mutate(asyncOp, change) get auto-rollback for free: the optimistic-queue entry settles as failed, removing the placeholder from the displayed state. Concurrent in-flight mutates each roll back independently - if A and B are both pending when the WS drops, both promises reject and both placeholders are removed in one display recompute.

Catch-up via initial fetch on resubscribe. Once the WS reconnects, every active stream re-runs its loader and broadcasts the result through the same merge strategy used on first subscribe. Pub/sub events that fired during the disconnect window arrive as part of the fresh fetch (they're materialized in the loader's data source). For tighter "no-frame-loss" guarantees use the delta configuration (see Delta sync and replay), which fills small gaps via the per-topic seq-numbered replay buffer and falls back to delta sync for larger gaps.

Triggering reconnect. The next RPC call after the WS drops triggers the adapter's reconnect logic. Most apps don't need to do anything special - the user clicks something, the RPC fires, the adapter reconnects, the call lands. If you want to display a reconnecting banner, watch the status store from svelte-adapter-uws/client for the 'reconnecting' -> 'open' transition.

Connection state stores

Four reactive stores re-export from svelte-realtime/client for rendering connection state without app-side WebSocket plumbing.

<script>
  import { status, failure, quiescent, health } from 'svelte-realtime/client';
</script>

{#if $health === 'degraded'}
  <Banner severity="warn">Real-time updates paused, reconnecting...</Banner>
{:else if $failure?.class === 'TERMINAL'}
  <Banner severity="error">Session expired. <a href="/login">Sign in again</a></Banner>
{:else if $failure?.class === 'THROTTLE'}
  <Banner severity="warn">Server is busy. Reconnecting more slowly...</Banner>
{:else if $status === 'disconnected'}
  <Banner severity="info">Reconnecting...</Banner>
{/if}

{#if !$quiescent}
  <Spinner />
{/if}

| Store | Type | Behavior | |---|---|---| | status | 'connecting' \| 'open' \| 'suspended' \| 'disconnected' \| 'failed' | Connection state machine. suspended = tab in background; failed = terminal (auth denied or close() called) | | failure | { kind, class, code, reason } \| null | Cause of the most recent non-open transition. class is TERMINAL (auth) / EXHAUSTED (max retries) / THROTTLE (4429) / RETRY / AUTH (HTTP preflight). Cleared on next 'open'. Not set on intentional close() | | quiescent | Readable<boolean> | true when every active stream has settled (initial load + all reconnects). Continuous signal - a false -> true transition after a reconnect cycle marks "everything caught up" | | health | 'healthy' \| 'degraded' | System-wide health, sourced from degraded / recovered events on the __realtime topic. Stays 'healthy' until something publishes - typically the extensions package's pub/sub bus circuit breaker |

failure and quiescent are pure additions; apps that don't use them pay nothing. health lazily subscribes to __realtime only on first read; never reading it = no subscription.

Apps that need richer health detail (reason strings, timestamps) can listen to the topic directly:

import { on } from 'svelte-adapter-uws/client';
on('__realtime').subscribe((envelope) => { /* full payload */ });

Svelte 5 store helpers

Generated $live/* stream stores work out of the box as Svelte 4 Readable<T> values via the $store auto-subscribe syntax. For Svelte 5 apps, two methods are exposed alongside the existing subscribe interface so component code stays terse without reaching for $derived.by(() => $store ?? []) boilerplate.

`store.rune()` - Svelte 5 reactive object

Returns an object with a single current getter, backed by Svelte's fromStore from svelte/store. Reading current inside an effect or component subscribes via Svelte's createSubscriber for fine-grained reactivity; reading it outside an effect synchronously returns the latest value.

<script>
  import { todos } from '$live/todos';
  const items = todos.rune();
</script>

<p>{items.current?.length ?? 0} items</p>
{#each items.current ?? [] as todo}
  <li>{todo.title}</li>
{/each}

rune() requires Svelte 5 (the fromStore export is not available in Svelte 4) and throws a descriptive error if called against an older runtime. Apps still on Svelte 4 use the $store auto-subscribe syntax instead.

`store.map(fn)` - per-item projection

Returns a mapped store with the same { subscribe, rune, map } shape as the source. Idiomatic alternative to $derived.by(() => ($stream ?? []).map(...)) and avoids the $derived(() => ...) footgun where storing a function reference instead of its return value silently breaks rendering.

<script>
  import { todos } from '$live/todos';
  const titles = todos.map(t => t.title);
  // Or compose with rune() for Svelte 5:
  const titlesRune = todos.map(t => t.title).rune();
</script>

{#each $titles as title}<li>{title}</li>{/each}

Semantics match the documented ($stream ?? []).map(fn) pattern: a null or undefined source emits []; an array source emits source.map(fn); a non-array source (set-merge stream, paginated wrapper) emits [] after a dev-mode console.warn pointing at the merge-strategy docs. Subscriptions are lazy: the source is only subscribed while at least one mapped consumer is active. Chains via further .map() calls preserve the same shape.

`empty` - bundled placeholder store

Every generated $live/<name>.js re-exports an empty store that holds undefined. Use it as the fallback for conditional streams without importing readable from svelte/store:

<script>
  import { todos, empty } from '$live/todos';
  let { user, orgId } = $props();
  const items = $derived(user ? todos(orgId) : empty);
</script>

{#each $items ?? [] as todo}
  <li>{todo.title}</li>
{/each}

Auto-imported alongside the stream itself; nothing extra to wire.

Error handling

The data store value never changes shape. It is always your data type or undefined while loading. Errors and connection status live on separate reactive stores so a network failure can never crash your UI:

| Property | Type | Description | |---|---|---| | $store | T \| undefined | Your data. Never replaced by an error object. On failure, the last loaded value is preserved. | | store.error | Readable<RpcError \| null> | Current error, or null when healthy. | | store.status | Readable<'loading' \| 'connected' \| 'reconnecting' \| 'error'> | Connection status. |

Handle loading in your template:

{#if $messages === undefined}
  <p>Loading...</p>
{:else}
  {#each $messages as msg (msg.id)}
    <p>{msg.text}</p>
  {/each}
{/if}

To show errors, subscribe to the .error store:

<script>
  import { messages } from '$live/chat';

  const err = messages.error;
  const status = messages.status;
</script>

{#if $err}
  <p>Error: {$err.message} ({$err.code})</p>
{/if}

{#if $status === 'reconnecting'}
  <p>Reconnecting...</p>
{/if}

Defensive patterns like ($store ?? []).filter(...) work correctly because $store is always an array or undefined.

For RPC calls, errors are thrown as RpcError with a code field:

import { sendMessage } from '$live/chat';

try {
  await sendMessage(text);
} catch (err) {
  if (err.code === 'VALIDATION') {
    // handle validation error - err.issues has details
  } else if (err.code === 'UNAUTHORIZED') {
    // redirect to login
  }
}

Terminal close codes

When the adapter's ready() promise rejects (terminal close codes 1008, 4401, 4403, exhausted retries, or explicit close()), svelte-realtime:

Rejects all pending RPCs immediately with RpcError('CONNECTION_CLOSED', ...)
Sets .error on all active stream stores (the data value is preserved)
Drains the offline queue with errors

RPCs called after a terminal close reject immediately without sending.

Reusable error boundary component

For Svelte 5, you can build a reusable boundary that handles loading and error states:

<!-- src/lib/StreamView.svelte -->
<script>
  /** @type {{ store: any, children: import('svelte').Snippet, loading?: import('svelte').Snippet, error?: import('svelte').Snippet<[any]> }} */
  let { store, children, loading, error } = $props();

  let value = $derived($store);
  const err = store.error;
</script>

{#if value === undefined}
  {#if loading}
    {@render loading()}
  {:else}
    <p>Loading...</p>
  {/if}
{:else if $err}
  {#if error}
    {@render error($err)}
  {:else}
    <p>Error: {$err.message}</p>
  {/if}
{:else}
  {@render children()}
{/if}

Use it to wrap any stream:

<script>
  import StreamView from '$lib/StreamView.svelte';
  import { messages, sendMessage } from '$live/chat';
</script>

<StreamView store={messages}>
  {#each $messages as msg (msg.id)}
    <p>{msg.text}</p>
  {/each}

  {#snippet loading()}
    <div class="skeleton-loader">Loading messages...</div>
  {/snippet}

  {#snippet error(err)}
    <div class="error-banner">
      <p>Could not load messages: {err.message}</p>
      <button onclick={() => location.reload()}>Retry</button>
    </div>
  {/snippet}
</StreamView>

With default slots, the minimal version is just:

<StreamView store={messages}>
  {#each $messages as msg (msg.id)}
    <p>{msg.text}</p>
  {/each}
</StreamView>

This removes the {#if}/{:else} boilerplate from every page that uses a stream.

Per-module auth

Every file in src/live/ can export a _guard that runs before all functions in that file.

// src/live/admin.js
import { live, guard, LiveError } from 'svelte-realtime/server';

export const _guard = guard((ctx) => {
  if (ctx.user?.role !== 'admin')
    throw new LiveError('FORBIDDEN', 'Admin only');
});

export const deleteUser = live(async (ctx, userId) => {
  await db.users.delete(userId);
});

export const banUser = live(async (ctx, userId) => {
  await db.users.ban(userId);
});

Both deleteUser and banUser require admin access. No need to check in each function.

guard() accepts multiple functions for composable middleware chains. They run in order, and earlier ones can enrich ctx for later ones:

export const _guard = guard(
  (ctx) => { if (!ctx.user) throw new LiveError('UNAUTHORIZED'); },
  (ctx) => { ctx.permissions = lookupPermissions(ctx.user.id); },
  (ctx) => { if (!ctx.permissions.includes('write')) throw new LiveError('FORBIDDEN'); }
);

Dynamic topics

Use a function instead of a string as the first argument to live.stream() for per-entity streams. The client-side stub becomes a factory function - call it with arguments to get a cached store for that entity.

The first argument's shape decides the client export. If the first argument is a string, the export is a Svelte store and you read it as $messages. If the first argument is a function, the export is a factory that returns a store, and you must call it first: const messages = roomMessages(data.roomId); ... $messages. Forgetting the call gives Svelte error: store_invalid_shape - 'roomMessages' is not a store with a 'subscribe' method during SSR. If the topic does not depend on arguments, prefer the string form.

// src/live/rooms.js
import { live } from 'svelte-realtime/server';

export const roomMessages = live.stream(
  (ctx, roomId) => 'chat:' + roomId,
  async (ctx, roomId) => db.messages.forRoom(roomId),
  { merge: 'crud', key: 'id' }
);

export const sendToRoom = live(async (ctx, roomId, text) => {
  const msg = await db.messages.insert({ roomId, userId: ctx.user.id, text });
  ctx.publish('chat:' + roomId, 'created', msg);
  return msg;
});

<!-- src/routes/rooms/[id]/+page.svelte -->
<script>
  import { roomMessages, sendToRoom } from '$live/rooms';
  let { data } = $props();

  // roomMessages is a function - call it with the room ID to get a store
  const messages = roomMessages(data.roomId);
</script>

{#each $messages as msg (msg.id)}
  <p>{msg.text}</p>
{/each}

Same arguments return the same cached store instance. The cache is cleaned up when all subscribers unsubscribe.

Topic registry

Centralize topic strings so the SQL trigger and the stream definition reference one source of truth. defineTopics(map) validates the map at boot and exposes __patterns for tooling.

// src/lib/topics.js
import { defineTopics } from 'svelte-realtime/server';

export const TOPICS = defineTopics({
  audit: (orgId) => `audit:${orgId}`,
  security: (orgId) => `security:${orgId}`,
  systemNotices: 'system:notices'
});

// Stream definitions reference the registry
live.stream((ctx, orgId) => TOPICS.audit(orgId), loadAudit, { merge: 'crud' });

// Tooling reads __patterns to derive shapes
TOPICS.__patterns;
// => { audit: 'audit:{arg0}', security: 'security:{arg0}', systemNotices: 'system:notices' }

Map values can be strings (static topics) or (...args) => string functions (dynamic topics). The helper validates non-empty strings and rejects reserved names (__patterns, __definedTopics). Pattern derivation calls each function with sentinel placeholders ({arg0}, {arg1}, ...) and falls back to '<dynamic>' if the function throws on placeholders or returns a non-string.

Build-time registry check

When the Vite plugin sees a defineTopics({...}) call anywhere under src/, it builds a registry of patterns and validates string-literal topics passed to live.stream(...) and live.channel(...) against it. A literal that does not match any registered pattern triggers a one-shot warning per (file, topic) pair:

[svelte-realtime] src/live/feed.js: live.stream topic 'mistyped-topic' is not in
your TOPICS registry. Either add it to defineTopics({...}) or call TOPICS.<name>(...)
instead of passing a string literal.

The check covers static-string patterns (feed: 'feed:notices') and arrow-return template literals (audit: (orgId) => \audit:${orgId}`); template interpolations match .+so'audit:org-123'and'audit:any-id'both pass against theauditpattern. Function references and other dynamic value shapes are silently skipped at parse time - the warning only fires for literal topics under a confidently parsed registry. If your project does not calldefineTopics` at all, the check is disabled.

Schema validation

Use live.validated(schema, fn) to validate the first argument against a schema before the function runs. Any Standard Schema-compatible validator is supported, including Zod, ArkType, Valibot, and others.

import { z } from 'zod';
import { live } from 'svelte-realtime/server';

const CreateTodo = z.object({
  text: z.string().min(1).max(200),
  priority: z.enum(['low', 'medium', 'high']).optional()
});

export const addTodo = live.validated(CreateTodo, async (ctx, input) => {
  const todo = await db.todos.insert({ ...input, userId: ctx.user.id });
  ctx.publish('todos', 'created', todo);
  return todo;
});

Because live.validated() uses the Standard Schema interface, you can swap in any compatible validator:

import { type } from 'arktype';
import { live } from 'svelte-realtime/server';

const CreateTodo = type({ text: 'string>0', priority: '"low"|"medium"|"high"|undefined' });

export const addTodo = live.validated(CreateTodo, async (ctx, input) => {
  const todo = await db.todos.insert({ ...input, userId: ctx.user.id });
  ctx.publish('todos', 'created', todo);
  return todo;
});

On the client, validated exports work like regular live() calls. Validation errors are thrown as RpcError with code: 'VALIDATION' and an issues array.

Validating stream arguments

Stream arguments validate via the args option on live.stream(). Validation runs BEFORE topic resolution, so a malformed argument can never reach a dynamic topic function.

import { z } from 'zod';

export const auditFeed = live.stream(
  (ctx, orgId) => `audit:${orgId}`,
  async (ctx, orgId) => loadAudit(orgId),
  { args: z.tuple([z.string().uuid()]) }
);

Validation failures reject the subscribe RPC with { code: 'VALIDATION', issues }. Both the WebSocket and .load() SSR paths apply the schema. Coerced values from the schema (e.g. Zod transforms) flow through to the loader and the topic function.

Channels

Ephemeral pub/sub topics with no database initialization. Clients subscribe and receive live events immediately.

// src/live/typing.js
import { live } from 'svelte-realtime/server';

export const typing = live.channel('typing:lobby', { merge: 'presence' });

<script>
  import { typing } from '$live/typing';
</script>

{#each $typing as user (user.key)}
  <span>{user.data.name} is typing...</span>
{/each}

Dynamic channels work the same way:

export const cursors = live.channel(
  (ctx, docId) => 'cursors:' + docId,
  { merge: 'cursor' }
);

For high-frequency streams where a missed frame is acceptable (typing indicators, telemetry pings, raw cursor positions you don't need replayed on reconnect), set volatile: true on the stream:

export const cursors = live.stream(
  (ctx, roomId) => `room:${roomId}:cursors`,
  loader,
  { merge: 'cursor', volatile: true }
);

Two effects: per-event seq stamping is skipped for the topic (so reconnect with lastSeenSeq won't try to backfill), and the option declares intent at the call site. Wire-level "drop on backpressure" is the adapter's default behavior already - uWS auto-skips a subscriber whose outbound buffer is over maxBackpressure (default 64 KB). Cannot combine with coalesceBy (queue vs drop - different intents) or replay (volatile messages aren't buffered for resume).

SSR hydration

Call live functions from +page.server.js to load data server-side, then hydrate the client-side stream store to avoid loading spinners.

// src/routes/chat/+page.server.js
export async function load({ platform, locals }) {
  const { messages } = await import('$live/chat');
  const data = await messages.load(platform, { user: locals.user });
  return { messages: data };
}

<!-- src/routes/chat/+page.svelte -->
<script>
  import { messages } from '$live/chat';
  let { data } = $props();

  // Pre-populate the store with SSR data - no loading spinner
  const msgs = messages.hydrate(data.messages);
</script>

{#each $msgs as msg (msg.id)}
  <p>{msg.text}</p>
{/each}

The hydrated store still subscribes for live updates on first render. It keeps the SSR data visible instead of showing undefined during the initial fetch. Guards still run during .load() calls. Pass { user } as the second argument if your guard or init function needs user data.

For dynamic streams (streams with a topic function), call the stream first to get the store, then hydrate:

// src/routes/team/[id]/+page.server.js
export async function load({ platform, locals, params }) {
  const { invitations } = await import('$live/invitation');
  const data = await invitations.load(platform, { args: [params.id], user: locals.user });
  return { invitations: data };
}

<!-- src/routes/team/[id]/+page.svelte -->
<script>
  import { invitations } from '$live/invitation';
  import { page } from '$app/state';
  let { data } = $props();

  const invites = invitations(page.params.id).hydrate(data.invitations);
</script>

{#each $invites as invite (invite.id)}
  <p>{invite.email}</p>
{/each}

Batching

Group multiple RPC calls into a single WebSocket frame to reduce round trips.

<script>
  import { batch } from 'svelte-realtime/client';
  import { createBoard, addColumn, addCard } from '$live/boards';

  async function setupBoard() {
    const [board, column, card] = await batch(() => [
      createBoard('My Board'),
      addColumn('To Do'),
      addCard('First task')
    ]);
  }
</script>

By default, calls in a batch run in parallel on the server. Pass { sequential: true } when order matters:

const [board, column] = await batch(() => [
  createBoard('My Board'),
  addColumn(boardId, 'To Do')
], { sequential: true });

Each call resolves or rejects independently - one failure does not cancel the others. Batches are limited to 50 calls - enforced both client-side (rejects before sending) and server-side.

A batch() containing only one call sends a bare RPC frame (no batch envelope) and the server replies through the normal RPC path - so the defensive "always wrap writes in batch() for symmetry" pattern pays no envelope overhead. Batches of 2+ keep the envelope.

Server-side batching

Use ctx.batch() inside RPC handlers to publish multiple messages in a single call:

export const resetBoard = live(async (ctx, boardId) => {
  await db.boards.reset(boardId);
  ctx.batch([
    { topic: `board:${boardId}`, event: 'set', data: [] },
    { topic: `board:${boardId}:presence`, event: 'set', data: [] }
  ]);
});

Volatile RPC (fire-and-forget)

For high-frequency one-way calls where the caller has no reply to await - cursor moves, drag updates, typing indicators, telemetry beacons, heartbeats - use live.volatile(fn) server-side + .fireAndForget(...args) client-side. The wire frame carries no id; the server runs the full handler chain (middleware, guards, rate limits, validation) but does not write a response.

// src/lib/realtime/cursors.js
import { live } from 'svelte-realtime';

export const moveCursor = live.volatile(async (ctx, boardId, pos) => {
  // ctx.publish / ctx.shed / guards all work normally
  ctx.publish(`board:${boardId}`, 'cursor', pos);
});

<script>
  import { moveCursor } from '$live/cursors';

  function onPointerMove(e) {
    moveCursor.fireAndForget(boardId, { x: e.clientX, y: e.clientY });
    // returns void synchronously - no Promise, no await
  }
</script>

What .fireAndForget() skips that a normal RPC does:

ID allocation (_nextId())
Promise allocation
Dedup-Map entry + queueMicrotask(delete)
Pending-Map entry
Timer allocation (per-call 30s timeout)
DevTools-pending entry

At 60-120Hz on a single hot path that is 100K+ short-lived heap allocations per second avoided on the client.

Safety:

Errors disappear silently from the caller. A volatile call that fails auth, validation, or throws still runs through metrics (_recordRpcMetrics) and server logs - operators see the failure - but the wire carries no reply, so the caller does not. Use live.volatile() only when this is the intended contract.
Backpressure drop. Before send, the client reads WS.bufferedAmount; if it exceeds volatileBackpressureBytes (default 4 MB, configurable via configure(...)), the frame is dropped silently and __devtools.volatileDropped ticks. Dev-mode emits a one-shot console.warn on first drop per session.
Offline drop. Volatile calls made while disconnected are silently dropped. They do not enter the offline queue (which is for awaited mutations).
Inside batch(). Throws in dev, no-op in prod. Volatile bypasses batching by design.

Server-side marker is recommended, not required. The wire shape (id absent) is the actual contract. A .fireAndForget() against a plain live() handler also works - server processes it, just skips the reply - but dev-mode emits a one-shot warning per such path naming the handler so accidental fire-and-forget surfaces. Mark intentional one-way handlers with live.volatile() to silence the warning and document intent. The marker can sit at any depth inside live.rateLimit / live.idempotent / live.breaker / live.validated / live.lock wrappers (the framework walks __wrappedFn to find it).

When NOT to use .fireAndForget():

The caller needs to know whether the call succeeded -> use the normal awaited RPC.
The call needs to be retried on failure -> use .with({ idempotencyKey }) + normal RPC.
The call should survive a disconnect -> use the offline queue via the normal RPC.
The call is sometimes one-way, sometimes interesting -> keep the handler live() (not live.volatile()) and choose at each call site.

live.notify vs .fireAndForget(). Both are fire-and-forget, but they go in opposite directions: live.notify(target, event, data) is server -> client (server-initiated push, no client reply expected), while .fireAndForget(...args) is client -> server (client-initiated RPC, no server reply emitted). Different surfaces, different use cases.

Optimistic updates

Apply changes to a stream store instantly, then roll back if the server call fails.

<script>
  import { todos, addTodo } from '$live/todos';

  async function add(text) {
    const tempId = 'temp-' + Date.now();
    const rollback = todos.optimistic('created', { id: tempId, text });

    try {
      await addTodo(text);
      // Server broadcasts the real 'created' event, which replaces the
      // optimistic entry (matched by key) with the confirmed data.
    } catch {
      rollback();
    }
  }
</script>

optimistic(event, data) returns a rollback function that restores the store to its previous state. It works with all merge strategies:

| Merge | Events | Behavior | |---|---|---| | crud | created, updated, deleted | Modifies array by key. Server event with same key replaces the optimistic entry. | | latest | any event name | Appends data to the ring buffer. | | set | any event name | Replaces the entire value. |

Auto-rollback with `store.mutate()`

store.mutate(asyncOp, optimisticChange) wraps the apply-await-rollback pattern. Applies the optimistic change synchronously, awaits the RPC, and on rejection rolls back and re-throws.

// Event-based: server's confirming event reconciles the placeholder by key
const todo = await todos.mutate(
  () => addTodo(text),
  { event: 'created', data: { id: tempId(), text } }
);

// Free-form mutator: arbitrary local change, no merge-strategy assumptions
await todos.mutate(
  () => removeTodo(id),
  (current) => current.filter((t) => t.id !== id)
);

Returns the result of asyncOp on success. The free-form mutator receives a shallow copy: top-level array shape changes (push, pop, filter, splice) roll back cleanly; in-place item field mutations (draft[0].name = 'x') do NOT, because the draft and the prior items share item references. Replace whole items instead: draft[i] = { ...draft[i], name: 'x' }.

Concurrent mutates roll back independently. Pending mutations are tracked in an in-flight queue and the displayed value is recomputed by replaying that queue against the un-overlaid server state after every server event and every settle. If mutate A and mutate B are both in flight and both fail, the displayed state returns to the latest server state with no phantom traces of either A or B. Server events with a key matching a queue entry's optimistic key absorb the entry, so the typical "client generates UUID, server confirms with same id" flow does not flicker.

RPC-bound shorthand: `rpc.createOptimistic()`

Every generated RPC stub also exposes a .createOptimistic(store, callArgs, optimisticChange) method. It threads callArgs into the optimistic-change callback so the call site doesn't have to capture them in a closure:

import { sendMessage, messages } from '$live/chat';

await sendMessage.createOptimistic(
  messages,
  ['Hello!'],
  (current, args) => [...current, { id: tempId(), text: args[0] }]
);

callArgs is always passed as an array (so multi-argument RPCs work the same as single-argument ones; pass [arg] for the single-arg case). The third argument accepts the same two shapes as store.mutate(): a (current, args) => newValue function or a { event, data } object. Equivalent to:

store.mutate(() => rpc(...callArgs), wrappedChange);

so behavior on success/rollback/server-confirmation is identical to store.mutate(). The shorthand is purely syntactic; reach for store.mutate() directly when the asyncOp isn't an RPC (third-party API call, multi-step flow, etc.).

Curried form - bind once, call many times. Pass two arguments instead of three (store, change) and createOptimistic returns a callable bound to that store + change:

const optimisticSend = sendMessage.createOptimistic(
  messages,
  (current, args) => [...current, { id: tempId(), text: args[0] }]
);
await optimisticSend('Hello!');
await optimisticSend('There!');

Stream-side spelling - the same flow can be expressed from the stream's perspective via store.createOptimistic(rpc, callArgs, change):

await messages.createOptimistic(
  sendMessage,
  ['Hello!'],
  (current, args) => [...current, { id: tempId(), text: args[0] }]
);

Identical semantics to the RPC-side spelling; pick whichever reads more naturally for your call site (stream-focused code prefers store.createOptimistic; RPC-focused code prefers rpc.createOptimistic).

Stream pagination

For large datasets, return { data, hasMore, cursor } from your stream init function to enable cursor-based pagination.

// src/live/feed.js
import { live } from 'svelte-realtime/server';

export const posts = live.stream('posts', async (ctx) => {
  const limit = 20;
  const rows = await db.posts.list({ limit: limit + 1, after: ctx.cursor });
  const hasMore = rows.length > limit;
  const data = hasMore ? rows.slice(0, limit) : rows;
  const cursor = data.length > 0 ? data[data.length - 1].id : null;
  return { data, hasMore, cursor };
}, { merge: 'crud', key: 'id' });

<script>
  import { posts } from '$live/feed';

  async function loadNext() {
    await posts.loadMore();
  }
</script>

{#each $posts as post (post.id)}
  <p>{post.title}</p>
{/each}

{#if posts.hasMore}
  <button onclick={loadNext}>Load more</button>
{/if}

The server detects the { data, hasMore } shape automatically. ctx.cursor contains the cursor value sent by the client on subsequent loadMore() calls (null on the first request).

Undo and redo

Stream stores support history tracking for undo/redo.

<script>
  import { todos } from '$live/todos';

  todos.enableHistory(100); // max 100 entries

  function handleUndo() {
    todos.undo();
  }
</script>

<button onclick={handleUndo} disabled={!todos.canUndo}>Undo</button>
<button onclick={() => todos.redo()} disabled={!todos.canRedo}>Redo</button>

History is recorded after every mutation (both live events and optimistic updates). Call enableHistory() once to start tracking.

Request deduplication

Identical RPC calls made within the same microtask are automatically coalesced into a single request.

// These two calls happen in the same microtask - only one request is sent
const [a, b] = await Promise.all([
  getUser(userId),
  getUser(userId) // same call, same args - reuses the first request
]);

To bypass deduplication and force a fresh request:

const result = await getUser.fresh(userId); // always sends a new request

Dev-mode coalesce warning. Stress tests and parallel-fan-out patterns like Promise.allSettled(Array.from({ length: 25 }, () => buyProduct('phone'))) expect N wire requests but get one - all 25 promises resolve to the same response, the server-side state shows a single decrement, and the failure is silent. To make this discoverable, the client logs a one-time console.warn on the first coalesce per RPC path per session, with a one-line pointer to .fresh(...). Dev-only (stripped under NODE_ENV=production); double-tap dedup on the same path warns once, never again.
[svelte-realtime] coalesced two or more identical calls to 'shop/buy' within one microtask - only one wire request was sent and all callers received the same response. ... If you wanted N parallel requests (stress test, fan-out), call `.fresh(...args)` on the rpc to bypass dedup. Warned once per path per session.

Idempotency keys

Microtask deduplication only collapses calls within the same tick. For durable safety against retries that span reconnects, tab reloads, or offline replay, wrap the handler with live.idempotent(config, fn). Identical calls (by key) return the cached result without re-running the handler.

// Server: server-derived key (Stripe-style)
import { live } from 'svelte-realtime/server';

export const createOrder = live.idempotent(
  { keyFrom: (ctx, input) => `order:${ctx.user.id}:${input.clientOrderId}`, ttl: 48 * 3600 },
  live.validated(OrderSchema, async (ctx, input) => {
    return db.orders.insert({ userId: ctx.user.id, ...input });
  })
);

// Client: envelope-supplied key (uuid per intent)
import { createOrder } from '$live/orders';

const intentId = crypto.randomUUID();
const order = await createOrder.with({ idempotencyKey: intentId })(payload);

Resolution: keyFrom(ctx, ...args) if defined, otherwise the client envelope's idempotencyKey, otherwise the wrapper is a no-op. Only successful results cache; throwing handlers abort the slot so the next caller re-runs. Default store is in-process and bounded; for multi-instance deployments pass store: createIdempotencyStore(redis) from svelte-adapter-uws-extensions.

Field name divergence with live.lock. live.idempotent uses keyFrom; live.lock uses key (which accepts a string OR a function). Mistakenly mirroring the wrong helper's shape used to silently fall through to the no-key bypass, breaking the one-per-key guarantee with no warning. Unknown config fields now throw at registration time with a cross-helper hint:
[svelte-realtime] live.idempotent: unknown config field 'key'. Allowed: keyFrom, store, ttl. Hint: live.lock uses 'key' but live.idempotent uses 'keyFrom' (the names diverged historically).

| Option | Default | Description | |---|---|---| | keyFrom | - | (ctx, ...args) => string \| null \| undefined. null/undefined falls back to the envelope key | | store | in-process | Any object exposing acquire(key, ttlSec) matching the extensions store contract | | ttl | 172800 (48h) | TTL in seconds. 0 skips the cache write (concurrent waiters re-run after the first finishes) |

__rpc().with({ ... }) composes options on the same surface:

// Compose idempotency + per-call timeout
await createOrder.with({ idempotencyKey: id, timeout: 60_000 })(payload);

| .with({}) option | Description | |---|---| | idempotencyKey | Carried in the envelope for the server's live.idempotent wrapper | | timeout | Per-call timeout in ms; overrides the global configure({ timeout }) default. Sleep-detect threshold scales with the override |

Offline queue

Queue RPC calls when the WebSocket is disconnected and replay them on reconnect.

import { configure } from 'svelte-realtime/client';

configure({
  offline: {
    queue: true,        // enable offline queuing
    maxQueue: 100,      // drop oldest if queue exceeds this (default: 100)
    maxAge: 60000,      // auto-reject queued calls older than this (ms)
    beforeReplay(call) {
      // Return false to drop stale mutations
      return Date.now() - call.queuedAt < 60000; // drop if older than 1 minute
    },
    onReplayError(call, error) {
      console.warn('Replay failed:', call.path, error);
    }
  }
});

When offline queuing is enabled, RPC calls made while disconnected return promises that resolve when the call is replayed after reconnection. If the queue overflows, the oldest entry is dropped and its promise rejects with QUEUE_FULL. If maxAge is set, queued calls older than that threshold are rejected with STALE at replay time.

Connection hooks

Use configure() on the client to react to WebSocket connection state changes.

<!-- src/routes/+layout.svelte -->
<script>
  import { configure } from 'svelte-realtime/client';

  configure({
    onConnect() {
      // Reconnected after a drop
      invalidateAll();
    },
    onDisconnect() {
      showBanner('Connection lost, reconnecting...');
    }
  });
</script>

Call configure() once at app startup. The hooks fire on state transitions only (not on the initial connection).

| Option | Description | |---|---| | url | Full WebSocket URL for cross-origin or native app usage (e.g. 'wss://api.example.com/ws') | | auth | true (or a custom path) to enable an HTTP preflight before each WebSocket upgrade so cookies set by the server's authenticate hook ride a normal HTTP response. Required behind Cloudflare Tunnel and other proxies that drop Set-Cookie on 101 responses. Requires svelte-adapter-uws >= 0.4.12. | | onConnect() | Called when the WebSocket connection opens after a reconnect | | onDisconnect() | Called when the WebSocket connection closes | | beforeReconnect() | Called before each reconnection attempt (can be async) | | timeout | Default RPC timeout in ms (default 30000). Per-call .with({ timeout }) overrides. | | resumeGraceMs | Stream resume-grace window in ms (default 60000). See Pause and resume without re-rehydrating below. Set to 0 to disable. |

Pause and resume without re-rehydrating

When the last subscriber of a stream unsubs, the stream releases its WebSocket subscription immediately (giving the server back its slot, dropping the in-flight counter) but keeps the in-memory data model -- currentValue, the last seen seq / version, the pagination cursor, and any history -- for resumeGraceMs (default 60 seconds). If a new subscribe() lands inside that window, the stream re-attaches its listeners and sends the retained cursor on the resume envelope, so the server can fill the gap from its bounded replay buffer (or delta.fromSeq, or a truncated-cache fall-through to a full rehydrate) instead of cold-starting.

This is the default for two reasons:

Pause/resume UIs work for free. A {#if active} <SubscribedComponent /> {/if} toggle, or an $effect whose subscribe-arm flips on user action, can pause and resume the subscription without re-loading from scratch. The events that arrived during the pause stream in via the replay buffer.
Browser back/forward feels instant. Navigating away and back within the grace window restores the previous data immediately, and any events the user missed are gap-filled by the server.

If the grace expires without a new subscriber, the data model resets and the next subscribe is a true cold start. Apps that prefer aggressive memory reclamation can shorten or disable the grace:

configure({ resumeGraceMs: 0 });        // every unsub is a full reset (pre-grace behavior)
configure({ resumeGraceMs: 5_000 });    // 5s grace covers brief toggles
configure({ resumeGraceMs: 300_000 });  // 5min grace for navigation-heavy apps

The grace only affects local data retention. The server's replay buffer and delta.fromSeq window are independent and govern how far back the gap-fill can reach.

Cross-origin and native app usage

When using svelte-realtime from a client that runs on a different origin (Svelte Native, React Native, or any standalone app), pass the url option to point at your SvelteKit backend:

import { configure, __rpc, __stream } from 'svelte-realtime/client';

configure({ url: 'wss://my-sveltekit-app.com/ws' });

// Call a live function (equivalent to $live/chat.sendMessage, but untyped)
const sendMessage = __rpc('chat/sendMessage');
await sendMessage('hello');

// Subscribe to a stream (returns a Svelte store)
const messages = __stream('chat/messages', { merge: 'crud', key: 'id' });
messages.subscribe((value) => console.log(value));

The typed $live/* imports and stream hydration are generated by the Vite plugin and only work inside a SvelteKit project. Outside SvelteKit, use __rpc() and __stream() directly. You get the same reconnection, offline queue, and batching - just without codegen and types.

When url is set, the default same-origin WebSocket URL is bypassed entirely. Requires svelte-adapter-uws 0.4.8+.

Browser clients authenticate via cookies set during login. Native clients typically use a token instead. Your upgrade hook can support both:

// src/hooks.ws.js
export { message } from 'svelte-realtime/server';

export function upgrade({ cookies, url }) {
  // Browser - cookie auth
  const session = cookies.session_id;
  if (session) return validateSession(session);

  // Native app - token auth via query string
  const token = new URL(url, 'http://n').searchParams.get('token');
  if (token) return validateToken(token);

  return false;
}

The native client passes the token in the URL:

configure({ url: 'wss://my-sveltekit-app.com/ws?token=...' });

Refreshing session cookies on connect (Cloudflare Tunnel and friends)

Cloudflare Tunnel and other strict edge proxies silently drop the Set-Cookie header on WebSocket 101 Switching Protocols responses. The connection appears to open server-side, then the client immediately sees close 1006 and never receives a single frame. The classic symptom for this in production: WebSockets work locally and on a bare server, then break the moment you put Cloudflare in front.

Fix it in three pieces:

Export an authenticate hook from src/hooks.ws.{js,ts}. It runs as a normal HTTP POST /__ws/auth before every upgrade (including reconnects), so cookies you set ride a 204 No Content response that proxies route correctly.
Opt into the client preflight with configure({ auth: true }).
Use svelte-adapter-uws >= 0.4.12.

// src/hooks.ws.js
export { message, close, unsubscribe } from 'svelte-realtime/server';

export function upgrade({ cookies }) {
  const session = validateSession(cookies.session_id);
  return session ? { id: session.userId, name: session.name } : false;
}

export function authenticate({ cookies }) {
  const session = validateSession(cookies.get('session_id'));
  if (!session) return false;

  if (shouldRotate(session)) {
    cookies.set('session_id', rotate(session), {
      httpOnly: true,
      secure: true,
      sameSite: 'lax',
      path: '/'
    });
  }
  return { id: session.userId, name: session.name };
}

<!-- src/routes/+layout.svelte -->
<script>
  import { configure } from 'svelte-realtime/client';
  configure({ auth: true });
</script>

The client coalesces concurrent connects into a single in-flight preflight, treats 4xx as terminal, and falls back to normal reconnect backoff on 5xx and network errors.

Detector: if the client sees two consecutive WebSocket open->close cycles inside one second with no traffic, it logs a one-shot console.warn pointing at this section. That is the Cloudflare-Tunnel-eating-cookies fingerprint.

SvelteKit transport

realtimeTransport() from svelte-realtime/hooks is a SvelteKit transport-hook preset that auto-registers RpcError and LiveError serialization across the SSR / client boundary. Without it, typed errors thrown from +page.server.js load() arrive at +error.svelte as plain Error instances and lose their code field.

// src/hooks.js  (NOT hooks.server.js - the shared hook is required)
import { realtimeTransport } from 'svelte-realtime/hooks';

export const transport = realtimeTransport();

Compose with app-defined types via the optional extras parameter (user entries appear after defaults so they win on key conflict):

// src/hooks.js
import { realtimeTransport } from 'svelte-realtime/hooks';
import { Vector } from '$lib/geometry';

export const transport = realtimeTransport({
  Vector: {
    encode: (v) => v instanceof Vector && [v.x, v.y],
    decode: ([x, y]) => new Vector(x, y)
  }
});

Wire from src/hooks.js, NOT hooks.server.js. SvelteKit's transport primitive needs both encode (server-side) and decode (client-side hydration) visible at build time - the wrong file silently half-works (encode runs but decode never reaches the client). RpcError's optional issues field (carried by live.validated() failures) survives the round-trip. Validation runs at registration: malformed extras throw immediately so misconfiguration fails fast at app boot.

Combine stores

Compose multiple stream stores into a single derived store. When any source updates, the combining function re-runs.

<script>
  import { combine } from 'svelte-realtime/client';
  import { orders, inventory } from '$live/dashboard';

  const dashboard = combine(orders, inventory, (o, i) => ({
    pendingOrders: o?.filter(x => x.status === 'pending').length ?? 0,
    lowStock: i?.filter(x => x.qty < 10) ?? []
  }));
</script>

<p>Pending: {$dashboard.pendingOrders}</p>

combine() accepts 2-6 stores with typed overloads, plus a variadic fallback for more. Zero network overhead - all computation happens client-side.

Global middleware

Use live.middleware() to register cross-cutting logic that runs before per-module guards on every RPC and stream call.

import { live, LiveError } from 'svelte-realtime/server';

// Logging middleware
live.middleware(async (ctx, next) => {
  const start = Date.now();
  const result = await next();
  console.log(`[${ctx.user?.id}] took ${Date.now() - start}ms`);
  return result;
});

// Auth middleware - rejects unauthenticated requests globally
live.middleware(async (ctx, next) => {
  if (!ctx.user) throw new LiveError('UNAUTHORIZED', 'Login required');
  return next();
});

Middleware runs in registration order. Each must call next() to continue the chain. When no middleware is registered, there is zero overhead.

Throttle and debounce

Use ctx.throttle() and ctx.debounce() inside any live() function to rate-limit publishes.

export const updatePosition = live(async (ctx, x, y) => {
  // Throttle: publishes immediately, then at most once per 50ms (trailing edge guaranteed)
  ctx.throttle('cursors', 'update', { key: ctx.user.id, x, y }, 50);
});

export const saveSearch = live(async (ctx, query) => {
  // Debounce: waits for 300ms of silence before publishing
  ctx.debounce('search:' + ctx.user.id, 'set', { query }, 300);
});

ctx.throttle publishes the first call immediately, stores subsequent calls, and sends the last value when the interval expires (trailing edge). ctx.debounce resets the timer on each call and only publishes after silence.

Stream lifecycle hooks

Use onSubscribe and onUnsubscribe in stream options to run logic when clients join or leave a stream.

export const presence = live.stream('room:lobby', async (ctx) => {
  return db.presence.list('lobby');
}, {
  merge: 'presence',
  onSubscribe(ctx, topic) {
    ctx.publish(topic, 'join', { key: ctx.user.id, name: ctx.user.name });
  },
  onUnsubscribe(ctx, topic, remainingSubscribers) {
    ctx.publish(topic, 'leave', { key: ctx.user.id });
    if (remainingSubscribers === 0) stopUpstreamFeed(topic);
  }
});

onSubscribe fires after ws.subscribe(topic) and the initial data fetch. onUnsubscribe fires in real time when a client unsubscribes from a topic (adapter 0.4.0+), and also when the WebSocket closes for any remaining topics. Export both hooks from your hooks.ws.js:

export { message, close, unsubscribe } from 'svelte-realtime/server';

onUnsubscribe fires for both static and dynamic topics. For dynamic topics, the server tracks which stream produced each subscription and fires the correct hook. The unsubscribe hook fires as soon as the client drops a topic; close only fires for topics still active at disconnect time. There is no double-firing.

The third argument remainingSubscribers counts OTHER WebSockets still holding a realtime-stream subscription to the topic after the current one drops. Use it to tear down upstream feeds (CDC connections, polling loops, external pub/sub follows) when the count reaches zero. Existing 2-argument `(ctx, t