1067 lines
36 KiB
Go
1067 lines
36 KiB
Go
package store
|
|
|
|
import (
|
|
"context"
|
|
"crypto/sha256"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"fmt"
|
|
"sort"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/redis/go-redis/v9"
|
|
"github.com/tinode/chat/server/store/types"
|
|
)
|
|
|
|
type redisConfig struct {
|
|
Addr string `json:"addr"`
|
|
Username string `json:"username"`
|
|
Password string `json:"password"`
|
|
DB int `json:"db"`
|
|
KeyPrefix string `json:"key_prefix"`
|
|
DialMS int `json:"dial_timeout_ms"`
|
|
ReadMS int `json:"read_timeout_ms"`
|
|
WriteMS int `json:"write_timeout_ms"`
|
|
PendingTTL int `json:"pending_ttl_seconds"`
|
|
DoneTTL int `json:"done_ttl_seconds"`
|
|
FailedTTL int `json:"failed_ttl_seconds"`
|
|
HotMsgTTL int `json:"hot_message_ttl_seconds"`
|
|
PresenceTTL int `json:"presence_ttl_seconds"`
|
|
CacheTTL int `json:"cache_ttl_seconds"`
|
|
ShortTTL int `json:"short_state_ttl_seconds"`
|
|
GroupFanout bool `json:"group_fanout_enabled"`
|
|
}
|
|
|
|
type groupMessageUniqueRecord struct {
|
|
Status string `json:"status"`
|
|
Seq int64 `json:"seq,omitempty"`
|
|
MessageID string `json:"message_id,omitempty"`
|
|
CreatedAt time.Time `json:"created_at"`
|
|
}
|
|
|
|
type MessageUniqueClaim struct {
|
|
Status string
|
|
Seq int64
|
|
}
|
|
|
|
type groupFanoutEnvelope struct {
|
|
OriginNodeID string `json:"origin_node_id"`
|
|
Message json.RawMessage `json:"message"`
|
|
}
|
|
|
|
type groupMessageRuntime interface {
|
|
Close() error
|
|
Claim(tenantID types.TenantID, topic string, from types.Uid, clientMsgID string) (*groupMessageUniqueRecord, bool, error)
|
|
NextSeq(tenantID types.TenantID, topic string, baseline int64) (int64, error)
|
|
MarkPending(tenantID types.TenantID, topic string, from types.Uid, clientMsgID string, seq int64) error
|
|
MarkDone(tenantID types.TenantID, topic string, from types.Uid, clientMsgID string, seq int64, msgID string) error
|
|
MarkFailed(tenantID types.TenantID, topic string, from types.Uid, clientMsgID string) error
|
|
CacheHotMessage(tenantID types.TenantID, msg *types.Message) error
|
|
CachePendingHotMessage(tenantID types.TenantID, msg *types.Message) error
|
|
GetHotMessages(tenantID types.TenantID, topic string, opts *types.QueryOpt, maxLimit int) ([]types.Message, bool, error)
|
|
InvalidateHotMessages(tenantID types.TenantID, topic string, ranges []types.Range) error
|
|
MarkUserOnline(tenantID types.TenantID, uid types.Uid, sid, nodeID, device, userAgent string) error
|
|
RefreshUserPresence(tenantID types.TenantID, uid types.Uid, sid, nodeID string) error
|
|
MarkUserOffline(tenantID types.TenantID, uid types.Uid, sid, nodeID string) error
|
|
GetUserPresence(tenantID types.TenantID, uid types.Uid) (*UserPresence, error)
|
|
MarkTopicMemberOnline(tenantID types.TenantID, topic string, uid types.Uid, nodeID string) error
|
|
MarkTopicMemberOffline(tenantID types.TenantID, topic string, uid types.Uid, nodeID string) error
|
|
GetTopicNodeOnlineMembers(tenantID types.TenantID, topic, nodeID string) ([]types.Uid, error)
|
|
PublishGroupFanout(originNodeID string, message any) error
|
|
StartGroupFanoutSubscriber(ctx context.Context, nodeID string, handler func([]byte)) error
|
|
CacheObject(tenantID types.TenantID, namespace, id string, value any) error
|
|
GetCachedObject(tenantID types.TenantID, namespace, id string, value any) (bool, error)
|
|
DeleteCachedObject(tenantID types.TenantID, namespace, id string) error
|
|
DeleteCachedObjectsByPrefix(tenantID types.TenantID, namespace, idPrefix string) error
|
|
AddCachedSetMember(tenantID types.TenantID, namespace, id, member string) error
|
|
RemoveCachedSetMember(tenantID types.TenantID, namespace, id, member string) error
|
|
CachedSetContains(tenantID types.TenantID, namespace, id, member string) (bool, error)
|
|
AllowRate(tenantID types.TenantID, scope string, limit int, window time.Duration) (bool, int, time.Duration, error)
|
|
ClaimShortState(namespace, id string, ttl time.Duration) (bool, error)
|
|
ClaimShortStateValue(namespace, id, value string, ttl time.Duration) (bool, bool, error)
|
|
ShortStateExists(namespace, id string) (bool, error)
|
|
}
|
|
|
|
var groupMsgRuntime groupMessageRuntime
|
|
|
|
func ClaimMessageUnique(tenantID types.TenantID, topic string, from types.Uid, clientMsgID string) (*MessageUniqueClaim, bool, bool, error) {
|
|
if groupMsgRuntime == nil || !tenantID.IsValid() || topic == "" || from.IsZero() || clientMsgID == "" {
|
|
return nil, true, false, nil
|
|
}
|
|
rec, claimed, err := groupMsgRuntime.Claim(tenantID, topic, from, clientMsgID)
|
|
if rec == nil {
|
|
return nil, claimed, true, err
|
|
}
|
|
return &MessageUniqueClaim{Status: rec.Status, Seq: rec.Seq}, claimed, true, err
|
|
}
|
|
|
|
func ReserveMessageSeq(tenantID types.TenantID, topic string, baseline int64) (int64, bool, error) {
|
|
if groupMsgRuntime == nil || !tenantID.IsValid() || topic == "" {
|
|
return 0, false, nil
|
|
}
|
|
seq, err := groupMsgRuntime.NextSeq(tenantID, topic, baseline)
|
|
return seq, true, err
|
|
}
|
|
|
|
func MarkMessageUniqueDone(tenantID types.TenantID, topic string, from types.Uid, clientMsgID string, seq int64, msgID string) error {
|
|
if groupMsgRuntime == nil || !tenantID.IsValid() || topic == "" || from.IsZero() || clientMsgID == "" {
|
|
return nil
|
|
}
|
|
return groupMsgRuntime.MarkDone(tenantID, topic, from, clientMsgID, seq, msgID)
|
|
}
|
|
|
|
func MarkMessageUniquePending(tenantID types.TenantID, topic string, from types.Uid, clientMsgID string, seq int64) error {
|
|
if groupMsgRuntime == nil || !tenantID.IsValid() || topic == "" || from.IsZero() || clientMsgID == "" || seq <= 0 {
|
|
return nil
|
|
}
|
|
return groupMsgRuntime.MarkPending(tenantID, topic, from, clientMsgID, seq)
|
|
}
|
|
|
|
func MarkMessageUniqueFailed(tenantID types.TenantID, topic string, from types.Uid, clientMsgID string) error {
|
|
if groupMsgRuntime == nil || !tenantID.IsValid() || topic == "" || from.IsZero() || clientMsgID == "" {
|
|
return nil
|
|
}
|
|
return groupMsgRuntime.MarkFailed(tenantID, topic, from, clientMsgID)
|
|
}
|
|
|
|
type UserPresence struct {
|
|
Online bool `json:"online"`
|
|
NodeID string `json:"node_id"`
|
|
SessionCount int `json:"session_count"`
|
|
LastActiveAt time.Time `json:"last_active_at"`
|
|
LastSeenAt time.Time `json:"last_seen_at"`
|
|
}
|
|
|
|
func MarkUserOnline(tenantID types.TenantID, uid types.Uid, sid, nodeID, device, userAgent string) error {
|
|
if groupMsgRuntime == nil || !tenantID.IsValid() || uid.IsZero() || sid == "" {
|
|
return nil
|
|
}
|
|
return groupMsgRuntime.MarkUserOnline(tenantID, uid, sid, nodeID, device, userAgent)
|
|
}
|
|
|
|
func ClaimShortState(namespace, id string, ttl time.Duration) (bool, bool, error) {
|
|
if groupMsgRuntime == nil || namespace == "" || id == "" {
|
|
return false, false, nil
|
|
}
|
|
ok, err := groupMsgRuntime.ClaimShortState(namespace, id, ttl)
|
|
return ok, true, err
|
|
}
|
|
|
|
func ClaimShortStateValue(namespace, id, value string, ttl time.Duration) (bool, bool, bool, error) {
|
|
if groupMsgRuntime == nil || namespace == "" || id == "" || value == "" {
|
|
return false, false, false, nil
|
|
}
|
|
claimed, sameValue, err := groupMsgRuntime.ClaimShortStateValue(namespace, id, value, ttl)
|
|
return claimed, sameValue, true, err
|
|
}
|
|
|
|
func RevokeLoginToken(tenantID types.TenantID, token []byte, ttl time.Duration) (bool, error) {
|
|
if groupMsgRuntime == nil || !tenantID.IsValid() || len(token) == 0 || ttl <= 0 {
|
|
return false, nil
|
|
}
|
|
digest := sha256.Sum256(token)
|
|
id := strconv.FormatInt(int64(tenantID), 10) + ":" + hex.EncodeToString(digest[:])
|
|
_, err := groupMsgRuntime.ClaimShortState("revoked_login_token", id, ttl)
|
|
return true, err
|
|
}
|
|
|
|
func IsLoginTokenRevoked(tenantID types.TenantID, token []byte) (bool, bool, error) {
|
|
if groupMsgRuntime == nil || !tenantID.IsValid() || len(token) == 0 {
|
|
return false, false, nil
|
|
}
|
|
digest := sha256.Sum256(token)
|
|
id := strconv.FormatInt(int64(tenantID), 10) + ":" + hex.EncodeToString(digest[:])
|
|
revoked, err := groupMsgRuntime.ShortStateExists("revoked_login_token", id)
|
|
return revoked, true, err
|
|
}
|
|
|
|
func RefreshUserPresence(tenantID types.TenantID, uid types.Uid, sid, nodeID string) error {
|
|
if groupMsgRuntime == nil || !tenantID.IsValid() || uid.IsZero() || sid == "" {
|
|
return nil
|
|
}
|
|
return groupMsgRuntime.RefreshUserPresence(tenantID, uid, sid, nodeID)
|
|
}
|
|
|
|
func MarkUserOffline(tenantID types.TenantID, uid types.Uid, sid, nodeID string) error {
|
|
if groupMsgRuntime == nil || !tenantID.IsValid() || uid.IsZero() || sid == "" {
|
|
return nil
|
|
}
|
|
return groupMsgRuntime.MarkUserOffline(tenantID, uid, sid, nodeID)
|
|
}
|
|
|
|
func GetUserPresence(tenantID types.TenantID, uid types.Uid) (*UserPresence, error) {
|
|
if groupMsgRuntime == nil || !tenantID.IsValid() || uid.IsZero() {
|
|
return &UserPresence{}, nil
|
|
}
|
|
return groupMsgRuntime.GetUserPresence(tenantID, uid)
|
|
}
|
|
|
|
func UserPresenceEnabled() bool {
|
|
return groupMsgRuntime != nil
|
|
}
|
|
|
|
func MarkTopicMemberOnline(tenantID types.TenantID, topic string, uid types.Uid, nodeID string) error {
|
|
if groupMsgRuntime == nil || !tenantID.IsValid() || topic == "" || uid.IsZero() {
|
|
return nil
|
|
}
|
|
return groupMsgRuntime.MarkTopicMemberOnline(tenantID, topic, uid, nodeID)
|
|
}
|
|
|
|
func MarkTopicMemberOffline(tenantID types.TenantID, topic string, uid types.Uid, nodeID string) error {
|
|
if groupMsgRuntime == nil || !tenantID.IsValid() || topic == "" || uid.IsZero() {
|
|
return nil
|
|
}
|
|
return groupMsgRuntime.MarkTopicMemberOffline(tenantID, topic, uid, nodeID)
|
|
}
|
|
|
|
func GetTopicNodeOnlineMembers(tenantID types.TenantID, topic, nodeID string) ([]types.Uid, error) {
|
|
if groupMsgRuntime == nil || !tenantID.IsValid() || topic == "" {
|
|
return nil, nil
|
|
}
|
|
return groupMsgRuntime.GetTopicNodeOnlineMembers(tenantID, topic, nodeID)
|
|
}
|
|
|
|
func PublishGroupFanout(originNodeID string, message any) error {
|
|
if groupMsgRuntime == nil || originNodeID == "" || message == nil {
|
|
return nil
|
|
}
|
|
return groupMsgRuntime.PublishGroupFanout(originNodeID, message)
|
|
}
|
|
|
|
func StartGroupFanoutSubscriber(ctx context.Context, nodeID string, handler func([]byte)) error {
|
|
if groupMsgRuntime == nil || nodeID == "" || handler == nil {
|
|
return nil
|
|
}
|
|
return groupMsgRuntime.StartGroupFanoutSubscriber(ctx, nodeID, handler)
|
|
}
|
|
|
|
func CacheObject(tenantID types.TenantID, namespace, id string, value any) error {
|
|
if groupMsgRuntime == nil || !tenantID.IsValid() || namespace == "" || id == "" || value == nil {
|
|
return nil
|
|
}
|
|
return groupMsgRuntime.CacheObject(tenantID, namespace, id, value)
|
|
}
|
|
|
|
func GetCachedObject(tenantID types.TenantID, namespace, id string, value any) (bool, error) {
|
|
if groupMsgRuntime == nil || !tenantID.IsValid() || namespace == "" || id == "" || value == nil {
|
|
return false, nil
|
|
}
|
|
return groupMsgRuntime.GetCachedObject(tenantID, namespace, id, value)
|
|
}
|
|
|
|
func DeleteCachedObject(tenantID types.TenantID, namespace, id string) error {
|
|
if groupMsgRuntime == nil || !tenantID.IsValid() || namespace == "" || id == "" {
|
|
return nil
|
|
}
|
|
return groupMsgRuntime.DeleteCachedObject(tenantID, namespace, id)
|
|
}
|
|
|
|
func DeleteCachedObjectsByPrefix(tenantID types.TenantID, namespace, idPrefix string) error {
|
|
if groupMsgRuntime == nil || !tenantID.IsValid() || namespace == "" || idPrefix == "" {
|
|
return nil
|
|
}
|
|
return groupMsgRuntime.DeleteCachedObjectsByPrefix(tenantID, namespace, idPrefix)
|
|
}
|
|
|
|
func CacheHotMessage(tenantID types.TenantID, msg *types.Message) error {
|
|
if groupMsgRuntime == nil || !tenantID.IsValid() || msg == nil || msg.Topic == "" || msg.SeqId <= 0 {
|
|
return nil
|
|
}
|
|
return groupMsgRuntime.CacheHotMessage(tenantID, msg)
|
|
}
|
|
|
|
func CachePendingHotMessage(tenantID types.TenantID, msg *types.Message) error {
|
|
if groupMsgRuntime == nil || !tenantID.IsValid() || msg == nil || msg.Topic == "" || msg.SeqId <= 0 {
|
|
return nil
|
|
}
|
|
return groupMsgRuntime.CachePendingHotMessage(tenantID, msg)
|
|
}
|
|
|
|
func GetHotMessages(tenantID types.TenantID, topic string, opts *types.QueryOpt, maxLimit int) ([]types.Message, bool, error) {
|
|
if groupMsgRuntime == nil || !tenantID.IsValid() || topic == "" {
|
|
return nil, false, nil
|
|
}
|
|
return groupMsgRuntime.GetHotMessages(tenantID, topic, opts, maxLimit)
|
|
}
|
|
|
|
func InvalidateHotMessages(tenantID types.TenantID, topic string, ranges []types.Range) error {
|
|
if groupMsgRuntime == nil || !tenantID.IsValid() || topic == "" {
|
|
return nil
|
|
}
|
|
return groupMsgRuntime.InvalidateHotMessages(tenantID, topic, ranges)
|
|
}
|
|
|
|
func AddCachedSetMember(tenantID types.TenantID, namespace, id, member string) error {
|
|
if groupMsgRuntime == nil || !tenantID.IsValid() || namespace == "" || id == "" || member == "" {
|
|
return nil
|
|
}
|
|
return groupMsgRuntime.AddCachedSetMember(tenantID, namespace, id, member)
|
|
}
|
|
|
|
func RemoveCachedSetMember(tenantID types.TenantID, namespace, id, member string) error {
|
|
if groupMsgRuntime == nil || !tenantID.IsValid() || namespace == "" || id == "" || member == "" {
|
|
return nil
|
|
}
|
|
return groupMsgRuntime.RemoveCachedSetMember(tenantID, namespace, id, member)
|
|
}
|
|
|
|
func CachedSetContains(tenantID types.TenantID, namespace, id, member string) (bool, error) {
|
|
if groupMsgRuntime == nil || !tenantID.IsValid() || namespace == "" || id == "" || member == "" {
|
|
return false, nil
|
|
}
|
|
return groupMsgRuntime.CachedSetContains(tenantID, namespace, id, member)
|
|
}
|
|
|
|
func AllowRate(tenantID types.TenantID, scope string, limit int, window time.Duration) (bool, int, time.Duration, error) {
|
|
if groupMsgRuntime == nil || !tenantID.IsValid() || scope == "" || limit <= 0 || window <= 0 {
|
|
return true, 0, 0, nil
|
|
}
|
|
return groupMsgRuntime.AllowRate(tenantID, scope, limit, window)
|
|
}
|
|
|
|
func initGroupMessageRuntime(raw json.RawMessage) error {
|
|
closeGroupMessageRuntime()
|
|
if len(raw) == 0 || string(raw) == "null" {
|
|
return nil
|
|
}
|
|
var cfg redisConfig
|
|
if err := json.Unmarshal(raw, &cfg); err != nil {
|
|
return err
|
|
}
|
|
cfg.Addr = strings.TrimSpace(cfg.Addr)
|
|
if cfg.Addr == "" {
|
|
return nil
|
|
}
|
|
if cfg.KeyPrefix == "" {
|
|
cfg.KeyPrefix = "im"
|
|
}
|
|
if cfg.PendingTTL <= 0 {
|
|
cfg.PendingTTL = 300
|
|
}
|
|
if cfg.DoneTTL <= 0 {
|
|
cfg.DoneTTL = 86400
|
|
}
|
|
if cfg.FailedTTL <= 0 {
|
|
cfg.FailedTTL = 600
|
|
}
|
|
if cfg.HotMsgTTL <= 0 {
|
|
cfg.HotMsgTTL = 86400
|
|
}
|
|
if cfg.PresenceTTL <= 0 {
|
|
cfg.PresenceTTL = 120
|
|
}
|
|
if cfg.CacheTTL <= 0 {
|
|
cfg.CacheTTL = 300
|
|
}
|
|
if cfg.ShortTTL <= 0 {
|
|
cfg.ShortTTL = 600
|
|
}
|
|
dialTimeout := timeoutOrDefault(cfg.DialMS, 2*time.Second)
|
|
readTimeout := timeoutOrDefault(cfg.ReadMS, 2*time.Second)
|
|
writeTimeout := timeoutOrDefault(cfg.WriteMS, 2*time.Second)
|
|
client := redis.NewClient(&redis.Options{
|
|
Addr: cfg.Addr,
|
|
Username: cfg.Username,
|
|
Password: cfg.Password,
|
|
DB: cfg.DB,
|
|
DialTimeout: dialTimeout,
|
|
ReadTimeout: readTimeout,
|
|
WriteTimeout: writeTimeout,
|
|
})
|
|
ctx, cancel := context.WithTimeout(context.Background(), dialTimeout)
|
|
defer cancel()
|
|
if err := client.Ping(ctx).Err(); err != nil {
|
|
_ = client.Close()
|
|
return err
|
|
}
|
|
groupMsgRuntime = &redisGroupMessageRuntime{client: client, prefix: cfg.KeyPrefix, cfg: cfg}
|
|
return nil
|
|
}
|
|
|
|
func closeGroupMessageRuntime() {
|
|
if groupMsgRuntime != nil {
|
|
_ = groupMsgRuntime.Close()
|
|
groupMsgRuntime = nil
|
|
}
|
|
}
|
|
|
|
func timeoutOrDefault(ms int, fallback time.Duration) time.Duration {
|
|
if ms <= 0 {
|
|
return fallback
|
|
}
|
|
return time.Duration(ms) * time.Millisecond
|
|
}
|
|
|
|
type redisGroupMessageRuntime struct {
|
|
client *redis.Client
|
|
prefix string
|
|
cfg redisConfig
|
|
}
|
|
|
|
func (r *redisGroupMessageRuntime) Close() error {
|
|
return r.client.Close()
|
|
}
|
|
|
|
func (r *redisGroupMessageRuntime) Claim(tenantID types.TenantID, topic string, from types.Uid, clientMsgID string) (*groupMessageUniqueRecord, bool, error) {
|
|
if clientMsgID == "" {
|
|
return nil, true, nil
|
|
}
|
|
key := r.uniqueKey(tenantID, topic, from, clientMsgID)
|
|
rec := &groupMessageUniqueRecord{Status: "pending", CreatedAt: time.Now().UTC()}
|
|
raw, _ := json.Marshal(rec)
|
|
ctx := context.Background()
|
|
ok, err := r.client.SetNX(ctx, key, raw, time.Duration(r.cfg.PendingTTL)*time.Second).Result()
|
|
if err != nil {
|
|
return nil, false, err
|
|
}
|
|
if ok {
|
|
return rec, true, nil
|
|
}
|
|
existing, err := r.client.Get(ctx, key).Bytes()
|
|
if err == redis.Nil {
|
|
return nil, false, nil
|
|
}
|
|
if err != nil {
|
|
return nil, false, err
|
|
}
|
|
var current groupMessageUniqueRecord
|
|
if err = json.Unmarshal(existing, ¤t); err != nil {
|
|
return nil, false, err
|
|
}
|
|
if current.Status == "failed" {
|
|
_ = r.client.Del(ctx, key).Err()
|
|
ok, err = r.client.SetNX(ctx, key, raw, time.Duration(r.cfg.PendingTTL)*time.Second).Result()
|
|
if err != nil {
|
|
return nil, false, err
|
|
}
|
|
if ok {
|
|
return rec, true, nil
|
|
}
|
|
}
|
|
return ¤t, false, nil
|
|
}
|
|
|
|
func (r *redisGroupMessageRuntime) NextSeq(tenantID types.TenantID, topic string, baseline int64) (int64, error) {
|
|
if baseline < 0 {
|
|
baseline = 0
|
|
}
|
|
seqKey := r.seqKey(tenantID, topic)
|
|
seq, err := r.client.Eval(context.Background(), `
|
|
local current = redis.call("GET", KEYS[1])
|
|
if not current then
|
|
redis.call("SET", KEYS[1], ARGV[1])
|
|
elseif tonumber(current) < tonumber(ARGV[1]) then
|
|
redis.call("SET", KEYS[1], ARGV[1])
|
|
end
|
|
return redis.call("INCR", KEYS[1])
|
|
`, []string{seqKey}, baseline).Int64()
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
return seq, nil
|
|
}
|
|
|
|
func (r *redisGroupMessageRuntime) MarkDone(tenantID types.TenantID, topic string, from types.Uid, clientMsgID string, seq int64, msgID string) error {
|
|
if clientMsgID == "" {
|
|
return nil
|
|
}
|
|
rec := &groupMessageUniqueRecord{Status: "done", Seq: seq, MessageID: msgID, CreatedAt: time.Now().UTC()}
|
|
raw, _ := json.Marshal(rec)
|
|
return r.client.Set(context.Background(), r.uniqueKey(tenantID, topic, from, clientMsgID), raw, time.Duration(r.cfg.DoneTTL)*time.Second).Err()
|
|
}
|
|
|
|
func (r *redisGroupMessageRuntime) MarkPending(tenantID types.TenantID, topic string, from types.Uid, clientMsgID string, seq int64) error {
|
|
if clientMsgID == "" || seq <= 0 {
|
|
return nil
|
|
}
|
|
rec := &groupMessageUniqueRecord{Status: "pending", Seq: seq, CreatedAt: time.Now().UTC()}
|
|
raw, _ := json.Marshal(rec)
|
|
return r.client.Set(context.Background(), r.uniqueKey(tenantID, topic, from, clientMsgID), raw, time.Duration(r.cfg.PendingTTL)*time.Second).Err()
|
|
}
|
|
|
|
func (r *redisGroupMessageRuntime) MarkFailed(tenantID types.TenantID, topic string, from types.Uid, clientMsgID string) error {
|
|
if clientMsgID == "" {
|
|
return nil
|
|
}
|
|
rec := &groupMessageUniqueRecord{Status: "failed", CreatedAt: time.Now().UTC()}
|
|
raw, _ := json.Marshal(rec)
|
|
return r.client.Set(context.Background(), r.uniqueKey(tenantID, topic, from, clientMsgID), raw, time.Duration(r.cfg.FailedTTL)*time.Second).Err()
|
|
}
|
|
|
|
func (r *redisGroupMessageRuntime) CacheHotMessage(tenantID types.TenantID, msg *types.Message) error {
|
|
return r.cacheHotMessage(tenantID, msg, time.Duration(r.cfg.HotMsgTTL)*time.Second)
|
|
}
|
|
|
|
func (r *redisGroupMessageRuntime) CachePendingHotMessage(tenantID types.TenantID, msg *types.Message) error {
|
|
return r.cacheHotMessage(tenantID, msg, time.Duration(r.cfg.PendingTTL)*time.Second)
|
|
}
|
|
|
|
func (r *redisGroupMessageRuntime) cacheHotMessage(tenantID types.TenantID, msg *types.Message, ttl time.Duration) error {
|
|
if msg == nil || msg.Topic == "" || msg.SeqId <= 0 {
|
|
return types.ErrMalformed
|
|
}
|
|
if ttl <= 0 {
|
|
return types.ErrMalformed
|
|
}
|
|
payload, err := json.Marshal(msg)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
ctx := context.Background()
|
|
seq := strconv.FormatInt(msg.SeqId, 10)
|
|
pipe := r.client.Pipeline()
|
|
pipe.Set(ctx, r.hotMsgKey(tenantID, msg.Topic, msg.SeqId), payload, ttl)
|
|
pipe.ZAdd(ctx, r.hotSeqKey(tenantID, msg.Topic), redis.Z{Score: float64(msg.SeqId), Member: seq})
|
|
pipe.Expire(ctx, r.hotSeqKey(tenantID, msg.Topic), ttl+time.Hour)
|
|
_, err = pipe.Exec(ctx)
|
|
return err
|
|
}
|
|
|
|
func (r *redisGroupMessageRuntime) GetHotMessages(tenantID types.TenantID, topic string, opts *types.QueryOpt, maxLimit int) ([]types.Message, bool, error) {
|
|
if topic == "" {
|
|
return nil, false, types.ErrMalformed
|
|
}
|
|
limit := maxLimit
|
|
if limit <= 0 {
|
|
limit = 100
|
|
}
|
|
if opts != nil && opts.Limit > 0 && opts.Limit < limit {
|
|
limit = opts.Limit
|
|
}
|
|
seqs, completeRange, err := r.hotSeqIDs(tenantID, topic, opts, limit)
|
|
if err != nil || len(seqs) == 0 || !completeRange {
|
|
return nil, false, err
|
|
}
|
|
keys := make([]string, 0, len(seqs))
|
|
for _, seq := range seqs {
|
|
keys = append(keys, r.hotMsgKey(tenantID, topic, seq))
|
|
}
|
|
values, err := r.client.MGet(context.Background(), keys...).Result()
|
|
if err != nil {
|
|
return nil, false, err
|
|
}
|
|
if len(values) != len(seqs) {
|
|
return nil, false, nil
|
|
}
|
|
msgs := make([]types.Message, 0, len(values))
|
|
for _, value := range values {
|
|
if value == nil {
|
|
return nil, false, nil
|
|
}
|
|
raw, ok := value.(string)
|
|
if !ok {
|
|
return nil, false, nil
|
|
}
|
|
var msg types.Message
|
|
if err := json.Unmarshal([]byte(raw), &msg); err != nil {
|
|
return nil, false, err
|
|
}
|
|
msg.TenantID = tenantID
|
|
msgs = append(msgs, msg)
|
|
}
|
|
return msgs, true, nil
|
|
}
|
|
|
|
func (r *redisGroupMessageRuntime) InvalidateHotMessages(tenantID types.TenantID, topic string, ranges []types.Range) error {
|
|
if topic == "" {
|
|
return types.ErrMalformed
|
|
}
|
|
seqs := expandRanges(ranges, int(^uint(0)>>1))
|
|
ctx := context.Background()
|
|
if len(seqs) == 0 {
|
|
pipe := r.client.Pipeline()
|
|
pipe.Del(ctx, r.hotSeqKey(tenantID, topic))
|
|
keys, err := r.client.Keys(ctx, r.hotMsgKey(tenantID, topic, "*")).Result()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if len(keys) > 0 {
|
|
pipe.Del(ctx, keys...)
|
|
}
|
|
_, err = pipe.Exec(ctx)
|
|
return err
|
|
}
|
|
pipe := r.client.Pipeline()
|
|
members := make([]any, 0, len(seqs))
|
|
for _, seq := range seqs {
|
|
pipe.Del(ctx, r.hotMsgKey(tenantID, topic, seq))
|
|
members = append(members, strconv.FormatInt(int64(seq), 10))
|
|
}
|
|
pipe.ZRem(ctx, r.hotSeqKey(tenantID, topic), members...)
|
|
_, err := pipe.Exec(ctx)
|
|
return err
|
|
}
|
|
|
|
func (r *redisGroupMessageRuntime) MarkUserOnline(tenantID types.TenantID, uid types.Uid, sid, nodeID, device, userAgent string) error {
|
|
if nodeID == "" {
|
|
nodeID = "local"
|
|
}
|
|
now := time.Now().UTC()
|
|
ttl := time.Duration(r.cfg.PresenceTTL) * time.Second
|
|
ctx := context.Background()
|
|
userKey := r.presenceUserKey(tenantID, uid)
|
|
sessionsKey := r.presenceSessionsKey(tenantID, uid)
|
|
sessionKey := r.presenceSessionKey(tenantID, uid, sid)
|
|
nodeKey := r.presenceNodeUsersKey(tenantID, nodeID)
|
|
member := nodeID + ":" + sid
|
|
pipe := r.client.Pipeline()
|
|
pipe.HSet(ctx, sessionKey, map[string]any{
|
|
"node_id": nodeID,
|
|
"session_id": sid,
|
|
"device": device,
|
|
"user_agent": userAgent,
|
|
"last_active_at": now.Format(time.RFC3339Nano),
|
|
})
|
|
pipe.Expire(ctx, sessionKey, ttl)
|
|
pipe.SAdd(ctx, sessionsKey, member)
|
|
pipe.Expire(ctx, sessionsKey, ttl)
|
|
pipe.SAdd(ctx, nodeKey, uid.UserId())
|
|
pipe.Expire(ctx, nodeKey, ttl)
|
|
pipe.HSet(ctx, userKey, map[string]any{
|
|
"online": "1",
|
|
"node_id": nodeID,
|
|
"last_active_at": now.Format(time.RFC3339Nano),
|
|
"last_seen_at": now.Format(time.RFC3339Nano),
|
|
})
|
|
pipe.Expire(ctx, userKey, ttl)
|
|
if _, err := pipe.Exec(ctx); err != nil {
|
|
return err
|
|
}
|
|
count, err := r.client.SCard(ctx, sessionsKey).Result()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return r.client.HSet(ctx, userKey, "session_count", count).Err()
|
|
}
|
|
|
|
func (r *redisGroupMessageRuntime) RefreshUserPresence(tenantID types.TenantID, uid types.Uid, sid, nodeID string) error {
|
|
if nodeID == "" {
|
|
nodeID = "local"
|
|
}
|
|
now := time.Now().UTC()
|
|
ttl := time.Duration(r.cfg.PresenceTTL) * time.Second
|
|
ctx := context.Background()
|
|
userKey := r.presenceUserKey(tenantID, uid)
|
|
sessionsKey := r.presenceSessionsKey(tenantID, uid)
|
|
sessionKey := r.presenceSessionKey(tenantID, uid, sid)
|
|
nodeKey := r.presenceNodeUsersKey(tenantID, nodeID)
|
|
pipe := r.client.Pipeline()
|
|
pipe.HSet(ctx, sessionKey, "last_active_at", now.Format(time.RFC3339Nano), "node_id", nodeID)
|
|
pipe.Expire(ctx, sessionKey, ttl)
|
|
pipe.Expire(ctx, sessionsKey, ttl)
|
|
pipe.SAdd(ctx, nodeKey, uid.UserId())
|
|
pipe.Expire(ctx, nodeKey, ttl)
|
|
pipe.HSet(ctx, userKey, "online", "1", "node_id", nodeID, "last_active_at", now.Format(time.RFC3339Nano), "last_seen_at", now.Format(time.RFC3339Nano))
|
|
pipe.Expire(ctx, userKey, ttl)
|
|
_, err := pipe.Exec(ctx)
|
|
return err
|
|
}
|
|
|
|
func (r *redisGroupMessageRuntime) MarkUserOffline(tenantID types.TenantID, uid types.Uid, sid, nodeID string) error {
|
|
if nodeID == "" {
|
|
nodeID = "local"
|
|
}
|
|
now := time.Now().UTC()
|
|
ttl := time.Duration(r.cfg.PresenceTTL) * time.Second
|
|
ctx := context.Background()
|
|
userKey := r.presenceUserKey(tenantID, uid)
|
|
sessionsKey := r.presenceSessionsKey(tenantID, uid)
|
|
sessionKey := r.presenceSessionKey(tenantID, uid, sid)
|
|
nodeKey := r.presenceNodeUsersKey(tenantID, nodeID)
|
|
member := nodeID + ":" + sid
|
|
pipe := r.client.Pipeline()
|
|
pipe.Del(ctx, sessionKey)
|
|
pipe.SRem(ctx, sessionsKey, member)
|
|
pipe.SRem(ctx, nodeKey, uid.UserId())
|
|
if _, err := pipe.Exec(ctx); err != nil {
|
|
return err
|
|
}
|
|
count, err := r.client.SCard(ctx, sessionsKey).Result()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if count > 0 {
|
|
pipe = r.client.Pipeline()
|
|
pipe.HSet(ctx, userKey, "online", "1", "session_count", count, "last_seen_at", now.Format(time.RFC3339Nano))
|
|
pipe.Expire(ctx, userKey, ttl)
|
|
pipe.Expire(ctx, sessionsKey, ttl)
|
|
_, err = pipe.Exec(ctx)
|
|
return err
|
|
}
|
|
pipe = r.client.Pipeline()
|
|
pipe.HSet(ctx, userKey, "online", "0", "session_count", 0, "node_id", nodeID, "last_seen_at", now.Format(time.RFC3339Nano))
|
|
pipe.Expire(ctx, userKey, ttl)
|
|
pipe.Del(ctx, sessionsKey)
|
|
_, err = pipe.Exec(ctx)
|
|
return err
|
|
}
|
|
|
|
func (r *redisGroupMessageRuntime) GetUserPresence(tenantID types.TenantID, uid types.Uid) (*UserPresence, error) {
|
|
raw, err := r.client.HGetAll(context.Background(), r.presenceUserKey(tenantID, uid)).Result()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
presence := &UserPresence{}
|
|
if len(raw) == 0 {
|
|
return presence, nil
|
|
}
|
|
presence.Online = raw["online"] == "1"
|
|
presence.NodeID = raw["node_id"]
|
|
presence.SessionCount, _ = strconv.Atoi(raw["session_count"])
|
|
presence.LastActiveAt, _ = time.Parse(time.RFC3339Nano, raw["last_active_at"])
|
|
presence.LastSeenAt, _ = time.Parse(time.RFC3339Nano, raw["last_seen_at"])
|
|
return presence, nil
|
|
}
|
|
|
|
func (r *redisGroupMessageRuntime) MarkTopicMemberOnline(tenantID types.TenantID, topic string, uid types.Uid, nodeID string) error {
|
|
if nodeID == "" {
|
|
nodeID = "local"
|
|
}
|
|
ttl := time.Duration(r.cfg.PresenceTTL) * time.Second
|
|
ctx := context.Background()
|
|
pipe := r.client.Pipeline()
|
|
pipe.SAdd(ctx, r.topicOnlineMembersKey(tenantID, topic), uid.UserId())
|
|
pipe.Expire(ctx, r.topicOnlineMembersKey(tenantID, topic), ttl)
|
|
pipe.SAdd(ctx, r.topicNodeOnlineMembersKey(tenantID, topic, nodeID), uid.UserId())
|
|
pipe.Expire(ctx, r.topicNodeOnlineMembersKey(tenantID, topic, nodeID), ttl)
|
|
_, err := pipe.Exec(ctx)
|
|
return err
|
|
}
|
|
|
|
func (r *redisGroupMessageRuntime) MarkTopicMemberOffline(tenantID types.TenantID, topic string, uid types.Uid, nodeID string) error {
|
|
if nodeID == "" {
|
|
nodeID = "local"
|
|
}
|
|
ctx := context.Background()
|
|
pipe := r.client.Pipeline()
|
|
pipe.SRem(ctx, r.topicOnlineMembersKey(tenantID, topic), uid.UserId())
|
|
pipe.SRem(ctx, r.topicNodeOnlineMembersKey(tenantID, topic, nodeID), uid.UserId())
|
|
_, err := pipe.Exec(ctx)
|
|
return err
|
|
}
|
|
|
|
func (r *redisGroupMessageRuntime) GetTopicNodeOnlineMembers(tenantID types.TenantID, topic, nodeID string) ([]types.Uid, error) {
|
|
if nodeID == "" {
|
|
nodeID = "local"
|
|
}
|
|
members, err := r.client.SMembers(context.Background(), r.topicNodeOnlineMembersKey(tenantID, topic, nodeID)).Result()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
uids := make([]types.Uid, 0, len(members))
|
|
for _, member := range members {
|
|
uid := types.ParseUserId(member)
|
|
if !uid.IsZero() {
|
|
uids = append(uids, uid)
|
|
}
|
|
}
|
|
return uids, nil
|
|
}
|
|
|
|
func (r *redisGroupMessageRuntime) PublishGroupFanout(originNodeID string, message any) error {
|
|
if !r.cfg.GroupFanout {
|
|
return nil
|
|
}
|
|
payload, err := json.Marshal(message)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
envelope, err := json.Marshal(groupFanoutEnvelope{
|
|
OriginNodeID: originNodeID,
|
|
Message: payload,
|
|
})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return r.client.Publish(context.Background(), r.groupFanoutChannel(), envelope).Err()
|
|
}
|
|
|
|
func (r *redisGroupMessageRuntime) StartGroupFanoutSubscriber(ctx context.Context, nodeID string, handler func([]byte)) error {
|
|
if !r.cfg.GroupFanout {
|
|
return nil
|
|
}
|
|
pubsub := r.client.Subscribe(ctx, r.groupFanoutChannel())
|
|
if _, err := pubsub.Receive(ctx); err != nil {
|
|
_ = pubsub.Close()
|
|
return err
|
|
}
|
|
go func() {
|
|
defer pubsub.Close()
|
|
ch := pubsub.Channel()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case msg, ok := <-ch:
|
|
if !ok {
|
|
return
|
|
}
|
|
var envelope groupFanoutEnvelope
|
|
if err := json.Unmarshal([]byte(msg.Payload), &envelope); err != nil || envelope.OriginNodeID == nodeID || len(envelope.Message) == 0 {
|
|
continue
|
|
}
|
|
handler(envelope.Message)
|
|
}
|
|
}
|
|
}()
|
|
return nil
|
|
}
|
|
|
|
func (r *redisGroupMessageRuntime) CacheObject(tenantID types.TenantID, namespace, id string, value any) error {
|
|
payload, err := json.Marshal(value)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return r.client.Set(context.Background(), r.objectCacheKey(tenantID, namespace, id), payload, r.cacheTTL(namespace)).Err()
|
|
}
|
|
|
|
func (r *redisGroupMessageRuntime) GetCachedObject(tenantID types.TenantID, namespace, id string, value any) (bool, error) {
|
|
payload, err := r.client.Get(context.Background(), r.objectCacheKey(tenantID, namespace, id)).Bytes()
|
|
if err == redis.Nil {
|
|
return false, nil
|
|
}
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
if err = json.Unmarshal(payload, value); err != nil {
|
|
_ = r.client.Del(context.Background(), r.objectCacheKey(tenantID, namespace, id)).Err()
|
|
return false, err
|
|
}
|
|
return true, nil
|
|
}
|
|
|
|
func (r *redisGroupMessageRuntime) DeleteCachedObject(tenantID types.TenantID, namespace, id string) error {
|
|
return r.client.Del(context.Background(), r.objectCacheKey(tenantID, namespace, id)).Err()
|
|
}
|
|
|
|
func (r *redisGroupMessageRuntime) DeleteCachedObjectsByPrefix(tenantID types.TenantID, namespace, idPrefix string) error {
|
|
ctx := context.Background()
|
|
pattern := r.objectCacheKey(tenantID, namespace, idPrefix+"*")
|
|
var cursor uint64
|
|
for {
|
|
keys, next, err := r.client.Scan(ctx, cursor, pattern, 100).Result()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if len(keys) > 0 {
|
|
if err = r.client.Del(ctx, keys...).Err(); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
cursor = next
|
|
if cursor == 0 {
|
|
return nil
|
|
}
|
|
}
|
|
}
|
|
|
|
func (r *redisGroupMessageRuntime) AddCachedSetMember(tenantID types.TenantID, namespace, id, member string) error {
|
|
key := r.setCacheKey(tenantID, namespace, id)
|
|
ctx := context.Background()
|
|
pipe := r.client.Pipeline()
|
|
pipe.SAdd(ctx, key, member)
|
|
pipe.Expire(ctx, key, r.cacheTTL(namespace))
|
|
_, err := pipe.Exec(ctx)
|
|
return err
|
|
}
|
|
|
|
func (r *redisGroupMessageRuntime) RemoveCachedSetMember(tenantID types.TenantID, namespace, id, member string) error {
|
|
key := r.setCacheKey(tenantID, namespace, id)
|
|
ctx := context.Background()
|
|
pipe := r.client.Pipeline()
|
|
pipe.SRem(ctx, key, member)
|
|
pipe.Expire(ctx, key, r.cacheTTL(namespace))
|
|
_, err := pipe.Exec(ctx)
|
|
return err
|
|
}
|
|
|
|
func (r *redisGroupMessageRuntime) CachedSetContains(tenantID types.TenantID, namespace, id, member string) (bool, error) {
|
|
return r.client.SIsMember(context.Background(), r.setCacheKey(tenantID, namespace, id), member).Result()
|
|
}
|
|
|
|
func (r *redisGroupMessageRuntime) AllowRate(tenantID types.TenantID, scope string, limit int, window time.Duration) (bool, int, time.Duration, error) {
|
|
key := r.rateKey(tenantID, scope)
|
|
ctx := context.Background()
|
|
count, err := r.client.Incr(ctx, key).Result()
|
|
if err != nil {
|
|
return true, 0, 0, err
|
|
}
|
|
if count == 1 {
|
|
if err = r.client.Expire(ctx, key, window).Err(); err != nil {
|
|
return true, int(count), 0, err
|
|
}
|
|
}
|
|
ttl, _ := r.client.TTL(ctx, key).Result()
|
|
return count <= int64(limit), int(count), ttl, nil
|
|
}
|
|
|
|
func (r *redisGroupMessageRuntime) ClaimShortState(namespace, id string, ttl time.Duration) (bool, error) {
|
|
if ttl <= 0 {
|
|
ttl = time.Duration(r.cfg.ShortTTL) * time.Second
|
|
}
|
|
return r.client.SetNX(context.Background(), r.shortStateKey(namespace, id), "1", ttl).Result()
|
|
}
|
|
|
|
func (r *redisGroupMessageRuntime) ClaimShortStateValue(namespace, id, value string, ttl time.Duration) (bool, bool, error) {
|
|
if ttl <= 0 {
|
|
ttl = time.Duration(r.cfg.ShortTTL) * time.Second
|
|
}
|
|
key := r.shortStateKey(namespace, id)
|
|
ctx := context.Background()
|
|
claimed, err := r.client.SetNX(ctx, key, value, ttl).Result()
|
|
if err != nil || claimed {
|
|
return claimed, false, err
|
|
}
|
|
existing, err := r.client.Get(ctx, key).Result()
|
|
if err != nil {
|
|
return false, false, err
|
|
}
|
|
return false, existing == value, nil
|
|
}
|
|
|
|
func (r *redisGroupMessageRuntime) ShortStateExists(namespace, id string) (bool, error) {
|
|
count, err := r.client.Exists(context.Background(), r.shortStateKey(namespace, id)).Result()
|
|
return count > 0, err
|
|
}
|
|
|
|
func (r *redisGroupMessageRuntime) hotSeqIDs(tenantID types.TenantID, topic string, opts *types.QueryOpt, limit int) ([]int64, bool, error) {
|
|
if opts != nil && len(opts.IdRanges) > 0 {
|
|
seqs := expandRanges(opts.IdRanges, limit)
|
|
sort.Slice(seqs, func(i, j int) bool { return seqs[i] > seqs[j] })
|
|
return seqs, len(seqs) > 0 && len(seqs) <= limit, nil
|
|
}
|
|
min := "-inf"
|
|
max := "+inf"
|
|
if opts != nil {
|
|
if opts.Since > 0 {
|
|
min = strconv.FormatInt(opts.Since, 10)
|
|
}
|
|
if opts.Before > 1 {
|
|
max = strconv.FormatInt(opts.Before-1, 10)
|
|
}
|
|
}
|
|
raw, err := r.client.ZRevRangeByScore(context.Background(), r.hotSeqKey(tenantID, topic), &redis.ZRangeBy{
|
|
Min: min,
|
|
Max: max,
|
|
Offset: 0,
|
|
Count: int64(limit),
|
|
}).Result()
|
|
if err != nil {
|
|
return nil, false, err
|
|
}
|
|
if len(raw) < limit {
|
|
return nil, false, nil
|
|
}
|
|
seqs := make([]int64, 0, len(raw))
|
|
for _, value := range raw {
|
|
seq, err := strconv.ParseInt(value, 10, 64)
|
|
if err != nil {
|
|
return nil, false, err
|
|
}
|
|
seqs = append(seqs, seq)
|
|
}
|
|
return seqs, true, nil
|
|
}
|
|
|
|
func expandRanges(ranges []types.Range, limit int) []int64 {
|
|
if limit <= 0 {
|
|
return nil
|
|
}
|
|
seqs := make([]int64, 0)
|
|
for _, rng := range ranges {
|
|
if rng.Low <= 0 {
|
|
continue
|
|
}
|
|
if rng.Hi == 0 {
|
|
seqs = append(seqs, rng.Low)
|
|
} else {
|
|
for seq := rng.Low; seq < rng.Hi; seq++ {
|
|
seqs = append(seqs, seq)
|
|
if len(seqs) >= limit {
|
|
return seqs
|
|
}
|
|
}
|
|
}
|
|
if len(seqs) >= limit {
|
|
return seqs
|
|
}
|
|
}
|
|
return seqs
|
|
}
|
|
|
|
func (r *redisGroupMessageRuntime) seqKey(tenantID types.TenantID, topic string) string {
|
|
return r.key(tenantID, "topic", topic, "last_seq")
|
|
}
|
|
|
|
func (r *redisGroupMessageRuntime) uniqueKey(tenantID types.TenantID, topic string, from types.Uid, clientMsgID string) string {
|
|
return r.key(tenantID, "topic", topic, "uniq", "msg", from.UserId(), clientMsgID)
|
|
}
|
|
|
|
func (r *redisGroupMessageRuntime) hotSeqKey(tenantID types.TenantID, topic string) string {
|
|
return r.key(tenantID, "topic", topic, "hot_msg_seqs")
|
|
}
|
|
|
|
func (r *redisGroupMessageRuntime) hotMsgKey(tenantID types.TenantID, topic string, seq any) string {
|
|
return r.key(tenantID, "topic", topic, "hot_msg", fmt.Sprint(seq))
|
|
}
|
|
|
|
func (r *redisGroupMessageRuntime) presenceUserKey(tenantID types.TenantID, uid types.Uid) string {
|
|
return r.key(tenantID, "presence", "user", uid.UserId())
|
|
}
|
|
|
|
func (r *redisGroupMessageRuntime) presenceSessionsKey(tenantID types.TenantID, uid types.Uid) string {
|
|
return r.key(tenantID, "presence", "user", uid.UserId(), "sessions")
|
|
}
|
|
|
|
func (r *redisGroupMessageRuntime) presenceSessionKey(tenantID types.TenantID, uid types.Uid, sid string) string {
|
|
return r.key(tenantID, "presence", "user", uid.UserId(), "session", sid)
|
|
}
|
|
|
|
func (r *redisGroupMessageRuntime) presenceNodeUsersKey(tenantID types.TenantID, nodeID string) string {
|
|
return r.key(tenantID, "presence", "node", nodeID, "users")
|
|
}
|
|
|
|
func (r *redisGroupMessageRuntime) topicOnlineMembersKey(tenantID types.TenantID, topic string) string {
|
|
return r.key(tenantID, "topic", topic, "online_members")
|
|
}
|
|
|
|
func (r *redisGroupMessageRuntime) topicNodeOnlineMembersKey(tenantID types.TenantID, topic, nodeID string) string {
|
|
return r.key(tenantID, "topic", topic, "node", nodeID, "online_members")
|
|
}
|
|
|
|
func (r *redisGroupMessageRuntime) groupFanoutChannel() string {
|
|
return r.key(types.ZeroTenantID, "group_fanout")
|
|
}
|
|
|
|
func (r *redisGroupMessageRuntime) objectCacheKey(tenantID types.TenantID, namespace, id string) string {
|
|
return r.key(tenantID, "cache", namespace, id)
|
|
}
|
|
|
|
func (r *redisGroupMessageRuntime) setCacheKey(tenantID types.TenantID, namespace, id string) string {
|
|
return r.key(tenantID, "set", namespace, id)
|
|
}
|
|
|
|
func (r *redisGroupMessageRuntime) rateKey(tenantID types.TenantID, scope string) string {
|
|
return r.key(tenantID, "rate", scope)
|
|
}
|
|
|
|
func (r *redisGroupMessageRuntime) shortStateKey(namespace, id string) string {
|
|
return r.key(types.ZeroTenantID, "short", namespace, id)
|
|
}
|
|
|
|
func (r *redisGroupMessageRuntime) cacheTTL(namespace string) time.Duration {
|
|
ttl := r.cfg.CacheTTL
|
|
if namespace == "verification" || namespace == "action_token" {
|
|
ttl = r.cfg.ShortTTL
|
|
}
|
|
return time.Duration(ttl) * time.Second
|
|
}
|
|
|
|
func (r *redisGroupMessageRuntime) key(tenantID types.TenantID, parts ...string) string {
|
|
all := []string{strings.TrimSuffix(r.prefix, ":"), strconv.FormatInt(int64(tenantID), 10)}
|
|
all = append(all, parts...)
|
|
return strings.Join(all, ":")
|
|
}
|