friend incr sync

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)
 	}

go.mod

@@ -178,6 +178,6 @@ require (
 	gopkg.in/ini.v1 v1.67.0 // indirect
 )
 
-//replace (
+replace (
-//	github.com/openimsdk/protocol => /Users/chao/Desktop/project/protocol
+	github.com/openimsdk/protocol => /Users/chao/Desktop/project/protocol
-//)
+)

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

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
+	if err := s.userRpcClient.Access(ctx, req.UserID); err != nil {
-	panic("implement me")
+		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,

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 {
+	oldUser, err := s.db.GetUserByID(ctx, req.UserInfo.UserID)
-		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.db.UpdateByMap(ctx, req.UserInfo.UserID, data); err != nil {
-		if err = s.groupRpcClient.NotificationUserInfoUpdate(ctx, req.UserInfo.UserID); err != nil {
+		return nil, err
-			return nil, err
-		}
-	}
-	for _, friendID := range friends {
-		s.friendNotificationSender.FriendInfoUpdatedNotification(ctx, req.UserInfo.UserID, friendID)
 	}
+	//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 {
+	oldUser, err := s.db.GetUserByID(ctx, req.UserInfo.UserID)
-		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.db.UpdateByMap(ctx, req.UserInfo.UserID, data); err != nil {
-		if err := s.groupRpcClient.NotificationUserInfoUpdate(ctx, req.UserInfo.UserID); err != nil {
+		return nil, err
-			return nil, err
-		}
-	}
-	for _, friendID := range friends {
-		s.friendNotificationSender.FriendInfoUpdatedNotification(ctx, req.UserInfo.UserID, friendID)
 	}
+	//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[:]...)
+}

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())
+
+}

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 {
+func UserDB2Pb(user *relationtb.UserModel) *sdkws.UserInfo {
-	result := make([]*sdkws.UserInfo, 0, len(users))
+	return &sdkws.UserInfo{
-	for _, user := range users {
+		UserID:           user.UserID,
-		userPb := &sdkws.UserInfo{
+		Nickname:         user.Nickname,
-			UserID:           user.UserID,
+		FaceURL:          user.FaceURL,
-			Nickname:         user.Nickname,
+		Ex:               user.Ex,
-			FaceURL:          user.FaceURL,
+		CreateTime:       user.CreateTime.UnixMilli(),
-			Ex:               user.Ex,
+		AppMangerLevel:   user.AppMangerLevel,
-			CreateTime:       user.CreateTime.UnixMilli(),
+		GlobalRecvMsgOpt: user.GlobalRecvMsgOpt,
-			AppMangerLevel:   user.AppMangerLevel,
-			GlobalRecvMsgOpt: user.GlobalRecvMsgOpt,
-		}
-		result = append(result, userPb)
 	}
-	return result
+}
+
+func UsersDB2Pb(users []*relationtb.UserModel) []*sdkws.UserInfo {
+	return datautil.Slice(users, UserDB2Pb)
 }
 
 func UserPb2DB(user *sdkws.UserInfo) *relationtb.UserModel {

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)

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)
+}

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 w.queryDoc || w.Version == 0 || len(w.Logs) != w.LogLen
-		return true
-	}
-	return 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) {

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 {

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)
 }