Distributed WebSocket framework backed by Redis. Handles connections, rooms, presence, pub/sub channels, versioned record sync, collections, structured commands, persistence, and automatic reconnection across multiple server instances.
npm install @prsm/realtime
importexpressfrom'express'import{createServer}from'node:http'import{RealtimeServer}from'@prsm/realtime'constapp=express()consthttpServer=createServer(app)constrealtime=newRealtimeServer({redis: {host: '127.0.0.1',port: 6379},authenticateConnection: (req)=>{return{user: 'amara',role: 'admin'}},})realtime.exposeChannel(/^notifications$/)realtime.exposeRecord(/^doc:.+$/)realtime.exposeWritableRecord(/^doc:.+$/)realtime.exposeCollection(/^inbox$/,()=>[{id: 'msg:1'},{id: 'msg:2'}])realtime.trackPresence(/^room:.+$/)realtime.exposeCommand('echo',async(ctx)=>({echoed: ctx.payload}))awaitrealtime.attach(httpServer,{port: 3000})awaitrealtime.writeChannel('notifications',{text: 'system online'})awaitrealtime.writeRecord('doc:welcome',{title: 'Welcome'})A single Redis instance coordinates connections across any number of server instances. Connections, room membership, presence, records, and collections are all visible cluster-wide.
import{RealtimeClient}from'@prsm/realtime/client'constclient=newRealtimeClient('ws://localhost:3000')awaitclient.connect()awaitclient.joinRoom('lobby')awaitclient.subscribeRecord('doc:welcome',(update)=>{console.log('record updated:',update.full??update.value)})awaitclient.subscribeChannel('notifications',(message)=>{console.log('notification:',message)})awaitclient.publishPresenceState('lobby',{state: {status: 'online'}})const{ echoed }=awaitclient.command('echo',{hello: 'world'})client.close()The client handles automatic reconnection with backoff, queued commands while disconnected, and re-subscription on reconnect.
@prsm/realtime/vue ships composables and renderless components that wrap the imperative client with reactive state and automatic lifecycle (subscribe on mount, unsubscribe on unmount, switch subscription when reactive keys change).
Create a RealtimeClient once, connect it, and make it available to descendant components. The recommended pattern is to do this at the root of the app:
<!-- App.vue -->
<script setup>import { RealtimeClient } from'@prsm/realtime/client'import { provideRealtime } from'@prsm/realtime/vue'constclient=newRealtimeClient('ws://localhost:3000')awaitclient.connect()provideRealtime(client)</script>
<template>
<router-view />
</template>provideRealtime(client) is a one-line helper that calls Vue's provide() with the right injection key. Every composable below this component automatically picks up the client via inject() - you don't have to thread client through props or pass it to each composable.
Inside any component descended from provideRealtime(client):
<script setup>import { useRoom, useRecord, useChannel, useCollection, usePresence } from'@prsm/realtime/vue'// auto-joins the room on mount, leaves on unmountconst { members, presence } =useRoom('lobby')// reactive value; updates flow in from the server; write() pushes backconst { value:doc, write } =useRecord('doc:welcome')// bounded message log; new messages append; oldest drop after 50const { messages } =useChannel('notifications', { max:50 })// resolves the collection's record IDs and keeps an items list in syncconst { items } =useCollection('inbox')// `me` is a ref<state> that publishes to the server on change;// `others` is the live map of other connections' statesconst { me, others } =usePresence('lobby', { initial: { status:'online' } })</script>
<template>
<p>{{ members.length }} in the room</p>
<inputv-model="me.status"placeholder="status..." />
<pre>{{ doc }}</pre>
</template>If you can't use the provide tree (tests, isolated components, a second connection), pass the client directly to any composable:
useRoom('lobby',{ client })useRecord('doc:1',{ client })The provide pattern is just sugar over this - pick whichever fits.
All composables mount cleanly: subscribing on onMounted, unsubscribing on onBeforeUnmount. Switching the reactive key (e.g. useRoom(activeRoom) where activeRoom is a ref) tears down the previous subscription and starts a new one automatically.
For the cases where you want the side effect to live in the template:
<RealtimeRoom name="lobby" v-slot="{ members }">
{{ members.length }} online
</RealtimeRoom>
<RealtimeRecord id="doc:welcome" v-slot="{ value, write }">
<input :value="value?.title" @input="write({ title: $event.target.value })" />
</RealtimeRecord>
<RealtimePresence room="lobby" :state="{ status, cursor }" />useConnection exposes the client's connection as reactive state, and RealtimeStatus is its renderless wrapper. These observe an existing client - they do not open or manage the connection. The RealtimeClient connects on its own (and reconnects on its own); you still create and connect it as shown in Setup. Use these only when you want to react to connection state in the UI.
<script setup>import { useConnection, useConnectionMetadata } from'@prsm/realtime/vue'const { status, isOnline, isReconnecting, latency, hasConnected, isStable } =useConnection()// local source of truth for this connection's metadata; set() writes through// to the server and the value is re-pushed automatically after a reconnectconst { metadata, set } =useConnectionMetadata({ initial: { name:'ada' } })</script>status is one of 'online', 'connecting', 'reconnecting', 'offline'. hasConnected becomes true after the first successful connect and stays true. isStable tracks isOnline but honors a grace window: when the connection drops it stays true for grace milliseconds (default 0), and a reconnect inside that window keeps it true so dependent UI never unmounts on a brief blip.
RealtimeStatus gates rendering on isStable through named slots, with grace as a prop. It reports connection state, it does not open the connection:
<RealtimeStatus :grace="2000">
<template #online="{ latency }">
<ChatRoom />
</template>
<template #reconnecting>
<p>reconnecting...</p>
</template>
<template #offline>
<p>offline</p>
</template>
</RealtimeStatus>Because the subscription composables queue commands while offline and replay them on reconnect, you don't need to gate them to keep subscriptions working - gate only when you genuinely want the children unmounted.
vue is an optional peer dependency. The /vue subpath only loads if you import from it.
Named groupings of connections. Used to scope presence and broadcasts.
awaitclient.joinRoom('lobby')awaitserver.broadcastRoom('lobby','announcement',{text: 'welcome'})awaitclient.leaveRoom('lobby')Server-to-client pub/sub. Multiple subscribers, fanned out across instances via Redis.
server.exposeChannel(/^chat:.+$/)awaitserver.writeChannel('chat:general',{author: 'amara',text: 'hi'})awaitclient.subscribeChannel('chat:general',(msg)=>{/* ... */})Versioned shared documents. Subscribers can choose full mode (every change ships the whole document) or patch mode (server diffs and ships JSON Patches).
server.exposeRecord(/^doc:.+$/)server.exposeWritableRecord(/^doc:.+$/)awaitserver.writeRecord('doc:42',{title: 'Hello',body: '...'})awaitclient.subscribeRecord('doc:42',(update)=>{console.log(update.full??update.patch)},{mode: 'patch'})awaitclient.writeRecord('doc:42',{title: 'Hello',body: '... updated'})Indexes over records, resolved per-connection at subscribe time.
server.exposeRecord(/^msg:.+$/)server.exposeCollection(/^inbox$/,(connection)=>[{id: 'msg:1'},{id: 'msg:2'},])awaitclient.subscribeCollection('inbox',{onDiff: ({ added, removed, changed })=>{/* ... */},})Per-room state broadcast to other members of the same room.
server.trackPresence(/^room:.+$/)awaitclient.joinRoom('room:design')awaitclient.publishPresenceState('room:design',{state: {cursor: {x: 100,y: 200}}})awaitclient.subscribePresence('room:design',(update)=>{// update.states is the full snapshot when first received// subsequent updates carry { connectionId, state } or { connectionId, removed }})Structured RPC. The server exposes named commands; the client invokes them and receives a response.
server.exposeCommand('order:create',async(ctx)=>{const{ user }=ctx.connection.authDataconstid=awaitdb.createOrder(user,ctx.payload)return{ id }})const{ id }=awaitclient.command('order:create',{items: [...]})Optional adapters keep record state durable across server restarts.
import{createSqliteAdapter}from'@prsm/realtime/sqlite'import{createPostgresAdapter}from'@prsm/realtime/postgres'newRealtimeServer({redis: {/* ... */},persistence: createPostgresAdapter({connectionString: 'postgres://...'}),})newRealtimeServer({authenticateConnection: async(req)=>{consturl=newURL(req.url,'http://x')consttoken=url.searchParams.get('token')constuser=awaitverifyToken(token)if(!user)thrownewError('unauthorized')return{userId: user.id,role: user.role}},})// expose guards:realtime.exposeChannel(/^chat:.+$/,(channel,connection)=>{returnconnection.authData.role==='member'})Pass a @prsm/trace tracer to the server and every command, record write, and channel publish becomes a span in the active trace.
import{createTracer}from'@prsm/trace'consttracer=createTracer({service: 'realtime-api'})newRealtimeServer({redis: {/* ... */}, tracer })make up # start Redis and Postgres
make test # run tests
make down # stop containers
Redis must be running on localhost:6379 for tests. Postgres is only needed for persistence adapter tests.