-rpi-sendspin/third_party/sendspin-go/pkg/sendspin/server.go

// ABOUTME: High-level Server API for Sendspin streaming
// ABOUTME: Wraps server components into a simple, user-friendly interface
package sendspin

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"net"
	"net/http"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/Sendspin/sendspin-go/internal/discovery"
	"github.com/Sendspin/sendspin-go/internal/server"
	"github.com/Sendspin/sendspin-go/pkg/protocol"
	"github.com/google/uuid"
	"github.com/gorilla/websocket"
)

const (
	ProtocolVersion = 1

	// Binary message type IDs per spec (bits 7-2 for role, bits 1-0 for slot)
	// Player role: 000001xx (4-7), slot 0 = 4
	AudioChunkMessageType = 4

	DefaultSampleRate = 192000
	DefaultChannels   = 2
	DefaultBitDepth   = 24

	ChunkDurationMs = 20  // 20ms chunks
	BufferAheadMs   = 500 // Send audio 500ms ahead
)

type ServerConfig struct {
	// Port to listen on (default: 8927)
	Port int

	Name string

	// Audio source to stream (required)
	Source AudioSource

	// EnableMDNS enables mDNS service advertisement (default: true)
	EnableMDNS bool

	Debug bool

	// DiscoverClients enables server-initiated discovery: browse for
	// clients advertising _sendspin._tcp and dial out to them.
	// See https://www.sendspin-audio.com/spec/ — "server-initiated" mode.
	DiscoverClients bool

	// SupportedRoles lists the role families this server activates.
	// When nil, defaults to ["player", "metadata"] for backward compat.
	// Roles registered via Group.RegisterRole are also activated
	// regardless of this list.
	SupportedRoles []string

	// ClientFilter, if non-nil, is invoked after a client's hello has
	// been parsed and before the server/hello reply is sent. Returning
	// false causes the server to close the connection without joining
	// the client to the default group. Used by multi-preset servers to
	// admit only the speakers belonging to the currently active preset.
	ClientFilter func(clientID string) bool

	// OnClientHello, if non-nil, is invoked for every parsed
	// client/hello, regardless of whether ClientFilter accepts the
	// connection. Lets the host application maintain a roster of
	// observed (clientID, name) pairs — useful when filtering rejects
	// most connections but the operator still wants to discover which
	// speakers exist on the network and what their stable IDs are.
	OnClientHello func(clientID, name string)
}

type Server struct {
	config   ServerConfig
	serverID string

	upgrader websocket.Upgrader

	httpServer *http.Server
	mux        *http.ServeMux
	boundAddr  atomic.Pointer[net.TCPAddr]

	clients      map[string]*ServerClient
	clientsMu    sync.RWMutex
	defaultGroup *Group

	clockStart time.Time // monotonic microseconds origin

	audioSource         AudioSource
	consecutiveReadErrs int

	// Accumulates bytes-on-wire across one streamAudio rate window
	// (encoded chunk size, not raw PCM) so the periodic stats line can
	// report both chunks/sec and kbps. Touched only from the streamAudio
	// goroutine so no locking needed.
	lastWindowBytes int64

	mdnsManager *discovery.Manager

	// server-initiated discovery dialer cancel
	dialerCancel context.CancelFunc

	stopChan   chan struct{}
	stopOnce   sync.Once
	shutdownMu sync.RWMutex
	isShutdown bool
	wg         sync.WaitGroup
}

type ClientInfo struct {
	ID     string
	Name   string
	State  string
	Volume int
	Muted  bool
	Codec  string
}

func NewServer(config ServerConfig) (*Server, error) {
	// Port == 0 is honored as "let the OS pick an ephemeral port". The
	// CLI / config-file layers supply 8927 as the documented default
	// before reaching this point; library callers that omit Port get
	// ephemeral binding (and can read it back via Server.Addr after
	// Start). This is the substitution-free behavior tests rely on for
	// flake-free Port: 0 binding.
	if config.Name == "" {
		config.Name = "Sendspin Server"
	}
	if config.Source == nil {
		return nil, fmt.Errorf("audio source is required")
	}
	if config.SupportedRoles == nil {
		config.SupportedRoles = []string{"player", "metadata"}
	}

	mux := http.NewServeMux()

	s := &Server{
		config:      config,
		serverID:    uuid.New().String(),
		mux:         mux,
		audioSource: config.Source,
		upgrader: websocket.Upgrader{
			CheckOrigin: func(r *http.Request) bool {
				// TODO: For production, implement proper origin validation.
				// Currently permissive for local-network deployments.
				return true
			},
		},
		clients:    make(map[string]*ServerClient),
		clockStart: time.Now(),
		stopChan:   make(chan struct{}),
	}

	s.defaultGroup = NewGroup(s.serverID)

	s.defaultGroup.RegisterRole(NewMetadataRole(MetadataConfig{
		GetMetadata: func() (string, string, string) {
			return s.audioSource.Metadata()
		},
		ClockMicros: s.getClockMicros,
	}))

	s.defaultGroup.RegisterRole(NewPlayerRole(PlayerRoleConfig{
		SampleRate: s.audioSource.SampleRate(),
		Channels:   s.audioSource.Channels(),
		BitDepth:   DefaultBitDepth,
		NewEncoder: func(sampleRate, channels, chunkSamples int) (*server.OpusEncoder, error) {
			return server.NewOpusEncoder(sampleRate, channels, chunkSamples)
		},
		NewFLACEncoder: func(sampleRate, channels, bitDepth, blockSize int) (*server.FLACEncoder, error) {
			return server.NewFLACEncoder(sampleRate, channels, bitDepth, blockSize)
		},
	}))

	return s, nil
}

func (s *Server) Start() error {
	log.Printf("Server starting: %s (ID: %s)", s.config.Name, s.serverID)
	log.Printf("Audio source: %dHz/%dbit/%dch",
		s.audioSource.SampleRate(),
		DefaultBitDepth,
		s.audioSource.Channels())

	if s.config.EnableMDNS {
		s.mdnsManager = discovery.NewManager(discovery.Config{
			ServiceName: s.config.Name,
			Port:        s.config.Port,
			ServerMode:  true,
		})

		if err := s.mdnsManager.Advertise(); err != nil {
			log.Printf("Failed to start mDNS advertisement: %v", err)
		} else {
			log.Printf("mDNS advertisement started")
		}
	}

	if s.config.DiscoverClients {
		if s.mdnsManager == nil {
			// If mDNS isn't running for advertising, start a manager just for browsing.
			s.mdnsManager = discovery.NewManager(discovery.Config{
				ServiceName: s.config.Name,
				Port:        s.config.Port,
				ServerMode:  true,
			})
		}

		if err := s.mdnsManager.BrowseClients(); err != nil {
			log.Printf("Failed to start client discovery: %v", err)
		} else {
			log.Printf("Browsing for clients advertising _sendspin._tcp")

			dialCtx, cancel := context.WithCancel(context.Background())
			s.dialerCancel = cancel

			dialer := newClientDialer(s.mdnsManager.Clients(), func(ctx context.Context, info *discovery.ClientInfo) error {
				return dialAndHandle(ctx, info, s.handleConnection)
			})

			s.wg.Add(1)
			go func() {
				defer s.wg.Done()
				dialer.run(dialCtx)
			}()
		}
	}

	s.mux.HandleFunc("/sendspin", s.handleWebSocket)

	s.wg.Add(1)
	go func() {
		defer s.wg.Done()
		s.streamAudio()
	}()

	addr := fmt.Sprintf(":%d", s.config.Port)

	listener, err := net.Listen("tcp", addr)
	if err != nil {
		return fmt.Errorf("listen on %s: %w", addr, err)
	}
	tcpAddr, _ := listener.Addr().(*net.TCPAddr)
	s.boundAddr.Store(tcpAddr)
	log.Printf("WebSocket server listening on %s", listener.Addr())

	s.httpServer = &http.Server{
		Handler: s.mux,
	}

	errChan := make(chan error, 1)
	go func() {
		if err := s.httpServer.Serve(listener); err != http.ErrServerClosed {
			errChan <- err
		}
	}()

	select {
	case <-s.stopChan:
		log.Printf("Server shutting down...")
	case err := <-errChan:
		log.Printf("HTTP server error: %v", err)
		return err
	}

	s.shutdownMu.Lock()
	s.isShutdown = true
	s.shutdownMu.Unlock()

	// Cancel in-flight client dials before stopping mDNS so they observe
	// context cancellation ahead of the discovery channel closing.
	if s.dialerCancel != nil {
		s.dialerCancel()
	}

	if s.mdnsManager != nil {
		s.mdnsManager.Stop()
	}

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	if err := s.httpServer.Shutdown(ctx); err != nil {
		log.Printf("HTTP server shutdown error: %v", err)
	}

	if err := s.audioSource.Close(); err != nil {
		log.Printf("Error closing audio source: %v", err)
	}

	if s.defaultGroup != nil {
		s.defaultGroup.Close()
	}

	s.wg.Wait()
	log.Printf("Server stopped cleanly")

	return nil
}

func (s *Server) Stop() {
	s.stopOnce.Do(func() {
		close(s.stopChan)
	})
}

// getClockMicros returns server uptime in microseconds (monotonic, not wall clock).
func (s *Server) getClockMicros() int64 {
	return time.Since(s.clockStart).Microseconds()
}

// Group returns the server's default playback group. For M2 there is
// exactly one implicit group per Server; the accessor exists so future
// GroupRole implementations (M3) can subscribe to its event bus.
func (s *Server) Group() *Group {
	return s.defaultGroup
}

// Addr returns the network address the server is listening on, or nil if
// Start has not yet bound a listener. Useful in tests that configure
// Port: 0 (OS-assigned ephemeral port) and need to know the actual port
// after Start runs.
func (s *Server) Addr() net.Addr {
	if a := s.boundAddr.Load(); a != nil {
		return a
	}
	return nil
}

// DisconnectClient closes the WebSocket for the given client_id, causing
// the per-client read loop to exit and standard teardown to run. Returns
// true if a matching client was found.
func (s *Server) DisconnectClient(id string) bool {
	s.clientsMu.RLock()
	c, ok := s.clients[id]
	s.clientsMu.RUnlock()
	if !ok {
		return false
	}
	// Closing the underlying connection unblocks the conn.ReadMessage
	// loop in handleConnection, which then runs removeClient via defer.
	_ = c.conn.Close()
	return true
}

// NotifyStreamStartAll re-runs the PlayerGroupRole join hook for every
// connected client. Use after Resume() to send a fresh stream/start
// (and recreate encoders) when transitioning paused → playing.
func (s *Server) NotifyStreamStartAll() {
	if s.defaultGroup == nil {
		return
	}
	s.clientsMu.RLock()
	clients := make([]*ServerClient, 0, len(s.clients))
	for _, c := range s.clients {
		clients = append(clients, c)
	}
	s.clientsMu.RUnlock()

	s.defaultGroup.mu.RLock()
	roles := make([]GroupRole, 0, len(s.defaultGroup.roles))
	for _, r := range s.defaultGroup.roles {
		roles = append(roles, r)
	}
	s.defaultGroup.mu.RUnlock()

	for _, c := range clients {
		for _, r := range roles {
			r.OnClientJoin(c)
		}
	}
}

// NotifyStreamEndAll sends stream/end to every connected player client.
// Use when transitioning playing → paused so clients flush buffers and
// stop emitting audio while the WebSocket stays open.
func (s *Server) NotifyStreamEndAll() {
	s.notifyStreamEnd()
}

func (s *Server) Clients() []ClientInfo {
	s.clientsMu.RLock()
	defer s.clientsMu.RUnlock()

	clients := make([]ClientInfo, 0, len(s.clients))
	for _, c := range s.clients {
		clients = append(clients, ClientInfo{
			ID:     c.ID(),
			Name:   c.Name(),
			State:  c.State(),
			Volume: c.Volume(),
			Muted:  c.Muted(),
			Codec:  c.Codec(),
		})
	}

	return clients
}

// ErrClientFiltered is returned by handleConnection when ClientFilter
// rejected the client_id from client/hello. The server-initiated dialer
// uses this sentinel to distinguish "filter rejection — try again
// later" from a successfully-handled session that should be latched.
var ErrClientFiltered = fmt.Errorf("client rejected by ClientFilter")

func (s *Server) handleWebSocket(w http.ResponseWriter, r *http.Request) {
	conn, err := s.upgrader.Upgrade(w, r, nil)
	if err != nil {
		log.Printf("WebSocket upgrade error: %v", err)
		return
	}

	log.Printf("New WebSocket connection from %s", r.RemoteAddr)
	_ = s.handleConnection(conn)
}

func (s *Server) handleConnection(conn *websocket.Conn) error {
	defer conn.Close()
	conn.SetReadLimit(1 << 20) // 1MB

	s.shutdownMu.RLock()
	if s.isShutdown {
		s.shutdownMu.RUnlock()
		log.Printf("Rejecting connection during shutdown")
		return nil
	}
	s.shutdownMu.RUnlock()

	_, data, err := conn.ReadMessage()
	if err != nil {
		log.Printf("Error reading hello: %v", err)
		return nil
	}

	var msg protocol.Message
	if err := json.Unmarshal(data, &msg); err != nil {
		log.Printf("Error unmarshaling message: %v", err)
		return nil
	}

	if msg.Type != "client/hello" {
		log.Printf("Expected client/hello, got %s", msg.Type)
		return nil
	}

	helloData, err := json.Marshal(msg.Payload)
	if err != nil {
		log.Printf("Error marshaling hello payload: %v", err)
		return nil
	}

	var hello protocol.ClientHello
	if err := json.Unmarshal(helloData, &hello); err != nil {
		log.Printf("Error unmarshaling client hello: %v", err)
		return nil
	}

	if hello.ClientID == "" || hello.Name == "" {
		log.Printf("Client hello missing required fields")
		return nil
	}
	if len(hello.ClientID) > 256 || len(hello.Name) > 256 || len(hello.SupportedRoles) > 20 {
		log.Printf("Client hello fields exceed size limits")
		return nil
	}

	log.Printf("Client hello: %s (ID: %s, Roles: %v)", hello.Name, hello.ClientID, hello.SupportedRoles)

	if s.config.OnClientHello != nil {
		s.config.OnClientHello(hello.ClientID, hello.Name)
	}

	if s.config.ClientFilter != nil && !s.config.ClientFilter(hello.ClientID) {
		log.Printf("Rejecting client %s (filtered out by ClientFilter)", hello.ClientID)
		return ErrClientFiltered
	}

	c := &ServerClient{
		id:           hello.ClientID,
		name:         hello.Name,
		conn:         conn,
		roles:        hello.SupportedRoles,
		capabilities: hello.PlayerV1Support,
		state:        "synchronized",
		volume:       100,
		muted:        false,
		sendChan:     make(chan interface{}, 100),
		done:         make(chan struct{}),
	}

	s.clientsMu.Lock()
	if _, exists := s.clients[hello.ClientID]; exists {
		s.clientsMu.Unlock()
		log.Printf("Client ID %s already connected, rejecting duplicate", hello.ClientID)
		return nil
	}
	s.clients[c.id] = c
	s.clientsMu.Unlock()

	defer func() {
		s.removeClient(c)
		log.Printf("Client disconnected: %s", c.name)
	}()

	activeRoles := s.activateRoles(hello.SupportedRoles)
	serverHello := protocol.ServerHello{
		ServerID:         s.serverID,
		Name:             s.config.Name,
		Version:          ProtocolVersion,
		ActiveRoles:      activeRoles,
		ConnectionReason: "playback",
	}

	if err := c.Send("server/hello", serverHello); err != nil {
		log.Printf("Error sending server hello: %v", err)
		return nil
	}

	s.wg.Add(1)
	go func() {
		defer s.wg.Done()
		s.clientWriter(c)
	}()

	// Add to group AFTER server/hello and writer start so that:
	// 1. server/hello is the first message the client receives
	// 2. The writer goroutine is running to deliver group/update
	//    (from addClient) and role-dispatched messages (stream/start,
	//    server/state) via sendChan.
	s.defaultGroup.addClient(c)

	for {
		_, data, err := conn.ReadMessage()
		if err != nil {
			if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) {
				log.Printf("WebSocket error: %v", err)
			}
			break
		}

		s.handleClientMessage(c, data)
	}
	return nil
}

func (s *Server) clientWriter(c *ServerClient) {
	ticker := time.NewTicker(30 * time.Second)
	defer ticker.Stop()

	const writeDeadline = 10 * time.Second

	for {
		select {
		case msg := <-c.sendChan:
			queueDepth := len(c.sendChan)
			switch v := msg.(type) {
			case []byte:
				c.conn.SetWriteDeadline(time.Now().Add(writeDeadline))
				wStart := time.Now()
				if err := c.conn.WriteMessage(websocket.BinaryMessage, v); err != nil {
					log.Printf("clientWriter %s: binary write error: %v (queue %d)", c.name, err, queueDepth)
					return
				}
				if wDur := time.Since(wStart); wDur > 5*time.Millisecond || queueDepth > 5 {
					log.Printf("clientWriter %s: %s for %d bytes (queue %d)",
						c.name, wDur.Round(time.Microsecond), len(v), queueDepth)
				}
			default:
				data, err := json.Marshal(v)
				if err != nil {
					continue
				}
				c.conn.SetWriteDeadline(time.Now().Add(writeDeadline))
				if err := c.conn.WriteMessage(websocket.TextMessage, data); err != nil {
					log.Printf("clientWriter %s: text write error: %v", c.name, err)
					return
				}
			}

		case <-ticker.C:
			if err := c.conn.WriteControl(websocket.PingMessage, []byte{}, time.Now().Add(10*time.Second)); err != nil {
				return
			}

		case <-c.done:
			return
		}
	}
}

func (s *Server) removeClient(c *ServerClient) {
	c.mu.Lock()
	if c.opusEncoder != nil {
		c.opusEncoder.Close()
		c.opusEncoder = nil
	}
	if c.flacEncoder != nil {
		c.flacEncoder.Close()
		c.flacEncoder = nil
	}
	c.resampler = nil
	c.mu.Unlock()

	s.clientsMu.Lock()
	delete(s.clients, c.id)
	s.clientsMu.Unlock()

	s.defaultGroup.removeClient(c)

	close(c.done)
}

// activateRoles filters a client's advertised role list down to the roles
// this server supports (via config + registered GroupRoles), keeping only
// the first version of each role family so "player@v1" wins over a later
// "player@v2" entry in the same hello.
func (s *Server) activateRoles(supportedRoles []string) []string {
	// Build the set of role families this server supports.
	allowed := make(map[string]bool)
	for _, family := range s.config.SupportedRoles {
		allowed[family] = true
	}

	// Roles registered on the Group are also allowed.
	if s.defaultGroup != nil {
		s.defaultGroup.mu.RLock()
		for family := range s.defaultGroup.roles {
			allowed[family] = true
		}
		s.defaultGroup.mu.RUnlock()
	}

	seen := make(map[string]bool)
	result := make([]string, 0, len(supportedRoles))

	for _, role := range supportedRoles {
		family := role
		if idx := strings.Index(role, "@"); idx > 0 {
			family = role[:idx]
		}

		if seen[family] {
			continue
		}

		if allowed[family] {
			seen[family] = true
			result = append(result, role)
		}
	}

	return result
}