Message-Push-Nest/service/cron_msg_service/startup_cron_msg.go

package cron_msg_service

import (
	"fmt"
	"message-nest/models"
	"message-nest/pkg/constant"
	"message-nest/service/cron_service"
	"message-nest/service/send_message_service"
	"strings"

	"github.com/sirupsen/logrus"
)

type MsgCronTask struct {
}

func (s MsgCronTask) Register() {
	// 获取所有的定时消息任务
	limit := 10000
	filter := make(map[string]interface{})
	filter["enable"] = 1
	data, err := models.GetCronMessages(0, limit, "", filter)
	if err != nil {
		logrus.Errorf("获取定时消息任务失败！原因：%s", err.Error())
		return
	}
	if len(data) == 0 {
		logrus.Infof("没有定时消息任务需要注册")
		return
	}
	//注册定时任务
	for _, msg := range data {
		AddCronMsgToCronServer(msg)
	}
	length := len(data)
	if length > 0 {
		logrus.Infof("完成用户自定义的定时消息注册，注册个数：%d", length)
	}
}

// AddCronMsgToCronServer 注册定时任务到定时服务
func AddCronMsgToCronServer(msg models.CronMessages) {
	if msg.Enable != 1 {
		return
	}
	taskId := cron_service.AddTask(cron_service.ScheduledTask{
		Schedule: msg.Cron,
		Job: func() {
			CronMsgSendF(msg)
		},
	})
	constant.CronMsgIdMapMemoryCache[msg.ID] = taskId
	logrus.Infof("新增定时消息成功，消息id: %s，消息名: %s，当前任务总数：%d", msg.ID, msg.Name, len(constant.CronMsgIdMapMemoryCache))
}

// 执行任务的构造函数
func CronMsgSendF(msg models.CronMessages) {
	logrus.Infof("开始只能执行定时消息发送任务: %s，消息名: %s", msg.ID, msg.Name)
	task, err := models.GetTaskByID(msg.TaskID)
	if err != nil {
		logrus.Infof("消息任务不存在: %s ", msg.TaskID)
		return
	}
	sender := send_message_service.SendMessageService{
		TaskID:   task.ID,
		SendMode: "task",
		Title:    msg.Title,
		Text:     msg.Content,
		URL:      msg.Url,
		CallerIp: fmt.Sprintf("[CrondTask] [%s] ID: %s", task.Name, task.ID),
		DefaultLogger: logrus.WithFields(logrus.Fields{
			"prefix": "[Cron Message]",
		}),
	}
	taskData, _ := sender.SendPreCheck()
	_, err = sender.Send(taskData)
	if err != nil {
		logrus.Error("执行定时消息失败：%s", err.Error())
		return
	}
}

// UpdateCronMsgToCronServer 更新定时服务的任务
func UpdateCronMsgToCronServer(msg models.CronMessages) {
	if entryId, ok := constant.CronMsgIdMapMemoryCache[msg.ID]; ok {
		// 先删除之前的定时任务
		delete(constant.CronMsgIdMapMemoryCache, msg.ID)
		cron_service.RemoveTask(entryId)
		// 再注册新的定时任务
		AddCronMsgToCronServer(msg)
	} else {
		// 注册新的定时任务
		AddCronMsgToCronServer(msg)
	}
	logrus.Infof("完成定时消息的定时更新，消息id: %s，当前任务总数：%d", msg.ID, len(constant.CronMsgIdMapMemoryCache))
}

// RemoveCronMsgToCronServer 删除定时任务中心的任务
func RemoveCronMsgToCronServer(msg models.CronMessages) {
	if entryId, ok := constant.CronMsgIdMapMemoryCache[msg.ID]; ok {
		// 先删除之前的定时任务
		delete(constant.CronMsgIdMapMemoryCache, msg.ID)
		cron_service.RemoveTask(entryId)
	}
	logrus.Infof("删除定时消息完成，消息id: %s，剩余任务总数：%d", msg.ID, len(constant.CronMsgIdMapMemoryCache))
}

// StartUpMsgCronTask 启动注册定时任务
func StartUpUserSetupMsgCronTask() {
	MsgCronTask{}.Register()
}

// SendCronMessage 发送定时消息（用于立即发送）
func SendCronMessage(msg models.CronMessages, callerIP string) error {
	// 查询关联的发信任务
	task, err := models.GetTaskByID(msg.TaskID)
	if err != nil {
		return fmt.Errorf("发信任务不存在: %s", msg.TaskID)
	}

	// 创建发送服务
	sender := send_message_service.SendMessageService{
		TaskID:   task.ID,
		Title:    msg.Title,
		Text:     msg.Content,
		URL:      msg.Url,
		CallerIp: callerIP,
		DefaultLogger: logrus.WithFields(logrus.Fields{
			"prefix": "[Manual Send Cron Message]",
		}),
	}

	// 预检查
	taskData, err := sender.SendPreCheck()
	if err != nil {
		errMsg := err.Error()
		// 如果是没有关联实例的错误，返回更友好的提示
		if strings.Contains(errMsg, "没有关联任何实例") {
			return fmt.Errorf("该发信任务尚未配置发送实例，请先在【发信任务】页面为任务 [%s] 添加至少一个发送实例", task.Name)
		}
		return fmt.Errorf("发送预检查失败: %s", errMsg)
	}

	// 发送消息
	_, err = sender.Send(taskData)
	if err != nil {
		return fmt.Errorf("发送失败: %s", err.Error())
	}

	logrus.Infof("立即发送定时消息成功，消息id: %s，消息名: %s", msg.ID, msg.Name)
	return nil
}