WebSocket Realtime Sync Engine

WebSocket Realtime Sync Engine

1. System Architecture & Prerequisites

This skill builds a resilient bidirectional WebSocket/SSE sync layer:

• Exponential backoff reconnection — delay = min(cap, base * 2^attempt) with jitter so a fleet of clients does not thundering-herd a restarting hub.
• Heartbeat keep-alive — periodic {type:"ping"} frames with a lastPong watchdog that force-reconnects a dead peer before the socket rots silently.
• Delta-based state reconciliation — deltas are JSON-Patch-like ops (set/merge/replace/delete on a path) applied to a local state tree; on reconnect a version vector {device_id, seq} drives a resync_request so missed deltas are replayed exactly once.

Prerequisites:

• Node.js 18+ (ESM). Client WebSocket is global in Node 22+; for older Node install the dependency-light ws package (npm i ws) and the module fallback loader picks it up automatically.
• The demo server uses ws WebSocketServer; without it, main() prints a graceful hint instead of crashing.
• No build step, no TypeScript compiler required — the file is plain ESM that runs with node realtime-client.js.

2. Input/Output Data Contracts

RealtimeClient constructor options

{
  "url": "ws://host:port/path",
  "WebSocketImpl": "optional WebSocket constructor override",
  "baseDelay": 1000,
  "maxDelay": 30000,
  "jitter": 0.3,
  "maxAttempts": 1000000000,
  "heartbeatIntervalMs": 15000,
  "heartbeatTimeoutMs": 5000,
  "deviceId": "dev-xxxxxxxx"
}

Delta operations (wire + local)

Wire envelope:

{ "type": "delta", "op": "merge", "path": ["users","1"], "value": { "online": true }, "seq": 12, "device_id": "dev-abc" }

Control frames

• {type:"ping", device_id, ts} / {type:"pong", device_id, ts}
• {type:"hello", device_id, version:{seq}} on open
• {type:"resync_request", request_id, version:{device_id, seq}, last_server_seq}
• {type:"resync_ack", request_id, last_server_seq, applied}
• {type:"batch", deltas:[...]} — replay of missed deltas

Output/state

RealtimeClient.getState() returns the locally reconciled state tree, and resync() resolves to {last_server_seq, applied} once the server has replayed the missed window.

3. Production Reference Implementation

Save as realtime-client.js. Dependency-light ESM, guards the WebSocket implementation, exports helpers suitable for unit testing.

// realtime-client.js — resilient bidirectional sync client.
// Node 18+ ESM. Prefers the global WebSocket (Node 22+); falls back to 'ws'.
import { createRequire } from "module";
import { pathToFileURL } from "url";

const nodeRequire = createRequire(import.meta.url);

/** Resolve a WebSocket implementation, trying global first, then 'ws'. */
function resolveWebSocket() {
  const wsGlobal = globalThis.WebSocket;
  if (typeof wsGlobal === "function" && wsGlobal) {
    return wsGlobal;
  }
  try {
    const mod = nodeRequire("ws");
    return mod.WebSocket || mod;
  } catch {
    throw new Error(
      "No WebSocket implementation available: run Node >= 22 or `npm i ws`."
    );
  }
}

/**
 * Exponential backoff with full jitter.
 * delay = min(cap, base * 2**attempt) * (1 + jitter * [-1..1])
 */
function backoffDelay(baseDelay, attempt, maxDelay, jitter) {
  const exponential = Math.min(baseDelay * 2 ** attempt, maxDelay);
  const wiggle = 1 + (Math.random() * 2 - 1) * jitter;
  return Math.max(25, Math.round(exponential * wiggle));
}

/* ---------- delta primitives (pure functions, unit-testable) ---------- */

function setAtPath(state, path, value) {
  if (!Array.isArray(path) || path.length === 0) {
    return { ...(state ?? {}), ...value };
  }
  const root = { ...(state ?? {}) };
  let node = root;
  for (let i = 0; i < path.length - 1; i += 1) {
    const key = path[i];
    if (node[key] === null || typeof node[key] !== "object") {
      node[key] = Array.isArray(node[key]) ? [] : {};
    }
    node[key] = { ...node[key] };
    node = node[key];
  }
  node[path[path.length - 1]] = value;
  return root;
}

function mergeInto(target, value) {
  if (target && value && typeof target === "object" && typeof value === "object") {
    for (const [key, val] of Object.entries(value)) {
      if (
        val && typeof val === "object" && !Array.isArray(val) &&
        target[key] && typeof target[key] === "object" && !Array.isArray(target[key])
      ) {
        mergeInto(target[key], val);
      } else {
        target[key] = val;
      }
    }
  }
  return target;
}

function deleteAtPath(state, path) {
  if (!Array.isArray(path) || path.length === 0) {
    return state;
  }
  const root = { ...(state ?? {}) };
  let node = root;
  for (let i = 0; i < path.length - 1; i += 1) {
    const key = path[i];
    if (node[key] === null || typeof node[key] !== "object") {
      return state; // nothing to delete
    }
    node[key] = { ...node[key] };
    node = node[key];
  }
  delete node[path[path.length - 1]];
  return root;
}

/** Apply a JSON-Patch-like delta to the state tree and return the new state. */
function applyDelta(state, delta) {
  const { type, path = [], value } = delta;
  switch (type) {
    case "set":
    case "replace":
      return setAtPath(state, path, value);
    case "merge": {
      const root = { ...(state ?? {}) };
      let node = root;
      let key = null;
      if (Array.isArray(path) && path.length > 0) {
        for (let i = 0; i < path.length - 1; i += 1) {
          const k = path[i];
          if (node[k] === null || typeof node[k] !== "object") {
            node[k] = {};
          }
          node[k] = { ...node[k] };
          node = node[k];
        }
        key = path[path.length - 1];
        if (node[key] === null || typeof node[key] !== "object") {
          node[key] = {};
        } else {
          node[key] = { ...node[key] };
        }
        mergeInto(node[key], value);
      } else {
        mergeInto(root, value);
      }
      return root;
    }
    case "delete":
      return deleteAtPath(state, path);
    default:
      throw new Error(`Unknown delta type: ${type}`);
  }
}

/* ---------------------------- the client ---------------------------- */

export class RealtimeClient {
  constructor(options = {}) {
    this.url = options.url;
    this.WebSocket = options.WebSocketImpl || resolveWebSocket();
    this.baseDelay = options.baseDelay ?? 1000;
    this.maxDelay = options.maxDelay ?? 30000;
    this.jitter = options.jitter ?? 0.3;
    this.maxAttempts = options.maxAttempts ?? Infinity;
    this.heartbeatIntervalMs = options.heartbeatIntervalMs ?? 15000;
    this.heartbeatTimeoutMs = options.heartbeatTimeoutMs ?? 5000;
    this.deviceId =
      options.deviceId ||
      `dev-${Math.random().toString(36).slice(2, 10)}`;

    this.state = {};
    this.seq = 0;            // outbound delta counter (this device's vector)
    this.lastServerSeq = 0;  // highest server delta we have applied
    this.attempts = 0;
    this.connected = false;
    this.manualClose = false;
    this.socket = null;
    this.pending = [];
    this.lastPong = Date.now();
    this._resolveOpen = null;
    this._rejectOpen = null;
    this._resyncWaiters = [];
    this._timers = { heartbeat: null, watchdog: null, reconnect: null };
    this._events = new Map();
  }

  on(event, handler) {
    if (!this._events.has(event)) this._events.set(event, new Set());
    this._events.get(event).add(handler);
    return () => this._events.get(event)?.delete(handler);
  }

  emit(event, payload) {
    for (const handler of this._events.get(event) ?? []) {
      handler(payload);
    }
  }

  getState() {
    return this.state;
  }

  connect() {
    return new Promise((resolve, reject) => {
      this._resolveOpen = resolve;
      this._rejectOpen = reject;
      this.manualClose = false;
      this._openSocket();
    });
  }

  _openSocket() {
    if (this.manualClose) return;
    this.attempts += 1;
    this.emit("connecting", { attempt: this.attempts });

    let socket;
    try {
      socket = new this.WebSocket(this.url);
    } catch (err) {
      this._scheduleReconnect();
      throw err;
    }
    this.socket = socket;

    socket.onopen = () => {
      this.attempts = 0;
      this.connected = true;
      this.lastPong = Date.now();
      this.emit("open", { deviceId: this.deviceId });
      if (this._resolveOpen) {
        this._resolveOpen(true);
        this._resolveOpen = null;
        this._rejectOpen = null;
      }
      this.startHeartbeat(this.heartbeatIntervalMs);
      // Ask for the missed window before replaying our queued writes.
      this.resync().catch(() => {});
      this._flushPending();
      this._sendRaw({
        type: "hello",
        device_id: this.deviceId,
        version: { seq: this.seq },
      });
    };

    socket.onmessage = (event) => {
      let raw = event.data;
      if (typeof raw !== "string" && raw != null) {
        raw = Buffer.from(raw).toString("utf8");
      }
      this._onMessage(raw);
    };

    socket.onerror = () => {
      // onclose drives teardown; keep handlers simple.
    };

    socket.onclose = (info) => {
      this.connected = false;
      this.stopHeartbeat();
      this.emit("close", { code: info && info.code, reason: info && info.reason });
      if (!this.manualClose) this._scheduleReconnect();
    };
  }

  _scheduleReconnect() {
    if (this.manualClose || this.connected) return;
    if (this.attempts >= this.maxAttempts) {
      this.emit("error", new Error("max reconnect attempts reached"));
      if (this._rejectOpen) {
        const err = new Error("connection failed: max attempts reached");
        this._rejectOpen(err);
        this._resolveOpen = null;
        this._rejectOpen = null;
      }
      return;
    }
    const delay = backoffDelay(
      this.baseDelay, this.attempts, this.maxDelay, this.jitter
    );
    this.emit("reconnecting", { attempt: this.attempts, delayMs: delay });
    this._timers.reconnect = setTimeout(() => this._openSocket(), delay);
  }

  /* ------------------------- outbound deltas ------------------------- */

  sendDelta(delta) {
    const packet = {
      type: "delta",
      op: delta.type,
      path: delta.path ?? [],
      value: delta.value,
      seq: ++this.seq,
      device_id: this.deviceId,
    };
    this.state = applyDelta(this.state, {
      type: delta.type, path: delta.path, value: delta.value,
    });
    this.emit("delta", { ...packet, local: true });
    if (this.connected && this.socket && this.socket.readyState === 1) {
      this.socket.send(JSON.stringify(packet));
    } else {
      this.pending.push(packet);
    }
    return this.seq;
  }

  _flushPending() {
    if (this.pending.length === 0) return;
    const queued = this.pending;
    this.pending = [];
    for (const packet of queued) {
      if (this.socket && this.socket.readyState === 1) {
        this.socket.send(JSON.stringify(packet));
      } else {
        this.pending.push(packet);
      }
    }
  }

  _sendRaw(payload) {
    if (this.connected && this.socket && this.socket.readyState === 1) {
      this.socket.send(JSON.stringify(payload));
      return true;
    }
    return false;
  }

  /* --------------------------- heartbeat ---------------------------- */

  startHeartbeat(intervalMs = this.heartbeatIntervalMs) {
    this.stopHeartbeat();
    this.lastPong = Date.now();
    this._timers.heartbeat = setInterval(() => {
      if (this.connected && this.socket && this.socket.readyState === 1) {
        this.socket.send(JSON.stringify({
          type: "ping", ts: Date.now(), device_id: this.deviceId,
        }));
      }
    }, intervalMs);
    this._timers.watchdog = setInterval(() => {
      if (this.connected &&
          Date.now() - this.lastPong > this.heartbeatTimeoutMs) {
        this.forceReconnect();
      }
    }, Math.min(intervalMs, Math.max(250, this.heartbeatTimeoutMs / 2)));
  }

  stopHeartbeat() {
    clearInterval(this._timers.heartbeat);
    clearInterval(this._timers.watchdog);
    this._timers.heartbeat = null;
    this._timers.watchdog = null;
  }

  forceReconnect() {
    this.emit("error", new Error("heartbeat timed out, reconnecting"));
    if (this.socket) {
      try {
        this.socket.close(4000, "heartbeat timeout");
      } catch {
        /* closing an already-dead socket is fine */
      }
    }
    this.connected = false;
    if (!this.manualClose) this._scheduleReconnect();
  }

  /* ----------------------- inbound message bus ---------------------- */

  _onMessage(raw) {
    if (typeof raw !== "string" || raw.length === 0) return;
    let msg;
    try {
      msg = JSON.parse(raw);
    } catch {
      return; // ignore non-JSON frames (binary, chat payloads, etc.)
    }
    switch (msg.type) {
      case "pong":
        this.lastPong = Date.now();
        break;
      case "ping":
        this._sendRaw({ type: "pong", ts: Date.now(), device_id: this.deviceId });
        break;
      case "delta": {
        this._applyWireDelta(msg);
        break;
      }
      case "batch": {
        for (const delta of msg.deltas ?? []) {
          this._applyWireDelta(delta);
        }
        break;
      }
      case "resync_ack": {
        this._settleResync(msg);
        break;
      }
      case "set":
      case "merge":
      case "replace":
      case "delete": {
        this.state = applyDelta(this.state, msg);
        break;
      }
      default:
        break; // unknown control frames are tolerated
    }
  }

  _applyWireDelta(msg) {
    const op = { type: msg.op ?? "set", path: msg.path ?? [], value: msg.value };
    this.state = applyDelta(this.state, op);
    if (Number.isFinite(msg.seq) && msg.seq > this.lastServerSeq) {
      this.lastServerSeq = msg.seq;
    }
    this.emit("delta", { ...msg, local: false });
  }

  /* ------------------ version-vector resync ------------------ */

  resync() {
    return new Promise((resolve, reject) => {
      const requestId =
        `rs-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`;
      const waiter = {
        requestId,
        resolve,
        reject,
        timer: setTimeout(() => {
          this._dropResyncWaiter(requestId);
          reject(new Error("resync timed out"));
        }, 5000),
      };
      this._resyncWaiters.push(waiter);
      const fingerprint = { device_id: this.deviceId, seq: this.seq };
      const sent = this._sendRaw({
        type: "resync_request",
        request_id: requestId,
        version: fingerprint,
        last_server_seq: this.lastServerSeq,
      });
      if (!sent) {
        clearTimeout(waiter.timer);
        this._dropResyncWaiter(requestId);
        reject(new Error("not connected — cannot resync"));
      }
    });
  }

  _dropResyncWaiter(requestId) {
    const idx = this._resyncWaiters.findIndex((w) => w.requestId === requestId);
    if (idx >= 0) this._resyncWaiters.splice(idx, 1);
  }

  _settleResync(msg) {
    const idx = this._resyncWaiters.findIndex(
      (w) => w.requestId === msg.request_id
    );
    if (idx < 0) return;
    const [waiter] = this._resyncWaiters.splice(idx, 1);
    clearTimeout(waiter.timer);
    waiter.resolve({
      last_server_seq: msg.last_server_seq,
      applied: msg.applied ?? 0,
    });
  }

  close() {
    this.manualClose = true;
    this.stopHeartbeat();
    clearTimeout(this._timers.reconnect);
    if (this.socket) {
      try {
        this.socket.close(1000, "client bye");
      } catch {
        /* already closed */
      }
    }
    this.connected = false;
    this.socket = null;
  }
}

/* ---------------------- in-memory echo demo ---------------------- */

async function runDemo() {
  let wsModule = null;
  try {
    wsModule = nodeRequire("ws");
  } catch {
    wsModule = null;
  }
  if (!wsModule) {
    console.log(
      "Demo needs a WebSocket server: run `npm i ws` first (client detects " +
      "the global WebSocket on Node 22+ automatically)."
    );
    return;
  }

  const { WebSocketServer } = wsModule;
  let serverState = { tasks: {} };
  const serverLog = [];
  let nextSeq = 0;

  const wss = new WebSocketServer({ port: 0 });
  const port = wss.address().port;

  wss.on("connection", (socket) => {
    socket.send(JSON.stringify({ type: "replace", path: [], value: serverState }));
    socket.on("message", (data) => {
      let msg;
      try {
        msg = JSON.parse(data.toString("utf8"));
      } catch {
        return;
      }
      switch (msg.type) {
        case "ping":
          socket.send(JSON.stringify({ type: "pong", ts: Date.now() }));
          break;
        case "delta": {
          const delta = { op: msg.op, path: msg.path, value: msg.value };
          serverLog.push({ ...delta, seq: ++nextSeq });
          serverState = applyDelta(serverState, { type: msg.op, path: msg.path, value: msg.value });
          break;
        }
        case "resync_request": {
          const seen = msg.last_server_seq ?? 0;
          const missed = serverLog.filter((d) => d.seq > seen);
          socket.send(JSON.stringify({ type: "batch", deltas: missed }));
          socket.send(JSON.stringify({
            type: "resync_ack",
            request_id: msg.request_id,
            last_server_seq: nextSeq,
            applied: missed.length,
          }));
          break;
        }
        default:
          break;
      }
    });
  });

  const client = new RealtimeClient({
    url: `ws://127.0.0.1:${port}`,
    heartbeatIntervalMs: 2000,
    heartbeatTimeoutMs: 3000,
  });
  await client.connect();

  client.sendDelta({ type: "set", path: ["tasks", "1"], value: { title: "write skill 14", done: false } });
  client.sendDelta({ type: "merge", path: ["tasks", "1"], value: { done: true } });

  // Simulate an interrupted session: pretend we missed every server delta,
  // then request a resync. The server replays the whole window.
  client.lastServerSeq = 0;
  const summary = await client.resync();

  console.log("resync:", JSON.stringify(summary));
  console.log("client state:", JSON.stringify(client.getState(), null, 2));
  console.log("server state:", JSON.stringify(serverState, null, 2));

  const reconciled =
    JSON.stringify(client.getState()) === JSON.stringify(serverState);
  console.log(reconciled ? "RECONCILED OK" : "STATE MISMATCH");

  client.close();
  await new Promise((resolve) => wss.close(resolve));
}

const isMain =
  process.argv[1] &&
  import.meta.url === pathToFileURL(process.argv[1]).href;

if (isMain) {
  runDemo().catch((err) => {
    console.error(err);
    process.exitCode = 1;
  });
}

export { applyDelta, backoffDelay, deleteAtPath, mergeInto, resolveWebSocket, setAtPath };

Run it:

node --version        # >= 18
npm i ws              # only needed for the demo server on Node < 22
node realtime-client.js

4. Execution Protocol & Step-by-Step Workflow

1. Provision the transport. Confirm the server exposes a WebSocket (or SSE fallback) endpoint that speaks the frame types above: ping/pong, delta, batch, resync_request/resync_ack.
2. Wrap the module. import { RealtimeClient } from "./realtime-client.js", construct with the server URL, then await client.connect().
3. Establish ground truth. On the first open, request replace of the full state (full-state bootstrap) and cache the resulting version.seq. Subsequent reconnects only need the incremental window.
4. Emit deltas locally-first. Call sendDelta for every local mutation — the local tree applies optimistically; the server batch/echo reconciles.
5. Reconcile on restore. on("open") already triggers resync(). If you need a manual pull (e.g. after a long tab background), call resync() directly and only then re-enable the UI's dirty flag.
6. Tune the backoff. Set maxDelay ≤ typical hub recovery time and keep maxAttempts bounded; track reconnecting events for observability.
7. Monitor liveness. heartbeatIntervalMs/heartbeatTimeoutMs defaults (15s/5s) suit most health probes; tighten on flaky mobile networks.
8. Shut down cleanly. Call close() on unload to mark manualClose so the reconnect loop does not keep the tab alive after the user leaves.

Server-side reconciliation contract (reference)

The demo server (ws WebSocketServer) implements the required server semantics: it tags every accepted delta with a monotonically increasing seq, persists a serverLog, answers ping with pong, answers resync_request with a batch of deltas whose seq > last_server_seq, then a matching resync_ack. Any production hub should mirror at least those four behaviors plus per-connection device_id scoping.

5. Edge Cases & Error Handling

• No WebSocket implementation — resolveWebSocket() throws a descriptive error (run Node ≥ 22 or npm i ws). The demo degrades to a printed hint instead of a stack trace.
• Reconnect storms — exponential backoff with jitter spreads the herd; the min(cap, base * 2**attempt) cap prevents unbounded delay growth. Cap maxAttempts for long-lived mobile sessions.
• Heartbeat death — the watchdog observes now - lastPong; if it exceeds heartbeatTimeoutMs the socket is force-closed and the reconnect loop starts even though onclose may not fire promptly.
• readyState flakiness during reconnect — outbound deltas while disconnected are queued in pending and flushed on open, so a burst of user edits during a blip is never dropped from the local sequence.
• Slow/dropped resync — resync() rejects on a 5s timeout or when offline; callers treat it as non-fatal (a later open auto-resyncs).
• Unknown messages — non-JSON frames and unknown type values are ignored (forward-compatibility); _applyWireDelta guards seq with Number.isFinite. Bad payloads are never allowed to crash state parsing.
• State type drift — merge deep-merges only plain-object subtrees and overwrites arrays/leaf values wholesale; document that arrays are replaced, not element-merged, to avoid silent inconsistent lists.
• Version skew on the client — lastServerSeq is bumped only on Number.isFinite higher seqs, so out-of-order delivery cannot rewind the watermark and force useless full replays.
• WebSocket + SSE — if a proxy strips WebSocket upgrades, switch the transport to SSE and keep RealtimeClient interfaces (sendDelta, resync, getState) identical; only _openSocket and _onMessage change.