feat: implement true batchGetIncrGroupMember RPC method and dependency methods. (#2417)

* update wip contents.

* update protocol pkg.

* feat: add BatchOption struct and method.

* fix: remove unnecessary field.

* feat: implement true BatchGetIncrGroupMember RPC method and corresponding dependency methods.

* fix: update mongo version collection have unique index.

* optimize method structures.

* update resp in add sortVersion field.

* fix uncorrect condition.

* add errs pkg.

internal/rpc/conversation/sync.go

@@ -2,6 +2,7 @@ package conversation
 
 import (
 	"context"
+
 	"github.com/openimsdk/open-im-server/v3/internal/rpc/incrversion"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
 	"github.com/openimsdk/open-im-server/v3/pkg/util/hashutil"
@@ -40,7 +41,6 @@ func (c *conversationServer) GetIncrementalConversation(ctx context.Context, req
 		Find: func(ctx context.Context, conversationIDs []string) ([]*conversation.Conversation, error) {
 			return c.getConversations(ctx, req.UserID, conversationIDs)
 		},
-		ID: func(elem *conversation.Conversation) string { return elem.GroupID },
 		Resp: func(version *model.VersionLog, delIDs []string, insertList, updateList []*conversation.Conversation, full bool) *conversation.GetIncrementalConversationResp {
 			return &conversation.GetIncrementalConversationResp{
 				VersionID: version.ID.Hex(),

internal/rpc/friend/sync.go

@@ -78,7 +78,6 @@ func (s *friendServer) GetIncrementalFriends(ctx context.Context, req *relation.
 		Find: func(ctx context.Context, ids []string) ([]*sdkws.FriendInfo, error) {
 			return s.getFriend(ctx, req.UserID, ids)
 		},
-		ID: func(elem *sdkws.FriendInfo) string { return elem.FriendUser.UserID },
 		Resp: func(version *model.VersionLog, deleteIds []string, insertList, updateList []*sdkws.FriendInfo, full bool) *relation.GetIncrementalFriendsResp {
 			return &relation.GetIncrementalFriendsResp{
 				VersionID:   version.ID.Hex(),

internal/rpc/group/sync.go

@@ -2,6 +2,7 @@ package group
 
 import (
 	"context"
+
 	"github.com/openimsdk/open-im-server/v3/internal/rpc/incrversion"
 	"github.com/openimsdk/open-im-server/v3/pkg/authverify"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/servererrs"
@@ -10,13 +11,10 @@ import (
 	"github.com/openimsdk/protocol/constant"
 	pbgroup "github.com/openimsdk/protocol/group"
 	"github.com/openimsdk/protocol/sdkws"
+	"github.com/openimsdk/tools/errs"
+	"github.com/openimsdk/tools/log"
 )
 
-func (s *groupServer) BatchGetIncrementalGroupMember(ctx context.Context, req *pbgroup.BatchGetIncrementalGroupMemberReq) (*pbgroup.BatchGetIncrementalGroupMemberResp, error) {
-	//TODO implement me
-	panic("implement me")
-}
-
 func (s *groupServer) GetFullGroupMemberUserIDs(ctx context.Context, req *pbgroup.GetFullGroupMemberUserIDsReq) (*pbgroup.GetFullGroupMemberUserIDsResp, error) {
 	vl, err := s.db.FindMaxGroupMemberVersionCache(ctx, req.GroupID)
 	if err != nil {
@@ -104,7 +102,6 @@ func (s *groupServer) GetIncrementalGroupMember(ctx context.Context, req *pbgrou
 		Find: func(ctx context.Context, ids []string) ([]*sdkws.GroupMemberFullInfo, error) {
 			return s.getGroupMembersInfo(ctx, req.GroupID, ids)
 		},
-		ID: func(elem *sdkws.GroupMemberFullInfo) string { return elem.UserID },
 		Resp: func(version *model.VersionLog, delIDs []string, insertList, updateList []*sdkws.GroupMemberFullInfo, full bool) *pbgroup.GetIncrementalGroupMemberResp {
 			return &pbgroup.GetIncrementalGroupMemberResp{
 				VersionID:   version.ID.Hex(),
@@ -135,6 +132,150 @@ func (s *groupServer) GetIncrementalGroupMember(ctx context.Context, req *pbgrou
 	return resp, nil
 }
 
+func (s *groupServer) BatchGetIncrementalGroupMember(ctx context.Context, req *pbgroup.BatchGetIncrementalGroupMemberReq) (resp *pbgroup.BatchGetIncrementalGroupMemberResp, err error) {
+	type VersionInfo struct {
+		GroupID       string
+		VersionID     string
+		VersionNumber uint64
+	}
+
+	var groupIDs []string
+
+	groupsVersionMap := make(map[string]*VersionInfo)
+	groupsMap := make(map[string]*model.Group)
+	hasGroupUpdateMap := make(map[string]bool)
+	sortVersionMap := make(map[string]uint64)
+
+	var targetKeys, versionIDs []string
+	var versionNumbers []uint64
+
+	var requestBodyLen int
+
+	for _, group := range req.ReqList {
+		groupsVersionMap[group.GroupID] = &VersionInfo{
+			GroupID:       group.GroupID,
+			VersionID:     group.VersionID,
+			VersionNumber: group.Version,
+		}
+
+		groupIDs = append(groupIDs, group.GroupID)
+	}
+
+	groups, err := s.db.FindGroup(ctx, groupIDs)
+	if err != nil {
+		return nil, errs.Wrap(err)
+	}
+
+	for _, group := range groups {
+		if group.Status == constant.GroupStatusDismissed {
+			err = servererrs.ErrDismissedAlready.Wrap()
+			log.ZError(ctx, "This group is Dismissed Already", err, "group is", group.GroupID)
+
+			delete(groupsVersionMap, group.GroupID)
+		} else {
+			groupsMap[group.GroupID] = group
+		}
+	}
+
+	for groupID, vInfo := range groupsVersionMap {
+		targetKeys = append(targetKeys, groupID)
+		versionIDs = append(versionIDs, vInfo.VersionID)
+		versionNumbers = append(versionNumbers, vInfo.VersionNumber)
+	}
+
+	opt := incrversion.BatchOption[[]*sdkws.GroupMemberFullInfo, pbgroup.BatchGetIncrementalGroupMemberResp]{
+		Ctx:            ctx,
+		TargetKeys:     targetKeys,
+		VersionIDs:     versionIDs,
+		VersionNumbers: versionNumbers,
+		Versions: func(ctx context.Context, groupIDs []string, versions []uint64, limits []int) (map[string]*model.VersionLog, error) {
+			vLogs, err := s.db.BatchFindMemberIncrVersion(ctx, groupIDs, versions, limits)
+			if err != nil {
+				return nil, errs.Wrap(err)
+			}
+
+			for groupID, vlog := range vLogs {
+				vlogElems := make([]model.VersionLogElem, 0, len(vlog.Logs))
+				for i, log := range vlog.Logs {
+					switch log.EID {
+					case model.VersionGroupChangeID:
+						vlog.LogLen--
+						hasGroupUpdateMap[groupID] = true
+					case model.VersionSortChangeID:
+						vlog.LogLen--
+						sortVersionMap[groupID] = uint64(log.Version)
+					default:
+						vlogElems = append(vlogElems, vlog.Logs[i])
+					}
+				}
+				vlog.Logs = vlogElems
+				if vlog.LogLen > 0 {
+					hasGroupUpdateMap[groupID] = true
+				}
+			}
+
+			return vLogs, nil
+		},
+		CacheMaxVersions: s.db.BatchFindMaxGroupMemberVersionCache,
+		Find: func(ctx context.Context, groupID string, ids []string) ([]*sdkws.GroupMemberFullInfo, error) {
+			memberInfo, err := s.getGroupMembersInfo(ctx, groupID, ids)
+			if err != nil {
+				return nil, err
+			}
+
+			return memberInfo, err
+		},
+		Resp: func(versions map[string]*model.VersionLog, deleteIdsMap map[string][]string, insertListMap, updateListMap map[string][]*sdkws.GroupMemberFullInfo, fullMap map[string]bool) *pbgroup.BatchGetIncrementalGroupMemberResp {
+			resList := make(map[string]*pbgroup.GetIncrementalGroupMemberResp)
+
+			for groupID, versionLog := range versions {
+				resList[groupID] = &pbgroup.GetIncrementalGroupMemberResp{
+					VersionID:   versionLog.ID.Hex(),
+					Version:     uint64(versionLog.Version),
+					Full:        fullMap[groupID],
+					Delete:      deleteIdsMap[groupID],
+					Insert:      insertListMap[groupID],
+					Update:      updateListMap[groupID],
+					SortVersion: sortVersionMap[groupID],
+				}
+
+				requestBodyLen += len(insertListMap[groupID]) + len(updateListMap[groupID]) + len(deleteIdsMap[groupID])
+				if requestBodyLen > 200 {
+					break
+				}
+			}
+
+			return &pbgroup.BatchGetIncrementalGroupMemberResp{
+				RespList: resList,
+			}
+		},
+	}
+
+	resp, err = opt.Build()
+	if err != nil {
+		return nil, errs.Wrap(err)
+	}
+
+	for groupID, val := range resp.RespList {
+		if val.Full || hasGroupUpdateMap[groupID] {
+			count, err := s.db.FindGroupMemberNum(ctx, groupID)
+			if err != nil {
+				return nil, err
+			}
+
+			owner, err := s.db.TakeGroupOwner(ctx, groupID)
+			if err != nil {
+				return nil, err
+			}
+
+			resp.RespList[groupID].Group = s.groupDB2PB(groupsMap[groupID], owner.UserID, count)
+		}
+	}
+
+	return resp, nil
+
+}
+
 func (s *groupServer) GetIncrementalJoinGroup(ctx context.Context, req *pbgroup.GetIncrementalJoinGroupReq) (*pbgroup.GetIncrementalJoinGroupResp, error) {
 	if err := authverify.CheckAccessV3(ctx, req.UserID, s.config.Share.IMAdminUserID); err != nil {
 		return nil, err
@@ -147,7 +288,6 @@ func (s *groupServer) GetIncrementalJoinGroup(ctx context.Context, req *pbgroup.
 		Version:         s.db.FindJoinIncrVersion,
 		CacheMaxVersion: s.db.FindMaxJoinGroupVersionCache,
 		Find:            s.getGroupsInfo,
-		ID:              func(elem *sdkws.GroupInfo) string { return elem.GroupID },
 		Resp: func(version *model.VersionLog, delIDs []string, insertList, updateList []*sdkws.GroupInfo, full bool) *pbgroup.GetIncrementalJoinGroupResp {
 			return &pbgroup.GetIncrementalJoinGroupResp{
 				VersionID: version.ID.Hex(),

internal/rpc/incrversion/batch_option.go

@@ -0,0 +1,207 @@
+package incrversion
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
+	"github.com/openimsdk/tools/errs"
+	"go.mongodb.org/mongo-driver/bson/primitive"
+)
+
+type BatchOption[A, B any] struct {
+	Ctx            context.Context
+	TargetKeys     []string
+	VersionIDs     []string
+	VersionNumbers []uint64
+	//SyncLimit       int
+	Versions         func(ctx context.Context, dIds []string, versions []uint64, limits []int) (map[string]*model.VersionLog, error)
+	CacheMaxVersions func(ctx context.Context, dIds []string) (map[string]*model.VersionLog, error)
+	Find             func(ctx context.Context, dId string, ids []string) (A, error)
+	Resp             func(versionsMap map[string]*model.VersionLog, deleteIdsMap map[string][]string, insertListMap, updateListMap map[string]A, fullMap map[string]bool) *B
+}
+
+func (o *BatchOption[A, B]) newError(msg string) error {
+	return errs.ErrInternalServer.WrapMsg(msg)
+}
+
+func (o *BatchOption[A, B]) check() error {
+	if o.Ctx == nil {
+		return o.newError("opt ctx is nil")
+	}
+	if len(o.TargetKeys) == 0 {
+		return o.newError("targetKeys is empty")
+	}
+	if o.Versions == nil {
+		return o.newError("func versions is nil")
+	}
+	if o.Find == nil {
+		return o.newError("func find is nil")
+	}
+	if o.Resp == nil {
+		return o.newError("func resp is nil")
+	}
+	return nil
+}
+
+func (o *BatchOption[A, B]) validVersions() []bool {
+	valids := make([]bool, len(o.VersionIDs))
+	for i, versionID := range o.VersionIDs {
+		objID, err := primitive.ObjectIDFromHex(versionID)
+		valids[i] = (err == nil && (!objID.IsZero()) && o.VersionNumbers[i] > 0)
+	}
+	return valids
+}
+
+func (o *BatchOption[A, B]) equalIDs(objIDs []primitive.ObjectID) []bool {
+	equals := make([]bool, len(o.VersionIDs))
+	for i, versionID := range o.VersionIDs {
+		equals[i] = versionID == objIDs[i].Hex()
+	}
+	return equals
+}
+
+func (o *BatchOption[A, B]) getVersions(tags *[]int) (versions map[string]*model.VersionLog, err error) {
+	var dIDs []string
+	var versionNums []uint64
+	var limits []int
+
+	valids := o.validVersions()
+
+	if o.CacheMaxVersions == nil {
+		for i, valid := range valids {
+			if valid {
+				(*tags)[i] = tagQuery
+				dIDs = append(dIDs, o.TargetKeys[i])
+				versionNums = append(versionNums, o.VersionNumbers[i])
+				limits = append(limits, syncLimit)
+			} else {
+				(*tags)[i] = tagFull
+				dIDs = append(dIDs, o.TargetKeys[i])
+				versionNums = append(versionNums, 0)
+				limits = append(limits, 0)
+			}
+		}
+
+		versions, err = o.Versions(o.Ctx, dIDs, versionNums, limits)
+		if err != nil {
+			return nil, errs.Wrap(err)
+		}
+		return versions, nil
+
+	} else {
+		caches, err := o.CacheMaxVersions(o.Ctx, o.TargetKeys)
+		if err != nil {
+			return nil, errs.Wrap(err)
+		}
+
+		objIDs := make([]primitive.ObjectID, len(o.VersionIDs))
+
+		for i, versionID := range o.VersionIDs {
+			objID, _ := primitive.ObjectIDFromHex(versionID)
+			objIDs[i] = objID
+		}
+
+		equals := o.equalIDs(objIDs)
+		for i, valid := range valids {
+			if !valid {
+				(*tags)[i] = tagFull
+			} else if !equals[i] {
+				(*tags)[i] = tagFull
+			} else if o.VersionNumbers[i] == uint64(caches[o.TargetKeys[i]].Version) {
+				(*tags)[i] = tagEqual
+			} else {
+				(*tags)[i] = tagQuery
+				dIDs = append(dIDs, o.TargetKeys[i])
+				versionNums = append(versionNums, o.VersionNumbers[i])
+				limits = append(limits, syncLimit)
+
+				delete(caches, o.TargetKeys[i])
+			}
+		}
+
+		if dIDs != nil {
+			versionMap, err := o.Versions(o.Ctx, dIDs, versionNums, limits)
+			if err != nil {
+				return nil, errs.Wrap(err)
+			}
+
+			for k, v := range versionMap {
+				caches[k] = v
+			}
+		}
+
+		versions = caches
+	}
+	return versions, nil
+}
+
+func (o *BatchOption[A, B]) Build() (*B, error) {
+	if err := o.check(); err != nil {
+		return nil, errs.Wrap(err)
+	}
+
+	tags := make([]int, len(o.TargetKeys))
+	versions, err := o.getVersions(&tags)
+	if err != nil {
+		return nil, errs.Wrap(err)
+	}
+
+	fullMap := make(map[string]bool)
+	for i, tag := range tags {
+		switch tag {
+		case tagQuery:
+			vLog := versions[o.TargetKeys[i]]
+			fullMap[o.TargetKeys[i]] = vLog.ID.Hex() != o.VersionIDs[i] || uint64(vLog.Version) < o.VersionNumbers[i] || len(vLog.Logs) != vLog.LogLen
+		case tagFull:
+			fullMap[o.TargetKeys[i]] = true
+		case tagEqual:
+			fullMap[o.TargetKeys[i]] = false
+		default:
+			panic(fmt.Errorf("undefined tag %d", tag))
+		}
+	}
+
+	var (
+		insertIdsMap = make(map[string][]string)
+		deleteIdsMap = make(map[string][]string)
+		updateIdsMap = make(map[string][]string)
+	)
+
+	for _, targetKey := range o.TargetKeys {
+		if !fullMap[targetKey] {
+			version := versions[targetKey]
+			insertIds, deleteIds, updateIds := version.DeleteAndChangeIDs()
+			insertIdsMap[targetKey] = insertIds
+			deleteIdsMap[targetKey] = deleteIds
+			updateIdsMap[targetKey] = updateIds
+		}
+	}
+
+	var (
+		insertListMap = make(map[string]A)
+		updateListMap = make(map[string]A)
+	)
+
+	for targetKey, insertIds := range insertIdsMap {
+		if len(insertIds) > 0 {
+			insertList, err := o.Find(o.Ctx, targetKey, insertIds)
+			if err != nil {
+				return nil, errs.Wrap(err)
+			}
+			insertListMap[targetKey] = insertList
+		}
+	}
+
+	for targetKey, updateIds := range updateIdsMap {
+		if len(updateIds) > 0 {
+			updateList, err := o.Find(o.Ctx, targetKey, updateIds)
+			if err != nil {
+				return nil, errs.Wrap(err)
+			}
+			updateListMap[targetKey] = updateList
+		}
+	}
+
+	return o.Resp(versions, deleteIdsMap, insertListMap, updateListMap, fullMap), nil
+}

internal/rpc/incrversion/option.go

@@ -3,6 +3,7 @@ package incrversion
 import (
 	"context"
 	"fmt"
+
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
 	"github.com/openimsdk/tools/errs"
 	"go.mongodb.org/mongo-driver/bson/primitive"
@@ -20,7 +21,7 @@ const syncLimit = 200
 const (
 	tagQuery = iota + 1
 	tagFull
-	tageEqual
+	tagEqual
 )
 
 type Option[A, B any] struct {
@@ -33,7 +34,6 @@ type Option[A, B any] struct {
 	Version         func(ctx context.Context, dId string, version uint, limit int) (*model.VersionLog, error)
 	//SortID          func(ctx context.Context, dId string) ([]string, error)
 	Find func(ctx context.Context, ids []string) ([]A, error)
-	ID   func(elem A) string
 	Resp func(version *model.VersionLog, deleteIds []string, insertList, updateList []A, full bool) *B
 }
 
@@ -60,9 +60,6 @@ func (o *Option[A, B]) check() error {
 	if o.Find == nil {
 		return o.newError("func find is nil")
 	}
-	if o.ID == nil {
-		return o.newError("func id is nil")
-	}
 	if o.Resp == nil {
 		return o.newError("func resp is nil")
 	}
@@ -100,7 +97,7 @@ func (o *Option[A, B]) getVersion(tag *int) (*model.VersionLog, error) {
 			return cache, nil
 		}
 		if o.VersionNumber == uint64(cache.Version) {
-			*tag = tageEqual
+			*tag = tagEqual
 			return cache, nil
 		}
 		*tag = tagQuery
@@ -123,7 +120,7 @@ func (o *Option[A, B]) Build() (*B, error) {
 		full = version.ID.Hex() != o.VersionID || uint64(version.Version) < o.VersionNumber || len(version.Logs) != version.LogLen
 	case tagFull:
 		full = true
-	case tageEqual:
+	case tagEqual:
 		full = false
 	default:
 		panic(fmt.Errorf("undefined tag %d", tag))