navidrome/plugins/host_websocket.go

package plugins

import (
	"context"
	"errors"
	"fmt"
	"maps"
	"net"
	"net/http"
	"net/url"
	"strings"
	"sync"
	"syscall"
	"time"

	"github.com/gorilla/websocket"
	"github.com/navidrome/navidrome/log"
	"github.com/navidrome/navidrome/model/id"
	"github.com/navidrome/navidrome/plugins/capabilities"
	"github.com/navidrome/navidrome/plugins/host"
)

// CapabilityWebSocket indicates the plugin can receive WebSocket callbacks.
// Detected when the plugin exports any of the WebSocket callback functions.
const CapabilityWebSocket Capability = "WebSocket"

// webSocketCallbackTimeout is the maximum duration allowed for a WebSocket callback.
const webSocketCallbackTimeout = 30 * time.Second

// WebSocket callback function names
const (
	FuncWebSocketOnTextMessage   = "nd_websocket_on_text_message"
	FuncWebSocketOnBinaryMessage = "nd_websocket_on_binary_message"
	FuncWebSocketOnError         = "nd_websocket_on_error"
	FuncWebSocketOnClose         = "nd_websocket_on_close"
)

func init() {
	registerCapability(
		CapabilityWebSocket,
		FuncWebSocketOnTextMessage,
		FuncWebSocketOnBinaryMessage,
		FuncWebSocketOnError,
		FuncWebSocketOnClose,
	)
}

// wsConnection represents an active WebSocket connection.
type wsConnection struct {
	conn     *websocket.Conn
	done     chan struct{}
	closeMu  sync.Mutex
	isClosed bool
}

// webSocketServiceImpl implements host.WebSocketService.
// It provides plugins with WebSocket communication capabilities.
type webSocketServiceImpl struct {
	baseCtx       context.Context //nolint:containedctx // bounds the read loops, which outlive the Connect() call
	pluginName    string
	manager       *Manager
	requiredHosts []string

	mu          sync.RWMutex
	connections map[string]*wsConnection
}

// newWebSocketService creates a new WebSocketService for a plugin.
func newWebSocketService(ctx context.Context, pluginName string, manager *Manager, permission *WebSocketPermission) *webSocketServiceImpl {
	return &webSocketServiceImpl{
		baseCtx:       ctx,
		pluginName:    pluginName,
		manager:       manager,
		requiredHosts: permission.RequiredHosts,
		connections:   make(map[string]*wsConnection),
	}
}

func (s *webSocketServiceImpl) Connect(ctx context.Context, urlStr string, headers map[string]string, connectionID string) (string, error) {
	// Parse and validate URL
	parsedURL, err := url.Parse(urlStr)
	if err != nil {
		return "", fmt.Errorf("invalid URL: %w", err)
	}

	// Validate scheme
	if parsedURL.Scheme != "ws" && parsedURL.Scheme != "wss" {
		return "", fmt.Errorf("invalid URL scheme: must be ws:// or wss://")
	}

	// Validate host against allowed hosts
	if !s.isHostAllowed(parsedURL.Host) {
		return "", fmt.Errorf("host %q is not allowed", parsedURL.Host)
	}

	// Generate connection ID if not provided
	if connectionID == "" {
		connectionID = id.NewRandom()
	}

	s.mu.Lock()
	if _, exists := s.connections[connectionID]; exists {
		s.mu.Unlock()
		return "", fmt.Errorf("connection ID %q already exists", connectionID)
	}
	s.mu.Unlock()

	// Create HTTP headers for handshake
	httpHeaders := http.Header{}
	for k, v := range headers {
		httpHeaders.Set(k, v)
	}

	// Establish WebSocket connection
	dialer := websocket.Dialer{
		HandshakeTimeout: 30 * time.Second,
		NetDialContext:   (&net.Dialer{Control: s.dialControl, Resolver: dialResolver}).DialContext,
	}

	conn, resp, err := dialer.DialContext(ctx, urlStr, httpHeaders)
	if resp != nil && resp.Body != nil {
		_ = resp.Body.Close()
	}
	if err != nil {
		return "", fmt.Errorf("failed to connect: %w", err)
	}

	wsConn := &wsConnection{
		conn: conn,
		done: make(chan struct{}),
	}

	s.mu.Lock()
	s.connections[connectionID] = wsConn
	s.mu.Unlock()

	// Start read goroutine with the service's base context instead of the
	// caller's ctx, because the readLoop must outlive the Connect() call.
	// Connections are closed by Close() when the plugin is unloaded, which ends
	// the readLoop; the base context is a backstop that also ends it on server
	// shutdown (it is never cancelled in one-shot CLI runs).
	go s.readLoop(s.baseCtx, connectionID, wsConn)

	log.Debug(ctx, "WebSocket connected", "plugin", s.pluginName, "connectionID", connectionID, "url", urlStr)
	return connectionID, nil
}

func (s *webSocketServiceImpl) SendText(ctx context.Context, connectionID, message string) error {
	wsConn, err := s.getConnection(connectionID)
	if err != nil {
		return err
	}

	if err := wsConn.conn.WriteMessage(websocket.TextMessage, []byte(message)); err != nil {
		return fmt.Errorf("failed to send text message: %w", err)
	}

	return nil
}

func (s *webSocketServiceImpl) SendBinary(ctx context.Context, connectionID string, data []byte) error {
	wsConn, err := s.getConnection(connectionID)
	if err != nil {
		return err
	}

	if err := wsConn.conn.WriteMessage(websocket.BinaryMessage, data); err != nil {
		return fmt.Errorf("failed to send binary message: %w", err)
	}

	return nil
}

func (s *webSocketServiceImpl) CloseConnection(ctx context.Context, connectionID string, code int32, reason string) error {
	s.mu.Lock()
	wsConn, exists := s.connections[connectionID]
	if !exists {
		s.mu.Unlock()
		return fmt.Errorf("connection ID %q not found", connectionID)
	}
	delete(s.connections, connectionID)
	s.mu.Unlock()

	// Mark as closed to prevent callback
	wsConn.closeMu.Lock()
	wsConn.isClosed = true
	wsConn.closeMu.Unlock()

	// Send close message
	closeMsg := websocket.FormatCloseMessage(int(code), reason)
	_ = wsConn.conn.WriteControl(websocket.CloseMessage, closeMsg, time.Now().Add(5*time.Second))
	_ = wsConn.conn.Close()

	// Signal read goroutine to stop
	close(wsConn.done)

	// Invoke close callback
	s.invokeOnClose(ctx, connectionID, code, reason)

	log.Debug(ctx, "WebSocket connection closed", "plugin", s.pluginName, "connectionID", connectionID, "code", code)
	return nil
}

// Close closes all connections for this plugin.
// This is called when the plugin is unloaded.
func (s *webSocketServiceImpl) Close() error {
	s.mu.Lock()
	connections := make(map[string]*wsConnection, len(s.connections))
	maps.Copy(connections, s.connections)
	s.connections = make(map[string]*wsConnection)
	s.mu.Unlock()

	ctx := context.Background()
	for connID, wsConn := range connections {
		wsConn.closeMu.Lock()
		wsConn.isClosed = true
		wsConn.closeMu.Unlock()

		closeMsg := websocket.FormatCloseMessage(websocket.CloseGoingAway, "plugin unloaded")
		err := wsConn.conn.WriteControl(websocket.CloseMessage, closeMsg, time.Now().Add(2*time.Second))
		if err != nil {
			log.Warn("Failed to send WebSocket close message on plugin unload", "plugin", s.pluginName, "connectionID", connID, "error", err)
		}
		err = wsConn.conn.Close()
		if err != nil {
			log.Warn("Failed to close WebSocket connection on plugin unload", "plugin", s.pluginName, "connectionID", connID, "error", err)
		}
		close(wsConn.done)

		s.invokeOnClose(ctx, connID, websocket.CloseGoingAway, "plugin unloaded")
		log.Debug("WebSocket connection closed on plugin unload", "plugin", s.pluginName, "connectionID", connID)
	}

	return nil
}

func (s *webSocketServiceImpl) getConnection(connectionID string) (*wsConnection, error) {
	s.mu.RLock()
	defer s.mu.RUnlock()

	wsConn, exists := s.connections[connectionID]
	if !exists {
		return nil, fmt.Errorf("connection ID %q not found", connectionID)
	}
	return wsConn, nil
}

func (s *webSocketServiceImpl) isHostAllowed(host string) bool {
	return isHostInAllowlist(s.requiredHosts, extractHostname(host))
}

func (s *webSocketServiceImpl) dialControl(_, address string, _ syscall.RawConn) error {
	return checkPrivateDial(s.requiredHosts, address)
}

// matchHostPattern matches a host against a pattern.
// Supports "*" (allow all) and wildcards like "*.example.com".
func matchHostPattern(pattern, host string) bool {
	if pattern == "*" {
		return true
	}
	if pattern == host {
		return true
	}

	// Handle wildcard patterns like *.example.com
	if strings.HasPrefix(pattern, "*.") {
		suffix := pattern[1:] // Get .example.com
		return strings.HasSuffix(host, suffix)
	}

	return false
}

func (s *webSocketServiceImpl) readLoop(ctx context.Context, connectionID string, wsConn *wsConnection) {
	defer func() {
		// Remove connection if still present
		s.mu.Lock()
		delete(s.connections, connectionID)
		s.mu.Unlock()
	}()

	for {
		select {
		case <-wsConn.done:
			return
		default:
		}

		messageType, data, err := wsConn.conn.ReadMessage()
		if err != nil {
			wsConn.closeMu.Lock()
			isClosed := wsConn.isClosed
			wsConn.closeMu.Unlock()

			if isClosed {
				return
			}

			// Check if it's a close error
			if websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseGoingAway, websocket.CloseNoStatusReceived) {
				closeCode := websocket.CloseNoStatusReceived
				closeReason := ""
				if ce, ok := errors.AsType[*websocket.CloseError](err); ok {
					closeCode = ce.Code
					closeReason = ce.Text
				}
				s.invokeOnClose(ctx, connectionID, int32(closeCode), closeReason)
				return
			}

			// Other read error
			s.invokeOnError(ctx, connectionID, err.Error())
			return
		}

		switch messageType {
		case websocket.TextMessage:
			s.invokeOnTextMessage(ctx, connectionID, string(data))
		case websocket.BinaryMessage:
			s.invokeOnBinaryMessage(ctx, connectionID, data)
		}
	}
}

// invokeWebSocketCallback is a generic helper that handles the common callback invocation pattern.
func invokeWebSocketCallback[I any](ctx context.Context, s *webSocketServiceImpl, funcName string, input I, callbackName string, connectionID string) {
	instance := s.getPluginInstance()
	if instance == nil {
		return
	}

	callbackCtx, cancel := context.WithTimeout(ctx, webSocketCallbackTimeout)
	defer cancel()

	start := time.Now()
	err := callPluginFunctionNoOutput(callbackCtx, instance, funcName, input)
	if err != nil {
		if !errors.Is(errFunctionNotFound, err) {
			log.Error(ctx, "WebSocket "+callbackName+" callback failed", "plugin", s.pluginName, "connectionID", connectionID, "duration", time.Since(start), err)
		}
	}
}

func (s *webSocketServiceImpl) invokeOnTextMessage(ctx context.Context, connectionID, message string) {
	invokeWebSocketCallback(ctx, s, FuncWebSocketOnTextMessage, capabilities.OnTextMessageRequest{
		ConnectionID: connectionID,
		Message:      message,
	}, "text message", connectionID)
}

func (s *webSocketServiceImpl) invokeOnBinaryMessage(ctx context.Context, connectionID string, data []byte) {
	invokeWebSocketCallback(ctx, s, FuncWebSocketOnBinaryMessage, capabilities.OnBinaryMessageRequest{
		ConnectionID: connectionID,
		Data:         data,
	}, "binary message", connectionID)
}

func (s *webSocketServiceImpl) invokeOnError(ctx context.Context, connectionID, errorMsg string) {
	invokeWebSocketCallback(ctx, s, FuncWebSocketOnError, capabilities.OnErrorRequest{
		ConnectionID: connectionID,
		Error:        errorMsg,
	}, "error", connectionID)
}

func (s *webSocketServiceImpl) invokeOnClose(ctx context.Context, connectionID string, code int32, reason string) {
	invokeWebSocketCallback(ctx, s, FuncWebSocketOnClose, capabilities.OnCloseRequest{
		ConnectionID: connectionID,
		Code:         code,
		Reason:       reason,
	}, "close", connectionID)
}

func (s *webSocketServiceImpl) getPluginInstance() *plugin {
	s.manager.mu.RLock()
	instance, ok := s.manager.plugins[s.pluginName]
	s.manager.mu.RUnlock()

	if !ok {
		log.Warn("Plugin not loaded for WebSocket callback", "plugin", s.pluginName)
		return nil
	}

	return instance
}

// Verify interface implementation
var _ host.WebSocketService = (*webSocketServiceImpl)(nil)