Merge branch 'main' of https://github.com/openimsdk/open-im-server
skiffer-git
6 months ago
commit
eb766ebd94
@ -1,371 +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 cache
|
||||
|
||||
import (
|
||||
"context"
|
||||
"math/big"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/dtm-labs/rockscache"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/cachekey"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/config"
|
||||
relationtb "github.com/openimsdk/open-im-server/v3/pkg/common/db/table/relation"
|
||||
"github.com/openimsdk/tools/log"
|
||||
"github.com/openimsdk/tools/utils/datautil"
|
||||
"github.com/openimsdk/tools/utils/encrypt"
|
||||
"github.com/redis/go-redis/v9"
|
||||
)
|
||||
|
||||
const (
|
||||
// ConversationKey = "CONVERSATION:"
|
||||
// conversationIDsKey = "CONVERSATION_IDS:"
|
||||
// conversationIDsHashKey = "CONVERSATION_IDS_HASH:"
|
||||
// conversationHasReadSeqKey = "CONVERSATION_HAS_READ_SEQ:"
|
||||
// recvMsgOptKey = "RECV_MSG_OPT:"
|
||||
// superGroupRecvMsgNotNotifyUserIDsKey = "SUPER_GROUP_RECV_MSG_NOT_NOTIFY_USER_IDS:"
|
||||
// superGroupRecvMsgNotNotifyUserIDsHashKey = "SUPER_GROUP_RECV_MSG_NOT_NOTIFY_USER_IDS_HASH:"
|
||||
// conversationNotReceiveMessageUserIDsKey = "CONVERSATION_NOT_RECEIVE_MESSAGE_USER_IDS:".
|
||||
|
||||
conversationExpireTime = time.Second * 60 * 60 * 12
|
||||
)
|
||||
|
||||
// arg fn will exec when no data in msgCache.
|
||||
type ConversationCache interface {
|
||||
metaCache
|
||||
NewCache() ConversationCache
|
||||
// get user's conversationIDs from msgCache
|
||||
GetUserConversationIDs(ctx context.Context, ownerUserID string) ([]string, error)
|
||||
DelConversationIDs(userIDs ...string) ConversationCache
|
||||
|
||||
GetUserConversationIDsHash(ctx context.Context, ownerUserID string) (hash uint64, err error)
|
||||
DelUserConversationIDsHash(ownerUserIDs ...string) ConversationCache
|
||||
|
||||
// get one conversation from msgCache
|
||||
GetConversation(ctx context.Context, ownerUserID, conversationID string) (*relationtb.ConversationModel, error)
|
||||
DelConversations(ownerUserID string, conversationIDs ...string) ConversationCache
|
||||
DelUsersConversation(conversationID string, ownerUserIDs ...string) ConversationCache
|
||||
// get one conversation from msgCache
|
||||
GetConversations(ctx context.Context, ownerUserID string,
|
||||
conversationIDs []string) ([]*relationtb.ConversationModel, error)
|
||||
// get one user's all conversations from msgCache
|
||||
GetUserAllConversations(ctx context.Context, ownerUserID string) ([]*relationtb.ConversationModel, error)
|
||||
// get user conversation recv msg from msgCache
|
||||
GetUserRecvMsgOpt(ctx context.Context, ownerUserID, conversationID string) (opt int, err error)
|
||||
DelUserRecvMsgOpt(ownerUserID, conversationID string) ConversationCache
|
||||
// get one super group recv msg but do not notification userID list
|
||||
// GetSuperGroupRecvMsgNotNotifyUserIDs(ctx context.Context, groupID string) (userIDs []string, err error)
|
||||
DelSuperGroupRecvMsgNotNotifyUserIDs(groupID string) ConversationCache
|
||||
// get one super group recv msg but do not notification userID list hash
|
||||
// GetSuperGroupRecvMsgNotNotifyUserIDsHash(ctx context.Context, groupID string) (hash uint64, err error)
|
||||
DelSuperGroupRecvMsgNotNotifyUserIDsHash(groupID string) ConversationCache
|
||||
|
||||
// GetUserAllHasReadSeqs(ctx context.Context, ownerUserID string) (map[string]int64, error)
|
||||
DelUserAllHasReadSeqs(ownerUserID string, conversationIDs ...string) ConversationCache
|
||||
|
||||
GetConversationsByConversationID(ctx context.Context,
|
||||
conversationIDs []string) ([]*relationtb.ConversationModel, error)
|
||||
DelConversationByConversationID(conversationIDs ...string) ConversationCache
|
||||
GetConversationNotReceiveMessageUserIDs(ctx context.Context, conversationID string) ([]string, error)
|
||||
DelConversationNotReceiveMessageUserIDs(conversationIDs ...string) ConversationCache
|
||||
}
|
||||
|
||||
func NewConversationRedis(rdb redis.UniversalClient, localCache *config.LocalCache, opts rockscache.Options, db relationtb.ConversationModelInterface) ConversationCache {
|
||||
rcClient := rockscache.NewClient(rdb, opts)
|
||||
mc := NewMetaCacheRedis(rcClient)
|
||||
c := localCache.Conversation
|
||||
log.ZDebug(context.Background(), "black local cache init", "Topic", c.Topic, "SlotNum", c.SlotNum, "SlotSize", c.SlotSize, "enable", c.Enable())
|
||||
mc.SetTopic(c.Topic)
|
||||
mc.SetRawRedisClient(rdb)
|
||||
return &ConversationRedisCache{
|
||||
rcClient: rcClient,
|
||||
metaCache: mc,
|
||||
conversationDB: db,
|
||||
expireTime: conversationExpireTime,
|
||||
}
|
||||
}
|
||||
|
||||
type ConversationRedisCache struct {
|
||||
metaCache
|
||||
rcClient *rockscache.Client
|
||||
conversationDB relationtb.ConversationModelInterface
|
||||
expireTime time.Duration
|
||||
}
|
||||
|
||||
// func NewNewConversationRedis(
|
||||
// rdb redis.UniversalClient,
|
||||
// conversationDB *relation.ConversationGorm,
|
||||
// options rockscache.Options,
|
||||
// ) ConversationCache {
|
||||
// rcClient := rockscache.NewClient(rdb, options)
|
||||
//
|
||||
// return &ConversationRedisCache{
|
||||
// rcClient: rcClient,
|
||||
// metaCache: NewMetaCacheRedis(rcClient),
|
||||
// conversationDB: conversationDB,
|
||||
// expireTime: conversationExpireTime,
|
||||
// }
|
||||
//}
|
||||
|
||||
func (c *ConversationRedisCache) NewCache() ConversationCache {
|
||||
return &ConversationRedisCache{
|
||||
rcClient: c.rcClient,
|
||||
metaCache: c.Copy(),
|
||||
conversationDB: c.conversationDB,
|
||||
expireTime: c.expireTime,
|
||||
}
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) getConversationKey(ownerUserID, conversationID string) string {
|
||||
return cachekey.GetConversationKey(ownerUserID, conversationID)
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) getConversationIDsKey(ownerUserID string) string {
|
||||
return cachekey.GetConversationIDsKey(ownerUserID)
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) getSuperGroupRecvNotNotifyUserIDsKey(groupID string) string {
|
||||
return cachekey.GetSuperGroupRecvNotNotifyUserIDsKey(groupID)
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) getRecvMsgOptKey(ownerUserID, conversationID string) string {
|
||||
return cachekey.GetRecvMsgOptKey(ownerUserID, conversationID)
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) getSuperGroupRecvNotNotifyUserIDsHashKey(groupID string) string {
|
||||
return cachekey.GetSuperGroupRecvNotNotifyUserIDsHashKey(groupID)
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) getConversationHasReadSeqKey(ownerUserID, conversationID string) string {
|
||||
return cachekey.GetConversationHasReadSeqKey(ownerUserID, conversationID)
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) getConversationNotReceiveMessageUserIDsKey(conversationID string) string {
|
||||
return cachekey.GetConversationNotReceiveMessageUserIDsKey(conversationID)
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) getUserConversationIDsHashKey(ownerUserID string) string {
|
||||
return cachekey.GetUserConversationIDsHashKey(ownerUserID)
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) GetUserConversationIDs(ctx context.Context, ownerUserID string) ([]string, error) {
|
||||
return getCache(ctx, c.rcClient, c.getConversationIDsKey(ownerUserID), c.expireTime, func(ctx context.Context) ([]string, error) {
|
||||
return c.conversationDB.FindUserIDAllConversationID(ctx, ownerUserID)
|
||||
})
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) DelConversationIDs(userIDs ...string) ConversationCache {
|
||||
keys := make([]string, 0, len(userIDs))
|
||||
for _, userID := range userIDs {
|
||||
keys = append(keys, c.getConversationIDsKey(userID))
|
||||
}
|
||||
cache := c.NewCache()
|
||||
cache.AddKeys(keys...)
|
||||
|
||||
return cache
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) GetUserConversationIDsHash(ctx context.Context, ownerUserID string) (hash uint64, err error) {
|
||||
return getCache(
|
||||
ctx,
|
||||
c.rcClient,
|
||||
c.getUserConversationIDsHashKey(ownerUserID),
|
||||
c.expireTime,
|
||||
func(ctx context.Context) (uint64, error) {
|
||||
conversationIDs, err := c.GetUserConversationIDs(ctx, ownerUserID)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
datautil.Sort(conversationIDs, true)
|
||||
bi := big.NewInt(0)
|
||||
bi.SetString(encrypt.Md5(strings.Join(conversationIDs, ";"))[0:8], 16)
|
||||
return bi.Uint64(), nil
|
||||
},
|
||||
)
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) DelUserConversationIDsHash(ownerUserIDs ...string) ConversationCache {
|
||||
keys := make([]string, 0, len(ownerUserIDs))
|
||||
for _, ownerUserID := range ownerUserIDs {
|
||||
keys = append(keys, c.getUserConversationIDsHashKey(ownerUserID))
|
||||
}
|
||||
cache := c.NewCache()
|
||||
cache.AddKeys(keys...)
|
||||
|
||||
return cache
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) GetConversation(ctx context.Context, ownerUserID, conversationID string) (*relationtb.ConversationModel, error) {
|
||||
return getCache(ctx, c.rcClient, c.getConversationKey(ownerUserID, conversationID), c.expireTime, func(ctx context.Context) (*relationtb.ConversationModel, error) {
|
||||
return c.conversationDB.Take(ctx, ownerUserID, conversationID)
|
||||
})
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) DelConversations(ownerUserID string, conversationIDs ...string) ConversationCache {
|
||||
keys := make([]string, 0, len(conversationIDs))
|
||||
for _, conversationID := range conversationIDs {
|
||||
keys = append(keys, c.getConversationKey(ownerUserID, conversationID))
|
||||
}
|
||||
cache := c.NewCache()
|
||||
cache.AddKeys(keys...)
|
||||
|
||||
return cache
|
||||
}
|
||||
|
||||
// func (c *ConversationRedisCache) getConversationIndex(convsation *relationtb.ConversationModel, keys []string) (int, error) {
|
||||
// key := c.getConversationKey(convsation.OwnerUserID, convsation.ConversationID)
|
||||
// for _i, _key := range keys {
|
||||
// if _key == key {
|
||||
// return _i, nil
|
||||
// }
|
||||
// }
|
||||
|
||||
// return 0, errs.New("not found key:" + key + " in keys")
|
||||
// }
|
||||
|
||||
func (c *ConversationRedisCache) GetConversations(ctx context.Context, ownerUserID string, conversationIDs []string) ([]*relationtb.ConversationModel, error) {
|
||||
// var keys []string
|
||||
// for _, conversarionID := range conversationIDs {
|
||||
// keys = append(keys, c.getConversationKey(ownerUserID, conversarionID))
|
||||
//}
|
||||
// return batchGetCache(
|
||||
// ctx,
|
||||
// c.rcClient,
|
||||
// keys,
|
||||
// c.expireTime,
|
||||
// c.getConversationIndex,
|
||||
// func(ctx context.Context) ([]*relationtb.ConversationModel, error) {
|
||||
// return c.conversationDB.Find(ctx, ownerUserID, conversationIDs)
|
||||
// },
|
||||
//)
|
||||
return batchGetCache2(ctx, c.rcClient, c.expireTime, conversationIDs, func(conversationID string) string {
|
||||
return c.getConversationKey(ownerUserID, conversationID)
|
||||
}, func(ctx context.Context, conversationID string) (*relationtb.ConversationModel, error) {
|
||||
return c.conversationDB.Take(ctx, ownerUserID, conversationID)
|
||||
})
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) GetUserAllConversations(ctx context.Context, ownerUserID string) ([]*relationtb.ConversationModel, error) {
|
||||
conversationIDs, err := c.GetUserConversationIDs(ctx, ownerUserID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
// var keys []string
|
||||
// for _, conversarionID := range conversationIDs {
|
||||
// keys = append(keys, c.getConversationKey(ownerUserID, conversarionID))
|
||||
//}
|
||||
// return batchGetCache(
|
||||
// ctx,
|
||||
// c.rcClient,
|
||||
// keys,
|
||||
// c.expireTime,
|
||||
// c.getConversationIndex,
|
||||
// func(ctx context.Context) ([]*relationtb.ConversationModel, error) {
|
||||
// return c.conversationDB.FindUserIDAllConversations(ctx, ownerUserID)
|
||||
// },
|
||||
//)
|
||||
return c.GetConversations(ctx, ownerUserID, conversationIDs)
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) GetUserRecvMsgOpt(ctx context.Context, ownerUserID, conversationID string) (opt int, err error) {
|
||||
return getCache(ctx, c.rcClient, c.getRecvMsgOptKey(ownerUserID, conversationID), c.expireTime, func(ctx context.Context) (opt int, err error) {
|
||||
return c.conversationDB.GetUserRecvMsgOpt(ctx, ownerUserID, conversationID)
|
||||
})
|
||||
}
|
||||
|
||||
// func (c *ConversationRedisCache) GetSuperGroupRecvMsgNotNotifyUserIDs(ctx context.Context, groupID string) (userIDs []string, err error) {
|
||||
// return getCache(ctx, c.rcClient, c.getSuperGroupRecvNotNotifyUserIDsKey(groupID), c.expireTime, func(ctx context.Context) (userIDs []string, err error) {
|
||||
// return c.conversationDB.FindSuperGroupRecvMsgNotNotifyUserIDs(ctx, groupID)
|
||||
// })
|
||||
//}
|
||||
|
||||
func (c *ConversationRedisCache) DelUsersConversation(conversationID string, ownerUserIDs ...string) ConversationCache {
|
||||
keys := make([]string, 0, len(ownerUserIDs))
|
||||
for _, ownerUserID := range ownerUserIDs {
|
||||
keys = append(keys, c.getConversationKey(ownerUserID, conversationID))
|
||||
}
|
||||
cache := c.NewCache()
|
||||
cache.AddKeys(keys...)
|
||||
|
||||
return cache
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) DelUserRecvMsgOpt(ownerUserID, conversationID string) ConversationCache {
|
||||
cache := c.NewCache()
|
||||
cache.AddKeys(c.getRecvMsgOptKey(ownerUserID, conversationID))
|
||||
|
||||
return cache
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) DelSuperGroupRecvMsgNotNotifyUserIDs(groupID string) ConversationCache {
|
||||
cache := c.NewCache()
|
||||
cache.AddKeys(c.getSuperGroupRecvNotNotifyUserIDsKey(groupID))
|
||||
|
||||
return cache
|
||||
}
|
||||
|
||||
// func (c *ConversationRedisCache) GetSuperGroupRecvMsgNotNotifyUserIDsHash(ctx context.Context, groupID string) (hash uint64, err error) {
|
||||
// return getCache(ctx, c.rcClient, c.getSuperGroupRecvNotNotifyUserIDsHashKey(groupID), c.expireTime, func(ctx context.Context) (hash uint64, err error) {
|
||||
// userIDs, err := c.GetSuperGroupRecvMsgNotNotifyUserIDs(ctx, groupID)
|
||||
// if err != nil {
|
||||
// return 0, err
|
||||
// }
|
||||
// utils.Sort(userIDs, true)
|
||||
// bi := big.NewInt(0)
|
||||
// bi.SetString(utils.Md5(strings.Join(userIDs, ";"))[0:8], 16)
|
||||
// return bi.Uint64(), nil
|
||||
// },
|
||||
// )
|
||||
//}
|
||||
|
||||
func (c *ConversationRedisCache) DelSuperGroupRecvMsgNotNotifyUserIDsHash(groupID string) ConversationCache {
|
||||
cache := c.NewCache()
|
||||
cache.AddKeys(c.getSuperGroupRecvNotNotifyUserIDsHashKey(groupID))
|
||||
|
||||
return cache
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) DelUserAllHasReadSeqs(ownerUserID string, conversationIDs ...string) ConversationCache {
|
||||
cache := c.NewCache()
|
||||
for _, conversationID := range conversationIDs {
|
||||
cache.AddKeys(c.getConversationHasReadSeqKey(ownerUserID, conversationID))
|
||||
}
|
||||
|
||||
return cache
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) GetConversationsByConversationID(ctx context.Context, conversationIDs []string) ([]*relationtb.ConversationModel, error) {
|
||||
panic("implement me")
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) DelConversationByConversationID(conversationIDs ...string) ConversationCache {
|
||||
panic("implement me")
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) GetConversationNotReceiveMessageUserIDs(ctx context.Context, conversationID string) ([]string, error) {
|
||||
return getCache(ctx, c.rcClient, c.getConversationNotReceiveMessageUserIDsKey(conversationID), c.expireTime, func(ctx context.Context) ([]string, error) {
|
||||
return c.conversationDB.GetConversationNotReceiveMessageUserIDs(ctx, conversationID)
|
||||
})
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) DelConversationNotReceiveMessageUserIDs(conversationIDs ...string) ConversationCache {
|
||||
cache := c.NewCache()
|
||||
for _, conversationID := range conversationIDs {
|
||||
cache.AddKeys(c.getConversationNotReceiveMessageUserIDsKey(conversationID))
|
||||
}
|
||||
|
||||
return cache
|
||||
}
|
||||
@ -1,284 +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 cache
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/dtm-labs/rockscache"
|
||||
"github.com/openimsdk/tools/errs"
|
||||
"github.com/openimsdk/tools/log"
|
||||
"github.com/openimsdk/tools/mw/specialerror"
|
||||
"github.com/openimsdk/tools/utils/datautil"
|
||||
"github.com/redis/go-redis/v9"
|
||||
)
|
||||
|
||||
const (
|
||||
scanCount = 3000
|
||||
maxRetryTimes = 5
|
||||
retryInterval = time.Millisecond * 100
|
||||
)
|
||||
|
||||
var errIndex = errs.New("err index")
|
||||
|
||||
type metaCache interface {
|
||||
ExecDel(ctx context.Context, distinct ...bool) error
|
||||
// delete key rapid
|
||||
DelKey(ctx context.Context, key string) error
|
||||
AddKeys(keys ...string)
|
||||
ClearKeys()
|
||||
GetPreDelKeys() []string
|
||||
SetTopic(topic string)
|
||||
SetRawRedisClient(cli redis.UniversalClient)
|
||||
Copy() metaCache
|
||||
}
|
||||
|
||||
func NewMetaCacheRedis(rcClient *rockscache.Client, keys ...string) metaCache {
|
||||
return &metaCacheRedis{rcClient: rcClient, keys: keys, maxRetryTimes: maxRetryTimes, retryInterval: retryInterval}
|
||||
}
|
||||
|
||||
type metaCacheRedis struct {
|
||||
topic string
|
||||
rcClient *rockscache.Client
|
||||
keys []string
|
||||
maxRetryTimes int
|
||||
retryInterval time.Duration
|
||||
redisClient redis.UniversalClient
|
||||
}
|
||||
|
||||
func (m *metaCacheRedis) Copy() metaCache {
|
||||
var keys []string
|
||||
if len(m.keys) > 0 {
|
||||
keys = make([]string, 0, len(m.keys)*2)
|
||||
keys = append(keys, m.keys...)
|
||||
}
|
||||
return &metaCacheRedis{
|
||||
topic: m.topic,
|
||||
rcClient: m.rcClient,
|
||||
keys: keys,
|
||||
maxRetryTimes: m.maxRetryTimes,
|
||||
retryInterval: m.retryInterval,
|
||||
redisClient: m.redisClient,
|
||||
}
|
||||
}
|
||||
|
||||
func (m *metaCacheRedis) SetTopic(topic string) {
|
||||
m.topic = topic
|
||||
}
|
||||
|
||||
func (m *metaCacheRedis) SetRawRedisClient(cli redis.UniversalClient) {
|
||||
m.redisClient = cli
|
||||
}
|
||||
|
||||
func (m *metaCacheRedis) ExecDel(ctx context.Context, distinct ...bool) error {
|
||||
if len(distinct) > 0 && distinct[0] {
|
||||
m.keys = datautil.Distinct(m.keys)
|
||||
}
|
||||
if len(m.keys) > 0 {
|
||||
log.ZDebug(ctx, "delete cache", "topic", m.topic, "keys", m.keys)
|
||||
for _, key := range m.keys {
|
||||
for i := 0; i < m.maxRetryTimes; i++ {
|
||||
if err := m.rcClient.TagAsDeleted(key); err != nil {
|
||||
log.ZError(ctx, "delete cache failed", err, "key", key)
|
||||
time.Sleep(m.retryInterval)
|
||||
continue
|
||||
}
|
||||
break
|
||||
}
|
||||
}
|
||||
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", 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", pk)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *metaCacheRedis) DelKey(ctx context.Context, key string) error {
|
||||
return m.rcClient.TagAsDeleted2(ctx, key)
|
||||
}
|
||||
|
||||
func (m *metaCacheRedis) AddKeys(keys ...string) {
|
||||
m.keys = append(m.keys, keys...)
|
||||
}
|
||||
|
||||
func (m *metaCacheRedis) ClearKeys() {
|
||||
m.keys = []string{}
|
||||
}
|
||||
|
||||
func (m *metaCacheRedis) GetPreDelKeys() []string {
|
||||
return m.keys
|
||||
}
|
||||
|
||||
func GetDefaultOpt() rockscache.Options {
|
||||
opts := rockscache.NewDefaultOptions()
|
||||
opts.StrongConsistency = true
|
||||
opts.RandomExpireAdjustment = 0.2
|
||||
|
||||
return opts
|
||||
}
|
||||
|
||||
func getCache[T any](ctx context.Context, rcClient *rockscache.Client, key string, expire time.Duration, fn func(ctx context.Context) (T, error)) (T, error) {
|
||||
var t T
|
||||
var write bool
|
||||
v, err := rcClient.Fetch2(ctx, key, expire, func() (s string, err error) {
|
||||
t, err = fn(ctx)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
bs, err := json.Marshal(t)
|
||||
if err != nil {
|
||||
return "", errs.WrapMsg(err, "marshal failed")
|
||||
}
|
||||
write = true
|
||||
|
||||
return string(bs), nil
|
||||
})
|
||||
if err != nil {
|
||||
return t, errs.Wrap(err)
|
||||
}
|
||||
if write {
|
||||
return t, nil
|
||||
}
|
||||
if v == "" {
|
||||
return t, errs.ErrRecordNotFound.WrapMsg("cache is not found")
|
||||
}
|
||||
err = json.Unmarshal([]byte(v), &t)
|
||||
if err != nil {
|
||||
errInfo := fmt.Sprintf("cache json.Unmarshal failed, key:%s, value:%s, expire:%s", key, v, expire)
|
||||
return t, errs.WrapMsg(err, errInfo)
|
||||
}
|
||||
|
||||
return t, nil
|
||||
}
|
||||
|
||||
// func batchGetCache[T any](ctx context.Context, rcClient *rockscache.Client, keys []string, expire time.Duration, keyIndexFn func(t T, keys []string) (int, error), fn func(ctx context.Context) ([]T,
|
||||
// error)) ([]T, error) {
|
||||
// batchMap, err := rcClient.FetchBatch2(ctx, keys, expire, func(idxs []int) (m map[int]string, err error) {
|
||||
// values := make(map[int]string)
|
||||
// tArrays, err := fn(ctx)
|
||||
// if err != nil {
|
||||
// return nil, err
|
||||
// }
|
||||
// for _, v := range tArrays {
|
||||
// index, err := keyIndexFn(v, keys)
|
||||
// if err != nil {
|
||||
// continue
|
||||
// }
|
||||
// bs, err := json.Marshal(v)
|
||||
// if err != nil {
|
||||
// return nil, utils.Wrap(err, "marshal failed")
|
||||
// }
|
||||
// values[index] = string(bs)
|
||||
// }
|
||||
// return values, nil
|
||||
// })
|
||||
// if err != nil {
|
||||
// return nil, err
|
||||
// }
|
||||
// var tArrays []T
|
||||
// for _, v := range batchMap {
|
||||
// if v != "" {
|
||||
// var t T
|
||||
// err = json.Unmarshal([]byte(v), &t)
|
||||
// if err != nil {
|
||||
// return nil, utils.Wrap(err, "unmarshal failed")
|
||||
// }
|
||||
// tArrays = append(tArrays, t)
|
||||
// }
|
||||
// }
|
||||
// return tArrays, nil
|
||||
//}
|
||||
|
||||
func batchGetCache2[T any, K comparable](
|
||||
ctx context.Context,
|
||||
rcClient *rockscache.Client,
|
||||
expire time.Duration,
|
||||
keys []K,
|
||||
keyFn func(key K) string,
|
||||
fns func(ctx context.Context, key K) (T, error),
|
||||
) ([]T, error) {
|
||||
if len(keys) == 0 {
|
||||
return nil, nil
|
||||
}
|
||||
res := make([]T, 0, len(keys))
|
||||
for _, key := range keys {
|
||||
val, err := getCache(ctx, rcClient, keyFn(key), expire, func(ctx context.Context) (T, error) {
|
||||
return fns(ctx, key)
|
||||
})
|
||||
if err != nil {
|
||||
if errs.ErrRecordNotFound.Is(specialerror.ErrCode(errs.Unwrap(err))) {
|
||||
continue
|
||||
}
|
||||
return nil, errs.Wrap(err)
|
||||
}
|
||||
res = append(res, val)
|
||||
}
|
||||
|
||||
return res, nil
|
||||
}
|
||||
|
||||
// func batchGetCacheMap[T any](
|
||||
// ctx context.Context,
|
||||
// rcClient *rockscache.Client,
|
||||
// keys, originKeys []string,
|
||||
// expire time.Duration,
|
||||
// keyIndexFn func(s string, keys []string) (int, error),
|
||||
// fn func(ctx context.Context) (map[string]T, error),
|
||||
// ) (map[string]T, error) {
|
||||
// batchMap, err := rcClient.FetchBatch2(ctx, keys, expire, func(idxs []int) (m map[int]string, err error) {
|
||||
// tArrays, err := fn(ctx)
|
||||
// if err != nil {
|
||||
// return nil, err
|
||||
// }
|
||||
// values := make(map[int]string)
|
||||
// for k, v := range tArrays {
|
||||
// index, err := keyIndexFn(k, originKeys)
|
||||
// if err != nil {
|
||||
// continue
|
||||
// }
|
||||
// bs, err := json.Marshal(v)
|
||||
// if err != nil {
|
||||
// return nil, utils.Wrap(err, "marshal failed")
|
||||
// }
|
||||
// values[index] = string(bs)
|
||||
// }
|
||||
// return values, nil
|
||||
// })
|
||||
// if err != nil {
|
||||
// return nil, err
|
||||
// }
|
||||
// tMap := make(map[string]T)
|
||||
// for i, v := range batchMap {
|
||||
// if v != "" {
|
||||
// var t T
|
||||
// err = json.Unmarshal([]byte(v), &t)
|
||||
// if err != nil {
|
||||
// return nil, utils.Wrap(err, "unmarshal failed")
|
||||
// }
|
||||
// tMap[originKeys[i]] = t
|
||||
// }
|
||||
// }
|
||||
// return tMap, nil
|
||||
//}
|
||||
@ -1,85 +0,0 @@
|
||||
package cache
|
||||
|
||||
import (
|
||||
"context"
|
||||
"github.com/openimsdk/tools/errs"
|
||||
"github.com/redis/go-redis/v9"
|
||||
"strconv"
|
||||
"time"
|
||||
)
|
||||
|
||||
type ThirdCache interface {
|
||||
SetFcmToken(ctx context.Context, account string, platformID int, fcmToken string, expireTime int64) (err error)
|
||||
GetFcmToken(ctx context.Context, account string, platformID int) (string, error)
|
||||
DelFcmToken(ctx context.Context, account string, platformID int) error
|
||||
IncrUserBadgeUnreadCountSum(ctx context.Context, userID string) (int, error)
|
||||
SetUserBadgeUnreadCountSum(ctx context.Context, userID string, value int) error
|
||||
GetUserBadgeUnreadCountSum(ctx context.Context, userID string) (int, error)
|
||||
SetGetuiToken(ctx context.Context, token string, expireTime int64) error
|
||||
GetGetuiToken(ctx context.Context) (string, error)
|
||||
SetGetuiTaskID(ctx context.Context, taskID string, expireTime int64) error
|
||||
GetGetuiTaskID(ctx context.Context) (string, error)
|
||||
}
|
||||
|
||||
func NewThirdCache(rdb redis.UniversalClient) ThirdCache {
|
||||
return &thirdCache{rdb: rdb}
|
||||
}
|
||||
|
||||
type thirdCache struct {
|
||||
rdb redis.UniversalClient
|
||||
}
|
||||
|
||||
func (c *thirdCache) SetFcmToken(ctx context.Context, account string, platformID int, fcmToken string, expireTime int64) (err error) {
|
||||
return errs.Wrap(c.rdb.Set(ctx, FCM_TOKEN+account+":"+strconv.Itoa(platformID), fcmToken, time.Duration(expireTime)*time.Second).Err())
|
||||
}
|
||||
|
||||
func (c *thirdCache) GetFcmToken(ctx context.Context, account string, platformID int) (string, error) {
|
||||
val, err := c.rdb.Get(ctx, FCM_TOKEN+account+":"+strconv.Itoa(platformID)).Result()
|
||||
if err != nil {
|
||||
return "", errs.Wrap(err)
|
||||
}
|
||||
return val, nil
|
||||
}
|
||||
|
||||
func (c *thirdCache) DelFcmToken(ctx context.Context, account string, platformID int) error {
|
||||
return errs.Wrap(c.rdb.Del(ctx, FCM_TOKEN+account+":"+strconv.Itoa(platformID)).Err())
|
||||
}
|
||||
|
||||
func (c *thirdCache) IncrUserBadgeUnreadCountSum(ctx context.Context, userID string) (int, error) {
|
||||
seq, err := c.rdb.Incr(ctx, userBadgeUnreadCountSum+userID).Result()
|
||||
|
||||
return int(seq), errs.Wrap(err)
|
||||
}
|
||||
|
||||
func (c *thirdCache) SetUserBadgeUnreadCountSum(ctx context.Context, userID string, value int) error {
|
||||
return errs.Wrap(c.rdb.Set(ctx, userBadgeUnreadCountSum+userID, value, 0).Err())
|
||||
}
|
||||
|
||||
func (c *thirdCache) GetUserBadgeUnreadCountSum(ctx context.Context, userID string) (int, error) {
|
||||
val, err := c.rdb.Get(ctx, userBadgeUnreadCountSum+userID).Int()
|
||||
return val, errs.Wrap(err)
|
||||
}
|
||||
|
||||
func (c *thirdCache) SetGetuiToken(ctx context.Context, token string, expireTime int64) error {
|
||||
return errs.Wrap(c.rdb.Set(ctx, getuiToken, token, time.Duration(expireTime)*time.Second).Err())
|
||||
}
|
||||
|
||||
func (c *thirdCache) GetGetuiToken(ctx context.Context) (string, error) {
|
||||
val, err := c.rdb.Get(ctx, getuiToken).Result()
|
||||
if err != nil {
|
||||
return "", errs.Wrap(err)
|
||||
}
|
||||
return val, nil
|
||||
}
|
||||
|
||||
func (c *thirdCache) SetGetuiTaskID(ctx context.Context, taskID string, expireTime int64) error {
|
||||
return errs.Wrap(c.rdb.Set(ctx, getuiTaskID, taskID, time.Duration(expireTime)*time.Second).Err())
|
||||
}
|
||||
|
||||
func (c *thirdCache) GetGetuiTaskID(ctx context.Context) (string, error) {
|
||||
val, err := c.rdb.Get(ctx, getuiTaskID).Result()
|
||||
if err != nil {
|
||||
return "", errs.Wrap(err)
|
||||
}
|
||||
return val, nil
|
||||
}
|
||||
@ -1,15 +0,0 @@
|
||||
// Copyright © 2024 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 relation // import "github.com/openimsdk/open-im-server/v3/pkg/common/db/table/relation"
|
||||
@ -0,0 +1,17 @@
|
||||
package cache
|
||||
|
||||
import (
|
||||
"context"
|
||||
)
|
||||
|
||||
// BatchDeleter interface defines a set of methods for batch deleting cache and publishing deletion information.
|
||||
type BatchDeleter interface {
|
||||
//ChainExecDel method is used for chain calls and must call Clone to prevent memory pollution.
|
||||
ChainExecDel(ctx context.Context) error
|
||||
//ExecDelWithKeys method directly takes keys for deletion.
|
||||
ExecDelWithKeys(ctx context.Context, keys []string) error
|
||||
//Clone method creates a copy of the BatchDeleter to avoid modifying the original object.
|
||||
Clone() BatchDeleter
|
||||
//AddKeys method adds keys to be deleted.
|
||||
AddKeys(keys ...string)
|
||||
}
|
||||
@ -0,0 +1,27 @@
|
||||
// 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 cache
|
||||
|
||||
import (
|
||||
"context"
|
||||
)
|
||||
|
||||
type BlackCache interface {
|
||||
BatchDeleter
|
||||
CloneBlackCache() BlackCache
|
||||
GetBlackIDs(ctx context.Context, userID string) (blackIDs []string, err error)
|
||||
// del user's blackIDs msgCache, exec when a user's black list changed
|
||||
DelBlackIDs(ctx context.Context, userID string) BlackCache
|
||||
}
|
||||
@ -0,0 +1,70 @@
|
||||
// Copyright © 2024 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 cachekey
|
||||
|
||||
import (
|
||||
"github.com/openimsdk/protocol/constant"
|
||||
"strconv"
|
||||
)
|
||||
|
||||
const (
|
||||
messageCache = "MESSAGE_CACHE:"
|
||||
messageDelUserList = "MESSAGE_DEL_USER_LIST:"
|
||||
userDelMessagesList = "USER_DEL_MESSAGES_LIST:"
|
||||
sendMsgFailedFlag = "SEND_MSG_FAILED_FLAG:"
|
||||
exTypeKeyLocker = "EX_LOCK:"
|
||||
reactionExSingle = "EX_SINGLE_"
|
||||
reactionWriteGroup = "EX_GROUP_"
|
||||
reactionReadGroup = "EX_SUPER_GROUP_"
|
||||
reactionNotification = "EX_NOTIFICATION_"
|
||||
)
|
||||
|
||||
func GetAllMessageCacheKey(conversationID string) string {
|
||||
return messageCache + conversationID + "_*"
|
||||
}
|
||||
|
||||
func GetMessageCacheKey(conversationID string, seq int64) string {
|
||||
return messageCache + conversationID + "_" + strconv.Itoa(int(seq))
|
||||
}
|
||||
|
||||
func GetMessageDelUserListKey(conversationID string, seq int64) string {
|
||||
return messageDelUserList + conversationID + ":" + strconv.Itoa(int(seq))
|
||||
}
|
||||
|
||||
func GetUserDelListKey(conversationID, userID string) string {
|
||||
return userDelMessagesList + conversationID + ":" + userID
|
||||
}
|
||||
|
||||
func GetMessageReactionExKey(clientMsgID string, sessionType int32) string {
|
||||
switch sessionType {
|
||||
case constant.SingleChatType:
|
||||
return reactionExSingle + clientMsgID
|
||||
case constant.WriteGroupChatType:
|
||||
return reactionWriteGroup + clientMsgID
|
||||
case constant.ReadGroupChatType:
|
||||
return reactionReadGroup + clientMsgID
|
||||
case constant.NotificationChatType:
|
||||
return reactionNotification + clientMsgID
|
||||
}
|
||||
|
||||
return ""
|
||||
}
|
||||
func GetLockMessageTypeKey(clientMsgID string, TypeKey string) string {
|
||||
return exTypeKeyLocker + clientMsgID + "_" + TypeKey
|
||||
}
|
||||
|
||||
func GetSendMsgKey(id string) string {
|
||||
return sendMsgFailedFlag + id
|
||||
}
|
||||
@ -0,0 +1,40 @@
|
||||
// Copyright © 2024 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 cachekey
|
||||
|
||||
import "strconv"
|
||||
|
||||
const (
|
||||
object = "OBJECT:"
|
||||
s3 = "S3:"
|
||||
minioImageInfo = "MINIO:IMAGE:"
|
||||
minioThumbnail = "MINIO:THUMBNAIL:"
|
||||
)
|
||||
|
||||
func GetObjectKey(engine string, name string) string {
|
||||
return object + engine + ":" + name
|
||||
}
|
||||
|
||||
func GetS3Key(engine string, name string) string {
|
||||
return s3 + engine + ":" + name
|
||||
}
|
||||
|
||||
func GetObjectImageInfoKey(key string) string {
|
||||
return minioImageInfo + key
|
||||
}
|
||||
|
||||
func GetMinioImageThumbnailKey(key string, format string, width int, height int) string {
|
||||
return minioThumbnail + format + ":w" + strconv.Itoa(width) + ":h" + strconv.Itoa(height) + ":" + key
|
||||
}
|
||||
@ -0,0 +1,38 @@
|
||||
// Copyright © 2024 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 cachekey
|
||||
|
||||
const (
|
||||
maxSeq = "MAX_SEQ:"
|
||||
minSeq = "MIN_SEQ:"
|
||||
conversationUserMinSeq = "CON_USER_MIN_SEQ:"
|
||||
hasReadSeq = "HAS_READ_SEQ:"
|
||||
)
|
||||
|
||||
func GetMaxSeqKey(conversationID string) string {
|
||||
return maxSeq + conversationID
|
||||
}
|
||||
|
||||
func GetMinSeqKey(conversationID string) string {
|
||||
return minSeq + conversationID
|
||||
}
|
||||
|
||||
func GetHasReadSeqKey(conversationID string, userID string) string {
|
||||
return hasReadSeq + userID + ":" + conversationID
|
||||
}
|
||||
|
||||
func GetConversationUserMinSeqKey(conversationID, userID string) string {
|
||||
return conversationUserMinSeq + conversationID + "u:" + userID
|
||||
}
|
||||
@ -0,0 +1,41 @@
|
||||
// Copyright © 2024 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 cachekey
|
||||
|
||||
import (
|
||||
"strconv"
|
||||
)
|
||||
|
||||
const (
|
||||
getuiToken = "GETUI_TOKEN"
|
||||
getuiTaskID = "GETUI_TASK_ID"
|
||||
fmcToken = "FCM_TOKEN:"
|
||||
userBadgeUnreadCountSum = "USER_BADGE_UNREAD_COUNT_SUM:"
|
||||
)
|
||||
|
||||
func GetFcmAccountTokenKey(account string, platformID int) string {
|
||||
return fmcToken + account + ":" + strconv.Itoa(platformID)
|
||||
}
|
||||
|
||||
func GetUserBadgeUnreadCountSumKey(userID string) string {
|
||||
return userBadgeUnreadCountSum + userID
|
||||
}
|
||||
|
||||
func GetGetuiTokenKey() string {
|
||||
return getuiToken
|
||||
}
|
||||
func GetGetuiTaskIDKey() string {
|
||||
return getuiTaskID
|
||||
}
|
||||
@ -0,0 +1,60 @@
|
||||
// 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 cache
|
||||
|
||||
import (
|
||||
"context"
|
||||
relationtb "github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
|
||||
)
|
||||
|
||||
// arg fn will exec when no data in msgCache.
|
||||
type ConversationCache interface {
|
||||
BatchDeleter
|
||||
CloneConversationCache() ConversationCache
|
||||
// get user's conversationIDs from msgCache
|
||||
GetUserConversationIDs(ctx context.Context, ownerUserID string) ([]string, error)
|
||||
DelConversationIDs(userIDs ...string) ConversationCache
|
||||
|
||||
GetUserConversationIDsHash(ctx context.Context, ownerUserID string) (hash uint64, err error)
|
||||
DelUserConversationIDsHash(ownerUserIDs ...string) ConversationCache
|
||||
|
||||
// get one conversation from msgCache
|
||||
GetConversation(ctx context.Context, ownerUserID, conversationID string) (*relationtb.Conversation, error)
|
||||
DelConversations(ownerUserID string, conversationIDs ...string) ConversationCache
|
||||
DelUsersConversation(conversationID string, ownerUserIDs ...string) ConversationCache
|
||||
// get one conversation from msgCache
|
||||
GetConversations(ctx context.Context, ownerUserID string,
|
||||
conversationIDs []string) ([]*relationtb.Conversation, error)
|
||||
// get one user's all conversations from msgCache
|
||||
GetUserAllConversations(ctx context.Context, ownerUserID string) ([]*relationtb.Conversation, error)
|
||||
// get user conversation recv msg from msgCache
|
||||
GetUserRecvMsgOpt(ctx context.Context, ownerUserID, conversationID string) (opt int, err error)
|
||||
DelUserRecvMsgOpt(ownerUserID, conversationID string) ConversationCache
|
||||
// get one super group recv msg but do not notification userID list
|
||||
// GetSuperGroupRecvMsgNotNotifyUserIDs(ctx context.Context, groupID string) (userIDs []string, err error)
|
||||
DelSuperGroupRecvMsgNotNotifyUserIDs(groupID string) ConversationCache
|
||||
// get one super group recv msg but do not notification userID list hash
|
||||
// GetSuperGroupRecvMsgNotNotifyUserIDsHash(ctx context.Context, groupID string) (hash uint64, err error)
|
||||
DelSuperGroupRecvMsgNotNotifyUserIDsHash(groupID string) ConversationCache
|
||||
|
||||
// GetUserAllHasReadSeqs(ctx context.Context, ownerUserID string) (map[string]int64, error)
|
||||
DelUserAllHasReadSeqs(ownerUserID string, conversationIDs ...string) ConversationCache
|
||||
|
||||
GetConversationsByConversationID(ctx context.Context,
|
||||
conversationIDs []string) ([]*relationtb.Conversation, error)
|
||||
DelConversationByConversationID(conversationIDs ...string) ConversationCache
|
||||
GetConversationNotReceiveMessageUserIDs(ctx context.Context, conversationID string) ([]string, error)
|
||||
DelConversationNotReceiveMessageUserIDs(conversationIDs ...string) ConversationCache
|
||||
}
|
||||
@ -0,0 +1,35 @@
|
||||
// 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 cache
|
||||
|
||||
import (
|
||||
"context"
|
||||
relationtb "github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
|
||||
)
|
||||
|
||||
// FriendCache is an interface for caching friend-related data.
|
||||
type FriendCache interface {
|
||||
BatchDeleter
|
||||
CloneFriendCache() FriendCache
|
||||
GetFriendIDs(ctx context.Context, ownerUserID string) (friendIDs []string, err error)
|
||||
// Called when friendID list changed
|
||||
DelFriendIDs(ownerUserID ...string) FriendCache
|
||||
// Get single friendInfo from the cache
|
||||
GetFriend(ctx context.Context, ownerUserID, friendUserID string) (friend *relationtb.Friend, err error)
|
||||
// Delete friend when friend info changed
|
||||
DelFriend(ownerUserID, friendUserID string) FriendCache
|
||||
// Delete friends when friends' info changed
|
||||
DelFriends(ownerUserID string, friendUserIDs []string) FriendCache
|
||||
}
|
||||
@ -0,0 +1,62 @@
|
||||
// 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 cache
|
||||
|
||||
import (
|
||||
"context"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/common"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
|
||||
)
|
||||
|
||||
type GroupHash interface {
|
||||
GetGroupHash(ctx context.Context, groupID string) (uint64, error)
|
||||
}
|
||||
|
||||
type GroupCache interface {
|
||||
BatchDeleter
|
||||
CloneGroupCache() GroupCache
|
||||
GetGroupsInfo(ctx context.Context, groupIDs []string) (groups []*model.Group, err error)
|
||||
GetGroupInfo(ctx context.Context, groupID string) (group *model.Group, err error)
|
||||
DelGroupsInfo(groupIDs ...string) GroupCache
|
||||
|
||||
GetGroupMembersHash(ctx context.Context, groupID string) (hashCode uint64, err error)
|
||||
GetGroupMemberHashMap(ctx context.Context, groupIDs []string) (map[string]*common.GroupSimpleUserID, error)
|
||||
DelGroupMembersHash(groupID string) GroupCache
|
||||
|
||||
GetGroupMemberIDs(ctx context.Context, groupID string) (groupMemberIDs []string, err error)
|
||||
GetGroupsMemberIDs(ctx context.Context, groupIDs []string) (groupMemberIDs map[string][]string, err error)
|
||||
|
||||
DelGroupMemberIDs(groupID string) GroupCache
|
||||
|
||||
GetJoinedGroupIDs(ctx context.Context, userID string) (joinedGroupIDs []string, err error)
|
||||
DelJoinedGroupID(userID ...string) GroupCache
|
||||
|
||||
GetGroupMemberInfo(ctx context.Context, groupID, userID string) (groupMember *model.GroupMember, err error)
|
||||
GetGroupMembersInfo(ctx context.Context, groupID string, userID []string) (groupMembers []*model.GroupMember, err error)
|
||||
GetAllGroupMembersInfo(ctx context.Context, groupID string) (groupMembers []*model.GroupMember, err error)
|
||||
GetGroupMembersPage(ctx context.Context, groupID string, userID []string, showNumber, pageNumber int32) (total uint32, groupMembers []*model.GroupMember, err error)
|
||||
FindGroupMemberUser(ctx context.Context, groupIDs []string, userID string) ([]*model.GroupMember, error)
|
||||
|
||||
GetGroupRoleLevelMemberIDs(ctx context.Context, groupID string, roleLevel int32) ([]string, error)
|
||||
GetGroupOwner(ctx context.Context, groupID string) (*model.GroupMember, error)
|
||||
GetGroupsOwner(ctx context.Context, groupIDs []string) ([]*model.GroupMember, error)
|
||||
DelGroupRoleLevel(groupID string, roleLevel []int32) GroupCache
|
||||
DelGroupAllRoleLevel(groupID string) GroupCache
|
||||
DelGroupMembersInfo(groupID string, userID ...string) GroupCache
|
||||
GetGroupRoleLevelMemberInfo(ctx context.Context, groupID string, roleLevel int32) ([]*model.GroupMember, error)
|
||||
GetGroupRolesLevelMemberInfo(ctx context.Context, groupID string, roleLevels []int32) ([]*model.GroupMember, error)
|
||||
GetGroupMemberNum(ctx context.Context, groupID string) (memberNum int64, err error)
|
||||
DelGroupsMemberNum(groupID ...string) GroupCache
|
||||
}
|
||||
@ -0,0 +1,43 @@
|
||||
// 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 cache
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
"github.com/openimsdk/protocol/sdkws"
|
||||
)
|
||||
|
||||
type MsgCache interface {
|
||||
GetMessagesBySeq(ctx context.Context, conversationID string, seqs []int64) (seqMsg []*sdkws.MsgData, failedSeqList []int64, err error)
|
||||
SetMessageToCache(ctx context.Context, conversationID string, msgs []*sdkws.MsgData) (int, error)
|
||||
UserDeleteMsgs(ctx context.Context, conversationID string, seqs []int64, userID string) error
|
||||
DelUserDeleteMsgsList(ctx context.Context, conversationID string, seqs []int64)
|
||||
DeleteMessages(ctx context.Context, conversationID string, seqs []int64) error
|
||||
GetUserDelList(ctx context.Context, userID, conversationID string) (seqs []int64, err error)
|
||||
CleanUpOneConversationAllMsg(ctx context.Context, conversationID string) error
|
||||
DelMsgFromCache(ctx context.Context, userID string, seqList []int64) error
|
||||
SetSendMsgStatus(ctx context.Context, id string, status int32) error
|
||||
GetSendMsgStatus(ctx context.Context, id string) (int32, error)
|
||||
JudgeMessageReactionExist(ctx context.Context, clientMsgID string, sessionType int32) (bool, error)
|
||||
GetOneMessageAllReactionList(ctx context.Context, clientMsgID string, sessionType int32) (map[string]string, error)
|
||||
DeleteOneMessageKey(ctx context.Context, clientMsgID string, sessionType int32, subKey string) error
|
||||
SetMessageReactionExpire(ctx context.Context, clientMsgID string, sessionType int32, expiration time.Duration) (bool, error)
|
||||
GetMessageTypeKeyValue(ctx context.Context, clientMsgID string, sessionType int32, typeKey string) (string, error)
|
||||
SetMessageTypeKeyValue(ctx context.Context, clientMsgID string, sessionType int32, typeKey, value string) error
|
||||
LockMessageTypeKey(ctx context.Context, clientMsgID string, TypeKey string) error
|
||||
UnLockMessageTypeKey(ctx context.Context, clientMsgID string, TypeKey string) error
|
||||
}
|
||||
@ -0,0 +1,211 @@
|
||||
// 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 redis
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"github.com/dtm-labs/rockscache"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/localcache"
|
||||
"github.com/openimsdk/tools/errs"
|
||||
"github.com/openimsdk/tools/log"
|
||||
"github.com/openimsdk/tools/mw/specialerror"
|
||||
"github.com/openimsdk/tools/utils/datautil"
|
||||
"github.com/redis/go-redis/v9"
|
||||
"time"
|
||||
)
|
||||
|
||||
// BatchDeleterRedis is a concrete implementation of the BatchDeleter interface based on Redis and RocksCache.
|
||||
type BatchDeleterRedis struct {
|
||||
redisClient redis.UniversalClient
|
||||
keys []string
|
||||
rocksClient *rockscache.Client
|
||||
redisPubTopics []string
|
||||
}
|
||||
|
||||
// NewBatchDeleterRedis creates a new BatchDeleterRedis instance.
|
||||
func NewBatchDeleterRedis(redisClient redis.UniversalClient, options *rockscache.Options, redisPubTopics []string) *BatchDeleterRedis {
|
||||
return &BatchDeleterRedis{
|
||||
redisClient: redisClient,
|
||||
rocksClient: rockscache.NewClient(redisClient, *options),
|
||||
redisPubTopics: redisPubTopics,
|
||||
}
|
||||
}
|
||||
|
||||
// ExecDelWithKeys directly takes keys for batch deletion and publishes deletion information.
|
||||
func (c *BatchDeleterRedis) ExecDelWithKeys(ctx context.Context, keys []string) error {
|
||||
distinctKeys := datautil.Distinct(keys)
|
||||
return c.execDel(ctx, distinctKeys)
|
||||
}
|
||||
|
||||
// ChainExecDel is used for chain calls for batch deletion. It must call Clone to prevent memory pollution.
|
||||
func (c *BatchDeleterRedis) ChainExecDel(ctx context.Context) error {
|
||||
distinctKeys := datautil.Distinct(c.keys)
|
||||
return c.execDel(ctx, distinctKeys)
|
||||
}
|
||||
|
||||
// execDel performs batch deletion and publishes the keys that have been deleted to update the local cache information of other nodes.
|
||||
func (c *BatchDeleterRedis) execDel(ctx context.Context, keys []string) error {
|
||||
if len(keys) > 0 {
|
||||
log.ZDebug(ctx, "delete cache", "topic", c.redisPubTopics, "keys", keys)
|
||||
slotMapKeys, err := groupKeysBySlot(ctx, c.redisClient, keys)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
// Batch delete keys
|
||||
for slot, singleSlotKeys := range slotMapKeys {
|
||||
if err := c.rocksClient.TagAsDeletedBatch2(ctx, singleSlotKeys); err != nil {
|
||||
log.ZWarn(ctx, "Batch delete cache failed", err, "slot", slot, "keys", singleSlotKeys)
|
||||
continue
|
||||
}
|
||||
}
|
||||
// Publish the keys that have been deleted to Redis to update the local cache information of other nodes
|
||||
if len(c.redisPubTopics) > 0 && len(keys) > 0 {
|
||||
keysByTopic := localcache.GetPublishKeysByTopic(c.redisPubTopics, keys)
|
||||
for topic, keys := range keysByTopic {
|
||||
if len(keys) > 0 {
|
||||
data, err := json.Marshal(keys)
|
||||
if err != nil {
|
||||
log.ZWarn(ctx, "keys json marshal failed", err, "topic", topic, "keys", keys)
|
||||
} else {
|
||||
if err := c.redisClient.Publish(ctx, topic, string(data)).Err(); err != nil {
|
||||
log.ZWarn(ctx, "redis publish cache delete error", err, "topic", topic, "keys", keys)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Clone creates a copy of BatchDeleterRedis for chain calls to prevent memory pollution.
|
||||
func (c *BatchDeleterRedis) Clone() cache.BatchDeleter {
|
||||
return &BatchDeleterRedis{
|
||||
redisClient: c.redisClient,
|
||||
keys: c.keys,
|
||||
rocksClient: c.rocksClient,
|
||||
redisPubTopics: c.redisPubTopics,
|
||||
}
|
||||
}
|
||||
|
||||
// AddKeys adds keys to be deleted.
|
||||
func (c *BatchDeleterRedis) AddKeys(keys ...string) {
|
||||
c.keys = append(c.keys, keys...)
|
||||
}
|
||||
|
||||
// GetRocksCacheOptions returns the default configuration options for RocksCache.
|
||||
func GetRocksCacheOptions() *rockscache.Options {
|
||||
opts := rockscache.NewDefaultOptions()
|
||||
opts.StrongConsistency = true
|
||||
opts.RandomExpireAdjustment = 0.2
|
||||
|
||||
return &opts
|
||||
}
|
||||
|
||||
// groupKeysBySlot groups keys by their Redis cluster hash slots.
|
||||
func groupKeysBySlot(ctx context.Context, redisClient redis.UniversalClient, keys []string) (map[int64][]string, error) {
|
||||
slots := make(map[int64][]string)
|
||||
clusterClient, isCluster := redisClient.(*redis.ClusterClient)
|
||||
if isCluster {
|
||||
pipe := clusterClient.Pipeline()
|
||||
cmds := make([]*redis.IntCmd, len(keys))
|
||||
for i, key := range keys {
|
||||
cmds[i] = pipe.ClusterKeySlot(ctx, key)
|
||||
}
|
||||
_, err := pipe.Exec(ctx)
|
||||
if err != nil {
|
||||
return nil, errs.WrapMsg(err, "get slot err")
|
||||
}
|
||||
|
||||
for i, cmd := range cmds {
|
||||
slot, err := cmd.Result()
|
||||
if err != nil {
|
||||
log.ZWarn(ctx, "some key get slot err", err, "key", keys[i])
|
||||
continue
|
||||
}
|
||||
slots[slot] = append(slots[slot], keys[i])
|
||||
}
|
||||
} else {
|
||||
// If not a cluster client, put all keys in the same slot (0)
|
||||
slots[0] = keys
|
||||
}
|
||||
|
||||
return slots, nil
|
||||
}
|
||||
|
||||
func getCache[T any](ctx context.Context, rcClient *rockscache.Client, key string, expire time.Duration, fn func(ctx context.Context) (T, error)) (T, error) {
|
||||
var t T
|
||||
var write bool
|
||||
v, err := rcClient.Fetch2(ctx, key, expire, func() (s string, err error) {
|
||||
t, err = fn(ctx)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
bs, err := json.Marshal(t)
|
||||
if err != nil {
|
||||
return "", errs.WrapMsg(err, "marshal failed")
|
||||
}
|
||||
write = true
|
||||
|
||||
return string(bs), nil
|
||||
})
|
||||
if err != nil {
|
||||
return t, errs.Wrap(err)
|
||||
}
|
||||
if write {
|
||||
return t, nil
|
||||
}
|
||||
if v == "" {
|
||||
return t, errs.ErrRecordNotFound.WrapMsg("cache is not found")
|
||||
}
|
||||
err = json.Unmarshal([]byte(v), &t)
|
||||
if err != nil {
|
||||
errInfo := fmt.Sprintf("cache json.Unmarshal failed, key:%s, value:%s, expire:%s", key, v, expire)
|
||||
return t, errs.WrapMsg(err, errInfo)
|
||||
}
|
||||
|
||||
return t, nil
|
||||
}
|
||||
|
||||
func batchGetCache[T any, K comparable](
|
||||
ctx context.Context,
|
||||
rcClient *rockscache.Client,
|
||||
expire time.Duration,
|
||||
keys []K,
|
||||
keyFn func(key K) string,
|
||||
fns func(ctx context.Context, key K) (T, error),
|
||||
) ([]T, error) {
|
||||
if len(keys) == 0 {
|
||||
return nil, nil
|
||||
}
|
||||
res := make([]T, 0, len(keys))
|
||||
for _, key := range keys {
|
||||
val, err := getCache(ctx, rcClient, keyFn(key), expire, func(ctx context.Context) (T, error) {
|
||||
return fns(ctx, key)
|
||||
})
|
||||
if err != nil {
|
||||
if errs.ErrRecordNotFound.Is(specialerror.ErrCode(errs.Unwrap(err))) {
|
||||
continue
|
||||
}
|
||||
return nil, errs.Wrap(err)
|
||||
}
|
||||
res = append(res, val)
|
||||
}
|
||||
|
||||
return res, nil
|
||||
}
|
||||
@ -0,0 +1,246 @@
|
||||
// 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 redis
|
||||
|
||||
import (
|
||||
"context"
|
||||
"github.com/dtm-labs/rockscache"
|
||||
"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/open-im-server/v3/pkg/common/storage/database"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
|
||||
"github.com/openimsdk/tools/log"
|
||||
"github.com/openimsdk/tools/utils/datautil"
|
||||
"github.com/openimsdk/tools/utils/encrypt"
|
||||
"github.com/redis/go-redis/v9"
|
||||
"math/big"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
const (
|
||||
conversationExpireTime = time.Second * 60 * 60 * 12
|
||||
)
|
||||
|
||||
func NewConversationRedis(rdb redis.UniversalClient, localCache *config.LocalCache, opts *rockscache.Options, db database.Conversation) cache.ConversationCache {
|
||||
batchHandler := NewBatchDeleterRedis(rdb, opts, []string{localCache.Conversation.Topic})
|
||||
c := localCache.Conversation
|
||||
log.ZDebug(context.Background(), "black local cache init", "Topic", c.Topic, "SlotNum", c.SlotNum, "SlotSize", c.SlotSize, "enable", c.Enable())
|
||||
return &ConversationRedisCache{
|
||||
BatchDeleter: batchHandler,
|
||||
rcClient: rockscache.NewClient(rdb, *opts),
|
||||
conversationDB: db,
|
||||
expireTime: conversationExpireTime,
|
||||
}
|
||||
}
|
||||
|
||||
type ConversationRedisCache struct {
|
||||
cache.BatchDeleter
|
||||
rcClient *rockscache.Client
|
||||
conversationDB database.Conversation
|
||||
expireTime time.Duration
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) CloneConversationCache() cache.ConversationCache {
|
||||
return &ConversationRedisCache{
|
||||
BatchDeleter: c.BatchDeleter.Clone(),
|
||||
rcClient: c.rcClient,
|
||||
conversationDB: c.conversationDB,
|
||||
expireTime: c.expireTime,
|
||||
}
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) getConversationKey(ownerUserID, conversationID string) string {
|
||||
return cachekey.GetConversationKey(ownerUserID, conversationID)
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) getConversationIDsKey(ownerUserID string) string {
|
||||
return cachekey.GetConversationIDsKey(ownerUserID)
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) getSuperGroupRecvNotNotifyUserIDsKey(groupID string) string {
|
||||
return cachekey.GetSuperGroupRecvNotNotifyUserIDsKey(groupID)
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) getRecvMsgOptKey(ownerUserID, conversationID string) string {
|
||||
return cachekey.GetRecvMsgOptKey(ownerUserID, conversationID)
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) getSuperGroupRecvNotNotifyUserIDsHashKey(groupID string) string {
|
||||
return cachekey.GetSuperGroupRecvNotNotifyUserIDsHashKey(groupID)
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) getConversationHasReadSeqKey(ownerUserID, conversationID string) string {
|
||||
return cachekey.GetConversationHasReadSeqKey(ownerUserID, conversationID)
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) getConversationNotReceiveMessageUserIDsKey(conversationID string) string {
|
||||
return cachekey.GetConversationNotReceiveMessageUserIDsKey(conversationID)
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) getUserConversationIDsHashKey(ownerUserID string) string {
|
||||
return cachekey.GetUserConversationIDsHashKey(ownerUserID)
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) GetUserConversationIDs(ctx context.Context, ownerUserID string) ([]string, error) {
|
||||
return getCache(ctx, c.rcClient, c.getConversationIDsKey(ownerUserID), c.expireTime, func(ctx context.Context) ([]string, error) {
|
||||
return c.conversationDB.FindUserIDAllConversationID(ctx, ownerUserID)
|
||||
})
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) DelConversationIDs(userIDs ...string) cache.ConversationCache {
|
||||
keys := make([]string, 0, len(userIDs))
|
||||
for _, userID := range userIDs {
|
||||
keys = append(keys, c.getConversationIDsKey(userID))
|
||||
}
|
||||
cache := c.CloneConversationCache()
|
||||
cache.AddKeys(keys...)
|
||||
|
||||
return cache
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) GetUserConversationIDsHash(ctx context.Context, ownerUserID string) (hash uint64, err error) {
|
||||
return getCache(
|
||||
ctx,
|
||||
c.rcClient,
|
||||
c.getUserConversationIDsHashKey(ownerUserID),
|
||||
c.expireTime,
|
||||
func(ctx context.Context) (uint64, error) {
|
||||
conversationIDs, err := c.GetUserConversationIDs(ctx, ownerUserID)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
datautil.Sort(conversationIDs, true)
|
||||
bi := big.NewInt(0)
|
||||
bi.SetString(encrypt.Md5(strings.Join(conversationIDs, ";"))[0:8], 16)
|
||||
return bi.Uint64(), nil
|
||||
},
|
||||
)
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) DelUserConversationIDsHash(ownerUserIDs ...string) cache.ConversationCache {
|
||||
keys := make([]string, 0, len(ownerUserIDs))
|
||||
for _, ownerUserID := range ownerUserIDs {
|
||||
keys = append(keys, c.getUserConversationIDsHashKey(ownerUserID))
|
||||
}
|
||||
cache := c.CloneConversationCache()
|
||||
cache.AddKeys(keys...)
|
||||
|
||||
return cache
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) GetConversation(ctx context.Context, ownerUserID, conversationID string) (*model.Conversation, error) {
|
||||
return getCache(ctx, c.rcClient, c.getConversationKey(ownerUserID, conversationID), c.expireTime, func(ctx context.Context) (*model.Conversation, error) {
|
||||
return c.conversationDB.Take(ctx, ownerUserID, conversationID)
|
||||
})
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) DelConversations(ownerUserID string, conversationIDs ...string) cache.ConversationCache {
|
||||
keys := make([]string, 0, len(conversationIDs))
|
||||
for _, conversationID := range conversationIDs {
|
||||
keys = append(keys, c.getConversationKey(ownerUserID, conversationID))
|
||||
}
|
||||
cache := c.CloneConversationCache()
|
||||
cache.AddKeys(keys...)
|
||||
|
||||
return cache
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) GetConversations(ctx context.Context, ownerUserID string, conversationIDs []string) ([]*model.Conversation, error) {
|
||||
return batchGetCache(ctx, c.rcClient, c.expireTime, conversationIDs, func(conversationID string) string {
|
||||
return c.getConversationKey(ownerUserID, conversationID)
|
||||
}, func(ctx context.Context, conversationID string) (*model.Conversation, error) {
|
||||
return c.conversationDB.Take(ctx, ownerUserID, conversationID)
|
||||
})
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) GetUserAllConversations(ctx context.Context, ownerUserID string) ([]*model.Conversation, error) {
|
||||
conversationIDs, err := c.GetUserConversationIDs(ctx, ownerUserID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return c.GetConversations(ctx, ownerUserID, conversationIDs)
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) GetUserRecvMsgOpt(ctx context.Context, ownerUserID, conversationID string) (opt int, err error) {
|
||||
return getCache(ctx, c.rcClient, c.getRecvMsgOptKey(ownerUserID, conversationID), c.expireTime, func(ctx context.Context) (opt int, err error) {
|
||||
return c.conversationDB.GetUserRecvMsgOpt(ctx, ownerUserID, conversationID)
|
||||
})
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) DelUsersConversation(conversationID string, ownerUserIDs ...string) cache.ConversationCache {
|
||||
keys := make([]string, 0, len(ownerUserIDs))
|
||||
for _, ownerUserID := range ownerUserIDs {
|
||||
keys = append(keys, c.getConversationKey(ownerUserID, conversationID))
|
||||
}
|
||||
cache := c.CloneConversationCache()
|
||||
cache.AddKeys(keys...)
|
||||
|
||||
return cache
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) DelUserRecvMsgOpt(ownerUserID, conversationID string) cache.ConversationCache {
|
||||
cache := c.CloneConversationCache()
|
||||
cache.AddKeys(c.getRecvMsgOptKey(ownerUserID, conversationID))
|
||||
|
||||
return cache
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) DelSuperGroupRecvMsgNotNotifyUserIDs(groupID string) cache.ConversationCache {
|
||||
cache := c.CloneConversationCache()
|
||||
cache.AddKeys(c.getSuperGroupRecvNotNotifyUserIDsKey(groupID))
|
||||
|
||||
return cache
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) DelSuperGroupRecvMsgNotNotifyUserIDsHash(groupID string) cache.ConversationCache {
|
||||
cache := c.CloneConversationCache()
|
||||
cache.AddKeys(c.getSuperGroupRecvNotNotifyUserIDsHashKey(groupID))
|
||||
|
||||
return cache
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) DelUserAllHasReadSeqs(ownerUserID string, conversationIDs ...string) cache.ConversationCache {
|
||||
cache := c.CloneConversationCache()
|
||||
for _, conversationID := range conversationIDs {
|
||||
cache.AddKeys(c.getConversationHasReadSeqKey(ownerUserID, conversationID))
|
||||
}
|
||||
|
||||
return cache
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) GetConversationsByConversationID(ctx context.Context, conversationIDs []string) ([]*model.Conversation, error) {
|
||||
panic("implement me")
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) DelConversationByConversationID(conversationIDs ...string) cache.ConversationCache {
|
||||
panic("implement me")
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) GetConversationNotReceiveMessageUserIDs(ctx context.Context, conversationID string) ([]string, error) {
|
||||
return getCache(ctx, c.rcClient, c.getConversationNotReceiveMessageUserIDsKey(conversationID), c.expireTime, func(ctx context.Context) ([]string, error) {
|
||||
return c.conversationDB.GetConversationNotReceiveMessageUserIDs(ctx, conversationID)
|
||||
})
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) DelConversationNotReceiveMessageUserIDs(conversationIDs ...string) cache.ConversationCache {
|
||||
cache := c.CloneConversationCache()
|
||||
for _, conversationID := range conversationIDs {
|
||||
cache.AddKeys(c.getConversationNotReceiveMessageUserIDsKey(conversationID))
|
||||
}
|
||||
|
||||
return cache
|
||||
}
|
||||
@ -0,0 +1,15 @@
|
||||
// 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 redis
|
||||
@ -0,0 +1,15 @@
|
||||
// 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 redis
|
||||
@ -0,0 +1,103 @@
|
||||
// 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 redis
|
||||
|
||||
import (
|
||||
"context"
|
||||
"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/redis/go-redis/v9"
|
||||
"time"
|
||||
)
|
||||
|
||||
func NewThirdCache(rdb redis.UniversalClient) cache.ThirdCache {
|
||||
return &thirdCache{rdb: rdb}
|
||||
}
|
||||
|
||||
type thirdCache struct {
|
||||
rdb redis.UniversalClient
|
||||
}
|
||||
|
||||
func (c *thirdCache) getGetuiTokenKey() string {
|
||||
return cachekey.GetGetuiTokenKey()
|
||||
}
|
||||
|
||||
func (c *thirdCache) getGetuiTaskIDKey() string {
|
||||
return cachekey.GetGetuiTaskIDKey()
|
||||
}
|
||||
|
||||
func (c *thirdCache) getUserBadgeUnreadCountSumKey(userID string) string {
|
||||
return cachekey.GetUserBadgeUnreadCountSumKey(userID)
|
||||
}
|
||||
|
||||
func (c *thirdCache) getFcmAccountTokenKey(account string, platformID int) string {
|
||||
return cachekey.GetFcmAccountTokenKey(account, platformID)
|
||||
}
|
||||
|
||||
func (c *thirdCache) SetFcmToken(ctx context.Context, account string, platformID int, fcmToken string, expireTime int64) (err error) {
|
||||
return errs.Wrap(c.rdb.Set(ctx, c.getFcmAccountTokenKey(account, platformID), fcmToken, time.Duration(expireTime)*time.Second).Err())
|
||||
}
|
||||
|
||||
func (c *thirdCache) GetFcmToken(ctx context.Context, account string, platformID int) (string, error) {
|
||||
val, err := c.rdb.Get(ctx, c.getFcmAccountTokenKey(account, platformID)).Result()
|
||||
if err != nil {
|
||||
return "", errs.Wrap(err)
|
||||
}
|
||||
return val, nil
|
||||
}
|
||||
|
||||
func (c *thirdCache) DelFcmToken(ctx context.Context, account string, platformID int) error {
|
||||
return errs.Wrap(c.rdb.Del(ctx, c.getFcmAccountTokenKey(account, platformID)).Err())
|
||||
}
|
||||
|
||||
func (c *thirdCache) IncrUserBadgeUnreadCountSum(ctx context.Context, userID string) (int, error) {
|
||||
seq, err := c.rdb.Incr(ctx, c.getUserBadgeUnreadCountSumKey(userID)).Result()
|
||||
|
||||
return int(seq), errs.Wrap(err)
|
||||
}
|
||||
|
||||
func (c *thirdCache) SetUserBadgeUnreadCountSum(ctx context.Context, userID string, value int) error {
|
||||
return errs.Wrap(c.rdb.Set(ctx, c.getUserBadgeUnreadCountSumKey(userID), value, 0).Err())
|
||||
}
|
||||
|
||||
func (c *thirdCache) GetUserBadgeUnreadCountSum(ctx context.Context, userID string) (int, error) {
|
||||
val, err := c.rdb.Get(ctx, c.getUserBadgeUnreadCountSumKey(userID)).Int()
|
||||
return val, errs.Wrap(err)
|
||||
}
|
||||
|
||||
func (c *thirdCache) SetGetuiToken(ctx context.Context, token string, expireTime int64) error {
|
||||
return errs.Wrap(c.rdb.Set(ctx, c.getGetuiTokenKey(), token, time.Duration(expireTime)*time.Second).Err())
|
||||
}
|
||||
|
||||
func (c *thirdCache) GetGetuiToken(ctx context.Context) (string, error) {
|
||||
val, err := c.rdb.Get(ctx, c.getGetuiTokenKey()).Result()
|
||||
if err != nil {
|
||||
return "", errs.Wrap(err)
|
||||
}
|
||||
return val, nil
|
||||
}
|
||||
|
||||
func (c *thirdCache) SetGetuiTaskID(ctx context.Context, taskID string, expireTime int64) error {
|
||||
return errs.Wrap(c.rdb.Set(ctx, c.getGetuiTaskIDKey(), taskID, time.Duration(expireTime)*time.Second).Err())
|
||||
}
|
||||
|
||||
func (c *thirdCache) GetGetuiTaskID(ctx context.Context) (string, error) {
|
||||
val, err := c.rdb.Get(ctx, c.getGetuiTaskIDKey()).Result()
|
||||
if err != nil {
|
||||
return "", errs.Wrap(err)
|
||||
}
|
||||
return val, nil
|
||||
}
|
||||
@ -1,31 +1,38 @@
|
||||
package cache
|
||||
// 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 redis
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/cachekey"
|
||||
"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/stringutil"
|
||||
"github.com/redis/go-redis/v9"
|
||||
)
|
||||
|
||||
func NewTokenCacheModel(rdb redis.UniversalClient) TokenModel {
|
||||
type tokenCache struct {
|
||||
rdb redis.UniversalClient
|
||||
}
|
||||
|
||||
func NewTokenCacheModel(rdb redis.UniversalClient) cache.TokenModel {
|
||||
return &tokenCache{
|
||||
rdb: rdb,
|
||||
}
|
||||
}
|
||||
|
||||
type TokenModel interface {
|
||||
AddTokenFlag(ctx context.Context, userID string, platformID int, token string, flag int) error
|
||||
GetTokensWithoutError(ctx context.Context, userID string, platformID int) (map[string]int, error)
|
||||
SetTokenMapByUidPid(ctx context.Context, userID string, platformID int, m map[string]int) error
|
||||
DeleteTokenByUidPid(ctx context.Context, userID string, platformID int, fields []string) error
|
||||
}
|
||||
|
||||
type tokenCache struct {
|
||||
rdb redis.UniversalClient
|
||||
}
|
||||
|
||||
func (c *tokenCache) AddTokenFlag(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())
|
||||
}
|
||||
@ -0,0 +1,51 @@
|
||||
// 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 cache
|
||||
|
||||
import (
|
||||
"context"
|
||||
relationtb "github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
|
||||
"github.com/openimsdk/tools/s3"
|
||||
)
|
||||
|
||||
type ObjectCache interface {
|
||||
BatchDeleter
|
||||
CloneObjectCache() ObjectCache
|
||||
GetName(ctx context.Context, engine string, name string) (*relationtb.Object, error)
|
||||
DelObjectName(engine string, names ...string) ObjectCache
|
||||
}
|
||||
|
||||
type S3Cache interface {
|
||||
BatchDeleter
|
||||
GetKey(ctx context.Context, engine string, key string) (*s3.ObjectInfo, error)
|
||||
DelS3Key(engine string, keys ...string) S3Cache
|
||||
}
|
||||
|
||||
// TODO integrating minio.Cache and MinioCache interfaces.
|
||||
type MinioCache interface {
|
||||
BatchDeleter
|
||||
GetImageObjectKeyInfo(ctx context.Context, key string, fn func(ctx context.Context) (*MinioImageInfo, error)) (*MinioImageInfo, error)
|
||||
GetThumbnailKey(ctx context.Context, key string, format string, width int, height int, minioCache func(ctx context.Context) (string, error)) (string, error)
|
||||
DelObjectImageInfoKey(keys ...string) MinioCache
|
||||
DelImageThumbnailKey(key string, format string, width int, height int) MinioCache
|
||||
}
|
||||
|
||||
type MinioImageInfo struct {
|
||||
IsImg bool `json:"isImg"`
|
||||
Width int `json:"width"`
|
||||
Height int `json:"height"`
|
||||
Format string `json:"format"`
|
||||
Etag string `json:"etag"`
|
||||
}
|
||||
@ -0,0 +1,30 @@
|
||||
package cache
|
||||
|
||||
import (
|
||||
"context"
|
||||
)
|
||||
|
||||
type SeqCache interface {
|
||||
SetMaxSeq(ctx context.Context, conversationID string, maxSeq int64) error
|
||||
GetMaxSeqs(ctx context.Context, conversationIDs []string) (map[string]int64, error)
|
||||
GetMaxSeq(ctx context.Context, conversationID string) (int64, error)
|
||||
SetMinSeq(ctx context.Context, conversationID string, minSeq int64) error
|
||||
SetMinSeqs(ctx context.Context, seqs map[string]int64) error
|
||||
GetMinSeqs(ctx context.Context, conversationIDs []string) (map[string]int64, error)
|
||||
GetMinSeq(ctx context.Context, conversationID string) (int64, error)
|
||||
GetConversationUserMinSeq(ctx context.Context, conversationID string, userID string) (int64, error)
|
||||
GetConversationUserMinSeqs(ctx context.Context, conversationID string, userIDs []string) (map[string]int64, error)
|
||||
SetConversationUserMinSeq(ctx context.Context, conversationID string, userID string, minSeq int64) error
|
||||
// seqs map: key userID value minSeq
|
||||
SetConversationUserMinSeqs(ctx context.Context, conversationID string, seqs map[string]int64) (err error)
|
||||
// seqs map: key conversationID value minSeq
|
||||
SetUserConversationsMinSeqs(ctx context.Context, userID string, seqs map[string]int64) error
|
||||
// has read seq
|
||||
SetHasReadSeq(ctx context.Context, userID string, conversationID string, hasReadSeq int64) error
|
||||
// k: user, v: seq
|
||||
SetHasReadSeqs(ctx context.Context, conversationID string, hasReadSeqs map[string]int64) error
|
||||
// k: conversation, v :seq
|
||||
UserSetHasReadSeqs(ctx context.Context, userID string, hasReadSeqs map[string]int64) error
|
||||
GetHasReadSeqs(ctx context.Context, userID string, conversationIDs []string) (map[string]int64, error)
|
||||
GetHasReadSeq(ctx context.Context, userID string, conversationID string) (int64, error)
|
||||
}
|
||||
@ -0,0 +1,18 @@
|
||||
package cache
|
||||
|
||||
import (
|
||||
"context"
|
||||
)
|
||||
|
||||
type ThirdCache interface {
|
||||
SetFcmToken(ctx context.Context, account string, platformID int, fcmToken string, expireTime int64) (err error)
|
||||
GetFcmToken(ctx context.Context, account string, platformID int) (string, error)
|
||||
DelFcmToken(ctx context.Context, account string, platformID int) error
|
||||
IncrUserBadgeUnreadCountSum(ctx context.Context, userID string) (int, error)
|
||||
SetUserBadgeUnreadCountSum(ctx context.Context, userID string, value int) error
|
||||
GetUserBadgeUnreadCountSum(ctx context.Context, userID string) (int, error)
|
||||
SetGetuiToken(ctx context.Context, token string, expireTime int64) error
|
||||
GetGetuiToken(ctx context.Context) (string, error)
|
||||
SetGetuiTaskID(ctx context.Context, taskID string, expireTime int64) error
|
||||
GetGetuiTaskID(ctx context.Context) (string, error)
|
||||
}
|
||||
@ -0,0 +1,12 @@
|
||||
package cache
|
||||
|
||||
import (
|
||||
"context"
|
||||
)
|
||||
|
||||
type TokenModel interface {
|
||||
AddTokenFlag(ctx context.Context, userID string, platformID int, token string, flag int) error
|
||||
GetTokensWithoutError(ctx context.Context, userID string, platformID int) (map[string]int, error)
|
||||
SetTokenMapByUidPid(ctx context.Context, userID string, platformID int, m map[string]int) error
|
||||
DeleteTokenByUidPid(ctx context.Context, userID string, platformID int, fields []string) error
|
||||
}
|
||||
@ -0,0 +1,33 @@
|
||||
// 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 cache
|
||||
|
||||
import (
|
||||
"context"
|
||||
relationtb "github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
|
||||
"github.com/openimsdk/protocol/user"
|
||||
)
|
||||
|
||||
type UserCache interface {
|
||||
BatchDeleter
|
||||
CloneUserCache() UserCache
|
||||
GetUserInfo(ctx context.Context, userID string) (userInfo *relationtb.User, err error)
|
||||
GetUsersInfo(ctx context.Context, userIDs []string) ([]*relationtb.User, error)
|
||||
DelUsersInfo(userIDs ...string) UserCache
|
||||
GetUserGlobalRecvMsgOpt(ctx context.Context, userID string) (opt int, err error)
|
||||
DelUsersGlobalRecvMsgOpt(userIDs ...string) UserCache
|
||||
GetUserStatus(ctx context.Context, userIDs []string) ([]*user.OnlineStatus, error)
|
||||
SetUserStatus(ctx context.Context, userID string, status, platformID int32) error
|
||||
}
|
||||
@ -0,0 +1,26 @@
|
||||
// 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 common
|
||||
|
||||
type BatchUpdateGroupMember struct {
|
||||
GroupID string
|
||||
UserID string
|
||||
Map map[string]any
|
||||
}
|
||||
|
||||
type GroupSimpleUserID struct {
|
||||
Hash uint64
|
||||
MemberNum uint32
|
||||
}
|
||||
@ -0,0 +1,15 @@
|
||||
// Copyright © 2024 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 database // import "github.com/openimsdk/open-im-server/v3/pkg/common/storage/model/relation"
|
||||
@ -0,0 +1,35 @@
|
||||
// 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 database
|
||||
|
||||
import (
|
||||
"context"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
|
||||
"github.com/openimsdk/tools/db/pagination"
|
||||
"time"
|
||||
)
|
||||
|
||||
type Group interface {
|
||||
Create(ctx context.Context, groups []*model.Group) (err error)
|
||||
UpdateMap(ctx context.Context, groupID string, args map[string]any) (err error)
|
||||
UpdateStatus(ctx context.Context, groupID string, status int32) (err error)
|
||||
Find(ctx context.Context, groupIDs []string) (groups []*model.Group, err error)
|
||||
Take(ctx context.Context, groupID string) (group *model.Group, err error)
|
||||
Search(ctx context.Context, keyword string, pagination pagination.Pagination) (total int64, groups []*model.Group, err error)
|
||||
// Get Group total quantity
|
||||
CountTotal(ctx context.Context, before *time.Time) (count int64, err error)
|
||||
// Get Group total quantity every day
|
||||
CountRangeEverydayTotal(ctx context.Context, start time.Time, end time.Time) (map[string]int64, error)
|
||||
}
|
||||
@ -0,0 +1,29 @@
|
||||
// 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 database
|
||||
|
||||
import (
|
||||
"context"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
|
||||
"github.com/openimsdk/tools/db/pagination"
|
||||
"time"
|
||||
)
|
||||
|
||||
type Log interface {
|
||||
Create(ctx context.Context, log []*model.Log) error
|
||||
Search(ctx context.Context, keyword string, start time.Time, end time.Time, pagination pagination.Pagination) (int64, []*model.Log, error)
|
||||
Delete(ctx context.Context, logID []string, userID string) error
|
||||
Get(ctx context.Context, logIDs []string, userID string) ([]*model.Log, error)
|
||||
}
|
||||
Some files were not shown because too many files have changed in this diff Show More
Loading…
Reference in new issue