110 lines
3.5 KiB
Go
110 lines
3.5 KiB
Go
package service
|
||
|
||
import (
|
||
"context"
|
||
"time"
|
||
|
||
"hyapp/pkg/xerr"
|
||
"hyapp/services/room-service/internal/room/command"
|
||
"hyapp/services/room-service/internal/room/outbox"
|
||
"hyapp/services/room-service/internal/room/rank"
|
||
"hyapp/services/room-service/internal/room/state"
|
||
)
|
||
|
||
func (s *Service) mutateRoom(ctx context.Context, cmd command.Command, replayable bool, mutate func(now time.Time, current *state.RoomState, localRank *rank.LocalRank) (mutationResult, []outbox.Record, error)) (mutationResult, error) {
|
||
if cmd.RoomID() == "" || cmd.ID() == "" || cmd.ActorUserID() == 0 {
|
||
// command meta 是幂等、路由、审计和权限判断的共同基础，缺一不可。
|
||
return mutationResult{}, xerr.New(xerr.InvalidArgument, "command meta is incomplete")
|
||
}
|
||
|
||
if seen, err := s.repository.HasCommand(ctx, cmd.RoomID(), cmd.ID()); err != nil {
|
||
return mutationResult{}, err
|
||
} else if seen {
|
||
// 幂等命令不重新执行 mutate，也不会再次调用 wallet 或写 outbox。
|
||
snapshot, err := s.currentSnapshot(ctx, cmd.RoomID())
|
||
if err != nil {
|
||
return mutationResult{}, err
|
||
}
|
||
|
||
return mutationResult{snapshot: snapshot}, nil
|
||
}
|
||
|
||
// 确保当前节点持有房间 lease，并在没有内存 Cell 时完成恢复。
|
||
roomCell, err := s.ensureCell(ctx, cmd.RoomID())
|
||
if err != nil {
|
||
return mutationResult{}, err
|
||
}
|
||
|
||
value, err := roomCell.Do(ctx, func(current *state.RoomState, currentRank *rank.LocalRank) (any, error) {
|
||
// 在副本上执行命令，只有 command log/outbox 保存成功后才整体替换当前状态。
|
||
nextState := current.Clone()
|
||
nextRank := currentRank.Clone()
|
||
now := s.clock.Now()
|
||
|
||
result, outboxRecords, err := mutate(now, nextState, nextRank)
|
||
if err != nil {
|
||
// 校验、钱包或业务失败时不提交状态，不写 command log。
|
||
return nil, err
|
||
}
|
||
|
||
if result.snapshot == nil {
|
||
// 无特殊返回需求的命令默认返回变更后的快照。
|
||
result.snapshot = nextState.ToProto()
|
||
}
|
||
|
||
if result.snapshot.GetVersion() != current.Version {
|
||
// 版本变化代表本命令产生实际状态变更，需要持久化命令和事件。
|
||
payload, err := command.Serialize(cmd)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
if err := s.repository.SaveCommand(ctx, CommandRecord{
|
||
RoomID: cmd.RoomID(),
|
||
RoomVersion: nextState.Version,
|
||
CommandID: cmd.ID(),
|
||
CommandType: cmd.Type(),
|
||
Payload: payload,
|
||
Replayable: replayable,
|
||
CreatedAt: now,
|
||
}); err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
if len(outboxRecords) > 0 {
|
||
// outbox 与 command log 共同构成恢复和房间外投递来源。
|
||
if err := s.repository.SaveOutbox(ctx, outboxRecords); err != nil {
|
||
return nil, err
|
||
}
|
||
}
|
||
|
||
// 持久化成功后再替换内存状态，避免内存提交领先于恢复来源。
|
||
current.ReplaceWith(nextState)
|
||
currentRank.ReplaceWith(nextRank)
|
||
result.snapshot = current.ToProto()
|
||
result.applied = true
|
||
}
|
||
|
||
return result, nil
|
||
})
|
||
if err != nil {
|
||
return mutationResult{}, err
|
||
}
|
||
|
||
result := value.(mutationResult)
|
||
|
||
if result.applied && shouldSnapshot(result.snapshot, s.snapshotEveryN) {
|
||
// 快照失败时返回错误，因为没有快照会增加恢复成本并可能影响 guard 查询。
|
||
if err := s.persistSnapshot(ctx, result.snapshot); err != nil {
|
||
return mutationResult{}, err
|
||
}
|
||
}
|
||
|
||
if result.applied && result.syncEvent != nil {
|
||
// 同步腾讯云 IM 失败不回滚房间状态；outbox 后续负责补偿投递。
|
||
_ = s.syncPublisher.PublishRoomEvent(ctx, *result.syncEvent)
|
||
}
|
||
|
||
return result, nil
|
||
}
|