@type-r/live-websocket

Production-oriented WebSocket live updates plugin for Type-R collections.

This package is intentionally separate from @type-r/endpoints. The core endpoints package contains only the basic WebSocket transport. @type-r/live-websocket is a plugin-level package with peer dependencies on Type-R packages, so an application uses the same @type-r/models and @type-r/endpoints instances that it already has installed.

This package extends the basic @type-r/endpoints WebSocket transport with:

• subscribe acknowledgement;
• one WebSocket connection shared by multiple collections;
• reconnect and resubscribe;
• heartbeat;
• per-collection version cursors;
• since replay after reconnect;
• batch replay events;
• gap detection and replay requests;
• resync_required / replay_failed handling;
• structured protocol errors;
• graceful server shutdown handling;
• REST snapshot coordination.

Installation

yarn add @type-r/live-websocket

Peer dependencies expected in the host application:

yarn add @type-r/models @type-r/endpoints

Client endpoint

import { auto, CollectionConstructor, define, Record } from '@type-r/models'
import { restfulWebsocketIO } from '@type-r/live-websocket'

const apiRoot = 'http://localhost:3000'
const liveUrl = 'ws://localhost:3000/live'

function liveEndpoint(restUrl: string, channel: string) {
    return restfulWebsocketIO(restUrl, liveUrl, {
        protocolVersion: 1,

        subscribeMessage: () => ({
            type: 'subscribe',
            channel
        }),

        unsubscribeMessage: () => ({
            type: 'unsubscribe',
            channel
        }),

        subscribeAck: message =>
            message.type === 'subscribed' &&
            message.channel === channel,

        ackVersion: ack => ack.version,
        sinceParam: 'since',
        version: event => event.version,
        enforceEventOrder: true,

        replayRequestMessage: (collection, from, to) => ({
            type: 'replay_request',
            channel,
            from,
            to
        }),

        replayFailed: message => message.type === 'replay_failed',
        resyncRequired: message => message.type === 'resync_required',

        snapshotParam: 'at',
        snapshotVersion: ack => ack.version,
        snapshotData: json => json.items,

        match: message =>
            message.channel === channel ||
            message.type === 'server_shutdown' ||
            message.type === 'pong',

        errorMessage: message =>
            message.type === 'error'
                ? { code: message.code, message: message.message }
                : undefined,

        serverShutdown: message =>
            message.type === 'server_shutdown'
                ? message.retry_after
                : undefined,

        heartbeat: {
            interval: 30000,
            timeout: 10000,
            ping: { type: 'ping' },
            pong: message => message.type === 'pong'
        },

        reconnect: {
            attempts: Infinity,
            minDelay: 1000,
            maxDelay: 30000,
            factor: 2
        }
    })
}

@define
class User extends Record {
    static Collection: CollectionConstructor<User>
    static endpoint = liveEndpoint(`${apiRoot}/api/users`, 'users')

    @auto name: string
}

Snapshot ordering

collection.fetch({ liveUpdates: true }) performs these steps:

1. Open or reuse the shared WebSocket connection.
2. Send subscribe.
3. Wait for subscribed acknowledgement.
4. Run REST GET with ?at=<ack.version> when snapshotParam and snapshotVersion are configured.
5. Apply later WebSocket events.

This avoids the usual race where an event can happen between REST fetch and WebSocket subscribe.

Reconnect replay

When reconnecting, the plugin sends the last applied version:

{
    "protocol": 1,
    "type": "subscribe",
    "channel": "users",
    "since": 42
}

The server should replay events newer than since:

{
    "type": "batch",
    "channel": "users",
    "events": [
        {
            "type": "updated",
            "version": 43,
            "payload": { "id": "1", "name": "Ann" }
        }
    ]
}

Gap replay

With enforceEventOrder: true, if the client has version 42 and receives event 44, it does not apply 44 immediately. It sends:

{
    "type": "replay_request",
    "channel": "users",
    "from": 43,
    "to": 43
}

The server responds with a batch containing the missing range. If it cannot, it sends:

{
    "type": "replay_failed",
    "channel": "users",
    "reason": "gap_not_available"
}

The plugin then calls resync(). By default it calls collection.fetch({ liveUpdates: true }).

Protocol errors

Structured errors are converted to ProtocolError:

{
    "type": "error",
    "channel": "users",
    "code": "forbidden",
    "message": "Forbidden"
}

users.on('error', (collection, error) => {
    console.error(error.code, error.message)
})

Graceful shutdown

Before deploy or process shutdown, server can send:

{
    "type": "server_shutdown",
    "retry_after": 1000
}

The client closes the socket and reconnects after at least retry_after milliseconds.

REST production headers

@type-r/endpoints REST methods support:

• idempotencyKey -> Idempotency-Key;
• expectedVersion -> If-Match.

await user.save({
    idempotencyKey: crypto.randomUUID(),
    expectedVersion: user._version
} as any)

Server example

See examples/python-backend.md for a focused FastAPI backend implementation.

See examples/client-server.md for a full client/server example with Russian comments.