Merge pull request 'developing' (#10) from developing into master

Reviewed-on: #10

src/app/editor/[projectId]/FlowchartEditor.tsx

@@ -46,8 +46,8 @@ import type { FlowchartData, FlowchartNode, FlowchartEdge, Character, Variable,
 // LocalStorage key prefix for draft saves
 const DRAFT_KEY_PREFIX = 'vnwrite-draft-'
 
-// Debounce delay in ms
-const AUTOSAVE_DEBOUNCE_MS = 1000
+// Debounce delay for LocalStorage draft saves
+const AUTOSAVE_DEBOUNCE_MS = 5000
 
 type ContextMenuState = {
   x: number
@@ -527,6 +527,12 @@ function FlowchartEditorInner({ projectId, projectName, userId, userDisplayName,
 
   const [characters, setCharacters] = useState<Character[]>(migratedData.characters)
   const [variables, setVariables] = useState<Variable[]>(migratedData.variables)
+
+  // Refs to always have the latest characters/variables for the CRDT persist callback
+  const charactersRef = useRef(characters)
+  const variablesRef = useRef(variables)
+  charactersRef.current = characters
+  variablesRef.current = variables
   const [showSettings, setShowSettings] = useState(false)
   const [showShare, setShowShare] = useState(false)
   const [showHistory, setShowHistory] = useState(false)
@@ -540,7 +546,6 @@ function FlowchartEditorInner({ projectId, projectName, userId, userDisplayName,
   const realtimeRef = useRef<RealtimeConnection | null>(null)
   const crdtRef = useRef<CRDTManager | null>(null)
   const auditRef = useRef<AuditTrailRecorder | null>(null)
-  const isRemoteUpdateRef = useRef(false)
   const cursorThrottleRef = useRef<number>(0)
   const [collaborationNotifications, setCollaborationNotifications] = useState<CollaborationNotification[]>([])
   const [nodeLocks, setNodeLocks] = useState<Map<string, NodeLockInfo>>(new Map())
@@ -554,25 +559,26 @@ function FlowchartEditorInner({ projectId, projectName, userId, userDisplayName,
 
     const crdtManager = new CRDTManager({
       onNodesChange: (crdtNodes: FlowchartNode[]) => {
-        isRemoteUpdateRef.current = true
         setNodes(toReactFlowNodes(crdtNodes))
-        isRemoteUpdateRef.current = false
       },
       onEdgesChange: (crdtEdges: FlowchartEdge[]) => {
-        isRemoteUpdateRef.current = true
         setEdges(toReactFlowEdges(crdtEdges))
-        isRemoteUpdateRef.current = false
       },
       onPersist: async (persistNodes: FlowchartNode[], persistEdges: FlowchartEdge[]) => {
         try {
+          // Auto-persist saves the current state for durability. We do NOT
+          // broadcast state-refresh here because CRDT already syncs nodes/edges
+          // via yjs-update broadcasts. Broadcasting here causes ping-pong:
+          // other clients fetch from DB, overwrite their local variables/characters,
+          // then their persist writes stale data back, causing a loop.
           await supabase
             .from('projects')
             .update({
               flowchart_data: {
                 nodes: persistNodes,
                 edges: persistEdges,
-                characters,
-                variables,
+                characters: charactersRef.current,
+                variables: variablesRef.current,
               },
             })
             .eq('id', projectId)
@@ -599,6 +605,8 @@ function FlowchartEditorInner({ projectId, projectName, userId, userDisplayName,
           ...prev,
           { id: nanoid(), displayName: user.displayName, type: 'join', color: getUserColor(user.userId) },
         ])
+        // Send full CRDT state so the joining client gets caught up
+        crdtManager.broadcastFullState()
       },
       onPresenceLeave: (user) => {
         setCollaborationNotifications((prev) => [
@@ -633,6 +641,29 @@ function FlowchartEditorInner({ projectId, projectName, userId, userDisplayName,
           return next
         })
       },
+      onCRDTUpdate: (update: number[]) => {
+        crdtManager.applyRemoteUpdate(update)
+      },
+      onStateRefresh: async () => {
+        try {
+          const sb = createClient()
+          const { data } = await sb
+            .from('projects')
+            .select('flowchart_data')
+            .eq('id', projectId)
+            .single()
+          if (data?.flowchart_data) {
+            const fd = data.flowchart_data as FlowchartData
+            setNodes(toReactFlowNodes(fd.nodes))
+            setEdges(toReactFlowEdges(fd.edges))
+            setCharacters(fd.characters)
+            setVariables(fd.variables)
+            crdtManager.refreshFromData(fd.nodes, fd.edges)
+          }
+        } catch {
+          // Non-critical: user can still manually refresh
+        }
+      },
       onChannelSubscribed: (channel) => {
         crdtManager.connectChannel(channel)
       },
@@ -656,6 +687,38 @@ function FlowchartEditorInner({ projectId, projectName, userId, userDisplayName,
   // eslint-disable-next-line react-hooks/exhaustive-deps
   }, [projectId, userId, userDisplayName])
 
+  // Manage connection lifecycle based on visibility and user activity
+  useEffect(() => {
+    const handleVisibilityChange = () => {
+      if (!document.hidden) {
+        realtimeRef.current?.notifyVisibilityResumed()
+      }
+    }
+
+    // Throttle activity notifications to avoid excessive calls
+    let activityThrottled = false
+    const throttledActivity = () => {
+      if (activityThrottled) return
+      activityThrottled = true
+      realtimeRef.current?.notifyActivity()
+      setTimeout(() => { activityThrottled = false }, 10_000)
+    }
+
+    document.addEventListener('visibilitychange', handleVisibilityChange)
+    document.addEventListener('mousedown', throttledActivity)
+    document.addEventListener('keydown', throttledActivity)
+    document.addEventListener('scroll', throttledActivity, true)
+    document.addEventListener('mousemove', throttledActivity)
+
+    return () => {
+      document.removeEventListener('visibilitychange', handleVisibilityChange)
+      document.removeEventListener('mousedown', throttledActivity)
+      document.removeEventListener('keydown', throttledActivity)
+      document.removeEventListener('scroll', throttledActivity, true)
+      document.removeEventListener('mousemove', throttledActivity)
+    }
+  }, [])
+
   // Sync local React Flow state changes to CRDT (skip remote-originated updates)
   const nodesForCRDT = useMemo(() => {
     return nodes.map((node) => ({
@@ -678,7 +741,6 @@ function FlowchartEditorInner({ projectId, projectName, userId, userDisplayName,
   }, [edges])
 
   useEffect(() => {
-    if (isRemoteUpdateRef.current) return
     crdtRef.current?.updateNodes(nodesForCRDT)
     if (!isRevertingRef.current) {
       auditRef.current?.recordNodeChanges(nodesForCRDT)
@@ -686,7 +748,6 @@ function FlowchartEditorInner({ projectId, projectName, userId, userDisplayName,
   }, [nodesForCRDT])
 
   useEffect(() => {
-    if (isRemoteUpdateRef.current) return
     crdtRef.current?.updateEdges(edgesForCRDT)
     if (!isRevertingRef.current) {
       auditRef.current?.recordEdgeChanges(edgesForCRDT)
@@ -1141,6 +1202,9 @@ function FlowchartEditorInner({ projectId, projectName, userId, userDisplayName,
       // Update last saved data ref to mark as not dirty
       lastSavedDataRef.current = flowchartData
 
+      // Notify other connected clients to refresh from the database
+      realtimeRef.current?.broadcastStateRefresh()
+
       setToast({ message: 'Project saved successfully', type: 'success' })
     } catch (error) {
       console.error('Failed to save project:', error)

src/lib/collaboration/crdt.ts

@@ -2,7 +2,7 @@ import * as Y from 'yjs'
 import type { RealtimeChannel } from '@supabase/supabase-js'
 import type { FlowchartNode, FlowchartEdge } from '@/types/flowchart'
 
-const PERSIST_DEBOUNCE_MS = 2000
+const PERSIST_DEBOUNCE_MS = 30_000 // 30s - DB persist is for crash recovery only; CRDT handles real-time sync
 const BROADCAST_EVENT = 'yjs-update'
 
 export type CRDTCallbacks = {
@@ -19,6 +19,7 @@ export class CRDTManager {
   private callbacks: CRDTCallbacks
   private persistTimer: ReturnType<typeof setTimeout> | null = null
   private isApplyingRemote = false
+  private isSuppressed = false // suppress broadcast/persist during init or refresh
   private isDestroyed = false
 
   constructor(callbacks: CRDTCallbacks) {
@@ -29,24 +30,25 @@ export class CRDTManager {
 
     // Schedule persistence on local Yjs document changes
     this.nodesMap.observe(() => {
-      if (this.isApplyingRemote) return
+      if (this.isApplyingRemote || this.isSuppressed) return
       this.schedulePersist()
     })
 
     this.edgesMap.observe(() => {
-      if (this.isApplyingRemote) return
+      if (this.isApplyingRemote || this.isSuppressed) return
       this.schedulePersist()
     })
 
     // Broadcast local updates to other clients
     this.doc.on('update', (update: Uint8Array, origin: unknown) => {
-      if (origin === 'remote') return // Don't re-broadcast remote updates
+      if (origin === 'remote' || this.isSuppressed) return
       this.broadcastUpdate(update)
     })
   }
 
   /** Initialize the Yjs document from database state */
   initializeFromData(nodes: FlowchartNode[], edges: FlowchartEdge[]): void {
+    this.isSuppressed = true
     this.doc.transact(() => {
       nodes.forEach((node) => {
         this.nodesMap.set(node.id, JSON.stringify(node))
@@ -55,29 +57,72 @@ export class CRDTManager {
         this.edgesMap.set(edge.id, JSON.stringify(edge))
       })
     }, 'init')
+    this.isSuppressed = false
   }
 
-  /** Connect to a Supabase Realtime channel for syncing updates */
+  /** Connect to a Supabase Realtime channel for outbound broadcasts */
   connectChannel(channel: RealtimeChannel): void {
     this.channel = channel
+    // Broadcast full state so other clients merge any updates they missed
+    this.broadcastFullState()
+  }
 
-    // Listen for broadcast updates from other clients
-    channel.on('broadcast', { event: BROADCAST_EVENT }, (payload) => {
-      if (this.isDestroyed) return
-      const data = payload.payload as { update?: number[] } | undefined
-      if (data?.update) {
-        const update = new Uint8Array(data.update)
-        this.isApplyingRemote = true
-        Y.applyUpdate(this.doc, update, 'remote')
-        this.isApplyingRemote = false
-        // Notify React state of remote changes
-        this.notifyNodesChange()
-        this.notifyEdgesChange()
-        this.schedulePersist()
-      }
+  /** Broadcast the full Yjs document state to sync all connected clients */
+  broadcastFullState(): void {
+    if (!this.channel || this.isDestroyed) return
+    const state = Y.encodeStateAsUpdate(this.doc)
+    this.channel.send({
+      type: 'broadcast',
+      event: BROADCAST_EVENT,
+      payload: { update: Array.from(state) },
     })
   }
 
+  /** Apply a remote CRDT update received via broadcast */
+  applyRemoteUpdate(updateData: number[]): void {
+    if (this.isDestroyed) return
+    const update = new Uint8Array(updateData)
+    this.isApplyingRemote = true
+    Y.applyUpdate(this.doc, update, 'remote')
+    this.isApplyingRemote = false
+    // Notify React state of remote changes
+    this.notifyNodesChange()
+    this.notifyEdgesChange()
+    // Note: we do NOT schedulePersist here. Only the originating client
+    // persists its own changes to avoid write races and stale data overwrites.
+  }
+
+  /** Replace CRDT state from a database refresh without broadcasting */
+  refreshFromData(nodes: FlowchartNode[], edges: FlowchartEdge[]): void {
+    this.isSuppressed = true
+    this.doc.transact(() => {
+      // Sync nodes
+      const nodeIds = new Set(nodes.map((n) => n.id))
+      Array.from(this.nodesMap.keys()).forEach((id) => {
+        if (!nodeIds.has(id)) this.nodesMap.delete(id)
+      })
+      nodes.forEach((node) => {
+        const serialized = JSON.stringify(node)
+        if (this.nodesMap.get(node.id) !== serialized) {
+          this.nodesMap.set(node.id, serialized)
+        }
+      })
+
+      // Sync edges
+      const edgeIds = new Set(edges.map((e) => e.id))
+      Array.from(this.edgesMap.keys()).forEach((id) => {
+        if (!edgeIds.has(id)) this.edgesMap.delete(id)
+      })
+      edges.forEach((edge) => {
+        const serialized = JSON.stringify(edge)
+        if (this.edgesMap.get(edge.id) !== serialized) {
+          this.edgesMap.set(edge.id, serialized)
+        }
+      })
+    }, 'remote')
+    this.isSuppressed = false
+  }
+
   /** Apply local node changes to the Yjs document */
   updateNodes(nodes: FlowchartNode[]): void {
     if (this.isApplyingRemote) return

src/lib/collaboration/realtime.ts

@@ -35,22 +35,31 @@ type RealtimeCallbacks = {
   onChannelSubscribed?: (channel: RealtimeChannel) => void
   onCursorUpdate?: (cursor: RemoteCursor) => void
   onNodeLockUpdate?: (lock: NodeLock | null, userId: string) => void
+  onCRDTUpdate?: (update: number[]) => void
+  onStateRefresh?: () => void
 }
 
 const HEARTBEAT_INTERVAL_MS = 30_000
 const RECONNECT_BASE_DELAY_MS = 1000
 const RECONNECT_MAX_DELAY_MS = 30_000
+const INACTIVITY_TIMEOUT_MS = 5 * 60_000 // 5 minutes of inactivity before pausing
+const CONNECTION_TIMEOUT_MS = 15_000 // 15s timeout for initial connection
+const STALE_THRESHOLD_MS = 3 * 60_000 // 3 minutes hidden before forcing fresh reconnect
 
 export class RealtimeConnection {
   private channel: RealtimeChannel | null = null
   private heartbeatTimer: ReturnType<typeof setInterval> | null = null
   private reconnectTimer: ReturnType<typeof setTimeout> | null = null
+  private inactivityTimer: ReturnType<typeof setTimeout> | null = null
+  private connectionTimer: ReturnType<typeof setTimeout> | null = null
   private reconnectAttempts = 0
   private projectId: string
   private userId: string
   private displayName: string
   private callbacks: RealtimeCallbacks
   private isDestroyed = false
+  private isPaused = false
+  private lastActivityTime = Date.now()
   private supabase = createClient()
 
   constructor(projectId: string, userId: string, displayName: string, callbacks: RealtimeCallbacks) {
@@ -62,7 +71,36 @@ export class RealtimeConnection {
 
   async connect(): Promise<void> {
     if (this.isDestroyed) return
+    this.isPaused = false
+    this.lastActivityTime = Date.now()
     this.callbacks.onConnectionStateChange('connecting')
+    this.resetInactivityTimer()
+    this.clearConnectionTimer()
+
+    // Ensure the Supabase client has a valid auth session before connecting.
+    // On initial page load, the session may still be loading from cookies.
+    try {
+      const { data: { session } } = await this.supabase.auth.getSession()
+      if (!session && !this.isDestroyed) {
+        // No session yet - wait briefly and retry
+        this.scheduleReconnect()
+        return
+      }
+    } catch {
+      // Session check failed - proceed anyway, channel will handle auth errors
+    }
+
+    if (this.isDestroyed || this.isPaused) return
+
+    // Set a timeout: if we don't connect within CONNECTION_TIMEOUT_MS, retry
+    this.connectionTimer = setTimeout(() => {
+      if (this.isDestroyed || this.isPaused) return
+      if (this.channel) {
+        this.supabase.removeChannel(this.channel)
+        this.channel = null
+      }
+      this.scheduleReconnect()
+    }, CONNECTION_TIMEOUT_MS)
 
     this.channel = this.supabase.channel(`project:${this.projectId}`, {
       config: { presence: { key: this.userId } },
@@ -131,8 +169,17 @@ export class RealtimeConnection {
           this.callbacks.onNodeLockUpdate?.(null, payload.userId)
         }
       })
+      .on('broadcast', { event: 'yjs-update' }, ({ payload }) => {
+        if (payload?.update) {
+          this.callbacks.onCRDTUpdate?.(payload.update)
+        }
+      })
+      .on('broadcast', { event: 'state-refresh' }, () => {
+        this.callbacks.onStateRefresh?.()
+      })
       .subscribe(async (status) => {
         if (this.isDestroyed) return
+        this.clearConnectionTimer()
 
         if (status === 'SUBSCRIBED') {
           this.reconnectAttempts = 0
@@ -152,15 +199,24 @@ export class RealtimeConnection {
           this.callbacks.onConnectionStateChange('reconnecting')
           this.scheduleReconnect()
         } else if (status === 'CLOSED') {
-          this.callbacks.onConnectionStateChange('disconnected')
+          if (!this.isPaused) {
+            // Unexpected close - attempt to reconnect
+            this.callbacks.onConnectionStateChange('reconnecting')
+            this.scheduleReconnect()
+          } else {
+            this.callbacks.onConnectionStateChange('disconnected')
+          }
         }
       })
   }
 
   async disconnect(): Promise<void> {
     this.isDestroyed = true
+    this.isPaused = false
     this.stopHeartbeat()
     this.clearReconnectTimer()
+    this.clearInactivityTimer()
+    this.clearConnectionTimer()
 
     if (this.channel) {
       await this.deleteSession()
@@ -171,6 +227,92 @@ export class RealtimeConnection {
     this.callbacks.onConnectionStateChange('disconnected')
   }
 
+  /**
+   * Pause the connection (e.g. on inactivity or tab hidden).
+   * Unlike disconnect(), this allows resuming later.
+   */
+  async pause(): Promise<void> {
+    if (this.isDestroyed || this.isPaused) return
+    this.isPaused = true
+    this.stopHeartbeat()
+    this.clearReconnectTimer()
+    this.clearInactivityTimer()
+    this.clearConnectionTimer()
+
+    if (this.channel) {
+      await this.deleteSession()
+      this.supabase.removeChannel(this.channel)
+      this.channel = null
+    }
+
+    this.callbacks.onConnectionStateChange('disconnected')
+  }
+
+  /**
+   * Resume the connection after it was paused.
+   * Re-establishes the channel and presence.
+   */
+  async resume(): Promise<void> {
+    if (this.isDestroyed || !this.isPaused) return
+    this.isPaused = false
+    this.reconnectAttempts = 0
+    await this.connect()
+  }
+
+  /**
+   * Notify that user activity has occurred, resetting the inactivity timer.
+   * If the connection was paused due to inactivity, it will resume.
+   */
+  notifyActivity(): void {
+    if (this.isDestroyed) return
+    this.lastActivityTime = Date.now()
+    this.resetInactivityTimer()
+
+    if (this.isPaused) {
+      this.resume()
+    }
+  }
+
+  /**
+   * Called when the tab becomes visible again. If the tab was hidden for a
+   * significant period, force a fresh reconnect to handle stale WebSockets.
+   */
+  notifyVisibilityResumed(): void {
+    if (this.isDestroyed) return
+    const now = Date.now()
+    const hiddenDuration = now - this.lastActivityTime
+    this.lastActivityTime = now
+    this.resetInactivityTimer()
+
+    if (this.isPaused) {
+      this.resume()
+    } else if (hiddenDuration > STALE_THRESHOLD_MS && this.channel) {
+      this.forceReconnect()
+    }
+  }
+
+  /**
+   * Force a fresh reconnect by tearing down the current channel and reconnecting.
+   */
+  private async forceReconnect(): Promise<void> {
+    if (this.isDestroyed || this.isPaused) return
+    this.stopHeartbeat()
+    this.clearReconnectTimer()
+    this.clearConnectionTimer()
+
+    if (this.channel) {
+      this.supabase.removeChannel(this.channel)
+      this.channel = null
+    }
+
+    this.reconnectAttempts = 0
+    await this.connect()
+  }
+
+  getIsPaused(): boolean {
+    return this.isPaused
+  }
+
   getChannel(): RealtimeChannel | null {
     return this.channel
   }
@@ -189,6 +331,15 @@ export class RealtimeConnection {
     })
   }
 
+  broadcastStateRefresh(): void {
+    if (!this.channel) return
+    this.channel.send({
+      type: 'broadcast',
+      event: 'state-refresh',
+      payload: {},
+    })
+  }
+
   broadcastNodeLock(nodeId: string | null): void {
     if (!this.channel) return
     this.channel.send({
@@ -235,19 +386,44 @@ export class RealtimeConnection {
 
   private startHeartbeat(): void {
     this.stopHeartbeat()
+    let consecutiveFailures = 0
     this.heartbeatTimer = setInterval(async () => {
-      if (this.isDestroyed) {
+      if (this.isDestroyed || this.isPaused) {
         this.stopHeartbeat()
         return
       }
+
+      // Check if the channel is still in a healthy state
+      if (this.channel) {
+        const state = (this.channel as unknown as { state?: string }).state
+        if (state && state !== 'joined' && state !== 'joining') {
+          // Channel is in an unhealthy state - trigger reconnect
+          this.callbacks.onConnectionStateChange('reconnecting')
+          this.supabase.removeChannel(this.channel)
+          this.channel = null
+          this.stopHeartbeat()
+          this.scheduleReconnect()
+          return
+        }
+      }
+
       try {
         await this.supabase
           .from('collaboration_sessions')
           .update({ last_heartbeat: new Date().toISOString() })
           .eq('project_id', this.projectId)
           .eq('user_id', this.userId)
+        consecutiveFailures = 0
       } catch {
-        // Heartbeat failure is non-critical
+        consecutiveFailures++
+        // If heartbeat fails 3 times in a row, the connection is likely dead
+        if (consecutiveFailures >= 3 && this.channel) {
+          this.callbacks.onConnectionStateChange('reconnecting')
+          this.supabase.removeChannel(this.channel)
+          this.channel = null
+          this.stopHeartbeat()
+          this.scheduleReconnect()
+        }
       }
     }, HEARTBEAT_INTERVAL_MS)
   }
@@ -288,4 +464,28 @@ export class RealtimeConnection {
       this.reconnectTimer = null
     }
   }
+
+  private resetInactivityTimer(): void {
+    this.clearInactivityTimer()
+    if (this.isDestroyed) return
+    this.inactivityTimer = setTimeout(() => {
+      if (!this.isDestroyed && !this.isPaused) {
+        this.pause()
+      }
+    }, INACTIVITY_TIMEOUT_MS)
+  }
+
+  private clearInactivityTimer(): void {
+    if (this.inactivityTimer) {
+      clearTimeout(this.inactivityTimer)
+      this.inactivityTimer = null
+    }
+  }
+
+  private clearConnectionTimer(): void {
+    if (this.connectionTimer) {
+      clearTimeout(this.connectionTimer)
+      this.connectionTimer = null
+    }
+  }
 }