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

// ABOUTME: ServerClient represents one accepted connection on a sendspin Server
// ABOUTME: Exposes ID/Name/Roles/Send surface for the Group/GroupRole layer
package sendspin

import (
	"encoding/binary"
	"encoding/json"
	"fmt"
	"log"
	"strings"
	"sync"
	"time"

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

// ServerClient represents one accepted WebSocket connection on a Server.
// It owns the client's negotiated roles, capabilities, playback state,
// per-codec encoder state, and the outbound send channel.
//
// Exposed accessors (ID, Name, Roles, HasRole, Send, SendBinary) are the
// stable surface the future Group/GroupRole layer depends on. Internal
// fields remain unexported so the server package can mutate them through
// the mutex without leaking that detail to callers.
type ServerClient struct {
	id           string
	name         string
	conn         *websocket.Conn
	roles        []string
	capabilities *protocol.PlayerV1Support

	state  string
	volume int
	muted  bool

	codec         string
	opusEncoder   *server.OpusEncoder
	flacEncoder   *server.FLACEncoder
	resampler     *audio.Resampler // non-nil only when source rate != 48kHz
	bufferTracker *BufferTracker

	sendChan chan interface{}
	done     chan struct{}

	mu sync.RWMutex
}

// ID returns the client-supplied unique identifier from client/hello.
func (c *ServerClient) ID() string { return c.id }

// Name returns the human-friendly name from client/hello.
func (c *ServerClient) Name() string { return c.name }

// Roles returns the client's advertised role list as a fresh slice.
// Callers may mutate the returned slice without affecting the ServerClient.
func (c *ServerClient) Roles() []string {
	out := make([]string, len(c.roles))
	copy(out, c.roles)
	return out
}

// HasRole reports whether this client advertised a given role family.
// It matches both exact ("player") and versioned ("player@v1") forms so
// callers can query by family without tracking versions.
func (c *ServerClient) HasRole(role string) bool {
	for _, r := range c.roles {
		if r == role || strings.HasPrefix(r, role+"@") {
			return true
		}
	}
	return false
}

// Send enqueues a typed control message for transmission. Returns an
// error immediately if the client's send buffer is full rather than
// blocking — callers decide whether to drop or disconnect.
//
// A nil return means the message was enqueued, not that it was delivered:
// after the client's writer goroutine has exited (e.g., post-disconnect),
// enqueued messages are dropped silently. Callers should treat this as
// best-effort once they've observed a client leaving.
func (c *ServerClient) Send(msgType string, payload interface{}) error {
	msg := protocol.Message{
		Type:    msgType,
		Payload: payload,
	}
	select {
	case c.sendChan <- msg:
		return nil
	default:
		return fmt.Errorf("client send buffer full")
	}
}

// SendBinary enqueues a raw binary frame (e.g., an audio chunk) for
// transmission. Same non-blocking semantics as Send.
//
// A nil return means the frame was enqueued, not that it was delivered:
// after the client's writer goroutine has exited (e.g., post-disconnect),
// enqueued frames are dropped silently. Callers should treat this as
// best-effort once they've observed a client leaving.
func (c *ServerClient) SendBinary(data []byte) error {
	// Fast path: queue has room.
	select {
	case c.sendChan <- data:
		return nil
	default:
	}

	// Queue full. Drop-oldest is timestamp-aware in effect: the engine
	// pushes chunks in monotonically increasing playbackTime order, so
	// the head of the queue has the earliest scheduled time = the chunk
	// most likely to be stale by the time the writer catches up.
	//
	// We use the *new* chunk's playbackTime as our "now + BufferAheadMs"
	// reference (the engine just stamped it that way), so we don't need
	// an external clock to decide what's stale. The actual log includes
	// how far past the scheduling headroom the dropped chunks were.
	newTS, newOK := extractAudioPlaybackTime(data)
	const bufferAheadUs = int64(BufferAheadMs) * 1000

	dropped := 0
	var maxLatenessUs int64
	for dropped < cap(c.sendChan) {
		select {
		case head := <-c.sendChan:
			if newOK {
				if headBytes, ok := head.([]byte); ok {
					if headTS, headOK := extractAudioPlaybackTime(headBytes); headOK {
						// Lateness = how long past "now" this chunk's playback
						// time was. "Now" ≈ newTS - bufferAhead.
						late := (newTS - bufferAheadUs) - headTS
						if late > maxLatenessUs {
							maxLatenessUs = late
						}
					}
				}
			}
			dropped++
		default:
		}
		select {
		case c.sendChan <- data:
			if dropped > 0 {
				log.Printf("client %s send pressure: dropped %d stale chunks (oldest was %s past schedule)",
					c.name, dropped,
					time.Duration(maxLatenessUs)*time.Microsecond)
			}
			return nil
		default:
		}
	}
	return fmt.Errorf("client send buffer full (depth=%d, cap=%d)",
		len(c.sendChan), cap(c.sendChan))
}

// extractAudioPlaybackTime returns the playbackTime (in microseconds) embedded
// in a sendspin audio chunk's 9-byte header. Returns false for non-audio frames.
func extractAudioPlaybackTime(data []byte) (int64, bool) {
	const audioChunkHeaderSize = 9
	if len(data) < audioChunkHeaderSize || data[0] != AudioChunkMessageType {
		return 0, false
	}
	return int64(binary.BigEndian.Uint64(data[1:audioChunkHeaderSize])), true
}

// State returns the client's current playback state ("synchronized",
// "playing", "paused"). Safe to call concurrently with state updates.
func (c *ServerClient) State() string {
	c.mu.RLock()
	defer c.mu.RUnlock()
	return c.state
}

// Volume returns the client's reported volume (0-100). Safe to call
// concurrently with state updates.
func (c *ServerClient) Volume() int {
	c.mu.RLock()
	defer c.mu.RUnlock()
	return c.volume
}

// Muted returns the client's mute state. Safe to call concurrently with
// state updates.
func (c *ServerClient) Muted() bool {
	c.mu.RLock()
	defer c.mu.RUnlock()
	return c.muted
}

// Codec returns the currently-negotiated codec name ("pcm", "opus", "").
// Returns the empty string before stream negotiation completes. Safe to
// call concurrently with state updates.
func (c *ServerClient) Codec() string {
	c.mu.RLock()
	defer c.mu.RUnlock()
	return c.codec
}

// NewServerClientFromConn wraps an existing WebSocket connection in a
// ServerClient and starts a background writer goroutine. This is the
// entry point for code that accepts its own WebSocket connections
// (e.g., conformance adapters) and wants to use the typed Send/SendBinary
// API without constructing a full Server.
//
// The caller is responsible for reading from the connection; the writer
// goroutine handles outbound messages via Send/SendBinary. Call Close()
// when done to stop the writer and release resources.
func NewServerClientFromConn(conn *websocket.Conn, id, name string, roles []string, capabilities *protocol.PlayerV1Support) *ServerClient {
	c := &ServerClient{
		id:           id,
		name:         name,
		conn:         conn,
		roles:        roles,
		capabilities: capabilities,
		state:        "synchronized",
		volume:       100,
		sendChan:     make(chan interface{}, 100),
		done:         make(chan struct{}),
	}
	go c.runWriter()
	return c
}

// runWriter drains the sendChan and writes messages to the WebSocket.
// Exits when the done channel is closed (via Close). This is the
// standalone equivalent of Server.clientWriter for ServerClients
// created via NewServerClientFromConn.
func (c *ServerClient) runWriter() {
	ticker := time.NewTicker(30 * time.Second)
	defer ticker.Stop()

	const writeDeadline = 10 * time.Second

	exit := func(reason string, err error) {
		log.Printf("runWriter %s exiting: %s: %v (queue depth %d)",
			c.name, reason, err, len(c.sendChan))
	}

	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 {
					exit("binary write", err)
					return
				}
				wDur := time.Since(wStart)
				if wDur > 5*time.Millisecond || queueDepth > 5 {
					log.Printf("ws write %s: %s for %d bytes (queue depth %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 {
					exit("text write", err)
					return
				}
			}

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

		case <-c.done:
			return
		}
	}
}

// Close stops the writer goroutine and signals that this ServerClient
// is done. Safe to call multiple times. Does NOT close the underlying
// WebSocket connection — the caller owns that lifecycle.
func (c *ServerClient) Close() {
	c.mu.Lock()
	defer c.mu.Unlock()
	select {
	case <-c.done:
		// Already closed
	default:
		close(c.done)
	}
}