im_server/server/store/group_message_runtime.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"`

	// Sentinel-aware connection. When set, the runtime uses a failover client
	// instead of the single Addr. Required whenever Redis runs behind Sentinel:
	// a plain client pinned to one address will hit replicas after failover,
	// where writes fail with READONLY and reads return stale values (a stale
	// seq counter can hand out already-persisted seq values).
	SentinelMasterName string   `json:"sentinel_master_name"`
	SentinelAddrs      []string `json:"sentinel_addrs"`
}

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
}

// GroupMessageShardRoute describes the Seq range stored in one monthly table.
type GroupMessageShardRoute struct {
	Month  string `db:"month"`
	MinSeq int64  `db:"min_seq"`
	MaxSeq int64  `db:"max_seq"`
	Count  int64  `db:"count"`
}

type groupFanoutEnvelope struct {
	OriginNodeID string          `json:"origin_node_id"`
	Message      json.RawMessage `json:"message"`
}

type RealtimeFanoutEnvelope struct {
	OriginNodeID string          `json:"origin_node_id"`
	TenantID     types.TenantID  `json:"tenant_id"`
	Kind         string          `json:"kind"`
	TargetType   string          `json:"target_type"`
	Target       string          `json:"target"`
	SkipSid      string          `json:"skip_sid,omitempty"`
	DedupKey     string          `json:"dedup_key,omitempty"`
	Message      json.RawMessage `json:"message"`
}

type LoginSession struct {
	UserID     int64  `json:"user_id"`
	TenantID   int64  `json:"tenant_id"`
	TenantCode string `json:"tenant_code"`
	DeviceID   string `json:"device_id,omitempty"`
	AuthLevel  string `json:"auth_level"`
	State      string `json:"state"`
	ExpiresAt  string `json:"expires_at"`
	IssuedAt   string `json:"issued_at"`
}

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, candidateMonth string) (int64, string, error)
	PeekSeq(tenantID types.TenantID, topic string) (int64, error)
	BumpSeqFloor(tenantID types.TenantID, topic string, seq int64) error
	MarkPending(tenantID types.TenantID, topic string, from types.Uid, clientMsgID string, seq int64) error
	PutShardRoute(tenantID types.TenantID, topic string, route GroupMessageShardRoute) error
	GetShardRoutes(tenantID types.TenantID, topic string) ([]GroupMessageShardRoute, bool, error)
	DeleteShardRoutes(tenantID types.TenantID, topic string) 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
	PublishRealtimeFanout(originNodeID string, tenantID types.TenantID, kind, targetType, target, skipSid, dedupKey string, message any) error
	StartRealtimeFanoutSubscriber(ctx context.Context, nodeID string, handler func(*RealtimeFanoutEnvelope)) 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)
	StoreLoginSession(token []byte, session LoginSession, ttl time.Duration) 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, candidateMonth string) (int64, string, bool, error) {
	if groupMsgRuntime == nil || !tenantID.IsValid() || topic == "" {
		return 0, candidateMonth, false, nil
	}
	seq, month, err := nextMessageSeqWithFloor(tenantID, topic, baseline, candidateMonth)
	return seq, month, true, err
}

func PutGroupMessageShardRoute(tenantID types.TenantID, topic string, route GroupMessageShardRoute) error {
	if groupMsgRuntime == nil || !tenantID.IsValid() || topic == "" || route.Month == "" {
		return nil
	}
	return groupMsgRuntime.PutShardRoute(tenantID, topic, route)
}

func GetGroupMessageShardRoutes(tenantID types.TenantID, topic string) ([]GroupMessageShardRoute, bool, error) {
	if groupMsgRuntime == nil || !tenantID.IsValid() || topic == "" {
		return nil, false, nil
	}
	return groupMsgRuntime.GetShardRoutes(tenantID, topic)
}

func DeleteGroupMessageShardRoutes(tenantID types.TenantID, topic string) error {
	if groupMsgRuntime == nil || !tenantID.IsValid() || topic == "" {
		return nil
	}
	return groupMsgRuntime.DeleteShardRoutes(tenantID, topic)
}

// TopicLastSeqID returns the seq last persisted for the topic in the DB, or 0
// when the topic is unknown. It is used as a floor when re-seeding the Redis
// seq counter.
func TopicLastSeqID(tenantID types.TenantID, topic string) int64 {
	if Topics == nil || !tenantID.IsValid() || topic == "" {
		return 0
	}
	stopic, err := Topics.Get(tenantID, topic)
	if err != nil || stopic == nil {
		return 0
	}
	return stopic.SeqId
}

// nextMessageSeqWithFloor allocates the next seq for the topic. When the Redis
// counter key is missing (Redis restart, eviction or flush) it re-seeds from
// the DB topic seq instead of trusting the caller's in-memory baseline, which
// may be stale and would otherwise hand out already-persisted seq values.
func nextMessageSeqWithFloor(tenantID types.TenantID, topic string, baseline int64, candidateMonth string) (int64, string, error) {
	if baseline < 0 {
		baseline = 0
	}
	if current, err := groupMsgRuntime.PeekSeq(tenantID, topic); err == nil && current <= 0 {
		if floor := TopicLastSeqID(tenantID, topic); floor > baseline {
			baseline = floor
		}
	}
	return groupMsgRuntime.NextSeq(tenantID, topic, baseline, candidateMonth)
}

// BumpMessageSeqFloor pushes the Redis seq counter forward to seq (never
// backward). Used after the async persister re-allocates a seq following a
// uniqueness conflict so subsequent allocations do not collide again.
func BumpMessageSeqFloor(tenantID types.TenantID, topic string, seq int64) error {
	if groupMsgRuntime == nil || !tenantID.IsValid() || topic == "" || seq <= 0 {
		return nil
	}
	return groupMsgRuntime.BumpSeqFloor(tenantID, topic, seq)
}

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 StoreLoginSession(token []byte, session LoginSession, ttl time.Duration) (bool, error) {
	if groupMsgRuntime == nil || len(token) == 0 || ttl <= 0 {
		return false, nil
	}
	return true, groupMsgRuntime.StoreLoginSession(token, session, ttl)
}

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 PublishRealtimeFanout(originNodeID string, tenantID types.TenantID, kind, targetType, target, skipSid, dedupKey string, message any) error {
	if groupMsgRuntime == nil || originNodeID == "" || !tenantID.IsValid() || targetType == "" || target == "" || message == nil {
		return nil
	}
	return groupMsgRuntime.PublishRealtimeFanout(originNodeID, tenantID, kind, targetType, target, skipSid, dedupKey, message)
}

func StartRealtimeFanoutSubscriber(ctx context.Context, nodeID string, handler func(*RealtimeFanoutEnvelope)) error {
	if groupMsgRuntime == nil || nodeID == "" || handler == nil {
		return nil
	}
	return groupMsgRuntime.StartRealtimeFanoutSubscriber(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)

	var client redis.UniversalClient
	if cfg.SentinelMasterName != "" && len(cfg.SentinelAddrs) > 0 {
		client = redis.NewFailoverClient(&redis.FailoverOptions{
			MasterName:       cfg.SentinelMasterName,
			SentinelAddrs:    cfg.SentinelAddrs,
			SentinelPassword: cfg.Password,
			Username:         cfg.Username,
			Password:         cfg.Password,
			DB:               cfg.DB,
			DialTimeout:      dialTimeout,
			ReadTimeout:      readTimeout,
			WriteTimeout:     writeTimeout,
		})
	} else {
		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.UniversalClient
	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, candidateMonth string) (int64, string, error) {
	if baseline < 0 {
		baseline = 0
	}
	redisNow, err := r.client.Time(context.Background()).Result()
	if err != nil {
		return 0, "", err
	}
	candidateMonth = redisNow.UTC().Format("200601")
	if !validShardMonth(candidateMonth) {
		return 0, "", types.ErrMalformed
	}
	seqKey := r.seqKey(tenantID, topic)
	result, 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

local active_month = redis.call("GET", KEYS[2])
if not active_month or active_month < ARGV[2] then
  redis.call("SET", KEYS[2], ARGV[2])
  active_month = ARGV[2]
end

local seq = redis.call("INCR", KEYS[1])
return {seq, active_month}
`, []string{seqKey, r.activeMonthKey(tenantID)}, baseline, candidateMonth).Slice()
	if err != nil {
		return 0, "", err
	}
	if len(result) != 2 {
		return 0, "", fmt.Errorf("redis group message sequence returned %d values", len(result))
	}
	seq, err := strconv.ParseInt(fmt.Sprint(result[0]), 10, 64)
	if err != nil {
		return 0, "", err
	}
	month := fmt.Sprint(result[1])
	if !validShardMonth(month) {
		return 0, "", fmt.Errorf("redis returned invalid active month %q", month)
	}
	return seq, month, nil
}

func (r *redisGroupMessageRuntime) PutShardRoute(tenantID types.TenantID, topic string, route GroupMessageShardRoute) error {
	if !validShardMonth(route.Month) || route.MinSeq <= 0 || route.MaxSeq < route.MinSeq {
		return types.ErrMalformed
	}
	_, err := r.client.Eval(context.Background(), `
local min_seq = tonumber(ARGV[2])
local max_seq = tonumber(ARGV[3])
local count = tonumber(ARGV[4])
local meta = redis.call("HGET", KEYS[2], ARGV[1])
if meta then
  local old_min, old_max, old_count = string.match(meta, "^(%d+):(%d+):(%d+)$")
  if old_min then
    min_seq = math.min(min_seq, tonumber(old_min))
    max_seq = math.max(max_seq, tonumber(old_max))
    if count == 1 then
      count = tonumber(old_count) + 1
    else
      count = math.max(count, tonumber(old_count))
    end
  end
end
redis.call("ZADD", KEYS[1], min_seq, ARGV[1])
redis.call("HSET", KEYS[2], ARGV[1], min_seq .. ":" .. max_seq .. ":" .. count)
return 1
`, []string{r.shardRouteKey(tenantID, topic), r.shardRouteMetaKey(tenantID, topic)},
		route.Month, route.MinSeq, route.MaxSeq, route.Count).Result()
	return err
}

func (r *redisGroupMessageRuntime) GetShardRoutes(tenantID types.TenantID, topic string) ([]GroupMessageShardRoute, bool, error) {
	ctx := context.Background()
	months, err := r.client.ZRevRange(ctx, r.shardRouteKey(tenantID, topic), 0, -1).Result()
	if err != nil {
		return nil, false, err
	}
	if len(months) == 0 {
		return nil, false, nil
	}
	values, err := r.client.HMGet(ctx, r.shardRouteMetaKey(tenantID, topic), months...).Result()
	if err != nil {
		return nil, false, err
	}
	routes := make([]GroupMessageShardRoute, 0, len(months))
	for i, month := range months {
		if values[i] == nil {
			continue
		}
		parts := strings.Split(fmt.Sprint(values[i]), ":")
		if len(parts) != 3 {
			continue
		}
		minSeq, minErr := strconv.ParseInt(parts[0], 10, 64)
		maxSeq, maxErr := strconv.ParseInt(parts[1], 10, 64)
		count, countErr := strconv.ParseInt(parts[2], 10, 64)
		if minErr == nil && maxErr == nil && countErr == nil && validShardMonth(month) {
			routes = append(routes, GroupMessageShardRoute{Month: month, MinSeq: minSeq, MaxSeq: maxSeq, Count: count})
		}
	}
	return routes, len(routes) > 0, nil
}

func (r *redisGroupMessageRuntime) DeleteShardRoutes(tenantID types.TenantID, topic string) error {
	return r.client.Del(context.Background(), r.shardRouteKey(tenantID, topic), r.shardRouteMetaKey(tenantID, topic)).Err()
}

func validShardMonth(month string) bool {
	if len(month) != 6 {
		return false
	}
	for i := range month {
		if month[i] < '0' || month[i] > '9' {
			return false
		}
	}
	parsed, err := time.Parse("200601", month)
	return err == nil && parsed.Format("200601") == month
}

func (r *redisGroupMessageRuntime) PeekSeq(tenantID types.TenantID, topic string) (int64, error) {
	value, err := r.client.Get(context.Background(), r.seqKey(tenantID, topic)).Result()
	if err == redis.Nil {
		return 0, nil
	}
	if err != nil {
		return 0, err
	}
	seq, err := strconv.ParseInt(value, 10, 64)
	if err != nil {
		return 0, err
	}
	return seq, nil
}

func (r *redisGroupMessageRuntime) BumpSeqFloor(tenantID types.TenantID, topic string, seq int64) error {
	if seq <= 0 {
		return nil
	}
	_, err := r.client.Eval(context.Background(), `
local current = redis.call("GET", KEYS[1])
if not current or tonumber(current) < tonumber(ARGV[1]) then
  redis.call("SET", KEYS[1], ARGV[1])
end
return redis.call("GET", KEYS[1])
`, []string{r.seqKey(tenantID, topic)}, seq).Int64()
	return err
}

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 {
	return r.PublishRealtimeFanout(originNodeID, types.ZeroTenantID, "message", "topic", "", "", "", message)
}

func (r *redisGroupMessageRuntime) PublishRealtimeFanout(originNodeID string, tenantID types.TenantID, kind, targetType, target, skipSid, dedupKey string, message any) error {
	if !r.cfg.GroupFanout {
		return nil
	}
	if originNodeID == "" || message == nil {
		return nil
	}
	payload, err := json.Marshal(message)
	if err != nil {
		return err
	}
	envelope, err := json.Marshal(RealtimeFanoutEnvelope{
		OriginNodeID: originNodeID,
		TenantID:     tenantID,
		Kind:         kind,
		TargetType:   targetType,
		Target:       target,
		SkipSid:      skipSid,
		DedupKey:     dedupKey,
		Message:      payload,
	})
	if err != nil {
		return err
	}
	return r.client.Publish(context.Background(), r.realtimeFanoutChannel(tenantID), envelope).Err()
}

func (r *redisGroupMessageRuntime) StartGroupFanoutSubscriber(ctx context.Context, nodeID string, handler func([]byte)) error {
	return r.StartRealtimeFanoutSubscriber(ctx, nodeID, func(envelope *RealtimeFanoutEnvelope) {
		if envelope != nil {
			handler(envelope.Message)
		}
	})
}

func (r *redisGroupMessageRuntime) StartRealtimeFanoutSubscriber(ctx context.Context, nodeID string, handler func(*RealtimeFanoutEnvelope)) error {
	if !r.cfg.GroupFanout {
		return nil
	}
	pubsub := r.client.PSubscribe(ctx, r.realtimeFanoutPattern())
	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 RealtimeFanoutEnvelope
				if err := json.Unmarshal([]byte(msg.Payload), &envelope); err != nil || envelope.OriginNodeID == nodeID || len(envelope.Message) == 0 {
					continue
				}
				if envelope.TenantID.IsValid() && envelope.DedupKey != "" {
					id := realtimeFanoutDedupClaimID(envelope.TenantID, nodeID, envelope.DedupKey)
					claimed, err := r.ClaimShortState("rt_dedup", id, 5*time.Minute)
					if err != nil || !claimed {
						continue
					}
				}
				if envelope.TargetType == "" {
					envelope.TargetType = "topic"
				}
				if envelope.Kind == "" {
					envelope.Kind = "message"
				}
				if !envelope.TenantID.IsValid() && envelope.Target == "" {
					continue
				}
				handler(&envelope)
			}
		}
	}()
	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) StoreLoginSession(token []byte, session LoginSession, ttl time.Duration) error {
	if len(token) == 0 || ttl <= 0 {
		return nil
	}
	digest := sha256.Sum256(token)
	payload, err := json.Marshal(session)
	if err != nil {
		return err
	}
	return r.client.Set(context.Background(), r.loginSessionKey(hex.EncodeToString(digest[:])), payload, ttl).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.shardKey(tenantID, "topic", topic, "last_seq")
}

func (r *redisGroupMessageRuntime) activeMonthKey(tenantID types.TenantID) string {
	return r.shardKey(tenantID, "groupmsg", "active_month")
}

func (r *redisGroupMessageRuntime) shardRouteKey(tenantID types.TenantID, topic string) string {
	return r.shardKey(tenantID, "groupmsg", "route", topic)
}

func (r *redisGroupMessageRuntime) shardRouteMetaKey(tenantID types.TenantID, topic string) string {
	return r.shardKey(tenantID, "groupmsg", "route_meta", topic)
}

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) realtimeFanoutChannel(tenantID types.TenantID) string {
	if tenantID.IsValid() {
		return r.key(tenantID, "rt", "fanout")
	}
	return r.groupFanoutChannel()
}

func (r *redisGroupMessageRuntime) realtimeFanoutPattern() string {
	return strings.TrimSuffix(r.prefix, ":") + ":*:rt:fanout"
}

func realtimeFanoutDedupClaimID(tenantID types.TenantID, nodeID, dedupKey string) string {
	if nodeID == "" {
		nodeID = "local"
	}
	return fmt.Sprintf("%d:%s:%s", tenantID, nodeID, dedupKey)
}

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) loginSessionKey(tokenHash string) string {
	return strings.TrimSuffix(r.prefix, ":") + ":session:token:" + tokenHash
}

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, ":")
}

func (r *redisGroupMessageRuntime) shardKey(tenantID types.TenantID, parts ...string) string {
	prefix := strings.TrimSuffix(r.prefix, ":")
	slot := "{" + strconv.FormatInt(int64(tenantID), 10) + "}"
	all := []string{prefix, slot}
	all = append(all, parts...)
	return strings.Join(all, ":")
}