hyapp-server/services/room-service/internal/room/service/service.go

// Package service 承载 room-service 的 Room Cell 管理、命令流水线、恢复、快照和 outbox 补偿。
package service

import (
	"errors"
	"sync"
	"sync/atomic"
	"time"

	roomv1 "hyapp.local/api/proto/room/v1"
	"hyapp/pkg/clock"
	"hyapp/pkg/tencentim"
	"hyapp/services/room-service/internal/integration"
	"hyapp/services/room-service/internal/room/cell"
	"hyapp/services/room-service/internal/room/outbox"
	"hyapp/services/room-service/internal/router"
)

// Config 收敛 room-service 领域服务的运行参数。
type Config struct {
	// NodeID 是当前 room-service 实例标识，Redis lease 以它作为房间 owner。
	NodeID string
	// LeaseTTL 是房间 owner 租约有效期，过期后其他节点可以接管并恢复房间。
	LeaseTTL time.Duration
	// RankLimit 是房间本地礼物榜保留长度。
	RankLimit int
	// SnapshotEveryN 表示每 N 个房间版本保存一次快照。
	SnapshotEveryN int64
	// PresenceStaleAfter 控制业务 presence 多久未刷新后由后台 worker 清理。
	PresenceStaleAfter time.Duration
	// MicPublishTimeout 控制 MicUp 后等待 RTC 发流确认的最长时间。
	MicPublishTimeout time.Duration
	// Clock 允许测试注入稳定时间，生产默认使用系统时钟。
	Clock clock.Clock
}

// Service 承载 room-service 的核心房间领域逻辑。
type Service struct {
	// nodeID 是当前实例 ID，必须和 Redis lease 中的 owner 对齐。
	nodeID string
	// leaseTTL 控制当前节点持有房间执行权的时间窗口。
	leaseTTL time.Duration
	// rankLimit 控制每个 Room Cell 的 LocalRank top N。
	rankLimit int
	// snapshotEveryN 控制快照频率，值越小恢复越快但写放大越高。
	snapshotEveryN int64
	// presenceStaleAfter 是断线或无心跳用户从 room-service presence 中移除的阈值。
	presenceStaleAfter time.Duration
	// micPublishTimeout 是 pending_publish 自动下麦的确认窗口。
	micPublishTimeout time.Duration
	// clock 是房间命令、事件和快照的统一时间来源。
	clock clock.Clock
	// directory 负责 room_id -> node_id lease，生产路径是 Redis。
	directory router.Directory
	// repository 是 MySQL 持久化边界，保存 meta、snapshot、command log 和 outbox。
	repository Repository
	// wallet 是 SendGift 的同步扣费依赖，扣费失败不能进入 Room Cell 状态变更。
	wallet integration.WalletClient
	// syncPublisher 是 room-service 到腾讯云 IM 的低时延房间系统事件桥。
	syncPublisher integration.RoomEventPublisher
	// outboxPublisher 是补偿 worker 对外部消费者的异步投递抽象。
	outboxPublisher integration.OutboxPublisher
	// eventPublisher 是命令提交后的房间事实消费者投递通道，不参与 Room Cell 提交回滚。
	eventPublisher integration.OutboxPublisher
	// draining 表示当前节点正在下线，只允许已经进入执行链路的命令收尾。
	draining atomic.Bool

	// mu 保护本进程内 room_id -> RoomCell 注册表。
	mu sync.Mutex
	// cells 保存当前节点已经装载的单房间执行单元。
	cells map[string]*cell.RoomCell
}

// mutationResult 是命令在 Room Cell 内执行后的统一结果包。
type mutationResult struct {
	// applied 表示本次命令确实改变并提交了房间状态。
	applied bool
	// snapshot 是命令执行后对外返回的房间快照。
	snapshot *roomv1.RoomSnapshot
	// user 是 JoinRoom 等命令需要返回的用户投影。
	user *roomv1.RoomUser
	// seatNo 是麦位命令返回的目标或原麦位编号。
	seatNo int32
	// micSessionID 是 MicUp 创建的发流会话 ID，客户端确认发流时必须带回。
	micSessionID string
	// publishDeadlineMS 是 pending_publish 自动释放的截止时间。
	publishDeadlineMS int64
	// billingReceiptID 是 SendGift 从 wallet-service 得到的账务回执。
	billingReceiptID string
	// roomHeat 是 SendGift 后的房间热度。
	roomHeat int64
	// giftRank 是 SendGift 后的本地礼物榜投影。
	giftRank []*roomv1.RankItem
	// commandPayload 允许少数命令把外部依赖的结算快照写入 command log，用于恢复。
	commandPayload []byte
	// syncEvent 是需要同步推给腾讯云 IM 的低时延房间系统事件。
	syncEvent *tencentim.RoomEvent
	// outboxRecords 是本次命令已经持久化的房间事实，提交后可投给 activity 等外部消费者。
	outboxRecords []outbox.Record
	// roomStatus 非空时代表命令提交时需要同步更新 rooms 和列表投影生命周期状态。
	roomStatus string
	// roomSeatCount 非 nil 时代表命令提交时需要同步更新 rooms 元数据座位数。
	roomSeatCount *int32
}

// New 初始化 room-service 领域服务。
func New(cfg Config, directory router.Directory, repository Repository, wallet integration.WalletClient, syncPublisher integration.RoomEventPublisher, outboxPublisher integration.OutboxPublisher) *Service {
	usedClock := cfg.Clock
	if usedClock == nil {
		// 生产默认系统时钟，测试可注入固定时钟保证断言稳定。
		usedClock = clock.System{}
	}

	snapshotEveryN := cfg.SnapshotEveryN
	if snapshotEveryN <= 0 {
		// 默认每次状态变更都保存快照，优先保证首版恢复简单。
		snapshotEveryN = 1
	}

	rankLimit := cfg.RankLimit
	if rankLimit <= 0 {
		// 默认只保留房间内前 20 名礼物榜，避免快照和广播载荷过大。
		rankLimit = 20
	}

	presenceStaleAfter := cfg.PresenceStaleAfter
	if presenceStaleAfter <= 0 {
		// 默认 2 分钟清理无刷新 presence，避免重连窗口过短导致误踢。
		presenceStaleAfter = 2 * time.Minute
	}

	micPublishTimeout := cfg.MicPublishTimeout
	if micPublishTimeout <= 0 {
		// MicUp 只是业务占麦，客户端必须在短窗口内确认 RTC 发流成功。
		micPublishTimeout = 15 * time.Second
	}

	return &Service{
		nodeID:             cfg.NodeID,
		leaseTTL:           cfg.LeaseTTL,
		rankLimit:          rankLimit,
		snapshotEveryN:     snapshotEveryN,
		presenceStaleAfter: presenceStaleAfter,
		micPublishTimeout:  micPublishTimeout,
		clock:              usedClock,
		directory:          directory,
		repository:         repository,
		wallet:             wallet,
		syncPublisher:      syncPublisher,
		outboxPublisher:    outboxPublisher,
		cells:              make(map[string]*cell.RoomCell),
	}
}

// SetEventPublisher 注入 room outbox 事实消费者；失败只记录日志，不改变房间主链路。
func (s *Service) SetEventPublisher(publisher integration.OutboxPublisher) {
	if s != nil {
		s.eventPublisher = publisher
	}
}

// HealthCheck 校验 room-service 核心状态 owner 和持久化依赖已经完成装配。
func (s *Service) HealthCheck() error {
	if s == nil {
		// nil service 说明 App 装配失败，live/ready 都不能放行。
		return errors.New("room service is not initialized")
	}
	if s.directory == nil {
		// 没有 directory 就无法保证单房间单活 owner。
		return errors.New("room directory is not initialized")
	}
	if s.repository == nil {
		// 没有 repository 就无法保证 command log、snapshot 和 outbox 持久化。
		return errors.New("room repository is not initialized")
	}
	if s.cells == nil {
		// cells 注册表缺失会导致已恢复房间无法复用同一个 Room Cell。
		return errors.New("room cell registry is not initialized")
	}

	return nil
}

// MarkDraining 让领域服务停止接收新的 Room Cell 命令和新 lease 接管。
func (s *Service) MarkDraining() {
	if s == nil {
		return
	}
	s.draining.Store(true)
}

func (s *Service) isDraining() bool {
	return s != nil && s.draining.Load()
}