feat: write listening history as songs start and members join

internal/adapter/in/api/api_test.go

@@ -39,7 +39,7 @@ const allowedOrigin = "https://allowed.example"
 func services(store *porttest.Store, rooms port.RoomDirectory, adminToken string) Services {
 	music := newCatalog(rooms, 10)
 	node := porttest.NewRedis().Node("a")
-	_, gateway := realtime.New(store, rooms, store, music, porttest.NewPreparer(), realtime.Cluster{Instance: "a", Leases: node, Liveness: node, Bus: node}, slog.New(slog.DiscardHandler))
+	_, gateway := realtime.New(store, rooms, store, store, music, porttest.NewPreparer(), realtime.Cluster{Instance: "a", Leases: node, Liveness: node, Bus: node}, slog.New(slog.DiscardHandler))
 	l := lobby.New(rooms, node)
 	return Services{
 		Identity: identity.New(store, rooms, store, gateway),

internal/adapter/in/api/room_test.go

@@ -43,7 +43,7 @@ func newRoomFixture(t *testing.T) roomFixture {
 	store := porttest.New()
 	music := newCatalog(store, 10)
 	node := porttest.NewRedis().Node("a")
-	host, gateway := realtime.New(store, store, store, music, porttest.NewPreparer(), realtime.Cluster{Instance: "a", Leases: node, Liveness: node, Bus: node}, slog.New(slog.DiscardHandler))
+	host, gateway := realtime.New(store, store, store, store, music, porttest.NewPreparer(), realtime.Cluster{Instance: "a", Leases: node, Liveness: node, Bus: node}, slog.New(slog.DiscardHandler))
 	ctx, stop := context.WithCancel(context.Background())
 	wait, err := realtime.Start(ctx, host, gateway)
 	if err != nil {

internal/adapter/in/api/ws_test.go

@@ -134,7 +134,7 @@ func newWSFixtureWithTickets(t *testing.T, store *porttest.Store, tickets port.T
 	music := newCatalog(directory, 10)
 	node := porttest.NewRedis().Node("a")
 	preparer := porttest.NewPreparer()
-	host, gateway := realtime.New(snapshots, store, store, music, preparer, realtime.Cluster{Instance: "a", Leases: node, Liveness: node, Bus: node}, slog.New(slog.DiscardHandler))
+	host, gateway := realtime.New(snapshots, store, store, store, music, preparer, realtime.Cluster{Instance: "a", Leases: node, Liveness: node, Bus: node}, slog.New(slog.DiscardHandler))
 	ctx, stop := context.WithCancel(context.Background())
 	wait, err := realtime.Start(ctx, host, gateway)
 	if err != nil {

internal/app/admin/admin_test.go

@@ -34,7 +34,7 @@ func newService(t *testing.T, token string) fixture {
 	t.Helper()
 	store := porttest.New()
 	node := porttest.NewRedis().Node("a")
-	host, gateway := realtime.New(store, store, store, nil, porttest.NewPreparer(), realtime.Cluster{Instance: "a", Leases: node, Liveness: node, Bus: node}, slog.New(slog.DiscardHandler))
+	host, gateway := realtime.New(store, store, store, store, nil, porttest.NewPreparer(), realtime.Cluster{Instance: "a", Leases: node, Liveness: node, Bus: node}, slog.New(slog.DiscardHandler))
 	ctx, stop := context.WithCancel(context.Background())
 	wait, err := realtime.Start(ctx, host, gateway)
 	if err != nil {

internal/app/identity/identity_test.go

@@ -28,7 +28,7 @@ func newService(t *testing.T) (*Service, *porttest.Store) {
 
 // 没有调用 realtime.Start：这些测试不运行房间。
 func newGateway(store *porttest.Store, node *porttest.Node, bus port.Bus) *realtime.Gateway {
-	_, gateway := realtime.New(store, store, store, nil, porttest.NewPreparer(), realtime.Cluster{Instance: "a", Leases: node, Liveness: node, Bus: bus}, slog.New(slog.DiscardHandler))
+	_, gateway := realtime.New(store, store, store, store, nil, porttest.NewPreparer(), realtime.Cluster{Instance: "a", Leases: node, Liveness: node, Bus: bus}, slog.New(slog.DiscardHandler))
 	return gateway
 }
 
@@ -206,7 +206,7 @@ func TestEnterRoomLeavesWhenTicketIsRevokedWhileJoining(t *testing.T) {
 	synctest.Test(t, func(t *testing.T) {
 		store := porttest.New()
 		node := porttest.NewRedis().Node("a")
-		host, gateway := realtime.New(store, store, store, nil, porttest.NewPreparer(), realtime.Cluster{Instance: "a", Leases: node, Liveness: node, Bus: node}, slog.New(slog.DiscardHandler))
+		host, gateway := realtime.New(store, store, store, store, nil, porttest.NewPreparer(), realtime.Cluster{Instance: "a", Leases: node, Liveness: node, Bus: node}, slog.New(slog.DiscardHandler))
 		ctx, stop := context.WithCancel(t.Context())
 		wait, err := realtime.Start(ctx, host, gateway)
 		if err != nil {

internal/app/lobby/lobby_test.go

@@ -14,7 +14,7 @@ import (
 func newService(t *testing.T, store *porttest.Store) (*Service, *realtime.Gateway) {
 	t.Helper()
 	node := porttest.NewRedis().Node("a")
-	host, gateway := realtime.New(store, store, store, nil, porttest.NewPreparer(), realtime.Cluster{Instance: "a", Leases: node, Liveness: node, Bus: node}, slog.New(slog.DiscardHandler))
+	host, gateway := realtime.New(store, store, store, store, nil, porttest.NewPreparer(), realtime.Cluster{Instance: "a", Leases: node, Liveness: node, Bus: node}, slog.New(slog.DiscardHandler))
 	ctx, stop := context.WithCancel(context.Background())
 	wait, err := realtime.Start(ctx, host, gateway)
 	if err != nil {

internal/app/realtime/cluster_test.go

@@ -169,7 +169,7 @@ func newFlakyBusInstance(t *testing.T, store *porttest.Store, logger *slog.Logge
 	songs := newSongs()
 	node := porttest.NewRedis().Node("a")
 	bus := &flakyBus{Bus: node}
-	h, g := New(store, store, unrevoked{}, songs.music, songs.preparer, Cluster{Instance: "a", Leases: node, Liveness: node, Bus: bus}, logger)
+	h, g := New(store, store, store, unrevoked{}, songs.music, songs.preparer, Cluster{Instance: "a", Leases: node, Liveness: node, Bus: bus}, logger)
 	return h, g, node, bus
 }
 
@@ -381,7 +381,7 @@ func TestGatewayRejoinsAfterItsAliveKeyExpired(t *testing.T) {
 		a := startInstance(t, redis, "a", store, store, songs)
 		node := redis.Node("b")
 		liveness := &flakyLiveness{Liveness: node}
-		h, g := New(store, store, unrevoked{}, songs.music, songs.preparer, Cluster{Instance: "b", Leases: node, Liveness: liveness, Bus: node}, slog.New(slog.DiscardHandler))
+		h, g := New(store, store, store, unrevoked{}, songs.music, songs.preparer, Cluster{Instance: "b", Leases: node, Liveness: liveness, Bus: node}, slog.New(slog.DiscardHandler))
 		ctx, stop := context.WithCancel(context.Background())
 		wait, err := Start(ctx, h, g)
 		if err != nil {
@@ -457,7 +457,7 @@ func startFlakyInstance(t *testing.T, redis *porttest.Redis, id string, snapshot
 	t.Helper()
 	node := redis.Node(id)
 	leases, bus := &flakyOwners{Leases: node}, &flakyForwards{Bus: node}
-	h, g := New(snapshots, store, unrevoked{}, songs.music, songs.preparer, Cluster{Instance: id, Leases: leases, Liveness: node, Bus: bus}, slog.New(slog.DiscardHandler))
+	h, g := New(snapshots, store, store, unrevoked{}, songs.music, songs.preparer, Cluster{Instance: id, Leases: leases, Liveness: node, Bus: bus}, slog.New(slog.DiscardHandler))
 	ctx, stop := context.WithCancel(context.Background())
 	wait, err := Start(ctx, h, g)
 	if err != nil {
@@ -722,7 +722,7 @@ func TestCommandFollowsOwnerAfterLosingTakeover(t *testing.T) {
 		node := f.redis.Node("c")
 		leases := &staleLeases{Leases: node}
 		songs := newSongs()
-		h, g := New(store, store, unrevoked{}, songs.music, songs.preparer, Cluster{Instance: "c", Leases: leases, Liveness: node, Bus: node}, slog.New(slog.DiscardHandler))
+		h, g := New(store, store, store, unrevoked{}, songs.music, songs.preparer, Cluster{Instance: "c", Leases: leases, Liveness: node, Bus: node}, slog.New(slog.DiscardHandler))
 		ctx, stop := context.WithCancel(context.Background())
 		wait, err := Start(ctx, h, g)
 		if err != nil {

internal/app/realtime/context_test.go

@@ -76,7 +76,7 @@ func TestPreparationKeepsRequestValues(t *testing.T) {
 		node := redis.Node("a")
 		preparer := contextPreparer{started: make(chan context.Context, 1)}
 		f := startHost(t, &hostFixture{store: store, redis: redis, node: node, spy: spy{redis.Node("spy")}},
-			newHost(store, store, preparer, newSongs().music, cluster(node, "a"), slog.New(slog.DiscardHandler)))
+			newHost(store, store, store, preparer, newSongs().music, cluster(node, "a"), slog.New(slog.DiscardHandler)))
 		id := f.createRoom(t, room.Persistent)
 
 		ctx, cancel := context.WithCancel(context.WithValue(context.Background(), requestValue{}, "点歌"))

internal/app/realtime/gateway.go

@@ -87,8 +87,8 @@ type roomLink struct {
 	confirmed uint64
 }
 
-func New(snapshots port.RoomSnapshots, directory port.RoomDirectory, tickets port.ExistingTickets, music *catalog.Service, preparer port.Preparer, cluster Cluster, logger *slog.Logger) (*Host, *Gateway) {
-	h := newHost(snapshots, directory, preparer, music, cluster, logger)
+func New(snapshots port.RoomSnapshots, directory port.RoomDirectory, listens port.ListenRecorder, tickets port.ExistingTickets, music *catalog.Service, preparer port.Preparer, cluster Cluster, logger *slog.Logger) (*Host, *Gateway) {
+	h := newHost(snapshots, directory, listens, preparer, music, cluster, logger)
 	g := &Gateway{
 		self:      cluster.Instance,
 		host:      h,

internal/app/realtime/gateway_test.go

@@ -43,7 +43,7 @@ func startInstance(t *testing.T, redis *porttest.Redis, id string, snapshots por
 func startLoggingInstance(t *testing.T, redis *porttest.Redis, id string, snapshots port.RoomSnapshots, store *porttest.Store, tickets port.ExistingTickets, songs *songs, logger *slog.Logger) *instance {
 	t.Helper()
 	node := redis.Node(id)
-	h, g := New(snapshots, store, tickets, songs.music, songs.preparer, cluster(node, id), logger)
+	h, g := New(snapshots, store, store, tickets, songs.music, songs.preparer, cluster(node, id), logger)
 	ctx, stop := context.WithCancel(context.Background())
 	wait, err := Start(ctx, h, g)
 	if err != nil {
@@ -603,7 +603,7 @@ func startCountingGateway(t *testing.T, store *porttest.Store) (*Gateway, *portt
 	songs := newSongs()
 	c := cluster(node, "a")
 	c.Bus = bus
-	h, g := New(store, store, unrevoked{}, songs.music, songs.preparer, c, slog.New(slog.DiscardHandler))
+	h, g := New(store, store, store, unrevoked{}, songs.music, songs.preparer, c, slog.New(slog.DiscardHandler))
 	ctx, stop := context.WithCancel(context.Background())
 	wait, err := Start(ctx, h, g)
 	if err != nil {

internal/app/realtime/host.go

@@ -49,6 +49,7 @@ type Host struct {
 	self      string
 	snapshots port.RoomSnapshots
 	directory port.RoomDirectory
+	listens   port.ListenRecorder
 	preparer  port.Preparer
 	catalog   *catalog.Service
 	leases    port.Leases
@@ -120,11 +121,12 @@ type hostedRoom struct {
 	timers      map[room.Token]*time.Timer
 }
 
-func newHost(snapshots port.RoomSnapshots, directory port.RoomDirectory, preparer port.Preparer, music *catalog.Service, cluster Cluster, logger *slog.Logger) *Host {
+func newHost(snapshots port.RoomSnapshots, directory port.RoomDirectory, listens port.ListenRecorder, preparer port.Preparer, music *catalog.Service, cluster Cluster, logger *slog.Logger) *Host {
 	return &Host{
 		self:       cluster.Instance,
 		snapshots:  snapshots,
 		directory:  directory,
+		listens:    listens,
 		preparer:   preparer,
 		catalog:    music,
 		leases:     cluster.Leases,
@@ -592,6 +594,8 @@ func (h *Host) execute(ctx context.Context, r *hostedRoom, effects []room.Effect
 			r.info = e.Snapshot.Info
 		case room.RecordPlay:
 			// 已经和 Persist 在同一个事务中写入。
+		case room.RecordListen:
+			h.recordListen(ctx, e)
 		case room.Emit:
 			if m, ok := e.Event.(room.MembersChanged); ok && !h.setOnline(ctx, r, len(m.Sessions)) {
 				return false
@@ -679,6 +683,16 @@ func (h *Host) persist(ctx context.Context, r *hostedRoom, s room.Snapshot, play
 	return h.snapshots.SaveRoom(ctx, r.term, s, plays)
 }
 
+// 播放历史不是房间的状态：写入失败不放弃房间。与落盘一样不随 ctx 取消。
+func (h *Host) recordListen(ctx context.Context, l room.RecordListen) {
+	writeCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), storeTimeout)
+	defer cancel()
+	listen := port.Listen{Track: l.Entry.Track, Meta: l.Entry.Meta, HeardAt: l.StartedAt}
+	if err := h.listens.RecordListen(writeCtx, l.Members, listen); err != nil {
+		h.logger.WarnContext(ctx, "写入播放历史失败", "error", err)
+	}
+}
+
 func (h *Host) remove(ctx context.Context, r *hostedRoom) error {
 	ctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), storeTimeout)
 	defer cancel()

internal/app/realtime/host_test.go

@@ -161,7 +161,7 @@ func newHostFixture(t *testing.T, snapshots port.RoomSnapshots, store *porttest.
 	redis := porttest.NewRedis()
 	node := redis.Node("a")
 	return startHost(t, &hostFixture{store: store, redis: redis, node: node, spy: spy{redis.Node("spy")}},
-		newHost(snapshots, store, songs.preparer, songs.music, cluster(node, "a"), slog.New(slog.DiscardHandler)))
+		newHost(snapshots, store, store, songs.preparer, songs.music, cluster(node, "a"), slog.New(slog.DiscardHandler)))
 }
 
 func startHost(t *testing.T, f *hostFixture, h *Host) *hostFixture {
@@ -289,6 +289,128 @@ func TestPersistFailureAbandonsRoom(t *testing.T) {
 	})
 }
 
+// 写入时把此前发布的房间消息和这次写入按发生的先后记进同一个日志。
+type orderedListens struct {
+	*porttest.Store
+	spy  spy
+	fail bool
+
+	mu  sync.Mutex
+	log []string
+}
+
+func (l *orderedListens) RecordListen(ctx context.Context, memberIDs []string, listen port.Listen) error {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	for _, m := range l.spy.take() {
+		l.log = append(l.log, describe(m))
+	}
+	l.log = append(l.log, fmt.Sprintf("listen %d %s", len(memberIDs), listen.Track.ID))
+	if l.fail {
+		return errors.New("连接中断")
+	}
+	return l.Store.RecordListen(ctx, memberIDs, listen)
+}
+
+func (l *orderedListens) messages() []string {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	for _, m := range l.spy.take() {
+		l.log = append(l.log, describe(m))
+	}
+	log := l.log
+	l.log = nil
+	return log
+}
+
+func newListenFixture(t *testing.T, fail bool, logger *slog.Logger) (*hostFixture, *orderedListens) {
+	t.Helper()
+	store := porttest.New()
+	songs := newSongs()
+	songs.add(sunny, sunnyM, sunnyP)
+	redis := porttest.NewRedis()
+	node := redis.Node("a")
+	listens := &orderedListens{Store: store, spy: spy{redis.Node("spy")}, fail: fail}
+	f := startHost(t, &hostFixture{store: store, redis: redis, node: node, spy: listens.spy},
+		newHost(store, store, listens, songs.preparer, songs.music, cluster(node, "a"), logger))
+	return f, listens
+}
+
+func createMembers(t *testing.T, store *porttest.Store, names ...string) []member.Member {
+	t.Helper()
+	members := make([]member.Member, len(names))
+	for i, name := range names {
+		m, err := store.CreateMember(t.Context(), []byte(name), name)
+		if err != nil {
+			t.Fatal(err)
+		}
+		members[i] = m
+	}
+	return members
+}
+
+func checkLog(t *testing.T, got []string, want ...string) {
+	t.Helper()
+	if !slices.Equal(got, want) {
+		t.Errorf("房间消息和写入的先后：\n%s\n期望：\n%s", strings.Join(got, "\n"), strings.Join(want, "\n"))
+	}
+}
+
+func TestListensAreWrittenBeforeTheBroadcast(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		f, listens := newListenFixture(t, false, slog.New(slog.DiscardHandler))
+		members := createMembers(t, f.store, "alice", "bob")
+		a, b := members[0], members[1]
+		id := f.createRoom(t, room.Persistent)
+		start := time.Now()
+
+		f.send(t, id, room.Join{Session: "s1", Member: a})
+		f.send(t, id, command("r1", a, room.Enqueue{Track: sunny, Meta: sunnyM}))
+		checkLog(t, listens.messages(),
+			"started 1",
+			"1/1 room.MembersChanged",
+			"1/2 room.Activity",
+			"1/3 room.QueueChanged",
+			"reply r1 room.EnqueueResult",
+			"listen 1 186016",
+			"1/4 room.QueueChanged",
+			"1/5 room.PlaybackChanged",
+		)
+		f.send(t, id, room.Join{Session: "s2", Member: b})
+		checkLog(t, listens.messages(), "listen 1 186016", "1/6 room.MembersChanged")
+
+		startedAt := start.Add(200 * time.Millisecond)
+		for _, m := range members {
+			got, err := f.store.MemberListens(t.Context(), m.ID)
+			if err != nil || len(got) != 1 || got[0].Track != sunny || got[0].Meta.Title != sunnyM.Title || !got[0].HeardAt.Equal(startedAt) {
+				t.Errorf("%s 的历史 = %+v, %v，期望《晴天》，时间是开播的 %v", m.Name, got, err, startedAt)
+			}
+		}
+	})
+}
+
+func TestListenFailureKeepsRoomRunning(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		var logs logBuffer
+		f, listens := newListenFixture(t, true, slog.New(slog.NewJSONHandler(&logs, nil)))
+		a := createMembers(t, f.store, "alice")[0]
+		id := f.createRoom(t, room.Persistent)
+
+		f.send(t, id, room.Join{Session: "s1", Member: a})
+		f.send(t, id, command("r1", a, room.Enqueue{Track: sunny, Meta: sunnyM}))
+		if got := listens.messages(); !slices.Contains(got, "1/5 room.PlaybackChanged") {
+			t.Errorf("写入失败之后的房间消息 = %v，期望照常广播开播", got)
+		}
+		if owner := f.owner(t, id); owner != "a" {
+			t.Errorf("写入失败之后房间的 owner = %q，期望不放弃房间", owner)
+		}
+		entries := logs.entries(t, "写入播放历史失败")
+		if len(entries) != 1 || entries[0]["level"] != "WARN" {
+			t.Errorf("写入失败的日志 = %v，期望一条 WARN", entries)
+		}
+	})
+}
+
 // 每次 Restore 的 token 都从 1 开始：放弃房间时如果不停下计时器，它会提前关闭重新接管后的房间。
 func TestTimersOfAbandonedRoomAreDiscarded(t *testing.T) {
 	synctest.Test(t, func(t *testing.T) {
@@ -341,7 +463,7 @@ func TestPrepareOfAbandonedRunIsDiscarded(t *testing.T) {
 		redis := porttest.NewRedis()
 		node := redis.Node("a")
 		f := startHost(t, &hostFixture{store: store.Store, redis: redis, node: node, spy: spy{redis.Node("spy")}},
-			newHost(store, store.Store, preparer, newSongs().music, cluster(node, "a"), slog.New(slog.DiscardHandler)))
+			newHost(store, store.Store, store.Store, preparer, newSongs().music, cluster(node, "a"), slog.New(slog.DiscardHandler)))
 		id := f.createRoom(t, room.Persistent)
 		f.send(t, id, command("r1", alice, room.Enqueue{Track: sunny, Meta: sunnyM}))
 		f.spy.take()
@@ -461,7 +583,7 @@ func TestHostAcceptsRoomsOnlyBetweenStartAndShutdown(t *testing.T) {
 		songs := newSongs()
 		redis := porttest.NewRedis()
 		watcher := spy{redis.Node("spy")}
-		h := newHost(store, store, songs.preparer, songs.music, cluster(redis.Node("a"), "a"), slog.New(slog.DiscardHandler))
+		h := newHost(store, store, store, songs.preparer, songs.music, cluster(redis.Node("a"), "a"), slog.New(slog.DiscardHandler))
 		id, err := store.CreateRoom(t.Context(), port.NewRoom{Kind: room.Persistent, Name: "房间"})
 		if err != nil {
 			t.Fatal(err)
@@ -673,7 +795,7 @@ func newFlakyFixture(t *testing.T) (*hostFixture, *flakyLeases) {
 	node := redis.Node("a")
 	leases := &flakyLeases{Leases: node, resume: make(chan struct{})}
 	f := startHost(t, &hostFixture{store: store, redis: redis, node: node, spy: spy{redis.Node("spy")}},
-		newHost(store, store, songs.preparer, songs.music, Cluster{Instance: "a", Leases: leases, Liveness: node, Bus: node}, slog.New(slog.DiscardHandler)))
+		newHost(store, store, store, songs.preparer, songs.music, Cluster{Instance: "a", Leases: leases, Liveness: node, Bus: node}, slog.New(slog.DiscardHandler)))
 	return f, leases
 }
 
@@ -876,7 +998,7 @@ func TestFailedReleaseNoticeIsLogged(t *testing.T) {
 		node := redis.Node("a")
 		var logs logBuffer
 		f := startHost(t, &hostFixture{store: store, redis: redis, node: node, spy: spy{redis.Node("spy")}},
-			newHost(store, store, songs.preparer, songs.music, Cluster{Instance: "a", Leases: node, Liveness: node, Bus: noReleaseNotice{node}}, slog.New(logctx.NewHandler(slog.NewJSONHandler(&logs, nil)))))
+			newHost(store, store, store, songs.preparer, songs.music, Cluster{Instance: "a", Leases: node, Liveness: node, Bus: noReleaseNotice{node}}, slog.New(logctx.NewHandler(slog.NewJSONHandler(&logs, nil)))))
 		id := f.createRoom(t, room.Persistent)
 		f.send(t, id, room.Join{Session: "s1", Member: alice})
 		if err := f.host.Shutdown(t.Context()); err != nil {
@@ -1000,7 +1122,7 @@ func TestTakeoverFailuresAreLoggedWhenTheyStart(t *testing.T) {
 		node := redis.Node("a")
 		var logs logBuffer
 		f := startHost(t, &hostFixture{store: store.Store, redis: redis, node: node, spy: spy{redis.Node("spy")}},
-			newHost(store, store, songs.preparer, songs.music, cluster(node, "a"), slog.New(slog.NewJSONHandler(&logs, nil))))
+			newHost(store, store, store, songs.preparer, songs.music, cluster(node, "a"), slog.New(slog.NewJSONHandler(&logs, nil))))
 		for range 3 {
 			f.createRoom(t, room.Persistent)
 		}
@@ -1087,7 +1209,7 @@ func TestRecoveredGatewayKeepsItsSessions(t *testing.T) {
 		gateway, unreachable := redis.Node("b"), redis.Node("c")
 		liveness := &recoveringLiveness{Liveness: node, gateway: gateway, unreachable: "c"}
 		f := startHost(t, &hostFixture{store: store, redis: redis, node: node, spy: spy{redis.Node("spy")}},
-			newHost(store, store, songs.preparer, songs.music, Cluster{Instance: "a", Leases: node, Liveness: liveness, Bus: node}, slog.New(slog.DiscardHandler)))
+			newHost(store, store, store, songs.preparer, songs.music, Cluster{Instance: "a", Leases: node, Liveness: liveness, Bus: node}, slog.New(slog.DiscardHandler)))
 		id := f.createRoom(t, room.Persistent)
 		for _, g := range []*porttest.Node{gateway, unreachable} {
 			if _, err := g.Heartbeat(t.Context(), aliveTTL); err != nil {

internal/app/realtime/observer_test.go

@@ -343,7 +343,7 @@ func TestRevocationNoticesAreBestEffort(t *testing.T) {
 	}
 	store := porttest.New()
 	var logs logBuffer
-	_, g := New(store, store, unrevoked{}, nil, porttest.NewPreparer(), Cluster{Instance: "a", Leases: node, Liveness: node, Bus: contextBus{node}}, slog.New(slog.NewJSONHandler(&logs, nil)))
+	_, g := New(store, store, store, unrevoked{}, nil, porttest.NewPreparer(), Cluster{Instance: "a", Leases: node, Liveness: node, Bus: contextBus{node}}, slog.New(slog.NewJSONHandler(&logs, nil)))
 	revokeAll := func(ctx context.Context) {
 		g.RevokeSessions(ctx, [][]byte{[]byte("s")})
 		g.Revoke(ctx, "r")
@@ -430,7 +430,7 @@ func TestObserverSkipsSnapshotOfEarlierRun(t *testing.T) {
 		redis := porttest.NewRedis()
 		node := redis.Node("a")
 		bus := &heldEmits{Bus: node, release: make(chan struct{})}
-		h, g := New(store, store, unrevoked{}, songs.music, songs.preparer, Cluster{Instance: "a", Leases: node, Liveness: node, Bus: bus}, slog.New(slog.DiscardHandler))
+		h, g := New(store, store, store, unrevoked{}, songs.music, songs.preparer, Cluster{Instance: "a", Leases: node, Liveness: node, Bus: bus}, slog.New(slog.DiscardHandler))
 		ctx, stop := context.WithCancel(context.Background())
 		wait, err := Start(ctx, h, g)
 		if err != nil {

internal/bootstrap/public.go

@@ -76,7 +76,7 @@ func newPublic(cfg serveConfig, pool *pgxpool.Pool, rdb *goredis.Client, instanc
 		Bus:      redis.NewBus(rdb, instance, version, metrics, tracer, base.With("component", "bus")),
 	}
 	runtimeLogger := base.With("component", "realtime")
-	host, gateway := realtime.New(measure(media.NewSnapshots(store, objects, runtimeLogger), metrics), store, store, music, preparer, cluster, runtimeLogger)
+	host, gateway := realtime.New(measure(media.NewSnapshots(store, objects, runtimeLogger), metrics), store, store, store, music, preparer, cluster, runtimeLogger)
 	registerRuntimeMetrics(metrics, host, gateway)
 	rooms := lobby.New(store, leases)
 	services := api.Services{