fix: in standalone mode, the user online status is wrong

internal/msggateway/ws_server.go

@@ -49,7 +49,7 @@ type WsServer struct {
 	unregisterChan    chan *Client
 	kickHandlerChan   chan *kickHandler
 	clients           UserMap
-	online            *rpccache.OnlineCache
+	online            rpccache.OnlineCache
 	subscription      *Subscription
 	clientPool        sync.Pool
 	onlineUserNum     atomic.Int64

internal/msgtransfer/online_msg_to_mongo_handler.go

@@ -48,15 +48,7 @@ func (mc *OnlineHistoryMongoConsumerHandler) HandleChatWs2Mongo(ctx context.Cont
 	log.ZDebug(ctx, "mongo consumer recv msg", "msgs", msgFromMQ.String())
 	err = mc.msgTransferDatabase.BatchInsertChat2DB(ctx, msgFromMQ.ConversationID, msgFromMQ.MsgData, msgFromMQ.LastSeq)
 	if err != nil {
-		log.ZError(
+		log.ZError(ctx, "single data insert to mongo err", err, "msg", msgFromMQ.MsgData, "conversationID", msgFromMQ.ConversationID)
-			ctx,
-			"single data insert to mongo err",
-			err,
-			"msg",
-			msgFromMQ.MsgData,
-			"conversationID",
-			msgFromMQ.ConversationID,
-		)
 		prommetrics.MsgInsertMongoFailedCounter.Inc()
 	} else {
 		prommetrics.MsgInsertMongoSuccessCounter.Inc()

internal/push/push_handler.go

@@ -33,7 +33,7 @@ type ConsumerHandler struct {
 	offlinePusher          offlinepush.OfflinePusher
 	onlinePusher           OnlinePusher
 	pushDatabase           controller.PushDatabase
-	onlineCache            *rpccache.OnlineCache
+	onlineCache            rpccache.OnlineCache
 	groupLocalCache        *rpccache.GroupLocalCache
 	conversationLocalCache *rpccache.ConversationLocalCache
 	webhookClient          *webhook.Client
@@ -120,11 +120,7 @@ func (c *ConsumerHandler) HandleMs2PsChat(ctx context.Context, msg []byte) {
 }
 
 func (c *ConsumerHandler) WaitCache() {
-	c.onlineCache.Lock.Lock()
+	c.onlineCache.WaitCache()
-	for c.onlineCache.CurrentPhase.Load() < rpccache.DoSubscribeOver {
-		c.onlineCache.Cond.Wait()
-	}
-	c.onlineCache.Lock.Unlock()
 }
 
 // Push2User Suitable for two types of conversations, one is SingleChatType and the other is NotificationChatType.

pkg/rpccache/online.go

@@ -9,6 +9,7 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"github.com/openimsdk/open-im-server/v3/pkg/rpcli"
 	"github.com/openimsdk/protocol/constant"
 	"github.com/openimsdk/protocol/user"
@@ -23,9 +24,25 @@ import (
 	"github.com/redis/go-redis/v9"
 )
 
-func NewOnlineCache(client *rpcli.UserClient, group *GroupLocalCache, rdb redis.UniversalClient, fullUserCache bool, fn func(ctx context.Context, userID string, platformIDs []int32)) (*OnlineCache, error) {
+const (
+	Begin uint32 = iota
+	DoOnlineStatusOver
+	DoSubscribeOver
+)
+
+type OnlineCache interface {
+	GetUserOnlinePlatform(ctx context.Context, userID string) ([]int32, error)
+	GetUserOnline(ctx context.Context, userID string) (bool, error)
+	GetUsersOnline(ctx context.Context, userIDs []string) ([]string, []string, error)
+	WaitCache()
+}
+
+func NewOnlineCache(client *rpcli.UserClient, group *GroupLocalCache, rdb redis.UniversalClient, fullUserCache bool, fn func(ctx context.Context, userID string, platformIDs []int32)) (OnlineCache, error) {
+	if config.Standalone() {
+		return disableOnlineCache{client: client}, nil
+	}
 	l := &sync.Mutex{}
-	x := &OnlineCache{
+	x := &defaultOnlineCache{
 		client:        client,
 		group:         group,
 		fullUserCache: fullUserCache,
@@ -60,13 +77,7 @@ func NewOnlineCache(client *rpcli.UserClient, group *GroupLocalCache, rdb redis.
 	return x, nil
 }
 
-const (
+type defaultOnlineCache struct {
-	Begin uint32 = iota
-	DoOnlineStatusOver
-	DoSubscribeOver
-)
-
-type OnlineCache struct {
 	client *rpcli.UserClient
 	group  *GroupLocalCache
 
@@ -82,7 +93,7 @@ type OnlineCache struct {
 	CurrentPhase atomic.Uint32
 }
 
-func (o *OnlineCache) initUsersOnlineStatus(ctx context.Context) (err error) {
+func (o *defaultOnlineCache) initUsersOnlineStatus(ctx context.Context) (err error) {
 	log.ZDebug(ctx, "init users online status begin")
 
 	var (
@@ -135,7 +146,7 @@ func (o *OnlineCache) initUsersOnlineStatus(ctx context.Context) (err error) {
 	return nil
 }
 
-func (o *OnlineCache) doSubscribe(ctx context.Context, rdb redis.UniversalClient, fn func(ctx context.Context, userID string, platformIDs []int32)) {
+func (o *defaultOnlineCache) doSubscribe(ctx context.Context, rdb redis.UniversalClient, fn func(ctx context.Context, userID string, platformIDs []int32)) {
 	o.Lock.Lock()
 	ch := rdb.Subscribe(ctx, cachekey.OnlineChannel).Channel()
 	for o.CurrentPhase.Load() < DoOnlineStatusOver {
@@ -186,7 +197,7 @@ func (o *OnlineCache) doSubscribe(ctx context.Context, rdb redis.UniversalClient
 	}
 }
 
-func (o *OnlineCache) getUserOnlinePlatform(ctx context.Context, userID string) ([]int32, error) {
+func (o *defaultOnlineCache) getUserOnlinePlatform(ctx context.Context, userID string) ([]int32, error) {
 	platformIDs, err := o.lruCache.Get(userID, func() ([]int32, error) {
 		return o.client.GetUserOnlinePlatform(ctx, userID)
 	})
@@ -198,7 +209,7 @@ func (o *OnlineCache) getUserOnlinePlatform(ctx context.Context, userID string)
 	return platformIDs, nil
 }
 
-func (o *OnlineCache) GetUserOnlinePlatform(ctx context.Context, userID string) ([]int32, error) {
+func (o *defaultOnlineCache) GetUserOnlinePlatform(ctx context.Context, userID string) ([]int32, error) {
 	platformIDs, err := o.getUserOnlinePlatform(ctx, userID)
 	if err != nil {
 		return nil, err
@@ -208,7 +219,7 @@ func (o *OnlineCache) GetUserOnlinePlatform(ctx context.Context, userID string)
 	return platformIDs, nil
 }
 
-func (o *OnlineCache) GetUserOnline(ctx context.Context, userID string) (bool, error) {
+func (o *defaultOnlineCache) GetUserOnline(ctx context.Context, userID string) (bool, error) {
 	platformIDs, err := o.getUserOnlinePlatform(ctx, userID)
 	if err != nil {
 		return false, err
@@ -216,7 +227,7 @@ func (o *OnlineCache) GetUserOnline(ctx context.Context, userID string) (bool, e
 	return len(platformIDs) > 0, nil
 }
 
-func (o *OnlineCache) getUserOnlinePlatformBatch(ctx context.Context, userIDs []string) (map[string][]int32, error) {
+func (o *defaultOnlineCache) getUserOnlinePlatformBatch(ctx context.Context, userIDs []string) (map[string][]int32, error) {
 	if len(userIDs) == 0 {
 		return nil, nil
 	}
@@ -240,7 +251,7 @@ func (o *OnlineCache) getUserOnlinePlatformBatch(ctx context.Context, userIDs []
 	return platformIDsMap, nil
 }
 
-func (o *OnlineCache) GetUsersOnline(ctx context.Context, userIDs []string) ([]string, []string, error) {
+func (o *defaultOnlineCache) GetUsersOnline(ctx context.Context, userIDs []string) ([]string, []string, error) {
 	t := time.Now()
 
 	var (
@@ -276,7 +287,7 @@ func (o *OnlineCache) GetUsersOnline(ctx context.Context, userIDs []string) ([]s
 	return onlineUserIDs, offlineUserIDs, nil
 }
 
-func (o *OnlineCache) setUserOnline(userID string, platformIDs []int32) {
+func (o *defaultOnlineCache) setUserOnline(userID string, platformIDs []int32) {
 	switch o.fullUserCache {
 	case true:
 		o.mapCache.Store(userID, platformIDs)
@@ -285,6 +296,51 @@ func (o *OnlineCache) setUserOnline(userID string, platformIDs []int32) {
 	}
 }
 
-func (o *OnlineCache) setHasUserOnline(userID string, platformIDs []int32) bool {
+func (o *defaultOnlineCache) setHasUserOnline(userID string, platformIDs []int32) bool {
 	return o.lruCache.SetHas(userID, platformIDs)
 }
+
+func (o *defaultOnlineCache) WaitCache() {
+	o.Lock.Lock()
+	for o.CurrentPhase.Load() < DoSubscribeOver {
+		o.Cond.Wait()
+	}
+	o.Lock.Unlock()
+}
+
+type disableOnlineCache struct {
+	client *rpcli.UserClient
+}
+
+func (o disableOnlineCache) GetUserOnlinePlatform(ctx context.Context, userID string) ([]int32, error) {
+	return o.client.GetUserOnlinePlatform(ctx, userID)
+}
+
+func (o disableOnlineCache) GetUserOnline(ctx context.Context, userID string) (bool, error) {
+	onlinePlatform, err := o.client.GetUserOnlinePlatform(ctx, userID)
+	if err != nil {
+		return false, err
+	}
+	return len(onlinePlatform) > 0, err
+}
+
+func (o disableOnlineCache) GetUsersOnline(ctx context.Context, userIDs []string) ([]string, []string, error) {
+	var (
+		onlineUserIDs  = make([]string, 0, len(userIDs))
+		offlineUserIDs = make([]string, 0, len(userIDs))
+	)
+	for _, userID := range userIDs {
+		online, err := o.GetUserOnline(ctx, userID)
+		if err != nil {
+			return nil, nil, err
+		}
+		if online {
+			onlineUserIDs = append(onlineUserIDs, userID)
+		} else {
+			offlineUserIDs = append(offlineUserIDs, userID)
+		}
+	}
+	return onlineUserIDs, offlineUserIDs, nil
+}
+
+func (o disableOnlineCache) WaitCache() {}