@riftway/ws-adapter

A strongly-typed WebSocket adapter for Riftway and Zustand stores with comprehensive type safety, automatic reconnection, message batching, and flexible codec support.

Features

• 🔒 Full Type Safety - No implicit any types, strongly-typed message handling
• 🔄 Automatic Reconnection - Exponential backoff with jitter to prevent thundering herd
• 📦 Message Batching - Optional batching for high-frequency scenarios
• 🔌 Pluggable Codecs - JSON by default, bring your own for binary/custom formats
• 🏪 Store Agnostic - Works with both Riftway and Zustand stores
• 🎯 Framework Agnostic - No React dependency, works anywhere
• 📊 Connection Status - Real-time status tracking with subscription support
• 🛡️ Error Recovery - Graceful error handling with comprehensive lifecycle hooks

Installation

npm install @riftway/ws-adapter
# or
pnpm add @riftway/ws-adapter
# or
yarn add @riftway/ws-adapter

Quick Start

With Zustand

import { create } from 'zustand';
import { createWsStoreAdapter } from '@riftway/ws-adapter';

// Define your action types
type ServerActions = {
  'chat/message': { id: string; text: string; user: string };
  'user/joined': { user: string };
  'user/left': { user: string };
};

type ClientActions = {
  'chat/send': { text: string };
  'user/ping': { timestamp: number };
};

// Create your Zustand store
type ChatState = { 
  messages: Array<{ id: string; text: string; user: string }>;
  users: string[];
  connected: boolean;
};

const useChat = create<ChatState>(() => ({
  messages: [],
  users: [],
  connected: false
}));

// Define message handlers
const handlers = {
  'chat/message': (state, payload) => ({
    messages: [...state.messages, payload]
  }),
  'user/joined': (state, payload) => ({
    users: [...state.users, payload.user]
  }),
  'user/left': (state, payload) => ({
    users: state.users.filter(u => u !== payload.user)
  })
};

// Create and connect the adapter
const adapter = createWsStoreAdapter(
  { getState: useChat.getState, setState: useChat.setState },
  handlers,
  {
    url: 'ws://localhost:8080/chat',
    reconnect: { attempts: 5, backoffMs: 1000, jitter: true },
    batch: { windowMs: 16, max: 100 }
  }
);

adapter.connect();

// Send messages
adapter.send('chat/send', { text: 'Hello, world!' });
adapter.send('user/ping', { timestamp: Date.now() });

// Monitor connection status
adapter.subscribeStatus((status) => {
  console.log('Connection status:', status);
  useChat.setState({ connected: status === 'open' });
});

With Riftway Store

import { createStore } from '@riftway/core';
import { createWsStoreAdapter } from '@riftway/ws-adapter';

// Create your Riftway store
const store = createStore<ChatState>({
  messages: [],
  users: [],
  connected: false
});

// Create adapter with Riftway store integration
const adapter = createWsStoreAdapter(
  {
    getState: store.getState,
    setState: (partial) => store.setState(
      typeof partial === 'function' 
        ? (prev) => ({ ...prev, ...partial(prev) })
        : (prev) => ({ ...prev, ...partial })
    )
  },
  handlers,
  options
);

Advanced Usage

Custom Codec

import { jsonCodec } from '@riftway/ws-adapter/utils/codec';

// Use the built-in JSON codec
const adapter = createWsStoreAdapter(store, handlers, {
  url: 'ws://localhost:8080',
  codec: jsonCodec<InboundMessage, OutboundMessage>()
});

// Or create a custom codec
const binaryCodec = {
  encode: (msg) => new TextEncoder().encode(JSON.stringify(msg)),
  decode: (data) => JSON.parse(new TextDecoder().decode(data))
};

Message Transformation

const adapter = createWsStoreAdapter(store, handlers, {
  url: 'ws://localhost:8080',
  transformIn: (msg) => {
    // Normalize message format
    return {
      ...msg,
      timestamp: msg.timestamp || Date.now()
    };
  }
});

Lifecycle Hooks

const adapter = createWsStoreAdapter(store, handlers, {
  url: 'ws://localhost:8080',
  onOpen: (event) => console.log('Connected'),
  onClose: (event) => console.log('Disconnected'),
  onError: (error) => console.error('WebSocket error:', error),
  onMessage: (msg) => console.log('Received:', msg),
  beforeApply: (msg, state) => console.log('Before applying:', msg),
  afterApply: (msg, state) => console.log('After applying:', msg)
});

Batching Configuration

// Time-based batching
const adapter = createWsStoreAdapter(store, handlers, {
  url: 'ws://localhost:8080',
  batch: { windowMs: 16 } // Batch messages for 16ms
});

// Count-based batching
const adapter = createWsStoreAdapter(store, handlers, {
  url: 'ws://localhost:8080',
  batch: { max: 50 } // Batch up to 50 messages
});

// Combined batching
const adapter = createWsStoreAdapter(store, handlers, {
  url: 'ws://localhost:8080',
  batch: { windowMs: 16, max: 50 } // Whichever comes first
});

// Disable batching
const adapter = createWsStoreAdapter(store, handlers, {
  url: 'ws://localhost:8080',
  batch: false
});

Reconnection Configuration

const adapter = createWsStoreAdapter(store, handlers, {
  url: 'ws://localhost:8080',
  reconnect: {
    attempts: 10,        // Maximum reconnection attempts
    backoffMs: 1000,     // Base delay between attempts
    jitter: true         // Add random variance to prevent thundering herd
  }
});

URL Parameters and Protocols

const adapter = createWsStoreAdapter(store, handlers, {
  url: () => `ws://localhost:8080/chat?token=${getAuthToken()}`, // Dynamic URL
  protocols: ['chat-protocol-v1'],
  params: { room: 'general', version: '1.0' } // Added to query string
});

API Reference

createWsStoreAdapter(store, handlers, options)

Creates a new WebSocket store adapter.

Parameters

• store: StoreApi<S> - Store instance with getState and setState methods
• handlers: Handlers<S, A, M> - Message handlers for each action type
• options: Options<S, A, ClientA, InMsg, OutMsg> - Configuration options

Returns

WsStoreAdapter<S, A, ClientA, MIn, MOut, InMsg, OutMsg> - The adapter instance

Adapter Methods

connect(): void

Start the WebSocket connection.

close(): void

Close the connection and stop reconnection attempts.

send<K>(action: K, payload: ClientA[K], meta?: MOut): void

Send a typed message to the server.

sendRaw(msg: OutMsg): void

Send a raw message envelope.

getStatus(): Status

Get the current connection status.

subscribeStatus(callback: (status: Status) => void): () => void

Subscribe to status changes. Returns an unsubscribe function.

Types

Status

type Status = 'idle' | 'connecting' | 'open' | 'closed' | 'reconnecting';

StoreApi<S>

type StoreApi<S> = {
  getState: () => S;
  setState: (partial: Partial<S> | ((s: S) => Partial<S>)) => void;
};

Handlers<S, A, M>

type Handlers<S, A, M> = {
  [K in keyof A]?: (state: S, payload: A[K], meta: M) => Partial<S> | void;
} & {
  __unknown__?: (state: S, msg: LooseEnvelope<M>) => Partial<S> | void;
};

Error Handling

The adapter provides comprehensive error handling:

• Connection errors trigger automatic reconnection
• Codec errors are passed to the onError hook
• Handler errors don't crash the adapter
• Malformed messages are gracefully ignored

const adapter = createWsStoreAdapter(store, handlers, {
  url: 'ws://localhost:8080',
  onError: (error) => {
    if (error instanceof Error) {
      console.error('Adapter error:', error.message);
    } else {
      console.error('WebSocket error:', error);
    }
  }
});

Performance Tips

1. Use batching for high-frequency messages to reduce store updates
2. Enable jitter in reconnection settings to prevent thundering herd
3. Implement __unknown__ handler to gracefully handle unexpected messages
4. Use message transformation to normalize data before processing
5. Monitor connection status to provide user feedback

TypeScript Support

The adapter is built with TypeScript and provides full type safety:

• Action names are validated at compile time
• Payload types are enforced for each action
• Handler return types are checked
• No implicit any types anywhere in the codebase

License

MIT