From 6ffcfbe68e358191407147dd257786cef68192e3 Mon Sep 17 00:00:00 2001 From: withchao <48119764+withchao@users.noreply.github.com> Date: Mon, 23 Oct 2023 08:24:22 -0500 Subject: [PATCH 1/3] fix: not support redis cluster. CROSSSLOT Keys in request don't hash to the same slot (#1258) * feat: v2 to v3 data conversion * feat: v2 to v3 data conversion * fix: CallbackBeforeCreateGroup * fix: NotificationUserInfoUpdate * fix: NotificationUserInfoUpdate * chore: update pkg github.com/OpenIMSDK/protocol v0.0.26 * chore: code format * update pkg * feat: QuitGroup support administrator operations * feat: QuitGroup support administrator operations * fix: checkMongo uri * fix: k8s minio prefix * fix: k8s minio prefix * fix: k8s minio prefix * fix: k8s minio prefix test * fix: k8s minio prefix test * fix: GetUsersInfo cache * fix: redis cache * fix: redis remove pipeline * fix: redis remove pipeline --- pkg/common/db/cache/black.go | 19 +- pkg/common/db/cache/conversation.go | 266 ++++++---------- pkg/common/db/cache/friend.go | 40 +-- pkg/common/db/cache/group.go | 266 ++++++---------- pkg/common/db/cache/meta_cache.go | 229 +++++++------- pkg/common/db/cache/msg.go | 374 +++++++++++++---------- pkg/common/db/cache/user.go | 49 +-- pkg/common/db/controller/conversation.go | 8 +- 8 files changed, 550 insertions(+), 701 deletions(-) diff --git a/pkg/common/db/cache/black.go b/pkg/common/db/cache/black.go index d1abe945c..5a70097ed 100644 --- a/pkg/common/db/cache/black.go +++ b/pkg/common/db/cache/black.go @@ -62,12 +62,7 @@ func NewBlackCacheRedis( } func (b *BlackCacheRedis) NewCache() BlackCache { - return &BlackCacheRedis{ - expireTime: b.expireTime, - rcClient: b.rcClient, - blackDB: b.blackDB, - metaCache: NewMetaCacheRedis(b.rcClient, b.metaCache.GetPreDelKeys()...), - } + return &BlackCacheRedis{expireTime: b.expireTime, rcClient: b.rcClient, blackDB: b.blackDB, metaCache: NewMetaCacheRedis(b.rcClient, b.metaCache.GetPreDelKeys()...)} } func (b *BlackCacheRedis) getBlackIDsKey(ownerUserID string) string { @@ -75,15 +70,9 @@ func (b *BlackCacheRedis) getBlackIDsKey(ownerUserID string) string { } func (b *BlackCacheRedis) GetBlackIDs(ctx context.Context, userID string) (blackIDs []string, err error) { - return getCache( - ctx, - b.rcClient, - b.getBlackIDsKey(userID), - b.expireTime, - func(ctx context.Context) ([]string, error) { - return b.blackDB.FindBlackUserIDs(ctx, userID) - }, - ) + return getCache(ctx, b.rcClient, b.getBlackIDsKey(userID), b.expireTime, func(ctx context.Context) ([]string, error) { + return b.blackDB.FindBlackUserIDs(ctx, userID) + }) } func (b *BlackCacheRedis) DelBlackIDs(ctx context.Context, userID string) BlackCache { diff --git a/pkg/common/db/cache/conversation.go b/pkg/common/db/cache/conversation.go index 890552160..9c0bcfae4 100644 --- a/pkg/common/db/cache/conversation.go +++ b/pkg/common/db/cache/conversation.go @@ -73,7 +73,7 @@ type ConversationCache interface { GetSuperGroupRecvMsgNotNotifyUserIDsHash(ctx context.Context, groupID string) (hash uint64, err error) DelSuperGroupRecvMsgNotNotifyUserIDsHash(groupID string) ConversationCache - GetUserAllHasReadSeqs(ctx context.Context, ownerUserID string) (map[string]int64, error) + //GetUserAllHasReadSeqs(ctx context.Context, ownerUserID string) (map[string]int64, error) DelUserAllHasReadSeqs(ownerUserID string, conversationIDs ...string) ConversationCache GetConversationsByConversationID(ctx context.Context, @@ -83,11 +83,7 @@ type ConversationCache interface { DelConversationNotReceiveMessageUserIDs(conversationIDs ...string) ConversationCache } -func NewConversationRedis( - rdb redis.UniversalClient, - opts rockscache.Options, - db relationtb.ConversationModelInterface, -) ConversationCache { +func NewConversationRedis(rdb redis.UniversalClient, opts rockscache.Options, db relationtb.ConversationModelInterface) ConversationCache { rcClient := rockscache.NewClient(rdb, opts) return &ConversationRedisCache{ @@ -158,15 +154,9 @@ func (c *ConversationRedisCache) getConversationNotReceiveMessageUserIDsKey(conv } func (c *ConversationRedisCache) GetUserConversationIDs(ctx context.Context, ownerUserID string) ([]string, error) { - return getCache( - ctx, - c.rcClient, - c.getConversationIDsKey(ownerUserID), - c.expireTime, - func(ctx context.Context) ([]string, error) { - return c.conversationDB.FindUserIDAllConversationID(ctx, ownerUserID) - }, - ) + return getCache(ctx, c.rcClient, c.getConversationIDsKey(ownerUserID), c.expireTime, func(ctx context.Context) ([]string, error) { + return c.conversationDB.FindUserIDAllConversationID(ctx, ownerUserID) + }) } func (c *ConversationRedisCache) DelConversationIDs(userIDs ...string) ConversationCache { @@ -184,10 +174,7 @@ func (c *ConversationRedisCache) getUserConversationIDsHashKey(ownerUserID strin return conversationIDsHashKey + ownerUserID } -func (c *ConversationRedisCache) GetUserConversationIDsHash( - ctx context.Context, - ownerUserID string, -) (hash uint64, err error) { +func (c *ConversationRedisCache) GetUserConversationIDsHash(ctx context.Context, ownerUserID string) (hash uint64, err error) { return getCache( ctx, c.rcClient, @@ -201,7 +188,6 @@ func (c *ConversationRedisCache) GetUserConversationIDsHash( utils.Sort(conversationIDs, true) bi := big.NewInt(0) bi.SetString(utils.Md5(strings.Join(conversationIDs, ";"))[0:8], 16) - return bi.Uint64(), nil }, ) @@ -218,19 +204,10 @@ func (c *ConversationRedisCache) DelUserConversationIDsHash(ownerUserIDs ...stri return cache } -func (c *ConversationRedisCache) GetConversation( - ctx context.Context, - ownerUserID, conversationID string, -) (*relationtb.ConversationModel, error) { - return getCache( - ctx, - c.rcClient, - c.getConversationKey(ownerUserID, conversationID), - c.expireTime, - func(ctx context.Context) (*relationtb.ConversationModel, error) { - return c.conversationDB.Take(ctx, ownerUserID, conversationID) - }, - ) +func (c *ConversationRedisCache) GetConversation(ctx context.Context, ownerUserID, conversationID string) (*relationtb.ConversationModel, error) { + return getCache(ctx, c.rcClient, c.getConversationKey(ownerUserID, conversationID), c.expireTime, func(ctx context.Context) (*relationtb.ConversationModel, error) { + return c.conversationDB.Take(ctx, ownerUserID, conversationID) + }) } func (c *ConversationRedisCache) DelConversations(ownerUserID string, conversationIDs ...string) ConversationCache { @@ -244,10 +221,7 @@ func (c *ConversationRedisCache) DelConversations(ownerUserID string, conversati return cache } -func (c *ConversationRedisCache) getConversationIndex( - convsation *relationtb.ConversationModel, - keys []string, -) (int, error) { +func (c *ConversationRedisCache) getConversationIndex(convsation *relationtb.ConversationModel, keys []string) (int, error) { key := c.getConversationKey(convsation.OwnerUserID, convsation.ConversationID) for _i, _key := range keys { if _key == key { @@ -258,81 +232,60 @@ func (c *ConversationRedisCache) getConversationIndex( return 0, errors.New("not found key:" + key + " in keys") } -func (c *ConversationRedisCache) GetConversations( - ctx context.Context, - ownerUserID string, - conversationIDs []string, -) ([]*relationtb.ConversationModel, error) { - keys := make([]string, 0, len(conversationIDs)) - for _, conversarionID := range conversationIDs { - keys = append(keys, c.getConversationKey(ownerUserID, conversarionID)) - } - - return batchGetCache( - ctx, - c.rcClient, - keys, - c.expireTime, - c.getConversationIndex, - func(ctx context.Context) ([]*relationtb.ConversationModel, error) { - return c.conversationDB.Find(ctx, ownerUserID, conversationIDs) - }, - ) -} - -func (c *ConversationRedisCache) GetUserAllConversations( - ctx context.Context, - ownerUserID string, -) ([]*relationtb.ConversationModel, error) { +func (c *ConversationRedisCache) GetConversations(ctx context.Context, ownerUserID string, conversationIDs []string) ([]*relationtb.ConversationModel, error) { + //var keys []string + //for _, conversarionID := range conversationIDs { + // keys = append(keys, c.getConversationKey(ownerUserID, conversarionID)) + //} + //return batchGetCache( + // ctx, + // c.rcClient, + // keys, + // c.expireTime, + // c.getConversationIndex, + // func(ctx context.Context) ([]*relationtb.ConversationModel, error) { + // return c.conversationDB.Find(ctx, ownerUserID, conversationIDs) + // }, + //) + return batchGetCache2(ctx, c.rcClient, c.expireTime, conversationIDs, func(conversationID string) string { + return c.getConversationKey(ownerUserID, conversationID) + }, func(ctx context.Context, conversationID string) (*relationtb.ConversationModel, error) { + return c.conversationDB.Take(ctx, ownerUserID, conversationID) + }) +} + +func (c *ConversationRedisCache) GetUserAllConversations(ctx context.Context, ownerUserID string) ([]*relationtb.ConversationModel, error) { conversationIDs, err := c.GetUserConversationIDs(ctx, ownerUserID) if err != nil { return nil, err } - keys := make([]string, 0, len(conversationIDs)) - for _, conversarionID := range conversationIDs { - keys = append(keys, c.getConversationKey(ownerUserID, conversarionID)) - } - - return batchGetCache( - ctx, - c.rcClient, - keys, - c.expireTime, - c.getConversationIndex, - func(ctx context.Context) ([]*relationtb.ConversationModel, error) { - return c.conversationDB.FindUserIDAllConversations(ctx, ownerUserID) - }, - ) -} - -func (c *ConversationRedisCache) GetUserRecvMsgOpt( - ctx context.Context, - ownerUserID, conversationID string, -) (opt int, err error) { - return getCache( - ctx, - c.rcClient, - c.getRecvMsgOptKey(ownerUserID, conversationID), - c.expireTime, - func(ctx context.Context) (opt int, err error) { - return c.conversationDB.GetUserRecvMsgOpt(ctx, ownerUserID, conversationID) - }, - ) -} - -func (c *ConversationRedisCache) GetSuperGroupRecvMsgNotNotifyUserIDs( - ctx context.Context, - groupID string, -) (userIDs []string, err error) { - return getCache( - ctx, - c.rcClient, - c.getSuperGroupRecvNotNotifyUserIDsKey(groupID), - c.expireTime, - func(ctx context.Context) (userIDs []string, err error) { - return c.conversationDB.FindSuperGroupRecvMsgNotNotifyUserIDs(ctx, groupID) - }, - ) + //var keys []string + //for _, conversarionID := range conversationIDs { + // keys = append(keys, c.getConversationKey(ownerUserID, conversarionID)) + //} + //return batchGetCache( + // ctx, + // c.rcClient, + // keys, + // c.expireTime, + // c.getConversationIndex, + // func(ctx context.Context) ([]*relationtb.ConversationModel, error) { + // return c.conversationDB.FindUserIDAllConversations(ctx, ownerUserID) + // }, + //) + return c.GetConversations(ctx, ownerUserID, conversationIDs) +} + +func (c *ConversationRedisCache) GetUserRecvMsgOpt(ctx context.Context, ownerUserID, conversationID string) (opt int, err error) { + return getCache(ctx, c.rcClient, c.getRecvMsgOptKey(ownerUserID, conversationID), c.expireTime, func(ctx context.Context) (opt int, err error) { + return c.conversationDB.GetUserRecvMsgOpt(ctx, ownerUserID, conversationID) + }) +} + +func (c *ConversationRedisCache) GetSuperGroupRecvMsgNotNotifyUserIDs(ctx context.Context, groupID string) (userIDs []string, err error) { + return getCache(ctx, c.rcClient, c.getSuperGroupRecvNotNotifyUserIDsKey(groupID), c.expireTime, func(ctx context.Context) (userIDs []string, err error) { + return c.conversationDB.FindSuperGroupRecvMsgNotNotifyUserIDs(ctx, groupID) + }) } func (c *ConversationRedisCache) DelUsersConversation(conversationID string, ownerUserIDs ...string) ConversationCache { @@ -360,26 +313,17 @@ func (c *ConversationRedisCache) DelSuperGroupRecvMsgNotNotifyUserIDs(groupID st return cache } -func (c *ConversationRedisCache) GetSuperGroupRecvMsgNotNotifyUserIDsHash( - ctx context.Context, - groupID string, -) (hash uint64, err error) { - return getCache( - ctx, - c.rcClient, - c.getSuperGroupRecvNotNotifyUserIDsHashKey(groupID), - c.expireTime, - func(ctx context.Context) (hash uint64, err error) { - userIDs, err := c.GetSuperGroupRecvMsgNotNotifyUserIDs(ctx, groupID) - if err != nil { - return 0, err - } - utils.Sort(userIDs, true) - bi := big.NewInt(0) - bi.SetString(utils.Md5(strings.Join(userIDs, ";"))[0:8], 16) - - return bi.Uint64(), nil - }, +func (c *ConversationRedisCache) GetSuperGroupRecvMsgNotNotifyUserIDsHash(ctx context.Context, groupID string) (hash uint64, err error) { + return getCache(ctx, c.rcClient, c.getSuperGroupRecvNotNotifyUserIDsHashKey(groupID), c.expireTime, func(ctx context.Context) (hash uint64, err error) { + userIDs, err := c.GetSuperGroupRecvMsgNotNotifyUserIDs(ctx, groupID) + if err != nil { + return 0, err + } + utils.Sort(userIDs, true) + bi := big.NewInt(0) + bi.SetString(utils.Md5(strings.Join(userIDs, ";"))[0:8], 16) + return bi.Uint64(), nil + }, ) } @@ -390,10 +334,7 @@ func (c *ConversationRedisCache) DelSuperGroupRecvMsgNotNotifyUserIDsHash(groupI return cache } -func (c *ConversationRedisCache) getUserAllHasReadSeqsIndex( - conversationID string, - conversationIDs []string, -) (int, error) { +func (c *ConversationRedisCache) getUserAllHasReadSeqsIndex(conversationID string, conversationIDs []string) (int, error) { for _i, _conversationID := range conversationIDs { if _conversationID == conversationID { return _i, nil @@ -403,35 +344,21 @@ func (c *ConversationRedisCache) getUserAllHasReadSeqsIndex( return 0, errors.New("not found key:" + conversationID + " in keys") } -func (c *ConversationRedisCache) GetUserAllHasReadSeqs( - ctx context.Context, - ownerUserID string, -) (map[string]int64, error) { - conversationIDs, err := c.GetUserConversationIDs(ctx, ownerUserID) - if err != nil { - return nil, err - } - keys := make([]string, 0, len(conversationIDs)) - for _, conversarionID := range conversationIDs { - keys = append(keys, c.getConversationHasReadSeqKey(ownerUserID, conversarionID)) - } - - return batchGetCacheMap( - ctx, - c.rcClient, - keys, - conversationIDs, - c.expireTime, - c.getUserAllHasReadSeqsIndex, - func(ctx context.Context) (map[string]int64, error) { - return c.conversationDB.GetUserAllHasReadSeqs(ctx, ownerUserID) - }, - ) -} - -func (c *ConversationRedisCache) DelUserAllHasReadSeqs(ownerUserID string, - conversationIDs ...string, -) ConversationCache { +//func (c *ConversationRedisCache) GetUserAllHasReadSeqs(ctx context.Context, ownerUserID string) (map[string]int64, error) { +// conversationIDs, err := c.GetUserConversationIDs(ctx, ownerUserID) +// if err != nil { +// return nil, err +// } +// var keys []string +// for _, conversarionID := range conversationIDs { +// keys = append(keys, c.getConversationHasReadSeqKey(ownerUserID, conversarionID)) +// } +// return batchGetCacheMap(ctx, c.rcClient, keys, conversationIDs, c.expireTime, c.getUserAllHasReadSeqsIndex, func(ctx context.Context) (map[string]int64, error) { +// return c.conversationDB.GetUserAllHasReadSeqs(ctx, ownerUserID) +// }) +//} + +func (c *ConversationRedisCache) DelUserAllHasReadSeqs(ownerUserID string, conversationIDs ...string) ConversationCache { cache := c.NewCache() for _, conversationID := range conversationIDs { cache.AddKeys(c.getConversationHasReadSeqKey(ownerUserID, conversationID)) @@ -440,10 +367,7 @@ func (c *ConversationRedisCache) DelUserAllHasReadSeqs(ownerUserID string, return cache } -func (c *ConversationRedisCache) GetConversationsByConversationID( - ctx context.Context, - conversationIDs []string, -) ([]*relationtb.ConversationModel, error) { +func (c *ConversationRedisCache) GetConversationsByConversationID(ctx context.Context, conversationIDs []string) ([]*relationtb.ConversationModel, error) { panic("implement me") } @@ -452,15 +376,9 @@ func (c *ConversationRedisCache) DelConversationByConversationID(conversationIDs } func (c *ConversationRedisCache) GetConversationNotReceiveMessageUserIDs(ctx context.Context, conversationID string) ([]string, error) { - return getCache( - ctx, - c.rcClient, - c.getConversationNotReceiveMessageUserIDsKey(conversationID), - c.expireTime, - func(ctx context.Context) ([]string, error) { - return c.conversationDB.GetConversationNotReceiveMessageUserIDs(ctx, conversationID) - }, - ) + return getCache(ctx, c.rcClient, c.getConversationNotReceiveMessageUserIDsKey(conversationID), c.expireTime, func(ctx context.Context) ([]string, error) { + return c.conversationDB.GetConversationNotReceiveMessageUserIDs(ctx, conversationID) + }) } func (c *ConversationRedisCache) DelConversationNotReceiveMessageUserIDs(conversationIDs ...string) ConversationCache { diff --git a/pkg/common/db/cache/friend.go b/pkg/common/db/cache/friend.go index 37f5b0a98..28bb971b4 100644 --- a/pkg/common/db/cache/friend.go +++ b/pkg/common/db/cache/friend.go @@ -53,11 +53,7 @@ type FriendCacheRedis struct { rcClient *rockscache.Client } -func NewFriendCacheRedis( - rdb redis.UniversalClient, - friendDB relationtb.FriendModelInterface, - options rockscache.Options, -) FriendCache { +func NewFriendCacheRedis(rdb redis.UniversalClient, friendDB relationtb.FriendModelInterface, options rockscache.Options) FriendCache { rcClient := rockscache.NewClient(rdb, options) return &FriendCacheRedis{ @@ -90,15 +86,9 @@ func (f *FriendCacheRedis) getFriendKey(ownerUserID, friendUserID string) string } func (f *FriendCacheRedis) GetFriendIDs(ctx context.Context, ownerUserID string) (friendIDs []string, err error) { - return getCache( - ctx, - f.rcClient, - f.getFriendIDsKey(ownerUserID), - f.expireTime, - func(ctx context.Context) ([]string, error) { - return f.friendDB.FindFriendUserIDs(ctx, ownerUserID) - }, - ) + return getCache(ctx, f.rcClient, f.getFriendIDsKey(ownerUserID), f.expireTime, func(ctx context.Context) ([]string, error) { + return f.friendDB.FindFriendUserIDs(ctx, ownerUserID) + }) } func (f *FriendCacheRedis) DelFriendIDs(ownerUserIDs ...string) FriendCache { @@ -113,10 +103,7 @@ func (f *FriendCacheRedis) DelFriendIDs(ownerUserIDs ...string) FriendCache { } // todo. -func (f *FriendCacheRedis) GetTwoWayFriendIDs( - ctx context.Context, - ownerUserID string, -) (twoWayFriendIDs []string, err error) { +func (f *FriendCacheRedis) GetTwoWayFriendIDs(ctx context.Context, ownerUserID string) (twoWayFriendIDs []string, err error) { friendIDs, err := f.GetFriendIDs(ctx, ownerUserID) if err != nil { return nil, err @@ -141,19 +128,10 @@ func (f *FriendCacheRedis) DelTwoWayFriendIDs(ctx context.Context, ownerUserID s return newFriendCache } -func (f *FriendCacheRedis) GetFriend( - ctx context.Context, - ownerUserID, friendUserID string, -) (friend *relationtb.FriendModel, err error) { - return getCache( - ctx, - f.rcClient, - f.getFriendKey(ownerUserID, friendUserID), - f.expireTime, - func(ctx context.Context) (*relationtb.FriendModel, error) { - return f.friendDB.Take(ctx, ownerUserID, friendUserID) - }, - ) +func (f *FriendCacheRedis) GetFriend(ctx context.Context, ownerUserID, friendUserID string) (friend *relationtb.FriendModel, err error) { + return getCache(ctx, f.rcClient, f.getFriendKey(ownerUserID, friendUserID), f.expireTime, func(ctx context.Context) (*relationtb.FriendModel, error) { + return f.friendDB.Take(ctx, ownerUserID, friendUserID) + }) } func (f *FriendCacheRedis) DelFriend(ownerUserID, friendUserID string) FriendCache { diff --git a/pkg/common/db/cache/group.go b/pkg/common/db/cache/group.go index 0505241d0..d505772eb 100644 --- a/pkg/common/db/cache/group.go +++ b/pkg/common/db/cache/group.go @@ -65,22 +65,10 @@ type GroupCache interface { GetJoinedGroupIDs(ctx context.Context, userID string) (joinedGroupIDs []string, err error) DelJoinedGroupID(userID ...string) GroupCache - GetGroupMemberInfo( - ctx context.Context, - groupID, userID string, - ) (groupMember *relationtb.GroupMemberModel, err error) - GetGroupMembersInfo( - ctx context.Context, - groupID string, - userID []string, - ) (groupMembers []*relationtb.GroupMemberModel, err error) + GetGroupMemberInfo(ctx context.Context, groupID, userID string) (groupMember *relationtb.GroupMemberModel, err error) + GetGroupMembersInfo(ctx context.Context, groupID string, userID []string) (groupMembers []*relationtb.GroupMemberModel, err error) GetAllGroupMembersInfo(ctx context.Context, groupID string) (groupMembers []*relationtb.GroupMemberModel, err error) - GetGroupMembersPage( - ctx context.Context, - groupID string, - userID []string, - showNumber, pageNumber int32, - ) (total uint32, groupMembers []*relationtb.GroupMemberModel, err error) + GetGroupMembersPage(ctx context.Context, groupID string, userID []string, showNumber, pageNumber int32) (total uint32, groupMembers []*relationtb.GroupMemberModel, err error) DelGroupMembersInfo(groupID string, userID ...string) GroupCache @@ -186,37 +174,25 @@ func (g *GroupCacheRedis) GetGroupMemberIndex(groupMember *relationtb.GroupMembe } // / groupInfo. -func (g *GroupCacheRedis) GetGroupsInfo( - ctx context.Context, - groupIDs []string, -) (groups []*relationtb.GroupModel, err error) { - keys := make([]string, 0, len(groupIDs)) - for _, group := range groupIDs { - keys = append(keys, g.getGroupInfoKey(group)) - } - - return batchGetCache( - ctx, - g.rcClient, - keys, - g.expireTime, - g.GetGroupIndex, - func(ctx context.Context) ([]*relationtb.GroupModel, error) { - return g.groupDB.Find(ctx, groupIDs) - }, - ) +func (g *GroupCacheRedis) GetGroupsInfo(ctx context.Context, groupIDs []string) (groups []*relationtb.GroupModel, err error) { + //var keys []string + //for _, group := range groupIDs { + // keys = append(keys, g.getGroupInfoKey(group)) + //} + //return batchGetCache(ctx, g.rcClient, keys, g.expireTime, g.GetGroupIndex, func(ctx context.Context) ([]*relationtb.GroupModel, error) { + // return g.groupDB.Find(ctx, groupIDs) + //}) + return batchGetCache2(ctx, g.rcClient, g.expireTime, groupIDs, func(groupID string) string { + return g.getGroupInfoKey(groupID) + }, func(ctx context.Context, groupID string) (*relationtb.GroupModel, error) { + return g.groupDB.Take(ctx, groupID) + }) } func (g *GroupCacheRedis) GetGroupInfo(ctx context.Context, groupID string) (group *relationtb.GroupModel, err error) { - return getCache( - ctx, - g.rcClient, - g.getGroupInfoKey(groupID), - g.expireTime, - func(ctx context.Context) (*relationtb.GroupModel, error) { - return g.groupDB.Take(ctx, groupID) - }, - ) + return getCache(ctx, g.rcClient, g.getGroupInfoKey(groupID), g.expireTime, func(ctx context.Context) (*relationtb.GroupModel, error) { + return g.groupDB.Take(ctx, groupID) + }) } func (g *GroupCacheRedis) DelGroupsInfo(groupIDs ...string) GroupCache { @@ -230,53 +206,38 @@ func (g *GroupCacheRedis) DelGroupsInfo(groupIDs ...string) GroupCache { return newGroupCache } -func (g *GroupCacheRedis) GetJoinedSuperGroupIDs( - ctx context.Context, - userID string, -) (joinedSuperGroupIDs []string, err error) { - return getCache( - ctx, - g.rcClient, - g.getJoinedSuperGroupsIDKey(userID), - g.expireTime, - func(ctx context.Context) ([]string, error) { - userGroup, err := g.mongoDB.GetSuperGroupByUserID(ctx, userID) - if err != nil { - return nil, err - } - - return userGroup.GroupIDs, nil - }, +func (g *GroupCacheRedis) GetJoinedSuperGroupIDs(ctx context.Context, userID string) (joinedSuperGroupIDs []string, err error) { + return getCache(ctx, g.rcClient, g.getJoinedSuperGroupsIDKey(userID), g.expireTime, func(ctx context.Context) ([]string, error) { + userGroup, err := g.mongoDB.GetSuperGroupByUserID(ctx, userID) + if err != nil { + return nil, err + } + return userGroup.GroupIDs, nil + }, ) } -func (g *GroupCacheRedis) GetSuperGroupMemberIDs( - ctx context.Context, - groupIDs ...string, -) (models []*unrelationtb.SuperGroupModel, err error) { - keys := make([]string, 0, len(groupIDs)) - for _, group := range groupIDs { - keys = append(keys, g.getSuperGroupMemberIDsKey(group)) - } - - return batchGetCache( - ctx, - g.rcClient, - keys, - g.expireTime, - func(model *unrelationtb.SuperGroupModel, keys []string) (int, error) { - for i, key := range keys { - if g.getSuperGroupMemberIDsKey(model.GroupID) == key { - return i, nil - } - } - - return 0, errIndex - }, - func(ctx context.Context) ([]*unrelationtb.SuperGroupModel, error) { - return g.mongoDB.FindSuperGroup(ctx, groupIDs) - }, - ) +func (g *GroupCacheRedis) GetSuperGroupMemberIDs(ctx context.Context, groupIDs ...string) (models []*unrelationtb.SuperGroupModel, err error) { + //var keys []string + //for _, group := range groupIDs { + // keys = append(keys, g.getSuperGroupMemberIDsKey(group)) + //} + //return batchGetCache(ctx, g.rcClient, keys, g.expireTime, func(model *unrelationtb.SuperGroupModel, keys []string) (int, error) { + // for i, key := range keys { + // if g.getSuperGroupMemberIDsKey(model.GroupID) == key { + // return i, nil + // } + // } + // return 0, errIndex + //}, + // func(ctx context.Context) ([]*unrelationtb.SuperGroupModel, error) { + // return g.mongoDB.FindSuperGroup(ctx, groupIDs) + // }) + return batchGetCache2(ctx, g.rcClient, g.expireTime, groupIDs, func(groupID string) string { + return g.getSuperGroupMemberIDsKey(groupID) + }, func(ctx context.Context, groupID string) (*unrelationtb.SuperGroupModel, error) { + return g.mongoDB.TakeSuperGroup(ctx, groupID) + }) } // userJoinSuperGroup. @@ -361,10 +322,7 @@ func (g *GroupCacheRedis) GetGroupMembersHash(ctx context.Context, groupID strin //) } -func (g *GroupCacheRedis) GetGroupMemberHashMap( - ctx context.Context, - groupIDs []string, -) (map[string]*relationtb.GroupSimpleUserID, error) { +func (g *GroupCacheRedis) GetGroupMemberHashMap(ctx context.Context, groupIDs []string) (map[string]*relationtb.GroupSimpleUserID, error) { res := make(map[string]*relationtb.GroupSimpleUserID) for _, groupID := range groupIDs { hash, err := g.GetGroupMembersHash(ctx, groupID) @@ -391,15 +349,9 @@ func (g *GroupCacheRedis) DelGroupMembersHash(groupID string) GroupCache { // groupMemberIDs. func (g *GroupCacheRedis) GetGroupMemberIDs(ctx context.Context, groupID string) (groupMemberIDs []string, err error) { - return getCache( - ctx, - g.rcClient, - g.getGroupMemberIDsKey(groupID), - g.expireTime, - func(ctx context.Context) ([]string, error) { - return g.groupMemberDB.FindMemberUserID(ctx, groupID) - }, - ) + return getCache(ctx, g.rcClient, g.getGroupMemberIDsKey(groupID), g.expireTime, func(ctx context.Context) ([]string, error) { + return g.groupMemberDB.FindMemberUserID(ctx, groupID) + }) } func (g *GroupCacheRedis) GetGroupsMemberIDs(ctx context.Context, groupIDs []string) (map[string][]string, error) { @@ -423,15 +375,9 @@ func (g *GroupCacheRedis) DelGroupMemberIDs(groupID string) GroupCache { } func (g *GroupCacheRedis) GetJoinedGroupIDs(ctx context.Context, userID string) (joinedGroupIDs []string, err error) { - return getCache( - ctx, - g.rcClient, - g.getJoinedGroupsKey(userID), - g.expireTime, - func(ctx context.Context) ([]string, error) { - return g.groupMemberDB.FindUserJoinedGroupID(ctx, userID) - }, - ) + return getCache(ctx, g.rcClient, g.getJoinedGroupsKey(userID), g.expireTime, func(ctx context.Context) ([]string, error) { + return g.groupMemberDB.FindUserJoinedGroupID(ctx, userID) + }) } func (g *GroupCacheRedis) DelJoinedGroupID(userIDs ...string) GroupCache { @@ -445,49 +391,28 @@ func (g *GroupCacheRedis) DelJoinedGroupID(userIDs ...string) GroupCache { return cache } -func (g *GroupCacheRedis) GetGroupMemberInfo( - ctx context.Context, - groupID, userID string, -) (groupMember *relationtb.GroupMemberModel, err error) { - return getCache( - ctx, - g.rcClient, - g.getGroupMemberInfoKey(groupID, userID), - g.expireTime, - func(ctx context.Context) (*relationtb.GroupMemberModel, error) { - return g.groupMemberDB.Take(ctx, groupID, userID) - }, - ) +func (g *GroupCacheRedis) GetGroupMemberInfo(ctx context.Context, groupID, userID string) (groupMember *relationtb.GroupMemberModel, err error) { + return getCache(ctx, g.rcClient, g.getGroupMemberInfoKey(groupID, userID), g.expireTime, func(ctx context.Context) (*relationtb.GroupMemberModel, error) { + return g.groupMemberDB.Take(ctx, groupID, userID) + }) } -func (g *GroupCacheRedis) GetGroupMembersInfo( - ctx context.Context, - groupID string, - userIDs []string, -) ([]*relationtb.GroupMemberModel, error) { - keys := make([]string, 0, len(userIDs)) - for _, userID := range userIDs { - keys = append(keys, g.getGroupMemberInfoKey(groupID, userID)) - } - - return batchGetCache( - ctx, - g.rcClient, - keys, - g.expireTime, - g.GetGroupMemberIndex, - func(ctx context.Context) ([]*relationtb.GroupMemberModel, error) { - return g.groupMemberDB.Find(ctx, []string{groupID}, userIDs, nil) - }, - ) +func (g *GroupCacheRedis) GetGroupMembersInfo(ctx context.Context, groupID string, userIDs []string) ([]*relationtb.GroupMemberModel, error) { + //var keys []string + //for _, userID := range userIDs { + // keys = append(keys, g.getGroupMemberInfoKey(groupID, userID)) + //} + //return batchGetCache(ctx, g.rcClient, keys, g.expireTime, g.GetGroupMemberIndex, func(ctx context.Context) ([]*relationtb.GroupMemberModel, error) { + // return g.groupMemberDB.Find(ctx, []string{groupID}, userIDs, nil) + //}) + return batchGetCache2(ctx, g.rcClient, g.expireTime, userIDs, func(userID string) string { + return g.getGroupMemberInfoKey(groupID, userID) + }, func(ctx context.Context, userID string) (*relationtb.GroupMemberModel, error) { + return g.groupMemberDB.Take(ctx, groupID, userID) + }) } -func (g *GroupCacheRedis) GetGroupMembersPage( - ctx context.Context, - groupID string, - userIDs []string, - showNumber, pageNumber int32, -) (total uint32, groupMembers []*relationtb.GroupMemberModel, err error) { +func (g *GroupCacheRedis) GetGroupMembersPage(ctx context.Context, groupID string, userIDs []string, showNumber, pageNumber int32) (total uint32, groupMembers []*relationtb.GroupMemberModel, err error) { groupMemberIDs, err := g.GetGroupMemberIDs(ctx, groupID) if err != nil { return 0, nil, err @@ -502,10 +427,7 @@ func (g *GroupCacheRedis) GetGroupMembersPage( return uint32(len(userIDs)), groupMembers, err } -func (g *GroupCacheRedis) GetAllGroupMembersInfo( - ctx context.Context, - groupID string, -) (groupMembers []*relationtb.GroupMemberModel, err error) { +func (g *GroupCacheRedis) GetAllGroupMembersInfo(ctx context.Context, groupID string) (groupMembers []*relationtb.GroupMemberModel, err error) { groupMemberIDs, err := g.GetGroupMemberIDs(ctx, groupID) if err != nil { return nil, err @@ -514,29 +436,19 @@ func (g *GroupCacheRedis) GetAllGroupMembersInfo( return g.GetGroupMembersInfo(ctx, groupID, groupMemberIDs) } -func (g *GroupCacheRedis) GetAllGroupMemberInfo( - ctx context.Context, - groupID string, -) ([]*relationtb.GroupMemberModel, error) { +func (g *GroupCacheRedis) GetAllGroupMemberInfo(ctx context.Context, groupID string) ([]*relationtb.GroupMemberModel, error) { groupMemberIDs, err := g.GetGroupMemberIDs(ctx, groupID) if err != nil { return nil, err } - keys := make([]string, 0, len(groupMemberIDs)) - for _, groupMemberID := range groupMemberIDs { - keys = append(keys, g.getGroupMemberInfoKey(groupID, groupMemberID)) - } - - return batchGetCache( - ctx, - g.rcClient, - keys, - g.expireTime, - g.GetGroupMemberIndex, - func(ctx context.Context) ([]*relationtb.GroupMemberModel, error) { - return g.groupMemberDB.Find(ctx, []string{groupID}, groupMemberIDs, nil) - }, - ) + //var keys []string + //for _, groupMemberID := range groupMemberIDs { + // keys = append(keys, g.getGroupMemberInfoKey(groupID, groupMemberID)) + //} + //return batchGetCache(ctx, g.rcClient, keys, g.expireTime, g.GetGroupMemberIndex, func(ctx context.Context) ([]*relationtb.GroupMemberModel, error) { + // return g.groupMemberDB.Find(ctx, []string{groupID}, groupMemberIDs, nil) + //}) + return g.GetGroupMembersInfo(ctx, groupID, groupMemberIDs) } func (g *GroupCacheRedis) DelGroupMembersInfo(groupID string, userIDs ...string) GroupCache { @@ -551,15 +463,9 @@ func (g *GroupCacheRedis) DelGroupMembersInfo(groupID string, userIDs ...string) } func (g *GroupCacheRedis) GetGroupMemberNum(ctx context.Context, groupID string) (memberNum int64, err error) { - return getCache( - ctx, - g.rcClient, - g.getGroupMemberNumKey(groupID), - g.expireTime, - func(ctx context.Context) (int64, error) { - return g.groupMemberDB.TakeGroupMemberNum(ctx, groupID) - }, - ) + return getCache(ctx, g.rcClient, g.getGroupMemberNumKey(groupID), g.expireTime, func(ctx context.Context) (int64, error) { + return g.groupMemberDB.TakeGroupMemberNum(ctx, groupID) + }) } func (g *GroupCacheRedis) DelGroupsMemberNum(groupID ...string) GroupCache { diff --git a/pkg/common/db/cache/meta_cache.go b/pkg/common/db/cache/meta_cache.go index 3d62255a7..5cff3df7f 100644 --- a/pkg/common/db/cache/meta_cache.go +++ b/pkg/common/db/cache/meta_cache.go @@ -18,7 +18,6 @@ import ( "context" "encoding/json" "errors" - "fmt" "time" "github.com/dtm-labs/rockscache" @@ -59,26 +58,37 @@ type metaCacheRedis struct { func (m *metaCacheRedis) ExecDel(ctx context.Context) error { if len(m.keys) > 0 { log.ZDebug(ctx, "delete cache", "keys", m.keys) - retryTimes := 0 - for { - if err := m.rcClient.TagAsDeletedBatch2(ctx, m.keys); err != nil { - if retryTimes >= m.maxRetryTimes { - err = errs.ErrInternalServer.Wrap( - fmt.Sprintf( - "delete cache error: %v, keys: %v, retry times %d, please check redis server", - err, - m.keys, - retryTimes, - ), - ) - log.ZWarn(ctx, "delete cache failed, please handle keys", err, "keys", m.keys) - - return err + for _, key := range m.keys { + for i := 0; i < m.maxRetryTimes; i++ { + if err := m.rcClient.TagAsDeleted(key); err != nil { + log.ZError(ctx, "delete cache failed", err, "key", key) + time.Sleep(m.retryInterval) + continue } - retryTimes++ - } else { break } + + //retryTimes := 0 + //for { + // m.rcClient.TagAsDeleted() + // if err := m.rcClient.TagAsDeletedBatch2(ctx, []string{key}); err != nil { + // if retryTimes >= m.maxRetryTimes { + // err = errs.ErrInternalServer.Wrap( + // fmt.Sprintf( + // "delete cache error: %v, keys: %v, retry times %d, please check redis server", + // err, + // key, + // retryTimes, + // ), + // ) + // log.ZWarn(ctx, "delete cache failed, please handle keys", err, "keys", key) + // return err + // } + // retryTimes++ + // } else { + // break + // } + //} } } @@ -109,13 +119,7 @@ func GetDefaultOpt() rockscache.Options { return opts } -func getCache[T any]( - ctx context.Context, - rcClient *rockscache.Client, - key string, - expire time.Duration, - fn func(ctx context.Context) (T, error), -) (T, error) { +func getCache[T any](ctx context.Context, rcClient *rockscache.Client, key string, expire time.Duration, fn func(ctx context.Context) (T, error)) (T, error) { var t T var write bool v, err := rcClient.Fetch2(ctx, key, expire, func() (s string, err error) { @@ -150,94 +154,101 @@ func getCache[T any]( return t, nil } -func batchGetCache[T any]( - ctx context.Context, - rcClient *rockscache.Client, - keys []string, - expire time.Duration, - keyIndexFn func(t T, keys []string) (int, error), - fn func(ctx context.Context) ([]T, error), -) ([]T, error) { - batchMap, err := rcClient.FetchBatch2(ctx, keys, expire, func(idxs []int) (m map[int]string, err error) { - values := make(map[int]string) - tArrays, err := fn(ctx) - if err != nil { - return nil, err - } - for _, v := range tArrays { - index, err := keyIndexFn(v, keys) - if err != nil { - continue - } - bs, err := json.Marshal(v) - if err != nil { - return nil, utils.Wrap(err, "marshal failed") - } - values[index] = string(bs) - } - - return values, nil - }) - if err != nil { - return nil, err - } - var tArrays []T - for _, v := range batchMap { - if v != "" { - var t T - err = json.Unmarshal([]byte(v), &t) - if err != nil { - return nil, utils.Wrap(err, "unmarshal failed") - } - tArrays = append(tArrays, t) - } +//func batchGetCache[T any](ctx context.Context, rcClient *rockscache.Client, keys []string, expire time.Duration, keyIndexFn func(t T, keys []string) (int, error), fn func(ctx context.Context) ([]T, error)) ([]T, error) { +// batchMap, err := rcClient.FetchBatch2(ctx, keys, expire, func(idxs []int) (m map[int]string, err error) { +// values := make(map[int]string) +// tArrays, err := fn(ctx) +// if err != nil { +// return nil, err +// } +// for _, v := range tArrays { +// index, err := keyIndexFn(v, keys) +// if err != nil { +// continue +// } +// bs, err := json.Marshal(v) +// if err != nil { +// return nil, utils.Wrap(err, "marshal failed") +// } +// values[index] = string(bs) +// } +// return values, nil +// }) +// if err != nil { +// return nil, err +// } +// var tArrays []T +// for _, v := range batchMap { +// if v != "" { +// var t T +// err = json.Unmarshal([]byte(v), &t) +// if err != nil { +// return nil, utils.Wrap(err, "unmarshal failed") +// } +// tArrays = append(tArrays, t) +// } +// } +// return tArrays, nil +//} + +func batchGetCache2[T any, K comparable](ctx context.Context, rcClient *rockscache.Client, expire time.Duration, keys []K, keyFn func(key K) string, fns func(ctx context.Context, key K) (T, error)) ([]T, error) { + if len(keys) == 0 { + return nil, nil } - - return tArrays, nil -} - -func batchGetCacheMap[T any]( - ctx context.Context, - rcClient *rockscache.Client, - keys, originKeys []string, - expire time.Duration, - keyIndexFn func(s string, keys []string) (int, error), - fn func(ctx context.Context) (map[string]T, error), -) (map[string]T, error) { - batchMap, err := rcClient.FetchBatch2(ctx, keys, expire, func(idxs []int) (m map[int]string, err error) { - tArrays, err := fn(ctx) + res := make([]T, 0, len(keys)) + for _, key := range keys { + val, err := getCache(ctx, rcClient, keyFn(key), expire, func(ctx context.Context) (T, error) { + return fns(ctx, key) + }) if err != nil { return nil, err } - values := make(map[int]string) - for k, v := range tArrays { - index, err := keyIndexFn(k, originKeys) - if err != nil { - continue - } - bs, err := json.Marshal(v) - if err != nil { - return nil, utils.Wrap(err, "marshal failed") - } - values[index] = string(bs) - } - - return values, nil - }) - if err != nil { - return nil, err - } - tMap := make(map[string]T) - for i, v := range batchMap { - if v != "" { - var t T - err = json.Unmarshal([]byte(v), &t) - if err != nil { - return nil, utils.Wrap(err, "unmarshal failed") - } - tMap[originKeys[i]] = t - } + res = append(res, val) } - return tMap, nil + return res, nil } + +//func batchGetCacheMap[T any]( +// ctx context.Context, +// rcClient *rockscache.Client, +// keys, originKeys []string, +// expire time.Duration, +// keyIndexFn func(s string, keys []string) (int, error), +// fn func(ctx context.Context) (map[string]T, error), +//) (map[string]T, error) { +// batchMap, err := rcClient.FetchBatch2(ctx, keys, expire, func(idxs []int) (m map[int]string, err error) { +// tArrays, err := fn(ctx) +// if err != nil { +// return nil, err +// } +// values := make(map[int]string) +// for k, v := range tArrays { +// index, err := keyIndexFn(k, originKeys) +// if err != nil { +// continue +// } +// bs, err := json.Marshal(v) +// if err != nil { +// return nil, utils.Wrap(err, "marshal failed") +// } +// values[index] = string(bs) +// } +// return values, nil +// }) +// if err != nil { +// return nil, err +// } +// tMap := make(map[string]T) +// for i, v := range batchMap { +// if v != "" { +// var t T +// err = json.Unmarshal([]byte(v), &t) +// if err != nil { +// return nil, utils.Wrap(err, "unmarshal failed") +// } +// tMap[originKeys[i]] = t +// } +// } +// return tMap, nil +//} diff --git a/pkg/common/db/cache/msg.go b/pkg/common/db/cache/msg.go index 66161c424..b55a76f62 100644 --- a/pkg/common/db/cache/msg.go +++ b/pkg/common/db/cache/msg.go @@ -17,6 +17,8 @@ package cache import ( "context" "errors" + "github.com/dtm-labs/rockscache" + unrelationtb "github.com/openimsdk/open-im-server/v3/pkg/common/db/table/unrelation" "strconv" "time" @@ -103,11 +105,7 @@ type MsgModel interface { GetTokensWithoutError(ctx context.Context, userID string, platformID int) (map[string]int, error) SetTokenMapByUidPid(ctx context.Context, userID string, platformID int, m map[string]int) error DeleteTokenByUidPid(ctx context.Context, userID string, platformID int, fields []string) error - GetMessagesBySeq( - ctx context.Context, - conversationID string, - seqs []int64, - ) (seqMsg []*sdkws.MsgData, failedSeqList []int64, err error) + GetMessagesBySeq(ctx context.Context, conversationID string, seqs []int64) (seqMsg []*sdkws.MsgData, failedSeqList []int64, err error) SetMessageToCache(ctx context.Context, conversationID string, msgs []*sdkws.MsgData) (int, error) UserDeleteMsgs(ctx context.Context, conversationID string, seqs []int64, userID string) error DelUserDeleteMsgsList(ctx context.Context, conversationID string, seqs []int64) @@ -120,12 +118,7 @@ type MsgModel interface { JudgeMessageReactionExist(ctx context.Context, clientMsgID string, sessionType int32) (bool, error) GetOneMessageAllReactionList(ctx context.Context, clientMsgID string, sessionType int32) (map[string]string, error) DeleteOneMessageKey(ctx context.Context, clientMsgID string, sessionType int32, subKey string) error - SetMessageReactionExpire( - ctx context.Context, - clientMsgID string, - sessionType int32, - expiration time.Duration, - ) (bool, error) + SetMessageReactionExpire(ctx context.Context, clientMsgID string, sessionType int32, expiration time.Duration) (bool, error) GetMessageTypeKeyValue(ctx context.Context, clientMsgID string, sessionType int32, typeKey string) (string, error) SetMessageTypeKeyValue(ctx context.Context, clientMsgID string, sessionType int32, typeKey, value string) error LockMessageTypeKey(ctx context.Context, clientMsgID string, TypeKey string) error @@ -138,10 +131,10 @@ func NewMsgCacheModel(client redis.UniversalClient) MsgModel { type msgCache struct { metaCache - rdb redis.UniversalClient - // expireTime time.Duration - // rcClient *rockscache.Client - // msgDocDatabase unrelationtb.MsgDocModelInterface + rdb redis.UniversalClient + expireTime time.Duration + rcClient *rockscache.Client + msgDocDatabase unrelationtb.MsgDocModelInterface } func (c *msgCache) getMaxSeqKey(conversationID string) string { @@ -156,53 +149,51 @@ func (c *msgCache) getHasReadSeqKey(conversationID string, userID string) string return hasReadSeq + userID + ":" + conversationID } -func (c *msgCache) setSeq( - ctx context.Context, - conversationID string, - seq int64, - getkey func(conversationID string) string, -) error { +func (c *msgCache) setSeq(ctx context.Context, conversationID string, seq int64, getkey func(conversationID string) string) error { return utils.Wrap1(c.rdb.Set(ctx, getkey(conversationID), seq, 0).Err()) } -func (c *msgCache) getSeq( - ctx context.Context, - conversationID string, - getkey func(conversationID string) string, -) (int64, error) { +func (c *msgCache) getSeq(ctx context.Context, conversationID string, getkey func(conversationID string) string) (int64, error) { return utils.Wrap2(c.rdb.Get(ctx, getkey(conversationID)).Int64()) } -func (c *msgCache) getSeqs( - ctx context.Context, - items []string, - getkey func(s string) string, -) (m map[string]int64, err error) { - pipe := c.rdb.Pipeline() - for _, v := range items { - err2 := pipe.Get(ctx, getkey(v)).Err() - if err2 != nil && !errors.Is(err2, redis.Nil) { - return nil, errs.Wrap(err2) - } - } - result, err := pipe.Exec(ctx) - if err != nil && !errors.Is(err, redis.Nil) { - return nil, errs.Wrap(err) - } +func (c *msgCache) getSeqs(ctx context.Context, items []string, getkey func(s string) string) (m map[string]int64, err error) { m = make(map[string]int64, len(items)) - for i, v := range result { - seq := v.(*redis.StringCmd) - - if seq.Err() != nil && !errors.Is(seq.Err(), redis.Nil) { - return nil, errs.Wrap(v.Err()) + for i, v := range items { + res, err := c.rdb.Get(ctx, getkey(v)).Result() + if err != nil && err != redis.Nil { + return nil, errs.Wrap(err) } - val := utils.StringToInt64(seq.Val()) + val := utils.StringToInt64(res) if val != 0 { m[items[i]] = val } } return m, nil + + //pipe := c.rdb.Pipeline() + //for _, v := range items { + // if err := pipe.Get(ctx, getkey(v)).Err(); err != nil && err != redis.Nil { + // return nil, errs.Wrap(err) + // } + //} + //result, err := pipe.Exec(ctx) + //if err != nil && err != redis.Nil { + // return nil, errs.Wrap(err) + //} + //m = make(map[string]int64, len(items)) + //for i, v := range result { + // seq := v.(*redis.StringCmd) + // if seq.Err() != nil && seq.Err() != redis.Nil { + // return nil, errs.Wrap(v.Err()) + // } + // val := utils.StringToInt64(seq.Val()) + // if val != 0 { + // m[items[i]] = val + // } + //} + //return m, nil } func (c *msgCache) SetMaxSeq(ctx context.Context, conversationID string, maxSeq int64) error { @@ -222,16 +213,21 @@ func (c *msgCache) SetMinSeq(ctx context.Context, conversationID string, minSeq } func (c *msgCache) setSeqs(ctx context.Context, seqs map[string]int64, getkey func(key string) string) error { - pipe := c.rdb.Pipeline() - for k, seq := range seqs { - err := pipe.Set(ctx, getkey(k), seq, 0).Err() - if err != nil { + for conversationID, seq := range seqs { + if err := c.rdb.Set(ctx, getkey(conversationID), seq, 0).Err(); err != nil { return errs.Wrap(err) } } - _, err := pipe.Exec(ctx) - - return err + return nil + //pipe := c.rdb.Pipeline() + //for k, seq := range seqs { + // err := pipe.Set(ctx, getkey(k), seq, 0).Err() + // if err != nil { + // return errs.Wrap(err) + // } + //} + //_, err := pipe.Exec(ctx) + //return err } func (c *msgCache) SetMinSeqs(ctx context.Context, seqs map[string]int64) error { @@ -254,30 +250,17 @@ func (c *msgCache) GetConversationUserMinSeq(ctx context.Context, conversationID return utils.Wrap2(c.rdb.Get(ctx, c.getConversationUserMinSeqKey(conversationID, userID)).Int64()) } -func (c *msgCache) GetConversationUserMinSeqs( - ctx context.Context, - conversationID string, - userIDs []string, -) (m map[string]int64, err error) { +func (c *msgCache) GetConversationUserMinSeqs(ctx context.Context, conversationID string, userIDs []string) (m map[string]int64, err error) { return c.getSeqs(ctx, userIDs, func(userID string) string { return c.getConversationUserMinSeqKey(conversationID, userID) }) } -func (c *msgCache) SetConversationUserMinSeq( - ctx context.Context, - conversationID string, - userID string, - minSeq int64, -) error { +func (c *msgCache) SetConversationUserMinSeq(ctx context.Context, conversationID string, userID string, minSeq int64) error { return utils.Wrap1(c.rdb.Set(ctx, c.getConversationUserMinSeqKey(conversationID, userID), minSeq, 0).Err()) } -func (c *msgCache) SetConversationUserMinSeqs( - ctx context.Context, - conversationID string, - seqs map[string]int64, -) (err error) { +func (c *msgCache) SetConversationUserMinSeqs(ctx context.Context, conversationID string, seqs map[string]int64) (err error) { return c.setSeqs(ctx, seqs, func(userID string) string { return c.getConversationUserMinSeqKey(conversationID, userID) }) @@ -305,11 +288,7 @@ func (c *msgCache) UserSetHasReadSeqs(ctx context.Context, userID string, hasRea }) } -func (c *msgCache) GetHasReadSeqs( - ctx context.Context, - userID string, - conversationIDs []string, -) (map[string]int64, error) { +func (c *msgCache) GetHasReadSeqs(ctx context.Context, userID string, conversationIDs []string) (map[string]int64, error) { return c.getSeqs(ctx, conversationIDs, func(conversationID string) string { return c.getHasReadSeqKey(conversationID, userID) }) @@ -363,62 +342,86 @@ func (c *msgCache) allMessageCacheKey(conversationID string) string { return messageCache + conversationID + "_*" } -func (c *msgCache) GetMessagesBySeq( - ctx context.Context, - conversationID string, - seqs []int64, -) (seqMsgs []*sdkws.MsgData, failedSeqs []int64, err error) { - pipe := c.rdb.Pipeline() - for _, v := range seqs { - // MESSAGE_CACHE:169.254.225.224_reliability1653387820_0_1 - key := c.getMessageCacheKey(conversationID, v) - err2 := pipe.Get(ctx, key).Err() - if err2 != nil && errors.Is(err2, redis.Nil) { - return nil, nil, err2 +func (c *msgCache) GetMessagesBySeq(ctx context.Context, conversationID string, seqs []int64) (seqMsgs []*sdkws.MsgData, failedSeqs []int64, err error) { + for _, seq := range seqs { + res, err := c.rdb.Get(ctx, c.getMessageCacheKey(conversationID, seq)).Result() + if err != nil { + log.ZError(ctx, "GetMessagesBySeq failed", err, "conversationID", conversationID, "seq", seq) + failedSeqs = append(failedSeqs, seq) + continue } - } - result, err := pipe.Exec(ctx) - for i, v := range result { - cmd := v.(*redis.StringCmd) - if cmd.Err() != nil { - failedSeqs = append(failedSeqs, seqs[i]) - } else { - msg := sdkws.MsgData{} - err = msgprocessor.String2Pb(cmd.Val(), &msg) - if err == nil { - if msg.Status != constant.MsgDeleted { - seqMsgs = append(seqMsgs, &msg) - - continue - } - } else { - log.ZWarn(ctx, "UnmarshalString failed", err, "conversationID", conversationID, "seq", seqs[i], "msg", cmd.Val()) - } - failedSeqs = append(failedSeqs, seqs[i]) + msg := sdkws.MsgData{} + if err = msgprocessor.String2Pb(res, &msg); err != nil { + log.ZError(ctx, "GetMessagesBySeq Unmarshal failed", err, "res", res, "conversationID", conversationID, "seq", seq) + failedSeqs = append(failedSeqs, seq) + continue } + if msg.Status == constant.MsgDeleted { + failedSeqs = append(failedSeqs, seq) + continue + } + seqMsgs = append(seqMsgs, &msg) } - return seqMsgs, failedSeqs, err + return + //pipe := c.rdb.Pipeline() + //for _, v := range seqs { + // // MESSAGE_CACHE:169.254.225.224_reliability1653387820_0_1 + // key := c.getMessageCacheKey(conversationID, v) + // if err := pipe.Get(ctx, key).Err(); err != nil && err != redis.Nil { + // return nil, nil, err + // } + //} + //result, err := pipe.Exec(ctx) + //for i, v := range result { + // cmd := v.(*redis.StringCmd) + // if cmd.Err() != nil { + // failedSeqs = append(failedSeqs, seqs[i]) + // } else { + // msg := sdkws.MsgData{} + // err = msgprocessor.String2Pb(cmd.Val(), &msg) + // if err == nil { + // if msg.Status != constant.MsgDeleted { + // seqMsgs = append(seqMsgs, &msg) + // continue + // } + // } else { + // log.ZWarn(ctx, "UnmarshalString failed", err, "conversationID", conversationID, "seq", seqs[i], "msg", cmd.Val()) + // } + // failedSeqs = append(failedSeqs, seqs[i]) + // } + //} + //return seqMsgs, failedSeqs, err } func (c *msgCache) SetMessageToCache(ctx context.Context, conversationID string, msgs []*sdkws.MsgData) (int, error) { - pipe := c.rdb.Pipeline() - var failedMsgs []*sdkws.MsgData for _, msg := range msgs { - key := c.getMessageCacheKey(conversationID, msg.Seq) s, err := msgprocessor.Pb2String(msg) if err != nil { return 0, errs.Wrap(err) } - err = pipe.Set(ctx, key, s, time.Duration(config.Config.MsgCacheTimeout)*time.Second).Err() - if err != nil { - failedMsgs = append(failedMsgs, msg) - log.ZWarn(ctx, "set msg 2 cache failed", err, "msg", failedMsgs) + key := c.getMessageCacheKey(conversationID, msg.Seq) + if err := c.rdb.Set(ctx, key, s, time.Duration(config.Config.MsgCacheTimeout)*time.Second).Err(); err != nil { + return 0, errs.Wrap(err) } } - _, err := pipe.Exec(ctx) - - return len(failedMsgs), err + return len(msgs), nil + //pipe := c.rdb.Pipeline() + //var failedMsgs []*sdkws.MsgData + //for _, msg := range msgs { + // key := c.getMessageCacheKey(conversationID, msg.Seq) + // s, err := msgprocessor.Pb2String(msg) + // if err != nil { + // return 0, errs.Wrap(err) + // } + // err = pipe.Set(ctx, key, s, time.Duration(config.Config.MsgCacheTimeout)*time.Second).Err() + // if err != nil { + // failedMsgs = append(failedMsgs, msg) + // log.ZWarn(ctx, "set msg 2 cache failed", err, "msg", failedMsgs) + // } + //} + //_, err := pipe.Exec(ctx) + //return len(failedMsgs), err } func (c *msgCache) getMessageDelUserListKey(conversationID string, seq int64) string { @@ -430,28 +433,47 @@ func (c *msgCache) getUserDelList(conversationID, userID string) string { } func (c *msgCache) UserDeleteMsgs(ctx context.Context, conversationID string, seqs []int64, userID string) error { - pipe := c.rdb.Pipeline() for _, seq := range seqs { delUserListKey := c.getMessageDelUserListKey(conversationID, seq) userDelListKey := c.getUserDelList(conversationID, userID) - err := pipe.SAdd(ctx, delUserListKey, userID).Err() + err := c.rdb.SAdd(ctx, delUserListKey, userID).Err() if err != nil { return errs.Wrap(err) } - err = pipe.SAdd(ctx, userDelListKey, seq).Err() + err = c.rdb.SAdd(ctx, userDelListKey, seq).Err() if err != nil { return errs.Wrap(err) } - if err := pipe.Expire(ctx, delUserListKey, time.Duration(config.Config.MsgCacheTimeout)*time.Second).Err(); err != nil { + if err := c.rdb.Expire(ctx, delUserListKey, time.Duration(config.Config.MsgCacheTimeout)*time.Second).Err(); err != nil { return errs.Wrap(err) } - if err := pipe.Expire(ctx, userDelListKey, time.Duration(config.Config.MsgCacheTimeout)*time.Second).Err(); err != nil { + if err := c.rdb.Expire(ctx, userDelListKey, time.Duration(config.Config.MsgCacheTimeout)*time.Second).Err(); err != nil { return errs.Wrap(err) } } - _, err := pipe.Exec(ctx) - return errs.Wrap(err) + return nil + //pipe := c.rdb.Pipeline() + //for _, seq := range seqs { + // delUserListKey := c.getMessageDelUserListKey(conversationID, seq) + // userDelListKey := c.getUserDelList(conversationID, userID) + // err := pipe.SAdd(ctx, delUserListKey, userID).Err() + // if err != nil { + // return errs.Wrap(err) + // } + // err = pipe.SAdd(ctx, userDelListKey, seq).Err() + // if err != nil { + // return errs.Wrap(err) + // } + // if err := pipe.Expire(ctx, delUserListKey, time.Duration(config.Config.MsgCacheTimeout)*time.Second).Err(); err != nil { + // return errs.Wrap(err) + // } + // if err := pipe.Expire(ctx, userDelListKey, time.Duration(config.Config.MsgCacheTimeout)*time.Second).Err(); err != nil { + // return errs.Wrap(err) + // } + //} + //_, err := pipe.Exec(ctx) + //return errs.Wrap(err) } func (c *msgCache) GetUserDelList(ctx context.Context, userID, conversationID string) (seqs []int64, err error) { @@ -476,47 +498,74 @@ func (c *msgCache) DelUserDeleteMsgsList(ctx context.Context, conversationID str continue } if len(delUsers) > 0 { - pipe := c.rdb.Pipeline() var failedFlag bool for _, userID := range delUsers { - err = pipe.SRem(ctx, c.getUserDelList(conversationID, userID), seq).Err() + err = c.rdb.SRem(ctx, c.getUserDelList(conversationID, userID), seq).Err() if err != nil { failedFlag = true - log.ZWarn( - ctx, - "DelUserDeleteMsgsList failed", - err, - "conversationID", - conversationID, - "seq", - seq, - "userID", - userID, - ) + log.ZWarn(ctx, "DelUserDeleteMsgsList failed", err, "conversationID", conversationID, "seq", seq, "userID", userID) } } if !failedFlag { - if err := pipe.Del(ctx, c.getMessageDelUserListKey(conversationID, seq)).Err(); err != nil { + if err := c.rdb.Del(ctx, c.getMessageDelUserListKey(conversationID, seq)).Err(); err != nil { log.ZWarn(ctx, "DelUserDeleteMsgsList failed", err, "conversationID", conversationID, "seq", seq) } } - if _, err := pipe.Exec(ctx); err != nil { - log.ZError(ctx, "pipe exec failed", err, "conversationID", conversationID, "seq", seq) - } } } + //for _, seq := range seqs { + // delUsers, err := c.rdb.SMembers(ctx, c.getMessageDelUserListKey(conversationID, seq)).Result() + // if err != nil { + // log.ZWarn(ctx, "DelUserDeleteMsgsList failed", err, "conversationID", conversationID, "seq", seq) + // continue + // } + // if len(delUsers) > 0 { + // pipe := c.rdb.Pipeline() + // var failedFlag bool + // for _, userID := range delUsers { + // err = pipe.SRem(ctx, c.getUserDelList(conversationID, userID), seq).Err() + // if err != nil { + // failedFlag = true + // log.ZWarn( + // ctx, + // "DelUserDeleteMsgsList failed", + // err, + // "conversationID", + // conversationID, + // "seq", + // seq, + // "userID", + // userID, + // ) + // } + // } + // if !failedFlag { + // if err := pipe.Del(ctx, c.getMessageDelUserListKey(conversationID, seq)).Err(); err != nil { + // log.ZWarn(ctx, "DelUserDeleteMsgsList failed", err, "conversationID", conversationID, "seq", seq) + // } + // } + // if _, err := pipe.Exec(ctx); err != nil { + // log.ZError(ctx, "pipe exec failed", err, "conversationID", conversationID, "seq", seq) + // } + // } + //} } func (c *msgCache) DeleteMessages(ctx context.Context, conversationID string, seqs []int64) error { - pipe := c.rdb.Pipeline() for _, seq := range seqs { - if err := pipe.Del(ctx, c.getMessageCacheKey(conversationID, seq)).Err(); err != nil { + if err := c.rdb.Del(ctx, c.getMessageCacheKey(conversationID, seq)).Err(); err != nil { return errs.Wrap(err) } } - _, err := pipe.Exec(ctx) - - return errs.Wrap(err) + return nil + //pipe := c.rdb.Pipeline() + //for _, seq := range seqs { + // if err := pipe.Del(ctx, c.getMessageCacheKey(conversationID, seq)).Err(); err != nil { + // return errs.Wrap(err) + // } + //} + //_, err := pipe.Exec(ctx) + //return errs.Wrap(err) } func (c *msgCache) CleanUpOneConversationAllMsg(ctx context.Context, conversationID string) error { @@ -527,16 +576,20 @@ func (c *msgCache) CleanUpOneConversationAllMsg(ctx context.Context, conversatio if err != nil { return errs.Wrap(err) } - pipe := c.rdb.Pipeline() for _, v := range vals { - err2 := pipe.Del(ctx, v).Err() - if err2 != nil { - return errs.Wrap(err2) + if err := c.rdb.Del(ctx, v).Err(); err != nil { + return errs.Wrap(err) } } - _, err = pipe.Exec(ctx) - - return errs.Wrap(err) + return nil + //pipe := c.rdb.Pipeline() + //for _, v := range vals { + // if err := pipe.Del(ctx, v).Err(); err != nil { + // return errs.Wrap(err) + // } + //} + //_, err = pipe.Exec(ctx) + //return errs.Wrap(err) } func (c *msgCache) DelMsgFromCache(ctx context.Context, userID string, seqs []int64) error { @@ -594,17 +647,8 @@ func (c *msgCache) GetSendMsgStatus(ctx context.Context, id string) (int32, erro return int32(result), errs.Wrap(err) } -func (c *msgCache) SetFcmToken( - ctx context.Context, - account string, - platformID int, - fcmToken string, - expireTime int64, -) (err error) { - return errs.Wrap( - c.rdb.Set(ctx, fcmToken+account+":"+strconv.Itoa(platformID), fcmToken, time.Duration(expireTime)*time.Second). - Err(), - ) +func (c *msgCache) SetFcmToken(ctx context.Context, account string, platformID int, fcmToken string, expireTime int64) (err error) { + return errs.Wrap(c.rdb.Set(ctx, fcmToken+account+":"+strconv.Itoa(platformID), fcmToken, time.Duration(expireTime)*time.Second).Err()) } func (c *msgCache) GetFcmToken(ctx context.Context, account string, platformID int) (string, error) { diff --git a/pkg/common/db/cache/user.go b/pkg/common/db/cache/user.go index 0afbd595e..d1164f2c0 100644 --- a/pkg/common/db/cache/user.go +++ b/pkg/common/db/cache/user.go @@ -112,29 +112,32 @@ func (u *UserCacheRedis) GetUserInfo(ctx context.Context, userID string) (userIn } func (u *UserCacheRedis) GetUsersInfo(ctx context.Context, userIDs []string) ([]*relationtb.UserModel, error) { - keys := make([]string, 0, len(userIDs)) - for _, userID := range userIDs { - keys = append(keys, u.getUserInfoKey(userID)) - } - - return batchGetCache( - ctx, - u.rcClient, - keys, - u.expireTime, - func(user *relationtb.UserModel, keys []string) (int, error) { - for i, key := range keys { - if key == u.getUserInfoKey(user.UserID) { - return i, nil - } - } - - return 0, errIndex - }, - func(ctx context.Context) ([]*relationtb.UserModel, error) { - return u.userDB.Find(ctx, userIDs) - }, - ) + //var keys []string + //for _, userID := range userIDs { + // keys = append(keys, u.getUserInfoKey(userID)) + //} + //return batchGetCache( + // ctx, + // u.rcClient, + // keys, + // u.expireTime, + // func(user *relationtb.UserModel, keys []string) (int, error) { + // for i, key := range keys { + // if key == u.getUserInfoKey(user.UserID) { + // return i, nil + // } + // } + // return 0, errIndex + // }, + // func(ctx context.Context) ([]*relationtb.UserModel, error) { + // return u.userDB.Find(ctx, userIDs) + // }, + //) + return batchGetCache2(ctx, u.rcClient, u.expireTime, userIDs, func(userID string) string { + return u.getUserInfoKey(userID) + }, func(ctx context.Context, userID string) (*relationtb.UserModel, error) { + return u.userDB.Take(ctx, userID) + }) } func (u *UserCacheRedis) DelUsersInfo(userIDs ...string) UserCache { diff --git a/pkg/common/db/controller/conversation.go b/pkg/common/db/controller/conversation.go index e68eb25ba..29cd0d152 100644 --- a/pkg/common/db/controller/conversation.go +++ b/pkg/common/db/controller/conversation.go @@ -50,7 +50,7 @@ type ConversationDatabase interface { GetConversationIDs(ctx context.Context, userID string) ([]string, error) GetUserConversationIDsHash(ctx context.Context, ownerUserID string) (hash uint64, err error) GetAllConversationIDs(ctx context.Context) ([]string, error) - GetUserAllHasReadSeqs(ctx context.Context, ownerUserID string) (map[string]int64, error) + //GetUserAllHasReadSeqs(ctx context.Context, ownerUserID string) (map[string]int64, error) GetConversationsByConversationID(ctx context.Context, conversationIDs []string) ([]*relationtb.ConversationModel, error) GetConversationIDsNeedDestruct(ctx context.Context) ([]*relationtb.ConversationModel, error) GetConversationNotReceiveMessageUserIDs(ctx context.Context, conversationID string) ([]string, error) @@ -306,9 +306,9 @@ func (c *conversationDatabase) GetAllConversationIDs(ctx context.Context) ([]str return c.conversationDB.GetAllConversationIDs(ctx) } -func (c *conversationDatabase) GetUserAllHasReadSeqs(ctx context.Context, ownerUserID string) (map[string]int64, error) { - return c.cache.GetUserAllHasReadSeqs(ctx, ownerUserID) -} +//func (c *conversationDatabase) GetUserAllHasReadSeqs(ctx context.Context, ownerUserID string) (map[string]int64, error) { +// return c.cache.GetUserAllHasReadSeqs(ctx, ownerUserID) +//} func (c *conversationDatabase) GetConversationsByConversationID(ctx context.Context, conversationIDs []string) ([]*relationtb.ConversationModel, error) { return c.conversationDB.GetConversationsByConversationID(ctx, conversationIDs) From e4f3e34249eace95af2290148e7e626c455b3b81 Mon Sep 17 00:00:00 2001 From: Gordon <46924906+FGadvancer@users.noreply.github.com> Date: Mon, 23 Oct 2023 21:26:35 +0800 Subject: [PATCH 2/3] fix: api send messages for notification conversation. (#1254) * fix: to start im or chat, ZooKeeper must be started first. * fix: msg gateway start output err info Signed-off-by: Gordon <1432970085@qq.com> * fix: msg gateway start output err info Signed-off-by: Gordon <1432970085@qq.com> * chore: package path changes Signed-off-by: withchao <993506633@qq.com> * fix: go mod update Signed-off-by: Gordon <1432970085@qq.com> * fix: token update Signed-off-by: Gordon <1432970085@qq.com> * chore: package path changes Signed-off-by: withchao <993506633@qq.com> * chore: package path changes Signed-off-by: withchao <993506633@qq.com> * fix: token update Signed-off-by: Gordon <1432970085@qq.com> * fix: token update Signed-off-by: Gordon <1432970085@qq.com> * fix: token update Signed-off-by: Gordon <1432970085@qq.com> * fix: token update Signed-off-by: Gordon <1432970085@qq.com> * fix: token update Signed-off-by: Gordon <1432970085@qq.com> * fix: token update Signed-off-by: Gordon <1432970085@qq.com> * fix: get all userID Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: msggateway add online status call Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * refactor: log change Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * refactor: log change Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * chore: network mode change Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * cicd: robot automated Change Signed-off-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com> * cicd: robot automated Change Signed-off-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com> * cicd: robot automated Change Signed-off-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com> * feat: add api of get server time Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * feat: remove go work sum Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * cicd: robot automated Change Signed-off-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com> * fix: pull message add isRead field Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: check msg-transfer script Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: script update Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: script update Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: script update Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: script update Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: script update Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: script update Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: script update Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: script update Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: script update Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: script update Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: script update Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: script update Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: script update Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: script update Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: script update Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: script update Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: script update Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: start don't kill old process Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * cicd: robot automated Change Signed-off-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com> * fix: check component Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: pull message set isRead only message come from single. Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * cicd: robot automated Change Signed-off-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com> * fix: multiple gateway kick user each other. Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: multiple gateway kick user each other. Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: multiple gateway kick user each other. Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: multiple gateway kick user each other. Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: multiple gateway kick user each other. Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: multiple gateway kick user each other. Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * cicd: robot automated Change Signed-off-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com> * fix: multiple gateway kick user each other. Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: add ex field to update group info. Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * cicd: robot automated Change * cicd: robot automated Change * refactor: change project module name. Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * refactor: change project module name. Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * refactor: change project module name. Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * cicd: robot automated Change * test: for pressure test. Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * test: for pressure test. Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * test: for pressure test. Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * test: message log. Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * cicd: robot automated Change * fxi: component check output valid info. Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fxi: component check output valid info. Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * test: send message test log. Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * cicd: robot automated Change * cicd: robot automated Change * test: remove info log. Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * feat: api of send message add sendTime field. Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: add callback for update user's info. * cicd: robot automated Change * fix: change callback command name. * cicd: robot automated Change * fix: single chat unread status change. * fix: single chat unread status change. * fix: single chat unread status change. * fix: user status change. * cicd: robot automated Change * fix: user status change. * fix: user status change. * fix: user status change. * cicd: robot automated Change * fix: ws close when user logout. * fix: remove repeat platform on online status. * cicd: robot automated Change * fix: api send messages for notification conversation . * fix: api send messages for notification conversation . * fix: api send messages for notification conversation . * fix: api send messages for notification conversation . * fix: api send messages for notification conversation . * fix: api send messages for notification conversation. * fix: api send messages for notification conversation. * fix: api send messages for notification conversation. * fix: api send messages for notification conversation. * fix: api send messages for notification conversation. * fix: api send messages for notification conversation. * re: remove router of unsubscribeStatus. * re: remove router of unsubscribeStatus. * re: remove router of unsubscribeStatus. * re: remove router of unsubscribeStatus. --------- Signed-off-by: Gordon <1432970085@qq.com> Signed-off-by: withchao <993506633@qq.com> Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> Signed-off-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com> Co-authored-by: withchao <993506633@qq.com> Co-authored-by: Xinwei Xiong <3293172751NSS@gmail.com> Co-authored-by: FGadvancer --- .golangci.yml | 2 +- cmd/openim-api/main.go | 1 - internal/api/msg.go | 32 +++------ internal/api/route.go | 1 - internal/api/user.go | 5 -- internal/msgtransfer/init.go | 4 +- .../msgtransfer/online_history_msg_handler.go | 26 ++++--- .../online_msg_to_mongo_handler.go | 2 +- internal/push/push_handler.go | 1 + internal/push/push_to_client.go | 9 ++- internal/rpc/conversation/conversaion.go | 72 ++++++++++++------- internal/rpc/friend/friend.go | 3 +- internal/rpc/msg/as_read.go | 5 +- internal/tools/msg.go | 6 +- pkg/apistruct/manage.go | 2 +- pkg/rpcclient/conversation.go | 13 ++-- pkg/rpcclient/msg.go | 11 +-- 17 files changed, 104 insertions(+), 91 deletions(-) diff --git a/.golangci.yml b/.golangci.yml index 050025b6e..8785b72d8 100644 --- a/.golangci.yml +++ b/.golangci.yml @@ -171,7 +171,7 @@ linters-settings: # exclude_godoc_examples: false funlen: lines: 150 - statements: 50 + statements: 80 gci: # put imports beginning with prefix after 3rd-party packages; # only support one prefix diff --git a/cmd/openim-api/main.go b/cmd/openim-api/main.go index f6db87353..d1f5cb3f8 100644 --- a/cmd/openim-api/main.go +++ b/cmd/openim-api/main.go @@ -32,7 +32,6 @@ import ( "github.com/openimsdk/open-im-server/v3/pkg/common/db/cache" ) - func main() { apiCmd := cmd.NewApiCmd() apiCmd.AddPortFlag() diff --git a/internal/api/msg.go b/internal/api/msg.go index b9b10e98e..028d77ca4 100644 --- a/internal/api/msg.go +++ b/internal/api/msg.go @@ -57,6 +57,10 @@ func (m MessageApi) newUserSendMsgReq(c *gin.Context, params *apistruct.SendMsg) var newContent string options := make(map[string]bool, 5) switch params.ContentType { + case constant.OANotification: + notification := sdkws.NotificationElem{} + notification.Detail = utils.StructToJsonString(params.Content) + newContent = utils.StructToJsonString(¬ification) case constant.Text: fallthrough case constant.Picture: @@ -69,10 +73,6 @@ func (m MessageApi) newUserSendMsgReq(c *gin.Context, params *apistruct.SendMsg) fallthrough case constant.File: fallthrough - case constant.CustomNotTriggerConversation: - fallthrough - case constant.CustomOnlineOnly: - fallthrough default: newContent = utils.StructToJsonString(params.Content) } @@ -82,11 +82,6 @@ func (m MessageApi) newUserSendMsgReq(c *gin.Context, params *apistruct.SendMsg) if params.NotOfflinePush { utils.SetSwitchFromOptions(options, constant.IsOfflinePush, false) } - if params.ContentType == constant.CustomOnlineOnly { - m.SetOptions(options, false) - } else if params.ContentType == constant.CustomNotTriggerConversation { - utils.SetSwitchFromOptions(options, constant.IsConversationUpdate, false) - } pbData := msg.SendMsgReq{ MsgData: &sdkws.MsgData{ SendID: params.SendID, @@ -105,14 +100,6 @@ func (m MessageApi) newUserSendMsgReq(c *gin.Context, params *apistruct.SendMsg) OfflinePushInfo: params.OfflinePushInfo, }, } - //if params.ContentType == constant.OANotification { - // var tips sdkws.TipsComm - // tips.JsonDetail = utils.StructToJsonString(params.Content) - // pbData.MsgData.Content, err = proto.Marshal(&tips) - // if err != nil { - // log.ZError(c, "Marshal failed ", err, "tips", tips.String()) - // } - //} return &pbData } @@ -180,15 +167,13 @@ func (m *MessageApi) getSendMsgReq(c *gin.Context, req apistruct.SendMsg) (sendM data = apistruct.FileElem{} case constant.Custom: data = apistruct.CustomElem{} - case constant.Revoke: - data = apistruct.RevokeElem{} case constant.OANotification: data = apistruct.OANotificationElem{} req.SessionType = constant.NotificationChatType - case constant.CustomNotTriggerConversation: - data = apistruct.CustomElem{} - case constant.CustomOnlineOnly: - data = apistruct.CustomElem{} + if !authverify.IsManagerUserID(req.SendID) { + return nil, errs.ErrNoPermission. + Wrap("only app manager can as sender send OANotificationElem") + } default: return nil, errs.ErrArgs.WithDetail("not support err contentType") } @@ -212,7 +197,6 @@ func (m *MessageApi) SendMessage(c *gin.Context) { apiresp.GinError(c, errs.ErrNoPermission.Wrap("only app manager can send message")) return } - sendMsgReq, err := m.getSendMsgReq(c, req.SendMsg) if err != nil { log.ZError(c, "decodeData failed", err) diff --git a/internal/api/route.go b/internal/api/route.go index 9a639a2de..d714270b4 100644 --- a/internal/api/route.go +++ b/internal/api/route.go @@ -83,7 +83,6 @@ func NewGinRouter(discov discoveryregistry.SvcDiscoveryRegistry, rdb redis.Unive userRouterGroup.POST("/get_users_online_status", ParseToken, u.GetUsersOnlineStatus) userRouterGroup.POST("/get_users_online_token_detail", ParseToken, u.GetUsersOnlineTokenDetail) userRouterGroup.POST("/subscribe_users_status", ParseToken, u.SubscriberStatus) - userRouterGroup.POST("/unsubscribe_users_status", ParseToken, u.UnSubscriberStatus) userRouterGroup.POST("/get_users_status", ParseToken, u.GetUserStatus) userRouterGroup.POST("/get_subscribe_users_status", ParseToken, u.GetSubscribeUsersStatus) } diff --git a/internal/api/user.go b/internal/api/user.go index 0a5a12698..86b7c0b0b 100644 --- a/internal/api/user.go +++ b/internal/api/user.go @@ -190,11 +190,6 @@ func (u *UserApi) SubscriberStatus(c *gin.Context) { a2r.Call(user.UserClient.SubscribeOrCancelUsersStatus, u.Client, c) } -// UnSubscriberStatus Unsubscribe a user's presence. -func (u *UserApi) UnSubscriberStatus(c *gin.Context) { - a2r.Call(user.UserClient.SubscribeOrCancelUsersStatus, u.Client, c) -} - // GetUserStatus Get the online status of the user. func (u *UserApi) GetUserStatus(c *gin.Context) { a2r.Call(user.UserClient.GetUserStatus, u.Client, c) diff --git a/internal/msgtransfer/init.go b/internal/msgtransfer/init.go index 7babd9a07..e5066633a 100644 --- a/internal/msgtransfer/init.go +++ b/internal/msgtransfer/init.go @@ -16,12 +16,13 @@ package msgtransfer import ( "fmt" - "sync" "google.golang.org/grpc" "google.golang.org/grpc/credentials/insecure" + "github.com/openimsdk/open-im-server/v3/pkg/common/discovery_register" + "github.com/OpenIMSDK/tools/mw" "github.com/openimsdk/open-im-server/v3/pkg/common/config" @@ -30,7 +31,6 @@ import ( "github.com/openimsdk/open-im-server/v3/pkg/common/db/relation" relationtb "github.com/openimsdk/open-im-server/v3/pkg/common/db/table/relation" "github.com/openimsdk/open-im-server/v3/pkg/common/db/unrelation" - "github.com/openimsdk/open-im-server/v3/pkg/common/discovery_register" "github.com/openimsdk/open-im-server/v3/pkg/common/prome" "github.com/openimsdk/open-im-server/v3/pkg/rpcclient" ) diff --git a/internal/msgtransfer/online_history_msg_handler.go b/internal/msgtransfer/online_history_msg_handler.go index e83939a4c..b4556634c 100644 --- a/internal/msgtransfer/online_history_msg_handler.go +++ b/internal/msgtransfer/online_history_msg_handler.go @@ -283,20 +283,30 @@ func (och *OnlineHistoryRedisConsumerHandler) handleMsg( return } if isNewConversation { - if storageList[0].SessionType == constant.SuperGroupChatType { - log.ZInfo(ctx, "group chat first create conversation", "conversationID", conversationID) + switch storageList[0].SessionType { + case constant.SuperGroupChatType: + log.ZInfo(ctx, "group chat first create conversation", "conversationID", + conversationID) userIDs, err := och.groupRpcClient.GetGroupMemberIDs(ctx, storageList[0].GroupID) if err != nil { - log.ZWarn(ctx, "get group member ids error", err, "conversationID", conversationID) + log.ZWarn(ctx, "get group member ids error", err, "conversationID", + conversationID) } else { - if err := och.conversationRpcClient.GroupChatFirstCreateConversation(ctx, storageList[0].GroupID, userIDs); err != nil { - log.ZWarn(ctx, "single chat first create conversation error", err, "conversationID", conversationID) + if err := och.conversationRpcClient.GroupChatFirstCreateConversation(ctx, + storageList[0].GroupID, userIDs); err != nil { + log.ZWarn(ctx, "single chat first create conversation error", err, + "conversationID", conversationID) } } - } else { - if err := och.conversationRpcClient.SingleChatFirstCreateConversation(ctx, storageList[0].RecvID, storageList[0].SendID); err != nil { - log.ZWarn(ctx, "single chat first create conversation error", err, "conversationID", conversationID) + case constant.SingleChatType, constant.NotificationChatType: + if err := och.conversationRpcClient.SingleChatFirstCreateConversation(ctx, storageList[0].RecvID, + storageList[0].SendID, conversationID, storageList[0].SessionType); err != nil { + log.ZWarn(ctx, "single chat or notification first create conversation error", err, + "conversationID", conversationID, "sessionType", storageList[0].SessionType) } + default: + log.ZWarn(ctx, "unknown session type", nil, "sessionType", + storageList[0].SessionType) } } diff --git a/internal/msgtransfer/online_msg_to_mongo_handler.go b/internal/msgtransfer/online_msg_to_mongo_handler.go index 8099d39d7..bfea6c433 100644 --- a/internal/msgtransfer/online_msg_to_mongo_handler.go +++ b/internal/msgtransfer/online_msg_to_mongo_handler.go @@ -62,7 +62,7 @@ func (mc *OnlineHistoryMongoConsumerHandler) handleChatWs2Mongo( log.ZError(ctx, "msgFromMQ.MsgData is empty", nil, "cMsg", cMsg) return } - log.ZInfo(ctx, "mongo consumer recv msg", "msgs", msgFromMQ.MsgData) + log.ZInfo(ctx, "mongo consumer recv msg", "msgs", msgFromMQ.String()) err = mc.msgDatabase.BatchInsertChat2DB(ctx, msgFromMQ.ConversationID, msgFromMQ.MsgData, msgFromMQ.LastSeq) if err != nil { log.ZError( diff --git a/internal/push/push_handler.go b/internal/push/push_handler.go index 91363282f..a1a9ff08e 100644 --- a/internal/push/push_handler.go +++ b/internal/push/push_handler.go @@ -58,6 +58,7 @@ func (c *ConsumerHandler) handleMs2PsChat(ctx context.Context, msg []byte) { } sec := msgFromMQ.MsgData.SendTime / 1000 nowSec := utils.GetCurrentTimestampBySecond() + log.ZDebug(ctx, "push msg", "msg", pbData.String(), "sec", sec, "nowSec", nowSec) if nowSec-sec > 10 { return } diff --git a/internal/push/push_to_client.go b/internal/push/push_to_client.go index 66b003eaa..ba0d65b39 100644 --- a/internal/push/push_to_client.go +++ b/internal/push/push_to_client.go @@ -126,13 +126,12 @@ func (p *Pusher) Push2User(ctx context.Context, userIDs []string, msg *sdkws.Msg } func (p *Pusher) UnmarshalNotificationElem(bytes []byte, t interface{}) error { - var notificationElem struct { - Detail string `json:"detail,omitempty"` - } - if err := json.Unmarshal(bytes, ¬ificationElem); err != nil { + var notification sdkws.NotificationElem + if err := json.Unmarshal(bytes, ¬ification); err != nil { return err } - return json.Unmarshal([]byte(notificationElem.Detail), t) + + return json.Unmarshal([]byte(notification.Detail), t) } func (p *Pusher) Push2SuperGroup(ctx context.Context, groupID string, msg *sdkws.MsgData) (err error) { diff --git a/internal/rpc/conversation/conversaion.go b/internal/rpc/conversation/conversaion.go index e76f83008..0ea7d54be 100644 --- a/internal/rpc/conversation/conversaion.go +++ b/internal/rpc/conversation/conversaion.go @@ -17,8 +17,6 @@ package conversation import ( "context" - "github.com/openimsdk/open-im-server/v3/pkg/msgprocessor" - "google.golang.org/grpc" "github.com/OpenIMSDK/protocol/constant" @@ -114,7 +112,10 @@ func (c *conversationServer) SetConversation(ctx context.Context, req *pbconvers return resp, nil } -func (c *conversationServer) SetConversations(ctx context.Context, req *pbconversation.SetConversationsReq) (*pbconversation.SetConversationsResp, error) { +//nolint +func (c *conversationServer) SetConversations(ctx context.Context, + req *pbconversation.SetConversationsReq, +) (*pbconversation.SetConversationsResp, error) { if req.Conversation == nil { return nil, errs.ErrArgs.Wrap("conversation must not be nil") } @@ -124,14 +125,8 @@ func (c *conversationServer) SetConversations(ctx context.Context, req *pbconver return nil, err } if groupInfo.Status == constant.GroupStatusDismissed { - return nil, err + return nil, errs.ErrDismissedAlready.Wrap("group dismissed") } - // for _, userID := range req.UserIDs { - // if _, err := c.groupRpcClient.GetGroupMemberCache(ctx, req.Conversation.GroupID, userID); err != nil { - // log.ZError(ctx, "user not in group", err, "userID", userID, "groupID", req.Conversation.GroupID) - // return nil, err - // } - // } } var unequal int var conv tablerelation.ConversationModel @@ -205,7 +200,14 @@ func (c *conversationServer) SetConversations(ctx context.Context, req *pbconver return nil, err } for _, userID := range req.UserIDs { - c.conversationNotificationSender.ConversationSetPrivateNotification(ctx, userID, req.Conversation.UserID, req.Conversation.IsPrivateChat.Value, req.Conversation.ConversationID) + err := c.conversationNotificationSender.ConversationSetPrivateNotification(ctx, userID, req.Conversation.UserID, + req.Conversation.IsPrivateChat.Value, req.Conversation.ConversationID) + if err != nil { + log.ZWarn(ctx, "send conversation set private notification failed", err, + "userID", userID, "conversationID", req.Conversation.ConversationID) + + continue + } } } if req.Conversation.BurnDuration != nil { @@ -235,24 +237,40 @@ func (c *conversationServer) GetRecvMsgNotNotifyUserIDs(ctx context.Context, req } // create conversation without notification for msg redis transfer. -func (c *conversationServer) CreateSingleChatConversations(ctx context.Context, req *pbconversation.CreateSingleChatConversationsReq) (*pbconversation.CreateSingleChatConversationsResp, error) { - var conversation tablerelation.ConversationModel - conversation.ConversationID = msgprocessor.GetConversationIDBySessionType(constant.SingleChatType, req.RecvID, req.SendID) - conversation.ConversationType = constant.SingleChatType - conversation.OwnerUserID = req.SendID - conversation.UserID = req.RecvID - err := c.conversationDatabase.CreateConversation(ctx, []*tablerelation.ConversationModel{&conversation}) - if err != nil { - log.ZWarn(ctx, "create conversation failed", err, "conversation", conversation) - } +func (c *conversationServer) CreateSingleChatConversations(ctx context.Context, + req *pbconversation.CreateSingleChatConversationsReq, +) (*pbconversation.CreateSingleChatConversationsResp, error) { + switch req.ConversationType { + case constant.SingleChatType: + var conversation tablerelation.ConversationModel + conversation.ConversationID = req.ConversationID + conversation.ConversationType = req.ConversationType + conversation.OwnerUserID = req.SendID + conversation.UserID = req.RecvID + err := c.conversationDatabase.CreateConversation(ctx, []*tablerelation.ConversationModel{&conversation}) + if err != nil { + log.ZWarn(ctx, "create conversation failed", err, "conversation", conversation) + } - conversation2 := conversation - conversation2.OwnerUserID = req.RecvID - conversation2.UserID = req.SendID - err = c.conversationDatabase.CreateConversation(ctx, []*tablerelation.ConversationModel{&conversation2}) - if err != nil { - log.ZWarn(ctx, "create conversation failed", err, "conversation2", conversation) + conversation2 := conversation + conversation2.OwnerUserID = req.RecvID + conversation2.UserID = req.SendID + err = c.conversationDatabase.CreateConversation(ctx, []*tablerelation.ConversationModel{&conversation2}) + if err != nil { + log.ZWarn(ctx, "create conversation failed", err, "conversation2", conversation) + } + case constant.NotificationChatType: + var conversation tablerelation.ConversationModel + conversation.ConversationID = req.ConversationID + conversation.ConversationType = req.ConversationType + conversation.OwnerUserID = req.RecvID + conversation.UserID = req.SendID + err := c.conversationDatabase.CreateConversation(ctx, []*tablerelation.ConversationModel{&conversation}) + if err != nil { + log.ZWarn(ctx, "create conversation failed", err, "conversation2", conversation) + } } + return &pbconversation.CreateSingleChatConversationsResp{}, nil } diff --git a/internal/rpc/friend/friend.go b/internal/rpc/friend/friend.go index 1524a7f27..c563f77fe 100644 --- a/internal/rpc/friend/friend.go +++ b/internal/rpc/friend/friend.go @@ -252,7 +252,8 @@ func (s *friendServer) GetDesignatedFriends( return resp, nil } -func (s *friendServer) GetDesignatedFriendsApply(ctx context.Context, req *pbfriend.GetDesignatedFriendsApplyReq) (resp *pbfriend.GetDesignatedFriendsApplyResp, err error) { +func (s *friendServer) GetDesignatedFriendsApply(ctx context.Context, + req *pbfriend.GetDesignatedFriendsApplyReq) (resp *pbfriend.GetDesignatedFriendsApplyResp, err error) { friendRequests, err := s.friendDatabase.FindBothFriendRequests(ctx, req.FromUserID, req.ToUserID) if err != nil { return nil, err diff --git a/internal/rpc/msg/as_read.go b/internal/rpc/msg/as_read.go index efa322774..55c3ba088 100644 --- a/internal/rpc/msg/as_read.go +++ b/internal/rpc/msg/as_read.go @@ -147,7 +147,6 @@ func (m *msgServer) MarkConversationAsRead( for i := hasReadSeq + 1; i <= req.HasReadSeq; i++ { seqs = append(seqs, i) } - if len(seqs) > 0 { log.ZDebug(ctx, "MarkConversationAsRead", "seqs", seqs, "conversationID", req.ConversationID) if err = m.MsgDatabase.MarkSingleChatMsgsAsRead(ctx, req.UserID, req.ConversationID, seqs); err != nil { @@ -165,7 +164,9 @@ func (m *msgServer) MarkConversationAsRead( m.conversationAndGetRecvID(conversation, req.UserID), seqs, hasReadSeq); err != nil { return nil, err } - } else if conversation.ConversationType == constant.SuperGroupChatType { + + } else if conversation.ConversationType == constant.SuperGroupChatType || + conversation.ConversationType == constant.NotificationChatType { if req.HasReadSeq > hasReadSeq { err = m.MsgDatabase.SetHasReadSeq(ctx, req.UserID, req.ConversationID, req.HasReadSeq) if err != nil { diff --git a/internal/tools/msg.go b/internal/tools/msg.go index 94ce2dec0..ca095051c 100644 --- a/internal/tools/msg.go +++ b/internal/tools/msg.go @@ -23,6 +23,8 @@ import ( "google.golang.org/grpc" "google.golang.org/grpc/credentials/insecure" + "github.com/openimsdk/open-im-server/v3/pkg/common/discovery_register" + "github.com/OpenIMSDK/tools/errs" "github.com/OpenIMSDK/tools/log" "github.com/OpenIMSDK/tools/mcontext" @@ -35,7 +37,6 @@ import ( "github.com/openimsdk/open-im-server/v3/pkg/common/db/controller" "github.com/openimsdk/open-im-server/v3/pkg/common/db/relation" "github.com/openimsdk/open-im-server/v3/pkg/common/db/unrelation" - "github.com/openimsdk/open-im-server/v3/pkg/common/discovery_register" "github.com/openimsdk/open-im-server/v3/pkg/rpcclient" "github.com/openimsdk/open-im-server/v3/pkg/rpcclient/notification" ) @@ -77,7 +78,8 @@ func InitMsgTool() (*MsgTool, error) { /* discov, err := zookeeper.NewClient(config.Config.Zookeeper.ZkAddr, config.Config.Zookeeper.Schema, zookeeper.WithFreq(time.Hour), zookeeper.WithRoundRobin(), zookeeper.WithUserNameAndPassword(config.Config.Zookeeper.Username, - config.Config.Zookeeper.Password), zookeeper.WithTimeout(10), zookeeper.WithLogger(log.NewZkLogger()))*/if err != nil { + config.Config.Zookeeper.Password), zookeeper.WithTimeout(10), zookeeper.WithLogger(log.NewZkLogger()))*/ + if err != nil { return nil, err } discov.AddOption(mw.GrpcClient(), grpc.WithTransportCredentials(insecure.NewCredentials())) diff --git a/pkg/apistruct/manage.go b/pkg/apistruct/manage.go index 7d30d2151..1238b4757 100644 --- a/pkg/apistruct/manage.go +++ b/pkg/apistruct/manage.go @@ -41,7 +41,7 @@ type SendMsgReq struct { type BatchSendMsgReq struct { SendMsg IsSendAll bool `json:"isSendAll"` - RecvIDs []string `json:"recvIDs"` + RecvIDs []string `json:"recvIDs" binding:"required"` } type BatchSendMsgResp struct { diff --git a/pkg/rpcclient/conversation.go b/pkg/rpcclient/conversation.go index 30b0b4b77..df01bcb8f 100644 --- a/pkg/rpcclient/conversation.go +++ b/pkg/rpcclient/conversation.go @@ -39,7 +39,6 @@ func NewConversation(discov discoveryregistry.SvcDiscoveryRegistry) *Conversatio panic(err) } client := pbconversation.NewConversationClient(conn) - return &Conversation{discov: discov, conn: conn, Client: client} } @@ -57,13 +56,17 @@ func (c *ConversationRpcClient) GetSingleConversationRecvMsgOpt(ctx context.Cont if err != nil { return 0, err } - return conversation.GetConversation().RecvMsgOpt, err } -func (c *ConversationRpcClient) SingleChatFirstCreateConversation(ctx context.Context, recvID, sendID string) error { - _, err := c.Client.CreateSingleChatConversations(ctx, &pbconversation.CreateSingleChatConversationsReq{RecvID: recvID, SendID: sendID}) - +func (c *ConversationRpcClient) SingleChatFirstCreateConversation(ctx context.Context, recvID, sendID, + conversationID string, conversationType int32, +) error { + _, err := c.Client.CreateSingleChatConversations(ctx, + &pbconversation.CreateSingleChatConversationsReq{ + RecvID: recvID, SendID: sendID, ConversationID: conversationID, + ConversationType: conversationType, + }) return err } diff --git a/pkg/rpcclient/msg.go b/pkg/rpcclient/msg.go index 51e29c7d8..00b0fa3f1 100644 --- a/pkg/rpcclient/msg.go +++ b/pkg/rpcclient/msg.go @@ -221,12 +221,13 @@ func WithRpcGetUserName() NotificationOptions { } } -func (s *NotificationSender) NotificationWithSesstionType(ctx context.Context, sendID, recvID string, contentType, sesstionType int32, m proto.Message, opts ...NotificationOptions) (err error) { +func (s *NotificationSender) NotificationWithSesstionType(ctx context.Context, sendID, recvID string, + contentType, sesstionType int32, m proto.Message, opts ...NotificationOptions) (err error) { n := sdkws.NotificationElem{Detail: utils.StructToJsonString(m)} content, err := json.Marshal(&n) if err != nil { - log.ZError(ctx, "MsgClient Notification json.Marshal failed", err, "sendID", sendID, "recvID", recvID, "contentType", contentType, "msg", m) - + log.ZError(ctx, "MsgClient Notification json.Marshal failed", err, "sendID", + sendID, "recvID", recvID, "contentType", contentType, "msg", m) return err } notificationOpt := ¬ificationOpt{} @@ -235,8 +236,8 @@ func (s *NotificationSender) NotificationWithSesstionType(ctx context.Context, s } var req msg.SendMsgReq var msg sdkws.MsgData + var userInfo *sdkws.UserInfo if notificationOpt.WithRpcGetUsername && s.getUserInfo != nil { - var userInfo *sdkws.UserInfo userInfo, err = s.getUserInfo(ctx, sendID) if err != nil { log.ZWarn(ctx, "getUserInfo failed", err, "sendID", sendID) @@ -259,7 +260,7 @@ func (s *NotificationSender) NotificationWithSesstionType(ctx context.Context, s msg.CreateTime = utils.GetCurrentTimestampByMill() msg.ClientMsgID = utils.GetMsgID(sendID) optionsConfig := s.contentTypeConf[contentType] - if sesstionType == constant.SuperGroupChatType && contentType == constant.HasReadReceipt { + if sendID == recvID && contentType == constant.HasReadReceipt { optionsConfig.ReliabilityLevel = constant.UnreliableNotification } options := config.GetOptionsByNotification(optionsConfig) From 8a13017665ac0203a06f65aaf2d7a020e3e3ade8 Mon Sep 17 00:00:00 2001 From: a3d21 <93191329+a3d21@users.noreply.github.com> Date: Mon, 23 Oct 2023 21:27:27 +0800 Subject: [PATCH 3/3] fix update user.FaceURL do not trigger GroupMemberInfoSetNotification (#1267) * fix update user.FaceURL do not trigger GroupMemberInfoSetNotification * cicd: robot automated Change --------- Co-authored-by: a3d21 --- cmd/openim-api/main.go | 1 + internal/rpc/group/group.go | 51 +++++++++++++++++---------- pkg/common/db/controller/msg.go | 18 ++++++++-- pkg/common/db/s3/minio/minio.go | 8 ++++- pkg/common/db/table/relation/group.go | 2 +- 5 files changed, 58 insertions(+), 22 deletions(-) diff --git a/cmd/openim-api/main.go b/cmd/openim-api/main.go index d1f5cb3f8..174300dc7 100644 --- a/cmd/openim-api/main.go +++ b/cmd/openim-api/main.go @@ -26,6 +26,7 @@ import ( "github.com/OpenIMSDK/protocol/constant" "github.com/OpenIMSDK/tools/discoveryregistry" "github.com/OpenIMSDK/tools/log" + "github.com/openimsdk/open-im-server/v3/internal/api" "github.com/openimsdk/open-im-server/v3/pkg/common/cmd" "github.com/openimsdk/open-im-server/v3/pkg/common/config" diff --git a/internal/rpc/group/group.go b/internal/rpc/group/group.go index de08d2eac..38dd0c32c 100644 --- a/internal/rpc/group/group.go +++ b/internal/rpc/group/group.go @@ -154,16 +154,16 @@ func (s *groupServer) CheckGroupAdmin(ctx context.Context, groupID string) error return nil } -func (s *groupServer) GetUsernameMap(ctx context.Context, userIDs []string, complete bool) (map[string]string, error) { +func (s *groupServer) GetPublicUserInfoMap(ctx context.Context, userIDs []string, complete bool) (map[string]*sdkws.PublicUserInfo, error) { if len(userIDs) == 0 { - return map[string]string{}, nil + return map[string]*sdkws.PublicUserInfo{}, nil } users, err := s.User.GetPublicUserInfos(ctx, userIDs, complete) if err != nil { return nil, err } - return utils.SliceToMapAny(users, func(e *sdkws.PublicUserInfo) (string, string) { - return e.UserID, e.Nickname + return utils.SliceToMapAny(users, func(e *sdkws.PublicUserInfo) (string, *sdkws.PublicUserInfo) { + return e.UserID, e }), nil } @@ -468,15 +468,18 @@ func (s *groupServer) GetGroupAllMember(ctx context.Context, req *pbgroup.GetGro if err != nil { return nil, err } - nameMap, err := s.GetUsernameMap(ctx, utils.Filter(members, func(e *relationtb.GroupMemberModel) (string, bool) { - return e.UserID, e.Nickname == "" + publicUserInfoMap, err := s.GetPublicUserInfoMap(ctx, utils.Filter(members, func(e *relationtb.GroupMemberModel) (string, bool) { + return e.UserID, e.Nickname == "" || e.FaceURL == "" }), true) if err != nil { return nil, err } resp.Members = utils.Slice(members, func(e *relationtb.GroupMemberModel) *sdkws.GroupMemberFullInfo { if e.Nickname == "" { - e.Nickname = nameMap[e.UserID] + e.Nickname = publicUserInfoMap[e.UserID].Nickname + } + if e.FaceURL == "" { + e.FaceURL = publicUserInfoMap[e.UserID].FaceURL } return convert.Db2PbGroupMember(e) }) @@ -616,15 +619,18 @@ func (s *groupServer) GetGroupMembersInfo(ctx context.Context, req *pbgroup.GetG if err != nil { return nil, err } - nameMap, err := s.GetUsernameMap(ctx, utils.Filter(members, func(e *relationtb.GroupMemberModel) (string, bool) { - return e.UserID, e.Nickname == "" + publicUserInfoMap, err := s.GetPublicUserInfoMap(ctx, utils.Filter(members, func(e *relationtb.GroupMemberModel) (string, bool) { + return e.UserID, e.Nickname == "" || e.FaceURL == "" }), true) if err != nil { return nil, err } resp.Members = utils.Slice(members, func(e *relationtb.GroupMemberModel) *sdkws.GroupMemberFullInfo { if e.Nickname == "" { - e.Nickname = nameMap[e.UserID] + e.Nickname = publicUserInfoMap[e.UserID].Nickname + } + if e.FaceURL == "" { + e.FaceURL = publicUserInfoMap[e.UserID].FaceURL } return convert.Db2PbGroupMember(e) }) @@ -1067,15 +1073,18 @@ func (s *groupServer) GetGroupMembersCMS(ctx context.Context, req *pbgroup.GetGr return nil, err } resp.Total = total - nameMap, err := s.GetUsernameMap(ctx, utils.Filter(members, func(e *relationtb.GroupMemberModel) (string, bool) { - return e.UserID, e.Nickname == "" + nameMap, err := s.GetPublicUserInfoMap(ctx, utils.Filter(members, func(e *relationtb.GroupMemberModel) (string, bool) { + return e.UserID, e.Nickname == "" || e.FaceURL == "" }), true) if err != nil { return nil, err } resp.Members = utils.Slice(members, func(e *relationtb.GroupMemberModel) *sdkws.GroupMemberFullInfo { if e.Nickname == "" { - e.Nickname = nameMap[e.UserID] + e.Nickname = nameMap[e.UserID].Nickname + } + if e.FaceURL == "" { + e.FaceURL = nameMap[e.UserID].FaceURL } return convert.Db2PbGroupMember(e) }) @@ -1461,7 +1470,7 @@ func (s *groupServer) GetUserInGroupMembers(ctx context.Context, req *pbgroup.Ge if err != nil { return nil, err } - nameMap, err := s.GetUsernameMap(ctx, utils.Filter(members, func(e *relationtb.GroupMemberModel) (string, bool) { + publicUserInfoMap, err := s.GetPublicUserInfoMap(ctx, utils.Filter(members, func(e *relationtb.GroupMemberModel) (string, bool) { return e.UserID, e.Nickname == "" }), true) if err != nil { @@ -1469,7 +1478,10 @@ func (s *groupServer) GetUserInGroupMembers(ctx context.Context, req *pbgroup.Ge } resp.Members = utils.Slice(members, func(e *relationtb.GroupMemberModel) *sdkws.GroupMemberFullInfo { if e.Nickname == "" { - e.Nickname = nameMap[e.UserID] + e.Nickname = publicUserInfoMap[e.UserID].Nickname + } + if e.FaceURL == "" { + e.FaceURL = publicUserInfoMap[e.UserID].FaceURL } return convert.Db2PbGroupMember(e) }) @@ -1494,15 +1506,18 @@ func (s *groupServer) GetGroupMemberRoleLevel(ctx context.Context, req *pbgroup. if err != nil { return nil, err } - nameMap, err := s.GetUsernameMap(ctx, utils.Filter(members, func(e *relationtb.GroupMemberModel) (string, bool) { - return e.UserID, e.Nickname == "" + publicUserInfoMap, err := s.GetPublicUserInfoMap(ctx, utils.Filter(members, func(e *relationtb.GroupMemberModel) (string, bool) { + return e.UserID, e.Nickname == "" || e.FaceURL == "" }), true) if err != nil { return nil, err } resp.Members = utils.Slice(members, func(e *relationtb.GroupMemberModel) *sdkws.GroupMemberFullInfo { if e.Nickname == "" { - e.Nickname = nameMap[e.UserID] + e.Nickname = publicUserInfoMap[e.UserID].Nickname + } + if e.FaceURL == "" { + e.FaceURL = publicUserInfoMap[e.UserID].FaceURL } return convert.Db2PbGroupMember(e) }) diff --git a/pkg/common/db/controller/msg.go b/pkg/common/db/controller/msg.go index 1bbf4cdf6..cda660afe 100644 --- a/pkg/common/db/controller/msg.go +++ b/pkg/common/db/controller/msg.go @@ -701,7 +701,14 @@ func (db *commonMsgDatabase) DeleteConversationMsgsAndSetMinSeq(ctx context.Cont return db.cache.SetMinSeq(ctx, conversationID, minSeq) } -func processMsgDocModel(ctx context.Context, msgDocModel *unrelationtb.MsgDocModel, userID, conversationID string, index int64, destructTime int64, lastMsgDestructTime time.Time) (seqs []int64, over bool) { +func processMsgDocModel( + ctx context.Context, + msgDocModel *unrelationtb.MsgDocModel, + userID, conversationID string, + index int64, + destructTime int64, + lastMsgDestructTime time.Time, +) (seqs []int64, over bool) { if len(msgDocModel.Msg) > 0 { i := 0 for _, msg := range msgDocModel.Msg { @@ -823,7 +830,14 @@ func handleFullAndExpiredForDeleteMsgRecursion(ctx context.Context, msgDocModel delStruct.minSeq = msgDocModel.Msg[len(msgDocModel.Msg)-1].Msg.Seq } -func handleNotFullAndExpiredForDeleteMsgRecursion(ctx context.Context, msgDocModel *unrelationtb.MsgDocModel, remainTime, index int64, conversationID string, delStruct *delMsgRecursionStruct, db *commonMsgDatabase) { +func handleNotFullAndExpiredForDeleteMsgRecursion( + ctx context.Context, + msgDocModel *unrelationtb.MsgDocModel, + remainTime, index int64, + conversationID string, + delStruct *delMsgRecursionStruct, + db *commonMsgDatabase, +) { var delMsgIndexs []int for i, MsgInfoModel := range msgDocModel.Msg { if MsgInfoModel != nil && MsgInfoModel.Msg != nil { diff --git a/pkg/common/db/s3/minio/minio.go b/pkg/common/db/s3/minio/minio.go index a84b8c3f7..78ed381d5 100644 --- a/pkg/common/db/s3/minio/minio.go +++ b/pkg/common/db/s3/minio/minio.go @@ -430,7 +430,13 @@ func (m *Minio) presignedGetObject(ctx context.Context, name string, expire time return rawURL.String(), nil } -func (m *Minio) getImageInfoForAccessURL(ctx context.Context, name string, expire time.Duration, opt *s3.AccessURLOption, reqParams url.Values) (fileInfo *s3.ObjectInfo, objectInfoPath, msg string, err error) { +func (m *Minio) getImageInfoForAccessURL( + ctx context.Context, + name string, + expire time.Duration, + opt *s3.AccessURLOption, + reqParams url.Values, +) (fileInfo *s3.ObjectInfo, objectInfoPath, msg string, err error) { if opt != nil { if opt.ContentType != "" { reqParams.Set("response-content-type", opt.ContentType) diff --git a/pkg/common/db/table/relation/group.go b/pkg/common/db/table/relation/group.go index 24a75173d..309114ee2 100644 --- a/pkg/common/db/table/relation/group.go +++ b/pkg/common/db/table/relation/group.go @@ -30,7 +30,7 @@ type GroupModel struct { Introduction string `gorm:"column:introduction;size:255" json:"introduction"` FaceURL string `gorm:"column:face_url;size:255" json:"faceURL"` CreateTime time.Time `gorm:"column:create_time;index:create_time;autoCreateTime"` - Ex string `gorm:"column:ex;size:1024" json:"ex"` + Ex string `gorm:"column:ex;size:1024" json:"ex"` Status int32 `gorm:"column:status"` CreatorUserID string `gorm:"column:creator_user_id;size:64"` GroupType int32 `gorm:"column:group_type"`