362 lines
12 KiB
Go
362 lines
12 KiB
Go
package service
|
||
|
||
import (
|
||
"context"
|
||
"log/slog"
|
||
"strings"
|
||
"time"
|
||
|
||
roomeventsv1 "hyapp.local/api/proto/events/room/v1"
|
||
roomv1 "hyapp.local/api/proto/room/v1"
|
||
"hyapp/pkg/appcode"
|
||
"hyapp/pkg/idgen"
|
||
"hyapp/pkg/logx"
|
||
"hyapp/pkg/tencentim"
|
||
"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"
|
||
)
|
||
|
||
// JoinRoom 把用户业务态接入房间。
|
||
func (s *Service) JoinRoom(ctx context.Context, req *roomv1.JoinRoomRequest) (*roomv1.JoinRoomResponse, error) {
|
||
ctx = contextFromMeta(ctx, req.GetMeta())
|
||
password := strings.TrimSpace(req.GetPassword())
|
||
// JoinRoom 只维护 room-service presence，不创建任何本地长连接订阅。
|
||
cmd := command.JoinRoom{
|
||
Base: baseFromMeta(req.GetMeta()),
|
||
Role: req.GetRole(),
|
||
}
|
||
|
||
if cmd.Role == "" {
|
||
// 客户端不传角色时按普通观众进入房间。
|
||
cmd.Role = "audience"
|
||
}
|
||
if err := s.requireRoomOpenForEntry(ctx, cmd.RoomID()); err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
result, err := s.mutateRoom(ctx, cmd, true, func(now time.Time, current *state.RoomState, _ *rank.LocalRank) (mutationResult, []outbox.Record, error) {
|
||
if ban, exists := current.BanUsers[cmd.ActorUserID()]; exists {
|
||
if ban.ActiveAt(now.UnixMilli()) {
|
||
// 被踢或封禁用户在 ban 未过期前不能重新进入业务 presence。
|
||
return mutationResult{}, nil, xerr.New(xerr.PermissionDenied, "user is banned")
|
||
}
|
||
// 限时 ban 过期后由下一次进房命令清理，避免定时任务介入 Room Cell owner 状态。
|
||
delete(current.BanUsers, cmd.ActorUserID())
|
||
}
|
||
|
||
if existing, exists := current.OnlineUsers[cmd.ActorUserID()]; exists {
|
||
// 重复 JoinRoom 表示重连或刷新业务 presence，只更新 last_seen，不重复发进房系统消息。
|
||
existing.LastSeenAtMS = now.UnixMilli()
|
||
current.Version++
|
||
|
||
snapshot := current.ToProto()
|
||
return mutationResult{
|
||
snapshot: snapshot,
|
||
user: findProtoUser(snapshot, cmd.ActorUserID()),
|
||
}, nil, nil
|
||
}
|
||
if current.RoomPasswordHash != "" && current.OwnerUserID != cmd.ActorUserID() && !roomPasswordMatches(current.RoomPasswordHash, password) {
|
||
// 锁房只拦截新进入的非房主用户；已在房内的用户和 owner 不因为锁房状态重设被挤出。
|
||
return mutationResult{}, nil, xerr.New(xerr.PermissionDenied, "room password is invalid")
|
||
}
|
||
|
||
current.OnlineUsers[cmd.ActorUserID()] = &state.RoomUserState{
|
||
UserID: cmd.ActorUserID(),
|
||
Role: cmd.Role,
|
||
JoinedAtMS: now.UnixMilli(),
|
||
LastSeenAtMS: now.UnixMilli(),
|
||
}
|
||
current.Version++
|
||
|
||
// JoinRoom 成功需要 outbox 事件，腾讯云 IM 房间系统消息由 worker 异步投递。
|
||
joinedEvent, err := outbox.Build(current.RoomID, "RoomUserJoined", current.Version, now, &roomeventsv1.RoomUserJoined{
|
||
UserId: cmd.ActorUserID(),
|
||
Role: cmd.Role,
|
||
VisibleRegionId: current.VisibleRegionID,
|
||
})
|
||
if err != nil {
|
||
return mutationResult{}, nil, err
|
||
}
|
||
|
||
snapshot := current.ToProto()
|
||
|
||
return mutationResult{
|
||
snapshot: snapshot,
|
||
user: findProtoUser(snapshot, cmd.ActorUserID()),
|
||
syncEvent: &tencentim.RoomEvent{
|
||
EventID: joinedEvent.EventID,
|
||
RoomID: current.RoomID,
|
||
EventType: "room_user_joined",
|
||
ActorUserID: cmd.ActorUserID(),
|
||
TargetUserID: cmd.ActorUserID(),
|
||
RoomVersion: current.Version,
|
||
},
|
||
}, []outbox.Record{joinedEvent}, nil
|
||
})
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
s.projectRoomVisitBestEffort(ctx, cmd.ActorUserID(), cmd.RoomID())
|
||
|
||
return &roomv1.JoinRoomResponse{
|
||
Result: commandResult(result.applied, result.snapshot.GetVersion(), s.clock.Now()),
|
||
User: result.user,
|
||
Room: result.snapshot,
|
||
}, nil
|
||
}
|
||
|
||
// projectRoomVisitBestEffort 记录 Mine/Visited tab 的房间访问索引。
|
||
// 这是列表读模型，不参与 Room Cell 提交；失败只记录日志，不能影响已经成功的 JoinRoom。
|
||
func (s *Service) projectRoomVisitBestEffort(ctx context.Context, userID int64, roomID string) {
|
||
if userID <= 0 || roomID == "" {
|
||
return
|
||
}
|
||
nowMS := s.clock.Now().UnixMilli()
|
||
if err := s.repository.UpsertRoomUserFeedEntry(ctx, RoomUserFeedEntry{
|
||
AppCode: appcode.FromContext(ctx),
|
||
UserID: userID,
|
||
FeedType: roomListTabVisited,
|
||
RoomID: roomID,
|
||
CreatedAtMS: nowMS,
|
||
UpdatedAtMS: nowMS,
|
||
}); err != nil {
|
||
logx.Error(ctx, "room_visit_feed_upsert_failed", err,
|
||
slog.String("component", "room_user_feed_projector"),
|
||
slog.String("room_id", roomID),
|
||
slog.Int64("user_id", userID),
|
||
)
|
||
}
|
||
}
|
||
|
||
// RoomHeartbeat 显式刷新已在房间内用户的业务 presence。
|
||
func (s *Service) RoomHeartbeat(ctx context.Context, req *roomv1.RoomHeartbeatRequest) (*roomv1.RoomHeartbeatResponse, error) {
|
||
ctx = contextFromMeta(ctx, req.GetMeta())
|
||
cmd := command.RoomHeartbeat{
|
||
Base: baseFromMeta(req.GetMeta()),
|
||
}
|
||
|
||
result, err := s.mutateRoom(ctx, cmd, true, func(now time.Time, current *state.RoomState, _ *rank.LocalRank) (mutationResult, []outbox.Record, error) {
|
||
if err := requireActiveRoom(current); err != nil {
|
||
return mutationResult{}, nil, err
|
||
}
|
||
if ban, exists := current.BanUsers[cmd.ActorUserID()]; exists {
|
||
if ban.ActiveAt(now.UnixMilli()) {
|
||
// ban 用户不能通过心跳续住业务 presence；KickUser 已经清理 presence，这里是防御脏状态。
|
||
return mutationResult{}, nil, xerr.New(xerr.PermissionDenied, "user is banned")
|
||
}
|
||
// 过期 ban 不再阻断心跳；保守清理后继续按 presence 是否存在判断。
|
||
delete(current.BanUsers, cmd.ActorUserID())
|
||
}
|
||
|
||
existing, exists := current.OnlineUsers[cmd.ActorUserID()]
|
||
if !exists {
|
||
// 心跳只刷新已存在 presence，不能把已离房用户重新加入房间。
|
||
return mutationResult{}, nil, xerr.New(xerr.NotFound, "user not in room")
|
||
}
|
||
|
||
existing.LastSeenAtMS = now.UnixMilli()
|
||
current.Version++
|
||
|
||
snapshot := current.ToProto()
|
||
return mutationResult{
|
||
snapshot: snapshot,
|
||
user: findProtoUser(snapshot, cmd.ActorUserID()),
|
||
}, nil, nil
|
||
})
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
return &roomv1.RoomHeartbeatResponse{
|
||
Result: commandResult(result.applied, result.snapshot.GetVersion(), s.clock.Now()),
|
||
User: result.user,
|
||
Room: result.snapshot,
|
||
}, nil
|
||
}
|
||
|
||
// LeaveRoom 把用户从房间业务态移出。
|
||
func (s *Service) LeaveRoom(ctx context.Context, req *roomv1.LeaveRoomRequest) (*roomv1.LeaveRoomResponse, error) {
|
||
ctx = contextFromMeta(ctx, req.GetMeta())
|
||
// LeaveRoom 只移除 room-service 业务 presence，真实 IM 连接态由腾讯云 IM SDK 处理。
|
||
cmd := command.LeaveRoom{
|
||
Base: baseFromMeta(req.GetMeta()),
|
||
}
|
||
|
||
result, err := s.leaveRoom(ctx, cmd, "")
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
return &roomv1.LeaveRoomResponse{
|
||
Result: commandResult(result.applied, result.snapshot.GetVersion(), s.clock.Now()),
|
||
Room: result.snapshot,
|
||
}, nil
|
||
}
|
||
|
||
func (s *Service) leaveRoom(ctx context.Context, cmd command.LeaveRoom, reason string) (mutationResult, error) {
|
||
return s.mutateRoom(ctx, cmd, true, func(now time.Time, current *state.RoomState, _ *rank.LocalRank) (mutationResult, []outbox.Record, error) {
|
||
_, inRoom := current.OnlineUsers[cmd.ActorUserID()]
|
||
seat, onSeat := current.SeatByUser(cmd.ActorUserID())
|
||
if !inRoom && !onSeat {
|
||
// 离房幂等：没有业务 presence 且不占麦时不写命令日志和 outbox。
|
||
return mutationResult{snapshot: current.ToProto()}, nil, nil
|
||
}
|
||
|
||
delete(current.OnlineUsers, cmd.ActorUserID())
|
||
var micSessionID string
|
||
if onSeat {
|
||
// 离房必须同步释放麦位；同时写 RoomMicChanged/down，给时长聚合一个稳定闭合事件。
|
||
index := current.SeatIndex(seat.SeatNo)
|
||
if index >= 0 {
|
||
micSessionID = current.MicSeats[index].MicSessionID
|
||
current.ClearMicSession(index)
|
||
}
|
||
}
|
||
current.Version++
|
||
|
||
records := make([]outbox.Record, 0, 2)
|
||
if onSeat && micSessionID != "" {
|
||
micReason := reason
|
||
if micReason == "" {
|
||
micReason = "leave"
|
||
}
|
||
micEvent, err := buildMicDownEventForSeat(current.RoomID, current.Version, now, cmd.ActorUserID(), cmd.ActorUserID(), seat.SeatNo, micSessionID, micReason)
|
||
if err != nil {
|
||
return mutationResult{}, nil, err
|
||
}
|
||
records = append(records, micEvent)
|
||
}
|
||
|
||
// 离房事件进入 outbox；stale 清理会通过 attributes 标记原因，腾讯云 IM 由 worker 异步投递。
|
||
leftEvent, err := outbox.Build(current.RoomID, "RoomUserLeft", current.Version, now, &roomeventsv1.RoomUserLeft{
|
||
UserId: cmd.ActorUserID(),
|
||
})
|
||
if err != nil {
|
||
return mutationResult{}, nil, err
|
||
}
|
||
records = append(records, leftEvent)
|
||
|
||
attributes := map[string]string{}
|
||
if reason != "" {
|
||
attributes["reason"] = reason
|
||
}
|
||
|
||
return mutationResult{
|
||
snapshot: current.ToProto(),
|
||
syncEvent: &tencentim.RoomEvent{
|
||
EventID: leftEvent.EventID,
|
||
RoomID: current.RoomID,
|
||
EventType: "room_user_left",
|
||
ActorUserID: cmd.ActorUserID(),
|
||
TargetUserID: cmd.ActorUserID(),
|
||
RoomVersion: current.Version,
|
||
Attributes: attributes,
|
||
},
|
||
}, records, nil
|
||
})
|
||
}
|
||
|
||
// RunPresenceStaleWorker 周期性清理本节点已装载房间中的过期业务 presence。
|
||
// worker 只调用 room-service 命令链路，不触碰腾讯云 IM 连接态，客户端重连后仍需重新 JoinRoom。
|
||
func (s *Service) RunPresenceStaleWorker(ctx context.Context, interval time.Duration) {
|
||
if interval <= 0 || s.presenceStaleAfter <= 0 {
|
||
// 非正数表示显式关闭 worker，测试或特殊环境可用。
|
||
return
|
||
}
|
||
|
||
ticker := time.NewTicker(interval)
|
||
defer ticker.Stop()
|
||
|
||
for {
|
||
select {
|
||
case <-ctx.Done():
|
||
return
|
||
case <-ticker.C:
|
||
_ = s.SweepStalePresence(ctx)
|
||
}
|
||
}
|
||
}
|
||
|
||
// SweepStalePresence 扫描并清理已经超过 last_seen 阈值的业务 presence。
|
||
// 该方法暴露给测试和后台 worker 复用，所有清理都复用 LeaveRoom 的持久化、快照和 outbox 语义。
|
||
func (s *Service) SweepStalePresence(ctx context.Context) error {
|
||
if s.presenceStaleAfter <= 0 {
|
||
return nil
|
||
}
|
||
|
||
now := s.clock.Now()
|
||
cutoffMs := now.Add(-s.presenceStaleAfter).UnixMilli()
|
||
for _, roomRef := range s.loadedRoomRefs() {
|
||
roomCtx := appcode.WithContext(ctx, roomRef.AppCode)
|
||
lease, owned, err := s.loadedRoomLease(roomCtx, roomRef, now)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if !owned {
|
||
continue
|
||
}
|
||
|
||
roomCell := s.loadCell(roomCtx, roomRef.RoomID)
|
||
if roomCell == nil {
|
||
continue
|
||
}
|
||
|
||
currentState, _, err := roomCell.Snapshot(roomCtx)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
owned, err = s.verifyLoadedRoomLease(roomCtx, roomRef, lease, s.clock.Now())
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if !owned {
|
||
continue
|
||
}
|
||
|
||
for _, userID := range stalePresenceUserIDs(currentState, cutoffMs) {
|
||
owned, err = s.verifyLoadedRoomLease(roomCtx, roomRef, lease, s.clock.Now())
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if !owned {
|
||
break
|
||
}
|
||
|
||
cmd := command.LeaveRoom{
|
||
Base: command.Base{
|
||
RequestID: idgen.New("req_presence_stale"),
|
||
CommandID: idgen.New("cmd_presence_stale"),
|
||
ActorID: userID,
|
||
Room: roomRef.RoomID,
|
||
GatewayNodeID: s.nodeID,
|
||
SessionID: "presence-stale",
|
||
SentAtMS: now.UnixMilli(),
|
||
},
|
||
}
|
||
if _, err := s.leaveRoom(roomCtx, cmd, "stale_timeout"); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
func stalePresenceUserIDs(current *state.RoomState, cutoffMs int64) []int64 {
|
||
if current == nil {
|
||
return nil
|
||
}
|
||
|
||
userIDs := make([]int64, 0)
|
||
for userID, userState := range current.OnlineUsers {
|
||
if userState != nil && userState.LastSeenAtMS <= cutoffMs {
|
||
userIDs = append(userIDs, userID)
|
||
}
|
||
}
|
||
|
||
return userIDs
|
||
}
|