MessageCenter/rpc/internal/common/messageService.go

package common

import (
	"app.yhyue.com/moapp/MessageCenter/entity"
	"app.yhyue.com/moapp/MessageCenter/rpc/type/message"
	"app.yhyue.com/moapp/MessageCenter/util"
	qutil "app.yhyue.com/moapp/jybase/common"
	"app.yhyue.com/moapp/jybase/redis"
	"context"
	"errors"
	"fmt"
	"log"
	"strings"
	"time"
)

type MessageService struct{}

// 修改消息阅读状态
func (service *MessageService) ChangeReadStatus(data *message.ChangeReadStatusReq) error {
	msg := entity.Mysql.FindOne("message_send_log", map[string]interface{}{"id": data.Id}, "group_id", "")
	if msg != nil && len(*msg) > 0 {
		groupId := qutil.IntAll((*msg)["group_id"])
		//更新用户未读消息bitmap
		sql := fmt.Sprintf(`alter table message_user_summary UPDATE readMsg = bitmapOr(readMsg,bitmapBuild([toUInt64(%d)])) where userId = '%s'`, data.Id, data.UserId)
		fmt.Println(sql)
		err1 := entity.ClickhouseConn.Exec(context.Background(), sql)
		if err1 != nil {
			return err1
		}
		//清缓存
		keyString := fmt.Sprintf(MsgCountKey, data.UserId, groupId)
		if redis.GetInt(redisModule, keyString) > 0 {
			redis.Decrby(redisModule, keyString, 1)
		}
	} else {
		return errors.New(fmt.Sprintf("消息不存在：%d", data.Id))
	}
	return nil
}

// 未读消息合计 isRedis 是否需要初始化redis
func (service *MessageService) CountUnread(userId string, isRedis bool) (map[string]int64, int64) {
	var (
		count              int64
		msgTypes, groupIds []string
	)
	data := make(map[string]int64)
	for _, v := range entity.MessageColumn {
		if util.IntAll(v["group_id"]) > 0 && util.IntAll(v["group_id"]) < 999 {
			//去除全部与私信
			msgTypes = append(msgTypes, fmt.Sprintf(`"%s"`, qutil.InterfaceToStr(v["group_id"])))
			key := fmt.Sprintf(MsgCountKey, userId, util.IntAll(v["group_id"]))
			groupIds = append(groupIds, qutil.InterfaceToStr(v["group_id"]))
			if exists, _ := redis.Exists(redisModule, key); exists {
				ct := util.Int64All(redis.GetInt(redisModule, key))
				data[qutil.InterfaceToStr(v["group_id"])] = ct
				count += ct
			}
		}
	}
	if len(msgTypes) > 0 && len(msgTypes) != len(data) {
		count = 0
		query := entity.Mysql.SelectBySql(fmt.Sprintf("SELECT group_id,COUNT(CASE WHEN isRead=0 THEN 1 END) as count  FROM message where receive_userid=? and isdel=1 and group_id IS NOT NULL GROUP BY group_id  ORDER BY FIELD(`group_id`,%s)", strings.Join(msgTypes, ",")), userId)
		if query != nil && len(*query) > 0 {
			for _, v := range *query {
				unread := qutil.Int64All(v["count"])
				data[qutil.InterfaceToStr(v["group_id"])] = unread
				count += unread
			}
		}
		if isRedis { //初始化未读数
			for _, v1 := range groupIds {
				key := fmt.Sprintf(MsgCountKey, userId, qutil.IntAll(v1))
				redis.Put(redisModule, key, data[v1], -1)
			}
		}
	}

	return data, count
}
func (service *MessageService) CountClassUnread(userId string, groupId int64) (classCount map[string]int64, total int64) {
	var (
		count int64
	)
	data := make(map[string]int64)
	if _, ok := entity.ClassSearchMap[groupId]; !ok {
		return
	}
	for i := 0; i < len(entity.ClassSearchMap[groupId]); i++ {
		msgClass := entity.ClassSearchMap[groupId][i]
		key := fmt.Sprintf(MsgClassCountKey, userId, msgClass.MsgType)
		if exists, _ := redis.Exists(redisModule, key); exists {
			ct := util.Int64All(redis.GetInt(redisModule, key))
			data[fmt.Sprintf("%d", msgClass.MsgType)] = ct
			count += ct
		} else {
			q := "select count(*) from message where receive_userid=? and isdel=1 and msg_type=?"
			classCount := entity.Mysql.CountBySql(q, userId, msgClass.MsgType)
			if classCount != -1 {
				redis.Put(redisModule, key, classCount, -1)
				data[fmt.Sprintf("%d", msgClass.MsgType)] = classCount
				count += classCount
			} else {
				log.Println("查询classCount失败：", classCount, q)
			}
		}
	}

	return data, count
}

// 查询消息详情
func FindMessageDetail(id, msgLogId int64, userId string) (msg *map[string]interface{}, err error) {
	//直接查询message_send_log
	msg = entity.Mysql.FindOne("message_send_log", map[string]interface{}{"id": msgLogId}, "", "")
	if msg != nil && len(*msg) > 0 {
		return msg, nil
	}
	return nil, errors.New("没有查询到消息")
}

// GetMsgType 消息的分类
/*func (service *MessageService) GetMsgType() (data []*message.MsgTypes, err error) {
	types := entity.Mysql.SelectBySql("SELECT * FROM `message_group` WHERE group_id > 0 ORDER BY sequence ASC")
	if types != nil && len(*types) > 0 {
		for _, val := range *types {
			data = append(data, &message.MsgTypes{
				MsgType:         qutil.Int64All(val["group_id"]),
				Name:            qutil.ObjToString(val["name"]),
				Img:             qutil.ObjToString(val["img"]),
				Code:            qutil.ObjToString(val["switch"]),
				DisplayPlatform: qutil.ObjToString(val["display_platform"]),
			})
		}
		return data, nil
	}
	return nil, nil
}*/

func (service *MessageService) MsgOpenLog(platFrom, msgLogId int64, userId string) error {
	//判断用户是否已经在pc端打开过
	sql := fmt.Sprintf("SELECT COUNT(*) FROM message_open_log WHERE msg_log_id = %d and platform = %d and userid = '%s'", msgLogId, platFrom, userId)
	row := entity.ClickhouseConn.QueryRow(context.Background(), sql)
	var count uint64
	row.Scan(&count)
	if count <= 0 {
		tmp := map[string]interface{}{
			"msg_log_id": msgLogId,
			"platform":   platFrom,
			"userid":     userId,
			"createtime": time.Now().Format("2006-01-02 15:04:05"),
		}
		SaveCache <- tmp
	}
	return nil
}