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.

1 files changed, 123 insertions, 17 deletions

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
 	}
 }