236 lines
8.3 KiB
Go
236 lines
8.3 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 = "audio_started"
|
||
rtcEventAudioStopped = "audio_stopped"
|
||
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())
|
||
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 {
|
||
return mutationResult{}, nil, xerr.New(xerr.InvalidArgument, "rtc event is incomplete")
|
||
}
|
||
|
||
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
|
||
}
|
||
|
||
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 {
|
||
return mutationResult{snapshot: current.ToProto()}, nil, nil
|
||
}
|
||
if !rtcEventMatchesCurrentMicSession(seat, cmd.EventTimeMS, s.micPublishTimeout) {
|
||
return mutationResult{snapshot: current.ToProto(), seatNo: seat.SeatNo}, nil, nil
|
||
}
|
||
if seat.PublishState == state.MicPublishPublishing {
|
||
return mutationResult{snapshot: current.ToProto(), seatNo: seat.SeatNo}, nil, nil
|
||
}
|
||
if seat.PublishState != state.MicPublishPending {
|
||
return mutationResult{snapshot: current.ToProto(), seatNo: seat.SeatNo}, nil, nil
|
||
}
|
||
|
||
index := current.SeatIndex(seat.SeatNo)
|
||
current.MicSeats[index].PublishState = state.MicPublishPublishing
|
||
current.MicSeats[index].LastPublishEventTimeMS = cmd.EventTimeMS
|
||
current.Version++
|
||
|
||
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,
|
||
})
|
||
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,
|
||
},
|
||
},
|
||
}, []outbox.Record{micEvent}, nil
|
||
}
|
||
|
||
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 {
|
||
return mutationResult{snapshot: current.ToProto()}, nil, nil
|
||
}
|
||
if !rtcStopEventMatchesCurrentMicSession(seat, cmd.EventTimeMS, s.micPublishTimeout) {
|
||
return mutationResult{snapshot: current.ToProto(), seatNo: seat.SeatNo}, nil, nil
|
||
}
|
||
|
||
index := current.SeatIndex(seat.SeatNo)
|
||
micSessionID := current.MicSeats[index].MicSessionID
|
||
current.ClearMicSession(index)
|
||
current.Version++
|
||
|
||
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,
|
||
})
|
||
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,
|
||
},
|
||
},
|
||
}, []outbox.Record{micEvent}, nil
|
||
}
|
||
|
||
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)
|
||
}
|
||
|
||
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 {
|
||
return false
|
||
}
|
||
|
||
return true
|
||
}
|
||
|
||
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 {
|
||
return false
|
||
}
|
||
|
||
return true
|
||
}
|
||
|
||
func rtcMicSessionCreatedAtMS(seat state.MicSeat, timeout time.Duration) int64 {
|
||
timeoutMS := timeout.Milliseconds()
|
||
if seat.PublishDeadlineMS <= 0 || timeoutMS <= 0 {
|
||
return 0
|
||
}
|
||
|
||
return seat.PublishDeadlineMS - timeoutMS
|
||
}
|