fix: can not receive OnJoinedGroupAdded notification (#1082)

WhereAreBugs · web-flow · commit 17b84d03cff5 · 2025-12-30T08:32:35.000Z
* refactor(group):let GetGroupMembersInfo use DataFetcher. * fix: can not receive OnJoinedGroupAdded notification (#1081) * fix: can not receive OnJoinedGroupAdded notification * refactor(group):let GetGroupMembersInfo use DataFetcher.
diff --git a/internal/conversation_msg/conversation_msg.go b/internal/conversation_msg/conversation_msg.go
@@ -851,7 +851,7 @@ func (c *Conversation) batchAddFaceURLAndName(ctx context.Context, conversations
 		return err
 	}
 
-	groupInfoList, err := c.group.GetSpecifiedGroupsInfo(ctx, groupIDs)
+	groupInfoList, err := c.group.GetSpecifiedGroupsInfoSafe(ctx, groupIDs)
 	if err != nil {
 		return err
 	}
diff --git a/internal/group/api.go b/internal/group/api.go
@@ -228,6 +228,74 @@ func (g *Group) GetSpecifiedGroupsInfo(ctx context.Context, groupIDs []string) (
 	return dataFetcher.FetchMissingAndCombineLocal(ctx, groupIDs)
 }
 
+// GetSpecifiedGroupsInfoSafe fetches group info without writing to local storage or touching version sync.
+func (g *Group) GetSpecifiedGroupsInfoSafe(ctx context.Context, groupIDs []string) ([]*model_struct.LocalGroup, error) {
+	if len(groupIDs) == 0 {
+		return nil, nil
+	}
+
+	dataFetcher := datafetcher.NewDataFetcher(
+		g.db,
+		g.groupTableName(),
+		g.loginUserID,
+		func(localGroup *model_struct.LocalGroup) string {
+			return localGroup.GroupID
+		},
+		func(ctx context.Context, values []*model_struct.LocalGroup) error {
+			return nil
+		},
+		func(ctx context.Context, groupIDs []string) ([]*model_struct.LocalGroup, bool, error) {
+			var (
+				res    []*model_struct.LocalGroup
+				needDB []string
+			)
+
+			for _, groupID := range groupIDs {
+				if v, ok := g.groupInfoCache.Load(groupID); ok {
+					res = append(res, v)
+				} else {
+					needDB = append(needDB, groupID)
+				}
+			}
+
+			if len(needDB) == 0 {
+				return res, false, nil
+			}
+
+			localGroups, err := g.db.GetGroups(ctx, needDB)
+			if err != nil {
+				return nil, false, err
+			}
+
+			localMap := datautil.SliceToMap(localGroups, func(e *model_struct.LocalGroup) string {
+				return e.GroupID
+			})
+			for _, info := range localGroups {
+				g.groupInfoCache.Store(info.GroupID, info)
+				res = append(res, info)
+			}
+
+			if len(localMap) == len(needDB) {
+				return res, false, nil
+			}
+
+			return res, true, nil
+		},
+		func(ctx context.Context, groupIDs []string) ([]*model_struct.LocalGroup, error) {
+			serverGroupInfo, err := g.getGroupsInfoFromServer(ctx, groupIDs)
+			if err != nil {
+				return nil, err
+			}
+			converted := datautil.Batch(ServerGroupToLocalGroup, serverGroupInfo)
+			for _, info := range converted {
+				g.groupInfoCache.Store(info.GroupID, info)
+			}
+			return converted, nil
+		},
+	)
+	return dataFetcher.FetchMissingAndCombineLocal(ctx, groupIDs)
+}
+
 func (g *Group) SearchGroups(ctx context.Context, param sdk_params_callback.SearchGroupsParam) ([]*model_struct.LocalGroup, error) {
 	if len(param.KeywordList) == 0 || (!param.IsSearchGroupName && !param.IsSearchGroupID) {
 		return nil, sdkerrs.ErrArgs.WrapMsg("keyword is null or search field all false")
diff --git a/internal/group/cache.go b/internal/group/cache.go
@@ -3,6 +3,7 @@ package group
 import (
 	"context"
 
+	"github.com/openimsdk/openim-sdk-core/v3/pkg/datafetcher"
 	"github.com/openimsdk/openim-sdk-core/v3/pkg/db/model_struct"
 	"github.com/openimsdk/tools/log"
 	"github.com/openimsdk/tools/utils/datautil"
@@ -45,25 +46,79 @@ func (g *Group) GetGroupMembersInfoFunc(ctx context.Context, groupID string, use
 }
 
 func (g *Group) GetGroupMembersInfo(ctx context.Context, groupID string, userIDs []string) (map[string]*model_struct.LocalGroupMember, error) {
-	return g.GetGroupMembersInfoFunc(ctx, groupID, userIDs, func(ctx context.Context, dbKeys []string) ([]*model_struct.LocalGroupMember, error) {
-		if len(dbKeys) == 0 {
-			return nil, nil
-		}
-		dbData, err := g.db.GetGroupSomeMemberInfo(ctx, groupID, dbKeys)
-		if err != nil {
-			return nil, err
-		}
-		queryKeys := datautil.SliceSubAny(dbKeys, dbData, func(t *model_struct.LocalGroupMember) string {
-			return t.UserID
-		})
-		if len(queryKeys) != 0 {
-			queryData, err := g.getDesignatedGroupMembers(ctx, groupID, queryKeys)
+	res := make(map[string]*model_struct.LocalGroupMember)
+	if len(userIDs) == 0 {
+		return res, nil
+	}
+
+	dataFetcher := datafetcher.NewDataFetcher(
+		g.db,
+		g.groupAndMemberVersionTableName(),
+		groupID,
+		func(member *model_struct.LocalGroupMember) string {
+			return member.UserID
+		},
+		func(ctx context.Context, values []*model_struct.LocalGroupMember) error {
+			return nil
+		},
+		func(ctx context.Context, userIDs []string) ([]*model_struct.LocalGroupMember, bool, error) {
+			var (
+				localData []*model_struct.LocalGroupMember
+				needDB    []string
+			)
+
+			for _, userID := range userIDs {
+				key := g.buildGroupMemberKey(groupID, userID)
+				if member, ok := g.groupMemberCache.Load(key); ok {
+					localData = append(localData, member)
+				} else {
+					needDB = append(needDB, userID)
+				}
+			}
+
+			if len(needDB) == 0 {
+				return localData, false, nil
+			}
+
+			dbData, err := g.db.GetGroupSomeMemberInfo(ctx, groupID, needDB)
+			if err != nil {
+				return nil, false, err
+			}
+
+			for _, member := range dbData {
+				g.groupMemberCache.Store(g.buildGroupMemberKey(groupID, member.UserID), member)
+				localData = append(localData, member)
+			}
+
+			if len(dbData) == len(needDB) {
+				return localData, false, nil
+			}
+
+			return localData, true, nil
+		},
+		func(ctx context.Context, userIDs []string) ([]*model_struct.LocalGroupMember, error) {
+			if len(userIDs) == 0 {
+				return nil, nil
+			}
+			queryData, err := g.getDesignatedGroupMembers(ctx, groupID, userIDs)
 			if err != nil {
 				return nil, err
 			}
+			converted := datautil.Batch(ServerGroupMemberToLocalGroupMember, queryData)
+			for _, member := range converted {
+				g.groupMemberCache.Store(g.buildGroupMemberKey(groupID, member.UserID), member)
+			}
+			return converted, nil
+		},
+	)
 
-			dbData = append(dbData, datautil.Batch(ServerGroupMemberToLocalGroupMember, queryData)...)
-		}
-		return dbData, nil
-	})
+	fetchData, err := dataFetcher.FetchMissingAndCombineLocal(ctx, userIDs)
+	if err != nil {
+		return nil, err
+	}
+
+	for _, member := range fetchData {
+		res[member.UserID] = member
+	}
+	return res, nil
 }
diff --git a/internal/group/group.go b/internal/group/group.go
@@ -52,6 +52,7 @@ func NewGroup(
 	}
 	g.initSyncer()
 	g.groupMemberCache = cache.NewCache[string, *model_struct.LocalGroupMember]()
+	g.groupInfoCache = cache.NewCache[string, *model_struct.LocalGroup]()
 	return g
 }
 
@@ -65,15 +66,21 @@ type Group struct {
 	groupSyncMutex         sync.Mutex
 	listenerForService     open_im_sdk_callback.OnListenerForService
 	groupMemberCache       *cache.Cache[string, *model_struct.LocalGroupMember]
+	groupInfoCache         *cache.Cache[string, *model_struct.LocalGroup]
 	filter                 *NotificationFilter
 }
 
 func (g *Group) initSyncer() {
 	g.groupSyncer = syncer.New2[*model_struct.LocalGroup, group.GetJoinedGroupListResp, string](
 		syncer.WithInsert[*model_struct.LocalGroup, group.GetJoinedGroupListResp, string](func(ctx context.Context, value *model_struct.LocalGroup) error {
-			return g.db.InsertGroup(ctx, value)
+			if err := g.db.InsertGroup(ctx, value); err != nil {
+				return err
+			}
+			g.groupInfoCache.Store(value.GroupID, value)
+			return nil
 		}),
 		syncer.WithDelete[*model_struct.LocalGroup, group.GetJoinedGroupListResp, string](func(ctx context.Context, value *model_struct.LocalGroup) error {
+			g.groupInfoCache.Delete(value.GroupID)
 			if err := g.db.DeleteGroupAllMembers(ctx, value.GroupID); err != nil {
 				return err
 			}
@@ -84,7 +91,11 @@ func (g *Group) initSyncer() {
 		}),
 		syncer.WithUpdate[*model_struct.LocalGroup, group.GetJoinedGroupListResp, string](func(ctx context.Context, server, local *model_struct.LocalGroup) error {
 			log.ZInfo(ctx, "groupSyncer trigger update function", "groupID", server.GroupID, "server", server, "local", local)
-			return g.db.UpdateGroup(ctx, server)
+			if err := g.db.UpdateGroup(ctx, server); err != nil {
+				return err
+			}
+			g.groupInfoCache.Store(server.GroupID, server)
+			return nil
 		}),
 		syncer.WithUUID[*model_struct.LocalGroup, group.GetJoinedGroupListResp, string](func(value *model_struct.LocalGroup) string {
 			return value.GroupID
@@ -130,9 +141,14 @@ func (g *Group) initSyncer() {
 		}),
 
 		syncer.WithBatchInsert[*model_struct.LocalGroup, group.GetJoinedGroupListResp, string](func(ctx context.Context, values []*model_struct.LocalGroup) error {
-			return g.db.BatchInsertGroup(ctx, values)
+			if err := g.db.BatchInsertGroup(ctx, values); err != nil {
+				return err
+			}
+			g.groupInfoCache.StoreAll(func(v *model_struct.LocalGroup) string { return v.GroupID }, values)
+			return nil
 		}),
 		syncer.WithDeleteAll[*model_struct.LocalGroup, group.GetJoinedGroupListResp, string](func(ctx context.Context, _ string) error {
+			g.groupInfoCache.DeleteAll()
 			return g.db.DeleteAllGroup(ctx)
 		}),
 		syncer.WithBatchPageReq[*model_struct.LocalGroup, group.GetJoinedGroupListResp, string](func(entityID string) page.PageReq {

Original file line number	Diff line number	Diff line change
`@@ -851,7 +851,7 @@ func (c *Conversation) batchAddFaceURLAndName(ctx context.Context, conversations`
`851`	`851`	`return err`
`852`	`852`	`}`
`853`	`853`
`854`		`- groupInfoList, err := c.group.GetSpecifiedGroupsInfo(ctx, groupIDs)`
	`854`	`+ groupInfoList, err := c.group.GetSpecifiedGroupsInfoSafe(ctx, groupIDs)`
`855`	`855`	`if err != nil {`
`856`	`856`	`return err`
`857`	`857`	`}`