360 lines
14 KiB
Go
360 lines
14 KiB
Go
package core
|
||
|
||
import (
|
||
"fmt"
|
||
"time"
|
||
|
||
"git.huangwc.com/pig/pig-farm-controller/internal/app/service"
|
||
"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/device"
|
||
domain_notify "git.huangwc.com/pig/pig-farm-controller/internal/domain/notify"
|
||
"git.huangwc.com/pig/pig-farm-controller/internal/domain/pig"
|
||
"git.huangwc.com/pig/pig-farm-controller/internal/domain/scheduler"
|
||
"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/database"
|
||
"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/notify"
|
||
"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/transport/lora"
|
||
"gorm.io/gorm"
|
||
)
|
||
|
||
// Infrastructure 聚合了所有基础设施层的组件。
|
||
type Infrastructure struct {
|
||
Storage database.Storage
|
||
Repos *Repositories
|
||
Lora *LoraComponents
|
||
NotifyService domain_notify.Service
|
||
TokenService token.Service
|
||
}
|
||
|
||
// initInfrastructure 初始化所有基础设施层组件。
|
||
func initInfrastructure(cfg *config.Config, logger *logs.Logger) (*Infrastructure, error) {
|
||
storage, err := initStorage(cfg.Database, logger)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
repos := initRepositories(storage.GetDB(), logger)
|
||
|
||
lora, err := initLora(cfg, logger, repos)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
notifyService, err := initNotifyService(cfg.Notify, logger, repos.UserRepo, repos.NotificationRepo)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("初始化通知服务失败: %w", err)
|
||
}
|
||
|
||
tokenService := token.NewTokenService([]byte(cfg.App.JWTSecret))
|
||
|
||
return &Infrastructure{
|
||
Storage: storage,
|
||
Repos: repos,
|
||
Lora: lora,
|
||
NotifyService: notifyService,
|
||
TokenService: tokenService,
|
||
}, nil
|
||
}
|
||
|
||
// Repositories 聚合了所有的仓库实例。
|
||
type Repositories struct {
|
||
UserRepo repository.UserRepository
|
||
DeviceRepo repository.DeviceRepository
|
||
AreaControllerRepo repository.AreaControllerRepository
|
||
DeviceTemplateRepo repository.DeviceTemplateRepository
|
||
PlanRepo repository.PlanRepository
|
||
PendingTaskRepo repository.PendingTaskRepository
|
||
ExecutionLogRepo repository.ExecutionLogRepository
|
||
SensorDataRepo repository.SensorDataRepository
|
||
DeviceCommandLogRepo repository.DeviceCommandLogRepository
|
||
PendingCollectionRepo repository.PendingCollectionRepository
|
||
UserActionLogRepo repository.UserActionLogRepository
|
||
PigBatchRepo repository.PigBatchRepository
|
||
PigBatchLogRepo repository.PigBatchLogRepository
|
||
PigFarmRepo repository.PigFarmRepository
|
||
PigPenRepo repository.PigPenRepository
|
||
PigTransferLogRepo repository.PigTransferLogRepository
|
||
PigTradeRepo repository.PigTradeRepository
|
||
PigSickPigLogRepo repository.PigSickLogRepository
|
||
MedicationLogRepo repository.MedicationLogRepository
|
||
RawMaterialRepo repository.RawMaterialRepository
|
||
NotificationRepo repository.NotificationRepository
|
||
UnitOfWork repository.UnitOfWork
|
||
}
|
||
|
||
// initRepositories 初始化所有的仓库。
|
||
func initRepositories(db *gorm.DB, logger *logs.Logger) *Repositories {
|
||
return &Repositories{
|
||
UserRepo: repository.NewGormUserRepository(db),
|
||
DeviceRepo: repository.NewGormDeviceRepository(db),
|
||
AreaControllerRepo: repository.NewGormAreaControllerRepository(db),
|
||
DeviceTemplateRepo: repository.NewGormDeviceTemplateRepository(db),
|
||
PlanRepo: repository.NewGormPlanRepository(db),
|
||
PendingTaskRepo: repository.NewGormPendingTaskRepository(db),
|
||
ExecutionLogRepo: repository.NewGormExecutionLogRepository(db),
|
||
SensorDataRepo: repository.NewGormSensorDataRepository(db),
|
||
DeviceCommandLogRepo: repository.NewGormDeviceCommandLogRepository(db),
|
||
PendingCollectionRepo: repository.NewGormPendingCollectionRepository(db),
|
||
UserActionLogRepo: repository.NewGormUserActionLogRepository(db),
|
||
PigBatchRepo: repository.NewGormPigBatchRepository(db),
|
||
PigBatchLogRepo: repository.NewGormPigBatchLogRepository(db),
|
||
PigFarmRepo: repository.NewGormPigFarmRepository(db),
|
||
PigPenRepo: repository.NewGormPigPenRepository(db),
|
||
PigTransferLogRepo: repository.NewGormPigTransferLogRepository(db),
|
||
PigTradeRepo: repository.NewGormPigTradeRepository(db),
|
||
PigSickPigLogRepo: repository.NewGormPigSickLogRepository(db),
|
||
MedicationLogRepo: repository.NewGormMedicationLogRepository(db),
|
||
RawMaterialRepo: repository.NewGormRawMaterialRepository(db),
|
||
NotificationRepo: repository.NewGormNotificationRepository(db),
|
||
UnitOfWork: repository.NewGormUnitOfWork(db, logger),
|
||
}
|
||
}
|
||
|
||
// DomainServices 聚合了所有的领域服务实例。
|
||
type DomainServices struct {
|
||
PigPenTransferManager pig.PigPenTransferManager
|
||
PigTradeManager pig.PigTradeManager
|
||
PigSickManager pig.SickPigManager
|
||
PigBatchDomain pig.PigBatchService
|
||
GeneralDeviceService device.Service
|
||
taskFactory scheduler.TaskFactory
|
||
AnalysisPlanTaskManager *scheduler.AnalysisPlanTaskManager
|
||
Scheduler *scheduler.Scheduler
|
||
}
|
||
|
||
// initDomainServices 初始化所有的领域服务。
|
||
func initDomainServices(cfg *config.Config, infra *Infrastructure, logger *logs.Logger) *DomainServices {
|
||
// 猪群管理相关
|
||
pigPenTransferManager := pig.NewPigPenTransferManager(infra.Repos.PigPenRepo, infra.Repos.PigTransferLogRepo, infra.Repos.PigBatchRepo)
|
||
pigTradeManager := pig.NewPigTradeManager(infra.Repos.PigTradeRepo)
|
||
pigSickManager := pig.NewSickPigManager(infra.Repos.PigSickPigLogRepo, infra.Repos.MedicationLogRepo)
|
||
pigBatchDomain := pig.NewPigBatchService(infra.Repos.PigBatchRepo, infra.Repos.PigBatchLogRepo, infra.Repos.UnitOfWork,
|
||
pigPenTransferManager, pigTradeManager, pigSickManager)
|
||
|
||
// 通用设备服务
|
||
generalDeviceService := device.NewGeneralDeviceService(
|
||
infra.Repos.DeviceRepo,
|
||
infra.Repos.DeviceCommandLogRepo,
|
||
infra.Repos.PendingCollectionRepo,
|
||
logger,
|
||
infra.Lora.Comm,
|
||
)
|
||
|
||
// 计划任务管理器
|
||
analysisPlanTaskManager := scheduler.NewAnalysisPlanTaskManager(infra.Repos.PlanRepo, infra.Repos.PendingTaskRepo, infra.Repos.ExecutionLogRepo, logger)
|
||
|
||
// 任务工厂
|
||
taskFactory := task.NewTaskFactory(logger, infra.Repos.SensorDataRepo, infra.Repos.DeviceRepo, generalDeviceService)
|
||
|
||
// 任务执行器
|
||
planScheduler := scheduler.NewScheduler(
|
||
infra.Repos.PendingTaskRepo,
|
||
infra.Repos.ExecutionLogRepo,
|
||
infra.Repos.DeviceRepo,
|
||
infra.Repos.SensorDataRepo,
|
||
infra.Repos.PlanRepo,
|
||
analysisPlanTaskManager,
|
||
taskFactory,
|
||
logger,
|
||
generalDeviceService,
|
||
time.Duration(cfg.Task.Interval)*time.Second,
|
||
cfg.Task.NumWorkers,
|
||
)
|
||
|
||
return &DomainServices{
|
||
PigPenTransferManager: pigPenTransferManager,
|
||
PigTradeManager: pigTradeManager,
|
||
PigSickManager: pigSickManager,
|
||
PigBatchDomain: pigBatchDomain,
|
||
GeneralDeviceService: generalDeviceService,
|
||
AnalysisPlanTaskManager: analysisPlanTaskManager,
|
||
taskFactory: taskFactory,
|
||
Scheduler: planScheduler,
|
||
}
|
||
}
|
||
|
||
// AppServices 聚合了所有的应用服务实例。
|
||
type AppServices struct {
|
||
PigFarmService service.PigFarmService
|
||
PigBatchService service.PigBatchService
|
||
MonitorService service.MonitorService
|
||
AuditService audit.Service
|
||
}
|
||
|
||
// initAppServices 初始化所有的应用服务。
|
||
func initAppServices(infra *Infrastructure, domainServices *DomainServices, logger *logs.Logger) *AppServices {
|
||
pigFarmService := service.NewPigFarmService(infra.Repos.PigFarmRepo, infra.Repos.PigPenRepo, infra.Repos.PigBatchRepo, domainServices.PigBatchDomain, infra.Repos.UnitOfWork, logger)
|
||
pigBatchService := service.NewPigBatchService(domainServices.PigBatchDomain, logger)
|
||
monitorService := service.NewMonitorService(
|
||
infra.Repos.SensorDataRepo,
|
||
infra.Repos.DeviceCommandLogRepo,
|
||
infra.Repos.ExecutionLogRepo,
|
||
infra.Repos.PlanRepo,
|
||
infra.Repos.PendingCollectionRepo,
|
||
infra.Repos.UserActionLogRepo,
|
||
infra.Repos.RawMaterialRepo,
|
||
infra.Repos.MedicationLogRepo,
|
||
infra.Repos.PigBatchRepo,
|
||
infra.Repos.PigBatchLogRepo,
|
||
infra.Repos.PigTransferLogRepo,
|
||
infra.Repos.PigSickPigLogRepo,
|
||
infra.Repos.PigTradeRepo,
|
||
infra.Repos.NotificationRepo,
|
||
)
|
||
auditService := audit.NewService(infra.Repos.UserActionLogRepo, logger)
|
||
|
||
return &AppServices{
|
||
PigFarmService: pigFarmService,
|
||
PigBatchService: pigBatchService,
|
||
MonitorService: monitorService,
|
||
AuditService: auditService,
|
||
}
|
||
}
|
||
|
||
// LoraComponents 聚合了所有 LoRa 相关组件。
|
||
type LoraComponents struct {
|
||
ListenHandler webhook.ListenHandler
|
||
Comm transport.Communicator
|
||
LoraListener transport.Listener
|
||
}
|
||
|
||
// initLora 根据配置初始化 LoRa 相关组件。
|
||
func initLora(
|
||
cfg *config.Config,
|
||
logger *logs.Logger,
|
||
repos *Repositories,
|
||
) (*LoraComponents, error) {
|
||
var listenHandler webhook.ListenHandler
|
||
var comm transport.Communicator
|
||
var loraListener transport.Listener
|
||
|
||
if cfg.Lora.Mode == config.LoraMode_LoRaWAN {
|
||
logger.Info("当前运行模式: lora_wan。初始化 ChirpStack 监听器和传输层。")
|
||
listenHandler = webhook.NewChirpStackListener(logger, repos.SensorDataRepo, repos.DeviceRepo, repos.AreaControllerRepo, repos.DeviceCommandLogRepo, repos.PendingCollectionRepo)
|
||
comm = lora.NewChirpStackTransport(cfg.ChirpStack, logger)
|
||
loraListener = lora.NewPlaceholderTransport(logger)
|
||
} else {
|
||
logger.Info("当前运行模式: lora_mesh。初始化 LoRa Mesh 传输层和占位符监听器。")
|
||
listenHandler = webhook.NewPlaceholderListener(logger)
|
||
tp, err := lora.NewLoRaMeshUartPassthroughTransport(cfg.LoraMesh, logger, repos.AreaControllerRepo, repos.PendingCollectionRepo, repos.DeviceRepo, repos.SensorDataRepo)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("无法初始化 LoRa Mesh 模块: %w", err)
|
||
}
|
||
loraListener = tp
|
||
comm = tp
|
||
}
|
||
|
||
return &LoraComponents{
|
||
ListenHandler: listenHandler,
|
||
Comm: comm,
|
||
LoraListener: loraListener,
|
||
}, nil
|
||
}
|
||
|
||
// initNotifyService 根据配置初始化并返回一个通知领域服务。
|
||
// 它确保至少有一个 LogNotifier 总是可用，并根据配置启用其他通知器。
|
||
func initNotifyService(
|
||
cfg config.NotifyConfig,
|
||
log *logs.Logger,
|
||
userRepo repository.UserRepository,
|
||
notificationRepo repository.NotificationRepository,
|
||
) (domain_notify.Service, error) {
|
||
var availableNotifiers []notify.Notifier
|
||
|
||
// 1. 总是创建 LogNotifier 作为所有告警的最终记录渠道
|
||
logNotifier := notify.NewLogNotifier(log)
|
||
availableNotifiers = append(availableNotifiers, logNotifier)
|
||
log.Info("Log通知器已启用 (作为所有告警的最终记录渠道)")
|
||
|
||
// 2. 根据配置，按需创建并收集所有启用的其他 Notifier 实例
|
||
if cfg.SMTP.Enabled {
|
||
smtpNotifier := notify.NewSMTPNotifier(
|
||
cfg.SMTP.Host,
|
||
cfg.SMTP.Port,
|
||
cfg.SMTP.Username,
|
||
cfg.SMTP.Password,
|
||
cfg.SMTP.Sender,
|
||
)
|
||
availableNotifiers = append(availableNotifiers, smtpNotifier)
|
||
log.Info("SMTP通知器已启用")
|
||
}
|
||
|
||
if cfg.WeChat.Enabled {
|
||
wechatNotifier := notify.NewWechatNotifier(
|
||
cfg.WeChat.CorpID,
|
||
cfg.WeChat.AgentID,
|
||
cfg.WeChat.Secret,
|
||
)
|
||
availableNotifiers = append(availableNotifiers, wechatNotifier)
|
||
log.Info("企业微信通知器已启用")
|
||
}
|
||
|
||
if cfg.Lark.Enabled {
|
||
larkNotifier := notify.NewLarkNotifier(
|
||
cfg.Lark.AppID,
|
||
cfg.Lark.AppSecret,
|
||
)
|
||
availableNotifiers = append(availableNotifiers, larkNotifier)
|
||
log.Info("飞书通知器已启用")
|
||
}
|
||
|
||
// 3. 动态确定首选通知器
|
||
var primaryNotifier notify.Notifier
|
||
primaryNotifierType := notify.NotifierType(cfg.Primary)
|
||
|
||
// 检查用户指定的主渠道是否已启用
|
||
for _, n := range availableNotifiers {
|
||
if n.Type() == primaryNotifierType {
|
||
primaryNotifier = n
|
||
break
|
||
}
|
||
}
|
||
|
||
// 如果用户指定的主渠道未启用或未指定，则自动选择第一个可用的 (这将是 LogNotifier，如果其他都未启用)
|
||
if primaryNotifier == nil {
|
||
primaryNotifier = availableNotifiers[0] // 确保总能找到一个，因为 LogNotifier 总是存在的
|
||
log.Warnf("配置的首选渠道 '%s' 未启用或未指定，已自动降级使用 '%s' 作为首选渠道。", cfg.Primary, primaryNotifier.Type())
|
||
}
|
||
|
||
// 4. 使用创建的 Notifier 列表和 notificationRepo 来组装领域服务
|
||
notifyService, err := domain_notify.NewFailoverService(
|
||
log,
|
||
userRepo,
|
||
availableNotifiers,
|
||
primaryNotifier.Type(),
|
||
cfg.FailureThreshold,
|
||
notificationRepo,
|
||
)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("创建故障转移通知服务失败: %w", err)
|
||
}
|
||
|
||
log.Infof("通知服务初始化成功，首选渠道: %s, 故障阈值: %d", primaryNotifier.Type(), cfg.FailureThreshold)
|
||
return notifyService, nil
|
||
}
|
||
|
||
// initStorage 封装了数据库的初始化、连接和迁移逻辑。
|
||
func initStorage(cfg config.DatabaseConfig, logger *logs.Logger) (database.Storage, error) {
|
||
// 创建存储实例
|
||
storage := database.NewStorage(cfg, logger)
|
||
if err := storage.Connect(); err != nil {
|
||
// 错误已在 Connect 内部被记录，这里只需包装并返回
|
||
return nil, fmt.Errorf("数据库连接失败: %w", err)
|
||
}
|
||
|
||
// 执行数据库迁移
|
||
if err := storage.Migrate(models.GetAllModels()...); err != nil {
|
||
return nil, fmt.Errorf("数据库迁移失败: %w", err)
|
||
}
|
||
|
||
logger.Info("数据库初始化完成。")
|
||
return storage, nil
|
||
}
|