package inbound import ( "context" "encoding/json" "fmt" "sort" "strings" "time" "github.com/redis/go-redis/v9" ) const defaultPresenceTTL = 3 * time.Minute type PresenceStore interface { TouchAccount(ctx context.Context, snapshot DownstreamPresence) error RemoveAccount(ctx context.Context, account string) error ListAccounts(ctx context.Context) ([]DownstreamPresence, error) } type DownstreamPresence struct { Account string `json:"account"` SrcID string `json:"srcId,omitempty"` RemoteIP string `json:"remoteIp,omitempty"` GatewayInstanceID string `json:"gatewayInstanceId,omitempty"` State string `json:"state"` ConnectedAt time.Time `json:"connectedAt"` UpdatedAt time.Time `json:"updatedAt"` LastSubmitAt time.Time `json:"lastSubmitAt,omitempty"` LastDeliverAt time.Time `json:"lastDeliverAt,omitempty"` } type RedisPresenceStore struct { Client *redis.Client TTL time.Duration Prefix string } func NewRedisPresenceStore(redisURL string) (*RedisPresenceStore, error) { if redisURL == "" { redisURL = "redis://127.0.0.1:6379" } options, err := redis.ParseURL(redisURL) if err != nil { return nil, err } return &RedisPresenceStore{ Client: redis.NewClient(options), TTL: defaultPresenceTTL, Prefix: "gateway:downstream:presence", }, nil } func (s *RedisPresenceStore) TouchAccount(ctx context.Context, snapshot DownstreamPresence) error { if s == nil || s.Client == nil || snapshot.Account == "" { return nil } if snapshot.UpdatedAt.IsZero() { snapshot.UpdatedAt = time.Now().UTC() } if snapshot.ConnectedAt.IsZero() { snapshot.ConnectedAt = snapshot.UpdatedAt } payload, err := json.Marshal(snapshot) if err != nil { return err } ttl := s.ttl() pipe := s.Client.TxPipeline() pipe.Set(ctx, s.accountKey(snapshot.Account), payload, ttl) pipe.SAdd(ctx, s.accountsKey(), snapshot.Account) pipe.Expire(ctx, s.accountsKey(), ttl*4) _, err = pipe.Exec(ctx) return err } func (s *RedisPresenceStore) RemoveAccount(ctx context.Context, account string) error { if s == nil || s.Client == nil || account == "" { return nil } pipe := s.Client.TxPipeline() pipe.Del(ctx, s.accountKey(account)) pipe.SRem(ctx, s.accountsKey(), account) _, err := pipe.Exec(ctx) return err } func (s *RedisPresenceStore) ListAccounts(ctx context.Context) ([]DownstreamPresence, error) { if s == nil || s.Client == nil { return nil, nil } accounts, err := s.Client.SMembers(ctx, s.accountsKey()).Result() if err != nil { return nil, err } result := make([]DownstreamPresence, 0, len(accounts)) for _, account := range accounts { payload, getErr := s.Client.Get(ctx, s.accountKey(account)).Bytes() if getErr == redis.Nil { _ = s.Client.SRem(ctx, s.accountsKey(), account).Err() continue } if getErr != nil { return nil, getErr } var snapshot DownstreamPresence if err := json.Unmarshal(payload, &snapshot); err != nil { return nil, fmt.Errorf("decode presence %s: %w", account, err) } result = append(result, snapshot) } return result, nil } func (s *RedisPresenceStore) ttl() time.Duration { if s != nil && s.TTL > 0 { return s.TTL } return defaultPresenceTTL } func (s *RedisPresenceStore) accountKey(account string) string { return fmt.Sprintf("%s:account:%s", s.prefix(), account) } func (s *RedisPresenceStore) accountsKey() string { return fmt.Sprintf("%s:accounts", s.prefix()) } func (s *RedisPresenceStore) prefix() string { if s != nil && s.Prefix != "" { return s.Prefix } return "gateway:downstream:presence" } func ListRecoveryCandidates(ctx context.Context, store PresenceStore) ([]DownstreamPresence, error) { candidateMap := map[string]DownstreamPresence{} if store != nil { snapshots, err := store.ListAccounts(ctx) if err != nil { return nil, err } for _, snapshot := range snapshots { account := strings.TrimSpace(snapshot.Account) if account == "" { continue } candidateMap[account] = snapshot } } for _, account := range onlineAccounts() { account = strings.TrimSpace(account) if account == "" { continue } current := candidateMap[account] current.Account = account if strings.TrimSpace(current.State) == "" { current.State = "connected" } if current.UpdatedAt.IsZero() { current.UpdatedAt = time.Now().UTC() } candidateMap[account] = current } result := make([]DownstreamPresence, 0, len(candidateMap)) for _, snapshot := range candidateMap { result = append(result, snapshot) } sort.Slice(result, func(i, j int) bool { if result[i].UpdatedAt.Equal(result[j].UpdatedAt) { return result[i].Account < result[j].Account } return result[i].UpdatedAt.After(result[j].UpdatedAt) }) return result, nil }