From 8640da55e11a0dcfd27307b97b974af4ebff4b49 Mon Sep 17 00:00:00 2001 From: DiheChen Date: Wed, 4 Nov 2026 23:50:20 +0800 Subject: [PATCH] fix: send prepared results only for the playing entry between instances --- internal/adapter/out/redis/bus_test.go | 43 ++++++++++++++++++++++++-- internal/adapter/out/redis/codec.go | 40 ++++++++++++++++++++++-- 2 files changed, 79 insertions(+), 4 deletions(-) diff --git a/internal/adapter/out/redis/bus_test.go b/internal/adapter/out/redis/bus_test.go index 1c0748f..648c933 100644 --- a/internal/adapter/out/redis/bus_test.go +++ b/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: ¤t}), emitted(room.PlaybackChanged{Current: ¤t})}, + {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") diff --git a/internal/adapter/out/redis/codec.go b/internal/adapter/out/redis/codec.go index 0757fef..d7112b0 100644 --- a/internal/adapter/out/redis/codec.go +++ b/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: