-rpi-sendspin/internal/live/manager.go

package live

import (
	"log"
	"sync"

	"github.com/Sendspin/sendspin-go/pkg/sendspin"
)

type Playback string

const (
	Off     Playback = "off"
	Paused  Playback = "paused"
	Playing Playback = "playing"
)

// StateChange is published to subscribers (MQTT bridge) whenever the
// manager's externally-visible state changes.
type StateChange struct {
	ActivePreset string
	Playback     Playback
	Volume       int
	Presets      []Preset
}

// MemberState is per-speaker detail used by /api/state.
type MemberState struct {
	ClientID  string `json:"client_id"`
	Connected bool   `json:"connected"`
	Muted     bool   `json:"muted"`
	Volume    int    `json:"volume"`
}

// Manager is the single source of truth for the live-audio server. All
// state transitions are serialized on its internal mutex.
type Manager struct {
	store  *Store
	server *sendspin.Server
	src    *Source

	mu             sync.Mutex
	active         string // preset id, "" when off
	playback       Playback
	sourceOverride string // when non-empty, wins over preset.Source

	subscribers   []chan StateChange
	subscribersMu sync.Mutex
}

func NewManager(store *Store, server *sendspin.Server, src *Source) *Manager {
	m := &Manager{
		store:    store,
		server:   server,
		src:      src,
		playback: Off,
	}
	// React to preset edits: if the active preset's content changed (or
	// it was deleted), reapply current state.
	store.OnWrite(m.reapply)
	return m
}

// SetSourceOverride pins the captured PipeWire node, ignoring whatever
// `source` field the active preset carries. Passing "" reverts to
// preset-driven behavior. Safe to call before or after activation; the
// new value takes effect on the next preset transition (or immediately
// if a preset is already active).
func (m *Manager) SetSourceOverride(node string) {
	m.mu.Lock()
	m.sourceOverride = node
	active := m.active
	m.mu.Unlock()

	if active == "" {
		return
	}
	// Reapply so the running pw-cat picks up the new target.
	m.reapply()
}

// resolveSource returns the PipeWire node to capture for the given
// preset, honoring sourceOverride when set. Caller must hold m.mu.
func (m *Manager) resolveSource(p Preset) string {
	if m.sourceOverride != "" {
		return m.sourceOverride
	}
	return p.Source
}

// ClientAllowed is intended to be plugged into ServerConfig.ClientFilter.
// It admits a client only if it belongs to the currently active preset.
func (m *Manager) ClientAllowed(clientID string) bool {
	m.mu.Lock()
	defer m.mu.Unlock()
	if m.active == "" {
		return false
	}
	p, ok := m.store.Get(m.active)
	if !ok {
		return false
	}
	for _, id := range p.Members {
		if id == clientID {
			return true
		}
	}
	return false
}

// Subscribe returns a channel that receives a StateChange whenever the
// externally-visible state changes. The channel is buffered; slow
// subscribers drop events.
func (m *Manager) Subscribe() <-chan StateChange {
	ch := make(chan StateChange, 8)
	m.subscribersMu.Lock()
	m.subscribers = append(m.subscribers, ch)
	m.subscribersMu.Unlock()
	return ch
}

func (m *Manager) publish() {
	m.subscribersMu.Lock()
	subs := append([]chan StateChange(nil), m.subscribers...)
	m.subscribersMu.Unlock()

	state := StateChange{
		ActivePreset: m.active,
		Playback:     m.playback,
		Volume:       m.groupVolumeLocked(),
		Presets:      m.store.List(),
	}
	for _, ch := range subs {
		select {
		case ch <- state:
		default:
		}
	}
}

// SetActivePreset transitions to the given preset id ("" → off). Members
// of the new preset are claimed; members no longer needed are released.
// Carries over playback state where possible.
func (m *Manager) SetActivePreset(id string) error {
	m.mu.Lock()
	defer m.mu.Unlock()

	if id == m.active {
		return nil
	}

	var nextMembers map[string]struct{}
	var nextSource string
	if id != "" {
		p, ok := m.store.Get(id)
		if !ok {
			return errNotFound
		}
		nextMembers = make(map[string]struct{}, len(p.Members))
		for _, id := range p.Members {
			nextMembers[id] = struct{}{}
		}
		nextSource = m.resolveSource(p)
	}

	// Drop currently-connected clients that don't belong to the new preset.
	for _, c := range m.server.Clients() {
		if _, ok := nextMembers[c.ID]; !ok {
			m.server.DisconnectClient(c.ID)
		}
	}

	prev := m.active
	m.active = id

	if id == "" {
		// off
		if m.playback == Playing {
			m.server.NotifyStreamEndAll()
		}
		m.playback = Off
		m.src.SetActive(false)
		m.src.SetTarget("")
		log.Printf("live: %s → off", prev)
		m.publish()
		return nil
	}

	// On entry from off, auto-start playback so picking a preset is a
	// one-click action. Switching between presets carries over the
	// current playback state.
	if prev == "" {
		m.playback = Playing
	}

	m.src.SetTarget(nextSource)
	if m.playback == Playing {
		m.src.SetActive(true)
		m.server.NotifyStreamStartAll()
	} else {
		m.src.SetActive(false)
	}

	log.Printf("live: active preset %q (%s)", id, m.playback)
	m.publish()
	return nil
}

// SetPlayback applies a play/pause/stop command. Stop sets the active
// preset to "".
func (m *Manager) SetPlayback(cmd string) error {
	switch cmd {
	case "stop":
		return m.SetActivePreset("")
	case "play":
		return m.setPlayback(Playing)
	case "pause":
		return m.setPlayback(Paused)
	default:
		return errBadRequest
	}
}

func (m *Manager) setPlayback(next Playback) error {
	m.mu.Lock()
	defer m.mu.Unlock()
	if m.active == "" {
		return errNoActivePreset
	}
	if m.playback == next {
		return nil
	}
	prev := m.playback
	m.playback = next

	switch next {
	case Playing:
		m.src.SetActive(true)
		// If transitioning from paused, clients are already connected
		// but received stream/end. Resend stream/start.
		if prev == Paused {
			m.server.NotifyStreamStartAll()
		}
	case Paused:
		m.src.SetActive(false)
		if prev == Playing {
			m.server.NotifyStreamEndAll()
		}
	}

	log.Printf("live: %s → %s", prev, next)
	m.publish()
	return nil
}

// SetVolume applies the spec's group-volume algorithm to the currently
// connected player members. No-op when off.
func (m *Manager) SetVolume(target int) error {
	if target < 0 {
		target = 0
	}
	if target > 100 {
		target = 100
	}

	m.mu.Lock()
	defer m.mu.Unlock()
	if m.active == "" {
		return nil
	}

	clients := m.server.Clients()
	players := make([]sendspin.ClientInfo, 0, len(clients))
	for _, c := range clients {
		players = append(players, c)
	}
	if len(players) == 0 {
		return nil
	}

	// Per spec ("Setting group volume", §Controller messages): apply
	// the *full* delta to every player, clamp out-of-range values, then
	// redistribute the lost delta across the still-eligible players.
	// Iterate until nothing clamps or every player is at a boundary.
	proposed := make(map[string]int, len(players))
	for _, c := range players {
		proposed[c.ID] = c.Volume
	}

	sum := 0
	for _, c := range players {
		sum += proposed[c.ID]
	}
	delta := target - sum/len(players)
	for _, c := range players {
		proposed[c.ID] += delta
	}

	clamped := make(map[string]bool, len(players))
	for range 10 {
		lost := 0
		any := false
		for _, c := range players {
			v := proposed[c.ID]
			if v < 0 {
				lost += v // negative
				proposed[c.ID] = 0
				if !clamped[c.ID] {
					any = true
					clamped[c.ID] = true
				}
			} else if v > 100 {
				lost += v - 100 // positive
				proposed[c.ID] = 100
				if !clamped[c.ID] {
					any = true
					clamped[c.ID] = true
				}
			}
		}
		if !any || lost == 0 {
			break
		}

		eligible := make([]string, 0, len(players))
		for _, c := range players {
			if !clamped[c.ID] {
				eligible = append(eligible, c.ID)
			}
		}
		if len(eligible) == 0 {
			break
		}

		share := lost / len(eligible)
		rem := lost - share*len(eligible)
		for _, id := range eligible {
			d := share
			if rem > 0 {
				d++
				rem--
			} else if rem < 0 {
				d--
				rem++
			}
			proposed[id] += d
		}
	}

	for _, c := range m.serverClientsByID(players) {
		v := proposed[c.ID()]
		_ = c.Send("server/command", map[string]any{
			"player": map[string]any{
				"command": "volume",
				"volume":  v,
			},
		})
	}

	m.publish()
	return nil
}

// reapply is invoked by the preset store after a write so the manager
// can re-evaluate which speakers should be connected and what source to
// capture (e.g. the active preset's member list was edited, or it was
// deleted).
func (m *Manager) reapply() {
	m.mu.Lock()
	active := m.active
	m.mu.Unlock()

	if active == "" {
		return
	}
	p, ok := m.store.Get(active)
	if !ok {
		_ = m.SetActivePreset("")
		return
	}

	m.mu.Lock()
	defer m.mu.Unlock()
	allowed := make(map[string]struct{}, len(p.Members))
	for _, id := range p.Members {
		allowed[id] = struct{}{}
	}
	for _, c := range m.server.Clients() {
		if _, ok := allowed[c.ID]; !ok {
			m.server.DisconnectClient(c.ID)
		}
	}
	m.src.SetTarget(m.resolveSource(p))
	m.publish()
}

// State returns a read-only snapshot for the HTTP /api/state endpoint.
func (m *Manager) State() (string, Playback, int, []MemberState) {
	m.mu.Lock()
	defer m.mu.Unlock()

	members := make([]MemberState, 0)
	allowed := map[string]struct{}{}
	if p, ok := m.store.Get(m.active); ok {
		for _, id := range p.Members {
			allowed[id] = struct{}{}
			members = append(members, MemberState{ClientID: id})
		}
	}
	for _, c := range m.server.Clients() {
		if _, ok := allowed[c.ID]; !ok {
			continue
		}
		for i, mem := range members {
			if mem.ClientID == c.ID {
				members[i].Connected = true
				members[i].Muted = c.Muted
				members[i].Volume = c.Volume
			}
		}
	}
	return m.active, m.playback, m.groupVolumeLocked(), members
}

func (m *Manager) groupVolumeLocked() int {
	if m.active == "" {
		return 0
	}
	clients := m.server.Clients()
	if len(clients) == 0 {
		return 0
	}
	sum := 0
	for _, c := range clients {
		sum += c.Volume
	}
	return sum / len(clients)
}

// serverClientsByID returns the live *ServerClient instances matching
// the IDs in the snapshot. Used so SetVolume can call .Send() rather
// than going through the snapshot, which is value-typed.
func (m *Manager) serverClientsByID(snapshot []sendspin.ClientInfo) []*sendspin.ServerClient {
	if m.server.Group() == nil {
		return nil
	}
	want := make(map[string]struct{}, len(snapshot))
	for _, c := range snapshot {
		want[c.ID] = struct{}{}
	}
	out := make([]*sendspin.ServerClient, 0, len(snapshot))
	for _, c := range m.server.Group().Clients() {
		if _, ok := want[c.ID()]; ok {
			out = append(out, c)
		}
	}
	return out
}