MessageCenter/rpc/internal/common/newSendMsgService.go

package common

import (
	"app.yhyue.com/moapp/MessageCenter/entity"
	"app.yhyue.com/moapp/MessageCenter/rpc/type/message"
	"app.yhyue.com/moapp/MessageCenter/util"
	"app.yhyue.com/moapp/jybase/common"
	"context"
	"errors"
	"fmt"
	"github.com/zeromicro/go-zero/core/logx"
	"log"
	"strconv"
	"strings"
	"sync"
	"time"
)

var MsgGroupIdMap map[int]int

func SetMsgSummary(newMsg, groupId, msgType int64) error {
	//更新所有消息
	if groupId == 11 {
		//groupId = msgType
		//根据msgType更新待办二级分类汇总
		err1 := entity.ClickhouseConn.Exec(context.Background(), fmt.Sprintf(`alter table message_summary UPDATE msg_bitmap = bitmapOr(msg_bitmap,bitmapBuild([toUInt64(%d)])) where group_id = %d`, newMsg, msgType))
		if err1 != nil {
			//插入失败
			return err1
		}
	}
	err := entity.ClickhouseConn.Exec(context.Background(), fmt.Sprintf(`alter table message_summary UPDATE msg_bitmap = bitmapOr(msg_bitmap,bitmapBuild([toUInt64(%d)])) where group_id = %d`, newMsg, groupId))
	if err != nil {
		return err
	}
	return nil
}

func MultSave11(this *message.MultipleSaveMsgReq) (int64, string) {
	userIdArr := strings.Split(this.UserIds, ",")
	positionIdArr := strings.Split(this.PositionIds, ",")
	if len(userIdArr) == 0 {
		return 0, "无效的用户id"
	}
	wg := &sync.WaitGroup{}
	group_id := MsgGroupIdMap[int(this.MsgType)]

	for i := 0; i < len(userIdArr); i++ {
		if userIdArr[i] == "" {
			continue
		}
		//查询
		wg.Add(1)
		entity.SaveConcurrencyChan <- 1
		var positionId int64
		if len(positionIdArr) == len(userIdArr) {
			positionId = common.Int64All(positionIdArr[i])
		}

		go func(v string, positionId int64) {
			defer func() {
				<-entity.SaveConcurrencyChan
				wg.Done()
			}()
			//消息数组
			nTime := time.Now().Format("2006-01-02 15:04:05")
			sql3 := `INSERT INTO message(appid,receive_userid,receive_name,send_userid,send_name,title,content,msg_type,link,cite_id,createtime,isRead,isdel,msg_log_id,show_buoy,show_content,group_id,position_id) values ("%s",'%s','%s','%s','%s','%s','%s',%d,'%s',0,'%s',0,1,%d,%d,'%s',%d,?);`
			sql3 = fmt.Sprintf(sql3, this.Appid, v, "", this.SendUserId, this.SendName, this.Title, this.Content, this.MsgType, this.Link, nTime, this.MsgLogId, this.ShowBuoy, this.ShowContent, group_id)
			in := entity.Mysql.InsertBySql(sql3, common.If(positionId != 0, positionId, nil))
			logx.Info("插入消息返回 in1 id:", in, "消息类型：", this.MsgType, "用户id：", v)
			if in > -1 {
				ok := MsgCountAdd(v, this.Appid, util.Int64All(group_id), this.MsgType)
				if !ok {
					log.Println("存redis：", ok, v)
				}
			}

			if in > -1 {
				//微信推送模板消息、app push
				pushData := WxTmplAndPush{
					MsgType:     this.MsgType,
					Title:       this.Title,
					Content:     this.Content,
					WxPushUrl:   this.WxPushUrl,
					AppPushUrl:  this.AppPushUrl,
					ProductName: this.ProductName,
					OrderId:     this.OrderId,
					OrderMoney:  this.OrderMoney,
					Row4:        this.Row4,
					SendUserId:  this.SendUserId,
				}
				SentWxTmplAndAppPush(pushData, v, group_id)
			} else {
				logx.Error(fmt.Sprintf("SendAppMsg uId %s  发送消息失败", v))
			}
		}(userIdArr[i], positionId)
	}
	wg.Wait()
	return 0, ""
}

func NewUserSendMsg(in *message.NewUserInsertMsgReq) string {
	userIdArr := strings.Split(in.UserIds, ",")
	positionIdArr := strings.Split(in.PositionIds, ",")
	if len(userIdArr) == 0 {
		return "无效的用户id"
	}
	wg := &sync.WaitGroup{}
	group_id := MsgGroupIdMap[int(in.MsgType)]
	for i := 0; i < len(userIdArr); i++ {
		if userIdArr[i] == "" {
			continue
		}
		//查询
		wg.Add(1)
		entity.SaveConcurrencyChan <- 1
		var positionId int64
		if len(positionIdArr) == len(userIdArr) {
			positionId = common.Int64All(positionIdArr[i])
		}

		go func(v string, positionId int64) {
			defer func() {
				<-entity.SaveConcurrencyChan
				wg.Done()
			}()

			row := entity.ClickhouseConn.QueryRow(context.Background(), fmt.Sprintf("SELECT COUNT(*) from message_user_summary where userId = '%s'", v))
			var count uint64
			row.Scan(&count)
			if count > 0 { //存在则更新
				err1 := entity.ClickhouseConn.Exec(context.Background(), fmt.Sprintf(`alter table message_user_summary UPDATE allMsg = bitmapOr(allMsg,bitmapBuild([toUInt64(%d)])) where userId = '%s'`, in.MsgLogId, v))
				if err1 != nil {
					log.Println("新用户update message_user_summary表出错，error", err1)
					return
				}
			} else {
				//新用户需要insert
				sql := "INSERT INTO message_user_summary values "
				sql += fmt.Sprintf(" ('%s',bitmapBuild([toUInt64(%d)]),bitmapBuild([toUInt64(0)])) ", v, int(in.MsgLogId))
				fmt.Println("sql", sql)
				err1 := entity.ClickhouseConn.Exec(context.Background(), sql)
				if err1 != nil {
					//插入失败
					log.Println("新用户insert message_user_summary表出错，error", err1)
					return
				}
			}
			//微信推送模板消息、app push
			pushData := WxTmplAndPush{
				MsgType:     in.MsgType,
				Title:       in.Title,
				Content:     in.Content,
				WxPushUrl:   in.WxPushUrl,
				AppPushUrl:  in.AppPushUrl,
				ProductName: in.ProductName,
				OrderId:     in.OrderId,
				OrderMoney:  in.OrderMoney,
				Row4:        in.Row4,
			}
			SentWxTmplAndAppPush(pushData, v, group_id)
		}(userIdArr[i], positionId)
	}
	wg.Wait()
	return ""
}

func UpdateUserMsgSummary(in *message.BitmapSaveMsgReq) error {
	userIdArr := strings.Split(in.UserIds, ",")
	//positionIdArr := strings.Split(in.PositionIds, ",")
	if len(userIdArr) == 0 {
		return errors.New("无效的用户id")
	}
	wg := &sync.WaitGroup{}
	group_id := MsgGroupIdMap[int(in.MsgType)]
	str := ""
	for k, v := range userIdArr {
		if k != 0 {
			str += ","
		}
		str += "'" + v + "'"
	}
	fmt.Println(fmt.Sprintf(`alter table message_user_summary UPDATE allMsg = bitmapOr(allMsg,bitmapBuild([toUInt64(%d)])) where userId in (%s)`, in.MsgLogId, str))
	err1 := entity.ClickhouseConn.Exec(context.Background(), fmt.Sprintf(`alter table message_user_summary UPDATE allMsg = bitmapOr(allMsg,bitmapBuild([toUInt64(%d)])) where userId in (%s)`, in.MsgLogId, str))
	if err1 != nil {
		return err1
	}
	for i := 0; i < len(userIdArr); i++ {
		if userIdArr[i] == "" {
			continue
		}
		//查询
		wg.Add(1)
		entity.SaveConcurrencyChan <- 1
		/*var positionId int64
		if len(positionIdArr) == len(userIdArr) {
			positionId = common.Int64All(positionIdArr[i])
		}*/
		go func(v string) {
			defer func() {
				<-entity.SaveConcurrencyChan
				wg.Done()
			}()
			//微信推送模板消息、app push
			pushData := WxTmplAndPush{
				MsgType:     in.MsgType,
				Title:       in.Title,
				Content:     in.Content,
				WxPushUrl:   in.WxPushUrl,
				AppPushUrl:  in.AppPushUrl,
				ProductName: in.ProductName,
				OrderId:     in.OrderId,
				OrderMoney:  in.OrderMoney,
				Row4:        in.Row4,
			}
			SentWxTmplAndAppPush(pushData, v, group_id)
		}(userIdArr[i])
	}
	wg.Wait()
	return nil
}

func ConvertToBitmap(num int) (res []uint32) {
	binary := strconv.FormatInt(int64(num), 2)
	total := len(binary)
	for i := total - 1; i >= 0; i-- {
		if binary[i] == '1' {
			res = append(res, uint32(total-i))
		}
	}
	return
}

type WxTmplAndPush struct {
	MsgType     int64
	Title       string
	Content     string
	WxPushUrl   string
	AppPushUrl  string
	ProductName string
	OrderId     string
	OrderMoney  string
	Row4        string
	SendUserId  string
}

func SentWxTmplAndAppPush(this WxTmplAndPush, v string, group_id int) {
	nTime := time.Now().Format("2006-01-02 15:04:05")
	//发送消息成功，推送微信、app
	//fmt.Println("this.MsgType", this.MsgType)
	pushConfig, err := GetWxTmplConfig(this.MsgType)
	if err != nil {
		logx.Error(fmt.Sprintf("SendWxTmplMsg uId %s  Error %s", v, err.Error()))
	}
	p := &WxTmplPush{
		Config: pushConfig,
	}
	p.MgoId = v
	if this.MsgType == 10 {
		this.Title = this.ProductName
		this.Content = this.OrderId
		nTime = this.OrderMoney
	}
	// 消息模版 工单类型 {{thing19.DATA}} 工单标题 {{thing6.DATA}} 项目名称 {{thing13.DATA}} 服务时间 {{time25.DATA}} 服务地址 {{thing26.DATA}}
	if this.MsgType != 1 && this.MsgType != 10 {
		err = p.SendMsg(this.WxPushUrl, this.Title, this.Content, nTime, this.Row4)
		if err != nil {
			logx.Error(fmt.Sprintf("SendWxTmplMsg uId %s  Error %s", v, err.Error()))
		} else {
			logx.Infof("SendWxTmplMsg uId success %s ", v)
		}
	}
	if this.MsgType == 1 {
		mst := new(WxTmplConfig)
		mst.Switch = AppPushMsgType[group_id]
		p.Config = mst
	}
	uData := p.GetUserPushInfo()
	//app推送
	if this.MsgType != 10 {
		category := ""
		if this.SendUserId == "cbgl" {
			category = "服务通知_工作事项"
		}
		if err = AppPushMsg(uData, AppPushMsgType[group_id], this.AppPushUrl, this.Title, this.Content, this.MsgType, category); err != nil {
			logx.Error(fmt.Sprintf("SendAppMsg uId %s  Error %s", v, err.Error()))
		}
	}
}