通知媒介增加任务队列查看功能

go.mod

@@ -4,6 +4,8 @@ go 1.15
 
 replace github.com/TeaOSLab/EdgeCommon => ../EdgeCommon
 
+
+
 require (
 	github.com/StackExchange/wmi v0.0.0-20190523213315-cbe66965904d // indirect
 	github.com/TeaOSLab/EdgeCommon v0.0.0-00010101000000-000000000000
@@ -14,7 +16,7 @@ require (
 	github.com/go-sql-driver/mysql v1.5.0
 	github.com/go-yaml/yaml v2.1.0+incompatible
 	github.com/golang/protobuf v1.5.2
-	github.com/iwind/TeaGo v0.0.0-20210809112119-a57ed0e84e34
+	github.com/iwind/TeaGo v0.0.0-20210824034952-1a56ad7d0b5e
 	github.com/iwind/gosock v0.0.0-20210722083328-12b2d66abec3
 	github.com/json-iterator/go v1.1.11 // indirect
 	github.com/lionsoul2014/ip2region v2.2.0-release+incompatible

go.sum

@@ -186,6 +186,8 @@ github.com/iwind/TeaGo v0.0.0-20210806054428-5534da0db9d1 h1:AZKkwTNEZYrpyv62zIk
 github.com/iwind/TeaGo v0.0.0-20210806054428-5534da0db9d1/go.mod h1:KU4mS7QNiZ7QWEuDBk1zw0/Q2LrAPZv3tycEFBsuUwc=
 github.com/iwind/TeaGo v0.0.0-20210809112119-a57ed0e84e34 h1:ZCNQXLiGF5Z1cV3Pi03zCWzwwjPfsI5XhcrNhTvCFIU=
 github.com/iwind/TeaGo v0.0.0-20210809112119-a57ed0e84e34/go.mod h1:KU4mS7QNiZ7QWEuDBk1zw0/Q2LrAPZv3tycEFBsuUwc=
+github.com/iwind/TeaGo v0.0.0-20210824034952-1a56ad7d0b5e h1:GDCU57lQD6W9u5KT2834MmK022FSeAbskb7H0p2eaJY=
+github.com/iwind/TeaGo v0.0.0-20210824034952-1a56ad7d0b5e/go.mod h1:KU4mS7QNiZ7QWEuDBk1zw0/Q2LrAPZv3tycEFBsuUwc=
 github.com/iwind/gosock v0.0.0-20210722083328-12b2d66abec3 h1:aBSonas7vFcgTj9u96/bWGILGv1ZbUSTLiOzcI1ZT6c=
 github.com/iwind/gosock v0.0.0-20210722083328-12b2d66abec3/go.mod h1:H5Q7SXwbx3a97ecJkaS2sD77gspzE7HFUafBO0peEyA=
 github.com/jmespath/go-jmespath v0.0.0-20180206201540-c2b33e8439af/go.mod h1:Nht3zPeWKUH0NzdCt2Blrr5ys8VGpn0CEB0cQHVjt7k=

internal/db/models/message_dao.go

@@ -106,11 +106,7 @@ func (this *MessageDAO) CreateClusterMessage(tx *dbs.Tx, role string, clusterId
 	}
 
 	// 发送给媒介接收人
-	err = SharedMessageTaskDAO.CreateMessageTasks(tx, MessageTaskTarget{
+	err = SharedMessageTaskDAO.CreateMessageTasks(tx, role, 0, 0, 0, messageType, subject, body)
-		ClusterId: clusterId,
-		NodeId:    0,
-		ServerId:  0,
-	}, messageType, subject, body)
 	if err != nil {
 		return err
 	}
@@ -138,31 +134,12 @@ func (this *MessageDAO) CreateNodeMessage(tx *dbs.Tx, role string, clusterId int
 		return err
 	}
 
-	// TODO 目前只支持边缘节点发送消息，将来要支持NS节点
-	if role == nodeconfigs.NodeRoleNode {
 	// 发送给媒介接收人 - 集群
-		err = SharedMessageTaskDAO.CreateMessageTasks(tx, MessageTaskTarget{
+	err = SharedMessageTaskDAO.CreateMessageTasks(tx, role, clusterId, nodeId, 0, messageType, subject, body)
-			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
 }
 

internal/db/models/message_media_instance_dao.go

@@ -7,6 +7,7 @@ import (
 	"github.com/iwind/TeaGo/Tea"
 	"github.com/iwind/TeaGo/dbs"
 	"github.com/iwind/TeaGo/maps"
+	"github.com/iwind/TeaGo/types"
 )
 
 const (
@@ -35,7 +36,7 @@ func init() {
 	})
 }
 
-// 启用条目
+// EnableMessageMediaInstance 启用条目
 func (this *MessageMediaInstanceDAO) EnableMessageMediaInstance(tx *dbs.Tx, id int64) error {
 	_, err := this.Query(tx).
 		Pk(id).
@@ -44,7 +45,7 @@ func (this *MessageMediaInstanceDAO) EnableMessageMediaInstance(tx *dbs.Tx, id i
 	return err
 }
 
-// 禁用条目
+// DisableMessageMediaInstance 禁用条目
 func (this *MessageMediaInstanceDAO) DisableMessageMediaInstance(tx *dbs.Tx, id int64) error {
 	_, err := this.Query(tx).
 		Pk(id).
@@ -53,19 +54,31 @@ func (this *MessageMediaInstanceDAO) DisableMessageMediaInstance(tx *dbs.Tx, id
 	return err
 }
 
-// 查找启用中的条目
+// FindEnabledMessageMediaInstance 查找启用中的条目
-func (this *MessageMediaInstanceDAO) FindEnabledMessageMediaInstance(tx *dbs.Tx, id int64) (*MessageMediaInstance, error) {
+func (this *MessageMediaInstanceDAO) FindEnabledMessageMediaInstance(tx *dbs.Tx, instanceId int64, cacheMap maps.Map) (*MessageMediaInstance, error) {
+	if cacheMap == nil {
+		cacheMap = maps.Map{}
+	}
+	var cacheKey = this.Table + ":record:" + types.String(instanceId)
+	var cache = cacheMap.Get(cacheKey)
+	if cache != nil {
+		return cache.(*MessageMediaInstance), nil
+	}
+
 	result, err := this.Query(tx).
-		Pk(id).
+		Pk(instanceId).
 		Attr("state", MessageMediaInstanceStateEnabled).
 		Find()
 	if result == nil {
 		return nil, err
 	}
+
+	cacheMap[cacheKey] = result
+
 	return result.(*MessageMediaInstance), err
 }
 
-// 创建媒介实例
+// CreateMediaInstance 创建媒介实例
 func (this *MessageMediaInstanceDAO) CreateMediaInstance(tx *dbs.Tx, name string, mediaType string, params maps.Map, description string) (int64, error) {
 	op := NewMessageMediaInstanceOperator()
 	op.Name = name
@@ -88,7 +101,7 @@ func (this *MessageMediaInstanceDAO) CreateMediaInstance(tx *dbs.Tx, name string
 	return this.SaveInt64(tx, op)
 }
 
-// 修改媒介实例
+// UpdateMediaInstance 修改媒介实例
 func (this *MessageMediaInstanceDAO) UpdateMediaInstance(tx *dbs.Tx, instanceId int64, name string, mediaType string, params maps.Map, description string, isOn bool) error {
 	if instanceId <= 0 {
 		return errors.New("invalid instanceId")
@@ -114,7 +127,7 @@ func (this *MessageMediaInstanceDAO) UpdateMediaInstance(tx *dbs.Tx, instanceId
 	return this.Save(tx, op)
 }
 
-// 计算接收人数量
+// CountAllEnabledMediaInstances 计算接收人数量
 func (this *MessageMediaInstanceDAO) CountAllEnabledMediaInstances(tx *dbs.Tx, mediaType string, keyword string) (int64, error) {
 	query := this.Query(tx)
 	if len(mediaType) > 0 {
@@ -130,7 +143,7 @@ func (this *MessageMediaInstanceDAO) CountAllEnabledMediaInstances(tx *dbs.Tx, m
 		Count()
 }
 
-// 列出单页接收人
+// ListAllEnabledMediaInstances 列出单页接收人
 func (this *MessageMediaInstanceDAO) ListAllEnabledMediaInstances(tx *dbs.Tx, mediaType string, keyword string, offset int64, size int64) (result []*MessageMediaInstance, err error) {
 	query := this.Query(tx)
 	if len(mediaType) > 0 {

internal/db/models/message_receiver_dao.go

@@ -75,11 +75,12 @@ func (this *MessageReceiverDAO) DisableReceivers(tx *dbs.Tx, clusterId int64, no
 }
 
 // CreateReceiver 创建接收人
-func (this *MessageReceiverDAO) CreateReceiver(tx *dbs.Tx, target MessageTaskTarget, messageType MessageType, params maps.Map, recipientId int64, recipientGroupId int64) (int64, error) {
+func (this *MessageReceiverDAO) CreateReceiver(tx *dbs.Tx, role string, clusterId int64, nodeId int64, serverId int64, messageType MessageType, params maps.Map, recipientId int64, recipientGroupId int64) (int64, error) {
 	op := NewMessageReceiverOperator()
-	op.ClusterId = target.ClusterId
+	op.Role = role
-	op.NodeId = target.NodeId
+	op.ClusterId = clusterId
-	op.ServerId = target.ServerId
+	op.NodeId = nodeId
+	op.ServerId = serverId
 	op.Type = messageType
 
 	if params == nil {
@@ -98,63 +99,120 @@ func (this *MessageReceiverDAO) CreateReceiver(tx *dbs.Tx, target MessageTaskTar
 }
 
 // FindAllEnabledReceivers 查询接收人
-func (this *MessageReceiverDAO) FindAllEnabledReceivers(tx *dbs.Tx, target MessageTaskTarget, messageType string) (result []*MessageReceiver, err error) {
+func (this *MessageReceiverDAO) FindAllEnabledReceivers(tx *dbs.Tx, role string, clusterId int64, nodeId int64, serverId int64, 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("role", role).
-		Attr("nodeId", target.NodeId).
+		Attr("clusterId", clusterId).
-		Attr("serverId", target.ServerId).
+		Attr("nodeId", nodeId).
+		Attr("serverId", serverId).
 		State(MessageReceiverStateEnabled).
 		AscPk().
 		Slice(&result).
 		FindAll()
-	if err != nil {
-		return nil, err
-	}
-
-	if len(result) == 0 {
-		// 去掉类型再试试
-		query := this.Query(tx)
-		_, err = query.
-			Attr("clusterId", target.ClusterId).
-			Attr("nodeId", target.NodeId).
-			Attr("serverId", target.ServerId).
-			State(MessageReceiverStateEnabled).
-			AscPk().
-			Slice(&result).
-			FindAll()
-		if err != nil {
-			return nil, err
-		}
-
-		// 去掉服务和节点再试试
-		if len(result) == 0 {
-			query := this.Query(tx)
-			_, err = query.
-				Attr("clusterId", target.ClusterId).
-				State(MessageReceiverStateEnabled).
-				AscPk().
-				Slice(&result).
-				FindAll()
-		}
-	}
-
 	return
 }
 
 // CountAllEnabledReceivers 计算接收人数量
-func (this *MessageReceiverDAO) CountAllEnabledReceivers(tx *dbs.Tx, target MessageTaskTarget, messageType string) (int64, error) {
+func (this *MessageReceiverDAO) CountAllEnabledReceivers(tx *dbs.Tx, role string, clusterId int64, nodeId int64, serverId int64, messageType string) (int64, error) {
 	query := this.Query(tx)
 	if len(messageType) > 0 {
 		query.Attr("type", []string{"*", messageType}) // *表示所有的
 	}
 	return query.
-		Attr("clusterId", target.ClusterId).
+		Attr("role", role).
-		Attr("nodeId", target.NodeId).
+		Attr("clusterId", clusterId).
-		Attr("serverId", target.ServerId).
+		Attr("nodeId", nodeId).
+		Attr("serverId", serverId).
 		State(MessageReceiverStateEnabled).
 		Count()
 }
+
+// FindEnabledBestFitReceivers 查询最适合的接收人
+func (this *MessageReceiverDAO) FindEnabledBestFitReceivers(tx *dbs.Tx, role string, clusterId int64, nodeId int64, serverId int64, messageType string) (result []*MessageReceiver, err error) {
+	// serverId优先
+	query := this.Query(tx)
+	if len(messageType) > 0 {
+		query.Attr("type", []string{"*", messageType}) // *表示所有的
+	}
+	if len(role) > 0 {
+		query.Attr("role", role)
+	}
+	if serverId > 0 {
+		query.Attr("serverId", serverId)
+	} else if nodeId > 0 {
+		query.Attr("nodeId", nodeId)
+	} else if clusterId > 0 {
+		query.Attr("clusterId", clusterId)
+	}
+	_, err = query.
+		State(MessageReceiverStateEnabled).
+		AscPk().
+		Slice(&result).
+		FindAll()
+	if err != nil || len(result) > 0 {
+		return
+	}
+
+	// nodeId优先
+	if serverId > 0 && nodeId > 0 {
+		query = this.Query(tx)
+		if len(messageType) > 0 {
+			query.Attr("type", []string{"*", messageType}) // *表示所有的
+		}
+		if len(role) > 0 {
+			query.Attr("role", role)
+		}
+		query.Attr("nodeId", nodeId)
+		_, err = query.
+			State(MessageReceiverStateEnabled).
+			AscPk().
+			Slice(&result).
+			FindAll()
+		if err != nil || len(result) > 0 {
+			return
+		}
+	}
+
+	// clusterId优先
+	if (serverId > 0 || nodeId > 0) && clusterId > 0 {
+		query = this.Query(tx)
+		if len(messageType) > 0 {
+			query.Attr("type", []string{"*", messageType}) // *表示所有的
+		}
+		if len(role) > 0 {
+			query.Attr("role", role)
+		}
+		query.Attr("clusterId", clusterId)
+		_, err = query.
+			State(MessageReceiverStateEnabled).
+			AscPk().
+			Slice(&result).
+			FindAll()
+		if err != nil || len(result) > 0 {
+			return
+		}
+	}
+
+	// 去掉集群ID
+	query = this.Query(tx)
+	if len(messageType) > 0 {
+		query.Attr("type", []string{"*", messageType}) // *表示所有的
+	}
+	if len(role) > 0 {
+		query.Attr("role", role)
+	}
+	_, err = query.
+		State(MessageReceiverStateEnabled).
+		AscPk().
+		Slice(&result).
+		FindAll()
+	if err != nil || len(result) > 0 {
+		return
+	}
+
+	return
+}

internal/db/models/message_receiver_dao_test.go

@@ -1,6 +1,30 @@
 package models
 
 import (
+	"github.com/TeaOSLab/EdgeCommon/pkg/nodeconfigs"
 	_ "github.com/go-sql-driver/mysql"
 	_ "github.com/iwind/TeaGo/bootstrap"
+	"github.com/iwind/TeaGo/dbs"
+	"github.com/iwind/TeaGo/logs"
+	"testing"
 )
+
+func TestMessageReceiverDAO_FindEnabledBestFitReceivers(t *testing.T) {
+	var tx *dbs.Tx
+
+	{
+		receivers, err := NewMessageReceiverDAO().FindEnabledBestFitReceivers(tx, nodeconfigs.NodeRoleNode, 18, 1, 2, "*")
+		if err != nil {
+			t.Fatal(err)
+		}
+		logs.PrintAsJSON(receivers, t)
+	}
+
+	{
+		receivers, err := NewMessageReceiverDAO().FindEnabledBestFitReceivers(tx, nodeconfigs.NodeRoleNode, 30, 1, 2, "*")
+		if err != nil {
+			t.Fatal(err)
+		}
+		logs.PrintAsJSON(receivers, t)
+	}
+}

internal/db/models/message_receiver_model.go

@@ -3,6 +3,7 @@ package models
 // MessageReceiver 消息通知接收人
 type MessageReceiver struct {
 	Id               uint32 `field:"id"`               // ID
+	Role             string `field:"role"`             // 节点角色
 	ClusterId        uint32 `field:"clusterId"`        // 集群ID
 	NodeId           uint32 `field:"nodeId"`           // 节点ID
 	ServerId         uint32 `field:"serverId"`         // 服务ID
@@ -15,6 +16,7 @@ type MessageReceiver struct {
 
 type MessageReceiverOperator struct {
 	Id               interface{} // ID
+	Role             interface{} // 节点角色
 	ClusterId        interface{} // 集群ID
 	NodeId           interface{} // 节点ID
 	ServerId         interface{} // 服务ID

internal/db/models/message_recipient_dao.go

@@ -7,6 +7,8 @@ import (
 	_ "github.com/go-sql-driver/mysql"
 	"github.com/iwind/TeaGo/Tea"
 	"github.com/iwind/TeaGo/dbs"
+	"github.com/iwind/TeaGo/maps"
+	"github.com/iwind/TeaGo/types"
 )
 
 const (
@@ -54,14 +56,27 @@ func (this *MessageRecipientDAO) DisableMessageRecipient(tx *dbs.Tx, id int64) e
 }
 
 // FindEnabledMessageRecipient 查找启用中的条目
-func (this *MessageRecipientDAO) FindEnabledMessageRecipient(tx *dbs.Tx, id int64) (*MessageRecipient, error) {
+func (this *MessageRecipientDAO) FindEnabledMessageRecipient(tx *dbs.Tx, recipientId int64, cacheMap maps.Map,
+) (*MessageRecipient, error) {
+	if cacheMap == nil {
+		cacheMap = maps.Map{}
+	}
+	var cacheKey = this.Table + ":record:" + types.String(recipientId)
+	var cache = cacheMap.Get(cacheKey)
+	if cache != nil {
+		return cache.(*MessageRecipient), nil
+	}
+
 	result, err := this.Query(tx).
-		Pk(id).
+		Pk(recipientId).
 		Attr("state", MessageRecipientStateEnabled).
 		Find()
 	if result == nil {
 		return nil, err
 	}
+
+	cacheMap[cacheKey] = result
+
 	return result.(*MessageRecipient), err
 }
 

internal/db/models/message_target.go

@@ -1,11 +0,0 @@
-// Copyright 2021 Liuxiangchao iwind.liu@gmail.com. All rights reserved.
-
-package models
-
-// MessageTaskTarget 消息接收对象
-// 每个字段不一定都有值
-type MessageTaskTarget struct {
-	ClusterId int64 // 集群ID
-	NodeId    int64 // 节点ID
-	ServerId  int64 // 服务ID
-}

internal/db/models/message_task_dao.go

@@ -2,9 +2,13 @@ package models
 
 import (
 	"github.com/TeaOSLab/EdgeAPI/internal/errors"
+	"github.com/TeaOSLab/EdgeAPI/internal/remotelogs"
+	"github.com/TeaOSLab/EdgeCommon/pkg/nodeconfigs"
 	_ "github.com/go-sql-driver/mysql"
 	"github.com/iwind/TeaGo/Tea"
 	"github.com/iwind/TeaGo/dbs"
+	"github.com/iwind/TeaGo/rands"
+	timeutil "github.com/iwind/TeaGo/utils/time"
 	"time"
 )
 
@@ -35,6 +39,21 @@ func NewMessageTaskDAO() *MessageTaskDAO {
 
 var SharedMessageTaskDAO *MessageTaskDAO
 
+func init() {
+	dbs.OnReadyDone(func() {
+		// 清理数据任务
+		var ticker = time.NewTicker(time.Duration(rands.Int(24, 48)) * time.Hour)
+		go func() {
+			for range ticker.C {
+				err := SharedMessageTaskDAO.CleanExpiredMessageTasks(nil, 30) // 只保留30天
+				if err != nil {
+					remotelogs.Error("SharedMessageTaskDAO", "clean expired data failed: "+err.Error())
+				}
+			}
+		}()
+	})
+}
+
 func init() {
 	dbs.OnReady(func() {
 		SharedMessageTaskDAO = NewMessageTaskDAO()
@@ -80,6 +99,7 @@ func (this *MessageTaskDAO) CreateMessageTask(tx *dbs.Tx, recipientId int64, ins
 	op.Subject = subject
 	op.Body = body
 	op.IsPrimary = isPrimary
+	op.Day = timeutil.Format("Ymd")
 	op.Status = MessageTaskStatusNone
 	op.State = MessageTaskStateEnabled
 	return this.SaveInt64(tx, op)
@@ -101,6 +121,28 @@ func (this *MessageTaskDAO) FindSendingMessageTasks(tx *dbs.Tx, size int64) (res
 	return
 }
 
+// CountMessageTasksWithStatus 根据状态计算任务数量
+func (this *MessageTaskDAO) CountMessageTasksWithStatus(tx *dbs.Tx, status MessageTaskStatus) (int64, error) {
+	return this.Query(tx).
+		State(MessageTaskStateEnabled).
+		Attr("status", status).
+		Count()
+}
+
+// ListMessageTasksWithStatus 根据状态列出单页任务
+func (this *MessageTaskDAO) ListMessageTasksWithStatus(tx *dbs.Tx, status MessageTaskStatus, offset int64, size int64) (result []*MessageTask, err error) {
+	_, err = this.Query(tx).
+		State(MessageTaskStateEnabled).
+		Attr("status", status).
+		Desc("isPrimary").
+		AscPk().
+		Offset(offset).
+		Limit(size).
+		Slice(&result).
+		FindAll()
+	return
+}
+
 // UpdateMessageTaskStatus 设置发送的状态
 func (this *MessageTaskDAO) UpdateMessageTaskStatus(tx *dbs.Tx, taskId int64, status MessageTaskStatus, result []byte) error {
 	if taskId <= 0 {
@@ -117,8 +159,8 @@ func (this *MessageTaskDAO) UpdateMessageTaskStatus(tx *dbs.Tx, taskId int64, st
 }
 
 // CreateMessageTasks 从集群、节点或者服务中创建任务
-func (this *MessageTaskDAO) CreateMessageTasks(tx *dbs.Tx, target MessageTaskTarget, messageType MessageType, subject string, body string) error {
+func (this *MessageTaskDAO) CreateMessageTasks(tx *dbs.Tx, role nodeconfigs.NodeRole, clusterId int64, nodeId int64, serverId int64, messageType MessageType, subject string, body string) error {
-	receivers, err := SharedMessageReceiverDAO.FindAllEnabledReceivers(tx, target, messageType)
+	receivers, err := SharedMessageReceiverDAO.FindEnabledBestFitReceivers(tx, role, clusterId, nodeId, serverId, messageType)
 	if err != nil {
 		return err
 	}
@@ -150,3 +192,16 @@ func (this *MessageTaskDAO) CreateMessageTasks(tx *dbs.Tx, target MessageTaskTar
 
 	return nil
 }
+
+// CleanExpiredMessageTasks 清理
+func (this *MessageTaskDAO) CleanExpiredMessageTasks(tx *dbs.Tx, days int) error {
+	if days <= 0 {
+		days = 30
+	}
+	var day = timeutil.Format("Ymd", time.Now().AddDate(0, 0, -days))
+	_, err := this.Query(tx).
+		Where("(day IS NULL OR day<:day)").
+		Param("day", day).
+		Delete()
+	return err
+}

internal/db/models/message_task_log_dao.go

@@ -1,13 +1,32 @@
 package models
 
 import (
+	"github.com/TeaOSLab/EdgeAPI/internal/remotelogs"
 	_ "github.com/go-sql-driver/mysql"
 	"github.com/iwind/TeaGo/Tea"
 	"github.com/iwind/TeaGo/dbs"
+	"github.com/iwind/TeaGo/rands"
+	timeutil "github.com/iwind/TeaGo/utils/time"
+	"time"
 )
 
 type MessageTaskLogDAO dbs.DAO
 
+func init() {
+	dbs.OnReadyDone(func() {
+		// 清理数据任务
+		var ticker = time.NewTicker(time.Duration(rands.Int(24, 48)) * time.Hour)
+		go func() {
+			for range ticker.C {
+				err := SharedMessageTaskLogDAO.CleanExpiredLogs(nil, 30) // 只保留30天
+				if err != nil {
+					remotelogs.Error("SharedMessageTaskLogDAO", "clean expired data failed: "+err.Error())
+				}
+			}
+		}()
+	})
+}
+
 func NewMessageTaskLogDAO() *MessageTaskLogDAO {
 	return dbs.NewDAO(&MessageTaskLogDAO{
 		DAOObject: dbs.DAOObject{
@@ -27,25 +46,28 @@ func init() {
 	})
 }
 
-// 创建日志
+// CreateLog 创建日志
 func (this *MessageTaskLogDAO) CreateLog(tx *dbs.Tx, taskId int64, isOk bool, errMsg string, response string) error {
 	op := NewMessageTaskLogOperator()
 	op.TaskId = taskId
 	op.IsOk = isOk
 	op.Error = errMsg
 	op.Response = response
+	op.Day = timeutil.Format("Ymd")
 	return this.Save(tx, op)
 }
 
-// 计算日志数量
+// CountLogs 计算日志数量
 func (this *MessageTaskLogDAO) CountLogs(tx *dbs.Tx) (int64, error) {
 	return this.Query(tx).
+		Where("taskId IN (SELECT id FROM " + SharedMessageTaskDAO.Table + ")").
 		Count()
 }
 
-// 列出单页日志
+// ListLogs 列出单页日志
 func (this *MessageTaskLogDAO) ListLogs(tx *dbs.Tx, offset int64, size int64) (result []*MessageTaskLog, err error) {
 	_, err = this.Query(tx).
+		Where("taskId IN (SELECT id FROM " + SharedMessageTaskDAO.Table + ")").
 		Offset(offset).
 		Limit(size).
 		DescPk().
@@ -53,3 +75,16 @@ func (this *MessageTaskLogDAO) ListLogs(tx *dbs.Tx, offset int64, size int64) (r
 		FindAll()
 	return
 }
+
+// CleanExpiredLogs 清理
+func (this *MessageTaskLogDAO) CleanExpiredLogs(tx *dbs.Tx, days int) error {
+	if days <= 0 {
+		days = 30
+	}
+	var day = timeutil.Format("Ymd", time.Now().AddDate(0, 0, -days))
+	_, err := this.Query(tx).
+		Where("(day IS NULL OR day<:day)").
+		Param("day", day).
+		Delete()
+	return err
+}

internal/db/models/message_task_log_model.go

@@ -1,6 +1,6 @@
 package models
 
-// 消息发送日志
+// MessageTaskLog 消息发送日志
 type MessageTaskLog struct {
 	Id        uint64 `field:"id"`        // ID
 	TaskId    uint64 `field:"taskId"`    // 任务ID
@@ -8,6 +8,7 @@ type MessageTaskLog struct {
 	IsOk      uint8  `field:"isOk"`      // 是否成功
 	Error     string `field:"error"`     // 错误信息
 	Response  string `field:"response"`  // 响应信息
+	Day       string `field:"day"`       // YYYYMMDD
 }
 
 type MessageTaskLogOperator struct {
@@ -17,6 +18,7 @@ type MessageTaskLogOperator struct {
 	IsOk      interface{} // 是否成功
 	Error     interface{} // 错误信息
 	Response  interface{} // 响应信息
+	Day       interface{} // YYYYMMDD
 }
 
 func NewMessageTaskLogOperator() *MessageTaskLogOperator {

internal/db/models/message_task_model.go

@@ -1,6 +1,6 @@
 package models
 
-//
+// MessageTask 消息发送相关任务
 type MessageTask struct {
 	Id          uint64 `field:"id"`          // ID
 	RecipientId uint32 `field:"recipientId"` // 接收人ID
@@ -13,6 +13,7 @@ type MessageTask struct {
 	SentAt      uint64 `field:"sentAt"`      // 最后一次发送时间
 	State       uint8  `field:"state"`       // 状态
 	Result      string `field:"result"`      // 结果
+	Day         string `field:"day"`         // YYYYMMDD
 	IsPrimary   uint8  `field:"isPrimary"`   // 是否优先
 }
 
@@ -28,6 +29,7 @@ type MessageTaskOperator struct {
 	SentAt      interface{} // 最后一次发送时间
 	State       interface{} // 状态
 	Result      interface{} // 结果
+	Day         interface{} // YYYYMMDD
 	IsPrimary   interface{} // 是否优先
 }
 

internal/db/models/node_log_dao.go

@@ -2,6 +2,7 @@ package models
 
 import (
 	"github.com/TeaOSLab/EdgeAPI/internal/errors"
+	"github.com/TeaOSLab/EdgeAPI/internal/remotelogs"
 	"github.com/TeaOSLab/EdgeCommon/pkg/configutils"
 	"github.com/TeaOSLab/EdgeCommon/pkg/nodeconfigs"
 	_ "github.com/go-sql-driver/mysql"
@@ -33,6 +34,9 @@ var SharedNodeLogDAO *NodeLogDAO
 func init() {
 	dbs.OnReady(func() {
 		SharedNodeLogDAO = NewNodeLogDAO()
+
+		// 设置日志存储
+		remotelogs.SetDAO(SharedNodeLogDAO)
 	})
 }
 

internal/remotelogs/dao_interface.go

@@ -0,0 +1,12 @@
+// Copyright 2021 Liuxiangchao iwind.liu@gmail.com. All rights reserved.
+
+package remotelogs
+
+import (
+	"github.com/TeaOSLab/EdgeCommon/pkg/nodeconfigs"
+	"github.com/iwind/TeaGo/dbs"
+)
+
+type DAOInterface interface {
+	CreateLog(tx *dbs.Tx, nodeRole nodeconfigs.NodeRole, nodeId int64, serverId int64, originId int64, level string, tag string, description string, createdAt int64) error
+}

internal/remotelogs/utils.go

@@ -3,7 +3,6 @@ package remotelogs
 import (
 	"github.com/TeaOSLab/EdgeAPI/internal/configs"
 	teaconst "github.com/TeaOSLab/EdgeAPI/internal/const"
-	"github.com/TeaOSLab/EdgeAPI/internal/db/models"
 	"github.com/TeaOSLab/EdgeCommon/pkg/nodeconfigs"
 	"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
 	"github.com/iwind/TeaGo/logs"
@@ -11,6 +10,7 @@ import (
 )
 
 var logChan = make(chan *pb.NodeLog, 1024)
+var sharedDAO DAOInterface
 
 func init() {
 	// 定期上传日志
@@ -25,7 +25,7 @@ func init() {
 	}()
 }
 
-// 打印普通信息
+// Println 打印普通信息
 func Println(tag string, description string) {
 	logs.Println("[" + tag + "]" + description)
 
@@ -48,7 +48,7 @@ func Println(tag string, description string) {
 	}
 }
 
-// 打印警告信息
+// Warn 打印警告信息
 func Warn(tag string, description string) {
 	logs.Println("[" + tag + "]" + description)
 
@@ -71,7 +71,7 @@ func Warn(tag string, description string) {
 	}
 }
 
-// 打印错误信息
+// Error 打印错误信息
 func Error(tag string, description string) {
 	logs.Println("[" + tag + "]" + description)
 
@@ -94,13 +94,22 @@ func Error(tag string, description string) {
 	}
 }
 
+// SetDAO 设置存储接口
+func SetDAO(dao DAOInterface) {
+	sharedDAO = dao
+}
+
 // 上传日志
 func uploadLogs() error {
+	if sharedDAO == nil {
+		return nil
+	}
+
 Loop:
 	for {
 		select {
 		case log := <-logChan:
-			err := models.SharedNodeLogDAO.CreateLog(nil, nodeconfigs.NodeRoleAPI, log.NodeId, log.ServerId, log.OriginId, log.Level, log.Tag, log.Description, log.CreatedAt)
+			err := sharedDAO.CreateLog(nil, nodeconfigs.NodeRoleAPI, log.NodeId, log.ServerId, log.OriginId, log.Level, log.Tag, log.Description, log.CreatedAt)
 			if err != nil {
 				return err
 			}

internal/rpc/services/service_firewall.go

@@ -7,6 +7,7 @@ import (
 	"github.com/TeaOSLab/EdgeAPI/internal/db/models"
 	"github.com/TeaOSLab/EdgeAPI/internal/db/models/stats"
 	"github.com/TeaOSLab/EdgeAPI/internal/utils"
+	"github.com/TeaOSLab/EdgeCommon/pkg/nodeconfigs"
 	"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
 	"github.com/iwind/TeaGo/types"
 	timeutil "github.com/iwind/TeaGo/utils/time"
@@ -224,11 +225,7 @@ func (this *FirewallService) NotifyHTTPFirewallEvent(ctx context.Context, req *p
 		"\n规则分组：" + ruleGroupName +
 		"\n规则集：" + ruleSetName +
 		"\n时间：" + timeutil.FormatTime("Y-m-d H:i:s", req.CreatedAt)
-	err = models.SharedMessageTaskDAO.CreateMessageTasks(tx, models.MessageTaskTarget{
+	err = models.SharedMessageTaskDAO.CreateMessageTasks(tx, nodeconfigs.NodeRoleNode, clusterId, nodeId, req.ServerId, models.MessageTypeFirewallEvent, "触发防火墙事件", msg)
-		ClusterId: clusterId,
-		NodeId:    nodeId,
-		ServerId:  req.ServerId,
-	}, models.MessageTypeFirewallEvent, "发生防火墙事件", msg)
 	if err != nil {
 		return nil, err
 	}

internal/rpc/services/service_message_media_instance.go

@@ -8,12 +8,12 @@ import (
 	"github.com/iwind/TeaGo/maps"
 )
 
-// 消息媒介实例服务
+// MessageMediaInstanceService 消息媒介实例服务
 type MessageMediaInstanceService struct {
 	BaseService
 }
 
-// 创建消息媒介实例
+// CreateMessageMediaInstance 创建消息媒介实例
 func (this *MessageMediaInstanceService) CreateMessageMediaInstance(ctx context.Context, req *pb.CreateMessageMediaInstanceRequest) (*pb.CreateMessageMediaInstanceResponse, error) {
 	_, err := this.ValidateAdmin(ctx, 0)
 	if err != nil {
@@ -38,7 +38,7 @@ func (this *MessageMediaInstanceService) CreateMessageMediaInstance(ctx context.
 	return &pb.CreateMessageMediaInstanceResponse{MessageMediaInstanceId: instanceId}, nil
 }
 
-// 修改消息实例
+// UpdateMessageMediaInstance 修改消息实例
 func (this *MessageMediaInstanceService) UpdateMessageMediaInstance(ctx context.Context, req *pb.UpdateMessageMediaInstanceRequest) (*pb.RPCSuccess, error) {
 	_, err := this.ValidateAdmin(ctx, 0)
 	if err != nil {
@@ -62,7 +62,7 @@ func (this *MessageMediaInstanceService) UpdateMessageMediaInstance(ctx context.
 	return this.Success()
 }
 
-// 删除媒介实例
+// DeleteMessageMediaInstance 删除媒介实例
 func (this *MessageMediaInstanceService) DeleteMessageMediaInstance(ctx context.Context, req *pb.DeleteMessageMediaInstanceRequest) (*pb.RPCSuccess, error) {
 	_, err := this.ValidateAdmin(ctx, 0)
 	if err != nil {
@@ -77,7 +77,7 @@ func (this *MessageMediaInstanceService) DeleteMessageMediaInstance(ctx context.
 	return this.Success()
 }
 
-// 计算媒介实例数量
+// CountAllEnabledMessageMediaInstances 计算媒介实例数量
 func (this *MessageMediaInstanceService) CountAllEnabledMessageMediaInstances(ctx context.Context, req *pb.CountAllEnabledMessageMediaInstancesRequest) (*pb.RPCCountResponse, error) {
 	_, err := this.ValidateAdmin(ctx, 0)
 	if err != nil {
@@ -93,7 +93,7 @@ func (this *MessageMediaInstanceService) CountAllEnabledMessageMediaInstances(ct
 	return this.SuccessCount(count)
 }
 
-// 列出单页媒介实例
+// ListEnabledMessageMediaInstances 列出单页媒介实例
 func (this *MessageMediaInstanceService) ListEnabledMessageMediaInstances(ctx context.Context, req *pb.ListEnabledMessageMediaInstancesRequest) (*pb.ListEnabledMessageMediaInstancesResponse, error) {
 	_, err := this.ValidateAdmin(ctx, 0)
 	if err != nil {
@@ -137,7 +137,7 @@ func (this *MessageMediaInstanceService) ListEnabledMessageMediaInstances(ctx co
 	return &pb.ListEnabledMessageMediaInstancesResponse{MessageMediaInstances: pbInstances}, nil
 }
 
-// 查找单个媒介实例信息
+// FindEnabledMessageMediaInstance 查找单个媒介实例信息
 func (this *MessageMediaInstanceService) FindEnabledMessageMediaInstance(ctx context.Context, req *pb.FindEnabledMessageMediaInstanceRequest) (*pb.FindEnabledMessageMediaInstanceResponse, error) {
 	_, err := this.ValidateAdmin(ctx, 0)
 	if err != nil {
@@ -145,7 +145,8 @@ func (this *MessageMediaInstanceService) FindEnabledMessageMediaInstance(ctx con
 	}
 
 	var tx = this.NullTx()
-	instance, err := models.SharedMessageMediaInstanceDAO.FindEnabledMessageMediaInstance(tx, req.MessageMediaInstanceId)
+	var cacheMap = maps.Map{}
+	instance, err := models.SharedMessageMediaInstanceDAO.FindEnabledMessageMediaInstance(tx, req.MessageMediaInstanceId, cacheMap)
 	if err != nil {
 		return nil, err
 	}

internal/rpc/services/service_message_receiver.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"github.com/TeaOSLab/EdgeAPI/internal/db/models"
+	"github.com/TeaOSLab/EdgeCommon/pkg/nodeconfigs"
 	"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
 	"github.com/iwind/TeaGo/dbs"
 	"github.com/iwind/TeaGo/maps"
@@ -21,6 +22,10 @@ func (this *MessageReceiverService) UpdateMessageReceivers(ctx context.Context,
 		return nil, err
 	}
 
+	if len(req.Role) == 0 {
+		req.Role = nodeconfigs.NodeRoleNode
+	}
+
 	params := maps.Map{}
 	if len(req.ParamsJSON) > 0 {
 		err = json.Unmarshal(req.ParamsJSON, &params)
@@ -37,11 +42,7 @@ func (this *MessageReceiverService) UpdateMessageReceivers(ctx context.Context,
 
 		for messageType, options := range req.RecipientOptions {
 			for _, option := range options.RecipientOptions {
-				_, err := models.SharedMessageReceiverDAO.CreateReceiver(tx, models.MessageTaskTarget{
+				_, err := models.SharedMessageReceiverDAO.CreateReceiver(tx, req.Role, req.NodeClusterId, req.NodeId, req.ServerId, messageType, params, option.MessageRecipientId, option.MessageRecipientGroupId)
-					ClusterId: req.NodeClusterId,
-					NodeId:    req.NodeId,
-					ServerId:  req.ServerId,
-				}, messageType, params, option.MessageRecipientId, option.MessageRecipientGroupId)
 				if err != nil {
 					return err
 				}
@@ -63,12 +64,13 @@ func (this *MessageReceiverService) FindAllEnabledMessageReceivers(ctx context.C
 		return nil, err
 	}
 
+	if len(req.Role) == 0 {
+		req.Role = nodeconfigs.NodeRoleNode
+	}
+
 	var tx = this.NullTx()
-	receivers, err := models.SharedMessageReceiverDAO.FindAllEnabledReceivers(tx, models.MessageTaskTarget{
+	var cacheMap = maps.Map{}
-		ClusterId: req.NodeClusterId,
+	receivers, err := models.SharedMessageReceiverDAO.FindAllEnabledReceivers(tx, req.Role, req.NodeClusterId, req.NodeId, req.ServerId, "")
-		NodeId:    req.NodeId,
-		ServerId:  req.ServerId,
-	}, "")
 	if err != nil {
 		return nil, err
 	}
@@ -78,7 +80,7 @@ func (this *MessageReceiverService) FindAllEnabledMessageReceivers(ctx context.C
 
 		// 接收人
 		if receiver.RecipientId > 0 {
-			recipient, err := models.SharedMessageRecipientDAO.FindEnabledMessageRecipient(tx, int64(receiver.RecipientId))
+			recipient, err := models.SharedMessageRecipientDAO.FindEnabledMessageRecipient(tx, int64(receiver.RecipientId), cacheMap)
 			if err != nil {
 				return nil, err
 			}
@@ -96,7 +98,7 @@ func (this *MessageReceiverService) FindAllEnabledMessageReceivers(ctx context.C
 			}
 
 			// 接收人
-			instance, err := models.SharedMessageMediaInstanceDAO.FindEnabledMessageMediaInstance(tx, int64(recipient.InstanceId))
+			instance, err := models.SharedMessageMediaInstanceDAO.FindEnabledMessageMediaInstance(tx, int64(recipient.InstanceId), cacheMap)
 			if err != nil {
 				return nil, err
 			}
@@ -177,12 +179,12 @@ func (this *MessageReceiverService) CountAllEnabledMessageReceivers(ctx context.
 		return nil, err
 	}
 
+	if len(req.Role) == 0 {
+		req.Role = nodeconfigs.NodeRoleNode
+	}
+
 	var tx = this.NullTx()
-	count, err := models.SharedMessageReceiverDAO.CountAllEnabledReceivers(tx, models.MessageTaskTarget{
+	count, err := models.SharedMessageReceiverDAO.CountAllEnabledReceivers(tx, req.Role, req.NodeClusterId, req.NodeId, req.ServerId, "")
-		ClusterId: req.NodeClusterId,
-		NodeId:    req.NodeId,
-		ServerId:  req.ServerId,
-	}, "")
 	if err != nil {
 		return nil, err
 	}

internal/rpc/services/service_message_recipient.go

@@ -4,14 +4,15 @@ import (
 	"context"
 	"github.com/TeaOSLab/EdgeAPI/internal/db/models"
 	"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
+	"github.com/iwind/TeaGo/maps"
 )
 
-// 消息接收人服务
+// MessageRecipientService 消息接收人服务
 type MessageRecipientService struct {
 	BaseService
 }
 
-// 创建接收人
+// CreateMessageRecipient 创建接收人
 func (this *MessageRecipientService) CreateMessageRecipient(ctx context.Context, req *pb.CreateMessageRecipientRequest) (*pb.CreateMessageRecipientResponse, error) {
 	_, err := this.ValidateAdmin(ctx, 0)
 	if err != nil {
@@ -27,7 +28,7 @@ func (this *MessageRecipientService) CreateMessageRecipient(ctx context.Context,
 	return &pb.CreateMessageRecipientResponse{MessageRecipientId: recipientId}, nil
 }
 
-// 修改接收人
+// UpdateMessageRecipient 修改接收人
 func (this *MessageRecipientService) UpdateMessageRecipient(ctx context.Context, req *pb.UpdateMessageRecipientRequest) (*pb.RPCSuccess, error) {
 	_, err := this.ValidateAdmin(ctx, 0)
 	if err != nil {
@@ -43,7 +44,7 @@ func (this *MessageRecipientService) UpdateMessageRecipient(ctx context.Context,
 	return this.Success()
 }
 
-// 删除接收人
+// DeleteMessageRecipient 删除接收人
 func (this *MessageRecipientService) DeleteMessageRecipient(ctx context.Context, req *pb.DeleteMessageRecipientRequest) (*pb.RPCSuccess, error) {
 	_, err := this.ValidateAdmin(ctx, 0)
 	if err != nil {
@@ -58,7 +59,7 @@ func (this *MessageRecipientService) DeleteMessageRecipient(ctx context.Context,
 	return this.Success()
 }
 
-// 计算接收人数量
+// CountAllEnabledMessageRecipients 计算接收人数量
 func (this *MessageRecipientService) CountAllEnabledMessageRecipients(ctx context.Context, req *pb.CountAllEnabledMessageRecipientsRequest) (*pb.RPCCountResponse, error) {
 	_, err := this.ValidateAdmin(ctx, 0)
 	if err != nil {
@@ -74,7 +75,7 @@ func (this *MessageRecipientService) CountAllEnabledMessageRecipients(ctx contex
 	return this.SuccessCount(count)
 }
 
-// 列出单页接收人
+// ListEnabledMessageRecipients 列出单页接收人
 func (this *MessageRecipientService) ListEnabledMessageRecipients(ctx context.Context, req *pb.ListEnabledMessageRecipientsRequest) (*pb.ListEnabledMessageRecipientsResponse, error) {
 	_, err := this.ValidateAdmin(ctx, 0)
 	if err != nil {
@@ -82,6 +83,7 @@ func (this *MessageRecipientService) ListEnabledMessageRecipients(ctx context.Co
 	}
 
 	var tx = this.NullTx()
+	var cacheMap = maps.Map{}
 	recipients, err := models.SharedMessageRecipientDAO.ListAllEnabledRecipients(tx, req.AdminId, req.MessageRecipientGroupId, req.MediaType, req.Keyword, req.Offset, req.Size)
 	if err != nil {
 		return nil, err
@@ -104,7 +106,7 @@ func (this *MessageRecipientService) ListEnabledMessageRecipients(ctx context.Co
 		}
 
 		// 媒介实例
-		instance, err := models.SharedMessageMediaInstanceDAO.FindEnabledMessageMediaInstance(tx, int64(recipient.InstanceId))
+		instance, err := models.SharedMessageMediaInstanceDAO.FindEnabledMessageMediaInstance(tx, int64(recipient.InstanceId), cacheMap)
 		if err != nil {
 			return nil, err
 		}
@@ -151,7 +153,7 @@ func (this *MessageRecipientService) ListEnabledMessageRecipients(ctx context.Co
 	return &pb.ListEnabledMessageRecipientsResponse{MessageRecipients: pbRecipients}, nil
 }
 
-// 查找单个接收人信息
+// FindEnabledMessageRecipient 查找单个接收人信息
 func (this *MessageRecipientService) FindEnabledMessageRecipient(ctx context.Context, req *pb.FindEnabledMessageRecipientRequest) (*pb.FindEnabledMessageRecipientResponse, error) {
 	_, err := this.ValidateAdmin(ctx, 0)
 	if err != nil {
@@ -159,7 +161,8 @@ func (this *MessageRecipientService) FindEnabledMessageRecipient(ctx context.Con
 	}
 
 	var tx = this.NullTx()
-	recipient, err := models.SharedMessageRecipientDAO.FindEnabledMessageRecipient(tx, req.MessageRecipientId)
+	var cacheMap = maps.Map{}
+	recipient, err := models.SharedMessageRecipientDAO.FindEnabledMessageRecipient(tx, req.MessageRecipientId, cacheMap)
 	if err != nil {
 		return nil, err
 	}
@@ -183,7 +186,7 @@ func (this *MessageRecipientService) FindEnabledMessageRecipient(ctx context.Con
 	}
 
 	// 媒介实例
-	instance, err := models.SharedMessageMediaInstanceDAO.FindEnabledMessageMediaInstance(tx, int64(recipient.InstanceId))
+	instance, err := models.SharedMessageMediaInstanceDAO.FindEnabledMessageMediaInstance(tx, int64(recipient.InstanceId), cacheMap)
 	if err != nil {
 		return nil, err
 	}

internal/rpc/services/service_message_task.go

@@ -37,6 +37,7 @@ func (this *MessageTaskService) FindSendingMessageTasks(ctx context.Context, req
 	}
 
 	var tx = this.NullTx()
+	var cacheMap = maps.Map{}
 	tasks, err := models.SharedMessageTaskDAO.FindSendingMessageTasks(tx, req.Size)
 	if err != nil {
 		return nil, err
@@ -45,8 +46,7 @@ func (this *MessageTaskService) FindSendingMessageTasks(ctx context.Context, req
 	for _, task := range tasks {
 		var pbRecipient *pb.MessageRecipient
 		if task.RecipientId > 0 {
-			// TODO 需要缓存以提升性能
+			recipient, err := models.SharedMessageRecipientDAO.FindEnabledMessageRecipient(tx, int64(task.RecipientId), cacheMap)
-			recipient, err := models.SharedMessageRecipientDAO.FindEnabledMessageRecipient(tx, int64(task.RecipientId))
 			if err != nil {
 				return nil, err
 			}
@@ -60,8 +60,7 @@ func (this *MessageTaskService) FindSendingMessageTasks(ctx context.Context, req
 			}
 
 			// 媒介
-			// TODO 需要缓存以提升性能
+			instance, err := models.SharedMessageMediaInstanceDAO.FindEnabledMessageMediaInstance(tx, int64(recipient.InstanceId), cacheMap)
-			instance, err := models.SharedMessageMediaInstanceDAO.FindEnabledMessageMediaInstance(tx, int64(recipient.InstanceId))
 			if err != nil {
 				return nil, err
 			}
@@ -87,8 +86,7 @@ func (this *MessageTaskService) FindSendingMessageTasks(ctx context.Context, req
 			}
 		} else { // 没有指定既定的接收人
 			// 媒介
-			// TODO 需要缓存以提升性能
+			instance, err := models.SharedMessageMediaInstanceDAO.FindEnabledMessageMediaInstance(tx, int64(task.InstanceId), cacheMap)
-			instance, err := models.SharedMessageMediaInstanceDAO.FindEnabledMessageMediaInstance(tx, int64(task.InstanceId))
 			if err != nil {
 				return nil, err
 			}
@@ -186,6 +184,7 @@ func (this *MessageTaskService) FindEnabledMessageTask(ctx context.Context, req
 	}
 
 	var tx = this.NullTx()
+	var cacheMap = maps.Map{}
 	task, err := models.SharedMessageTaskDAO.FindEnabledMessageTask(tx, req.MessageTaskId)
 	if err != nil {
 		return nil, err
@@ -194,10 +193,9 @@ func (this *MessageTaskService) FindEnabledMessageTask(ctx context.Context, req
 		return &pb.FindEnabledMessageTaskResponse{MessageTask: nil}, nil
 	}
 
-	// TODO 需要缓存以提升性能
 	var pbRecipient *pb.MessageRecipient
 	if task.RecipientId > 0 {
-		recipient, err := models.SharedMessageRecipientDAO.FindEnabledMessageRecipient(tx, int64(task.RecipientId))
+		recipient, err := models.SharedMessageRecipientDAO.FindEnabledMessageRecipient(tx, int64(task.RecipientId), cacheMap)
 		if err != nil {
 			return nil, err
 		}
@@ -211,8 +209,7 @@ func (this *MessageTaskService) FindEnabledMessageTask(ctx context.Context, req
 		}
 
 		// 媒介
-		// TODO 需要缓存以提升性能
+		instance, err := models.SharedMessageMediaInstanceDAO.FindEnabledMessageMediaInstance(tx, int64(recipient.InstanceId), cacheMap)
-		instance, err := models.SharedMessageMediaInstanceDAO.FindEnabledMessageMediaInstance(tx, int64(recipient.InstanceId))
 		if err != nil {
 			return nil, err
 		}
@@ -237,8 +234,7 @@ func (this *MessageTaskService) FindEnabledMessageTask(ctx context.Context, req
 		}
 	} else { // 没有指定既定的接收人
 		// 媒介
-		// TODO 需要缓存以提升性能
+		instance, err := models.SharedMessageMediaInstanceDAO.FindEnabledMessageMediaInstance(tx, int64(task.InstanceId), cacheMap)
-		instance, err := models.SharedMessageMediaInstanceDAO.FindEnabledMessageMediaInstance(tx, int64(task.InstanceId))
 		if err != nil {
 			return nil, err
 		}
@@ -282,3 +278,127 @@ func (this *MessageTaskService) FindEnabledMessageTask(ctx context.Context, req
 		Result:           result,
 	}}, nil
 }
+
+// CountMessageTasksWithStatus 计算某个状态的消息任务数量
+func (this *MessageTaskService) CountMessageTasksWithStatus(ctx context.Context, req *pb.CountMessageTasksWithStatusRequest) (*pb.RPCCountResponse, error) {
+	_, err := this.ValidateAdmin(ctx, 0)
+	if err != nil {
+		return nil, err
+	}
+
+	var tx = this.NullTx()
+	count, err := models.SharedMessageTaskDAO.CountMessageTasksWithStatus(tx, types.Int(req.Status))
+	if err != nil {
+		return nil, err
+	}
+
+	return this.SuccessCount(count)
+}
+
+// ListMessageTasksWithStatus 根据状态列出某页任务
+func (this *MessageTaskService) ListMessageTasksWithStatus(ctx context.Context, req *pb.ListMessageTasksWithStatusRequest) (*pb.ListMessageTasksWithStatusResponse, error) {
+	_, err := this.ValidateAdmin(ctx, 0)
+	if err != nil {
+		return nil, err
+	}
+
+	var tx = this.NullTx()
+	var cacheMap = maps.Map{}
+	tasks, err := models.SharedMessageTaskDAO.ListMessageTasksWithStatus(tx, types.Int(req.Status), req.Offset, req.Size)
+	if err != nil {
+		return nil, err
+	}
+
+	var pbTasks = []*pb.MessageTask{}
+	for _, task := range tasks {
+		var pbRecipient *pb.MessageRecipient
+		if task.RecipientId > 0 {
+			recipient, err := models.SharedMessageRecipientDAO.FindEnabledMessageRecipient(tx, int64(task.RecipientId), cacheMap)
+			if err != nil {
+				return nil, err
+			}
+			if recipient == nil || recipient.IsOn == 0 {
+				// 如果发送人已经删除或者禁用，则删除此消息
+				err = models.SharedMessageTaskDAO.DisableMessageTask(tx, int64(task.Id))
+				if err != nil {
+					return nil, err
+				}
+				continue
+			}
+
+			// 媒介
+			instance, err := models.SharedMessageMediaInstanceDAO.FindEnabledMessageMediaInstance(tx, int64(recipient.InstanceId), cacheMap)
+			if err != nil {
+				return nil, err
+			}
+			if instance == nil || instance.IsOn == 0 {
+				// 如果媒介实例已经删除或者禁用，则删除此消息
+				err = models.SharedMessageTaskDAO.DisableMessageTask(tx, int64(task.Id))
+				if err != nil {
+					return nil, err
+				}
+				continue
+			}
+
+			pbRecipient = &pb.MessageRecipient{
+				Id:   int64(recipient.Id),
+				User: recipient.User,
+				MessageMediaInstance: &pb.MessageMediaInstance{
+					Id:   int64(instance.Id),
+					Name: instance.Name,
+					MessageMedia: &pb.MessageMedia{
+						Type: instance.MediaType,
+					},
+					ParamsJSON: []byte(instance.Params),
+				},
+			}
+		} else { // 没有指定既定的接收人
+			// 媒介
+			instance, err := models.SharedMessageMediaInstanceDAO.FindEnabledMessageMediaInstance(tx, int64(task.InstanceId), cacheMap)
+			if err != nil {
+				return nil, err
+			}
+			if instance == nil || instance.IsOn == 0 {
+				// 如果媒介实例已经删除或者禁用，则删除此消息
+				err = models.SharedMessageTaskDAO.DisableMessageTask(tx, int64(task.Id))
+				if err != nil {
+					return nil, err
+				}
+				continue
+			}
+			pbRecipient = &pb.MessageRecipient{
+				Id: 0,
+				MessageMediaInstance: &pb.MessageMediaInstance{
+					Id:   int64(instance.Id),
+					Name: instance.Name,
+					MessageMedia: &pb.MessageMedia{
+						Type: instance.MediaType,
+					},
+					ParamsJSON: []byte(instance.Params),
+				},
+			}
+		}
+
+		var result = &pb.MessageTaskResult{}
+		if len(task.Result) > 0 {
+			err = json.Unmarshal([]byte(task.Result), result)
+			if err != nil {
+				return nil, err
+			}
+		}
+
+		pbTasks = append(pbTasks, &pb.MessageTask{
+			Id:               int64(task.Id),
+			MessageRecipient: pbRecipient,
+			User:             task.User,
+			Subject:          task.Subject,
+			Body:             task.Body,
+			CreatedAt:        int64(task.CreatedAt),
+			Status:           types.Int32(task.Status),
+			SentAt:           int64(task.SentAt),
+			Result:           result,
+		})
+	}
+
+	return &pb.ListMessageTasksWithStatusResponse{MessageTasks: pbTasks}, nil
+}

internal/rpc/services/service_message_task_log.go

@@ -4,14 +4,15 @@ import (
 	"context"
 	"github.com/TeaOSLab/EdgeAPI/internal/db/models"
 	"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
+	"github.com/iwind/TeaGo/maps"
 )
 
-// 消息发送日志相关服务
+// MessageTaskLogService 消息发送日志相关服务
 type MessageTaskLogService struct {
 	BaseService
 }
 
-// 计算日志数量
+// CountMessageTaskLogs 计算日志数量
 func (this *MessageTaskLogService) CountMessageTaskLogs(ctx context.Context, req *pb.CountMessageTaskLogsRequest) (*pb.RPCCountResponse, error) {
 	_, err := this.ValidateAdmin(ctx, 0)
 	if err != nil {
@@ -32,6 +33,7 @@ func (this *MessageTaskLogService) ListMessageTaskLogs(ctx context.Context, req
 		return nil, err
 	}
 	var tx = this.NullTx()
+	var cacheMap = maps.Map{}
 	logs, err := models.SharedMessageTaskLogDAO.ListLogs(tx, req.Offset, req.Size)
 	if err != nil {
 		return nil, err
@@ -49,7 +51,7 @@ func (this *MessageTaskLogService) ListMessageTaskLogs(ctx context.Context, req
 
 		var pbRecipient *pb.MessageRecipient
 		if task.RecipientId > 0 {
-			recipient, err := models.SharedMessageRecipientDAO.FindEnabledMessageRecipient(tx, int64(task.RecipientId))
+			recipient, err := models.SharedMessageRecipientDAO.FindEnabledMessageRecipient(tx, int64(task.RecipientId), cacheMap)
 			if err != nil {
 				return nil, err
 			}
@@ -62,7 +64,7 @@ func (this *MessageTaskLogService) ListMessageTaskLogs(ctx context.Context, req
 			}
 		}
 
-		instance, err := models.SharedMessageMediaInstanceDAO.FindEnabledMessageMediaInstance(tx, int64(task.InstanceId))
+		instance, err := models.SharedMessageMediaInstanceDAO.FindEnabledMessageMediaInstance(tx, int64(task.InstanceId), cacheMap)
 		if err != nil {
 			return nil, err
 		}

internal/rpc/services/service_node_cluster.go

@@ -951,9 +951,7 @@ func (this *NodeClusterService) FindEnabledNodeClusterConfigInfo(ctx context.Con
 	result.HasThresholds = countThresholds > 0
 
 	// message receivers
-	countReceivers, err := models.SharedMessageReceiverDAO.CountAllEnabledReceivers(tx, models.MessageTaskTarget{
+	countReceivers, err := models.SharedMessageReceiverDAO.CountAllEnabledReceivers(tx, nodeconfigs.NodeRoleNode, req.NodeClusterId, 0, 0, "")
-		ClusterId: req.NodeClusterId,
-	}, "")
 	if err != nil {
 		return nil, err
 	}