Compare commits

...

7 Commits

Author SHA1 Message Date
dependabot[bot] 8670fbf31f build(deps): bump github.com/golang-jwt/jwt/v4 from 4.5.1 to 4.5.2
Bumps [github.com/golang-jwt/jwt/v4](https://github.com/golang-jwt/jwt) from 4.5.1 to 4.5.2.
- [Release notes](https://github.com/golang-jwt/jwt/releases)
- [Changelog](https://github.com/golang-jwt/jwt/blob/main/VERSION_HISTORY.md)
- [Commits](https://github.com/golang-jwt/jwt/compare/v4.5.1...v4.5.2)

---
updated-dependencies:
- dependency-name: github.com/golang-jwt/jwt/v4
  dependency-version: 4.5.2
  dependency-type: direct:production
...

Signed-off-by: dependabot[bot] <support@github.com>
2025-08-28 03:50:32 +00:00
Monet Lee 6856a864d0 feat: implement auth local cache. (#3533)
* feat: add auth local cache.

* feat: implement auth local cache.

* feat: improve auth localcache.
2025-08-28 03:26:02 +00:00
Monet Lee 349a8cd9af fix: nil pointer when chatlog not found. (#3531) 2025-08-19 09:22:09 +00:00
Monet Lee 929c6704f3 fix: solve incorrect batchGetIncrGroupMember when group dismissed. (#3526) 2025-08-19 09:20:58 +00:00
icey-yu 68a735ba99 feat: enable redis aof-use-rdb-preamble && disable auto rdb (#3529) 2025-08-19 08:23:27 +00:00
OpenIM-Gordon 123abe9803 fix: fill in the most recent sendTime for a gap message to prevent the client from repeatedly retrieving the same message due to sendTime being 0. (#3522) 2025-08-14 06:16:50 +00:00
Monet Lee 34971c8b96 fix: optimize to lru local cache. (#3514)
* fix: optimize to lru  local cache.

* revert lock timing.
2025-08-14 02:19:11 +00:00
21 changed files with 309 additions and 65 deletions
+7
View File
@@ -1,3 +1,10 @@
auth:
topic: DELETE_CACHE_AUTH
slotNum: 100
slotSize: 2000
successExpire: 300
failedExpire: 5
user:
topic: DELETE_CACHE_USER
slotNum: 100
+6 -1
View File
@@ -63,7 +63,12 @@ services:
restart: always
sysctls:
net.core.somaxconn: 1024
command: redis-server /usr/local/redis/config/redis.conf --requirepass openIM123 --appendonly yes
command: >
redis-server
--requirepass openIM123
--appendonly yes
--aof-use-rdb-preamble yes
--save ""
networks:
- openim
+2 -2
View File
@@ -8,11 +8,11 @@ require (
github.com/gin-gonic/gin v1.9.1
github.com/go-playground/validator/v10 v10.20.0
github.com/gogo/protobuf v1.3.2 // indirect
github.com/golang-jwt/jwt/v4 v4.5.1
github.com/golang-jwt/jwt/v4 v4.5.2
github.com/gorilla/websocket v1.5.1
github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0
github.com/mitchellh/mapstructure v1.5.0
github.com/openimsdk/protocol v0.0.73-alpha.12
github.com/openimsdk/protocol v0.0.73-alpha.14
github.com/openimsdk/tools v0.0.50-alpha.97
github.com/pkg/errors v0.9.1 // indirect
github.com/prometheus/client_golang v1.18.0
+4 -4
View File
@@ -171,8 +171,8 @@ github.com/godbus/dbus/v5 v5.0.4/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5x
github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q=
github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q=
github.com/golang-jwt/jwt/v4 v4.4.2/go.mod h1:m21LjoU+eqJr34lmDMbreY2eSTRJ1cv77w39/MY0Ch0=
github.com/golang-jwt/jwt/v4 v4.5.1 h1:JdqV9zKUdtaa9gdPlywC3aeoEsR681PlKC+4F5gQgeo=
github.com/golang-jwt/jwt/v4 v4.5.1/go.mod h1:m21LjoU+eqJr34lmDMbreY2eSTRJ1cv77w39/MY0Ch0=
github.com/golang-jwt/jwt/v4 v4.5.2 h1:YtQM7lnr8iZ+j5q71MGKkNw9Mn7AjHM68uc9g5fXeUI=
github.com/golang-jwt/jwt/v4 v4.5.2/go.mod h1:m21LjoU+eqJr34lmDMbreY2eSTRJ1cv77w39/MY0Ch0=
github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q=
github.com/golang/groupcache v0.0.0-20200121045136-8c9f03a8e57e/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc=
github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da h1:oI5xCqsCo564l8iNU+DwB5epxmsaqB+rhGL0m5jtYqE=
@@ -347,8 +347,8 @@ github.com/onsi/gomega v1.25.0 h1:Vw7br2PCDYijJHSfBOWhov+8cAnUf8MfMaIOV323l6Y=
github.com/onsi/gomega v1.25.0/go.mod h1:r+zV744Re+DiYCIPRlYOTxn0YkOLcAnW8k1xXdMPGhM=
github.com/openimsdk/gomake v0.0.15-alpha.11 h1:PQudYDRESYeYlUYrrLLJhYIlUPO5x7FAx+o5El9U/Bw=
github.com/openimsdk/gomake v0.0.15-alpha.11/go.mod h1:PndCozNc2IsQIciyn9mvEblYWZwJmAI+06z94EY+csI=
github.com/openimsdk/protocol v0.0.73-alpha.12 h1:2NYawXeHChYUeSme6QJ9pOLh+Empce2WmwEtbP4JvKk=
github.com/openimsdk/protocol v0.0.73-alpha.12/go.mod h1:WF7EuE55vQvpyUAzDXcqg+B+446xQyEba0X35lTINmw=
github.com/openimsdk/protocol v0.0.73-alpha.14 h1:lv9wNiPRm6G7q74TfpMobKrSfeTaBlZ+Ps3O6UFPmaE=
github.com/openimsdk/protocol v0.0.73-alpha.14/go.mod h1:WF7EuE55vQvpyUAzDXcqg+B+446xQyEba0X35lTINmw=
github.com/openimsdk/tools v0.0.50-alpha.97 h1:6ik5w3PpgDG6VjSo3nb3FT/fxN3JX7iIARVxVu9g7VY=
github.com/openimsdk/tools v0.0.50-alpha.97/go.mod h1:n2poR3asX1e1XZce4O+MOWAp+X02QJRFvhcLCXZdzRo=
github.com/pelletier/go-toml/v2 v2.2.2 h1:aYUidT7k73Pcl9nb2gScu7NSrKCSHIDE89b3+6Wq+LM=
+38 -10
View File
@@ -18,10 +18,13 @@ import (
"context"
"errors"
"github.com/openimsdk/open-im-server/v3/pkg/common/convert"
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache"
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/mcache"
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database/mgo"
"github.com/openimsdk/open-im-server/v3/pkg/dbbuild"
"github.com/openimsdk/open-im-server/v3/pkg/localcache"
"github.com/openimsdk/open-im-server/v3/pkg/rpccache"
"github.com/openimsdk/open-im-server/v3/pkg/rpcli"
"github.com/openimsdk/open-im-server/v3/pkg/common/config"
@@ -46,6 +49,7 @@ import (
type authServer struct {
pbauth.UnimplementedAuthServer
authDatabase controller.AuthDatabase
AuthLocalCache *rpccache.AuthLocalCache
RegisterCenter discovery.Conn
config *Config
userClient *rpcli.UserClient
@@ -53,11 +57,12 @@ type authServer struct {
}
type Config struct {
RpcConfig config.Auth
RedisConfig config.Redis
MongoConfig config.Mongo
Share config.Share
Discovery config.Discovery
RpcConfig config.Auth
RedisConfig config.Redis
MongoConfig config.Mongo
Share config.Share
LocalCacheConfig config.LocalCache
Discovery config.Discovery
}
func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryRegistry, server grpc.ServiceRegistrar) error {
@@ -78,12 +83,19 @@ func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryReg
}
token = mcache.NewTokenCacheModel(mc, config.RpcConfig.TokenPolicy.Expire)
} else {
token = redis2.NewTokenCacheModel(rdb, config.RpcConfig.TokenPolicy.Expire)
token = redis2.NewTokenCacheModel(rdb, &config.LocalCacheConfig, config.RpcConfig.TokenPolicy.Expire)
}
userConn, err := client.GetConn(ctx, config.Discovery.RpcService.User)
if err != nil {
return err
}
authConn, err := client.GetConn(ctx, config.Discovery.RpcService.Auth)
if err != nil {
return err
}
localcache.InitLocalCache(&config.LocalCacheConfig)
pbauth.RegisterAuthServer(server, &authServer{
RegisterCenter: client,
authDatabase: controller.NewAuthDatabase(
@@ -93,9 +105,10 @@ func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryReg
config.Share.MultiLogin,
config.Share.IMAdminUser.UserIDs,
),
config: config,
userClient: rpcli.NewUserClient(userConn),
adminUserIDs: config.Share.IMAdminUser.UserIDs,
AuthLocalCache: rpccache.NewAuthLocalCache(rpcli.NewAuthClient(authConn), &config.LocalCacheConfig, rdb),
config: config,
userClient: rpcli.NewUserClient(userConn),
adminUserIDs: config.Share.IMAdminUser.UserIDs,
})
return nil
}
@@ -121,6 +134,7 @@ func (s *authServer) GetAdminToken(ctx context.Context, req *pbauth.GetAdminToke
}
prommetrics.UserLoginCounter.Inc()
resp.Token = token
resp.ExpireTimeSeconds = s.config.RpcConfig.TokenPolicy.Expire * 24 * 60 * 60
return &resp, nil
@@ -151,20 +165,34 @@ func (s *authServer) GetUserToken(ctx context.Context, req *pbauth.GetUserTokenR
if err != nil {
return nil, err
}
resp.Token = token
resp.ExpireTimeSeconds = s.config.RpcConfig.TokenPolicy.Expire * 24 * 60 * 60
return &resp, nil
}
func (s *authServer) GetExistingToken(ctx context.Context, req *pbauth.GetExistingTokenReq) (*pbauth.GetExistingTokenResp, error) {
m, err := s.authDatabase.GetTokensWithoutError(ctx, req.UserID, int(req.PlatformID))
if err != nil {
return nil, err
}
return &pbauth.GetExistingTokenResp{
TokenStates: convert.TokenMapDB2Pb(m),
}, nil
}
func (s *authServer) parseToken(ctx context.Context, tokensString string) (claims *tokenverify.Claims, err error) {
claims, err = tokenverify.GetClaimFromToken(tokensString, authverify.Secret(s.config.Share.Secret))
if err != nil {
return nil, err
}
m, err := s.authDatabase.GetTokensWithoutError(ctx, claims.UserID, claims.PlatformID)
m, err := s.AuthLocalCache.GetExistingToken(ctx, claims.UserID, claims.PlatformID)
if err != nil {
return nil, err
}
if len(m) == 0 {
isAdmin := authverify.CheckUserIsAdmin(ctx, claims.UserID)
if isAdmin {
+14 -1
View File
@@ -132,6 +132,7 @@ func (c *conversationServer) GetConversation(ctx context.Context, req *pbconvers
return resp, nil
}
// Deprecated: Use `GetConversations` instead.
func (c *conversationServer) GetSortedConversationList(ctx context.Context, req *pbconversation.GetSortedConversationListReq) (resp *pbconversation.GetSortedConversationListResp, err error) {
if err := authverify.CheckAccess(ctx, req.UserID); err != nil {
return nil, err
@@ -183,9 +184,21 @@ func (c *conversationServer) GetSortedConversationList(ctx context.Context, req
conversation_isPinTime := make(map[int64]string)
conversation_notPinTime := make(map[int64]string)
for _, v := range conversations {
conversationID := v.ConversationID
time := conversationMsg[conversationID].MsgInfo.LatestMsgRecvTime
var time int64
if _, ok := conversationMsg[conversationID]; ok {
time = conversationMsg[conversationID].MsgInfo.LatestMsgRecvTime
} else {
conversationMsg[conversationID] = &pbconversation.ConversationElem{
ConversationID: conversationID,
IsPinned: v.IsPinned,
MsgInfo: nil,
}
time = v.CreateTime.UnixMilli()
}
conversationMsg[conversationID].RecvMsgOpt = v.RecvMsgOpt
if v.IsPinned {
conversationMsg[conversationID].IsPinned = v.IsPinned
+9
View File
@@ -2,6 +2,7 @@ package group
import (
"context"
"errors"
"github.com/openimsdk/open-im-server/v3/internal/rpc/incrversion"
"github.com/openimsdk/open-im-server/v3/pkg/authverify"
@@ -11,6 +12,7 @@ import (
"github.com/openimsdk/protocol/constant"
pbgroup "github.com/openimsdk/protocol/group"
"github.com/openimsdk/protocol/sdkws"
"github.com/openimsdk/tools/log"
)
const versionSyncLimit = 500
@@ -170,19 +172,26 @@ func (g *groupServer) GetIncrementalJoinGroup(ctx context.Context, req *pbgroup.
func (g *groupServer) BatchGetIncrementalGroupMember(ctx context.Context, req *pbgroup.BatchGetIncrementalGroupMemberReq) (*pbgroup.BatchGetIncrementalGroupMemberResp, error) {
var num int
resp := make(map[string]*pbgroup.GetIncrementalGroupMemberResp)
for _, memberReq := range req.ReqList {
if _, ok := resp[memberReq.GroupID]; ok {
continue
}
memberResp, err := g.GetIncrementalGroupMember(ctx, memberReq)
if err != nil {
if errors.Is(err, servererrs.ErrDismissedAlready) {
log.ZWarn(ctx, "Failed to get incremental group member", err, "groupID", memberReq.GroupID, "request", memberReq)
continue
}
return nil, err
}
resp[memberReq.GroupID] = memberResp
num += len(memberResp.Insert) + len(memberResp.Update) + len(memberResp.Delete)
if num >= versionSyncLimit {
break
}
}
return &pbgroup.BatchGetIncrementalGroupMemberResp{RespList: resp}, nil
}
+1
View File
@@ -40,6 +40,7 @@ func NewAuthRpcCmd() *AuthRpcCmd {
config.RedisConfigFileName: &authConfig.RedisConfig,
config.MongodbConfigFileName: &authConfig.MongoConfig,
config.ShareFileName: &authConfig.Share,
config.LocalCacheConfigFileName: &authConfig.LocalCacheConfig,
config.DiscoveryConfigFilename: &authConfig.Discovery,
}
ret.RootCmd = NewRootCmd(program.GetProcessName(), WithConfigMap(ret.configMap))
+1
View File
@@ -43,6 +43,7 @@ type CacheConfig struct {
}
type LocalCache struct {
Auth CacheConfig `yaml:"auth"`
User CacheConfig `yaml:"user"`
Group CacheConfig `yaml:"group"`
Friend CacheConfig `yaml:"friend"`
+25
View File
@@ -0,0 +1,25 @@
package convert
func TokenMapDB2Pb(tokenMapDB map[string]int) map[string]int32 {
if tokenMapDB == nil {
return nil
}
tokenMapPB := make(map[string]int32, len(tokenMapDB))
for k, v := range tokenMapDB {
tokenMapPB[k] = int32(v)
}
return tokenMapPB
}
func TokenMapPb2DB(tokenMapPB map[string]int32) map[string]int {
if tokenMapPB == nil {
return nil
}
tokenMapDB := make(map[string]int, len(tokenMapPB))
for k, v := range tokenMapPB {
tokenMapDB[k] = int(v)
}
return tokenMapDB
}
+1 -1
View File
@@ -24,7 +24,7 @@ var (
func NewMsgCache(cache database.Cache, msgDocDatabase database.Msg) cache.MsgCache {
initMemMsgCache.Do(func() {
memMsgCache = lru.NewLayLRU[string, *model.MsgInfoModel](1024*8, time.Hour, time.Second*10, localcache.EmptyTarget{}, nil)
memMsgCache = lru.NewLazyLRU[string, *model.MsgInfoModel](1024*8, time.Hour, time.Second*10, localcache.EmptyTarget{}, nil)
})
return &msgCache{
cache: cache,
+74 -5
View File
@@ -2,13 +2,16 @@ package redis
import (
"context"
"encoding/json"
"strconv"
"sync"
"time"
"github.com/openimsdk/open-im-server/v3/pkg/common/config"
"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/log"
"github.com/openimsdk/tools/utils/datautil"
"github.com/redis/go-redis/v9"
)
@@ -16,16 +19,26 @@ import (
type tokenCache struct {
rdb redis.UniversalClient
accessExpire time.Duration
localCache *config.LocalCache
}
func NewTokenCacheModel(rdb redis.UniversalClient, accessExpire int64) cache.TokenModel {
c := &tokenCache{rdb: rdb}
func NewTokenCacheModel(rdb redis.UniversalClient, localCache *config.LocalCache, accessExpire int64) cache.TokenModel {
c := &tokenCache{rdb: rdb, localCache: localCache}
c.accessExpire = c.getExpireTime(accessExpire)
return c
}
func (c *tokenCache) SetTokenFlag(ctx context.Context, userID string, platformID int, token string, flag int) error {
return errs.Wrap(c.rdb.HSet(ctx, cachekey.GetTokenKey(userID, platformID), token, flag).Err())
key := cachekey.GetTokenKey(userID, platformID)
if err := c.rdb.HSet(ctx, key, token, flag).Err(); err != nil {
return errs.Wrap(err)
}
if c.localCache != nil {
c.removeLocalTokenCache(ctx, key)
}
return nil
}
// SetTokenFlagEx set token and flag with expire time
@@ -37,6 +50,11 @@ func (c *tokenCache) SetTokenFlagEx(ctx context.Context, userID string, platform
if err := c.rdb.Expire(ctx, key, c.accessExpire).Err(); err != nil {
return errs.Wrap(err)
}
if c.localCache != nil {
c.removeLocalTokenCache(ctx, key)
}
return nil
}
@@ -106,7 +124,17 @@ func (c *tokenCache) SetTokenMapByUidPid(ctx context.Context, userID string, pla
for k, v := range m {
mm[k] = v
}
return errs.Wrap(c.rdb.HSet(ctx, cachekey.GetTokenKey(userID, platformID), mm).Err())
err := c.rdb.HSet(ctx, cachekey.GetTokenKey(userID, platformID), mm).Err()
if err != nil {
return errs.Wrap(err)
}
if c.localCache != nil {
c.removeLocalTokenCache(ctx, cachekey.GetTokenKey(userID, platformID))
}
return nil
}
func (c *tokenCache) BatchSetTokenMapByUidPid(ctx context.Context, tokens map[string]map[string]any) error {
@@ -124,11 +152,23 @@ func (c *tokenCache) BatchSetTokenMapByUidPid(ctx context.Context, tokens map[st
}); err != nil {
return err
}
if c.localCache != nil {
c.removeLocalTokenCache(ctx, keys...)
}
return nil
}
func (c *tokenCache) DeleteTokenByUidPid(ctx context.Context, userID string, platformID int, fields []string) error {
return errs.Wrap(c.rdb.HDel(ctx, cachekey.GetTokenKey(userID, platformID), fields...).Err())
key := cachekey.GetTokenKey(userID, platformID)
if err := c.rdb.HDel(ctx, key, fields...).Err(); err != nil {
return errs.Wrap(err)
}
if c.localCache != nil {
c.removeLocalTokenCache(ctx, key)
}
return nil
}
func (c *tokenCache) getExpireTime(t int64) time.Duration {
@@ -161,6 +201,11 @@ func (c *tokenCache) DeleteTokenByTokenMap(ctx context.Context, userID string, t
return err
}
// Remove local cache for the token
if c.localCache != nil {
c.removeLocalTokenCache(ctx, keys...)
}
return nil
}
@@ -175,5 +220,29 @@ func (c *tokenCache) DeleteAndSetTemporary(ctx context.Context, userID string, p
if err := c.rdb.HDel(ctx, key, fields...).Err(); err != nil {
return errs.Wrap(err)
}
if c.localCache != nil {
c.removeLocalTokenCache(ctx, key)
}
return nil
}
func (c *tokenCache) removeLocalTokenCache(ctx context.Context, keys ...string) {
if len(keys) == 0 {
return
}
topic := c.localCache.Auth.Topic
if topic == "" {
return
}
data, err := json.Marshal(keys)
if err != nil {
log.ZWarn(ctx, "keys json marshal failed", err, "topic", topic, "keys", keys)
} else {
if err := c.rdb.Publish(ctx, topic, string(data)).Err(); err != nil {
log.ZWarn(ctx, "redis publish cache delete error", err, "topic", topic, "keys", keys)
}
}
}
-1
View File
@@ -194,7 +194,6 @@ func (g *groupDatabase) CreateGroup(ctx context.Context, groups []*model.Group,
}
for _, group := range groups {
c = c.DelGroupsInfo(group.GroupID).
DelGroupMembersHash(group.GroupID).
DelGroupMembersHash(group.GroupID).
DelGroupsMemberNum(group.GroupID).
DelGroupMemberIDs(group.GroupID).
+7 -6
View File
@@ -5,6 +5,11 @@ import (
"fmt"
"time"
"go.mongodb.org/mongo-driver/bson"
"go.mongodb.org/mongo-driver/bson/primitive"
"go.mongodb.org/mongo-driver/mongo"
"go.mongodb.org/mongo-driver/mongo/options"
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database"
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
"github.com/openimsdk/protocol/constant"
@@ -14,10 +19,6 @@ import (
"github.com/openimsdk/tools/errs"
"github.com/openimsdk/tools/utils/datautil"
"github.com/openimsdk/tools/utils/jsonutil"
"go.mongodb.org/mongo-driver/bson"
"go.mongodb.org/mongo-driver/bson/primitive"
"go.mongodb.org/mongo-driver/mongo"
"go.mongodb.org/mongo-driver/mongo/options"
)
func NewMsgMongo(db *mongo.Database) (database.Msg, error) {
@@ -1154,7 +1155,7 @@ func (m *MsgMgo) findBeforeDocSendTime(ctx context.Context, docID string, limit
if err != nil {
return 0, 0, err
}
for i := len(res) - 1; i > 0; i-- {
for i := len(res) - 1; i >= 0; i-- {
v := res[i]
if v.Msgs != nil && v.Msgs.Msg != nil && v.Msgs.Msg.SendTime > 0 {
return v.Msgs.Msg.Seq, v.Msgs.Msg.SendTime, nil
@@ -1169,7 +1170,7 @@ func (m *MsgMgo) findBeforeSendTime(ctx context.Context, conversationID string,
limit := int64(-1)
if first {
first = false
limit = m.model.GetMsgIndex(seq)
limit = m.model.GetLimitForSingleDoc(seq)
}
docID := m.model.BuildDocIDByIndex(conversationID, i)
msgSeq, msgSendTime, err := m.findBeforeDocSendTime(ctx, docID, limit)
+4
View File
@@ -132,6 +132,10 @@ func (*MsgDocModel) GetMsgIndex(seq int64) int64 {
return (seq - 1) % singleGocMsgNum
}
func (*MsgDocModel) GetLimitForSingleDoc(seq int64) int64 {
return seq % singleGocMsgNum
}
func (*MsgDocModel) indexGen(conversationID string, seqSuffix int64) string {
return conversationID + ":" + strconv.FormatInt(seqSuffix, 10)
}
+5 -3
View File
@@ -47,15 +47,15 @@ func New[V any](opts ...Option) Cache[V] {
if opt.localSlotNum > 0 && opt.localSlotSize > 0 {
createSimpleLRU := func() lru.LRU[string, V] {
if opt.expirationEvict {
return lru.NewExpirationLRU[string, V](opt.localSlotSize, opt.localSuccessTTL, opt.localFailedTTL, opt.target, c.onEvict)
return lru.NewExpirationLRU(opt.localSlotSize, opt.localSuccessTTL, opt.localFailedTTL, opt.target, c.onEvict)
} else {
return lru.NewLayLRU[string, V](opt.localSlotSize, opt.localSuccessTTL, opt.localFailedTTL, opt.target, c.onEvict)
return lru.NewLazyLRU(opt.localSlotSize, opt.localSuccessTTL, opt.localFailedTTL, opt.target, c.onEvict)
}
}
if opt.localSlotNum == 1 {
c.local = createSimpleLRU()
} else {
c.local = lru.NewSlotLRU[string, V](opt.localSlotNum, LRUStringHash, createSimpleLRU)
c.local = lru.NewSlotLRU(opt.localSlotNum, LRUStringHash, createSimpleLRU)
}
if opt.linkSlotNum > 0 {
c.link = link.New(opt.linkSlotNum)
@@ -71,6 +71,8 @@ type cache[V any] struct {
}
func (c *cache[V]) onEvict(key string, value V) {
_ = value
if c.link != nil {
lks := c.link.Del(key)
for k := range lks {
+7 -2
View File
@@ -15,10 +15,11 @@
package localcache
import (
"github.com/openimsdk/open-im-server/v3/pkg/common/config"
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/cachekey"
"strings"
"sync"
"github.com/openimsdk/open-im-server/v3/pkg/common/config"
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/cachekey"
)
var (
@@ -32,6 +33,10 @@ func InitLocalCache(localCache *config.LocalCache) {
Local config.CacheConfig
Keys []string
}{
{
Local: localCache.Auth,
Keys: []string{cachekey.UidPidToken},
},
{
Local: localCache.User,
Keys: []string{cachekey.UserInfoKey, cachekey.UserGlobalRecvMsgOptKey},
+31 -25
View File
@@ -21,25 +21,25 @@ import (
"github.com/hashicorp/golang-lru/v2/simplelru"
)
type layLruItem[V any] struct {
type lazyLruItem[V any] struct {
lock sync.Mutex
expires int64
err error
value V
}
func NewLayLRU[K comparable, V any](size int, successTTL, failedTTL time.Duration, target Target, onEvict EvictCallback[K, V]) *LayLRU[K, V] {
var cb simplelru.EvictCallback[K, *layLruItem[V]]
func NewLazyLRU[K comparable, V any](size int, successTTL, failedTTL time.Duration, target Target, onEvict EvictCallback[K, V]) *LazyLRU[K, V] {
var cb simplelru.EvictCallback[K, *lazyLruItem[V]]
if onEvict != nil {
cb = func(key K, value *layLruItem[V]) {
cb = func(key K, value *lazyLruItem[V]) {
onEvict(key, value.value)
}
}
core, err := simplelru.NewLRU[K, *layLruItem[V]](size, cb)
core, err := simplelru.NewLRU[K, *lazyLruItem[V]](size, cb)
if err != nil {
panic(err)
}
return &LayLRU[K, V]{
return &LazyLRU[K, V]{
core: core,
successTTL: successTTL,
failedTTL: failedTTL,
@@ -47,15 +47,15 @@ func NewLayLRU[K comparable, V any](size int, successTTL, failedTTL time.Duratio
}
}
type LayLRU[K comparable, V any] struct {
type LazyLRU[K comparable, V any] struct {
lock sync.Mutex
core *simplelru.LRU[K, *layLruItem[V]]
core *simplelru.LRU[K, *lazyLruItem[V]]
successTTL time.Duration
failedTTL time.Duration
target Target
}
func (x *LayLRU[K, V]) Get(key K, fetch func() (V, error)) (V, error) {
func (x *LazyLRU[K, V]) Get(key K, fetch func() (V, error)) (V, error) {
x.lock.Lock()
v, ok := x.core.Get(key)
if ok {
@@ -68,7 +68,7 @@ func (x *LayLRU[K, V]) Get(key K, fetch func() (V, error)) (V, error) {
return value, err
}
} else {
v = &layLruItem[V]{}
v = &lazyLruItem[V]{}
x.core.Add(key, v)
v.lock.Lock()
x.lock.Unlock()
@@ -88,15 +88,15 @@ func (x *LayLRU[K, V]) Get(key K, fetch func() (V, error)) (V, error) {
return v.value, v.err
}
func (x *LayLRU[K, V]) GetBatch(keys []K, fetch func(keys []K) (map[K]V, error)) (map[K]V, error) {
func (x *LazyLRU[K, V]) GetBatch(keys []K, fetch func(keys []K) (map[K]V, error)) (map[K]V, error) {
var (
err error
once sync.Once
)
res := make(map[K]V)
queries := make([]K, 0)
setVs := make(map[K]*layLruItem[V])
queries := make([]K, 0, len(keys))
for _, key := range keys {
x.lock.Lock()
v, ok := x.core.Get(key)
@@ -118,14 +118,20 @@ func (x *LayLRU[K, V]) GetBatch(keys []K, fetch func(keys []K) (map[K]V, error))
}
queries = append(queries, key)
}
values, err1 := fetch(queries)
if err1 != nil {
if len(queries) == 0 {
return res, err
}
values, fetchErr := fetch(queries)
if fetchErr != nil {
once.Do(func() {
err = err1
err = fetchErr
})
}
for key, val := range values {
v := &layLruItem[V]{}
v := &lazyLruItem[V]{}
v.value = val
if err == nil {
@@ -135,7 +141,7 @@ func (x *LayLRU[K, V]) GetBatch(keys []K, fetch func(keys []K) (map[K]V, error))
v.expires = time.Now().Add(x.failedTTL).UnixMilli()
x.target.IncrGetFailed()
}
setVs[key] = v
x.lock.Lock()
x.core.Add(key, v)
x.lock.Unlock()
@@ -145,29 +151,29 @@ func (x *LayLRU[K, V]) GetBatch(keys []K, fetch func(keys []K) (map[K]V, error))
return res, err
}
//func (x *LayLRU[K, V]) Has(key K) bool {
//func (x *LazyLRU[K, V]) Has(key K) bool {
// x.lock.Lock()
// defer x.lock.Unlock()
// return x.core.Contains(key)
//}
func (x *LayLRU[K, V]) Set(key K, value V) {
func (x *LazyLRU[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()})
x.core.Add(key, &lazyLruItem[V]{value: value, expires: time.Now().Add(x.successTTL).UnixMilli()})
}
func (x *LayLRU[K, V]) SetHas(key K, value V) bool {
func (x *LazyLRU[K, V]) SetHas(key K, value V) bool {
x.lock.Lock()
defer x.lock.Unlock()
if x.core.Contains(key) {
x.core.Add(key, &layLruItem[V]{value: value, expires: time.Now().Add(x.successTTL).UnixMilli()})
x.core.Add(key, &lazyLruItem[V]{value: value, expires: time.Now().Add(x.successTTL).UnixMilli()})
return true
}
return false
}
func (x *LayLRU[K, V]) Del(key K) bool {
func (x *LazyLRU[K, V]) Del(key K) bool {
x.lock.Lock()
ok := x.core.Remove(key)
x.lock.Unlock()
@@ -179,6 +185,6 @@ func (x *LayLRU[K, V]) Del(key K) bool {
return ok
}
func (x *LayLRU[K, V]) Stop() {
func (x *LazyLRU[K, V]) Stop() {
}
+3 -3
View File
@@ -35,7 +35,7 @@ type slotLRU[K comparable, V any] struct {
func (x *slotLRU[K, V]) GetBatch(keys []K, fetch func(keys []K) (map[K]V, error)) (map[K]V, error) {
var (
slotKeys = make(map[uint64][]K)
vs = make(map[K]V)
kVs = make(map[K]V)
)
for _, k := range keys {
@@ -49,10 +49,10 @@ func (x *slotLRU[K, V]) GetBatch(keys []K, fetch func(keys []K) (map[K]V, error)
return nil, err
}
for key, value := range batches {
vs[key] = value
kVs[key] = value
}
}
return vs, nil
return kVs, nil
}
func (x *slotLRU[K, V]) getIndex(k K) uint64 {
+69
View File
@@ -0,0 +1,69 @@
package rpccache
import (
"context"
"time"
"github.com/openimsdk/open-im-server/v3/pkg/common/config"
"github.com/openimsdk/open-im-server/v3/pkg/common/convert"
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/cachekey"
"github.com/openimsdk/open-im-server/v3/pkg/localcache"
"github.com/openimsdk/open-im-server/v3/pkg/rpcli"
"github.com/openimsdk/protocol/auth"
"github.com/openimsdk/tools/log"
"github.com/redis/go-redis/v9"
)
func NewAuthLocalCache(client *rpcli.AuthClient, localCache *config.LocalCache, cli redis.UniversalClient) *AuthLocalCache {
lc := localCache.Auth
log.ZDebug(context.Background(), "AuthLocalCache", "topic", lc.Topic, "slotNum", lc.SlotNum, "slotSize", lc.SlotSize, "enable", lc.Enable())
x := &AuthLocalCache{
client: client,
local: localcache.New[[]byte](
localcache.WithLocalSlotNum(lc.SlotNum),
localcache.WithLocalSlotSize(lc.SlotSize),
localcache.WithLinkSlotNum(lc.SlotNum),
localcache.WithLocalSuccessTTL(lc.Success()),
localcache.WithLocalFailedTTL(lc.Failed()),
),
}
if lc.Enable() {
go subscriberRedisDeleteCache(context.Background(), cli, lc.Topic, x.local.DelLocal)
}
return x
}
type AuthLocalCache struct {
client *rpcli.AuthClient
local localcache.Cache[[]byte]
}
func (a *AuthLocalCache) GetExistingToken(ctx context.Context, userID string, platformID int) (val map[string]int, err error) {
resp, err := a.getExistingToken(ctx, userID, platformID)
if err != nil {
return nil, err
}
res := convert.TokenMapPb2DB(resp.TokenStates)
return res, nil
}
func (a *AuthLocalCache) getExistingToken(ctx context.Context, userID string, platformID int) (val *auth.GetExistingTokenResp, err error) {
start := time.Now()
log.ZDebug(ctx, "AuthLocalCache GetExistingToken req", "userID", userID, "platformID", platformID)
defer func() {
if err != nil {
log.ZError(ctx, "AuthLocalCache GetExistingToken error", err, "cost", time.Since(start), "userID", userID, "platformID", platformID)
} else {
log.ZDebug(ctx, "AuthLocalCache GetExistingToken resp", "cost", time.Since(start), "userID", userID, "platformID", platformID, "val", val)
}
}()
var cache cacheProto[auth.GetExistingTokenResp]
return cache.Unmarshal(a.local.Get(ctx, cachekey.GetTokenKey(userID, platformID), func(ctx context.Context) ([]byte, error) {
log.ZDebug(ctx, "AuthLocalCache GetExistingToken call rpc", "userID", userID, "platformID", platformID)
return cache.Marshal(a.client.AuthClient.GetExistingToken(ctx, &auth.GetExistingTokenReq{UserID: userID, PlatformID: int32(platformID)}))
}))
}
+1 -1
View File
@@ -64,7 +64,7 @@ func NewOnlineCache(client *rpcli.UserClient, group *GroupLocalCache, rdb redis.
case false:
log.ZDebug(ctx, "fullUserCache is 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) {})
return lru.NewLazyLRU[string, []int32](2048, cachekey.OnlineExpire/2, time.Second*3, localcache.EmptyTarget{}, func(key string, value []int32) {})
})
x.CurrentPhase.Store(DoSubscribeOver)
x.Cond.Broadcast()