实现基础的通知媒介管理

internal/rpc/services/service_api_node.go

@@ -65,7 +65,7 @@ func (this *APINodeService) DeleteAPINode(ctx context.Context, req *pb.DeleteAPI
 
 // 列出所有可用API节点
 func (this *APINodeService) FindAllEnabledAPINodes(ctx context.Context, req *pb.FindAllEnabledAPINodesRequest) (*pb.FindAllEnabledAPINodesResponse, error) {
-	_, _, err := rpcutils.ValidateRequest(ctx, rpcutils.UserTypeAdmin, rpcutils.UserTypeUser, rpcutils.UserTypeNode)
+	_, _, err := rpcutils.ValidateRequest(ctx, rpcutils.UserTypeAdmin, rpcutils.UserTypeUser, rpcutils.UserTypeNode, rpcutils.UserTypeMonitor, rpcutils.UserTypeDNS)
 	if err != nil {
 		return nil, err
 	}

internal/rpc/services/service_base.go

@@ -82,6 +82,12 @@ func (this *BaseService) ValidateUser(ctx context.Context) (userId int64, err er
 	return
 }
 
+// 校验监控节点
+func (this *BaseService) ValidateMonitor(ctx context.Context) (nodeId int64, err error) {
+	_, nodeId, err = rpcutils.ValidateRequest(ctx, rpcutils.UserTypeMonitor)
+	return
+}
+
 // 获取节点ID
 func (this *BaseService) ValidateNodeId(ctx context.Context, roles ...rpcutils.UserType) (role rpcutils.UserType, nodeIntId int64, err error) {
 	if ctx == nil {
@@ -170,6 +176,8 @@ func (this *BaseService) ValidateNodeId(ctx context.Context, roles ...rpcutils.U
 		nodeIntId, err = models.SharedUserNodeDAO.FindEnabledUserNodeIdWithUniqueId(nil, nodeId)
 	case rpcutils.UserTypeAdmin:
 		nodeIntId = 0
+	case rpcutils.UserTypeMonitor:
+		nodeIntId, err = models.SharedMonitorNodeDAO.FindEnabledMonitorNodeIdWithUniqueId(nil, nodeId)
 	default:
 		err = errors.New("unsupported user role '" + apiToken.Role + "'")
 	}

internal/rpc/services/service_message_media.go

@@ -0,0 +1,64 @@
+package services
+
+import (
+	"context"
+	"github.com/TeaOSLab/EdgeAPI/internal/db/models"
+	"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
+	"github.com/iwind/TeaGo/maps"
+)
+
+// 消息媒介服务
+type MessageMediaService struct {
+	BaseService
+}
+
+// 获取所有支持的媒介
+func (this *MessageMediaService) FindAllMessageMedias(ctx context.Context, req *pb.FindAllMessageMediasRequest) (*pb.FindAllMessageMediasResponse, error) {
+	_, err := this.ValidateAdmin(ctx, 0)
+	if err != nil {
+		return nil, err
+	}
+	var tx = this.NullTx()
+	medias, err := models.SharedMessageMediaDAO.FindAllEnabledMessageMedias(tx)
+	if err != nil {
+		return nil, err
+	}
+	pbMedias := []*pb.MessageMedia{}
+	for _, media := range medias {
+		pbMedias = append(pbMedias, &pb.MessageMedia{
+			Id:              int64(media.Id),
+			Type:            media.Type,
+			Name:            media.Name,
+			Description:     media.Description,
+			UserDescription: media.UserDescription,
+			IsOn:            media.IsOn == 1,
+		})
+	}
+	return &pb.FindAllMessageMediasResponse{MessageMedias: pbMedias}, nil
+}
+
+// 设置所有支持的媒介
+func (this *MessageMediaService) UpdateMessageMedias(ctx context.Context, req *pb.UpdateMessageMediasRequest) (*pb.RPCSuccess, error) {
+	_, err := this.ValidateMonitor(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	mediaMaps := []maps.Map{}
+	for _, media := range req.MessageMedias {
+		mediaMaps = append(mediaMaps, maps.Map{
+			"name":            media.Name,
+			"type":            media.Type,
+			"description":     media.Description,
+			"userDescription": media.UserDescription,
+			"isOn":            media.IsOn,
+		})
+	}
+
+	var tx = this.NullTx()
+	err = models.SharedMessageMediaDAO.UpdateMessageMedias(tx, mediaMaps)
+	if err != nil {
+		return nil, err
+	}
+	return this.Success()
+}

internal/rpc/services/service_message_media_instance.go

@@ -0,0 +1,181 @@
+package services
+
+import (
+	"context"
+	"encoding/json"
+	"github.com/TeaOSLab/EdgeAPI/internal/db/models"
+	"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
+	"github.com/iwind/TeaGo/maps"
+)
+
+// 消息媒介实例服务
+type MessageMediaInstanceService struct {
+	BaseService
+}
+
+// 创建消息媒介实例
+func (this *MessageMediaInstanceService) CreateMessageMediaInstance(ctx context.Context, req *pb.CreateMessageMediaInstanceRequest) (*pb.CreateMessageMediaInstanceResponse, error) {
+	_, err := this.ValidateAdmin(ctx, 0)
+	if err != nil {
+		return nil, err
+	}
+
+	var tx = this.NullTx()
+
+	params := maps.Map{}
+	if len(req.ParamsJSON) > 0 {
+		err = json.Unmarshal(req.ParamsJSON, &params)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	instanceId, err := models.SharedMessageMediaInstanceDAO.CreateMediaInstance(tx, req.Name, req.MediaType, params, req.Description)
+	if err != nil {
+		return nil, err
+	}
+
+	return &pb.CreateMessageMediaInstanceResponse{MessageMediaInstanceId: instanceId}, nil
+}
+
+// 修改消息实例
+func (this *MessageMediaInstanceService) UpdateMessageMediaInstance(ctx context.Context, req *pb.UpdateMessageMediaInstanceRequest) (*pb.RPCSuccess, error) {
+	_, err := this.ValidateAdmin(ctx, 0)
+	if err != nil {
+		return nil, err
+	}
+
+	params := maps.Map{}
+	if len(req.ParamsJSON) > 0 {
+		err = json.Unmarshal(req.ParamsJSON, &params)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	var tx = this.NullTx()
+	err = models.SharedMessageMediaInstanceDAO.UpdateMediaInstance(tx, req.MessageMediaInstanceId, req.Name, req.MediaType, params, req.Description, req.IsOn)
+	if err != nil {
+		return nil, err
+	}
+
+	return this.Success()
+}
+
+// 删除媒介实例
+func (this *MessageMediaInstanceService) DeleteMessageMediaInstance(ctx context.Context, req *pb.DeleteMessageMediaInstanceRequest) (*pb.RPCSuccess, error) {
+	_, err := this.ValidateAdmin(ctx, 0)
+	if err != nil {
+		return nil, err
+	}
+
+	var tx = this.NullTx()
+	err = models.SharedMessageMediaInstanceDAO.DisableMessageMediaInstance(tx, req.MessageMediaInstanceId)
+	if err != nil {
+		return nil, err
+	}
+	return this.Success()
+}
+
+// 计算媒介实例数量
+func (this *MessageMediaInstanceService) CountAllEnabledMessageMediaInstances(ctx context.Context, req *pb.CountAllEnabledMessageMediaInstancesRequest) (*pb.RPCCountResponse, error) {
+	_, err := this.ValidateAdmin(ctx, 0)
+	if err != nil {
+		return nil, err
+	}
+
+	var tx = this.NullTx()
+	count, err := models.SharedMessageMediaInstanceDAO.CountAllEnabledMediaInstances(tx, req.MediaType, req.Keyword)
+	if err != nil {
+		return nil, err
+	}
+
+	return this.SuccessCount(count)
+}
+
+// 列出单页媒介实例
+func (this *MessageMediaInstanceService) ListEnabledMessageMediaInstances(ctx context.Context, req *pb.ListEnabledMessageMediaInstancesRequest) (*pb.ListEnabledMessageMediaInstancesResponse, error) {
+	_, err := this.ValidateAdmin(ctx, 0)
+	if err != nil {
+		return nil, err
+	}
+
+	var tx = this.NullTx()
+	instances, err := models.SharedMessageMediaInstanceDAO.ListAllEnabledMediaInstances(tx, req.MediaType, req.Keyword, req.Offset, req.Size)
+	if err != nil {
+		return nil, err
+	}
+	pbInstances := []*pb.MessageMediaInstance{}
+	for _, instance := range instances {
+		// 媒介
+		media, err := models.SharedMessageMediaDAO.FindEnabledMediaWithType(tx, instance.MediaType)
+		if err != nil {
+			return nil, err
+		}
+		if media == nil {
+			continue
+		}
+		pbMedia := &pb.MessageMedia{
+			Id:              int64(media.Id),
+			Type:            media.Type,
+			Name:            media.Name,
+			Description:     media.Description,
+			UserDescription: media.UserDescription,
+			IsOn:            media.IsOn == 1,
+		}
+
+		pbInstances = append(pbInstances, &pb.MessageMediaInstance{
+			Id:           int64(instance.Id),
+			Name:         instance.Name,
+			IsOn:         instance.IsOn == 1,
+			MessageMedia: pbMedia,
+			ParamsJSON:   []byte(instance.Params),
+			Description:  instance.Description,
+		})
+	}
+
+	return &pb.ListEnabledMessageMediaInstancesResponse{MessageMediaInstances: pbInstances}, nil
+}
+
+// 查找单个媒介实例信息
+func (this *MessageMediaInstanceService) FindEnabledMessageMediaInstance(ctx context.Context, req *pb.FindEnabledMessageMediaInstanceRequest) (*pb.FindEnabledMessageMediaInstanceResponse, error) {
+	_, err := this.ValidateAdmin(ctx, 0)
+	if err != nil {
+		return nil, err
+	}
+
+	var tx = this.NullTx()
+	instance, err := models.SharedMessageMediaInstanceDAO.FindEnabledMessageMediaInstance(tx, req.MessageMediaInstanceId)
+	if err != nil {
+		return nil, err
+	}
+	if instance == nil {
+		return &pb.FindEnabledMessageMediaInstanceResponse{MessageMediaInstance: nil}, nil
+	}
+
+	// 媒介
+	media, err := models.SharedMessageMediaDAO.FindEnabledMediaWithType(tx, instance.MediaType)
+	if err != nil {
+		return nil, err
+	}
+	if media == nil {
+		return &pb.FindEnabledMessageMediaInstanceResponse{MessageMediaInstance: nil}, nil
+	}
+	pbMedia := &pb.MessageMedia{
+		Id:              int64(media.Id),
+		Type:            media.Type,
+		Name:            media.Name,
+		Description:     media.Description,
+		UserDescription: media.UserDescription,
+		IsOn:            media.IsOn == 1,
+	}
+
+	return &pb.FindEnabledMessageMediaInstanceResponse{MessageMediaInstance: &pb.MessageMediaInstance{
+		Id:           int64(instance.Id),
+		Name:         instance.Name,
+		IsOn:         instance.IsOn == 1,
+		MessageMedia: pbMedia,
+		ParamsJSON:   []byte(instance.Params),
+		Description:  instance.Description,
+	}}, nil
+}

internal/rpc/services/service_message_recipient.go

@@ -0,0 +1,228 @@
+package services
+
+import (
+	"context"
+	"github.com/TeaOSLab/EdgeAPI/internal/db/models"
+	"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
+)
+
+// 消息接收人服务
+type MessageRecipientService struct {
+	BaseService
+}
+
+// 创建接收人
+func (this *MessageRecipientService) CreateMessageRecipient(ctx context.Context, req *pb.CreateMessageRecipientRequest) (*pb.CreateMessageRecipientResponse, error) {
+	_, err := this.ValidateAdmin(ctx, 0)
+	if err != nil {
+		return nil, err
+	}
+
+	var tx = this.NullTx()
+	recipientId, err := models.SharedMessageRecipientDAO.CreateRecipient(tx, req.AdminId, req.InstanceId, req.User, req.GroupIds, req.Description)
+	if err != nil {
+		return nil, err
+	}
+
+	return &pb.CreateMessageRecipientResponse{MessageRecipientId: recipientId}, nil
+}
+
+// 修改接收人
+func (this *MessageRecipientService) UpdateMessageRecipient(ctx context.Context, req *pb.UpdateMessageRecipientRequest) (*pb.RPCSuccess, error) {
+	_, err := this.ValidateAdmin(ctx, 0)
+	if err != nil {
+		return nil, err
+	}
+
+	var tx = this.NullTx()
+	err = models.SharedMessageRecipientDAO.UpdateRecipient(tx, req.MessageRecipientId, req.AdminId, req.InstanceId, req.User, req.GroupIds, req.Description, req.IsOn)
+	if err != nil {
+		return nil, err
+	}
+
+	return this.Success()
+}
+
+// 删除接收人
+func (this *MessageRecipientService) DeleteMessageRecipient(ctx context.Context, req *pb.DeleteMessageRecipientRequest) (*pb.RPCSuccess, error) {
+	_, err := this.ValidateAdmin(ctx, 0)
+	if err != nil {
+		return nil, err
+	}
+
+	var tx = this.NullTx()
+	err = models.SharedMessageRecipientDAO.DisableMessageRecipient(tx, req.MessageRecipientId)
+	if err != nil {
+		return nil, err
+	}
+	return this.Success()
+}
+
+// 计算接收人数量
+func (this *MessageRecipientService) CountAllEnabledMessageRecipients(ctx context.Context, req *pb.CountAllEnabledMessageRecipientsRequest) (*pb.RPCCountResponse, error) {
+	_, err := this.ValidateAdmin(ctx, 0)
+	if err != nil {
+		return nil, err
+	}
+
+	var tx = this.NullTx()
+	count, err := models.SharedMessageRecipientDAO.CountAllEnabledRecipients(tx, req.AdminId, req.GroupId, req.MediaType, req.Keyword)
+	if err != nil {
+		return nil, err
+	}
+
+	return this.SuccessCount(count)
+}
+
+// 列出单页接收人
+func (this *MessageRecipientService) ListEnabledMessageRecipients(ctx context.Context, req *pb.ListEnabledMessageRecipientsRequest) (*pb.ListEnabledMessageRecipientsResponse, error) {
+	_, err := this.ValidateAdmin(ctx, 0)
+	if err != nil {
+		return nil, err
+	}
+
+	var tx = this.NullTx()
+	recipients, err := models.SharedMessageRecipientDAO.ListAllEnabledRecipients(tx, req.AdminId, req.GroupId, req.MediaType, req.Keyword, req.Offset, req.Size)
+	if err != nil {
+		return nil, err
+	}
+	pbRecipients := []*pb.MessageRecipient{}
+	for _, recipient := range recipients {
+		// admin
+		admin, err := models.SharedAdminDAO.FindEnabledAdmin(tx, int64(recipient.AdminId))
+		if err != nil {
+			return nil, err
+		}
+		if admin == nil {
+			continue
+		}
+		pbAdmin := &pb.Admin{
+			Id:       int64(admin.Id),
+			Fullname: admin.Fullname,
+			Username: admin.Username,
+			IsOn:     admin.IsOn == 1,
+		}
+
+		// 媒介实例
+		instance, err := models.SharedMessageMediaInstanceDAO.FindEnabledMessageMediaInstance(tx, int64(recipient.InstanceId))
+		if err != nil {
+			return nil, err
+		}
+		if instance == nil {
+			continue
+		}
+		pbInstance := &pb.MessageMediaInstance{
+			Id:          int64(instance.Id),
+			IsOn:        instance.IsOn == 1,
+			Name:        instance.Name,
+			Description: instance.Description,
+		}
+
+		// 分组
+		pbGroups := []*pb.MessageRecipientGroup{}
+		groupIds := recipient.DecodeGroupIds()
+		if len(groupIds) > 0 {
+			for _, groupId := range groupIds {
+				group, err := models.SharedMessageRecipientGroupDAO.FindEnabledMessageRecipientGroup(tx, groupId)
+				if err != nil {
+					return nil, err
+				}
+				if group != nil {
+					pbGroups = append(pbGroups, &pb.MessageRecipientGroup{
+						Id:   int64(group.Id),
+						Name: group.Name,
+						IsOn: group.IsOn == 1,
+					})
+				}
+			}
+		}
+
+		pbRecipients = append(pbRecipients, &pb.MessageRecipient{
+			Id:                     int64(recipient.Id),
+			Admin:                  pbAdmin,
+			User:                   recipient.User,
+			MessageMediaInstance:   pbInstance,
+			IsOn:                   recipient.IsOn == 1,
+			MessageRecipientGroups: pbGroups,
+			Description:            recipient.Description,
+		})
+	}
+
+	return &pb.ListEnabledMessageRecipientsResponse{MessageRecipients: pbRecipients}, nil
+}
+
+// 查找单个接收人信息
+func (this *MessageRecipientService) FindEnabledMessageRecipient(ctx context.Context, req *pb.FindEnabledMessageRecipientRequest) (*pb.FindEnabledMessageRecipientResponse, error) {
+	_, err := this.ValidateAdmin(ctx, 0)
+	if err != nil {
+		return nil, err
+	}
+
+	var tx = this.NullTx()
+	recipient, err := models.SharedMessageRecipientDAO.FindEnabledMessageRecipient(tx, req.MessageRecipientId)
+	if err != nil {
+		return nil, err
+	}
+	if recipient == nil {
+		return &pb.FindEnabledMessageRecipientResponse{MessageRecipient: nil}, nil
+	}
+
+	// admin
+	admin, err := models.SharedAdminDAO.FindEnabledAdmin(tx, int64(recipient.AdminId))
+	if err != nil {
+		return nil, err
+	}
+	if admin == nil {
+		return &pb.FindEnabledMessageRecipientResponse{MessageRecipient: nil}, nil
+	}
+	pbAdmin := &pb.Admin{
+		Id:       int64(admin.Id),
+		Fullname: admin.Fullname,
+		Username: admin.Username,
+		IsOn:     admin.IsOn == 1,
+	}
+
+	// 媒介实例
+	instance, err := models.SharedMessageMediaInstanceDAO.FindEnabledMessageMediaInstance(tx, int64(recipient.InstanceId))
+	if err != nil {
+		return nil, err
+	}
+	if instance == nil {
+		return &pb.FindEnabledMessageRecipientResponse{MessageRecipient: nil}, nil
+	}
+	pbInstance := &pb.MessageMediaInstance{
+		Id:          int64(instance.Id),
+		IsOn:        instance.IsOn == 1,
+		Name:        instance.Name,
+		Description: instance.Description,
+	}
+
+	// 分组
+	pbGroups := []*pb.MessageRecipientGroup{}
+	groupIds := recipient.DecodeGroupIds()
+	if len(groupIds) > 0 {
+		for _, groupId := range groupIds {
+			group, err := models.SharedMessageRecipientGroupDAO.FindEnabledMessageRecipientGroup(tx, groupId)
+			if err != nil {
+				return nil, err
+			}
+			if group != nil {
+				pbGroups = append(pbGroups, &pb.MessageRecipientGroup{
+					Id:   int64(group.Id),
+					Name: group.Name,
+					IsOn: group.IsOn == 1,
+				})
+			}
+		}
+	}
+
+	return &pb.FindEnabledMessageRecipientResponse{MessageRecipient: &pb.MessageRecipient{
+		Id:                     int64(recipient.Id),
+		User:                   recipient.User,
+		Admin:                  pbAdmin,
+		MessageMediaInstance:   pbInstance,
+		IsOn:                   recipient.IsOn == 1,
+		MessageRecipientGroups: pbGroups,
+		Description:            recipient.Description,
+	}}, nil
+}

internal/rpc/services/service_message_recipient_group.go

@@ -0,0 +1,106 @@
+package services
+
+import (
+	"context"
+	"github.com/TeaOSLab/EdgeAPI/internal/db/models"
+	"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
+)
+
+// 消息接收人分组
+type MessageRecipientGroupService struct {
+	BaseService
+}
+
+// 创建分组
+func (this *MessageRecipientGroupService) CreateMessageRecipientGroup(ctx context.Context, req *pb.CreateMessageRecipientGroupRequest) (*pb.CreateMessageRecipientGroupResponse, error) {
+	_, err := this.ValidateAdmin(ctx, 0)
+	if err != nil {
+		return nil, err
+	}
+
+	var tx = this.NullTx()
+	groupId, err := models.SharedMessageRecipientGroupDAO.CreateGroup(tx, req.Name)
+	if err != nil {
+		return nil, err
+	}
+
+	return &pb.CreateMessageRecipientGroupResponse{MessageRecipientGroupId: groupId}, nil
+}
+
+// 修改分组
+func (this *MessageRecipientGroupService) UpdateMessageRecipientGroup(ctx context.Context, req *pb.UpdateMessageRecipientGroupRequest) (*pb.RPCSuccess, error) {
+	_, err := this.ValidateAdmin(ctx, 0)
+	if err != nil {
+		return nil, err
+	}
+
+	var tx = this.NullTx()
+	err = models.SharedMessageRecipientGroupDAO.UpdateGroup(tx, req.MessageRecipientGroupId, req.Name, req.IsOn)
+	if err != nil {
+		return nil, err
+	}
+
+	return this.Success()
+}
+
+// 查找所有可用的分组
+func (this *MessageRecipientGroupService) FindAllEnabledMessageRecipientGroups(ctx context.Context, req *pb.FindAllEnabledMessageRecipientGroupsRequest) (*pb.FindAllEnabledMessageRecipientGroupsResponse, error) {
+	_, err := this.ValidateAdmin(ctx, 0)
+	if err != nil {
+		return nil, err
+	}
+
+	var tx = this.NullTx()
+	groups, err := models.SharedMessageRecipientGroupDAO.FindAllEnabledGroups(tx)
+	if err != nil {
+		return nil, err
+	}
+	pbGroups := []*pb.MessageRecipientGroup{}
+	for _, group := range groups {
+		pbGroups = append(pbGroups, &pb.MessageRecipientGroup{
+			Id:   int64(group.Id),
+			Name: group.Name,
+			IsOn: group.IsOn == 1,
+		})
+	}
+
+	return &pb.FindAllEnabledMessageRecipientGroupsResponse{MessageRecipientGroups: pbGroups}, nil
+}
+
+// 删除分组
+func (this *MessageRecipientGroupService) DeleteMessageRecipientGroup(ctx context.Context, req *pb.DeleteMessageRecipientGroupRequest) (*pb.RPCSuccess, error) {
+	_, err := this.ValidateAdmin(ctx, 0)
+	if err != nil {
+		return nil, err
+	}
+
+	var tx = this.NullTx()
+	err = models.SharedMessageRecipientGroupDAO.DisableMessageRecipientGroup(tx, req.MessageRecipientGroupId)
+	if err != nil {
+		return nil, err
+	}
+	return this.Success()
+}
+
+// 查找单个分组信息
+func (this *MessageRecipientGroupService) FindEnabledMessageRecipientGroup(ctx context.Context, req *pb.FindEnabledMessageRecipientGroupRequest) (*pb.FindEnabledMessageRecipientGroupResponse, error) {
+	_, err := this.ValidateAdmin(ctx, 0)
+	if err != nil {
+		return nil, err
+	}
+
+	var tx = this.NullTx()
+	group, err := models.SharedMessageRecipientGroupDAO.FindEnabledMessageRecipientGroup(tx, req.MessageRecipientGroupId)
+	if err != nil {
+		return nil, err
+	}
+	if group == nil {
+		return &pb.FindEnabledMessageRecipientGroupResponse{MessageRecipientGroup: nil}, nil
+	}
+	pbGroup := &pb.MessageRecipientGroup{
+		Id:   int64(group.Id),
+		IsOn: group.IsOn == 1,
+		Name: group.Name,
+	}
+	return &pb.FindEnabledMessageRecipientGroupResponse{MessageRecipientGroup: pbGroup}, nil
+}

internal/rpc/services/service_message_task.go

@@ -0,0 +1,284 @@
+package services
+
+import (
+	"context"
+	"encoding/json"
+	"github.com/TeaOSLab/EdgeAPI/internal/db/models"
+	"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
+	"github.com/iwind/TeaGo/maps"
+	"github.com/iwind/TeaGo/types"
+)
+
+// 消息发送任务服务
+type MessageTaskService struct {
+	BaseService
+}
+
+// 创建任务
+func (this *MessageTaskService) CreateMessageTask(ctx context.Context, req *pb.CreateMessageTaskRequest) (*pb.CreateMessageTaskResponse, error) {
+	_, err := this.ValidateAdmin(ctx, 0)
+	if err != nil {
+		return nil, err
+	}
+
+	var tx = this.NullTx()
+	taskId, err := models.SharedMessageTaskDAO.CreateMessageTask(tx, req.RecipientId, req.InstanceId, req.User, req.Subject, req.Body, req.IsPrimary)
+	if err != nil {
+		return nil, err
+	}
+	return &pb.CreateMessageTaskResponse{MessageTaskId: taskId}, nil
+}
+
+// 查找要发送的任务
+func (this *MessageTaskService) FindSendingMessageTasks(ctx context.Context, req *pb.FindSendingMessageTasksRequest) (*pb.FindSendingMessageTasksResponse, error) {
+	_, err := this.ValidateMonitor(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	var tx = this.NullTx()
+	tasks, err := models.SharedMessageTaskDAO.FindSendingMessageTasks(tx, req.Size)
+	if err != nil {
+		return nil, err
+	}
+	pbTasks := []*pb.MessageTask{}
+	for _, task := range tasks {
+		var pbRecipient *pb.MessageRecipient
+		if task.RecipientId > 0 {
+			// TODO 需要缓存以提升性能
+			recipient, err := models.SharedMessageRecipientDAO.FindEnabledMessageRecipient(tx, int64(task.RecipientId))
+			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
+			}
+
+			// 媒介
+			// TODO 需要缓存以提升性能
+			instance, err := models.SharedMessageMediaInstanceDAO.FindEnabledMessageMediaInstance(tx, int64(recipient.InstanceId))
+			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),
+					MessageMedia: &pb.MessageMedia{
+						Type: instance.MediaType,
+					},
+					ParamsJSON: []byte(instance.Params),
+				},
+			}
+		} else { // 没有指定既定的接收人
+			// 媒介
+			// TODO 需要缓存以提升性能
+			instance, err := models.SharedMessageMediaInstanceDAO.FindEnabledMessageMediaInstance(tx, int64(task.InstanceId))
+			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),
+					MessageMedia: &pb.MessageMedia{
+						Type: instance.MediaType,
+					},
+					ParamsJSON: []byte(instance.Params),
+				},
+			}
+		}
+
+		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),
+		})
+	}
+	return &pb.FindSendingMessageTasksResponse{MessageTasks: pbTasks}, nil
+}
+
+// 修改任务状态
+func (this *MessageTaskService) UpdateMessageTaskStatus(ctx context.Context, req *pb.UpdateMessageTaskStatusRequest) (*pb.RPCSuccess, error) {
+	_, err := this.ValidateMonitor(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	var tx = this.NullTx()
+	resultJSON := []byte{}
+	if req.Result != nil {
+		resultJSON, err = json.Marshal(maps.Map{
+			"isOk":     req.Result.IsOk,
+			"error":    req.Result.Error,
+			"response": req.Result.Response,
+		})
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	err = models.SharedMessageTaskDAO.UpdateMessageTaskStatus(tx, req.MessageTaskId, int(req.Status), resultJSON)
+	if err != nil {
+		return nil, err
+	}
+
+	// 创建发送记录
+	if (int(req.Status) == models.MessageTaskStatusSuccess || int(req.Status) == models.MessageTaskStatusFailed) && req.Result != nil {
+		err = models.SharedMessageTaskLogDAO.CreateLog(tx, req.MessageTaskId, req.Result.IsOk, req.Result.Error, req.Result.Response)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	return this.Success()
+}
+
+// 删除消息任务
+func (this *MessageTaskService) DeleteMessageTask(ctx context.Context, req *pb.DeleteMessageTaskRequest) (*pb.RPCSuccess, error) {
+	_, err := this.ValidateAdmin(ctx, 0)
+	if err != nil {
+		return nil, err
+	}
+
+	var tx = this.NullTx()
+	err = models.SharedMessageTaskDAO.DisableMessageTask(tx, req.MessageTaskId)
+	if err != nil {
+		return nil, err
+	}
+
+	return this.Success()
+}
+
+// 读取消息任务状态
+func (this *MessageTaskService) FindEnabledMessageTask(ctx context.Context, req *pb.FindEnabledMessageTaskRequest) (*pb.FindEnabledMessageTaskResponse, error) {
+	_, err := this.ValidateAdmin(ctx, 0)
+	if err != nil {
+		return nil, err
+	}
+
+	var tx = this.NullTx()
+	task, err := models.SharedMessageTaskDAO.FindEnabledMessageTask(tx, req.MessageTaskId)
+	if err != nil {
+		return nil, err
+	}
+	if task == nil {
+		return &pb.FindEnabledMessageTaskResponse{MessageTask: nil}, nil
+	}
+
+	// TODO 需要缓存以提升性能
+	var pbRecipient *pb.MessageRecipient
+	if task.RecipientId > 0 {
+		recipient, err := models.SharedMessageRecipientDAO.FindEnabledMessageRecipient(tx, int64(task.RecipientId))
+		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
+			}
+			return &pb.FindEnabledMessageTaskResponse{MessageTask: nil}, nil
+		}
+
+		// 媒介
+		// TODO 需要缓存以提升性能
+		instance, err := models.SharedMessageMediaInstanceDAO.FindEnabledMessageMediaInstance(tx, int64(recipient.InstanceId))
+		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
+			}
+			return &pb.FindEnabledMessageTaskResponse{MessageTask: nil}, nil
+		}
+
+		pbRecipient = &pb.MessageRecipient{
+
+			MessageMediaInstance: &pb.MessageMediaInstance{
+				Id: int64(instance.Id),
+				MessageMedia: &pb.MessageMedia{
+					Type: instance.MediaType,
+				},
+				ParamsJSON: []byte(instance.Params),
+			},
+		}
+	} else { // 没有指定既定的接收人
+		// 媒介
+		// TODO 需要缓存以提升性能
+		instance, err := models.SharedMessageMediaInstanceDAO.FindEnabledMessageMediaInstance(tx, int64(task.InstanceId))
+		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
+			}
+			return &pb.FindEnabledMessageTaskResponse{MessageTask: nil}, nil
+		}
+		pbRecipient = &pb.MessageRecipient{
+			Id: 0,
+			MessageMediaInstance: &pb.MessageMediaInstance{
+				Id: int64(instance.Id),
+				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
+		}
+	}
+
+	return &pb.FindEnabledMessageTaskResponse{MessageTask: &pb.MessageTask{
+		Id:               int64(task.Id),
+		MessageRecipient: pbRecipient,
+		User:             task.User,
+		Subject:          task.Subject,
+		Body:             task.Body,
+		CreatedAt:        int64(task.CreatedAt),
+		Status:           int32(task.Status),
+		SentAt:           int64(task.SentAt),
+		Result:           result,
+	}}, nil
+}

internal/rpc/services/service_message_task_log.go

@@ -0,0 +1,96 @@
+package services
+
+import (
+	"context"
+	"github.com/TeaOSLab/EdgeAPI/internal/db/models"
+	"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
+)
+
+// 消息发送日志相关服务
+type MessageTaskLogService struct {
+	BaseService
+}
+
+// 计算日志数量
+func (this *MessageTaskLogService) CountMessageTaskLogs(ctx context.Context, req *pb.CountMessageTaskLogsRequest) (*pb.RPCCountResponse, error) {
+	_, err := this.ValidateAdmin(ctx, 0)
+	if err != nil {
+		return nil, err
+	}
+	var tx = this.NullTx()
+	count, err := models.SharedMessageTaskLogDAO.CountLogs(tx)
+	if err != nil {
+		return nil, err
+	}
+	return this.SuccessCount(count)
+}
+
+// 列出当页日志
+func (this *MessageTaskLogService) ListMessageTaskLogs(ctx context.Context, req *pb.ListMessageTaskLogsRequest) (*pb.ListMessageTaskLogsResponse, error) {
+	_, err := this.ValidateAdmin(ctx, 0)
+	if err != nil {
+		return nil, err
+	}
+	var tx = this.NullTx()
+	logs, err := models.SharedMessageTaskLogDAO.ListLogs(tx, req.Offset, req.Size)
+	if err != nil {
+		return nil, err
+	}
+
+	pbLogs := []*pb.MessageTaskLog{}
+	for _, log := range logs {
+		task, err := models.SharedMessageTaskDAO.FindEnabledMessageTask(tx, int64(log.TaskId))
+		if err != nil {
+			return nil, err
+		}
+		if task == nil {
+			continue
+		}
+
+		var pbRecipient *pb.MessageRecipient
+		if task.RecipientId > 0 {
+			recipient, err := models.SharedMessageRecipientDAO.FindEnabledMessageRecipient(tx, int64(task.RecipientId))
+			if err != nil {
+				return nil, err
+			}
+			if recipient != nil {
+				pbRecipient = &pb.MessageRecipient{
+					Id:   int64(recipient.Id),
+					User: recipient.User,
+				}
+			}
+		}
+
+		instance, err := models.SharedMessageMediaInstanceDAO.FindEnabledMessageMediaInstance(tx, int64(task.InstanceId))
+		if err != nil {
+			return nil, err
+		}
+		if instance == nil {
+			continue
+		}
+
+		pbLogs = append(pbLogs, &pb.MessageTaskLog{
+			Id:        int64(log.Id),
+			CreatedAt: int64(log.CreatedAt),
+			IsOk:      log.IsOk == 1,
+			Error:     log.Error,
+			Response:  log.Response,
+			MessageTask: &pb.MessageTask{
+				Id:               int64(task.Id),
+				MessageRecipient: pbRecipient,
+				MessageMediaInstance: &pb.MessageMediaInstance{
+					Id:   int64(instance.Id),
+					Name: instance.Name,
+				},
+				User:      task.User,
+				Subject:   task.Subject,
+				Body:      task.Body,
+				CreatedAt: int64(task.CreatedAt),
+				Status:    int32(task.Status),
+				SentAt:    int64(task.SentAt),
+				Result:    nil,
+			},
+		})
+	}
+	return &pb.ListMessageTaskLogsResponse{MessageTaskLogs: pbLogs}, nil
+}

internal/rpc/services/service_monitor_node.go

@@ -0,0 +1,233 @@
+package services
+
+import (
+	"context"
+	"github.com/TeaOSLab/EdgeAPI/internal/db/models"
+	"github.com/TeaOSLab/EdgeAPI/internal/errors"
+	rpcutils "github.com/TeaOSLab/EdgeAPI/internal/rpc/utils"
+	"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
+	"google.golang.org/grpc/metadata"
+)
+
+type MonitorNodeService struct {
+	BaseService
+}
+
+// 创建监控节点
+func (this *MonitorNodeService) CreateMonitorNode(ctx context.Context, req *pb.CreateMonitorNodeRequest) (*pb.CreateMonitorNodeResponse, error) {
+	_, _, err := rpcutils.ValidateRequest(ctx, rpcutils.UserTypeAdmin)
+	if err != nil {
+		return nil, err
+	}
+
+	tx := this.NullTx()
+
+	nodeId, err := models.SharedMonitorNodeDAO.CreateMonitorNode(tx, req.Name, req.Description, req.IsOn)
+	if err != nil {
+		return nil, err
+	}
+
+	return &pb.CreateMonitorNodeResponse{NodeId: nodeId}, nil
+}
+
+// 修改监控节点
+func (this *MonitorNodeService) UpdateMonitorNode(ctx context.Context, req *pb.UpdateMonitorNodeRequest) (*pb.RPCSuccess, error) {
+	_, _, err := rpcutils.ValidateRequest(ctx, rpcutils.UserTypeAdmin)
+	if err != nil {
+		return nil, err
+	}
+
+	tx := this.NullTx()
+
+	err = models.SharedMonitorNodeDAO.UpdateMonitorNode(tx, req.NodeId, req.Name, req.Description, req.IsOn)
+	if err != nil {
+		return nil, err
+	}
+
+	return this.Success()
+}
+
+// 删除监控节点
+func (this *MonitorNodeService) DeleteMonitorNode(ctx context.Context, req *pb.DeleteMonitorNodeRequest) (*pb.RPCSuccess, error) {
+	_, _, err := rpcutils.ValidateRequest(ctx, rpcutils.UserTypeAdmin)
+	if err != nil {
+		return nil, err
+	}
+
+	tx := this.NullTx()
+
+	err = models.SharedMonitorNodeDAO.DisableMonitorNode(tx, req.NodeId)
+	if err != nil {
+		return nil, err
+	}
+
+	return this.Success()
+}
+
+// 列出所有可用监控节点
+func (this *MonitorNodeService) FindAllEnabledMonitorNodes(ctx context.Context, req *pb.FindAllEnabledMonitorNodesRequest) (*pb.FindAllEnabledMonitorNodesResponse, error) {
+	_, _, err := rpcutils.ValidateRequest(ctx, rpcutils.UserTypeAdmin)
+	if err != nil {
+		return nil, err
+	}
+
+	tx := this.NullTx()
+
+	nodes, err := models.SharedMonitorNodeDAO.FindAllEnabledMonitorNodes(tx)
+	if err != nil {
+		return nil, err
+	}
+
+	result := []*pb.MonitorNode{}
+	for _, node := range nodes {
+		result = append(result, &pb.MonitorNode{
+			Id:          int64(node.Id),
+			IsOn:        node.IsOn == 1,
+			UniqueId:    node.UniqueId,
+			Secret:      node.Secret,
+			Name:        node.Name,
+			Description: node.Description,
+		})
+	}
+
+	return &pb.FindAllEnabledMonitorNodesResponse{Nodes: result}, nil
+}
+
+// 计算监控节点数量
+func (this *MonitorNodeService) CountAllEnabledMonitorNodes(ctx context.Context, req *pb.CountAllEnabledMonitorNodesRequest) (*pb.RPCCountResponse, error) {
+	_, _, err := rpcutils.ValidateRequest(ctx, rpcutils.UserTypeAdmin)
+	if err != nil {
+		return nil, err
+	}
+
+	tx := this.NullTx()
+
+	count, err := models.SharedMonitorNodeDAO.CountAllEnabledMonitorNodes(tx)
+	if err != nil {
+		return nil, err
+	}
+
+	return this.SuccessCount(count)
+}
+
+// 列出单页的监控节点
+func (this *MonitorNodeService) ListEnabledMonitorNodes(ctx context.Context, req *pb.ListEnabledMonitorNodesRequest) (*pb.ListEnabledMonitorNodesResponse, error) {
+	_, _, err := rpcutils.ValidateRequest(ctx, rpcutils.UserTypeAdmin)
+	if err != nil {
+		return nil, err
+	}
+
+	tx := this.NullTx()
+
+	nodes, err := models.SharedMonitorNodeDAO.ListEnabledMonitorNodes(tx, req.Offset, req.Size)
+	if err != nil {
+		return nil, err
+	}
+
+	result := []*pb.MonitorNode{}
+	for _, node := range nodes {
+		result = append(result, &pb.MonitorNode{
+			Id:          int64(node.Id),
+			IsOn:        node.IsOn == 1,
+			UniqueId:    node.UniqueId,
+			Secret:      node.Secret,
+			Name:        node.Name,
+			Description: node.Description,
+			StatusJSON:  []byte(node.Status),
+		})
+	}
+
+	return &pb.ListEnabledMonitorNodesResponse{Nodes: result}, nil
+}
+
+// 根据ID查找节点
+func (this *MonitorNodeService) FindEnabledMonitorNode(ctx context.Context, req *pb.FindEnabledMonitorNodeRequest) (*pb.FindEnabledMonitorNodeResponse, error) {
+	_, _, err := rpcutils.ValidateRequest(ctx, rpcutils.UserTypeAdmin)
+	if err != nil {
+		return nil, err
+	}
+
+	tx := this.NullTx()
+
+	node, err := models.SharedMonitorNodeDAO.FindEnabledMonitorNode(tx, req.NodeId)
+	if err != nil {
+		return nil, err
+	}
+
+	if node == nil {
+		return &pb.FindEnabledMonitorNodeResponse{Node: nil}, nil
+	}
+
+	result := &pb.MonitorNode{
+		Id:          int64(node.Id),
+		IsOn:        node.IsOn == 1,
+		UniqueId:    node.UniqueId,
+		Secret:      node.Secret,
+		Name:        node.Name,
+		Description: node.Description,
+	}
+	return &pb.FindEnabledMonitorNodeResponse{Node: result}, nil
+}
+
+// 获取当前监控节点的版本
+func (this *MonitorNodeService) FindCurrentMonitorNode(ctx context.Context, req *pb.FindCurrentMonitorNodeRequest) (*pb.FindCurrentMonitorNodeResponse, error) {
+	_, err := this.ValidateMonitor(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	tx := this.NullTx()
+
+	md, ok := metadata.FromIncomingContext(ctx)
+	if !ok {
+		return nil, errors.New("context: need 'nodeId'")
+	}
+	nodeIds := md.Get("nodeid")
+	if len(nodeIds) == 0 {
+		return nil, errors.New("invalid 'nodeId'")
+	}
+	nodeId := nodeIds[0]
+	node, err := models.SharedMonitorNodeDAO.FindEnabledMonitorNodeWithUniqueId(tx, nodeId)
+	if err != nil {
+		return nil, err
+	}
+
+	if node == nil {
+		return &pb.FindCurrentMonitorNodeResponse{Node: nil}, nil
+	}
+
+	result := &pb.MonitorNode{
+		Id:          int64(node.Id),
+		IsOn:        node.IsOn == 1,
+		UniqueId:    node.UniqueId,
+		Secret:      node.Secret,
+		Name:        node.Name,
+		Description: node.Description,
+	}
+	return &pb.FindCurrentMonitorNodeResponse{Node: result}, nil
+}
+
+// 更新节点状态
+func (this *MonitorNodeService) UpdateMonitorNodeStatus(ctx context.Context, req *pb.UpdateMonitorNodeStatusRequest) (*pb.RPCSuccess, error) {
+	// 校验节点
+	_, nodeId, err := this.ValidateNodeId(ctx, rpcutils.UserTypeMonitor)
+	if err != nil {
+		return nil, err
+	}
+
+	if req.NodeId > 0 {
+		nodeId = req.NodeId
+	}
+
+	if nodeId <= 0 {
+		return nil, errors.New("'nodeId' should be greater than 0")
+	}
+
+	tx := this.NullTx()
+
+	err = models.SharedMonitorNodeDAO.UpdateNodeStatus(tx, nodeId, req.StatusJSON)
+	if err != nil {
+		return nil, err
+	}
+	return this.Success()
+}

internal/rpc/services/service_sys_locker.go

@@ -15,7 +15,10 @@ type SysLockerService struct {
 func (this *SysLockerService) SysLockerLock(ctx context.Context, req *pb.SysLockerLockRequest) (*pb.SysLockerLockResponse, error) {
 	_, userId, err := this.ValidateAdminAndUser(ctx, 0, 0)
 	if err != nil {
-		return nil, err
+		_, err = this.ValidateMonitor(ctx)
+		if err != nil {
+			return nil, err
+		}
 	}
 
 	key := req.Key
@@ -42,7 +45,10 @@ func (this *SysLockerService) SysLockerLock(ctx context.Context, req *pb.SysLock
 func (this *SysLockerService) SysLockerUnlock(ctx context.Context, req *pb.SysLockerUnlockRequest) (*pb.RPCSuccess, error) {
 	_, userId, err := this.ValidateAdminAndUser(ctx, 0, 0)
 	if err != nil {
-		return nil, err
+		_, err = this.ValidateMonitor(ctx)
+		if err != nil {
+			return nil, err
+		}
 	}
 
 	key := req.Key

internal/rpc/utils/utils.go

@@ -154,6 +154,8 @@ func ValidateRequest(ctx context.Context, userTypes ...UserType) (userType UserT
 		}
 		nodeUserId = clusterId
 	case UserTypeUser:
+	case UserTypeMonitor:
+	case UserTypeDNS:
 	}
 
 	if nodeUserId > 0 {