Austin
753a04ed12
Logout left contacts seeing the user online until the me topic idle timeout fired presUsersOfInterest(off), and even that frame was silently dropped for p2p routes by handlePresence's xoriginal check. MarkUserOffline now returns the remaining session count and cleanUp notifies users of interest immediately once it hits zero; p2p presence is delivered straight to the contact's me topic, mirroring presSingleUserOffline. Zentao #22.
1472 lines
50 KiB
Go
1472 lines
50 KiB
Go
package store
|
|
|
|
import (
|
|
"context"
|
|
"crypto/sha256"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"fmt"
|
|
"reflect"
|
|
"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"`
|
|
SkipTopic string `json:"skip_topic,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) (int, 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) (int, error) {
|
|
if groupMsgRuntime == nil || !tenantID.IsValid() || uid.IsZero() || sid == "" {
|
|
return 0, 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, ¤t); err != nil {
|
|
return nil, false, err
|
|
}
|
|
if current.Status == "failed" {
|
|
_ = r.client.Del(ctx, key).Err()
|
|
ok, err = r.client.SetNX(ctx, key, raw, time.Duration(r.cfg.PendingTTL)*time.Second).Result()
|
|
if err != nil {
|
|
return nil, false, err
|
|
}
|
|
if ok {
|
|
return rec, true, nil
|
|
}
|
|
}
|
|
return ¤t, false, nil
|
|
}
|
|
|
|
func (r *redisGroupMessageRuntime) NextSeq(tenantID types.TenantID, topic string, baseline int64, 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) (int, 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 0, err
|
|
}
|
|
count, err := r.client.SCard(ctx, sessionsKey).Result()
|
|
if err != nil {
|
|
return 0, 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 int(count), 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 0, 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,
|
|
SkipTopic: realtimeFanoutSkipTopic(message),
|
|
DedupKey: dedupKey,
|
|
Message: payload,
|
|
})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return r.client.Publish(context.Background(), r.realtimeFanoutChannel(tenantID), envelope).Err()
|
|
}
|
|
|
|
func realtimeFanoutSkipTopic(message any) string {
|
|
if message == nil {
|
|
return ""
|
|
}
|
|
v := reflect.ValueOf(message)
|
|
if v.Kind() == reflect.Pointer {
|
|
if v.IsNil() {
|
|
return ""
|
|
}
|
|
v = v.Elem()
|
|
}
|
|
if v.Kind() != reflect.Struct {
|
|
return ""
|
|
}
|
|
for _, fieldName := range []string{"Pres", "Info"} {
|
|
field := v.FieldByName(fieldName)
|
|
if !field.IsValid() || field.IsNil() {
|
|
continue
|
|
}
|
|
if field.Kind() == reflect.Pointer {
|
|
field = field.Elem()
|
|
}
|
|
if field.Kind() != reflect.Struct {
|
|
continue
|
|
}
|
|
skip := field.FieldByName("SkipTopic")
|
|
if skip.IsValid() && skip.Kind() == reflect.String {
|
|
return strings.TrimSpace(skip.String())
|
|
}
|
|
}
|
|
return ""
|
|
}
|
|
|
|
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, ":")
|
|
}
|