271 lines
11 KiB
Go
271 lines
11 KiB
Go
package service
|
||
|
||
import (
|
||
"context"
|
||
"strings"
|
||
"time"
|
||
|
||
roomeventsv1 "hyapp.local/api/proto/events/room/v1"
|
||
roomv1 "hyapp.local/api/proto/room/v1"
|
||
"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"
|
||
)
|
||
|
||
const (
|
||
// rtcEventAudioStarted 表示腾讯 RTC 侧观测到用户开始发布音频。
|
||
rtcEventAudioStarted = "audio_started"
|
||
// rtcEventAudioStopped 表示腾讯 RTC 侧观测到用户停止发布音频。
|
||
rtcEventAudioStopped = "audio_stopped"
|
||
// rtcEventRoomExited 表示用户离开 RTC 媒体房间，不等价于 App 业务离房。
|
||
rtcEventRoomExited = "room_exited"
|
||
)
|
||
|
||
// ApplyRTCEvent 把已验签的腾讯 RTC 服务端事件落到 Room Cell。
|
||
func (s *Service) ApplyRTCEvent(ctx context.Context, req *roomv1.ApplyRTCEventRequest) (*roomv1.ApplyRTCEventResponse, error) {
|
||
ctx = contextFromMeta(ctx, req.GetMeta())
|
||
// gateway 负责验签和 user_id 映射，room-service 只处理可信且已标准化的媒体事件。
|
||
cmd := command.ApplyRTCEvent{
|
||
Base: baseFromMeta(req.GetMeta()),
|
||
TargetUserID: req.GetTargetUserId(),
|
||
EventType: strings.TrimSpace(req.GetEventType()),
|
||
EventTimeMS: req.GetEventTimeMs(),
|
||
Reason: strings.TrimSpace(req.GetReason()),
|
||
Source: strings.TrimSpace(req.GetSource()),
|
||
ExternalEventID: strings.TrimSpace(req.GetExternalEventId()),
|
||
}
|
||
if cmd.TargetUserID == 0 {
|
||
// gateway 正常会显式传 target_user_id；兜底使用 actor 保持内部调用更易测试。
|
||
cmd.TargetUserID = cmd.ActorUserID()
|
||
}
|
||
if cmd.Source == "" {
|
||
cmd.Source = "rtc_webhook"
|
||
}
|
||
|
||
result, err := s.mutateRoom(ctx, cmd, true, func(now time.Time, current *state.RoomState, _ *rank.LocalRank) (mutationResult, []outbox.Record, error) {
|
||
if cmd.TargetUserID <= 0 || cmd.EventType == "" || cmd.EventTimeMS <= 0 {
|
||
// RTC 事件必须能定位用户、事件类型和事件时间，否则无法做乱序/旧事件过滤。
|
||
return mutationResult{}, nil, xerr.New(xerr.InvalidArgument, "rtc event is incomplete")
|
||
}
|
||
|
||
// 不同 RTC 事件的业务语义不同：开始发布只确认麦位，停止发布释放麦位，退房只处理媒体状态。
|
||
switch cmd.EventType {
|
||
case rtcEventAudioStarted:
|
||
return s.applyRTCAudioStarted(now, current, cmd)
|
||
case rtcEventAudioStopped:
|
||
return s.applyRTCAudioStopped(now, current, cmd)
|
||
case rtcEventRoomExited:
|
||
return s.applyRTCRoomExited(now, current, cmd)
|
||
default:
|
||
return mutationResult{}, nil, xerr.New(xerr.InvalidArgument, "rtc event type is unsupported")
|
||
}
|
||
})
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
return &roomv1.ApplyRTCEventResponse{
|
||
Result: commandResult(result.applied, result.snapshot.GetVersion(), s.clock.Now()),
|
||
SeatNo: result.seatNo,
|
||
Room: result.snapshot,
|
||
}, nil
|
||
}
|
||
|
||
// applyRTCAudioStarted 把当前 mic_session 从 pending_publish 确认为 publishing。
|
||
// 它只接受匹配当前麦位会话、未超时且时间更新的事件，腾讯回调重试或乱序事件都按 no-op 处理。
|
||
func (s *Service) applyRTCAudioStarted(now time.Time, current *state.RoomState, cmd command.ApplyRTCEvent) (mutationResult, []outbox.Record, error) {
|
||
if err := requireActiveRoom(current); err != nil {
|
||
return mutationResult{}, nil, err
|
||
}
|
||
if _, exists := current.OnlineUsers[cmd.TargetUserID]; !exists {
|
||
// RTC 事件不拥有业务进房事实；用户不在业务 presence 时只忽略旧事件。
|
||
return mutationResult{snapshot: current.ToProto()}, nil, nil
|
||
}
|
||
|
||
seat, exists := current.SeatByUser(cmd.TargetUserID)
|
||
if !exists {
|
||
// 用户不在麦上时，audio_started 只能是旧回调或客户端已经下麦后的迟到事件。
|
||
return mutationResult{snapshot: current.ToProto()}, nil, nil
|
||
}
|
||
if !rtcEventMatchesCurrentMicSession(seat, cmd.EventTimeMS, s.micPublishTimeout) {
|
||
// 会话不匹配、事件时间倒退或超过确认 deadline 时不能推进当前麦位状态。
|
||
return mutationResult{snapshot: current.ToProto(), seatNo: seat.SeatNo}, nil, nil
|
||
}
|
||
if seat.PublishState == state.MicPublishPublishing {
|
||
// 腾讯可能重试同一事件，已确认发流时保持幂等 no-op。
|
||
return mutationResult{snapshot: current.ToProto(), seatNo: seat.SeatNo}, nil, nil
|
||
}
|
||
if seat.PublishState != state.MicPublishPending {
|
||
// 非 pending 状态不应该被 audio_started 重新打开，避免旧事件复活已释放会话。
|
||
return mutationResult{snapshot: current.ToProto(), seatNo: seat.SeatNo}, nil, nil
|
||
}
|
||
|
||
// 只有通过所有防旧校验后，才把麦位推进到 publishing 并记录最新媒体事件时间。
|
||
index := current.SeatIndex(seat.SeatNo)
|
||
current.MicSeats[index].PublishState = state.MicPublishPublishing
|
||
current.MicSeats[index].LastPublishEventTimeMS = cmd.EventTimeMS
|
||
current.MicSeats[index].MicHeartbeatAtMS = now.UnixMilli()
|
||
current.Version++
|
||
seatStatus := current.MicSeats[index].SeatStatus()
|
||
|
||
micEvent, err := outbox.Build(current.RoomID, "RoomMicChanged", current.Version, now, &roomeventsv1.RoomMicChanged{
|
||
ActorUserId: cmd.ActorUserID(),
|
||
TargetUserId: cmd.TargetUserID,
|
||
FromSeat: seat.SeatNo,
|
||
ToSeat: seat.SeatNo,
|
||
Action: "publish_confirmed",
|
||
MicSessionId: seat.MicSessionID,
|
||
PublishState: state.MicPublishPublishing,
|
||
PublishDeadlineMs: seat.PublishDeadlineMS,
|
||
PublishEventTimeMs: cmd.EventTimeMS,
|
||
MicMuted: current.MicSeats[index].MicMuted,
|
||
SeatStatus: seatStatus,
|
||
})
|
||
if err != nil {
|
||
return mutationResult{}, nil, err
|
||
}
|
||
|
||
return mutationResult{
|
||
snapshot: current.ToProto(),
|
||
seatNo: seat.SeatNo,
|
||
syncEvent: &tencentim.RoomEvent{
|
||
EventID: micEvent.EventID,
|
||
RoomID: current.RoomID,
|
||
EventType: "room_mic_publish_confirmed",
|
||
ActorUserID: cmd.ActorUserID(),
|
||
TargetUserID: cmd.TargetUserID,
|
||
SeatNo: seat.SeatNo,
|
||
RoomVersion: current.Version,
|
||
Attributes: map[string]string{
|
||
"action": "publish_confirmed",
|
||
"publish_state": state.MicPublishPublishing,
|
||
"mic_session_id": seat.MicSessionID,
|
||
"publish_event_time_ms": int64String(cmd.EventTimeMS),
|
||
"source": cmd.Source,
|
||
"external_event_id": cmd.ExternalEventID,
|
||
"mic_muted": boolString(current.MicSeats[index].MicMuted),
|
||
"seat_status": seatStatus,
|
||
},
|
||
},
|
||
}, []outbox.Record{micEvent}, nil
|
||
}
|
||
|
||
// applyRTCAudioStopped 释放当前 mic_session 占用的麦位。
|
||
// 停止发布事件不检查业务 presence，因为 RTC 退流可能先于 LeaveRoom 或 stale 清理到达。
|
||
func (s *Service) applyRTCAudioStopped(now time.Time, current *state.RoomState, cmd command.ApplyRTCEvent) (mutationResult, []outbox.Record, error) {
|
||
seat, exists := current.SeatByUser(cmd.TargetUserID)
|
||
if !exists {
|
||
// 麦位已释放时，迟到的停止发布事件按 no-op 收敛。
|
||
return mutationResult{snapshot: current.ToProto()}, nil, nil
|
||
}
|
||
if !rtcStopEventMatchesCurrentMicSession(seat, cmd.EventTimeMS, s.micPublishTimeout) {
|
||
// 旧事件不能释放新 mic_session，防止用户快速重上麦时被迟到 webhook 踢下。
|
||
return mutationResult{snapshot: current.ToProto(), seatNo: seat.SeatNo}, nil, nil
|
||
}
|
||
|
||
// 清麦和 RoomMicChanged/down 必须同版本提交，给时长聚合一个稳定闭合点。
|
||
index := current.SeatIndex(seat.SeatNo)
|
||
micSessionID := current.MicSeats[index].MicSessionID
|
||
current.ClearMicSession(index)
|
||
current.Version++
|
||
seatStatus := current.MicSeats[index].SeatStatus()
|
||
|
||
micEvent, err := outbox.Build(current.RoomID, "RoomMicChanged", current.Version, now, &roomeventsv1.RoomMicChanged{
|
||
ActorUserId: cmd.ActorUserID(),
|
||
TargetUserId: cmd.TargetUserID,
|
||
FromSeat: seat.SeatNo,
|
||
ToSeat: 0,
|
||
Action: "down",
|
||
MicSessionId: micSessionID,
|
||
PublishState: state.MicPublishIdle,
|
||
Reason: cmd.Reason,
|
||
PublishEventTimeMs: cmd.EventTimeMS,
|
||
SeatStatus: seatStatus,
|
||
})
|
||
if err != nil {
|
||
return mutationResult{}, nil, err
|
||
}
|
||
|
||
return mutationResult{
|
||
snapshot: current.ToProto(),
|
||
seatNo: seat.SeatNo,
|
||
syncEvent: &tencentim.RoomEvent{
|
||
EventID: micEvent.EventID,
|
||
RoomID: current.RoomID,
|
||
EventType: "room_mic_down",
|
||
ActorUserID: cmd.ActorUserID(),
|
||
TargetUserID: cmd.TargetUserID,
|
||
SeatNo: seat.SeatNo,
|
||
RoomVersion: current.Version,
|
||
Attributes: map[string]string{
|
||
"action": "down",
|
||
"publish_state": state.MicPublishIdle,
|
||
"mic_session_id": micSessionID,
|
||
"reason": cmd.Reason,
|
||
"publish_event_time_ms": int64String(cmd.EventTimeMS),
|
||
"source": cmd.Source,
|
||
"external_event_id": cmd.ExternalEventID,
|
||
"seat_status": seatStatus,
|
||
},
|
||
},
|
||
}, []outbox.Record{micEvent}, nil
|
||
}
|
||
|
||
// applyRTCRoomExited 只处理 RTC 媒体房间退出事实，不删除 room-service 业务 presence。
|
||
func (s *Service) applyRTCRoomExited(now time.Time, current *state.RoomState, cmd command.ApplyRTCEvent) (mutationResult, []outbox.Record, error) {
|
||
if user, exists := current.OnlineUsers[cmd.TargetUserID]; exists && user.LastSeenAtMS > cmd.EventTimeMS {
|
||
// 腾讯可能重试或延迟投递旧退房事件；如果业务 presence 已被更新到事件之后，保留当前重连和麦位状态。
|
||
return mutationResult{snapshot: current.ToProto()}, nil, nil
|
||
}
|
||
|
||
_, onSeat := current.SeatByUser(cmd.TargetUserID)
|
||
if !onSeat {
|
||
// RTC 退房只代表媒体频道状态，不拥有 room-service 业务 presence。
|
||
return mutationResult{snapshot: current.ToProto()}, nil, nil
|
||
}
|
||
|
||
// 用户仍可能停留在房间页继续重连；这里只释放当前 mic_session，业务离房只能由 LeaveRoom、kick、close 或 stale worker 触发。
|
||
return s.applyRTCAudioStopped(now, current, cmd)
|
||
}
|
||
|
||
// rtcEventMatchesCurrentMicSession 校验音频开始事件是否仍属于当前 pending mic_session。
|
||
func rtcEventMatchesCurrentMicSession(seat state.MicSeat, eventTimeMS int64, timeout time.Duration) bool {
|
||
if !rtcStopEventMatchesCurrentMicSession(seat, eventTimeMS, timeout) {
|
||
return false
|
||
}
|
||
if seat.PublishDeadlineMS > 0 && eventTimeMS > seat.PublishDeadlineMS {
|
||
// audio_started 过了确认 deadline 不再接受，超时 worker 会按 mic_session_id 释放麦位。
|
||
return false
|
||
}
|
||
|
||
return true
|
||
}
|
||
|
||
// rtcStopEventMatchesCurrentMicSession 校验停止/退房事件是否能作用于当前 mic_session。
|
||
func rtcStopEventMatchesCurrentMicSession(seat state.MicSeat, eventTimeMS int64, timeout time.Duration) bool {
|
||
if seat.MicSessionID == "" || eventTimeMS <= seat.LastPublishEventTimeMS {
|
||
// 没有会话或事件时间不递增时，事件不能改变当前麦位。
|
||
return false
|
||
}
|
||
if createdAtMS := rtcMicSessionCreatedAtMS(seat, timeout); createdAtMS > 0 && eventTimeMS < createdAtMS {
|
||
// 小于会话创建时间的 webhook 一定属于旧会话。
|
||
return false
|
||
}
|
||
|
||
return true
|
||
}
|
||
|
||
// rtcMicSessionCreatedAtMS 从发布 deadline 反推出会话创建时间，用于过滤旧 RTC 事件。
|
||
func rtcMicSessionCreatedAtMS(seat state.MicSeat, timeout time.Duration) int64 {
|
||
timeoutMS := timeout.Milliseconds()
|
||
if seat.PublishDeadlineMS <= 0 || timeoutMS <= 0 {
|
||
// 没有 deadline 或 timeout 配置时无法反推创建时间，调用方只能使用事件时间递增校验。
|
||
return 0
|
||
}
|
||
|
||
return seat.PublishDeadlineMS - timeoutMS
|
||
}
|