sync

sync() synchronizes state across the network in real-time. Pass it a transport, options, and boxed proxies from tosi() — local changes are throttled and sent as batched deltas, and inbound messages from other clients are applied to the local state automatically.

import { tosi, sync } from 'tosijs'

const { game } = tosi({
  game: { players: {}, ball: { x: 0, y: 0 } }
})

const ws = new WebSocket('wss://my-server.example/sync')

const { disconnect } = await sync(
  websocketTransport(ws),
  { throttleInterval: 50 },
  game
)

// Later, to disconnect:
disconnect()

Transport interface

sync() is transport-agnostic. You provide an object that satisfies SyncTransport:

interface SyncTransport {
  send(messages: SyncMessage[]): void
  onReceive(handler: (messages: SyncMessage[]) => void): void
  connect(): Promise<void> | void
  disconnect(): void
}

interface SyncMessage {
  path: string
  value: any
}

Messages are batched — send() receives an array of accumulated deltas flushed at the throttle interval, and onReceive() delivers batches from the server.

WebSocket transport helper

function websocketTransport(ws) {
  let handler = null
  let pingInterval = null

  return {
    connect() {
      return new Promise((resolve, reject) => {
        if (ws.readyState === WebSocket.OPEN) return resolve()
        ws.addEventListener('open', () => resolve(), { once: true })
        ws.addEventListener('error', reject, { once: true })
      })
    },
    send(messages) {
      if (ws.readyState === WebSocket.OPEN) {
        ws.send(JSON.stringify(messages))
      }
    },
    onReceive(h) {
      handler = h
      ws.addEventListener('message', (event) => {
        handler(JSON.parse(event.data))
      })
      // Keep alive: send empty batch periodically so the server
      // knows we're still here (see idleTimeout in sync-server.ts)
      pingInterval = setInterval(() => {
        if (ws.readyState === WebSocket.OPEN) ws.send('[]')
      }, 15000)
    },
    disconnect() {
      clearInterval(pingInterval)
      ws.close()
    },
  }
}

Firebase Realtime Database transport

For Firebase, implement SyncTransport using onValue for inbound and update for outbound. This is a sketch — adapt to your data model:

import { ref, onValue, update } from 'firebase/database'

function firebaseTransport(db, rootPath) {
  let handler = null
  let unsubscribe = null

  return {
    connect() {
      const dbRef = ref(db, rootPath)
      unsubscribe = onValue(dbRef, (snapshot) => {
        const data = snapshot.val()
        if (data && handler) {
          // Convert Firebase snapshot to SyncMessage[]
          const messages = Object.entries(data).map(
            ([path, value]) => ({ path, value })
          )
          handler(messages)
        }
      })
    },
    send(messages) {
      const updates = {}
      for (const msg of messages) {
        updates[`${rootPath}/${msg.path}`] = msg.value
      }
      update(ref(db), updates)
    },
    onReceive(h) { handler = h },
    disconnect() { if (unsubscribe) unsubscribe() },
  }
}

Server architecture

The transport carries SyncMessage[] arrays. The server decides:

For most realistic applications, use a custom socket server as the single source of truth. See examples/sync-server.ts for a minimal Bun WebSocket relay server.

API

sync(
  transport: SyncTransport,
  options: SyncOptions,
  ...proxies: (BoxedProxy | string)[]
): Promise<{ disconnect: () => void }>

SyncOptions:

The returned disconnect() removes all observers and calls transport.disconnect().