实现发送消息到媒介

internal/db/models/message_dao.go

@@ -25,12 +25,12 @@ const (
 type MessageType = string
 
 const (
-	MessageTypeHealthCheckFailed          MessageType = "HealthCheckFailed"
-	MessageTypeHealthCheckNodeUp          MessageType = "HealthCheckNodeUp"
-	MessageTypeHealthCheckNodeDown        MessageType = "HealthCheckNodeDown"
-	MessageTypeNodeInactive               MessageType = "NodeInactive"
-	MessageTypeNodeActive                 MessageType = "NodeActive"
-	MessageTypeClusterDNSSyncFailed       MessageType = "ClusterDNSSyncFailed"
+	MessageTypeHealthCheckFailed          MessageType = "HealthCheckFailed"          // 节点健康检查失败
+	MessageTypeHealthCheckNodeUp          MessageType = "HealthCheckNodeUp"          // 因健康检查节点上线
+	MessageTypeHealthCheckNodeDown        MessageType = "HealthCheckNodeDown"        // 因健康检查节点下线
+	MessageTypeNodeInactive               MessageType = "NodeInactive"               // 节点不活跃
+	MessageTypeNodeActive                 MessageType = "NodeActive"                 // 节点活跃
+	MessageTypeClusterDNSSyncFailed       MessageType = "ClusterDNSSyncFailed"       // DNS同步失败
 	MessageTypeSSLCertExpiring            MessageType = "SSLCertExpiring"            // SSL证书即将过期
 	MessageTypeSSLCertACMETaskFailed      MessageType = "SSLCertACMETaskFailed"      // SSL证书任务执行失败
 	MessageTypeSSLCertACMETaskSuccess     MessageType = "SSLCertACMETaskSuccess"     // SSL证书任务执行成功
@@ -60,7 +60,7 @@ func init() {
 	})
 }
 
-// 启用条目
+// EnableMessage 启用条目
 func (this *MessageDAO) EnableMessage(tx *dbs.Tx, id int64) error {
 	_, err := this.Query(tx).
 		Pk(id).
@@ -69,7 +69,7 @@ func (this *MessageDAO) EnableMessage(tx *dbs.Tx, id int64) error {
 	return err
 }
 
-// 禁用条目
+// DisableMessage 禁用条目
 func (this *MessageDAO) DisableMessage(tx *dbs.Tx, id int64) error {
 	_, err := this.Query(tx).
 		Pk(id).
@@ -78,7 +78,7 @@ func (this *MessageDAO) DisableMessage(tx *dbs.Tx, id int64) error {
 	return err
 }
 
-// 查找启用中的条目
+// FindEnabledMessage 查找启用中的条目
 func (this *MessageDAO) FindEnabledMessage(tx *dbs.Tx, id int64) (*Message, error) {
 	result, err := this.Query(tx).
 		Pk(id).
@@ -90,20 +90,60 @@ func (this *MessageDAO) FindEnabledMessage(tx *dbs.Tx, id int64) (*Message, erro
 	return result.(*Message), err
 }
 
-// 创建集群消息
-func (this *MessageDAO) CreateClusterMessage(tx *dbs.Tx, clusterId int64, messageType MessageType, level string, body string, paramsJSON []byte) error {
-	_, err := this.createMessage(tx, clusterId, 0, messageType, level, body, paramsJSON)
-	return err
+// CreateClusterMessage 创建集群消息
+func (this *MessageDAO) CreateClusterMessage(tx *dbs.Tx, clusterId int64, messageType MessageType, level string, subject string, body string, paramsJSON []byte) error {
+	_, err := this.createMessage(tx, clusterId, 0, messageType, level, subject, body, paramsJSON)
+	if err != nil {
+		return err
+	}
+
+	// 发送给媒介接收人
+	err = SharedMessageTaskDAO.CreateMessageTasks(tx, MessageTaskTarget{
+		ClusterId: clusterId,
+		NodeId:    0,
+		ServerId:  0,
+	}, messageType, subject, body)
+	if err != nil {
+		return err
+	}
+
+	return nil
 }
 
-// 创建节点消息
-func (this *MessageDAO) CreateNodeMessage(tx *dbs.Tx, clusterId int64, nodeId int64, messageType MessageType, level string, body string, paramsJSON []byte) error {
-	_, err := this.createMessage(tx, clusterId, nodeId, messageType, level, body, paramsJSON)
-	return err
+// CreateNodeMessage 创建节点消息
+func (this *MessageDAO) CreateNodeMessage(tx *dbs.Tx, clusterId int64, nodeId int64, messageType MessageType, level string, subject string, body string, paramsJSON []byte) error {
+	_, err := this.createMessage(tx, clusterId, nodeId, messageType, level, subject, body, paramsJSON)
+	if err != nil {
+		return err
+	}
+
+	// 发送给媒介接收人 - 集群
+	err = SharedMessageTaskDAO.CreateMessageTasks(tx, MessageTaskTarget{
+		ClusterId: clusterId,
+		NodeId:    0,
+		ServerId:  0,
+	}, messageType, subject, body)
+	if err != nil {
+		return err
+	}
+
+	// 发送给媒介接收人 - 节点
+	if nodeId > 0 {
+		err = SharedMessageTaskDAO.CreateMessageTasks(tx, MessageTaskTarget{
+			ClusterId: clusterId,
+			NodeId:    nodeId,
+			ServerId:  0,
+		}, messageType, subject, body)
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
 }
 
-// 创建普通消息
-func (this *MessageDAO) CreateMessage(tx *dbs.Tx, adminId int64, userId int64, messageType MessageType, level string, body string, paramsJSON []byte) error {
+// CreateMessage 创建普通消息
+func (this *MessageDAO) CreateMessage(tx *dbs.Tx, adminId int64, userId int64, messageType MessageType, level string, subject string, body string, paramsJSON []byte) error {
 	h := md5.New()
 	h.Write([]byte(body))
 	h.Write(paramsJSON)
@@ -114,6 +154,14 @@ func (this *MessageDAO) CreateMessage(tx *dbs.Tx, adminId int64, userId int64, m
 	op.UserId = userId
 	op.Type = messageType
 	op.Level = level
+
+	subjectRunes := []rune(subject)
+	if len(subjectRunes) > 100 {
+		op.Subject = string(subjectRunes[:100]) + "..."
+	} else {
+		op.Subject = subject
+	}
+
 	op.Body = body
 	if len(paramsJSON) > 0 {
 		op.Params = paramsJSON
@@ -123,10 +171,14 @@ func (this *MessageDAO) CreateMessage(tx *dbs.Tx, adminId int64, userId int64, m
 	op.Day = timeutil.Format("Ymd")
 	op.Hash = hash
 	err := this.Save(tx, op)
-	return err
+	if err != nil {
+		return err
+	}
+
+	return nil
 }
 
-// 删除某天之前的消息
+// DeleteMessagesBeforeDay 删除某天之前的消息
 func (this *MessageDAO) DeleteMessagesBeforeDay(tx *dbs.Tx, dayTime time.Time) error {
 	day := timeutil.Format("Ymd", dayTime)
 	_, err := this.Query(tx).
@@ -136,7 +188,7 @@ func (this *MessageDAO) DeleteMessagesBeforeDay(tx *dbs.Tx, dayTime time.Time) e
 	return err
 }
 
-// 计算未读消息数量
+// CountUnreadMessages 计算未读消息数量
 func (this *MessageDAO) CountUnreadMessages(tx *dbs.Tx, adminId int64, userId int64) (int64, error) {
 	query := this.Query(tx).
 		Attr("isRead", false)
@@ -149,7 +201,7 @@ func (this *MessageDAO) CountUnreadMessages(tx *dbs.Tx, adminId int64, userId in
 	return query.Count()
 }
 
-// 列出单页未读消息
+// ListUnreadMessages 列出单页未读消息
 func (this *MessageDAO) ListUnreadMessages(tx *dbs.Tx, adminId int64, userId int64, offset int64, size int64) (result []*Message, err error) {
 	query := this.Query(tx).
 		Attr("isRead", false)
@@ -168,7 +220,7 @@ func (this *MessageDAO) ListUnreadMessages(tx *dbs.Tx, adminId int64, userId int
 	return
 }
 
-// 设置消息已读状态
+// UpdateMessageRead 设置消息已读状态
 func (this *MessageDAO) UpdateMessageRead(tx *dbs.Tx, messageId int64, b bool) error {
 	if messageId <= 0 {
 		return errors.New("invalid messageId")
@@ -180,7 +232,7 @@ func (this *MessageDAO) UpdateMessageRead(tx *dbs.Tx, messageId int64, b bool) e
 	return err
 }
 
-// 设置一组消息为已读状态
+// UpdateMessagesRead 设置一组消息为已读状态
 func (this *MessageDAO) UpdateMessagesRead(tx *dbs.Tx, messageIds []int64, b bool) error {
 	// 这里我们一个一个更改，因为In语句不容易Prepare，且效率不高
 	for _, messageId := range messageIds {
@@ -192,7 +244,7 @@ func (this *MessageDAO) UpdateMessagesRead(tx *dbs.Tx, messageIds []int64, b boo
 	return nil
 }
 
-// 设置所有消息为已读
+// UpdateAllMessagesRead 设置所有消息为已读
 func (this *MessageDAO) UpdateAllMessagesRead(tx *dbs.Tx, adminId int64, userId int64) error {
 	query := this.Query(tx).
 		Attr("isRead", false)
@@ -208,7 +260,7 @@ func (this *MessageDAO) UpdateAllMessagesRead(tx *dbs.Tx, adminId int64, userId
 	return err
 }
 
-// 检查消息权限
+// CheckMessageUser 检查消息权限
 func (this *MessageDAO) CheckMessageUser(tx *dbs.Tx, messageId int64, adminId int64, userId int64) (bool, error) {
 	if messageId <= 0 || (adminId <= 0 && userId <= 0) {
 		return false, nil
@@ -225,7 +277,7 @@ func (this *MessageDAO) CheckMessageUser(tx *dbs.Tx, messageId int64, adminId in
 }
 
 // 创建消息
-func (this *MessageDAO) createMessage(tx *dbs.Tx, clusterId int64, nodeId int64, messageType MessageType, level string, body string, paramsJSON []byte) (int64, error) {
+func (this *MessageDAO) createMessage(tx *dbs.Tx, clusterId int64, nodeId int64, messageType MessageType, level string, subject string, body string, paramsJSON []byte) (int64, error) {
 	h := md5.New()
 	h.Write([]byte(body))
 	h.Write(paramsJSON)
@@ -241,6 +293,14 @@ func (this *MessageDAO) createMessage(tx *dbs.Tx, clusterId int64, nodeId int64,
 	op.NodeId = nodeId
 	op.Type = messageType
 	op.Level = level
+
+	subjectRunes := []rune(subject)
+	if len(subjectRunes) > 100 {
+		op.Subject = string(subjectRunes[:100]) + "..."
+	} else {
+		op.Subject = subject
+	}
+
 	op.Body = body
 	if len(paramsJSON) > 0 {
 		op.Params = paramsJSON

internal/db/models/message_model.go

@@ -1,6 +1,6 @@
 package models
 
-// 消息通知
+// Message 消息通知
 type Message struct {
 	Id        uint64 `field:"id"`        // ID
 	AdminId   uint32 `field:"adminId"`   // 管理员ID
@@ -8,6 +8,7 @@ type Message struct {
 	ClusterId uint32 `field:"clusterId"` // 集群ID
 	NodeId    uint32 `field:"nodeId"`    // 节点ID
 	Level     string `field:"level"`     // 级别
+	Subject   string `field:"subject"`   // 标题
 	Body      string `field:"body"`      // 内容
 	Type      string `field:"type"`      // 消息类型
 	Params    string `field:"params"`    // 额外的参数
@@ -25,6 +26,7 @@ type MessageOperator struct {
 	ClusterId interface{} // 集群ID
 	NodeId    interface{} // 节点ID
 	Level     interface{} // 级别
+	Subject   interface{} // 标题
 	Body      interface{} // 内容
 	Type      interface{} // 消息类型
 	Params    interface{} // 额外的参数

internal/db/models/message_receiver_dao.go

@@ -0,0 +1,115 @@
+package models
+
+import (
+	"encoding/json"
+	_ "github.com/go-sql-driver/mysql"
+	"github.com/iwind/TeaGo/Tea"
+	"github.com/iwind/TeaGo/dbs"
+	"github.com/iwind/TeaGo/maps"
+)
+
+const (
+	MessageReceiverStateEnabled  = 1 // 已启用
+	MessageReceiverStateDisabled = 0 // 已禁用
+)
+
+type MessageReceiverDAO dbs.DAO
+
+func NewMessageReceiverDAO() *MessageReceiverDAO {
+	return dbs.NewDAO(&MessageReceiverDAO{
+		DAOObject: dbs.DAOObject{
+			DB:     Tea.Env,
+			Table:  "edgeMessageReceivers",
+			Model:  new(MessageReceiver),
+			PkName: "id",
+		},
+	}).(*MessageReceiverDAO)
+}
+
+var SharedMessageReceiverDAO *MessageReceiverDAO
+
+func init() {
+	dbs.OnReady(func() {
+		SharedMessageReceiverDAO = NewMessageReceiverDAO()
+	})
+}
+
+// EnableMessageReceiver 启用条目
+func (this *MessageReceiverDAO) EnableMessageReceiver(tx *dbs.Tx, id int64) error {
+	_, err := this.Query(tx).
+		Pk(id).
+		Set("state", MessageReceiverStateEnabled).
+		Update()
+	return err
+}
+
+// DisableMessageReceiver 禁用条目
+func (this *MessageReceiverDAO) DisableMessageReceiver(tx *dbs.Tx, id int64) error {
+	_, err := this.Query(tx).
+		Pk(id).
+		Set("state", MessageReceiverStateDisabled).
+		Update()
+	return err
+}
+
+// FindEnabledMessageReceiver 查找启用中的条目
+func (this *MessageReceiverDAO) FindEnabledMessageReceiver(tx *dbs.Tx, id int64) (*MessageReceiver, error) {
+	result, err := this.Query(tx).
+		Pk(id).
+		Attr("state", MessageReceiverStateEnabled).
+		Find()
+	if result == nil {
+		return nil, err
+	}
+	return result.(*MessageReceiver), err
+}
+
+// DisableReceivers 禁用一组接收人
+func (this *MessageReceiverDAO) DisableReceivers(tx *dbs.Tx, clusterId int64, nodeId int64, serverId int64) error {
+	return this.Query(tx).
+		Attr("clusterId", clusterId).
+		Attr("nodeId", nodeId).
+		Attr("serverId", serverId).
+		Set("state", MessageReceiverStateDisabled).
+		UpdateQuickly()
+}
+
+// CreateReceiver 创建接收人
+func (this *MessageReceiverDAO) CreateReceiver(tx *dbs.Tx, target MessageTaskTarget, messageType MessageType, params maps.Map, recipientId int64, recipientGroupId int64) (int64, error) {
+	op := NewMessageReceiverOperator()
+	op.ClusterId = target.ClusterId
+	op.NodeId = target.NodeId
+	op.ServerId = target.ServerId
+	op.Type = messageType
+
+	if params == nil {
+		params = maps.Map{}
+	}
+	paramsJSON, err := json.Marshal(params)
+	if err != nil {
+		return 0, err
+	}
+	op.Params = paramsJSON
+
+	op.RecipientId = recipientId
+	op.RecipientGroupId = recipientGroupId
+	op.State = MessageReceiverStateEnabled
+	return this.SaveInt64(tx, op)
+}
+
+// FindAllReceivers 查询接收人
+func (this *MessageReceiverDAO) FindAllReceivers(tx *dbs.Tx, target MessageTaskTarget, messageType string) (result []*MessageReceiver, err error) {
+	query := this.Query(tx)
+	if len(messageType) > 0 {
+		query.Attr("type", []string{"*", messageType}) // *表示所有的
+	}
+	_, err = query.
+		Attr("clusterId", target.ClusterId).
+		Attr("nodeId", target.NodeId).
+		Attr("serverId", target.ServerId).
+		State(MessageReceiverStateEnabled).
+		AscPk().
+		Slice(&result).
+		FindAll()
+	return
+}

internal/db/models/message_receiver_dao_test.go

@@ -0,0 +1,6 @@
+package models
+
+import (
+	_ "github.com/go-sql-driver/mysql"
+	_ "github.com/iwind/TeaGo/bootstrap"
+)

internal/db/models/message_receiver_model.go

@@ -0,0 +1,30 @@
+package models
+
+// MessageReceiver 消息通知接收人
+type MessageReceiver struct {
+	Id               uint32 `field:"id"`               // ID
+	ClusterId        uint32 `field:"clusterId"`        // 集群ID
+	NodeId           uint32 `field:"nodeId"`           // 节点ID
+	ServerId         uint32 `field:"serverId"`         // 服务ID
+	Type             string `field:"type"`             // 类型
+	Params           string `field:"params"`           // 参数
+	RecipientId      uint32 `field:"recipientId"`      // 接收人ID
+	RecipientGroupId uint32 `field:"recipientGroupId"` // 接收人分组ID
+	State            uint8  `field:"state"`            // 状态
+}
+
+type MessageReceiverOperator struct {
+	Id               interface{} // ID
+	ClusterId        interface{} // 集群ID
+	NodeId           interface{} // 节点ID
+	ServerId         interface{} // 服务ID
+	Type             interface{} // 类型
+	Params           interface{} // 参数
+	RecipientId      interface{} // 接收人ID
+	RecipientGroupId interface{} // 接收人分组ID
+	State            interface{} // 状态
+}
+
+func NewMessageReceiverOperator() *MessageReceiverOperator {
+	return &MessageReceiverOperator{}
+}

internal/db/models/message_receiver_model_ext.go

@@ -0,0 +1 @@
+package models

internal/db/models/message_recipient_dao.go

@@ -3,6 +3,7 @@ package models
 import (
 	"encoding/json"
 	"github.com/TeaOSLab/EdgeAPI/internal/errors"
+	"github.com/TeaOSLab/EdgeAPI/internal/utils/numberutils"
 	_ "github.com/go-sql-driver/mysql"
 	"github.com/iwind/TeaGo/Tea"
 	"github.com/iwind/TeaGo/dbs"
@@ -34,7 +35,7 @@ func init() {
 	})
 }
 
-// 启用条目
+// EnableMessageRecipient 启用条目
 func (this *MessageRecipientDAO) EnableMessageRecipient(tx *dbs.Tx, id int64) error {
 	_, err := this.Query(tx).
 		Pk(id).
@@ -43,7 +44,7 @@ func (this *MessageRecipientDAO) EnableMessageRecipient(tx *dbs.Tx, id int64) er
 	return err
 }
 
-// 禁用条目
+// DisableMessageRecipient 禁用条目
 func (this *MessageRecipientDAO) DisableMessageRecipient(tx *dbs.Tx, id int64) error {
 	_, err := this.Query(tx).
 		Pk(id).
@@ -52,7 +53,7 @@ func (this *MessageRecipientDAO) DisableMessageRecipient(tx *dbs.Tx, id int64) e
 	return err
 }
 
-// 查找启用中的条目
+// FindEnabledMessageRecipient 查找启用中的条目
 func (this *MessageRecipientDAO) FindEnabledMessageRecipient(tx *dbs.Tx, id int64) (*MessageRecipient, error) {
 	result, err := this.Query(tx).
 		Pk(id).
@@ -64,7 +65,7 @@ func (this *MessageRecipientDAO) FindEnabledMessageRecipient(tx *dbs.Tx, id int6
 	return result.(*MessageRecipient), err
 }
 
-// 创建接收人
+// CreateRecipient 创建接收人
 func (this *MessageRecipientDAO) CreateRecipient(tx *dbs.Tx, adminId int64, instanceId int64, user string, groupIds []int64, description string) (int64, error) {
 	op := NewMessageRecipientOperator()
 	op.AdminId = adminId
@@ -87,7 +88,7 @@ func (this *MessageRecipientDAO) CreateRecipient(tx *dbs.Tx, adminId int64, inst
 	return this.SaveInt64(tx, op)
 }
 
-// 修改接收人
+// UpdateRecipient 修改接收人
 func (this *MessageRecipientDAO) UpdateRecipient(tx *dbs.Tx, recipientId int64, adminId int64, instanceId int64, user string, groupIds []int64, description string, isOn bool) error {
 	if recipientId <= 0 {
 		return errors.New("invalid recipientId")
@@ -114,14 +115,15 @@ func (this *MessageRecipientDAO) UpdateRecipient(tx *dbs.Tx, recipientId int64,
 	return this.Save(tx, op)
 }
 
-// 计算接收人数量
+// CountAllEnabledRecipients 计算接收人数量
 func (this *MessageRecipientDAO) CountAllEnabledRecipients(tx *dbs.Tx, adminId int64, groupId int64, mediaType string, keyword string) (int64, error) {
 	query := this.Query(tx)
 	if adminId > 0 {
 		query.Attr("adminId", adminId)
 	}
 	if groupId > 0 {
-		query.Attr("groupId", groupId)
+		query.Where("JSON_CONTAINS(groupIds, :groupId)").
+			Param("groupId", numberutils.FormatInt64(groupId))
 	}
 	if len(mediaType) > 0 {
 		query.Where("instanceId IN (SELECT id FROM "+SharedMessageMediaInstanceDAO.Table+" WHERE state=1 AND mediaType=:mediaType)").
@@ -138,14 +140,15 @@ func (this *MessageRecipientDAO) CountAllEnabledRecipients(tx *dbs.Tx, adminId i
 		Count()
 }
 
-// 列出单页接收人
+// ListAllEnabledRecipients 列出单页接收人
 func (this *MessageRecipientDAO) ListAllEnabledRecipients(tx *dbs.Tx, adminId int64, groupId int64, mediaType string, keyword string, offset int64, size int64) (result []*MessageRecipient, err error) {
 	query := this.Query(tx)
 	if adminId > 0 {
 		query.Attr("adminId", adminId)
 	}
 	if groupId > 0 {
-		query.Attr("groupId", groupId)
+		query.Where("JSON_CONTAINS(groupIds, :groupId)").
+			Param("groupId", numberutils.FormatInt64(groupId))
 	}
 	if len(mediaType) > 0 {
 		query.Where("instanceId IN (SELECT id FROM "+SharedMessageMediaInstanceDAO.Table+" WHERE state=1 AND mediaType=:mediaType)").
@@ -166,3 +169,21 @@ func (this *MessageRecipientDAO) ListAllEnabledRecipients(tx *dbs.Tx, adminId in
 		FindAll()
 	return
 }
+
+// FindAllEnabledAndOnRecipientIdsWithGroup 查找某个分组下的所有可用接收人ID
+func (this *MessageRecipientDAO) FindAllEnabledAndOnRecipientIdsWithGroup(tx *dbs.Tx, groupId int64) ([]int64, error) {
+	ones, err := this.Query(tx).
+		Where("JSON_CONTAINS(groupIds, :groupId)").
+		Param("groupId", numberutils.FormatInt64(groupId)).
+		State(MessageRecipientStateEnabled).
+		ResultPk().
+		FindAll()
+	if err != nil {
+		return nil, err
+	}
+	result := []int64{}
+	for _, one := range ones {
+		result = append(result, int64(one.(*MessageRecipient).Id))
+	}
+	return result, nil
+}

internal/db/models/message_recipient_dao_test.go

@@ -3,4 +3,15 @@ package models
 import (
 	_ "github.com/go-sql-driver/mysql"
 	_ "github.com/iwind/TeaGo/bootstrap"
+	"github.com/iwind/TeaGo/dbs"
+	"testing"
 )
+
+func TestMessageRecipientDAO_FindAllEnabledAndOnRecipientIdsWithGroup(t *testing.T) {
+	dbs.NotifyReady()
+	recipientIds, err := SharedMessageRecipientDAO.FindAllEnabledAndOnRecipientIdsWithGroup(nil, 4)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Log(recipientIds)
+}

internal/db/models/message_target.go

@@ -0,0 +1,9 @@
+// Copyright 2021 Liuxiangchao iwind.liu@gmail.com. All rights reserved.
+
+package models
+
+type MessageTaskTarget struct {
+	ClusterId int64
+	NodeId    int64
+	ServerId  int64
+}

internal/db/models/message_task_dao.go

@@ -41,7 +41,7 @@ func init() {
 	})
 }
 
-// 启用条目
+// EnableMessageTask 启用条目
 func (this *MessageTaskDAO) EnableMessageTask(tx *dbs.Tx, id int64) error {
 	_, err := this.Query(tx).
 		Pk(id).
@@ -50,7 +50,7 @@ func (this *MessageTaskDAO) EnableMessageTask(tx *dbs.Tx, id int64) error {
 	return err
 }
 
-// 禁用条目
+// DisableMessageTask 禁用条目
 func (this *MessageTaskDAO) DisableMessageTask(tx *dbs.Tx, id int64) error {
 	_, err := this.Query(tx).
 		Pk(id).
@@ -59,7 +59,7 @@ func (this *MessageTaskDAO) DisableMessageTask(tx *dbs.Tx, id int64) error {
 	return err
 }
 
-// 查找启用中的条目
+// FindEnabledMessageTask 查找启用中的条目
 func (this *MessageTaskDAO) FindEnabledMessageTask(tx *dbs.Tx, id int64) (*MessageTask, error) {
 	result, err := this.Query(tx).
 		Pk(id).
@@ -71,7 +71,7 @@ func (this *MessageTaskDAO) FindEnabledMessageTask(tx *dbs.Tx, id int64) (*Messa
 	return result.(*MessageTask), err
 }
 
-// 创建任务
+// CreateMessageTask 创建任务
 func (this *MessageTaskDAO) CreateMessageTask(tx *dbs.Tx, recipientId int64, instanceId int64, user string, subject string, body string, isPrimary bool) (int64, error) {
 	op := NewMessageTaskOperator()
 	op.RecipientId = recipientId
@@ -85,7 +85,7 @@ func (this *MessageTaskDAO) CreateMessageTask(tx *dbs.Tx, recipientId int64, ins
 	return this.SaveInt64(tx, op)
 }
 
-// 查找需要发送的任务
+// FindSendingMessageTasks 查找需要发送的任务
 func (this *MessageTaskDAO) FindSendingMessageTasks(tx *dbs.Tx, size int64) (result []*MessageTask, err error) {
 	if size <= 0 {
 		return nil, nil
@@ -101,7 +101,7 @@ func (this *MessageTaskDAO) FindSendingMessageTasks(tx *dbs.Tx, size int64) (res
 	return
 }
 
-// 设置发送的状态
+// UpdateMessageTaskStatus 设置发送的状态
 func (this *MessageTaskDAO) UpdateMessageTaskStatus(tx *dbs.Tx, taskId int64, status MessageTaskStatus, result []byte) error {
 	if taskId <= 0 {
 		return errors.New("invalid taskId")
@@ -115,3 +115,31 @@ func (this *MessageTaskDAO) UpdateMessageTaskStatus(tx *dbs.Tx, taskId int64, st
 	}
 	return this.Save(tx, op)
 }
+
+// CreateMessageTasks 从集群、节点或者服务中创建任务
+func (this *MessageTaskDAO) CreateMessageTasks(tx *dbs.Tx, target MessageTaskTarget, messageType MessageType, subject string, body string) error {
+	receivers, err := SharedMessageReceiverDAO.FindAllReceivers(tx, target, messageType)
+	if err != nil {
+		return err
+	}
+	for _, receiver := range receivers {
+		if receiver.RecipientId > 0 {
+			_, err := this.CreateMessageTask(tx, int64(receiver.RecipientId), 0, "", subject, body, false)
+			if err != nil {
+				return err
+			}
+		} else if receiver.RecipientGroupId > 0 {
+			recipientIds, err := SharedMessageRecipientDAO.FindAllEnabledAndOnRecipientIdsWithGroup(tx, int64(receiver.RecipientGroupId))
+			if err != nil {
+				return err
+			}
+			for _, recipientId := range recipientIds {
+				_, err := this.CreateMessageTask(tx, recipientId, 0, "", subject, body, false)
+				if err != nil {
+					return err
+				}
+			}
+		}
+	}
+	return nil
+}

internal/db/models/node_dao_test.go

@@ -18,9 +18,9 @@ func TestNodeDAO_FindAllNodeIdsMatch(t *testing.T) {
 func TestNodeDAO_UpdateNodeUp(t *testing.T) {
 	dbs.NotifyReady()
 	var tx *dbs.Tx
-	isChanged, err := SharedNodeDAO.UpdateNodeUp(tx, 57, false, 3, 3)
+	err := SharedNodeDAO.UpdateNodeUp(tx, 57, false)
 	if err != nil {
 		t.Fatal(err)
 	}
-	t.Log("changed:", isChanged)
+	t.Log("ok")
 }