public/grafana
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Implement cursor sync via websocket broadcast

Browse Source
This commit is contained in:
Christian Simon
2025-12-02 10:47:10 +00:00
parent b0c7930d6d
commit f6e547a7e4
9 changed files with 374 additions and 163 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/services/exploremap/realtime/channels.go
+54 -1
View File
@@ -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)
public/app/features/explore-map/components/ExploreMapCanvas.tsx
+27 -16
View File
@@ -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<HTMLDivElement>) => {
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 (
<div className={styles.canvasWrapper}>
<div className={styles.canvasWrapper} onMouseMove={handleCanvasMouseMove}>
<TransformWrapper
ref={contextTransformRef}
initialScale={viewport.zoom}
@@ -228,7 +240,6 @@ export function ExploreMapCanvas() {
className={styles.canvas}
onClick={handleCanvasClick}
onMouseDown={handleCanvasMouseDown}
onMouseMove={handleCanvasMouseMove}
onMouseUp={handleCanvasMouseUp}
onKeyDown={(e) => {
if (e.key === 'Escape') {
@@ -244,7 +255,7 @@ export function ExploreMapCanvas() {
return <ExploreMapPanelContainer key={`${panel.id}-v${panel.remoteVersion || 0}`} panel={panel} />;
})}
{Object.values(cursors).map((cursor) => (
<UserCursor key={cursor.userId} cursor={cursor} />
<UserCursor key={cursor.sessionId} cursor={cursor} />
))}
{selectionRect && (
<div
public/app/features/explore-map/components/UserCursor.tsx
+5 -1
View File
@@ -50,7 +50,11 @@ const getStyles = (theme: GrafanaTheme2) => {
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',
public/app/features/explore-map/hooks/useCursorSync.ts
+237
View File
@@ -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<Unsubscribable | null>(null);
const channelAddressRef = useRef<LiveChannelAddress | null>(null);
const cursorColorRef = useRef<string>(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<CursorMessage>(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)];
}
public/app/features/explore-map/hooks/useMockCursors.ts
-136
View File
@@ -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<Omit<MockUser, 'targetX' | 'targetY' | 'currentX' | 'currentY' | 'pauseUntil'>> = 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<MockUser[]>([]);
const animationFrameRef = useRef<number>();
const lastUpdateRef = useRef<number>(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]);
}
public/app/features/explore-map/realtime/useRealtimeSync.ts
+9 -1
View File
@@ -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) {
public/app/features/explore-map/state/crdtSlice.ts
+14 -6
View File
@@ -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<UserCursor>) => {
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];
},
/**
public/app/features/explore-map/state/selectors.ts
+27 -2
View File
@@ -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<string, ExploreMapPanel> => {
[
(state: ExploreMapCRDTState) => state.crdtStateJSON,
(state: ExploreMapCRDTState) => state.nodeId,
(state: ExploreMapCRDTState) => state.uid,
],
(crdtStateJSON, nodeId, uid): Record<string, ExploreMapPanel> => {
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<string, ExploreMapPanel> = {};
@@ -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
*/
public/app/features/explore-map/state/types.ts
+1
View File
@@ -47,6 +47,7 @@ export interface CanvasViewport {
export interface UserCursor {
userId: string;
sessionId: string;
userName: string;
color: string;
x: number;