rtc禁言

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"`

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) {

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,

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 状态重试。

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) {

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,

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

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 {

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)
+}

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),