feat: add redis cache and fanout support

server/store/group_message_runtime.go

@@ -0,0 +1,866 @@
+package store
+
+import (
+	"context"
+	"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 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)
+	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
+	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
+	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)
+}
+
+var groupMsgRuntime groupMessageRuntime
+
+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 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 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 = 120
+	}
+	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, &current); 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 &current, 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(), `
+if redis.call("EXISTS", KEYS[1]) == 0 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) 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 {
+	if msg == nil || msg.Topic == "" || msg.SeqId <= 0 {
+		return types.ErrMalformed
+	}
+	payload, err := json.Marshal(msg)
+	if err != nil {
+		return err
+	}
+	ctx := context.Background()
+	ttl := time.Duration(r.cfg.HotMsgTTL) * time.Second
+	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) 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) 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) 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, ":")
+}