path: root/wrms.go

package main

import (
	"context"
	"encoding/json"
	"sync"
	"sync/atomic"

	"muhq.space/go/wrms/llog"

	"github.com/google/uuid"

	"nhooyr.io/websocket"
)

type Song struct {
	Title     string                 `json:"title"`
	Artist    string                 `json:"artist"`
	Source    string                 `json:"source"`
	Uri       string                 `json:"uri"`
	Weight    int                    `json:"weight"`
	index     int                    `json:"-"` // used by heap.Interface
	Upvotes   map[uuid.UUID]struct{} `json:"-"`
	Downvotes map[uuid.UUID]struct{} `json:"-"`
}

func NewSong(title, artist, source, uri string) Song {
	return Song{title, artist, source, uri, 0, 0, map[uuid.UUID]struct{}{}, map[uuid.UUID]struct{}{}}
}

func NewSongFromJson(data []byte) (Song, error) {
	var s Song
	err := json.Unmarshal(data, &s)
	if err != nil {
		llog.Error("Failed to parse song data %s with %s", string(data), err)
		return s, err
	}

	s.Upvotes = map[uuid.UUID]struct{}{}
	s.Downvotes = map[uuid.UUID]struct{}{}
	return s, nil
}

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

func newEvent(event string, songs []Song) Event {
	return Event{Event: event, Songs: songs}
}

func newNotification(notification string) Event {
	return Event{Event: notification}
}

type Connection struct {
	Id        uuid.UUID
	closing   atomic.Bool
	refs      atomic.Int64
	Events    chan Event
	ws        *websocket.Conn
	ctx       context.Context
	cancel    func()
	nextEvent uint64
}

const EVENT_BUFFER_SIZE = 3

func newConnection(id uuid.UUID, ws *websocket.Conn, ctx context.Context, cncl func()) *Connection {

	return &Connection{
		Id:     id,
		Events: make(chan Event, EVENT_BUFFER_SIZE),
		ws:     ws,
		ctx:    ctx,
		cancel: cncl,
	}
}

func (c *Connection) Send(ev Event) {
	llog.DDebug("Sending %v to %v", ev, c.Id)
	// Connection is closing -> not write to it
	if c.closing.Load() {
		return
	}
	// Register us as sender
	c.refs.Add(1)
	// Send the event
	c.Events <- ev
	// Deregister us as sender
	c.refs.Add(-1)
}

func (conn *Connection) serve() {
	// Map to store future events to send them in order
	toSend := make(map[uint64]*Event)

	for {
		var ev Event
		// We have the next event -> send it
		if toSend[conn.nextEvent] != nil {
			ev = *toSend[conn.nextEvent]
			delete(toSend, conn.nextEvent)
			// We have not received the next event yet -> receive a new event
		} else {
			ev = <-conn.Events
			// We received a future object -> store it and continue
			if ev.Id > conn.nextEvent {
				toSend[ev.Id] = &ev
				continue
			}
		}

		conn.nextEvent++

		data, err := json.Marshal(ev)
		if err != nil {
			llog.Error("Encoding the %s event failed with %s", ev.Event, err)
			return
		}

		llog.Debug("Sending ev %s to %s", string(data), conn.Id)
		err = conn.ws.Write(conn.ctx, websocket.MessageText, data)
		if err != nil {
			llog.Warning("Sending the ev %s to %d failed with %s", string(data), conn.Id, err)
			return
		}
	}
}

func (c *Connection) Close() {
	llog.Info("Closing connection %s", c.Id)
	// Remove the closing connection from the map
	wrms.delConn(c)
	// Announce that the connection is going to be closed
	c.closing.Store(true)

	// Consume all events from the registered senders
	for c.refs.Load() > 0 {
		select {
		case <-c.Events:
		default:
		}
	}

	close(c.Events)
}

type Wrms struct {
	Connections sync.Map
	// 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 = NewPlayer(&wrms, config.Backends)
	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)
	wrms.Connections.Delete(conn.Id)
}

func (wrms *Wrms) initConn(conn *Connection) error {
	wrms.addConn(conn)

	wrms.rwlock.RLock()
	defer wrms.rwlock.RUnlock()

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

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

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

		for _, song := range wrms.Songs {
			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, newEvent("upvoted", upvoted))
	}

	if len(downvoted) > 0 {
		initialCmds = append(initialCmds, newEvent("downvoted", downvoted))
	}

	for _, cmd := range initialCmds {
		llog.Info("Sending initial cmd %v", cmd)
		data, err := json.Marshal(cmd)
		if err != nil {
			llog.Warning("Encoding the initial command %v failed with %s", cmd, err)
			return err
		}

		err = conn.ws.Write(conn.ctx, websocket.MessageText, data)
		if err != nil {
			llog.Warning("Sending the initial commands failed with %s", err)
			return err
		}
	}

	return nil
}

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

func (wrms *Wrms) Broadcast(ev Event) {
	// Tag the event to create a total event order
	ev.Id = wrms.eventId.Add(1)

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

func (wrms *Wrms) startPlaying() {
	currentSong := wrms.CurrentSong.Load()
	wrms.Broadcast(newEvent("play", []Song{*currentSong}))
}

func (wrms *Wrms) AddSong(song Song) {
	startPlayingAgain := wrms.Playing && wrms.CurrentSong.Load() == nil
	wrms.Songs = append(wrms.Songs, song)
	s := &wrms.Songs[len(wrms.Songs)-1]
	wrms.queue.Add(s)
	llog.Info("Added song %s (ptr=%p) to Songs", s.Uri, s)
	wrms.Broadcast(newEvent("add", []Song{song}))

	if startPlayingAgain {
		wrms.Next()
	}
}

func (wrms *Wrms) DeleteSong(songUri string) {
	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)
		wrms.Broadcast(newEvent("delete", []Song{*s}))
		break
	}
}

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

	next := wrms.queue.PopSong()
	if next == nil {
		wrms.CurrentSong.Store(nil)
		wrms.Broadcast(newNotification("stop"))
		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
		}
	}

	if wrms.Playing {
		wrms.Broadcast(newEvent("play", []Song{*next}))
		wrms.Player.Play(next)
	} else if wrms.Player.mpv != nil {
		wrms.Broadcast(newEvent("next", []Song{*next}))
		wrms.Player.terminateMpv()
	}
}

func (wrms *Wrms) PlayPause() {
	wrms.Playing = !wrms.Playing

	if wrms.Playing {
		currentSong := wrms.CurrentSong.Load()
		if currentSong == nil {
			llog.Info("No song currently playing play the next")
			wrms.Next()
		} else {
			wrms.Broadcast(newEvent("play", []Song{*currentSong}))
			if wrms.Player.mpv != nil {
				wrms.Player.Continue()
			} else {
				wrms.Player.Play(currentSong)
			}
		}
	} else {
		wrms.Broadcast(newNotification("pause"))
		wrms.Player.Pause()
	}
}

func (wrms *Wrms) AdjustSongWeight(connId uuid.UUID, songUri string, vote string) {
	for i := 0; i < len(wrms.Songs); i++ {
		s := &wrms.Songs[i]
		if s.Uri != songUri {
			continue
		}

		llog.Info("Adjusting song %s (ptr=%p)", s.Uri, 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
			} else {
				s.Weight += 1
			}

			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)
				s.Weight -= 2
			} else {
				s.Weight -= 1
			}

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

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

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

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

		wrms.queue.Adjust(s)
		wrms.Broadcast(newEvent("update", []Song{*s}))
		break
	}
}