pig/pig-farm-controller
1
0
Fork 0
Code Issues 11 Pull Requests Actions Packages Projects Releases Wiki Activity

调整目录结构

Browse Source
This commit is contained in:
huang
2025-10-02 00:18:13 +08:00
parent 0b8b37511e
commit 829f0a6253
24 changed files with 85 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
internal/app/api/api.go
View File

@@ -20,10 +20,10 @@ import (
"git.huangwc.com/pig/pig-farm-controller/internal/app/controller/plan"
"git.huangwc.com/pig/pig-farm-controller/internal/app/controller/user"
"git.huangwc.com/pig/pig-farm-controller/internal/app/middleware"
"git.huangwc.com/pig/pig-farm-controller/internal/app/service/audit"
"git.huangwc.com/pig/pig-farm-controller/internal/app/service/task"
"git.huangwc.com/pig/pig-farm-controller/internal/app/service/token"
"git.huangwc.com/pig/pig-farm-controller/internal/app/service/transport"
"git.huangwc.com/pig/pig-farm-controller/internal/app/webhook"
"git.huangwc.com/pig/pig-farm-controller/internal/domain/audit"
"git.huangwc.com/pig/pig-farm-controller/internal/domain/task"
"git.huangwc.com/pig/pig-farm-controller/internal/domain/token"
"git.huangwc.com/pig/pig-farm-controller/internal/infra/config"
"git.huangwc.com/pig/pig-farm-controller/internal/infra/logs"
"git.huangwc.com/pig/pig-farm-controller/internal/infra/repository"
@@ -45,7 +45,7 @@ type API struct {
userController *user.Controller // 用户控制器实例
deviceController *device.Controller // 设备控制器实例
planController *plan.Controller // 计划控制器实例
listenHandler transport.ListenHandler // 设备上行事件监听器
listenHandler webhook.ListenHandler // 设备上行事件监听器
analysisTaskManager *task.AnalysisPlanTaskManager // 计划触发器管理器实例
}
@@ -61,7 +61,7 @@ func NewAPI(cfg config.ServerConfig,
userActionLogRepository repository.UserActionLogRepository,
tokenService token.TokenService,
auditService audit.Service, // 注入审计服务
listenHandler transport.ListenHandler,
listenHandler webhook.ListenHandler,
analysisTaskManager *task.AnalysisPlanTaskManager) *API {
// 设置 Gin 模式，例如 gin.ReleaseMode (生产模式) 或 gin.DebugMode (开发模式)
// 从配置中获取 Gin 模式
@@ -127,10 +127,7 @@ func (a *API) setupRoutes() {
a.logger.Info("pprof 接口注册成功")
// 上行事件监听路由
a.engine.POST("/upstream", func(c *gin.Context) {
h := a.listenHandler.Handler()
h.ServeHTTP(c.Writer, c.Request)
})
a.engine.POST("/upstream", gin.WrapH(a.listenHandler.Handler()))
a.logger.Info("上行事件监听接口注册成功")
// 添加 Swagger UI 路由, Swagger UI可在 /swagger/index.html 上找到

2
internal/app/controller/plan/plan_controller.go
View File

@@ -5,7 +5,7 @@ import (
"strconv"
"git.huangwc.com/pig/pig-farm-controller/internal/app/controller"
task "git.huangwc.com/pig/pig-farm-controller/internal/app/service/task"
"git.huangwc.com/pig/pig-farm-controller/internal/domain/task"
"git.huangwc.com/pig/pig-farm-controller/internal/infra/logs"
"git.huangwc.com/pig/pig-farm-controller/internal/infra/models"
"git.huangwc.com/pig/pig-farm-controller/internal/infra/repository"

2
internal/app/controller/user/user_controller.go
View File

@@ -5,7 +5,7 @@ import (
"time"
"git.huangwc.com/pig/pig-farm-controller/internal/app/controller"
"git.huangwc.com/pig/pig-farm-controller/internal/app/service/token"
"git.huangwc.com/pig/pig-farm-controller/internal/domain/token"
"git.huangwc.com/pig/pig-farm-controller/internal/infra/logs"
"git.huangwc.com/pig/pig-farm-controller/internal/infra/models"
"git.huangwc.com/pig/pig-farm-controller/internal/infra/repository"

2
internal/app/middleware/audit.go
View File

@@ -6,7 +6,7 @@ import (
"io"
"strconv"
"git.huangwc.com/pig/pig-farm-controller/internal/app/service/audit"
"git.huangwc.com/pig/pig-farm-controller/internal/domain/audit"
"git.huangwc.com/pig/pig-farm-controller/internal/infra/models"
"github.com/gin-gonic/gin"
)

2
internal/app/middleware/auth.go
View File

@@ -5,7 +5,7 @@ import (
"net/http"
"strings"
"git.huangwc.com/pig/pig-farm-controller/internal/app/service/token"
"git.huangwc.com/pig/pig-farm-controller/internal/domain/token"
"git.huangwc.com/pig/pig-farm-controller/internal/infra/models"
"git.huangwc.com/pig/pig-farm-controller/internal/infra/repository"
"github.com/gin-gonic/gin"

69
internal/app/service/audit/service.go
View File

@@ -1,69 +0,0 @@
// Package audit 提供了用户操作审计相关的功能
package audit
import (
"time"
"git.huangwc.com/pig/pig-farm-controller/internal/infra/logs"
"git.huangwc.com/pig/pig-farm-controller/internal/infra/models"
"git.huangwc.com/pig/pig-farm-controller/internal/infra/repository"
// 移除对 "github.com/gin-gonic/gin" 的直接依赖
)
// RequestContext 封装了审计日志所需的请求上下文信息
type RequestContext struct {
ClientIP string
HTTPPath string
HTTPMethod string
}
// Service 定义了审计服务的接口
type Service interface {
LogAction(user *models.User, reqCtx RequestContext, actionType, description string, targetResource interface{}, status models.AuditStatus, resultDetails string)
}
// service 是 Service 接口的实现
type service struct {
userActionLogRepository repository.UserActionLogRepository
logger *logs.Logger
}
// NewService 创建一个新的审计服务实例
func NewService(repo repository.UserActionLogRepository, logger *logs.Logger) Service {
return &service{userActionLogRepository: repo, logger: logger}
}
// LogAction 记录一个用户操作。它在一个新的 goroutine 中异步执行，以避免阻塞主请求。
func (s *service) LogAction(user *models.User, reqCtx RequestContext, actionType, description string, targetResource interface{}, status models.AuditStatus, resultDetails string) {
// 不再从 context 中获取用户信息，直接使用传入的 user 对象
if user == nil {
s.logger.Warnw("无法记录审计日志：传入的用户对象为 nil")
return
}
log := &models.UserActionLog{
Time: time.Now(),
UserID: user.ID,
Username: user.Username, // 用户名快照
SourceIP: reqCtx.ClientIP,
ActionType: actionType,
Description: description,
Status: status,
HTTPPath: reqCtx.HTTPPath,
HTTPMethod: reqCtx.HTTPMethod,
ResultDetails: resultDetails,
}
// 使用模型提供的方法来设置 TargetResource
if err := log.SetTargetResource(targetResource); err != nil {
s.logger.Errorw("无法记录审计日志：序列化 targetResource 失败", "error", err)
// 即使序列化失败，我们可能仍然希望记录操作本身，所以不在此处 return
}
// 异步写入数据库，不阻塞当前请求
go func() {
if err := s.userActionLogRepository.Create(log); err != nil {
s.logger.Errorw("异步保存审计日志失败", "error", err)
}
}()
}

42
internal/app/service/device/device_service.go
View File

@@ -1,42 +0,0 @@
package device
import "git.huangwc.com/pig/pig-farm-controller/internal/infra/models"
// 设备行为
type DeviceAction string
var (
DeviceActionStart DeviceAction = "start" // 启动
DeviceActionStop DeviceAction = "stop" // 停止
)
// 指令类型
type Method string
var (
MethodSwitch Method = "switch" // 启停指令
)
// Service 抽象了一组方法用于控制设备行为
type Service interface {
// Switch 用于切换指定设备的状态, 比如启动和停止
Switch(device *models.Device, action DeviceAction) error
// Collect 用于发起对指定区域主控下的多个设备的批量采集请求。
Collect(regionalControllerID uint, devicesToCollect []*models.Device) error
}
// 设备操作指令通用结构(最外层)
type DeviceRequest struct {
MessageID int // 消息ID, 用于后续匹配响应
Method string // 这是什么指令
Data interface{} // 指令参数
}
// 设备操作指令通用响应结构(最外层)
type DeviceResponse struct {
MessageID int // 消息ID, 用于匹配这是哪一个请求的响应
Message string
Data interface{} // 响应内容
}

249
internal/app/service/device/general_device_service.go
View File

@@ -1,249 +0,0 @@
package device
import (
"errors"
"fmt"
"time"
"git.huangwc.com/pig/pig-farm-controller/internal/app/service/device/proto"
"git.huangwc.com/pig/pig-farm-controller/internal/infra/logs"
"git.huangwc.com/pig/pig-farm-controller/internal/infra/models"
"git.huangwc.com/pig/pig-farm-controller/internal/infra/repository"
"git.huangwc.com/pig/pig-farm-controller/internal/infra/transport"
"git.huangwc.com/pig/pig-farm-controller/internal/infra/utils/command_generater"
"github.com/google/uuid"
gproto "google.golang.org/protobuf/proto"
"google.golang.org/protobuf/types/known/anypb"
)
type GeneralDeviceService struct {
deviceRepo repository.DeviceRepository
deviceCommandLogRepo repository.DeviceCommandLogRepository
pendingCollectionRepo repository.PendingCollectionRepository
logger *logs.Logger
comm transport.Communicator
}
// NewGeneralDeviceService 创建一个通用设备服务
func NewGeneralDeviceService(
deviceRepo repository.DeviceRepository,
deviceCommandLogRepo repository.DeviceCommandLogRepository,
pendingCollectionRepo repository.PendingCollectionRepository,
logger *logs.Logger,
comm transport.Communicator,
) Service {
return &GeneralDeviceService{
deviceRepo: deviceRepo,
deviceCommandLogRepo: deviceCommandLogRepo,
pendingCollectionRepo: pendingCollectionRepo,
logger: logger,
comm: comm,
}
}
func (g *GeneralDeviceService) Switch(device *models.Device, action DeviceAction) error {
// 1. 依赖模型自身的 SelfCheck 进行全面校验
if err := device.SelfCheck(); err != nil {
return fmt.Errorf("设备 %v(id=%v) 未通过自检: %w", device.Name, device.ID, err)
}
if err := device.DeviceTemplate.SelfCheck(); err != nil {
return fmt.Errorf("设备 %v(id=%v) 的模板未通过自检: %w", device.Name, device.ID, err)
}
// 2. 检查预加载的 AreaController 是否有效
areaController := &device.AreaController
if err := areaController.SelfCheck(); err != nil {
return fmt.Errorf("区域主控 %v(id=%v) 未通过自检: %w", areaController.Name, areaController.ID, err)
}
// 3. 使用模型层预定义的 Bus485Properties 结构体解析设备属性
var deviceProps models.Bus485Properties
if err := device.ParseProperties(&deviceProps); err != nil {
return fmt.Errorf("解析设备 %v(id=%v) 的属性失败: %w", device.Name, device.ID, err)
}
// 4. 解析设备模板中的开关指令参数
var switchCmd models.SwitchCommands
if err := device.DeviceTemplate.ParseCommands(&switchCmd); err != nil {
return fmt.Errorf("解析设备 %v(id=%v) 的开关指令失败: %w", device.Name, device.ID, err)
}
// 5. 根据 action 生成 Modbus RTU 写入指令
onOffState := true // 默认为开启
if action == DeviceActionStop { // 如果是停止动作，则设置为关闭
onOffState = false
}
modbusCommandBytes, err := command_generater.GenerateModbusRTUSwitchCommand(
deviceProps.BusAddress,
switchCmd.ModbusStartAddress,
onOffState,
)
if err != nil {
return fmt.Errorf("生成Modbus RTU写入指令失败: %w", err)
}
// 6. 构建 Protobuf Raw485Command，包含总线号
raw485Cmd := &proto.Raw485Command{
BusNumber: int32(deviceProps.BusNumber), // 添加总线号
CommandBytes: modbusCommandBytes,
}
data, err := anypb.New(raw485Cmd)
if err != nil {
return fmt.Errorf("创建 Raw485Command Any Protobuf 失败: %w", err)
}
instruction := &proto.Instruction{
Method: proto.MethodType_INSTRUCTION, // 使用通用指令类型
Data: data,
}
message, err := gproto.Marshal(instruction)
if err != nil {
return fmt.Errorf("序列化指令失败: %w", err)
}
// 7. 发送指令
networkID := areaController.NetworkID
sendResult, err := g.comm.Send(networkID, message)
if err != nil {
return fmt.Errorf("发送指令到 %s 失败: %w", networkID, err)
}
// 8. 创建并保存命令日志
logRecord := &models.DeviceCommandLog{
MessageID: sendResult.MessageID,
DeviceID: areaController.ID,
SentAt: time.Now(),
}
if err := g.deviceCommandLogRepo.Create(logRecord); err != nil {
// 记录日志失败是一个需要关注的问题，但可能不应该中断主流程。
// 我们记录一个错误日志，然后成功返回。
g.logger.Errorf("创建指令日志失败 (MessageID: %s): %v", sendResult.MessageID, err)
}
g.logger.Infof("成功发送指令到 %s 并创建日志 (MessageID: %s)", networkID, sendResult.MessageID)
return nil
}
// Collect 实现了 Service 接口，用于发起对指定区域主控下的多个设备的批量采集请求。
func (g *GeneralDeviceService) Collect(regionalControllerID uint, devicesToCollect []*models.Device) error {
if len(devicesToCollect) == 0 {
g.logger.Info("待采集设备列表为空，无需执行采集任务。")
return nil
}
// 1. 从设备列表中获取预加载的区域主控，并进行校验
regionalController := &devicesToCollect[0].AreaController
if regionalController.ID != regionalControllerID {
return fmt.Errorf("设备列表与指定的区域主控ID (%d) 不匹配", regionalControllerID)
}
if err := regionalController.SelfCheck(); err != nil {
return fmt.Errorf("区域主控 (ID: %d) 未通过自检: %w", regionalControllerID, err)
}
// 2. 准备采集任务列表
var childDeviceIDs []uint
var collectTasks []*proto.CollectTask
for _, dev := range devicesToCollect {
// 依赖模型自身的 SelfCheck 进行全面校验
if err := dev.SelfCheck(); err != nil {
g.logger.Warnf("跳过设备 %d，因其未通过自检: %v", dev.ID, err)
continue
}
if err := dev.DeviceTemplate.SelfCheck(); err != nil {
g.logger.Warnf("跳过设备 %d，因其设备模板未通过自检: %v", dev.ID, err)
continue
}
// 使用模板的 ParseCommands 方法获取传感器指令参数
var sensorCmd models.SensorCommands
if err := dev.DeviceTemplate.ParseCommands(&sensorCmd); err != nil {
g.logger.Warnf("跳过设备 %d，因其模板指令无法解析为 SensorCommands: %v", dev.ID, err)
continue
}
// 使用模型层预定义的 Bus485Properties 结构体解析设备属性
var deviceProps models.Bus485Properties
if err := dev.ParseProperties(&deviceProps); err != nil {
g.logger.Warnf("跳过设备 %d，因其属性解析失败: %v", dev.ID, err)
continue
}
// 生成 Modbus RTU 读取指令
modbusCommandBytes, err := command_generater.GenerateModbusRTUReadCommand(
deviceProps.BusAddress,
sensorCmd.ModbusFunctionCode,
sensorCmd.ModbusStartAddress,
sensorCmd.ModbusQuantity,
)
if err != nil {
g.logger.Warnf("跳过设备 %d，因生成Modbus RTU读取指令失败: %v", dev.ID, err)
continue
}
// 构建 Raw485Command，包含总线号
raw485Cmd := &proto.Raw485Command{
BusNumber: int32(deviceProps.BusNumber), // 添加总线号
CommandBytes: modbusCommandBytes,
}
collectTasks = append(collectTasks, &proto.CollectTask{
Command: raw485Cmd,
})
childDeviceIDs = append(childDeviceIDs, dev.ID)
}
if len(childDeviceIDs) == 0 {
return errors.New("经过滤后，没有可通过自检的有效设备")
}
// 3. 构建并发送指令
networkID := regionalController.NetworkID
// 4. 创建待处理请求记录
correlationID := uuid.New().String()
pendingReq := &models.PendingCollection{
CorrelationID: correlationID,
DeviceID: regionalController.ID,
CommandMetadata: childDeviceIDs,
Status: models.PendingStatusPending,
CreatedAt: time.Now(),
}
if err := g.pendingCollectionRepo.Create(pendingReq); err != nil {
g.logger.Errorf("创建待采集请求失败 (CorrelationID: %s): %v", correlationID, err)
return err
}
g.logger.Infof("成功创建待采集请求 (CorrelationID: %s, DeviceID: %d)", correlationID, regionalController.ID)
// 5. 构建最终的空中载荷
batchCmd := &proto.BatchCollectCommand{
CorrelationId: correlationID,
Tasks: collectTasks,
}
anyData, err := anypb.New(batchCmd)
if err != nil {
g.logger.Errorf("创建 Any Protobuf 失败 (CorrelationID: %s): %v", correlationID, err)
return err
}
instruction := &proto.Instruction{
Method: proto.MethodType_COLLECT, // 使用 COLLECT 指令类型
Data: anyData,
}
payload, err := gproto.Marshal(instruction)
if err != nil {
g.logger.Errorf("序列化采集指令失败 (CorrelationID: %s): %v", correlationID, err)
return err
}
if _, err := g.comm.Send(networkID, payload); err != nil {
g.logger.DPanicf("待采集请求 (CorrelationID: %s) 已创建，但发送到设备失败: %v。数据可能不一致！", correlationID, err)
return err
}
g.logger.Infof("成功将采集请求 (CorrelationID: %s) 发送到设备 %s", correlationID, networkID)
return nil
}

417
internal/app/service/device/proto/device.pb.go
View File

@@ -1,417 +0,0 @@
// Code generated by protoc-gen-go. DO NOT EDIT.
// versions:
// protoc-gen-go v1.36.9
// protoc v6.32.1
// source: device.proto
package proto
import (
protoreflect "google.golang.org/protobuf/reflect/protoreflect"
protoimpl "google.golang.org/protobuf/runtime/protoimpl"
anypb "google.golang.org/protobuf/types/known/anypb"
reflect "reflect"
sync "sync"
unsafe "unsafe"
)
const (
// Verify that this generated code is sufficiently up-to-date.
_ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion)
// Verify that runtime/protoimpl is sufficiently up-to-date.
_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
)
// 指令类型
type MethodType int32
const (
MethodType_INSTRUCTION MethodType = 0 // 下发指令
MethodType_COLLECT MethodType = 1 // 批量采集
)
// Enum value maps for MethodType.
var (
MethodType_name = map[int32]string{
0: "INSTRUCTION",
1: "COLLECT",
}
MethodType_value = map[string]int32{
"INSTRUCTION": 0,
"COLLECT": 1,
}
)
func (x MethodType) Enum() *MethodType {
p := new(MethodType)
*p = x
return p
}
func (x MethodType) String() string {
return protoimpl.X.EnumStringOf(x.Descriptor(), protoreflect.EnumNumber(x))
}
func (MethodType) Descriptor() protoreflect.EnumDescriptor {
return file_device_proto_enumTypes[0].Descriptor()
}
func (MethodType) Type() protoreflect.EnumType {
return &file_device_proto_enumTypes[0]
}
func (x MethodType) Number() protoreflect.EnumNumber {
return protoreflect.EnumNumber(x)
}
// Deprecated: Use MethodType.Descriptor instead.
func (MethodType) EnumDescriptor() ([]byte, []int) {
return file_device_proto_rawDescGZIP(), []int{0}
}
// 平台生成的原始485指令，单片机直接发送到总线
type Raw485Command struct {
state protoimpl.MessageState `protogen:"open.v1"`
BusNumber int32 `protobuf:"varint,1,opt,name=bus_number,json=busNumber,proto3" json:"bus_number,omitempty"` // 总线号，用于指示单片机将指令发送到哪个总线
CommandBytes []byte `protobuf:"bytes,2,opt,name=command_bytes,json=commandBytes,proto3" json:"command_bytes,omitempty"` // 原始485指令的字节数组
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *Raw485Command) Reset() {
*x = Raw485Command{}
mi := &file_device_proto_msgTypes[0]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *Raw485Command) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*Raw485Command) ProtoMessage() {}
func (x *Raw485Command) ProtoReflect() protoreflect.Message {
mi := &file_device_proto_msgTypes[0]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use Raw485Command.ProtoReflect.Descriptor instead.
func (*Raw485Command) Descriptor() ([]byte, []int) {
return file_device_proto_rawDescGZIP(), []int{0}
}
func (x *Raw485Command) GetBusNumber() int32 {
if x != nil {
return x.BusNumber
}
return 0
}
func (x *Raw485Command) GetCommandBytes() []byte {
if x != nil {
return x.CommandBytes
}
return nil
}
// 指令 (所有空中数据都会被包装在这里面)
// data字段现在可以包含 Raw485Command，表示平台生成的原始485指令。
type Instruction struct {
state protoimpl.MessageState `protogen:"open.v1"`
Method MethodType `protobuf:"varint,1,opt,name=method,proto3,enum=device.MethodType" json:"method,omitempty"`
Data *anypb.Any `protobuf:"bytes,2,opt,name=data,proto3" json:"data,omitempty"` // 可以是 Switch, Raw485Command 等
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *Instruction) Reset() {
*x = Instruction{}
mi := &file_device_proto_msgTypes[1]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *Instruction) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*Instruction) ProtoMessage() {}
func (x *Instruction) ProtoReflect() protoreflect.Message {
mi := &file_device_proto_msgTypes[1]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use Instruction.ProtoReflect.Descriptor instead.
func (*Instruction) Descriptor() ([]byte, []int) {
return file_device_proto_rawDescGZIP(), []int{1}
}
func (x *Instruction) GetMethod() MethodType {
if x != nil {
return x.Method
}
return MethodType_INSTRUCTION
}
func (x *Instruction) GetData() *anypb.Any {
if x != nil {
return x.Data
}
return nil
}
// BatchCollectCommand
// 用于在平台内部构建一个完整的、包含所有元数据的批量采集任务。
// 这个消息本身不会被发送到设备。
type BatchCollectCommand struct {
state protoimpl.MessageState `protogen:"open.v1"`
CorrelationId string `protobuf:"bytes,1,opt,name=correlation_id,json=correlationId,proto3" json:"correlation_id,omitempty"` // 用于关联请求和响应的唯一ID
Tasks []*CollectTask `protobuf:"bytes,2,rep,name=tasks,proto3" json:"tasks,omitempty"` // 采集任务列表
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *BatchCollectCommand) Reset() {
*x = BatchCollectCommand{}
mi := &file_device_proto_msgTypes[2]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *BatchCollectCommand) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*BatchCollectCommand) ProtoMessage() {}
func (x *BatchCollectCommand) ProtoReflect() protoreflect.Message {
mi := &file_device_proto_msgTypes[2]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use BatchCollectCommand.ProtoReflect.Descriptor instead.
func (*BatchCollectCommand) Descriptor() ([]byte, []int) {
return file_device_proto_rawDescGZIP(), []int{2}
}
func (x *BatchCollectCommand) GetCorrelationId() string {
if x != nil {
return x.CorrelationId
}
return ""
}
func (x *BatchCollectCommand) GetTasks() []*CollectTask {
if x != nil {
return x.Tasks
}
return nil
}
// CollectTask
// 定义了单个采集任务的“意图”。现在直接包含平台生成的原始485指令，并带上总线号。
type CollectTask struct {
state protoimpl.MessageState `protogen:"open.v1"`
Command *Raw485Command `protobuf:"bytes,2,opt,name=command,proto3" json:"command,omitempty"` // 平台生成的原始485指令
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *CollectTask) Reset() {
*x = CollectTask{}
mi := &file_device_proto_msgTypes[3]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *CollectTask) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*CollectTask) ProtoMessage() {}
func (x *CollectTask) ProtoReflect() protoreflect.Message {
mi := &file_device_proto_msgTypes[3]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use CollectTask.ProtoReflect.Descriptor instead.
func (*CollectTask) Descriptor() ([]byte, []int) {
return file_device_proto_rawDescGZIP(), []int{3}
}
func (x *CollectTask) GetCommand() *Raw485Command {
if x != nil {
return x.Command
}
return nil
}
// CollectResult
// 这是设备响应的、极致精简的数据包。
type CollectResult struct {
state protoimpl.MessageState `protogen:"open.v1"`
CorrelationId string `protobuf:"bytes,1,opt,name=correlation_id,json=correlationId,proto3" json:"correlation_id,omitempty"` // 从下行指令中原样返回的关联ID
Values []float32 `protobuf:"fixed32,2,rep,packed,name=values,proto3" json:"values,omitempty"` // 按预定顺序排列的采集值
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *CollectResult) Reset() {
*x = CollectResult{}
mi := &file_device_proto_msgTypes[4]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *CollectResult) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*CollectResult) ProtoMessage() {}
func (x *CollectResult) ProtoReflect() protoreflect.Message {
mi := &file_device_proto_msgTypes[4]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use CollectResult.ProtoReflect.Descriptor instead.
func (*CollectResult) Descriptor() ([]byte, []int) {
return file_device_proto_rawDescGZIP(), []int{4}
}
func (x *CollectResult) GetCorrelationId() string {
if x != nil {
return x.CorrelationId
}
return ""
}
func (x *CollectResult) GetValues() []float32 {
if x != nil {
return x.Values
}
return nil
}
var File_device_proto protoreflect.FileDescriptor
const file_device_proto_rawDesc = "" +
"\n" +
"\fdevice.proto\x12\x06device\x1a\x19google/protobuf/any.proto\"S\n" +
"\rRaw485Command\x12\x1d\n" +
"\n" +
"bus_number\x18\x01 \x01(\x05R\tbusNumber\x12#\n" +
"\rcommand_bytes\x18\x02 \x01(\fR\fcommandBytes\"c\n" +
"\vInstruction\x12*\n" +
"\x06method\x18\x01 \x01(\x0e2\x12.device.MethodTypeR\x06method\x12(\n" +
"\x04data\x18\x02 \x01(\v2\x14.google.protobuf.AnyR\x04data\"g\n" +
"\x13BatchCollectCommand\x12%\n" +
"\x0ecorrelation_id\x18\x01 \x01(\tR\rcorrelationId\x12)\n" +
"\x05tasks\x18\x02 \x03(\v2\x13.device.CollectTaskR\x05tasks\">\n" +
"\vCollectTask\x12/\n" +
"\acommand\x18\x02 \x01(\v2\x15.device.Raw485CommandR\acommand\"N\n" +
"\rCollectResult\x12%\n" +
"\x0ecorrelation_id\x18\x01 \x01(\tR\rcorrelationId\x12\x16\n" +
"\x06values\x18\x02 \x03(\x02R\x06values**\n" +
"\n" +
"MethodType\x12\x0f\n" +
"\vINSTRUCTION\x10\x00\x12\v\n" +
"\aCOLLECT\x10\x01B#Z!internal/app/service/device/protob\x06proto3"
var (
file_device_proto_rawDescOnce sync.Once
file_device_proto_rawDescData []byte
)
func file_device_proto_rawDescGZIP() []byte {
file_device_proto_rawDescOnce.Do(func() {
file_device_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_device_proto_rawDesc), len(file_device_proto_rawDesc)))
})
return file_device_proto_rawDescData
}
var file_device_proto_enumTypes = make([]protoimpl.EnumInfo, 1)
var file_device_proto_msgTypes = make([]protoimpl.MessageInfo, 5)
var file_device_proto_goTypes = []any{
(MethodType)(0), // 0: device.MethodType
(*Raw485Command)(nil), // 1: device.Raw485Command
(*Instruction)(nil), // 2: device.Instruction
(*BatchCollectCommand)(nil), // 3: device.BatchCollectCommand
(*CollectTask)(nil), // 4: device.CollectTask
(*CollectResult)(nil), // 5: device.CollectResult
(*anypb.Any)(nil), // 6: google.protobuf.Any
}
var file_device_proto_depIdxs = []int32{
0, // 0: device.Instruction.method:type_name -> device.MethodType
6, // 1: device.Instruction.data:type_name -> google.protobuf.Any
4, // 2: device.BatchCollectCommand.tasks:type_name -> device.CollectTask
1, // 3: device.CollectTask.command:type_name -> device.Raw485Command
4, // [4:4] is the sub-list for method output_type
4, // [4:4] is the sub-list for method input_type
4, // [4:4] is the sub-list for extension type_name
4, // [4:4] is the sub-list for extension extendee
0, // [0:4] is the sub-list for field type_name
}
func init() { file_device_proto_init() }
func file_device_proto_init() {
if File_device_proto != nil {
return
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_device_proto_rawDesc), len(file_device_proto_rawDesc)),
NumEnums: 1,
NumMessages: 5,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_device_proto_goTypes,
DependencyIndexes: file_device_proto_depIdxs,
EnumInfos: file_device_proto_enumTypes,
MessageInfos: file_device_proto_msgTypes,
}.Build()
File_device_proto = out.File
file_device_proto_goTypes = nil
file_device_proto_depIdxs = nil
}

51
internal/app/service/device/proto/device.proto
View File

@@ -1,51 +0,0 @@
syntax = "proto3";
package device;
import "google/protobuf/any.proto";
option go_package = "internal/app/service/device/proto";
// --- 通用指令结构 ---
// 指令类型
enum MethodType {
INSTRUCTION = 0; // 下发指令
COLLECT = 1; // 批量采集
}
// 平台生成的原始485指令，单片机直接发送到总线
message Raw485Command {
int32 bus_number = 1; // 总线号，用于指示单片机将指令发送到哪个总线
bytes command_bytes = 2; // 原始485指令的字节数组
}
// 指令 (所有空中数据都会被包装在这里面)
// data字段现在可以包含 Raw485Command，表示平台生成的原始485指令。
message Instruction {
MethodType method = 1;
google.protobuf.Any data = 2; // 可以是 Switch, Raw485Command 等
}
// --- 批量采集相关结构 ---
// BatchCollectCommand
// 用于在平台内部构建一个完整的、包含所有元数据的批量采集任务。
// 这个消息本身不会被发送到设备。
message BatchCollectCommand {
string correlation_id = 1; // 用于关联请求和响应的唯一ID
repeated CollectTask tasks = 2; // 采集任务列表
}
// CollectTask
// 定义了单个采集任务的“意图”。现在直接包含平台生成的原始485指令，并带上总线号。
message CollectTask {
Raw485Command command = 2; // 平台生成的原始485指令
}
// CollectResult
// 这是设备响应的、极致精简的数据包。
message CollectResult {
string correlation_id = 1; // 从下行指令中原样返回的关联ID
repeated float values = 2; // 按预定顺序排列的采集值
}

316
internal/app/service/task/analysis_plan_task_manager.go
View File

@@ -1,316 +0,0 @@
package task
import (
"fmt"
"sync"
"time"
"git.huangwc.com/pig/pig-farm-controller/internal/infra/logs"
"git.huangwc.com/pig/pig-farm-controller/internal/infra/models"
"git.huangwc.com/pig/pig-farm-controller/internal/infra/repository"
"git.huangwc.com/pig/pig-farm-controller/internal/infra/utils"
)
// AnalysisPlanTaskManager 负责管理分析计划的触发器任务。
// 它确保数据库中可执行的计划在待执行队列中有对应的触发器，并移除无效的触发器。
// 这是一个有状态的组件，包含一个互斥锁以确保并发安全。
type AnalysisPlanTaskManager struct {
planRepo repository.PlanRepository
pendingTaskRepo repository.PendingTaskRepository
executionLogRepo repository.ExecutionLogRepository
logger *logs.Logger
mu sync.Mutex
}
// NewAnalysisPlanTaskManager 是 AnalysisPlanTaskManager 的构造函数。
func NewAnalysisPlanTaskManager(
planRepo repository.PlanRepository,
pendingTaskRepo repository.PendingTaskRepository,
executionLogRepo repository.ExecutionLogRepository,
logger *logs.Logger,
) *AnalysisPlanTaskManager {
return &AnalysisPlanTaskManager{
planRepo: planRepo,
pendingTaskRepo: pendingTaskRepo,
executionLogRepo: executionLogRepo,
logger: logger,
}
}
// Refresh 同步数据库中的计划状态和待执行队列中的触发器任务。
// 这是一个编排方法，将复杂的逻辑分解到多个内部方法中。
func (m *AnalysisPlanTaskManager) Refresh() error {
m.mu.Lock()
defer m.mu.Unlock()
m.logger.Info("开始同步计划任务管理器...")
// 1. 一次性获取所有需要的数据
runnablePlans, invalidPlanIDs, pendingTasks, err := m.getRefreshData()
if err != nil {
return fmt.Errorf("获取刷新数据失败: %w", err)
}
// 2. 清理所有与失效计划相关的待执行任务
if err := m.cleanupInvalidTasks(invalidPlanIDs, pendingTasks); err != nil {
// 仅记录错误，清理失败不应阻止新任务的添加
m.logger.Errorf("清理无效任务时出错: %v", err)
}
// 3. 添加或更新触发器
if err := m.addOrUpdateTriggers(runnablePlans, pendingTasks); err != nil {
return fmt.Errorf("添加或更新触发器时出错: %w", err)
}
m.logger.Info("计划任务管理器同步完成.")
return nil
}
// CreateOrUpdateTrigger 为给定的 planID 创建其关联的触发任务。
// 如果触发器已存在，会根据计划类型更新其执行时间。
func (m *AnalysisPlanTaskManager) CreateOrUpdateTrigger(planID uint) error {
m.mu.Lock()
defer m.mu.Unlock()
// 检查计划是否可执行
plan, err := m.planRepo.GetBasicPlanByID(planID)
if err != nil {
return fmt.Errorf("获取计划基本信息失败: %w", err)
}
if plan.Status != models.PlanStatusEnabled {
return fmt.Errorf("计划 #%d 当前状态为 '%d'，无法创建或更新触发器", planID, plan.Status)
}
// 查找现有触发器
existingTrigger, err := m.pendingTaskRepo.FindPendingTriggerByPlanID(planID)
if err != nil {
return fmt.Errorf("查找现有触发器失败: %w", err)
}
// 如果触发器已存在，则根据计划类型更新其执行时间
if existingTrigger != nil {
var expectedExecuteAt time.Time
if plan.ExecutionType == models.PlanExecutionTypeManual {
// 手动计划，如果再次触发，则立即执行
expectedExecuteAt = time.Now()
} else { // 自动计划
// 自动计划，根据 Cron 表达式计算下一次执行时间
next, err := utils.GetNextCronTime(plan.CronExpression)
if err != nil {
m.logger.Errorf("为计划 #%d 解析Cron表达式失败，无法更新触发器: %v", plan.ID, err)
return fmt.Errorf("解析 Cron 表达式失败: %w", err)
}
expectedExecuteAt = next
}
// 如果计算出的执行时间与当前待执行任务的时间不一致，则更新
if !existingTrigger.ExecuteAt.Equal(expectedExecuteAt) {
m.logger.Infof("计划 #%d 的执行时间已变更，正在更新触发器 #%d 的执行时间从 %v 到 %v...", plan.ID, existingTrigger.ID, existingTrigger.ExecuteAt, expectedExecuteAt)
if err := m.pendingTaskRepo.UpdatePendingTaskExecuteAt(existingTrigger.ID, expectedExecuteAt); err != nil {
m.logger.Errorf("更新触发器 #%d 的执行时间失败: %v", existingTrigger.ID, err)
return fmt.Errorf("更新触发器执行时间失败: %w", err)
}
} else {
m.logger.Infof("计划 #%d 的触发器已存在且执行时间无需更新。", plan.ID)
}
return nil // 触发器已存在且已处理更新，直接返回
}
// 如果触发器不存在，则创建新的触发器
m.logger.Infof("为计划 #%d 创建新的触发器...", planID)
return m.createTriggerTask(plan)
}
// EnsureAnalysisTaskDefinition 确保计划的分析任务定义存在于 tasks 表中。
// 如果不存在，则会自动创建。此方法不涉及待执行队列。
func (m *AnalysisPlanTaskManager) EnsureAnalysisTaskDefinition(planID uint) error {
m.mu.Lock()
defer m.mu.Unlock()
plan, err := m.planRepo.GetBasicPlanByID(planID)
if err != nil {
return fmt.Errorf("确保分析任务定义失败：获取计划 #%d 基本信息时出错: %w", planID, err)
}
analysisTask, err := m.planRepo.FindPlanAnalysisTaskByPlanID(plan.ID)
if err != nil {
return fmt.Errorf("确保分析任务定义失败：查找计划 #%d 的分析任务时出错: %w", plan.ID, err)
}
if analysisTask == nil {
m.logger.Infof("未找到计划 #%d 关联的 'plan_analysis' 任务定义，将自动创建...", plan.ID)
_, err := m.planRepo.CreatePlanAnalysisTask(plan) // CreatePlanAnalysisTask returns *models.Task, error
if err != nil {
return fmt.Errorf("自动创建 'plan_analysis' 任务定义失败: %w", err)
}
m.logger.Infof("已成功为计划 #%d 创建 'plan_analysis' 任务定义。", plan.ID)
} else {
m.logger.Infof("计划 #%d 的 'plan_analysis' 任务定义已存在。", plan.ID)
}
return nil
}
// --- 内部私有方法 ---
// getRefreshData 从数据库获取刷新所需的所有数据。
func (m *AnalysisPlanTaskManager) getRefreshData() (runnablePlans []*models.Plan, invalidPlanIDs []uint, pendingTasks []models.PendingTask, err error) {
runnablePlans, err = m.planRepo.FindRunnablePlans()
if err != nil {
m.logger.Errorf("获取可执行计划列表失败: %v", err)
return
}
invalidPlans, err := m.planRepo.FindInactivePlans()
if err != nil {
m.logger.Errorf("获取失效计划列表失败: %v", err)
return
}
invalidPlanIDs = make([]uint, len(invalidPlans))
for i, p := range invalidPlans {
invalidPlanIDs[i] = p.ID
}
pendingTasks, err = m.pendingTaskRepo.FindAllPendingTasks()
if err != nil {
m.logger.Errorf("获取所有待执行任务失败: %v", err)
return
}
return
}
// cleanupInvalidTasks 清理所有与失效计划相关的待执行任务。
func (m *AnalysisPlanTaskManager) cleanupInvalidTasks(invalidPlanIDs []uint, allPendingTasks []models.PendingTask) error {
if len(invalidPlanIDs) == 0 {
return nil // 没有需要清理的计划
}
invalidPlanIDSet := make(map[uint]struct{}, len(invalidPlanIDs))
for _, id := range invalidPlanIDs {
invalidPlanIDSet[id] = struct{}{}
}
var tasksToDeleteIDs []uint
var logsToCancelIDs []uint
for _, pt := range allPendingTasks {
if pt.Task == nil { // 防御性编程，确保 Task 被预加载
continue
}
if _, isInvalid := invalidPlanIDSet[pt.Task.PlanID]; isInvalid {
tasksToDeleteIDs = append(tasksToDeleteIDs, pt.ID)
logsToCancelIDs = append(logsToCancelIDs, pt.TaskExecutionLogID)
}
}
if len(tasksToDeleteIDs) == 0 {
return nil // 没有找到需要清理的任务
}
m.logger.Infof("准备从待执行队列中清理 %d 个与失效计划相关的任务...", len(tasksToDeleteIDs))
// 批量删除待执行任务
if err := m.pendingTaskRepo.DeletePendingTasksByIDs(tasksToDeleteIDs); err != nil {
return fmt.Errorf("批量删除待执行任务失败: %w", err)
}
// 批量更新相关执行日志状态为“已取消”
if err := m.executionLogRepo.UpdateTaskExecutionLogStatusByIDs(logsToCancelIDs, models.ExecutionStatusCancelled); err != nil {
// 这是一个非关键性错误，只记录日志
m.logger.Warnf("批量更新日志状态为 'Cancelled' 失败: %v", err)
}
return nil
}
// addOrUpdateTriggers 检查、更新或创建触发器。
func (m *AnalysisPlanTaskManager) addOrUpdateTriggers(runnablePlans []*models.Plan, allPendingTasks []models.PendingTask) error {
// 创建一个映射，存放所有已在队列中的计划触发器
pendingTriggersMap := make(map[uint]models.PendingTask)
for _, pt := range allPendingTasks {
if pt.Task != nil && pt.Task.Type == models.TaskPlanAnalysis {
pendingTriggersMap[pt.Task.PlanID] = pt
}
}
for _, plan := range runnablePlans {
existingTrigger, exists := pendingTriggersMap[plan.ID]
if exists {
// --- 新增逻辑：检查并更新现有触发器 ---
// 只对自动计划检查时间更新
if plan.ExecutionType == models.PlanExecutionTypeAutomatic {
next, err := utils.GetNextCronTime(plan.CronExpression)
if err != nil {
m.logger.Errorf("为计划 #%d 解析Cron表达式失败，跳过更新: %v", plan.ID, err)
continue
}
// 如果数据库中记录的执行时间与根据当前Cron表达式计算出的下一次时间不一致，则更新
if !existingTrigger.ExecuteAt.Equal(next) {
m.logger.Infof("计划 #%d 的执行时间已变更，正在更新触发器 #%d 的执行时间从 %v 到 %v...", plan.ID, existingTrigger.ID, existingTrigger.ExecuteAt, next)
if err := m.pendingTaskRepo.UpdatePendingTaskExecuteAt(existingTrigger.ID, next); err != nil {
m.logger.Errorf("更新触发器 #%d 的执行时间失败: %v", existingTrigger.ID, err)
}
}
}
} else {
// --- 原有逻辑：为缺失的计划创建新触发器 ---
m.logger.Infof("发现应执行但队列中缺失的计划 #%d，正在为其创建触发器...", plan.ID)
if err := m.createTriggerTask(plan); err != nil {
m.logger.Errorf("为计划 #%d 创建触发器失败: %v", plan.ID, err)
// 继续处理下一个，不因单点失败而中断
}
}
}
return nil
}
// createTriggerTask 是创建触发器任务的内部核心逻辑。
func (m *AnalysisPlanTaskManager) createTriggerTask(plan *models.Plan) error {
analysisTask, err := m.planRepo.FindPlanAnalysisTaskByPlanID(plan.ID)
if err != nil {
return fmt.Errorf("查找计划分析任务失败: %w", err)
}
// --- 如果触发器任务定义不存在，则自动创建 ---
if analysisTask == nil {
m.logger.Warnf("未找到计划 #%d 关联的 'plan_analysis' 任务定义，将自动创建...", plan.ID)
newAnalysisTask, err := m.planRepo.CreatePlanAnalysisTask(plan)
if err != nil {
return fmt.Errorf("自动创建 'plan_analysis' 任务定义失败: %w", err)
}
analysisTask = newAnalysisTask
m.logger.Infof("已成功为计划 #%d 创建 'plan_analysis' 任务定义 (ID: %d)", plan.ID, analysisTask.ID)
}
var executeAt time.Time
if plan.ExecutionType == models.PlanExecutionTypeManual {
executeAt = time.Now()
} else {
next, err := utils.GetNextCronTime(plan.CronExpression)
if err != nil {
return fmt.Errorf("解析 Cron 表达式 '%s' 失败: %w", plan.CronExpression, err)
}
executeAt = next
}
taskLog := &models.TaskExecutionLog{
TaskID: analysisTask.ID,
Status: models.ExecutionStatusWaiting,
}
if err := m.executionLogRepo.CreateTaskExecutionLog(taskLog); err != nil {
return fmt.Errorf("创建任务执行日志失败: %w", err)
}
pendingTask := &models.PendingTask{
TaskID: analysisTask.ID,
ExecuteAt: executeAt,
TaskExecutionLogID: taskLog.ID,
}
if err := m.pendingTaskRepo.CreatePendingTask(pendingTask); err != nil {
return fmt.Errorf("创建待执行任务失败: %w", err)
}
m.logger.Infof("成功为计划 #%d 创建触发器 (任务ID: %d)，执行时间: %v", plan.ID, analysisTask.ID, executeAt)
return nil
}

66
internal/app/service/task/delay_task.go
View File

@@ -1,66 +0,0 @@
package task
import (
"fmt"
"time"
"git.huangwc.com/pig/pig-farm-controller/internal/infra/logs"
"git.huangwc.com/pig/pig-farm-controller/internal/infra/models"
)
type DelayTaskParams struct {
DelayDuration float64 `json:"delay_duration"`
}
// DelayTask 是一个用于模拟延迟的 Task 实现
type DelayTask struct {
executionTask *models.TaskExecutionLog
duration time.Duration
logger *logs.Logger
}
func NewDelayTask(logger *logs.Logger, executionTask *models.TaskExecutionLog) Task {
return &DelayTask{
executionTask: executionTask,
logger: logger,
}
}
// Execute 执行延迟任务，等待指定的时间
func (d *DelayTask) Execute() error {
if err := d.parseParameters(); err != nil {
return err
}
d.logger.Infof("任务 %v: 开始延迟 %v...", d.executionTask.TaskID, d.duration)
time.Sleep(d.duration)
d.logger.Infof("任务 %v: 延迟结束。", d.executionTask.TaskID)
return nil
}
func (d *DelayTask) parseParameters() error {
if d.executionTask.Task.Parameters == nil {
d.logger.Errorf("任务 %v: 缺少参数", d.executionTask.TaskID)
return fmt.Errorf("任务 %v: 参数不全", d.executionTask.TaskID)
}
var params DelayTaskParams
err := d.executionTask.Task.ParseParameters(&params)
if err != nil {
d.logger.Errorf("任务 %v: 解析参数失败: %v", d.executionTask.TaskID, err)
return fmt.Errorf("任务 %v: 解析参数失败: %v", d.executionTask.TaskID, err)
}
if params.DelayDuration <= 0 {
d.logger.Errorf("任务 %v: 参数 delay_duration 缺失或无效 (必须大于0)", d.executionTask.TaskID)
return fmt.Errorf("任务 %v: 参数 delay_duration 缺失或无效 (必须大于0)", d.executionTask.TaskID)
}
d.duration = time.Duration(params.DelayDuration) * time.Second
return nil
}
func (d *DelayTask) OnFailure(executeErr error) {
d.logger.Errorf("任务 %v: 执行失败: %v", d.executionTask.TaskID, executeErr)
}

61
internal/app/service/task/delay_task_test.go
View File

@@ -1,61 +0,0 @@
package task_test
import (
"fmt"
"testing"
"time"
"git.huangwc.com/pig/pig-farm-controller/internal/app/service/task"
)
func TestNewDelayTask(t *testing.T) {
id := "test-delay-task-1"
duration := 100 * time.Millisecond
priority := 1
dt := task.NewDelayTask(id, duration, priority)
if dt.GetID() != id {
t.Errorf("期望任务ID为 %s, 实际为 %s", id, dt.GetID())
}
if dt.GetPriority() != priority {
t.Errorf("期望任务优先级为 %d, 实际为 %d", priority, dt.GetPriority())
}
if dt.IsDone() != false {
t.Error("任务初始状态不应为已完成")
}
// 动态生成的描述，需要匹配 GetDescription 的实现
expectedDesc := fmt.Sprintf("延迟任务，ID: %s，延迟时间: %s", id, duration)
if dt.GetDescription() != expectedDesc {
t.Errorf("期望任务描述为 %s, 实际为 %s", expectedDesc, dt.GetDescription())
}
}
func TestDelayTaskExecute(t *testing.T) {
id := "test-delay-task-execute"
duration := 50 * time.Millisecond // 使用较短的延迟以加快测试速度
priority := 1
dt := task.NewDelayTask(id, duration, priority)
if dt.IsDone() {
t.Error("任务执行前不应为已完成状态")
}
startTime := time.Now()
err := dt.Execute()
endTime := time.Now()
if err != nil {
t.Errorf("Execute 方法返回错误: %v", err)
}
if !dt.IsDone() {
t.Error("任务执行后应为已完成状态")
}
// 验证延迟时间大致正确，允许一些误差
elapsed := endTime.Sub(startTime)
if elapsed < duration || elapsed > duration*2 {
t.Errorf("期望执行时间在 %v 左右, 但实际耗时 %v", duration, elapsed)
}
}

167
internal/app/service/task/release_feed_weight_task.go
View File

@@ -1,167 +0,0 @@
package task
import (
"encoding/json"
"fmt"
"time"
"git.huangwc.com/pig/pig-farm-controller/internal/app/service/device"
"git.huangwc.com/pig/pig-farm-controller/internal/infra/logs"
"git.huangwc.com/pig/pig-farm-controller/internal/infra/models"
"git.huangwc.com/pig/pig-farm-controller/internal/infra/repository"
)
// ReleaseFeedWeightTaskParams 定义了 ReleaseFeedWeightTask 的参数结构
type ReleaseFeedWeightTaskParams struct {
ReleaseWeight float64 `json:"release_weight"` // 需要释放的重量
FeedPortDeviceID uint `json:"feed_port_device_id"` // 下料口ID
MixingTankDeviceID uint `json:"mixing_tank_device_id"` // 称重传感器ID
}
// ReleaseFeedWeightTask 是一个控制下料口释放指定重量的任务
type ReleaseFeedWeightTask struct {
deviceRepo repository.DeviceRepository
sensorDataRepo repository.SensorDataRepository
claimedLog *models.TaskExecutionLog
feedPortDevice *models.Device
releaseWeight float64
mixingTankDeviceID uint
feedPort device.Service
logger *logs.Logger
}
// NewReleaseFeedWeightTask 创建一个新的 ReleaseFeedWeightTask 实例
func NewReleaseFeedWeightTask(
claimedLog *models.TaskExecutionLog,
sensorDataRepo repository.SensorDataRepository,
deviceRepo repository.DeviceRepository,
deviceService device.Service,
logger *logs.Logger,
) Task {
return &ReleaseFeedWeightTask{
claimedLog: claimedLog,
deviceRepo: deviceRepo,
sensorDataRepo: sensorDataRepo,
feedPort: deviceService, // 直接注入
logger: logger,
}
}
func (r *ReleaseFeedWeightTask) Execute() error {
r.logger.Infof("任务 %v: 开始执行, 日志ID: %v", r.claimedLog.TaskID, r.claimedLog.ID)
if err := r.parseParameters(); err != nil {
return err
}
weight, err := r.getNowWeight()
if err != nil {
return err
}
if err = r.feedPort.Switch(r.feedPortDevice, device.DeviceActionStart); err != nil {
r.logger.Errorf("启动下料口(id=%v)失败: %v , 日志ID: %v", r.feedPortDevice.ID, err, r.claimedLog.ID)
return err
}
targetWeight := weight - r.releaseWeight
errCount := 1
// TODO 这个判断有延迟, 尤其是LoRa通信本身延迟较高, 可以考虑根据信号质量或其他指标提前发送停止命令
for targetWeight <= weight {
weight, err = r.getNowWeight()
if err != nil {
errCount++
if errCount > 3 { // 如果连续三次没成功采集到重量数据，则认为计划执行失败
r.logger.Errorf("获取当前计划执行日志(id=%v)的当前搅拌罐重量失败: %v, 任务结束", r.claimedLog.ID, err)
return err
}
r.logger.Warnf("第%v次尝试获取当前计划执行日志(id=%v)的当前搅拌罐重量失败: %v", errCount, r.claimedLog.ID, err)
continue
}
time.Sleep(100 * time.Millisecond)
}
if err = r.feedPort.Switch(r.feedPortDevice, device.DeviceActionStop); err != nil {
r.logger.Errorf("关闭下料口(id=%v)失败: %v , 日志ID: %v", r.feedPortDevice.ID, err, r.claimedLog.ID)
return err
}
r.logger.Infof("完成计划执行日志(id=%v)的当前计划, 完成下料 %vkg, 搅拌罐剩余重量 %vkg", r.claimedLog.ID, r.releaseWeight, weight)
return nil
}
// 获取当前搅拌罐重量
func (r *ReleaseFeedWeightTask) getNowWeight() (float64, error) {
sensorData, err := r.sensorDataRepo.GetLatestSensorDataByDeviceIDAndSensorType(r.mixingTankDeviceID, models.SensorTypeWeight)
if err != nil {
r.logger.Errorf("获取设备 %v 最新传感器数据失败: %v , 日志ID: %v", r.mixingTankDeviceID, err, r.claimedLog.ID)
return 0, err
}
if sensorData == nil {
return 0, fmt.Errorf("未找到设备 %v 的最新重量传感器数据", r.mixingTankDeviceID)
}
wg := &models.WeightData{}
err = json.Unmarshal(sensorData.Data, wg)
if err != nil {
r.logger.Errorf("反序列化设备 %v 最新传感器数据失败: %v , 日志ID: %v", r.mixingTankDeviceID, err, r.claimedLog.ID)
return 0, err
}
return wg.WeightKilograms, nil
}
func (r *ReleaseFeedWeightTask) parseParameters() error {
if r.claimedLog.Task.Parameters == nil {
r.logger.Errorf("任务 %v: 缺少参数", r.claimedLog.TaskID)
return fmt.Errorf("任务 %v: 参数不全", r.claimedLog.TaskID)
}
var params ReleaseFeedWeightTaskParams
err := r.claimedLog.Task.ParseParameters(&params)
if err != nil {
r.logger.Errorf("任务 %v: 解析参数失败: %v", r.claimedLog.TaskID, err)
return fmt.Errorf("任务 %v: 解析参数失败: %v", r.claimedLog.TaskID, err)
}
// 校验参数是否存在
if params.ReleaseWeight == 0 {
r.logger.Errorf("任务 %v: 参数 release_weight 缺失或无效", r.claimedLog.TaskID)
return fmt.Errorf("任务 %v: 参数 release_weight 缺失或无效", r.claimedLog.TaskID)
}
if params.FeedPortDeviceID == 0 {
r.logger.Errorf("任务 %v: 参数 feed_port_device_id 缺失或无效", r.claimedLog.TaskID)
return fmt.Errorf("任务 %v: 参数 feed_port_device_id 缺失或无效", r.claimedLog.TaskID)
}
if params.MixingTankDeviceID == 0 {
r.logger.Errorf("任务 %v: 参数 mixing_tank_device_id 缺失或无效", r.claimedLog.TaskID)
return fmt.Errorf("任务 %v: 参数 mixing_tank_device_id 缺失或无效", r.claimedLog.TaskID)
}
r.releaseWeight = params.ReleaseWeight
r.mixingTankDeviceID = params.MixingTankDeviceID
r.feedPortDevice, err = r.deviceRepo.FindByID(params.FeedPortDeviceID)
if err != nil {
r.logger.Errorf("任务 %v: 获取设备信息失败: %v", r.claimedLog.TaskID, err)
return fmt.Errorf("任务 %v: 获取设备信息失败: %v", r.claimedLog.TaskID, err)
}
return nil
}
func (r *ReleaseFeedWeightTask) OnFailure(executeErr error) {
r.logger.Errorf("开始善后处理, 日志ID:%v", r.claimedLog.ID)
if r.feedPort != nil {
err := r.feedPort.Switch(r.feedPortDevice, device.DeviceActionStop)
if err != nil {
r.logger.Errorf("[严重] 下料口停止失败, 日志ID: %v, 错误: %v", r.claimedLog.ID, err)
}
} else {
r.logger.Warnf("[警告] 下料口通信器尚未初始化, 不进行任何操作, 日志ID: %v", r.claimedLog.ID)
}
r.logger.Errorf("善后处理完成, 日志ID:%v", r.claimedLog.ID)
}

462
internal/app/service/task/scheduler.go
View File

@@ -1,462 +0,0 @@
package task
import (
"errors"
"sync"
"time"
"git.huangwc.com/pig/pig-farm-controller/internal/app/service/device"
"git.huangwc.com/pig/pig-farm-controller/internal/infra/logs"
"git.huangwc.com/pig/pig-farm-controller/internal/infra/models"
"git.huangwc.com/pig/pig-farm-controller/internal/infra/repository"
"github.com/panjf2000/ants/v2"
"gorm.io/gorm"
)
// ProgressTracker 仅用于在内存中提供计划执行的并发锁
type ProgressTracker struct {
mu sync.Mutex
cond *sync.Cond // 用于实现阻塞锁
runningPlans map[uint]bool // key: planExecutionLogID, value: true (用作内存锁)
}
// NewProgressTracker 创建一个新的进度跟踪器
func NewProgressTracker() *ProgressTracker {
t := &ProgressTracker{
runningPlans: make(map[uint]bool),
}
t.cond = sync.NewCond(&t.mu)
return t
}
// TryLock (非阻塞) 尝试锁定一个计划。如果计划未被锁定，则锁定并返回 true。
func (t *ProgressTracker) TryLock(planLogID uint) bool {
t.mu.Lock()
defer t.mu.Unlock()
if t.runningPlans[planLogID] {
return false // 已被锁定
}
t.runningPlans[planLogID] = true
return true
}
// Lock (阻塞) 获取一个计划的执行锁。如果锁已被占用，则会一直等待直到锁被释放。
func (t *ProgressTracker) Lock(planLogID uint) {
t.mu.Lock()
// 当计划正在运行时，调用 t.cond.Wait() 会原子地解锁 mu 并挂起当前协程。
// 当被唤醒时，它会重新锁定 mu 并再次检查循环条件。
for t.runningPlans[planLogID] {
t.cond.Wait()
}
// 获取到锁
t.runningPlans[planLogID] = true
t.mu.Unlock()
}
// Unlock 解锁一个计划，并唤醒所有正在等待此锁的协程。
func (t *ProgressTracker) Unlock(planLogID uint) {
t.mu.Lock()
defer t.mu.Unlock()
delete(t.runningPlans, planLogID)
// 唤醒所有在此条件上等待的协程
t.cond.Broadcast()
}
// GetRunningPlanIDs 获取当前所有正在执行的计划ID列表
func (t *ProgressTracker) GetRunningPlanIDs() []uint {
t.mu.Lock()
defer t.mu.Unlock()
ids := make([]uint, 0, len(t.runningPlans))
for id := range t.runningPlans {
ids = append(ids, id)
}
return ids
}
// Scheduler 是核心的、持久化的任务调度器
type Scheduler struct {
logger *logs.Logger
pollingInterval time.Duration
workers int
pendingTaskRepo repository.PendingTaskRepository
executionLogRepo repository.ExecutionLogRepository
deviceRepo repository.DeviceRepository
sensorDataRepo repository.SensorDataRepository
planRepo repository.PlanRepository
analysisPlanTaskManager *AnalysisPlanTaskManager
progressTracker *ProgressTracker
deviceService device.Service
pool *ants.Pool // 使用 ants 协程池来管理并发
wg sync.WaitGroup
stopChan chan struct{} // 用于停止主循环的信号通道
}
// NewScheduler 创建一个新的调度器实例
func NewScheduler(
pendingTaskRepo repository.PendingTaskRepository,
executionLogRepo repository.ExecutionLogRepository,
deviceRepo repository.DeviceRepository,
sensorDataRepo repository.SensorDataRepository,
planRepo repository.PlanRepository,
analysisPlanTaskManager *AnalysisPlanTaskManager,
logger *logs.Logger,
deviceService device.Service,
interval time.Duration,
numWorkers int,
) *Scheduler {
return &Scheduler{
pendingTaskRepo: pendingTaskRepo,
executionLogRepo: executionLogRepo,
deviceRepo: deviceRepo,
sensorDataRepo: sensorDataRepo,
planRepo: planRepo,
analysisPlanTaskManager: analysisPlanTaskManager,
logger: logger,
deviceService: deviceService,
pollingInterval: interval,
workers: numWorkers,
progressTracker: NewProgressTracker(),
stopChan: make(chan struct{}), // 初始化停止信号通道
}
}
// Start 启动调度器，包括初始化协程池和启动主轮询循环
func (s *Scheduler) Start() {
s.logger.Warnf("任务调度器正在启动，工作协程数: %d...", s.workers)
pool, err := ants.NewPool(s.workers, ants.WithPanicHandler(func(err interface{}) {
s.logger.Errorf("[严重] 任务执行时发生 panic: %v", err)
}))
if err != nil {
panic("初始化协程池失败: " + err.Error())
}
s.pool = pool
s.wg.Add(1)
go s.run()
s.logger.Warnf("任务调度器已成功启动")
}
// Stop 优雅地停止调度器
func (s *Scheduler) Stop() {
s.logger.Warnf("正在停止任务调度器...")
close(s.stopChan) // 1. 发出停止信号，停止主循环
s.wg.Wait() // 2. 等待主循环完成
s.pool.Release() // 3. 释放 ants 池 (等待所有已提交的任务执行完毕)
s.logger.Warnf("任务调度器已安全停止")
}
// run 是主轮询循环，负责从数据库认领任务并提交到协程池
func (s *Scheduler) run() {
defer s.wg.Done()
ticker := time.NewTicker(s.pollingInterval)
defer ticker.Stop()
for {
select {
case <-s.stopChan:
// 收到停止信号，退出循环
return
case <-ticker.C:
// 定时触发任务认领和提交
go s.claimAndSubmit()
}
}
}
// claimAndSubmit 实现了最终的“认领-锁定-执行 或 等待-放回”的健壮逻辑
func (s *Scheduler) claimAndSubmit() {
runningPlanIDs := s.progressTracker.GetRunningPlanIDs()
claimedLog, pendingTask, err := s.pendingTaskRepo.ClaimNextAvailableTask(runningPlanIDs)
if err != nil {
if !errors.Is(err, gorm.ErrRecordNotFound) {
s.logger.Errorf("认领任务时发生错误: %v", err)
}
// gorm.ErrRecordNotFound 说明没任务要执行
return
}
// 尝试获取内存执行锁
if s.progressTracker.TryLock(claimedLog.PlanExecutionLogID) {
// 成功获取锁，正常派发任务
err = s.pool.Submit(func() {
defer s.progressTracker.Unlock(claimedLog.PlanExecutionLogID)
s.processTask(claimedLog)
})
if err != nil {
s.logger.Errorf("向协程池提交任务失败: %v", err)
// 提交失败，必须释放刚刚获取的锁
s.progressTracker.Unlock(claimedLog.PlanExecutionLogID)
// 同样需要将任务安全放回
s.handleRequeue(claimedLog.PlanExecutionLogID, pendingTask)
}
} else {
// 获取锁失败，说明有“兄弟”任务正在执行。执行“锁定并安全放回”逻辑。
s.handleRequeue(claimedLog.PlanExecutionLogID, pendingTask)
}
}
// handleRequeue 同步地、安全地将一个无法立即执行的任务放回队列。
func (s *Scheduler) handleRequeue(planExecutionLogID uint, taskToRequeue *models.PendingTask) {
s.logger.Warnf("计划 %d 正在执行，任务 %d (TaskID: %d) 将等待并重新入队...", planExecutionLogID, taskToRequeue.ID, taskToRequeue.TaskID)
// 1. 阻塞式地等待，直到可以获取到该计划的锁。
s.progressTracker.Lock(planExecutionLogID)
defer s.progressTracker.Unlock(planExecutionLogID)
// 2. 在持有锁的情况下，将任务安全地放回队列。
if err := s.pendingTaskRepo.RequeueTask(taskToRequeue); err != nil {
s.logger.Errorf("[严重] 任务重新入队失败, 原始PendingTaskID: %d, 错误: %v", taskToRequeue.ID, err)
return
}
s.logger.Warnf("任务 (原始ID: %d) 已成功重新入队，并已释放计划 %d 的锁。", taskToRequeue.ID, planExecutionLogID)
}
// processTask 处理单个任务的逻辑
func (s *Scheduler) processTask(claimedLog *models.TaskExecutionLog) {
s.logger.Warnf("开始处理任务, 日志ID: %d, 任务ID: %d, 任务名称: %s, 描述: %s",
claimedLog.ID, claimedLog.TaskID, claimedLog.Task.Name, claimedLog.Task.Description)
claimedLog.StartedAt = time.Now()
claimedLog.Status = models.ExecutionStatusCompleted // 先乐观假定任务成功, 后续失败了再改
defer s.updateTaskExecutionLogStatus(claimedLog)
// 执行任务
err := s.runTask(claimedLog)
if err != nil {
claimedLog.Status = models.ExecutionStatusFailed
claimedLog.Output = err.Error()
// 任务失败时，调用统一的终止服务
s.handlePlanTermination(claimedLog.PlanExecutionLogID, "子任务执行失败: "+err.Error())
return
}
// 如果是计划分析任务，它的职责是解析和分发任务，到此即完成，不参与后续的计划完成度检查。
if claimedLog.Task.Type == models.TaskPlanAnalysis {
s.logger.Warnf("完成计划分析任务, 日志ID: %d", claimedLog.ID)
return
}
// --- 以下是常规任务的完成逻辑 ---
s.logger.Warnf("完成任务, 日志ID: %d", claimedLog.ID)
// 检查是否是最后一个任务
incompleteCount, err := s.executionLogRepo.CountIncompleteTasksByPlanLogID(claimedLog.PlanExecutionLogID)
if err != nil {
s.logger.Errorf("检查计划 %d 的未完成任务数时出错: %v", claimedLog.PlanExecutionLogID, err)
return
}
// 如果此计划执行中，未完成的任务只剩下当前这一个（因为当前任务的状态此时在数据库中仍为 'started'），
// 则认为整个计划已完成。
if incompleteCount == 1 {
s.handlePlanCompletion(claimedLog.PlanExecutionLogID)
}
}
// runTask 用于执行具体任务
func (s *Scheduler) runTask(claimedLog *models.TaskExecutionLog) error {
// 这是个特殊任务, 用于解析Plan并将解析出的任务队列添加到待执行队列中
if claimedLog.Task.Type == models.TaskPlanAnalysis {
// 解析plan
err := s.analysisPlan(claimedLog)
if err != nil {
// TODO 这里要处理一下, 比如再插一个新的触发器回去
s.logger.Errorf("[严重] 计划解析失败, 日志ID: %d, 错误: %v", claimedLog.ID, err)
return err
}
} else {
// 执行普通任务
task := s.taskFactory(claimedLog)
if err := task.Execute(); err != nil {
s.logger.Errorf("[严重] 任务执行失败, 日志ID: %d, 错误: %v", claimedLog.ID, err)
task.OnFailure(err)
return err
}
}
return nil
}
// taskFactory 会根据任务类型初始化对应任务
func (s *Scheduler) taskFactory(claimedLog *models.TaskExecutionLog) Task {
switch claimedLog.Task.Type {
case models.TaskTypeWaiting:
return NewDelayTask(s.logger, claimedLog)
case models.TaskTypeReleaseFeedWeight:
return NewReleaseFeedWeightTask(claimedLog, s.sensorDataRepo, s.deviceRepo, s.deviceService, s.logger)
default:
// TODO 这里直接panic合适吗? 不过这个场景确实不该出现任何异常的任务类型
panic("不支持的任务类型")
}
}
// analysisPlan 解析Plan并将解析出的Task列表插入待执行队列中
func (s *Scheduler) analysisPlan(claimedLog *models.TaskExecutionLog) error {
// 创建Plan执行记录
// 从任务的 Parameters 中解析出真实的 PlanID
var params struct {
PlanID uint `json:"plan_id"`
}
if err := claimedLog.Task.ParseParameters(&params); err != nil {
s.logger.Errorf("解析任务参数中的计划ID失败，日志ID: %d, 错误: %v", claimedLog.ID, err)
return err
}
realPlanID := params.PlanID
planLog := &models.PlanExecutionLog{
PlanID: realPlanID, // 使用从参数中解析出的真实 PlanID
Status: models.ExecutionStatusStarted,
StartedAt: time.Now(),
}
if err := s.executionLogRepo.CreatePlanExecutionLog(planLog); err != nil {
s.logger.Errorf("[严重] 创建计划执行日志失败, 日志ID: %d, 错误: %v", claimedLog.ID, err)
return err
}
// 解析出Task列表
tasks, err := s.planRepo.FlattenPlanTasks(realPlanID)
if err != nil {
s.logger.Errorf("[严重] 解析计划失败, 日志ID: %d, 错误: %v", claimedLog.ID, err)
return err
}
// 写入执行历史
taskLogs := make([]*models.TaskExecutionLog, len(tasks))
for i, task := range tasks {
taskLogs[i] = &models.TaskExecutionLog{
PlanExecutionLogID: planLog.ID,
TaskID: task.ID,
Status: models.ExecutionStatusWaiting,
}
}
err = s.executionLogRepo.CreateTaskExecutionLogsInBatch(taskLogs)
if err != nil {
s.logger.Errorf("[严重] 写入执行历史, 日志ID: %d, 错误: %v", claimedLog.ID, err)
return err
}
// 写入待执行队列
pendingTasks := make([]*models.PendingTask, len(tasks))
for i, task := range tasks {
pendingTasks[i] = &models.PendingTask{
TaskID: task.ID,
TaskExecutionLogID: taskLogs[i].ID, // 使用正确的 TaskExecutionLogID
// 待执行队列是通过任务触发时间排序的, 且只要在调度器获取的时间点之前的都可以被触发
ExecuteAt: time.Now().Add(time.Duration(i) * time.Second),
}
}
err = s.pendingTaskRepo.CreatePendingTasksInBatch(pendingTasks)
if err != nil {
s.logger.Errorf("[严重] 写入待执行队列, 日志ID: %d, 错误: %v", claimedLog.ID, err)
return err
}
// --- 处理空计划的边缘情况 ---
// 如果一个计划被解析后，发现其任务列表为空，
// 那么它实际上已经“执行”完毕了，我们需要在这里手动为它创建下一次的触发器。
if len(tasks) == 0 {
s.handlePlanCompletion(planLog.ID)
}
return nil
}
// updateTaskExecutionLogStatus 修改任务历史中的执行状态
func (s *Scheduler) updateTaskExecutionLogStatus(claimedLog *models.TaskExecutionLog) error {
claimedLog.EndedAt = time.Now()
if err := s.executionLogRepo.UpdateTaskExecutionLog(claimedLog); err != nil {
s.logger.Errorf("[严重] 更新任务执行日志失败, 日志ID: %d, 错误: %v", claimedLog.ID, err)
return err
}
return nil
}
// handlePlanTermination 集中处理计划的终止逻辑（失败或取消）
func (s *Scheduler) handlePlanTermination(planLogID uint, reason string) {
// 1. 从待执行队列中删除所有相关的子任务
if err := s.pendingTaskRepo.DeletePendingTasksByPlanLogID(planLogID); err != nil {
s.logger.Errorf("从待执行队列中删除计划 %d 的后续任务时出错: %v", planLogID, err)
}
// 2. 将父计划的执行日志标记为失败
if err := s.executionLogRepo.FailPlanExecution(planLogID, reason); err != nil {
s.logger.Errorf("标记计划执行日志 %d 为失败时出错: %v", planLogID, err)
}
// 3. 将所有未完成的子任务日志标记为已取消
if err := s.executionLogRepo.CancelIncompleteTasksByPlanLogID(planLogID, "父计划失败或被取消"); err != nil {
s.logger.Errorf("取消计划 %d 的后续任务日志时出错: %v", planLogID, err)
}
// 4. 将计划本身的状态更新为失败
planLog, err := s.executionLogRepo.FindPlanExecutionLogByID(planLogID)
if err != nil {
s.logger.Errorf("无法找到计划执行日志 %d 以更新父计划状态: %v", planLogID, err)
return
}
if err := s.planRepo.UpdatePlanStatus(planLog.PlanID, models.PlanStatusFailed); err != nil {
s.logger.Errorf("更新计划 %d 状态为 '失败' 时出错: %v", planLog.PlanID, err)
}
}
// handlePlanCompletion 集中处理计划成功完成后的所有逻辑
func (s *Scheduler) handlePlanCompletion(planLogID uint) {
s.logger.Infof("计划执行 %d 的所有任务已完成，开始处理计划完成逻辑...", planLogID)
// 1. 通过 PlanExecutionLog 反查正确的顶层 PlanID
planExecutionLog, err := s.executionLogRepo.FindPlanExecutionLogByID(planLogID)
if err != nil {
s.logger.Errorf("获取计划执行日志 %d 失败: %v", planLogID, err)
return
}
topLevelPlanID := planExecutionLog.PlanID // 这才是正确的顶层计划ID
// 2. 获取计划的最新数据，这里我们只需要基本信息来判断执行类型和次数
plan, err := s.planRepo.GetBasicPlanByID(topLevelPlanID)
if err != nil {
s.logger.Errorf("获取计划 %d 的基本信息失败: %v", topLevelPlanID, err)
return
}
// 3. 在内存中计算新的计数值和状态
newExecuteCount := plan.ExecuteCount + 1
newStatus := plan.Status // 默认为当前状态
// 如果是自动计划且达到执行次数上限，或计划是手动类型，则更新计划状态为已停止
if (plan.ExecutionType == models.PlanExecutionTypeAutomatic && plan.ExecuteNum > 0 && newExecuteCount >= plan.ExecuteNum) || plan.ExecutionType == models.PlanExecutionTypeManual {
newStatus = models.PlanStatusStopped
s.logger.Infof("计划 %d 已完成执行，状态更新为 '执行完毕'。", topLevelPlanID)
}
// 4. 使用专门的方法来原子性地更新计数值和状态
if err := s.planRepo.UpdatePlanStateAfterExecution(topLevelPlanID, newExecuteCount, newStatus); err != nil {
s.logger.Errorf("更新计划 %d 的执行后状态失败: %v", topLevelPlanID, err)
return
}
// 5. 更新计划执行日志状态为完成
if err := s.executionLogRepo.UpdatePlanExecutionLogStatus(planLogID, models.ExecutionStatusCompleted); err != nil {
s.logger.Errorf("更新计划执行日志 %d 状态为 '完成' 失败: %v", planLogID, err)
}
// 6. 调用共享的 Manager 来处理触发器更新逻辑
// 只有当计划在本次执行后仍然是 Enabled 状态时，才需要创建下一次的触发器。
if newStatus == models.PlanStatusEnabled {
if err := s.analysisPlanTaskManager.CreateOrUpdateTrigger(topLevelPlanID); err != nil {
s.logger.Errorf("为计划 %d 创建/更新触发器失败: %v", topLevelPlanID, err)
}
} else {
s.logger.Infof("计划 %d 状态为 '%d'，无需创建下一次触发器。", topLevelPlanID, newStatus)
}
}

30
internal/app/service/task/task.go
View File

@@ -1,30 +0,0 @@
package task
import (
"git.huangwc.com/pig/pig-farm-controller/internal/infra/models"
)
// Task 定义了所有可被调度器执行的任务必须实现的接口。
type Task interface {
// Execute 是任务的核心执行逻辑。
// ctx: 用于控制任务的超时或取消。
// log: 包含了当前任务执行的完整上下文信息，包括从数据库中加载的任务参数等。
// 返回的 error 表示任务是否执行成功。调度器会根据返回的 error 是否为 nil 来决定任务状态。
Execute() error
// OnFailure 定义了当 Execute 方法返回错误时，需要执行的回滚或清理逻辑。
// log: 任务执行的上下文。
// executeErr: 从 Execute 方法返回的原始错误。
OnFailure(executeErr error)
}
// TaskFactory 是一个任务组装工厂, 可以根据Task类型获取到对应的初始化函数
var TaskFactory = func(tt models.TaskType) Task {
switch tt {
case models.TaskTypeWaiting:
return &DelayTask{}
default:
// 出现位置任务类型说明业务逻辑出现重大问题, 一个异常任务被创建了出来
panic("发现未知任务类型")
}
}

68
internal/app/service/token/token_service.go
View File

@@ -1,68 +0,0 @@
package token
import (
"fmt"
"time"
"github.com/golang-jwt/jwt/v5"
)
// Claims 定义了 JWT 的声明结构
type Claims struct {
UserID uint `json:"user_id"`
jwt.RegisteredClaims
}
// TokenService 定义了 token 操作的接口
type TokenService interface {
GenerateToken(userID uint) (string, error)
ParseToken(tokenString string) (*Claims, error)
}
// tokenService 是 TokenService 接口的实现
type tokenService struct {
secret []byte
}
// NewTokenService 创建并返回一个新的 TokenService 实例
func NewTokenService(secret []byte) TokenService {
return &tokenService{secret: secret}
}
// GenerateToken 生成一个新的 JWT token
func (s *tokenService) GenerateToken(userID uint) (string, error) {
nowTime := time.Now()
expireTime := nowTime.Add(24 * time.Hour) // Token 有效期为 24 小时
claims := Claims{
UserID: userID,
RegisteredClaims: jwt.RegisteredClaims{
ExpiresAt: jwt.NewNumericDate(expireTime),
Issuer: "pig-farm-controller",
},
}
tokenClaims := jwt.NewWithClaims(jwt.SigningMethodHS256, claims)
token, err := tokenClaims.SignedString(s.secret)
return token, err
}
// ParseToken 解析并验证 JWT token
func (s *tokenService) ParseToken(tokenString string) (*Claims, error) {
token, err := jwt.ParseWithClaims(tokenString, &Claims{}, func(token *jwt.Token) (interface{}, error) {
return s.secret, nil
})
// 优先检查解析过程中是否发生错误
if err != nil {
return nil, err
}
// 只有当 token 对象有效时，才尝试获取 Claims 并验证
if claims, ok := token.Claims.(*Claims); ok && token.Valid {
return claims, nil
}
// 如果 token 无效（例如，过期但没有返回错误，或者 Claims 类型不匹配），则返回一个通用错误
return nil, fmt.Errorf("token is invalid")
}

107
internal/app/service/token/token_service_test.go
View File

@@ -1,107 +0,0 @@
package token_test
import (
"errors"
"testing"
"time"
"git.huangwc.com/pig/pig-farm-controller/internal/app/service/token"
"github.com/golang-jwt/jwt/v5"
)
func TestGenerateToken(t *testing.T) {
// 使用一个测试密钥初始化 TokenService
testSecret := []byte("test_secret_key")
service := token.NewTokenService(testSecret)
userID := uint(123)
tokenString, err := service.GenerateToken(userID)
if err != nil {
t.Fatalf("生成令牌失败: %v", err)
}
if tokenString == "" {
t.Fatal("生成的令牌字符串为空")
}
// 解析 token 以确保其有效性及声明
claims, err := service.ParseToken(tokenString)
if err != nil {
t.Fatalf("生成后解析令牌失败: %v", err)
}
if claims.UserID != userID {
t.Errorf("期望用户ID %d, 实际为 %d", userID, claims.UserID)
}
// 检查 token 是否未过期 (在合理范围内)
if claims.ExpiresAt == nil || claims.ExpiresAt.Time.Before(time.Now().Add(-time.Minute)) {
t.Errorf("令牌过期时间无效或已过期")
}
if claims.Issuer != "pig-farm-controller" {
t.Errorf("期望签发者 \"pig-farm-controller\", 实际为 \"%s\"", claims.Issuer)
}
}
func TestParseToken(t *testing.T) {
// 使用两个不同的测试密钥
correctSecret := []byte("the_correct_secret")
wrongSecret := []byte("a_very_wrong_secret")
serviceWithCorrectKey := token.NewTokenService(correctSecret)
serviceWithWrongKey := token.NewTokenService(wrongSecret)
userID := uint(456)
// 1. 生成一个有效的 token
validToken, err := serviceWithCorrectKey.GenerateToken(userID)
if err != nil {
t.Fatalf("为解析测试生成有效令牌失败: %v", err)
}
// 测试用例 1: 使用正确的密钥成功解析
claims, err := serviceWithCorrectKey.ParseToken(validToken)
if err != nil {
t.Errorf("使用正确密钥解析有效令牌失败: %v", err)
}
if claims.UserID != userID {
t.Errorf("解析有效令牌时期望用户ID %d, 实际为 %d", userID, claims.UserID)
}
// 测试用例 2: 无效 token (例如, 格式错误的字符串)
invalidTokenString := "this.is.not.a.valid.jwt"
_, err = serviceWithCorrectKey.ParseToken(invalidTokenString)
if err == nil {
t.Error("解析格式错误的令牌意外成功")
}
// 测试用C:\Users\divano\Desktop\work\AA-Pig\pig-farm-controller\internal\infra\repository\plan_repository_test.go例 3: 过期 token
expiredClaims := token.Claims{
UserID: userID,
RegisteredClaims: jwt.RegisteredClaims{
ExpiresAt: jwt.NewNumericDate(time.Now().Add(-time.Hour)), // 1 小时前
Issuer: "pig-farm-controller",
},
}
expiredTokenClaims := jwt.NewWithClaims(jwt.SigningMethodHS256, expiredClaims)
expiredTokenString, err := expiredTokenClaims.SignedString(correctSecret)
if err != nil {
t.Fatalf("生成过期令牌失败: %v", err)
}
_, err = serviceWithCorrectKey.ParseToken(expiredTokenString)
if err == nil {
t.Error("解析过期令牌意外成功")
}
// 新增测试用例 4: 使用错误的密钥解析
_, err = serviceWithWrongKey.ParseToken(validToken)
if err == nil {
t.Error("使用错误密钥解析令牌意外成功")
}
// 我们可以更精确地检查错误类型，以确保它是签名错误
if !errors.Is(err, jwt.ErrTokenSignatureInvalid) {
t.Errorf("期望得到签名无效错误 (ErrTokenSignatureInvalid)，但得到了: %v", err)
}
}

4
internal/app/service/transport/chirp_stack.go → internal/app/webhook/chirp_stack.go
View File

@@ -1,4 +1,4 @@
package transport
package webhook
import (
"encoding/base64"
@@ -7,7 +7,7 @@ import (
"net/http"
"time"
"git.huangwc.com/pig/pig-farm-controller/internal/app/service/device/proto"
"git.huangwc.com/pig/pig-farm-controller/internal/domain/device/proto"
"git.huangwc.com/pig/pig-farm-controller/internal/infra/logs"
"git.huangwc.com/pig/pig-farm-controller/internal/infra/models"
"git.huangwc.com/pig/pig-farm-controller/internal/infra/repository"

2
internal/app/service/transport/chirp_stack_types.go → internal/app/webhook/chirp_stack_types.go
View File

@@ -1,4 +1,4 @@
package transport
package webhook
import (
"encoding/json"

2
internal/app/service/transport/transport.go → internal/app/webhook/transport.go
View File

@@ -1,4 +1,4 @@
package transport
package webhook
import "net/http"