feat: group members, friends sorting version, client online subscription (#2427)

* fix: GroupApplicationAcceptedNotification

* fix: GroupApplicationAcceptedNotification

* fix: NotificationUserInfoUpdate

* cicd: robot automated Change

* fix: component

* fix: getConversationInfo

* feat: cron task

* feat: cron task

* feat: cron task

* feat: cron task

* feat: cron task

* fix: minio config url recognition error

* new mongo

* new mongo

* new mongo

* new mongo

* new mongo

* new mongo

* new mongo

* new mongo

* friend incr sync

* friend incr sync

* friend incr sync

* friend incr sync

* friend incr sync

* mage

* optimization version log

* optimization version log

* sync

* sync

* sync

* group sync

* sync option

* sync option

* refactor: replace `friend` package with `realtion`.

* refactor: update lastest commit to relation.

* sync option

* sync option

* sync option

* sync

* sync

* go.mod

* seq

* update: go mod

* refactor: change incremental to full

* feat: get full friend user ids

* feat: api and config

* seq

* group version

* merge

* seq

* seq

* seq

* fix: sort by id avoid unstable sort friends.

* group

* group

* group

* fix: sort by id avoid unstable sort friends.

* fix: sort by id avoid unstable sort friends.

* fix: sort by id avoid unstable sort friends.

* user version

* seq

* seq

* seq user

* user online

* implement minio expire delete.

* user online

* config

* fix

* fix

* implement minio expire delete logic.

* online cache

* online cache

* online cache

* online cache

* online cache

* online cache

* online cache

* online cache

* online cache

* online cache

* online cache

* online cache

* feat: implement scheduled delete outdated object in minio.

* update gomake version

* update gomake version

* implement FindExpires pagination.

* remove unnesseary incr.

* fix uncorrect args call.

* online push

* online push

* online push

* resolving conflicts

* resolving conflicts

* test

* api prommetrics

* api prommetrics

* api prommetrics

* api prommetrics

* api prommetrics

* rpc prommetrics

* rpc prommetrics

* online status

* online status

* online status

* online status

* sub

* conversation version incremental

* merge seq

* merge online

* merge online

* merge online

* merge seq

* GetOwnerConversation

* fix: change incremental syncer router name.

* rockscache batch get

* rockscache seq batch get

* fix: GetMsgDocModelByIndex bug

* update go.mod

* update go.mod

* merge

* feat: prometheus

* feat: prometheus

* group member sort

* sub

* sub

* fix: seq conversion bug

* fix: redis pipe exec

* sort version

* sort version

* sort version

* remove old version online subscription

* remove old version online subscription

* version log index

---------

Co-authored-by: withchao <withchao@users.noreply.github.com>
Co-authored-by: Monet Lee <monet_lee@163.com>
Co-authored-by: OpenIM-Gordon <46924906+FGadvancer@users.noreply.github.com>
Co-authored-by: icey-yu <1186114839@qq.com>

internal/msggateway/client.go

@@ -75,8 +75,8 @@ type Client struct {
 	token          string
 	hbCtx          context.Context
 	hbCancel       context.CancelFunc
-	subLock        sync.Mutex
-	subUserIDs     map[string]struct{}
+	subLock        *sync.Mutex
+	subUserIDs     map[string]struct{} // client conn subscription list
 }
 
 // ResetClient updates the client's state with new connection and context information.
@@ -94,6 +94,11 @@ func (c *Client) ResetClient(ctx *UserConnContext, conn LongConn, longConnServer
 	c.closedErr = nil
 	c.token = ctx.GetToken()
 	c.hbCtx, c.hbCancel = context.WithCancel(c.ctx)
+	c.subLock = new(sync.Mutex)
+	if c.subUserIDs != nil {
+		clear(c.subUserIDs)
+	}
+	c.subUserIDs = make(map[string]struct{})
 }
 
 func (c *Client) pingHandler(appData string) error {
@@ -246,13 +251,11 @@ func (c *Client) setAppBackgroundStatus(ctx context.Context, req *Req) ([]byte,
 }
 
 func (c *Client) close() {
+	c.w.Lock()
+	defer c.w.Unlock()
 	if c.closed.Load() {
 		return
 	}
-
-	c.w.Lock()
-	defer c.w.Unlock()
-
 	c.closed.Store(true)
 	c.conn.Close()
 	c.hbCancel() // Close server-initiated heartbeat.
@@ -313,6 +316,14 @@ func (c *Client) KickOnlineMessage() error {
 	return err
 }
 
+func (c *Client) PushUserOnlineStatus(data []byte) error {
+	resp := Resp{
+		ReqIdentifier: WsSubUserOnlineStatus,
+		Data:          data,
+	}
+	return c.writeBinaryMsg(resp)
+}
+
 func (c *Client) writeBinaryMsg(resp Resp) error {
 	if c.closed.Load() {
 		return nil

internal/msggateway/subscription.go

@@ -2,15 +2,11 @@ package msggateway
 
 import (
 	"context"
-	"encoding/json"
-	"github.com/openimsdk/protocol/constant"
 	"github.com/openimsdk/protocol/sdkws"
 	"github.com/openimsdk/tools/log"
 	"github.com/openimsdk/tools/utils/datautil"
-	"github.com/openimsdk/tools/utils/idutil"
 	"google.golang.org/protobuf/proto"
 	"sync"
-	"time"
 )
 
 func (ws *WsServer) subscriberUserOnlineStatusChanges(ctx context.Context, userID string, platformIDs []int32) {
@@ -45,33 +41,19 @@ func (ws *WsServer) SubUserOnlineStatus(ctx context.Context, client *Client, dat
 	return proto.Marshal(&resp)
 }
 
-type subClient struct {
-	clients map[string]*Client
-}
-
 func newSubscription() *Subscription {
 	return &Subscription{
 		userIDs: make(map[string]*subClient),
 	}
 }
 
-type Subscription struct {
-	lock    sync.RWMutex
-	userIDs map[string]*subClient
+type subClient struct {
+	clients map[string]*Client
 }
 
-func (s *Subscription) GetClient(userID string) []*Client {
-	s.lock.RLock()
-	defer s.lock.RUnlock()
-	cs, ok := s.userIDs[userID]
-	if !ok {
-		return nil
-	}
-	clients := make([]*Client, 0, len(cs.clients))
-	for _, client := range cs.clients {
-		clients = append(clients, client)
-	}
-	return clients
+type Subscription struct {
+	lock    sync.RWMutex
+	userIDs map[string]*subClient // subscribe to the user's client connection
 }
 
 func (s *Subscription) DelClient(client *Client) {
@@ -99,6 +81,20 @@ func (s *Subscription) DelClient(client *Client) {
 	}
 }
 
+func (s *Subscription) GetClient(userID string) []*Client {
+	s.lock.RLock()
+	defer s.lock.RUnlock()
+	cs, ok := s.userIDs[userID]
+	if !ok {
+		return nil
+	}
+	clients := make([]*Client, 0, len(cs.clients))
+	for _, client := range cs.clients {
+		clients = append(clients, client)
+	}
+	return clients
+}
+
 func (s *Subscription) Sub(client *Client, addUserIDs, delUserIDs []string) {
 	if len(addUserIDs)+len(delUserIDs) == 0 {
 		return
@@ -121,6 +117,7 @@ func (s *Subscription) Sub(client *Client, addUserIDs, delUserIDs []string) {
 			continue
 		}
 		client.subUserIDs[userID] = struct{}{}
+		add[userID] = struct{}{}
 	}
 	client.subLock.Unlock()
 	if len(del)+len(add) == 0 {
@@ -154,28 +151,16 @@ func (ws *WsServer) pushUserIDOnlineStatus(ctx context.Context, userID string, p
 	if len(clients) == 0 {
 		return
 	}
-	msgContent, err := json.Marshal(platformIDs)
+	onlineStatus, err := proto.Marshal(&sdkws.SubUserOnlineStatusTips{
+		Subscribers: []*sdkws.SubUserOnlineStatusElem{{UserID: userID, OnlinePlatformIDs: platformIDs}},
+	})
 	if err != nil {
 		log.ZError(ctx, "pushUserIDOnlineStatus json.Marshal", err)
 		return
 	}
-	now := time.Now().UnixMilli()
-	msgID := idutil.GetMsgIDByMD5(userID)
-	msg := &sdkws.MsgData{
-		SendID:           userID,
-		ClientMsgID:      msgID,
-		ServerMsgID:      msgID,
-		SenderPlatformID: constant.AdminPlatformID,
-		SessionType:      constant.NotificationChatType,
-		ContentType:      constant.UserSubscribeOnlineStatusNotification,
-		Content:          msgContent,
-		SendTime:         now,
-		CreateTime:       now,
-	}
 	for _, client := range clients {
-		msg.RecvID = client.UserID
-		if err := client.PushMessage(ctx, msg); err != nil {
-			log.ZError(ctx, "UserSubscribeOnlineStatusNotification push failed", err, "userID", client.UserID, "platformID", client.PlatformID, "changeUserID", userID, "content", msgContent)
+		if err := client.PushUserOnlineStatus(onlineStatus); err != nil {
+			log.ZError(ctx, "UserSubscribeOnlineStatusNotification push failed", err, "userID", client.UserID, "platformID", client.PlatformID, "changeUserID", userID, "changePlatformID", platformIDs)
 		}
 	}
 }

internal/msggateway/ws_server.go

@@ -358,9 +358,7 @@ func (ws *WsServer) unregisterClient(client *Client) {
 		prommetrics.OnlineUserGauge.Dec()
 	}
 	ws.onlineUserConnNum.Add(-1)
-	client.subLock.Lock()
-	clear(client.subUserIDs)
-	client.subLock.Unlock()
+	ws.subscription.DelClient(client)
 	//ws.SetUserOnlineStatus(client.ctx, client, constant.Offline)
 	log.ZInfo(client.ctx, "user offline", "close reason", client.closedErr, "online user Num",
 		ws.onlineUserNum.Load(), "online user conn Num",

internal/rpc/friend/notification.go

@@ -16,6 +16,8 @@ package friend
 
 import (
 	"context"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/versionctx"
 
 	relationtb "github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
 
@@ -191,10 +193,37 @@ func (f *FriendNotificationSender) FriendDeletedNotification(ctx context.Context
 	f.Notification(ctx, req.OwnerUserID, req.FriendUserID, constant.FriendDeletedNotification, &tips)
 }
 
+func (f *FriendNotificationSender) setVersion(ctx context.Context, version *uint64, versionID *string, collName string, id string) {
+	versions := versionctx.GetVersionLog(ctx).Get()
+	for _, coll := range versions {
+		if coll.Name == collName && coll.Doc.DID == id {
+			*version = uint64(coll.Doc.Version)
+			*versionID = coll.Doc.ID.Hex()
+			return
+		}
+	}
+}
+
+func (f *FriendNotificationSender) setSortVersion(ctx context.Context, version *uint64, versionID *string, collName string, id string, sortVersion *uint64) {
+	versions := versionctx.GetVersionLog(ctx).Get()
+	for _, coll := range versions {
+		if coll.Name == collName && coll.Doc.DID == id {
+			*version = uint64(coll.Doc.Version)
+			*versionID = coll.Doc.ID.Hex()
+			for _, elem := range coll.Doc.Logs {
+				if elem.EID == relationtb.VersionSortChangeID {
+					*sortVersion = uint64(elem.Version)
+				}
+			}
+		}
+	}
+}
+
 func (f *FriendNotificationSender) FriendRemarkSetNotification(ctx context.Context, fromUserID, toUserID string) {
 	tips := sdkws.FriendInfoChangedTips{FromToUserID: &sdkws.FromToUserID{}}
 	tips.FromToUserID.FromUserID = fromUserID
 	tips.FromToUserID.ToUserID = toUserID
+	f.setSortVersion(ctx, &tips.FriendVersion, &tips.FriendVersionID, database.FriendVersionName, toUserID, &tips.FriendSortVersion)
 	f.Notification(ctx, fromUserID, toUserID, constant.FriendRemarkSetNotification, &tips)
 }
 

internal/rpc/friend/sync.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"github.com/openimsdk/open-im-server/v3/pkg/util/hashutil"
 	"github.com/openimsdk/protocol/sdkws"
+	"slices"
 
 	"github.com/openimsdk/open-im-server/v3/internal/rpc/incrversion"
 	"github.com/openimsdk/open-im-server/v3/pkg/authverify"
@@ -52,12 +53,27 @@ func (s *friendServer) GetIncrementalFriends(ctx context.Context, req *relation.
 	if err := authverify.CheckAccessV3(ctx, req.UserID, s.config.Share.IMAdminUserID); err != nil {
 		return nil, err
 	}
+	var sortVersion uint64
 	opt := incrversion.Option[*sdkws.FriendInfo, relation.GetIncrementalFriendsResp]{
-		Ctx:             ctx,
-		VersionKey:      req.UserID,
-		VersionID:       req.VersionID,
-		VersionNumber:   req.Version,
-		Version:         s.db.FindFriendIncrVersion,
+		Ctx:           ctx,
+		VersionKey:    req.UserID,
+		VersionID:     req.VersionID,
+		VersionNumber: req.Version,
+		Version: func(ctx context.Context, ownerUserID string, version uint, limit int) (*model.VersionLog, error) {
+			vl, err := s.db.FindFriendIncrVersion(ctx, ownerUserID, version, limit)
+			if err != nil {
+				return nil, err
+			}
+			vl.Logs = slices.DeleteFunc(vl.Logs, func(elem model.VersionLogElem) bool {
+				if elem.EID == model.VersionSortChangeID {
+					vl.LogLen--
+					sortVersion = uint64(elem.Version)
+					return true
+				}
+				return false
+			})
+			return vl, nil
+		},
 		CacheMaxVersion: s.db.FindMaxFriendVersionCache,
 		Find: func(ctx context.Context, ids []string) ([]*sdkws.FriendInfo, error) {
 			return s.getFriend(ctx, req.UserID, ids)
@@ -65,12 +81,13 @@ func (s *friendServer) GetIncrementalFriends(ctx context.Context, req *relation.
 		ID: func(elem *sdkws.FriendInfo) string { return elem.FriendUser.UserID },
 		Resp: func(version *model.VersionLog, deleteIds []string, insertList, updateList []*sdkws.FriendInfo, full bool) *relation.GetIncrementalFriendsResp {
 			return &relation.GetIncrementalFriendsResp{
-				VersionID: version.ID.Hex(),
-				Version:   uint64(version.Version),
-				Full:      full,
-				Delete:    deleteIds,
-				Insert:    insertList,
-				Update:    updateList,
+				VersionID:   version.ID.Hex(),
+				Version:     uint64(version.Version),
+				Full:        full,
+				Delete:      deleteIds,
+				Insert:      insertList,
+				Update:      updateList,
+				SortVersion: sortVersion,
 			}
 		},
 	}

internal/rpc/group/notification.go

@@ -306,6 +306,21 @@ func (g *GroupNotificationSender) setVersion(ctx context.Context, version *uint6
 	}
 }
 
+func (g *GroupNotificationSender) setSortVersion(ctx context.Context, version *uint64, versionID *string, collName string, id string, sortVersion *uint64) {
+	versions := versionctx.GetVersionLog(ctx).Get()
+	for _, coll := range versions {
+		if coll.Name == collName && coll.Doc.DID == id {
+			*version = uint64(coll.Doc.Version)
+			*versionID = coll.Doc.ID.Hex()
+			for _, elem := range coll.Doc.Logs {
+				if elem.EID == model.VersionSortChangeID {
+					*sortVersion = uint64(elem.Version)
+				}
+			}
+		}
+	}
+}
+
 func (g *GroupNotificationSender) GroupCreatedNotification(ctx context.Context, tips *sdkws.GroupCreatedTips) {
 	var err error
 	defer func() {
@@ -707,7 +722,7 @@ func (g *GroupNotificationSender) GroupMemberInfoSetNotification(ctx context.Con
 	if err = g.fillOpUser(ctx, &tips.OpUser, tips.Group.GroupID); err != nil {
 		return
 	}
-	g.setVersion(ctx, &tips.GroupMemberVersion, &tips.GroupMemberVersionID, database.GroupMemberVersionName, tips.Group.GroupID)
+	g.setSortVersion(ctx, &tips.GroupMemberVersion, &tips.GroupMemberVersionID, database.GroupMemberVersionName, tips.Group.GroupID, &tips.GroupSortVersion)
 	g.Notification(ctx, mcontext.GetOpUserID(ctx), group.GroupID, constant.GroupMemberInfoSetNotification, tips)
 }
 

internal/rpc/group/sync.go

@@ -10,9 +10,13 @@ import (
 	"github.com/openimsdk/protocol/constant"
 	pbgroup "github.com/openimsdk/protocol/group"
 	"github.com/openimsdk/protocol/sdkws"
-	"slices"
 )
 
+func (s *groupServer) BatchGetIncrementalGroupMember(ctx context.Context, req *pbgroup.BatchGetIncrementalGroupMemberReq) (*pbgroup.BatchGetIncrementalGroupMemberResp, error) {
+	//TODO implement me
+	panic("implement me")
+}
+
 func (s *groupServer) GetFullGroupMemberUserIDs(ctx context.Context, req *pbgroup.GetFullGroupMemberUserIDsReq) (*pbgroup.GetFullGroupMemberUserIDsResp, error) {
 	vl, err := s.db.FindMaxGroupMemberVersionCache(ctx, req.GroupID)
 	if err != nil {
@@ -63,7 +67,10 @@ func (s *groupServer) GetIncrementalGroupMember(ctx context.Context, req *pbgrou
 	if group.Status == constant.GroupStatusDismissed {
 		return nil, servererrs.ErrDismissedAlready.Wrap()
 	}
-	var hasGroupUpdate bool
+	var (
+		hasGroupUpdate bool
+		sortVersion    uint64
+	)
 	opt := incrversion.Option[*sdkws.GroupMemberFullInfo, pbgroup.GetIncrementalGroupMemberResp]{
 		Ctx:           ctx,
 		VersionKey:    req.GroupID,
@@ -74,14 +81,20 @@ func (s *groupServer) GetIncrementalGroupMember(ctx context.Context, req *pbgrou
 			if err != nil {
 				return nil, err
 			}
-			vl.Logs = slices.DeleteFunc(vl.Logs, func(elem model.VersionLogElem) bool {
-				if elem.EID == "" {
+			logs := make([]model.VersionLogElem, 0, len(vl.Logs))
+			for i, log := range vl.Logs {
+				switch log.EID {
+				case model.VersionGroupChangeID:
 					vl.LogLen--
 					hasGroupUpdate = true
-					return true
+				case model.VersionSortChangeID:
+					vl.LogLen--
+					sortVersion = uint64(log.Version)
+				default:
+					logs = append(logs, vl.Logs[i])
 				}
-				return false
-			})
+			}
+			vl.Logs = logs
 			if vl.LogLen > 0 {
 				hasGroupUpdate = true
 			}
@@ -94,12 +107,13 @@ func (s *groupServer) GetIncrementalGroupMember(ctx context.Context, req *pbgrou
 		ID: func(elem *sdkws.GroupMemberFullInfo) string { return elem.UserID },
 		Resp: func(version *model.VersionLog, delIDs []string, insertList, updateList []*sdkws.GroupMemberFullInfo, full bool) *pbgroup.GetIncrementalGroupMemberResp {
 			return &pbgroup.GetIncrementalGroupMemberResp{
-				VersionID: version.ID.Hex(),
-				Version:   uint64(version.Version),
-				Full:      full,
-				Delete:    delIDs,
-				Insert:    insertList,
-				Update:    updateList,
+				VersionID:   version.ID.Hex(),
+				Version:     uint64(version.Version),
+				Full:        full,
+				Delete:      delIDs,
+				Insert:      insertList,
+				Update:      updateList,
+				SortVersion: sortVersion,
 			}
 		},
 	}

internal/rpc/user/online.go

@@ -3,7 +3,6 @@ package user
 import (
 	"context"
 	"github.com/openimsdk/protocol/constant"
-	"github.com/openimsdk/protocol/sdkws"
 	pbuser "github.com/openimsdk/protocol/user"
 )
 
@@ -38,23 +37,6 @@ func (s *userServer) getUsersOnlineStatus(ctx context.Context, userIDs []string)
 
 // SubscribeOrCancelUsersStatus Subscribe online or cancel online users.
 func (s *userServer) SubscribeOrCancelUsersStatus(ctx context.Context, req *pbuser.SubscribeOrCancelUsersStatusReq) (*pbuser.SubscribeOrCancelUsersStatusResp, error) {
-	if req.Genre == constant.SubscriberUser {
-		err := s.db.SubscribeUsersStatus(ctx, req.UserID, req.UserIDs)
-		if err != nil {
-			return nil, err
-		}
-		var status []*pbuser.OnlineStatus
-		status, err = s.getUsersOnlineStatus(ctx, req.UserIDs)
-		if err != nil {
-			return nil, err
-		}
-		return &pbuser.SubscribeOrCancelUsersStatusResp{StatusList: status}, nil
-	} else if req.Genre == constant.Unsubscribe {
-		err := s.db.UnsubscribeUsersStatus(ctx, req.UserID, req.UserIDs)
-		if err != nil {
-			return nil, err
-		}
-	}
 	return &pbuser.SubscribeOrCancelUsersStatusResp{}, nil
 }
 
@@ -82,34 +64,12 @@ func (s *userServer) SetUserStatus(ctx context.Context, req *pbuser.SetUserStatu
 	if err := s.online.SetUserOnline(ctx, req.UserID, online, offline); err != nil {
 		return nil, err
 	}
-	list, err := s.db.GetSubscribedList(ctx, req.UserID)
-	if err != nil {
-		return nil, err
-	}
-	for _, userID := range list {
-		tips := &sdkws.UserStatusChangeTips{
-			FromUserID: req.UserID,
-			ToUserID:   userID,
-			Status:     req.Status,
-			PlatformID: req.PlatformID,
-		}
-		s.userNotificationSender.UserStatusChangeNotification(ctx, tips)
-	}
-
 	return &pbuser.SetUserStatusResp{}, nil
 }
 
 // GetSubscribeUsersStatus Get the online status of subscribers.
 func (s *userServer) GetSubscribeUsersStatus(ctx context.Context, req *pbuser.GetSubscribeUsersStatusReq) (*pbuser.GetSubscribeUsersStatusResp, error) {
-	userList, err := s.db.GetAllSubscribeList(ctx, req.UserID)
-	if err != nil {
-		return nil, err
-	}
-	onlineStatusList, err := s.getUsersOnlineStatus(ctx, userList)
-	if err != nil {
-		return nil, err
-	}
-	return &pbuser.GetSubscribeUsersStatusResp{StatusList: onlineStatusList}, nil
+	return &pbuser.GetSubscribeUsersStatusResp{}, nil
 }
 
 func (s *userServer) SetUserOnlineStatus(ctx context.Context, req *pbuser.SetUserOnlineStatusReq) (*pbuser.SetUserOnlineStatusResp, error) {

internal/rpc/user/user.go

@@ -93,8 +93,7 @@ func Start(ctx context.Context, config *Config, client registry.SvcDiscoveryRegi
 		return err
 	}
 	userCache := redis.NewUserCacheRedis(rdb, &config.LocalCacheConfig, userDB, redis.GetRocksCacheOptions())
-	userMongoDB := mgo.NewUserMongoDriver(mgocli.GetDB())
-	database := controller.NewUserDatabase(userDB, userCache, mgocli.GetTx(), userMongoDB)
+	database := controller.NewUserDatabase(userDB, userCache, mgocli.GetTx())
 	friendRpcClient := rpcclient.NewFriendRpcClient(client, config.Share.RpcRegisterName.Friend)
 	groupRpcClient := rpcclient.NewGroupRpcClient(client, config.Share.RpcRegisterName.Group)
 	msgRpcClient := rpcclient.NewMessageRpcClient(client, config.Share.RpcRegisterName.Msg)