Merge branch 'fix-err' of github.com:icey-yu/open-im-server into fix-err

pkg/common/config/config.go

@@ -220,6 +220,7 @@ type Push struct {
 		BadgeCount bool   `mapstructure:"badgeCount"`
 		Production bool   `mapstructure:"production"`
 	} `mapstructure:"iosPush"`
+	FullUserCache bool `mapstructure:"fullUserCache"`
 }
 
 type Auth struct {

pkg/localcache/lru/lru.go

@@ -20,6 +20,7 @@ type EvictCallback[K comparable, V any] simplelru.EvictCallback[K, V]
 
 type LRU[K comparable, V any] interface {
 	Get(key K, fetch func() (V, error)) (V, error)
+	Set(key K, value V)
 	GetBatch(key []K, fetchBatch func([]K) (map[string]V, error)) (map[string]V, error)
 	SetHas(key K, value V) bool
 	Del(key K) bool

pkg/localcache/lru/lru_expiration.go

@@ -99,5 +99,11 @@ func (x *ExpirationLRU[K, V]) SetHas(key K, value V) bool {
 	return false
 }
 
+func (x *ExpirationLRU[K, V]) Set(key K, value V) {
+	x.lock.Lock()
+	defer x.lock.Unlock()
+	x.core.Add(key, &expirationLruItem[V]{value: value})
+}
+
 func (x *ExpirationLRU[K, V]) Stop() {
 }

pkg/localcache/lru/lru_lazy.go

@@ -176,6 +176,12 @@ func (x *LayLRU[K, V]) SetHasBatch(data map[K]V) {
 //	return x.core.Contains(key)
 //}
 
+func (x *LayLRU[K, V]) Set(key K, value V) {
+	x.lock.Lock()
+	defer x.lock.Unlock()
+	x.core.Add(key, &layLruItem[V]{value: value, expires: time.Now().Add(x.successTTL).UnixMilli()})
+}
+
 func (x *LayLRU[K, V]) SetHas(key K, value V) bool {
 	x.lock.Lock()
 	defer x.lock.Unlock()

pkg/localcache/lru/lru_slot.go

@@ -40,6 +40,10 @@ func (x *slotLRU[K, V]) Get(key K, fetch func() (V, error)) (V, error) {
 	return x.slots[x.getIndex(key)].Get(key, fetch)
 }
 
+func (x *slotLRU[K, V]) Set(key K, value V) {
+	x.slots[x.getIndex(key)].Set(key, value)
+}
+
 func (x *slotLRU[K, V]) SetHas(key K, value V) bool {
 	return x.slots[x.getIndex(key)].SetHas(key, value)
 }

pkg/rpccache/online.go

@@ -11,45 +11,112 @@ import (
 	"github.com/openimsdk/open-im-server/v3/pkg/localcache/lru"
 	"github.com/openimsdk/open-im-server/v3/pkg/rpcclient"
 	"github.com/openimsdk/open-im-server/v3/pkg/util/useronline"
+	"github.com/openimsdk/protocol/constant"
+	"github.com/openimsdk/tools/db/cacheutil"
 	"github.com/openimsdk/tools/log"
 	"github.com/openimsdk/tools/mcontext"
 	"github.com/redis/go-redis/v9"
 )
 
-func NewOnlineCache(user rpcclient.UserRpcClient, group *GroupLocalCache, rdb redis.UniversalClient, fn func(ctx context.Context, userID string, platformIDs []int32)) *OnlineCache {
+func NewOnlineCache(user rpcclient.UserRpcClient, group *GroupLocalCache, rdb redis.UniversalClient, fullUserCache bool, fn func(ctx context.Context, userID string, platformIDs []int32)) (*OnlineCache, error) {
 	x := &OnlineCache{
-		user:  user,
-		group: group,
-		local: lru.NewSlotLRU(1024, localcache.LRUStringHash, func() lru.LRU[string, []int32] {
-			return lru.NewLayLRU[string, []int32](2048, cachekey.OnlineExpire/2, time.Second*3, localcache.EmptyTarget{}, func(key string, value []int32) {})
-		}),
+		user:          user,
+		group:         group,
+		fullUserCache: fullUserCache,
 	}
+
+	switch x.fullUserCache {
+	case true:
+		x.mapCache = cacheutil.NewCache[string, []int32]()
+		if err := x.initUsersOnlineStatus(mcontext.SetOperationID(context.TODO(), strconv.FormatInt(time.Now().UnixNano()+int64(rand.Uint32()), 10))); err != nil {
+			return nil, err
+		}
+	case false:
+		x.lruCache = lru.NewSlotLRU(1024, localcache.LRUStringHash, func() lru.LRU[string, []int32] {
+			return lru.NewLayLRU[string, []int32](2048, cachekey.OnlineExpire/2, time.Second*3, localcache.EmptyTarget{}, func(key string, value []int32) {})
+		})
+	}
+
 	go func() {
 		ctx := mcontext.SetOperationID(context.Background(), cachekey.OnlineChannel+strconv.FormatUint(rand.Uint64(), 10))
 		for message := range rdb.Subscribe(ctx, cachekey.OnlineChannel).Channel() {
 			userID, platformIDs, err := useronline.ParseUserOnlineStatus(message.Payload)
 			if err != nil {
-				log.ZError(ctx, "OnlineCache setUserOnline redis subscribe parseUserOnlineStatus", err, "payload", message.Payload, "channel", message.Channel)
+				log.ZError(ctx, "OnlineCache setHasUserOnline redis subscribe parseUserOnlineStatus", err, "payload", message.Payload, "channel", message.Channel)
 				continue
 			}
-			storageCache := x.setUserOnline(userID, platformIDs)
-			log.ZDebug(ctx, "OnlineCache setUserOnline", "userID", userID, "platformIDs", platformIDs, "payload", message.Payload, "storageCache", storageCache)
-			if fn != nil {
-				fn(ctx, userID, platformIDs)
+
+			switch x.fullUserCache {
+			case true:
+				if len(platformIDs) == 0 {
+					// offline
+					x.mapCache.Delete(userID)
+				} else {
+					x.mapCache.Store(userID, platformIDs)
+				}
+			case false:
+				storageCache := x.setHasUserOnline(userID, platformIDs)
+				log.ZDebug(ctx, "OnlineCache setHasUserOnline", "userID", userID, "platformIDs", platformIDs, "payload", message.Payload, "storageCache", storageCache)
+				if fn != nil {
+					fn(ctx, userID, platformIDs)
+				}
 			}
+
 		}
 	}()
-	return x
+	return x, nil
 }
 
 type OnlineCache struct {
 	user  rpcclient.UserRpcClient
 	group *GroupLocalCache
-	local lru.LRU[string, []int32]
+
+	// fullUserCache if enabled, caches the online status of all users using mapCache;
+	// otherwise, only a portion of users' online statuses (regardless of whether they are online) will be cached using lruCache.
+	fullUserCache bool
+
+	lruCache lru.LRU[string, []int32]
+	mapCache *cacheutil.Cache[string, []int32]
+}
+
+func (o *OnlineCache) initUsersOnlineStatus(ctx context.Context) error {
+	log.ZDebug(ctx, "init users online status begin")
+
+	var (
+		totalSet int
+	)
+
+	defer func(t time.Time) {
+		log.ZWarn(ctx, "init users online status end", nil, "cost", time.Since(t), "totalSet", totalSet)
+	}(time.Now())
+
+	for page := int32(1); ; page++ {
+		resp, err := o.user.GetAllUserID(ctx, page, constant.ParamMaxLength)
+		if err != nil {
+			return err
+		}
+
+		usersStatus, err := o.user.GetUsersOnlinePlatform(ctx, resp.UserIDs)
+		if err != nil {
+			return err
+		}
+
+		for _, user := range usersStatus {
+			if user.Status == constant.Online {
+				o.setUserOnline(user.UserID, user.PlatformIDs)
+			}
+			totalSet++
+		}
+
+		if len(resp.UserIDs) < constant.ParamMaxLength {
+			break
+		}
+	}
+	return nil
 }
 
 func (o *OnlineCache) getUserOnlinePlatform(ctx context.Context, userID string) ([]int32, error) {
-	platformIDs, err := o.local.Get(userID, func() ([]int32, error) {
+	platformIDs, err := o.lruCache.Get(userID, func() ([]int32, error) {
 		return o.user.GetUserOnlinePlatform(ctx, userID)
 	})
 	if err != nil {
@@ -91,7 +158,7 @@ func (o *OnlineCache) GetUserOnline(ctx context.Context, userID string) (bool, e
 // ----------------------
 
 func (o *OnlineCache) getUserOnlinePlatformBatch(ctx context.Context, userIDs []string) (map[string][]int32, error) {
-	platformIDsMap, err := o.local.GetBatch(userIDs, func(missingUsers []string) (map[string][]int32, error) {
+	platformIDsMap, err := o.lruCache.GetBatch(userIDs, func(missingUsers []string) (map[string][]int32, error) {
 		platformIDsMap := make(map[string][]int32)
 
 		usersStatus, err := o.user.GetUsersOnlinePlatform(ctx, missingUsers)
@@ -134,6 +201,19 @@ func (o *OnlineCache) GetUsersOnline(ctx context.Context, userIDs []string) ([]s
 		}
 	}
 
+	switch o.fullUserCache {
+	case true:
+		for _, userID := range userIDs {
+			if _, ok := o.mapCache.Load(userID); ok {
+				onlineUserIDs = append(onlineUserIDs, userID)
+			} else {
+				offlineUserIDs = append(offlineUserIDs, userID)
+			}
+		}
+	case false:
+	}
+
+	log.ZWarn(ctx, "get users online", nil, "online users length", len(onlineUserIDs), "offline users length", len(offlineUserIDs))
 	return onlineUserIDs, offlineUserIDs, nil
 }
 
@@ -171,6 +251,15 @@ func (o *OnlineCache) GetUsersOnline(ctx context.Context, userIDs []string) ([]s
 //	return onlineUserIDs, nil
 //}
 
-func (o *OnlineCache) setUserOnline(userID string, platformIDs []int32) bool {
-	return o.local.SetHas(userID, platformIDs)
+func (o *OnlineCache) setUserOnline(userID string, platformIDs []int32) {
+	switch o.fullUserCache {
+	case true:
+		o.mapCache.Store(userID, platformIDs)
+	case false:
+		o.lruCache.Set(userID, platformIDs)
+	}
+}
+
+func (o *OnlineCache) setHasUserOnline(userID string, platformIDs []int32) bool {
+	return o.lruCache.SetHas(userID, platformIDs)
 }

pkg/rpcclient/user.go

@@ -169,6 +169,15 @@ func (u *UserRpcClient) Access(ctx context.Context, ownerUserID string) error {
 	return authverify.CheckAccessV3(ctx, ownerUserID, u.imAdminUserID)
 }
 
+// GetAllUserID retrieves all user IDs with pagination options.
+func (u *UserRpcClient) GetAllUserID(ctx context.Context, pageNumber, showNumber int32) (*user.GetAllUserIDResp, error) {
+	resp, err := u.Client.GetAllUserID(ctx, &user.GetAllUserIDReq{Pagination: &sdkws.RequestPagination{PageNumber: pageNumber, ShowNumber: showNumber}})
+	if err != nil {
+		return nil, err
+	}
+	return resp, nil
+}
+
 // GetAllUserIDs retrieves all user IDs with pagination options.
 func (u *UserRpcClient) GetAllUserIDs(ctx context.Context, pageNumber, showNumber int32) ([]string, error) {
 	resp, err := u.Client.GetAllUserID(ctx, &user.GetAllUserIDReq{Pagination: &sdkws.RequestPagination{PageNumber: pageNumber, ShowNumber: showNumber}})