# System: Real-Time Infrastructure

**Date:** 2026-05-31  
**Status:** Draft  
**Layer:** 1 — System Core  
**Spec #:** 45  
**Depends on:** `foundation-monorepo`, `foundation-auth-rbac`, `notification-center`  
**Referenced by:** `app-shell` (spec 7, notification badge), `tasks-board-engine` (spec 11, board refresh), `crm-support-center` (spec 14, ticket updates), `home-dashboard` (spec 30, activity feed), `time-management` (spec 13, timer sync)

---

## Overview

Defines the server-push real-time infrastructure for Zync. Enables the server to push events to connected browser tabs without polling. All Zync real-time communication is unidirectional: server → client. Client → server continues via standard REST calls.

The chosen pattern: **WebSocket connections to a Durable Object per tenant**. The tenant's DO maintains all active tab connections, receives events from a Cloudflare Queue, and fans out to all connected tabs. This is the correct Cloudflare-native pattern for stateful connection management across a distributed Worker fleet.

---

## Why These Choices

### WebSocket (not SSE)

Although SSE was initially considered, Cloudflare Workers do not support long-lived streaming HTTP responses in the way SSE requires — Cloudflare terminates idle connections and Workers cannot hold open responses across PoP boundaries without Durable Objects. Using WebSocket with a Durable Object gives the same server-push semantics with better Cloudflare-native support and built-in reconnection handling.

SSE `EventSource` reconnect is browser-managed and sends `Last-Event-ID` — but without a stateful connection holder, the Worker still cannot maintain an open stream. The DO makes this work.

### Durable Object (not stateless Workers + KV)

Stateless Workers cannot hold open WebSocket connections between requests. KV-based connection registries introduce propagation delay and are not designed for connection state. One DO per tenant is low-cost, naturally scoped, and the Cloudflare-recommended pattern for this use case.

### No persistent event table

Real-time events are ephemeral. If a tab is not connected when an event fires, it re-fetches on reconnect. Persistent notification records live in the `notifications` table (spec 35) — the real-time layer delivers the push signal only.

---

## Architecture

```
Server action (e.g. task moved, new message)
          │
          ▼
   Write event to Queue
   `zync-realtime` (tenant_id + event payload)
          │
          ▼
   Queue consumer Worker
   → looks up tenant's Durable Object
   → DO.fetch(event) to fan out
          │
          ▼
   TenantRealtimeDO
   → iterates all open WebSocket connections for this tenant
   → sends event JSON to each connected tab
          │
          ▼
   Browser tab
   → parses event type
   → triggers UI update (refetch, optimistic update, badge increment)
```

### Deployment topology

All three logical components run **inside the `zync-api` Worker** — not as separate
deployables: the upgrade proxy is the `/api/realtime/connect` route, the queue consumer is
a `batch.queue === 'zync-realtime'` case in the Worker's `queue()` handler, and
`TenantRealtimeDO` is a DO class defined and bound by zync-api (`DO_REALTIME`). A single
Worker can own all three; no `script_name` cross-binding is needed. tasks-detail-communication
publishes task events into the same queue.

### Deferred (known gap)

Admin observability (`/admin/realtime` metrics dashboard) is **not** in this slice: it needs
per-tenant connection counts (a KV registry the DO does not yet maintain) plus CF Analytics
Engine / Queue-depth APIs not modeled here. Add a KV connection registry + a
`GET /api/admin/realtime/metrics` route before reintroducing that page.

Broad `notification.new` real-time fan-out is also deferred: notifications are created by
the db-layer `createNotification` helper (no queue binding), so per-notification push needs
the queue threaded through that helper or its call sites — each `notification.new` MUST carry
`targetUserId` (the recipient) so personal notifications never broadcast tenant-wide. Today no
code publishes `notification.new`, so there is no leak; task assignment uses the user-scoped
`task.assigned` event instead.

---

## Durable Object: `TenantRealtimeDO`

One DO instance per tenant. DO ID: `tenant:{tenantId}` (derived from tenant UUID via `idFromName`).

### Connection lifecycle

```ts
// workers/realtime-do/src/index.ts

export class TenantRealtimeDO implements DurableObject {
  private connections = new Map<string, { ws: WebSocket; userId: string; connId: string }>()

  async fetch(request: Request): Promise<Response> {
    const url = new URL(request.url)

    if (url.pathname === '/connect') {
      return this.handleConnect(request)
    }
    if (url.pathname === '/fanout') {
      return this.handleFanout(request)
    }
    return new Response('Not found', { status: 404 })
  }

  private async handleConnect(request: Request): Promise<Response> {
    // Identity is established by the connect route (session cookie) and passed
    // as a trusted server-set header. The DO is never reachable directly.
    const userId = request.headers.get('X-User-Id')
    if (!userId) return new Response('Forbidden', { status: 403 })

    // Upgrade to WebSocket (Hibernation API). Tag the socket with userId so
    // getWebSockets(userId) targets it for targetUserId-scoped events.
    const { 0: client, 1: server } = new WebSocketPair()
    this.ctx.acceptWebSocket(server, [userId])

    return new Response(null, { status: 101, webSocket: client })
  }

  async webSocketClose(ws: WebSocket): Promise<void> {
    const attachment = ws.deserializeAttachment()
    if (attachment?.connId) {
      this.connections.delete(attachment.connId)
    }
  }

  async webSocketError(ws: WebSocket): Promise<void> {
    const attachment = ws.deserializeAttachment()
    if (attachment?.connId) {
      this.connections.delete(attachment.connId)
    }
  }

  private async handleFanout(request: Request): Promise<Response> {
    // Called by Queue consumer to deliver an event
    const event: RealtimeEvent = await request.json()
    const payload = JSON.stringify(event)

    for (const [, conn] of this.connections) {
      // User-scoped events: only send to matching user
      if (event.targetUserId && conn.userId !== event.targetUserId) continue
      try {
        conn.ws.send(payload)
      } catch {
        // Connection dead — will be cleaned up on next webSocketClose
      }
    }

    return new Response('ok')
  }
}
```

### DO hibernation

The DO uses Cloudflare's WebSocket Hibernation API (`ctx.acceptWebSocket`). The DO hibernates when no messages are being processed, waking on incoming messages. This means idle tenants (no active connections) do not incur DO compute cost — the DO only runs when there are events to fan out or connections to manage.

---

## Client Connection

### Connection URL

```
wss://app.zync.is/api/realtime/connect
```

The Worker upgrades the HTTP request to WebSocket and proxies to the tenant's DO.

### Auth

Authentication is the **same-origin HttpOnly session cookie** — no token in the query
string. The app, API, and WebSocket endpoint are all served under `app.zync.is`, so the
browser sends the session cookie automatically on the `wss://app.zync.is/api/realtime/connect`
handshake (cookies are sent on same-origin upgrades; the "cannot set custom headers"
limitation applies to custom headers, not cookies). The connect route runs the normal
`authMiddleware`, then proxies the upgrade to the tenant DO with a trusted server-set
`X-User-Id` header so the DO can tag the socket for `targetUserId`-scoped delivery.
This keeps the client tokenless — the app has no access JWT (sessions are cookie-only).

### Client-side manager

```ts
// apps/zync-app/src/lib/realtime/client.ts

export class RealtimeClient {
  private ws: WebSocket | null = null
  private handlers = new Map<string, Set<(event: RealtimeEvent) => void>>()
  private reconnectDelay = 1_000
  private reconnectTimer: ReturnType<typeof setTimeout> | null = null

  connect() {
    // Same-origin: the session cookie is sent automatically — no token in the URL.
    const url = `wss://${window.location.host}/api/realtime/connect`
    this.ws = new WebSocket(url)

    this.ws.addEventListener('message', (e) => {
      const event: RealtimeEvent = JSON.parse(e.data)
      const eventHandlers = this.handlers.get(event.type)
      eventHandlers?.forEach(h => h(event))
      // Also call wildcard handlers
      this.handlers.get('*')?.forEach(h => h(event))
    })

    this.ws.addEventListener('close', () => this.scheduleReconnect())
    this.ws.addEventListener('error', () => this.scheduleReconnect())

    // Reset backoff on successful open
    this.ws.addEventListener('open', () => { this.reconnectDelay = 1_000 })
  }

  on(eventType: string, handler: (event: RealtimeEvent) => void): () => void {
    if (!this.handlers.has(eventType)) this.handlers.set(eventType, new Set())
    this.handlers.get(eventType)!.add(handler)
    return () => this.handlers.get(eventType)?.delete(handler)  // unsubscribe fn
  }

  private scheduleReconnect() {
    this.reconnectTimer = setTimeout(() => {
      this.reconnectDelay = Math.min(this.reconnectDelay * 2, 30_000)  // max 30s backoff
      this.connect()
    }, this.reconnectDelay)
  }

  disconnect() {
    if (this.reconnectTimer) clearTimeout(this.reconnectTimer)
    this.ws?.close()
    this.ws = null
  }
}

// Singleton + React context provider
export const realtimeClient = new RealtimeClient()
```

### React integration

```ts
// apps/zync-app/src/lib/realtime/hooks.ts

export function useRealtimeEvent(
  eventType: string,
  handler: (event: RealtimeEvent) => void,
  deps: React.DependencyList = []
) {
  useEffect(() => {
    const unsubscribe = realtimeClient.on(eventType, handler)
    return unsubscribe
  }, [eventType, ...deps])
}
```

---

## Event Schema

All events share a common envelope:

```ts
interface RealtimeEvent {
  id: string              // UUID — idempotency for duplicate delivery
  type: RealtimeEventType
  tenantId: string
  targetUserId?: string   // NULL = broadcast to all tenant connections; set = user-scoped
  payload: Record<string, unknown>
  timestamp: string       // ISO 8601
}

type RealtimeEventType =
  | 'notification.new'
  | 'task.status_changed'
  | 'task.assigned'
  | 'ticket.message_added'
  | 'ticket.status_changed'
  | 'activity.new'
  | 'time_entry.started'
  | 'time_entry.stopped'
```

### Event payload shapes

```ts
// notification.new
{ notificationId: string; type: string; title: string; unreadCount: number }

// task.status_changed
{ taskId: string; boardId: string; fromStatus: string; toStatus: string; movedByUserId: string }

// task.assigned
{ taskId: string; assignedToUserId: string; assignedByUserId: string }

// ticket.message_added
{ ticketId: string; messageId: string; authorId: string; authorName: string; preview: string }

// ticket.status_changed
{ ticketId: string; fromStatus: string; toStatus: string }

// activity.new
{ activityId: string; entityType: string; entityId: string; actorId: string; verb: string }

// time_entry.started
{ timeEntryId: string; taskId?: string; startedAt: string }

// time_entry.stopped
{ timeEntryId: string; durationSeconds: number }
```

---

## Event Publication

Server-side code (route handlers, queue consumers) publishes events via:

```ts
// packages/realtime/src/publish.ts

export async function publishRealtimeEvent(
  queue: Queue,
  event: Omit<RealtimeEvent, 'id' | 'timestamp'>
): Promise<void> {
  await queue.send({
    ...event,
    id: crypto.randomUUID(),
    timestamp: new Date().toISOString(),
  })
}
```

### Usage in route handlers

```ts
// Example: task status update route
await db.update(tasks).set({ status: newStatus }).where(eq(tasks.id, taskId))

await publishRealtimeEvent(env.REALTIME_QUEUE, {
  type: 'task.status_changed',
  tenantId: session.tid,
  payload: { taskId, boardId, fromStatus: task.status, toStatus: newStatus, movedByUserId: session.sub },
})
```

Publication is fire-and-forget — it must not block the response. Route handlers do not `await` the queue send in the critical path; it is wrapped in `ctx.waitUntil` where latency sensitivity is high:

```ts
ctx.waitUntil(publishRealtimeEvent(env.REALTIME_QUEUE, event))
```

---

## Queue Consumer

```ts
// workers/realtime-consumer/src/index.ts

export default {
  async queue(batch: MessageBatch<RealtimeEvent>, env: Env): Promise<void> {
    for (const message of batch.messages) {
      const event = message.body
      const doId = env.DO_REALTIME.idFromName(`tenant:${event.tenantId}`)
      const stub = env.DO_REALTIME.get(doId)

      try {
        const response = await stub.fetch('https://do-internal/fanout', {
          method: 'POST',
          headers: { 'Content-Type': 'application/json' },
          body: JSON.stringify(event),
        })
        if (response.ok) {
          message.ack()
        } else {
          message.retry()
        }
      } catch {
        message.retry()
      }
    }
  }
}
```

Queue consumer retries on DO fetch failure. Max retries: 3 (configured in `wrangler.toml`). Dead-letter: events failing all retries are discarded — real-time events are ephemeral; stale events should not accumulate.

---

## Worker: Connection Upgrade Proxy

```ts
// workers/realtime-proxy/src/index.ts
// Handles: GET /api/realtime/connect

export default {
  async fetch(request: Request, env: Env): Promise<Response> {
    if (request.headers.get('Upgrade') !== 'websocket') {
      return new Response('Expected WebSocket upgrade', { status: 426 })
    }

    // Validate JWT from query param
    const url = new URL(request.url)
    const token = url.searchParams.get('token')
    if (!token) return new Response('Missing token', { status: 401 })

    const session = await verifyJWT(token, env.JWT_SECRET)
    if (!session || !session.tid) return new Response('Invalid token', { status: 401 })

    // Route to tenant DO
    const doId = env.DO_REALTIME.idFromName(`tenant:${session.tid}`)
    const stub = env.DO_REALTIME.get(doId)

    // Forward to DO with user info in header (token stripped from URL)
    const doRequest = new Request('https://do-internal/connect', {
      headers: {
        'Upgrade': 'websocket',
        'X-User-Id': session.sub,
        'X-Tenant-Id': session.tid,
      },
    })

    return stub.fetch(doRequest)
  }
}
```

---

## API Endpoints

| Method | Path | Description |
|--------|------|-------------|
| `GET` | `/api/realtime/connect` | WebSocket upgrade endpoint; proxies to tenant DO |

No other HTTP endpoints. The real-time layer is entirely event-driven.

### Admin visibility

The admin dashboard (`/admin/realtime`) shows:
- Active DO instances (estimated via CF Analytics Engine `DO_REALTIME` metric)
- Connected WebSocket count per tenant (sourced from DO hibernation metrics)
- Queue depth for `zync-realtime`
- Event throughput: events/min by type (past 1h)

This is observability only — no admin actions on real-time connections.

---

## Data Model

No persistent tables for the real-time layer. Events are ephemeral.

Related tables owned by other specs:
- `notifications` (spec 35) — persistent notification records; real-time delivers the push signal
- `audit_log` (spec 28) — does not apply; real-time events are not audited

### KV: connection count (optional, for admin UI)

```
KV key: `rt:active:{tenantId}`
Value: integer (connection count, approximate)
TTL: 300s
```

Updated by the DO on connect/disconnect. Used only for admin UI display — not authoritative for routing.

---

## Reconnect and Missed Events

When a client reconnects after a disconnect, it may have missed events. The strategy per use case:

| Event type | On reconnect |
|-----------|-------------|
| `notification.new` | Fetch `/api/notifications?unread=true` — notifications table has durable state |
| `task.status_changed` | Refetch board data if board is currently open |
| `ticket.message_added` | Refetch ticket messages if ticket is open |
| `activity.new` | Refetch activity feed (last 20 items) |
| `time_entry.started/stopped` | Refetch active timer state |

The client triggers a full-data refetch on reconnect for any module currently in the viewport. This is simpler and more reliable than `Last-Event-ID` replay for ephemeral events.

```ts
// In RealtimeClient
this.ws.addEventListener('open', () => {
  this.reconnectDelay = 1_000
  // Notify app to refetch active views
  this.handlers.get('reconnect')?.forEach(h => h({ type: 'reconnect' } as any))
})
```

### Exponential Backoff with Jitter

Reconnect without jitter causes thundering herd: all clients disconnect simultaneously (server deploy, network blip) and all reconnect at the same intervals. Under heavy tenant load this causes a reconnect storm.

```ts
function reconnectDelay(attempt: number): number {
  const base = Math.min(1000 * 2 ** attempt, 30_000)  // cap at 30s
  const jitter = Math.random() * base * 0.3            // ±30% jitter
  return Math.floor(base + jitter - (base * 0.15))     // symmetric jitter
}

// Reconnect schedule (approx):
// attempt 0:  1.0s ±0.3s
// attempt 1:  2.0s ±0.6s
// attempt 2:  4.0s ±1.2s
// attempt 3:  8.0s ±2.4s
// attempt 4: 16.0s ±4.8s
// attempt 5+: 30.0s ±9.0s (capped)
```

**Invariant:** no reconnect algorithm may use `Math.random() * delay` without the base (pure random collapses to thundering herd at attempt 0). Always start from exponential base.

### Tab Deduplication (BroadcastChannel)

Multiple browser tabs for the same user should not each maintain a WebSocket connection to the server. Use `BroadcastChannel` to elect a single "primary" tab that owns the connection; secondary tabs receive events via the channel.

```ts
const channel = new BroadcastChannel('zync:ws')

// On tab focus / visibility:
channel.postMessage({ type: 'CLAIM_PRIMARY' })
channel.onmessage = (e) => {
  if (e.data.type === 'CLAIM_PRIMARY') {
    // Another tab claimed primary — downgrade to secondary, close WS
    ws?.close()
    ws = null
  }
}

// Primary tab: forward all inbound WS messages to channel
ws.onmessage = (e) => {
  processLocally(e.data)
  channel.postMessage({ type: 'EVENT', payload: e.data })
}

// Secondary tab: receive from primary via channel
channel.onmessage = (e) => {
  if (e.data.type === 'EVENT') processLocally(e.data.payload)
}
```

**Invariant:** at most 1 active WebSocket per user session across all tabs. Primary tab re-elected on tab close (via `beforeunload`) or if heartbeat from primary times out (5s).

---

## Permissions

| Access | Rule |
|--------|------|
| `/api/realtime/connect` | Requires valid session JWT; any authenticated tenant user |
| DO `/connect` | JWT validated in proxy worker; DO trusts X-User-Id header (internal only) |
| DO `/fanout` | Called only by queue consumer Worker; not publicly reachable |
| Admin observability | System admin session |

---

## Cost Estimate

| Component | Unit cost | Estimate at 1,000 tenants |
|-----------|-----------|--------------------------|
| Durable Objects requests | $0.15/million | ~3 connected tabs avg × 60 events/hr × 730hr = ~131M events/mo → ~$20/mo |
| DO compute (WebSocket Hibernation) | $12.50/million GB-s | Hibernating DOs: minimal; active: ~$5/mo |
| DO storage | $0.20/GB-month | Minimal (connection map is in-memory, not storage) |
| Queue messages | $0.40/million | Same as DO events → ~$53/mo |
| **Total estimated** | | **~$80/month at 1,000 tenants** |

WebSocket Hibernation is critical to cost: DOs sleep when no WebSocket messages are flowing, eliminating idle compute charges.

---

## Architecture Decisions

| Decision | Choice | Reason |
|----------|--------|--------|
| WebSocket vs SSE | WebSocket | CF Workers cannot hold open SSE streams without a DO; WebSocket with DO is the native Cloudflare pattern |
| DO per tenant | Yes | Natural isolation; low cost with hibernation; scales with tenant count not connection count |
| DO hibernation API | `ctx.acceptWebSocket` | Zero compute cost when idle; CF manages the WS lifecycle across isolate restarts |
| No persistent event store | Yes | Events are ephemeral; clients refetch on reconnect; notification records are in `notifications` table |
| Queue for event delivery | `zync-realtime` CF Queue | Decouples event publication from DO fanout; handles backpressure; at-least-once delivery |
| Auth via query param | JWT in `?token=` | Browser WebSocket API cannot set custom headers; single-use for handshake only |
| Reconnect strategy | Refetch on open | Simpler than event replay; correct for all event types in Zync (no streaming diff needed) |
| Exponential backoff | 1s → 30s max | Prevents thundering herd on server restart; max 30s acceptable UX for real-time features |
| targetUserId filtering | In DO fanout | Avoids broadcasting sensitive events (e.g. personal notifications) to all tenant tabs |
| No Durable Object for SSE | N/A (chose WS) | Consistent: DO is used for both; SSE option dropped |

---

## Foundation Deltas

### Bindings

| Binding | Type | Purpose | Spec |
|---------|------|---------|------|
| `DO_REALTIME` | Durable Object (`TenantRealtimeDO`) | Per-tenant WebSocket hub | This spec |
| `REALTIME_QUEUE` | Queue (producer) | Publish real-time events from route handlers | This spec |

> `DO_REALTIME` was already listed in the foundation index from specs 5 and 12. This spec is the owner — the binding is defined in `workers/realtime-do/wrangler.toml` and referenced from all other Workers that need to produce events.

### Queues

| Queue | Consumer | Purpose |
|-------|----------|---------|
| `zync-realtime` | `realtime-consumer` Worker | Fan out real-time events to tenant DOs |

### Packages

| Package | Purpose |
|---------|---------|
| `@zync/realtime` | `packages/realtime/` — `publishRealtimeEvent` helper and event type definitions; imported by all Workers that produce events |