Compare commits
12
Commits
@@ -46,8 +46,8 @@ import type { FlowchartData, FlowchartNode, FlowchartEdge, Character, Variable,
|
|||||||
// LocalStorage key prefix for draft saves
|
// LocalStorage key prefix for draft saves
|
||||||
const DRAFT_KEY_PREFIX = 'vnwrite-draft-'
|
const DRAFT_KEY_PREFIX = 'vnwrite-draft-'
|
||||||
|
|
||||||
// Debounce delay in ms
|
// Debounce delay for LocalStorage draft saves
|
||||||
const AUTOSAVE_DEBOUNCE_MS = 1000
|
const AUTOSAVE_DEBOUNCE_MS = 5000
|
||||||
|
|
||||||
type ContextMenuState = {
|
type ContextMenuState = {
|
||||||
x: number
|
x: number
|
||||||
@@ -527,6 +527,12 @@ function FlowchartEditorInner({ projectId, projectName, userId, userDisplayName,
|
|||||||
|
|
||||||
const [characters, setCharacters] = useState<Character[]>(migratedData.characters)
|
const [characters, setCharacters] = useState<Character[]>(migratedData.characters)
|
||||||
const [variables, setVariables] = useState<Variable[]>(migratedData.variables)
|
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 [showSettings, setShowSettings] = useState(false)
|
||||||
const [showShare, setShowShare] = useState(false)
|
const [showShare, setShowShare] = useState(false)
|
||||||
const [showHistory, setShowHistory] = useState(false)
|
const [showHistory, setShowHistory] = useState(false)
|
||||||
@@ -540,7 +546,6 @@ function FlowchartEditorInner({ projectId, projectName, userId, userDisplayName,
|
|||||||
const realtimeRef = useRef<RealtimeConnection | null>(null)
|
const realtimeRef = useRef<RealtimeConnection | null>(null)
|
||||||
const crdtRef = useRef<CRDTManager | null>(null)
|
const crdtRef = useRef<CRDTManager | null>(null)
|
||||||
const auditRef = useRef<AuditTrailRecorder | null>(null)
|
const auditRef = useRef<AuditTrailRecorder | null>(null)
|
||||||
const isRemoteUpdateRef = useRef(false)
|
|
||||||
const cursorThrottleRef = useRef<number>(0)
|
const cursorThrottleRef = useRef<number>(0)
|
||||||
const [collaborationNotifications, setCollaborationNotifications] = useState<CollaborationNotification[]>([])
|
const [collaborationNotifications, setCollaborationNotifications] = useState<CollaborationNotification[]>([])
|
||||||
const [nodeLocks, setNodeLocks] = useState<Map<string, NodeLockInfo>>(new Map())
|
const [nodeLocks, setNodeLocks] = useState<Map<string, NodeLockInfo>>(new Map())
|
||||||
@@ -554,25 +559,26 @@ function FlowchartEditorInner({ projectId, projectName, userId, userDisplayName,
|
|||||||
|
|
||||||
const crdtManager = new CRDTManager({
|
const crdtManager = new CRDTManager({
|
||||||
onNodesChange: (crdtNodes: FlowchartNode[]) => {
|
onNodesChange: (crdtNodes: FlowchartNode[]) => {
|
||||||
isRemoteUpdateRef.current = true
|
|
||||||
setNodes(toReactFlowNodes(crdtNodes))
|
setNodes(toReactFlowNodes(crdtNodes))
|
||||||
isRemoteUpdateRef.current = false
|
|
||||||
},
|
},
|
||||||
onEdgesChange: (crdtEdges: FlowchartEdge[]) => {
|
onEdgesChange: (crdtEdges: FlowchartEdge[]) => {
|
||||||
isRemoteUpdateRef.current = true
|
|
||||||
setEdges(toReactFlowEdges(crdtEdges))
|
setEdges(toReactFlowEdges(crdtEdges))
|
||||||
isRemoteUpdateRef.current = false
|
|
||||||
},
|
},
|
||||||
onPersist: async (persistNodes: FlowchartNode[], persistEdges: FlowchartEdge[]) => {
|
onPersist: async (persistNodes: FlowchartNode[], persistEdges: FlowchartEdge[]) => {
|
||||||
try {
|
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
|
await supabase
|
||||||
.from('projects')
|
.from('projects')
|
||||||
.update({
|
.update({
|
||||||
flowchart_data: {
|
flowchart_data: {
|
||||||
nodes: persistNodes,
|
nodes: persistNodes,
|
||||||
edges: persistEdges,
|
edges: persistEdges,
|
||||||
characters,
|
characters: charactersRef.current,
|
||||||
variables,
|
variables: variablesRef.current,
|
||||||
},
|
},
|
||||||
})
|
})
|
||||||
.eq('id', projectId)
|
.eq('id', projectId)
|
||||||
@@ -599,6 +605,8 @@ function FlowchartEditorInner({ projectId, projectName, userId, userDisplayName,
|
|||||||
...prev,
|
...prev,
|
||||||
{ id: nanoid(), displayName: user.displayName, type: 'join', color: getUserColor(user.userId) },
|
{ 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) => {
|
onPresenceLeave: (user) => {
|
||||||
setCollaborationNotifications((prev) => [
|
setCollaborationNotifications((prev) => [
|
||||||
@@ -633,6 +641,29 @@ function FlowchartEditorInner({ projectId, projectName, userId, userDisplayName,
|
|||||||
return next
|
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) => {
|
onChannelSubscribed: (channel) => {
|
||||||
crdtManager.connectChannel(channel)
|
crdtManager.connectChannel(channel)
|
||||||
},
|
},
|
||||||
@@ -656,6 +687,38 @@ function FlowchartEditorInner({ projectId, projectName, userId, userDisplayName,
|
|||||||
// eslint-disable-next-line react-hooks/exhaustive-deps
|
// eslint-disable-next-line react-hooks/exhaustive-deps
|
||||||
}, [projectId, userId, userDisplayName])
|
}, [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)
|
// Sync local React Flow state changes to CRDT (skip remote-originated updates)
|
||||||
const nodesForCRDT = useMemo(() => {
|
const nodesForCRDT = useMemo(() => {
|
||||||
return nodes.map((node) => ({
|
return nodes.map((node) => ({
|
||||||
@@ -678,7 +741,6 @@ function FlowchartEditorInner({ projectId, projectName, userId, userDisplayName,
|
|||||||
}, [edges])
|
}, [edges])
|
||||||
|
|
||||||
useEffect(() => {
|
useEffect(() => {
|
||||||
if (isRemoteUpdateRef.current) return
|
|
||||||
crdtRef.current?.updateNodes(nodesForCRDT)
|
crdtRef.current?.updateNodes(nodesForCRDT)
|
||||||
if (!isRevertingRef.current) {
|
if (!isRevertingRef.current) {
|
||||||
auditRef.current?.recordNodeChanges(nodesForCRDT)
|
auditRef.current?.recordNodeChanges(nodesForCRDT)
|
||||||
@@ -686,7 +748,6 @@ function FlowchartEditorInner({ projectId, projectName, userId, userDisplayName,
|
|||||||
}, [nodesForCRDT])
|
}, [nodesForCRDT])
|
||||||
|
|
||||||
useEffect(() => {
|
useEffect(() => {
|
||||||
if (isRemoteUpdateRef.current) return
|
|
||||||
crdtRef.current?.updateEdges(edgesForCRDT)
|
crdtRef.current?.updateEdges(edgesForCRDT)
|
||||||
if (!isRevertingRef.current) {
|
if (!isRevertingRef.current) {
|
||||||
auditRef.current?.recordEdgeChanges(edgesForCRDT)
|
auditRef.current?.recordEdgeChanges(edgesForCRDT)
|
||||||
@@ -1141,6 +1202,9 @@ function FlowchartEditorInner({ projectId, projectName, userId, userDisplayName,
|
|||||||
// Update last saved data ref to mark as not dirty
|
// Update last saved data ref to mark as not dirty
|
||||||
lastSavedDataRef.current = flowchartData
|
lastSavedDataRef.current = flowchartData
|
||||||
|
|
||||||
|
// Notify other connected clients to refresh from the database
|
||||||
|
realtimeRef.current?.broadcastStateRefresh()
|
||||||
|
|
||||||
setToast({ message: 'Project saved successfully', type: 'success' })
|
setToast({ message: 'Project saved successfully', type: 'success' })
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
console.error('Failed to save project:', error)
|
console.error('Failed to save project:', error)
|
||||||
|
|||||||
@@ -2,7 +2,7 @@ import * as Y from 'yjs'
|
|||||||
import type { RealtimeChannel } from '@supabase/supabase-js'
|
import type { RealtimeChannel } from '@supabase/supabase-js'
|
||||||
import type { FlowchartNode, FlowchartEdge } from '@/types/flowchart'
|
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'
|
const BROADCAST_EVENT = 'yjs-update'
|
||||||
|
|
||||||
export type CRDTCallbacks = {
|
export type CRDTCallbacks = {
|
||||||
@@ -19,6 +19,7 @@ export class CRDTManager {
|
|||||||
private callbacks: CRDTCallbacks
|
private callbacks: CRDTCallbacks
|
||||||
private persistTimer: ReturnType<typeof setTimeout> | null = null
|
private persistTimer: ReturnType<typeof setTimeout> | null = null
|
||||||
private isApplyingRemote = false
|
private isApplyingRemote = false
|
||||||
|
private isSuppressed = false // suppress broadcast/persist during init or refresh
|
||||||
private isDestroyed = false
|
private isDestroyed = false
|
||||||
|
|
||||||
constructor(callbacks: CRDTCallbacks) {
|
constructor(callbacks: CRDTCallbacks) {
|
||||||
@@ -29,24 +30,25 @@ export class CRDTManager {
|
|||||||
|
|
||||||
// Schedule persistence on local Yjs document changes
|
// Schedule persistence on local Yjs document changes
|
||||||
this.nodesMap.observe(() => {
|
this.nodesMap.observe(() => {
|
||||||
if (this.isApplyingRemote) return
|
if (this.isApplyingRemote || this.isSuppressed) return
|
||||||
this.schedulePersist()
|
this.schedulePersist()
|
||||||
})
|
})
|
||||||
|
|
||||||
this.edgesMap.observe(() => {
|
this.edgesMap.observe(() => {
|
||||||
if (this.isApplyingRemote) return
|
if (this.isApplyingRemote || this.isSuppressed) return
|
||||||
this.schedulePersist()
|
this.schedulePersist()
|
||||||
})
|
})
|
||||||
|
|
||||||
// Broadcast local updates to other clients
|
// Broadcast local updates to other clients
|
||||||
this.doc.on('update', (update: Uint8Array, origin: unknown) => {
|
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)
|
this.broadcastUpdate(update)
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
/** Initialize the Yjs document from database state */
|
/** Initialize the Yjs document from database state */
|
||||||
initializeFromData(nodes: FlowchartNode[], edges: FlowchartEdge[]): void {
|
initializeFromData(nodes: FlowchartNode[], edges: FlowchartEdge[]): void {
|
||||||
|
this.isSuppressed = true
|
||||||
this.doc.transact(() => {
|
this.doc.transact(() => {
|
||||||
nodes.forEach((node) => {
|
nodes.forEach((node) => {
|
||||||
this.nodesMap.set(node.id, JSON.stringify(node))
|
this.nodesMap.set(node.id, JSON.stringify(node))
|
||||||
@@ -55,27 +57,70 @@ export class CRDTManager {
|
|||||||
this.edgesMap.set(edge.id, JSON.stringify(edge))
|
this.edgesMap.set(edge.id, JSON.stringify(edge))
|
||||||
})
|
})
|
||||||
}, 'init')
|
}, '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 {
|
connectChannel(channel: RealtimeChannel): void {
|
||||||
this.channel = channel
|
this.channel = channel
|
||||||
|
// Broadcast full state so other clients merge any updates they missed
|
||||||
|
this.broadcastFullState()
|
||||||
|
}
|
||||||
|
|
||||||
// Listen for broadcast updates from other clients
|
/** Broadcast the full Yjs document state to sync all connected clients */
|
||||||
channel.on('broadcast', { event: BROADCAST_EVENT }, (payload) => {
|
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
|
if (this.isDestroyed) return
|
||||||
const data = payload.payload as { update?: number[] } | undefined
|
const update = new Uint8Array(updateData)
|
||||||
if (data?.update) {
|
|
||||||
const update = new Uint8Array(data.update)
|
|
||||||
this.isApplyingRemote = true
|
this.isApplyingRemote = true
|
||||||
Y.applyUpdate(this.doc, update, 'remote')
|
Y.applyUpdate(this.doc, update, 'remote')
|
||||||
this.isApplyingRemote = false
|
this.isApplyingRemote = false
|
||||||
// Notify React state of remote changes
|
// Notify React state of remote changes
|
||||||
this.notifyNodesChange()
|
this.notifyNodesChange()
|
||||||
this.notifyEdgesChange()
|
this.notifyEdgesChange()
|
||||||
this.schedulePersist()
|
// 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 */
|
/** Apply local node changes to the Yjs document */
|
||||||
|
|||||||
@@ -35,22 +35,31 @@ type RealtimeCallbacks = {
|
|||||||
onChannelSubscribed?: (channel: RealtimeChannel) => void
|
onChannelSubscribed?: (channel: RealtimeChannel) => void
|
||||||
onCursorUpdate?: (cursor: RemoteCursor) => void
|
onCursorUpdate?: (cursor: RemoteCursor) => void
|
||||||
onNodeLockUpdate?: (lock: NodeLock | null, userId: string) => void
|
onNodeLockUpdate?: (lock: NodeLock | null, userId: string) => void
|
||||||
|
onCRDTUpdate?: (update: number[]) => void
|
||||||
|
onStateRefresh?: () => void
|
||||||
}
|
}
|
||||||
|
|
||||||
const HEARTBEAT_INTERVAL_MS = 30_000
|
const HEARTBEAT_INTERVAL_MS = 30_000
|
||||||
const RECONNECT_BASE_DELAY_MS = 1000
|
const RECONNECT_BASE_DELAY_MS = 1000
|
||||||
const RECONNECT_MAX_DELAY_MS = 30_000
|
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 {
|
export class RealtimeConnection {
|
||||||
private channel: RealtimeChannel | null = null
|
private channel: RealtimeChannel | null = null
|
||||||
private heartbeatTimer: ReturnType<typeof setInterval> | null = null
|
private heartbeatTimer: ReturnType<typeof setInterval> | null = null
|
||||||
private reconnectTimer: ReturnType<typeof setTimeout> | 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 reconnectAttempts = 0
|
||||||
private projectId: string
|
private projectId: string
|
||||||
private userId: string
|
private userId: string
|
||||||
private displayName: string
|
private displayName: string
|
||||||
private callbacks: RealtimeCallbacks
|
private callbacks: RealtimeCallbacks
|
||||||
private isDestroyed = false
|
private isDestroyed = false
|
||||||
|
private isPaused = false
|
||||||
|
private lastActivityTime = Date.now()
|
||||||
private supabase = createClient()
|
private supabase = createClient()
|
||||||
|
|
||||||
constructor(projectId: string, userId: string, displayName: string, callbacks: RealtimeCallbacks) {
|
constructor(projectId: string, userId: string, displayName: string, callbacks: RealtimeCallbacks) {
|
||||||
@@ -62,7 +71,36 @@ export class RealtimeConnection {
|
|||||||
|
|
||||||
async connect(): Promise<void> {
|
async connect(): Promise<void> {
|
||||||
if (this.isDestroyed) return
|
if (this.isDestroyed) return
|
||||||
|
this.isPaused = false
|
||||||
|
this.lastActivityTime = Date.now()
|
||||||
this.callbacks.onConnectionStateChange('connecting')
|
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}`, {
|
this.channel = this.supabase.channel(`project:${this.projectId}`, {
|
||||||
config: { presence: { key: this.userId } },
|
config: { presence: { key: this.userId } },
|
||||||
@@ -131,8 +169,17 @@ export class RealtimeConnection {
|
|||||||
this.callbacks.onNodeLockUpdate?.(null, payload.userId)
|
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) => {
|
.subscribe(async (status) => {
|
||||||
if (this.isDestroyed) return
|
if (this.isDestroyed) return
|
||||||
|
this.clearConnectionTimer()
|
||||||
|
|
||||||
if (status === 'SUBSCRIBED') {
|
if (status === 'SUBSCRIBED') {
|
||||||
this.reconnectAttempts = 0
|
this.reconnectAttempts = 0
|
||||||
@@ -152,15 +199,24 @@ export class RealtimeConnection {
|
|||||||
this.callbacks.onConnectionStateChange('reconnecting')
|
this.callbacks.onConnectionStateChange('reconnecting')
|
||||||
this.scheduleReconnect()
|
this.scheduleReconnect()
|
||||||
} else if (status === 'CLOSED') {
|
} else if (status === 'CLOSED') {
|
||||||
|
if (!this.isPaused) {
|
||||||
|
// Unexpected close - attempt to reconnect
|
||||||
|
this.callbacks.onConnectionStateChange('reconnecting')
|
||||||
|
this.scheduleReconnect()
|
||||||
|
} else {
|
||||||
this.callbacks.onConnectionStateChange('disconnected')
|
this.callbacks.onConnectionStateChange('disconnected')
|
||||||
}
|
}
|
||||||
|
}
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
async disconnect(): Promise<void> {
|
async disconnect(): Promise<void> {
|
||||||
this.isDestroyed = true
|
this.isDestroyed = true
|
||||||
|
this.isPaused = false
|
||||||
this.stopHeartbeat()
|
this.stopHeartbeat()
|
||||||
this.clearReconnectTimer()
|
this.clearReconnectTimer()
|
||||||
|
this.clearInactivityTimer()
|
||||||
|
this.clearConnectionTimer()
|
||||||
|
|
||||||
if (this.channel) {
|
if (this.channel) {
|
||||||
await this.deleteSession()
|
await this.deleteSession()
|
||||||
@@ -171,6 +227,92 @@ export class RealtimeConnection {
|
|||||||
this.callbacks.onConnectionStateChange('disconnected')
|
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 {
|
getChannel(): RealtimeChannel | null {
|
||||||
return this.channel
|
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 {
|
broadcastNodeLock(nodeId: string | null): void {
|
||||||
if (!this.channel) return
|
if (!this.channel) return
|
||||||
this.channel.send({
|
this.channel.send({
|
||||||
@@ -235,19 +386,44 @@ export class RealtimeConnection {
|
|||||||
|
|
||||||
private startHeartbeat(): void {
|
private startHeartbeat(): void {
|
||||||
this.stopHeartbeat()
|
this.stopHeartbeat()
|
||||||
|
let consecutiveFailures = 0
|
||||||
this.heartbeatTimer = setInterval(async () => {
|
this.heartbeatTimer = setInterval(async () => {
|
||||||
if (this.isDestroyed) {
|
if (this.isDestroyed || this.isPaused) {
|
||||||
this.stopHeartbeat()
|
this.stopHeartbeat()
|
||||||
return
|
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 {
|
try {
|
||||||
await this.supabase
|
await this.supabase
|
||||||
.from('collaboration_sessions')
|
.from('collaboration_sessions')
|
||||||
.update({ last_heartbeat: new Date().toISOString() })
|
.update({ last_heartbeat: new Date().toISOString() })
|
||||||
.eq('project_id', this.projectId)
|
.eq('project_id', this.projectId)
|
||||||
.eq('user_id', this.userId)
|
.eq('user_id', this.userId)
|
||||||
|
consecutiveFailures = 0
|
||||||
} catch {
|
} 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)
|
}, HEARTBEAT_INTERVAL_MS)
|
||||||
}
|
}
|
||||||
@@ -288,4 +464,28 @@ export class RealtimeConnection {
|
|||||||
this.reconnectTimer = null
|
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