39 lines
1.5 KiB
Go
39 lines
1.5 KiB
Go
package service
|
||
|
||
import (
|
||
"context"
|
||
"fmt"
|
||
|
||
"google.golang.org/protobuf/proto"
|
||
roomeventsv1 "hyapp.local/api/proto/events/room/v1"
|
||
)
|
||
|
||
// syncRTCUserMuteFromOutbox 把持久化的房间禁言事实补偿到腾讯 RTC 媒体平面。
|
||
// 重试时必须读取 Room Cell 的最新禁言集合，而不能直接重放旧事件里的 muted；否则旧 mute
|
||
// 在 newer unmute 之后重试会重新关掉用户音频，形成跨系统状态回退。
|
||
func (s *Service) syncRTCUserMuteFromOutbox(ctx context.Context, envelope *roomeventsv1.EventEnvelope) error {
|
||
if s == nil || s.rtcUserAudioBlocker == nil || envelope == nil || envelope.GetEventType() != "RoomUserMuted" {
|
||
return nil
|
||
}
|
||
|
||
var event roomeventsv1.RoomUserMuted
|
||
if err := proto.Unmarshal(envelope.GetBody(), &event); err != nil {
|
||
return fmt.Errorf("decode rtc mute outbox event: %w", err)
|
||
}
|
||
if event.GetTargetUserId() <= 0 || envelope.GetRoomId() == "" {
|
||
return fmt.Errorf("rtc mute outbox event is incomplete")
|
||
}
|
||
|
||
snapshot, err := s.currentSnapshot(ctx, envelope.GetRoomId())
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if snapshot == nil {
|
||
// 房间已被物理清理时 RTC 连接也不再存在；把历史禁言事件视为终态成功，
|
||
// 避免一条不可执行的外部副作用永久阻塞同一 outbox worker。
|
||
return nil
|
||
}
|
||
blocked := containsID(snapshot.GetMuteUserIds(), event.GetTargetUserId())
|
||
return s.rtcUserAudioBlocker.SetUserAudioBlockedByStrRoomID(ctx, envelope.GetRoomId(), event.GetTargetUserId(), blocked)
|
||
}
|