fix: online status renewal (#2468)

* 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

* version log index

* batch push

* batch push

* seq void filling

* fix: batchGetMaxSeq

* fix: batchGetMaxSeq

* cache db error log

* 111

* fix bug

* fix: ImportFriends

* add online cache

* add some logs

* add some logs

* fix: onlineUserIDs

* add logs

* test

* test

* test

* test

---------

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/online.go

@@ -4,13 +4,16 @@ import (
 	"context"
 	"crypto/md5"
 	"encoding/binary"
+	"fmt"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/cachekey"
 	pbuser "github.com/openimsdk/protocol/user"
 	"github.com/openimsdk/tools/log"
 	"github.com/openimsdk/tools/mcontext"
 	"github.com/openimsdk/tools/utils/datautil"
 	"math/rand"
+	"os"
 	"strconv"
+	"sync/atomic"
 	"time"
 )
 
@@ -78,8 +81,10 @@ func (ws *WsServer) ChangeOnlineStatus(concurrent int) {
 		}
 	}
 
-	opIdCtx := mcontext.SetOperationID(context.Background(), "r"+strconv.FormatUint(rNum, 10))
+	var count atomic.Int64
+	operationIDPrefix := fmt.Sprintf("p_%d_", os.Getpid())
 	doRequest := func(req *pbuser.SetUserOnlineStatusReq) {
+		opIdCtx := mcontext.SetOperationID(context.Background(), operationIDPrefix+strconv.FormatInt(count.Add(1), 10))
 		ctx, cancel := context.WithTimeout(opIdCtx, time.Second*5)
 		defer cancel()
 		if _, err := ws.userClient.Client.SetUserOnlineStatus(ctx, req); err != nil {
@@ -102,7 +107,7 @@ func (ws *WsServer) ChangeOnlineStatus(concurrent int) {
 		case now := <-renewalTicker.C:
 			deadline := now.Add(-cachekey.OnlineExpire / 3)
 			users := ws.clients.GetAllUserStatus(deadline, now)
-			log.ZDebug(context.Background(), "renewal ticker", "deadline", deadline, "nowtime", now, "num", len(users))
+			log.ZDebug(context.Background(), "renewal ticker", "deadline", deadline, "nowtime", now, "num", len(users), "users", users)
 			pushUserState(users...)
 		case state := <-ws.clients.UserState():
 			log.ZDebug(context.Background(), "OnlineCache user online change", "userID", state.UserID, "online", state.Online, "offline", state.Offline)

internal/msggateway/user_map.go

@@ -1,6 +1,10 @@
 package msggateway
 
 import (
+	"context"
+	"fmt"
+	"github.com/openimsdk/protocol/constant"
+	"github.com/openimsdk/tools/log"
 	"github.com/openimsdk/tools/utils/datautil"
 	"sync"
 	"time"
@@ -117,6 +121,7 @@ func (u *userMap) Get(userID string, platformID int) ([]*Client, bool, bool) {
 }
 
 func (u *userMap) Set(userID string, client *Client) {
+	log.ZDebug(context.Background(), "userMap Set", "userID", userID, "platformID", client.PlatformID, "platform", constant.PlatformIDToName(client.PlatformID), "pointer", fmt.Sprintf("%p", client))
 	u.lock.Lock()
 	defer u.lock.Unlock()
 	result, ok := u.data[userID]
@@ -162,12 +167,12 @@ func (u *userMap) DeleteClients(userID string, clients []*Client) (isDeleteUser
 	return true
 }
 
-func (u *userMap) GetAllUserStatus(deadline time.Time, nowtime time.Time) []UserState {
+func (u *userMap) GetAllUserStatus(deadline time.Time, nowtime time.Time) (result []UserState) {
 	u.lock.RLock()
 	defer u.lock.RUnlock()
-	result := make([]UserState, 0, len(u.data))
+	result = make([]UserState, 0, len(u.data))
 	for userID, userPlatform := range u.data {
-		if userPlatform.Time.Before(deadline) {
+		if deadline.Before(userPlatform.Time) {
 			continue
 		}
 		userPlatform.Time = nowtime

internal/push/push_handler.go

@@ -202,7 +202,7 @@ func (c *ConsumerHandler) GetConnsAndOnlinePush(ctx context.Context, msg *sdkws.
 	var result []*msggateway.SingleMsgToUserResults
 	if len(onlineUserIDs) > 0 {
 		var err error
-		result, err = c.onlinePusher.GetConnsAndOnlinePush(ctx, msg, pushToUserIDs)
+		result, err = c.onlinePusher.GetConnsAndOnlinePush(ctx, msg, onlineUserIDs)
 		if err != nil {
 			return nil, err
 		}