send events in a total order

Move Connection code to wmrs.go. Introduce a RWMutex to synchronize
access to the internal Wrms state. Tag each event when it is broadcast
with the next event count. Senders will order the broadcasted events.
This guarantees that no obsoltete events are ever delivered.

3 files changed, 135 insertions, 109 deletions

diff --git a/TODO.md b/TODO.md
index 49f4a23..3bb9f64 100644
--- a/TODO.md
+++ b/TODO.md
@@ -14,7 +14,7 @@
 - [X] escape HTML 
 - [X] support incremental search results
   - [X] distinguish different search queries
-- [ ] do not apply obsolete events
+- [X] do not apply obsolete events
 
 ## Web Backend
 
diff --git a/main.go b/main.go
index 532cead..0ea10dd 100644
--- a/main.go
+++ b/main.go
@@ -2,7 +2,6 @@ package main
 
 import (
 	"context"
-	"encoding/json"
 	"flag"
 	"fmt"
 	"html/template"
@@ -190,114 +189,35 @@ func nextHandler(w http.ResponseWriter, r *http.Request) {
 }
 
 func wsEndpoint(w http.ResponseWriter, r *http.Request) {
-	c, err := websocket.Accept(w, r, nil)
-	if err != nil {
-		llog.Warning("Accepting the websocket failed with %s", err)
-		return
-	}
-
-	uuidCookie, err := r.Cookie("UUID")
+	connId, err := getConnId(w, r)
 	if err != nil {
-		llog.Error("websocket connection has no set UUID cookie")
-		http.Error(w, "websocket connection has no set UUID cookie", http.StatusBadRequest)
 		return
 	}
 
-	id, err := uuid.Parse(uuidCookie.Value)
+	ws, err := websocket.Accept(w, r, nil)
 	if err != nil {
-		llog.Error("Invalid UUID set in cookie")
-		http.Error(w, "Invalid UUID set in cookie", http.StatusBadRequest)
+		llog.Warning("Accepting the websocket failed with %s", err)
 		return
 	}
 
 	ctx, cancel := context.WithCancel(r.Context())
-	ctx = c.CloseRead(ctx)
+	ctx = ws.CloseRead(ctx)
 
 	defer func() {
-		llog.Info("cancel context of connection %v", id)
+		llog.Info("cancel context of connection %v", connId)
 		cancel()
 	}()
 
-	conn := newConnection(id, c)
-	wrms.addConn(conn)
+	conn := newConnection(connId, ws, ctx, cancel)
 	defer conn.Close()
 
-	llog.Info("New websocket connection with id %v", id)
-
-	initialCmds := [][]byte{}
-	if wrms.Playing {
-		var songs []Song
-		if currentSong := wrms.CurrentSong.Load(); currentSong != nil {
-			songs = []Song{*currentSong}
-		}
-		data, err := json.Marshal(wrms.newEvent("play", songs))
-		if err != nil {
-			llog.Error("Encoding the play event failed with %s", err)
-			return
-		}
-		initialCmds = append(initialCmds, data)
-	}
-
-	upvoted := []Song{}
-	downvoted := []Song{}
-	if len(wrms.Songs) > 0 {
-		data, err := json.Marshal(wrms.newEvent("add", wrms.Songs))
-		if err != nil {
-			llog.Error("Encoding the add event failed with %s", err)
-			return
-		}
-		initialCmds = append(initialCmds, data)
-
-		for _, song := range wrms.Songs {
-			if _, ok := song.Upvotes[id]; ok {
-				upvoted = append(upvoted, song)
-			} else if _, ok := song.Downvotes[id]; ok {
-				downvoted = append(downvoted, song)
-			}
-		}
-	}
-
-	if len(upvoted) > 0 {
-		data, err := json.Marshal(wrms.newEvent("upvoted", upvoted))
-		if err != nil {
-			llog.Error("Encoding the upvoted event failed with %s", err)
-			return
-		}
-		initialCmds = append(initialCmds, data)
-	}
-
-	if len(downvoted) > 0 {
-		data, err := json.Marshal(wrms.newEvent("downvoted", downvoted))
-		if err != nil {
-			llog.Error("Encoding the upvoted event failed with %s", err)
-			return
-		}
-		initialCmds = append(initialCmds, data)
-	}
-
-	for _, data := range initialCmds {
-		llog.Info("Sending initial cmd %s", string(data))
-		err = c.Write(ctx, websocket.MessageText, data)
-		if err != nil {
-			llog.Warning("Sending the initial commands failed with %s", err)
-			return
-		}
+	llog.Info("New websocket connection with id %v", connId)
+	err = wrms.initConn(conn)
+	if err != nil {
+		return
 	}
 
-	for ev := range conn.Events {
-		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), id)
-		err = c.Write(ctx, websocket.MessageText, data)
-		if err != nil {
-			llog.Warning("Sending the ev %s to %d failed with %s", string(data), id, err)
-			return
-		}
-	}
+	conn.serve()
 }
 
 func setupRoutes() {
diff --git a/wrms.go b/wrms.go
index c983eb5..7e45b97 100644
--- a/wrms.go
+++ b/wrms.go
@@ -1,6 +1,7 @@
 package main
 
 import (
+	"context"
 	"encoding/json"
 	"sync"
 	"sync/atomic"
@@ -46,29 +47,35 @@ type Event struct {
 	Songs []Song `json:"songs"`
 }
 
-func (wrms *Wrms) newEvent(event string, songs []Song) Event {
-	return Event{Event: event, Id: wrms.eventId.Add(1), Songs: songs}
+func newEvent(event string, songs []Song) Event {
+	return Event{Event: event, Songs: songs}
 }
 
 func newNotification(notification string) Event {
-	return Event{Event: notification, Id: wrms.eventId.Add(1)}
+	return Event{Event: notification}
 }
 
 type Connection struct {
-	Id      uuid.UUID
-	closing atomic.Bool
-	refs    atomic.Int64
-	Events  chan Event
-	C       *websocket.Conn
+	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) *Connection {
+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),
-		C:      ws,
+		ws:     ws,
+		ctx:    ctx,
+		cancel: cncl,
 	}
 }
 
@@ -86,6 +93,42 @@ func (c *Connection) Send(ev Event) {
 	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
+			}
+		}
+
+		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
+		}
+	}
+	conn.nextEvent++
+}
+
 func (c *Connection) Close() {
 	llog.Info("Closing connection %s", c.Id)
 	// Remove the closing connection from the map
@@ -106,6 +149,8 @@ func (c *Connection) Close() {
 
 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]
@@ -132,12 +177,73 @@ func (wrms *Wrms) delConn(conn *Connection) {
 	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)
@@ -147,7 +253,7 @@ func (wrms *Wrms) Broadcast(ev Event) {
 
 func (wrms *Wrms) startPlaying() {
 	currentSong := wrms.CurrentSong.Load()
-	wrms.Broadcast(wrms.newEvent("play", []Song{*currentSong}))
+	wrms.Broadcast(newEvent("play", []Song{*currentSong}))
 }
 
 func (wrms *Wrms) AddSong(song Song) {
@@ -156,7 +262,7 @@ func (wrms *Wrms) AddSong(song 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(wrms.newEvent("add", []Song{song}))
+	wrms.Broadcast(newEvent("add", []Song{song}))
 
 	if startPlayingAgain {
 		wrms.Next()
@@ -174,7 +280,7 @@ func (wrms *Wrms) DeleteSong(songUri string) {
 		wrms.Songs = wrms.Songs[:len(wrms.Songs)-1]
 
 		wrms.queue.RemoveSong(s)
-		wrms.Broadcast(wrms.newEvent("delete", []Song{*s}))
+		wrms.Broadcast(newEvent("delete", []Song{*s}))
 		break
 	}
 }
@@ -202,10 +308,10 @@ func (wrms *Wrms) Next() {
 	}
 
 	if wrms.Playing {
-		wrms.Broadcast(wrms.newEvent("play", []Song{*next}))
+		wrms.Broadcast(newEvent("play", []Song{*next}))
 		wrms.Player.Play(next)
 	} else if wrms.Player.mpv != nil {
-		wrms.Broadcast(wrms.newEvent("next", []Song{*next}))
+		wrms.Broadcast(newEvent("next", []Song{*next}))
 		wrms.Player.terminateMpv()
 	}
 }
@@ -219,7 +325,7 @@ func (wrms *Wrms) PlayPause() {
 			llog.Info("No song currently playing play the next")
 			wrms.Next()
 		} else {
-			wrms.Broadcast(wrms.newEvent("play", []Song{*currentSong}))
+			wrms.Broadcast(newEvent("play", []Song{*currentSong}))
 			if wrms.Player.mpv != nil {
 				wrms.Player.Continue()
 			} else {
@@ -290,7 +396,7 @@ func (wrms *Wrms) AdjustSongWeight(connId uuid.UUID, songUri string, vote string
 		}
 
 		wrms.queue.Adjust(s)
-		wrms.Broadcast(wrms.newEvent("update", []Song{*s}))
+		wrms.Broadcast(newEvent("update", []Song{*s}))
 		break
 	}
 }