feat: GroupApplicationAgreeMemberEnterNotification splitting (#3297)

* 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: GroupApplicationAgreeMemberEnterNotification splitting, rpc body size limit

* feat: GroupApplicationAgreeMemberEnterNotification splitting, rpc body size limit

config/share.yml

@@ -7,3 +7,7 @@ multiLogin:
   policy: 1
   # max num of tokens in one end
   maxNumOneEnd: 30
+
+rpcMaxBodySize:
+  requestMaxBodySize:  8388608
+  responseMaxBodySize: 8388608

internal/rpc/group/group.go

@@ -448,13 +448,26 @@ func (g *groupServer) InviteUserToGroup(ctx context.Context, req *pbgroup.Invite
 		return nil, err
 	}
 
-	if err := g.db.CreateGroup(ctx, nil, groupMembers); err != nil {
+	const singleQuantity = 50
+	for start := 0; start < len(groupMembers); start += singleQuantity {
+		end := start + singleQuantity
+		if end > len(groupMembers) {
+			end = len(groupMembers)
+		}
+		currentMembers := groupMembers[start:end]
+
+		if err := g.db.CreateGroup(ctx, nil, currentMembers); err != nil {
 			return nil, err
 		}
 
-	if err = g.notification.GroupApplicationAgreeMemberEnterNotification(ctx, req.GroupID, req.SendMessage, opUserID, req.InvitedUserIDs...); err != nil {
+		userIDs := datautil.Slice(currentMembers, func(e *model.GroupMember) string {
+			return e.UserID
+		})
+
+		if err = g.notification.GroupApplicationAgreeMemberEnterNotification(ctx, req.GroupID, req.SendMessage, opUserID, userIDs...); err != nil {
 			return nil, err
 		}
+	}
 	return &pbgroup.InviteUserToGroupResp{}, nil
 }
 

internal/rpc/group/notification.go

@@ -522,6 +522,10 @@ func (g *NotificationSender) MemberKickedNotification(ctx context.Context, tips
 }
 
 func (g *NotificationSender) GroupApplicationAgreeMemberEnterNotification(ctx context.Context, groupID string, SendMessage *bool, invitedOpUserID string, entrantUserID ...string) error {
+	return g.groupApplicationAgreeMemberEnterNotification(ctx, groupID, SendMessage, invitedOpUserID, entrantUserID...)
+}
+
+func (g *NotificationSender) groupApplicationAgreeMemberEnterNotification(ctx context.Context, groupID string, SendMessage *bool, invitedOpUserID string, entrantUserID ...string) error {
 	var err error
 	defer func() {
 		if err != nil {

pkg/common/config/config.go

@@ -351,6 +351,12 @@ type Share struct {
 	Secret         string         `yaml:"secret"`
 	IMAdminUserID  []string       `yaml:"imAdminUserID"`
 	MultiLogin     MultiLogin     `yaml:"multiLogin"`
+	RPCMaxBodySize MaxRequestBody `yaml:"rpcMaxBodySize"`
+}
+
+type MaxRequestBody struct {
+	RequestMaxBodySize  int `yaml:"requestMaxBodySize"`
+	ResponseMaxBodySize int `yaml:"responseMaxBodySize"`
 }
 
 type MultiLogin struct {

pkg/common/startrpc/start.go

@@ -21,6 +21,7 @@ import (
 	"net"
 	"os"
 	"os/signal"
+	"reflect"
 	"strconv"
 	"syscall"
 	"time"
@@ -45,6 +46,36 @@ 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 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,
@@ -55,7 +86,24 @@ func Start[T any](ctx context.Context, disc *conf.Discovery, prometheusConfig *c
 		conf.InitNotification(notification)
 	}
 
-	options = append(options, mw.GrpcServer())
+	maxRequestBody := getConfigRpcMaxRequestBody(reflect.ValueOf(config))
+
+	log.ZDebug(ctx, "rpc start", "rpcMaxRequestBody", maxRequestBody, "rpcRegisterName", rpcRegisterName, "registerIP", registerIP, "listenIP", listenIP)
+
+	options = append(options,
+		mw.GrpcServer(),
+	)
+	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 {
@@ -84,6 +132,9 @@ func Start[T any](ctx context.Context, disc *conf.Discovery, prometheusConfig *c
 		mw.GrpcClient(), grpc.WithTransportCredentials(insecure.NewCredentials()),
 		grpc.WithDefaultServiceConfig(fmt.Sprintf(`{"LoadBalancingPolicy": "%s"}`, "round_robin")),
 	)
+	if len(clientOptions) > 0 {
+		client.AddOption(clientOptions...)
+	}
 
 	ctx, cancel := context.WithCancelCause(ctx)