From 9ba22f30aeaffb5d5961b3cc8cf8b0304305b925 Mon Sep 17 00:00:00 2001 From: withchao <993506633@qq.com> Date: Thu, 30 May 2024 11:19:32 +0800 Subject: [PATCH] friend incr sync --- cmd/openim-rpc/openim-rpc-friend/main.go | 4 + go.mod | 6 +- internal/rpc/friend/friend.go | 3 + internal/rpc/friend/sync.go | 50 ++++++++++- internal/rpc/user/user.go | 108 ++++++++++++++++------- pkg/common/cmd/msg_gateway_test.go | 7 ++ pkg/common/convert/user.go | 28 +++--- pkg/common/db/cache/friend.go | 13 +++ pkg/common/db/controller/friend.go | 21 +++++ pkg/common/db/dataver/common.go | 13 +-- pkg/common/db/mgo/friend.go | 39 +++++++- pkg/common/db/table/relation/friend.go | 6 ++ 12 files changed, 239 insertions(+), 59 deletions(-) diff --git a/cmd/openim-rpc/openim-rpc-friend/main.go b/cmd/openim-rpc/openim-rpc-friend/main.go index 745c40553..4589fb30d 100644 --- a/cmd/openim-rpc/openim-rpc-friend/main.go +++ b/cmd/openim-rpc/openim-rpc-friend/main.go @@ -17,9 +17,13 @@ package main import ( "github.com/openimsdk/open-im-server/v3/pkg/common/cmd" "github.com/openimsdk/tools/system/program" + "os" ) func main() { + if len(os.Args) == 1 { + os.Args = []string{os.Args[0], "-i", "0", "-c", "/Users/chao/Desktop/project/open-im-server/config"} + } if err := cmd.NewFriendRpcCmd().Exec(); err != nil { program.ExitWithError(err) } diff --git a/go.mod b/go.mod index 0b2f3cb3b..0ae542cd2 100644 --- a/go.mod +++ b/go.mod @@ -178,6 +178,6 @@ require ( gopkg.in/ini.v1 v1.67.0 // indirect ) -//replace ( -// github.com/openimsdk/protocol => /Users/chao/Desktop/project/protocol -//) +replace ( + github.com/openimsdk/protocol => /Users/chao/Desktop/project/protocol +) diff --git a/internal/rpc/friend/friend.go b/internal/rpc/friend/friend.go index b49490f26..31965adb9 100644 --- a/internal/rpc/friend/friend.go +++ b/internal/rpc/friend/friend.go @@ -62,6 +62,9 @@ type Config struct { } func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryRegistry, server *grpc.Server) error { + if config.RpcConfig.FriendSyncCount < 1 { + config.RpcConfig.FriendSyncCount = constant.MaxSyncPullNumber + } mgocli, err := mongoutil.NewMongoDB(ctx, config.MongodbConfig.Build()) if err != nil { return err diff --git a/internal/rpc/friend/sync.go b/internal/rpc/friend/sync.go index a2f49b283..dd6e49250 100644 --- a/internal/rpc/friend/sync.go +++ b/internal/rpc/friend/sync.go @@ -8,11 +8,55 @@ import ( "github.com/openimsdk/open-im-server/v3/pkg/authverify" "github.com/openimsdk/open-im-server/v3/pkg/common/db/table/relation" pbfriend "github.com/openimsdk/protocol/friend" + "github.com/openimsdk/tools/errs" ) +func (s *friendServer) NotificationUserInfoUpdate(ctx context.Context, req *pbfriend.NotificationUserInfoUpdateReq) (*pbfriend.NotificationUserInfoUpdateResp, error) { + if req.NewUserInfo == nil { + var err error + req.NewUserInfo, err = s.userRpcClient.GetUserInfo(ctx, req.UserID) + if err != nil { + return nil, err + } + } + if req.UserID != req.NewUserInfo.UserID { + return nil, errs.ErrArgs.WrapMsg("req.UserID != req.NewUserInfo.UserID") + } + userIDs, err := s.friendDatabase.FindFriendUserID(ctx, req.UserID) + if err != nil { + return nil, err + } + if len(userIDs) > 0 { + if err := s.friendDatabase.UpdateFriendUserInfo(ctx, req.UserID, userIDs, req.NewUserInfo.Nickname, req.NewUserInfo.FaceURL); err != nil { + return nil, err + } + s.notificationSender.FriendsInfoUpdateNotification(ctx, req.UserID, userIDs) + } + return &pbfriend.NotificationUserInfoUpdateResp{}, nil +} + func (s *friendServer) SearchFriends(ctx context.Context, req *pbfriend.SearchFriendsReq) (*pbfriend.SearchFriendsResp, error) { - //TODO implement me - panic("implement me") + if err := s.userRpcClient.Access(ctx, req.UserID); err != nil { + return nil, err + } + if req.Keyword == "" { + total, friends, err := s.friendDatabase.PageOwnerFriends(ctx, req.UserID, req.Pagination) + if err != nil { + return nil, err + } + return &pbfriend.SearchFriendsResp{ + Total: total, + Friends: friendsDB2PB(friends), + }, nil + } + total, friends, err := s.friendDatabase.SearchFriend(ctx, req.UserID, req.Keyword, req.Pagination) + if err != nil { + return nil, err + } + return &pbfriend.SearchFriendsResp{ + Total: total, + Friends: friendsDB2PB(friends), + }, nil } func (s *friendServer) sortFriendUserIDsHash(userIDs []string) uint64 { @@ -54,7 +98,7 @@ func (s *friendServer) GetIncrementalFriends(ctx context.Context, req *pbfriend. } return &pbfriend.GetIncrementalFriendsResp{ Version: uint64(incrVer.Version), - VersionID: incrVer.ID.String(), + VersionID: incrVer.ID.Hex(), Full: incrVer.Full(), SyncCount: uint32(s.config.RpcConfig.FriendSyncCount), DeleteUserIds: deleteUserIDs, diff --git a/internal/rpc/user/user.go b/internal/rpc/user/user.go index 2296ee62f..92f897ea3 100644 --- a/internal/rpc/user/user.go +++ b/internal/rpc/user/user.go @@ -16,12 +16,16 @@ package user import ( "context" + "errors" "github.com/openimsdk/open-im-server/v3/internal/rpc/friend" "github.com/openimsdk/open-im-server/v3/pkg/common/config" "github.com/openimsdk/open-im-server/v3/pkg/common/webhook" + friendpb "github.com/openimsdk/protocol/friend" + "github.com/openimsdk/protocol/group" "github.com/openimsdk/tools/db/redisutil" "math/rand" "strings" + "sync" "time" "github.com/openimsdk/open-im-server/v3/pkg/authverify" @@ -56,11 +60,6 @@ type userServer struct { webhookClient *webhook.Client } -func (s *userServer) SearchUser(ctx context.Context, req *pbuser.SearchUserReq) (*pbuser.SearchUserResp, error) { - //TODO implement me - panic("implement me") -} - type Config struct { RpcConfig config.User RedisConfig config.Redis @@ -136,26 +135,29 @@ func (s *userServer) UpdateUserInfo(ctx context.Context, req *pbuser.UpdateUserI if err := s.webhookBeforeUpdateUserInfo(ctx, &s.config.WebhooksConfig.BeforeUpdateUserInfo, req); err != nil { return nil, err } - data := convert.UserPb2DBMap(req.UserInfo) - if err := s.db.UpdateByMap(ctx, req.UserInfo.UserID, data); err != nil { - return nil, err - } - s.friendNotificationSender.UserInfoUpdatedNotification(ctx, req.UserInfo.UserID) - friends, err := s.friendRpcClient.GetFriendIDs(ctx, req.UserInfo.UserID) + oldUser, err := s.db.GetUserByID(ctx, req.UserInfo.UserID) if err != nil { return nil, err } - if req.UserInfo.Nickname != "" || req.UserInfo.FaceURL != "" { - if err = s.groupRpcClient.NotificationUserInfoUpdate(ctx, req.UserInfo.UserID); err != nil { - return nil, err - } - } - for _, friendID := range friends { - s.friendNotificationSender.FriendInfoUpdatedNotification(ctx, req.UserInfo.UserID, friendID) + if err := s.db.UpdateByMap(ctx, req.UserInfo.UserID, data); err != nil { + return nil, err } + //s.friendNotificationSender.UserInfoUpdatedNotification(ctx, req.UserInfo.UserID) + //friends, err := s.friendRpcClient.GetFriendIDs(ctx, req.UserInfo.UserID) + //if err != nil { + // return nil, err + //} + //if req.UserInfo.Nickname != "" || req.UserInfo.FaceURL != "" { + // if err = s.NotificationUserInfoUpdate(ctx, req.UserInfo.UserID,oldUser); err != nil { + // return nil, err + // } + //} + //for _, friendID := range friends { + // s.friendNotificationSender.FriendInfoUpdatedNotification(ctx, req.UserInfo.UserID, friendID) + //} s.webhookAfterUpdateUserInfo(ctx, &s.config.WebhooksConfig.AfterUpdateUserInfo, req) - if err = s.groupRpcClient.NotificationUserInfoUpdate(ctx, req.UserInfo.UserID); err != nil { + if err = s.NotificationUserInfoUpdate(ctx, req.UserInfo.UserID, oldUser); err != nil { return nil, err } return resp, nil @@ -170,24 +172,28 @@ func (s *userServer) UpdateUserInfoEx(ctx context.Context, req *pbuser.UpdateUse return nil, err } data := convert.UserPb2DBMapEx(req.UserInfo) - if err = s.db.UpdateByMap(ctx, req.UserInfo.UserID, data); err != nil { - return nil, err - } - s.friendNotificationSender.UserInfoUpdatedNotification(ctx, req.UserInfo.UserID) - friends, err := s.friendRpcClient.GetFriendIDs(ctx, req.UserInfo.UserID) + oldUser, err := s.db.GetUserByID(ctx, req.UserInfo.UserID) if err != nil { return nil, err } - if req.UserInfo.Nickname != nil || req.UserInfo.FaceURL != nil { - if err := s.groupRpcClient.NotificationUserInfoUpdate(ctx, req.UserInfo.UserID); err != nil { - return nil, err - } - } - for _, friendID := range friends { - s.friendNotificationSender.FriendInfoUpdatedNotification(ctx, req.UserInfo.UserID, friendID) + if err = s.db.UpdateByMap(ctx, req.UserInfo.UserID, data); err != nil { + return nil, err } + //s.friendNotificationSender.UserInfoUpdatedNotification(ctx, req.UserInfo.UserID) + //friends, err := s.friendRpcClient.GetFriendIDs(ctx, req.UserInfo.UserID) + //if err != nil { + // return nil, err + //} + //if req.UserInfo.Nickname != nil || req.UserInfo.FaceURL != nil { + // if err := s.NotificationUserInfoUpdate(ctx, req.UserInfo.UserID); err != nil { + // return nil, err + // } + //} + //for _, friendID := range friends { + // s.friendNotificationSender.FriendInfoUpdatedNotification(ctx, req.UserInfo.UserID, friendID) + //} s.webhookAfterUpdateUserInfoEx(ctx, &s.config.WebhooksConfig.AfterUpdateUserInfoEx, req) - if err := s.groupRpcClient.NotificationUserInfoUpdate(ctx, req.UserInfo.UserID); err != nil { + if err := s.NotificationUserInfoUpdate(ctx, req.UserInfo.UserID, oldUser); err != nil { return nil, err } return resp, nil @@ -688,3 +694,41 @@ func (s *userServer) userModelToResp(users []*relation.UserModel, pagination pag return &pbuser.SearchNotificationAccountResp{Total: total, NotificationAccounts: notificationAccounts} } + +func (s *userServer) NotificationUserInfoUpdate(ctx context.Context, userID string, oldUser *relation.UserModel) error { + user, err := s.db.GetUserByID(ctx, userID) + if err != nil { + return err + } + if *user == *oldUser { + return nil + } + s.friendNotificationSender.UserInfoUpdatedNotification(ctx, userID) + if user.Nickname == oldUser.Nickname && user.FaceURL == oldUser.FaceURL { + return nil + } + oldUserInfo := convert.UserDB2Pb(oldUser) + newUserInfo := convert.UserDB2Pb(user) + var wg sync.WaitGroup + var es [2]error + wg.Add(len(es)) + go func() { + defer wg.Done() + _, es[0] = s.groupRpcClient.Client.NotificationUserInfoUpdate(ctx, &group.NotificationUserInfoUpdateReq{ + UserID: userID, + OldUserInfo: oldUserInfo, + NewUserInfo: newUserInfo, + }) + }() + + go func() { + defer wg.Done() + _, es[1] = s.friendRpcClient.Client.NotificationUserInfoUpdate(ctx, &friendpb.NotificationUserInfoUpdateReq{ + UserID: userID, + OldUserInfo: oldUserInfo, + NewUserInfo: newUserInfo, + }) + }() + wg.Wait() + return errors.Join(es[:]...) +} diff --git a/pkg/common/cmd/msg_gateway_test.go b/pkg/common/cmd/msg_gateway_test.go index d820627b5..2b68a3e3a 100644 --- a/pkg/common/cmd/msg_gateway_test.go +++ b/pkg/common/cmd/msg_gateway_test.go @@ -19,6 +19,7 @@ import ( "github.com/openimsdk/tools/apiresp" "github.com/openimsdk/tools/utils/jsonutil" "github.com/stretchr/testify/mock" + "go.mongodb.org/mongo-driver/bson/primitive" "math" "testing" ) @@ -59,3 +60,9 @@ func TestName(t *testing.T) { t.Logf("%+v\n", rReso) } + +func TestName1(t *testing.T) { + t.Log(primitive.NewObjectID().String()) + t.Log(primitive.NewObjectID().Hex()) + +} diff --git a/pkg/common/convert/user.go b/pkg/common/convert/user.go index a9378e1a0..c7c90edb3 100644 --- a/pkg/common/convert/user.go +++ b/pkg/common/convert/user.go @@ -15,27 +15,27 @@ package convert import ( + "github.com/openimsdk/tools/utils/datautil" "time" relationtb "github.com/openimsdk/open-im-server/v3/pkg/common/db/table/relation" "github.com/openimsdk/protocol/sdkws" ) -func UsersDB2Pb(users []*relationtb.UserModel) []*sdkws.UserInfo { - result := make([]*sdkws.UserInfo, 0, len(users)) - for _, user := range users { - userPb := &sdkws.UserInfo{ - UserID: user.UserID, - Nickname: user.Nickname, - FaceURL: user.FaceURL, - Ex: user.Ex, - CreateTime: user.CreateTime.UnixMilli(), - AppMangerLevel: user.AppMangerLevel, - GlobalRecvMsgOpt: user.GlobalRecvMsgOpt, - } - result = append(result, userPb) +func UserDB2Pb(user *relationtb.UserModel) *sdkws.UserInfo { + return &sdkws.UserInfo{ + UserID: user.UserID, + Nickname: user.Nickname, + FaceURL: user.FaceURL, + Ex: user.Ex, + CreateTime: user.CreateTime.UnixMilli(), + AppMangerLevel: user.AppMangerLevel, + GlobalRecvMsgOpt: user.GlobalRecvMsgOpt, } - return result +} + +func UsersDB2Pb(users []*relationtb.UserModel) []*sdkws.UserInfo { + return datautil.Slice(users, UserDB2Pb) } func UserPb2DB(user *sdkws.UserInfo) *relationtb.UserModel { diff --git a/pkg/common/db/cache/friend.go b/pkg/common/db/cache/friend.go index 4a51a8c17..eb984d8a4 100644 --- a/pkg/common/db/cache/friend.go +++ b/pkg/common/db/cache/friend.go @@ -52,6 +52,8 @@ type FriendCache interface { // Delete friends when friends' info changed DelFriends(ownerUserID string, friendUserIDs []string) FriendCache + DelOwner(friendUserID string, ownerUserIDs []string) FriendCache + FindSortFriendUserIDs(ctx context.Context, ownerUserID string) ([]string, error) FindFriendIncrVersion(ctx context.Context, ownerUserID string, version uint, limit int) (*dataver.WriteLog, error) @@ -196,6 +198,17 @@ func (f *FriendCacheRedis) DelFriends(ownerUserID string, friendUserIDs []string return newFriendCache } +func (f *FriendCacheRedis) DelOwner(friendUserID string, ownerUserIDs []string) FriendCache { + newFriendCache := f.NewCache() + + for _, ownerUserID := range ownerUserIDs { + key := f.getFriendKey(ownerUserID, friendUserID) + newFriendCache.AddKeys(key) // Assuming AddKeys marks the keys for deletion + } + + return newFriendCache +} + func (f *FriendCacheRedis) FindSortFriendUserIDs(ctx context.Context, ownerUserID string) ([]string, error) { return getCache(ctx, f.rcClient, f.getFriendSyncSortUserIDsKey(ownerUserID), f.expireTime, func(ctx context.Context) ([]string, error) { return f.friendDB.FindOwnerFriendUserIds(ctx, ownerUserID, f.syncCount) diff --git a/pkg/common/db/controller/friend.go b/pkg/common/db/controller/friend.go index 49c0cb990..0eef2bee3 100644 --- a/pkg/common/db/controller/friend.go +++ b/pkg/common/db/controller/friend.go @@ -80,6 +80,12 @@ type FriendDatabase interface { FindSortFriendUserIDs(ctx context.Context, ownerUserID string) ([]string, error) FindFriendIncrVersion(ctx context.Context, ownerUserID string, version uint, limit int) (*dataver.WriteLog, error) + + FindFriendUserID(ctx context.Context, friendUserID string) ([]string, error) + + UpdateFriendUserInfo(ctx context.Context, friendUserID string, ownerUserID []string, nickname string, faceURL string) error + + SearchFriend(ctx context.Context, ownerUserID, keyword string, pagination pagination.Pagination) (int64, []*relation.FriendModel, error) } type friendDatabase struct { @@ -357,3 +363,18 @@ func (f *friendDatabase) FindSortFriendUserIDs(ctx context.Context, ownerUserID func (f *friendDatabase) FindFriendIncrVersion(ctx context.Context, ownerUserID string, version uint, limit int) (*dataver.WriteLog, error) { return f.cache.FindFriendIncrVersion(ctx, ownerUserID, version, limit) } + +func (f *friendDatabase) FindFriendUserID(ctx context.Context, friendUserID string) ([]string, error) { + return f.friend.FindFriendUserID(ctx, friendUserID) +} + +func (f *friendDatabase) UpdateFriendUserInfo(ctx context.Context, friendUserID string, ownerUserIDs []string, nickname string, faceURL string) error { + if err := f.friend.UpdateFriendUserInfo(ctx, friendUserID, nickname, faceURL); err != nil { + return err + } + return f.cache.DelOwner(friendUserID, ownerUserIDs).ExecDel(ctx) +} + +func (f *friendDatabase) SearchFriend(ctx context.Context, ownerUserID, keyword string, pagination pagination.Pagination) (int64, []*relation.FriendModel, error) { + return f.friend.SearchFriend(ctx, ownerUserID, keyword, pagination) +} diff --git a/pkg/common/db/dataver/common.go b/pkg/common/db/dataver/common.go index e29ab2b96..5ade1fd58 100644 --- a/pkg/common/db/dataver/common.go +++ b/pkg/common/db/dataver/common.go @@ -26,13 +26,11 @@ type WriteLog struct { Deleted uint `bson:"deleted"` LastUpdate time.Time `bson:"last_update"` LogLen int `bson:"log_len"` + queryDoc bool `bson:"-"` } func (w *WriteLog) Full() bool { - if w.Version == 0 { - return true - } - return len(w.Logs) != w.LogLen + return w.queryDoc || w.Version == 0 || len(w.Logs) != w.LogLen } func (w *WriteLog) DeleteAndChangeIDs() (delIds []string, changeIds []string) { @@ -211,7 +209,12 @@ func (l *logModel) writeLogBatch(ctx context.Context, dId string, eIds []string, } func (l *logModel) findDoc(ctx context.Context, dId string) (*WriteLog, error) { - return mongoutil.FindOne[*WriteLog](ctx, l.coll, bson.M{"d_id": dId}, options.FindOne().SetProjection(bson.M{"logs": 0})) + res, err := mongoutil.FindOne[*WriteLog](ctx, l.coll, bson.M{"d_id": dId}, options.FindOne().SetProjection(bson.M{"logs": 0})) + if err != nil { + return nil, err + } + res.queryDoc = true + return res, nil } func (l *logModel) FindChangeLog(ctx context.Context, dId string, version uint, limit int) (*WriteLog, error) { diff --git a/pkg/common/db/mgo/friend.go b/pkg/common/db/mgo/friend.go index 37ba2d7ac..cf84acb1c 100644 --- a/pkg/common/db/mgo/friend.go +++ b/pkg/common/db/mgo/friend.go @@ -145,13 +145,20 @@ func (f *FriendMgo) FindReversalFriends(ctx context.Context, friendUserID string // FindOwnerFriends retrieves a paginated list of friends for a given owner. func (f *FriendMgo) FindOwnerFriends(ctx context.Context, ownerUserID string, pagination pagination.Pagination) (int64, []*relation.FriendModel, error) { filter := bson.M{"owner_user_id": ownerUserID} - opt := options.Find().SetSort(bson.A{bson.M{"friend_nickname": 1}, bson.M{"create_time": 1}}) + opt := options.Find().SetSort(bson.D{{"friend_nickname", 1}, {"create_time", 1}}) return mongoutil.FindPage[*relation.FriendModel](ctx, f.coll, filter, pagination, opt) } func (f *FriendMgo) FindOwnerFriendUserIds(ctx context.Context, ownerUserID string, limit int) ([]string, error) { filter := bson.M{"owner_user_id": ownerUserID} - opt := options.Find().SetProjection(bson.M{"_id": 0, "friend_user_id": 1}).SetSort(bson.A{bson.M{"friend_nickname": 1}, bson.M{"create_time": 1}}).SetLimit(int64(limit)) + opt := options.Find().SetProjection(bson.M{"_id": 0, "friend_user_id": 1}).SetSort(bson.D{{"friend_nickname", 1}, {"create_time", 1}}).SetLimit(int64(limit)) + //res, err := mongoutil.Find[string](ctx, f.coll, filter, opt) + //if err != nil { + // errMsg := err.Error() + // _ = errMsg + // return nil, err + //} + //return res, nil return mongoutil.Find[string](ctx, f.coll, filter, opt) } @@ -193,6 +200,34 @@ func (f *FriendMgo) FindIncrVersion(ctx context.Context, ownerUserID string, ver return f.owner.FindChangeLog(ctx, ownerUserID, version, limit) } +func (f *FriendMgo) FindFriendUserID(ctx context.Context, friendUserID string) ([]string, error) { + filter := bson.M{ + "friend_user_id": friendUserID, + } + return mongoutil.Find[string](ctx, f.coll, filter, options.Find().SetProjection(bson.M{"_id": 0, "owner_user_id": 1})) +} + +func (f *FriendMgo) UpdateFriendUserInfo(ctx context.Context, friendUserID string, nickname string, faceURL string) error { + filter := bson.M{ + "friend_user_id": friendUserID, + } + _, err := mongoutil.UpdateMany(ctx, f.coll, filter, bson.M{"$set": bson.M{"nickname": nickname, "face_url": faceURL}}) + return err +} + +func (f *FriendMgo) SearchFriend(ctx context.Context, ownerUserID, keyword string, pagination pagination.Pagination) (int64, []*relation.FriendModel, error) { + //where := bson.M{ + // "owner_user_id": ownerUserID, + // "$or": []bson.M{ + // {"remark": bson.M{"$regex": keyword, "$options": "i"}}, + // {"friend_user_id": bson.M{"$regex": keyword, "$options": "i"}}, + // {"nickname": bson.M{"$regex": keyword, "$options": "i"}}, + // }, + //} + //return f.aggregatePagination(ctx, where, pagination) + panic("todo") +} + func IncrVersion(dbs ...func() error) error { for _, fn := range dbs { if err := fn(); err != nil { diff --git a/pkg/common/db/table/relation/friend.go b/pkg/common/db/table/relation/friend.go index a28b2278d..4970a0420 100644 --- a/pkg/common/db/table/relation/friend.go +++ b/pkg/common/db/table/relation/friend.go @@ -66,4 +66,10 @@ type FriendModelInterface interface { UpdateFriends(ctx context.Context, ownerUserID string, friendUserIDs []string, val map[string]any) (err error) FindIncrVersion(ctx context.Context, ownerUserID string, version uint, limit int) (*dataver.WriteLog, error) + + FindFriendUserID(ctx context.Context, friendUserID string) ([]string, error) + + UpdateFriendUserInfo(ctx context.Context, friendUserID string, nickname string, faceURL string) error + + SearchFriend(ctx context.Context, ownerUserID, keyword string, pagination pagination.Pagination) (int64, []*FriendModel, error) }