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())) } } }