feat: notify collaborators to refresh from DB after save
Instead of relying on Yjs broadcast serialization (which has delivery issues), use a lightweight state-refresh broadcast event. When any client persists (manual save or CRDT auto-persist), it broadcasts state-refresh. Other clients fetch the latest flowchart_data from the database and update their local state and CRDT. Added isSuppressed flag to CRDTManager to prevent broadcast/persist loops during initialization and refresh operations. Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 4.5
parent
cdaae6b965
commit
eb86ccd291
@@ -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,6 +57,7 @@ export class CRDTManager {
|
||||
this.edgesMap.set(edge.id, JSON.stringify(edge))
|
||||
})
|
||||
}, 'init')
|
||||
this.isSuppressed = false
|
||||
}
|
||||
|
||||
/** Connect to a Supabase Realtime channel for outbound broadcasts */
|
||||
@@ -76,6 +79,37 @@ export class CRDTManager {
|
||||
// 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
|
||||
@@ -173,17 +207,6 @@ export class CRDTManager {
|
||||
this.callbacks.onEdgesChange(this.getEdges())
|
||||
}
|
||||
|
||||
/** Broadcast the full document state to sync all connected clients */
|
||||
broadcastFullState(): void {
|
||||
if (!this.channel || this.isDestroyed) return
|
||||
const fullState = Y.encodeStateAsUpdate(this.doc)
|
||||
this.channel.send({
|
||||
type: 'broadcast',
|
||||
event: BROADCAST_EVENT,
|
||||
payload: { update: Array.from(fullState) },
|
||||
})
|
||||
}
|
||||
|
||||
private broadcastUpdate(update: Uint8Array): void {
|
||||
if (!this.channel || this.isDestroyed) return
|
||||
this.channel.send({
|
||||
|
||||
@@ -36,6 +36,7 @@ type RealtimeCallbacks = {
|
||||
onCursorUpdate?: (cursor: RemoteCursor) => void
|
||||
onNodeLockUpdate?: (lock: NodeLock | null, userId: string) => void
|
||||
onCRDTUpdate?: (update: number[]) => void
|
||||
onStateRefresh?: () => void
|
||||
}
|
||||
|
||||
const HEARTBEAT_INTERVAL_MS = 30_000
|
||||
@@ -137,6 +138,9 @@ export class RealtimeConnection {
|
||||
this.callbacks.onCRDTUpdate?.(payload.update)
|
||||
}
|
||||
})
|
||||
.on('broadcast', { event: 'state-refresh' }, () => {
|
||||
this.callbacks.onStateRefresh?.()
|
||||
})
|
||||
.subscribe(async (status) => {
|
||||
if (this.isDestroyed) return
|
||||
|
||||
@@ -195,6 +199,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({
|
||||
|
||||
Reference in New Issue
Block a user