优化模块租户检查中间件，重构为函数式实现并添加超级管理员检查；新增MongoDB更新过滤构建工具；修复Redis消息消费错误处理

beans/user.go

@@ -0,0 +1,5 @@
+package beans
+
+type IsSuperAdminRes struct {
+	IsSuperAdmin bool `p:"isSuperAdmin"`
+}

message/redis.go

@@ -122,10 +122,10 @@ func readFromStream(ctx context.Context, msg QueueMessage) error {
 		for {
 			result, err := getRedisClient().Do(ctx, "XREADGROUP", "GROUP", msg.GroupName, msg.ConsumerName, "COUNT", msg.BatchSize, "BLOCK", 0, "STREAMS", msg.StreamKey, ">")
 			if err != nil {
-				select {
+				//select {
-				case <-ctx.Done():
+				//case <-ctx.Done():
-					return
+				//	return
-				}
+				//}
 				time.Sleep(time.Second)
 				goto RECONNECT
 			}

middleware/module_tenant_check.go

@@ -9,7 +9,6 @@ import (
 	"gitee.com/red-future---jilin-g/common/message"
 	"gitee.com/red-future---jilin-g/common/utils"
 	"github.com/gogf/gf/v2/database/gredis"
-	"github.com/gogf/gf/v2/errors/gerror"
 	"github.com/gogf/gf/v2/frame/g"
 	"github.com/gogf/gf/v2/net/ghttp"
 	"github.com/gogf/gf/v2/os/gtime"
@@ -17,18 +16,27 @@ import (
 	"time"
 )
 
-type moduleTenant struct{}
+func ModuleTenantCheck(r *ghttp.Request) {
-
+	//将 http.Header 转换为 map[string]string
-var ModuleTenant = new(moduleTenant)
+	headers := make(map[string]string)
-
+	for k, v := range r.Request.Header {
-// ModuleTenantInfo 别名,引用admin-go的entity.ModuleTenant
+		if len(v) > 0 {
-type ModuleTenantInfo = beans.ModuleTenant
+			headers[k] = v[0]
-
+		}
-func (s *moduleTenant) ModuleTenantCheck(r *ghttp.Request) {
+	}
+	// 检查是否是超级管理员
+	IsSuperAdmin, err := IsSuperAdmin(r.Context(), headers)
+	if err != nil {
+		SetResponseInfo(r.Context(), r, err)
+	}
+	// 如果是超级管理员，则不进行模块租户检查
+	if IsSuperAdmin.IsSuperAdmin {
+		r.Middleware.Next()
+		return
+	}
 	getUserInfo, err := utils.GetUserInfo(r.Context())
 	if err != nil {
-		r.Response.WriteJson(err)
+		SetResponseInfo(r.Context(), r, err)
-		r.Exit()
 	}
 	exit := gconv.Int64(time.Minute * 1)
 	getEX, err := message.GetRedisClientTest("test").GetEX(r.Context(), fmt.Sprintf("module_tenant:tenantId-%v", getUserInfo.TenantId), gredis.GetEXOption{
@@ -37,16 +45,14 @@ func (s *moduleTenant) ModuleTenantCheck(r *ghttp.Request) {
 		},
 	})
 	if err != nil {
-		r.Response.WriteJson(err)
+		SetResponseInfo(r.Context(), r, err)
-		r.Exit()
 	}
 	// 获取模块key
 	moduleKey := g.Cfg().MustGet(context.Background(), "server.name")
 	if !g.IsEmpty(getEX.String()) {
-		list := make([]ModuleTenantInfo, 0)
+		list := make([]beans.ModuleTenant, 0)
 		if err = json.Unmarshal([]byte(getEX.String()), &list); err != nil {
-			r.Response.WriteJson(err)
+			SetResponseInfo(r.Context(), r, err)
-			r.Exit()
 		}
 		var expireAt *gtime.Time
 		for _, value := range list {
@@ -60,45 +66,45 @@ func (s *moduleTenant) ModuleTenantCheck(r *ghttp.Request) {
 			gt1 := gtime.New(time.Now())
 			gt2 := gtime.New(expireAt)
 			if !gt1.Before(gt2) {
-				r.Response.WriteJson(gerror.New("您访问的模块已过期，请续期后再使用"))
+				SetResponseInfo(r.Context(), r, "您访问的模块已过期，请续期后再使用")
-				r.Exit()
 			}
 		} else {
-			r.Response.WriteJson(gerror.New("您未开通此模块，请开通后再使用"))
+			SetResponseInfo(r.Context(), r, "您未开通此模块，请开通后再使用")
-			r.Exit()
 		}
 	} else {
-		//将 http.Header 转换为 map[string]string
-		headers := make(map[string]string)
-		for k, v := range r.Request.Header {
-			if len(v) > 0 {
-				headers[k] = v[0]
-			}
-		}
 		// 缓存为空，调用admin-go的Check接口检查模块开通状态
-		res, err := s.Check(r.Context(), headers, beans.ModuleTenantCheckReq{
+		res, err := Check(r.Context(), headers, beans.ModuleTenantCheckReq{
 			ModuleKey: moduleKey.String(),
 			TenantId:  gconv.Uint64(getUserInfo.TenantId),
 		})
 		if err != nil {
-			r.Response.WriteJson(err)
+			SetResponseInfo(r.Context(), r, err)
-			r.Exit()
 		}
 		// 根据检查结果判断是否允许访问
 		if res.Status == "not_activated" {
-			r.Response.WriteJson(gerror.New("您未开通此模块，请开通后再使用"))
+			SetResponseInfo(r.Context(), r, "您未开通此模块，请开通后再使用")
-			r.Exit()
 		} else if res.Status == "expired" {
-			r.Response.WriteJson(gerror.New("您访问的模块已过期，请续期后再使用"))
+			SetResponseInfo(r.Context(), r, "您访问的模块已过期，请续期后再使用")
-			r.Exit()
 		}
 	}
 	r.Middleware.Next() // 继续执行后续中间件和路由处理
 }
 
+// SetResponseInfo 设置响应信息
+func SetResponseInfo(ctx context.Context, r *ghttp.Request, message any) {
+	_ = ctx
+	r.Response.Status = 503
+	r.Response.WriteJsonExit(map[string]interface{}{
+		"success": false,
+		"code":    503,
+		"message": fmt.Sprintf("服务不可用：%s", message),
+	})
+	r.Exit()
+}
+
 // Check 调用admin-go服务检查模块开通状态
-func (s *moduleTenant) Check(ctx context.Context, headers map[string]string, req beans.ModuleTenantCheckReq) (res *beans.ModuleTenantCheckRes, err error) {
+func Check(ctx context.Context, headerMap map[string]string, req beans.ModuleTenantCheckReq) (res *beans.ModuleTenantCheckRes, err error) {
-	if err = http.Get(ctx, "admin-go/api/v1/system/moduleTenant/check", headers, &res,
+	if err = http.Get(ctx, "admin-go/api/v1/system/moduleTenant/check", headerMap, &res,
 		"moduleKey", req.ModuleKey,
 		"tenantId", req.TenantId,
 	); err != nil {
@@ -106,3 +112,11 @@ func (s *moduleTenant) Check(ctx context.Context, headers map[string]string, req
 	}
 	return
 }
+
+// IsSuperAdmin 调用admin-go服务检查是否是超级管理员
+func IsSuperAdmin(ctx context.Context, headerMap map[string]string) (res *beans.IsSuperAdminRes, err error) {
+	if err = http.Get(ctx, "admin-go/api/v1/system/user/checkIsSuperAdmin", headerMap, &res); err != nil {
+		return
+	}
+	return
+}

mongo/mongo.go

@@ -544,6 +544,18 @@ func (m *MongoDB) Count(ctx context.Context, filter bson.M, collection string) (
 	return
 }
 
+func BuildUpdateFilter(ctx context.Context, req interface{}) (filter bson.M, err error) {
+	_ = ctx
+	filter = bson.M{}
+	reqMap := gconv.Map(req)
+	for mk, mv := range reqMap {
+		if mk != "id" && !g.IsEmpty(mv) {
+			filter[mk] = mv
+		}
+	}
+	return
+}
+
 // EntityToBson 将 *entity/entity 转换为 bson.M
 func EntityToBson(entity interface{}) (bson.M, error) {
 	return EntityToBsonWithFilter(entity, false)