MessageCenter/rpc/internal/common/sendWxTmplMsg.go

package common

import (
	"app.yhyue.com/moapp/MessageCenter/entity"
	"app.yhyue.com/moapp/MessageCenter/rpc/internal/config"
	"app.yhyue.com/moapp/MessageCenter/util"
	"app.yhyue.com/moapp/jybase/common"
	dataFormat "app.yhyue.com/moapp/jybase/date"
	m "app.yhyue.com/moapp/jybase/mongodb"
	"app.yhyue.com/moapp/jybase/redis"
	qrpc "app.yhyue.com/moapp/jybase/rpc"
	"fmt"
	"strings"
	"time"
)

type WxTmplPush struct {
	MgoId, OpenId, Position string //UserId 用户mgoId, OpenId 微信id, Position 职位id
	MessageClass            string //对应是否开启推送的key
	PushTmplId              string //推送模版id
}

const CacheDb = "msgCount"

func MessageType() (func() map[string]string, []map[string]interface{}) {
	var data []map[string]interface{}
	rData := entity.Mysql.SelectBySql("SELECT * FROM message_column ORDER BY sequence ASC")
	switchName := map[string]string{}
	if rData != nil && len(*rData) > 0 {
		data = *rData
		for _, mData := range *rData {
			if settingKey, messageName := util.ObjToString(mData["switch"]), util.ObjToString(mData["name"]); settingKey == "" && messageName == "" {
				switchName[settingKey] = messageName
			}
		}
	}
	return func() map[string]string {
		return switchName
	}, data
}

var AllMsgType func() map[string]string

var getSendTotalRedisKey = func(uFlag ...string) string {
	if len(uFlag) == 0 { //统计当日信息总发送量
		return fmt.Sprintf("messageCenter_SendWxMsgTotal_%s", time.Now().Format(dataFormat.Date_yyyyMMdd))
	} //统计单用户今日发送量
	return fmt.Sprintf("messageCenter_SendWxMsgTotal_%s_%s", time.Now().Format(dataFormat.Date_yyyyMMdd), uFlag[0])
}

func (stm *WxTmplPush) SendMsg(link string, getMsg func() map[string]*qrpc.TmplItem) error {
	if stm.PushTmplId == "" || stm.MessageClass == "" || (stm.MgoId != "" && stm.OpenId != "" && stm.Position != "") {
		return fmt.Errorf("缺少参数")
	}
	if AllMsgType()[stm.MessageClass] == "" {
		return fmt.Errorf("未知消息类型")
	}
	if err := stm.getUserOpenIdAndWxPushState(); err != nil {
		return err
	}
	//校验发送量及频率
	if err := stm.Check(); err != nil {
		return err
	}
	// 发送信息
	if _, err := stm.Send(link, getMsg()); err != nil {
		return err
	}
	// 发送数量计数
	stm.IncrCount()
	return nil
}

func (stm *WxTmplPush) Check() error {
	uCache, allCache := getSendTotalRedisKey(stm.OpenId), getSendTotalRedisKey()
	//校验当日微信模版发送总量
	if total := redis.GetInt(CacheDb, allCache); total > config.ConfigJson.WxTmplConfig.Limit.Total {
		return fmt.Errorf("已达发送总量上限")
	}
	//校验当日单用户发送总量
	if uTotal := redis.GetInt(CacheDb, uCache); uTotal > config.ConfigJson.WxTmplConfig.Limit.OneDayLimit {
		return fmt.Errorf("已达单该用户发送总量上限")
	}
	//校验发送间隔
	if sendWait, _ := redis.Exists(CacheDb, fmt.Sprintf("%s_sendwait", uCache)); sendWait {
		return fmt.Errorf("发送模版消息频繁，稍后重试")
	}
	return nil
}

func (stm *WxTmplPush) IncrCount() {
	uCache, allCache := getSendTotalRedisKey(stm.OpenId), getSendTotalRedisKey() //当日微信模版消息发送总量
	var total int64
	if total = redis.Incr(CacheDb, allCache); total == 1 {
		_ = redis.SetExpire(CacheDb, allCache, 60*60*24*7)
	}
	redis.Incr(CacheDb, uCache)                                                                                   //当日用户发送数量
	redis.Put(CacheDb, fmt.Sprintf("%s_sendwait", uCache), 1, config.ConfigJson.WxTmplConfig.Limit.DuringMine*60) //下次发送时间

	for _, num := range config.ConfigJson.WxTmplConfig.Limit.Alert.Nums {
		if total == num {
			util.SendRetryMail(3, strings.Join(config.ConfigJson.WxTmplConfig.Limit.Alert.ToMail, ","), strings.Join(config.ConfigJson.WxTmplConfig.Limit.Alert.ToMail, ","),
				"剑鱼微信模版告警邮件", fmt.Sprintf("今日发送微信模版信息数量已达%d条，总量%d条", total, config.ConfigJson.WxTmplConfig.Limit.Total), entity.GmailAuth)
		}
	}
}

// getUserOpenIdAndWxPushState 查询微信openid微信消息通知状态
// mId mongoUserid、oId 用户openid、pId positionId 用户职位id
func (stm *WxTmplPush) getUserOpenIdAndWxPushState() error {
	uData := func() map[string]interface{} {
		query := map[string]interface{}{}
		if stm.OpenId != "" {
			query["s_m_openid"] = stm.OpenId
		} else if stm.MgoId != "" {
			query["_id"] = m.StringTOBsonId(stm.MgoId)
		} else if stm.Position != "" {
			uInfo := entity.Mysql.SelectBySql("SELECT user_id FROM base_service.base_position WHERE id = ? ", stm.Position)
			if uInfo != nil && len(*uInfo) > 0 {
				if baseUserId := common.Int64All((*uInfo)[0]["user_id"]); baseUserId != 0 {
					query["base_user_id"] = baseUserId
				}
			}
		}
		if len(query) > 0 {
			rData, _ := entity.MQFW.FindOneByField("user", query, fmt.Sprintf(`{"s_m_openid":1,"o_pushset.%s.i_wxpush":1}`, stm.MessageClass))
			if rData != nil && len(*rData) > 0 {
				return *rData
			}
		}
		return nil
	}()
	if uData == nil {
		return fmt.Errorf("未查询到用户微信信息")
	}
	if pushSetMap := common.ObjToMap(uData["o_pushset"]); pushSetMap != nil && len(*pushSetMap) > 0 {
		if pushKeyMap := common.ObjToMap((*pushSetMap)[stm.MessageClass]); pushKeyMap != nil && len(*pushKeyMap) > 0 {
			if common.Int64All((*pushKeyMap)["i_wxpush"]) == 1 {
				return nil
			}
		}
	}
	return fmt.Errorf("未开启推送设置")
}

// Send 发送微信模版消息
func (stm *WxTmplPush) Send(link string, msg map[string]*qrpc.TmplItem) (pushOk bool, err error) {
	return qrpc.WxSendTmplMsg(config.ConfigJson.WxTmplConfig.RpcAddr, &qrpc.WxTmplMsg{
		OpenId:   stm.OpenId,
		TplId:    stm.PushTmplId,
		TmplData: msg,
		Url:      link,
	})
}