feat: local cache

pkg/common/db/cache/black.go

@@ -16,7 +16,6 @@ package cache
 
 import (
 	"context"
-	"github.com/OpenIMSDK/tools/log"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/cachekey"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"time"
@@ -56,9 +55,7 @@ func NewBlackCacheRedis(
 ) BlackCache {
 	rcClient := rockscache.NewClient(rdb, options)
 	mc := NewMetaCacheRedis(rcClient)
-	f := config.Config.LocalCache.Friend
-	log.ZDebug(context.Background(), "black local cache init", "Topic", f.Topic, "SlotNum", f.SlotNum, "SlotSize", f.SlotSize)
-	mc.SetTopic(f.Topic)
+	mc.SetTopic(config.Config.LocalCache.Friend.Topic)
 	mc.SetRawRedisClient(rdb)
 	return &BlackCacheRedis{
 		expireTime: blackExpireTime,
@@ -73,7 +70,7 @@ func (b *BlackCacheRedis) NewCache() BlackCache {
 		expireTime: b.expireTime,
 		rcClient:   b.rcClient,
 		blackDB:    b.blackDB,
-		metaCache:  b.metaCache.Copy(),
+		metaCache:  b.Copy(),
 	}
 }
 

pkg/common/db/cache/config.go

@@ -0,0 +1,62 @@
+package cache
+
+import (
+	"github.com/openimsdk/open-im-server/v3/pkg/common/cachekey"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
+	"strings"
+	"sync"
+)
+
+var (
+	once      sync.Once
+	subscribe map[string][]string
+)
+
+func getPublishKey(topic string, key []string) []string {
+	if topic == "" || len(key) == 0 {
+		return nil
+	}
+	once.Do(func() {
+		list := []struct {
+			Local config.LocalCache
+			Keys  []string
+		}{
+			{
+				Local: config.Config.LocalCache.Group,
+				Keys:  []string{cachekey.GroupMemberIDsKey},
+			},
+			{
+				Local: config.Config.LocalCache.Friend,
+				Keys:  []string{cachekey.FriendIDsKey, cachekey.BlackIDsKey},
+			},
+			{
+				Local: config.Config.LocalCache.Conversation,
+				Keys:  []string{cachekey.ConversationIDsKey},
+			},
+		}
+		subscribe = make(map[string][]string)
+		for _, v := range list {
+			if v.Local.Enable() {
+				subscribe[v.Local.Topic] = v.Keys
+			}
+		}
+	})
+	prefix, ok := subscribe[topic]
+	if !ok {
+		return nil
+	}
+	res := make([]string, 0, len(key))
+	for _, k := range key {
+		var exist bool
+		for _, p := range prefix {
+			if strings.HasPrefix(k, p) {
+				exist = true
+				break
+			}
+		}
+		if exist {
+			res = append(res, k)
+		}
+	}
+	return res
+}

pkg/common/db/cache/conversation.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"errors"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/cachekey"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"math/big"
 	"strings"
 	"time"
@@ -85,10 +86,12 @@ type ConversationCache interface {
 
 func NewConversationRedis(rdb redis.UniversalClient, opts rockscache.Options, db relationtb.ConversationModelInterface) ConversationCache {
 	rcClient := rockscache.NewClient(rdb, opts)
-
+	mc := NewMetaCacheRedis(rcClient)
+	mc.SetTopic(config.Config.LocalCache.Conversation.Topic)
+	mc.SetRawRedisClient(rdb)
 	return &ConversationRedisCache{
 		rcClient:       rcClient,
-		metaCache:      NewMetaCacheRedis(rcClient),
+		metaCache:      mc,
 		conversationDB: db,
 		expireTime:     conversationExpireTime,
 	}
@@ -119,7 +122,7 @@ type ConversationRedisCache struct {
 func (c *ConversationRedisCache) NewCache() ConversationCache {
 	return &ConversationRedisCache{
 		rcClient:       c.rcClient,
-		metaCache:      NewMetaCacheRedis(c.rcClient, c.metaCache.GetPreDelKeys()...),
+		metaCache:      c.Copy(),
 		conversationDB: c.conversationDB,
 		expireTime:     c.expireTime,
 	}

pkg/common/db/cache/friend.go

@@ -80,7 +80,7 @@ func NewFriendCacheRedis(rdb redis.UniversalClient, friendDB relationtb.FriendMo
 func (f *FriendCacheRedis) NewCache() FriendCache {
 	return &FriendCacheRedis{
 		rcClient:   f.rcClient,
-		metaCache:  f.metaCache.Copy(),
+		metaCache:  f.Copy(),
 		friendDB:   f.friendDB,
 		expireTime: f.expireTime,
 	}

pkg/common/db/cache/group.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"fmt"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/cachekey"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"time"
 
 	"github.com/OpenIMSDK/protocol/constant"
@@ -104,12 +105,14 @@ func NewGroupCacheRedis(
 	opts rockscache.Options,
 ) GroupCache {
 	rcClient := rockscache.NewClient(rdb, opts)
-
+	mc := NewMetaCacheRedis(rcClient)
+	mc.SetTopic(config.Config.LocalCache.Group.Topic)
+	mc.SetRawRedisClient(rdb)
 	return &GroupCacheRedis{
 		rcClient: rcClient, expireTime: groupExpireTime,
 		groupDB: groupDB, groupMemberDB: groupMemberDB, groupRequestDB: groupRequestDB,
 		groupHash: hashCode,
-		metaCache: NewMetaCacheRedis(rcClient),
+		metaCache: mc,
 	}
 }
 
@@ -120,7 +123,7 @@ func (g *GroupCacheRedis) NewCache() GroupCache {
 		groupDB:        g.groupDB,
 		groupMemberDB:  g.groupMemberDB,
 		groupRequestDB: g.groupRequestDB,
-		metaCache:      NewMetaCacheRedis(g.rcClient, g.metaCache.GetPreDelKeys()...),
+		metaCache:      g.Copy(),
 	}
 }
 

pkg/common/db/cache/meta_cache.go

@@ -103,13 +103,13 @@ func (m *metaCacheRedis) ExecDel(ctx context.Context, distinct ...bool) error {
 				break
 			}
 		}
-		if m.topic != "" && m.redisClient != nil {
-			data, err := json.Marshal(m.keys)
+		if pk := getPublishKey(m.topic, m.keys); len(pk) > 0 {
+			data, err := json.Marshal(pk)
 			if err != nil {
-				log.ZError(ctx, "keys json marshal failed", err, "topic", m.topic, "keys", m.keys)
+				log.ZError(ctx, "keys json marshal failed", err, "topic", m.topic, "keys", pk)
 			} else {
 				if err := m.redisClient.Publish(ctx, m.topic, string(data)).Err(); err != nil {
-					log.ZError(ctx, "redis publish cache delete error", err, "topic", m.topic, "keys", m.keys)
+					log.ZError(ctx, "redis publish cache delete error", err, "topic", m.topic, "keys", pk)
 				}
 			}
 		}

pkg/common/db/localcache/conversation.go

@@ -1,87 +0,0 @@
-// Copyright © 2023 OpenIM. All rights reserved.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package localcache
-
-import (
-	"context"
-	"sync"
-
-	"github.com/OpenIMSDK/protocol/conversation"
-
-	"github.com/openimsdk/open-im-server/v3/pkg/rpcclient"
-)
-
-type ConversationLocalCache struct {
-	lock                              sync.Mutex
-	superGroupRecvMsgNotNotifyUserIDs map[string]Hash
-	conversationIDs                   map[string]Hash
-	client                            *rpcclient.ConversationRpcClient
-}
-
-type Hash struct {
-	hash uint64
-	ids  []string
-}
-
-func NewConversationLocalCache(client *rpcclient.ConversationRpcClient) *ConversationLocalCache {
-	return &ConversationLocalCache{
-		superGroupRecvMsgNotNotifyUserIDs: make(map[string]Hash),
-		conversationIDs:                   make(map[string]Hash),
-		client:                            client,
-	}
-}
-
-func (g *ConversationLocalCache) GetRecvMsgNotNotifyUserIDs(ctx context.Context, groupID string) ([]string, error) {
-	resp, err := g.client.Client.GetRecvMsgNotNotifyUserIDs(ctx, &conversation.GetRecvMsgNotNotifyUserIDsReq{
-		GroupID: groupID,
-	})
-	if err != nil {
-		return nil, err
-	}
-	return resp.UserIDs, nil
-}
-
-func (g *ConversationLocalCache) GetConversationIDs(ctx context.Context, userID string) ([]string, error) {
-	resp, err := g.client.Client.GetUserConversationIDsHash(ctx, &conversation.GetUserConversationIDsHashReq{
-		OwnerUserID: userID,
-	})
-	if err != nil {
-		return nil, err
-	}
-
-	g.lock.Lock()
-	hash, ok := g.conversationIDs[userID]
-	g.lock.Unlock()
-
-	if !ok || hash.hash != resp.Hash {
-		conversationIDsResp, err := g.client.Client.GetConversationIDs(ctx, &conversation.GetConversationIDsReq{
-			UserID: userID,
-		})
-		if err != nil {
-			return nil, err
-		}
-
-		g.lock.Lock()
-		defer g.lock.Unlock()
-		g.conversationIDs[userID] = Hash{
-			hash: resp.Hash,
-			ids:  conversationIDsResp.ConversationIDs,
-		}
-
-		return conversationIDsResp.ConversationIDs, nil
-	}
-
-	return hash.ids, nil
-}

pkg/common/db/localcache/doc.go

@@ -1,15 +0,0 @@
-// Copyright © 2023 OpenIM. All rights reserved.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package localcache // import "github.com/openimsdk/open-im-server/v3/pkg/common/db/localcache"

pkg/common/db/localcache/group.go

@@ -1,78 +0,0 @@
-// Copyright © 2023 OpenIM. All rights reserved.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package localcache
-
-import (
-	"context"
-	"sync"
-
-	"github.com/OpenIMSDK/protocol/group"
-	"github.com/OpenIMSDK/tools/errs"
-
-	"github.com/openimsdk/open-im-server/v3/pkg/rpcclient"
-)
-
-type GroupLocalCache struct {
-	lock   sync.Mutex
-	cache  map[string]GroupMemberIDsHash
-	client *rpcclient.GroupRpcClient
-}
-
-type GroupMemberIDsHash struct {
-	memberListHash uint64
-	userIDs        []string
-}
-
-func NewGroupLocalCache(client *rpcclient.GroupRpcClient) *GroupLocalCache {
-	return &GroupLocalCache{
-		cache:  make(map[string]GroupMemberIDsHash, 0),
-		client: client,
-	}
-}
-
-func (g *GroupLocalCache) GetGroupMemberIDs(ctx context.Context, groupID string) ([]string, error) {
-	resp, err := g.client.Client.GetGroupAbstractInfo(ctx, &group.GetGroupAbstractInfoReq{
-		GroupIDs: []string{groupID},
-	})
-	if err != nil {
-		return nil, err
-	}
-	if len(resp.GroupAbstractInfos) < 1 {
-		return nil, errs.ErrGroupIDNotFound
-	}
-
-	g.lock.Lock()
-	localHashInfo, ok := g.cache[groupID]
-	if ok && localHashInfo.memberListHash == resp.GroupAbstractInfos[0].GroupMemberListHash {
-		g.lock.Unlock()
-		return localHashInfo.userIDs, nil
-	}
-	g.lock.Unlock()
-
-	groupMembersResp, err := g.client.Client.GetGroupMemberUserIDs(ctx, &group.GetGroupMemberUserIDsReq{
-		GroupID: groupID,
-	})
-	if err != nil {
-		return nil, err
-	}
-
-	g.lock.Lock()
-	defer g.lock.Unlock()
-	g.cache[groupID] = GroupMemberIDsHash{
-		memberListHash: resp.GroupAbstractInfos[0].GroupMemberListHash,
-		userIDs:        groupMembersResp.UserIDs,
-	}
-	return g.cache[groupID].userIDs, nil
-}

pkg/common/db/localcache/meta_local_cache.go

@@ -1,15 +0,0 @@
-// Copyright © 2023 OpenIM. All rights reserved.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package localcache