package websocket

import (
	"fmt"
	"gitee.com/zhonglin-liu/dao_socket/pkg/utils"
	"github.com/go-redis/redis"
	"time"
)

type RedisClient struct {
	MaxCount int
	MaxTtl   int
	redis    Redis
	read     chan int64
	write    chan *Broadcast
	logger   Logger
}

var redisClient *RedisClient

func NewRedisCache(maxCount, maxTtl int, redis Redis) *RedisClient {
	if redisClient == nil {
		redisClient = &RedisClient{
			MaxCount: maxCount,
			MaxTtl:   maxTtl,
			redis:    redis,
			read:     make(chan int64, 100),
			write:    make(chan *Broadcast, 100),
		}
	}
	go redisClient.run()
	return redisClient
}

func (f *RedisClient) run() {
	goNum := make(chan struct{}, 1024)
	for {
		select {
		case uid := <-f.read:
			goNum <- struct{}{}
			go func() {
				if err := f.syncRead(uid); err != nil {
					f.logger.Error(err)
				}
				<-goNum
			}()

		case msg := <-f.write:
			go func() {
				goNum <- struct{}{}
				if err := f.add(msg); err != nil {
					f.logger.Error(err)
				}
				<-goNum
			}()
		}
	}
}

func (f *RedisClient) add(msg *Broadcast) error {
	key := f.uidMessageKey(msg.uid)
	l, err := f.redis.ZCard(key).Result()
	if err != nil {
		return err
	}
	pipe := f.redis.Pipeline()
	if int(l) > f.MaxCount {
		pipe.ZRemRangeByRank(key, 0, l-int64(f.MaxCount))
	}
	msgId := fmt.Sprintf("%s:%d", utils.Md5(msg.msg), l)
	pipe.ZAdd(key, redis.Z{
		Score:  float64(msg.ts),
		Member: msgId,
	})
	pipe.Set(msgId, string(msg.msg), time.Second*time.Duration(f.MaxTtl))
	_, err = pipe.Exec()
	return err
}

func (f *RedisClient) uidMessageKey(uid int64) string {
	return fmt.Sprintf("uid_message:%d", uid)
}

// TODO test
func (f *RedisClient) syncRead(uid int64) (err error) {
	key := f.uidMessageKey(uid)
	total, err := f.redis.ZCard(key).Result()
	if err != nil {
		return err
	}
	pipe := f.redis.Pipeline()
	var readCount int64
	for readCount < total {
		l := total - readCount
		if l > 100 {
			l = 100
		}
		cmd := pipe.ZRangeWithScores(key, 0, l)
		pipe.ZRemRangeByRank(key, 0, l)
		if _, err = pipe.Exec(); err != nil {
			return err
		}
		zs, err := cmd.Result()
		if err != nil {
			return err
		}
		if len(zs) == 0 {
			break
		}
		msgIds := make([]string, 0)
		for _, val := range zs {
			msgIds = append(msgIds, val.Member.(string))
		}
		sliceCmd := pipe.MGet(msgIds...)
		pipe.Del(msgIds...)
		if _, err = pipe.Exec(); err != nil {
			return err
		}
		messages, err := sliceCmd.Result()
		if err != nil {
			return err
		}
		for i, msg := range messages {
			GetHub().send <- &Broadcast{
				msg: []byte(msg.(string)),
				uid: uid,
				ts:  int64(zs[i].Score),
			}
		}
		readCount += 100
	}
	return nil
}