eta_api/services/edb_monitor/edb_monitor_message.go

package edbmonitor

import (
	"encoding/json"
	"errors"
	"eta/eta_api/global"
	"eta/eta_api/models"
	edbmonitor "eta/eta_api/models/edb_monitor"
	"eta/eta_api/models/edb_monitor/response"
	"eta/eta_api/utils"
	"fmt"
	"time"

	"github.com/rdlucklib/rdluck_tools/paging"
)

func ReadEdbMonitorMessage(messageId, adminId int) (msg string, err error) {
	message, err := edbmonitor.GetEdbMonitorMessageById(messageId)
	if err != nil {
		if utils.IsErrNoRow(err) {
			msg = "消息不存在"
			return
		}
		msg = "获取消息失败"
		return
	}
	if message.AdminId != adminId {
		msg = "您没有权限查看该消息"
		err = errors.New("no permission")
		return
	}
	message.IsRead = 1
	err = message.Update([]string{"IsRead"})
	if err != nil {
		msg = "已读失败"
		return
	}
	return
}

func ReadEdbMonitorMessageList(messageId []int, adminId int) (msg string, err error) {
	err = edbmonitor.BatchModifyEdbMonitorMessageIsRead(messageId, adminId)
	if err != nil {
		msg = "已读失败"
		return
	}
	return
}

// func EdbMonitorMessageHealth(adminId int) (isClose bool, err error) {
// 	conn := MonitorMessageConn[adminId]
// 	if conn == nil {
// 		err = errors.New("no connection")
// 		isClose = true
// 		return
// 	}
// 	_, msg, err := conn.ReadMessage()
// 	if err != nil {
// 		isClose = true
// 		return
// 	}
// 	if string(msg) == "ping" {
// 		healthKey := EDB_MONITOR_MESSAGE_CONNECT_CACHE + strconv.Itoa(adminId)
// 		err = utils.Rc.Put(healthKey, "1", time.Minute*1)
// 		if err != nil {
// 			return
// 		}
// 	}
// 	return
// }

func LogMessage(content, uniqueCode string, triggerTime time.Time, edbInfoId, edbInfoType, adminId, isRead, classifyId int) (err error) {
	message := &edbmonitor.EdbMonitorMessage{
		EdbInfoId:          edbInfoId,
		EdbInfoType:        edbInfoType,
		EdbUniqueCode:      uniqueCode,
		EdbClassifyId:      classifyId,
		AdminId:            adminId,
		IsRead:             isRead,
		Message:            content + "触发预警",
		MonitorTriggerTime: triggerTime,
		CreateTime:         time.Now(),
	}
	_, err = message.Insert()

	return err
}

func SendMessages(adminId, edbInfoId, edbInfoType int, classifyId int, edbUniqueCode, message string, triggerTime string) (err error) {
	conn := global.AdminWebSocketConnMap[adminId]
	if conn == nil {
		err = errors.New("no connection")
		return
	}
	resp := models.WebsocketMessageResponse{
		MessageType: 0,
		Data: response.EdbMonitorMessageResp{
			EdbInfoId:     edbInfoId,
			EdbInfoType:   edbInfoType,
			EdbUniqueCode: edbUniqueCode,
			EdbClassifyId: classifyId,
			Message:       message,
			TriggerTime:   triggerTime,
		},
	}
	jsonData, err := json.Marshal(resp)
	if err != nil {
		err = fmt.Errorf("json marshal failed, err:%s", err.Error())
		return
	}
	ok := conn.Send(jsonData)
	if !ok {
		err = fmt.Errorf("send message failed, err:%s", err.Error())
		return
	}
	return
}

func GetHistoryMessages(adminId int) (items []*response.EdbMonitorMessageResp, err error) {
	messageList, err := edbmonitor.GetEdbMonitorMessageByAdminId(adminId)
	if err != nil {
		return
	}

	items = toEdbMonitorMessageResp(messageList)
	return
}

func GetMessageList(adminid int, currentIndex, pageSize int) (resp response.EdbMonitorMessageListResp, err error) {
	startSize := utils.StartIndex(currentIndex, pageSize)

	total, err := edbmonitor.GetEdbMonitorMessageCountByAdminId(adminid)
	if err != nil {
		return
	}
	if total == 0 {
		resp.List = make([]*response.EdbMonitorMessageResp, 0)
		resp.Paging = paging.GetPaging(currentIndex, pageSize, total)
		return
	}

	messageList, err := edbmonitor.GetEdbMonitorMessagePageByAdminId(adminid, startSize, pageSize)
	if err != nil {
		return
	}

	resp.List = toEdbMonitorMessageResp(messageList)
	resp.Paging = paging.GetPaging(currentIndex, pageSize, total)
	return
}

func toEdbMonitorMessageResp(items []*edbmonitor.EdbMonitorMessage) (list []*response.EdbMonitorMessageResp) {
	list = make([]*response.EdbMonitorMessageResp, 0)
	for _, message := range items {
		item := response.EdbMonitorMessageResp{
			EdbMonitorMessageId: message.EdbMonitorMessageId,
			EdbInfoId:           message.EdbInfoId,
			EdbInfoType:         message.EdbInfoType,
			EdbUniqueCode:       message.EdbUniqueCode,
			EdbClassifyId:       message.EdbClassifyId,
			IsRead:              message.IsRead,
			Message:             message.Message,
			TriggerTime:         utils.TimeTransferString(utils.FormatDateTime, message.MonitorTriggerTime),
		}
		list = append(list, &item)
	}
	return
}

func AutoCheckMonitorMessageList() (err error) {
	defer func() {
		if err != nil {
			utils.FileLog.Error("指标预警信息发送失败,err:%s", err.Error())
		}
	}()
	ticker := time.NewTicker(time.Second * 10)
	defer ticker.Stop()

	for {
		select {
		case <-ticker.C:
			autoCheckMonitorMessageList()
		}
	}
}

func autoCheckMonitorMessageList() (err error){
	defer func() {
		if err != nil {
			utils.FileLog.Error("指标预警信息检查失败,err:%s", err.Error())
		}
	}()
	utils.FileLog.Info("检查是否有预警信息")
	admins := make([]int, 0)
	for adminId, conn := range global.AdminWebSocketConnMap {
		if conn == nil {
			continue
		}
		admins = append(admins, adminId)
	}
	if len(admins) == 0 {
		return
	}
	messageList, er := edbmonitor.GetEdbMonitorMessageUnreadListByAdminIds(admins)
	if er != nil {
		err = fmt.Errorf("获取指标预警信息历史失败,err:%s", er.Error())
		return
	}
	readList := make([]int, 0)
	adminMsgMap := make(map[int]int)
	for _, msg := range messageList {
		if _, ok := adminMsgMap[msg.AdminId]; !ok {
			adminMsgMap[msg.AdminId] = msg.EdbMonitorMessageId
			triggerTime := utils.TimeTransferString(utils.FormatDateTime, msg.MonitorTriggerTime)
			err := SendMessages(msg.AdminId, msg.EdbInfoId, msg.EdbInfoType, msg.EdbClassifyId, msg.EdbUniqueCode, msg.Message, triggerTime)
			if err != nil {
				utils.FileLog.Error("指标预警信息发送失败,err:%s, adminId:%d", err.Error(), msg.AdminId)
			} 
		}else {
			readList = append(readList, msg.EdbMonitorMessageId)
		}
	}

	err = edbmonitor.SetEdbMonitorMessageReadByIds(readList)
	if err != nil {
		err = fmt.Errorf("指标预警信息已读失败,err:%s", err.Error())
		return
	}
	return
}

func AutoCheckMonitorMessageListByAdminId(adminId int) (err error){
	// 设置缓存防止重复发送
	cacheKey := fmt.Sprintf("%s%d", utils.CACHE_EDB_MONITOR_MESSAGE, adminId)
	if !utils.Rc.SetNX(cacheKey, 1, 10*time.Minute) {
		err = fmt.Errorf("系统处理中，请稍后重试！")
		utils.FileLog.Error("指标预警信息检查失败,err:%s", err.Error())
		return
	}
	defer func() {
		if err != nil {
			utils.FileLog.Error("指标预警信息检查失败,err:%s", err.Error())
		}
		_ = utils.Rc.Delete(cacheKey)
	}()
	utils.FileLog.Info("检查是否有预警信息")
	messageList, er := edbmonitor.GetEdbMonitorMessageUnreadByAdminId(adminId)
	if er != nil {
		err = fmt.Errorf("获取指标预警信息历史失败,err:%s", er.Error())
		return
	}
	if len(messageList) == 0 {
		return
	}
	readList := make([]int, 0)
	for k, msg := range messageList {
		if k == 0 {
			triggerTime := utils.TimeTransferString(utils.FormatDateTime, msg.MonitorTriggerTime)
			err := SendMessages(msg.AdminId, msg.EdbInfoId, msg.EdbInfoType, msg.EdbClassifyId, msg.EdbUniqueCode, msg.Message, triggerTime)
			if err != nil {
				utils.FileLog.Error("指标预警信息发送失败,err:%s, adminId:%d", err.Error(), msg.AdminId)
			} 
		}
		readList = append(readList, msg.EdbMonitorMessageId)
	}

	err = edbmonitor.SetEdbMonitorMessageReadByAdminId(adminId, readList)
	if err != nil {
		err = fmt.Errorf("指标预警信息已读失败,err:%s", err.Error())
		return
	}
	return
}