package common import ( "app.yhyue.com/moapp/MessageCenter/rpc/internal/config" "fmt" "log" "strconv" "strings" "sync" "time" "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" "app.yhyue.com/moapp/jybase/redis" "github.com/zeromicro/go-zero/core/logx" ) // 类型的顺序 const order = "1,4" const MsgCountKey = "count_%s_%d" //redis 消息未读数量 Count.用户id.消息类型=数量 const MsgClassCountKey = "msg_class_count_%s_%d" //redis 用户消息class分类消息数量 const redisModule = "msgCount" func FindUserMsg(this message.FindUserMsgReq, isClean bool) message.FindUserMsgRes { var err error var count int64 cquery := map[string]interface{}{ "receive_userid": this.UserId, "isdel": 1, "appid": this.Appid, } if this.MsgType > 0 { cquery["group_id"] = this.MsgType } if this.Read != -1 { cquery["isRead"] = this.Read } count = entity.Mysql.Count("message", cquery) data := message.FindUserMsgRes{} if this.PageSize == 5 { //从缓存里边取数据 pc_a, err := entity.GetData(this.UserId) if err == nil && pc_a != nil { // 缓存有值 if !isClean { data.Code = 1 data.Message = "查询成功" data.Data = pc_a.Data data.Count = pc_a.Count return data } } } count = entity.Mysql.Count("message", cquery) if count > 0 { res := entity.Mysql.Find("message", cquery, "", "createtime desc", (int(this.OffSet)-1)*int(this.PageSize), int(this.PageSize)) //log.Println("数据:", res) if res != nil && len(*res) > 0 { for _, v := range *res { _id := util.Int64All(v["id"]) id := strconv.FormatInt(_id, 10) data.Data = append(data.Data, &message.Messages{ Id: id, Appid: util.ObjToString(v["appId"]), ReceiveUserId: util.ObjToString(v["receive_userid"]), ReceiveName: util.ObjToString(v["receive_name"]), SendUserId: util.ObjToString(v["send_userid"]), SendName: util.ObjToString(v["send_name"]), Createtime: util.ObjToString(v["createtime"]), Title: util.ObjToString(v["title"]), MsgType: int64(util.IntAll(v["group_id"])), Link: util.ObjToString(v["link"]), CiteId: util.Int64All(v["cite_id"]), Content: util.ObjToString(v["content"]), IsRead: util.Int64All(v["isRead"]), MsgLogId: util.Int64All(v["msg_log_id"]), }) } } } data.Count = count if this.PageSize == 5 { redisData := map[string]interface{}{ "count": count, "data": data.Data, } entity.SetData(this.UserId, redisData, entity.SurvivalTime) } if err != nil { data.Code = 0 data.Message = "查询失败" } else { data.Code = 1 data.Message = "查询成功" } return data } func UserMsgList(this *message.UserMsgListReq) *message.UserMsgList { var ( unread, count int64 ) m := &MessageService{} data := new(message.UserMsgList) if !this.IsMsgList && !this.IsColumnNewMsg && !this.IsColumn { //消息未读数统计 //获取总未读数 初始化 _, unread = m.CountUnread(this.UserId, true) if this.IsContainLetter { //私信统计 unread += unreadMsg(this) } data.Unread = unread return data } cquery := map[string]interface{}{ "receive_userid": this.UserId, "isdel": 1, "appid": this.Appid, } // p436 细化分类时需要用msg_type 进行查询 if this.MsgType > 0 && this.IsClassSearch { cquery["msg_type"] = this.MsgType } else if this.MsgType > 0 { cquery["group_id"] = this.MsgType } if this.Read != -1 { cquery["isRead"] = this.Read } //获取栏目下的数据 sData := make(map[string][]*message.Messages) t := time.Now() if this.IsColumnNewMsg && this.SortSize > 0 { var sortData *[]map[string]interface{} if this.IsClassSearch { // p436 增加 sortDataQ := fmt.Sprintf(`SELECT title,createtime,msg_type as group_id ,id FROM ( SELECT title,createtime,msg_type,id, ROW_NUMBER() OVER (PARTITION BY msg_type, receive_userid ORDER BY createtime DESC) AS row_num FROM message WHERE receive_userid = '%s' and isdel = 1 and appid = %s and group_id=%d ) AS message_ranked WHERE row_num <=%d;`, this.UserId, this.Appid, this.MsgType, this.SortSize) sortData = entity.Mysql.SelectBySql(sortDataQ) } else { sortData = entity.Mysql.SelectBySql(fmt.Sprintf(`SELECT title,createtime,group_id,id FROM ( SELECT title,createtime,group_id,id, ROW_NUMBER() OVER (PARTITION BY group_id, receive_userid ORDER BY createtime DESC) AS row_num FROM message WHERE receive_userid = '%s' and isdel = 1 and appid = %s ) AS message_ranked WHERE row_num <=%d;`, this.UserId, this.Appid, this.SortSize)) } log.Println("消息列表耗时1:", time.Since(t)) if sortData != nil { for _, v := range *sortData { _id := util.Int64All(v["id"]) id := strconv.FormatInt(_id, 10) var msg = message.Messages{ Id: id, Createtime: common.InterfaceToStr(v["createtime"]), Title: common.InterfaceToStr(v["title"]), MsgType: int64(util.IntAll(v["group_id"])), } if sData[common.InterfaceToStr(v["group_id"])] == nil { sData[common.InterfaceToStr(v["group_id"])] = []*message.Messages{&msg} } else { sData[common.InterfaceToStr(v["group_id"])] = append(sData[common.InterfaceToStr(v["group_id"])], &msg) } } } } // 消息栏目下的最新消息 var columnData []*message.AllSortData if this.IsColumn && this.MsgType > 0 && this.IsClassSearch { // p436 处理消息细分分类要返回的数据 // 获取小分类下的未读数 sortUnread, _ := m.CountClassUnread(this.UserId, this.MsgType) columnArr := []entity.MsgClass{} if !this.IsColumnNewMsg { // 用于区分分类列表页和分类详情页 根据不同情况 columnArr = append(columnArr, entity.ClassMap[this.MsgType]) } else { columnArr = entity.ClassSearchMap[this.MsgType] } for i := 0; i < len(columnArr); i++ { tmp := columnArr[i] var column message.AllSortData column.Name = tmp.Name column.Img = fmt.Sprintf("/common-module/msgCenter/%s.png", tmp.Img) column.MsgType = tmp.MsgType // 消息未读数 msgType := common.InterfaceToStr(tmp.MsgType) column.UnreadMessages = sortUnread[msgType] unread += sortUnread[msgType] column.Data = sData[msgType] column.IsClassSearch = true columnData = append(columnData, &column) } // 未读数量 } else if this.IsColumn { //获取所有分类未读数 不初始化 sortUnread, _ := m.CountUnread(this.UserId, false) for _, v := range entity.MessageColumn { var column message.AllSortData column.Name = common.InterfaceToStr(v["name"]) column.Img = fmt.Sprintf("/common-module/msgCenter/%s.png", common.InterfaceToStr(v["img"])) column.MsgType = common.Int64All(v["group_id"]) if column.Name == "私信" { column.UnreadMessages = unreadMsg(this) } else if common.IntAll(v["group_id"]) > 0 { //消息未读数 msgType := common.InterfaceToStr(v["group_id"]) column.UnreadMessages = sortUnread[msgType] unread += sortUnread[msgType] column.Data = sData[msgType] } // p436 该groupId属于展示细化分类的 如待办 点击待办进入到待参加会议、待处理任务中间列表页等 // 该返回值用于前端后续传参使用 if _, ok := entity.ClassSearchMap[column.MsgType]; ok { column.IsClassSearch = true } columnData = append(columnData, &column) } } data.SortData = columnData count = entity.Mysql.Count("message", cquery) if this.IsMsgList { if count > 0 { if this.OffSet <= 0 { this.OffSet = 1 } res := entity.Mysql.Find("message", cquery, "", "createtime desc", (int(this.OffSet)-1)*int(this.PageSize), int(this.PageSize)) log.Println("消息列表耗时3:", time.Since(t)) if res != nil && len(*res) > 0 { for _, v := range *res { _id := util.Int64All(v["id"]) id := strconv.FormatInt(_id, 10) data.Data = append(data.Data, &message.Messages{ Id: id, Appid: common.InterfaceToStr(v["appId"]), ReceiveUserId: common.InterfaceToStr(v["receive_userid"]), ReceiveName: common.InterfaceToStr(v["receive_name"]), SendUserId: common.InterfaceToStr(v["send_userid"]), SendName: common.InterfaceToStr(v["send_name"]), Createtime: common.InterfaceToStr(v["createtime"]), Title: common.InterfaceToStr(v["title"]), MsgType: int64(util.IntAll(v["group_id"])), Link: common.InterfaceToStr(v["link"]), CiteId: util.Int64All(v["cite_id"]), Content: common.InterfaceToStr(v["content"]), IsRead: util.Int64All(v["isRead"]), MsgLogId: util.Int64All(v["msg_log_id"]), }) } } } } data.Count = count if this.Read == 0 { unread = count if this.IsContainLetter { //是否需要统计私信未读数 unread += unreadMsg(this) } } data.Unread = unread return data } func unreadMsg(this *message.UserMsgListReq) int64 { if this.PositionId <= 0 { return 0 } querySql := fmt.Sprintf("SELECT b.*,(SELECT SUM( a.unread) FROM %s a "+ "LEFT JOIN %s b ON a.message_id = b.id "+ "WHERE a.unread > 0 "+ "AND ( a.my_position_id = %d OR a.user_id = %d )) AS unread "+ "FROM %s a "+ "LEFT JOIN %s b ON a.message_id = b.id "+ "WHERE a.unread > 0 "+ "AND ( a.my_position_id = %d OR a.user_id = %d ) "+ "ORDER BY a.TIMESTAMP DESC LIMIT 0,1", "socialize_summary", "socialize_message", this.PositionId, this.NewUserId, "socialize_summary", "socialize_message", this.PositionId, this.NewUserId) log.Println("查询sql", querySql) msgUnread := entity.BaseMysql.SelectBySql(querySql) if msgUnread != nil && len(*msgUnread) > 0 { return common.Int64All((*msgUnread)[0]["unread"]) } return 0 } func UserUnreadMsgList(this *message.UserUnreadMsgListReq) (int64, []*message.Messages) { count := 0 data := []*message.Messages{} types := entity.Mysql.Find("message_group", map[string]interface{}{}, `"group_id"`, "", -1, -1) if types != nil && len(*types) > 0 { for _, v := range *types { key := fmt.Sprintf(MsgCountKey, this.UserId, util.IntAll(v["group_id"])) if exists, _ := redis.Exists(redisModule, key); exists { count += redis.GetInt(redisModule, key) } } } if this.IsNeedData == 1 && count > 0 { query := map[string]interface{}{ "receive_userid": this.UserId, "isdel": 1, "appid": this.Appid, "isRead": 0, } res := entity.Mysql.Find("message", query, "", "createtime desc", (int(this.OffSet)-1)*int(this.PageSize), int(this.PageSize)) if res != nil && len(*res) > 0 { for _, val := range *res { _id := util.Int64All(val["id"]) id := strconv.FormatInt(_id, 10) links4 := common.InterfaceToStr(val["link"]) link4, androidUrl4, iosUrl4, weChatUrl4 := util.LinkSplit(links4) url := map[string]string{ "androidUrl": androidUrl4, "iosUrl": iosUrl4, "weChatUrl": weChatUrl4, } data = append(data, &message.Messages{ Id: id, Appid: common.InterfaceToStr(val["appid"]), ReceiveUserId: common.InterfaceToStr(val["receive_userid"]), ReceiveName: common.InterfaceToStr(val["receive_name"]), SendUserId: common.InterfaceToStr(val["send_userid"]), SendName: common.InterfaceToStr(val["send_name"]), Createtime: common.InterfaceToStr(val["createtime"]), Title: common.InterfaceToStr(val["title"]), MsgType: common.Int64All(val["group_id"]), Link: link4, CiteId: common.Int64All(val["cite_id"]), Content: common.InterfaceToStr(val["content"]), IsRead: common.Int64All(val["isRead"]), MsgLogId: common.Int64All(val["msg_log_id"]), Url: url, }) } } } return util.Int64All(count), data } func MessageGetLast(this *message.UserMsgListReq) *message.Messages { if !this.IsMsgList && !this.IsColumnNewMsg && !this.IsColumn { return nil } query := map[string]interface{}{ "receive_userid": this.UserId, "isdel": 1, "appid": this.Appid, "isRead": 0, "group_id": 1, } lastMsg := entity.Mysql.FindOne("message", query, "", "createtime desc") if lastMsg != nil && len(*lastMsg) > 0 { _id := util.Int64All((*lastMsg)["id"]) id := strconv.FormatInt(_id, 10) msg := message.Messages{ Id: id, Appid: common.InterfaceToStr((*lastMsg)["appid"]), ReceiveUserId: common.InterfaceToStr((*lastMsg)["receive_userid"]), ReceiveName: common.InterfaceToStr((*lastMsg)["receive_name"]), SendUserId: common.InterfaceToStr((*lastMsg)["send_userid"]), SendName: common.InterfaceToStr((*lastMsg)["send_name"]), Createtime: common.InterfaceToStr((*lastMsg)["createtime"]), Title: common.InterfaceToStr((*lastMsg)["title"]), MsgType: common.Int64All((*lastMsg)["group_id"]), Link: common.InterfaceToStr((*lastMsg)["link"]), CiteId: common.Int64All((*lastMsg)["cite_id"]), Content: common.InterfaceToStr((*lastMsg)["content"]), IsRead: common.Int64All((*lastMsg)["isRead"]), MsgLogId: common.Int64All((*lastMsg)["msg_log_id"]), } return &msg } return nil } // MsgCountAdd 消息未读数量加1 func MsgCountAdd(userId, appId string, msgType int64, msgClassType int64) bool { keyString := fmt.Sprintf(MsgCountKey, userId, msgType) classKeyString := fmt.Sprintf(MsgClassCountKey, userId, msgClassType) if exist, _ := redis.Exists(redisModule, classKeyString); exist { redis.Incr(redisModule, classKeyString) } exists, _ := redis.Exists(redisModule, keyString) if exists { in := redis.Incr(redisModule, keyString) FindUserMsg(message.FindUserMsgReq{ UserId: userId, Appid: appId, OffSet: 1, PageSize: 5, MsgType: -1, Read: 0, }, true) return in > 0 } return true } // MsgCountMinusOne 根据消息类型未读消息数量减1 func MsgCountMinusOne(userId, appId string, msgType int64, msgClassType int64) bool { classKeyString := fmt.Sprintf(MsgClassCountKey, userId, msgClassType) if exist, _ := redis.Exists(redisModule, classKeyString); exist { if redis.GetInt(redisModule, classKeyString) > 0 { redis.Decrby(redisModule, classKeyString, 1) } } keyString := fmt.Sprintf(MsgCountKey, userId, msgType) exists, _ := redis.Exists(redisModule, keyString) if exists { FindUserMsg(message.FindUserMsgReq{ UserId: userId, Appid: appId, OffSet: 1, PageSize: 5, MsgType: -1, Read: 0, }, true) if redis.GetInt(redisModule, keyString) <= 0 { return true } in := redis.Decrby(redisModule, keyString, 1) return in > 0 } return true } // MsgCountZero 把该消息类型未读数量置0 func MsgCountZero(userId, appId string, msgType int64) bool { if msgType > 0 && msgType < 999 { //全部私信不统计 keyString := fmt.Sprintf(MsgCountKey, userId, msgType) fool := redis.Put(redisModule, keyString, 0, -1) FindUserMsg(message.FindUserMsgReq{ UserId: userId, Appid: appId, OffSet: 1, PageSize: 5, MsgType: -1, Read: 0, }, true) return fool } return true } func MultSave(this message.MultipleSaveMsgReq) (int64, string) { userIdArr := strings.Split(this.UserIds, ",") userNameArr := strings.Split(this.UserNames, ",") positionIdArr := strings.Split(this.PositionIds, ",") if len(userIdArr) == 0 { return 0, "无效的用户id" } wg := &sync.WaitGroup{} var group_id int class := entity.Mysql.FindOne("message_class", map[string]interface{}{"msg_type": this.MsgType}, "group_id", "") if class != nil && len(*class) > 0 { group_id = util.IntAll((*class)["group_id"]) } //p459 特殊处理 传过来的消息内容格式为 消息内容#jy#微信模板项目名称#jy#服务地址 equityName, equityAddr := "", "" if this.MsgType == config.ConfigJson.EquityInfoMsgType { equityRs := strings.Split(this.Content, "#jy#") if len(equityRs) != 3 { log.Println("消息内容格式有误:", this.Content) return 0, "无效的消息内容格式" } this.Content = equityRs[0] equityName = equityRs[1] equityAddr = equityRs[2] } for i := 0; i < len(userIdArr); i++ { if userIdArr[i] == "" { continue } name := userNameArr[i] wg.Add(1) entity.SaveConcurrencyChan <- 1 var positionId int64 if len(positionIdArr) == len(userIdArr) { positionId = common.Int64All(positionIdArr[i]) } go func(v, userName string, positionId int64) { defer func() { <-entity.SaveConcurrencyChan wg.Done() }() //消息数组 nTime := time.Now().Format("2006-01-02 15:04:05") //c := entity.Mysql.Count("conversation", map[string]interface{}{"receive_id": v, "send_id": this.SendUserId}) 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, userName, this.SendUserId, this.SendName, this.Title, this.Content, this.MsgType, this.Link, nTime, this.MsgLogId, this.ShowBuoy, this.ShowContent, group_id) var in int64 /*if c <= 0 { sql1 := `INSERT INTO conversation(appid,secret_key,user_id,receive_id,receive_name,send_id,send_name,sort,createtime) values ('%s','','%s','%s','%s','%s','%s',0,'%s');` sql1 = fmt.Sprintf(sql1, this.Appid, this.SendUserId, v, userName, this.SendUserId, this.SendName, nTime) //插入会话表 in1 := entity.Mysql.InsertBySql(sql1) sql2 := `INSERT INTO conversation(appid,secret_key,user_id,receive_id,receive_name,send_id,send_name,sort,createtime) values ('%s','','%s','%s','%s','%s','%s',0,'%s');` sql2 = fmt.Sprintf(sql2, this.Appid, v, this.SendUserId, this.SendName, v, userName, nTime) in2 := entity.Mysql.InsertBySql(sql2) //插入消息表 in = entity.Mysql.InsertBySql(sql3, common.If(positionId != 0, positionId, nil)) logx.Info(in1, in2, in) if in1 > -1 && in2 > -1 && in > -1 { ok1 := MsgCountAdd(v, this.Appid, this.MsgType) if !ok1 { log.Println("存redis:", ok1, v) } } } else {*/ 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 pushConfig, err := GetWxTmplConfig(this.MsgType) if err != nil { logx.Error(fmt.Sprintf("SendWxTmplMsg uId %s Error %s", v, err.Error())) } p := &WxTmplPush{ Config: pushConfig, CustomWxTpl: this.CustomWxTpl, } 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 { if this.MsgType == config.ConfigJson.EquityInfoMsgType { // p459 服务地址特殊处理 err = p.SendMsg(this.WxPushUrl, this.Title, equityName, nTime, this.Row4, equityAddr) } else { 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 } //app推送 if this.MsgType != 10 { uData := p.GetUserPushInfo() 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())) } } } }(userIdArr[i], name, positionId) } wg.Wait() return 0, "" }