131 lines
3.8 KiB
Go
131 lines
3.8 KiB
Go
package model
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"strconv"
|
|
|
|
"github.com/redis/go-redis/v9"
|
|
|
|
"spark-server/internal/module/inbox/entity"
|
|
)
|
|
|
|
// RedisStore is C7 dialog list / unread / preview. No MySQL.
|
|
type RedisStore struct {
|
|
rdb *redis.Client
|
|
}
|
|
|
|
func NewRedisStore(rdb *redis.Client) *RedisStore {
|
|
return &RedisStore{rdb: rdb}
|
|
}
|
|
|
|
func unreadKey(uid int64) string { return fmt.Sprintf("im:dialog:unread:%d", uid) }
|
|
func readKey(uid int64) string { return fmt.Sprintf("im:dialog:read:%d", uid) }
|
|
func countKey(uid int64) string { return fmt.Sprintf("im:dialog:unreadcnt:%d", uid) }
|
|
func previewKey(cid int64) string {
|
|
return fmt.Sprintf("im:dialog:preview:%d", cid)
|
|
}
|
|
|
|
func (s *RedisStore) TouchUnread(ctx context.Context, userID, conversationID, score int64) error {
|
|
if s == nil || s.rdb == nil {
|
|
return nil
|
|
}
|
|
member := strconv.FormatInt(conversationID, 10)
|
|
pipe := s.rdb.Pipeline()
|
|
pipe.ZAdd(ctx, unreadKey(userID), redis.Z{Score: float64(score), Member: member})
|
|
pipe.ZRem(ctx, readKey(userID), member)
|
|
_, err := pipe.Exec(ctx)
|
|
return err
|
|
}
|
|
|
|
func (s *RedisStore) TouchRead(ctx context.Context, userID, conversationID, score int64) error {
|
|
if s == nil || s.rdb == nil {
|
|
return nil
|
|
}
|
|
member := strconv.FormatInt(conversationID, 10)
|
|
pipe := s.rdb.Pipeline()
|
|
pipe.ZAdd(ctx, readKey(userID), redis.Z{Score: float64(score), Member: member})
|
|
pipe.ZRem(ctx, unreadKey(userID), member)
|
|
_, err := pipe.Exec(ctx)
|
|
return err
|
|
}
|
|
|
|
func (s *RedisStore) IncrUnread(ctx context.Context, userID, conversationID, delta int64) error {
|
|
if s == nil || s.rdb == nil {
|
|
return nil
|
|
}
|
|
return s.rdb.HIncrBy(ctx, countKey(userID), strconv.FormatInt(conversationID, 10), delta).Err()
|
|
}
|
|
|
|
func (s *RedisStore) SetPreview(ctx context.Context, conversationID int64, preview []byte) error {
|
|
if s == nil || s.rdb == nil {
|
|
return nil
|
|
}
|
|
return s.rdb.Set(ctx, previewKey(conversationID), preview, 0).Err()
|
|
}
|
|
|
|
func (s *RedisStore) ListUpdatedSince(ctx context.Context, userID, score int64, limit int) ([]entity.DialogHit, error) {
|
|
if s == nil || s.rdb == nil {
|
|
return nil, nil
|
|
}
|
|
if limit <= 0 {
|
|
limit = 200
|
|
}
|
|
minScore := fmt.Sprintf("(%d", score)
|
|
unread, err := s.rdb.ZRangeByScoreWithScores(ctx, unreadKey(userID), &redis.ZRangeBy{Min: minScore, Max: "+inf", Count: int64(limit)}).Result()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
read, err := s.rdb.ZRangeByScoreWithScores(ctx, readKey(userID), &redis.ZRangeBy{Min: minScore, Max: "+inf", Count: int64(limit)}).Result()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
out := make([]entity.DialogHit, 0, len(unread)+len(read))
|
|
seen := map[int64]struct{}{}
|
|
for _, part := range [][]redis.Z{unread, read} {
|
|
for _, z := range part {
|
|
member, _ := z.Member.(string)
|
|
id, convErr := strconv.ParseInt(member, 10, 64)
|
|
if convErr != nil {
|
|
continue
|
|
}
|
|
if _, ok := seen[id]; ok {
|
|
continue
|
|
}
|
|
seen[id] = struct{}{}
|
|
out = append(out, entity.DialogHit{ConversationID: id, Score: int64(z.Score)})
|
|
}
|
|
}
|
|
if len(out) > limit {
|
|
out = out[:limit]
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
func (s *RedisStore) LoadDialogs(ctx context.Context, userID int64, hits []entity.DialogHit) ([]entity.Dialog, error) {
|
|
if s == nil || s.rdb == nil || len(hits) == 0 {
|
|
return nil, nil
|
|
}
|
|
pipe := s.rdb.Pipeline()
|
|
previewCmds := make([]*redis.StringCmd, len(hits))
|
|
countCmds := make([]*redis.StringCmd, len(hits))
|
|
ck := countKey(userID)
|
|
for i, hit := range hits {
|
|
previewCmds[i] = pipe.Get(ctx, previewKey(hit.ConversationID))
|
|
countCmds[i] = pipe.HGet(ctx, ck, strconv.FormatInt(hit.ConversationID, 10))
|
|
}
|
|
_, _ = pipe.Exec(ctx)
|
|
out := make([]entity.Dialog, 0, len(hits))
|
|
for i, hit := range hits {
|
|
d := entity.Dialog{ConversationID: hit.ConversationID, Score: hit.Score}
|
|
if raw, err := previewCmds[i].Bytes(); err == nil {
|
|
d.Preview = raw
|
|
}
|
|
if n, err := countCmds[i].Int64(); err == nil {
|
|
d.Unread = n
|
|
}
|
|
out = append(out, d)
|
|
}
|
|
return out, nil
|
|
}
|