Merge 8cd57d68535f83c6495423e777f6a30c88cbe0f1 into d817cb0ffd4ad9c540a36b358c71c081e800920b

* fix: performance issues with Kafka caused by encapsulating the MQ interface

* fix: admin token in standalone mode

internal/api/router.go

@@ -97,6 +97,11 @@ func newGinRouter(ctx context.Context, client discovery.SvcDiscoveryRegistry, cf
 	case BestSpeed:
 		r.Use(gzip.Gzip(gzip.BestSpeed))
 	}
+	if config.Standalone() {
+		r.Use(func(c *gin.Context) {
+			c.Set(authverify.CtxAdminUserIDsKey, cfg.Share.IMAdminUser.UserIDs)
+		})
+	}
 	r.Use(api.GinLogger(), prommetricsGin(), gin.RecoveryWithWriter(gin.DefaultErrorWriter, mw.GinPanicErr), mw.CorsHandler(),
 		mw.GinParseOperationID(), GinParseToken(rpcli.NewAuthClient(authConn)), setGinIsAdmin(cfg.Share.IMAdminUser.UserIDs))
 

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

@@ -76,6 +76,27 @@ func batchGetCache2[K comparable, V any](ctx context.Context, rcClient *rocksCac
 	if err != nil {
 		return nil, err
 	}
+	values, err := fn(ctx, ids)
+	if err != nil {
+		log.ZError(ctx, "batchGetCache query database failed", err, "ids", ids)
+		return nil, err
+	}
+	idToValue := make(map[K]*V)
+	for _, value := range values {
+		idToValue[vId(value)] = value
+	}
+	getSlotValues := func(slotIds []K) []*V {
+		if len(slotIds) == 0 {
+			return nil
+		}
+		slotValues := make([]*V, 0, len(slotIds))
+		for _, id := range slotIds {
+			if value, ok := idToValue[id]; ok {
+				slotValues = append(slotValues, value)
+			}
+		}
+		return slotValues
+	}
 	result := make([]*V, 0, len(findKeys))
 	for _, keys := range slotKeys {
 		indexCache, err := rcClient.GetClient().FetchBatch2(ctx, keys, expire, func(idx []int) (map[int]string, error) {
@@ -86,16 +107,12 @@ func batchGetCache2[K comparable, V any](ctx context.Context, rcClient *rocksCac
 				idIndex[id] = index
 				queryIds = append(queryIds, id)
 			}
-			values, err := fn(ctx, queryIds)
-			if err != nil {
-				log.ZError(ctx, "batchGetCache query database failed", err, "keys", keys, "queryIds", queryIds)
-				return nil, err
-			}
-			if len(values) == 0 {
+			slotValues := getSlotValues(queryIds)
+			if len(slotValues) == 0 {
 				return map[int]string{}, nil
 			}
 			cacheIndex := make(map[int]string)
-			for _, value := range values {
+			for _, value := range slotValues {
 				id := vId(value)
 				index, ok := idIndex[id]
 				if !ok {

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

@@ -165,16 +165,15 @@ func (c *tokenCache) DeleteTokenByTokenMap(ctx context.Context, userID string, t
 }
 
 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)
 		}
 	}
-
+	key := cachekey.GetTokenKey(userID, platformID)
+	if err := c.rdb.HDel(ctx, key, fields...).Err(); err != nil {
+		return errs.Wrap(err)
+	}
 	return nil
 }