@@ -2,87 +2,39 @@ package redis
22
33import (
44 "context"
5+ "encoding/json"
6+ "github.com/dtm-labs/rockscache"
57 "github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache"
68 "github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/cachekey"
7- "github.com/openimsdk/open-im-server/v3/pkg/msgprocessor "
8- "github.com/openimsdk/protocol/sdkws "
9+ "github.com/openimsdk/open-im-server/v3/pkg/common/storage/database "
10+ "github.com/openimsdk/open-im-server/v3/pkg/common/storage/model "
911 "github.com/openimsdk/tools/errs"
1012 "github.com/openimsdk/tools/utils/datautil"
1113 "github.com/redis/go-redis/v9"
1214 "time"
1315) //
1416
1517// msgCacheTimeout is expiration time of message cache, 86400 seconds
16- const msgCacheTimeout = 86400
18+ const msgCacheTimeout = time . Hour * 24
1719
18- func NewMsgCache (client redis.UniversalClient ) cache.MsgCache {
19- return & msgCache {rdb : client }
20+ func NewMsgCache (client redis.UniversalClient , db database.Msg ) cache.MsgCache {
21+ return & msgCache {
22+ rdb : client ,
23+ rcClient : rockscache .NewClient (client , * GetRocksCacheOptions ()),
24+ msgDocDatabase : db ,
25+ }
2026}
2127
2228type msgCache struct {
23- rdb redis.UniversalClient
24- }
25-
26- func (c * msgCache ) getMessageCacheKey (conversationID string , seq int64 ) string {
27- return cachekey .GetMessageCacheKey (conversationID , seq )
28- }
29- func (c * msgCache ) getMessageDelUserListKey (conversationID string , seq int64 ) string {
30- return cachekey .GetMessageDelUserListKey (conversationID , seq )
31- }
32-
33- func (c * msgCache ) getUserDelList (conversationID , userID string ) string {
34- return cachekey .GetUserDelListKey (conversationID , userID )
29+ rdb redis.UniversalClient
30+ rcClient * rockscache.Client
31+ msgDocDatabase database.Msg
3532}
3633
3734func (c * msgCache ) getSendMsgKey (id string ) string {
3835 return cachekey .GetSendMsgKey (id )
3936}
4037
41- func (c * msgCache ) getLockMessageTypeKey (clientMsgID string , TypeKey string ) string {
42- return cachekey .GetLockMessageTypeKey (clientMsgID , TypeKey )
43- }
44-
45- func (c * msgCache ) getMessageReactionExPrefix (clientMsgID string , sessionType int32 ) string {
46- return cachekey .GetMessageReactionExKey (clientMsgID , sessionType )
47- }
48-
49- func (c * msgCache ) SetMessagesToCache (ctx context.Context , conversationID string , msgs []* sdkws.MsgData ) (int , error ) {
50- msgMap := datautil .SliceToMap (msgs , func (msg * sdkws.MsgData ) string {
51- return c .getMessageCacheKey (conversationID , msg .Seq )
52- })
53- keys := datautil .Slice (msgs , func (msg * sdkws.MsgData ) string {
54- return c .getMessageCacheKey (conversationID , msg .Seq )
55- })
56- err := ProcessKeysBySlot (ctx , c .rdb , keys , func (ctx context.Context , slot int64 , keys []string ) error {
57- var values []string
58- for _ , key := range keys {
59- if msg , ok := msgMap [key ]; ok {
60- s , err := msgprocessor .Pb2String (msg )
61- if err != nil {
62- return err
63- }
64- values = append (values , s )
65- }
66- }
67- return LuaSetBatchWithCommonExpire (ctx , c .rdb , keys , values , msgCacheTimeout )
68- })
69- if err != nil {
70- return 0 , err
71- }
72- return len (msgs ), nil
73- }
74-
75- func (c * msgCache ) DeleteMessagesFromCache (ctx context.Context , conversationID string , seqs []int64 ) error {
76- var keys []string
77- for _ , seq := range seqs {
78- keys = append (keys , c .getMessageCacheKey (conversationID , seq ))
79- }
80-
81- return ProcessKeysBySlot (ctx , c .rdb , keys , func (ctx context.Context , slot int64 , keys []string ) error {
82- return LuaDeleteBatch (ctx , c .rdb , keys )
83- })
84- }
85-
8638func (c * msgCache ) SetSendMsgStatus (ctx context.Context , id string , status int32 ) error {
8739 return errs .Wrap (c .rdb .Set (ctx , c .getSendMsgKey (id ), status , time .Hour * 24 ).Err ())
8840}
@@ -92,81 +44,53 @@ func (c *msgCache) GetSendMsgStatus(ctx context.Context, id string) (int32, erro
9244 return int32 (result ), errs .Wrap (err )
9345}
9446
95- func (c * msgCache ) LockMessageTypeKey (ctx context.Context , clientMsgID string , TypeKey string ) error {
96- key := c .getLockMessageTypeKey (clientMsgID , TypeKey )
97- return errs .Wrap (c .rdb .SetNX (ctx , key , 1 , time .Minute ).Err ())
98- }
99-
100- func (c * msgCache ) UnLockMessageTypeKey (ctx context.Context , clientMsgID string , TypeKey string ) error {
101- key := c .getLockMessageTypeKey (clientMsgID , TypeKey )
102- return errs .Wrap (c .rdb .Del (ctx , key ).Err ())
47+ func (c * msgCache ) GetMessageBySeqs (ctx context.Context , conversationID string , seqs []int64 ) ([]* model.MsgInfoModel , error ) {
48+ if len (seqs ) == 0 {
49+ return nil , nil
50+ }
51+ getKey := func (seq int64 ) string {
52+ return cachekey .GetMsgCacheKey (conversationID , seq )
53+ }
54+ getMsgID := func (msg * model.MsgInfoModel ) int64 {
55+ return msg .Msg .Seq
56+ }
57+ find := func (ctx context.Context , seqs []int64 ) ([]* model.MsgInfoModel , error ) {
58+ return c .msgDocDatabase .FindSeqs (ctx , conversationID , seqs )
59+ }
60+ return batchGetCache2 (ctx , c .rcClient , msgCacheTimeout , seqs , getKey , getMsgID , find )
10361}
10462
105- func (c * msgCache ) JudgeMessageReactionExist (ctx context.Context , clientMsgID string , sessionType int32 ) (bool , error ) {
106- n , err := c .rdb .Exists (ctx , c .getMessageReactionExPrefix (clientMsgID , sessionType )).Result ()
63+ func (c * msgCache ) DelMessageBySeqs (ctx context.Context , conversationID string , seqs []int64 ) error {
64+ if len (seqs ) == 0 {
65+ return nil
66+ }
67+ keys := datautil .Slice (seqs , func (seq int64 ) string {
68+ return cachekey .GetMsgCacheKey (conversationID , seq )
69+ })
70+ slotKeys , err := groupKeysBySlot (ctx , getRocksCacheRedisClient (c .rcClient ), keys )
10771 if err != nil {
108- return false , errs . Wrap ( err )
72+ return err
10973 }
110-
111- return n > 0 , nil
112- }
113-
114- func (c * msgCache ) SetMessageTypeKeyValue (ctx context.Context , clientMsgID string , sessionType int32 , typeKey , value string ) error {
115- return errs .Wrap (c .rdb .HSet (ctx , c .getMessageReactionExPrefix (clientMsgID , sessionType ), typeKey , value ).Err ())
116- }
117-
118- func (c * msgCache ) SetMessageReactionExpire (ctx context.Context , clientMsgID string , sessionType int32 , expiration time.Duration ) (bool , error ) {
119- val , err := c .rdb .Expire (ctx , c .getMessageReactionExPrefix (clientMsgID , sessionType ), expiration ).Result ()
120- return val , errs .Wrap (err )
121- }
122-
123- func (c * msgCache ) GetMessageTypeKeyValue (ctx context.Context , clientMsgID string , sessionType int32 , typeKey string ) (string , error ) {
124- val , err := c .rdb .HGet (ctx , c .getMessageReactionExPrefix (clientMsgID , sessionType ), typeKey ).Result ()
125- return val , errs .Wrap (err )
126- }
127-
128- func (c * msgCache ) GetOneMessageAllReactionList (ctx context.Context , clientMsgID string , sessionType int32 ) (map [string ]string , error ) {
129- val , err := c .rdb .HGetAll (ctx , c .getMessageReactionExPrefix (clientMsgID , sessionType )).Result ()
130- return val , errs .Wrap (err )
131- }
132-
133- func (c * msgCache ) DeleteOneMessageKey (ctx context.Context , clientMsgID string , sessionType int32 , subKey string ) error {
134- return errs .Wrap (c .rdb .HDel (ctx , c .getMessageReactionExPrefix (clientMsgID , sessionType ), subKey ).Err ())
74+ for _ , keys := range slotKeys {
75+ if err := c .rcClient .TagAsDeletedBatch2 (ctx , keys ); err != nil {
76+ return err
77+ }
78+ }
79+ return nil
13580}
13681
137- func (c * msgCache ) GetMessagesBySeq (ctx context.Context , conversationID string , seqs []int64 ) (seqMsgs []* sdkws.MsgData , failedSeqs []int64 , err error ) {
138- var keys []string
139- keySeqMap := make (map [string ]int64 , 10 )
140- for _ , seq := range seqs {
141- key := c .getMessageCacheKey (conversationID , seq )
142- keys = append (keys , key )
143- keySeqMap [key ] = seq
144- }
145- err = ProcessKeysBySlot (ctx , c .rdb , keys , func (ctx context.Context , slot int64 , keys []string ) error {
146- result , err := LuaGetBatch (ctx , c .rdb , keys )
82+ func (c * msgCache ) SetMessageBySeqs (ctx context.Context , conversationID string , msgs []* model.MsgInfoModel ) error {
83+ for _ , msg := range msgs {
84+ if msg == nil || msg .Msg == nil || msg .Msg .Seq <= 0 {
85+ continue
86+ }
87+ data , err := json .Marshal (msg )
14788 if err != nil {
14889 return err
14990 }
150- for i , value := range result {
151- seq := keySeqMap [keys [i ]]
152- if value == nil {
153- failedSeqs = append (failedSeqs , seq )
154- continue
155- }
156-
157- msg := & sdkws.MsgData {}
158- msgString , ok := value .(string )
159- if ! ok || msgprocessor .String2Pb (msgString , msg ) != nil {
160- failedSeqs = append (failedSeqs , seq )
161- continue
162- }
163- seqMsgs = append (seqMsgs , msg )
164-
91+ if err := c .rcClient .RawSet (ctx , cachekey .GetMsgCacheKey (conversationID , msg .Msg .Seq ), string (data ), msgCacheTimeout ); err != nil {
92+ return err
16593 }
166- return nil
167- })
168- if err != nil {
169- return nil , nil , err
17094 }
171- return seqMsgs , failedSeqs , nil
95+ return nil
17296}
0 commit comments