fix: send prepared results only for the playing entry between instances

internal/adapter/out/redis/bus_test.go

@@ -104,10 +104,10 @@ var (
 	waiting = room.Entry{
 		ID: 12, Track: sunny, By: alice, Likes: 2, Ready: true, AddedAt: when,
 		Meta: track.Meta{Title: "晴天", Artist: "周杰伦", Album: "叶惠美", CoverURL: "https://p1.music.126.net/example.jpg", PageURL: "https://music.163.com/#/song?id=186016", Duration: 269 * time.Second},
-		Prep: track.Prepared{Duration: 269500 * time.Millisecond, Gain: 0.75, Lyric: "[00:01.00]故事的小黄花"},
 	}
+	playing = withPrep(waiting)
 	link    = room.Entry{ID: 13, Track: track.Track{Source: track.Link, ID: "https://www.youtube.com/watch?v=x"}, Meta: track.Meta{Title: "https://www.youtube.com/watch?v=x"}, Auto: true, AddedAt: when}
-	current = room.NowPlaying{Entry: waiting, StartedAt: when.Add(time.Minute)}
+	current = room.NowPlaying{Entry: playing, StartedAt: when.Add(time.Minute)}
 	view    = room.View{
 		Term: 3, Seq: 128,
 		Snapshot: room.Snapshot{
@@ -122,6 +122,11 @@ var (
 	}
 )
 
+func withPrep(e room.Entry) room.Entry {
+	e.Prep = track.Prepared{Duration: 269500 * time.Millisecond, Gain: 0.75, Lyric: "[00:01.00]故事的小黄花"}
+	return e
+}
+
 func roomMessageSamples() []port.RoomMessage {
 	return []port.RoomMessage{
 		port.OwnerChanged{Term: 3},
@@ -167,6 +172,40 @@ func TestRoomMessagesArriveInOrder(t *testing.T) {
 	}
 }
 
+func TestOnlyPlayingEntryKeepsPreparedResultAcrossInstances(t *testing.T) {
+	client := newClient(t)
+	a, b := newBus(t, client, "v1"), newBus(t, client, "v1")
+	roomID := randomRoom()
+	if err := b.bus.SubscribeRoom(t.Context(), roomID); err != nil {
+		t.Fatal(err)
+	}
+	pending, queue := playing, []room.Entry{link, playing}
+	sentView, wantView := view, view
+	sentView.Snapshot.Pending, sentView.Snapshot.Waiting = &pending, queue
+	wantView.Snapshot.Waiting = []room.Entry{link, waiting}
+	emitted := func(e room.Event) port.RoomMessage { return port.Emitted{Emit: room.Emit{Term: 3, Seq: 1, Event: e}} }
+	replied := func(v room.View) port.RoomMessage {
+		return port.Replied{Reply: room.Reply{RequestID: "a/r1", Result: room.SyncResult{View: v}}}
+	}
+	for _, c := range []struct{ sent, want port.RoomMessage }{
+		{emitted(room.PlaybackChanged{Current: &current}), emitted(room.PlaybackChanged{Current: &current})},
+		{emitted(room.QueueChanged{Pending: &pending, Waiting: queue}), emitted(room.QueueChanged{Pending: &waiting, Waiting: []room.Entry{link, waiting}})},
+		{emitted(room.Activity{Kind: room.ActivityLiked, Member: alice, Entry: playing}), emitted(room.Activity{Kind: room.ActivityLiked, Member: alice, Entry: waiting})},
+		{replied(sentView), replied(wantView)},
+	} {
+		if err := a.bus.PublishRoom(t.Context(), roomID, c.sent); err != nil {
+			t.Fatalf("发布 %T：%v", c.sent, err)
+		}
+		if got, want := b.next(t), (port.RoomDelivery{Room: roomID, Message: c.want}); !reflect.DeepEqual(got, want) {
+			t.Errorf("收到 %#v\n期望 %#v", got, want)
+		}
+	}
+	// 房间状态和 effect 共用这些切片和指针。
+	if pending != playing || !slices.Equal(queue, []room.Entry{link, playing}) {
+		t.Error("编码时改写了房间状态")
+	}
+}
+
 func TestRoomInputs(t *testing.T) {
 	client := newClient(t)
 	a, b := newBus(t, client, "v1"), newBus(t, client, "v1")

internal/adapter/out/redis/codec.go

@@ -5,10 +5,12 @@ import (
 	"errors"
 	"fmt"
 	"reflect"
+	"slices"
 
 	"github.com/dihemoe/alisten/internal/app/port"
 	"github.com/dihemoe/alisten/internal/domain/member"
 	"github.com/dihemoe/alisten/internal/domain/room"
+	"github.com/dihemoe/alisten/internal/domain/track"
 )
 
 // 只有同一构建版本的实例互相通信，格式不需要跨版本兼容。
@@ -109,7 +111,7 @@ func encodeRoomMessage(m port.RoomMessage) (string, json.RawMessage, error) {
 	case port.Emitted:
 		w := wireEmit{Term: m.Emit.Term, Seq: m.Emit.Seq}
 		var err error
-		if w.Event, w.Data, err = events.encode(m.Emit.Event); err != nil {
+		if w.Event, w.Data, err = events.encode(lightEvent(m.Emit.Event)); err != nil {
 			return "", nil, err
 		}
 		data, err := json.Marshal(w)
@@ -118,7 +120,7 @@ func encodeRoomMessage(m port.RoomMessage) (string, json.RawMessage, error) {
 		w := wireReply{RequestID: m.Reply.RequestID, Err: m.Reply.Err}
 		if m.Reply.Err == nil {
 			var err error
-			if w.Result, w.Data, err = results.encode(m.Reply.Result); err != nil {
+			if w.Result, w.Data, err = results.encode(lightResult(m.Reply.Result)); err != nil {
 				return "", nil, err
 			}
 		}
@@ -128,6 +130,40 @@ func encodeRoomMessage(m port.RoomMessage) (string, json.RawMessage, error) {
 	return roomMessages.encode(m)
 }
 
+// 收到消息的实例只用正在播放的条目的准备结果。其他条目如果也带上，队列每次变化都要把整个队列的歌词发给所有实例。
+func lightEvent(e room.Event) room.Event {
+	switch e := e.(type) {
+	case room.QueueChanged:
+		e.Pending, e.Waiting = withoutPrep(e.Pending, e.Waiting)
+		return e
+	case room.Activity:
+		e.Entry.Prep = track.Prepared{}
+		return e
+	}
+	return e
+}
+
+func lightResult(r room.Result) room.Result {
+	if s, ok := r.(room.SyncResult); ok {
+		s.View.Snapshot.Pending, s.View.Snapshot.Waiting = withoutPrep(s.View.Snapshot.Pending, s.View.Snapshot.Waiting)
+		return s
+	}
+	return r
+}
+
+func withoutPrep(pending *room.Entry, waiting []room.Entry) (*room.Entry, []room.Entry) {
+	if pending != nil {
+		p := *pending
+		p.Prep = track.Prepared{}
+		pending = &p
+	}
+	waiting = slices.Clone(waiting)
+	for i := range waiting {
+		waiting[i].Prep = track.Prepared{}
+	}
+	return pending, waiting
+}
+
 func decodeRoomMessage(kind string, data json.RawMessage) (port.RoomMessage, error) {
 	switch kind {
 	case kindEmitted: