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 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) (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
	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) (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 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)
	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(), `
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 {
	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.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) 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, ":")
}