From 00db43b01ba6ef3381e220953cf02331946b0a9c Mon Sep 17 00:00:00 2001 From: zhx Date: Tue, 21 Jul 2026 18:57:09 +0800 Subject: [PATCH] =?UTF-8?q?=E7=9B=B8=E5=85=B3im=E6=94=B9=E5=8A=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pkg/tencentim/rest_client.go | 34 ++++ pkg/tencentim/rest_client_test.go | 38 ++++ .../internal/client/room_client.go | 2 +- .../http/callbackapi/tencent_im_callback.go | 138 +++++++++------ .../internal/transport/http/response_test.go | 89 +++++++++- services/room-service/internal/app/app.go | 3 + .../internal/integration/clients.go | 6 + .../internal/integration/tencent_im.go | 24 ++- .../internal/integration/tencent_im_test.go | 65 +++++++ .../internal/room/service/guard.go | 2 +- .../internal/room/service/im_moderation.go | 162 ++++++++++++++++++ .../internal/room/service/mic_test.go | 134 +++++++++++---- .../internal/room/service/moderation.go | 17 +- .../internal/room/service/outbox_worker.go | 33 ++-- .../internal/room/service/service.go | 12 ++ .../internal/transport/grpc/server.go | 2 +- 16 files changed, 641 insertions(+), 120 deletions(-) create mode 100644 services/room-service/internal/room/service/im_moderation.go diff --git a/pkg/tencentim/rest_client.go b/pkg/tencentim/rest_client.go index d7612877..8d812b82 100644 --- a/pkg/tencentim/rest_client.go +++ b/pkg/tencentim/rest_client.go @@ -37,8 +37,12 @@ const ( destroyGroupCommand = "v4/group_open_http_svc/destroy_group" // deleteGroupMemberCommand 用于把已离房用户从腾讯群里移除，防止 IM 群成员残留扩大消息面。 deleteGroupMemberCommand = "v4/group_open_http_svc/delete_group_member" + // forbidSendMsgCommand 只修改指定群和指定成员的腾讯 IM 禁言状态，不进入任何消息发送前回调链路。 + forbidSendMsgCommand = "v4/group_open_http_svc/forbid_send_msg" // kickUserCommand 让指定账号现有 IM 登录态失效；被封禁用户需要立即从腾讯 IM SDK 下线。 kickUserCommand = "v4/im_open_login_svc/kick" + // PermanentGroupMuteSeconds 是腾讯 IM 约定的永久禁言秒数；解除禁言使用 0。 + PermanentGroupMuteSeconds uint32 = 1<<32 - 1 ) // RESTConfig 保存服务端调用腾讯云 IM REST API 所需配置。 @@ -464,6 +468,30 @@ func (c *RESTClient) DeleteRoomGroupMember(ctx context.Context, roomID string, u return c.DeleteGroupMember(ctx, roomID, userID) } +// SetGroupMemberMuted 通过腾讯 IM 原生群成员禁言能力投影 Room Cell 状态。 +// 该接口只在禁言事实变化时调用，不参与普通群消息发送，因此不会给共享 SDKAppID 的其它项目增加发送时延。 +func (c *RESTClient) SetGroupMemberMuted(ctx context.Context, groupID string, userID int64, muted bool) error { + groupID = strings.TrimSpace(groupID) + if groupID == "" || userID <= 0 { + return fmt.Errorf("group member mute is incomplete") + } + + muteTime := uint32(0) + if muted { + muteTime = PermanentGroupMuteSeconds + } + request := forbidSendMsgRequest{ + GroupID: groupID, + MemberAccounts: []string{FormatUserID(userID)}, + MuteTime: muteTime, + } + var response restResponse + if err := c.post(ctx, forbidSendMsgCommand, request, &response); err != nil { + return err + } + return response.err() +} + // KickUser 失效指定用户的腾讯云 IM 登录态。 // 该接口不会禁止未来用新 UserSig 登录，因此调用方必须先在业务状态和 token denylist 上完成封禁。 func (c *RESTClient) KickUser(ctx context.Context, userID int64) error { @@ -667,6 +695,12 @@ type deleteGroupMemberRequest struct { Reason string `json:"Reason,omitempty"` } +type forbidSendMsgRequest struct { + GroupID string `json:"GroupId"` + MemberAccounts []string `json:"Members_Account"` + MuteTime uint32 `json:"MuteTime"` +} + type kickUserRequest struct { UserID string `json:"UserID"` } diff --git a/pkg/tencentim/rest_client_test.go b/pkg/tencentim/rest_client_test.go index 5cb4984b..98d2199c 100644 --- a/pkg/tencentim/rest_client_test.go +++ b/pkg/tencentim/rest_client_test.go @@ -328,6 +328,44 @@ func TestRESTClientDeleteRoomGroupMemberBuildsTencentRequest(t *testing.T) { } } +// TestRESTClientSetGroupMemberMutedUsesNativeForbidSendMsg 锁定禁言和解除禁言只调用目标群成员接口。 +func TestRESTClientSetGroupMemberMutedUsesNativeForbidSendMsg(t *testing.T) { + paths := make([]string, 0, 2) + bodies := make([]map[string]any, 0, 2) + client := newTestRESTClient(t, func(request *http.Request) (*http.Response, error) { + paths = append(paths, request.URL.Path) + 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) + return okRESTResponse(), nil + }) + + if err := client.SetGroupMemberMuted(context.Background(), "room-1001", 10002, true); err != nil { + t.Fatalf("SetGroupMemberMuted(true) failed: %v", err) + } + if err := client.SetGroupMemberMuted(context.Background(), "room-1001", 10002, false); err != nil { + t.Fatalf("SetGroupMemberMuted(false) failed: %v", err) + } + + for index, path := range paths { + if path != "/"+forbidSendMsgCommand { + t.Fatalf("request %d path mismatch: got %q", index, path) + } + members, ok := bodies[index]["Members_Account"].([]any) + if !ok || len(members) != 1 || members[0] != "10002" || bodies[index]["GroupId"] != "room-1001" { + t.Fatalf("request %d target mismatch: %+v", index, bodies[index]) + } + } + if bodies[0]["MuteTime"] != float64(PermanentGroupMuteSeconds) { + t.Fatalf("mute must use permanent Tencent duration: %+v", bodies[0]) + } + if bodies[1]["MuteTime"] != float64(0) { + t.Fatalf("unmute must use MuteTime=0: %+v", bodies[1]) + } +} + // TestRESTClientKickUserBuildsTencentRequest 锁定封禁链路使用腾讯云 IM 失效账号登录态接口。 func TestRESTClientKickUserBuildsTencentRequest(t *testing.T) { var capturedPath string diff --git a/services/gateway-service/internal/client/room_client.go b/services/gateway-service/internal/client/room_client.go index a4201f03..fb82c7d7 100644 --- a/services/gateway-service/internal/client/room_client.go +++ b/services/gateway-service/internal/client/room_client.go @@ -36,7 +36,7 @@ type RoomClient interface { } // RoomGuardClient 抽象 gateway 对 room-service 实时守卫 RPC 的依赖。 -// 腾讯云 IM 回调只做字段解析和鉴权，房间 presence、禁言和踢人判断必须回查 room-service。 +// RoomGuardClient 暴露房间权威状态查询；腾讯 IM 发消息不再同步调用该接口，避免把共享 SDKAppID 的发送链路串到 gateway。 type RoomGuardClient interface { CheckSpeakPermission(ctx context.Context, req *roomv1.CheckSpeakPermissionRequest) (*roomv1.CheckSpeakPermissionResponse, error) VerifyRoomPresence(ctx context.Context, req *roomv1.VerifyRoomPresenceRequest) (*roomv1.VerifyRoomPresenceResponse, error) diff --git a/services/gateway-service/internal/transport/http/callbackapi/tencent_im_callback.go b/services/gateway-service/internal/transport/http/callbackapi/tencent_im_callback.go index 53bbbc10..f264d9fb 100644 --- a/services/gateway-service/internal/transport/http/callbackapi/tencent_im_callback.go +++ b/services/gateway-service/internal/transport/http/callbackapi/tencent_im_callback.go @@ -17,6 +17,7 @@ import ( userv1 "hyapp.local/api/proto/user/v1" "hyapp/pkg/appcode" "hyapp/pkg/imgroup" + "hyapp/pkg/xerr" ) const ( @@ -95,7 +96,9 @@ func (h *Handler) handleTencentIMCallback(writer http.ResponseWriter, request *h case tencentCallbackJoinGroup: h.handleTencentIMJoinCallback(writer, request, body) case tencentCallbackSendMsg: - h.handleTencentIMSendCallback(writer, request, body) + // 公屏禁言已经改为 room_outbox 异步调用腾讯 IM forbid_send_msg。 + // 即使控制台误开启旧回调，gateway 也必须纯透传，不能再引入 room-service/数据库同步依赖。 + writeTencentCallback(writer, http.StatusOK, 0, "") case tencentCallbackC2CSent: h.handleTencentIMC2CSentCallback(writer, request, body) default: @@ -152,43 +155,68 @@ func (h *Handler) handleTencentIMC2CSentCallback(writer http.ResponseWriter, req func (h *Handler) handleTencentIMJoinCallback(writer http.ResponseWriter, request *http.Request, body tencentIMCallbackBody) { parsed := imgroup.ParseWithPrefix(body.GroupID, h.tencentIM.GroupIDPrefix) - userID, ok := parseTencentIMUserID(body.Requestor) - if !ok { - writeTencentCallback(writer, http.StatusOK, 1, "join guard input invalid") - return - } - switch parsed.Kind { case imgroup.KindGlobalBroadcast, imgroup.KindRegionBroadcast: + userID, ok := parseTencentIMUserID(body.Requestor) + if !ok { + writeTencentCallback(writer, http.StatusOK, 1, "join guard input invalid") + return + } // 播报群不属于 Room Cell，准入只看 user-service 的用户状态、资料完成度、app_code 和区域归属。 h.handleTencentIMBroadcastJoinCallback(writer, request, parsed, userID) return case imgroup.KindRoom: - // 房间群仍走 room-service guard，保证只有当前房间 presence 用户能加入/留在房间 IM 群。 - h.handleTencentIMRoomJoinCallback(writer, request, parsed.RoomID, userID) + callbackAppCode, managed, err := h.resolveTencentIMRoomOwnership(request, parsed.RoomID) + if err != nil { + h.logTencentIMRoomCallbackDecision(request, tencentCallbackJoinGroup, body.GroupID, "rejected", "room_owner_unavailable", "") + writeTencentCallback(writer, http.StatusOK, 1, "room guard unavailable") + return + } + if !managed { + // 腾讯 IM 回调按 SDKAppID 全局生效；共享 SDKAppID 下，非本服务房间必须透传给实际项目。 + h.logTencentIMRoomCallbackDecision(request, tencentCallbackJoinGroup, body.GroupID, "passthrough", "room_not_managed", "") + writeTencentCallback(writer, http.StatusOK, 0, "") + return + } + userID, ok := parseTencentIMUserID(body.Requestor) + if !ok { + writeTencentCallback(writer, http.StatusOK, 1, "join guard input invalid") + return + } + // 原始 room GroupID 没有租户前缀，必须先由 room-service 持久事实解析真实 app_code 再校验 presence。 + h.handleTencentIMRoomJoinCallback(writer, request, parsed.RoomID, userID, callbackAppCode) return default: - writeTencentCallback(writer, http.StatusOK, 1, "group_id is invalid") + if h.tencentIMGroupIDReserved(body.GroupID) { + h.logTencentIMRoomCallbackDecision(request, tencentCallbackJoinGroup, body.GroupID, "rejected", "reserved_group_invalid", "") + writeTencentCallback(writer, http.StatusOK, 1, "group_id is invalid") + return + } + // 不属于当前环境 HyApp 保留命名空间的群由共享 SDKAppID 中的其它项目负责，当前回调不能阻断。 + h.logTencentIMRoomCallbackDecision(request, tencentCallbackJoinGroup, body.GroupID, "passthrough", "group_not_managed", "") + writeTencentCallback(writer, http.StatusOK, 0, "") return } } -func (h *Handler) handleTencentIMRoomJoinCallback(writer http.ResponseWriter, request *http.Request, roomID string, userID int64) { +func (h *Handler) handleTencentIMRoomJoinCallback(writer http.ResponseWriter, request *http.Request, roomID string, userID int64, callbackAppCode string) { if roomID == "" || h.roomGuardClient == nil { writeTencentCallback(writer, http.StatusOK, 1, "join guard input invalid") return } - resp, err := h.roomGuardClient.VerifyRoomPresence(request.Context(), &roomv1.VerifyRoomPresenceRequest{ + resp, err := h.roomGuardClient.VerifyRoomPresence(appcode.WithContext(request.Context(), callbackAppCode), &roomv1.VerifyRoomPresenceRequest{ RoomId: roomID, UserId: userID, RequestId: httpkit.RequestIDFromContext(request.Context()), - AppCode: appcode.FromContext(request.Context()), + AppCode: callbackAppCode, }) if err != nil || !resp.GetPresent() { + h.logTencentIMRoomCallbackDecision(request, tencentCallbackJoinGroup, roomID, "rejected", tencentGuardErrorInfo(err, resp.GetReason()), callbackAppCode) writeTencentCallback(writer, http.StatusOK, 1, tencentGuardErrorInfo(err, resp.GetReason())) return } + h.logTencentIMRoomCallbackDecision(request, tencentCallbackJoinGroup, roomID, "allowed", "presence_verified", callbackAppCode) writeTencentCallback(writer, http.StatusOK, 0, "") } @@ -207,48 +235,60 @@ func (h *Handler) handleTencentIMBroadcastJoinCallback(writer http.ResponseWrite writeTencentCallback(writer, http.StatusOK, 0, "") } -func (h *Handler) handleTencentIMSendCallback(writer http.ResponseWriter, request *http.Request, body tencentIMCallbackBody) { - parsed := imgroup.ParseWithPrefix(body.GroupID, h.tencentIM.GroupIDPrefix) - switch parsed.Kind { - case imgroup.KindGlobalBroadcast, imgroup.KindRegionBroadcast: - if h.tencentIMCallbackSenderIsServer(body) { - // 播报群是服务端只写通道；管理员账号发送的 REST 消息由腾讯回调放行。 - writeTencentCallback(writer, http.StatusOK, 0, "") - return - } - writeTencentCallback(writer, http.StatusOK, 1, "broadcast group is server-only") - return - case imgroup.KindRoom: - userID, ok := parseTencentIMUserID(body.FromAccount) - if !ok { - writeTencentCallback(writer, http.StatusOK, 1, "send guard input invalid") - return - } - h.handleTencentIMRoomSendCallback(writer, request, parsed.RoomID, userID) - return - default: - writeTencentCallback(writer, http.StatusOK, 1, "group_id is invalid") - return +// resolveTencentIMRoomOwnership 只把 room-service 持久表中存在的 room_id 认作本项目房间。 +// NOT_FOUND 是共享 SDKAppID 的正常外部流量；其余错误可能隐藏真实 HyApp 房间，必须保持 fail-closed。 +func (h *Handler) resolveTencentIMRoomOwnership(request *http.Request, roomID string) (string, bool, error) { + if h.roomGuardClient == nil { + return "", false, fmt.Errorf("room guard unavailable") } + resp, err := h.roomGuardClient.ResolveRoomAppCode(request.Context(), &roomv1.ResolveRoomAppCodeRequest{ + RequestId: httpkit.RequestIDFromContext(request.Context()), + RoomId: roomID, + }) + if err != nil { + if xerr.ReasonFromGRPC(err) == xerr.NotFound { + return "", false, nil + } + return "", false, err + } + callbackAppCode := appcode.Normalize(resp.GetAppCode()) + if strings.TrimSpace(resp.GetAppCode()) == "" { + return "", false, fmt.Errorf("room app_code is empty") + } + return callbackAppCode, true, nil } -func (h *Handler) handleTencentIMRoomSendCallback(writer http.ResponseWriter, request *http.Request, roomID string, userID int64) { - if roomID == "" || h.roomGuardClient == nil { - writeTencentCallback(writer, http.StatusOK, 1, "send guard input invalid") - return +// tencentIMGroupIDReserved 只保留当前环境自己的播报群命名空间。 +// 配置了环境前缀时，未携带该前缀的同名群属于其它环境或项目，不能被当前回调拒绝。 +func (h *Handler) tencentIMGroupIDReserved(groupID string) bool { + groupID = strings.TrimSpace(groupID) + prefix := strings.TrimSpace(h.tencentIM.GroupIDPrefix) + if prefix != "" { + var ok bool + groupID, ok = strings.CutPrefix(groupID, prefix) + if !ok { + return false + } } + return strings.HasPrefix(groupID, "hy_") && strings.Contains(groupID, "_bc_") +} - resp, err := h.roomGuardClient.CheckSpeakPermission(request.Context(), &roomv1.CheckSpeakPermissionRequest{ - RoomId: roomID, - UserId: userID, - AppCode: appcode.FromContext(request.Context()), - }) - if err != nil || !resp.GetAllowed() { - writeTencentCallback(writer, http.StatusOK, 1, tencentGuardErrorInfo(err, resp.GetReason())) - return +// logTencentIMRoomCallbackDecision 记录可审计但不含签名、消息正文和账号的路由结果。 +// HTTP access log 只能看到 200；该业务日志用于区分腾讯响应体中的允许、拒绝和共享项目透传。 +func (h *Handler) logTencentIMRoomCallbackDecision(request *http.Request, command string, groupID string, decision string, reason string, callbackAppCode string) { + loggedAppCode := strings.TrimSpace(callbackAppCode) + if loggedAppCode != "" { + loggedAppCode = appcode.Normalize(loggedAppCode) } - - writeTencentCallback(writer, http.StatusOK, 0, "") + slog.LogAttrs(request.Context(), slog.LevelInfo, "tencent_im_room_callback_decision", + slog.String("request_id", httpkit.RequestIDFromContext(request.Context())), + slog.String("callback_command", command), + slog.String("group_id", strings.TrimSpace(groupID)), + slog.String("app_code", loggedAppCode), + slog.String("decision", decision), + slog.String("reason", strings.TrimSpace(reason)), + slog.String("opt_platform", strings.TrimSpace(request.URL.Query().Get("OptPlatform"))), + ) } func (h *Handler) tencentIMCallbackUser(request *http.Request, callbackAppCode string, userID int64) (*userv1.User, error) { diff --git a/services/gateway-service/internal/transport/http/response_test.go b/services/gateway-service/internal/transport/http/response_test.go index 1986ce20..d832b458 100644 --- a/services/gateway-service/internal/transport/http/response_test.go +++ b/services/gateway-service/internal/transport/http/response_test.go @@ -2875,6 +2875,7 @@ type fakeRoomGuardClient struct { lastPresence *roomv1.VerifyRoomPresenceRequest lastRoomApp *roomv1.ResolveRoomAppCodeRequest err error + roomAppErr error } func (f *fakeRoomGuardClient) CheckSpeakPermission(_ context.Context, req *roomv1.CheckSpeakPermissionRequest) (*roomv1.CheckSpeakPermissionResponse, error) { @@ -2903,6 +2904,9 @@ func (f *fakeRoomGuardClient) VerifyRoomPresence(_ context.Context, req *roomv1. func (f *fakeRoomGuardClient) ResolveRoomAppCode(_ context.Context, req *roomv1.ResolveRoomAppCodeRequest) (*roomv1.ResolveRoomAppCodeResponse, error) { f.lastRoomApp = req + if f.roomAppErr != nil { + return nil, f.roomAppErr + } if f.err != nil { return nil, f.err } @@ -12116,9 +12120,15 @@ func TestTencentIMJoinCallbackUsesPresenceGuard(t *testing.T) { router.ServeHTTP(recorder, request) assertTencentCallback(t, recorder, http.StatusOK, 0) + if guard.lastRoomApp == nil || guard.lastRoomApp.GetRoomId() != "room-1" { + t.Fatalf("room ownership request mismatch: %+v", guard.lastRoomApp) + } if guard.lastPresence == nil || guard.lastPresence.GetRoomId() != "room-1" || guard.lastPresence.GetUserId() != 42 || guard.lastPresence.GetRequestId() != assertGeneratedRequestID(t, recorder, "req-im-join") { t.Fatalf("presence guard request mismatch: %+v", guard.lastPresence) } + if guard.lastPresence.GetAppCode() != "lalu" { + t.Fatalf("presence guard must use resolved app_code, got %+v", guard.lastPresence) + } guard.presenceResp = &roomv1.VerifyRoomPresenceResponse{Present: false, Reason: "not_in_room"} deniedRequest := httptest.NewRequest(http.MethodPost, "/api/v1/tencent-im/callback?SdkAppid=1400000000&CallbackCommand=Group.CallbackBeforeApplyJoinGroup&CallbackAuthToken=callback-token", bytes.NewReader(body)) @@ -12128,7 +12138,7 @@ func TestTencentIMJoinCallbackUsesPresenceGuard(t *testing.T) { assertTencentCallback(t, deniedRecorder, http.StatusOK, 1) } -func TestTencentIMSendCallbackUsesSpeakGuard(t *testing.T) { +func TestTencentIMSendCallbackNeverEntersRoomGuard(t *testing.T) { guard := &fakeRoomGuardClient{ speakResp: &roomv1.CheckSpeakPermissionResponse{Allowed: false, Reason: "user_muted"}, } @@ -12141,13 +12151,77 @@ func TestTencentIMSendCallbackUsesSpeakGuard(t *testing.T) { router.ServeHTTP(recorder, request) - assertTencentCallback(t, recorder, http.StatusOK, 1) - if guard.lastSpeak == nil || guard.lastSpeak.GetRoomId() != "room-1" || guard.lastSpeak.GetUserId() != 42 { - t.Fatalf("speak guard request mismatch: %+v", guard.lastSpeak) + assertTencentCallback(t, recorder, http.StatusOK, 0) + if guard.lastRoomApp != nil { + t.Fatalf("before-send callback must not resolve room ownership: %+v", guard.lastRoomApp) + } + if guard.lastSpeak != nil { + t.Fatalf("before-send callback must not enter room speak guard: %+v", guard.lastSpeak) } } -func TestTencentIMBroadcastCallbacksUseRegionGuardAndServerOnlySend(t *testing.T) { +func TestTencentIMRoomCallbacksPassthroughGroupsOutsideHyAppOwnership(t *testing.T) { + guard := &fakeRoomGuardClient{roomAppErr: status.Error(codes.NotFound, "room not found")} + router := NewHandlerWithConfig(&fakeRoomClient{}, guard, nil, nil, TencentIMConfig{ + SDKAppID: 1400000000, + }).Routes(auth.NewVerifier("secret")) + + sendBody := []byte(`{"CallbackCommand":"Group.CallbackBeforeSendMsg","GroupId":"other-project-room","From_Account":"external-user"}`) + sendRequest := httptest.NewRequest(http.MethodPost, "/api/v1/tencent-im/callback?SdkAppid=1400000000&CallbackCommand=Group.CallbackBeforeSendMsg&OptPlatform=RESTAPI", bytes.NewReader(sendBody)) + sendRecorder := httptest.NewRecorder() + router.ServeHTTP(sendRecorder, sendRequest) + assertTencentCallback(t, sendRecorder, http.StatusOK, 0) + if guard.lastSpeak != nil { + t.Fatalf("external room send must not reach HyApp speak guard: %+v", guard.lastSpeak) + } + + joinBody := []byte(`{"CallbackCommand":"Group.CallbackBeforeApplyJoinGroup","GroupId":"@TGS#external","Requestor_Account":"external-user"}`) + joinRequest := httptest.NewRequest(http.MethodPost, "/api/v1/tencent-im/callback?SdkAppid=1400000000&CallbackCommand=Group.CallbackBeforeApplyJoinGroup&OptPlatform=Android", bytes.NewReader(joinBody)) + joinRecorder := httptest.NewRecorder() + router.ServeHTTP(joinRecorder, joinRequest) + assertTencentCallback(t, joinRecorder, http.StatusOK, 0) + if guard.lastPresence != nil { + t.Fatalf("external group join must not reach HyApp presence guard: %+v", guard.lastPresence) + } +} + +func TestTencentIMRoomCallbacksFailClosedWhenOwnershipLookupFails(t *testing.T) { + guard := &fakeRoomGuardClient{roomAppErr: status.Error(codes.Unavailable, "room service unavailable")} + router := NewHandlerWithConfig(&fakeRoomClient{}, guard, nil, nil, TencentIMConfig{ + SDKAppID: 1400000000, + }).Routes(auth.NewVerifier("secret")) + body := []byte(`{"CallbackCommand":"Group.CallbackBeforeApplyJoinGroup","GroupId":"room-1","Requestor_Account":"42"}`) + request := httptest.NewRequest(http.MethodPost, "/api/v1/tencent-im/callback?SdkAppid=1400000000&CallbackCommand=Group.CallbackBeforeApplyJoinGroup", bytes.NewReader(body)) + recorder := httptest.NewRecorder() + router.ServeHTTP(recorder, request) + + assertTencentCallback(t, recorder, http.StatusOK, 1) + if guard.lastPresence != nil { + t.Fatalf("ownership lookup failure must stop before the presence guard: %+v", guard.lastPresence) + } +} + +func TestTencentIMRoomSendCallbackPassthroughDoesNotResolveOwnership(t *testing.T) { + guard := &fakeRoomGuardClient{roomAppResp: &roomv1.ResolveRoomAppCodeResponse{AppCode: "fami"}} + router := NewHandlerWithConfig(&fakeRoomClient{}, guard, nil, nil, TencentIMConfig{ + SDKAppID: 1400000000, + AdminIdentifier: "administrator", + }).Routes(auth.NewVerifier("secret")) + body := []byte(`{"CallbackCommand":"Group.CallbackBeforeSendMsg","GroupId":"room-1","From_Account":"administrator"}`) + request := httptest.NewRequest(http.MethodPost, "/api/v1/tencent-im/callback?SdkAppid=1400000000&CallbackCommand=Group.CallbackBeforeSendMsg&OptPlatform=RESTAPI", bytes.NewReader(body)) + recorder := httptest.NewRecorder() + router.ServeHTTP(recorder, request) + + assertTencentCallback(t, recorder, http.StatusOK, 0) + if guard.lastRoomApp != nil { + t.Fatalf("before-send passthrough must not resolve room ownership: %+v", guard.lastRoomApp) + } + if guard.lastSpeak != nil { + t.Fatalf("before-send passthrough must not evaluate room speech: %+v", guard.lastSpeak) + } +} + +func TestTencentIMBroadcastCallbacksUseRegionGuardAndSendPassthrough(t *testing.T) { profileClient := &fakeUserProfileClient{regionID: 1001} guard := &fakeRoomGuardClient{} router := NewHandlerWithConfig(&fakeRoomClient{}, guard, nil, nil, TencentIMConfig{ @@ -12175,7 +12249,7 @@ func TestTencentIMBroadcastCallbacksUseRegionGuardAndServerOnlySend(t *testing.T userSendRequest := httptest.NewRequest(http.MethodPost, "/api/v1/tencent-im/callback?SdkAppid=1400000000&CallbackCommand=Group.CallbackBeforeSendMsg", bytes.NewReader(userSendBody)) userSendRecorder := httptest.NewRecorder() router.ServeHTTP(userSendRecorder, userSendRequest) - assertTencentCallback(t, userSendRecorder, http.StatusOK, 1) + assertTencentCallback(t, userSendRecorder, http.StatusOK, 0) adminSendBody := []byte(`{"CallbackCommand":"Group.CallbackBeforeSendMsg","GroupId":"hy_lalu_bc_r_1001","From_Account":"administrator"}`) adminSendRequest := httptest.NewRequest(http.MethodPost, "/api/v1/tencent-im/callback?SdkAppid=1400000000&CallbackCommand=Group.CallbackBeforeSendMsg", bytes.NewReader(adminSendBody)) @@ -12202,7 +12276,8 @@ func TestTencentIMBroadcastCallbacksUseConfiguredGroupPrefix(t *testing.T) { oldGroupRequest := httptest.NewRequest(http.MethodPost, "/api/v1/tencent-im/callback?SdkAppid=1400000000&CallbackCommand=Group.CallbackBeforeApplyJoinGroup", bytes.NewReader(oldGroupBody)) oldGroupRecorder := httptest.NewRecorder() router.ServeHTTP(oldGroupRecorder, oldGroupRequest) - assertTencentCallback(t, oldGroupRecorder, http.StatusOK, 1) + // 配置 test_ 代表当前回调只拥有 test_ 命名空间；无前缀群可能属于共享 SDKAppID 的其它环境，必须透传。 + assertTencentCallback(t, oldGroupRecorder, http.StatusOK, 0) } func TestTencentIMCallbackAuthAndUnknownCommand(t *testing.T) { diff --git a/services/room-service/internal/app/app.go b/services/room-service/internal/app/app.go index 5e5827ed..1f8ccf73 100644 --- a/services/room-service/internal/app/app.go +++ b/services/room-service/internal/app/app.go @@ -166,6 +166,7 @@ func New(cfg config.Config) (*App, error) { outboxPublishers := make([]integration.OutboxPublisher, 0, 3) var rtcUserRemover integration.RTCUserRemover var rtcUserAudioBlocker integration.RTCUserAudioBlocker + var imUserMuter integration.IMUserMuter var tencentPublisher integration.OutboxPublisher if cfg.TencentIM.Enabled { // 腾讯云 IM 替代自研 IM；room-service 只负责服务端 REST 群消息桥接。 @@ -179,6 +180,7 @@ func New(cfg config.Config) (*App, error) { } concreteTencentPublisher := integration.NewTencentIMPublisher(tencentClient) tencentPublisher = concreteTencentPublisher + imUserMuter = concreteTencentPublisher roomPublisher = concreteTencentPublisher // 图片消息和 VIP 房间装扮都是用户可见的持久事实，必须由 durable room_outbox 补偿腾讯 IM； // 只过滤这两类事件，避免把现有 direct-IM 展示事件整体接入后制造双消息。 @@ -251,6 +253,7 @@ func New(cfg config.Config) (*App, error) { MicPublishTimeout: cfg.MicPublishTimeout, RTCUserRemover: rtcUserRemover, RTCUserAudioBlocker: rtcUserAudioBlocker, + IMUserMuter: imUserMuter, 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 35c5218e..d418e88e 100644 --- a/services/room-service/internal/integration/clients.go +++ b/services/room-service/internal/integration/clients.go @@ -67,6 +67,12 @@ type RTCUserAudioBlocker interface { SetUserAudioBlockedByStrRoomID(ctx context.Context, roomID string, userID int64, blocked bool) error } +// IMUserMuter 抽象腾讯 IM 群成员禁言投影。 +// Room Cell 仍是唯一权威状态；实现只修改指定房间群和指定用户，绝不能注册消息发送前回调。 +type IMUserMuter interface { + SetRoomGroupMemberMuted(ctx context.Context, roomID string, userID int64, muted bool) error +} + // OutboxPublisher 抽象 outbox worker 对外部消费者的异步投递。 type OutboxPublisher interface { // PublishOutboxEvent 投递 protobuf outbox 信封，失败由 room_outbox 转为 retryable 状态重试。 diff --git a/services/room-service/internal/integration/tencent_im.go b/services/room-service/internal/integration/tencent_im.go index 7a8ccc4c..032f6de6 100644 --- a/services/room-service/internal/integration/tencent_im.go +++ b/services/room-service/internal/integration/tencent_im.go @@ -64,6 +64,28 @@ func (p *TencentIMPublisher) PublishRoomEvent(ctx context.Context, event tencent return nil } +// SetRoomGroupMemberMuted 把 Room Cell 禁言状态投影到腾讯 IM 原生群成员禁言。 +// 群缺失时沿用房间消息补偿策略先补建再重试；其它错误返回 outbox worker 继续退避重试。 +func (p *TencentIMPublisher) SetRoomGroupMemberMuted(ctx context.Context, roomID string, userID int64, muted bool) error { + if p == nil || p.client == nil { + return nil + } + groupID, err := imgroup.RoomGroupID(roomID) + if err != nil { + return err + } + if err := p.client.SetGroupMemberMuted(ctx, groupID, userID, muted); err != nil { + if !tencentim.IsRESTErrorCode(err, 10010, 10015) { + return err + } + if ensureErr := p.client.EnsureRoomGroup(ctx, groupID, 0); ensureErr != nil { + return fmt.Errorf("ensure room im group after mute failure: %w", ensureErr) + } + return p.client.SetGroupMemberMuted(ctx, groupID, userID, muted) + } + return nil +} + // PublishOutboxEvent 补偿投递 room_outbox 中可转成房间系统消息的事件。 func (p *TencentIMPublisher) PublishOutboxEvent(ctx context.Context, envelope *roomeventsv1.EventEnvelope) error { if p == nil || p.client == nil || envelope == nil { @@ -295,7 +317,7 @@ func roomEventFromEnvelope(envelope *roomeventsv1.EventEnvelope) (tencentim.Room appendRoomMediaAttributes(base.Attributes, body.GetImage(), "image_") return base, true, nil case "RoomChatEnabledChanged": - // 公屏开关事件提示客户端更新输入态，真正发言仍由 CheckSpeakPermission 拦截。 + // 公屏开关事件只提示客户端更新输入态；单用户服务端禁言由 forbid_send_msg 的独立 outbox 投影负责。 var body roomeventsv1.RoomChatEnabledChanged if err := proto.Unmarshal(envelope.GetBody(), &body); err != nil { return tencentim.RoomEvent{}, false, err diff --git a/services/room-service/internal/integration/tencent_im_test.go b/services/room-service/internal/integration/tencent_im_test.go index 13639da4..d173ad22 100644 --- a/services/room-service/internal/integration/tencent_im_test.go +++ b/services/room-service/internal/integration/tencent_im_test.go @@ -621,6 +621,71 @@ func TestTencentIMPublisherEnsuresRoomGroupAndRetriesWhenGroupMissing(t *testing } } +func TestTencentIMPublisherEnsuresRoomGroupAndRetriesMemberMuteWhenGroupMissing(t *testing.T) { + paths := make([]string, 0, 3) + bodies := make([]map[string]any, 0, 3) + client, err := tencentim.NewRESTClient(tencentim.RESTConfig{ + SDKAppID: 1400000000, + SecretKey: "secret", + AdminIdentifier: "administrator", + AdminUserSigTTL: time.Hour, + HTTPClient: &http.Client{Transport: roundTripFunc(func(request *http.Request) (*http.Response, error) { + paths = append(paths, request.URL.Path) + 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) + if len(paths) == 1 { + return &http.Response{ + StatusCode: http.StatusOK, + Body: io.NopCloser(bytes.NewReader([]byte(`{"ActionStatus":"FAIL","ErrorCode":10010,"ErrorInfo":"group not exist"}`))), + Header: make(http.Header), + }, nil + } + return &http.Response{ + StatusCode: http.StatusOK, + Body: io.NopCloser(bytes.NewReader([]byte(`{"ActionStatus":"OK","ErrorCode":0,"ErrorInfo":""}`))), + Header: make(http.Header), + }, nil + })}, + }) + if err != nil { + t.Fatalf("NewRESTClient failed: %v", err) + } + + publisher := NewTencentIMPublisher(client) + if err := publisher.SetRoomGroupMemberMuted(context.Background(), "room-mute", 42, true); err != nil { + t.Fatalf("SetRoomGroupMemberMuted should retry after ensuring group: %v", err) + } + + wantPaths := []string{ + "/v4/group_open_http_svc/forbid_send_msg", + "/v4/group_open_http_svc/create_group", + "/v4/group_open_http_svc/forbid_send_msg", + } + if len(paths) != len(wantPaths) { + t.Fatalf("path count mismatch: paths=%+v bodies=%+v", paths, bodies) + } + for index, want := range wantPaths { + if paths[index] != want { + t.Fatalf("path[%d] mismatch: got %s want %s", index, paths[index], want) + } + if bodies[index]["GroupId"] != "room-mute" { + t.Fatalf("request %d must preserve room group id: %+v", index, bodies[index]) + } + } + for _, index := range []int{0, 2} { + if bodies[index]["MuteTime"] != float64(tencentim.PermanentGroupMuteSeconds) { + t.Fatalf("mute request %d must use Tencent permanent mute value: %+v", index, bodies[index]) + } + accounts, ok := bodies[index]["Members_Account"].([]any) + if !ok || len(accounts) != 1 || accounts[0] != "42" { + t.Fatalf("mute request %d must target only user 42: %+v", index, bodies[index]) + } + } +} + 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/room/service/guard.go b/services/room-service/internal/room/service/guard.go index c9419485..5ff41db9 100644 --- a/services/room-service/internal/room/service/guard.go +++ b/services/room-service/internal/room/service/guard.go @@ -7,7 +7,7 @@ import ( "hyapp/pkg/appcode" ) -// CheckSpeakPermission 给腾讯云 IM 回调或 gateway 提供同步发言守卫。 +// CheckSpeakPermission 给 gateway 等受控业务入口提供同步发言守卫；腾讯 IM 普通消息不再进入本服务。 func (s *Service) CheckSpeakPermission(ctx context.Context, req *roomv1.CheckSpeakPermissionRequest) (*roomv1.CheckSpeakPermissionResponse, error) { ctx = appcode.WithContext(ctx, req.GetAppCode()) closed, err := s.isRoomClosed(ctx, req.GetRoomId()) diff --git a/services/room-service/internal/room/service/im_moderation.go b/services/room-service/internal/room/service/im_moderation.go new file mode 100644 index 00000000..f015fb36 --- /dev/null +++ b/services/room-service/internal/room/service/im_moderation.go @@ -0,0 +1,162 @@ +package service + +import ( + "context" + "fmt" + "strings" + "time" + + "google.golang.org/protobuf/proto" + roomeventsv1 "hyapp.local/api/proto/events/room/v1" + "hyapp/pkg/appcode" +) + +const ( + // forbid_send_msg 官方单接口上限为 200 QPS；双节点各 100 QPS，避免共享 SDKAppID 被单服务打满。 + imMuteProjectionInterval = 10 * time.Millisecond + // 云 API 执行期间若 Room Cell 状态继续变化，最多在同一门闩内追赶三次，之后交给 outbox 退避重试。 + imMuteProjectionMaxAttempts = 3 + // REST client 单次上限为 5s；总预算额外覆盖进程内门闩、限速和最新快照复核。 + imMuteProjectionTimeout = 6 * time.Second +) + +type imMuteProjectionKey struct { + appCode string + roomID string + userID int64 +} + +// imMuteProjectionGate 让同一 app/room/user 的旧 mute 和新 unmute 在单进程内串行， +// 同时允许等待者遵守 context deadline；Room Cell 最新状态仍是每次云端写入的唯一输入。 +type imMuteProjectionGate struct { + token chan struct{} + refs int +} + +// syncIMUserMuteFromOutbox 只消费已经随 Room Cell 命令原子提交的 RoomUserMuted 事实。 +// 普通群消息不会调用该路径，因此腾讯 IM 回调关闭后，共享 SDKAppID 的其它项目不再依赖本服务延迟或可用性。 +func (s *Service) syncIMUserMuteFromOutbox(ctx context.Context, envelope *roomeventsv1.EventEnvelope) error { + if s == nil || s.imUserMuter == 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 im mute outbox event: %w", err) + } + if event.GetTargetUserId() <= 0 || strings.TrimSpace(envelope.GetRoomId()) == "" { + return fmt.Errorf("im mute outbox event is incomplete") + } + return s.reconcileIMUserMute(ctx, envelope.GetRoomId(), event.GetTargetUserId()) +} + +// reconcileIMUserMute 在每次 forbid_send_msg 前后重读 Room Cell。 +// outbox 可能因网络失败重放旧 mute 事件，不能直接采用历史 payload.muted，否则会覆盖较新的 unmute。 +func (s *Service) reconcileIMUserMute(parent context.Context, roomID string, userID int64) error { + if s == nil || s.imUserMuter == nil { + return nil + } + roomID = strings.TrimSpace(roomID) + if roomID == "" || userID <= 0 { + return fmt.Errorf("im mute reconcile target is incomplete") + } + + ctx, cancel := context.WithTimeout(parent, imMuteProjectionTimeout) + defer cancel() + release, err := s.acquireIMMuteProjection(ctx, imMuteProjectionKey{ + appCode: appcode.FromContext(ctx), + roomID: roomID, + userID: userID, + }) + if err != nil { + return err + } + defer release() + + for attempt := 1; attempt <= imMuteProjectionMaxAttempts; attempt++ { + before, err := s.currentSnapshot(ctx, roomID) + if err != nil { + return err + } + if before == nil { + // 房间已被物理清理时群也不再是活跃投递通道，历史禁言投影可以直接结束。 + return nil + } + desiredMuted := containsID(before.GetMuteUserIds(), userID) + if err := s.waitIMMuteProjectionRate(ctx); err != nil { + return err + } + if err := s.imUserMuter.SetRoomGroupMemberMuted(ctx, roomID, userID, desiredMuted); err != nil { + return err + } + + after, err := s.currentSnapshot(ctx, roomID) + if err != nil { + return err + } + if after == nil || containsID(after.GetMuteUserIds(), userID) == desiredMuted { + return nil + } + // 状态在腾讯请求飞行期间变化；保持同一 keyed gate 并立即投影最新值，避免本轮留下旧云端状态。 + } + return fmt.Errorf("im mute state kept changing during %d reconcile attempts", imMuteProjectionMaxAttempts) +} + +func (s *Service) acquireIMMuteProjection(ctx context.Context, key imMuteProjectionKey) (func(), error) { + s.imMuteGateMu.Lock() + if s.imMuteGates == nil { + s.imMuteGates = make(map[imMuteProjectionKey]*imMuteProjectionGate) + } + gate := s.imMuteGates[key] + if gate == nil { + gate = &imMuteProjectionGate{token: make(chan struct{}, 1)} + gate.token <- struct{}{} + s.imMuteGates[key] = gate + } + gate.refs++ + s.imMuteGateMu.Unlock() + + select { + case <-ctx.Done(): + s.releaseIMMuteProjectionRef(key, gate) + return nil, ctx.Err() + case <-gate.token: + return func() { + gate.token <- struct{}{} + s.releaseIMMuteProjectionRef(key, gate) + }, nil + } +} + +func (s *Service) releaseIMMuteProjectionRef(key imMuteProjectionKey, gate *imMuteProjectionGate) { + s.imMuteGateMu.Lock() + defer s.imMuteGateMu.Unlock() + gate.refs-- + if gate.refs == 0 && s.imMuteGates[key] == gate { + delete(s.imMuteGates, key) + } +} + +func (s *Service) waitIMMuteProjectionRate(ctx context.Context) error { + s.imMuteRateMu.Lock() + now := time.Now() + slot := now + if s.imMuteNextCall.After(slot) { + slot = s.imMuteNextCall + } + s.imMuteNextCall = slot.Add(imMuteProjectionInterval) + s.imMuteRateMu.Unlock() + + wait := time.Until(slot) + if wait <= 0 { + return nil + } + timer := time.NewTimer(wait) + defer timer.Stop() + select { + case <-ctx.Done(): + return ctx.Err() + case <-timer.C: + return nil + } +} diff --git a/services/room-service/internal/room/service/mic_test.go b/services/room-service/internal/room/service/mic_test.go index f539f36d..4c09fcdf 100644 --- a/services/room-service/internal/room/service/mic_test.go +++ b/services/room-service/internal/room/service/mic_test.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "strings" "sync" "testing" "time" @@ -24,6 +25,25 @@ type recordingRTCUserAudioBlocker struct { err error } +type recordingIMUserMuter struct { + mu sync.Mutex + calls []bool + err error +} + +func (m *recordingIMUserMuter) SetRoomGroupMemberMuted(_ context.Context, _ string, _ int64, muted bool) error { + m.mu.Lock() + defer m.mu.Unlock() + m.calls = append(m.calls, muted) + return m.err +} + +func (m *recordingIMUserMuter) snapshot() []bool { + m.mu.Lock() + defer m.mu.Unlock() + return append([]bool(nil), m.calls...) +} + func (b *recordingRTCUserAudioBlocker) SetUserAudioBlockedByStrRoomID(_ context.Context, _ string, _ int64, blocked bool) error { b.mu.Lock() defer b.mu.Unlock() @@ -42,6 +62,7 @@ func TestMutedUserCannotSpeakOrMicUpUntilUnmuted(t *testing.T) { repository := mysqltest.NewRepository(t) now := &fixedRoomRocketClock{now: time.Date(2026, 7, 21, 8, 0, 0, 0, time.UTC)} rtcBlocker := &recordingRTCUserAudioBlocker{} + imMuter := &recordingIMUserMuter{} svc := roomservice.New(roomservice.Config{ NodeID: "node-muted-rtc-test", LeaseTTL: 10 * time.Second, @@ -49,6 +70,7 @@ func TestMutedUserCannotSpeakOrMicUpUntilUnmuted(t *testing.T) { SnapshotEveryN: 1, Clock: now, RTCUserAudioBlocker: rtcBlocker, + IMUserMuter: imMuter, }, router.NewMemoryDirectory(), repository, &rocketTestWallet{}, integration.NewNoopRoomEventPublisher(), integration.NewNoopOutboxPublisher()) roomID := "room-muted-user-guard" @@ -109,10 +131,9 @@ func TestMutedUserCannotSpeakOrMicUpUntilUnmuted(t *testing.T) { }); err != nil { t.Fatalf("old mute command replay failed: %v", err) } - if got := rtcBlocker.snapshot(); len(got) != 3 || got[2] { - // pipeline 会向重放返回第一次 mute 的历史 result.snapshot；RTC 投影 - // 必须忽略它并重读 newer unmute，否则重试会把云端永久回退为禁音。 - t.Fatalf("old mute replay must reconcile latest unmuted state: %+v", got) + if got := rtcBlocker.snapshot(); len(got) != 0 { + // MuteUser 只提交 Room Cell 与 outbox；旧命令重放和新命令都不能把腾讯 RTC 慢调用带入同步请求。 + t.Fatalf("room mute command must not call Tencent RTC before durable outbox processing: %+v", got) } upResp, err := svc.MicUp(ctx, &roomv1.MicUpRequest{ Meta: rocketMeta(roomID, mutedUserID, "unmuted-mic-up"), @@ -188,11 +209,13 @@ func TestMutedUserCannotSpeakOrMicUpUntilUnmuted(t *testing.T) { } directCalls := rtcBlocker.snapshot() - if len(directCalls) != 6 || !directCalls[0] || directCalls[1] || directCalls[2] || !directCalls[3] || !directCalls[4] || directCalls[5] { - // 四次新 MuteUser 分别投影 true/false/true/false，旧 mute 重放必须仍投影 false； - // 中间被篡改客户端触发的 - // audio_started 必须额外 re-block=true，因此按调用顺序断言服务端防线完整。 - t.Fatalf("unexpected direct RTC audio block sequence: %+v", directCalls) + if len(directCalls) != 1 || !directCalls[0] { + // MuteUser 的 RTC 投影全部异步；只有被篡改客户端触发 audio_started 时，服务端回调会立即 re-block。 + t.Fatalf("unexpected pre-outbox RTC audio block sequence: %+v", directCalls) + } + if calls := imMuter.snapshot(); len(calls) != 0 { + // IM 禁言不能进入 MuteUser 同步路径；否则腾讯慢请求会拉长房间命令，且实现重新依赖外部服务可用性。 + t.Fatalf("room mute command must not call Tencent IM before durable outbox processing: %+v", calls) } if err := svc.ProcessPendingOutbox(ctx, roomservice.OutboxWorkerOptions{PublishTimeout: time.Second, BatchSize: 100}); err != nil { @@ -209,6 +232,17 @@ func TestMutedUserCannotSpeakOrMicUpUntilUnmuted(t *testing.T) { t.Fatalf("stale RTC mute outbox call %d rolled current state backward", index) } } + imCalls := imMuter.snapshot() + if len(imCalls) != 4 { + t.Fatalf("each durable RoomUserMuted event must reach Tencent IM projection exactly once: %+v", imCalls) + } + for index, muted := range imCalls { + // IM 投影只由 outbox 触发；处理四条历史事件时当前 Room Cell 已解除禁言， + // 因此每次都必须写 MuteTime=0，不能按旧事件载荷恢复永久禁言。 + if muted { + t.Fatalf("stale IM mute outbox call %d rolled current state backward", index) + } + } } type blockingRTCUserAudioBlocker struct { @@ -267,30 +301,30 @@ func TestRTCUserAudioReconcileConvergesSlowMuteAndNewerUnmute(t *testing.T) { createRocketRoom(t, ctx, svc, roomID, ownerID, 9101) joinRocketRoom(t, ctx, svc, roomID, targetUserID) - muteDone := make(chan error, 1) + if _, err := svc.MuteUser(ctx, &roomv1.MuteUserRequest{ + Meta: rocketMeta(roomID, ownerID, "slow-mute"), + TargetUserId: targetUserID, + Muted: true, + }); err != nil { + t.Fatalf("mute command failed: %v", err) + } + projectionDone := make(chan error, 1) go func() { - _, err := svc.MuteUser(ctx, &roomv1.MuteUserRequest{ - Meta: rocketMeta(roomID, ownerID, "slow-mute"), - TargetUserId: targetUserID, - Muted: true, - }) - muteDone <- err + projectionDone <- svc.ProcessPendingOutbox(ctx, roomservice.OutboxWorkerOptions{PublishTimeout: 3 * time.Second, BatchSize: 100}) }() select { case <-rtcBlocker.firstStarted: case <-time.After(2 * time.Second): - t.Fatal("slow mute never reached RTC blocker") + t.Fatal("slow mute outbox never reached RTC blocker") } - unmuteDone := make(chan error, 1) - go func() { - _, err := svc.MuteUser(ctx, &roomv1.MuteUserRequest{ - Meta: rocketMeta(roomID, ownerID, "fast-unmute"), - TargetUserId: targetUserID, - Muted: false, - }) - unmuteDone <- err - }() + if _, err := svc.MuteUser(ctx, &roomv1.MuteUserRequest{ + Meta: rocketMeta(roomID, ownerID, "fast-unmute"), + TargetUserId: targetUserID, + Muted: false, + }); err != nil { + t.Fatalf("newer unmute command failed: %v", err) + } deadline := time.Now().Add(2 * time.Second) for { permission, err := svc.CheckSpeakPermission(ctx, &roomv1.CheckSpeakPermissionRequest{RoomId: roomID, UserId: targetUserID, AppCode: appcode.Default}) @@ -303,11 +337,8 @@ func TestRTCUserAudioReconcileConvergesSlowMuteAndNewerUnmute(t *testing.T) { time.Sleep(5 * time.Millisecond) } close(rtcBlocker.releaseFirst) - if err := <-muteDone; err != nil { - t.Fatalf("slow mute request failed: %v", err) - } - if err := <-unmuteDone; err != nil { - t.Fatalf("newer unmute request failed: %v", err) + if err := <-projectionDone; err != nil { + t.Fatalf("slow mute outbox processing failed: %v", err) } calls, blocked := rtcBlocker.snapshot() if blocked || len(calls) < 2 || !calls[0] || calls[len(calls)-1] { @@ -315,11 +346,12 @@ func TestRTCUserAudioReconcileConvergesSlowMuteAndNewerUnmute(t *testing.T) { } } -func TestRTCUserAudioFailureDoesNotBlockRoomMuteOutboxPublisher(t *testing.T) { +func TestMuteProjectionFailuresDoNotBlockFactPublisherAndRemainRetryable(t *testing.T) { ctx := context.Background() repository := mysqltest.NewRepository(t) now := &fixedRoomRocketClock{now: time.Date(2026, 7, 21, 10, 0, 0, 0, time.UTC)} rtcBlocker := &recordingRTCUserAudioBlocker{err: errors.New("rtc unavailable")} + imMuter := &recordingIMUserMuter{err: errors.New("im unavailable")} outboxPublisher := newRecordingRoomDirectIMPublisher() svc := roomservice.New(roomservice.Config{ NodeID: "node-muted-rtc-outbox-test", @@ -327,6 +359,7 @@ func TestRTCUserAudioFailureDoesNotBlockRoomMuteOutboxPublisher(t *testing.T) { SnapshotEveryN: 1, Clock: now, RTCUserAudioBlocker: rtcBlocker, + IMUserMuter: imMuter, }, router.NewMemoryDirectory(), repository, &rocketTestWallet{}, integration.NewNoopRoomEventPublisher(), outboxPublisher) roomID := "room-muted-rtc-outbox" @@ -339,7 +372,27 @@ func TestRTCUserAudioFailureDoesNotBlockRoomMuteOutboxPublisher(t *testing.T) { TargetUserId: targetUserID, Muted: true, }); err != nil { - t.Fatalf("durable room mute must succeed despite direct RTC failure: %v", err) + t.Fatalf("durable room mute must not depend on RTC availability: %v", err) + } + if calls := rtcBlocker.snapshot(); len(calls) != 0 { + t.Fatalf("room mute request must return before RTC projection: %+v", calls) + } + if calls := imMuter.snapshot(); len(calls) != 0 { + t.Fatalf("room mute request must return before IM projection: %+v", calls) + } + pending, err := repository.ListPendingOutbox(ctx, 100) + if err != nil { + t.Fatalf("list pending room outbox: %v", err) + } + var muteEventID string + for _, record := range pending { + if record.EventType == "RoomUserMuted" { + muteEventID = record.EventID + break + } + } + if muteEventID == "" { + t.Fatal("RoomUserMuted outbox event not found") } if err := svc.ProcessPendingOutbox(ctx, roomservice.OutboxWorkerOptions{ PublishTimeout: time.Second, @@ -351,7 +404,20 @@ func TestRTCUserAudioFailureDoesNotBlockRoomMuteOutboxPublisher(t *testing.T) { t.Fatalf("process room outbox with RTC failure: %v", err) } if outboxPublisher.counts()["RoomUserMuted"] != 1 { - t.Fatalf("RTC failure must not prevent generic RoomUserMuted publication: counts=%+v", outboxPublisher.counts()) + t.Fatalf("IM/RTC failures must not prevent generic RoomUserMuted publication: counts=%+v", outboxPublisher.counts()) + } + if calls := rtcBlocker.snapshot(); len(calls) != 1 || !calls[0] { + t.Fatalf("RTC projection must attempt the latest muted state once: %+v", calls) + } + if calls := imMuter.snapshot(); len(calls) != 1 || !calls[0] { + t.Fatalf("IM projection must attempt the latest muted state once: %+v", calls) + } + record, exists := repository.OutboxRecord(muteEventID) + if !exists || record.Status != "retryable" || record.RetryCount != 1 { + t.Fatalf("failed cloud projections must keep durable event retryable: %+v exists=%t", record, exists) + } + if !strings.Contains(record.LastError, "im unavailable") || !strings.Contains(record.LastError, "rtc unavailable") { + t.Fatalf("retryable outbox must preserve both projection failures: %+v", record) } } diff --git a/services/room-service/internal/room/service/moderation.go b/services/room-service/internal/room/service/moderation.go index a56c4467..92d4f731 100644 --- a/services/room-service/internal/room/service/moderation.go +++ b/services/room-service/internal/room/service/moderation.go @@ -2,12 +2,10 @@ 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" @@ -19,7 +17,7 @@ import ( // MuteUser 修改用户禁言状态。 func (s *Service) MuteUser(ctx context.Context, req *roomv1.MuteUserRequest) (*roomv1.MuteUserResponse, error) { ctx = contextFromMeta(ctx, req.GetMeta()) - // 禁言状态由 room-service 持有：腾讯云 IM 发言前回调查 CheckSpeakPermission，RTC 麦位则同步为服务端静音态。 + // 禁言状态由 Room Cell 持有；提交后的 durable outbox 分别投影腾讯 IM 原生群成员禁言和 RTC 音频上行闸门。 cmd := command.MuteUser{ Base: baseFromMeta(req.GetMeta()), TargetUserID: req.GetTargetUserId(), @@ -120,19 +118,6 @@ func (s *Service) MuteUser(ctx context.Context, req *roomv1.MuteUserRequest) (*r return nil, err } - if s.rtcUserAudioBlocker != nil { - // 低延迟路径在 Room Cell 提交后立即收敛 RTC；必须重读当前状态， - // 不能使用 result.snapshot，因为旧 command_id 重放会返回历史结果， - // 在 newer unmute 后照旧投影就会把云端状态永久回退。 - if err := s.reconcileRTCUserAudio(ctx, cmd.RoomID(), cmd.TargetUserID, false); 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.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 8535bd9a..0f9f60c9 100644 --- a/services/room-service/internal/room/service/outbox_worker.go +++ b/services/room-service/internal/room/service/outbox_worker.go @@ -419,23 +419,36 @@ func (s *Service) publishOutboxRecord(ctx context.Context, record outbox.Record) } var publishErr error if s.outboxPublisher != nil { - // 通用事实通道必须先尝试发布；RTC 外部 API 慢或失败不能让 - // RoomUserMuted 在进入 MQ/活动消费者之前就被截断。RTC 失败时本记录仍会 + // 通用事实通道必须先尝试发布；IM/RTC 外部 API 慢或失败不能让 + // RoomUserMuted 在进入 MQ/活动消费者之前就被截断。投影失败时本记录仍会 // retry，MQ 可能重放由既有 event_id 幂等收敛。 publishErr = s.outboxPublisher.PublishOutboxEvent(ctx, record.Envelope) } + var imErr error var rtcErr error if record.Envelope != nil && record.Envelope.GetEventType() == "RoomUserMuted" { - // RTC 使用独立有界预算，不消耗通用 publisher 的 deadline；两侧都尝试后 - // 任一失败都让 outbox 保持 retryable，避免丢失媒体补偿或业务事实。 - rtcCtx, cancel := context.WithTimeout(appcode.WithContext(context.Background(), appcode.FromContext(ctx)), rtcMuteProjectionTimeout) - rtcErr = s.syncRTCUserMuteFromOutbox(rtcCtx, record.Envelope) - cancel() + // IM 与 RTC 使用彼此独立的有界预算并行投影，避免串行云请求把单条 outbox lease 拉长到 12 秒。 + // 两侧都完成后再汇总错误，任一失败都让事实保持 retryable，不会丢失文字禁言或媒体禁音。 + projectionContext := appcode.WithContext(context.Background(), appcode.FromContext(ctx)) + imResult := make(chan error, 1) + rtcResult := make(chan error, 1) + go func() { + imCtx, imCancel := context.WithTimeout(projectionContext, imMuteProjectionTimeout) + defer imCancel() + imResult <- s.syncIMUserMuteFromOutbox(imCtx, record.Envelope) + }() + go func() { + rtcCtx, rtcCancel := context.WithTimeout(projectionContext, rtcMuteProjectionTimeout) + defer rtcCancel() + rtcResult <- s.syncRTCUserMuteFromOutbox(rtcCtx, record.Envelope) + }() + imErr = <-imResult + rtcErr = <-rtcResult } - // Join 保留两个独立通道的排障语义；只要任一不为 nil，调用方就会 - // 将记录改为 retryable，下次同时重放 MQ event_id 和 latest-state RTC 投影。 - return errors.Join(publishErr, rtcErr) + // Join 保留三个独立通道的排障语义；只要任一不为 nil，调用方就会把记录改为 retryable， + // 下次重放 MQ event_id，并分别按 Room Cell 最新状态收敛 IM 与 RTC，不信任历史 muted 字段。 + return errors.Join(publishErr, imErr, rtcErr) } // outboxBackoff 根据失败次数计算指数退避，且永远不超过 MaxBackoff。 diff --git a/services/room-service/internal/room/service/service.go b/services/room-service/internal/room/service/service.go index ec7a7afc..4c0c4abb 100644 --- a/services/room-service/internal/room/service/service.go +++ b/services/room-service/internal/room/service/service.go @@ -35,6 +35,8 @@ type Config struct { RTCUserRemover integration.RTCUserRemover // RTCUserAudioBlocker 把 Room Cell 禁言事实投影到腾讯 RTC 服务端音频上行闸门。 RTCUserAudioBlocker integration.RTCUserAudioBlocker + // IMUserMuter 把 Room Cell 禁言事实异步投影到腾讯 IM 原生群成员禁言。 + IMUserMuter integration.IMUserMuter // RoomRocketLaunchScheduler 把倒计时发射唤醒交给外部延迟消息，避免只靠已加载 Cell 扫描。 RoomRocketLaunchScheduler integration.RoomRocketLaunchScheduler // RobotDisplayPublisher 专门 best-effort 投递机器人房间展示事件；失败只打日志，不进入 MySQL outbox。 @@ -121,6 +123,14 @@ type Service struct { rtcUserRemover integration.RTCUserRemover // rtcUserAudioBlocker 在房间禁言提交后关闭目标用户的 RTC 上行音频；状态仍只保存在 Room Cell。 rtcUserAudioBlocker integration.RTCUserAudioBlocker + // imUserMuter 只由 durable outbox worker 调用，普通 IM 发消息链路永远不会经过 room-service。 + imUserMuter integration.IMUserMuter + // imMuteGateMu 保护按 app/room/user 划分的 IM 投影门闩，避免同一用户的旧 mute 与新 unmute 并发写腾讯。 + imMuteGateMu sync.Mutex + imMuteGates map[imMuteProjectionKey]*imMuteProjectionGate + // IM forbid_send_msg 官方上限为 200 QPS；单节点节流给双节点部署保留共享 SDKAppID 余量。 + imMuteRateMu sync.Mutex + imMuteNextCall time.Time // rtcMuteGateMu 保护按 app/room/user 划分的 RTC 投影门闩；门闩只串行化本进程请求， // 跨实例竞态由每次云 API 返回后的 latest-state recheck 继续收敛。 rtcMuteGateMu sync.Mutex @@ -312,6 +322,8 @@ func New(cfg Config, directory router.Directory, repository Repository, wallet i luckyGiftSendLockTTL: luckyGiftSendLockTTL, rtcUserRemover: cfg.RTCUserRemover, rtcUserAudioBlocker: cfg.RTCUserAudioBlocker, + imUserMuter: cfg.IMUserMuter, + imMuteGates: make(map[imMuteProjectionKey]*imMuteProjectionGate), rtcMuteGates: make(map[rtcMuteProjectionKey]*rtcMuteProjectionGate), cells: make(map[string]*loadedCell), robotRuntimes: make(map[string]robotRoomRuntime), diff --git a/services/room-service/internal/transport/grpc/server.go b/services/room-service/internal/transport/grpc/server.go index f8a5abe0..f50acab0 100644 --- a/services/room-service/internal/transport/grpc/server.go +++ b/services/room-service/internal/transport/grpc/server.go @@ -377,7 +377,7 @@ func (s *Server) SetRoomAdmin(ctx context.Context, req *roomv1.SetRoomAdminReque // MuteUser 代理到领域服务。 func (s *Server) MuteUser(ctx context.Context, req *roomv1.MuteUserRequest) (*roomv1.MuteUserResponse, error) { - // 禁言状态是腾讯云 IM 发言回调的权威来源。 + // 禁言状态由 Room Cell 持有；腾讯 IM/RTC 仅消费 outbox 异步投影，查询接口保留给其它受控入口。 ctx = contextWithMetaApp(ctx, req.GetMeta()) if resp, forwarded, err := forwardCommand(s, ctx, req.GetMeta(), func(callCtx context.Context, client roomv1.RoomCommandServiceClient) (*roomv1.MuteUserResponse, error) { return client.MuteUser(callCtx, req)