DiheMoe/alisten
1
0 Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

feat: tell which members hear each song from the room domain

Browse Source
This commit is contained in:
DiheChen committed 2026-11-11 00:04:01 +08:00
1 parent 038cbb206b
commit a12ab44ec8
8 files changed
+152 -8
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

internal/domain/room/effect.go
+17 -8
View File
@@ -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() }
internal/domain/room/fuzz_test.go
+30
View File
@@ -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
}
}
internal/domain/room/golden_test.go
+2
View File
@@ -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:
internal/domain/room/room.go
+25
View File
@@ -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。
internal/domain/room/testdata/autodj.txtar
Vendored
+2
View File
@@ -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
internal/domain/room/testdata/listen.txtar
Vendored
+73
View File
@@ -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 -
internal/domain/room/testdata/sequential.txtar
Vendored
+2
View File
@@ -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
internal/domain/room/testdata/vote.txtar
Vendored
+1
View File
@@ -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