dy_server/service/main/redis/redis.go

package redis

import (
	"app/service/main/message/pb"
	"context"
	"encoding/json"
	"github.com/redis/go-redis/v9"
	"os"
)

var client *redis.Client

func init() {
	url := os.Getenv("REDIS_URL")
	opts, err := redis.ParseURL(url)
	if err != nil {
		panic(err)
	}
	client = redis.NewClient(opts)
}

func UpdateReport(appId string, scoreMap map[string]int32) error {
	ctx := context.Background()
	pip := client.Pipeline()
	for openId, score := range scoreMap {
		key := ThisWeekScoreKey(appId)
		pip.ZIncrBy(ctx, key, float64(score), openId)
	}
	_, err := pip.Exec(ctx)
	return err
}

func GetRank(appId string, topCount int32) ([]*pb.Audience, error) {
	ctx := context.Background()
	key := ThisWeekScoreKey(appId)
	cmd := client.ZRevRangeWithScores(ctx, key, 0, int64(topCount))
	result, err := cmd.Result()
	if err != nil {
		return nil, err
	}
	pip := client.Pipeline()
	for _, info := range result {
		key := ThisWeekScoreKey(appId)

		ret = append(ret, &RankInfo{
			OpenId: info.Member.(string),
			Score:  int32(info.Score),
		})
	}
	return rankList, nil
}

func SetAudienceBasic(appId string, data *pb.AudienceBasic) {
	ctx := context.Background()
	key := UserDataKey(appId, data.OpenId)
	exist, err := client.Exists(ctx, key).Result()
	if err != nil {
		return
	}
	if exist == 1 {
		return
	}

	jsonData, err := json.Marshal(data)
	if err != nil {
		return
	}
	client.Set(ctx, key, string(jsonData), -1)
}

func GetAudience(appId string, openId string) *pb.Audience {
	return &pb.Audience{
		AudienceBasic: GetAudienceBasic(appId, openId),
		AudienceInfo:  GetAudienceInfo(appId, openId),
	}
}

func GetAudienceBasic(appId string, openId string) *pb.AudienceBasic {
	ctx := context.Background()
	key := UserDataKey(appId, openId)
	result, err := client.Get(ctx, key).Result()
	if err != nil {
		return nil
	}
	data := &pb.AudienceBasic{}
	err = json.Unmarshal([]byte(result), data)
	if err != nil {
		return nil
	}
	return data
}

func GetAudienceInfo(appId string, openId string) *pb.AudienceInfo {
	ctx := context.Background()
	ret := &pb.AudienceInfo{
		OpenId: openId,
	}
	pip := client.Pipeline()
	keyScore := ThisWeekScoreKey(appId)
	keyScoreLast := LastWeekScoreKey(appId)

	var err error
	cmdScore := pip.ZScore(ctx, keyScore, openId)
	cmdRank := pip.ZRank(ctx, keyScore, openId)
	cmdRankLast := pip.ZRank(ctx, keyScoreLast, openId)
	result, err := pip.Exec(ctx)
	if err != nil {
		return ret
	}
	if len(result) != 3 {
		return ret
	}
	if result, err := cmdScore.Result(); err == nil {
		ret.Score = int32(result)
	}
	if result, err := cmdRank.Result(); err == nil {
		ret.Rank = int32(result)
	}
	if result, err := cmdRankLast.Result(); err == nil {
		ret.Score = int32(result)
	}
	return ret
}

func GetAudienceBasicList(appId string, openIdList []string) (ret []*pb.AudienceBasic) {
	ctx := context.Background()
	pip := client.Pipeline()
	var cmdList []*redis.StringCmd
	for _, openId := range openIdList {
		key := UserDataKey(appId, openId)
		cmdList = append(cmdList, pip.Get(ctx, key))
	}
	_, err := pip.Exec(ctx)
	if err != nil {
		return nil
	}
	for _, cmd := range cmdList {
		data := &pb.AudienceBasic{}
		if result, err := cmd.Result(); err == nil {
			_ = json.Unmarshal([]byte(result), data)
		}
		ret = append(ret, data)
	}
	return
}

func GetAudienceInfoList(appId string, openIdList []string) (ret []*pb.AudienceInfo) {
	ctx := context.Background()

}

func Subscribe(appId string, roomId string, pushFunc func(string), pushCloseChan chan struct{}) {
	ps := client.Subscribe(context.Background(), PublishKey(appId, roomId))
	go func() {
		for {
			msg, err := ps.ReceiveMessage(context.Background())
			if err != nil {
				return
			}
			pushFunc(msg.Payload)
		}
	}()
	go func() {
		<-pushCloseChan
		_ = ps.Unsubscribe(context.Background(), PublishKey(appId, roomId))
	}()
}

func Publish(appId string, roomId string, data *pb.NotifyAudienceAction) {
	jsonData, err := json.Marshal(data)
	if err != nil {
		return
	}
	client.Publish(context.Background(), PublishKey(appId, roomId), string(jsonData))
}