Merge pull request #3341 from openimsdk/cherry-pick-56c5c1f

fix: delete token by correct platformID && feat: adminToken can be re… [Created by @icey-yu from #3313]

pkg/common/storage/cache/cachekey/token.go

@@ -1,8 +1,9 @@
 package cachekey
 
 import (
-	"github.com/openimsdk/protocol/constant"
 	"strings"
+
+	"github.com/openimsdk/protocol/constant"
 )
 
 const (
@@ -13,6 +14,10 @@ func GetTokenKey(userID string, platformID int) string {
 	return UidPidToken + userID + ":" + constant.PlatformIDToName(platformID)
 }
 
+func GetTemporaryTokenKey(userID string, platformID int, token string) string {
+	return UidPidToken + ":TEMPORARY:" + userID + ":" + constant.PlatformIDToName(platformID) + ":" + token
+}
+
 func GetAllPlatformTokenKey(userID string) []string {
 	res := make([]string, len(constant.PlatformID2Name))
 	for k := range constant.PlatformID2Name {

pkg/common/storage/cache/mcache/token.go

@@ -0,0 +1,166 @@
+package mcache
+
+import (
+	"context"
+	"fmt"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/cachekey"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database"
+	"github.com/openimsdk/tools/errs"
+	"github.com/openimsdk/tools/log"
+)
+
+func NewTokenCacheModel(cache database.Cache, accessExpire int64) cache.TokenModel {
+	c := &tokenCache{cache: cache}
+	c.accessExpire = c.getExpireTime(accessExpire)
+	return c
+}
+
+type tokenCache struct {
+	cache        database.Cache
+	accessExpire time.Duration
+}
+
+func (x *tokenCache) getTokenKey(userID string, platformID int, token string) string {
+	return cachekey.GetTokenKey(userID, platformID) + ":" + token
+}
+
+func (x *tokenCache) SetTokenFlag(ctx context.Context, userID string, platformID int, token string, flag int) error {
+	return x.cache.Set(ctx, x.getTokenKey(userID, platformID, token), strconv.Itoa(flag), x.accessExpire)
+}
+
+// SetTokenFlagEx set token and flag with expire time
+func (x *tokenCache) SetTokenFlagEx(ctx context.Context, userID string, platformID int, token string, flag int) error {
+	return x.SetTokenFlag(ctx, userID, platformID, token, flag)
+}
+
+func (x *tokenCache) GetTokensWithoutError(ctx context.Context, userID string, platformID int) (map[string]int, error) {
+	prefix := x.getTokenKey(userID, platformID, "")
+	m, err := x.cache.Prefix(ctx, prefix)
+	if err != nil {
+		return nil, errs.Wrap(err)
+	}
+	mm := make(map[string]int)
+	for k, v := range m {
+		state, err := strconv.Atoi(v)
+		if err != nil {
+			log.ZError(ctx, "token value is not int", err, "value", v, "userID", userID, "platformID", platformID)
+			continue
+		}
+		mm[strings.TrimPrefix(k, prefix)] = state
+	}
+	return mm, nil
+}
+
+func (x *tokenCache) HasTemporaryToken(ctx context.Context, userID string, platformID int, token string) error {
+	key := cachekey.GetTemporaryTokenKey(userID, platformID, token)
+	if _, err := x.cache.Get(ctx, []string{key}); err != nil {
+		return err
+	}
+	return nil
+}
+
+func (x *tokenCache) GetAllTokensWithoutError(ctx context.Context, userID string) (map[int]map[string]int, error) {
+	prefix := cachekey.UidPidToken + userID + ":"
+	tokens, err := x.cache.Prefix(ctx, prefix)
+	if err != nil {
+		return nil, err
+	}
+	res := make(map[int]map[string]int)
+	for key, flagStr := range tokens {
+		flag, err := strconv.Atoi(flagStr)
+		if err != nil {
+			log.ZError(ctx, "token value is not int", err, "key", key, "value", flagStr, "userID", userID)
+			continue
+		}
+		arr := strings.SplitN(strings.TrimPrefix(key, prefix), ":", 2)
+		if len(arr) != 2 {
+			log.ZError(ctx, "token value is not int", err, "key", key, "value", flagStr, "userID", userID)
+			continue
+		}
+		platformID, err := strconv.Atoi(arr[0])
+		if err != nil {
+			log.ZError(ctx, "token value is not int", err, "key", key, "value", flagStr, "userID", userID)
+			continue
+		}
+		token := arr[1]
+		if token == "" {
+			log.ZError(ctx, "token value is not int", err, "key", key, "value", flagStr, "userID", userID)
+			continue
+		}
+		tk, ok := res[platformID]
+		if !ok {
+			tk = make(map[string]int)
+			res[platformID] = tk
+		}
+		tk[token] = flag
+	}
+	return res, nil
+}
+
+func (x *tokenCache) SetTokenMapByUidPid(ctx context.Context, userID string, platformID int, m map[string]int) error {
+	for token, flag := range m {
+		err := x.SetTokenFlag(ctx, userID, platformID, token, flag)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (x *tokenCache) BatchSetTokenMapByUidPid(ctx context.Context, tokens map[string]map[string]any) error {
+	for prefix, tokenFlag := range tokens {
+		for token, flag := range tokenFlag {
+			flagStr := fmt.Sprintf("%v", flag)
+			if err := x.cache.Set(ctx, prefix+":"+token, flagStr, x.accessExpire); err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+
+func (x *tokenCache) DeleteTokenByUidPid(ctx context.Context, userID string, platformID int, fields []string) error {
+	keys := make([]string, 0, len(fields))
+	for _, token := range fields {
+		keys = append(keys, x.getTokenKey(userID, platformID, token))
+	}
+	return x.cache.Del(ctx, keys)
+}
+
+func (x *tokenCache) getExpireTime(t int64) time.Duration {
+	return time.Hour * 24 * time.Duration(t)
+}
+
+func (x *tokenCache) DeleteTokenByTokenMap(ctx context.Context, userID string, tokens map[int][]string) error {
+	keys := make([]string, 0, len(tokens))
+	for platformID, ts := range tokens {
+		for _, t := range ts {
+			keys = append(keys, x.getTokenKey(userID, platformID, t))
+		}
+	}
+	return x.cache.Del(ctx, keys)
+}
+
+func (x *tokenCache) DeleteAndSetTemporary(ctx context.Context, userID string, platformID int, fields []string) error {
+	keys := make([]string, 0, len(fields))
+	for _, f := range fields {
+		keys = append(keys, x.getTokenKey(userID, platformID, f))
+	}
+	if err := x.cache.Del(ctx, keys); err != nil {
+		return err
+	}
+
+	for _, f := range fields {
+		k := cachekey.GetTemporaryTokenKey(userID, platformID, f)
+		if err := x.cache.Set(ctx, k, "", time.Minute*5); err != nil {
+			return errs.Wrap(err)
+		}
+	}
+
+	return nil
+}

pkg/common/storage/cache/redis/token.go

@@ -9,6 +9,7 @@ import (
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/cachekey"
 	"github.com/openimsdk/tools/errs"
+	"github.com/openimsdk/tools/utils/datautil"
 	"github.com/redis/go-redis/v9"
 )
 
@@ -55,6 +56,14 @@ func (c *tokenCache) GetTokensWithoutError(ctx context.Context, userID string, p
 	return mm, nil
 }
 
+func (c *tokenCache) HasTemporaryToken(ctx context.Context, userID string, platformID int, token string) error {
+	err := c.rdb.Get(ctx, cachekey.GetTemporaryTokenKey(userID, platformID, token)).Err()
+	if err != nil {
+		return errs.Wrap(err)
+	}
+	return nil
+}
+
 func (c *tokenCache) GetAllTokensWithoutError(ctx context.Context, userID string) (map[int]map[string]int, error) {
 	var (
 		res     = make(map[int]map[string]int)
@@ -101,13 +110,19 @@ func (c *tokenCache) SetTokenMapByUidPid(ctx context.Context, userID string, pla
 }
 
 func (c *tokenCache) BatchSetTokenMapByUidPid(ctx context.Context, tokens map[string]map[string]any) error {
-	pipe := c.rdb.Pipeline()
-	for k, v := range tokens {
-		pipe.HSet(ctx, k, v)
-	}
-	_, err := pipe.Exec(ctx)
-	if err != nil {
-		return errs.Wrap(err)
+	keys := datautil.Keys(tokens)
+	if err := ProcessKeysBySlot(ctx, c.rdb, keys, func(ctx context.Context, slot int64, keys []string) error {
+		pipe := c.rdb.Pipeline()
+		for k, v := range tokens {
+			pipe.HSet(ctx, k, v)
+		}
+		_, err := pipe.Exec(ctx)
+		if err != nil {
+			return errs.Wrap(err)
+		}
+		return nil
+	}); err != nil {
+		return err
 	}
 	return nil
 }
@@ -119,3 +134,47 @@ func (c *tokenCache) DeleteTokenByUidPid(ctx context.Context, userID string, pla
 func (c *tokenCache) getExpireTime(t int64) time.Duration {
 	return time.Hour * 24 * time.Duration(t)
 }
+
+// DeleteTokenByTokenMap tokens key is platformID, value is token slice
+func (c *tokenCache) DeleteTokenByTokenMap(ctx context.Context, userID string, tokens map[int][]string) error {
+	var (
+		keys   = make([]string, 0, len(tokens))
+		keyMap = make(map[string][]string)
+	)
+	for k, v := range tokens {
+		k1 := cachekey.GetTokenKey(userID, k)
+		keys = append(keys, k1)
+		keyMap[k1] = v
+	}
+
+	if err := ProcessKeysBySlot(ctx, c.rdb, keys, func(ctx context.Context, slot int64, keys []string) error {
+		pipe := c.rdb.Pipeline()
+		for k, v := range tokens {
+			pipe.HDel(ctx, cachekey.GetTokenKey(userID, k), v...)
+		}
+		_, err := pipe.Exec(ctx)
+		if err != nil {
+			return errs.Wrap(err)
+		}
+		return nil
+	}); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (c *tokenCache) DeleteAndSetTemporary(ctx context.Context, userID string, platformID int, fields []string) error {
+	key := cachekey.GetTokenKey(userID, platformID)
+	if err := c.rdb.HDel(ctx, key, fields...).Err(); err != nil {
+		return errs.Wrap(err)
+	}
+	for _, f := range fields {
+		k := cachekey.GetTemporaryTokenKey(userID, platformID, f)
+		if err := c.rdb.Set(ctx, k, "", time.Minute*5).Err(); err != nil {
+			return errs.Wrap(err)
+		}
+	}
+
+	return nil
+}

pkg/common/storage/cache/token.go

@@ -9,8 +9,11 @@ type TokenModel interface {
 	// SetTokenFlagEx set token and flag with expire time
 	SetTokenFlagEx(ctx context.Context, userID string, platformID int, token string, flag int) error
 	GetTokensWithoutError(ctx context.Context, userID string, platformID int) (map[string]int, error)
+	HasTemporaryToken(ctx context.Context, userID string, platformID int, token string) error
 	GetAllTokensWithoutError(ctx context.Context, userID string) (map[int]map[string]int, error)
 	SetTokenMapByUidPid(ctx context.Context, userID string, platformID int, m map[string]int) error
 	BatchSetTokenMapByUidPid(ctx context.Context, tokens map[string]map[string]any) error
 	DeleteTokenByUidPid(ctx context.Context, userID string, platformID int, fields []string) error
+	DeleteTokenByTokenMap(ctx context.Context, userID string, tokens map[int][]string) error
+	DeleteAndSetTemporary(ctx context.Context, userID string, platformID int, fields []string) error
 }