wechat_ipad_pro/db/redisOperation.go

/**
 * @author mii
 * @date 2020/2/29 0029
 */

package db

import (
	"errors"
	"fmt"
	"sync"
	"time"
	"xiawan/wx/clientsdk/baseinfo"
	"xiawan/wx/db/table"
	"xiawan/wx/srv"
	"xiawan/wx/srv/srvconfig"

	"github.com/gogf/gf/database/gredis"
	"github.com/gogf/gf/encoding/gjson"
	"github.com/gogf/gf/frame/g"
	"github.com/gogf/gf/util/gconv"
	"github.com/gomodule/redigo/redis"
)

const (
	REDIS_OPERATION_SET             = "SET"
	REDIS_OPERATION_GET             = "GET"
	REDIS_OPERATION_LIST_LLEN       = "llen"
	REDIS_OPERATION_LIST_LPSUH      = "LPUSH"
	REDIS_OPERATION_LIST_LPOP       = "lpop"
	REDIS_OPERATION_LIST_RPOP       = "rpop"
	REDIS_OPERATION_LIST_LTRIM      = "ltrim"
	REDIS_OPERATION_CHANNEL_PUBLISH = "publish"
	REDIS_OPERATION_EXISTS          = "EXISTS"
	REDIS_OPERATION_DELETE          = "DEL"
)

const (
	DEFAULT_GROUP_NAME = "default" // Default configuration group name.
	DEFAULT_REDIS_PORT = 6379      // Default redis port configuration if not passed.
)

type Obj struct {
	Value interface{}
}

func RedisSetup() {
	//配置redis
	// MaxIdle:          10,   // 设置最大空闲连接数
	// MaxActive:        100,  // 设置最大活跃连接数
	// IdleTimeout:      300,  // 空闲连接超时时间（单位：秒）
	// MaxConnLifetime:  300,  // 连接最大存活期（单位：秒）
	// WaitTimeout:      5,    // 连接等待超时时间（单位：秒）
	// ReadTimeout:      10,   // 读操作超时时间（单位：秒）
	// WriteTimeout:     10,   // 写操作超时时间（单位：秒）
	gredis.SetConfig(srvconfig.GlobalSetting.RedisConfig, DEFAULT_GROUP_NAME)
	conn := g.Redis().GetConn()
	err := conn.Err()
	if err != nil {
		//logger.Errorln(err)
		fmt.Println("failed to connect Redis:", err)
	}
	fmt.Println("connect Redis success")
	go scheduleSaveToRedis()
}

// 尝试获取锁
func AcquireLock(key string, timeout time.Duration) (bool, error) {
	redisConn := getRedisConn()
	defer redisConn.Close()

	lockValue := fmt.Sprintf("%d", time.Now().UnixNano()) // Unique lock value
	result, err := redisConn.Do(REDIS_OPERATION_SET, key, lockValue, "NX", "PX", int64(timeout.Seconds()*1000))
	if err != nil {
		return false, err
	}
	return result != nil, nil
}

// 释放锁
func ReleaseLock(key string) error {
	redisConn := getRedisConn()
	defer redisConn.Close()

	_, err := redisConn.Do(REDIS_OPERATION_DELETE, key)
	if err != nil {
		return err
	}
	return nil
}

// acquireLockWithRetry 尝试获取锁，带有重试机制
func acquireLockWithRetry(key string, timeout time.Duration, retryInterval time.Duration, maxRetries int) (bool, error) {
	for i := 0; i < maxRetries; i++ {
		lockAcquired, err := AcquireLock(key, timeout)
		if err != nil {
			return false, err
		}
		if lockAcquired {
			return true, nil
		}
		time.Sleep(retryInterval) // 等待一段时间后重试
	}
	return false, fmt.Errorf("could not acquire lock after %d retries", maxRetries)
}

// CreateSyncMsgList 创建一个同步信息保存列表
func CreateSyncMsgList(exId string) bool {
	ok, err := LPUSH(exId, "撒大声地")
	if err != nil {
		return false
	}
	return ok
}

// LPOPObj 从列表中取出对象然后反序列化成对象
func LPOPObj(k string, i interface{}) error {
	_var, err := LPOP(k)
	if err != nil {
		return err
	}
	err = gjson.DecodeTo(_var.Bytes(), &i)
	if err != nil {
		return err
	}
	return nil
}

// RPOPObj 从列表尾部 取出对象然后反序列化成对象
func RPOPObj(k string, i interface{}) error {
	_var, err := RPOP(k)
	if err != nil {
		return err
	}
	err = gjson.DecodeTo(_var.Bytes(), &i)
	if err != nil {
		return err
	}
	return nil
}

// LPUSHObj 将对象序列化成json保存
func LPUSHObj(k string, i interface{}) (bool, error) {
	iData, err := gjson.Encode(i)
	// 判断 k 的尾号是 _syncMsg

	if err != nil {
		return false, err
	}
	ok, err := LPUSH(k, iData)
	if err != nil {
		return ok, err
	}
	return ok, nil
}

func LPUSH(k string, i interface{}) (bool, error) {
	lockKey := k + "_lock"
	maxRetries := 200                      // 最大重试次数
	retryInterval := 50 * time.Millisecond // 重试间隔时间为500毫秒
	// 使用带重试机制的获取锁函数
	acquireLockWithRetry(lockKey, 5*time.Second, retryInterval, maxRetries)
	defer ReleaseLock(lockKey) // 确保方法结束时释放锁

	maxLength := 500 // 设置最大列表长度
	redisConn := getRedisConn()
	defer redisConn.Close()
	result, err := redisConn.Do(REDIS_OPERATION_LIST_LPSUH, k, i)
	if err != nil {
		//logger.Errorln(err)
		return false, err
	}
	_len, err := LLEN(k)
	if err != nil {
		return true, err
	}
	if _len == 0 {
		return true, nil
	}
	// 如果列表长度超过 maxLength，则循环执行 LPOP 操作
	for _len > int32(maxLength) {
		_, err = redisConn.Do(REDIS_OPERATION_LIST_LPOP, k)
		if err != nil {
			return false, err
		}
		_len = _len - int32(1)
	}
	return gconv.String(result) == "OK", nil
}

// LPOP 从列表中取出一个值
func LPOP(k string) (*g.Var, error) {
	redisConn := getRedisConn()
	defer redisConn.Close()
	r, err := redisConn.DoVar(REDIS_OPERATION_LIST_LPOP, k)
	if err != nil {
		return nil, err
	}

	return r, nil
}

// RPOP 从列表尾部 取出一个元素
func RPOP(k string) (*g.Var, error) {
	redisConn := getRedisConn()
	defer redisConn.Close()
	r, err := redisConn.DoVar(REDIS_OPERATION_LIST_RPOP, k)
	if err != nil {
		return nil, err
	}

	return r, nil
}

// LLEN 取列表长度
func LLEN(k string) (int32, error) {
	redisConn := getRedisConn()
	defer redisConn.Close()
	_var, err := redisConn.DoVar(REDIS_OPERATION_LIST_LLEN, k)
	if err != nil {
		//logger.Errorln(err)
		return 0, err
	}
	return _var.Int32(), nil
}

// PUBLISH 发布消息
func PUBLISH(k string, i interface{}) error {
	return PushQueue(k, i)
}

func SETExpirationObj(k string, i interface{}, expiration int64) error {
	redisConn := getRedisConn()
	defer redisConn.Close()

	iData, err := gjson.Encode(i)
	if err != nil {
		return err
	}
	var result interface{}
	if expiration > 0 {
		result, err = redisConn.Do(REDIS_OPERATION_SET, k, iData, "EX", expiration)
	} else {
		result, err = redisConn.Do(REDIS_OPERATION_SET, k, iData)
	}

	if err != nil {
		//logger.Errorln(err)
		return err
	}
	if gconv.String(result) == "OK" {
		return nil
	}
	return errors.New(gconv.String(result))
}

func SETObj(k string, i interface{}) error {
	redisConn := getRedisConn()
	defer redisConn.Close()

	iData, err := gjson.Encode(i)
	if err != nil {
		return err
	}
	result, err := redisConn.Do(REDIS_OPERATION_SET, k, iData)
	if err != nil {
		//logger.Errorln(err)
		return err
	}
	if gconv.String(result) == "OK" {
		return nil
	}
	return errors.New(gconv.String(result))
}

func GETObj(k string, i interface{}) error {
	redisConn := getRedisConn()
	defer redisConn.Close()

	_var, err := redisConn.Do(REDIS_OPERATION_GET, k)
	if err != nil {
		return err
	}
	err = gjson.DecodeTo(_var, &i)
	if err != nil {
		return err
	}
	return nil
}

func DelObj(k string) error {
	redisConn := getRedisConn()
	defer redisConn.Close()

	_, err := redisConn.Do(REDIS_OPERATION_DELETE, k)
	if err != nil {
		return err
	}
	return nil
}

func Exists(k string) (bool, error) {
	//检查是否存在key值
	redisConn := getRedisConn()
	defer redisConn.Close()

	exists, err := redisConn.Do(REDIS_OPERATION_EXISTS, k)
	if err != nil {
		fmt.Println("illegal exception")
		return false, err
	}
	//fmt.Printf("exists or not: %v \n", exists)
	if exists.(int64) == 1 {
		return true, nil
	}
	return false, nil
}

func getRedisConn() *gredis.Conn {
	return g.Redis().GetConn()
}

// PublishSyncMsgWxMessage 发布微信消息; 这里就是接收到 微信后台 推送来的消息(包括其他用户私聊消息、群聊消息、收款消息等等)
func PublishSyncMsgWxMessage(acc *srv.WXAccount, response table.SyncMessageResponse) error {
	if acc == nil {
		return errors.New("PublishSyncMsgWxMessage acc == nil")
	}
	response.UUID = acc.GetUserInfo().UUID
	response.UserName = acc.GetUserInfo().GetUserName()
	response.LoginState = acc.GetLoginState()
	response.Type = table.RedisPushSyncTypeWxMsg
	response.TargetIp = srvconfig.GlobalSetting.TargetIp
	if len(response.GetAddMsgs()) != 0 || len(response.GetContacts()) != 0 {
		if srvconfig.GlobalSetting.HttpSyncMsg {
			LPUSHObj(response.UUID+"_syncHttp", &response) //缓存reids
		}
	}
	return nil
}

// PublishFavItem 发布收藏消息
func PublishFavItem(acc *srv.WXAccount, favItem *baseinfo.FavItem) error {
	if acc == nil {
		return errors.New("PublishSyncMsgWxMessage acc == nil")
	}
	response := table.SyncMessageResponse{}
	response.UUID = acc.GetUserInfo().UUID
	response.UserName = acc.GetUserInfo().GetUserName()
	response.LoginState = acc.GetLoginState()
	response.TargetIp = srvconfig.GlobalSetting.TargetIp
	response.Type = table.RedisPushFavitem
	response.FavItem = favItem
	// 缓存reids
	if srvconfig.GlobalSetting.NewsSynWxId {
		// 缓存reids
		LPUSHObj(response.UUID+"_syncMsg", &response)
		return PUBLISH(response.UUID+"_wx_sync_msg_topic", &response)
	}
	return PUBLISH("wx_sync_msg_topic", &response)
}

// PublishSyncMsgLoginState 微信状态
func PublishSyncMsgLoginState(wxid string, state uint32) error {
	//fmt.Println("推送->wxid=="+wxid+"--->", state)
	response := table.SyncMessageResponse{}
	response.TargetIp = srvconfig.GlobalSetting.TargetIp
	response.Type = table.RedisPushSyncTypeLoginState
	response.UserName = wxid
	response.LoginState = state
	// 缓存reids
	if srvconfig.GlobalSetting.NewsSynWxId {
		// 缓存reids
		LPUSHObj(response.UUID+"_syncMsg", &response)
		return PUBLISH(response.UUID+"_wx_sync_msg_topic", &response)
	}
	return PUBLISH("wx_sync_msg_topic", &response)
}

// PublishWxInItOk 初始化完成
func PublishWxInItOk(UUID string, state uint32) error {
	if UUID == "" || state == 0 {
		return errors.New("uuid || state ==  nil")
	}
	response := table.SyncMessageResponse{}
	response.TargetIp = srvconfig.GlobalSetting.TargetIp
	response.Type = table.RedisPushWxInItOk
	response.LoginState = state
	response.UUID = UUID
	response.UserName = "初始化完成!"
	// 缓存reids
	if srvconfig.GlobalSetting.NewsSynWxId {
		_, _ = LPUSHObj(response.UUID+"_wx_sync_msg_topic", &response)
		return PUBLISH(response.UUID+"_wx_sync_msg_topic", &response)
	}
	return PUBLISH("wx_sync_msg_topic", &response)
}

// PublishSyncMsgCheckLogin 扫码结果
func PublishSyncMsgCheckLogin(UUID string, result *baseinfo.CheckLoginQrCodeResult) error {
	if UUID == "" || result == nil {
		return errors.New("uuid || result ==  nil")
	}
	response := table.SubMessageCheckLoginQrCode{}
	response.TargetIp = srvconfig.GlobalSetting.TargetIp
	response.Type = table.RedisPushSyncTypeCheckLogin
	response.CheckLoginResult = result
	response.UUID = UUID
	// 缓存reids
	if srvconfig.GlobalSetting.NewsSynWxId {
		_, _ = LPUSHObj(response.UUID+"_wx_sync_msg_topic", &response)
		return PUBLISH(response.UUID+"_wx_sync_msg_topic", &response)
	}
	return PUBLISH("wx_sync_msg_topic", &response)
}

// GETSyncMsg 获取指定号缓存在redis的消息
func GETSyncMsg(uuid string) (int, []*table.SyncMessageResponse, error) {
	_len, err := LLEN(uuid + "_syncMsg")
	if err != nil {
		return 0, nil, err
	}
	if _len == 0 {
		return 0, nil, nil
	}
	syncMsgLen := _len
	if syncMsgLen > 10 {
		syncMsgLen = 10
	}
	opValues := []*table.SyncMessageResponse{}
	for i := int32(0); i <= syncMsgLen; i++ {
		opVal := &table.SyncMessageResponse{}
		err = LPOPObj(uuid+"_wx_sync_msg_topic", opVal)
		if err != nil {
			continue
		}
		opValues = append(opValues, opVal)
	}
	ContinueFlag := 0
	if (_len % 10) > 0 {
		ContinueFlag = 1
	}

	return ContinueFlag, opValues, nil
}

// PublishLicenseKey 使用状态, 主要是过期时间和开始时间
func PublishLicenseKey(licenseKey *table.LicenseKey) error {
	lockKey := "uuid_syncLicenseKey_lock"
	maxRetries := 200                      // 最大重试次数
	retryInterval := 50 * time.Millisecond // 重试间隔时间为500毫秒
	// 使用带重试机制的获取锁函数
	acquireLockWithRetry(lockKey, 5*time.Second, retryInterval, maxRetries)

	defer ReleaseLock(lockKey) // 确保方法结束时释放锁
	// 缓存 reids
	LPUSHObj("uuid_syncLicenseKey", licenseKey)
	return nil
}

// HttpSyncLicenseKey HTTP-轮询 同步激活状态 获取 Redis 内的消息
func HttpSyncLicenseKey() ([]*table.LicenseKey, error) {
	lockKey := "uuid_syncLicenseKey_lock"
	maxRetries := 200                      // 最大重试次数
	retryInterval := 50 * time.Millisecond // 重试间隔时间为500毫秒
	// 使用带重试机制的获取锁函数
	acquireLockWithRetry(lockKey, 5*time.Second, retryInterval, maxRetries)

	defer ReleaseLock(lockKey) // 确保方法结束时释放锁

	syncMsgLen, err := LLEN("uuid_syncLicenseKey")
	if err != nil {
		return nil, err
	}

	msgInfos := []*table.LicenseKey{}
	if !(syncMsgLen > int32(0)) {
		return msgInfos, nil
	}

	for i := int32(0); i < syncMsgLen; i++ {
		msg := &table.LicenseKey{}
		err = RPOPObj("uuid_syncLicenseKey", msg)
		if err != nil {
			continue
		}
		msgInfos = append(msgInfos, msg)
	}
	return msgInfos, nil
}

// HttpSyncMsy HTTP-轮询 同步消息时 获取 Redis 内的消息
func HttpSyncMsy(uuid string, count int) ([]*table.SyncMessageResponse, error) {
	lockKey := uuid + "_syncHttp_lock"
	maxRetries := 200                      // 最大重试次数
	retryInterval := 50 * time.Millisecond // 重试间隔时间为500毫秒
	// 使用带重试机制的获取锁函数
	acquireLockWithRetry(lockKey, 5*time.Second, retryInterval, maxRetries)

	defer ReleaseLock(lockKey) // 确保方法结束时释放锁

	_len, err := LLEN(uuid + "_syncHttp")
	if err != nil {
		return nil, err
	}

	syncMsgLen := int32(count)
	if syncMsgLen == 0 { // count=0 时获取所有消息
		syncMsgLen = _len
	} else if syncMsgLen > _len {
		syncMsgLen = _len
	}

	msgInfos := []*table.SyncMessageResponse{}
	if _len == 0 {
		return msgInfos, nil
	}

	for i := int32(0); i < syncMsgLen; i++ {
		msg := &table.SyncMessageResponse{}
		err = RPOPObj(uuid+"_syncHttp", msg)
		if err != nil {
			continue
		}
		msgInfos = append(msgInfos, msg)
	}

	return msgInfos, nil
}

// PushQueue 发送队例
func PushQueue(queueName string, i interface{}) (err error) {
	con := getRedisConn()
	defer con.Close()

	maxLength := 200 // 最大 200 条
	if i == nil {
		return err
	}

	_, err = con.Do("rpush", queueName, gconv.String(i))
	if err != nil {
		return fmt.Errorf("PushQueue redis error: %s", err)
	}
	// 使用 ltrim 裁剪列表，以保持其最大长度
	_, err = con.Do("LTRIM", queueName, -maxLength, -1)
	return err
}

// ClearSyncMsgCache 清理所有以 nameKey(_syncMsg) 结尾的 Redis 缓存键
func ClearSyncMsgCache(nameKey string) error {
	redisConn := getRedisConn()
	defer redisConn.Close()

	// SCAN 命令用于增量迭代 key，效率较高
	iter := 0
	for {
		// 使用 redigo 的 Do 方法执行 SCAN 命令
		reply, err := redis.Values(redisConn.Do("SCAN", iter, "MATCH", "*"+nameKey))
		if err != nil {
			return fmt.Errorf("error during SCAN: %v", err)
		}

		// 解析出 SCAN 的结果集
		var keys []string
		_, err = redis.Scan(reply, &iter, &keys)
		if err != nil {
			return fmt.Errorf("error parsing SCAN result: %v", err)
		}

		// 删除匹配的 key
		for _, key := range keys {
			_, err = redisConn.Do(REDIS_OPERATION_DELETE, key)
			if err != nil {
				return fmt.Errorf("error deleting key %s: %v", key, err)
			}
		}

		// iter 为 0 时表示 SCAN 已迭代完所有 key
		if iter == 0 {
			break
		}
	}

	return nil
}

const (
	maxMessagesPerUser = 125
	saveInterval       = 5 * time.Second
)

type userCache struct {
	mutex sync.RWMutex
	msgs  map[string]struct{}
}

var (
	msgCache    = make(map[string]*userCache) // 存储每个用户的缓存和锁
	globalMutex sync.RWMutex                  // 保证对 msgCache 的安全访问
)

// AddOrUpdateMsgId 添加消息ID到 Redis 集合，如果集合中不存在则更新
func AddOrUpdateMsgId(userName string, newMsgId string) (bool, error) {

	// 获取或创建用户缓存
	userCache := getUserCache(userName)

	userCache.mutex.Lock()
	defer userCache.mutex.Unlock()

	// 检查消息 ID 是否存在
	if _, exists := userCache.msgs[newMsgId]; exists {
		return false, nil
	}

	// 添加消息 ID 到缓存
	userCache.msgs[newMsgId] = struct{}{}

	// 确保缓存不会超过 25 条 // 删除最旧的消息 ID
	if len(userCache.msgs) > maxMessagesPerUser {
		for msgID := range userCache.msgs {
			delete(userCache.msgs, msgID)
			if len(userCache.msgs) <= maxMessagesPerUser {
				break
			}
		}
	}

	return true, nil
}

func getUserCache(userName string) *userCache {
	// 使用全局锁来安全地访问或初始化用户缓存
	globalMutex.RLock()
	cache, exists := msgCache[userName]
	globalMutex.RUnlock()

	if !exists {
		// 使用写锁来修改 msgCache
		globalMutex.Lock()
		defer globalMutex.Unlock()

		// 再次检查，防止竞争条件
		if cache, exists = msgCache[userName]; !exists {
			cache = &userCache{
				msgs:  make(map[string]struct{}),
				mutex: sync.RWMutex{},
			}
			// 从 Redis 加载用户消息
			loadFromRedis(userName, cache.msgs)
			msgCache[userName] = cache
		}
	}

	return cache
}

func loadFromRedis(userName string, userMsgs map[string]struct{}) {
	redisConn := getRedisConn()
	defer redisConn.Close()

	key := fmt.Sprintf("user:%s:msg_ids", userName)
	msgs, err := redis.Strings(redisConn.Do("SMEMBERS", key))
	if err != nil {
		return
	}

	for _, msgID := range msgs {
		userMsgs[msgID] = struct{}{}
	}
}

func saveToRedis() {
	// 锁定整个操作以安全地遍历 msgCache
	globalMutex.RLock()
	defer globalMutex.RUnlock()

	redisConn := getRedisConn()
	defer redisConn.Close()

	for userName, userCache := range msgCache {
		// 锁定每个用户的缓存数据
		userCache.mutex.RLock()

		key := fmt.Sprintf("user:%s:msg_ids", userName)
		msgIDs := make([]interface{}, 0, len(userCache.msgs)+1)
		msgIDs = append(msgIDs, key)
		for msgID := range userCache.msgs {
			msgIDs = append(msgIDs, msgID)
		}

		// 更新 Redis
		redisConn.Do("DEL", key)
		_, err := redisConn.Do("SADD", msgIDs...)
		if err != nil {
			fmt.Println("Error writing to Redis:", err)
		}
		userCache.mutex.RUnlock()
	}
}

func scheduleSaveToRedis() {
	ticker := time.NewTicker(saveInterval)
	for range ticker.C {
		saveToRedis()
	}
}