Real-Time Sync

Flowstate uses Server-Sent Events (SSE) to push real-time updates from the server to connected clients. This keeps the canvas, rooms, and user presence synchronized across all devices without polling.

Architecture

The sync system uses a three-layer architecture:

Client (Browser/Desktop) ──> SSE Connection ──> Backend API ──> Redis Pub/Sub ──> All Connected Clients

Flow:

1. Client establishes SSE connection to /api/sync/subscribe
2. Server subscribes to user-specific Redis channel
3. When data changes (via API), server publishes event to Redis
4. Redis broadcasts event to all subscribed backend instances
5. Backend pushes event to connected clients via SSE
6. Clients update local state (Zustand stores)

Endpoint

GET /api/sync/subscribe
Accept: text/event-stream

Authentication: Required (cookie-based JWT)

Response Headers:

Content-Type: text/event-stream
Cache-Control: no-cache
Connection: keep-alive
X-Accel-Buffering: no

Connection Lifecycle

1. Establish Connection

const eventSource = new EventSource('/api/sync/subscribe', {
  withCredentials: true, // Include cookies
});

eventSource.onopen = () => {
  console.log('Connected to sync service');
};

2. Receive Events

eventSource.onmessage = (event) => {
  const data = JSON.parse(event.data);
  console.log('Received event:', data);

  // Update local state based on event type
  switch (data.type) {
    case 'spark:created':
      addSparkToStore(data.payload);
      break;
    case 'position:updated':
      updateTilePosition(data.payload);
      break;
    // ... handle other event types
  }
};

3. Handle Errors

eventSource.onerror = (error) => {
  console.error('SSE connection error:', error);

  if (eventSource.readyState === EventSource.CLOSED) {
    console.log('Connection closed, reconnecting...');
    // EventSource automatically reconnects
  }
};

4. Close Connection

eventSource.close();

Event Format

All events follow a consistent structure:

{
  "type": "spark:created",
  "payload": {
    "id": "spark_abc123",
    "title": "New Spark",
    "roomId": "room_xyz789",
    "x": 100,
    "y": 200,
    "createdBy": "user_def456",
    "createdAt": "2026-02-28T10:30:00.000Z"
  },
  "timestamp": "2026-02-28T10:30:00.123Z",
  "userId": "user_def456"
}

Fields:

• type - Event type (see Event Types below)
• payload - Event-specific data
• timestamp - Server timestamp (ISO 8601)
• userId - User who triggered the event (optional)

Event Types

Room Events

Event	Trigger	Payload
room:created	New room created	Full room object
room:updated	Room name/description changed	Updated room object
room:deleted	Room deleted	{ id: string }

Example:

{
  "type": "room:created",
  "payload": {
    "id": "room_abc123",
    "name": "Design Team",
    "description": "Collaborative workspace",
    "isPublic": false,
    "createdBy": "user_xyz789",
    "createdAt": "2026-02-28T10:30:00.000Z"
  },
  "timestamp": "2026-02-28T10:30:00.123Z",
  "userId": "user_xyz789"
}

Spark Events

Event	Trigger	Payload
spark:created	New spark created	Full spark object
spark:updated	Spark title/content changed	Updated spark object
spark:deleted	Spark deleted	{ id: string, roomId: string }

Example:

{
  "type": "spark:updated",
  "payload": {
    "id": "spark_abc123",
    "title": "Updated Title",
    "content": "Updated content",
    "roomId": "room_xyz789",
    "updatedAt": "2026-02-28T10:35:00.000Z"
  },
  "timestamp": "2026-02-28T10:35:00.456Z",
  "userId": "user_def456"
}

Office Events

Event	Trigger	Payload
office:created	New office created	Full office object
office:updated	Office settings changed	Updated office object
office:deleted	Office deleted	{ id: string }
office:member-joined	User joined office	{ officeId: string, userId: string, role: string }
office:member-left	User left office	{ officeId: string, userId: string }

Shelf Events

Event	Trigger	Payload
shelf:created	New shelf created	Full shelf object
shelf:updated	Shelf name/links changed	Updated shelf object
shelf:deleted	Shelf deleted	{ id: string }
shelf:link-added	Link added to shelf	{ shelfId: string, link: object }
shelf:link-removed	Link removed from shelf	{ shelfId: string, linkId: string }

Canvas Events

Event	Trigger	Payload
position:updated	Single tile moved	{ tileId: string, x: number, y: number }
positions:batch-updated	Multiple tiles moved	{ updates: Array<{ tileId, x, y }> }
layout:computed	Auto-layout applied	{ roomId: string, positions: Array<...> }
layout:validated	Layout validation completed	{ roomId: string, valid: boolean, issues: Array<...> }

Example:

{
  "type": "position:updated",
  "payload": {
    "tileId": "spark_abc123",
    "x": 500,
    "y": 600,
    "snapped": true
  },
  "timestamp": "2026-02-28T10:40:00.789Z",
  "userId": "user_xyz789"
}

Cursor Events

Event	Trigger	Payload
cursor:moved	User moved cursor on canvas	{ userId: string, x: number, y: number, roomId: string }
cursor:left	User left canvas	{ userId: string, roomId: string }

Example:

{
  "type": "cursor:moved",
  "payload": {
    "userId": "user_def456",
    "x": 1024,
    "y": 768,
    "roomId": "room_xyz789"
  },
  "timestamp": "2026-02-28T10:45:00.123Z"
}

Heartbeat

The server sends a heartbeat event every 30 seconds to keep the connection alive:

event: heartbeat
data: {"timestamp":"2026-02-28T10:30:00.000Z"}

Clients should monitor heartbeats to detect stale connections:

let lastHeartbeat = Date.now();

eventSource.addEventListener('heartbeat', () => {
  lastHeartbeat = Date.now();
});

// Check for stale connection every 60 seconds
setInterval(() => {
  if (Date.now() - lastHeartbeat > 60000) {
    console.warn('No heartbeat received, connection may be stale');
    eventSource.close();
    // Reconnect logic
  }
}, 60000);

Per-User Channels

Each user subscribes to their own Redis channel:

sync:user:<userId>

This ensures users only receive events relevant to them:

• Rooms they’re a member of
• Offices they belong to
• Sparks in their rooms
• Cursor movements in active rooms

Privacy: Users never receive events for resources they don’t have access to.

Position Sync Flow

Here’s how tile position updates propagate:

1. User drags tile on canvas
   ↓
2. Client calls PATCH /api/canvas/position
   ↓
3. Backend validates and persists to MongoDB
   ↓
4. Backend publishes event to Redis: sync:user:<userId>
   ↓
5. Redis broadcasts to all backend instances
   ↓
6. Each backend instance pushes to connected clients
   ↓
7. Clients receive position:updated event
   ↓
8. Clients update Zustand store (useCanvasStore)
   ↓
9. React re-renders tile at new position

Optimistic Updates: Clients can update local state immediately and rollback if the API call fails.

Graceful Cleanup

When a client disconnects, the server:

1. Unsubscribes from Redis channel
2. Removes connection from active connections map
3. Publishes cursor:left event (if on canvas)
4. Cleans up any temporary state

Server-side cleanup:

req.on('close', () => {
  redis.unsubscribe(`sync:user:${userId}`);
  activeConnections.delete(connectionId);
  publishEvent('cursor:left', { userId, roomId });
});

Error Handling

Connection Failures

EventSource automatically reconnects with exponential backoff. The browser handles this transparently.

Custom reconnection logic:

let reconnectAttempts = 0;
const maxReconnectDelay = 30000; // 30 seconds

function connect() {
  const eventSource = new EventSource('/api/sync/subscribe', {
    withCredentials: true,
  });

  eventSource.onerror = () => {
    reconnectAttempts++;
    const delay = Math.min(1000 * Math.pow(2, reconnectAttempts), maxReconnectDelay);

    console.log(`Reconnecting in ${delay}ms (attempt ${reconnectAttempts})`);

    setTimeout(() => {
      eventSource.close();
      connect();
    }, delay);
  };

  eventSource.onopen = () => {
    reconnectAttempts = 0; // Reset on successful connection
  };

  return eventSource;
}

Authentication Failures

If the session expires, the server closes the connection with a 401 status. Clients should redirect to login:

eventSource.onerror = (error) => {
  if (error.status === 401) {
    console.error('Session expired, redirecting to login');
    window.location.href = '/login';
  }
};

Performance Considerations

Event Batching

For high-frequency events (cursor movements), the server batches updates:

// Batch cursor movements every 100ms
const cursorBatch = [];
setInterval(() => {
  if (cursorBatch.length > 0) {
    publishEvent('cursors:batch-updated', { cursors: cursorBatch });
    cursorBatch.length = 0;
  }
}, 100);

Selective Subscriptions

Clients can filter events client-side to reduce processing:

eventSource.onmessage = (event) => {
  const data = JSON.parse(event.data);

  // Only process events for the active room
  if (data.payload.roomId !== activeRoomId) {
    return;
  }

  handleEvent(data);
};

Connection Limits

The server limits concurrent SSE connections per user to prevent abuse:

• Max connections per user: 5
• Max connections per IP: 20

Exceeding limits returns a 429 status.

Desktop App Integration

The Electron desktop app uses the same SSE endpoint but with Bearer token authentication:

const eventSource = new EventSource('/api/sync/subscribe', {
  headers: {
    Authorization: `Bearer ${accessToken}`,
  },
});

The desktop app maintains a persistent connection and handles token refresh automatically.

Debugging

Enable Verbose Logging

eventSource.onmessage = (event) => {
  console.log('[SSE] Received:', event.data);
  const data = JSON.parse(event.data);
  console.log('[SSE] Parsed:', data);
};

eventSource.onerror = (error) => {
  console.error('[SSE] Error:', error);
  console.log('[SSE] ReadyState:', eventSource.readyState);
};

Monitor Connection State

console.log('EventSource.CONNECTING:', EventSource.CONNECTING); // 0
console.log('EventSource.OPEN:', EventSource.OPEN); // 1
console.log('EventSource.CLOSED:', EventSource.CLOSED); // 2

console.log('Current state:', eventSource.readyState);

Server-Side Logging

The backend logs all published events in development:

[Sync] Publishing event: spark:created to channel: sync:user:user_abc123
[Sync] Event payload: {"id":"spark_def456","title":"New Spark",...}
[Sync] Active connections: 12

Best Practices

1. Always handle reconnection - Network failures are inevitable
2. Validate event payloads - Don’t trust incoming data blindly
3. Use optimistic updates - Update UI immediately, rollback on failure
4. Batch high-frequency events - Cursor movements, scroll positions
5. Close connections on unmount - Prevent memory leaks in React components
6. Monitor heartbeats - Detect stale connections early
7. Filter events client-side - Reduce unnecessary processing
8. Handle authentication errors - Redirect to login on 401
9. Implement exponential backoff - Don’t hammer the server on failures
10. Test offline scenarios - Ensure graceful degradation

Example: React Hook

import { useEffect } from 'react';
import { useCanvasStore } from '@lib/canvasStore';

export function useSyncSubscription() {
  useEffect(() => {
    const eventSource = new EventSource('/api/sync/subscribe', {
      withCredentials: true,
    });

    eventSource.onmessage = (event) => {
      const data = JSON.parse(event.data);

      switch (data.type) {
        case 'spark:created':
          useCanvasStore.getState().addSpark(data.payload);
          break;
        case 'position:updated':
          useCanvasStore.getState().updatePosition(data.payload);
          break;
        // ... handle other events
      }
    };

    eventSource.onerror = (error) => {
      console.error('Sync error:', error);
    };

    return () => {
      eventSource.close();
    };
  }, []);
}

Future Enhancements

• WebSocket support - Bidirectional communication for collaborative editing
• Event replay - Catch up on missed events after reconnection
• Compression - Reduce bandwidth for large payloads
• Selective subscriptions - Subscribe to specific rooms/offices only
• Presence indicators - Show who’s online in real-time