From ab6c9dca713877264f5c44652b661710e8dc9ad3 Mon Sep 17 00:00:00 2001 From: chao <48119764+withchao@users.noreply.github.com> Date: Mon, 3 Mar 2025 17:59:35 +0800 Subject: [PATCH] feat: optimizing BatchGetIncrementalGroupMember (#3180) * 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: optimizing BatchGetIncrementalGroupMember --- internal/rpc/group/sync.go | 168 +++++-------------------------------- 1 file changed, 22 insertions(+), 146 deletions(-) diff --git a/internal/rpc/group/sync.go b/internal/rpc/group/sync.go index 50ac8252f..d311ae076 100644 --- a/internal/rpc/group/sync.go +++ b/internal/rpc/group/sync.go @@ -11,10 +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" ) +const versionSyncLimit = 500 + func (g *groupServer) GetFullGroupMemberUserIDs(ctx context.Context, req *pbgroup.GetFullGroupMemberUserIDsReq) (*pbgroup.GetFullGroupMemberUserIDsResp, error) { vl, err := g.db.FindMaxGroupMemberVersionCache(ctx, req.GroupID) if err != nil { @@ -132,150 +132,6 @@ func (g *groupServer) GetIncrementalGroupMember(ctx context.Context, req *pbgrou return resp, nil } -func (g *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 := g.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 := g.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: g.db.BatchFindMaxGroupMemberVersionCache, - Find: func(ctx context.Context, groupID string, ids []string) ([]*sdkws.GroupMemberFullInfo, error) { - memberInfo, err := g.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 := g.db.FindGroupMemberNum(ctx, groupID) - if err != nil { - return nil, err - } - - owner, err := g.db.TakeGroupOwner(ctx, groupID) - if err != nil { - return nil, err - } - - resp.RespList[groupID].Group = g.groupDB2PB(groupsMap[groupID], owner.UserID, count) - } - } - - return resp, nil - -} - func (g *groupServer) GetIncrementalJoinGroup(ctx context.Context, req *pbgroup.GetIncrementalJoinGroupReq) (*pbgroup.GetIncrementalJoinGroupResp, error) { if err := authverify.CheckAccessV3(ctx, req.UserID, g.config.Share.IMAdminUserID); err != nil { return nil, err @@ -301,3 +157,23 @@ func (g *groupServer) GetIncrementalJoinGroup(ctx context.Context, req *pbgroup. } return opt.Build() } + +func (g *groupServer) BatchGetIncrementalGroupMember(ctx context.Context, req *pbgroup.BatchGetIncrementalGroupMemberReq) (*pbgroup.BatchGetIncrementalGroupMemberResp, error) { + var num int + resp := make(map[string]*pbgroup.GetIncrementalGroupMemberResp) + for _, memberReq := range req.ReqList { + if _, ok := resp[memberReq.GroupID]; ok { + continue + } + memberResp, err := g.GetIncrementalGroupMember(ctx, memberReq) + if err != nil { + return nil, err + } + resp[memberReq.GroupID] = memberResp + num += len(memberResp.Insert) + len(memberResp.Update) + len(memberResp.Delete) + if num >= versionSyncLimit { + break + } + } + return &pbgroup.BatchGetIncrementalGroupMemberResp{RespList: resp}, nil +}