From 6bd8a1aea86e3d12fd128ceb986344864e9ebbae Mon Sep 17 00:00:00 2001
From: DiheChen <michikawachin@outlook.com>
Date: Tue, 10 Nov 2026 12:08:12 +0800
Subject: [PATCH] feat: tell which members hear each song from the room domain

---
 internal/domain/room/effect.go                | 25 +++++--
 internal/domain/room/fuzz_test.go             | 30 ++++++++
 internal/domain/room/golden_test.go           |  2 +
 internal/domain/room/room.go                  | 25 +++++++
 internal/domain/room/testdata/autodj.txtar    |  2 +
 internal/domain/room/testdata/listen.txtar    | 73 +++++++++++++++++++
 .../domain/room/testdata/sequential.txtar     |  2 +
 internal/domain/room/testdata/vote.txtar      |  1 +
 8 files changed, 152 insertions(+), 8 deletions(-)
 create mode 100644 internal/domain/room/testdata/listen.txtar

diff --git a/internal/domain/room/effect.go b/internal/domain/room/effect.go
index 30b8960..d08742e 100644
--- a/internal/domain/room/effect.go
+++ b/internal/domain/room/effect.go
@@ -44,16 +44,25 @@ type RecordPlay struct {
 	EndedAt   time.Time
 }
 
+type RecordListen struct {
+	// 成员 ID，不重复。
+	Members []string
+	Entry   Entry
+	// 这首歌的开播时间。
+	StartedAt time.Time
+}
+
 type Close struct{}
 
-func (Persist) effect()    {}
-func (Emit) effect()       {}
-func (Reply) effect()      {}
-func (Prepare) effect()    {}
-func (Recommend) effect()  {}
-func (Schedule) effect()   {}
-func (RecordPlay) effect() {}
-func (Close) effect()      {}
+func (Persist) effect()      {}
+func (Emit) effect()         {}
+func (Reply) effect()        {}
+func (Prepare) effect()      {}
+func (Recommend) effect()    {}
+func (Schedule) effect()     {}
+func (RecordPlay) effect()   {}
+func (RecordListen) effect() {}
+func (Close) effect()        {}
 
 type Event interface{ event() }
 
diff --git a/internal/domain/room/fuzz_test.go b/internal/domain/room/fuzz_test.go
index 55d442d..d7792b3 100644
--- a/internal/domain/room/fuzz_test.go
+++ b/internal/domain/room/fuzz_test.go
@@ -227,6 +227,7 @@ func checkInvariants(t *testing.T, r *Room, before Snapshot, effects []Effect, t
 	} else if len(find[Persist](effects)) != 0 {
 		t.Fatalf("持久状态没变却落盘：%#v", effects)
 	}
+	checkListen(t, r, effects)
 	for _, e := range find[Emit](effects) {
 		if e.Term != term || e.Seq != *seq+1 {
 			t.Fatalf("任期 %d 中 seq %d 之后是任期 %d 的 %d", term, *seq, e.Term, e.Seq)
@@ -234,3 +235,32 @@ func checkInvariants(t *testing.T, r *Room, before Snapshot, effects []Effect, t
 		*seq = e.Seq
 	}
 }
+
+// 听歌记录只记正在播放的歌和在线的成员，排在落盘之后、广播之前。
+func checkListen(t *testing.T, r *Room, effects []Effect) {
+	t.Helper()
+	i := slices.IndexFunc(effects, func(e Effect) bool { _, ok := e.(RecordListen); return ok })
+	if i < 0 {
+		return
+	}
+	l := effects[i].(RecordListen)
+	if len(find[RecordListen](effects)) != 1 {
+		t.Fatalf("一批中有多个 RecordListen：%#v", effects)
+	}
+	if len(find[Emit](effects[:i])) != 0 || len(find[Persist](effects[i:])) != 0 {
+		t.Fatalf("RecordListen 应排在 Persist 之后、Emit 之前：%#v", effects)
+	}
+	if r.current == nil || l.Entry != r.current.Entry || !l.StartedAt.Equal(r.current.StartedAt) {
+		t.Fatalf("RecordListen 记的不是正在播放的歌：%#v，正在播放 %#v", l, r.current)
+	}
+	if len(l.Members) == 0 {
+		t.Fatal("RecordListen 没有成员")
+	}
+	seen := map[string]bool{}
+	for _, id := range l.Members {
+		if seen[id] || !slices.ContainsFunc(r.sessions, func(s Session) bool { return s.Member.ID == id }) {
+			t.Fatalf("RecordListen 的成员重复或不在线：%v，在线 %#v", l.Members, r.sessions)
+		}
+		seen[id] = true
+	}
+}
diff --git a/internal/domain/room/golden_test.go b/internal/domain/room/golden_test.go
index 2ebb3fa..f95c03d 100644
--- a/internal/domain/room/golden_test.go
+++ b/internal/domain/room/golden_test.go
@@ -224,6 +224,8 @@ func formatEffect(e Effect) string {
 			modeNames[s.Mode], s.EntrySeq, formatPlaying(s.Current), formatEntryPtr(s.Pending), formatEntries(s.Waiting))
 	case RecordPlay:
 		return fmt.Sprintf("record %s %s-%s", formatEntry(e.Entry), offset(e.StartedAt), offset(e.EndedAt))
+	case RecordListen:
+		return fmt.Sprintf("listen [%s] %s@%s", strings.Join(e.Members, " "), formatEntry(e.Entry), offset(e.StartedAt))
 	case Emit:
 		return fmt.Sprintf("emit %d.%d %s", e.Term, e.Seq, formatEvent(e.Event))
 	case Close:
diff --git a/internal/domain/room/room.go b/internal/domain/room/room.go
index 4be5be5..2786a09 100644
--- a/internal/domain/room/room.go
+++ b/internal/domain/room/room.go
@@ -164,6 +164,10 @@ func (r *Room) commit(b *batch) []Effect {
 	for _, p := range b.plays {
 		out = append(out, p)
 	}
+	// 排在广播之前：客户端看到开播时，听歌记录已经写好。
+	if l, ok := r.listened(b); ok {
+		out = append(out, l)
+	}
 	events := b.events
 	if after.Info != b.before.Info {
 		events = append(events, InfoChanged{Name: after.Info.Name, Description: after.Info.Description})
@@ -193,6 +197,27 @@ func (r *Room) commit(b *batch) []Effect {
 	return append(out, b.tasks...)
 }
 
+// 开播了一首新歌时记给所有在线成员；正在播放的歌没变时，只记给这一批之前没有会话的成员。
+func (r *Room) listened(b *batch) (RecordListen, bool) {
+	c := r.current
+	if c == nil {
+		return RecordListen{}, false
+	}
+	started := b.before.Current == nil || b.before.Current.Entry.ID != c.Entry.ID
+	var members []string
+	for _, s := range r.sessions {
+		id := s.Member.ID
+		if slices.Contains(members, id) || !started && slices.ContainsFunc(b.sessions, func(old Session) bool { return old.Member.ID == id }) {
+			continue
+		}
+		members = append(members, id)
+	}
+	if len(members) == 0 {
+		return RecordListen{}, false
+	}
+	return RecordListen{Members: members, Entry: c.Entry, StartedAt: c.StartedAt}, true
+}
+
 func (r *Room) timer(b *batch, in Timer) {
 	switch in.Token {
 	// 没有定时器时 endTimer 和 closeTimer 都是 0。
diff --git a/internal/domain/room/testdata/autodj.txtar b/internal/domain/room/testdata/autodj.txtar
index 0b49738..46caf54 100644
--- a/internal/domain/room/testdata/autodj.txtar
+++ b/internal/domain/room/testdata/autodj.txtar
@@ -15,6 +15,7 @@ prepare netease:a
 -- 04 --
 > 20s prepared netease:a 3m
 persist mode=sequential entrySeq=1 current=#1(netease:a,ready)@20.2s pending=- waiting=[]
+listen [m1] #1(netease:a,ready)@20.2s
 emit 1.4 queue pending=- waiting=[]
 emit 1.5 playback #1(netease:a,ready)@20.2s
 schedule 2 3m20.2s
@@ -33,6 +34,7 @@ emit 1.8 queue pending=- waiting=[#2(netease:b,ready,auto)]
 > 3m20.2s timer 2
 persist mode=sequential entrySeq=2 current=#2(netease:b,ready,auto)@3m20.4s pending=- waiting=[]
 record #1(netease:a,ready) 20.2s-3m20.2s
+listen [m1] #2(netease:b,ready,auto)@3m20.4s
 emit 1.9 queue pending=- waiting=[]
 emit 1.10 playback #2(netease:b,ready,auto)@3m20.4s
 schedule 4 5m20.4s
diff --git a/internal/domain/room/testdata/listen.txtar b/internal/domain/room/testdata/listen.txtar
new file mode 100644
index 0000000..14ff8f3
--- /dev/null
+++ b/internal/domain/room/testdata/listen.txtar
@@ -0,0 +1,73 @@
+听歌记录：开播时记给所有在线成员，同一成员的几个会话只记一次；没有在播的歌时进入不记；中途进入的成员只记自己；
+同一成员的第二台设备上线不记；断线重连时新会话先于旧会话的离开到达，不记；离开之后再进入，按同一个开播时间再记一次。
+-- 01 --
+> restore persistent
+-- 02 --
+> 0s join s1 m1
+emit 1.1 members [s1=m1]
+-- 03 --
+> 0s join s2 m1
+emit 1.2 members [s1=m1 s2=m1]
+-- 04 --
+> 1s enqueue m1 netease:a
+persist mode=sequential entrySeq=1 current=- pending=#1(netease:a) waiting=[]
+emit 1.3 activity added m1 #1(netease:a)
+emit 1.4 queue pending=#1(netease:a) waiting=[]
+reply q04 enqueued #1(netease:a)
+prepare netease:a
+-- 05 --
+> 2s enqueue m2 netease:b
+persist mode=sequential entrySeq=2 current=- pending=#1(netease:a) waiting=[#2(netease:b)]
+emit 1.5 activity added m2 #2(netease:b)
+emit 1.6 queue pending=#1(netease:a) waiting=[#2(netease:b)]
+reply q05 enqueued #2(netease:b)
+prepare netease:b
+-- 06 --
+> 3s prepared netease:b 1m
+persist mode=sequential entrySeq=2 current=- pending=#1(netease:a) waiting=[#2(netease:b,ready)]
+emit 1.7 queue pending=#1(netease:a) waiting=[#2(netease:b,ready)]
+-- 07 --
+> 4s prepared netease:a 2m
+persist mode=sequential entrySeq=2 current=#1(netease:a,ready)@4.2s pending=- waiting=[#2(netease:b,ready)]
+listen [m1] #1(netease:a,ready)@4.2s
+emit 1.8 queue pending=- waiting=[#2(netease:b,ready)]
+emit 1.9 playback #1(netease:a,ready)@4.2s
+schedule 1 2m4.2s
+-- 08 --
+> 10s join s3 m2
+listen [m2] #1(netease:a,ready)@4.2s
+emit 1.10 members [s1=m1 s2=m1 s3=m2]
+-- 09 --
+> 11s join s4 m2
+emit 1.11 members [s1=m1 s2=m1 s3=m2 s4=m2]
+-- 10 --
+> 12s join s5 m3
+listen [m3] #1(netease:a,ready)@4.2s
+emit 1.12 members [s1=m1 s2=m1 s3=m2 s4=m2 s5=m3]
+-- 11 --
+> 13s join s6 m3
+emit 1.13 members [s1=m1 s2=m1 s3=m2 s4=m2 s5=m3 s6=m3]
+-- 12 --
+> 14s leave s5
+emit 1.14 members [s1=m1 s2=m1 s3=m2 s4=m2 s6=m3]
+-- 13 --
+> 15s leave s6
+emit 1.15 members [s1=m1 s2=m1 s3=m2 s4=m2]
+-- 14 --
+> 16s join s7 m3
+listen [m3] #1(netease:a,ready)@4.2s
+emit 1.16 members [s1=m1 s2=m1 s3=m2 s4=m2 s7=m3]
+-- 15 --
+> 2m4.2s timer 1
+persist mode=sequential entrySeq=2 current=#2(netease:b,ready)@2m4.4s pending=- waiting=[]
+record #1(netease:a,ready) 4.2s-2m4.2s
+listen [m1 m2 m3] #2(netease:b,ready)@2m4.4s
+emit 1.17 queue pending=- waiting=[]
+emit 1.18 playback #2(netease:b,ready)@2m4.4s
+schedule 2 3m4.4s
+recommend 3
+-- 16 --
+> 3m4.4s timer 2
+persist mode=sequential entrySeq=2 current=- pending=- waiting=[]
+record #2(netease:b,ready) 2m4.4s-3m4.4s
+emit 1.19 playback -
diff --git a/internal/domain/room/testdata/sequential.txtar b/internal/domain/room/testdata/sequential.txtar
index 268767e..e55b7f3 100644
--- a/internal/domain/room/testdata/sequential.txtar
+++ b/internal/domain/room/testdata/sequential.txtar
@@ -25,6 +25,7 @@ emit 1.6 queue pending=#1(netease:a) waiting=[#2(netease:b,ready)]
 -- 06 --
 > 4s prepared netease:a 2m
 persist mode=sequential entrySeq=2 current=#1(netease:a,ready)@4.2s pending=- waiting=[#2(netease:b,ready)]
+listen [m1] #1(netease:a,ready)@4.2s
 emit 1.7 queue pending=- waiting=[#2(netease:b,ready)]
 emit 1.8 playback #1(netease:a,ready)@4.2s
 schedule 1 2m4.2s
@@ -32,6 +33,7 @@ schedule 1 2m4.2s
 > 2m4.2s timer 1
 persist mode=sequential entrySeq=2 current=#2(netease:b,ready)@2m4.4s pending=- waiting=[]
 record #1(netease:a,ready) 4.2s-2m4.2s
+listen [m1] #2(netease:b,ready)@2m4.4s
 emit 1.9 queue pending=- waiting=[]
 emit 1.10 playback #2(netease:b,ready)@2m4.4s
 schedule 2 3m4.4s
diff --git a/internal/domain/room/testdata/vote.txtar b/internal/domain/room/testdata/vote.txtar
index 5022935..e8cb19a 100644
--- a/internal/domain/room/testdata/vote.txtar
+++ b/internal/domain/room/testdata/vote.txtar
@@ -23,6 +23,7 @@ prepare netease:a
 -- 07 --
 > 1s prepared netease:a 3m
 persist mode=sequential entrySeq=1 current=#1(netease:a,ready)@1.2s pending=- waiting=[]
+listen [m1 m2 m3] #1(netease:a,ready)@1.2s
 emit 1.7 queue pending=- waiting=[]
 emit 1.8 playback #1(netease:a,ready)@1.2s
 schedule 1 3m1.2s