From cf431eb8d79d6dc274b8b7fbdfd379085abf6de9 Mon Sep 17 00:00:00 2001 From: zhx Date: Tue, 21 Jul 2026 13:11:11 +0800 Subject: [PATCH] =?UTF-8?q?rtc=E7=A6=81=E8=A8=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pkg/tencentrtc/rest_client.go | 74 ++++++++++++++++++- pkg/tencentrtc/rest_client_test.go | 70 ++++++++++++++++++ services/room-service/internal/app/app.go | 3 + .../internal/integration/clients.go | 6 ++ .../internal/room/service/mic_test.go | 68 ++++++++++++++++- .../internal/room/service/moderation.go | 16 ++++ .../internal/room/service/outbox_worker.go | 5 ++ .../internal/room/service/rtc_event.go | 12 +++ .../internal/room/service/rtc_moderation.go | 38 ++++++++++ .../internal/room/service/service.go | 5 ++ 10 files changed, 295 insertions(+), 2 deletions(-) create mode 100644 services/room-service/internal/room/service/rtc_moderation.go diff --git a/pkg/tencentrtc/rest_client.go b/pkg/tencentrtc/rest_client.go index 3fe08192..5770f732 100644 --- a/pkg/tencentrtc/rest_client.go +++ b/pkg/tencentrtc/rest_client.go @@ -8,6 +8,7 @@ import ( "crypto/sha256" "encoding/hex" "encoding/json" + "errors" "fmt" "io" "net/http" @@ -25,6 +26,9 @@ const ( trtcServiceName = "trtc" trtcAPIVersion = "2019-07-22" removeUserByStrRoomAction = "RemoveUserByStrRoomId" + setUserBlockedByStrAction = "SetUserBlockedByStrRoomId" + trtcUserMediaEnabled = 0 + trtcUserAudioDisabled = 2 tencentCloudTC3Algorithm = "TC3-HMAC-SHA256" tencentCloudTC3RequestType = "tc3_request" ) @@ -95,6 +99,38 @@ func (c *RESTClient) RemoveUserByStrRoomID(ctx context.Context, roomID string, u return c.post(ctx, removeUserByStrRoomAction, payload) } +// SetUserAudioBlockedByStrRoomID uses TRTC's server-side media gate instead of +// trusting the target client to stop publishing. Room-service remains the mute +// state owner; this call only projects that committed state onto the live RTC +// data plane and deliberately leaves receiving audio enabled. +func (c *RESTClient) SetUserAudioBlockedByStrRoomID(ctx context.Context, roomID string, userID int64, blocked bool) error { + roomID = strings.TrimSpace(roomID) + if roomID == "" || userID <= 0 { + return fmt.Errorf("rtc set user audio blocked is incomplete") + } + + muteMode := trtcUserMediaEnabled + if blocked { + // IsMute=2 disables only upstream audio. IsMute=1 would also disable + // video and would silently broaden the semantics of room mute. + muteMode = trtcUserAudioDisabled + } + payload := setUserBlockedByStrRoomRequest{ + SDKAppID: c.cfg.SDKAppID, + RoomID: roomID, + UserID: strconv.FormatInt(userID, 10), + IsMute: muteMode, + } + err := c.post(ctx, setUserBlockedByStrAction, payload) + if IsRESTErrorCode(err, "FailedOperation.RoomNotExist", "FailedOperation.UserNotExist", "InternalError.UserNotExist") { + // The desired media state is already satisfied when either the room or + // user has gone. Retrying these terminal absences would only poison the + // durable room outbox until it reaches the dead-letter threshold. + return nil + } + return err +} + func (c *RESTClient) post(ctx context.Context, action string, payload any) error { body, err := json.Marshal(payload) if err != nil { @@ -131,11 +167,40 @@ func (c *RESTClient) post(ctx context.Context, action string, payload any) error return err } if cloudResp.Response.Error != nil { - return fmt.Errorf("tencent rtc rest failed: code=%s message=%s", cloudResp.Response.Error.Code, cloudResp.Response.Error.Message) + return &RESTError{Code: cloudResp.Response.Error.Code, Message: cloudResp.Response.Error.Message} } return nil } +// RESTError preserves Tencent's stable business code so callers can classify +// terminal idempotent outcomes without matching a human-readable message. +type RESTError struct { + Code string + Message string +} + +func (e *RESTError) Error() string { + if e == nil { + return "" + } + return fmt.Sprintf("tencent rtc rest failed: code=%s message=%s", e.Code, e.Message) +} + +// IsRESTErrorCode reports whether err is a Tencent RTC business error with one +// of the supplied codes. Wrapped errors are supported for service-level context. +func IsRESTErrorCode(err error, codes ...string) bool { + var restErr *RESTError + if !errors.As(err, &restErr) { + return false + } + for _, code := range codes { + if restErr.Code == code { + return true + } + } + return false +} + func (c *RESTClient) sign(request *http.Request, body []byte, action string) { now := c.now().UTC() timestamp := now.Unix() @@ -198,6 +263,13 @@ type removeUserByStrRoomRequest struct { UserIDs []string `json:"UserIds"` } +type setUserBlockedByStrRoomRequest struct { + SDKAppID int64 `json:"SdkAppId"` + RoomID string `json:"StrRoomId"` + UserID string `json:"UserId"` + IsMute int `json:"IsMute"` +} + type cloudAPIResponse struct { Response struct { RequestID string `json:"RequestId"` diff --git a/pkg/tencentrtc/rest_client_test.go b/pkg/tencentrtc/rest_client_test.go index e7116e01..9092037f 100644 --- a/pkg/tencentrtc/rest_client_test.go +++ b/pkg/tencentrtc/rest_client_test.go @@ -56,6 +56,76 @@ func TestRESTClientRemoveUserByStrRoomIDBuildsTencentCloudRequest(t *testing.T) } } +// TestRESTClientSetUserAudioBlockedByStrRoomIDBuildsAudioOnlyRequest 锁定服务端禁言只关闭上行音频， +// 解除时恢复音视频能力，避免误用 IsMute=1 把未来的视频房能力一并关闭。 +func TestRESTClientSetUserAudioBlockedByStrRoomIDBuildsAudioOnlyRequest(t *testing.T) { + var bodies []map[string]any + var actions []string + client, err := NewRESTClient(RESTConfig{ + SDKAppID: 1400000001, + SecretID: "AKIDEXAMPLE", + SecretKey: "secret", + Region: "ap-singapore", + HTTPClient: &http.Client{Transport: roundTripFunc(func(request *http.Request) (*http.Response, error) { + var body map[string]any + if err := json.NewDecoder(request.Body).Decode(&body); err != nil { + t.Fatalf("decode request body failed: %v", err) + } + bodies = append(bodies, body) + actions = append(actions, request.Header.Get("X-TC-Action")) + return &http.Response{ + StatusCode: http.StatusOK, + Body: io.NopCloser(bytes.NewReader([]byte(`{"Response":{"RequestId":"req-block"}}`))), + Header: make(http.Header), + }, nil + })}, + Now: func() time.Time { return time.Unix(1_700_000_000, 0).UTC() }, + }) + if err != nil { + t.Fatalf("NewRESTClient failed: %v", err) + } + + if err := client.SetUserAudioBlockedByStrRoomID(context.Background(), "room_1001", 10002, true); err != nil { + t.Fatalf("block user audio failed: %v", err) + } + if err := client.SetUserAudioBlockedByStrRoomID(context.Background(), "room_1001", 10002, false); err != nil { + t.Fatalf("unblock user audio failed: %v", err) + } + + if len(bodies) != 2 || len(actions) != 2 { + t.Fatalf("unexpected request count: bodies=%d actions=%d", len(bodies), len(actions)) + } + for index, expectedMute := range []float64{trtcUserAudioDisabled, trtcUserMediaEnabled} { + if actions[index] != setUserBlockedByStrAction || bodies[index]["SdkAppId"] != float64(1400000001) || bodies[index]["StrRoomId"] != "room_1001" || bodies[index]["UserId"] != "10002" || bodies[index]["IsMute"] != expectedMute { + t.Fatalf("unexpected set user blocked request %d: action=%s body=%+v", index, actions[index], bodies[index]) + } + } +} + +// TestRESTClientSetUserAudioBlockedTreatsAbsentRTCUserAsConverged 防止用户先退 RTC 后的历史 +// RoomUserMuted 事件把 durable outbox 永久卡在无意义重试上。 +func TestRESTClientSetUserAudioBlockedTreatsAbsentRTCUserAsConverged(t *testing.T) { + client, err := NewRESTClient(RESTConfig{ + SDKAppID: 1400000001, + SecretID: "AKIDEXAMPLE", + SecretKey: "secret", + Region: "ap-singapore", + HTTPClient: &http.Client{Transport: roundTripFunc(func(*http.Request) (*http.Response, error) { + return &http.Response{ + StatusCode: http.StatusOK, + Body: io.NopCloser(bytes.NewReader([]byte(`{"Response":{"Error":{"Code":"FailedOperation.UserNotExist","Message":"user absent"},"RequestId":"req-absent"}}`))), + Header: make(http.Header), + }, nil + })}, + }) + if err != nil { + t.Fatalf("NewRESTClient failed: %v", err) + } + if err := client.SetUserAudioBlockedByStrRoomID(context.Background(), "room_1001", 10002, true); err != nil { + t.Fatalf("absent RTC user must be treated as converged: %v", err) + } +} + type roundTripFunc func(*http.Request) (*http.Response, error) func (f roundTripFunc) RoundTrip(request *http.Request) (*http.Response, error) { diff --git a/services/room-service/internal/app/app.go b/services/room-service/internal/app/app.go index acead580..5e5827ed 100644 --- a/services/room-service/internal/app/app.go +++ b/services/room-service/internal/app/app.go @@ -165,6 +165,7 @@ func New(cfg config.Config) (*App, error) { roomPublisher := integration.NewNoopRoomEventPublisher() outboxPublishers := make([]integration.OutboxPublisher, 0, 3) var rtcUserRemover integration.RTCUserRemover + var rtcUserAudioBlocker integration.RTCUserAudioBlocker var tencentPublisher integration.OutboxPublisher if cfg.TencentIM.Enabled { // 腾讯云 IM 替代自研 IM；room-service 只负责服务端 REST 群消息桥接。 @@ -197,6 +198,7 @@ func New(cfg config.Config) (*App, error) { return nil, err } rtcUserRemover = rtcClient + rtcUserAudioBlocker = rtcClient } mqProducers := make([]*rocketmqx.Producer, 0, 2) @@ -248,6 +250,7 @@ func New(cfg config.Config) (*App, error) { PresenceStaleAfter: cfg.PresenceStaleAfter, MicPublishTimeout: cfg.MicPublishTimeout, RTCUserRemover: rtcUserRemover, + RTCUserAudioBlocker: rtcUserAudioBlocker, RoomRocketLaunchScheduler: rocketLaunchScheduler, RobotDisplayPublisher: tencentPublisher, RoomDisplayPublisher: tencentPublisher, diff --git a/services/room-service/internal/integration/clients.go b/services/room-service/internal/integration/clients.go index 3b2625c8..35c5218e 100644 --- a/services/room-service/internal/integration/clients.go +++ b/services/room-service/internal/integration/clients.go @@ -61,6 +61,12 @@ type RTCUserRemover interface { RemoveUserByStrRoomID(ctx context.Context, roomID string, userID int64) error } +// RTCUserAudioBlocker 抽象腾讯 RTC 媒体平面的服务端音频开关。 +// Room Cell 仍是禁言事实 owner；实现只能投影当前状态，不能自行持有禁言集合。 +type RTCUserAudioBlocker interface { + SetUserAudioBlockedByStrRoomID(ctx context.Context, roomID string, userID int64, blocked bool) error +} + // OutboxPublisher 抽象 outbox worker 对外部消费者的异步投递。 type OutboxPublisher interface { // PublishOutboxEvent 投递 protobuf outbox 信封，失败由 room_outbox 转为 retryable 状态重试。 diff --git a/services/room-service/internal/room/service/mic_test.go b/services/room-service/internal/room/service/mic_test.go index f578ed84..178c6350 100644 --- a/services/room-service/internal/room/service/mic_test.go +++ b/services/room-service/internal/room/service/mic_test.go @@ -3,21 +3,51 @@ package service_test import ( "context" "fmt" + "sync" "testing" "time" roomv1 "hyapp.local/api/proto/room/v1" "hyapp/pkg/appcode" "hyapp/pkg/xerr" + "hyapp/services/room-service/internal/integration" "hyapp/services/room-service/internal/room/command" + roomservice "hyapp/services/room-service/internal/room/service" + "hyapp/services/room-service/internal/router" "hyapp/services/room-service/internal/testutil/mysqltest" ) +type recordingRTCUserAudioBlocker struct { + mu sync.Mutex + calls []bool +} + +func (b *recordingRTCUserAudioBlocker) SetUserAudioBlockedByStrRoomID(_ context.Context, _ string, _ int64, blocked bool) error { + b.mu.Lock() + defer b.mu.Unlock() + b.calls = append(b.calls, blocked) + return nil +} + +func (b *recordingRTCUserAudioBlocker) snapshot() []bool { + b.mu.Lock() + defer b.mu.Unlock() + return append([]bool(nil), b.calls...) +} + func TestMutedUserCannotSpeakOrMicUpUntilUnmuted(t *testing.T) { ctx := context.Background() repository := mysqltest.NewRepository(t) now := &fixedRoomRocketClock{now: time.Date(2026, 7, 21, 8, 0, 0, 0, time.UTC)} - svc := newRocketTestService(t, repository, &rocketTestWallet{}, now) + rtcBlocker := &recordingRTCUserAudioBlocker{} + svc := roomservice.New(roomservice.Config{ + NodeID: "node-muted-rtc-test", + LeaseTTL: 10 * time.Second, + RankLimit: 20, + SnapshotEveryN: 1, + Clock: now, + RTCUserAudioBlocker: rtcBlocker, + }, router.NewMemoryDirectory(), repository, &rocketTestWallet{}, integration.NewNoopRoomEventPublisher(), integration.NewNoopOutboxPublisher()) roomID := "room-muted-user-guard" ownerID := int64(8481) @@ -113,6 +143,20 @@ func TestMutedUserCannotSpeakOrMicUpUntilUnmuted(t *testing.T) { }); !xerr.IsCode(err, xerr.PermissionDenied) { t.Fatalf("room mute between MicUp and RTC confirmation must reject publishing: %v", err) } + rtcEventResp, err := svc.ApplyRTCEvent(ctx, &roomv1.ApplyRTCEventRequest{ + Meta: rocketMeta(roomID, mutedUserID, "muted-rtc-audio-started"), + TargetUserId: mutedUserID, + EventType: "audio_started", + EventTimeMs: now.Now().UnixMilli(), + Source: "tencent_rtc_callback", + ExternalEventId: "rtc-event-muted-audio-started", + }) + if err != nil { + t.Fatalf("muted RTC audio callback must be acknowledged after cloud re-block: %v", err) + } + if rtcEventResp.GetResult().GetApplied() { + t.Fatalf("muted RTC audio callback must not advance the seat to publishing: %+v", rtcEventResp) + } if _, err := svc.MuteUser(ctx, &roomv1.MuteUserRequest{ Meta: rocketMeta(roomID, ownerID, "unmute-current-rtc-seat"), @@ -128,6 +172,28 @@ func TestMutedUserCannotSpeakOrMicUpUntilUnmuted(t *testing.T) { }); err != nil { t.Fatalf("room unmute must restore explicit RTC mic control: %v", err) } + + directCalls := rtcBlocker.snapshot() + if len(directCalls) != 5 || !directCalls[0] || directCalls[1] || !directCalls[2] || !directCalls[3] || directCalls[4] { + // 四次 MuteUser 分别投影 true/false/true/false；中间被篡改客户端触发的 + // audio_started 必须额外 re-block=true，因此按调用顺序断言服务端防线完整。 + t.Fatalf("unexpected direct RTC audio block sequence: %+v", directCalls) + } + + if err := svc.ProcessPendingOutbox(ctx, roomservice.OutboxWorkerOptions{PublishTimeout: time.Second, BatchSize: 100}); err != nil { + t.Fatalf("process RTC mute outbox failed: %v", err) + } + allCalls := rtcBlocker.snapshot() + if len(allCalls)-len(directCalls) != 4 { + t.Fatalf("each durable RoomUserMuted event must reach RTC compensation: direct=%+v all=%+v", directCalls, allCalls) + } + for index, blocked := range allCalls[len(directCalls):] { + // outbox 中既有 mute 也有 unmute 历史事件；补偿必须读取当前已解除的 Room Cell + // 状态并全部收敛到 false，不能按旧事件载荷重新把用户禁言。 + if blocked { + t.Fatalf("stale RTC mute outbox call %d rolled current state backward", index) + } + } } func TestMicUpReturnsExistingSeatForDuplicateUserCommand(t *testing.T) { diff --git a/services/room-service/internal/room/service/moderation.go b/services/room-service/internal/room/service/moderation.go index e35bbdca..c1e27634 100644 --- a/services/room-service/internal/room/service/moderation.go +++ b/services/room-service/internal/room/service/moderation.go @@ -2,10 +2,12 @@ package service import ( "context" + "log/slog" "time" roomeventsv1 "hyapp.local/api/proto/events/room/v1" roomv1 "hyapp.local/api/proto/room/v1" + "hyapp/pkg/logx" "hyapp/pkg/tencentim" "hyapp/pkg/xerr" "hyapp/services/room-service/internal/room/command" @@ -118,6 +120,20 @@ func (s *Service) MuteUser(ctx context.Context, req *roomv1.MuteUserRequest) (*r return nil, err } + if s.rtcUserAudioBlocker != nil { + // 低延迟路径在 Room Cell 提交后立即更新 RTC；失败不能回滚已经落盘的禁言事实， + // 同一条 RoomUserMuted outbox 会按最新快照继续补偿，避免客户端暂时断线造成永久漏禁。 + blocked := containsID(result.snapshot.GetMuteUserIds(), cmd.TargetUserID) + if err := s.rtcUserAudioBlocker.SetUserAudioBlockedByStrRoomID(ctx, cmd.RoomID(), cmd.TargetUserID, blocked); err != nil { + logx.Warn(ctx, "rtc_user_audio_block_sync_failed", + slog.String("room_id", cmd.RoomID()), + slog.Int64("target_user_id", cmd.TargetUserID), + slog.Bool("blocked", blocked), + slog.String("error", err.Error()), + ) + } + } + return &roomv1.MuteUserResponse{ Result: commandResult(result.applied, result.snapshot.GetVersion(), s.clock.Now()), Room: result.snapshot, diff --git a/services/room-service/internal/room/service/outbox_worker.go b/services/room-service/internal/room/service/outbox_worker.go index e0b21332..ec2de989 100644 --- a/services/room-service/internal/room/service/outbox_worker.go +++ b/services/room-service/internal/room/service/outbox_worker.go @@ -413,6 +413,11 @@ func (s *Service) publishOutboxRecord(ctx context.Context, record outbox.Record) } return s.roomDisplayPublisher.PublishOutboxEvent(ctx, record.Envelope) } + // RTC 禁言是 RoomUserMuted 的外部媒体投影。先按当前 Room Cell 状态收敛，再发布通用 + // MQ/活动消费者；任何一侧失败都会保留同一 outbox 记录重试，且 RTC 调用本身幂等。 + if err := s.syncRTCUserMuteFromOutbox(ctx, record.Envelope); err != nil { + return err + } // Ignited 在 outbox worker 内安排延迟发射，保证 MQ 调度失败能通过 room_outbox retry；客户端 IM 展示已在 Room Cell 提交后直发。 if err := s.scheduleRoomRocketLaunchFromEnvelope(ctx, record.Envelope); err != nil { return err diff --git a/services/room-service/internal/room/service/rtc_event.go b/services/room-service/internal/room/service/rtc_event.go index 19c621fc..b4dc4ab7 100644 --- a/services/room-service/internal/room/service/rtc_event.go +++ b/services/room-service/internal/room/service/rtc_event.go @@ -66,6 +66,13 @@ func (s *Service) ApplyRTCEvent(ctx context.Context, req *roomv1.ApplyRTCEventRe if err != nil { return nil, err } + if cmd.EventType == rtcEventAudioStarted && s.rtcUserAudioBlocker != nil && containsID(result.snapshot.GetMuteUserIds(), cmd.TargetUserID) { + // 被禁言用户绕过客户端直接发流时，腾讯 RTC webhook 是第二道服务端防线。 + // 失败向 gateway 返回错误让腾讯重试；幂等 command replay 仍会再次执行本段外部投影。 + if err := s.rtcUserAudioBlocker.SetUserAudioBlockedByStrRoomID(ctx, cmd.RoomID(), cmd.TargetUserID, true); err != nil { + return nil, err + } + } return &roomv1.ApplyRTCEventResponse{ Result: commandResult(result.applied, result.snapshot.GetVersion(), s.clock.Now()), @@ -84,6 +91,11 @@ func (s *Service) applyRTCAudioStarted(now time.Time, current *state.RoomState, // RTC 事件不拥有业务进房事实；用户不在业务 presence 时只忽略旧事件。 return mutationResult{snapshot: current.ToProto()}, nil, nil } + if current.MuteUsers[cmd.TargetUserID] { + // 房间禁言是音频发布的权威权限；即使篡改客户端已经触发 RTC audio_started， + // 也不能把麦位推进 publishing，外层会同步调用 TRTC 服务端音频封禁并要求失败重试。 + return mutationResult{snapshot: current.ToProto()}, nil, nil + } seat, exists := current.SeatByUser(cmd.TargetUserID) if !exists { diff --git a/services/room-service/internal/room/service/rtc_moderation.go b/services/room-service/internal/room/service/rtc_moderation.go new file mode 100644 index 00000000..74b7e17a --- /dev/null +++ b/services/room-service/internal/room/service/rtc_moderation.go @@ -0,0 +1,38 @@ +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) +} diff --git a/services/room-service/internal/room/service/service.go b/services/room-service/internal/room/service/service.go index e7a98742..b2955256 100644 --- a/services/room-service/internal/room/service/service.go +++ b/services/room-service/internal/room/service/service.go @@ -33,6 +33,8 @@ type Config struct { MicPublishTimeout time.Duration // RTCUserRemover 是平台级封禁后把用户从腾讯 RTC 房间移出的外部边界。 RTCUserRemover integration.RTCUserRemover + // RTCUserAudioBlocker 把 Room Cell 禁言事实投影到腾讯 RTC 服务端音频上行闸门。 + RTCUserAudioBlocker integration.RTCUserAudioBlocker // RoomRocketLaunchScheduler 把倒计时发射唤醒交给外部延迟消息，避免只靠已加载 Cell 扫描。 RoomRocketLaunchScheduler integration.RoomRocketLaunchScheduler // RobotDisplayPublisher 专门 best-effort 投递机器人房间展示事件；失败只打日志，不进入 MySQL outbox。 @@ -117,6 +119,8 @@ type Service struct { luckyGiftSendLockTTL time.Duration // rtcUserRemover 在 Room Cell 提交驱逐后同步移除实时音频房连接。 rtcUserRemover integration.RTCUserRemover + // rtcUserAudioBlocker 在房间禁言提交后关闭目标用户的 RTC 上行音频；状态仍只保存在 Room Cell。 + rtcUserAudioBlocker integration.RTCUserAudioBlocker // draining 表示当前节点正在下线，只允许已经进入执行链路的命令收尾。 draining atomic.Bool @@ -299,6 +303,7 @@ func New(cfg Config, directory router.Directory, repository Repository, wallet i luckyGiftSendLocker: cfg.LuckyGiftSendLocker, luckyGiftSendLockTTL: luckyGiftSendLockTTL, rtcUserRemover: cfg.RTCUserRemover, + rtcUserAudioBlocker: cfg.RTCUserAudioBlocker, cells: make(map[string]*loadedCell), robotRuntimes: make(map[string]robotRoomRuntime), humanRobotRuntimes: make(map[string]robotRoomRuntime),