Compare commits
9
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
cd3ecc4507 | ||
|
|
3b19f58e26 | ||
|
|
9769a80381 | ||
|
|
c28b9ac565 | ||
|
|
fa8437d03a | ||
|
|
2d0c3b6df6 | ||
|
|
ec053d8126 | ||
|
|
cd73b31739 | ||
|
|
d2c48c0c9a |
@@ -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
|
||||
@@ -566,6 +566,11 @@ function FlowchartEditorInner({ projectId, projectName, userId, userDisplayName,
|
||||
},
|
||||
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({
|
||||
@@ -577,8 +582,6 @@ function FlowchartEditorInner({ projectId, projectName, userId, userDisplayName,
|
||||
},
|
||||
})
|
||||
.eq('id', projectId)
|
||||
// Notify other clients to refresh after successful auto-persist
|
||||
realtimeRef.current?.broadcastStateRefresh()
|
||||
} catch {
|
||||
// Persistence failure is non-critical; will retry on next change
|
||||
}
|
||||
@@ -602,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) => [
|
||||
@@ -682,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) => ({
|
||||
|
||||
@@ -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 = {
|
||||
@@ -63,6 +63,19 @@ export class CRDTManager {
|
||||
/** 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()
|
||||
}
|
||||
|
||||
/** 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 */
|
||||
|
||||
@@ -42,17 +42,24 @@ type RealtimeCallbacks = {
|
||||
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) {
|
||||
@@ -64,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 } },
|
||||
@@ -143,6 +179,7 @@ export class RealtimeConnection {
|
||||
})
|
||||
.subscribe(async (status) => {
|
||||
if (this.isDestroyed) return
|
||||
this.clearConnectionTimer()
|
||||
|
||||
if (status === 'SUBSCRIBED') {
|
||||
this.reconnectAttempts = 0
|
||||
@@ -162,15 +199,24 @@ export class RealtimeConnection {
|
||||
this.callbacks.onConnectionStateChange('reconnecting')
|
||||
this.scheduleReconnect()
|
||||
} else if (status === 'CLOSED') {
|
||||
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()
|
||||
@@ -181,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
|
||||
}
|
||||
@@ -254,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)
|
||||
}
|
||||
@@ -307,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
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user