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_%s" //redis 消息未读数量 Count.用户id.消息类型=数量
const redisModule = "msgCount"

/*
var (

	UserLockMap = map[string]*sync.Mutex{}
	//MainLock    = sync.Mutex{}

)
*/

func SendMsg(this message.SendMsgRequest) (int64, string) {
	r, err := entity.Mysql11.Query("select count(*) as c from conversation where receive_id = ? and send_id = ? ", this.ReceiveUserId, this.SendUserId)
	c := 0
	for r.Next() {
		err := r.Scan(&c)
		if err != nil {
			panic(err.Error())
		}
	}
	logx.Info("查询数量：", c)

	sql3 := `INSERT INTO message(appid,receive_userid,receive_name,send_userid,send_name,title,content,msg_type,link,cite_id,createtime,isRead,isdel)
		values ("%s",'%s','%s','%s','%s','%s','%s','%d','%s',0,'%s',0,1);`
	sql3 = fmt.Sprintf(sql3, this.Appid, this.ReceiveUserId, this.ReceiveName, this.SendUserId, this.SendName, this.Title, this.Content, this.MsgType, this.Link, time.Now().Format("2006-01-02 15:04:05"))
	if c < 1 {
		sql1 := `INSERT INTO conversation(appid,` + "`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, this.ReceiveUserId, this.ReceiveName, this.SendUserId, this.SendName, time.Now().Format("2006-01-02 15:04:05"))
		ok := entity.Mysql.ExecTx("发送消息事务", func(tx *sql.Tx) bool {
			//插入会话表
			_, err := entity.Mysql11.Exec(sql1)

			sql2 := `INSERT INTO conversation(appid,` + "`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, this.ReceiveUserId, this.SendUserId, this.SendName, this.ReceiveUserId, this.ReceiveName, time.Now().Format("2006-01-02 15:04:05"))
			_, err = entity.Mysql11.Exec(sql2)
			//插入消息表
			_, err = entity.Mysql11.Exec(sql3)
			if err != nil {
				return false
			}
			return true
		})
		if ok {
			return 1, "消息发送成功"
		}
	}
	_, err = entity.Mysql11.Exec(sql3)
	if err == nil {
		MsgCountAdd(this.ReceiveUserId, strconv.Itoa(int(this.MsgType)), this.Appid)
		return 1, "消息发送成功"
	}
	return 0, "消息发送失败"
}
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
}

// this.IsPc 是否pc端请求 this.MobileHome是否移动消息列表页
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
	if this.IsColumn {
		var msgTypes []string
		columnUnread := make(map[string]int64)
		for _, v := range entity.MessageColumn {
			msgTypes = append(msgTypes, fmt.Sprintf(`"%s"`, common.InterfaceToStr(v["msg_type"])))
		}
		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 = common.InterfaceToStr(v["img"])
			column.MsgType = common.Int64All(v["msg_type"])
			msgType := common.InterfaceToStr(v["msg_type"])
			//消息未读数
			column.UnreadMessages = columnUnread[msgType]
			unread += column.UnreadMessages
			column.Data = sData[msgType]
			columnData = append(columnData, &column)
		}
	}

	if this.IsMsgList {
		data.SortData = columnData
		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))
			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
	}
	data.Unread = unread
	return data
}

func UserUnreadMsgList(this *message.UserUnreadMsgListReq) (int64, []*message.Messages) {
	var count int64
	var data []*message.Messages
	cquery := map[string]interface{}{
		"receive_userid": this.UserId,
		"isdel":          1,
		"appid":          this.Appid,
		"isRead":         0,
	}
	log.Println(cquery)
	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))
		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 count, data
}

func MessageGetLast(this *message.UserMsgListReq) *message.Messages {
	if this.IsColumn {
		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
}

// 指定分类未读消息合计
func ClassCountUnread(msgType int, userId string, appId string) (int64, string, int64) {
	query := map[string]interface{}{
		"msg_type":       msgType,
		"receive_userid": userId,
		"isdel":          1,
		"appid":          appId,
		"isRead":         0,
	}
	count := entity.Mysql.Count("message", query)
	return 1, "查询指定分类未读消息成功", count
}

// MsgCountAdd 消息未读数量加1
func MsgCountAdd(userId, msgType, appId string) 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, msgType, appId string) 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, msgType, appId string) 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, strconv.Itoa(int(this.MsgType)), this.Appid)
					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, strconv.Itoa(int(this.MsgType)), this.Appid)
					if !ok {
						log.Println("存redis：", ok, v)
					}
				}
			}
		}(userIdArr[i], name, positionId)
	}
	wg.Wait()
	return 0, ""
}