fix: optimize grpc option and fix some interface permission checks (#3327)

* pb * fix: Modifying other fields while setting IsPrivateChat does not take effect * fix: quote message error revoke * refactoring scheduled tasks * refactoring scheduled tasks * refactoring scheduled tasks * refactoring scheduled tasks * refactoring scheduled tasks * refactoring scheduled tasks * upgrading pkg tools * fix * fix * optimize log output * feat: support GetLastMessage * feat: support GetLastMessage * feat: s3 switch * feat: s3 switch * fix: GetUsersOnline * feat: SendBusinessNotification supported configuration parameters * feat: SendBusinessNotification supported configuration parameters * feat: SendBusinessNotification supported configuration parameters * feat: seq conversion failed without exiting * fix: DeleteDoc crash * fix: fill send time * fix: fill send time * fix: crash caused by withdrawing messages from users who have left the group * fix: user msg timestamp * seq read config * seq read config * fix: the source message of the reference is withdrawn, and the referenced message is deleted * feat: optimize the default notification.yml * fix: shouldPushOffline * fix: the sorting is wrong after canceling the administrator in group settings * feat: Sending messages supports returning fields modified by webhook * feat: Sending messages supports returning fields modified by webhook * feat: Sending messages supports returning fields modified by webhook * fix: oss specifies content-type when uploading * fix: the version number contains a line break * fix: the version number contains a line break * feat: GetConversationsHasReadAndMaxSeq support pinned * feat: GetConversationsHasReadAndMaxSeq support pinned * feat: GetConversationsHasReadAndMaxSeq support pinned * fix: transferring the group owner to a muted member, incremental version error * feat: unified conversion code * feat: update gomake * feat: grpc mw * fix: permission verification * fix: optimizing the code * fix: optimize grpc option and fix some interface permission checks
2026-05-10 20:15:59 +08:00 · 2025-05-14 16:07:05 +08:00
parent c29e2a9a28
commit d071a6093c
26 changed files with 383 additions and 289 deletions
@@ -0,0 +1,15 @@
+package startrpc
+
+import (
+	"context"
+
+	"github.com/openimsdk/open-im-server/v3/pkg/authverify"
+	"google.golang.org/grpc"
+)
+
+func grpcServerIMAdminUserID(imAdminUserID []string) grpc.ServerOption {
+	return grpc.ChainUnaryInterceptor(func(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp any, err error) {
+		ctx = authverify.WithIMAdminUserIDs(ctx, imAdminUserID)
+		return handler(ctx, req)
+	})
+}
@@ -40,14 +40,75 @@ import (
 	"github.com/openimsdk/tools/discovery"
 	"github.com/openimsdk/tools/errs"
 	"github.com/openimsdk/tools/log"
-	"github.com/openimsdk/tools/mw"
-	"github.com/openimsdk/tools/utils/network"
+	grpccli "github.com/openimsdk/tools/mw/grpc/client"
+	grpcsrv "github.com/openimsdk/tools/mw/grpc/server"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
 )

-// Start rpc server.
-func Start[T any](ctx context.Context, discovery *conf.Discovery, prometheusConfig *conf.Prometheus, listenIP,
+func init() {
+	prommetrics.RegistryAll()
+}
+
+func getConfigRpcMaxRequestBody(value reflect.Value) *conf.MaxRequestBody {
+	for value.Kind() == reflect.Pointer {
+		value = value.Elem()
+	}
+	if value.Kind() == reflect.Struct {
+		num := value.NumField()
+		for i := 0; i < num; i++ {
+			field := value.Field(i)
+			if !field.CanInterface() {
+				continue
+			}
+			for field.Kind() == reflect.Pointer {
+				field = field.Elem()
+			}
+			switch elem := field.Interface().(type) {
+			case conf.Share:
+				return &elem.RPCMaxBodySize
+			case conf.MaxRequestBody:
+				return &elem
+			}
+			if field.Kind() == reflect.Struct {
+				if elem := getConfigRpcMaxRequestBody(field); elem != nil {
+					return elem
+				}
+			}
+		}
+	}
+	return nil
+}
+
+func getConfigShare(value reflect.Value) *conf.Share {
+	for value.Kind() == reflect.Pointer {
+		value = value.Elem()
+	}
+	if value.Kind() == reflect.Struct {
+		num := value.NumField()
+		for i := 0; i < num; i++ {
+			field := value.Field(i)
+			if !field.CanInterface() {
+				continue
+			}
+			for field.Kind() == reflect.Pointer {
+				field = field.Elem()
+			}
+			switch elem := field.Interface().(type) {
+			case conf.Share:
+				return &elem
+			}
+			if field.Kind() == reflect.Struct {
+				if elem := getConfigShare(field); elem != nil {
+					return elem
+				}
+			}
+		}
+	}
+	return nil
+}
+
+func Start[T any](ctx context.Context, disc *conf.Discovery, prometheusConfig *conf.Prometheus, listenIP,
 	registerIP string, autoSetPorts bool, rpcPorts []int, index int, rpcRegisterName string, notification *conf.Notification, config T,
 	watchConfigNames []string, watchServiceNames []string,
 	rpcFn func(ctx context.Context, config T, client discovery.SvcDiscoveryRegistry, server *grpc.Server) error,
@@ -65,6 +126,33 @@ func Start[T any](ctx context.Context, discovery *conf.Discovery, prometheusConf
 		conf.InitNotification(notification)
 	}

+	maxRequestBody := getConfigRpcMaxRequestBody(reflect.ValueOf(config))
+	shareConfig := getConfigShare(reflect.ValueOf(config))
+
+	log.ZDebug(ctx, "rpc start", "rpcMaxRequestBody", maxRequestBody, "rpcRegisterName", rpcRegisterName, "registerIP", registerIP, "listenIP", listenIP)
+
+	options = append(options,
+		grpcsrv.GrpcServerMetadataContext(),
+		grpcsrv.GrpcServerLogger(),
+		grpcsrv.GrpcServerErrorConvert(),
+		grpcsrv.GrpcServerRequestValidate(),
+		grpcsrv.GrpcServerPanicCapture(),
+	)
+	if shareConfig != nil && len(shareConfig.IMAdminUserID) > 0 {
+		options = append(options, grpcServerIMAdminUserID(shareConfig.IMAdminUserID))
+	}
+	var clientOptions []grpc.DialOption
+	if maxRequestBody != nil {
+		if maxRequestBody.RequestMaxBodySize > 0 {
+			options = append(options, grpc.MaxRecvMsgSize(maxRequestBody.RequestMaxBodySize))
+			clientOptions = append(clientOptions, grpc.WithDefaultCallOptions(grpc.MaxCallSendMsgSize(maxRequestBody.RequestMaxBodySize)))
+		}
+		if maxRequestBody.ResponseMaxBodySize > 0 {
+			options = append(options, grpc.MaxSendMsgSize(maxRequestBody.ResponseMaxBodySize))
+			clientOptions = append(clientOptions, grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(maxRequestBody.ResponseMaxBodySize)))
+		}
+	}
+
 	registerIP, err := network.GetRpcRegisterIP(registerIP)
 	if err != nil {
 		return err
@@ -101,15 +189,33 @@ func Start[T any](ctx context.Context, discovery *conf.Discovery, prometheusConf
 	}

 	defer client.Close()
-	client.AddOption(mw.GrpcClient(), grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithDefaultServiceConfig(fmt.Sprintf(`{"LoadBalancingPolicy": "%s"}`, "round_robin")))
+	client.AddOption(
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+		grpc.WithDefaultServiceConfig(fmt.Sprintf(`{"LoadBalancingPolicy": "%s"}`, "round_robin")),

-	// var reg *prometheus.Registry
-	// var metric *grpcprometheus.ServerMetrics
-	if prometheusConfig.Enable {
-		// cusMetrics := prommetrics.GetGrpcCusMetrics(rpcRegisterName, share)
-		// reg, metric, _ = prommetrics.NewGrpcPromObj(cusMetrics)
-		// options = append(options, mw.GrpcServer(), grpc.StreamInterceptor(metric.StreamServerInterceptor()),
-		//	grpc.UnaryInterceptor(metric.UnaryServerInterceptor()))
+		grpccli.GrpcClientLogger(),
+		grpccli.GrpcClientContext(),
+		grpccli.GrpcClientErrorConvert(),
+	)
+	if len(clientOptions) > 0 {
+		client.AddOption(clientOptions...)
+	}
+
+	ctx, cancel := context.WithCancelCause(ctx)
+
+	go func() {
+		sigs := make(chan os.Signal, 1)
+		signal.Notify(sigs, syscall.SIGTERM, syscall.SIGINT, syscall.SIGKILL)
+		select {
+		case <-ctx.Done():
+			return
+		case val := <-sigs:
+			log.ZDebug(ctx, "recv signal", "signal", val.String())
+			cancel(fmt.Errorf("signal %s", val.String()))
+		}
+	}()
+
+	if prometheusListenAddr != "" {
 		options = append(
 			options, mw.GrpcServer(),
 			prommetricsUnaryInterceptor(rpcRegisterName),