hy/hyapp-server
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

相关im改动

Browse Source
This commit is contained in:
zhx 2026-07-21 18:57:09 +08:00
parent 5d74fb1ff7
commit 00db43b01b
16 changed files with 641 additions and 120 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
pkg/tencentim/rest_client.go
View File

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

38
pkg/tencentim/rest_client_test.go
View File

@ -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

2
services/gateway-service/internal/client/room_client.go
View File

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

138
services/gateway-service/internal/transport/http/callbackapi/tencent_im_callback.go
View File

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

89
services/gateway-service/internal/transport/http/response_test.go
View File

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

3
services/room-service/internal/app/app.go
View File

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

6
services/room-service/internal/integration/clients.go
View File

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

24
services/room-service/internal/integration/tencent_im.go
View File

@ -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

65
services/room-service/internal/integration/tencent_im_test.go
View File

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

2
services/room-service/internal/room/service/guard.go
View File

@ -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())

162
services/room-service/internal/room/service/im_moderation.go Normal file
View File

@ -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
}
}

134
services/room-service/internal/room/service/mic_test.go
View File

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

17
services/room-service/internal/room/service/moderation.go
View File

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

33
services/room-service/internal/room/service/outbox_worker.go
View File

@ -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。

12
services/room-service/internal/room/service/service.go
View File

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

2
services/room-service/internal/transport/grpc/server.go
View File

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