mirror of
https://github.com/langgenius/dify.git
synced 2026-05-05 01:48:04 +08:00
Merge remote-tracking branch 'origin/p254' into p284
This commit is contained in:
hjlarry
@ -1,5 +1,7 @@
|
||||
import type { FC } from 'react'
|
||||
import { useViewport } from 'reactflow'
|
||||
import type { CursorPosition, OnlineUser } from '@/app/components/workflow/collaboration/types'
|
||||
import { getUserColor } from '../utils/user-color'
|
||||
|
||||
type UserCursorsProps = {
|
||||
cursors: Record<string, CursorPosition>
|
||||
@ -7,20 +9,20 @@ type UserCursorsProps = {
|
||||
onlineUsers: OnlineUser[]
|
||||
}
|
||||
|
||||
const getUserColor = (id: string) => {
|
||||
const colors = ['#3B82F6', '#EF4444', '#10B981', '#F59E0B', '#8B5CF6', '#EC4899', '#06B6D4', '#84CC16']
|
||||
const hash = id.split('').reduce((a, b) => {
|
||||
a = ((a << 5) - a) + b.charCodeAt(0)
|
||||
return a & a
|
||||
}, 0)
|
||||
return colors[Math.abs(hash) % colors.length]
|
||||
}
|
||||
|
||||
const UserCursors: FC<UserCursorsProps> = ({
|
||||
cursors,
|
||||
myUserId,
|
||||
onlineUsers,
|
||||
}) => {
|
||||
const viewport = useViewport()
|
||||
|
||||
const convertToScreenCoordinates = (cursor: CursorPosition) => {
|
||||
// Convert world coordinates to screen coordinates using current viewport
|
||||
const screenX = cursor.x * viewport.zoom + viewport.x
|
||||
const screenY = cursor.y * viewport.zoom + viewport.y
|
||||
|
||||
return { x: screenX, y: screenY }
|
||||
}
|
||||
return (
|
||||
<>
|
||||
{Object.entries(cursors || {}).map(([userId, cursor]) => {
|
||||
@ -30,14 +32,15 @@ const UserCursors: FC<UserCursorsProps> = ({
|
||||
const userInfo = onlineUsers.find(user => user.user_id === userId)
|
||||
const userName = userInfo?.username || `User ${userId.slice(-4)}`
|
||||
const userColor = getUserColor(userId)
|
||||
const screenPos = convertToScreenCoordinates(cursor)
|
||||
|
||||
return (
|
||||
<div
|
||||
key={userId}
|
||||
className="pointer-events-none absolute z-[10000] -translate-x-0.5 -translate-y-0.5 transition-all duration-150 ease-out"
|
||||
className="pointer-events-none absolute z-[10000] transition-all duration-150 ease-out"
|
||||
style={{
|
||||
left: cursor.x,
|
||||
top: cursor.y,
|
||||
left: screenPos.x,
|
||||
top: screenPos.y,
|
||||
}}
|
||||
>
|
||||
<svg
|
||||
@ -49,16 +52,16 @@ const UserCursors: FC<UserCursorsProps> = ({
|
||||
className="drop-shadow-md"
|
||||
>
|
||||
<path
|
||||
d="M3 3L16 8L9 10L7 17L3 3Z"
|
||||
d="M5 3L5 15L8 11.5L11 16L13 15L10 10.5L14 10.5L5 3Z"
|
||||
fill={userColor}
|
||||
stroke="white"
|
||||
strokeWidth="1"
|
||||
strokeWidth="1.5"
|
||||
strokeLinejoin="round"
|
||||
/>
|
||||
</svg>
|
||||
|
||||
<div
|
||||
className="absolute -top-0.5 left-[18px] max-w-[120px] overflow-hidden text-ellipsis whitespace-nowrap rounded px-1.5 py-0.5 text-[11px] font-medium text-white shadow-sm"
|
||||
className="absolute left-4 top-4 max-w-[120px] overflow-hidden text-ellipsis whitespace-nowrap rounded px-1.5 py-0.5 text-[11px] font-medium text-white shadow-sm"
|
||||
style={{
|
||||
backgroundColor: userColor,
|
||||
}}
|
||||
|
||||
@ -1,4 +1,4 @@
|
||||
import { LoroDoc } from 'loro-crdt'
|
||||
import { LoroDoc, UndoManager } from 'loro-crdt'
|
||||
import { isEqual } from 'lodash-es'
|
||||
import { webSocketClient } from './websocket-manager'
|
||||
import { CRDTProvider } from './crdt-provider'
|
||||
@ -8,6 +8,7 @@ import type { CollaborationState, CursorPosition, OnlineUser } from '../types/co
|
||||
|
||||
export class CollaborationManager {
|
||||
private doc: LoroDoc | null = null
|
||||
private undoManager: UndoManager | null = null
|
||||
private provider: CRDTProvider | null = null
|
||||
private nodesMap: any = null
|
||||
private edgesMap: any = null
|
||||
@ -17,6 +18,8 @@ export class CollaborationManager {
|
||||
private isLeader = false
|
||||
private leaderId: string | null = null
|
||||
private cursors: Record<string, CursorPosition> = {}
|
||||
private activeConnections = new Set<string>()
|
||||
private isUndoRedoInProgress = false
|
||||
|
||||
init = (appId: string, reactFlowStore: any): void => {
|
||||
if (!reactFlowStore) {
|
||||
@ -27,44 +30,141 @@ export class CollaborationManager {
|
||||
}
|
||||
|
||||
setNodes = (oldNodes: Node[], newNodes: Node[]): void => {
|
||||
if (!this.doc) return
|
||||
|
||||
// Don't track operations during undo/redo to prevent loops
|
||||
if (this.isUndoRedoInProgress) {
|
||||
console.log('Skipping setNodes during undo/redo')
|
||||
return
|
||||
}
|
||||
|
||||
console.log('Setting nodes with tracking')
|
||||
this.syncNodes(oldNodes, newNodes)
|
||||
if (this.doc)
|
||||
this.doc.commit()
|
||||
this.doc.commit()
|
||||
}
|
||||
|
||||
setEdges = (oldEdges: Edge[], newEdges: Edge[]): void => {
|
||||
if (!this.doc) return
|
||||
|
||||
// Don't track operations during undo/redo to prevent loops
|
||||
if (this.isUndoRedoInProgress) {
|
||||
console.log('Skipping setEdges during undo/redo')
|
||||
return
|
||||
}
|
||||
|
||||
console.log('Setting edges with tracking')
|
||||
this.syncEdges(oldEdges, newEdges)
|
||||
if (this.doc)
|
||||
this.doc.commit()
|
||||
this.doc.commit()
|
||||
}
|
||||
|
||||
destroy = (): void => {
|
||||
this.disconnect()
|
||||
}
|
||||
|
||||
async connect(appId: string, reactFlowStore: any): Promise<void> {
|
||||
if (this.currentAppId === appId && this.doc) return
|
||||
async connect(appId: string, reactFlowStore?: any): Promise<string> {
|
||||
const connectionId = Math.random().toString(36).substring(2, 11)
|
||||
|
||||
this.disconnect()
|
||||
this.activeConnections.add(connectionId)
|
||||
|
||||
if (this.currentAppId === appId && this.doc) {
|
||||
// Already connected to the same app, only update store if provided and we don't have one
|
||||
if (reactFlowStore && !this.reactFlowStore)
|
||||
this.reactFlowStore = reactFlowStore
|
||||
|
||||
return connectionId
|
||||
}
|
||||
|
||||
// Only disconnect if switching to a different app
|
||||
if (this.currentAppId && this.currentAppId !== appId)
|
||||
this.forceDisconnect()
|
||||
|
||||
this.currentAppId = appId
|
||||
this.reactFlowStore = reactFlowStore
|
||||
// Only set store if provided
|
||||
if (reactFlowStore)
|
||||
this.reactFlowStore = reactFlowStore
|
||||
|
||||
const socket = webSocketClient.connect(appId)
|
||||
|
||||
// Setup event listeners BEFORE any other operations
|
||||
this.setupSocketEventListeners(socket)
|
||||
|
||||
this.doc = new LoroDoc()
|
||||
this.nodesMap = this.doc.getMap('nodes')
|
||||
this.edgesMap = this.doc.getMap('edges')
|
||||
|
||||
// Initialize UndoManager for collaborative undo/redo
|
||||
this.undoManager = new UndoManager(this.doc, {
|
||||
maxUndoSteps: 100,
|
||||
mergeInterval: 500, // Merge operations within 500ms
|
||||
excludeOriginPrefixes: [], // Don't exclude anything - let UndoManager track all local operations
|
||||
onPush: (isUndo, range, event) => {
|
||||
console.log('UndoManager onPush:', { isUndo, range, event })
|
||||
// Store current selection state when an operation is pushed
|
||||
const selectedNode = this.reactFlowStore?.getState().getNodes().find((n: Node) => n.data.selected)
|
||||
|
||||
// Emit event to update UI button states when new operation is pushed
|
||||
setTimeout(() => {
|
||||
this.eventEmitter.emit('undoRedoStateChange', {
|
||||
canUndo: this.undoManager?.canUndo() || false,
|
||||
canRedo: this.undoManager?.canRedo() || false,
|
||||
})
|
||||
}, 0)
|
||||
|
||||
return {
|
||||
value: {
|
||||
selectedNodeId: selectedNode?.id || null,
|
||||
timestamp: Date.now(),
|
||||
},
|
||||
cursors: [],
|
||||
}
|
||||
},
|
||||
onPop: (isUndo, value, counterRange) => {
|
||||
console.log('UndoManager onPop:', { isUndo, value, counterRange })
|
||||
// Restore selection state when undoing/redoing
|
||||
if (value?.value && typeof value.value === 'object' && 'selectedNodeId' in value.value && this.reactFlowStore) {
|
||||
const selectedNodeId = (value.value as any).selectedNodeId
|
||||
if (selectedNodeId) {
|
||||
const { setNodes } = this.reactFlowStore.getState()
|
||||
const nodes = this.reactFlowStore.getState().getNodes()
|
||||
const newNodes = nodes.map((n: Node) => ({
|
||||
...n,
|
||||
data: {
|
||||
...n.data,
|
||||
selected: n.id === selectedNodeId,
|
||||
},
|
||||
}))
|
||||
setNodes(newNodes)
|
||||
}
|
||||
}
|
||||
},
|
||||
})
|
||||
|
||||
this.provider = new CRDTProvider(socket, this.doc)
|
||||
|
||||
this.setupSubscriptions()
|
||||
this.setupSocketEventListeners(socket)
|
||||
|
||||
// Force user_connect if already connected
|
||||
if (socket.connected)
|
||||
socket.emit('user_connect', { workflow_id: appId })
|
||||
|
||||
return connectionId
|
||||
}
|
||||
|
||||
disconnect = (): void => {
|
||||
disconnect = (connectionId?: string): void => {
|
||||
if (connectionId)
|
||||
this.activeConnections.delete(connectionId)
|
||||
|
||||
// Only disconnect when no more connections
|
||||
if (this.activeConnections.size === 0)
|
||||
this.forceDisconnect()
|
||||
}
|
||||
|
||||
private forceDisconnect = (): void => {
|
||||
if (this.currentAppId)
|
||||
webSocketClient.disconnect(this.currentAppId)
|
||||
|
||||
this.provider?.destroy()
|
||||
this.undoManager = null
|
||||
this.doc = null
|
||||
this.provider = null
|
||||
this.nodesMap = null
|
||||
@ -72,6 +172,17 @@ export class CollaborationManager {
|
||||
this.currentAppId = null
|
||||
this.reactFlowStore = null
|
||||
this.cursors = {}
|
||||
this.isUndoRedoInProgress = false
|
||||
|
||||
// Only reset leader status when actually disconnecting
|
||||
const wasLeader = this.isLeader
|
||||
this.isLeader = false
|
||||
this.leaderId = null
|
||||
|
||||
if (wasLeader)
|
||||
this.eventEmitter.emit('leaderChange', false)
|
||||
|
||||
this.activeConnections.clear()
|
||||
this.eventEmitter.removeAllListeners()
|
||||
}
|
||||
|
||||
@ -101,6 +212,38 @@ export class CollaborationManager {
|
||||
}
|
||||
}
|
||||
|
||||
emitSyncRequest(): void {
|
||||
if (!this.currentAppId || !webSocketClient.isConnected(this.currentAppId)) return
|
||||
|
||||
const socket = webSocketClient.getSocket(this.currentAppId)
|
||||
if (socket) {
|
||||
console.log('Emitting sync request to leader')
|
||||
socket.emit('collaboration_event', {
|
||||
type: 'syncRequest',
|
||||
data: { timestamp: Date.now() },
|
||||
timestamp: Date.now(),
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
emitWorkflowUpdate(appId: string): void {
|
||||
if (!this.currentAppId || !webSocketClient.isConnected(this.currentAppId)) return
|
||||
|
||||
const socket = webSocketClient.getSocket(this.currentAppId)
|
||||
if (socket) {
|
||||
console.log('Emitting Workflow update event')
|
||||
socket.emit('collaboration_event', {
|
||||
type: 'workflowUpdate',
|
||||
data: { appId, timestamp: Date.now() },
|
||||
timestamp: Date.now(),
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
onSyncRequest(callback: () => void): () => void {
|
||||
return this.eventEmitter.on('syncRequest', callback)
|
||||
}
|
||||
|
||||
onStateChange(callback: (state: Partial<CollaborationState>) => void): () => void {
|
||||
return this.eventEmitter.on('stateChange', callback)
|
||||
}
|
||||
@ -113,38 +256,277 @@ export class CollaborationManager {
|
||||
return this.eventEmitter.on('onlineUsers', callback)
|
||||
}
|
||||
|
||||
onWorkflowUpdate(callback: (update: { appId: string; timestamp: number }) => void): () => void {
|
||||
return this.eventEmitter.on('workflowUpdate', callback)
|
||||
}
|
||||
|
||||
onVarsAndFeaturesUpdate(callback: (update: any) => void): () => void {
|
||||
return this.eventEmitter.on('varsAndFeaturesUpdate', callback)
|
||||
}
|
||||
|
||||
onAppStateUpdate(callback: (update: any) => void): () => void {
|
||||
return this.eventEmitter.on('appStateUpdate', callback)
|
||||
}
|
||||
|
||||
onMcpServerUpdate(callback: (update: any) => void): () => void {
|
||||
return this.eventEmitter.on('mcpServerUpdate', callback)
|
||||
}
|
||||
|
||||
onLeaderChange(callback: (isLeader: boolean) => void): () => void {
|
||||
return this.eventEmitter.on('leaderChange', callback)
|
||||
}
|
||||
|
||||
onCommentsUpdate(callback: (update: { appId: string; timestamp: number }) => void): () => void {
|
||||
return this.eventEmitter.on('commentsUpdate', callback)
|
||||
}
|
||||
|
||||
emitCommentsUpdate(appId: string): void {
|
||||
if (!this.currentAppId || !webSocketClient.isConnected(this.currentAppId)) return
|
||||
|
||||
const socket = webSocketClient.getSocket(this.currentAppId)
|
||||
if (socket) {
|
||||
console.log('Emitting Comments update event')
|
||||
socket.emit('collaboration_event', {
|
||||
type: 'commentsUpdate',
|
||||
data: { appId, timestamp: Date.now() },
|
||||
timestamp: Date.now(),
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
onUndoRedoStateChange(callback: (state: { canUndo: boolean; canRedo: boolean }) => void): () => void {
|
||||
return this.eventEmitter.on('undoRedoStateChange', callback)
|
||||
}
|
||||
|
||||
getLeaderId(): string | null {
|
||||
return this.leaderId
|
||||
}
|
||||
|
||||
getIsLeader(): boolean {
|
||||
return this.isLeader
|
||||
}
|
||||
|
||||
// Collaborative undo/redo methods
|
||||
undo(): boolean {
|
||||
if (!this.undoManager) {
|
||||
console.log('UndoManager not initialized')
|
||||
return false
|
||||
}
|
||||
|
||||
const canUndo = this.undoManager.canUndo()
|
||||
console.log('Can undo:', canUndo)
|
||||
|
||||
if (canUndo) {
|
||||
this.isUndoRedoInProgress = true
|
||||
const result = this.undoManager.undo()
|
||||
|
||||
// After undo, manually update React state from CRDT without triggering collaboration
|
||||
if (result && this.reactFlowStore) {
|
||||
requestAnimationFrame(() => {
|
||||
// Get ReactFlow's native setters, not the collaborative ones
|
||||
const state = this.reactFlowStore.getState()
|
||||
const updatedNodes = Array.from(this.nodesMap.values())
|
||||
const updatedEdges = Array.from(this.edgesMap.values())
|
||||
console.log('Manually updating React state after undo')
|
||||
|
||||
// Call ReactFlow's native setters directly to avoid triggering collaboration
|
||||
state.setNodes(updatedNodes)
|
||||
state.setEdges(updatedEdges)
|
||||
|
||||
this.isUndoRedoInProgress = false
|
||||
|
||||
// Emit event to update UI button states
|
||||
this.eventEmitter.emit('undoRedoStateChange', {
|
||||
canUndo: this.undoManager?.canUndo() || false,
|
||||
canRedo: this.undoManager?.canRedo() || false,
|
||||
})
|
||||
})
|
||||
}
|
||||
else {
|
||||
this.isUndoRedoInProgress = false
|
||||
}
|
||||
|
||||
console.log('Undo result:', result)
|
||||
return result
|
||||
}
|
||||
|
||||
return false
|
||||
}
|
||||
|
||||
redo(): boolean {
|
||||
if (!this.undoManager) {
|
||||
console.log('RedoManager not initialized')
|
||||
return false
|
||||
}
|
||||
|
||||
const canRedo = this.undoManager.canRedo()
|
||||
console.log('Can redo:', canRedo)
|
||||
|
||||
if (canRedo) {
|
||||
this.isUndoRedoInProgress = true
|
||||
const result = this.undoManager.redo()
|
||||
|
||||
// After redo, manually update React state from CRDT without triggering collaboration
|
||||
if (result && this.reactFlowStore) {
|
||||
requestAnimationFrame(() => {
|
||||
// Get ReactFlow's native setters, not the collaborative ones
|
||||
const state = this.reactFlowStore.getState()
|
||||
const updatedNodes = Array.from(this.nodesMap.values())
|
||||
const updatedEdges = Array.from(this.edgesMap.values())
|
||||
console.log('Manually updating React state after redo')
|
||||
|
||||
// Call ReactFlow's native setters directly to avoid triggering collaboration
|
||||
state.setNodes(updatedNodes)
|
||||
state.setEdges(updatedEdges)
|
||||
|
||||
this.isUndoRedoInProgress = false
|
||||
|
||||
// Emit event to update UI button states
|
||||
this.eventEmitter.emit('undoRedoStateChange', {
|
||||
canUndo: this.undoManager?.canUndo() || false,
|
||||
canRedo: this.undoManager?.canRedo() || false,
|
||||
})
|
||||
})
|
||||
}
|
||||
else {
|
||||
this.isUndoRedoInProgress = false
|
||||
}
|
||||
|
||||
console.log('Redo result:', result)
|
||||
return result
|
||||
}
|
||||
|
||||
return false
|
||||
}
|
||||
|
||||
canUndo(): boolean {
|
||||
if (!this.undoManager) return false
|
||||
return this.undoManager.canUndo()
|
||||
}
|
||||
|
||||
canRedo(): boolean {
|
||||
if (!this.undoManager) return false
|
||||
return this.undoManager.canRedo()
|
||||
}
|
||||
|
||||
clearUndoStack(): void {
|
||||
if (!this.undoManager) return
|
||||
this.undoManager.clear()
|
||||
}
|
||||
|
||||
debugLeaderStatus(): void {
|
||||
console.log('=== Leader Status Debug ===')
|
||||
console.log('Current leader status:', this.isLeader)
|
||||
console.log('Current leader ID:', this.leaderId)
|
||||
console.log('Active connections:', this.activeConnections.size)
|
||||
console.log('Connected:', this.isConnected())
|
||||
console.log('Current app ID:', this.currentAppId)
|
||||
console.log('Has ReactFlow store:', !!this.reactFlowStore)
|
||||
console.log('========================')
|
||||
}
|
||||
|
||||
private syncNodes(oldNodes: Node[], newNodes: Node[]): void {
|
||||
if (!this.nodesMap) return
|
||||
if (!this.nodesMap || !this.doc) return
|
||||
|
||||
const oldNodesMap = new Map(oldNodes.map(node => [node.id, node]))
|
||||
const newNodesMap = new Map(newNodes.map(node => [node.id, node]))
|
||||
|
||||
// Delete removed nodes
|
||||
oldNodes.forEach((oldNode) => {
|
||||
if (!newNodesMap.has(oldNode.id))
|
||||
this.nodesMap.delete(oldNode.id)
|
||||
})
|
||||
|
||||
// Add or update nodes with fine-grained sync for data properties
|
||||
newNodes.forEach((newNode) => {
|
||||
const oldNode = oldNodesMap.get(newNode.id)
|
||||
|
||||
if (!oldNode) {
|
||||
const persistentData = this.getPersistentNodeData(newNode)
|
||||
const clonedData = JSON.parse(JSON.stringify(persistentData))
|
||||
this.nodesMap.set(newNode.id, clonedData)
|
||||
// New node - create as nested structure
|
||||
const nodeData: any = {
|
||||
id: newNode.id,
|
||||
type: newNode.type,
|
||||
position: { ...newNode.position },
|
||||
width: newNode.width,
|
||||
height: newNode.height,
|
||||
sourcePosition: newNode.sourcePosition,
|
||||
targetPosition: newNode.targetPosition,
|
||||
data: {},
|
||||
}
|
||||
|
||||
// Clone data properties, excluding private ones
|
||||
Object.entries(newNode.data).forEach(([key, value]) => {
|
||||
if (!key.startsWith('_') && value !== undefined)
|
||||
nodeData.data[key] = JSON.parse(JSON.stringify(value))
|
||||
})
|
||||
|
||||
this.nodesMap.set(newNode.id, nodeData)
|
||||
}
|
||||
else {
|
||||
const oldPersistentData = this.getPersistentNodeData(oldNode)
|
||||
const newPersistentData = this.getPersistentNodeData(newNode)
|
||||
if (!isEqual(oldPersistentData, newPersistentData)) {
|
||||
const clonedData = JSON.parse(JSON.stringify(newPersistentData))
|
||||
this.nodesMap.set(newNode.id, clonedData)
|
||||
// Get existing node from CRDT
|
||||
const existingNode = this.nodesMap.get(newNode.id)
|
||||
|
||||
if (existingNode) {
|
||||
// Create a deep copy to modify
|
||||
const updatedNode = JSON.parse(JSON.stringify(existingNode))
|
||||
|
||||
// Update position only if changed
|
||||
if (oldNode.position.x !== newNode.position.x || oldNode.position.y !== newNode.position.y)
|
||||
updatedNode.position = { ...newNode.position }
|
||||
|
||||
// Update dimensions only if changed
|
||||
if (oldNode.width !== newNode.width)
|
||||
updatedNode.width = newNode.width
|
||||
|
||||
if (oldNode.height !== newNode.height)
|
||||
updatedNode.height = newNode.height
|
||||
|
||||
// Ensure data object exists
|
||||
if (!updatedNode.data)
|
||||
updatedNode.data = {}
|
||||
|
||||
// Fine-grained update of data properties
|
||||
const oldData = oldNode.data || {}
|
||||
const newData = newNode.data || {}
|
||||
|
||||
// Only update changed properties in data
|
||||
Object.entries(newData).forEach(([key, value]) => {
|
||||
if (!key.startsWith('_')) {
|
||||
const oldValue = (oldData as any)[key]
|
||||
if (!isEqual(oldValue, value))
|
||||
updatedNode.data[key] = JSON.parse(JSON.stringify(value))
|
||||
}
|
||||
})
|
||||
|
||||
// Remove deleted properties from data
|
||||
Object.keys(oldData).forEach((key) => {
|
||||
if (!key.startsWith('_') && !(key in newData))
|
||||
delete updatedNode.data[key]
|
||||
})
|
||||
|
||||
// Only update in CRDT if something actually changed
|
||||
if (!isEqual(existingNode, updatedNode))
|
||||
this.nodesMap.set(newNode.id, updatedNode)
|
||||
}
|
||||
else {
|
||||
// Node exists locally but not in CRDT yet
|
||||
const nodeData: any = {
|
||||
id: newNode.id,
|
||||
type: newNode.type,
|
||||
position: { ...newNode.position },
|
||||
width: newNode.width,
|
||||
height: newNode.height,
|
||||
sourcePosition: newNode.sourcePosition,
|
||||
targetPosition: newNode.targetPosition,
|
||||
data: {},
|
||||
}
|
||||
|
||||
Object.entries(newNode.data).forEach(([key, value]) => {
|
||||
if (!key.startsWith('_') && value !== undefined)
|
||||
nodeData.data[key] = JSON.parse(JSON.stringify(value))
|
||||
})
|
||||
|
||||
this.nodesMap.set(newNode.id, nodeData)
|
||||
}
|
||||
}
|
||||
})
|
||||
@ -174,31 +556,45 @@ export class CollaborationManager {
|
||||
})
|
||||
}
|
||||
|
||||
private getPersistentNodeData(node: Node): any {
|
||||
const { data, ...rest } = node
|
||||
const filteredData = Object.fromEntries(
|
||||
Object.entries(data).filter(([key]) => !key.startsWith('_')),
|
||||
)
|
||||
return { ...rest, data: filteredData }
|
||||
}
|
||||
|
||||
private setupSubscriptions(): void {
|
||||
this.nodesMap?.subscribe((event: any) => {
|
||||
console.log('nodesMap subscription event:', event)
|
||||
if (event.by === 'import' && this.reactFlowStore) {
|
||||
// Don't update React nodes during undo/redo to prevent loops
|
||||
if (this.isUndoRedoInProgress) {
|
||||
console.log('Skipping nodes subscription update during undo/redo')
|
||||
return
|
||||
}
|
||||
|
||||
requestAnimationFrame(() => {
|
||||
const { setNodes } = this.reactFlowStore.getState()
|
||||
// Get ReactFlow's native setters, not the collaborative ones
|
||||
const state = this.reactFlowStore.getState()
|
||||
const updatedNodes = Array.from(this.nodesMap.values())
|
||||
setNodes(updatedNodes)
|
||||
console.log('Updating React nodes from subscription')
|
||||
|
||||
// Call ReactFlow's native setter directly to avoid triggering collaboration
|
||||
state.setNodes(updatedNodes)
|
||||
})
|
||||
}
|
||||
})
|
||||
|
||||
this.edgesMap?.subscribe((event: any) => {
|
||||
console.log('edgesMap subscription event:', event)
|
||||
if (event.by === 'import' && this.reactFlowStore) {
|
||||
// Don't update React edges during undo/redo to prevent loops
|
||||
if (this.isUndoRedoInProgress) {
|
||||
console.log('Skipping edges subscription update during undo/redo')
|
||||
return
|
||||
}
|
||||
|
||||
requestAnimationFrame(() => {
|
||||
const { setEdges } = this.reactFlowStore.getState()
|
||||
// Get ReactFlow's native setters, not the collaborative ones
|
||||
const state = this.reactFlowStore.getState()
|
||||
const updatedEdges = Array.from(this.edgesMap.values())
|
||||
setEdges(updatedEdges)
|
||||
console.log('Updating React edges from subscription')
|
||||
|
||||
// Call ReactFlow's native setter directly to avoid triggering collaboration
|
||||
state.setEdges(updatedEdges)
|
||||
})
|
||||
}
|
||||
})
|
||||
@ -209,8 +605,6 @@ export class CollaborationManager {
|
||||
|
||||
socket.on('collaboration_update', (update: any) => {
|
||||
if (update.type === 'mouseMove') {
|
||||
console.log('Processing mouseMove event:', update)
|
||||
|
||||
// Update cursor state for this user
|
||||
this.cursors[update.userId] = {
|
||||
x: update.data.x,
|
||||
@ -219,29 +613,81 @@ export class CollaborationManager {
|
||||
timestamp: update.timestamp,
|
||||
}
|
||||
|
||||
// Emit the complete cursor state
|
||||
console.log('Emitting complete cursor state:', this.cursors)
|
||||
this.eventEmitter.emit('cursors', { ...this.cursors })
|
||||
}
|
||||
else if (update.type === 'varsAndFeaturesUpdate') {
|
||||
console.log('Processing varsAndFeaturesUpdate event:', update)
|
||||
this.eventEmitter.emit('varsAndFeaturesUpdate', update)
|
||||
}
|
||||
else if (update.type === 'appStateUpdate') {
|
||||
console.log('Processing appStateUpdate event:', update)
|
||||
this.eventEmitter.emit('appStateUpdate', update)
|
||||
}
|
||||
else if (update.type === 'mcpServerUpdate') {
|
||||
console.log('Processing mcpServerUpdate event:', update)
|
||||
this.eventEmitter.emit('mcpServerUpdate', update)
|
||||
}
|
||||
else if (update.type === 'workflowUpdate') {
|
||||
console.log('Processing workflowUpdate event:', update)
|
||||
this.eventEmitter.emit('workflowUpdate', update.data)
|
||||
}
|
||||
else if (update.type === 'commentsUpdate') {
|
||||
console.log('Processing commentsUpdate event:', update)
|
||||
this.eventEmitter.emit('commentsUpdate', update.data)
|
||||
}
|
||||
else if (update.type === 'syncRequest') {
|
||||
console.log('Received sync request from another user')
|
||||
// Only process if we are the leader
|
||||
if (this.isLeader) {
|
||||
console.log('Leader received sync request, triggering sync')
|
||||
this.eventEmitter.emit('syncRequest', {})
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
socket.on('online_users', (data: { users: OnlineUser[]; leader: string }) => {
|
||||
const onlineUserIds = new Set(data.users.map(user => user.user_id))
|
||||
socket.on('online_users', (data: { users: OnlineUser[]; leader?: string }) => {
|
||||
try {
|
||||
if (!data || !Array.isArray(data.users)) {
|
||||
console.warn('Invalid online_users data structure:', data)
|
||||
return
|
||||
}
|
||||
|
||||
// Remove cursors for offline users
|
||||
Object.keys(this.cursors).forEach((userId) => {
|
||||
if (!onlineUserIds.has(userId))
|
||||
delete this.cursors[userId]
|
||||
})
|
||||
const onlineUserIds = new Set(data.users.map((user: OnlineUser) => user.user_id))
|
||||
|
||||
console.log('Updated online users and cleaned offline cursors:', data.users)
|
||||
this.leaderId = data.leader
|
||||
this.eventEmitter.emit('onlineUsers', data.users)
|
||||
this.eventEmitter.emit('cursors', { ...this.cursors })
|
||||
// Remove cursors for offline users
|
||||
Object.keys(this.cursors).forEach((userId) => {
|
||||
if (!onlineUserIds.has(userId))
|
||||
delete this.cursors[userId]
|
||||
})
|
||||
|
||||
// Update leader information
|
||||
if (data.leader && typeof data.leader === 'string')
|
||||
this.leaderId = data.leader
|
||||
|
||||
this.eventEmitter.emit('onlineUsers', data.users)
|
||||
this.eventEmitter.emit('cursors', { ...this.cursors })
|
||||
}
|
||||
catch (error) {
|
||||
console.error('Error processing online_users update:', error)
|
||||
}
|
||||
})
|
||||
|
||||
socket.on('status', (data: any) => {
|
||||
try {
|
||||
if (!data || typeof data.isLeader !== 'boolean') {
|
||||
console.warn('Invalid status data:', data)
|
||||
return
|
||||
}
|
||||
|
||||
const wasLeader = this.isLeader
|
||||
this.isLeader = data.isLeader
|
||||
|
||||
if (wasLeader !== this.isLeader)
|
||||
this.eventEmitter.emit('leaderChange', this.isLeader)
|
||||
}
|
||||
catch (error) {
|
||||
console.error('Error processing status update:', error)
|
||||
}
|
||||
})
|
||||
|
||||
socket.on('status', (data: { isLeader: boolean }) => {
|
||||
@ -261,11 +707,26 @@ export class CollaborationManager {
|
||||
})
|
||||
|
||||
socket.on('connect', () => {
|
||||
console.log('WebSocket connected successfully')
|
||||
this.eventEmitter.emit('stateChange', { isConnected: true })
|
||||
})
|
||||
|
||||
socket.on('disconnect', () => {
|
||||
socket.on('disconnect', (reason: string) => {
|
||||
console.log('WebSocket disconnected:', reason)
|
||||
this.cursors = {}
|
||||
this.isLeader = false
|
||||
this.leaderId = null
|
||||
this.eventEmitter.emit('stateChange', { isConnected: false })
|
||||
this.eventEmitter.emit('cursors', {})
|
||||
})
|
||||
|
||||
socket.on('connect_error', (error: any) => {
|
||||
console.error('WebSocket connection error:', error)
|
||||
this.eventEmitter.emit('stateChange', { isConnected: false, error: error.message })
|
||||
})
|
||||
|
||||
socket.on('error', (error: any) => {
|
||||
console.error('WebSocket error:', error)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@ -1,4 +1,5 @@
|
||||
import { useEffect, useRef, useState } from 'react'
|
||||
import type { ReactFlowInstance } from 'reactflow'
|
||||
import { collaborationManager } from '../core/collaboration-manager'
|
||||
import { CursorService } from '../services/cursor-service'
|
||||
import type { CollaborationState } from '../types/collaboration'
|
||||
@ -16,15 +17,13 @@ export function useCollaboration(appId: string, reactFlowStore?: any) {
|
||||
useEffect(() => {
|
||||
if (!appId) return
|
||||
|
||||
if (!cursorServiceRef.current) {
|
||||
cursorServiceRef.current = new CursorService({
|
||||
minMoveDistance: 10,
|
||||
throttleMs: 300,
|
||||
})
|
||||
}
|
||||
let connectionId: string | null = null
|
||||
|
||||
if (!cursorServiceRef.current)
|
||||
cursorServiceRef.current = new CursorService()
|
||||
|
||||
const initCollaboration = async () => {
|
||||
await collaborationManager.connect(appId, reactFlowStore)
|
||||
connectionId = await collaborationManager.connect(appId, reactFlowStore)
|
||||
setState((prev: any) => ({ ...prev, appId, isConnected: collaborationManager.isConnected() }))
|
||||
}
|
||||
|
||||
@ -36,7 +35,6 @@ export function useCollaboration(appId: string, reactFlowStore?: any) {
|
||||
})
|
||||
|
||||
const unsubscribeCursors = collaborationManager.onCursorUpdate((cursors: any) => {
|
||||
console.log('Cursor update received:', cursors)
|
||||
setState((prev: any) => ({ ...prev, cursors }))
|
||||
})
|
||||
|
||||
@ -46,6 +44,7 @@ export function useCollaboration(appId: string, reactFlowStore?: any) {
|
||||
})
|
||||
|
||||
const unsubscribeLeaderChange = collaborationManager.onLeaderChange((isLeader: boolean) => {
|
||||
console.log('Leader status changed:', isLeader)
|
||||
setState((prev: any) => ({ ...prev, isLeader }))
|
||||
})
|
||||
|
||||
@ -55,15 +54,16 @@ export function useCollaboration(appId: string, reactFlowStore?: any) {
|
||||
unsubscribeUsers()
|
||||
unsubscribeLeaderChange()
|
||||
cursorServiceRef.current?.stopTracking()
|
||||
collaborationManager.disconnect()
|
||||
if (connectionId)
|
||||
collaborationManager.disconnect(connectionId)
|
||||
}
|
||||
}, [appId, reactFlowStore])
|
||||
|
||||
const startCursorTracking = (containerRef: React.RefObject<HTMLElement>) => {
|
||||
const startCursorTracking = (containerRef: React.RefObject<HTMLElement>, reactFlowInstance?: ReactFlowInstance) => {
|
||||
if (cursorServiceRef.current) {
|
||||
cursorServiceRef.current.startTracking(containerRef, (position) => {
|
||||
collaborationManager.emitCursorMove(position)
|
||||
})
|
||||
}, reactFlowInstance)
|
||||
}
|
||||
}
|
||||
|
||||
@ -71,12 +71,15 @@ export function useCollaboration(appId: string, reactFlowStore?: any) {
|
||||
cursorServiceRef.current?.stopTracking()
|
||||
}
|
||||
|
||||
return {
|
||||
const result = {
|
||||
isConnected: state.isConnected || false,
|
||||
onlineUsers: state.onlineUsers || [],
|
||||
cursors: state.cursors || {},
|
||||
isLeader: state.isLeader || false,
|
||||
leaderId: collaborationManager.getLeaderId(),
|
||||
startCursorTracking,
|
||||
stopCursorTracking,
|
||||
}
|
||||
|
||||
return result
|
||||
}
|
||||
|
||||
@ -1,35 +1,29 @@
|
||||
import type { RefObject } from 'react'
|
||||
import type { CursorPosition } from '../types/collaboration'
|
||||
import type { ReactFlowInstance } from 'reactflow'
|
||||
|
||||
export type CursorServiceConfig = {
|
||||
minMoveDistance?: number
|
||||
throttleMs?: number
|
||||
}
|
||||
const CURSOR_MIN_MOVE_DISTANCE = 10
|
||||
const CURSOR_THROTTLE_MS = 500
|
||||
|
||||
export class CursorService {
|
||||
private containerRef: RefObject<HTMLElement> | null = null
|
||||
private reactFlowInstance: ReactFlowInstance | null = null
|
||||
private isTracking = false
|
||||
private onCursorUpdate: ((cursors: Record<string, CursorPosition>) => void) | null = null
|
||||
private onEmitPosition: ((position: CursorPosition) => void) | null = null
|
||||
private lastEmitTime = 0
|
||||
private lastPosition: { x: number; y: number } | null = null
|
||||
private config: Required<CursorServiceConfig>
|
||||
|
||||
constructor(config: CursorServiceConfig = {}) {
|
||||
this.config = {
|
||||
minMoveDistance: config.minMoveDistance ?? 5,
|
||||
throttleMs: config.throttleMs ?? 300,
|
||||
}
|
||||
}
|
||||
|
||||
startTracking(
|
||||
containerRef: RefObject<HTMLElement>,
|
||||
onEmitPosition: (position: CursorPosition) => void,
|
||||
reactFlowInstance?: ReactFlowInstance,
|
||||
): void {
|
||||
if (this.isTracking) this.stopTracking()
|
||||
|
||||
this.containerRef = containerRef
|
||||
this.onEmitPosition = onEmitPosition
|
||||
this.reactFlowInstance = reactFlowInstance || null
|
||||
this.isTracking = true
|
||||
|
||||
if (containerRef.current)
|
||||
@ -41,6 +35,7 @@ export class CursorService {
|
||||
this.containerRef.current.removeEventListener('mousemove', this.handleMouseMove)
|
||||
|
||||
this.containerRef = null
|
||||
this.reactFlowInstance = null
|
||||
this.onEmitPosition = null
|
||||
this.isTracking = false
|
||||
this.lastPosition = null
|
||||
@ -59,26 +54,35 @@ export class CursorService {
|
||||
if (!this.containerRef?.current || !this.onEmitPosition) return
|
||||
|
||||
const rect = this.containerRef.current.getBoundingClientRect()
|
||||
const x = event.clientX - rect.left
|
||||
const y = event.clientY - rect.top
|
||||
let x = event.clientX - rect.left
|
||||
let y = event.clientY - rect.top
|
||||
|
||||
if (x >= 0 && y >= 0 && x <= rect.width && y <= rect.height) {
|
||||
const now = Date.now()
|
||||
const timeThrottled = now - this.lastEmitTime > this.config.throttleMs
|
||||
const distanceThrottled = !this.lastPosition
|
||||
|| (Math.abs(x - this.lastPosition.x) > this.config.minMoveDistance
|
||||
|| Math.abs(y - this.lastPosition.y) > this.config.minMoveDistance)
|
||||
// Transform coordinates to ReactFlow world coordinates if ReactFlow instance is available
|
||||
if (this.reactFlowInstance) {
|
||||
const viewport = this.reactFlowInstance.getViewport()
|
||||
// Convert screen coordinates to world coordinates
|
||||
// World coordinates = (screen coordinates - viewport translation) / zoom
|
||||
x = (x - viewport.x) / viewport.zoom
|
||||
y = (y - viewport.y) / viewport.zoom
|
||||
}
|
||||
|
||||
if (timeThrottled && distanceThrottled) {
|
||||
this.lastPosition = { x, y }
|
||||
this.lastEmitTime = now
|
||||
this.onEmitPosition({
|
||||
x,
|
||||
y,
|
||||
userId: '',
|
||||
timestamp: now,
|
||||
})
|
||||
}
|
||||
// Always emit cursor position (remove boundary check since world coordinates can be negative)
|
||||
const now = Date.now()
|
||||
const timeThrottled = now - this.lastEmitTime > CURSOR_THROTTLE_MS
|
||||
const minDistance = CURSOR_MIN_MOVE_DISTANCE / (this.reactFlowInstance?.getZoom() || 1)
|
||||
const distanceThrottled = !this.lastPosition
|
||||
|| (Math.abs(x - this.lastPosition.x) > minDistance)
|
||||
|| (Math.abs(y - this.lastPosition.y) > minDistance)
|
||||
|
||||
if (timeThrottled && distanceThrottled) {
|
||||
this.lastPosition = { x, y }
|
||||
this.lastEmitTime = now
|
||||
this.onEmitPosition({
|
||||
x,
|
||||
y,
|
||||
userId: '',
|
||||
timestamp: now,
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -0,0 +1,12 @@
|
||||
/**
|
||||
* Generate a consistent color for a user based on their ID
|
||||
* Used for cursor colors and avatar backgrounds
|
||||
*/
|
||||
export const getUserColor = (id: string): string => {
|
||||
const colors = ['#155AEF', '#0BA5EC', '#444CE7', '#7839EE', '#4CA30D', '#0E9384', '#DD2590', '#FF4405', '#D92D20', '#F79009', '#828DAD']
|
||||
const hash = id.split('').reduce((a, b) => {
|
||||
a = ((a << 5) - a) + b.charCodeAt(0)
|
||||
return a & a
|
||||
}, 0)
|
||||
return colors[Math.abs(hash) % colors.length]
|
||||
}
|
||||
Reference in New Issue
Block a user