Add state syncing via CRDT

This commit is contained in:
Aleksandar Petrov
2025-12-01 17:08:56 -04:00
parent b9ae0c98b6
commit 71526b2ab9
36 changed files with 5635 additions and 66 deletions
+2 -2
View File
@@ -727,7 +727,7 @@ func Initialize(ctx context.Context, cfg *setting.Cfg, opts Options, apiOpts api
}
csrfCSRF := csrf.ProvideCSRFFilter(cfg)
playlistService := playlistimpl.ProvideService(sqlStore, tracingService)
exploremapService := exploremapimpl.ProvideService(sqlStore, tracingService)
exploremapService := exploremapimpl.ProvideService(sqlStore, tracingService, grafanaLive)
secretsMigrator := migrator2.ProvideSecretsMigrator(serviceService, secretsService, sqlStore, ossImpl, featureToggles)
dataSourceSecretMigrationService := migrations3.ProvideDataSourceMigrationService(service15, kvStore, featureToggles)
secretMigrationProviderImpl := migrations3.ProvideSecretMigrationProvider(serverLockService, dataSourceSecretMigrationService)
@@ -1378,7 +1378,7 @@ func InitializeForTest(ctx context.Context, t sqlutil.ITestDB, testingT interfac
}
csrfCSRF := csrf.ProvideCSRFFilter(cfg)
playlistService := playlistimpl.ProvideService(sqlStore, tracingService)
exploremapService := exploremapimpl.ProvideService(sqlStore, tracingService)
exploremapService := exploremapimpl.ProvideService(sqlStore, tracingService, grafanaLive)
secretsMigrator := migrator2.ProvideSecretsMigrator(serviceService, secretsService, sqlStore, ossImpl, featureToggles)
dataSourceSecretMigrationService := migrations3.ProvideDataSourceMigrationService(service15, kvStore, featureToggles)
secretMigrationProviderImpl := migrations3.ProvideSecretMigrationProvider(serverLockService, dataSourceSecretMigrationService)
+177
View File
@@ -0,0 +1,177 @@
package crdt
import (
"encoding/json"
"time"
)
// HLCTimestamp represents a hybrid logical clock timestamp
type HLCTimestamp struct {
LogicalTime int64 `json:"logicalTime"`
WallTime int64 `json:"wallTime"` // milliseconds since epoch
NodeID string `json:"nodeId"`
}
// HybridLogicalClock manages hybrid logical time for a node
type HybridLogicalClock struct {
logicalTime int64
lastWallTime int64
nodeID string
}
// NewHybridLogicalClock creates a new HLC with the given node ID
func NewHybridLogicalClock(nodeID string) *HybridLogicalClock {
return &HybridLogicalClock{
logicalTime: 0,
lastWallTime: time.Now().UnixMilli(),
nodeID: nodeID,
}
}
// Tick advances the clock for a local event and returns a new timestamp
func (h *HybridLogicalClock) Tick() HLCTimestamp {
now := time.Now().UnixMilli()
if now > h.lastWallTime {
// Physical clock advanced - use it
h.lastWallTime = now
h.logicalTime = 0
} else {
// Physical clock hasn't advanced - increment logical component
h.logicalTime++
}
return h.Clone()
}
// Update updates the clock based on a received timestamp from another node
func (h *HybridLogicalClock) Update(received HLCTimestamp) {
now := time.Now().UnixMilli()
maxWall := max(h.lastWallTime, received.WallTime, now)
if maxWall == h.lastWallTime && maxWall == received.WallTime {
// Same wall time - take max logical time and increment
h.logicalTime = max(h.logicalTime, received.LogicalTime, 0) + 1
} else if maxWall == received.WallTime {
// Received timestamp has newer wall time
h.lastWallTime = maxWall
h.logicalTime = received.LogicalTime + 1
} else {
// Our wall time or physical clock is newer
h.lastWallTime = maxWall
h.logicalTime = 0
}
}
// Clone returns a copy of the current timestamp
func (h *HybridLogicalClock) Clone() HLCTimestamp {
return HLCTimestamp{
LogicalTime: h.logicalTime,
WallTime: h.lastWallTime,
NodeID: h.nodeID,
}
}
// Now returns the current timestamp without advancing the clock
func (h *HybridLogicalClock) Now() HLCTimestamp {
return h.Clone()
}
// GetNodeID returns the node ID
func (h *HybridLogicalClock) GetNodeID() string {
return h.nodeID
}
// CompareHLC compares two HLC timestamps
// Returns: -1 if a < b, 0 if a == b, 1 if a > b
func CompareHLC(a, b HLCTimestamp) int {
// First compare wall time
if a.WallTime != b.WallTime {
if a.WallTime < b.WallTime {
return -1
}
return 1
}
// Then compare logical time
if a.LogicalTime != b.LogicalTime {
if a.LogicalTime < b.LogicalTime {
return -1
}
return 1
}
// Finally compare node IDs for deterministic tie-breaking
if a.NodeID < b.NodeID {
return -1
} else if a.NodeID > b.NodeID {
return 1
}
return 0
}
// HappensBefore checks if timestamp a happened before timestamp b
func HappensBefore(a, b HLCTimestamp) bool {
return CompareHLC(a, b) < 0
}
// HappensAfter checks if timestamp a happened after timestamp b
func HappensAfter(a, b HLCTimestamp) bool {
return CompareHLC(a, b) > 0
}
// TimestampEquals checks if two timestamps are equal
func TimestampEquals(a, b HLCTimestamp) bool {
return CompareHLC(a, b) == 0
}
// MaxTimestamp returns the later of two timestamps
func MaxTimestamp(a, b HLCTimestamp) HLCTimestamp {
if CompareHLC(a, b) >= 0 {
return a
}
return b
}
// MarshalJSON implements json.Marshaler
func (t HLCTimestamp) MarshalJSON() ([]byte, error) {
return json.Marshal(struct {
LogicalTime int64 `json:"logicalTime"`
WallTime int64 `json:"wallTime"`
NodeID string `json:"nodeId"`
}{
LogicalTime: t.LogicalTime,
WallTime: t.WallTime,
NodeID: t.NodeID,
})
}
// UnmarshalJSON implements json.Unmarshaler
func (t *HLCTimestamp) UnmarshalJSON(data []byte) error {
var tmp struct {
LogicalTime int64 `json:"logicalTime"`
WallTime int64 `json:"wallTime"`
NodeID string `json:"nodeId"`
}
if err := json.Unmarshal(data, &tmp); err != nil {
return err
}
t.LogicalTime = tmp.LogicalTime
t.WallTime = tmp.WallTime
t.NodeID = tmp.NodeID
return nil
}
func max(a, b, c int64) int64 {
result := a
if b > result {
result = b
}
if c > result {
result = c
}
return result
}
@@ -0,0 +1,82 @@
package crdt
import (
"encoding/json"
)
// LWWRegister represents a Last-Write-Wins Register CRDT
type LWWRegister struct {
value interface{}
timestamp HLCTimestamp
}
// LWWRegisterJSON is the JSON representation of LWWRegister
type LWWRegisterJSON struct {
Value interface{} `json:"value"`
Timestamp HLCTimestamp `json:"timestamp"`
}
// NewLWWRegister creates a new LWW-Register with an initial value and timestamp
func NewLWWRegister(value interface{}, timestamp HLCTimestamp) *LWWRegister {
return &LWWRegister{
value: value,
timestamp: timestamp,
}
}
// Set sets the register value if the new timestamp is greater than current
// Returns true if the value was updated, false if update was ignored
func (r *LWWRegister) Set(value interface{}, timestamp HLCTimestamp) bool {
// Only update if new timestamp is strictly greater
if CompareHLC(timestamp, r.timestamp) > 0 {
r.value = value
r.timestamp = timestamp
return true
}
return false
}
// Get returns the current value
func (r *LWWRegister) Get() interface{} {
return r.value
}
// GetTimestamp returns the current timestamp
func (r *LWWRegister) GetTimestamp() HLCTimestamp {
return r.timestamp
}
// Merge merges another LWW-Register into this one
// Keeps the value with the highest timestamp
// Returns true if this register's value was updated
func (r *LWWRegister) Merge(other *LWWRegister) bool {
return r.Set(other.value, other.timestamp)
}
// Clone creates a copy of the register
func (r *LWWRegister) Clone() *LWWRegister {
return &LWWRegister{
value: r.value,
timestamp: r.timestamp,
}
}
// MarshalJSON implements json.Marshaler
func (r *LWWRegister) MarshalJSON() ([]byte, error) {
return json.Marshal(LWWRegisterJSON{
Value: r.value,
Timestamp: r.timestamp,
})
}
// UnmarshalJSON implements json.Unmarshaler
func (r *LWWRegister) UnmarshalJSON(data []byte) error {
var tmp LWWRegisterJSON
if err := json.Unmarshal(data, &tmp); err != nil {
return err
}
r.value = tmp.Value
r.timestamp = tmp.Timestamp
return nil
}
+134
View File
@@ -0,0 +1,134 @@
package crdt
import (
"encoding/json"
)
// OperationType represents the type of CRDT operation
type OperationType string
const (
OpAddPanel OperationType = "add-panel"
OpRemovePanel OperationType = "remove-panel"
OpUpdatePanelPosition OperationType = "update-panel-position"
OpUpdatePanelSize OperationType = "update-panel-size"
OpUpdatePanelZIndex OperationType = "update-panel-zindex"
OpUpdatePanelExplore OperationType = "update-panel-explore-state"
OpUpdateTitle OperationType = "update-title"
OpBatch OperationType = "batch"
)
// Operation represents a CRDT operation
type Operation struct {
Type OperationType `json:"type"`
MapUID string `json:"mapUid"`
OperationID string `json:"operationId"`
Timestamp HLCTimestamp `json:"timestamp"`
NodeID string `json:"nodeId"`
Payload json.RawMessage `json:"payload"`
}
// AddPanelPayload represents the payload for add-panel operation
type AddPanelPayload struct {
PanelID string `json:"panelId"`
ExploreID string `json:"exploreId"`
Position PanelPosition `json:"position"`
}
// RemovePanelPayload represents the payload for remove-panel operation
type RemovePanelPayload struct {
PanelID string `json:"panelId"`
ObservedTags []string `json:"observedTags"`
}
// UpdatePanelPositionPayload represents the payload for update-panel-position operation
type UpdatePanelPositionPayload struct {
PanelID string `json:"panelId"`
X float64 `json:"x"`
Y float64 `json:"y"`
}
// UpdatePanelSizePayload represents the payload for update-panel-size operation
type UpdatePanelSizePayload struct {
PanelID string `json:"panelId"`
Width float64 `json:"width"`
Height float64 `json:"height"`
}
// UpdatePanelZIndexPayload represents the payload for update-panel-zindex operation
type UpdatePanelZIndexPayload struct {
PanelID string `json:"panelId"`
ZIndex int64 `json:"zIndex"`
}
// UpdatePanelExploreStatePayload represents the payload for update-panel-explore-state operation
type UpdatePanelExploreStatePayload struct {
PanelID string `json:"panelId"`
ExploreState interface{} `json:"exploreState"`
}
// UpdateTitlePayload represents the payload for update-title operation
type UpdateTitlePayload struct {
Title string `json:"title"`
}
// BatchPayload represents the payload for batch operation
type BatchPayload struct {
Operations []Operation `json:"operations"`
}
// PanelPosition represents the position and size of a panel
type PanelPosition struct {
X float64 `json:"x"`
Y float64 `json:"y"`
Width float64 `json:"width"`
Height float64 `json:"height"`
}
// ParsePayload parses the operation payload into the appropriate type
func (op *Operation) ParsePayload() (interface{}, error) {
switch op.Type {
case OpAddPanel:
var payload AddPanelPayload
err := json.Unmarshal(op.Payload, &payload)
return payload, err
case OpRemovePanel:
var payload RemovePanelPayload
err := json.Unmarshal(op.Payload, &payload)
return payload, err
case OpUpdatePanelPosition:
var payload UpdatePanelPositionPayload
err := json.Unmarshal(op.Payload, &payload)
return payload, err
case OpUpdatePanelSize:
var payload UpdatePanelSizePayload
err := json.Unmarshal(op.Payload, &payload)
return payload, err
case OpUpdatePanelZIndex:
var payload UpdatePanelZIndexPayload
err := json.Unmarshal(op.Payload, &payload)
return payload, err
case OpUpdatePanelExplore:
var payload UpdatePanelExploreStatePayload
err := json.Unmarshal(op.Payload, &payload)
return payload, err
case OpUpdateTitle:
var payload UpdateTitlePayload
err := json.Unmarshal(op.Payload, &payload)
return payload, err
case OpBatch:
var payload BatchPayload
err := json.Unmarshal(op.Payload, &payload)
return payload, err
default:
return nil, nil
}
}
+215
View File
@@ -0,0 +1,215 @@
package crdt
import (
"encoding/json"
)
// ORSet represents an Observed-Remove Set CRDT
type ORSet struct {
adds map[string]map[string]bool // element -> set of tags
removes map[string]bool // set of removed tags
}
// ORSetJSON is the JSON representation of ORSet
type ORSetJSON struct {
Adds map[string][]string `json:"adds"`
Removes []string `json:"removes"`
}
// NewORSet creates a new empty OR-Set
func NewORSet() *ORSet {
return &ORSet{
adds: make(map[string]map[string]bool),
removes: make(map[string]bool),
}
}
// Add adds an element to the set with a unique tag
func (s *ORSet) Add(element, tag string) {
if s.adds[element] == nil {
s.adds[element] = make(map[string]bool)
}
s.adds[element][tag] = true
}
// Remove removes an element from the set
// Only removes the specific tags that were observed
func (s *ORSet) Remove(element string, observedTags []string) {
for _, tag := range observedTags {
s.removes[tag] = true
}
// Clean up the element's tags
if elementTags, exists := s.adds[element]; exists {
for _, tag := range observedTags {
delete(elementTags, tag)
}
// If no tags remain, remove the element entry
if len(elementTags) == 0 {
delete(s.adds, element)
}
}
}
// Contains checks if an element is in the set
// Element is present if it has at least one non-removed tag
func (s *ORSet) Contains(element string) bool {
tags, exists := s.adds[element]
if !exists || len(tags) == 0 {
return false
}
// Element is present if it has at least one tag that hasn't been removed
for tag := range tags {
if !s.removes[tag] {
return true
}
}
return false
}
// GetTags returns all tags for an element (including removed ones)
func (s *ORSet) GetTags(element string) []string {
tags, exists := s.adds[element]
if !exists {
return []string{}
}
result := make([]string, 0, len(tags))
for tag := range tags {
result = append(result, tag)
}
return result
}
// Values returns all elements currently in the set
func (s *ORSet) Values() []string {
result := make([]string, 0, len(s.adds))
for element, tags := range s.adds {
// Include element if it has at least one non-removed tag
for tag := range tags {
if !s.removes[tag] {
result = append(result, element)
break
}
}
}
return result
}
// Size returns the number of elements in the set
func (s *ORSet) Size() int {
return len(s.Values())
}
// IsEmpty checks if the set is empty
func (s *ORSet) IsEmpty() bool {
return s.Size() == 0
}
// Merge merges another OR-Set into this one
// Takes the union of all adds and removes
func (s *ORSet) Merge(other *ORSet) {
// Merge adds (union of all tags)
for element, otherTags := range other.adds {
if s.adds[element] == nil {
s.adds[element] = make(map[string]bool)
}
for tag := range otherTags {
s.adds[element][tag] = true
}
}
// Merge removes (union of all removed tags)
for tag := range other.removes {
s.removes[tag] = true
}
// Clean up elements with all tags removed
for element, tags := range s.adds {
hasLiveTag := false
for tag := range tags {
if !s.removes[tag] {
hasLiveTag = true
break
}
}
if !hasLiveTag {
delete(s.adds, element)
}
}
}
// Clone creates a deep copy of the OR-Set
func (s *ORSet) Clone() *ORSet {
clone := NewORSet()
// Deep copy adds
for element, tags := range s.adds {
clone.adds[element] = make(map[string]bool)
for tag := range tags {
clone.adds[element][tag] = true
}
}
// Deep copy removes
for tag := range s.removes {
clone.removes[tag] = true
}
return clone
}
// Clear removes all elements from the set
func (s *ORSet) Clear() {
s.adds = make(map[string]map[string]bool)
s.removes = make(map[string]bool)
}
// MarshalJSON implements json.Marshaler
func (s *ORSet) MarshalJSON() ([]byte, error) {
adds := make(map[string][]string)
for element, tags := range s.adds {
tagList := make([]string, 0, len(tags))
for tag := range tags {
tagList = append(tagList, tag)
}
adds[element] = tagList
}
removes := make([]string, 0, len(s.removes))
for tag := range s.removes {
removes = append(removes, tag)
}
return json.Marshal(ORSetJSON{
Adds: adds,
Removes: removes,
})
}
// UnmarshalJSON implements json.Unmarshaler
func (s *ORSet) UnmarshalJSON(data []byte) error {
var tmp ORSetJSON
if err := json.Unmarshal(data, &tmp); err != nil {
return err
}
s.adds = make(map[string]map[string]bool)
for element, tags := range tmp.Adds {
s.adds[element] = make(map[string]bool)
for _, tag := range tags {
s.adds[element][tag] = true
}
}
s.removes = make(map[string]bool)
for _, tag := range tmp.Removes {
s.removes[tag] = true
}
return nil
}
+134
View File
@@ -0,0 +1,134 @@
package crdt
import (
"encoding/json"
)
// PNCounter represents a Positive-Negative Counter CRDT
type PNCounter struct {
increments map[string]int64 // nodeId -> count
decrements map[string]int64 // nodeId -> count
}
// PNCounterJSON is the JSON representation of PNCounter
type PNCounterJSON struct {
Increments map[string]int64 `json:"increments"`
Decrements map[string]int64 `json:"decrements"`
}
// NewPNCounter creates a new PN-Counter
func NewPNCounter() *PNCounter {
return &PNCounter{
increments: make(map[string]int64),
decrements: make(map[string]int64),
}
}
// Increment increments the counter for a specific node
func (c *PNCounter) Increment(nodeID string, delta int64) {
if delta < 0 {
panic("delta must be non-negative for increment")
}
c.increments[nodeID] += delta
}
// Decrement decrements the counter for a specific node
func (c *PNCounter) Decrement(nodeID string, delta int64) {
if delta < 0 {
panic("delta must be non-negative for decrement")
}
c.decrements[nodeID] += delta
}
// Value returns the current value of the counter
// Value = sum of all increments - sum of all decrements
func (c *PNCounter) Value() int64 {
var sum int64 = 0
// Add all increments
for _, count := range c.increments {
sum += count
}
// Subtract all decrements
for _, count := range c.decrements {
sum -= count
}
return sum
}
// Next gets the next value and increments the counter for a node
// This is useful for allocating sequential IDs (like z-indices)
func (c *PNCounter) Next(nodeID string) int64 {
nextValue := c.Value() + 1
c.Increment(nodeID, 1)
return nextValue
}
// Merge merges another PN-Counter into this one
// Takes the maximum value for each node's counters
func (c *PNCounter) Merge(other *PNCounter) {
// Merge increments (take max for each node)
for nodeID, count := range other.increments {
if current, exists := c.increments[nodeID]; !exists || count > current {
c.increments[nodeID] = count
}
}
// Merge decrements (take max for each node)
for nodeID, count := range other.decrements {
if current, exists := c.decrements[nodeID]; !exists || count > current {
c.decrements[nodeID] = count
}
}
}
// Clone creates a copy of the counter
func (c *PNCounter) Clone() *PNCounter {
clone := NewPNCounter()
for nodeID, count := range c.increments {
clone.increments[nodeID] = count
}
for nodeID, count := range c.decrements {
clone.decrements[nodeID] = count
}
return clone
}
// Reset resets the counter to zero
func (c *PNCounter) Reset() {
c.increments = make(map[string]int64)
c.decrements = make(map[string]int64)
}
// MarshalJSON implements json.Marshaler
func (c *PNCounter) MarshalJSON() ([]byte, error) {
return json.Marshal(PNCounterJSON{
Increments: c.increments,
Decrements: c.decrements,
})
}
// UnmarshalJSON implements json.Unmarshaler
func (c *PNCounter) UnmarshalJSON(data []byte) error {
var tmp PNCounterJSON
if err := json.Unmarshal(data, &tmp); err != nil {
return err
}
c.increments = tmp.Increments
if c.increments == nil {
c.increments = make(map[string]int64)
}
c.decrements = tmp.Decrements
if c.decrements == nil {
c.decrements = make(map[string]int64)
}
return nil
}
@@ -2,25 +2,71 @@ package exploremapimpl
import (
"context"
"time"
"github.com/grafana/grafana/pkg/infra/db"
"github.com/grafana/grafana/pkg/infra/tracing"
"github.com/grafana/grafana/pkg/services/exploremap"
"github.com/grafana/grafana/pkg/services/exploremap/realtime"
"github.com/grafana/grafana/pkg/services/live"
)
type Service struct {
store store
tracer tracing.Tracer
hub *realtime.OperationHub
}
var _ exploremap.Service = &Service{}
func ProvideService(db db.DB, tracer tracing.Tracer) exploremap.Service {
// storeAdapter adapts the internal store interface to the realtime Store interface
type storeAdapter struct {
store store
}
func (s *storeAdapter) Update(ctx context.Context, cmd *exploremap.UpdateExploreMapCommand) (*exploremap.ExploreMapDTO, error) {
return s.store.Update(ctx, cmd)
}
func (s *storeAdapter) Get(ctx context.Context, query *exploremap.GetExploreMapByUIDQuery) (*exploremap.ExploreMapDTO, error) {
// Get returns ExploreMap, but we need ExploreMapDTO
mapData, err := s.store.Get(ctx, query)
if err != nil {
return nil, err
}
return &exploremap.ExploreMapDTO{
UID: mapData.UID,
Title: mapData.Title,
Data: mapData.Data,
CreatedBy: mapData.CreatedBy,
UpdatedBy: mapData.UpdatedBy,
CreatedAt: mapData.CreatedAt,
UpdatedAt: mapData.UpdatedAt,
}, nil
}
func ProvideService(db db.DB, tracer tracing.Tracer, liveService *live.GrafanaLive) exploremap.Service {
store := &sqlStore{
db: db,
}
// Create adapter for realtime hub
adapter := &storeAdapter{store: store}
// Create operation hub for CRDT synchronization
hub := realtime.NewOperationHub(liveService, adapter)
// Register channel handler with Grafana Live
channelHandler := realtime.NewExploreMapChannelHandler(hub)
liveService.GrafanaScope.Features["explore-map"] = channelHandler
// Start background snapshot worker (saves CRDT state to SQL every 30 seconds)
go hub.StartSnapshotWorker(context.Background(), 30*time.Second)
return &Service{
tracer: tracer,
store: &sqlStore{
db: db,
},
store: store,
hub: hub,
}
}
@@ -0,0 +1,97 @@
package realtime
import (
"context"
"encoding/json"
"fmt"
"strings"
"github.com/grafana/grafana-plugin-sdk-go/backend"
"github.com/grafana/grafana/pkg/apimachinery/identity"
"github.com/grafana/grafana/pkg/services/exploremap/crdt"
"github.com/grafana/grafana/pkg/services/live/model"
)
// ExploreMapChannelHandler handles Grafana Live channels for Explore Maps
type ExploreMapChannelHandler struct {
hub *OperationHub
}
// NewExploreMapChannelHandler creates a new channel handler
func NewExploreMapChannelHandler(hub *OperationHub) *ExploreMapChannelHandler {
return &ExploreMapChannelHandler{
hub: hub,
}
}
// OnSubscribe is called when a client subscribes to the channel
func (h *ExploreMapChannelHandler) OnSubscribe(ctx context.Context, user identity.Requester, e model.SubscribeEvent) (model.SubscribeReply, backend.SubscribeStreamStatus, error) {
// Extract map UID from channel path
// Channel format: grafana/explore-map/{mapUid}
mapUID := extractMapUID(e.Channel)
if mapUID == "" {
return model.SubscribeReply{}, backend.SubscribeStreamStatusNotFound, nil
}
// TODO: Check access permissions
// For now, allow all authenticated users
// Get current CRDT state
state, err := h.hub.GetState(ctx, mapUID)
if err != nil {
return model.SubscribeReply{}, backend.SubscribeStreamStatusNotFound, nil
}
// Serialize state as initial data
state.mu.RLock()
stateData := map[string]interface{}{
"title": state.Title,
"panels": state.Panels,
"zIndex": state.ZIndex,
}
state.mu.RUnlock()
data, err := json.Marshal(stateData)
if err != nil {
return model.SubscribeReply{}, backend.SubscribeStreamStatusPermissionDenied, err
}
return model.SubscribeReply{
Data: data,
}, backend.SubscribeStreamStatusOK, nil
}
// 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
var op crdt.Operation
if err := json.Unmarshal(e.Data, &op); err != nil {
logger.Warn("Failed to parse operation", "error", err)
return model.PublishReply{}, backend.PublishStreamStatusPermissionDenied, fmt.Errorf("failed to parse operation: %w", err)
}
// TODO: Validate user has permission to modify the map
// Process operation through hub
if err := h.hub.HandleOperation(ctx, op); err != nil {
logger.Warn("Failed to handle operation", "error", err)
return model.PublishReply{}, backend.PublishStreamStatusPermissionDenied, err
}
return model.PublishReply{}, backend.PublishStreamStatusOK, nil
}
// GetHandlerForPath returns the channel handler for a given path
func (h *ExploreMapChannelHandler) GetHandlerForPath(path string) (model.ChannelHandler, error) {
return h, nil
}
// extractMapUID extracts the map UID from a channel path
// Channel format: grafana/explore-map/{mapUid}
func extractMapUID(channel string) string {
parts := strings.Split(channel, "/")
if len(parts) >= 3 && parts[1] == "explore-map" {
return parts[2]
}
return ""
}
+247
View File
@@ -0,0 +1,247 @@
package realtime
import (
"context"
"encoding/json"
"fmt"
"sync"
"time"
"github.com/grafana/grafana/pkg/infra/log"
"github.com/grafana/grafana/pkg/services/exploremap"
"github.com/grafana/grafana/pkg/services/exploremap/crdt"
"github.com/grafana/grafana/pkg/services/live"
)
var logger = log.New("exploremap.realtime")
// Store interface for saving map snapshots
type Store interface {
Update(ctx context.Context, cmd *exploremap.UpdateExploreMapCommand) (*exploremap.ExploreMapDTO, error)
Get(ctx context.Context, query *exploremap.GetExploreMapByUIDQuery) (*exploremap.ExploreMapDTO, error)
}
// OperationHub manages real-time CRDT operations
type OperationHub struct {
liveService *live.GrafanaLive
store Store
states *StateCache
mu sync.RWMutex
}
// StateCache holds in-memory CRDT states for active maps
type StateCache struct {
states map[string]*MapState
mu sync.RWMutex
}
// MapState represents the CRDT state for a single map
type MapState struct {
UID string
OrgID int64
Title *crdt.LWWRegister
Panels *crdt.ORSet
ZIndex *crdt.PNCounter
Updated time.Time
mu sync.RWMutex
}
// NewOperationHub creates a new operation hub
func NewOperationHub(liveService *live.GrafanaLive, store Store) *OperationHub {
return &OperationHub{
liveService: liveService,
store: store,
states: &StateCache{
states: make(map[string]*MapState),
},
}
}
// HandleOperation processes an incoming CRDT operation
func (h *OperationHub) HandleOperation(ctx context.Context, op crdt.Operation) error {
// Get or create state
state, err := h.states.GetOrCreate(op.MapUID)
if err != nil {
return fmt.Errorf("failed to get state: %w", err)
}
state.mu.Lock()
defer state.mu.Unlock()
// Apply operation to CRDT state
if err := h.applyOperation(state, op); err != nil {
return fmt.Errorf("failed to apply operation: %w", err)
}
state.Updated = time.Now()
// Broadcast to all connected clients
// Include grafana scope prefix to match subscription channel
channel := fmt.Sprintf("grafana/explore-map/%s", op.MapUID)
data, err := json.Marshal(op)
if err != nil {
return fmt.Errorf("failed to marshal operation: %w", err)
}
// TODO: Get actual orgID from map state or context
// For now, use 1 as the orgID (Grafana default org)
orgID := int64(1)
if err := h.liveService.Publish(orgID, channel, data); err != nil {
logger.Warn("Failed to broadcast operation", "error", err, "mapUid", op.MapUID, "channel", channel)
return fmt.Errorf("failed to broadcast operation: %w", err)
}
return nil
}
// applyOperation applies a CRDT operation to the state
func (h *OperationHub) applyOperation(state *MapState, op crdt.Operation) error {
payload, err := op.ParsePayload()
if err != nil {
return fmt.Errorf("failed to parse payload: %w", err)
}
switch op.Type {
case crdt.OpAddPanel:
p := payload.(crdt.AddPanelPayload)
state.Panels.Add(p.PanelID, op.OperationID)
// Allocate z-index
state.ZIndex.Next(op.NodeID)
case crdt.OpRemovePanel:
p := payload.(crdt.RemovePanelPayload)
state.Panels.Remove(p.PanelID, p.ObservedTags)
case crdt.OpUpdateTitle:
p := payload.(crdt.UpdateTitlePayload)
state.Title.Set(p.Title, op.Timestamp)
case crdt.OpBatch:
p := payload.(crdt.BatchPayload)
for _, subOp := range p.Operations {
if err := h.applyOperation(state, subOp); err != nil {
return err
}
}
// Other operation types don't need special handling in the hub
// They're applied at the client level
}
return nil
}
// GetState returns the current CRDT state for a map
func (h *OperationHub) GetState(ctx context.Context, mapUID string) (*MapState, error) {
return h.states.GetOrCreate(mapUID)
}
// SnapshotState persists the current CRDT state to the database
func (h *OperationHub) SnapshotState(ctx context.Context, mapUID string) error {
state, err := h.states.Get(mapUID)
if err != nil {
return err
}
state.mu.RLock()
defer state.mu.RUnlock()
// Skip if OrgID not set - map may not exist in database yet
if state.OrgID == 0 {
return exploremap.ErrExploreMapNotFound
}
orgID := state.OrgID
// Serialize state to JSON
stateData := map[string]interface{}{
"title": state.Title,
"panels": state.Panels,
"zIndex": state.ZIndex,
}
data, err := json.Marshal(stateData)
if err != nil {
return fmt.Errorf("failed to marshal state: %w", err)
}
// Update in database
_, err = h.store.Update(ctx, &exploremap.UpdateExploreMapCommand{
UID: mapUID,
OrgID: orgID,
Data: string(data),
})
return err
}
// StartSnapshotWorker starts a background worker that periodically snapshots states
func (h *OperationHub) StartSnapshotWorker(ctx context.Context, interval time.Duration) {
ticker := time.NewTicker(interval)
defer ticker.Stop()
for {
select {
case <-ticker.C:
h.snapshotAll(ctx)
case <-ctx.Done():
return
}
}
}
func (h *OperationHub) snapshotAll(ctx context.Context) {
h.states.mu.RLock()
mapUIDs := make([]string, 0, len(h.states.states))
for uid := range h.states.states {
mapUIDs = append(mapUIDs, uid)
}
h.states.mu.RUnlock()
for _, uid := range mapUIDs {
if err := h.SnapshotState(ctx, uid); err != nil {
// Ignore "not found" errors - map may have been deleted or not yet created
if err != exploremap.ErrExploreMapNotFound {
logger.Warn("Failed to snapshot state", "error", err, "mapUid", uid)
}
}
}
}
// StateCache methods
func (sc *StateCache) Get(mapUID string) (*MapState, error) {
sc.mu.RLock()
defer sc.mu.RUnlock()
state, exists := sc.states[mapUID]
if !exists {
return nil, fmt.Errorf("state not found for map: %s", mapUID)
}
return state, nil
}
func (sc *StateCache) GetOrCreate(mapUID string) (*MapState, error) {
sc.mu.Lock()
defer sc.mu.Unlock()
state, exists := sc.states[mapUID]
if !exists {
// Create new state
state = &MapState{
UID: mapUID,
Title: crdt.NewLWWRegister("Untitled Map", crdt.HLCTimestamp{}),
Panels: crdt.NewORSet(),
ZIndex: crdt.NewPNCounter(),
Updated: time.Now(),
}
sc.states[mapUID] = state
}
return state, nil
}
func (sc *StateCache) Remove(mapUID string) {
sc.mu.Lock()
defer sc.mu.Unlock()
delete(sc.states, mapUID)
}
@@ -1,5 +1,5 @@
import { css } from '@emotion/css';
import { useEffect, useRef } from 'react';
import { useCallback, useEffect, useRef } from 'react';
import { useParams } from 'react-router-dom-v5-compat';
import { ReactZoomPanPinchRef } from 'react-zoom-pan-pinch';
@@ -15,6 +15,7 @@ import { ExploreMapFloatingToolbar } from './components/ExploreMapFloatingToolba
import { ExploreMapToolbar } from './components/ExploreMapToolbar';
import { TransformProvider } from './context/TransformContext';
import { useCanvasPersistence } from './hooks/useCanvasPersistence';
import { useRealtimeSync } from './realtime/useRealtimeSync';
export default function ExploreMapPage(props: GrafanaRouteComponentProps<{ uid?: string }>) {
const styles = useStyles2(getStyles);
@@ -26,6 +27,28 @@ export default function ExploreMapPage(props: GrafanaRouteComponentProps<{ uid?:
// Initialize canvas persistence (with uid if available)
const { loading } = useCanvasPersistence({ uid });
// Stable callback references for realtime sync
const handleConnected = useCallback(() => {
// Connected to real-time sync
}, []);
const handleDisconnected = useCallback(() => {
// Disconnected from real-time sync
}, []);
const handleError = useCallback((error: Error) => {
console.error('CRDT sync error:', error);
}, []);
// Enable real-time CRDT synchronization when uid is available
useRealtimeSync({
mapUid: uid || '',
enabled: !!uid,
onConnected: handleConnected,
onDisconnected: handleDisconnected,
onError: handleError,
});
useEffect(() => {
chrome.update({
sectionNav: navModel,
@@ -8,7 +8,8 @@ import { useDispatch, useSelector } from 'app/types/store';
import { useTransformContext } from '../context/TransformContext';
import { useMockCursors } from '../hooks/useMockCursors';
import { selectMultiplePanels, selectPanel, updateViewport } from '../state/exploreMapSlice';
import { selectPanel as selectPanelCRDT, updateViewport as updateViewportCRDT, selectMultiplePanels as selectMultiplePanelsCRDT } from '../state/crdtSlice';
import { selectPanels, selectViewport, selectCursors, selectSelectedPanelIds } from '../state/selectors';
import { ExploreMapPanelContainer } from './ExploreMapPanelContainer';
import { UserCursor } from './UserCursor';
@@ -29,9 +30,10 @@ export function ExploreMapCanvas() {
const [isSelecting, setIsSelecting] = useState(false);
const justCompletedSelectionRef = useRef(false);
const panels = useSelector((state) => state.exploreMap.panels);
const viewport = useSelector((state) => state.exploreMap.viewport);
const cursors = useSelector((state) => state.exploreMap.cursors);
const panels = useSelector((state) => selectPanels(state.exploreMapCRDT));
const viewport = useSelector((state) => selectViewport(state.exploreMapCRDT));
const cursors = useSelector((state) => selectCursors(state.exploreMapCRDT));
const selectedPanelIds = useSelector((state) => selectSelectedPanelIds(state.exploreMapCRDT));
// Initialize mock cursors
useMockCursors();
@@ -44,12 +46,12 @@ export function ExploreMapCanvas() {
return;
}
// Only deselect if clicking directly on canvas (not on panels)
if (e.target === e.currentTarget) {
dispatch(selectPanel({ panelId: undefined }));
// Only deselect if clicking directly on canvas (not on panels) and there are panels to deselect
if (e.target === e.currentTarget && selectedPanelIds.length > 0) {
dispatch(selectPanelCRDT({ panelId: undefined }));
}
},
[dispatch]
[dispatch, selectedPanelIds]
);
const handleCanvasMouseDown = useCallback(
@@ -133,12 +135,12 @@ export function ExploreMapCanvas() {
if (selectedPanelIds.length > 0) {
// Select all panels at once
console.log('Dispatching selectMultiplePanels with:', { panelIds: selectedPanelIds, addToSelection: isAdditive });
dispatch(selectMultiplePanels({ panelIds: selectedPanelIds, addToSelection: isAdditive }));
dispatch(selectMultiplePanelsCRDT({ panelIds: selectedPanelIds, addToSelection: isAdditive }));
console.log('After dispatch');
justCompletedSelectionRef.current = true;
} else if (!isAdditive) {
// Clear selection if no panels selected and not holding modifier
dispatch(selectPanel({ panelId: undefined }));
dispatch(selectPanelCRDT({ panelId: undefined }));
}
setSelectionRect(null);
@@ -150,7 +152,7 @@ export function ExploreMapCanvas() {
const handleTransformChange = useCallback(
(ref: ReactZoomPanPinchRef) => {
dispatch(
updateViewport({
updateViewportCRDT({
zoom: ref.state.scale,
panX: ref.state.positionX,
panY: ref.state.positionY,
@@ -230,7 +232,7 @@ export function ExploreMapCanvas() {
onMouseUp={handleCanvasMouseUp}
onKeyDown={(e) => {
if (e.key === 'Escape') {
dispatch(selectPanel({ panelId: undefined }));
dispatch(selectPanelCRDT({ panelId: undefined }));
}
}}
role="button"
@@ -6,7 +6,7 @@ import { Trans } from '@grafana/i18n';
import { Button, useStyles2 } from '@grafana/ui';
import { useDispatch } from 'app/types/store';
import { addPanel } from '../state/exploreMapSlice';
import { addPanel } from '../state/crdtSlice';
export function ExploreMapFloatingToolbar() {
const styles = useStyles2(getStyles);
@@ -15,7 +15,9 @@ import {
selectPanel,
updateMultiplePanelPositions,
updatePanelPosition,
} from '../state/exploreMapSlice';
updatePanelSize,
} from '../state/crdtSlice';
import { selectSelectedPanelIds, selectViewport } from '../state/selectors';
import { ExploreMapPanel } from '../state/types';
import { ExploreMapPanelContent } from './ExploreMapPanelContent';
@@ -30,8 +32,8 @@ export function ExploreMapPanelContainer({ panel }: ExploreMapPanelContainerProp
const rndRef = useRef<Rnd>(null);
const [dragStartPos, setDragStartPos] = useState<{ x: number; y: number } | null>(null);
const selectedPanelIds = useSelector((state) => state.exploreMap.selectedPanelIds || []);
const viewport = useSelector((state) => state.exploreMap.viewport);
const selectedPanelIds = useSelector((state) => selectSelectedPanelIds(state.exploreMapCRDT));
const viewport = useSelector((state) => selectViewport(state.exploreMapCRDT));
const isSelected = selectedPanelIds.includes(panel.id);
const handleDragStart: RndDragCallback = useCallback(
@@ -85,7 +87,8 @@ export function ExploreMapPanelContainer({ panel }: ExploreMapPanelContainerProp
dispatch(
updatePanelPosition({
panelId: panel.id,
position: { x: data.x, y: data.y },
x: data.x,
y: data.y,
})
);
} else {
@@ -93,7 +96,8 @@ export function ExploreMapPanelContainer({ panel }: ExploreMapPanelContainerProp
dispatch(
updatePanelPosition({
panelId: panel.id,
position: { x: data.x, y: data.y },
x: data.x,
y: data.y,
})
);
}
@@ -108,15 +112,21 @@ export function ExploreMapPanelContainer({ panel }: ExploreMapPanelContainerProp
const newWidth = ref.offsetWidth;
const newHeight = ref.offsetHeight;
// Update position
dispatch(
updatePanelPosition({
panelId: panel.id,
position: {
x: position.x,
y: position.y,
width: newWidth,
height: newHeight,
},
x: position.x,
y: position.y,
})
);
// Update size
dispatch(
updatePanelSize({
panelId: panel.id,
width: newWidth,
height: newHeight,
})
);
@@ -134,7 +144,6 @@ export function ExploreMapPanelContainer({ panel }: ExploreMapPanelContainerProp
// If this panel is already selected and we're not multi-selecting,
// don't change selection (allows dragging multiple selected panels)
if (isSelected && !isMultiSelect) {
// Just bring to front, don't change selection
dispatch(bringPanelToFront({ panelId: panel.id }));
return;
}
@@ -212,6 +221,7 @@ export function ExploreMapPanelContainer({ panel }: ExploreMapPanelContainerProp
</div>
<div className={styles.panelContent}>
<ExploreMapPanelContent
panelId={panel.id}
exploreId={panel.exploreId}
width={panel.position.width}
height={panel.position.height - 36}
@@ -9,8 +9,12 @@ import { useDispatch, useSelector } from 'app/types/store';
import { ExplorePaneContainer } from '../../explore/ExplorePaneContainer';
import { DEFAULT_RANGE } from '../../explore/state/constants';
import { initializeExplore } from '../../explore/state/explorePane';
// import { useExploreStateReceiver } from '../hooks/useExploreStateReceiver';
// import { useExploreStateSync } from '../hooks/useExploreStateSync';
import { selectPanels } from '../state/selectors';
interface ExploreMapPanelContentProps {
panelId: string;
exploreId: string;
width: number;
height: number;
@@ -73,7 +77,7 @@ function patchGetBoundingClientRect() {
isPatched = true;
}
export function ExploreMapPanelContent({ exploreId, width, height }: ExploreMapPanelContentProps) {
export function ExploreMapPanelContent({ panelId, exploreId, width, height }: ExploreMapPanelContentProps) {
const styles = useStyles2(getStyles);
const dispatch = useDispatch();
const [isInitialized, setIsInitialized] = useState(false);
@@ -81,18 +85,31 @@ export function ExploreMapPanelContent({ exploreId, width, height }: ExploreMapP
// Create scoped event bus for this panel
const eventBus = useMemo(() => new EventBusSrv(), []);
// TODO: Re-enable once we fix the re-rendering issue
// Sync Explore state changes to CRDT (outgoing)
// useExploreStateSync({
// panelId,
// exploreId,
// enabled: isInitialized,
// });
// Receive and apply Explore state changes from CRDT (incoming)
// useExploreStateReceiver({
// panelId,
// exploreId,
// enabled: isInitialized,
// });
// Patch getBoundingClientRect on mount
useEffect(() => {
patchGetBoundingClientRect();
}, []);
// Check if the explore pane exists in Redux
const explorePane = useSelector((state) => state.explore?.panes?.[exploreId]);
// Find the panel with this exploreId to get saved state
const panel = useSelector((state) =>
Object.values(state.exploreMap.panels).find((p) => p.exploreId === exploreId)
);
// Get panel from CRDT state (which has the latest exploreState)
const panel = useSelector((state) => {
const panels = selectPanels(state.exploreMapCRDT);
return panels[panelId];
});
// Initialize Explore pane on mount
useEffect(() => {
@@ -100,6 +117,8 @@ export function ExploreMapPanelContent({ exploreId, width, height }: ExploreMapP
// Use saved state if available, otherwise defaults
const savedState = panel?.exploreState;
console.log('[ExploreMapPanelContent] Initializing with saved state:', savedState);
await dispatch(
initializeExplore({
exploreId,
@@ -119,10 +138,11 @@ export function ExploreMapPanelContent({ exploreId, width, height }: ExploreMapP
return () => {
eventBus.removeAllListeners();
};
}, [dispatch, exploreId, eventBus, panel?.exploreState]);
// eslint-disable-next-line react-hooks/exhaustive-deps
}, [dispatch, exploreId, eventBus]);
// Wait for Redux state to be initialized
if (!isInitialized || !explorePane) {
if (!isInitialized) {
return (
<div className={styles.container}>
<div className={styles.loading}>
@@ -8,7 +8,8 @@ import { useDispatch, useSelector } from 'app/types/store';
import { useTransformContext } from '../context/TransformContext';
import { useCanvasPersistence } from '../hooks/useCanvasPersistence';
import { resetCanvas, updateMapTitle } from '../state/exploreMapSlice';
import { updateMapTitle } from '../state/crdtSlice';
import { selectPanelCount, selectViewport, selectMapTitle } from '../state/selectors';
interface ExploreMapToolbarProps {
uid?: string;
@@ -24,9 +25,9 @@ export function ExploreMapToolbar({ uid }: ExploreMapToolbarProps) {
const [titleValue, setTitleValue] = useState('');
const titleInputRef = useRef<HTMLInputElement>(null);
const panelCount = useSelector((state) => Object.keys(state.exploreMap.panels).length);
const viewport = useSelector((state) => state.exploreMap.viewport);
const mapTitle = useSelector((state) => state.exploreMap.title);
const panelCount = useSelector((state) => selectPanelCount(state.exploreMapCRDT));
const viewport = useSelector((state) => selectViewport(state.exploreMapCRDT));
const mapTitle = useSelector((state) => selectMapTitle(state.exploreMapCRDT));
useEffect(() => {
if (mapTitle) {
@@ -46,7 +47,9 @@ export function ExploreMapToolbar({ uid }: ExploreMapToolbarProps) {
}, []);
const confirmResetCanvas = useCallback(() => {
dispatch(resetCanvas());
// TODO: Implement resetCanvas for CRDT state
// dispatch(resetCanvas());
console.warn('Reset canvas not yet implemented for CRDT state');
setShowResetConfirm(false);
}, [dispatch]);
+158
View File
@@ -0,0 +1,158 @@
/**
* Hybrid Logical Clock (HLC) implementation
*
* Combines logical time (Lamport timestamp) with physical wall-clock time
* to provide timestamps that are:
* - Monotonically increasing
* - Causally consistent
* - Approximately synchronized with real time
*
* Used for conflict resolution in LWW-Register CRDTs.
*/
export interface HLCTimestamp {
logicalTime: number; // Lamport timestamp component
wallTime: number; // Physical clock component (milliseconds since epoch)
nodeId: string; // Unique node identifier for tie-breaking
}
export class HybridLogicalClock {
private logicalTime: number = 0;
private lastWallTime: number = 0;
private nodeId: string;
constructor(nodeId: string) {
this.nodeId = nodeId;
this.lastWallTime = Date.now();
}
/**
* Advance the clock for a local event
* Returns a new timestamp representing "now"
*/
tick(): HLCTimestamp {
const now = Date.now();
if (now > this.lastWallTime) {
// Physical clock advanced - use it
this.lastWallTime = now;
this.logicalTime = 0;
} else {
// Physical clock hasn't advanced - increment logical component
this.logicalTime++;
}
return this.clone();
}
/**
* Update clock based on received timestamp from another node
* Ensures causal consistency: if A → B, then timestamp(A) < timestamp(B)
*/
update(receivedTimestamp: HLCTimestamp): void {
const now = Date.now();
const maxWall = Math.max(this.lastWallTime, receivedTimestamp.wallTime, now);
if (maxWall === this.lastWallTime && maxWall === receivedTimestamp.wallTime) {
// Same wall time - take max logical time and increment
this.logicalTime = Math.max(this.logicalTime, receivedTimestamp.logicalTime) + 1;
} else if (maxWall === receivedTimestamp.wallTime) {
// Received timestamp has newer wall time
this.lastWallTime = maxWall;
this.logicalTime = receivedTimestamp.logicalTime + 1;
} else {
// Our wall time or physical clock is newer
this.lastWallTime = maxWall;
this.logicalTime = 0;
}
}
/**
* Get current timestamp without advancing the clock
*/
now(): HLCTimestamp {
return this.clone();
}
/**
* Create a copy of the current timestamp
*/
clone(): HLCTimestamp {
return {
logicalTime: this.logicalTime,
wallTime: this.lastWallTime,
nodeId: this.nodeId,
};
}
/**
* Get the node ID
*/
getNodeId(): string {
return this.nodeId;
}
}
/**
* Compare two HLC timestamps
* Returns:
* < 0 if a < b
* > 0 if a > b
* = 0 if a == b
*/
export function compareHLC(a: HLCTimestamp, b: HLCTimestamp): number {
// First compare wall time
if (a.wallTime !== b.wallTime) {
return a.wallTime - b.wallTime;
}
// Then compare logical time
if (a.logicalTime !== b.logicalTime) {
return a.logicalTime - b.logicalTime;
}
// Finally compare node IDs for deterministic tie-breaking
return a.nodeId.localeCompare(b.nodeId);
}
/**
* Check if timestamp a happened before timestamp b
*/
export function happensBefore(a: HLCTimestamp, b: HLCTimestamp): boolean {
return compareHLC(a, b) < 0;
}
/**
* Check if timestamp a happened after timestamp b
*/
export function happensAfter(a: HLCTimestamp, b: HLCTimestamp): boolean {
return compareHLC(a, b) > 0;
}
/**
* Check if two timestamps are equal
*/
export function timestampEquals(a: HLCTimestamp, b: HLCTimestamp): boolean {
return compareHLC(a, b) === 0;
}
/**
* Get the maximum of two timestamps
*/
export function maxTimestamp(a: HLCTimestamp, b: HLCTimestamp): HLCTimestamp {
return compareHLC(a, b) >= 0 ? a : b;
}
/**
* Serialize timestamp to JSON
*/
export function serializeHLC(timestamp: HLCTimestamp): string {
return JSON.stringify(timestamp);
}
/**
* Deserialize timestamp from JSON
*/
export function deserializeHLC(json: string): HLCTimestamp {
return JSON.parse(json);
}
@@ -0,0 +1,39 @@
/**
* CRDT module exports
*
* Conflict-Free Replicated Data Types for collaborative Explore Map editing
*/
// Core CRDT types
export { HybridLogicalClock, compareHLC, happensBefore, happensAfter, timestampEquals, maxTimestamp } from './hlc';
export type { HLCTimestamp } from './hlc';
export { ORSet } from './orset';
export type { ORSetJSON } from './orset';
export { LWWRegister, createLWWRegister } from './lwwregister';
export type { LWWRegisterJSON } from './lwwregister';
export { PNCounter } from './pncounter';
export type { PNCounterJSON } from './pncounter';
// CRDT state manager
export { CRDTStateManager } from './state';
// Types
export type {
CRDTExploreMapState,
CRDTPanelData,
CRDTExploreMapStateJSON,
CRDTOperation,
CRDTOperationType,
AddPanelOperation,
RemovePanelOperation,
UpdatePanelPositionOperation,
UpdatePanelSizeOperation,
UpdatePanelZIndexOperation,
UpdatePanelExploreStateOperation,
UpdateTitleOperation,
BatchOperation,
OperationResult,
} from './types';
@@ -0,0 +1,127 @@
/**
* Last-Write-Wins Register (LWW-Register) CRDT implementation
*
* A register that stores a single value and resolves conflicts using timestamps.
* When concurrent updates occur, the one with the highest timestamp wins.
*
* Properties:
* - Deterministic conflict resolution via timestamp ordering
* - Idempotent: setting same value with same timestamp is safe
* - Commutative: can apply updates in any order, final state is consistent
*
* Used for panel properties: position, dimensions, exploreState, etc.
*/
import { HLCTimestamp, compareHLC } from './hlc';
export interface LWWRegisterJSON<T> {
value: T;
timestamp: HLCTimestamp;
}
export class LWWRegister<T> {
private value: T;
private timestamp: HLCTimestamp;
/**
* Create a new LWW-Register with an initial value and timestamp
*
* @param initialValue - The initial value
* @param initialTimestamp - The initial timestamp
*/
constructor(initialValue: T, initialTimestamp: HLCTimestamp) {
this.value = initialValue;
this.timestamp = initialTimestamp;
}
/**
* Set the register value if the new timestamp is greater than current
*
* @param value - The new value
* @param timestamp - The timestamp of this update
* @returns true if the value was updated, false if update was ignored
*/
set(value: T, timestamp: HLCTimestamp): boolean {
// Only update if new timestamp is strictly greater
if (compareHLC(timestamp, this.timestamp) > 0) {
this.value = value;
this.timestamp = timestamp;
return true;
}
return false;
}
/**
* Get the current value
*/
get(): T {
return this.value;
}
/**
* Get the current timestamp
*/
getTimestamp(): HLCTimestamp {
return this.timestamp;
}
/**
* Merge another LWW-Register into this one
* Keeps the value with the highest timestamp
*
* @param other - The register to merge
* @returns true if this register's value was updated
*/
merge(other: LWWRegister<T>): boolean {
return this.set(other.value, other.timestamp);
}
/**
* Create a copy of this register
*/
clone(): LWWRegister<T> {
return new LWWRegister(this.value, { ...this.timestamp });
}
/**
* Serialize to JSON
*/
toJSON(): LWWRegisterJSON<T> {
return {
value: this.value,
timestamp: this.timestamp,
};
}
/**
* Deserialize from JSON
*/
static fromJSON<T>(json: LWWRegisterJSON<T>): LWWRegister<T> {
return new LWWRegister(json.value, json.timestamp);
}
/**
* Get debug information
*/
debug(): {
value: T;
timestamp: HLCTimestamp;
} {
return {
value: this.value,
timestamp: this.timestamp,
};
}
}
/**
* Helper function to create a register with a zero timestamp
* Useful for initialization
*/
export function createLWWRegister<T>(value: T, nodeId: string): LWWRegister<T> {
return new LWWRegister(value, {
logicalTime: 0,
wallTime: 0,
nodeId,
});
}
@@ -0,0 +1,257 @@
/**
* Observed-Remove Set (OR-Set) CRDT implementation
*
* A set that handles concurrent add/remove operations correctly.
* Each element is tagged with unique identifiers, and removes only
* affect the specific tags they observed.
*
* Properties:
* - Add-wins semantics: concurrent add/remove results in element being present
* - Idempotent operations: applying same operation multiple times is safe
* - Commutative: operations can be applied in any order
*
* Used for tracking which panels exist on the canvas.
*/
export interface ORSetJSON<T> {
adds: Record<string, string[]>; // element -> array of unique tags
removes: string[]; // array of removed tags
}
export class ORSet<T extends string = string> {
private adds: Map<T, Set<string>>; // element -> set of unique tags
private removes: Set<string>; // set of removed tags
constructor() {
this.adds = new Map();
this.removes = new Set();
}
/**
* Add an element to the set with a unique tag
* The tag should be globally unique (e.g., operation ID)
*
* @param element - The element to add
* @param tag - Unique identifier for this add operation
*/
add(element: T, tag: string): void {
if (!this.adds.has(element)) {
this.adds.set(element, new Set());
}
this.adds.get(element)!.add(tag);
}
/**
* Remove an element from the set
* Only removes the specific tags that were observed
*
* @param element - The element to remove
* @param observedTags - The tags that were observed when the remove was issued
*/
remove(element: T, observedTags: string[]): void {
for (const tag of observedTags) {
this.removes.add(tag);
}
// Clean up the element's tags
const elementTags = this.adds.get(element);
if (elementTags) {
for (const tag of observedTags) {
elementTags.delete(tag);
}
// If no tags remain, remove the element entry
if (elementTags.size === 0) {
this.adds.delete(element);
}
}
}
/**
* Check if an element is in the set
* Element is present if it has at least one non-removed tag
*
* @param element - The element to check
* @returns true if element is in the set
*/
contains(element: T): boolean {
const tags = this.adds.get(element);
if (!tags || tags.size === 0) {
return false;
}
// Element is present if it has at least one tag that hasn't been removed
for (const tag of tags) {
if (!this.removes.has(tag)) {
return true;
}
}
return false;
}
/**
* Get all tags for an element (including removed ones)
*
* @param element - The element to get tags for
* @returns Array of tags, or empty array if element not found
*/
getTags(element: T): string[] {
const tags = this.adds.get(element);
return tags ? Array.from(tags) : [];
}
/**
* Get all elements currently in the set
*
* @returns Array of elements
*/
values(): T[] {
const result: T[] = [];
for (const [element, tags] of this.adds.entries()) {
// Include element if it has at least one non-removed tag
for (const tag of tags) {
if (!this.removes.has(tag)) {
result.push(element);
break;
}
}
}
return result;
}
/**
* Get the number of elements in the set
*/
size(): number {
return this.values().length;
}
/**
* Check if the set is empty
*/
isEmpty(): boolean {
return this.size() === 0;
}
/**
* Merge another OR-Set into this one
* Takes the union of all adds and removes
*
* @param other - The OR-Set to merge
* @returns This OR-Set (for chaining)
*/
merge(other: ORSet<T>): this {
// Merge adds (union of all tags)
for (const [element, otherTags] of other.adds.entries()) {
if (!this.adds.has(element)) {
this.adds.set(element, new Set());
}
const myTags = this.adds.get(element)!;
for (const tag of otherTags) {
myTags.add(tag);
}
}
// Merge removes (union of all removed tags)
for (const tag of other.removes) {
this.removes.add(tag);
}
// Clean up elements with all tags removed
for (const [element, tags] of this.adds.entries()) {
let hasLiveTag = false;
for (const tag of tags) {
if (!this.removes.has(tag)) {
hasLiveTag = true;
break;
}
}
if (!hasLiveTag) {
this.adds.delete(element);
}
}
return this;
}
/**
* Create a copy of this OR-Set
*/
clone(): ORSet<T> {
const copy = new ORSet<T>();
// Deep copy adds
for (const [element, tags] of this.adds.entries()) {
copy.adds.set(element, new Set(tags));
}
// Deep copy removes
copy.removes = new Set(this.removes);
return copy;
}
/**
* Clear all elements from the set (for testing/reset)
*/
clear(): void {
this.adds.clear();
this.removes.clear();
}
/**
* Serialize to JSON for network transmission or storage
*/
toJSON(): ORSetJSON<T> {
const adds: Record<string, string[]> = {};
for (const [element, tags] of this.adds.entries()) {
adds[element] = Array.from(tags);
}
return {
adds,
removes: Array.from(this.removes),
};
}
/**
* Deserialize from JSON
*/
static fromJSON<T extends string = string>(json: ORSetJSON<T>): ORSet<T> {
const set = new ORSet<T>();
// Restore adds
for (const [element, tags] of Object.entries(json.adds)) {
set.adds.set(element as T, new Set(tags));
}
// Restore removes
set.removes = new Set(json.removes);
return set;
}
/**
* Get debug information about the set
*/
debug(): {
elements: T[];
totalTags: number;
removedTags: number;
rawAdds: Map<T, Set<string>>;
rawRemoves: Set<string>;
} {
let totalTags = 0;
for (const tags of this.adds.values()) {
totalTags += tags.size;
}
return {
elements: this.values(),
totalTags,
removedTags: this.removes.size,
rawAdds: this.adds,
rawRemoves: this.removes,
};
}
}
@@ -0,0 +1,167 @@
/**
* Positive-Negative Counter (PN-Counter) CRDT implementation
*
* A counter that supports both increment and decrement operations.
* Each node maintains separate positive and negative counters.
*
* Properties:
* - Commutative: operations can be applied in any order
* - Idempotent: applying same operation multiple times (with dedup) is safe
* - Eventually consistent: all replicas converge to same value
*
* Used for allocating monotonically increasing z-indices.
* For our use case, we only need increments (positive counter).
*/
export interface PNCounterJSON {
increments: Record<string, number>; // nodeId -> count
decrements: Record<string, number>; // nodeId -> count
}
export class PNCounter {
private increments: Map<string, number>; // nodeId -> count
private decrements: Map<string, number>; // nodeId -> count
constructor() {
this.increments = new Map();
this.decrements = new Map();
}
/**
* Increment the counter for a specific node
*
* @param nodeId - The node performing the increment
* @param delta - The amount to increment (default: 1)
*/
increment(nodeId: string, delta: number = 1): void {
if (delta < 0) {
throw new Error('Delta must be non-negative for increment');
}
const current = this.increments.get(nodeId) || 0;
this.increments.set(nodeId, current + delta);
}
/**
* Decrement the counter for a specific node
*
* @param nodeId - The node performing the decrement
* @param delta - The amount to decrement (default: 1)
*/
decrement(nodeId: string, delta: number = 1): void {
if (delta < 0) {
throw new Error('Delta must be non-negative for decrement');
}
const current = this.decrements.get(nodeId) || 0;
this.decrements.set(nodeId, current + delta);
}
/**
* Get the current value of the counter
* Value = sum of all increments - sum of all decrements
*/
value(): number {
let sum = 0;
// Add all increments
for (const count of this.increments.values()) {
sum += count;
}
// Subtract all decrements
for (const count of this.decrements.values()) {
sum -= count;
}
return sum;
}
/**
* Get the next value and increment the counter for a node
* This is useful for allocating sequential IDs (like z-indices)
*
* @param nodeId - The node allocating the next value
* @returns The next available value
*/
next(nodeId: string): number {
const nextValue = this.value() + 1;
this.increment(nodeId, 1);
return nextValue;
}
/**
* Merge another PN-Counter into this one
* Takes the maximum value for each node's counters
*
* @param other - The counter to merge
*/
merge(other: PNCounter): this {
// Merge increments (take max for each node)
for (const [nodeId, count] of other.increments.entries()) {
const current = this.increments.get(nodeId) || 0;
this.increments.set(nodeId, Math.max(current, count));
}
// Merge decrements (take max for each node)
for (const [nodeId, count] of other.decrements.entries()) {
const current = this.decrements.get(nodeId) || 0;
this.decrements.set(nodeId, Math.max(current, count));
}
return this;
}
/**
* Create a copy of this counter
*/
clone(): PNCounter {
const copy = new PNCounter();
copy.increments = new Map(this.increments);
copy.decrements = new Map(this.decrements);
return copy;
}
/**
* Reset the counter to zero (for testing)
*/
reset(): void {
this.increments.clear();
this.decrements.clear();
}
/**
* Serialize to JSON
*/
toJSON(): PNCounterJSON {
return {
increments: Object.fromEntries(this.increments),
decrements: Object.fromEntries(this.decrements),
};
}
/**
* Deserialize from JSON
*/
static fromJSON(json: PNCounterJSON): PNCounter {
const counter = new PNCounter();
counter.increments = new Map(Object.entries(json.increments));
counter.decrements = new Map(Object.entries(json.decrements));
return counter;
}
/**
* Get debug information
*/
debug(): {
value: number;
increments: Record<string, number>;
decrements: Record<string, number>;
nodeCount: number;
} {
return {
value: this.value(),
increments: Object.fromEntries(this.increments),
decrements: Object.fromEntries(this.decrements),
nodeCount: new Set([...this.increments.keys(), ...this.decrements.keys()]).size,
};
}
}
@@ -0,0 +1,569 @@
/**
* CRDT State Manager for Explore Map
*
* This class manages the CRDT-based state and provides high-level
* operations for adding/removing/updating panels.
*/
import { v4 as uuidv4 } from 'uuid';
import { HybridLogicalClock } from './hlc';
import { LWWRegister, createLWWRegister } from './lwwregister';
import { ORSet } from './orset';
import { PNCounter } from './pncounter';
import {
CRDTExploreMapState,
CRDTPanelData,
CRDTOperation,
AddPanelOperation,
RemovePanelOperation,
UpdatePanelPositionOperation,
UpdatePanelSizeOperation,
UpdatePanelZIndexOperation,
UpdatePanelExploreStateOperation,
UpdateTitleOperation,
OperationResult,
CRDTExploreMapStateJSON,
} from './types';
import { SerializedExploreState } from '../state/types';
export class CRDTStateManager {
private state: CRDTExploreMapState;
private clock: HybridLogicalClock;
private nodeId: string;
private mapUid: string;
constructor(mapUid: string, nodeId?: string) {
this.mapUid = mapUid;
this.nodeId = nodeId || uuidv4();
this.clock = new HybridLogicalClock(this.nodeId);
// Initialize empty state
this.state = this.createInitialState();
}
private createInitialState(): CRDTExploreMapState {
return {
uid: this.mapUid,
title: createLWWRegister('Untitled Map', this.nodeId),
panels: new ORSet<string>(),
panelData: new Map(),
zIndexCounter: new PNCounter(),
local: {
viewport: {
zoom: 1,
panX: -4040,
panY: -4460,
},
selectedPanelIds: [],
cursors: {},
},
};
}
/**
* Get the current node ID
*/
getNodeId(): string {
return this.nodeId;
}
/**
* Get the current state
*/
getState(): CRDTExploreMapState {
return this.state;
}
/**
* Get all panel IDs currently in the set
*/
getPanelIds(): string[] {
return this.state.panels.values();
}
/**
* Get panel data by ID
*/
getPanelData(panelId: string): CRDTPanelData | undefined {
if (!this.state.panels.contains(panelId)) {
return undefined;
}
return this.state.panelData.get(panelId);
}
/**
* Get a plain object representation of a panel for UI rendering
*/
getPanelForUI(panelId: string) {
const data = this.getPanelData(panelId);
if (!data) {
return undefined;
}
return {
id: data.id,
exploreId: data.exploreId,
position: {
x: data.positionX.get(),
y: data.positionY.get(),
width: data.width.get(),
height: data.height.get(),
zIndex: data.zIndex.get(),
},
exploreState: data.exploreState.get(),
};
}
/**
* Get all panels for UI rendering
*/
getAllPanelsForUI() {
const panels: Record<string, any> = {};
for (const panelId of this.getPanelIds()) {
const panel = this.getPanelForUI(panelId);
if (panel) {
panels[panelId] = panel;
}
}
return panels;
}
/**
* Create an add panel operation
*/
createAddPanelOperation(
panelId: string,
exploreId: string,
position: { x: number; y: number; width: number; height: number }
): AddPanelOperation {
const timestamp = this.clock.tick();
return {
type: 'add-panel',
mapUid: this.mapUid,
operationId: uuidv4(),
timestamp,
nodeId: this.nodeId,
payload: {
panelId,
exploreId,
position,
},
};
}
/**
* Create a remove panel operation
*/
createRemovePanelOperation(panelId: string): RemovePanelOperation | null {
if (!this.state.panels.contains(panelId)) {
return null;
}
const timestamp = this.clock.tick();
const observedTags = this.state.panels.getTags(panelId);
return {
type: 'remove-panel',
mapUid: this.mapUid,
operationId: uuidv4(),
timestamp,
nodeId: this.nodeId,
payload: {
panelId,
observedTags,
},
};
}
/**
* Create an update panel position operation
*/
createUpdatePanelPositionOperation(
panelId: string,
x: number,
y: number
): UpdatePanelPositionOperation | null {
if (!this.state.panels.contains(panelId)) {
return null;
}
const timestamp = this.clock.tick();
return {
type: 'update-panel-position',
mapUid: this.mapUid,
operationId: uuidv4(),
timestamp,
nodeId: this.nodeId,
payload: {
panelId,
x,
y,
},
};
}
/**
* Create an update panel size operation
*/
createUpdatePanelSizeOperation(
panelId: string,
width: number,
height: number
): UpdatePanelSizeOperation | null {
if (!this.state.panels.contains(panelId)) {
return null;
}
const timestamp = this.clock.tick();
return {
type: 'update-panel-size',
mapUid: this.mapUid,
operationId: uuidv4(),
timestamp,
nodeId: this.nodeId,
payload: {
panelId,
width,
height,
},
};
}
/**
* Create an update panel z-index operation
*/
createUpdatePanelZIndexOperation(panelId: string): UpdatePanelZIndexOperation | null {
if (!this.state.panels.contains(panelId)) {
return null;
}
const timestamp = this.clock.tick();
const zIndex = this.state.zIndexCounter.next(this.nodeId);
return {
type: 'update-panel-zindex',
mapUid: this.mapUid,
operationId: uuidv4(),
timestamp,
nodeId: this.nodeId,
payload: {
panelId,
zIndex,
},
};
}
/**
* Create an update panel explore state operation
*/
createUpdatePanelExploreStateOperation(
panelId: string,
exploreState: SerializedExploreState | undefined
): UpdatePanelExploreStateOperation | null {
if (!this.state.panels.contains(panelId)) {
return null;
}
const timestamp = this.clock.tick();
return {
type: 'update-panel-explore-state',
mapUid: this.mapUid,
operationId: uuidv4(),
timestamp,
nodeId: this.nodeId,
payload: {
panelId,
exploreState,
},
};
}
/**
* Create an update title operation
*/
createUpdateTitleOperation(title: string): UpdateTitleOperation {
const timestamp = this.clock.tick();
return {
type: 'update-title',
mapUid: this.mapUid,
operationId: uuidv4(),
timestamp,
nodeId: this.nodeId,
payload: {
title,
},
};
}
/**
* Apply a CRDT operation to the state
*/
applyOperation(operation: CRDTOperation): OperationResult {
// Update clock with received timestamp
this.clock.update(operation.timestamp);
try {
switch (operation.type) {
case 'add-panel':
return this.applyAddPanel(operation);
case 'remove-panel':
return this.applyRemovePanel(operation);
case 'update-panel-position':
return this.applyUpdatePanelPosition(operation);
case 'update-panel-size':
return this.applyUpdatePanelSize(operation);
case 'update-panel-zindex':
return this.applyUpdatePanelZIndex(operation);
case 'update-panel-explore-state':
return this.applyUpdatePanelExploreState(operation);
case 'update-title':
return this.applyUpdateTitle(operation);
case 'batch':
return this.applyBatchOperation(operation);
default:
return {
success: false,
applied: false,
error: `Unknown operation type: ${(operation as any).type}`,
};
}
} catch (error) {
return {
success: false,
applied: false,
error: error instanceof Error ? error.message : String(error),
};
}
}
private applyAddPanel(operation: AddPanelOperation): OperationResult {
const { panelId, exploreId, position } = operation.payload;
// Add to OR-Set with operation ID as tag
this.state.panels.add(panelId, operation.operationId);
// Initialize panel data if it doesn't exist
if (!this.state.panelData.has(panelId)) {
const zIndex = this.state.zIndexCounter.next(operation.nodeId);
this.state.panelData.set(panelId, {
id: panelId,
exploreId,
positionX: new LWWRegister(position.x, operation.timestamp),
positionY: new LWWRegister(position.y, operation.timestamp),
width: new LWWRegister(position.width, operation.timestamp),
height: new LWWRegister(position.height, operation.timestamp),
zIndex: new LWWRegister(zIndex, operation.timestamp),
exploreState: new LWWRegister(undefined, operation.timestamp),
});
}
return { success: true, applied: true };
}
private applyRemovePanel(operation: RemovePanelOperation): OperationResult {
const { panelId, observedTags } = operation.payload;
// Remove from OR-Set
this.state.panels.remove(panelId, observedTags);
// Keep panel data as tombstone for CRDT correctness
// (Don't delete from panelData map - needed for merging)
return { success: true, applied: true };
}
private applyUpdatePanelPosition(operation: UpdatePanelPositionOperation): OperationResult {
const { panelId, x, y } = operation.payload;
const panelData = this.state.panelData.get(panelId);
if (!panelData) {
return { success: true, applied: false, error: 'Panel not found' };
}
const xUpdated = panelData.positionX.set(x, operation.timestamp);
const yUpdated = panelData.positionY.set(y, operation.timestamp);
return {
success: true,
applied: xUpdated || yUpdated,
};
}
private applyUpdatePanelSize(operation: UpdatePanelSizeOperation): OperationResult {
const { panelId, width, height } = operation.payload;
const panelData = this.state.panelData.get(panelId);
if (!panelData) {
return { success: true, applied: false, error: 'Panel not found' };
}
const widthUpdated = panelData.width.set(width, operation.timestamp);
const heightUpdated = panelData.height.set(height, operation.timestamp);
return {
success: true,
applied: widthUpdated || heightUpdated,
};
}
private applyUpdatePanelZIndex(operation: UpdatePanelZIndexOperation): OperationResult {
const { panelId, zIndex } = operation.payload;
const panelData = this.state.panelData.get(panelId);
if (!panelData) {
return { success: true, applied: false, error: 'Panel not found' };
}
const updated = panelData.zIndex.set(zIndex, operation.timestamp);
return {
success: true,
applied: updated,
};
}
private applyUpdatePanelExploreState(operation: UpdatePanelExploreStateOperation): OperationResult {
const { panelId, exploreState } = operation.payload;
const panelData = this.state.panelData.get(panelId);
if (!panelData) {
return { success: true, applied: false, error: 'Panel not found' };
}
const updated = panelData.exploreState.set(exploreState, operation.timestamp);
return {
success: true,
applied: updated,
};
}
private applyUpdateTitle(operation: UpdateTitleOperation): OperationResult {
const { title } = operation.payload;
const updated = this.state.title.set(title, operation.timestamp);
return {
success: true,
applied: updated,
};
}
private applyBatchOperation(operation: any): OperationResult {
let anyApplied = false;
const errors: string[] = [];
for (const subOp of operation.payload.operations) {
const result = this.applyOperation(subOp);
if (!result.success) {
errors.push(result.error || 'Unknown error');
}
if (result.applied) {
anyApplied = true;
}
}
return {
success: errors.length === 0,
applied: anyApplied,
error: errors.length > 0 ? errors.join('; ') : undefined,
};
}
/**
* Merge another CRDT state into this one
*/
mergeState(other: CRDTExploreMapState): void {
// Merge title
this.state.title.merge(other.title);
// Merge panel OR-Set
this.state.panels.merge(other.panels);
// Merge panel data
for (const [panelId, otherPanelData] of other.panelData.entries()) {
const myPanelData = this.state.panelData.get(panelId);
if (!myPanelData) {
// Panel doesn't exist locally - copy it
this.state.panelData.set(panelId, {
id: otherPanelData.id,
exploreId: otherPanelData.exploreId,
positionX: otherPanelData.positionX.clone(),
positionY: otherPanelData.positionY.clone(),
width: otherPanelData.width.clone(),
height: otherPanelData.height.clone(),
zIndex: otherPanelData.zIndex.clone(),
exploreState: otherPanelData.exploreState.clone(),
});
} else {
// Merge each LWW register
myPanelData.positionX.merge(otherPanelData.positionX);
myPanelData.positionY.merge(otherPanelData.positionY);
myPanelData.width.merge(otherPanelData.width);
myPanelData.height.merge(otherPanelData.height);
myPanelData.zIndex.merge(otherPanelData.zIndex);
myPanelData.exploreState.merge(otherPanelData.exploreState);
}
}
// Merge z-index counter
this.state.zIndexCounter.merge(other.zIndexCounter);
}
/**
* Serialize state to JSON
*/
toJSON(): CRDTExploreMapStateJSON {
const panelData: Record<string, any> = {};
for (const [panelId, data] of this.state.panelData.entries()) {
panelData[panelId] = {
id: data.id,
exploreId: data.exploreId,
positionX: data.positionX.toJSON(),
positionY: data.positionY.toJSON(),
width: data.width.toJSON(),
height: data.height.toJSON(),
zIndex: data.zIndex.toJSON(),
exploreState: data.exploreState.toJSON(),
};
}
return {
uid: this.state.uid,
title: this.state.title.toJSON(),
panels: this.state.panels.toJSON(),
panelData,
zIndexCounter: this.state.zIndexCounter.toJSON(),
};
}
/**
* Load state from JSON
*/
static fromJSON(json: CRDTExploreMapStateJSON, nodeId?: string): CRDTStateManager {
const manager = new CRDTStateManager(json.uid || '', nodeId);
manager.state.uid = json.uid;
manager.state.title = LWWRegister.fromJSON(json.title);
manager.state.panels = ORSet.fromJSON(json.panels);
manager.state.zIndexCounter = PNCounter.fromJSON(json.zIndexCounter);
// Load panel data
for (const [panelId, data] of Object.entries(json.panelData)) {
manager.state.panelData.set(panelId, {
id: data.id,
exploreId: data.exploreId,
positionX: LWWRegister.fromJSON(data.positionX),
positionY: LWWRegister.fromJSON(data.positionY),
width: LWWRegister.fromJSON(data.width),
height: LWWRegister.fromJSON(data.height),
zIndex: LWWRegister.fromJSON(data.zIndex),
exploreState: LWWRegister.fromJSON(data.exploreState),
});
}
return manager;
}
}
@@ -0,0 +1,236 @@
/**
* CRDT-based Explore Map state types
*
* This file defines the CRDT-enhanced data structures for the Explore Map
* feature, enabling conflict-free collaborative editing.
*/
import { LWWRegister } from './lwwregister';
import { ORSet } from './orset';
import { PNCounter } from './pncounter';
import { HLCTimestamp } from './hlc';
import { SerializedExploreState } from '../state/types';
/**
* CRDT state for a single panel
*/
export interface CRDTPanelData {
// Stable identifiers
id: string;
exploreId: string;
// CRDT-replicated position properties
positionX: LWWRegister<number>;
positionY: LWWRegister<number>;
width: LWWRegister<number>;
height: LWWRegister<number>;
zIndex: LWWRegister<number>;
// CRDT-replicated explore state
exploreState: LWWRegister<SerializedExploreState | undefined>;
}
/**
* Complete CRDT-based Explore Map state
*/
export interface CRDTExploreMapState {
// Map metadata
uid?: string;
title: LWWRegister<string>;
// Panel collection (OR-Set for add/remove operations)
panels: ORSet<string>; // Set of panel IDs
// Panel data (position, size, content)
panelData: Map<string, CRDTPanelData>;
// Counter for allocating z-indices
zIndexCounter: PNCounter;
// Local-only state (not replicated via CRDT)
local: {
viewport: {
zoom: number;
panX: number;
panY: number;
};
selectedPanelIds: string[];
cursors: Record<string, {
userId: string;
userName: string;
color: string;
x: number;
y: number;
lastUpdated: number;
}>;
};
}
/**
* JSON-serializable version of CRDT state
*/
export interface CRDTExploreMapStateJSON {
uid?: string;
title: {
value: string;
timestamp: HLCTimestamp;
};
panels: {
adds: Record<string, string[]>;
removes: string[];
};
panelData: Record<string, {
id: string;
exploreId: string;
positionX: { value: number; timestamp: HLCTimestamp };
positionY: { value: number; timestamp: HLCTimestamp };
width: { value: number; timestamp: HLCTimestamp };
height: { value: number; timestamp: HLCTimestamp };
zIndex: { value: number; timestamp: HLCTimestamp };
exploreState: { value: SerializedExploreState | undefined; timestamp: HLCTimestamp };
}>;
zIndexCounter: {
increments: Record<string, number>;
decrements: Record<string, number>;
};
}
/**
* Operation types for CRDT updates
*/
export type CRDTOperationType =
| 'add-panel'
| 'remove-panel'
| 'update-panel-position'
| 'update-panel-size'
| 'update-panel-zindex'
| 'update-panel-explore-state'
| 'update-title'
| 'batch'; // For batching multiple operations
/**
* Base operation interface
*/
export interface CRDTOperationBase {
type: CRDTOperationType;
mapUid: string;
operationId: string; // Unique operation ID (UUID)
timestamp: HLCTimestamp;
nodeId: string; // Client/user ID
}
/**
* Add panel operation
*/
export interface AddPanelOperation extends CRDTOperationBase {
type: 'add-panel';
payload: {
panelId: string;
exploreId: string;
position: {
x: number;
y: number;
width: number;
height: number;
};
};
}
/**
* Remove panel operation
*/
export interface RemovePanelOperation extends CRDTOperationBase {
type: 'remove-panel';
payload: {
panelId: string;
observedTags: string[]; // Tags from OR-Set
};
}
/**
* Update panel position operation
*/
export interface UpdatePanelPositionOperation extends CRDTOperationBase {
type: 'update-panel-position';
payload: {
panelId: string;
x: number;
y: number;
};
}
/**
* Update panel size operation
*/
export interface UpdatePanelSizeOperation extends CRDTOperationBase {
type: 'update-panel-size';
payload: {
panelId: string;
width: number;
height: number;
};
}
/**
* Update panel z-index operation
*/
export interface UpdatePanelZIndexOperation extends CRDTOperationBase {
type: 'update-panel-zindex';
payload: {
panelId: string;
zIndex: number;
};
}
/**
* Update panel explore state operation
*/
export interface UpdatePanelExploreStateOperation extends CRDTOperationBase {
type: 'update-panel-explore-state';
payload: {
panelId: string;
exploreState: SerializedExploreState | undefined;
};
}
/**
* Update map title operation
*/
export interface UpdateTitleOperation extends CRDTOperationBase {
type: 'update-title';
payload: {
title: string;
};
}
/**
* Batch operation (multiple operations in one)
*/
export interface BatchOperation extends CRDTOperationBase {
type: 'batch';
payload: {
operations: CRDTOperation[];
};
}
/**
* Union type of all operations
*/
export type CRDTOperation =
| AddPanelOperation
| RemovePanelOperation
| UpdatePanelPositionOperation
| UpdatePanelSizeOperation
| UpdatePanelZIndexOperation
| UpdatePanelExploreStateOperation
| UpdateTitleOperation
| BatchOperation;
/**
* Result of applying an operation
*/
export interface OperationResult {
success: boolean;
applied: boolean; // Whether the operation made changes
error?: string;
}
@@ -6,7 +6,9 @@ import { createErrorNotification, createSuccessNotification } from 'app/core/cop
import { useDispatch, useSelector } from 'app/types/store';
import { exploreMapApi } from '../api/exploreMapApi';
import { initializeFromLegacyState } from '../state/crdtSlice';
import { loadCanvas } from '../state/exploreMapSlice';
import { selectPanels, selectMapTitle, selectViewport } from '../state/selectors';
import { ExploreMapState, initialExploreMapState, SerializedExploreState } from '../state/types';
const STORAGE_KEY = 'grafana.exploreMap.state';
@@ -20,6 +22,7 @@ export function useCanvasPersistence(options: UseMapPersistenceOptions = {}) {
const { uid } = options;
const dispatch = useDispatch();
const exploreMapState = useSelector((state) => state.exploreMap);
const crdtState = useSelector((state) => state.exploreMapCRDT);
const exploreState = useSelector((state) => state.explore);
const [loading, setLoading] = useState(!!uid);
const [saving, setSaving] = useState(false);
@@ -77,11 +80,33 @@ export function useCanvasPersistence(options: UseMapPersistenceOptions = {}) {
try {
setLoading(true);
const mapData = await exploreMapApi.getExploreMap(uid);
const parsed: ExploreMapState = JSON.parse(mapData.data);
// Use title from DB column, not from JSON data
parsed.uid = mapData.uid;
parsed.title = mapData.title;
// Handle empty or missing data (new maps)
let parsed: ExploreMapState;
if (!mapData.data || mapData.data.trim() === '') {
// Initialize with default empty state for new maps
parsed = {
...initialExploreMapState,
uid: mapData.uid,
title: mapData.title,
};
} else {
parsed = JSON.parse(mapData.data);
// Use title from DB column, not from JSON data
parsed.uid = mapData.uid;
parsed.title = mapData.title;
}
// Load into legacy state (for backward compatibility)
dispatch(loadCanvas(parsed));
// Initialize CRDT state from loaded data
dispatch(initializeFromLegacyState({
uid: parsed.uid,
title: parsed.title,
panels: parsed.panels || {},
viewport: parsed.viewport || initialExploreMapState.viewport,
}));
} catch (error) {
console.error('Failed to load map from API:', error);
dispatch(
@@ -102,7 +127,17 @@ export function useCanvasPersistence(options: UseMapPersistenceOptions = {}) {
const savedState = store.get(STORAGE_KEY);
if (savedState) {
const parsed: ExploreMapState = JSON.parse(savedState);
// Load into legacy state
dispatch(loadCanvas(parsed));
// Initialize CRDT state from loaded data
dispatch(initializeFromLegacyState({
uid: parsed.uid,
title: parsed.title,
panels: parsed.panels,
viewport: parsed.viewport,
}));
}
} catch (error) {
console.error('Failed to load canvas state from storage:', error);
@@ -115,14 +150,19 @@ export function useCanvasPersistence(options: UseMapPersistenceOptions = {}) {
// Auto-save to API or localStorage
useEffect(() => {
// Don't persist an empty canvas; this avoids removing a previously saved
// non-empty canvas when the in-memory state is still at its initial value.
if (!exploreMapState || Object.keys(exploreMapState.panels || {}).length === 0) {
// Skip auto-save during initial load
if (!initialLoadDone.current || loading) {
return;
}
// Skip auto-save during initial load
if (!initialLoadDone.current || loading) {
// Get current CRDT state as panels
const panels = selectPanels(crdtState);
const mapTitle = selectMapTitle(crdtState);
const viewport = selectViewport(crdtState);
// Don't persist an empty canvas; this avoids removing a previously saved
// non-empty canvas when the in-memory state is still at its initial value.
if (Object.keys(panels || {}).length === 0) {
return;
}
@@ -132,23 +172,29 @@ export function useCanvasPersistence(options: UseMapPersistenceOptions = {}) {
}
const saveState = async () => {
const enrichedState = enrichStateWithExploreData(exploreMapState);
// CRDT panels already contain exploreState from savePanelExploreState actions
// We don't need to enrich them with live Explore pane data
console.log('[Persistence] Saving panels:', panels);
const enrichedState: ExploreMapState = {
uid,
title: mapTitle,
viewport,
panels: panels, // Already contains exploreState from CRDT
selectedPanelIds: [],
nextZIndex: 1,
cursors: {},
};
if (uid) {
// Save to API with debounce
saveTimeoutRef.current = setTimeout(async () => {
try {
setSaving(true);
const titleToSave = exploreMapState.title || 'Untitled Map';
// Ensure data also has the correct title
const dataToSave = {
...enrichedState,
uid: exploreMapState.uid,
title: titleToSave,
};
const titleToSave = mapTitle || 'Untitled Map';
await exploreMapApi.updateExploreMap(uid, {
title: titleToSave,
data: dataToSave,
data: enrichedState,
});
setLastSaved(new Date());
} catch (error) {
@@ -176,7 +222,7 @@ export function useCanvasPersistence(options: UseMapPersistenceOptions = {}) {
clearTimeout(saveTimeoutRef.current);
}
};
}, [exploreMapState, exploreState, enrichStateWithExploreData, dispatch, loading, uid]);
}, [crdtState, dispatch, loading, uid]);
const exportCanvas = useCallback(() => {
try {
@@ -225,7 +271,18 @@ export function useCanvasPersistence(options: UseMapPersistenceOptions = {}) {
throw new Error('Invalid file content');
}
const parsed: ExploreMapState = JSON.parse(result);
// Load into legacy state
dispatch(loadCanvas(parsed));
// Initialize CRDT state from imported data
dispatch(initializeFromLegacyState({
uid: parsed.uid,
title: parsed.title,
panels: parsed.panels,
viewport: parsed.viewport,
}));
dispatch(notifyApp(createSuccessNotification('Canvas imported successfully')));
} catch (error) {
console.error('Failed to parse imported canvas:', error);
@@ -0,0 +1,106 @@
/**
* Hook to receive and apply Explore state changes from CRDT
*
* This hook watches for CRDT operations that update panel explore state
* and applies them to the local Explore pane so the user sees changes
* made by other collaborators in real-time.
*/
import { useEffect, useRef } from 'react';
import { useDispatch, useSelector } from 'app/types/store';
import { changeDatasource } from '../../explore/state/datasource';
import { setQueriesAction } from '../../explore/state/query';
import { updateTime } from '../../explore/state/time';
interface UseExploreStateReceiverOptions {
panelId: string;
exploreId: string;
enabled?: boolean;
}
/**
* Hook to receive and apply explore state changes from CRDT
*/
export function useExploreStateReceiver(options: UseExploreStateReceiverOptions) {
const { panelId, exploreId, enabled = true } = options;
const dispatch = useDispatch();
// Get the panel's explore state from CRDT
const panels = useSelector((state) => {
const crdtStateJSON = state.exploreMapCRDT.crdtStateJSON;
if (!crdtStateJSON) {
return {};
}
const parsed = JSON.parse(crdtStateJSON);
const panelData: Record<string, { exploreState?: { value: any } }> = {};
for (const [id, data] of Object.entries(parsed.panelData || {})) {
panelData[id] = data as { exploreState?: { value: any } };
}
return panelData;
});
const panel = panels[panelId];
const exploreState = panel?.exploreState?.value;
// Get current Explore pane state for comparison
const explorePane = useSelector((state) => state.explore?.panes?.[exploreId]);
// Track what we've already applied to avoid loops
const lastAppliedStateRef = useRef<string | null>(null);
useEffect(() => {
if (!enabled || !exploreState || !explorePane) {
return;
}
const exploreStateStr = JSON.stringify(exploreState);
// Skip if we've already applied this exact state
if (lastAppliedStateRef.current === exploreStateStr) {
return;
}
console.log('[ExploreReceiver] Received new explore state for panel', panelId, exploreState);
lastAppliedStateRef.current = exploreStateStr;
// Apply queries if they've changed
if (exploreState.queries && JSON.stringify(exploreState.queries) !== JSON.stringify(explorePane.queries)) {
console.log('[ExploreReceiver] Applying queries', exploreState.queries);
dispatch(setQueriesAction({
exploreId,
queries: exploreState.queries,
}));
}
// Apply datasource if it's changed
if (exploreState.datasourceUid && exploreState.datasourceUid !== explorePane.datasourceInstance?.uid) {
console.log('[ExploreReceiver] Changing datasource to', exploreState.datasourceUid);
dispatch(changeDatasource({
exploreId,
datasource: exploreState.datasourceUid,
}));
}
// Apply time range if it's changed
if (exploreState.range && JSON.stringify(exploreState.range) !== JSON.stringify(explorePane.range)) {
console.log('[ExploreReceiver] Updating time range', exploreState.range);
dispatch(updateTime({
exploreId,
rawRange: exploreState.range,
}));
}
// Note: We don't sync refreshInterval, panelsState, or compact mode automatically
// as those are more UI preference than data state
// eslint-disable-next-line react-hooks/exhaustive-deps
}, [
enabled,
panelId,
exploreId,
// Intentionally not including exploreState or explorePane to avoid excessive re-renders
// We use refs inside the effect to detect actual changes
dispatch,
]);
}
@@ -0,0 +1,91 @@
/**
* Hook to synchronize Explore pane state changes to CRDT
*
* This hook watches for changes to the Explore pane (queries, datasource, range, etc.)
* and broadcasts them via CRDT operations so other users see the changes in real-time.
*/
import { useEffect, useRef } from 'react';
import { useDispatch, useSelector } from 'app/types/store';
import { savePanelExploreState } from '../state/crdtSlice';
import { SerializedExploreState } from '../state/types';
interface UseExploreStateSyncOptions {
panelId: string;
exploreId: string;
enabled?: boolean;
}
/**
* Debounce delay for syncing explore state changes
* We don't want to broadcast every keystroke, so we wait a bit
*/
const SYNC_DELAY_MS = 1000;
export function useExploreStateSync(options: UseExploreStateSyncOptions) {
const { panelId, exploreId, enabled = true } = options;
const dispatch = useDispatch();
// Get the explore pane state from Redux
const explorePane = useSelector((state) => state.explore?.panes?.[exploreId]);
// Track previous state to detect changes
const previousStateRef = useRef<string | null>(null);
const syncTimeoutRef = useRef<NodeJS.Timeout | null>(null);
useEffect(() => {
if (!enabled || !explorePane) {
return;
}
// Serialize the current explore state
const currentState: SerializedExploreState = {
queries: explorePane.queries,
datasourceUid: explorePane.datasourceInstance?.uid,
range: explorePane.range,
refreshInterval: explorePane.refreshInterval,
panelsState: explorePane.panelsState,
compact: explorePane.compact,
};
const currentStateStr = JSON.stringify(currentState);
// Check if state has actually changed
if (previousStateRef.current === currentStateStr) {
return;
}
console.log('[ExploreSync] Explore state changed for panel', panelId, currentState);
previousStateRef.current = currentStateStr;
// Clear any pending sync
if (syncTimeoutRef.current) {
clearTimeout(syncTimeoutRef.current);
}
// Debounce the sync operation
syncTimeoutRef.current = setTimeout(() => {
console.log('[ExploreSync] Dispatching savePanelExploreState for panel', panelId);
dispatch(savePanelExploreState({
panelId,
exploreState: currentState,
}));
}, SYNC_DELAY_MS);
// Cleanup timeout on unmount
return () => {
if (syncTimeoutRef.current) {
clearTimeout(syncTimeoutRef.current);
}
};
// eslint-disable-next-line react-hooks/exhaustive-deps
}, [
enabled,
panelId,
// Intentionally not including explorePane to avoid excessive re-renders
// We check previousStateRef inside the effect to detect actual changes
dispatch,
]);
}
@@ -0,0 +1,265 @@
/**
* Operation creator functions
*
* Helper functions to create well-formed CRDT operations.
* These are used by the Redux layer to convert user actions into operations.
*/
import { v4 as uuidv4 } from 'uuid';
import { HLCTimestamp } from '../crdt/hlc';
import {
CRDTOperation,
AddPanelOperation,
RemovePanelOperation,
UpdatePanelPositionOperation,
UpdatePanelSizeOperation,
UpdatePanelZIndexOperation,
UpdatePanelExploreStateOperation,
UpdateTitleOperation,
BatchOperation,
} from '../crdt/types';
import { SerializedExploreState } from '../state/types';
/**
* Create an add panel operation
*/
export function createAddPanelOperation(
mapUid: string,
nodeId: string,
timestamp: HLCTimestamp,
payload: {
panelId: string;
exploreId: string;
position: {
x: number;
y: number;
width: number;
height: number;
};
}
): AddPanelOperation {
return {
type: 'add-panel',
mapUid,
operationId: uuidv4(),
timestamp,
nodeId,
payload,
};
}
/**
* Create a remove panel operation
*/
export function createRemovePanelOperation(
mapUid: string,
nodeId: string,
timestamp: HLCTimestamp,
payload: {
panelId: string;
observedTags: string[];
}
): RemovePanelOperation {
return {
type: 'remove-panel',
mapUid,
operationId: uuidv4(),
timestamp,
nodeId,
payload,
};
}
/**
* Create an update panel position operation
*/
export function createUpdatePanelPositionOperation(
mapUid: string,
nodeId: string,
timestamp: HLCTimestamp,
payload: {
panelId: string;
x: number;
y: number;
}
): UpdatePanelPositionOperation {
return {
type: 'update-panel-position',
mapUid,
operationId: uuidv4(),
timestamp,
nodeId,
payload,
};
}
/**
* Create an update panel size operation
*/
export function createUpdatePanelSizeOperation(
mapUid: string,
nodeId: string,
timestamp: HLCTimestamp,
payload: {
panelId: string;
width: number;
height: number;
}
): UpdatePanelSizeOperation {
return {
type: 'update-panel-size',
mapUid,
operationId: uuidv4(),
timestamp,
nodeId,
payload,
};
}
/**
* Create an update panel z-index operation
*/
export function createUpdatePanelZIndexOperation(
mapUid: string,
nodeId: string,
timestamp: HLCTimestamp,
payload: {
panelId: string;
zIndex: number;
}
): UpdatePanelZIndexOperation {
return {
type: 'update-panel-zindex',
mapUid,
operationId: uuidv4(),
timestamp,
nodeId,
payload,
};
}
/**
* Create an update panel explore state operation
*/
export function createUpdatePanelExploreStateOperation(
mapUid: string,
nodeId: string,
timestamp: HLCTimestamp,
payload: {
panelId: string;
exploreState: SerializedExploreState | undefined;
}
): UpdatePanelExploreStateOperation {
return {
type: 'update-panel-explore-state',
mapUid,
operationId: uuidv4(),
timestamp,
nodeId,
payload,
};
}
/**
* Create an update title operation
*/
export function createUpdateTitleOperation(
mapUid: string,
nodeId: string,
timestamp: HLCTimestamp,
payload: {
title: string;
}
): UpdateTitleOperation {
return {
type: 'update-title',
mapUid,
operationId: uuidv4(),
timestamp,
nodeId,
payload,
};
}
/**
* Create a batch operation containing multiple sub-operations
*/
export function createBatchOperation(
mapUid: string,
nodeId: string,
timestamp: HLCTimestamp,
operations: CRDTOperation[]
): BatchOperation {
return {
type: 'batch',
mapUid,
operationId: uuidv4(),
timestamp,
nodeId,
payload: {
operations,
},
};
}
/**
* Create an operation to move multiple panels together
* Returns a batch operation with position updates for all panels
*/
export function createMultiPanelMoveOperation(
mapUid: string,
nodeId: string,
timestamp: HLCTimestamp,
panelMoves: Array<{ panelId: string; x: number; y: number }>
): BatchOperation {
const operations = panelMoves.map((move) =>
createUpdatePanelPositionOperation(mapUid, nodeId, timestamp, move)
);
return createBatchOperation(mapUid, nodeId, timestamp, operations);
}
/**
* Create an operation to duplicate a panel
* Returns a batch operation that adds a new panel with the same properties
*/
export function createDuplicatePanelOperation(
mapUid: string,
nodeId: string,
timestamp: HLCTimestamp,
sourcePanel: {
id: string;
exploreId: string;
position: { x: number; y: number; width: number; height: number };
exploreState?: SerializedExploreState;
},
offset: { x: number; y: number }
): BatchOperation {
const newPanelId = uuidv4();
const newExploreId = `explore-${uuidv4()}`;
const operations: CRDTOperation[] = [
createAddPanelOperation(mapUid, nodeId, timestamp, {
panelId: newPanelId,
exploreId: newExploreId,
position: {
x: sourcePanel.position.x + offset.x,
y: sourcePanel.position.y + offset.y,
width: sourcePanel.position.width,
height: sourcePanel.position.height,
},
}),
];
// If source panel has explore state, copy it
if (sourcePanel.exploreState) {
operations.push(
createUpdatePanelExploreStateOperation(mapUid, nodeId, timestamp, {
panelId: newPanelId,
exploreState: sourcePanel.exploreState,
})
);
}
return createBatchOperation(mapUid, nodeId, timestamp, operations);
}
@@ -0,0 +1,43 @@
/**
* Operations module exports
*
* CRDT operation management: queue, validation, creation, serialization
*/
// Operation queue
export { OperationQueue } from './queue';
export type { QueuedOperation, OperationQueueStats } from './queue';
// Validators
export { validateOperation, quickValidate } from './validators';
export type { ValidationResult, ValidationOptions } from './validators';
// Operation creators
export {
createAddPanelOperation,
createRemovePanelOperation,
createUpdatePanelPositionOperation,
createUpdatePanelSizeOperation,
createUpdatePanelZIndexOperation,
createUpdatePanelExploreStateOperation,
createUpdateTitleOperation,
createBatchOperation,
createMultiPanelMoveOperation,
createDuplicatePanelOperation,
} from './creators';
// Serialization
export {
serializeOperation,
deserializeOperation,
serializeOperations,
deserializeOperations,
serializeForWebSocket,
deserializeFromWebSocket,
compressOperation,
decompressOperation,
estimateOperationSize,
batchOperations,
deduplicateOperations,
} from './serialization';
export type { WebSocketMessage } from './serialization';
@@ -0,0 +1,288 @@
/**
* Operation Queue for CRDT operations
*
* Manages local and remote operations, ensuring they are applied in causal order.
* Handles deduplication, ordering, and buffering of out-of-order operations.
*/
import { HybridLogicalClock, compareHLC, HLCTimestamp } from '../crdt/hlc';
import { CRDTOperation } from '../crdt/types';
export interface QueuedOperation {
operation: CRDTOperation;
source: 'local' | 'remote';
enqueuedAt: number; // Timestamp when added to queue
}
export interface OperationQueueStats {
pendingCount: number;
appliedCount: number;
localCount: number;
remoteCount: number;
}
/**
* Operation queue that maintains causal ordering of CRDT operations
*/
export class OperationQueue {
private clock: HybridLogicalClock;
private nodeId: string;
// Operations waiting to be applied (sorted by HLC timestamp)
private pendingQueue: QueuedOperation[] = [];
// Set of operation IDs that have been applied (for deduplication)
private appliedOperations: Set<string> = new Set();
// Maximum number of applied operation IDs to keep in memory
private readonly maxAppliedHistorySize = 10000;
constructor(nodeId: string) {
this.nodeId = nodeId;
this.clock = new HybridLogicalClock(nodeId);
}
/**
* Get the current node ID
*/
getNodeId(): string {
return this.nodeId;
}
/**
* Get the current HLC
*/
getClock(): HybridLogicalClock {
return this.clock;
}
/**
* Add a local operation to the queue
* Automatically assigns a new timestamp from the local clock
*/
addLocalOperation(operation: CRDTOperation): CRDTOperation {
// Tick clock for new local event
const timestamp = this.clock.tick();
// Create operation with new timestamp
const timedOperation: CRDTOperation = {
...operation,
timestamp,
nodeId: this.nodeId,
};
// Add to pending queue
this.enqueueOperation(timedOperation, 'local');
return timedOperation;
}
/**
* Add a remote operation to the queue
* Updates local clock based on received timestamp
*/
addRemoteOperation(operation: CRDTOperation): boolean {
// Check if already applied (deduplication)
if (this.appliedOperations.has(operation.operationId)) {
return false; // Already applied, ignore
}
// Check if already in pending queue
if (this.pendingQueue.some((q) => q.operation.operationId === operation.operationId)) {
return false; // Already queued, ignore
}
// Update clock with received timestamp
this.clock.update(operation.timestamp);
// Add to pending queue
this.enqueueOperation(operation, 'remote');
return true;
}
/**
* Internal method to enqueue an operation and maintain sorted order
*/
private enqueueOperation(operation: CRDTOperation, source: 'local' | 'remote'): void {
const queued: QueuedOperation = {
operation,
source,
enqueuedAt: Date.now(),
};
// Insert in sorted order by HLC timestamp
const insertIndex = this.findInsertIndex(operation.timestamp);
this.pendingQueue.splice(insertIndex, 0, queued);
}
/**
* Binary search to find insertion index for maintaining sorted order
*/
private findInsertIndex(timestamp: HLCTimestamp): number {
let left = 0;
let right = this.pendingQueue.length;
while (left < right) {
const mid = Math.floor((left + right) / 2);
const comparison = compareHLC(this.pendingQueue[mid].operation.timestamp, timestamp);
if (comparison < 0) {
left = mid + 1;
} else {
right = mid;
}
}
return left;
}
/**
* Dequeue the next operation to apply
* Returns undefined if queue is empty
*/
dequeue(): CRDTOperation | undefined {
const queued = this.pendingQueue.shift();
if (!queued) {
return undefined;
}
// Mark as applied
this.markApplied(queued.operation.operationId);
return queued.operation;
}
/**
* Peek at the next operation without removing it
*/
peek(): CRDTOperation | undefined {
return this.pendingQueue[0]?.operation;
}
/**
* Get all pending operations (without removing them)
*/
getPendingOperations(): CRDTOperation[] {
return this.pendingQueue.map((q) => q.operation);
}
/**
* Check if an operation has been applied
*/
hasApplied(operationId: string): boolean {
return this.appliedOperations.has(operationId);
}
/**
* Mark an operation as applied
*/
markApplied(operationId: string): void {
this.appliedOperations.add(operationId);
// Limit memory usage by removing old entries
if (this.appliedOperations.size > this.maxAppliedHistorySize) {
this.pruneAppliedHistory();
}
}
/**
* Prune old entries from applied operations set
* Removes the oldest 10% of entries
*/
private pruneAppliedHistory(): void {
const toRemove = Math.floor(this.maxAppliedHistorySize * 0.1);
const entries = Array.from(this.appliedOperations);
// Remove first 10% (oldest)
for (let i = 0; i < toRemove && i < entries.length; i++) {
this.appliedOperations.delete(entries[i]);
}
}
/**
* Get the number of pending operations
*/
getPendingCount(): number {
return this.pendingQueue.length;
}
/**
* Check if queue is empty
*/
isEmpty(): boolean {
return this.pendingQueue.length === 0;
}
/**
* Clear all pending operations (useful for testing/reset)
*/
clearPending(): void {
this.pendingQueue = [];
}
/**
* Clear applied operations history
*/
clearApplied(): void {
this.appliedOperations.clear();
}
/**
* Get queue statistics
*/
getStats(): OperationQueueStats {
const localCount = this.pendingQueue.filter((q) => q.source === 'local').length;
const remoteCount = this.pendingQueue.filter((q) => q.source === 'remote').length;
return {
pendingCount: this.pendingQueue.length,
appliedCount: this.appliedOperations.size,
localCount,
remoteCount,
};
}
/**
* Drain all operations from the queue in order
* Returns array of operations in causal order
*/
drainAll(): CRDTOperation[] {
const operations: CRDTOperation[] = [];
while (!this.isEmpty()) {
const op = this.dequeue();
if (op) {
operations.push(op);
}
}
return operations;
}
/**
* Get operations older than a certain age (in milliseconds)
* Useful for detecting stale operations
*/
getStaleOperations(maxAge: number): CRDTOperation[] {
const now = Date.now();
return this.pendingQueue
.filter((q) => now - q.enqueuedAt > maxAge)
.map((q) => q.operation);
}
/**
* Remove specific operations by ID
* Returns number of operations removed
*/
removeOperations(operationIds: string[]): number {
const idsToRemove = new Set(operationIds);
const initialLength = this.pendingQueue.length;
this.pendingQueue = this.pendingQueue.filter(
(q) => !idsToRemove.has(q.operation.operationId)
);
return initialLength - this.pendingQueue.length;
}
}
@@ -0,0 +1,290 @@
/**
* Operation serialization utilities
*
* Handles serialization/deserialization of CRDT operations for network transmission
* and storage. Supports both JSON and potential future binary formats.
*/
import { CRDTOperation } from '../crdt/types';
/**
* Serialize an operation to JSON string
*/
export function serializeOperation(operation: CRDTOperation): string {
return JSON.stringify(operation);
}
/**
* Deserialize an operation from JSON string
*/
export function deserializeOperation(json: string): CRDTOperation {
return JSON.parse(json);
}
/**
* Serialize multiple operations to JSON string
*/
export function serializeOperations(operations: CRDTOperation[]): string {
return JSON.stringify(operations);
}
/**
* Deserialize multiple operations from JSON string
*/
export function deserializeOperations(json: string): CRDTOperation[] {
return JSON.parse(json);
}
/**
* Serialize operation for WebSocket transmission
* Adds metadata for efficient routing
*/
export interface WebSocketMessage {
type: 'operation' | 'batch' | 'sync-request' | 'sync-response';
mapUid: string;
data: CRDTOperation | CRDTOperation[];
timestamp: number; // Message sent timestamp
}
export function serializeForWebSocket(
operation: CRDTOperation | CRDTOperation[]
): string {
const operations = Array.isArray(operation) ? operation : [operation];
const mapUid = operations[0]?.mapUid || '';
const message: WebSocketMessage = {
type: Array.isArray(operation) ? 'batch' : 'operation',
mapUid,
data: operation,
timestamp: Date.now(),
};
return JSON.stringify(message);
}
export function deserializeFromWebSocket(json: string): WebSocketMessage {
return JSON.parse(json);
}
/**
* Compress operation by removing redundant data
* Useful for bandwidth optimization
*/
export function compressOperation(operation: CRDTOperation): any {
// Remove verbose field names, use short aliases
const compressed: any = {
t: operation.type,
i: operation.operationId,
m: operation.mapUid,
n: operation.nodeId,
ts: {
l: operation.timestamp.logicalTime,
w: operation.timestamp.wallTime,
n: operation.timestamp.nodeId,
},
p: compressPayload(operation),
};
return compressed;
}
function compressPayload(operation: CRDTOperation): any {
const payload = (operation as any).payload;
if (!payload) {
return undefined;
}
// Type-specific compression
switch (operation.type) {
case 'add-panel':
return {
pi: payload.panelId,
ei: payload.exploreId,
pos: {
x: payload.position.x,
y: payload.position.y,
w: payload.position.width,
h: payload.position.height,
},
};
case 'remove-panel':
return {
pi: payload.panelId,
t: payload.observedTags,
};
case 'update-panel-position':
return {
pi: payload.panelId,
x: payload.x,
y: payload.y,
};
case 'update-panel-size':
return {
pi: payload.panelId,
w: payload.width,
h: payload.height,
};
case 'update-panel-zindex':
return {
pi: payload.panelId,
z: payload.zIndex,
};
case 'update-panel-explore-state':
return {
pi: payload.panelId,
es: payload.exploreState,
};
case 'update-title':
return {
t: payload.title,
};
case 'batch':
return {
ops: payload.operations.map(compressOperation),
};
default:
return payload;
}
}
/**
* Decompress operation from compressed format
*/
export function decompressOperation(compressed: any): CRDTOperation {
const base = {
type: compressed.t,
operationId: compressed.i,
mapUid: compressed.m,
nodeId: compressed.n,
timestamp: {
logicalTime: compressed.ts.l,
wallTime: compressed.ts.w,
nodeId: compressed.ts.n,
},
};
const payload = decompressPayload(compressed.t, compressed.p);
return {
...base,
payload,
} as CRDTOperation;
}
function decompressPayload(type: string, compressed: any): any {
if (!compressed) {
return undefined;
}
switch (type) {
case 'add-panel':
return {
panelId: compressed.pi,
exploreId: compressed.ei,
position: {
x: compressed.pos.x,
y: compressed.pos.y,
width: compressed.pos.w,
height: compressed.pos.h,
},
};
case 'remove-panel':
return {
panelId: compressed.pi,
observedTags: compressed.t,
};
case 'update-panel-position':
return {
panelId: compressed.pi,
x: compressed.x,
y: compressed.y,
};
case 'update-panel-size':
return {
panelId: compressed.pi,
width: compressed.w,
height: compressed.h,
};
case 'update-panel-zindex':
return {
panelId: compressed.pi,
zIndex: compressed.z,
};
case 'update-panel-explore-state':
return {
panelId: compressed.pi,
exploreState: compressed.es,
};
case 'update-title':
return {
title: compressed.t,
};
case 'batch':
return {
operations: compressed.ops.map(decompressOperation),
};
default:
return compressed;
}
}
/**
* Calculate approximate size of an operation in bytes
* Useful for monitoring bandwidth usage
*/
export function estimateOperationSize(operation: CRDTOperation): number {
const json = serializeOperation(operation);
// Approximate UTF-8 byte length (not exact, but close enough)
return new Blob([json]).size;
}
/**
* Batch multiple operations into a single message
* Useful for reducing WebSocket message overhead
*/
export function batchOperations(
operations: CRDTOperation[],
maxBatchSize: number = 10
): CRDTOperation[][] {
const batches: CRDTOperation[][] = [];
for (let i = 0; i < operations.length; i += maxBatchSize) {
batches.push(operations.slice(i, i + maxBatchSize));
}
return batches;
}
/**
* Deduplicate operations by operation ID
* Keeps the first occurrence of each unique operation ID
*/
export function deduplicateOperations(operations: CRDTOperation[]): CRDTOperation[] {
const seen = new Set<string>();
const deduplicated: CRDTOperation[] = [];
for (const operation of operations) {
if (!seen.has(operation.operationId)) {
seen.add(operation.operationId);
deduplicated.push(operation);
}
}
return deduplicated;
}
@@ -0,0 +1,385 @@
/**
* Operation validators
*
* Validates CRDT operations for correctness, security, and schema compliance.
*/
import { CRDTOperation } from '../crdt/types';
export interface ValidationResult {
valid: boolean;
errors: string[];
}
export interface ValidationOptions {
maxPanelSize?: { width: number; height: number };
minPanelSize?: { width: number; height: number };
maxTitleLength?: number;
allowNegativeCoordinates?: boolean;
maxCoordinate?: number;
}
const DEFAULT_OPTIONS: Required<ValidationOptions> = {
maxPanelSize: { width: 5000, height: 5000 },
minPanelSize: { width: 100, height: 100 },
maxTitleLength: 255,
allowNegativeCoordinates: false,
maxCoordinate: 20000,
};
/**
* Validate a CRDT operation
*/
export function validateOperation(
operation: CRDTOperation,
options: ValidationOptions = {}
): ValidationResult {
const opts = { ...DEFAULT_OPTIONS, ...options };
const errors: string[] = [];
// Validate common fields
if (!operation.operationId || typeof operation.operationId !== 'string') {
errors.push('Operation ID is required and must be a string');
}
if (!operation.mapUid || typeof operation.mapUid !== 'string') {
errors.push('Map UID is required and must be a string');
}
if (!operation.nodeId || typeof operation.nodeId !== 'string') {
errors.push('Node ID is required and must be a string');
}
if (!operation.timestamp) {
errors.push('Timestamp is required');
} else {
validateTimestamp(operation.timestamp, errors);
}
// Validate type-specific fields
switch (operation.type) {
case 'add-panel':
validateAddPanel(operation, opts, errors);
break;
case 'remove-panel':
validateRemovePanel(operation, errors);
break;
case 'update-panel-position':
validateUpdatePanelPosition(operation, opts, errors);
break;
case 'update-panel-size':
validateUpdatePanelSize(operation, opts, errors);
break;
case 'update-panel-zindex':
validateUpdatePanelZIndex(operation, errors);
break;
case 'update-panel-explore-state':
validateUpdatePanelExploreState(operation, errors);
break;
case 'update-title':
validateUpdateTitle(operation, opts, errors);
break;
case 'batch':
validateBatchOperation(operation, opts, errors);
break;
default:
errors.push(`Unknown operation type: ${(operation as any).type}`);
}
return {
valid: errors.length === 0,
errors,
};
}
function validateTimestamp(timestamp: any, errors: string[]): void {
if (typeof timestamp !== 'object' || timestamp === null) {
errors.push('Timestamp must be an object');
return;
}
if (typeof timestamp.logicalTime !== 'number' || timestamp.logicalTime < 0) {
errors.push('Timestamp logical time must be a non-negative number');
}
if (typeof timestamp.wallTime !== 'number' || timestamp.wallTime < 0) {
errors.push('Timestamp wall time must be a non-negative number');
}
if (typeof timestamp.nodeId !== 'string' || !timestamp.nodeId) {
errors.push('Timestamp node ID must be a non-empty string');
}
// Check for reasonable wall time (not too far in future)
const now = Date.now();
const maxFutureOffset = 60000; // 1 minute
if (timestamp.wallTime > now + maxFutureOffset) {
errors.push('Timestamp wall time is too far in the future');
}
}
function validateAddPanel(operation: any, opts: Required<ValidationOptions>, errors: string[]): void {
if (!operation.payload) {
errors.push('Add panel operation requires payload');
return;
}
const { panelId, exploreId, position } = operation.payload;
if (!panelId || typeof panelId !== 'string') {
errors.push('Panel ID is required and must be a string');
}
if (!exploreId || typeof exploreId !== 'string') {
errors.push('Explore ID is required and must be a string');
}
if (!position || typeof position !== 'object') {
errors.push('Position is required and must be an object');
return;
}
validatePosition(position, opts, errors);
}
function validateRemovePanel(operation: any, errors: string[]): void {
if (!operation.payload) {
errors.push('Remove panel operation requires payload');
return;
}
const { panelId, observedTags } = operation.payload;
if (!panelId || typeof panelId !== 'string') {
errors.push('Panel ID is required and must be a string');
}
if (!Array.isArray(observedTags)) {
errors.push('Observed tags must be an array');
} else if (observedTags.some((tag) => typeof tag !== 'string')) {
errors.push('All observed tags must be strings');
}
}
function validateUpdatePanelPosition(
operation: any,
opts: Required<ValidationOptions>,
errors: string[]
): void {
if (!operation.payload) {
errors.push('Update panel position operation requires payload');
return;
}
const { panelId, x, y } = operation.payload;
if (!panelId || typeof panelId !== 'string') {
errors.push('Panel ID is required and must be a string');
}
if (typeof x !== 'number') {
errors.push('X coordinate must be a number');
} else {
validateCoordinate('x', x, opts, errors);
}
if (typeof y !== 'number') {
errors.push('Y coordinate must be a number');
} else {
validateCoordinate('y', y, opts, errors);
}
}
function validateUpdatePanelSize(
operation: any,
opts: Required<ValidationOptions>,
errors: string[]
): void {
if (!operation.payload) {
errors.push('Update panel size operation requires payload');
return;
}
const { panelId, width, height } = operation.payload;
if (!panelId || typeof panelId !== 'string') {
errors.push('Panel ID is required and must be a string');
}
if (typeof width !== 'number') {
errors.push('Width must be a number');
} else {
validateDimension('width', width, opts, errors);
}
if (typeof height !== 'number') {
errors.push('Height must be a number');
} else {
validateDimension('height', height, opts, errors);
}
}
function validateUpdatePanelZIndex(operation: any, errors: string[]): void {
if (!operation.payload) {
errors.push('Update panel z-index operation requires payload');
return;
}
const { panelId, zIndex } = operation.payload;
if (!panelId || typeof panelId !== 'string') {
errors.push('Panel ID is required and must be a string');
}
if (typeof zIndex !== 'number' || zIndex < 0) {
errors.push('Z-index must be a non-negative number');
}
}
function validateUpdatePanelExploreState(operation: any, errors: string[]): void {
if (!operation.payload) {
errors.push('Update panel explore state operation requires payload');
return;
}
const { panelId, exploreState } = operation.payload;
if (!panelId || typeof panelId !== 'string') {
errors.push('Panel ID is required and must be a string');
}
// exploreState can be undefined, but if present must be an object
if (exploreState !== undefined && (typeof exploreState !== 'object' || exploreState === null)) {
errors.push('Explore state must be an object or undefined');
}
}
function validateUpdateTitle(operation: any, opts: Required<ValidationOptions>, errors: string[]): void {
if (!operation.payload) {
errors.push('Update title operation requires payload');
return;
}
const { title } = operation.payload;
if (typeof title !== 'string') {
errors.push('Title must be a string');
} else if (title.length > opts.maxTitleLength) {
errors.push(`Title exceeds maximum length of ${opts.maxTitleLength} characters`);
}
}
function validateBatchOperation(operation: any, opts: Required<ValidationOptions>, errors: string[]): void {
if (!operation.payload) {
errors.push('Batch operation requires payload');
return;
}
const { operations } = operation.payload;
if (!Array.isArray(operations)) {
errors.push('Batch operations must be an array');
return;
}
if (operations.length === 0) {
errors.push('Batch operation cannot be empty');
}
// Validate each sub-operation
operations.forEach((subOp: any, index: number) => {
const result = validateOperation(subOp, opts);
if (!result.valid) {
errors.push(`Batch operation[${index}]: ${result.errors.join(', ')}`);
}
});
}
function validatePosition(
position: any,
opts: Required<ValidationOptions>,
errors: string[]
): void {
const { x, y, width, height } = position;
if (typeof x !== 'number') {
errors.push('Position x must be a number');
} else {
validateCoordinate('x', x, opts, errors);
}
if (typeof y !== 'number') {
errors.push('Position y must be a number');
} else {
validateCoordinate('y', y, opts, errors);
}
if (typeof width !== 'number') {
errors.push('Position width must be a number');
} else {
validateDimension('width', width, opts, errors);
}
if (typeof height !== 'number') {
errors.push('Position height must be a number');
} else {
validateDimension('height', height, opts, errors);
}
}
function validateCoordinate(
name: string,
value: number,
opts: Required<ValidationOptions>,
errors: string[]
): void {
if (!Number.isFinite(value)) {
errors.push(`${name} must be a finite number`);
return;
}
if (!opts.allowNegativeCoordinates && value < 0) {
errors.push(`${name} cannot be negative`);
}
if (Math.abs(value) > opts.maxCoordinate) {
errors.push(`${name} exceeds maximum coordinate value of ${opts.maxCoordinate}`);
}
}
function validateDimension(
name: string,
value: number,
opts: Required<ValidationOptions>,
errors: string[]
): void {
if (!Number.isFinite(value)) {
errors.push(`${name} must be a finite number`);
return;
}
const minSize = name === 'width' ? opts.minPanelSize.width : opts.minPanelSize.height;
const maxSize = name === 'width' ? opts.maxPanelSize.width : opts.maxPanelSize.height;
if (value < minSize) {
errors.push(`${name} must be at least ${minSize}`);
}
if (value > maxSize) {
errors.push(`${name} cannot exceed ${maxSize}`);
}
}
/**
* Quick validation that only checks critical fields
* Useful for performance-sensitive paths
*/
export function quickValidate(operation: CRDTOperation): boolean {
return !!(
operation.operationId &&
operation.mapUid &&
operation.nodeId &&
operation.timestamp &&
operation.type
);
}
@@ -0,0 +1,216 @@
/**
* React hook for real-time CRDT synchronization via Grafana Live
*
* This hook connects to the Grafana Live WebSocket channel for a map
* and handles bidirectional operation synchronization.
*/
import { useEffect, useRef, useState } from 'react';
import { Unsubscribable } from 'rxjs';
import { LiveChannelAddress, LiveChannelScope, isLiveChannelMessageEvent } from '@grafana/data';
import { getGrafanaLiveSrv } from '@grafana/runtime';
import { StoreState, useDispatch, useSelector } from 'app/types/store';
import { CRDTOperation } from '../crdt/types';
import { OperationQueue } from '../operations/queue';
import { applyOperation, clearPendingOperations, setOnlineStatus } from '../state/crdtSlice';
import { selectPendingOperations, selectNodeId } from '../state/selectors';
export interface RealtimeSyncOptions {
mapUid: string;
enabled?: boolean;
onError?: (error: Error) => void;
onConnected?: () => void;
onDisconnected?: () => void;
}
export interface RealtimeSyncStatus {
isConnected: boolean;
isInitialized: boolean;
error?: Error;
}
/**
* Hook to synchronize CRDT state with Grafana Live
*/
export function useRealtimeSync(options: RealtimeSyncOptions): RealtimeSyncStatus {
const { mapUid, enabled = true, onError, onConnected, onDisconnected } = options;
const dispatch = useDispatch();
const nodeId = useSelector((state: StoreState) => selectNodeId(state.exploreMapCRDT));
const pendingOperations = useSelector((state: StoreState) => selectPendingOperations(state.exploreMapCRDT));
const [status, setStatus] = useState<RealtimeSyncStatus>({
isConnected: false,
isInitialized: false,
});
const subscriptionRef = useRef<Unsubscribable | null>(null);
const queueRef = useRef<OperationQueue>(new OperationQueue(nodeId));
const appliedOpsRef = useRef<Set<string>>(new Set());
const channelAddressRef = useRef<LiveChannelAddress | null>(null);
useEffect(() => {
if (!enabled || !mapUid) {
return;
}
let isSubscribed = true;
const connect = () => {
try {
const liveService = getGrafanaLiveSrv();
if (!liveService) {
throw new Error('Grafana Live service not available');
}
// 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<CRDTOperation>(channelAddress).subscribe({
next: (event) => {
if (!isSubscribed) {
return;
}
try {
// Handle message events
if (isLiveChannelMessageEvent(event)) {
const operation: CRDTOperation = event.message;
// Skip if this is our own operation
if (operation.nodeId === nodeId) {
return;
}
// Skip if already applied
if (appliedOpsRef.current.has(operation.operationId)) {
return;
}
// Add to queue and apply
const added = queueRef.current.addRemoteOperation(operation);
if (added) {
appliedOpsRef.current.add(operation.operationId);
dispatch(applyOperation({ operation }));
}
}
} catch (error) {
console.error('[CRDT] Failed to handle incoming operation:', error);
if (onError && error instanceof Error) {
onError(error);
}
}
},
error: (error) => {
console.error('[CRDT] Channel error:', error);
setStatus((prev) => ({
...prev,
isConnected: false,
error: error instanceof Error ? error : new Error(String(error)),
}));
dispatch(setOnlineStatus({ isOnline: false }));
if (onError && error instanceof Error) {
onError(error);
}
if (onDisconnected) {
onDisconnected();
}
},
complete: () => {
if (isSubscribed) {
setStatus((prev) => ({ ...prev, isConnected: false }));
dispatch(setOnlineStatus({ isOnline: false }));
if (onDisconnected) {
onDisconnected();
}
}
},
});
subscriptionRef.current = subscription;
// Mark as connected
setStatus({
isConnected: true,
isInitialized: true,
error: undefined,
});
dispatch(setOnlineStatus({ isOnline: true }));
if (onConnected) {
onConnected();
}
} catch (error) {
console.error('[CRDT] Failed to connect to Live channel:', error);
setStatus({
isConnected: false,
isInitialized: true,
error: error instanceof Error ? error : new Error(String(error)),
});
if (onError && error instanceof Error) {
onError(error);
}
}
};
connect();
return () => {
isSubscribed = false;
if (subscriptionRef.current) {
subscriptionRef.current.unsubscribe();
subscriptionRef.current = null;
}
channelAddressRef.current = null;
};
}, [mapUid, enabled, nodeId, dispatch, onError, onConnected, onDisconnected]);
// Broadcast pending operations
useEffect(() => {
if (!status.isConnected || !channelAddressRef.current || !pendingOperations || pendingOperations.length === 0) {
return;
}
const liveService = getGrafanaLiveSrv();
if (!liveService) {
console.error('[CRDT] Live service not available');
return;
}
const channelAddress = channelAddressRef.current;
// Broadcast each pending operation
for (const operation of pendingOperations) {
try {
// Mark as applied locally
appliedOpsRef.current.add(operation.operationId);
// Publish to channel
liveService.publish(channelAddress, operation).catch((error) => {
console.error('[CRDT] Failed to broadcast operation:', error);
});
} catch (error) {
console.error('[CRDT] Failed to broadcast operation:', error);
}
}
// Clear pending operations after broadcast
dispatch(clearPendingOperations());
}, [pendingOperations, status.isConnected, dispatch]);
return status;
}
@@ -0,0 +1,562 @@
/**
* CRDT-based Redux slice for Explore Map
*
* This slice wraps the CRDT state manager and provides Redux actions
* for applying operations and managing local UI state.
*/
import { createSlice, PayloadAction } from '@reduxjs/toolkit';
import { v4 as uuidv4 } from 'uuid';
import { generateExploreId } from 'app/core/utils/explore';
import { CRDTStateManager } from '../crdt/state';
import { CRDTOperation } from '../crdt/types';
import { CanvasViewport, SerializedExploreState, UserCursor } from './types';
/**
* Combined state: CRDT state + local UI state
*/
export interface ExploreMapCRDTState {
// Map metadata
uid?: string;
// CRDT state manager instance (serialized)
crdtStateJSON?: string;
// Node ID for this client
nodeId: string;
// Operation queue (not serialized, reconstructed on load)
pendingOperations: CRDTOperation[];
// Local UI state (not replicated)
local: {
viewport: CanvasViewport;
selectedPanelIds: string[];
cursors: Record<string, UserCursor>;
isOnline: boolean;
isSyncing: boolean;
};
}
const initialViewport: CanvasViewport = {
zoom: 1,
panX: -4040,
panY: -4460,
};
/**
* Create initial state with a new node ID
*/
export function createInitialCRDTState(mapUid?: string): ExploreMapCRDTState {
const nodeId = uuidv4();
const manager = new CRDTStateManager(mapUid || '', nodeId);
return {
uid: mapUid,
crdtStateJSON: JSON.stringify(manager.toJSON()),
nodeId,
pendingOperations: [],
local: {
viewport: initialViewport,
selectedPanelIds: [],
cursors: {},
isOnline: false,
isSyncing: false,
},
};
}
const initialState: ExploreMapCRDTState = createInitialCRDTState();
/**
* Helper to get CRDT manager from state
*/
function getCRDTManager(state: ExploreMapCRDTState): CRDTStateManager {
if (!state.crdtStateJSON) {
return new CRDTStateManager(state.uid || '', state.nodeId);
}
const json = JSON.parse(state.crdtStateJSON);
return CRDTStateManager.fromJSON(json, state.nodeId);
}
/**
* Helper to save CRDT manager to state
*/
function saveCRDTManager(state: ExploreMapCRDTState, manager: CRDTStateManager): void {
state.crdtStateJSON = JSON.stringify(manager.toJSON());
}
const crdtSlice = createSlice({
name: 'exploreMapCRDT',
initialState,
reducers: {
/**
* Initialize with a new map UID
*/
initializeMap: (state, action: PayloadAction<{ uid: string }>) => {
state.uid = action.payload.uid;
const manager = new CRDTStateManager(action.payload.uid, state.nodeId);
saveCRDTManager(state, manager);
},
/**
* Load CRDT state from server
*/
loadState: (state, action: PayloadAction<{ crdtState: any }>) => {
const manager = CRDTStateManager.fromJSON(action.payload.crdtState, state.nodeId);
saveCRDTManager(state, manager);
},
/**
* Initialize CRDT state from legacy ExploreMapState (for backward compatibility)
*/
initializeFromLegacyState: (state, action: PayloadAction<{
uid?: string;
title?: string;
panels: Record<string, {
id: string;
exploreId: string;
position: { x: number; y: number; width: number; height: number; zIndex: number };
exploreState?: SerializedExploreState;
}>;
viewport: CanvasViewport;
}>) => {
const { uid, title, panels, viewport } = action.payload;
// Create a new manager
const manager = new CRDTStateManager(uid || '', state.nodeId);
// Set map title if provided
if (title) {
const titleOp = manager.createUpdateTitleOperation(title);
manager.applyOperation(titleOp);
}
// Add all panels
for (const panel of Object.values(panels)) {
const addPanelOp = manager.createAddPanelOperation(
panel.id,
panel.exploreId,
{
x: panel.position.x,
y: panel.position.y,
width: panel.position.width,
height: panel.position.height,
}
);
manager.applyOperation(addPanelOp);
// Set z-index
const zIndexOp = manager.createUpdatePanelZIndexOperation(panel.id);
if (zIndexOp) {
manager.applyOperation(zIndexOp);
}
// Save explore state if present
if (panel.exploreState) {
const exploreStateOp = manager.createUpdatePanelExploreStateOperation(
panel.id,
panel.exploreState
);
if (exploreStateOp) {
manager.applyOperation(exploreStateOp);
}
}
}
// Update state
state.uid = uid;
saveCRDTManager(state, manager);
state.local.viewport = viewport;
},
/**
* Apply a local or remote operation
*/
applyOperation: (state, action: PayloadAction<{ operation: CRDTOperation }>) => {
const manager = getCRDTManager(state);
manager.applyOperation(action.payload.operation);
saveCRDTManager(state, manager);
},
/**
* Apply multiple operations in batch
*/
applyOperations: (state, action: PayloadAction<{ operations: CRDTOperation[] }>) => {
const manager = getCRDTManager(state);
for (const operation of action.payload.operations) {
manager.applyOperation(operation);
}
saveCRDTManager(state, manager);
},
/**
* Add a panel (creates and applies operation)
*/
addPanel: (state, action: PayloadAction<{
viewportSize?: { width: number; height: number };
position?: { x: number; y: number; width: number; height: number };
}>) => {
const manager = getCRDTManager(state);
// Calculate position
const viewportSize = action.payload.viewportSize || { width: 1920, height: 1080 };
const canvasCenterX = (-state.local.viewport.panX + viewportSize.width / 2) / state.local.viewport.zoom;
const canvasCenterY = (-state.local.viewport.panY + viewportSize.height / 2) / state.local.viewport.zoom;
const panelWidth = action.payload.position?.width || 600;
const panelHeight = action.payload.position?.height || 400;
const panelCount = manager.getPanelIds().length;
const offset = panelCount * 30;
const position = action.payload.position || {
x: canvasCenterX - panelWidth / 2 + offset,
y: canvasCenterY - panelHeight / 2 + offset,
width: panelWidth,
height: panelHeight,
};
// Create operation
const panelId = uuidv4();
const exploreId = generateExploreId();
const operation = manager.createAddPanelOperation(panelId, exploreId, position);
// Apply locally
manager.applyOperation(operation);
saveCRDTManager(state, manager);
// Add to pending operations for broadcast
state.pendingOperations.push(operation);
// Select the new panel
state.local.selectedPanelIds = [panelId];
},
/**
* Remove a panel
*/
removePanel: (state, action: PayloadAction<{ panelId: string }>) => {
const manager = getCRDTManager(state);
const operation = manager.createRemovePanelOperation(action.payload.panelId);
if (!operation) {
return; // Panel doesn't exist
}
// Apply locally
manager.applyOperation(operation);
saveCRDTManager(state, manager);
// Add to pending operations
state.pendingOperations.push(operation);
// Deselect the panel
state.local.selectedPanelIds = state.local.selectedPanelIds.filter(
(id) => id !== action.payload.panelId
);
},
/**
* Update panel position
*/
updatePanelPosition: (
state,
action: PayloadAction<{ panelId: string; x: number; y: number }>
) => {
const manager = getCRDTManager(state);
const operation = manager.createUpdatePanelPositionOperation(
action.payload.panelId,
action.payload.x,
action.payload.y
);
if (!operation) {
return;
}
manager.applyOperation(operation);
saveCRDTManager(state, manager);
state.pendingOperations.push(operation);
},
/**
* Update multiple panel positions (for moving selected panels together)
*/
updateMultiplePanelPositions: (
state,
action: PayloadAction<{ panelId: string; deltaX: number; deltaY: number }>
) => {
const manager = getCRDTManager(state);
const { panelId, deltaX, deltaY } = action.payload;
// Get all selected panels except the one being dragged
const panelsToMove = state.local.selectedPanelIds.filter((id) => id !== panelId);
// Update position for each panel
for (const id of panelsToMove) {
const panel = manager.getPanelForUI(id);
if (panel) {
const operation = manager.createUpdatePanelPositionOperation(
id,
panel.position.x + deltaX,
panel.position.y + deltaY
);
if (operation) {
manager.applyOperation(operation);
state.pendingOperations.push(operation);
}
}
}
saveCRDTManager(state, manager);
},
/**
* Update panel size
*/
updatePanelSize: (
state,
action: PayloadAction<{ panelId: string; width: number; height: number }>
) => {
const manager = getCRDTManager(state);
const operation = manager.createUpdatePanelSizeOperation(
action.payload.panelId,
action.payload.width,
action.payload.height
);
if (!operation) {
return;
}
manager.applyOperation(operation);
saveCRDTManager(state, manager);
state.pendingOperations.push(operation);
},
/**
* Bring panel to front
*/
bringPanelToFront: (state, action: PayloadAction<{ panelId: string }>) => {
const manager = getCRDTManager(state);
const operation = manager.createUpdatePanelZIndexOperation(action.payload.panelId);
if (!operation) {
return;
}
manager.applyOperation(operation);
saveCRDTManager(state, manager);
state.pendingOperations.push(operation);
},
/**
* Update panel explore state
*/
savePanelExploreState: (
state,
action: PayloadAction<{ panelId: string; exploreState: SerializedExploreState }>
) => {
const manager = getCRDTManager(state);
const operation = manager.createUpdatePanelExploreStateOperation(
action.payload.panelId,
action.payload.exploreState
);
if (!operation) {
return;
}
manager.applyOperation(operation);
saveCRDTManager(state, manager);
state.pendingOperations.push(operation);
},
/**
* Update map title
*/
updateMapTitle: (state, action: PayloadAction<{ title: string }>) => {
const manager = getCRDTManager(state);
const operation = manager.createUpdateTitleOperation(action.payload.title);
manager.applyOperation(operation);
saveCRDTManager(state, manager);
state.pendingOperations.push(operation);
},
/**
* Duplicate a panel
*/
duplicatePanel: (state, action: PayloadAction<{ panelId: string }>) => {
const manager = getCRDTManager(state);
const sourcePanel = manager.getPanelForUI(action.payload.panelId);
if (!sourcePanel) {
return;
}
// Create new panel with offset
const newPanelId = uuidv4();
const newExploreId = generateExploreId();
const offset = 30;
const addOperation = manager.createAddPanelOperation(
newPanelId,
newExploreId,
{
x: sourcePanel.position.x + offset,
y: sourcePanel.position.y + offset,
width: sourcePanel.position.width,
height: sourcePanel.position.height,
}
);
manager.applyOperation(addOperation);
state.pendingOperations.push(addOperation);
// Copy explore state if exists
if (sourcePanel.exploreState) {
const stateOperation = manager.createUpdatePanelExploreStateOperation(
newPanelId,
sourcePanel.exploreState
);
if (stateOperation) {
manager.applyOperation(stateOperation);
state.pendingOperations.push(stateOperation);
}
}
saveCRDTManager(state, manager);
state.local.selectedPanelIds = [newPanelId];
},
/**
* Clear pending operations (after broadcast)
*/
clearPendingOperations: (state) => {
state.pendingOperations = [];
},
// Local UI state actions (not replicated)
/**
* Update viewport (pan/zoom)
*/
updateViewport: (state, action: PayloadAction<Partial<CanvasViewport>>) => {
state.local.viewport = { ...state.local.viewport, ...action.payload };
},
/**
* Select panel(s)
*/
selectPanel: (state, action: PayloadAction<{ panelId?: string; addToSelection?: boolean }>) => {
const { panelId, addToSelection } = action.payload;
if (!panelId) {
state.local.selectedPanelIds = [];
return;
}
if (addToSelection) {
if (state.local.selectedPanelIds.includes(panelId)) {
state.local.selectedPanelIds = state.local.selectedPanelIds.filter((id) => id !== panelId);
} else {
state.local.selectedPanelIds.push(panelId);
}
} else {
state.local.selectedPanelIds = [panelId];
}
},
/**
* Select multiple panels
*/
selectMultiplePanels: (state, action: PayloadAction<{ panelIds: string[]; addToSelection?: boolean }>) => {
const { panelIds, addToSelection } = action.payload;
if (addToSelection) {
// Add to existing selection
const newSelections = panelIds.filter((id) => !state.local.selectedPanelIds.includes(id));
state.local.selectedPanelIds.push(...newSelections);
} else {
// Replace selection
state.local.selectedPanelIds = panelIds;
}
},
/**
* Update cursor position
*/
updateCursor: (state, action: PayloadAction<UserCursor>) => {
state.local.cursors[action.payload.userId] = action.payload;
},
/**
* Remove cursor
*/
removeCursor: (state, action: PayloadAction<{ userId: string }>) => {
delete state.local.cursors[action.payload.userId];
},
/**
* Set online status
*/
setOnlineStatus: (state, action: PayloadAction<{ isOnline: boolean }>) => {
state.local.isOnline = action.payload.isOnline;
},
/**
* Set syncing status
*/
setSyncingStatus: (state, action: PayloadAction<{ isSyncing: boolean }>) => {
state.local.isSyncing = action.payload.isSyncing;
},
/**
* Clear all state (reset)
*/
clearMap: (state) => {
const newState = createInitialCRDTState(state.uid);
Object.assign(state, newState);
},
},
});
export const {
initializeMap,
loadState,
initializeFromLegacyState,
applyOperation,
applyOperations,
addPanel,
removePanel,
updatePanelPosition,
updateMultiplePanelPositions,
updatePanelSize,
bringPanelToFront,
savePanelExploreState,
updateMapTitle,
duplicatePanel,
clearPendingOperations,
updateViewport,
selectPanel,
selectMultiplePanels,
updateCursor,
removeCursor,
setOnlineStatus,
setSyncingStatus,
clearMap,
} = crdtSlice.actions;
export const crdtReducer = crdtSlice.reducer;
@@ -0,0 +1,260 @@
/**
* Redux middleware for CRDT operations
*
* This middleware intercepts actions that modify state and:
* 1. Broadcasts pending operations to the WebSocket channel
* 2. Handles incoming remote operations
* 3. Manages operation queue and deduplication
*/
import { Middleware } from '@reduxjs/toolkit';
import { CRDTOperation } from '../crdt/types';
import { selectPendingOperations } from './selectors';
import { clearPendingOperations } from './crdtSlice';
export interface OperationBroadcaster {
/**
* Broadcast an operation to other clients
*/
broadcast(operation: CRDTOperation): void;
/**
* Broadcast multiple operations
*/
broadcastBatch(operations: CRDTOperation[]): void;
/**
* Subscribe to incoming operations
*/
subscribe(handler: (operation: CRDTOperation) => void): () => void;
}
/**
* Create CRDT operation middleware
*
* @param broadcaster - Interface for broadcasting operations (WebSocket, etc.)
*/
export function createOperationMiddleware(
broadcaster?: OperationBroadcaster
): Middleware {
return (store) => (next) => (action) => {
// Apply action first
const result = next(action);
// After state update, check for pending operations to broadcast
if (broadcaster && shouldBroadcast((action as any).type)) {
const state = store.getState().exploreMapCRDT;
const pendingOps = selectPendingOperations(state);
if (pendingOps.length > 0) {
// Broadcast all pending operations
for (const op of pendingOps) {
try {
broadcaster.broadcast(op);
} catch (error) {
console.error('Failed to broadcast operation:', error);
}
}
// Clear pending operations after successful broadcast
store.dispatch(clearPendingOperations());
}
}
return result;
};
}
/**
* Determine if an action should trigger broadcasting
*/
function shouldBroadcast(actionType: string): boolean {
const broadcastableActions = [
'exploreMapCRDT/addPanel',
'exploreMapCRDT/removePanel',
'exploreMapCRDT/updatePanelPosition',
'exploreMapCRDT/updatePanelSize',
'exploreMapCRDT/bringPanelToFront',
'exploreMapCRDT/savePanelExploreState',
'exploreMapCRDT/updateMapTitle',
'exploreMapCRDT/duplicatePanel',
];
return broadcastableActions.includes(actionType);
}
/**
* Mock broadcaster for testing without WebSocket
*/
export class MockBroadcaster implements OperationBroadcaster {
private handlers: Array<(operation: CRDTOperation) => void> = [];
public broadcastedOperations: CRDTOperation[] = [];
broadcast(operation: CRDTOperation): void {
this.broadcastedOperations.push(operation);
}
broadcastBatch(operations: CRDTOperation[]): void {
this.broadcastedOperations.push(...operations);
}
subscribe(handler: (operation: CRDTOperation) => void): () => void {
this.handlers.push(handler);
// Return unsubscribe function
return () => {
const index = this.handlers.indexOf(handler);
if (index > -1) {
this.handlers.splice(index, 1);
}
};
}
/**
* Simulate receiving a remote operation
*/
simulateRemoteOperation(operation: CRDTOperation): void {
for (const handler of this.handlers) {
handler(operation);
}
}
/**
* Clear broadcasted operations history
*/
clear(): void {
this.broadcastedOperations = [];
}
}
/**
* Throttle helper for rate-limiting broadcasts
*/
export class ThrottledBroadcaster implements OperationBroadcaster {
private broadcaster: OperationBroadcaster;
private throttleMs: number;
private pendingBatch: CRDTOperation[] = [];
private timeoutId?: ReturnType<typeof setTimeout>;
constructor(broadcaster: OperationBroadcaster, throttleMs: number = 100) {
this.broadcaster = broadcaster;
this.throttleMs = throttleMs;
}
broadcast(operation: CRDTOperation): void {
this.pendingBatch.push(operation);
if (!this.timeoutId) {
this.timeoutId = setTimeout(() => {
this.flush();
}, this.throttleMs);
}
}
broadcastBatch(operations: CRDTOperation[]): void {
this.pendingBatch.push(...operations);
if (!this.timeoutId) {
this.timeoutId = setTimeout(() => {
this.flush();
}, this.throttleMs);
}
}
subscribe(handler: (operation: CRDTOperation) => void): () => void {
return this.broadcaster.subscribe(handler);
}
private flush(): void {
if (this.pendingBatch.length > 0) {
this.broadcaster.broadcastBatch([...this.pendingBatch]);
this.pendingBatch = [];
}
this.timeoutId = undefined;
}
/**
* Immediately flush any pending operations
*/
flushNow(): void {
if (this.timeoutId) {
clearTimeout(this.timeoutId);
this.timeoutId = undefined;
}
this.flush();
}
}
/**
* Buffered broadcaster for offline support
* Buffers operations when offline and replays when back online
*/
export class BufferedBroadcaster implements OperationBroadcaster {
private broadcaster: OperationBroadcaster;
private buffer: CRDTOperation[] = [];
private isOnline: boolean = true;
private maxBufferSize: number;
constructor(broadcaster: OperationBroadcaster, maxBufferSize: number = 1000) {
this.broadcaster = broadcaster;
this.maxBufferSize = maxBufferSize;
}
broadcast(operation: CRDTOperation): void {
if (this.isOnline) {
this.broadcaster.broadcast(operation);
} else {
this.bufferOperation(operation);
}
}
broadcastBatch(operations: CRDTOperation[]): void {
if (this.isOnline) {
this.broadcaster.broadcastBatch(operations);
} else {
operations.forEach((op) => this.bufferOperation(op));
}
}
subscribe(handler: (operation: CRDTOperation) => void): () => void {
return this.broadcaster.subscribe(handler);
}
private bufferOperation(operation: CRDTOperation): void {
this.buffer.push(operation);
// Limit buffer size
if (this.buffer.length > this.maxBufferSize) {
this.buffer.shift(); // Remove oldest
}
}
/**
* Set online status
* When going online, replays buffered operations
*/
setOnline(online: boolean): void {
const wasOffline = !this.isOnline;
this.isOnline = online;
if (online && wasOffline && this.buffer.length > 0) {
// Replay buffered operations
this.broadcaster.broadcastBatch([...this.buffer]);
this.buffer = [];
}
}
/**
* Get current buffer size
*/
getBufferSize(): number {
return this.buffer.length;
}
/**
* Clear buffer
*/
clearBuffer(): void {
this.buffer = [];
}
}
@@ -1,5 +1,7 @@
import { exploreMapReducer } from './exploreMapSlice';
import { crdtReducer } from './crdtSlice';
export default {
exploreMap: exploreMapReducer,
exploreMapCRDT: crdtReducer,
};
@@ -0,0 +1,258 @@
/**
* Redux selectors for CRDT-based Explore Map state
*
* These selectors convert CRDT state into UI-friendly formats
* for React components to consume.
*/
import { createSelector } from '@reduxjs/toolkit';
import { CRDTStateManager } from '../crdt/state';
import { ExploreMapCRDTState } from './crdtSlice';
import { ExploreMapPanel } from './types';
/**
* Get CRDT manager from state
*/
function getCRDTManager(state: ExploreMapCRDTState): CRDTStateManager {
if (!state.crdtStateJSON) {
return new CRDTStateManager(state.uid || '', state.nodeId);
}
const json = JSON.parse(state.crdtStateJSON);
return CRDTStateManager.fromJSON(json, state.nodeId);
}
/**
* Select all panels as a Record (for compatibility with existing UI)
*/
export const selectPanels = createSelector(
[(state: ExploreMapCRDTState) => state],
(state): Record<string, ExploreMapPanel> => {
const manager = getCRDTManager(state);
const panels: Record<string, ExploreMapPanel> = {};
for (const panelId of manager.getPanelIds()) {
const panelData = manager.getPanelForUI(panelId);
if (panelData) {
panels[panelId] = panelData as ExploreMapPanel;
}
}
return panels;
}
);
/**
* Select a single panel by ID
*/
export const selectPanel = createSelector(
[
(state: ExploreMapCRDTState) => state,
(_state: ExploreMapCRDTState, panelId: string) => panelId,
],
(state, panelId): ExploreMapPanel | undefined => {
const manager = getCRDTManager(state);
const panelData = manager.getPanelForUI(panelId);
return panelData ? (panelData as ExploreMapPanel) : undefined;
}
);
/**
* Select panel IDs
*/
export const selectPanelIds = createSelector(
[(state: ExploreMapCRDTState) => state],
(state): string[] => {
const manager = getCRDTManager(state);
return manager.getPanelIds();
}
);
/**
* Select map title
*/
export const selectMapTitle = createSelector(
[(state: ExploreMapCRDTState) => state],
(state): string => {
const manager = getCRDTManager(state);
const crdtState = manager.getState();
return crdtState.title.get();
}
);
/**
* Select map UID
*/
export const selectMapUid = (state: ExploreMapCRDTState): string | undefined => {
return state.uid;
};
/**
* Select viewport
*/
export const selectViewport = (state: ExploreMapCRDTState) => {
return state.local.viewport;
};
/**
* Select selected panel IDs
*/
export const selectSelectedPanelIds = (state: ExploreMapCRDTState): string[] => {
return state.local.selectedPanelIds;
};
/**
* Select cursors
*/
export const selectCursors = (state: ExploreMapCRDTState) => {
return state.local.cursors;
};
/**
* Select online status
*/
export const selectIsOnline = (state: ExploreMapCRDTState): boolean => {
return state.local.isOnline;
};
/**
* Select syncing status
*/
export const selectIsSyncing = (state: ExploreMapCRDTState): boolean => {
return state.local.isSyncing;
};
/**
* Select pending operations
*/
export const selectPendingOperations = (state: ExploreMapCRDTState) => {
return state.pendingOperations;
};
/**
* Select whether there are pending operations to broadcast
*/
export const selectHasPendingOperations = (state: ExploreMapCRDTState): boolean => {
return state.pendingOperations.length > 0;
};
/**
* Select node ID
*/
export const selectNodeId = (state: ExploreMapCRDTState): string => {
return state.nodeId;
};
/**
* Select panel count
*/
export const selectPanelCount = createSelector(
[(state: ExploreMapCRDTState) => state],
(state): number => {
const manager = getCRDTManager(state);
return manager.getPanelIds().length;
}
);
/**
* Select selected panels
*/
export const selectSelectedPanels = createSelector(
[selectPanels, selectSelectedPanelIds],
(panels, selectedIds): ExploreMapPanel[] => {
return selectedIds.map((id) => panels[id]).filter(Boolean);
}
);
/**
* Check if a panel is selected
*/
export const selectIsPanelSelected = createSelector(
[
selectSelectedPanelIds,
(_state: ExploreMapCRDTState, panelId: string) => panelId,
],
(selectedIds, panelId): boolean => {
return selectedIds.includes(panelId);
}
);
/**
* Get the highest z-index (for bringing panels to front)
*/
export const selectMaxZIndex = createSelector(
[selectPanels],
(panels): number => {
let max = 0;
for (const panel of Object.values(panels)) {
if (panel.position.zIndex > max) {
max = panel.position.zIndex;
}
}
return max;
}
);
/**
* Get bounding box of all selected panels
*/
export const selectSelectedPanelsBounds = createSelector(
[selectSelectedPanels],
(panels): { minX: number; minY: number; maxX: number; maxY: number } | null => {
if (panels.length === 0) {
return null;
}
let minX = Infinity;
let minY = Infinity;
let maxX = -Infinity;
let maxY = -Infinity;
for (const panel of panels) {
const { x, y, width, height } = panel.position;
minX = Math.min(minX, x);
minY = Math.min(minY, y);
maxX = Math.max(maxX, x + width);
maxY = Math.max(maxY, y + height);
}
return { minX, minY, maxX, maxY };
}
);
/**
* Select entire CRDT state as JSON (for persistence)
*/
export const selectCRDTStateJSON = (state: ExploreMapCRDTState): string | undefined => {
return state.crdtStateJSON;
};
/**
* Select the entire legacy-compatible state
* Useful for gradual migration
*/
export const selectLegacyState = createSelector(
[
selectPanels,
selectMapTitle,
selectViewport,
selectSelectedPanelIds,
selectCursors,
selectMapUid,
(state: ExploreMapCRDTState) => state,
],
(panels, title, viewport, selectedPanelIds, cursors, uid, state) => {
const manager = getCRDTManager(state);
const crdtState = manager.getState();
return {
uid,
title,
viewport,
panels,
selectedPanelIds,
nextZIndex: crdtState.zIndexCounter.value() + 1,
cursors,
};
}
);
+3
View File
@@ -7,6 +7,7 @@ import { allMiddleware as allApiClientMiddleware } from '@grafana/api-clients/rt
import { legacyAPI } from 'app/api/clients/legacy';
import { browseDashboardsAPI } from 'app/features/browse-dashboards/api/browseDashboardsAPI';
import { publicDashboardApi } from 'app/features/dashboard/api/publicDashboardApi';
import { createOperationMiddleware } from 'app/features/explore-map/state/middleware';
import { StoreState } from 'app/types/store';
import { buildInitialState } from '../core/reducers/navModel';
@@ -45,6 +46,8 @@ export function configureStore(initialState?: Partial<StoreState>) {
browseDashboardsAPI.middleware,
legacyAPI.middleware,
...allApiClientMiddleware,
// CRDT operation middleware for Explore Maps
createOperationMiddleware(),
...extraMiddleware
),
devTools: process.env.NODE_ENV !== 'production',