Real-Time Sync Client

2 collapsed lines
1
// Types and constants
2
type ReconnectionOptions = {
3
  initialDelay?: number
4
  maxDelay?: number
5
  multiplier?: number
6
  jitter?: number
7
  maxAttempts?: number
8
}
9

10
class ReconnectionManager {
11
  private currentDelay: number
12
  private attempts = 0
13

14
  constructor(private options: Required<ReconnectionOptions>) {
15
    this.currentDelay = options.initialDelay
16
  }
17

18
  // Core logic: calculate next delay with jitter
19
  getNextDelay(): number | null {
20
    if (this.attempts >= this.options.maxAttempts) {
21
      return null // Give up
22
    }
23

24
    const jitterRange = this.currentDelay * this.options.jitter
25
    const jitter = Math.random() * jitterRange * 2 - jitterRange
26
    const delay = Math.min(this.currentDelay + jitter, this.options.maxDelay)
27

28
    this.currentDelay = Math.min(this.currentDelay * this.options.multiplier, this.options.maxDelay)
29
    this.attempts++
30

31
    return delay
32
  }
33

34
  reset(): void {
11 collapsed lines
35
    this.currentDelay = this.options.initialDelay
36
    this.attempts = 0
37
  }
38
}
39

40
// Usage
41
const manager = new ReconnectionManager({
42
  initialDelay: 1000,
43
  maxDelay: 30000,
44
  multiplier: 2,
45
  jitter: 0.1,
46
  maxAttempts: 10,
47
})

4 collapsed lines
1
// WebSocket wrapper with heartbeat
2
type HeartbeatOptions = {
3
  interval: number // How often to send ping
4
  timeout: number // How long to wait for pong
5
}
6

7
class HeartbeatConnection {
8
  private pingTimer: number | null = null
9
  private pongTimer: number | null = null
10
  private ws: WebSocket
11

12
  constructor(
13
    url: string,
14
    private options: HeartbeatOptions,
15
  ) {
16
    this.ws = new WebSocket(url)
17
    this.ws.onopen = () => this.startHeartbeat()
18
    this.ws.onmessage = (e) => this.handleMessage(e)
19
    this.ws.onclose = () => this.stopHeartbeat()
20
  }
21

22
  // Core heartbeat logic
23
  private startHeartbeat(): void {
24
    this.pingTimer = window.setInterval(() => {
25
      this.ws.send(JSON.stringify({ type: "ping", ts: Date.now() }))
26

27
      this.pongTimer = window.setTimeout(() => {
28
        // No pong received - connection is dead
29
        this.ws.close(4000, "Heartbeat timeout")
11 collapsed lines
30
      }, this.options.timeout)
31
    }, this.options.interval)
32
  }
33

34
  private handleMessage(event: MessageEvent): void {
35
    const data = JSON.parse(event.data)
36
    if (data.type === "pong") {
37
      if (this.pongTimer) clearTimeout(this.pongTimer)
38
      return
39
    }
40
    // Handle application messages...
41
  }
42

43
  private stopHeartbeat(): void {
44
    if (this.pingTimer) clearInterval(this.pingTimer)
45
    if (this.pongTimer) clearTimeout(this.pongTimer)
46
  }
47
}

5 collapsed lines
1
// Track last acknowledged sequence
2
interface Message {
3
  seq: number
4
  payload: unknown
5
}
6

7
class SequenceRecovery {
8
  private lastAckedSeq = 0
9
  private pendingMessages: Message[] = []
10

11
  send(payload: unknown): void {
12
    const msg: Message = {
13
      seq: this.pendingMessages.length + this.lastAckedSeq + 1,
14
      payload,
15
    }
16
    this.pendingMessages.push(msg)
17
    this.ws.send(JSON.stringify(msg))
18
  }
19

20
  // On reconnect, request replay from last ack
21
  onReconnect(): void {
22
    this.ws.send(
23
      JSON.stringify({
24
        type: "resume",
11 collapsed lines
25
        lastSeq: this.lastAckedSeq,
26
      }),
27
    )
28
  }
29

30
  onAck(seq: number): void {
31
    this.lastAckedSeq = seq
32
    this.pendingMessages = this.pendingMessages.filter((m) => m.seq > seq)
33
  }
34

35
  onServerReplay(messages: Message[]): void {
36
    // Server sends missed messages since lastSeq
37
    for (const msg of messages) {
38
      this.handleMessage(msg)
39
    }
40
  }
41
}

1
const eventSource = new EventSource("/events")
2

3
eventSource.onmessage = (event) => {
4
  // event.lastEventId contains the id from the server
5
  // On reconnect, browser automatically sends Last-Event-ID header
6
  processEvent(JSON.parse(event.data))
7
}

4 collapsed lines
1
// Per-sender message ordering
2
type SenderId = string
3

4
class OrderedMessageHandler {
5
  private lastSeq = new Map<SenderId, number>()
6
  private pending = new Map<SenderId, Map<number, unknown>>()
7

8
  handle(senderId: SenderId, seq: number, payload: unknown): void {
9
    const expected = (this.lastSeq.get(senderId) ?? 0) + 1
10

11
    if (seq === expected) {
12
      // In order - process and check pending
13
      this.process(payload)
14
      this.lastSeq.set(senderId, seq)
15
      this.processPending(senderId)
16
    } else if (seq > expected) {
17
      // Out of order - buffer
18
      const senderPending = this.pending.get(senderId) ?? new Map()
19
      senderPending.set(seq, payload)
20
      this.pending.set(senderId, senderPending)
21
    }
22
    // seq < expected means duplicate - ignore
23
  }
24

25
  private processPending(senderId: SenderId): void {
26
    const senderPending = this.pending.get(senderId)
27
    if (!senderPending) return
28

29
    let next = (this.lastSeq.get(senderId) ?? 0) + 1
30
    while (senderPending.has(next)) {
31
      this.process(senderPending.get(next))
32
      senderPending.delete(next)
33
      this.lastSeq.set(senderId, next)
34
      next++
35
    }
36
  }
37

38
  private process(payload: unknown): void {
39
    // Application-specific processing
40
  }
41
}

3 collapsed lines
1
// Time-bounded deduplication
2
const DEDUP_WINDOW_MS = 5 * 60 * 1000 // 5 minutes
3

4
class Deduplicator {
5
  private processed = new Map<string, number>() // id -> timestamp
6

7
  isDuplicate(messageId: string): boolean {
8
    this.cleanup()
9

10
    if (this.processed.has(messageId)) {
11
      return true
12
    }
13

14
    this.processed.set(messageId, Date.now())
15
    return false
16
  }
17

18
  private cleanup(): void {
19
    const cutoff = Date.now() - DEDUP_WINDOW_MS
20
    for (const [id, ts] of this.processed) {
21
      if (ts < cutoff) {
22
        this.processed.delete(id)
23
      }
24
    }
11 collapsed lines
25
  }
26
}
27

28
// Usage
29
const dedup = new Deduplicator()
30

31
function handleMessage(msg: { id: string; payload: unknown }): void {
32
  if (dedup.isDuplicate(msg.id)) {
33
    return // Already processed
34
  }
35
  processPayload(msg.payload)
36
}

8 collapsed lines
1
// Optimistic update with rollback
2
type Todo = { id: string; text: string; completed: boolean }
3
type TodoStore = {
4
  todos: Todo[]
5
  pendingUpdates: Map<string, { previous: Todo; current: Todo }>
6
}
7

8
const store: TodoStore = { todos: [], pendingUpdates: new Map() }
9

10
async function toggleTodo(id: string): Promise<void> {
11
  const todo = store.todos.find((t) => t.id === id)
12
  if (!todo) return
13

14
  // 1. Save previous state
15
  const previous = { ...todo }
16

17
  // 2. Apply optimistic update
18
  const updated = { ...todo, completed: !todo.completed }
19
  store.todos = store.todos.map((t) => (t.id === id ? updated : t))
20
  store.pendingUpdates.set(id, { previous, current: updated })
21

22
  // 3. Notify UI (immediate feedback)
23
  renderTodos()
24

25
  try {
26
    // 4. Send to server
27
    await fetch(`/api/todos/${id}`, {
28
      method: "PATCH",
29
      body: JSON.stringify({ completed: updated.completed }),
30
      headers: { "Idempotency-Key": `toggle-${id}-${Date.now()}` },
31
    })
32

33
    // 5. Success - remove from pending
34
    store.pendingUpdates.delete(id)
35
  } catch (error) {
36
    // 6. Failure - rollback
37
    store.todos = store.todos.map((t) => (t.id === id ? store.pendingUpdates.get(id)!.previous : t))
38
    store.pendingUpdates.delete(id)
39
    renderTodos()
40
    showError("Failed to update todo")
41
  }
42
}
43

44
// TanStack Query equivalent
15 collapsed lines
45
const mutation = useMutation({
46
  mutationFn: (todo: Todo) => updateTodo(todo),
47
  onMutate: async (newTodo) => {
48
    await queryClient.cancelQueries({ queryKey: ["todos"] })
49
    const previous = queryClient.getQueryData(["todos"])
50
    queryClient.setQueryData(["todos"], (old: Todo[]) => old.map((t) => (t.id === newTodo.id ? newTodo : t)))
51
    return { previous }
52
  },
53
  onError: (err, newTodo, context) => {
54
    queryClient.setQueryData(["todos"], context?.previous)
55
  },
56
  onSettled: () => {
57
    queryClient.invalidateQueries({ queryKey: ["todos"] })
58
  },
59
})

4 collapsed lines
1
// React 19 built-in optimistic updates
2
import { useOptimistic } from 'react';
3

4
function TodoItem({ todo }: { todo: Todo }) {
5
  const [optimisticTodo, setOptimisticTodo] = useOptimistic(
6
    todo,
7
    (current, completed: boolean) => ({ ...current, completed })
8
  );
9

10
  async function toggle() {
11
    setOptimisticTodo(!optimisticTodo.completed);
12
    await updateTodo({ ...todo, completed: !todo.completed });
13
  }
14

15
  return (
16
    <li style={{ opacity: optimisticTodo.completed !== todo.completed ? 0.5 : 1 }}>
17
      <input type="checkbox" checked={optimisticTodo.completed} onChange={toggle} />
18
      {todo.text}
19
    </li>
2 collapsed lines
20
  );
21
}

1
type LWWRegister<T> = {
2
  value: T
3
  timestamp: number
4
  clientId: string
5
}
6

7
function merge<T>(local: LWWRegister<T>, remote: LWWRegister<T>): LWWRegister<T> {
8
  if (remote.timestamp > local.timestamp) {
9
    return remote
10
  }
11
  if (remote.timestamp === local.timestamp) {
12
    // Tiebreaker: lexicographic client ID comparison
13
    return remote.clientId > local.clientId ? remote : local
14
  }
15
  return local
16
}

10 collapsed lines
1
// Simplified OT for text
2
type Insert = { type: "insert"; pos: number; char: string }
3
type Delete = { type: "delete"; pos: number }
4
type Op = Insert | Delete
5

6
function transform(op1: Op, op2: Op): Op {
7
  // Transform op1 assuming op2 has been applied
8
  if (op1.type === "insert" && op2.type === "insert") {
9
    if (op1.pos <= op2.pos) {
10
      return op1 // op1 is before op2, no change
11
    }
12
    return { ...op1, pos: op1.pos + 1 } // Shift right
13
  }
14

15
  if (op1.type === "insert" && op2.type === "delete") {
16
    if (op1.pos <= op2.pos) {
17
      return op1
18
    }
19
    return { ...op1, pos: op1.pos - 1 } // Shift left
20
  }
21

22
  if (op1.type === "delete" && op2.type === "insert") {
23
    if (op1.pos < op2.pos) {
24
      return op1
25
    }
26
    return { ...op1, pos: op1.pos + 1 }
27
  }
28

29
  if (op1.type === "delete" && op2.type === "delete") {
30
    if (op1.pos < op2.pos) {
31
      return op1
32
    }
33
    if (op1.pos > op2.pos) {
34
      return { ...op1, pos: op1.pos - 1 }
35
    }
36
    // Same position - op1 is a no-op (already deleted)
37
    return { type: "insert", pos: -1, char: "" } // No-op sentinel
38
  }
39

16 collapsed lines
40
  return op1
41
}
42

43
// OT requires central server for operation ordering
44
class OTClient {
45
  private pending: Op[] = []
46
  private serverVersion = 0
47

48
  applyLocal(op: Op): void {
49
    this.pending.push(op)
50
    this.sendToServer(op, this.serverVersion)
51
  }
52

53
  onServerAck(ackVersion: number): void {
54
    this.serverVersion = ackVersion
55
    this.pending.shift() // Remove acknowledged operation
56
  }
57

58
  onServerOp(op: Op): void {
59
    // Transform all pending operations against server operation
60
    for (let i = 0; i < this.pending.length; i++) {
61
      this.pending[i] = transform(this.pending[i], op)
62
    }
63
    this.applyToDocument(op)
64
    this.serverVersion++
65
  }
66
}

3 collapsed lines
1
// G-Counter CRDT
2
type NodeId = string
3

4
class GCounter {
5
  private counts = new Map<NodeId, number>()
6

7
  constructor(private nodeId: NodeId) {}
8

9
  increment(): void {
10
    const current = this.counts.get(this.nodeId) ?? 0
11
    this.counts.set(this.nodeId, current + 1)
12
  }
13

14
  value(): number {
15
    let sum = 0
16
    for (const count of this.counts.values()) {
17
      sum += count
18
    }
19
    return sum
20
  }
21

22
  // Merge is commutative, associative, idempotent
23
  merge(other: GCounter): void {
24
    for (const [nodeId, count] of other.counts) {
25
      const current = this.counts.get(nodeId) ?? 0
26
      this.counts.set(nodeId, Math.max(current, count))
27
    }
28
  }
29

30
  serialize(): Record<NodeId, number> {
31
    return Object.fromEntries(this.counts)
32
  }
33
}

10 collapsed lines
1
// State reconciliation with pending input replay
2
type Input = { id: string; action: string; params: unknown }
3
type State = {
4
  /* application state */
5
}
6

7
class ReconciliationEngine {
8
  private pendingInputs: Input[] = []
9
  private localState: State
10
  private lastAckedInputId: string | null = null
11

12
  applyInput(input: Input): void {
13
    // 1. Apply locally for immediate feedback
14
    this.localState = this.applyToState(this.localState, input)
15

16
    // 2. Track as pending
17
    this.pendingInputs.push(input)
18

19
    // 3. Send to server
20
    this.sendToServer(input)
21
  }
22

23
  onServerUpdate(serverState: State, lastAckedId: string): void {
24
    // 1. Remove acknowledged inputs
25
    const ackIndex = this.pendingInputs.findIndex((i) => i.id === lastAckedId)
26
    if (ackIndex !== -1) {
27
      this.pendingInputs = this.pendingInputs.slice(ackIndex + 1)
28
    }
29

30
    // 2. Start from authoritative server state
31
    let reconciledState = serverState
32

33
    // 3. Re-apply pending (unacknowledged) inputs
34
    for (const input of this.pendingInputs) {
35
      reconciledState = this.applyToState(reconciledState, input)
36
    }
37

38
    // 4. Update local state
39
    this.localState = reconciledState
40

41
    // 5. Re-render
42
    this.render()
43
  }
44

45
  private applyToState(state: State, input: Input): State {
46
    // Application-specific state transition
47
    return state
48
  }
49

8 collapsed lines
50
  private sendToServer(input: Input): void {
51
    // WebSocket send
52
  }
53

54
  private render(): void {
55
    // Update UI
56
  }
57
}

4 collapsed lines
1
// Simple full state sync on reconnect
2
async function reconnect(): Promise<void> {
3
  const state = await fetch("/api/state").then((r) => r.json())
4
  store.setState(state)
5
  render()
6
}
7

8
// With local mutation queue
9
async function reconnectWithQueue(): Promise<void> {
10
  // 1. Fetch server state
11
  const serverState = await fetch("/api/state").then((r) => r.json())
12

13
  // 2. Replay queued mutations
14
  for (const mutation of localQueue) {
15
    try {
16
      await fetch("/api/mutate", {
17
        method: "POST",
18
        body: JSON.stringify(mutation),
19
      })
11 collapsed lines
20
    } catch {
21
      // Handle permanent failure
22
    }
23
  }
24

25
  // 3. Fetch final state (includes replayed mutations)
26
  const finalState = await fetch("/api/state").then((r) => r.json())
27
  store.setState(finalState)
28
  localQueue.clear()
29
  render()
30
}

5 collapsed lines
1
// Heartbeat presence
2
const HEARTBEAT_INTERVAL = 15000 // 15 seconds
3
const TIMEOUT_THRESHOLD = 45000 // 3 missed heartbeats
4

5
class PresenceManager {
6
  private users = new Map<string, { lastSeen: number; metadata: unknown }>()
7
  private heartbeatTimer: number | null = null
8

9
  start(userId: string, metadata: unknown): void {
10
    this.heartbeatTimer = window.setInterval(() => {
11
      this.ws.send(
12
        JSON.stringify({
13
          type: "presence",
14
          userId,
15
          metadata,
16
          timestamp: Date.now(),
17
        }),
18
      )
19
    }, HEARTBEAT_INTERVAL)
20
  }
21

22
  onPresenceUpdate(userId: string, metadata: unknown): void {
23
    this.users.set(userId, { lastSeen: Date.now(), metadata })
24
    this.pruneStale()
25
    this.render()
26
  }
27

28
  private pruneStale(): void {
29
    const now = Date.now()
11 collapsed lines
30
    for (const [userId, data] of this.users) {
31
      if (now - data.lastSeen > TIMEOUT_THRESHOLD) {
32
        this.users.delete(userId)
33
      }
34
    }
35
  }
36

37
  stop(): void {
38
    if (this.heartbeatTimer) clearInterval(this.heartbeatTimer)
39
  }
40

41
  getOnlineUsers(): string[] {
42
    return Array.from(this.users.keys())
43
  }
44
}

8 collapsed lines
1
// Throttled cursor sharing
2
type CursorPosition = { x: number; y: number; userId: string }
3

4
class CursorPresence {
5
  private cursors = new Map<string, CursorPosition>()
6
  private throttledSend: (pos: CursorPosition) => void
7

8
  constructor(
9
    private ws: WebSocket,
10
    private userId: string,
11
  ) {
12
    // Throttle to 30 updates per second max
13
    this.throttledSend = throttle((pos: CursorPosition) => {
14
      ws.send(JSON.stringify({ type: "cursor", ...pos }))
15
    }, 33)
16
  }
17

18
  onLocalMove(x: number, y: number): void {
19
    this.throttledSend({ x, y, userId: this.userId })
20
  }
21

22
  onRemoteCursor(cursor: CursorPosition): void {
23
    this.cursors.set(cursor.userId, cursor)
24
    this.renderCursors()
25
  }
26

27
  onUserLeave(userId: string): void {
28
    this.cursors.delete(userId)
29
    this.renderCursors()
30
  }
31

32
  private renderCursors(): void {
33
    // Render remote cursors with user labels
34
  }
12 collapsed lines
35
}
36

37
function throttle<T extends (...args: unknown[]) => void>(fn: T, ms: number): T {
38
  let lastCall = 0
39
  return ((...args: Parameters<T>) => {
40
    const now = Date.now()
41
    if (now - lastCall >= ms) {
42
      lastCall = now
43
      fn(...args)
44
    }
45
  }) as T
46
}

1
// Main thread
2
const worker = new Worker("sync-worker.js")
3

4
worker.postMessage({ type: "parse", data: rawJson })
5

6
worker.onmessage = (e) => {
7
  // Parsed data ready
8
  updateState(e.data)
9
}
10

11
// sync-worker.js
12
self.onmessage = (e) => {
13
  if (e.data.type === "parse") {
14
    const parsed = JSON.parse(e.data.data)
15
    self.postMessage(parsed)
16
  }
17
}

1
// Chrome 124+ WebSocketStream with backpressure
2
const ws = new WebSocketStream("wss://example.com")
3
const { readable, writable } = await ws.opened
4

5
const reader = readable.getReader()
6
while (true) {
7
  const { value, done } = await reader.read()
8
  if (done) break
9

10
  // Backpressure: read() naturally pauses when we can't keep up
11
  await processMessage(value)
12
}

5 collapsed lines
1
// Handle quota exceeded gracefully
2
async function cacheData(key: string, value: unknown): Promise<void> {
3
  const db = await openDB("cache", 1)
4

5
  try {
6
    await db.put("data", { key, value, timestamp: Date.now() })
7
  } catch (e) {
8
    if (e.name === "QuotaExceededError") {
9
      // Evict oldest entries
10
      const all = await db.getAll("data")
11
      all.sort((a, b) => a.timestamp - b.timestamp)
12

13
      // Delete oldest 20%
14
      const toDelete = all.slice(0, Math.ceil(all.length * 0.2))
15
      for (const item of toDelete) {
16
        await db.delete("data", item.key)
17
      }
18

19
      // Retry
20
      await db.put("data", { key, value, timestamp: Date.now() })
21
    } else {
22
      throw e
23
    }
24
  }
25
}

10 collapsed lines
1
// Network-aware sync strategy
2
type ConnectionType = "wifi" | "cellular" | "none"
3

4
function getConnectionType(): ConnectionType {
5
  const connection = (navigator as unknown as { connection?: { type: string } }).connection
6
  if (!connection) return "wifi" // Assume best
7
  return connection.type === "wifi" ? "wifi" : "cellular"
8
}
9

10
class NetworkAwareSync {
11
  private batchBuffer: unknown[] = []
12
  private flushTimer: number | null = null
13

14
  send(message: unknown): void {
15
    this.batchBuffer.push(message)
16

17
    const delay = getConnectionType() === "cellular" ? 2000 : 100
18

19
    if (!this.flushTimer) {
20
      this.flushTimer = window.setTimeout(() => {
21
        this.flush()
22
      }, delay)
23
    }
24
  }
25

26
  private flush(): void {
27
    if (this.batchBuffer.length === 0) return
28

29
    this.ws.send(
30
      JSON.stringify({
31
        type: "batch",
32
        messages: this.batchBuffer,
33
      }),
34
    )
10 collapsed lines
35

36
    this.batchBuffer = []
37
    this.flushTimer = null
38
  }
39
}
40

41
// Detect network type changes
42
navigator.connection?.addEventListener("change", () => {
43
  adjustSyncStrategy(getConnectionType())
44
})