diff --git a/pkg/services/exploremap/realtime/channels.go b/pkg/services/exploremap/realtime/channels.go index d78589ec290..d843f911425 100644 --- a/pkg/services/exploremap/realtime/channels.go +++ b/pkg/services/exploremap/realtime/channels.go @@ -5,6 +5,7 @@ import ( "encoding/json" "fmt" "strings" + "time" "github.com/grafana/grafana-plugin-sdk-go/backend" "github.com/grafana/grafana/pkg/apimachinery/identity" @@ -61,9 +62,61 @@ func (h *ExploreMapChannelHandler) OnSubscribe(ctx context.Context, user identit }, backend.SubscribeStreamStatusOK, nil } +// messageType represents the type of message being sent +type messageType string + +const ( + MessageTypeCursorUpdate messageType = "cursor_update" + MessageTypeCursorLeave messageType = "cursor_leave" + MessageTypeViewportUpdate messageType = "viewport_update" +) + // OnPublish is called when a client publishes to the channel func (h *ExploreMapChannelHandler) OnPublish(ctx context.Context, user identity.Requester, e model.PublishEvent) (model.PublishReply, backend.PublishStreamStatus, error) { - // Parse operation + // First, peek at the message to see if it has a "type" field that matches cursor message types + var msgPeek struct { + Type messageType `json:"type"` + } + if err := json.Unmarshal(e.Data, &msgPeek); err == nil { + // Check if this is a cursor/viewport message + if msgPeek.Type == MessageTypeCursorUpdate || msgPeek.Type == MessageTypeCursorLeave || msgPeek.Type == MessageTypeViewportUpdate { + // This is a cursor message, enrich it with user info and broadcast + var msg struct { + Type messageType `json:"type"` + SessionID string `json:"sessionId"` + UserID string `json:"userId"` + UserName string `json:"userName"` + Timestamp int64 `json:"timestamp"` + Data json.RawMessage `json:"data"` + } + if err := json.Unmarshal(e.Data, &msg); err != nil { + logger.Warn("Failed to parse cursor message", "error", err) + return model.PublishReply{}, backend.PublishStreamStatusPermissionDenied, fmt.Errorf("invalid cursor message format") + } + + // Enrich with user info + msg.UserID = user.GetRawIdentifier() + msg.UserName = user.GetName() + if msg.UserName == "" { + msg.UserName = user.GetLogin() + } + msg.Timestamp = time.Now().UnixMilli() + + // Marshal enriched message + enrichedData, err := json.Marshal(msg) + if err != nil { + logger.Warn("Failed to marshal enriched cursor message", "error", err) + return model.PublishReply{}, backend.PublishStreamStatusPermissionDenied, fmt.Errorf("internal error") + } + + // Broadcast to all subscribers + return model.PublishReply{ + Data: enrichedData, + }, backend.PublishStreamStatusOK, nil + } + } + + // Not a cursor message, treat as CRDT operation var op crdt.Operation if err := json.Unmarshal(e.Data, &op); err != nil { logger.Warn("Failed to parse operation", "error", err) diff --git a/public/app/features/explore-map/components/ExploreMapCanvas.tsx b/public/app/features/explore-map/components/ExploreMapCanvas.tsx index 5a21eb88441..d84d4cf729b 100644 --- a/public/app/features/explore-map/components/ExploreMapCanvas.tsx +++ b/public/app/features/explore-map/components/ExploreMapCanvas.tsx @@ -7,9 +7,9 @@ import { useStyles2 } from '@grafana/ui'; import { useDispatch, useSelector } from 'app/types/store'; import { useTransformContext } from '../context/TransformContext'; -import { useMockCursors } from '../hooks/useMockCursors'; +import { useCursorSync } from '../hooks/useCursorSync'; import { selectPanel as selectPanelCRDT, updateViewport as updateViewportCRDT, selectMultiplePanels as selectMultiplePanelsCRDT } from '../state/crdtSlice'; -import { selectPanels, selectViewport, selectCursors, selectSelectedPanelIds } from '../state/selectors'; +import { selectPanels, selectViewport, selectCursors, selectSelectedPanelIds, selectMapUid } from '../state/selectors'; import { ExploreMapPanelContainer } from './ExploreMapPanelContainer'; import { UserCursor } from './UserCursor'; @@ -34,9 +34,13 @@ export function ExploreMapCanvas() { const viewport = useSelector((state) => selectViewport(state.exploreMapCRDT)); const cursors = useSelector((state) => selectCursors(state.exploreMapCRDT)); const selectedPanelIds = useSelector((state) => selectSelectedPanelIds(state.exploreMapCRDT)); + const mapUid = useSelector((state) => selectMapUid(state.exploreMapCRDT)); - // Initialize mock cursors - useMockCursors(); + // Initialize cursor sync + const { updatePosition } = useCursorSync({ + mapUid: mapUid || '', + enabled: !!mapUid, + }); const handleCanvasClick = useCallback( (e: React.MouseEvent) => { @@ -82,20 +86,28 @@ export function ExploreMapCanvas() { const handleCanvasMouseMove = useCallback( (e: React.MouseEvent) => { - if (!isSelecting || !selectionRect) { + // Get position relative to the canvas element, not the event target + if (!canvasRef.current) { return; } - const canvasX = e.nativeEvent.offsetX; - const canvasY = e.nativeEvent.offsetY; + const canvasRect = canvasRef.current.getBoundingClientRect(); + const canvasX = e.clientX - canvasRect.left; + const canvasY = e.clientY - canvasRect.top; - setSelectionRect({ - ...selectionRect, - currentX: canvasX, - currentY: canvasY, - }); + // Update cursor position for all sessions + updatePosition(canvasX, canvasY); + + // Handle selection rectangle if dragging + if (isSelecting && selectionRect) { + setSelectionRect({ + ...selectionRect, + currentX: canvasX, + currentY: canvasY, + }); + } }, - [isSelecting, selectionRect] + [isSelecting, selectionRect, updatePosition] ); const handleCanvasMouseUp = useCallback( @@ -201,7 +213,7 @@ export function ExploreMapCanvas() { ); return ( -
+
{ if (e.key === 'Escape') { @@ -244,7 +255,7 @@ export function ExploreMapCanvas() { return ; })} {Object.values(cursors).map((cursor) => ( - + ))} {selectionRect && (
{ position: 'absolute', pointerEvents: 'none', zIndex: 10000, - transition: 'left 0.15s ease-out, top 0.15s ease-out', + // Smooth transition matching the update frequency (100ms) + // Using linear for more predictive movement + transition: 'left 0.1s linear, top 0.1s linear', + // Will-change hint for better performance + willChange: 'left, top', }), cursorSvg: css({ display: 'block', diff --git a/public/app/features/explore-map/hooks/useCursorSync.ts b/public/app/features/explore-map/hooks/useCursorSync.ts new file mode 100644 index 00000000000..172f6271436 --- /dev/null +++ b/public/app/features/explore-map/hooks/useCursorSync.ts @@ -0,0 +1,237 @@ +/** + * React hook for real-time cursor synchronization via Grafana Live + * + * This hook handles sending and receiving cursor position updates + * for collaborative cursor sharing across all active sessions. + */ + +import { useCallback, useEffect, useRef } from 'react'; +import { Unsubscribable } from 'rxjs'; +import { throttle } from 'lodash'; + +import { LiveChannelAddress, LiveChannelScope, isLiveChannelMessageEvent } from '@grafana/data'; +import { getGrafanaLiveSrv } from '@grafana/runtime'; +import { StoreState, useDispatch, useSelector } from 'app/types/store'; + +import { updateCursor, removeCursor } from '../state/crdtSlice'; +import { selectSessionId } from '../state/selectors'; +import { UserCursor } from '../state/types'; + +export interface CursorSyncOptions { + mapUid: string; + enabled?: boolean; + throttleMs?: number; +} + +interface CursorUpdateMessage { + type: 'cursor_update'; + sessionId: string; + userId: string; + userName: string; + data: { + x: number; + y: number; + color: string; + }; + timestamp: number; +} + +interface CursorLeaveMessage { + type: 'cursor_leave'; + sessionId: string; + userId: string; + userName: string; + timestamp: number; +} + +type CursorMessage = CursorUpdateMessage | CursorLeaveMessage; + +/** + * Hook to synchronize cursor positions with Grafana Live + */ +export function useCursorSync(options: CursorSyncOptions) { + const { mapUid, enabled = true, throttleMs = 50 } = options; + + const dispatch = useDispatch(); + const sessionId = useSelector((state: StoreState) => selectSessionId(state.exploreMapCRDT)); + + const subscriptionRef = useRef(null); + const channelAddressRef = useRef(null); + const cursorColorRef = useRef(generateRandomColor()); + + // Initialize channel connection + useEffect(() => { + if (!enabled || !mapUid) { + return; + } + + let isSubscribed = true; + + const connect = () => { + try { + const liveService = getGrafanaLiveSrv(); + if (!liveService) { + return; + } + + // Create channel address for explore-map + const channelAddress: LiveChannelAddress = { + scope: LiveChannelScope.Grafana, + namespace: 'explore-map', + path: mapUid, + }; + + channelAddressRef.current = channelAddress; + + // Subscribe to the channel stream + const subscription = liveService.getStream(channelAddress).subscribe({ + next: (event) => { + if (!isSubscribed) { + return; + } + + try { + // Handle cursor message events + if (isLiveChannelMessageEvent(event)) { + const message = event.message as CursorMessage; + + // Skip our own messages (backend should filter, but double-check) + if (message.sessionId === sessionId) { + return; + } + + if (message.type === 'cursor_update') { + const cursor: UserCursor = { + userId: message.userId, + sessionId: message.sessionId, + userName: message.userName, + color: message.data.color, + x: message.data.x, + y: message.data.y, + lastUpdated: message.timestamp, + }; + dispatch(updateCursor(cursor)); + } else if (message.type === 'cursor_leave') { + dispatch(removeCursor({ sessionId: message.sessionId })); + } + } + } catch (error) { + // Silently ignore parsing errors + } + }, + error: () => { + // Channel error - connection will be retried automatically + }, + }); + + subscriptionRef.current = subscription; + } catch (error) { + // Failed to connect - will retry on next mount + } + }; + + connect(); + + return () => { + isSubscribed = false; + + // Send cursor leave message before disconnecting + if (channelAddressRef.current) { + sendCursorLeave(); + } + + if (subscriptionRef.current) { + subscriptionRef.current.unsubscribe(); + subscriptionRef.current = null; + } + channelAddressRef.current = null; + }; + }, [mapUid, enabled, sessionId, dispatch]); + + // Send cursor leave message + const sendCursorLeave = useCallback(() => { + if (!channelAddressRef.current) { + return; + } + + const liveService = getGrafanaLiveSrv(); + if (!liveService) { + return; + } + + const message: CursorLeaveMessage = { + type: 'cursor_leave', + sessionId, + userId: '', // Will be enriched by backend + userName: '', // Will be enriched by backend + timestamp: Date.now(), + }; + + liveService.publish(channelAddressRef.current, message, { useSocket: true }).catch(() => { + // Failed to send cursor leave - ignore silently + }); + }, [sessionId]); + + // Throttled cursor update function + const sendCursorUpdate = useRef( + throttle((x: number, y: number) => { + if (!channelAddressRef.current) { + return; + } + + const liveService = getGrafanaLiveSrv(); + if (!liveService) { + return; + } + + const message: CursorUpdateMessage = { + type: 'cursor_update', + sessionId, + userId: '', // Will be enriched by backend + userName: '', // Will be enriched by backend + data: { + x, + y, + color: cursorColorRef.current, + }, + timestamp: Date.now(), + }; + + liveService.publish(channelAddressRef.current, message, { useSocket: true }).catch(() => { + // Failed to send cursor update - ignore silently + }); + }, throttleMs) + ).current; + + // Update cursor position + const updatePosition = useCallback( + (x: number, y: number) => { + sendCursorUpdate(x, y); + }, + [sendCursorUpdate] + ); + + return { + updatePosition, + color: cursorColorRef.current, + }; +} + +/** + * Generate a random color for cursor + */ +function generateRandomColor(): string { + const colors = [ + '#FF6B6B', // Red + '#4ECDC4', // Teal + '#45B7D1', // Blue + '#FFA07A', // Orange + '#98D8C8', // Mint + '#F7DC6F', // Yellow + '#BB8FCE', // Purple + '#85C1E2', // Sky Blue + '#F8B739', // Amber + '#52C1B9', // Turquoise + ]; + return colors[Math.floor(Math.random() * colors.length)]; +} diff --git a/public/app/features/explore-map/hooks/useMockCursors.ts b/public/app/features/explore-map/hooks/useMockCursors.ts deleted file mode 100644 index eee33f4437c..00000000000 --- a/public/app/features/explore-map/hooks/useMockCursors.ts +++ /dev/null @@ -1,136 +0,0 @@ -import { useEffect, useRef } from 'react'; - -import { useDispatch } from 'app/types/store'; - -import { updateCursor } from '../state/exploreMapSlice'; - -interface MockUser { - userId: string; - userName: string; - color: string; - targetX: number; - targetY: number; - currentX: number; - currentY: number; - speed: number; - pauseUntil: number; -} - -const ENABLE_MOCK_USERS = false; - -let MOCK_USERS: Array> = ENABLE_MOCK_USERS ? [ - { userId: 'mock-1', userName: 'Christian', color: '#FF6B6B', speed: 8 }, - { userId: 'mock-2', userName: 'Ryan', color: '#4ECDC4', speed: 6 }, - { userId: 'mock-3', userName: 'Marc', color: '#45B7D1', speed: 10 }, -] : []; - -// Constrain cursors to upper left area of canvas -const MOVEMENT_AREA = { - minX: 100, - maxX: 1000, - minY: 100, - maxY: 800, -}; -const UPDATE_INTERVAL = 50; // ms - -function getRandomPosition() { - return { - x: Math.random() * (MOVEMENT_AREA.maxX - MOVEMENT_AREA.minX) + MOVEMENT_AREA.minX, - y: Math.random() * (MOVEMENT_AREA.maxY - MOVEMENT_AREA.minY) + MOVEMENT_AREA.minY, - }; -} - -function getRandomPauseDuration() { - return Math.random() * 1500 + 300; // 300-1800ms pause -} - -export function useMockCursors() { - const dispatch = useDispatch(); - const mockUsersRef = useRef([]); - const animationFrameRef = useRef(); - const lastUpdateRef = useRef(0); - - useEffect(() => { - // Initialize mock users - mockUsersRef.current = MOCK_USERS.map((user) => { - const startPos = getRandomPosition(); - const targetPos = getRandomPosition(); - return { - ...user, - currentX: startPos.x, - currentY: startPos.y, - targetX: targetPos.x, - targetY: targetPos.y, - pauseUntil: 0, - }; - }); - - const animate = (timestamp: number) => { - const deltaTime = timestamp - lastUpdateRef.current; - - if (deltaTime >= UPDATE_INTERVAL) { - lastUpdateRef.current = timestamp; - - mockUsersRef.current = mockUsersRef.current.map((user) => { - const now = Date.now(); - - // If paused, skip movement - if (now < user.pauseUntil) { - return user; - } - - const dx = user.targetX - user.currentX; - const dy = user.targetY - user.currentY; - const distance = Math.sqrt(dx * dx + dy * dy); - - // If close to target, set new target and maybe pause - if (distance < 10) { - const newTarget = getRandomPosition(); - const shouldPause = Math.random() > 0.5; // 50% chance to pause - - return { - ...user, - targetX: newTarget.x, - targetY: newTarget.y, - pauseUntil: shouldPause ? now + getRandomPauseDuration() : 0, - }; - } - - // Move towards target - const moveDistance = user.speed; - const ratio = moveDistance / distance; - const newX = user.currentX + dx * ratio; - const newY = user.currentY + dy * ratio; - - // Update cursor in Redux - dispatch( - updateCursor({ - userId: user.userId, - userName: user.userName, - color: user.color, - x: newX, - y: newY, - lastUpdated: now, - }) - ); - - return { - ...user, - currentX: newX, - currentY: newY, - }; - }); - } - - animationFrameRef.current = requestAnimationFrame(animate); - }; - - animationFrameRef.current = requestAnimationFrame(animate); - - return () => { - if (animationFrameRef.current) { - cancelAnimationFrame(animationFrameRef.current); - } - }; - }, [dispatch]); -} diff --git a/public/app/features/explore-map/realtime/useRealtimeSync.ts b/public/app/features/explore-map/realtime/useRealtimeSync.ts index 491d237fe0f..1bd1b221b0e 100644 --- a/public/app/features/explore-map/realtime/useRealtimeSync.ts +++ b/public/app/features/explore-map/realtime/useRealtimeSync.ts @@ -84,7 +84,15 @@ export function useRealtimeSync(options: RealtimeSyncOptions): RealtimeSyncStatu try { // Handle message events if (isLiveChannelMessageEvent(event)) { - const operation: CRDTOperation = event.message; + const message: any = event.message; + + // Skip cursor-related messages (handled by useCursorSync) + if (message.type === 'cursor_update' || message.type === 'cursor_leave' || message.type === 'viewport_update') { + return; + } + + // Handle CRDT operations + const operation: CRDTOperation = message; // Skip if this is our own operation if (operation.nodeId === nodeId) { diff --git a/public/app/features/explore-map/state/crdtSlice.ts b/public/app/features/explore-map/state/crdtSlice.ts index d2f49e42c15..c5726eacb07 100644 --- a/public/app/features/explore-map/state/crdtSlice.ts +++ b/public/app/features/explore-map/state/crdtSlice.ts @@ -28,6 +28,9 @@ export interface ExploreMapCRDTState { // Node ID for this client nodeId: string; + // Session ID for this browser tab (unique per tab) + sessionId: string; + // Operation queue (not serialized, reconstructed on load) pendingOperations: CRDTOperation[]; @@ -48,16 +51,18 @@ const initialViewport: CanvasViewport = { }; /** - * Create initial state with a new node ID + * Create initial state with a new node ID and session ID */ export function createInitialCRDTState(mapUid?: string): ExploreMapCRDTState { const nodeId = uuidv4(); + const sessionId = uuidv4(); const manager = new CRDTStateManager(mapUid || '', nodeId); return { uid: mapUid, crdtStateJSON: JSON.stringify(manager.toJSON()), nodeId, + sessionId, pendingOperations: [], local: { viewport: initialViewport, @@ -109,6 +114,9 @@ const crdtSlice = createSlice({ loadState: (state, action: PayloadAction<{ crdtState: any }>) => { const manager = CRDTStateManager.fromJSON(action.payload.crdtState, state.nodeId); saveCRDTManager(state, manager); + // Restore uid from the loaded CRDT state + const crdtState = manager.getState(); + state.uid = crdtState.uid; }, /** @@ -550,17 +558,17 @@ const crdtSlice = createSlice({ }, /** - * Update cursor position + * Update cursor position (keyed by sessionId to support multiple sessions per user) */ updateCursor: (state, action: PayloadAction) => { - state.local.cursors[action.payload.userId] = action.payload; + state.local.cursors[action.payload.sessionId] = action.payload; }, /** - * Remove cursor + * Remove cursor (by sessionId) */ - removeCursor: (state, action: PayloadAction<{ userId: string }>) => { - delete state.local.cursors[action.payload.userId]; + removeCursor: (state, action: PayloadAction<{ sessionId: string }>) => { + delete state.local.cursors[action.payload.sessionId]; }, /** diff --git a/public/app/features/explore-map/state/selectors.ts b/public/app/features/explore-map/state/selectors.ts index dc2e8c3f1cc..c29d655ad62 100644 --- a/public/app/features/explore-map/state/selectors.ts +++ b/public/app/features/explore-map/state/selectors.ts @@ -26,8 +26,26 @@ function getCRDTManager(state: ExploreMapCRDTState): CRDTStateManager { * Select all panels as a Record (for compatibility with existing UI) */ export const selectPanels = createSelector( - [(state: ExploreMapCRDTState) => state], - (state): Record => { + [ + (state: ExploreMapCRDTState) => state.crdtStateJSON, + (state: ExploreMapCRDTState) => state.nodeId, + (state: ExploreMapCRDTState) => state.uid, + ], + (crdtStateJSON, nodeId, uid): Record => { + const state: ExploreMapCRDTState = { + uid, + crdtStateJSON, + nodeId, + sessionId: '', // Not needed for panel selection + pendingOperations: [], + local: { + viewport: { zoom: 1, panX: 0, panY: 0 }, + selectedPanelIds: [], + cursors: {}, + isOnline: false, + isSyncing: false, + }, + }; const manager = getCRDTManager(state); const panels: Record = {}; @@ -143,6 +161,13 @@ export const selectNodeId = (state: ExploreMapCRDTState): string => { return state.nodeId; }; +/** + * Select session ID + */ +export const selectSessionId = (state: ExploreMapCRDTState): string => { + return state.sessionId; +}; + /** * Select panel count */ diff --git a/public/app/features/explore-map/state/types.ts b/public/app/features/explore-map/state/types.ts index 24e085c899f..bd1c237647d 100644 --- a/public/app/features/explore-map/state/types.ts +++ b/public/app/features/explore-map/state/types.ts @@ -47,6 +47,7 @@ export interface CanvasViewport { export interface UserCursor { userId: string; + sessionId: string; userName: string; color: string; x: number;