bookhoard/internal/sync/websocket.go

package sync

import (
	"context"
	"log"
	"sync"
	"time"

	"github.com/google/uuid"
)

// Message types for WebSocket communication
const (
	MessageTypeProgressUpdate   = "progress_update"
	MessageTypeAnnotationUpdate = "annotation_update"
	MessageTypeConflict         = "conflict"
	MessageTypeSyncComplete     = "sync_complete"
	MessageTypeHeartbeat        = "heartbeat"
	MessageTypeInitial          = "initial_state"
	MessageTypeScanProgress     = "scan_progress"
	MessageTypeScanComplete     = "scan_complete"
	MessageTypeScanError        = "scan_error"
	MessageTypeScanCancelled    = "scan_cancelled"
)

// BroadcastMessage represents a message to broadcast to connected clients
type BroadcastMessage struct {
	Type      string        `json:"type"`
	Timestamp string        `json:"timestamp"`
	Data      interface{}   `json:"data"`
	Source    *SourceDevice `json:"source_device,omitempty"`
}

// SourceDevice represents the device that sent the update
type SourceDevice struct {
	ID   string `json:"id"`
	Name string `json:"name"`
	Type string `json:"type"`
}

// DeviceConnection represents an active WebSocket connection
type DeviceConnection struct {
	ID           string
	UserID       string
	DeviceType   string
	DeviceName   string
	Connected    time.Time
	LastPing     time.Time
	Send         chan BroadcastMessage
	Disconnected chan struct{}
}

// ConnectionManager manages WebSocket connections
//
// CHOKEPOINT (2026-11-02): there is deliberately NO untargeted broadcast
// method. Every send must name its audience — the user whose data the
// event carries. The pre-fix Broadcast fanned progress, annotation, and
// conflict events out to every connected client on the instance,
// crossing user boundaries (a family-lens violation: hidden-library
// book progress and annotation payloads reached unrelated clients).
// The compiler now enforces what review used to.
type ConnectionManager struct {
	connections map[string]*DeviceConnection
	mu          sync.RWMutex
}

// NewConnectionManager creates a new connection manager
func NewConnectionManager() *ConnectionManager {
	return &ConnectionManager{
		connections: make(map[string]*DeviceConnection),
	}
}

// BroadcastToAll delivers an event to every connected client on the
// instance. This is the ONE sanctioned instance-wide fan-out, and using
// it is a decision: the payload must contain no user-derived data. Its
// only caller is the scan-event emitter — scan activity is instance
// activity (job ids, file counts, progress percentages), and the
// sitewide scan indicator in web/src/header.ts is built on receiving it
// in every browser. Anything that carries a user's books, notes, or
// annotations belongs in BroadcastToUser.
func (m *ConnectionManager) BroadcastToAll(msg BroadcastMessage) {
	m.mu.RLock()
	defer m.mu.RUnlock()

	for _, conn := range m.connections {
		select {
		case conn.Send <- msg:
		default:
			// Channel full, skip this connection
			log.Printf("WebSocket: Channel full for %s, skipping broadcast", conn.DeviceName)
		}
	}
}

// BroadcastToUser sends a message to all connections for a specific user
func (m *ConnectionManager) BroadcastToUser(userID string, msg BroadcastMessage) {
	m.mu.RLock()
	defer m.mu.RUnlock()

	for _, conn := range m.connections {
		if conn.UserID == userID {
			select {
			case conn.Send <- msg:
			default:
				// Channel full, skip this connection
				log.Printf("WebSocket: Channel full for %s, skipping broadcast", conn.DeviceName)
			}
		}
	}
}

// BroadcastProgressUpdate broadcasts a progress update to the acting
// user's connections. Progress is per-user data — nobody else's client
// receives it.
func (m *ConnectionManager) BroadcastProgressUpdate(userID string, bookID uuid.UUID, percentage float64, source SourceDevice) {
	msg := BroadcastMessage{
		Type:      MessageTypeProgressUpdate,
		Timestamp: time.Now().Format(time.RFC3339),
		Data: map[string]interface{}{
			"book_id":    bookID.String(),
			"percentage": percentage,
		},
		Source: &source,
	}
	m.BroadcastToUser(userID, msg)
}

// BroadcastAnnotationUpdate broadcasts an annotation update to the acting
// user's connections. Annotation payloads can carry note and highlight
// content — strictly per-user.
func (m *ConnectionManager) BroadcastAnnotationUpdate(userID string, bookID uuid.UUID, annotationType string, data interface{}, source SourceDevice) {
	msg := BroadcastMessage{
		Type:      MessageTypeAnnotationUpdate,
		Timestamp: time.Now().Format(time.RFC3339),
		Data: map[string]interface{}{
			"book_id":         bookID.String(),
			"annotation_type": annotationType,
			"data":            data,
		},
		Source: &source,
	}
	m.BroadcastToUser(userID, msg)
}

// BroadcastConflictNotification broadcasts a conflict notification to the
// acting user's connections.
func (m *ConnectionManager) BroadcastConflictNotification(userID string, bookID [16]byte, notificationType string, conflictID string) {
	msg := BroadcastMessage{
		Type:      MessageTypeConflict,
		Timestamp: time.Now().Format(time.RFC3339),
		Data: map[string]interface{}{
			"book_id":           uuid.UUID(bookID).String(),
			"notification_type": notificationType,
			"conflict_id":       conflictID,
		},
	}
	m.BroadcastToUser(userID, msg)
}

// AddConnection adds a new WebSocket connection
func (m *ConnectionManager) AddConnection(conn *DeviceConnection) {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.connections[conn.ID] = conn
	log.Printf("WebSocket: Connection added for device %s (user: %s, type: %s)", conn.DeviceName, conn.UserID, conn.DeviceType)
}

// RemoveConnection removes a WebSocket connection
func (m *ConnectionManager) RemoveConnection(connID string) {
	m.mu.Lock()
	defer m.mu.Unlock()
	if conn, exists := m.connections[connID]; exists {
		close(conn.Disconnected)
		delete(m.connections, connID)
		log.Printf("WebSocket: Connection removed for device %s", conn.DeviceName)
	}
}

// GetConnection retrieves a connection by ID
func (m *ConnectionManager) GetConnection(connID string) (*DeviceConnection, bool) {
	m.mu.RLock()
	defer m.mu.RUnlock()
	conn, exists := m.connections[connID]
	return conn, exists
}

// GetUserConnections returns all connections for a specific user
func (m *ConnectionManager) GetUserConnections(userID string) []*DeviceConnection {
	m.mu.RLock()
	defer m.mu.RUnlock()

	var conns []*DeviceConnection
	for _, conn := range m.connections {
		if conn.UserID == userID {
			conns = append(conns, conn)
		}
	}
	return conns
}

// GetConnectionCount returns the number of active connections
func (m *ConnectionManager) GetConnectionCount() int {
	m.mu.RLock()
	defer m.mu.RUnlock()
	return len(m.connections)
}

// CleanupStaleConnections removes connections that haven't sent a ping in 2 minutes
func (m *ConnectionManager) CleanupStaleConnections() {
	m.mu.Lock()
	defer m.mu.Unlock()

	staleThreshold := time.Now().Add(-2 * time.Minute)
	for id, conn := range m.connections {
		if conn.LastPing.Before(staleThreshold) {
			log.Printf("WebSocket: Cleaning up stale connection for %s", conn.DeviceName)
			close(conn.Disconnected)
			delete(m.connections, id)
		}
	}
}

// GetConnectionStats returns statistics about connections
func (m *ConnectionManager) GetConnectionStats() map[string]int {
	m.mu.RLock()
	defer m.mu.RUnlock()

	stats := make(map[string]int)
	for _, conn := range m.connections {
		stats[conn.DeviceType]++
	}
	return stats
}

// StartCleanupTask starts a background task to cleanup stale connections
func (m *ConnectionManager) StartCleanupTask() context.CancelFunc {
	ticker := time.NewTicker(1 * time.Minute)
	ctx, cancel := context.WithCancel(context.Background())

	go func() {
		for {
			select {
			case <-ctx.Done():
				ticker.Stop()
				return
			case <-ticker.C:
				m.CleanupStaleConnections()
			}
		}
	}()

	return cancel
}