增加消息管理

internal/db/models/db_node_initializer.go

@@ -156,7 +156,7 @@ func (this *DBNodeInitializer) loop() error {
 	accessLogLocker.Lock()
 	closingDbs := []*dbs.DB{}
 	for nodeId, db := range accessLogDBMapping {
-		if !this.containsInt64(nodeIds, nodeId) {
+		if !lists.ContainsInt64(nodeIds, nodeId) {
 			closingDbs = append(closingDbs, db)
 			delete(accessLogDBMapping, nodeId)
 			delete(accessLogDAOMapping, nodeId)
@@ -250,13 +250,3 @@ func (this *DBNodeInitializer) loop() error {
 
 	return nil
 }
-
-// 判断是否包含某数字
-func (this *DBNodeInitializer) containsInt64(values []int64, value int64) bool {
-	for _, v := range values {
-		if v == value {
-			return true
-		}
-	}
-	return false
-}

internal/db/models/http_web_dao.go

@@ -9,6 +9,7 @@ import (
 	_ "github.com/go-sql-driver/mysql"
 	"github.com/iwind/TeaGo/Tea"
 	"github.com/iwind/TeaGo/dbs"
+	"github.com/iwind/TeaGo/lists"
 	"github.com/iwind/TeaGo/types"
 	"strconv"
 )
@@ -548,7 +549,7 @@ func (this *HTTPWebDAO) FindAllWebIdsWithCachePolicyId(cachePolicyId int64) ([]i
 
 			// 如果非Location
 			if locationId == 0 {
-				if !this.containsInt64(result, webId) {
+				if !lists.ContainsInt64(result, webId) {
 					result = append(result, webId)
 				}
 				break
@@ -592,7 +593,7 @@ func (this *HTTPWebDAO) FindAllWebIdsWithHTTPFirewallPolicyId(firewallPolicyId i
 
 			// 如果非Location
 			if locationId == 0 {
-				if !this.containsInt64(result, webId) {
+				if !lists.ContainsInt64(result, webId) {
 					result = append(result, webId)
 				}
 				break
@@ -621,13 +622,3 @@ func (this *HTTPWebDAO) FindEnabledWebIdWithLocationId(locationId int64) (webId
 		Reuse(false). // 由于我们在JSON_CONTAINS()直接使用了变量，所以不能重用
 		FindInt64Col(0)
 }
-
-// 判断slice是否包含某个int64值
-func (this *HTTPWebDAO) containsInt64(values []int64, value int64) bool {
-	for _, v := range values {
-		if v == value {
-			return true
-		}
-	}
-	return false
-}

internal/db/models/message_dao.go

@@ -3,6 +3,7 @@ package models
 import (
 	"crypto/md5"
 	"fmt"
+	"github.com/TeaOSLab/EdgeAPI/internal/errors"
 	_ "github.com/go-sql-driver/mysql"
 	"github.com/iwind/TeaGo/Tea"
 	"github.com/iwind/TeaGo/dbs"
@@ -93,6 +94,58 @@ func (this *MessageDAO) DeleteMessagesBeforeDay(dayTime time.Time) error {
 	return err
 }
 
+// 计算未读消息数量
+func (this *MessageDAO) CountUnreadMessages() (int64, error) {
+	return this.Query().
+		Attr("isRead", false).
+		Count()
+}
+
+// 列出单页未读消息
+func (this *MessageDAO) ListUnreadMessages(offset int64, size int64) (result []*Message, err error) {
+	_, err = this.Query().
+		Attr("isRead", false).
+		Offset(offset).
+		Limit(size).
+		DescPk().
+		Slice(&result).
+		FindAll()
+	return
+}
+
+// 设置消息已读状态
+func (this *MessageDAO) UpdateMessageRead(messageId int64, b bool) error {
+	if messageId <= 0 {
+		return errors.New("invalid messageId")
+	}
+	op := NewMessageOperator()
+	op.Id = messageId
+	op.IsRead = b
+	_, err := this.Save(op)
+	return err
+}
+
+// 设置一组消息为已读状态
+func (this *MessageDAO) UpdateMessagesRead(messageIds []int64, b bool) error {
+	// 这里我们一个一个更改，因为In语句不容易Prepare，且效率不高
+	for _, messageId := range messageIds {
+		err := this.UpdateMessageRead(messageId, b)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// 设置所有消息为已读
+func (this *MessageDAO) UpdateAllMessagesRead() error {
+	_, err := this.Query().
+		Attr("isRead", false).
+		Set("isRead", true).
+		Update()
+	return err
+}
+
 // 创建消息
 func (this *MessageDAO) createMessage(clusterId int64, nodeId int64, messageType MessageType, level string, body string, paramsJSON []byte) (int64, error) {
 	h := md5.New()

internal/nodes/api_node.go

@@ -170,6 +170,7 @@ func (this *APINode) listenRPC(listener net.Listener, tlsConfig *tls.Config) err
 	pb.RegisterDBNodeServiceServer(rpcServer, &services.DBNodeService{})
 	pb.RegisterNodeLogServiceServer(rpcServer, &services.NodeLogService{})
 	pb.RegisterHTTPAccessLogServiceServer(rpcServer, &services.HTTPAccessLogService{})
+	pb.RegisterMessageServiceServer(rpcServer, &services.MessageService{})
 	err := rpcServer.Serve(listener)
 	if err != nil {
 		return errors.New("[API]start rpc failed: " + err.Error())

internal/rpc/services/service_message.go

@@ -0,0 +1,131 @@
+package services
+
+import (
+	"context"
+	"github.com/TeaOSLab/EdgeAPI/internal/db/models"
+	rpcutils "github.com/TeaOSLab/EdgeAPI/internal/rpc/utils"
+	"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
+)
+
+// 消息相关服务
+type MessageService struct {
+}
+
+// 计算未读消息数
+func (this *MessageService) CountUnreadMessages(ctx context.Context, req *pb.CountUnreadMessagesRequest) (*pb.CountUnreadMessagesResponse, error) {
+	// 校验请求
+	_, _, err := rpcutils.ValidateRequest(ctx, rpcutils.UserTypeAdmin)
+	if err != nil {
+		return nil, err
+	}
+
+	count, err := models.SharedMessageDAO.CountUnreadMessages()
+	if err != nil {
+		return nil, err
+	}
+	return &pb.CountUnreadMessagesResponse{Count: count}, nil
+}
+
+// 列出单页未读消息
+func (this *MessageService) ListUnreadMessages(ctx context.Context, req *pb.ListUnreadMessagesRequest) (*pb.ListUnreadMessagesResponse, error) {
+	// 校验请求
+	_, _, err := rpcutils.ValidateRequest(ctx, rpcutils.UserTypeAdmin)
+	if err != nil {
+		return nil, err
+	}
+
+	messages, err := models.SharedMessageDAO.ListUnreadMessages(req.Offset, req.Size)
+	if err != nil {
+		return nil, err
+	}
+	result := []*pb.Message{}
+	for _, message := range messages {
+		var pbCluster *pb.NodeCluster = nil
+		var pbNode *pb.Node = nil
+
+		if message.ClusterId > 0 {
+			cluster, err := models.SharedNodeClusterDAO.FindEnabledNodeCluster(int64(message.ClusterId))
+			if err != nil {
+				return nil, err
+			}
+			if cluster != nil {
+				pbCluster = &pb.NodeCluster{
+					Id:   int64(cluster.Id),
+					Name: cluster.Name,
+				}
+			}
+		}
+
+		if message.NodeId > 0 {
+			node, err := models.SharedNodeDAO.FindEnabledNode(int64(message.NodeId))
+			if err != nil {
+				return nil, err
+			}
+			if node != nil {
+				pbNode = &pb.Node{
+					Id:   int64(node.Id),
+					Name: node.Name,
+				}
+			}
+		}
+
+		result = append(result, &pb.Message{
+			Id:         int64(message.Id),
+			Type:       message.Type,
+			Body:       message.Body,
+			Level:      message.Level,
+			ParamsJSON: []byte(message.Params),
+			IsRead:     message.IsRead == 1,
+			CreatedAt:  int64(message.CreatedAt),
+			Cluster:    pbCluster,
+			Node:       pbNode,
+		})
+	}
+
+	return &pb.ListUnreadMessagesResponse{Messages: result}, nil
+}
+
+// 设置消息已读状态
+func (this *MessageService) UpdateMessageRead(ctx context.Context, req *pb.UpdateMessageReadRequest) (*pb.RPCUpdateSuccess, error) {
+	// 校验请求
+	_, _, err := rpcutils.ValidateRequest(ctx, rpcutils.UserTypeAdmin)
+	if err != nil {
+		return nil, err
+	}
+
+	err = models.SharedMessageDAO.UpdateMessageRead(req.MessageId, req.IsRead)
+	if err != nil {
+		return nil, err
+	}
+	return rpcutils.RPCUpdateSuccess()
+}
+
+// 设置一组消息已读状态
+func (this *MessageService) UpdateMessagesRead(ctx context.Context, req *pb.UpdateMessagesReadRequest) (*pb.RPCUpdateSuccess, error) {
+	// 校验请求
+	_, _, err := rpcutils.ValidateRequest(ctx, rpcutils.UserTypeAdmin)
+	if err != nil {
+		return nil, err
+	}
+
+	err = models.SharedMessageDAO.UpdateMessagesRead(req.MessageIds, req.IsRead)
+	if err != nil {
+		return nil, err
+	}
+	return rpcutils.RPCUpdateSuccess()
+}
+
+// 设置所有消息为已读
+func (this *MessageService) UpdateAllMessagesRead(ctx context.Context, req *pb.UpdateAllMessagesReadRequest) (*pb.RPCUpdateSuccess, error) {
+	// 校验请求
+	_, _, err := rpcutils.ValidateRequest(ctx, rpcutils.UserTypeAdmin)
+	if err != nil {
+		return nil, err
+	}
+
+	err = models.SharedMessageDAO.UpdateAllMessagesRead()
+	if err != nil {
+		return nil, err
+	}
+	return rpcutils.RPCUpdateSuccess()
+}