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" "app.yhyue.com/moapp/jybase/redis" "database/sql" "fmt" "github.com/zeromicro/go-zero/core/logx" "log" "strconv" "strings" "sync" "time" ) // 类型的顺序 const order = "1,4" const MsgCountKey = "count_%s_%d" //redis 消息未读数量 Count.用户id.消息类型=数量 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 != -1 { cquery["msg_type"] = 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["msg_type"])), 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 ) cquery := map[string]interface{}{ "receive_userid": this.UserId, "isdel": 1, "appid": this.Appid, } if this.MsgType > 0 { cquery["msg_type"] = this.MsgType } if this.Read != -1 { cquery["isRead"] = this.Read } data := new(message.UserMsgList) //获取栏目下的数据 sData := make(map[string][]*message.Messages) if this.IsColumnNewMsg && this.SortSize > 0 { sortData := entity.Mysql.SelectBySql(fmt.Sprintf(`SELECT * FROM ( SELECT *, 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 ) AS message_ranked WHERE row_num <=%d;`, this.UserId, this.Appid, this.SortSize)) if sortData != nil { for _, v := range *sortData { _id := util.Int64All(v["id"]) id := strconv.FormatInt(_id, 10) var msg = 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["msg_type"])), 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"]), } if sData[common.InterfaceToStr(v["msg_type"])] == nil { sData[common.InterfaceToStr(v["msg_type"])] = []*message.Messages{&msg} } else { sData[common.InterfaceToStr(v["msg_type"])] = append(sData[common.InterfaceToStr(v["msg_type"])], &msg) } } } } //消息栏目下的最新消息 var columnData []*message.AllSortData var msgTypes []string for _, v := range entity.MessageColumn { if common.IntAll(v["msg_type"]) > 0 { msgTypes = append(msgTypes, fmt.Sprintf(`"%s"`, common.InterfaceToStr(v["msg_type"]))) } } if this.IsColumn { columnUnread := make(map[string]int64) query := entity.Mysql.SelectBySql(fmt.Sprintf("SELECT msg_type,COUNT(CASE WHEN isRead=0 THEN 1 END) as count FROM message where receive_userid=? and isdel=1 and appid=? GROUP BY msg_type ORDER BY FIELD(`msg_type`,%s)", strings.Join(msgTypes, ",")), this.UserId, this.Appid) if query != nil && len(*query) > 0 { for _, v := range *query { columnUnread[common.InterfaceToStr(v["msg_type"])] = common.Int64All(v["count"]) } } 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["msg_type"]) if column.Name == "私信" { column.UnreadMessages = unreadMsg(this) } else { //消息未读数 msgType := common.InterfaceToStr(v["msg_type"]) column.UnreadMessages = columnUnread[msgType] column.Data = sData[msgType] } unread += column.UnreadMessages 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)) 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["msg_type"])), 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.IsColumn && 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_column", map[string]interface{}{}, `"msg_type"`, "", -1, -1) if types != nil && len(*types) > 0 { for _, v := range *types { key := fmt.Sprintf(MsgCountKey, this.UserId, util.IntAll(v["msg_type"])) 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, } log.Println(query) 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["msg_type"]), 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 { query := map[string]interface{}{ "receive_userid": this.UserId, "isdel": 1, "appid": this.Appid, "isRead": 0, "msg_type": 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)["msg_type"]), 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) bool { keyString := fmt.Sprintf(MsgCountKey, userId, msgType) in := redis.Incr(redisModule, keyString) FindUserMsg(message.FindUserMsgReq{ UserId: userId, Appid: appId, OffSet: 1, PageSize: 5, MsgType: -1, Read: 0, }, true) return in > 0 } // MsgCountMinusOne 根据消息类型未读消息数量减1 func MsgCountMinusOne(userId, appId string, msgType int64) bool { keyString := fmt.Sprintf(MsgCountKey, userId, msgType) FindUserMsg(message.FindUserMsgReq{ UserId: userId, Appid: appId, OffSet: 1, PageSize: 5, MsgType: -1, Read: 0, }, true) if redis.GetInt(redisModule, keyString) <= 0 { return redis.Put(redisModule, keyString, 0, -1) } in := redis.Decrby(redisModule, keyString, 1) return in > 0 } // MsgCountZero 把该消息类型未读数量置0 func MsgCountZero(userId, appId string, msgType int64) bool { 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 } 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{} 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() }() //消息数组 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,position_id) values ("%s",'%s','%s','%s','%s','%s','%s',%d,'%s',0,'%s',0,1,%d,%d,'%s',?);` sql3 = fmt.Sprintf(sql3, this.Appid, v, userName, this.SendUserId, this.SendName, this.Title, this.Content, this.MsgType, this.Link, time.Now().Format("2006-01-02 15:04:05"), this.MsgLogId, this.ShowBuoy, this.ShowContent) 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, time.Now().Format("2006-01-02 15:04:05")) ok := entity.Mysql.ExecTx("发送消息事务", func(tx *sql.Tx) bool { //插入会话表 in1 := entity.Mysql.InsertBySqlByTx(tx, 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, time.Now().Format("2006-01-02 15:04:05")) in2 := entity.Mysql.InsertBySqlByTx(tx, sql2) //插入消息表 in3 := entity.Mysql.InsertBySqlByTx(tx, sql3, common.If(positionId != 0, positionId, nil)) logx.Info(in1, in2, in3) return in1 > -1 && in2 > -1 && in3 > -1 }) logx.Info("执行事务是否成功:", ok) if ok { 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) if in > -1 { ok := MsgCountAdd(v, this.Appid, this.MsgType) if !ok { log.Println("存redis:", ok, v) } } } }(userIdArr[i], name, positionId) } wg.Wait() return 0, "" }