MessageCenter/rpc/internal/common/sendMsg.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"
	"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("/images/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 {
		if this.Read == 0 {
			unread = count
			if !this.IsMsgList && !this.IsColumnNewMsg {
				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, ""
}