lislgosms/gateway/internal/inbound/presence.go

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
}