wrms.go

package main

import (
	"sync"
	"sync/atomic"

	"muhq.space/go/wrms/llog"

	"github.com/google/uuid"
)

type Event struct {
	Event string  `json:"cmd"`
	Id    uint64  `json:"id"`
	Songs []*Song `json:"songs"`
}

func (wrms *Wrms) incEventId() uint64 {
	id := wrms.eventId.Add(1)
	llog.DDebug("Increment event id to: %d", id)
	return id
}

// A private event does not appear in the global event order and
// thus does not increment the global event count.
// Private events are used for example to report search results.
func (wrms *Wrms) newPrivateEvent(id uint64, event string, songs []*Song) Event {
	return Event{Id: id, Event: event, Songs: songs}
}

func (wrms *Wrms) newEvent(event string, songs []*Song) Event {
	return Event{Id: wrms.incEventId(), Event: event, Songs: songs}
}

func (wrms *Wrms) newNotification(notification string) Event {
	return wrms.newEvent(notification, nil)
}

type Wrms struct {
	Connections sync.Map
	nextConnNr  atomic.Uint64
	// The rwlock must be held when using most internal state
	rwlock      sync.RWMutex
	Songs       []*Song
	queue       Playlist
	CurrentSong atomic.Pointer[Song]
	Player      Player
	playing     bool
	Config      Config
	eventId     atomic.Uint64
}

func NewWrms(config Config) *Wrms {
	wrms := Wrms{}
	wrms.Config = config
	wrms.Player = NewMpvPlayer(&wrms, config.Backends)

	if len(wrms.Config.Playlists) > 0 {
		wrms.loadPlaylists(wrms.Config.Playlists)
	}

	return &wrms
}

func (wrms *Wrms) addConn(conn *Connection) {
	llog.DDebug("Adding Connection %s", conn.Id)
	wrms.Connections.Store(conn.Id, conn)
}

func (wrms *Wrms) delConn(conn *Connection) {
	llog.DDebug("Deleting Connection %s", conn.Id)
	_oldConn, _ := wrms.Connections.Load(conn.Id)
	oldConn := _oldConn.(*Connection)
	// Only delete the connection if no newer one is currently active
	if oldConn.nr == conn.nr {
		wrms.Connections.Delete(conn.Id)
	}
}

func (wrms *Wrms) initConn(conn *Connection) {
	wrms.rwlock.RLock()

	curEventId := wrms.eventId.Load()
	conn.nextEvent = curEventId + 1

	initialCmds := []interface{}{}

	if wrms.Config.TimeBonus != 0 {
		ev := map[string]any{"cmd": "timeBonus", "timeBonus": wrms.Config.TimeBonus}
		initialCmds = append(initialCmds, ev)
	}

	if wrms.playing {
		var songs []*Song
		if currentSong := wrms.CurrentSong.Load(); currentSong != nil {
			songs = []*Song{currentSong}
		}
		initialCmds = append(initialCmds, wrms.newPrivateEvent(curEventId, "play", songs))
	}

	upvoted := []*Song{}
	downvoted := []*Song{}
	if len(wrms.Songs) > 0 {
		initialCmds = append(initialCmds, wrms.newPrivateEvent(curEventId, "add", wrms.Songs))

		for _, song := range wrms.queue.OrderedList() {
			llog.DDebug("Looking at the votes of song: %v", song)
			if _, ok := song.Upvotes[conn.Id]; ok {
				upvoted = append(upvoted, song)
			} else if _, ok := song.Downvotes[conn.Id]; ok {
				downvoted = append(downvoted, song)
			}
		}
	}

	if len(upvoted) > 0 {
		initialCmds = append(initialCmds, wrms.newPrivateEvent(curEventId, "upvoted", upvoted))
	}

	if len(downvoted) > 0 {
		initialCmds = append(initialCmds, wrms.newPrivateEvent(curEventId, "downvoted", downvoted))
	}

	wrms.rwlock.RUnlock()
	llog.Info("Sending initial cmds %v", initialCmds)
	conn._send(initialCmds)
}

func (wrms *Wrms) GetConn(connId uuid.UUID) *Connection {
	conn, _ := wrms.Connections.Load(connId)
	return conn.(*Connection)
}

func (wrms *Wrms) Broadcast(ev Event) {
	llog.Info("Broadcasting %v", ev)
	wrms.Connections.Range(func(_, conn any) bool {
		conn.(*Connection).Send(ev)
		return true
	})
}

func (wrms *Wrms) _addSong(song *Song) {
	wrms.Songs = append(wrms.Songs, song)
	wrms.queue.Add(song)
}

func (wrms *Wrms) AddSong(song *Song) {
	wrms.rwlock.Lock()

	startPlayingAgain := wrms.playing && wrms.CurrentSong.Load() == nil

	if wrms.Config.TimeBonus != 0 {
		llog.Info("Apply time bonus %v", wrms.Config.TimeBonus)
		wrms.queue.applyTimeBonus(wrms.Config.TimeBonus)
	}

	wrms._addSong(song)

	ev := wrms.newEvent("add", []*Song{song})
	wrms.rwlock.Unlock()

	llog.Info("Added song %s (ptr=%p) to Songs", song.Uri, song)
	wrms.Broadcast(ev)

	if startPlayingAgain {
		wrms._lockedNext()
	}
}

func (wrms *Wrms) DeleteSong(songUri string) {
	wrms.rwlock.Lock()

	for i := 0; i < len(wrms.Songs); i++ {
		s := wrms.Songs[i]
		if s.Uri != songUri {
			continue
		}

		wrms.Songs[i] = wrms.Songs[len(wrms.Songs)-1]
		wrms.Songs = wrms.Songs[:len(wrms.Songs)-1]

		wrms.queue.RemoveSong(s)

		ev := wrms.newEvent("delete", []*Song{s})
		wrms.rwlock.Unlock()

		wrms.Broadcast(ev)
		break
	}
}

func (wrms *Wrms) Next() {
	wrms.rwlock.Lock()

	// Terminate the Player if it is currently playing
	if wrms.Player.Playing() {
		wrms.Player.Stop()
	}

	wrms._next()
}

func (wrms *Wrms) _lockedNext() {
	wrms.rwlock.Lock()
	wrms._next()
}

func (wrms *Wrms) _next() {
	llog.DDebug("Next Song")

	next := wrms.queue.PopSong()
	if next == nil {
		wrms.CurrentSong.Store(nil)
		wrms.Broadcast(wrms.newNotification("stop"))
		wrms.rwlock.Unlock()
		return
	}

	wrms.CurrentSong.Store(next)

	llog.Info("popped next song and removing it from the song list %v", next)

	for i, s := range wrms.Songs {
		if s.Uri == next.Uri {
			wrms.Songs[i] = wrms.Songs[len(wrms.Songs)-1]
			wrms.Songs = wrms.Songs[:len(wrms.Songs)-1]
			break
		}
	}

	cmd := "next"
	// We are playing -> start playing the next song
	if wrms.playing {
		wrms.Player.Play(next)
		cmd = "play"
	}

	ev := wrms.newEvent(cmd, []*Song{next})
	wrms.rwlock.Unlock()

	wrms.Broadcast(ev)
}

func (wrms *Wrms) PlayPause() {
	wrms.rwlock.Lock()
	// Toggle the playback state
	wrms.playing = !wrms.playing

	// Wrms was playing -> pause the player
	if !wrms.playing {
		wrms.Player.Pause()
		wrms.rwlock.Unlock()
		wrms.Broadcast(wrms.newNotification("pause"))
		return
	}

	// Wrms was not playing
	currentSong := wrms.CurrentSong.Load()

	// Wrms has no current song -> try to play the next song
	if currentSong == nil {
		llog.Info("No song currently playing play the next")
		// _next() releases the rwlock
		wrms._next()
		return
	}

	// The player is playing -> continue playing
	if wrms.Player.Playing() {
		wrms.Player.Continue()
		// The player is stopped -> start it
	} else {
		wrms.Player.Play(currentSong)
	}

	ev := wrms.newEvent("play", []*Song{currentSong})
	wrms.rwlock.Unlock()

	wrms.Broadcast(ev)
}

func (wrms *Wrms) AdjustSongWeight(connId uuid.UUID, songUri string, vote string) {
	wrms.rwlock.Lock()

	for i := 0; i < len(wrms.Songs); i++ {
		s := wrms.Songs[i]
		if s.Uri != songUri {
			continue
		}

		llog.Info("Adjusting song %v (ptr=%p)", s, s)
		switch vote {
		case "up":
			if _, ok := s.Upvotes[connId]; ok {
				llog.Error("Double upvote of song %s by connections %s", songUri, connId)
				return
			}

			if _, ok := s.Downvotes[connId]; ok {
				delete(s.Downvotes, connId)
				s.Weight += 2.0
				llog.Debug("Flip downvote")
			} else {
				s.Weight += 1.0
			}

			s.Upvotes[connId] = struct{}{}

		case "down":
			if _, ok := s.Downvotes[connId]; ok {
				llog.Error("Double downvote of song %s by connections %s", songUri, connId)
				return
			}

			if _, ok := s.Upvotes[connId]; ok {
				delete(s.Upvotes, connId)
				llog.Debug("Flip upvote")
				s.Weight -= 2.0
			} else {
				s.Weight -= 1.0
			}

			s.Downvotes[connId] = struct{}{}

		case "unvote":
			if _, ok := s.Downvotes[connId]; ok {
				delete(s.Downvotes, connId)
				s.Weight += 1.0

			} else if _, ok := s.Upvotes[connId]; ok {
				delete(s.Upvotes, connId)
				s.Weight -= 1.0
			} else {
				llog.Error("Double unvote of song %s by connections %s", songUri, connId)
				return
			}

		default:
			llog.Fatal("invalid vote")
		}

		wrms.queue.Adjust(s)
		ev := wrms.newEvent("update", []*Song{s})
		wrms.rwlock.Unlock()

		wrms.Broadcast(ev)
		break
	}
}

func (wrms *Wrms) Search(pattern map[string]string) chan []*Song {
	return wrms.Player.Search(pattern)
}

func (wrms *Wrms) loadPlaylists(playlists []string) {
	for _, playlist := range playlists {
		wrms.appendPlaylist(playlist)
	}
}

func (wrms *Wrms) appendPlaylist(playlist string) {
	songs := wrms.Player.LoadPlaylist(playlist)

	for _, song := range songs {
		wrms._addSong(song)
	}
}