RFC-002: リアルタイムコラボレーション機能の実装¶

// OT Engine
type OTEngine struct {
    history []Operation
    version int
}

// Transform function: transforms operation A against operation B
func Transform(opA, opB Operation) (Operation, Operation) {
    // Example: Two users editing the same field

    // Case 1: Both replace the same path
    if opA.Op == "replace" && opB.Op == "replace" && opA.Path == opB.Path {
        // Priority: Last-Write-Wins (based on timestamp or user priority)
        // Or: Merge values (for text fields, use OT text algorithm)
        if opA.Timestamp > opB.Timestamp {
            return opA, Operation{Op: "noop"}
        } else {
            return Operation{Op: "noop"}, opB
        }
    }

    // Case 2: One adds, one removes at the same array index
    if opA.Op == "add" && opB.Op == "remove" {
        // Adjust indices
        opA.Path = adjustPathForRemove(opA.Path, opB.Path)
        return opA, opB
    }

    // Case 3: Independent paths - no conflict
    if !pathsConflict(opA.Path, opB.Path) {
        return opA, opB
    }

    // ... more cases

    return opA, opB
}

// Apply operation to document state
func (e *OTEngine) Apply(op Operation, state *Event) (*Event, error) {
    // Validate operation
    if err := e.validate(op); err != nil {
        return nil, err
    }

    // Apply JSON Patch
    newState, err := applyJSONPatch(state, op)
    if err != nil {
        return nil, err
    }

    // Add to history
    e.history = append(e.history, op)
    e.version++

    return newState, nil
}

// CRDT-based Event model
type CRDTEvent struct {
    ID          LWWID              // Last-Write-Wins ID
    Title       LWWRegister        // Last-Write-Wins Register
    StartTime   LWWRegister
    EndTime     LWWRegister
    Attendees   ORSet              // Observed-Remove Set
    Tags        GSet               // Grow-only Set
    Description RGA                // Replicated Growable Array (for rich text)
}

// LWW Register (Last-Write-Wins)
type LWWRegister struct {
    Value     interface{}
    Timestamp int64
    ClientID  string
}

func (r *LWWRegister) Set(value interface{}, timestamp int64, clientID string) {
    if timestamp > r.Timestamp ||
       (timestamp == r.Timestamp && clientID > r.ClientID) {
        r.Value = value
        r.Timestamp = timestamp
        r.ClientID = clientID
    }
}

// ORSet (Observed-Remove Set) for attendees
type ORSet struct {
    adds    map[string][]string  // element -> [unique tags]
    removes map[string][]string
}

func (s *ORSet) Add(element string, tag string) {
    s.adds[element] = append(s.adds[element], tag)
}

func (s *ORSet) Remove(element string) {
    // Observe all current tags and mark for removal
    s.removes[element] = append(s.removes[element], s.adds[element]...)
}

func (s *ORSet) Contains(element string) bool {
    addTags := s.adds[element]
    removeTags := s.removes[element]

    // Element exists if any add tag is not in remove tags
    for _, addTag := range addTags {
        removed := false
        for _, removeTag := range removeTags {
            if addTag == removeTag {
                removed = true
                break
            }
        }
        if !removed {
            return true
        }
    }
    return false
}

// Presence Manager
type PresenceManager struct {
    sessions map[string]*Session  // resource_id -> sessions
    redis    *redis.Client
}

func (pm *PresenceManager) JoinResource(resourceID, userID, clientID string) {
    presence := Presence{
        UserID:   userID,
        ClientID: clientID,
        JoinedAt: time.Now().Unix(),
    }

    // Add to Redis (with TTL)
    key := fmt.Sprintf("presence:%s", resourceID)
    pm.redis.HSet(ctx, key, clientID, marshalJSON(presence))
    pm.redis.Expire(ctx, key, 5*time.Minute)

    // Broadcast to other users
    pm.broadcast(resourceID, Message{
        Type: "PRESENCE_JOIN",
        Payload: presence,
    })
}

func (pm *PresenceManager) UpdatePresence(resourceID, clientID string, update PresenceUpdate) {
    // Update Redis
    key := fmt.Sprintf("presence:%s", resourceID)
    pm.redis.HSet(ctx, key, clientID, marshalJSON(update))

    // Broadcast
    pm.broadcast(resourceID, Message{
        Type: "PRESENCE_UPDATE",
        Payload: update,
    })
}

// Heartbeat to keep presence alive
func (pm *PresenceManager) Heartbeat(resourceID, clientID string) {
    key := fmt.Sprintf("presence:%s", resourceID)
    pm.redis.Expire(ctx, key, 5*time.Minute)
}

// Delta Sync: Only send changes since last sync
type SyncRequest struct {
    ResourceID  string
    LastVersion int
}

type SyncResponse struct {
    Operations []Operation  // All ops since last version
    CurrentVersion int
    State Event  // Full state (optional, for full resync)
}

func (s *SyncService) Sync(req SyncRequest) (*SyncResponse, error) {
    // Get operations since last version
    ops, err := s.getOperationsSince(req.ResourceID, req.LastVersion)
    if err != nil {
        return nil, err
    }

    // If too many operations, send full state instead
    if len(ops) > 100 {
        state, err := s.getFullState(req.ResourceID)
        if err != nil {
            return nil, err
        }
        return &SyncResponse{
            State: state,
            CurrentVersion: state.Version,
        }, nil
    }

    return &SyncResponse{
        Operations: ops,
        CurrentVersion: s.getCurrentVersion(req.ResourceID),
    }, nil
}

// Prometheus metrics
var (
    wsConnections = prometheus.NewGauge(
        prometheus.GaugeOpts{
            Name: "websocket_connections_total",
            Help: "Total number of active WebSocket connections",
        },
    )

    wsMessagesSent = prometheus.NewCounter(
        prometheus.CounterOpts{
            Name: "websocket_messages_sent_total",
            Help: "Total number of messages sent",
        },
    )

    wsLatency = prometheus.NewHistogram(
        prometheus.HistogramOpts{
            Name:    "websocket_message_latency_seconds",
            Help:    "Latency of WebSocket message delivery",
            Buckets: prometheus.DefBuckets,
        },
    )
)

func (ws *WebSocketServer) authenticate(msg AuthMessage) error {
    // Validate JWT token
    claims, err := jwt.Verify(msg.Token)
    if err != nil {
        return errors.New("invalid token")
    }

    // Check token expiration
    if claims.ExpiresAt < time.Now().Unix() {
        return errors.New("token expired")
    }

    // Verify user permissions
    if !hasPermission(claims.UserID, msg.ResourceID) {
        return errors.New("unauthorized")
    }

    return nil
}

// Rate limiter per user
type RateLimiter struct {
    limiters map[string]*rate.Limiter
}

func (rl *RateLimiter) Allow(userID string) bool {
    limiter, exists := rl.limiters[userID]
    if !exists {
        // 100 operations per minute per user
        limiter = rate.NewLimiter(rate.Every(time.Minute/100), 10)
        rl.limiters[userID] = limiter
    }

    return limiter.Allow()
}

func validateOperation(op Operation) error {
    // Validate op type
    validOps := []string{"add", "remove", "replace", "move", "copy", "test"}
    if !contains(validOps, op.Op) {
        return errors.New("invalid operation type")
    }

    // Validate path (prevent path traversal)
    if !isValidJSONPointer(op.Path) {
        return errors.New("invalid JSON pointer")
    }

    // Sanitize value (XSS prevention)
    if op.Value != nil {
        op.Value = sanitize(op.Value)
    }

    // Validate against schema
    if err := validateSchema(op); err != nil {
        return err
    }

    return nil
}

func TestOperationalTransformation(t *testing.T) {
    tests := []struct {
        name string
        opA  Operation
        opB  Operation
        want []Operation
    }{
        {
            name: "concurrent title edits",
            opA:  Operation{Op: "replace", Path: "/title", Value: "Meeting A"},
            opB:  Operation{Op: "replace", Path: "/title", Value: "Meeting B"},
            want: []Operation{
                {Op: "replace", Path: "/title", Value: "Meeting B"},  // B wins
                {Op: "noop"},
            },
        },
        {
            name: "add and remove attendee",
            opA:  Operation{Op: "add", Path: "/attendees/-", Value: user1},
            opB:  Operation{Op: "remove", Path: "/attendees/0"},
            want: []Operation{
                {Op: "add", Path: "/attendees/-", Value: user1},
                {Op: "remove", Path: "/attendees/0"},
            },
        },
        // ... more test cases
    }

    for _, tt := range tests {
        t.Run(tt.name, func(t *testing.T) {
            got := Transform(tt.opA, tt.opB)
            assert.Equal(t, tt.want, got)
        })
    }
}