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) AddSong(song Song) { wrms.rwlock.Lock() startPlayingAgain := wrms.Playing && wrms.CurrentSong.Load() == nil wrms.Songs = append(wrms.Songs, song) s := &wrms.Songs[len(wrms.Songs)-1] wrms.queue.Add(s) wrms.rwlock.Unlock() 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) { 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) wrms.rwlock.Unlock() wrms.Broadcast(newEvent("delete", []Song{*s})) break } } func (wrms *Wrms) Next() { 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(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 } } var cmd string if wrms.Playing { wrms.Player.Play(next) cmd = "play" } else if wrms.Player.mpv != nil { wrms.Player.terminateMpv() cmd = "next" } wrms.rwlock.Unlock() wrms.Broadcast(newEvent(cmd, []Song{*next})) } func (wrms *Wrms) PlayPause() { wrms.rwlock.Lock() // Toggle the playback state defer func() { wrms.Playing = !wrms.Playing }() // Wrms is playing -> pause the player if wrms.Playing { wrms.Player.Pause() wrms.rwlock.Unlock() wrms.Broadcast(newNotification("pause")) return } // Wrms is 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 } // Wrms has an active mpv process -> continue playing if wrms.Player.mpv != nil { wrms.Player.Continue() // Wrms has a current song but no mpv process playing it -> start a new one } else { wrms.Player.Play(currentSong) } wrms.rwlock.Unlock() wrms.Broadcast(newEvent("play", []Song{*currentSong})) } 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 %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.rwlock.Unlock() wrms.Broadcast(newEvent("update", []Song{*s})) break } }