new feat: sync designated model data (#668)

* new feature: add batch send msg

* new feature: add batch send msg

* new feature: add batch send msg

* new feature: add batch send msg

* new feature: add batch send msg

* new feature: add batch send msg

* fix bug: multiple gateway kick user

* fix bug: multiple gateway kick user

* fix bug: multiple gateway kick user

* fix bug: multiple gateway kick user

* fix bug: multiple gateway kick user

* MsgDestructTime

* fix bug: msg destruct sql

* fix bug: msg destruct

* fix bug: msg destruct

* fix bug: msg destruct sql

* fix bug: msg destruct sql

* fix bug: msg destruct sql

* fix bug: msg destruct sql

* debug: print stack

* debug: print stack

* debug: print stack

* fix bug: msg destruct sql

Signed-off-by: wangchuxiao <wangchuxiao97@outlook.com>

* fix bug: msg notification self 2 self push twice

* fix bug: heartbeat get self notification

* fix bug: init grpc conn in one process

* fix bug: grpc conn

Signed-off-by: wangchuxiao <wangchuxiao97@outlook.com>

* fix bug: zk client recreate node when reconn

* fix bug: set friend mark args error

* fix bug: rpc client intercepter called twice

Signed-off-by: wangchuxiao <wangchuxiao97@outlook.com>

* cicd: robot automated Change

Signed-off-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>

* test: document msg num set 100

* new feat: sync designated model

* new feat: sync designated model

* new feat: sync designated model

* new feat: sync designated model

* new feat: sync designated model

* new feat: sync designated model

* new feat: sync designated model

* merge code

* merge code

---------

Signed-off-by: wangchuxiao <wangchuxiao97@outlook.com>
Signed-off-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>
Co-authored-by: wangchuxiao-dev <wangchuxiao-dev@users.noreply.github.com>

internal/api/friend.go

@@ -44,6 +44,10 @@ func (o *FriendApi) GetFriendApplyList(c *gin.Context) {
 	a2r.Call(friend.FriendClient.GetPaginationFriendsApplyTo, o.Client, c)
 }
 
+func (o *FriendApi) GetDesignatedFriendsApply(c *gin.Context) {
+	a2r.Call(friend.FriendClient.GetDesignatedFriendsApply, o.Client, c)
+}
+
 func (o *FriendApi) GetSelfApplyList(c *gin.Context) {
 	a2r.Call(friend.FriendClient.GetPaginationFriendsApplyFrom, o.Client, c)
 }
@@ -52,6 +56,10 @@ func (o *FriendApi) GetFriendList(c *gin.Context) {
 	a2r.Call(friend.FriendClient.GetPaginationFriends, o.Client, c)
 }
 
+func (o *FriendApi) GetDesignatedFriends(c *gin.Context) {
+	a2r.Call(friend.FriendClient.GetDesignatedFriends, o.Client, c)
+}
+
 func (o *FriendApi) SetFriendRemark(c *gin.Context) {
 	a2r.Call(friend.FriendClient.SetFriendRemark, o.Client, c)
 }

internal/api/group.go

@@ -60,6 +60,10 @@ func (o *GroupApi) GetUserReqGroupApplicationList(c *gin.Context) {
 	a2r.Call(group.GroupClient.GetUserReqApplicationList, o.Client, c)
 }
 
+func (o *GroupApi) GetGroupUsersReqApplicationList(c *gin.Context) {
+	a2r.Call(group.GroupClient.GetGroupUsersReqApplicationList, o.Client, c)
+}
+
 func (o *GroupApi) GetGroupsInfo(c *gin.Context) {
 	a2r.Call(group.GroupClient.GetGroupsInfo, o.Client, c)
 }

internal/api/route.go

@@ -86,8 +86,10 @@ func NewGinRouter(discov discoveryregistry.SvcDiscoveryRegistry, rdb redis.Unive
 		f := NewFriendApi(*friendRpc)
 		friendRouterGroup.POST("/delete_friend", f.DeleteFriend)
 		friendRouterGroup.POST("/get_friend_apply_list", f.GetFriendApplyList)
+		friendRouterGroup.POST("/get_designated_friend_apply", f.GetDesignatedFriendsApply)
 		friendRouterGroup.POST("/get_self_friend_apply_list", f.GetSelfApplyList)
 		friendRouterGroup.POST("/get_friend_list", f.GetFriendList)
+		friendRouterGroup.POST("/get_designated_friends", f.GetDesignatedFriends)
 		friendRouterGroup.POST("/add_friend", f.ApplyToAddFriend)
 		friendRouterGroup.POST("/add_friend_response", f.RespondFriendApply)
 		friendRouterGroup.POST("/set_friend_remark", f.SetFriendRemark)
@@ -108,6 +110,7 @@ func NewGinRouter(discov discoveryregistry.SvcDiscoveryRegistry, rdb redis.Unive
 		groupRouterGroup.POST("/transfer_group", g.TransferGroupOwner)
 		groupRouterGroup.POST("/get_recv_group_applicationList", g.GetRecvGroupApplicationList)
 		groupRouterGroup.POST("/get_user_req_group_applicationList", g.GetUserReqGroupApplicationList)
+		groupRouterGroup.POST("/get_group_users_req_application_list", g.GetGroupUsersReqApplicationList)
 		groupRouterGroup.POST("/get_groups_info", g.GetGroupsInfo)
 		groupRouterGroup.POST("/kick_group", g.KickGroupMember)
 		groupRouterGroup.POST("/get_group_members_info", g.GetGroupMembersInfo)

internal/rpc/conversation/conversaion.go

@@ -107,7 +107,7 @@ func (c *conversationServer) SetConversation(ctx context.Context, req *pbConvers
 	if err != nil {
 		return nil, err
 	}
-	_ = c.conversationNotificationSender.ConversationChangeNotification(ctx, req.Conversation.OwnerUserID)
+	_ = c.conversationNotificationSender.ConversationChangeNotification(ctx, req.Conversation.OwnerUserID, []string{req.Conversation.ConversationID})
 	resp := &pbConversation.SetConversationResp{}
 	return resp, nil
 }
@@ -170,7 +170,7 @@ 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)
+			c.conversationNotificationSender.ConversationSetPrivateNotification(ctx, userID, req.Conversation.UserID, req.Conversation.IsPrivateChat.Value, req.Conversation.ConversationID)
 		}
 	}
 	if req.Conversation.BurnDuration != nil {
@@ -181,7 +181,7 @@ func (c *conversationServer) SetConversations(ctx context.Context, req *pbConver
 		return nil, err
 	}
 	for _, v := range req.UserIDs {
-		c.conversationNotificationSender.ConversationChangeNotification(ctx, v)
+		c.conversationNotificationSender.ConversationChangeNotification(ctx, v, []string{req.Conversation.ConversationID})
 	}
 	return &pbConversation.SetConversationsResp{}, nil
 }

internal/rpc/friend/friend.go

@@ -83,11 +83,6 @@ func Start(client registry.SvcDiscoveryRegistry, server *grpc.Server) error {
 	return nil
 }
 
-func (s *friendServer) GetDesignatedFriendsApply(ctx context.Context, req *pbfriend.GetDesignatedFriendsApplyReq) (*pbfriend.GetDesignatedFriendsApplyResp, error) {
-	//TODO implement me
-	panic("implement me")
-}
-
 // ok.
 func (s *friendServer) ApplyToAddFriend(
 	ctx context.Context,
@@ -245,6 +240,19 @@ func (s *friendServer) GetDesignatedFriends(
 	return resp, nil
 }
 
+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
+	}
+	resp = &pbfriend.GetDesignatedFriendsApplyResp{}
+	resp.FriendRequests, err = convert.FriendRequestDB2Pb(ctx, friendRequests, s.userRpcClient.GetUsersInfoMap)
+	if err != nil {
+		return nil, err
+	}
+	return resp, nil
+}
+
 // ok 获取接收到的好友申请（即别人主动申请的）.
 func (s *friendServer) GetPaginationFriendsApplyTo(
 	ctx context.Context,
@@ -303,7 +311,6 @@ func (s *friendServer) IsFriend(
 	return resp, nil
 }
 
-// ok.
 func (s *friendServer) GetPaginationFriends(
 	ctx context.Context,
 	req *pbfriend.GetPaginationFriendsReq,

internal/rpc/group/group.go

@@ -95,11 +95,6 @@ type groupServer struct {
 	msgRpcClient          rpcclient.MessageRpcClient
 }
 
-func (s *groupServer) GetGroupUsersReqApplicationList(ctx context.Context, req *pbGroup.GetGroupUsersReqApplicationListReq) (*pbGroup.GetGroupUsersReqApplicationListResp, error) {
-	//TODO implement me
-	panic("implement me")
-}
-
 func (s *groupServer) CheckGroupAdmin(ctx context.Context, groupID string) error {
 	if !authverify.IsAppManagerUid(ctx) {
 		groupMember, err := s.GroupDatabase.TakeGroupMember(ctx, groupID, mcontext.GetOpUserID(ctx))
@@ -1445,3 +1440,46 @@ func (s *groupServer) GetGroupMemberRoleLevel(ctx context.Context, req *pbGroup.
 	})
 	return resp, nil
 }
+
+func (s *groupServer) GetGroupUsersReqApplicationList(ctx context.Context, req *pbGroup.GetGroupUsersReqApplicationListReq) (*pbGroup.GetGroupUsersReqApplicationListResp, error) {
+	resp := &pbGroup.GetGroupUsersReqApplicationListResp{}
+	total, requests, err := s.GroupDatabase.FindGroupRequests(ctx, req.GroupID, req.UserIDs)
+	if err != nil {
+		return nil, err
+	}
+	if len(requests) == 0 {
+		return resp, nil
+	}
+	groupIDs := utils.Distinct(utils.Slice(requests, func(e *relationTb.GroupRequestModel) string {
+		return e.GroupID
+	}))
+	groups, err := s.GroupDatabase.FindGroup(ctx, groupIDs)
+	if err != nil {
+		return nil, err
+	}
+	groupMap := utils.SliceToMap(groups, func(e *relationTb.GroupModel) string {
+		return e.GroupID
+	})
+	if ids := utils.Single(groupIDs, utils.Keys(groupMap)); len(ids) > 0 {
+		return nil, errs.ErrGroupIDNotFound.Wrap(strings.Join(ids, ","))
+	}
+	owners, err := s.FindGroupMember(ctx, groupIDs, nil, []int32{constant.GroupOwner})
+	if err != nil {
+		return nil, err
+	}
+	ownerMap := utils.SliceToMap(owners, func(e *relationTb.GroupMemberModel) string {
+		return e.GroupID
+	})
+	if ids := utils.Single(groupIDs, utils.Keys(ownerMap)); len(ids) > 0 {
+		return nil, errs.ErrData.Wrap("group no owner", strings.Join(ids, ","))
+	}
+	groupMemberNum, err := s.GroupDatabase.MapGroupMemberNum(ctx, groupIDs)
+	if err != nil {
+		return nil, err
+	}
+	resp.GroupRequests = utils.Slice(requests, func(e *relationTb.GroupRequestModel) *sdkws.GroupRequest {
+		return convert.Db2PbGroupRequest(e, nil, convert.Db2PbGroupInfo(groupMap[e.GroupID], ownerMap[e.GroupID].UserID, uint32(groupMemberNum[e.GroupID])))
+	})
+	resp.Total = total
+	return resp, nil
+}

internal/rpc/user/user.go

@@ -17,10 +17,11 @@ package user
 import (
 	"context"
 	"errors"
-	"github.com/OpenIMSDK/Open-IM-Server/pkg/authverify"
 	"strings"
 	"time"
 
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/authverify"
+
 	"github.com/OpenIMSDK/tools/log"
 
 	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/config"
@@ -50,11 +51,6 @@ type userServer struct {
 	RegisterCenter     registry.SvcDiscoveryRegistry
 }
 
-func (s *userServer) SubscribeOrCancelUsersStatus(ctx context.Context, req *pbuser.SubscribeOrCancelUsersStatusReq) (*pbuser.SubscribeOrCancelUsersStatusResp, error) {
-	//TODO implement me
-	panic("implement me")
-}
-
 func Start(client registry.SvcDiscoveryRegistry, server *grpc.Server) error {
 	db, err := relation.NewGormDB()
 	if err != nil {
@@ -246,3 +242,7 @@ func (s *userServer) GetAllUserID(ctx context.Context, req *pbuser.GetAllUserIDR
 	}
 	return &pbuser.GetAllUserIDResp{UserIDs: userIDs}, nil
 }
+
+func (s *userServer) SubscribeOrCancelUsersStatus(ctx context.Context, req *pbuser.SubscribeOrCancelUsersStatusReq) (resp *pbuser.SubscribeOrCancelUsersStatusResp, err error) {
+	panic("implement me")
+}

pkg/common/db/controller/friend.go

@@ -75,6 +75,7 @@ type FriendDatabase interface {
 		friendUserIDs []string,
 	) (friends []*relation.FriendModel, err error)
 	FindFriendUserIDs(ctx context.Context, ownerUserID string) (friendUserIDs []string, err error)
+	FindBothFriendRequests(ctx context.Context, fromUserID, toUserID string) (friends []*relation.FriendRequestModel, err error)
 }
 
 type friendDatabase struct {
@@ -363,3 +364,7 @@ func (f *friendDatabase) FindFriendUserIDs(
 ) (friendUserIDs []string, err error) {
 	return f.cache.GetFriendIDs(ctx, ownerUserID)
 }
+
+func (f *friendDatabase) FindBothFriendRequests(ctx context.Context, fromUserID, toUserID string) (friends []*relation.FriendRequestModel, err error) {
+	return f.friendRequest.FindBothFriendRequests(ctx, fromUserID, toUserID)
+}

pkg/common/db/controller/group.go

@@ -110,6 +110,7 @@ type GroupDatabase interface {
 	// GroupRequest
 	CreateGroupRequest(ctx context.Context, requests []*relationTb.GroupRequestModel) error
 	TakeGroupRequest(ctx context.Context, groupID string, userID string) (*relationTb.GroupRequestModel, error)
+	FindGroupRequests(ctx context.Context, groupID string, userIDs []string) (int64, []*relationTb.GroupRequestModel, error)
 	PageGroupRequestUser(
 		ctx context.Context,
 		userID string,
@@ -576,3 +577,7 @@ func (g *groupDatabase) CountTotal(ctx context.Context, before *time.Time) (coun
 func (g *groupDatabase) CountRangeEverydayTotal(ctx context.Context, start time.Time, end time.Time) (map[string]int64, error) {
 	return g.groupDB.CountRangeEverydayTotal(ctx, start, end)
 }
+
+func (g *groupDatabase) FindGroupRequests(ctx context.Context, groupID string, userIDs []string) (int64, []*relationTb.GroupRequestModel, error) {
+	return g.groupRequestDB.FindGroupRequests(ctx, groupID, userIDs)
+}

pkg/common/db/relation/friend_request_model.go

@@ -147,3 +147,14 @@ func (f *FriendRequestGorm) FindFromUserID(
 	)
 	return
 }
+
+func (f *FriendRequestGorm) FindBothFriendRequests(ctx context.Context, fromUserID, toUserID string) (friends []*relation.FriendRequestModel, err error) {
+	err = utils.Wrap(
+		f.db(ctx).
+			Where("(from_user_id = ? AND to_user_id = ?) OR (from_user_id = ? AND to_user_id = ?)", fromUserID, toUserID, toUserID, fromUserID).
+			Find(&friends).
+			Error,
+		"",
+	)
+	return
+}

pkg/common/db/relation/group_request_model.go

@@ -110,3 +110,8 @@ func (g *GroupRequestGorm) PageGroup(
 		showNumber,
 	)
 }
+
+func (g *GroupRequestGorm) FindGroupRequests(ctx context.Context, groupID string, userIDs []string) (total int64, groupRequests []*relation.GroupRequestModel, err error) {
+	err = g.DB.WithContext(ctx).Where("group_id = ? and user_id in ?", groupID, userIDs).Find(&groupRequests).Error
+	return int64(len(groupRequests)), groupRequests, utils.Wrap(err, utils.GetSelfFuncName())
+}

pkg/common/db/table/relation/friend_request.go

@@ -61,6 +61,6 @@ type FriendRequestModelInterface interface {
 		fromUserID string,
 		pageNumber, showNumber int32,
 	) (friendRequests []*FriendRequestModel, total int64, err error)
-
+	FindBothFriendRequests(ctx context.Context, fromUserID, toUserID string) (friends []*FriendRequestModel, err error)
 	NewTx(tx any) FriendRequestModelInterface
 }

pkg/common/db/table/relation/group_request.go

@@ -47,6 +47,7 @@ type GroupRequestModelInterface interface {
 	Delete(ctx context.Context, groupID string, userID string) (err error)
 	UpdateHandler(ctx context.Context, groupID string, userID string, handledMsg string, handleResult int32) (err error)
 	Take(ctx context.Context, groupID string, userID string) (groupRequest *GroupRequestModel, err error)
+	FindGroupRequests(ctx context.Context, groupID string, userIDs []string) (int64, []*GroupRequestModel, error)
 	Page(
 		ctx context.Context,
 		userID string,

pkg/rpcclient/notification/conevrsation.go

@@ -31,23 +31,22 @@ func NewConversationNotificationSender(msgRpcClient *rpcclient.MessageRpcClient)
 }
 
 // SetPrivate调用.
-func (c *ConversationNotificationSender) ConversationSetPrivateNotification(
+func (c *ConversationNotificationSender) ConversationSetPrivateNotification(ctx context.Context, sendID, recvID string,
-	ctx context.Context,
+	isPrivateChat bool, conversationID string) error {
-	sendID, recvID string,
-	isPrivateChat bool,
-) error {
 	tips := &sdkws.ConversationSetPrivateTips{
 		RecvID:         recvID,
 		SendID:         sendID,
 		IsPrivate:      isPrivateChat,
+		ConversationID: conversationID,
 	}
 	return c.Notification(ctx, sendID, recvID, constant.ConversationPrivateChatNotification, tips)
 }
 
 // 会话改变.
-func (c *ConversationNotificationSender) ConversationChangeNotification(ctx context.Context, userID string) error {
+func (c *ConversationNotificationSender) ConversationChangeNotification(ctx context.Context, userID string, conversationIDs []string) error {
 	tips := &sdkws.ConversationUpdateTips{
 		UserID:             userID,
+		ConversationIDList: conversationIDs,
 	}
 	return c.Notification(ctx, userID, userID, constant.ConversationChangeNotification, tips)
 }