Realtime Sync

Realtime Sync

Overview

Architects high-concurrency, sub-50ms latency synchronization between distributed clients and servers. Covers WebTransport (HTTP/3) bidirectional streaming, transactional outbox patterns for database-to-sync consistency, CRDTs for collaborative editing, and AI token stream orchestration.

Core principles: the database is the source of truth (real-time channels notify, not persist), CRDTs eliminate locking for concurrent edits, and backpressure management prevents UI jitter from high-frequency streams.

When to use: Real-time collaborative UIs, pub/sub messaging, WebTransport/WebSocket connections, live AI token streams, presence tracking, conflict resolution with CRDTs, multiplayer applications.

When NOT to use: Batch processing pipelines, static content delivery, request-response APIs without real-time requirements, offline-only applications.

Quick Reference

Common Mistakes

Delegation

If the local-first skill is available, delegate local-first architecture decisions, sync engine comparison, and client storage selection to it.

If the electricsql skill is available, delegate ElectricSQL shape-based Postgres sync patterns to it.

• Explore transport protocol options and latency benchmarks: Use Explore agent to compare WebTransport, SSE, and WebSocket tradeoffs
• Implement transactional outbox pattern with CDC pipeline: Use Task agent to set up database triggers, outbox table, and background worker
• Design real-time architecture for collaborative editing: Use Plan agent to map CRDT strategy, presence management, and conflict resolution flow
• Build multi-stream AI orchestration component: Use Task agent to implement parallel token stream rendering with backpressure
• Audit real-time connection lifecycle management: Use Review agent to check for stream leaks, missing cleanup, and reconnection handling

References

• WebTransport streaming and worker patterns
• Pub/sub messaging and guaranteed delivery
• CRDTs and collaborative editing
• AI stream orchestration patterns
• Collaborative undo and redo patterns

AI Stream Orchestration

Handling live LLM token streams in real-time UIs requires careful rendering strategies to prevent jitter, dropped frames, and memory pressure.

Token-by-Token Rendering

Subscribe to a real-time channel and batch token updates using requestAnimationFrame:

import { useState, useEffect, useCallback } from 'react';

function LiveAIResponse({ channelId }: { channelId: string }) {
  const [tokens, setTokens] = useState('');

  useEffect(() => {
    const channel = ably.channels.get(channelId);
    let buffer = '';
    let rafId: number;

    function flush() {
      if (buffer) {
        setTokens((prev) => prev + buffer);
        buffer = '';
      }
      rafId = requestAnimationFrame(flush);
    }

    rafId = requestAnimationFrame(flush);

    channel.subscribe('token', (msg) => {
      buffer += msg.data;
    });

    return () => {
      cancelAnimationFrame(rafId);
      channel.unsubscribe();
    };
  }, [channelId]);

  return <div className="whitespace-pre-wrap">{tokens}</div>;
}

The buffer collects tokens between animation frames, then flushes them in a single state update. This prevents one re-render per token.

useTransition for Non-Blocking Updates

For streams that update complex UI (syntax highlighting, markdown rendering), defer non-urgent updates:

import { useState, useTransition, useEffect } from 'react';

function AICodeResponse({ channelId }: { channelId: string }) {
  const [rawTokens, setRawTokens] = useState('');
  const [renderedContent, setRenderedContent] = useState('');
  const [isPending, startTransition] = useTransition();

  useEffect(() => {
    const channel = ably.channels.get(channelId);

    channel.subscribe('token', (msg) => {
      setRawTokens((prev) => {
        const next = prev + msg.data;
        startTransition(() => {
          setRenderedContent(highlightSyntax(next));
        });
        return next;
      });
    });

    return () => channel.unsubscribe();
  }, [channelId]);

  return (
    <div>
      <pre>{renderedContent}</pre>
      {isPending ? (
        <span className="text-muted-foreground">Processing...</span>
      ) : null}
    </div>
  );
}

startTransition marks the syntax highlighting as non-urgent, allowing the raw token accumulation to proceed without blocking.

Backpressure Handling

When token production outpaces rendering, apply backpressure:

const MAX_BUFFER_SIZE = 10000;

function createTokenBuffer(onFlush: (tokens: string) => void) {
  let buffer = '';
  let rafId: number;

  function flush() {
    if (buffer) {
      onFlush(buffer);
      buffer = '';
    }
    rafId = requestAnimationFrame(flush);
  }

  rafId = requestAnimationFrame(flush);

  return {
    push(token: string) {
      buffer += token;
      if (buffer.length > MAX_BUFFER_SIZE) {
        cancelAnimationFrame(rafId);
        onFlush(buffer);
        buffer = '';
        rafId = requestAnimationFrame(flush);
      }
    },
    destroy() {
      cancelAnimationFrame(rafId);
    },
  };
}

When the buffer exceeds the threshold, force an immediate flush to prevent memory buildup.

Multi-Stream Orchestration

Handle multiple concurrent AI responses (e.g., parallel tool calls):

import { useState, useEffect } from 'react';

type StreamState = {
  id: string;
  tokens: string;
  status: 'streaming' | 'complete' | 'error';
};

function MultiStreamView({ streamIds }: { streamIds: string[] }) {
  const [streams, setStreams] = useState<Map<string, StreamState>>(new Map());

  useEffect(() => {
    const channels = streamIds.map((id) => {
      const channel = ably.channels.get(`ai-stream-${id}`);

      channel.subscribe('token', (msg) => {
        setStreams((prev) => {
          const next = new Map(prev);
          const current = next.get(id) ?? {
            id,
            tokens: '',
            status: 'streaming' as const,
          };
          next.set(id, { ...current, tokens: current.tokens + msg.data });
          return next;
        });
      });

      channel.subscribe('done', () => {
        setStreams((prev) => {
          const next = new Map(prev);
          const current = next.get(id);
          if (current) next.set(id, { ...current, status: 'complete' });
          return next;
        });
      });

      return channel;
    });

    return () => channels.forEach((ch) => ch.unsubscribe());
  }, [streamIds]);

  return (
    <div className="space-y-4">
      {Array.from(streams.values()).map((stream) => (
        <div key={stream.id}>
          <pre>{stream.tokens}</pre>
          {stream.status === 'streaming' ? <span>Streaming...</span> : null}
        </div>
      ))}
    </div>
  );
}

Stream Lifecycle

Performance Guidelines

CRDTs and Collaborative Editing

Conflict-free Replicated Data Types (CRDTs) allow multiple users to edit the same data simultaneously without a central lock. All changes merge deterministically regardless of arrival order.

Engine Selection

Both engines produce "deltas" (small change descriptions) that can be sent over any transport layer.

Yjs Integration

Basic Document Setup

import * as Y from 'yjs';

const ydoc = new Y.Doc();
const ytext = ydoc.getText('editor');

ytext.insert(0, 'Hello, world!');

Syncing Yjs Over Pub/Sub

Bridge Yjs updates over a pub/sub channel for global persistence:

import * as Y from 'yjs';

const ydoc = new Y.Doc();
const channel = ably.channels.get('doc-123');

ydoc.on('update', (update: Uint8Array) => {
  channel.publish('yjs-update', update);
});

channel.subscribe('yjs-update', (msg) => {
  Y.applyUpdate(ydoc, new Uint8Array(msg.data));
});

Initial State Sync

When a new client joins, it needs the full document state:

async function syncInitialState(ydoc: Y.Doc, channel: Channel) {
  const stateVector = Y.encodeStateVector(ydoc);
  const response = await channel.history({ limit: 1 });

  if (response.items.length > 0) {
    const snapshot = response.items[0].data;
    Y.applyUpdate(ydoc, new Uint8Array(snapshot));
  }
}

Automerge Integration

For JSON-like collaborative state:

import * as Automerge from '@automerge/automerge';

let doc = Automerge.init<{ items: string[] }>();
doc = Automerge.change(doc, (d) => {
  d.items.push('New item');
});

const changes = Automerge.getAllChanges(doc);

Presence: Collaborative Cursors

Show where other users are working in the document.

Cursor Position Tracking

import { Awareness } from 'y-protocols/awareness';

const awareness = new Awareness(ydoc);

awareness.setLocalStateField('cursor', {
  anchor: { index: 42 },
  head: { index: 42 },
  user: { name: 'Alice', color: '#ff0000' },
});

awareness.on('change', () => {
  const states = awareness.getStates();
  states.forEach((state, clientId) => {
    if (clientId !== ydoc.clientID && state.cursor) {
      renderRemoteCursor(clientId, state.cursor);
    }
  });
});

Cursor Rendering Best Practices

Common CRDT Pitfalls

Document Bloat

CRDTs store edit history. Unbounded documents grow indefinitely.

Fix: Use snapshotting for fields that do not need history. For example, a window width preference should use Last-Writer-Wins (LWW) rather than a full CRDT:

const ymap = ydoc.getMap('preferences');
ymap.set('window_width', 1200);

Clock Drift

CRDTs do not require synchronized clocks, but massive drift between clients can cause text interleaving (characters appearing out of order in collaborative text editing).

Fix: Use logical clocks (Lamport timestamps) rather than wall clocks. Yjs handles this internally.

Security

CRDT updates are opaque binary data. A malicious client can send arbitrary state.

Fix: Validate the final document state server-side in an "auditor" service. Do not trust client-produced CRDT updates for security-sensitive data without validation.

Undo/Redo

CRDT undo is more complex than single-user undo because other users' changes may interleave.

Fix: Use Yjs UndoManager which tracks only the local user's operations:

const LOCAL_ORIGIN = 'local-user-input';

const undoManager = new Y.UndoManager(ytext, {
  trackedOrigins: new Set([LOCAL_ORIGIN]),
});

ydoc.transact(() => {
  ytext.insert(0, 'Hello');
}, LOCAL_ORIGIN);

undoManager.undo();
undoManager.redo();

Architecture Summary

Client A ──(delta)──> Pub/Sub Channel ──(delta)──> Client B
                          │
                          ├──(delta)──> Server Persistence
                          │
                          └──(delta)──> Client C

All clients converge to the same state regardless of message arrival order. The server persists snapshots periodically for new client onboarding.

Pub/Sub Messaging and Guaranteed Delivery

Pub/sub infrastructure provides the orchestration layer between transport and application state. Guaranteed delivery requires sequence tracking, transactional consistency, and reconnection recovery.

Sequence Tracking

Every message published to a channel carries a sequence ID. This enables detection of gaps (missed messages) and ordering guarantees.

Rewind on Reconnect

When a client reconnects after a network interruption, use the rewind parameter to fetch missed messages:

const channel = ably.channels.get('orders', {
  params: { rewind: '1m' },
});

channel.subscribe('update', (msg) => {
  applyUpdate(msg.data);
});

The rewind: '1m' parameter requests the last minute of message history, filling any gaps from the disconnection period.

Sequence Validation

On the client, track the last received sequence and detect gaps:

let lastSequence = 0;

function onMessage(msg: { sequence: number; data: unknown }) {
  if (msg.sequence > lastSequence + 1) {
    requestRewind(lastSequence + 1, msg.sequence - 1);
  }
  lastSequence = msg.sequence;
  applyUpdate(msg.data);
}

Transactional Outbox Pattern

Ensures database state and real-time notifications never drift. The outbox guarantees that a message is only published if the database transaction succeeds.

Database Schema

CREATE TABLE realtime_outbox (
  id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
  channel text NOT NULL,
  event_type text NOT NULL,
  payload jsonb NOT NULL,
  sequence_id uuid DEFAULT gen_random_uuid(),
  created_at timestamptz DEFAULT now(),
  published_at timestamptz
);

Transaction Pattern

Write the business data and outbox entry in a single transaction:

BEGIN;
  UPDATE orders SET status = 'shipped' WHERE id = 456;

  INSERT INTO realtime_outbox (channel, event_type, payload)
  VALUES (
    'order_updates',
    'status_changed',
    '{"order_id": 456, "status": "shipped"}'::jsonb
  );
COMMIT;

CDC Worker

A background process (Change Data Capture) reads unpublished outbox entries and pushes them to the real-time channel:

async function processOutbox() {
  const pending = await db.query(
    'SELECT * FROM realtime_outbox WHERE published_at IS NULL ORDER BY id LIMIT 100',
  );

  for (const entry of pending.rows) {
    await channel.publish(entry.event_type, entry.payload);
    await db.query(
      'UPDATE realtime_outbox SET published_at = now() WHERE id = $1',
      [entry.id],
    );
  }
}

Poll this function on a short interval (100-500ms) or trigger it via a database notification (LISTEN/NOTIFY).

Presence Management

Track which users are currently active in a channel:

const channel = ably.channels.get('document-123');

await channel.presence.enter({ cursor: { x: 0, y: 0 } });

channel.presence.subscribe('enter', (member) => {
  addUserCursor(member.clientId, member.data.cursor);
});

channel.presence.subscribe('leave', (member) => {
  removeUserCursor(member.clientId);
});

Zombie Prevention

Presence relies on heartbeats. When a client crashes without sending a "leave" message, it becomes a zombie presence entry.

Epidemic broadcast protocol: Peers periodically exchange presence state. If a peer reports a user absent that another peer still shows as present, the conflict resolves by removing the stale entry.

Heartbeat tuning: Mobile devices need longer heartbeat intervals to conserve battery. Desktop clients can use shorter intervals for faster detection:

Channel Multiplexing

Ably SDKs multiplex all channel traffic over a single transport connection. Subscribe to multiple channels without extra connections:

const orders = ably.channels.get('orders');
const notifications = ably.channels.get('notifications');
const presence = ably.channels.get('presence');

orders.subscribe('update', (msg) => handleOrder(msg.data));
notifications.subscribe('alert', (msg) => handleNotification(msg.data));
presence.subscribe('enter', (msg) => handlePresenceEvent(msg.data));

All subscriptions share one WebSocket connection, reducing overhead on mobile devices.

Quality Checklist

Yjs UndoManager

The UndoManager tracks changes to Yjs shared types and supports local-only undo — each user undoes their own changes without affecting other users' edits.

Basic Setup

import * as Y from 'yjs';

const doc = new Y.Doc();
const yText = doc.getText('content');
const yMap = doc.getMap('metadata');

const undoManager = new Y.UndoManager([yText, yMap], {
  captureTimeout: 500,
  trackedOrigins: new Set(['user-edit']),
});

Tracked Transactions

Only changes made within a doc.transact() call with a tracked origin are recorded in the undo stack.

doc.transact(() => {
  yText.insert(0, 'Hello ');
  yMap.set('lastEdited', Date.now());
}, 'user-edit');

Changes from remote sync (different origin) are not tracked, so undoing only reverses local edits.

Undo, Redo, and Stop Capturing

undoManager.undo();
undoManager.redo();

undoManager.stopCapturing();

stopCapturing() forces the next change to start a new undo stack item rather than merging with the previous one. Use it to create explicit undo boundaries.

yText.insert(0, 'First edit');
undoManager.stopCapturing();
yText.insert(11, ' Second edit');

undoManager.undo();

Stack Item Metadata

Attach metadata (cursor position, scroll state) to undo stack items for restoring UI state on undo/redo.

undoManager.on(
  'stack-item-added',
  (event: { stackItem: Y.UndoManagerStackItem; type: 'undo' | 'redo' }) => {
    event.stackItem.meta.set('cursor', getCursorPosition());
    event.stackItem.meta.set('scroll', getScrollPosition());
  },
);

undoManager.on(
  'stack-item-popped',
  (event: { stackItem: Y.UndoManagerStackItem; type: 'undo' | 'redo' }) => {
    const cursor = event.stackItem.meta.get('cursor') as number | undefined;
    const scroll = event.stackItem.meta.get('scroll') as number | undefined;

    if (cursor !== undefined) setCursorPosition(cursor);
    if (scroll !== undefined) setScrollPosition(scroll);
  },
);

Delete Filter

Prevent certain deletions from being tracked. Useful for protecting structural elements.

const undoManager = new Y.UndoManager([yText], {
  trackedOrigins: new Set(['user-edit']),
  deleteFilter: (item: Y.Item) => {
    if (
      item.parent instanceof Y.XmlElement &&
      item.parent.nodeName === 'TABLE'
    ) {
      return false;
    }
    return true;
  },
});

Return false to exclude a deletion from the undo stack.

Local Undo vs Global Undo

Yjs UndoManager implements local undo by default via trackedOrigins. For global undo, track all origins:

const globalUndoManager = new Y.UndoManager([yText], {
  trackedOrigins: new Set([null]),
});

Setting null as a tracked origin captures changes from doc.transact() calls with no explicit origin.

Automerge Undo

automerge-repo-undo-redo

import { AutomergeRepoUndoRedo } from 'automerge-repo-undo-redo';

const undoRedo = new AutomergeRepoUndoRedo(repo);

Making Tracked Changes

import { type DocHandle } from '@automerge/automerge-repo';

const handle: DocHandle<TodoDoc> = repo.find(docUrl);

undoRedo.change(handle, (doc) => {
  doc.todos.push({ id: '1', title: 'New todo', completed: false });
});

Undo and Redo

undoRedo.undo(handle);
undoRedo.redo(handle);

Transactions for Batch Undo

Group multiple changes into a single undoable operation.

undoRedo.transaction(handle, (tx) => {
  tx.change((doc) => {
    doc.title = 'Updated title';
  });
  tx.change((doc) => {
    doc.lastModified = Date.now();
  });
});

A single undo() call reverses both changes.

Scoped Undo Stacks

Create separate undo stacks for isolated UI contexts like modals or dialogs.

Yjs Scoped Origin

const MODAL_ORIGIN = 'modal-edit';
const MAIN_ORIGIN = 'main-edit';

const mainUndoManager = new Y.UndoManager([yText], {
  trackedOrigins: new Set([MAIN_ORIGIN]),
});

const modalUndoManager = new Y.UndoManager([yText], {
  trackedOrigins: new Set([MODAL_ORIGIN]),
});

doc.transact(() => {
  yText.insert(0, 'modal change');
}, MODAL_ORIGIN);

modalUndoManager.undo();

Automerge Scoped Stacks

const mainUndoRedo = new AutomergeRepoUndoRedo(repo);
const modalUndoRedo = new AutomergeRepoUndoRedo(repo);

Each AutomergeRepoUndoRedo instance maintains its own stack. Changes made through modalUndoRedo.change() are only undone by modalUndoRedo.undo().

Multi-Document Undo

Coordinate undo across multiple documents with a unified manager.

Yjs Multi-Document

class MultiDocUndoManager {
  private managers = new Map<string, Y.UndoManager>();

  register(id: string, types: Y.AbstractType<unknown>[], origin: string): void {
    this.managers.set(
      id,
      new Y.UndoManager(types, {
        trackedOrigins: new Set([origin]),
      }),
    );
  }

  undo(id: string): void {
    this.managers.get(id)?.undo();
  }

  redo(id: string): void {
    this.managers.get(id)?.redo();
  }

  undoAll(): void {
    for (const manager of this.managers.values()) {
      manager.undo();
    }
  }

  destroy(): void {
    for (const manager of this.managers.values()) {
      manager.destroy();
    }
    this.managers.clear();
  }
}

Automerge Multi-Document

class UndoRedoManager {
  private undoRedo: AutomergeRepoUndoRedo;
  private handles = new Map<string, DocHandle<unknown>>();

  constructor(repo: Repo) {
    this.undoRedo = new AutomergeRepoUndoRedo(repo);
  }

  register(id: string, handle: DocHandle<unknown>): void {
    this.handles.set(id, handle);
  }

  change<T>(id: string, changeFn: (doc: T) => void): void {
    const handle = this.handles.get(id) as DocHandle<T> | undefined;
    if (handle) {
      this.undoRedo.change(handle, changeFn);
    }
  }

  undo(id: string): void {
    const handle = this.handles.get(id);
    if (handle) {
      this.undoRedo.undo(handle);
    }
  }

  redo(id: string): void {
    const handle = this.handles.get(id);
    if (handle) {
      this.undoRedo.redo(handle);
    }
  }
}

WebTransport Streaming

WebTransport is the HTTP/3-powered replacement for WebSockets. It supports multiple streams within a single connection and both reliable and unreliable data delivery.

Stream Types

Basic Client Connection

const transport = new WebTransport('https://example.com/api/realtime');
await transport.ready;

const stream = await transport.createBidirectionalStream();
const writer = stream.writable.getWriter();
const reader = stream.readable.getReader();

await writer.write(new TextEncoder().encode(JSON.stringify({ type: 'join' })));

const { value } = await reader.read();
const response = JSON.parse(new TextDecoder().decode(value));

Datagram Channel (Unreliable/Fast)

For high-frequency updates where losing a frame is better than waiting for retransmission:

const transport = new WebTransport(serverUrl);
await transport.ready;

const datagramWriter = transport.datagrams.writable.getWriter();

function sendCursorPosition(x: number, y: number) {
  const payload = new Float32Array([x, y]);
  datagramWriter.write(new Uint8Array(payload.buffer));
}

async function readDatagrams() {
  const reader = transport.datagrams.readable.getReader();
  while (true) {
    const { value, done } = await reader.read();
    if (done) break;
    const coords = new Float32Array(value.buffer);
    updateRemoteCursor(coords[0], coords[1]);
  }
}

Web Worker Pattern

Handle WebTransport inside a Web Worker to prevent blocking the UI thread:

// transport.worker.ts
const transport = new WebTransport(url);
await transport.ready;

const stream = await transport.createUnidirectionalStream();
const writer = stream.getWriter();

self.onmessage = ({ data }) => {
  writer.write(new TextEncoder().encode(JSON.stringify(data)));
};

// main thread
const worker = new Worker(new URL('./transport.worker.ts', import.meta.url), {
  type: 'module',
});

worker.postMessage({ type: 'cursor', x: 100, y: 200 });

Connection Statistics

Monitor connection health using getStats():

const stats = await transport.getStats();
console.log('Smoothed RTT:', stats.smoothedRtt);
console.log('Bytes sent:', stats.bytesSent);
console.log('Packets lost:', stats.packetsLost);

High packet loss indicates network saturation. Reduce send frequency or switch to reliable streams.

WebSocket Fallback

Enterprise firewalls may block UDP (HTTP/3). Implement a detection and fallback:

async function createTransport(url: string) {
  if ('WebTransport' in globalThis) {
    try {
      const transport = new WebTransport(url);
      await transport.ready;
      return { type: 'webtransport' as const, transport };
    } catch {
      // WebTransport blocked or failed
    }
  }

  const ws = new WebSocket(url.replace('https:', 'wss:'));
  return new Promise<{ type: 'websocket'; transport: WebSocket }>((resolve) => {
    ws.onopen = () => resolve({ type: 'websocket', transport: ws });
  });
}

Common Pitfalls

Stream Lifecycle

Always clean up streams when they are no longer needed:

async function cleanup(transport: WebTransport) {
  try {
    transport.close({ closeCode: 0, reason: 'Client disconnecting' });
  } catch {
    // Transport may already be closed
  }
  await transport.closed;
}