eta_api/services/websocket_msg.go

package services

import (
	"encoding/json"
	"eta/eta_api/global"
	"eta/eta_api/models"
	"eta/eta_api/models/data_manage/edb_inspection"
	"eta/eta_api/services/data"
	edb_monitor "eta/eta_api/services/edb_monitor"
	"eta/eta_api/utils"
	"fmt"
	"runtime"
	"sync"
	"time"

	"context"
)

func DealWebSocketMsg(adminId int) {
	DealEdbInspectionMessageTest(adminId)

	//go DealEdbInspectionMessage(adminId)
}

// 处理巡检消息
func DealEdbInspectionMessage(adminId int) {
	utils.FileLog.Info("创建协程, adminId:%d", adminId)
	// 创建上下文用于控制 goroutine 生命周期
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	cacheKey := fmt.Sprintf("%s%d", utils.CACHE_EDB_INSPECTION_MESSAGE, adminId)

	// 添加错误恢复机制
	defer func() {
		if r := recover(); r != nil {
			utils.FileLog.Error("WebSocket handler recovered from panic: %v", r)
			// 清理资源
			cancel()
		}
	}()
	go func() {
		ticker := time.NewTicker(time.Minute)
		defer ticker.Stop()
		for {
			select {
			case <-ticker.C:
				utils.FileLog.Info("Current goroutine count: %d", runtime.NumGoroutine())
			case <-ctx.Done():
				return
			}
		}
	}()
	for {
		select {
		case <-ctx.Done():
			utils.FileLog.Info("DealEdbInspectionMessage 巡检消息处理协程结束, adminId:%d", adminId)
			return
		default:
			// 检查连接状态
			conn := global.MonitorMessageConn[adminId]
			if conn == nil {
				utils.FileLog.Error("检查连接状态 发送消息时发现连接已断开, adminId:%d", adminId)
				cancel()
				return
			}
			// 使用带超时的 Redis 操作
			val := utils.Rc.Get(cacheKey)
			if val == "" {
				//utils.FileLog.Info("巡检信息历史为空, adminId:%d", adminId)
				continue
			}
				utils.FileLog.Info("收到巡检信息开始处理, adminId:%d", adminId)
				messageList, err := data.GetHistoryInspectionMessages(adminId)
				if err != nil {
					utils.FileLog.Error("获取巡检信息历史失败,err:%s, adminId:%d", err.Error(), adminId)
					return
				}
				if len(messageList) == 0 {
					utils.FileLog.Info("巡检信息历史为空, adminId:%d", adminId)
					return
				}

				readList := make([]int64, 0)
				// 检查连接状态
				// conn := global.MonitorMessageConn[adminId]
				// if conn == nil {
				// 	utils.FileLog.Error("发送消息时发现连接已断开, adminId:%d", adminId)
				// 	cancel()
				// 	return
				// }
				// 只处理第一条消息的发送，其他消息只标记为已读
				for i, msg := range messageList {
					if i == 0 {
						respData, err := data.SendInspectionMessages(adminId, msg)
						if err != nil {
							utils.FileLog.Error("巡检信息发送失败,err:%s, adminId:%d", err.Error(), adminId)
							continue
						}

						resp := models.WebsocketMessageResponse{
							MessageType: 1,
							Data:       respData,
						}

						err, isClose := WriteWebSocketMessageAsync(ctx, adminId, resp)
						if err != nil {
							utils.FileLog.Error("巡检信息发送失败,err:%s, adminId:%d", err.Error(), adminId)
							cancel()
							continue
						}
						if isClose {
							utils.FileLog.Error("巡检信息发送失败,连接已断开, adminId:%d", adminId)
							cancel()
							return
						}

						utils.FileLog.Info("巡检信息发送成功,adminId:%d, messageId:%d", adminId, msg.MessageId)
					}
					readList = append(readList, msg.MessageId)
				}

				if len(readList) > 0 {
					_, err = data.ReadEdbInspectionMessageList(readList, adminId)
					if err != nil {
						utils.FileLog.Error("巡检信息已读失败,err:%s, adminId:%d", err.Error(), adminId)
					}
				}
			//})

			// if err != nil && err.Error() != "redis: nil" {
			// 	utils.FileLog.Error("Redis operation failed: %v", err)
			// 	continue
			// }else {
			// 	utils.FileLog.Info("巡检信息处理完成, adminId:%d", adminId)
			// }
		}
	}
}

func WriteWebSocketMessageAsync(ctx context.Context, adminId int, resp interface{}) (error, bool) {
	errChan := make(chan error, 1)
	var wsWriteMutex sync.Mutex
	isClose := false
	
	go func() {
		wsWriteMutex.Lock()
		defer wsWriteMutex.Unlock()
		
		conn := global.MonitorMessageConn[adminId]
		if conn == nil {
			isClose = true
			errChan <- fmt.Errorf("connection closed for adminId: %d", adminId)
			return
		}
		
		// 设置写超时
		//conn.SetWriteDeadline(time.Now().Add(10 * time.Second))
		errChan <- conn.WriteJSON(resp)
	}()
	
	select {
	case err := <-errChan:
		utils.FileLog.Error("WriteWebSocketMessageAsync errChan: %v", err)
		return err, isClose
	case <-ctx.Done():
		utils.FileLog.Error("WriteWebSocketMessageAsync ctx.Done(): %v", ctx.Err())
		return ctx.Err(), isClose
	}
}

func DealEdbInspectionMessageTest(adminId int) {
	messageList, err := data.GetHistoryInspectionMessages(adminId)
	if err != nil {
		utils.FileLog.Error("获取巡检信息历史失败,err:%s, adminId:%d", err.Error(), adminId)
	}
	success := make(chan int64, 10)
	go func() {
		defer close(success)
		for i, msg := range messageList {
			if i == 0 {
				// 多条消息仅发送最新一条
				respData, err := data.SendInspectionMessages(adminId, msg)
				if err != nil {
					utils.FileLog.Error("巡检信息发送失败,err:%s, adminId:%d", err.Error(), adminId)
				} else {
					resp := models.WebsocketMessageResponse{
						MessageType: 1,
						Data: respData,
					}
					conn := global.MonitorMessageConn[adminId]
					if conn == nil {
						utils.FileLog.Error("巡检信息发送失败,连接已断开, adminId:%d", adminId)
						return
					}
					err = conn.WriteJSON(resp)
					if err != nil {
						utils.FileLog.Error("巡检信息发送失败,err:%s, adminId:%d", err.Error(), adminId)
						} else {
							utils.FileLog.Info("巡检信息发送成功,adminId:%d, messageId:%d", adminId, msg.MessageId)
							success <- msg.MessageId
						}
					}
				} else {
					success <- msg.MessageId
				}
			}
	}()
	go func() {
		readList := make([]int64, 0)
		for {
			msgId, ok := <-success
			if !ok {
				break
			}
			readList = append(readList, msgId)
		}
		_, err = data.ReadEdbInspectionMessageList(readList, adminId)
		if err != nil {
			utils.FileLog.Error("巡检信息已读失败,err:%s, adminId:%d", err.Error(), adminId)
		}
	}()
}

func AutoCheckWebsocketMessageList() {
	ticker := time.NewTicker(time.Second * 10)
	defer ticker.Stop()

	for {
		select {
		case <-ticker.C:
			// 获取活跃连接
			admins := make([]int, 0)
			for adminId, conn := range global.AdminWebSocketConnMap {
				if conn == nil {
					continue
				}
				admins = append(admins, adminId)
			}
			
			// 如果没有活跃连接，继续等待下一个tick
			if len(admins) == 0 {
				//utils.FileLog.Info("当前没有活跃的WebSocket连接")
				continue
			}

			// 并发处理不同类型的消息
			var wg sync.WaitGroup
			wg.Add(2)

			// 处理巡检消息
			go func() {
				defer wg.Done()
				if err := AutoCheckInspectionMessageList(admins); err != nil {
					utils.FileLog.Error("处理巡检消息失败: %v", err)
				}
			}()

			// 处理监控消息
			go func() {
				defer wg.Done()
				if err := edb_monitor.AutoCheckMonitorMessageList(admins); err != nil {
					utils.FileLog.Error("处理监控消息失败: %v", err)
				}
			}()

			// 等待所有消息处理完成
			wg.Wait()
		}
	}
}

func AutoCheckInspectionMessageList(admins []int) (err error) {
	//utils.FileLog.Info("检查是否有巡检信息")
	// 设置redis缓存，防止消息重复处理
	cacheKey := fmt.Sprintf("%s", utils.CACHE_EDB_INSPECTION_MESSAGE)
	if !utils.Rc.SetNX(cacheKey, 1, 30*time.Second) {
		err = fmt.Errorf("系统处理中，请稍后重试！")
		utils.FileLog.Error("巡检信息检查失败,err:%s", err.Error())
		return
	}
	defer func() {
		_ = utils.Rc.Delete(cacheKey)
	}()
	messageList, er := edb_inspection.GetUnreadInspectionMessageList(admins)
	if er != nil {
		err = fmt.Errorf("获取巡检信息历史失败,err:%s", er.Error())
		return
	}
	readList := make([]int64, 0)
	for _, msg := range messageList {
		adminId := int(msg.AdminId)
		respData, er := data.SendInspectionMessages(adminId, msg)
		if er != nil {
			utils.FileLog.Error("巡检信息发送失败,err:%s, adminId:%d", er.Error(), adminId)
		} else {
			resp := models.WebsocketMessageResponse{
				MessageType: 1,
				Data: respData,
			}
			conn := global.AdminWebSocketConnMap[int(msg.AdminId)]
			if conn == nil {
				utils.FileLog.Error("巡检信息发送失败,连接已断开, adminId:%d", adminId)
				return
			}
			message, er := json.Marshal(resp)
			if er != nil {
				utils.FileLog.Error("巡检信息发送失败,err:%s, adminId:%d", er.Error(), adminId)
				return
			}
			ok := conn.Send(message)
			if !ok {
				utils.FileLog.Error("巡检信息发送失败,err:%s, adminId:%d",  adminId)
			}
		}
		readList = append(readList, msg.MessageId)
	}

	err = edb_inspection.SetEdbInspectionMessageRead(readList)
	if err != nil {
		err = fmt.Errorf("巡检信息已读失败,err:%s", err.Error())
		return
	}
	return
}

// 弃用
func AutoCheckInspectionMessageListByAdminId(adminId int) (err error) {
	defer func() {
		if err != nil {
			utils.FileLog.Error("巡检信息发送失败,err:%s", err.Error())
		}
	}()
	messageList, er := edb_inspection.GetUnreadInspectionMessageListByAdminId(adminId)
	if er != nil {
		err = fmt.Errorf("获取巡检信息历史失败,err:%s", er.Error())
		return
	}
	if len(messageList) == 0 {
		return
	}
	readList := make([]int64, 0)
	for _, msg := range messageList {
		adminId := int(msg.AdminId)
		respData, er := data.SendInspectionMessages(adminId, msg)
		if er != nil {
			utils.FileLog.Error("巡检信息发送失败,err:%s, adminId:%d", er.Error(), adminId)
		} else {
			resp := models.WebsocketMessageResponse{
				MessageType: 1,
				Data: respData,
			}
			conn := global.AdminWebSocketConnMap[adminId]
			if conn == nil {
				utils.FileLog.Error("巡检信息发送失败,连接已断开, adminId:%d", adminId)
				return
			}
			message, er := json.Marshal(resp)
			if er != nil {
				utils.FileLog.Error("巡检信息发送失败,err:%s, adminId:%d", er.Error(), adminId)
				return
			}
			ok := conn.Send(message)
			if !ok {
				utils.FileLog.Error("巡检信息发送失败,err:%s, adminId:%d",  adminId)
			}
		}
		readList = append(readList, msg.MessageId)
	}
	err = edb_inspection.SetEdbInspectionMessageRead(readList)
	if err != nil {
		err = fmt.Errorf("巡检信息已读失败,err:%s", err.Error())
		return
	}
	return
}