huang
aef20d2ab1
All checks were successful
构建并部署到测试环境（无 SSH） / build-and-deploy (push) Successful in 7m23s
- 新增 RedisPollingConfigChangedChannel 常量，用于跨进程配置变更通知 - ConfigService 注入 Redis，Create/Delete/UpdateStatus 后发布 Pub/Sub 事件 - PollingConfigManager 新增 WatchChanges() 方法，收到通知立即刷新内存缓存 - PollingInitializer 新增 Restart() 方法，支持初始化完成后安全重启（CAS 防并发） - Worker 订阅配置变更事件，configs 从空变为非空时触发 Initializer.Restart() 根因：启动时所有配置禁用导致分片队列为空，Initializer 是一次性的， 之后启用配置只更新 DB 但不重建队列，调度器无卡可处理
281 lines
9.0 KiB
Go
281 lines
9.0 KiB
Go
package polling
|
|
|
|
import (
|
|
"context"
|
|
"time"
|
|
|
|
"github.com/redis/go-redis/v9"
|
|
"go.uber.org/zap"
|
|
"gorm.io/gorm"
|
|
|
|
"github.com/break/junhong_cmp_fiber/internal/model"
|
|
"github.com/break/junhong_cmp_fiber/internal/model/dto"
|
|
"github.com/break/junhong_cmp_fiber/internal/store"
|
|
"github.com/break/junhong_cmp_fiber/internal/store/postgres"
|
|
"github.com/break/junhong_cmp_fiber/pkg/constants"
|
|
"github.com/break/junhong_cmp_fiber/pkg/errors"
|
|
"github.com/break/junhong_cmp_fiber/pkg/middleware"
|
|
)
|
|
|
|
// ConfigService 轮询配置服务
|
|
type ConfigService struct {
|
|
configStore *postgres.PollingConfigStore
|
|
redis *redis.Client
|
|
logger *zap.Logger
|
|
}
|
|
|
|
// NewConfigService 创建轮询配置服务实例
|
|
func NewConfigService(configStore *postgres.PollingConfigStore, redisClient *redis.Client, logger *zap.Logger) *ConfigService {
|
|
return &ConfigService{configStore: configStore, redis: redisClient, logger: logger}
|
|
}
|
|
|
|
func (s *ConfigService) notifyConfigChanged(ctx context.Context, event string) {
|
|
if s.redis == nil {
|
|
return
|
|
}
|
|
if err := s.redis.Publish(ctx, constants.RedisPollingConfigChangedChannel(), event).Err(); err != nil {
|
|
s.logger.Warn("通知轮询配置变更失败", zap.String("event", event), zap.Error(err))
|
|
}
|
|
}
|
|
|
|
// Create 创建轮询配置
|
|
func (s *ConfigService) Create(ctx context.Context, req *dto.CreatePollingConfigRequest) (*dto.PollingConfigResponse, error) {
|
|
currentUserID := middleware.GetUserIDFromContext(ctx)
|
|
if currentUserID == 0 {
|
|
return nil, errors.New(errors.CodeUnauthorized, "未授权访问")
|
|
}
|
|
|
|
// 验证配置名称唯一性
|
|
existing, _ := s.configStore.GetByName(ctx, req.ConfigName)
|
|
if existing != nil {
|
|
return nil, errors.New(errors.CodeInvalidParam, "配置名称已存在")
|
|
}
|
|
|
|
// 验证检查间隔（至少一个不为空）
|
|
if req.RealnameCheckInterval == nil && req.CarddataCheckInterval == nil &&
|
|
req.PackageCheckInterval == nil && req.ProtectCheckInterval == nil &&
|
|
req.CardStatusCheckInterval == nil {
|
|
return nil, errors.New(errors.CodeInvalidParam, "至少需要配置一种检查间隔")
|
|
}
|
|
|
|
config := &model.PollingConfig{
|
|
ConfigName: req.ConfigName,
|
|
CardCondition: req.CardCondition,
|
|
CardCategory: req.CardCategory,
|
|
CarrierID: req.CarrierID,
|
|
Priority: req.Priority,
|
|
RealnameCheckInterval: req.RealnameCheckInterval,
|
|
CarddataCheckInterval: req.CarddataCheckInterval,
|
|
PackageCheckInterval: req.PackageCheckInterval,
|
|
ProtectCheckInterval: req.ProtectCheckInterval,
|
|
CardStatusCheckInterval: req.CardStatusCheckInterval,
|
|
Status: 1,
|
|
Description: req.Description,
|
|
CreatedBy: ¤tUserID,
|
|
UpdatedBy: ¤tUserID,
|
|
}
|
|
|
|
if err := s.configStore.Create(ctx, config); err != nil {
|
|
return nil, errors.Wrap(errors.CodeInternalError, err, "创建轮询配置失败")
|
|
}
|
|
|
|
s.notifyConfigChanged(ctx, "created")
|
|
return s.toResponse(config), nil
|
|
}
|
|
|
|
// Get 获取轮询配置详情
|
|
func (s *ConfigService) Get(ctx context.Context, id uint) (*dto.PollingConfigResponse, error) {
|
|
config, err := s.configStore.GetByID(ctx, id)
|
|
if err != nil {
|
|
if err == gorm.ErrRecordNotFound {
|
|
return nil, errors.New(errors.CodePollingConfigNotFound, "轮询配置不存在")
|
|
}
|
|
return nil, errors.Wrap(errors.CodeInternalError, err, "获取轮询配置失败")
|
|
}
|
|
return s.toResponse(config), nil
|
|
}
|
|
|
|
// Update 更新轮询配置
|
|
func (s *ConfigService) Update(ctx context.Context, id uint, req *dto.UpdatePollingConfigRequest) (*dto.PollingConfigResponse, error) {
|
|
currentUserID := middleware.GetUserIDFromContext(ctx)
|
|
if currentUserID == 0 {
|
|
return nil, errors.New(errors.CodeUnauthorized, "未授权访问")
|
|
}
|
|
|
|
config, err := s.configStore.GetByID(ctx, id)
|
|
if err != nil {
|
|
if err == gorm.ErrRecordNotFound {
|
|
return nil, errors.New(errors.CodePollingConfigNotFound, "轮询配置不存在")
|
|
}
|
|
return nil, errors.Wrap(errors.CodeInternalError, err, "获取轮询配置失败")
|
|
}
|
|
|
|
// 更新字段
|
|
if req.ConfigName != nil {
|
|
// 检查名称唯一性
|
|
existing, _ := s.configStore.GetByName(ctx, *req.ConfigName)
|
|
if existing != nil && existing.ID != id {
|
|
return nil, errors.New(errors.CodeInvalidParam, "配置名称已存在")
|
|
}
|
|
config.ConfigName = *req.ConfigName
|
|
}
|
|
if req.CardCondition != nil {
|
|
config.CardCondition = *req.CardCondition
|
|
}
|
|
if req.CardCategory != nil {
|
|
config.CardCategory = *req.CardCategory
|
|
}
|
|
if req.CarrierID != nil {
|
|
config.CarrierID = req.CarrierID
|
|
}
|
|
if req.Priority != nil {
|
|
config.Priority = *req.Priority
|
|
}
|
|
if req.RealnameCheckInterval != nil {
|
|
config.RealnameCheckInterval = req.RealnameCheckInterval
|
|
}
|
|
if req.CarddataCheckInterval != nil {
|
|
config.CarddataCheckInterval = req.CarddataCheckInterval
|
|
}
|
|
if req.PackageCheckInterval != nil {
|
|
config.PackageCheckInterval = req.PackageCheckInterval
|
|
}
|
|
if req.ProtectCheckInterval != nil {
|
|
config.ProtectCheckInterval = req.ProtectCheckInterval
|
|
}
|
|
if req.CardStatusCheckInterval != nil {
|
|
config.CardStatusCheckInterval = req.CardStatusCheckInterval
|
|
}
|
|
if req.Description != nil {
|
|
config.Description = *req.Description
|
|
}
|
|
config.UpdatedBy = ¤tUserID
|
|
|
|
if err := s.configStore.Update(ctx, config); err != nil {
|
|
return nil, errors.Wrap(errors.CodeInternalError, err, "更新轮询配置失败")
|
|
}
|
|
|
|
return s.toResponse(config), nil
|
|
}
|
|
|
|
// Delete 删除轮询配置
|
|
func (s *ConfigService) Delete(ctx context.Context, id uint) error {
|
|
_, err := s.configStore.GetByID(ctx, id)
|
|
if err != nil {
|
|
if err == gorm.ErrRecordNotFound {
|
|
return errors.New(errors.CodePollingConfigNotFound, "轮询配置不存在")
|
|
}
|
|
return errors.Wrap(errors.CodeInternalError, err, "获取轮询配置失败")
|
|
}
|
|
|
|
if err := s.configStore.Delete(ctx, id); err != nil {
|
|
return errors.Wrap(errors.CodeInternalError, err, "删除轮询配置失败")
|
|
}
|
|
|
|
s.notifyConfigChanged(ctx, "deleted")
|
|
return nil
|
|
}
|
|
|
|
// List 列表查询轮询配置
|
|
func (s *ConfigService) List(ctx context.Context, req *dto.PollingConfigListRequest) ([]*dto.PollingConfigResponse, int64, error) {
|
|
opts := &store.QueryOptions{
|
|
Page: req.Page,
|
|
PageSize: req.PageSize,
|
|
OrderBy: "priority ASC, id DESC",
|
|
}
|
|
if opts.Page == 0 {
|
|
opts.Page = 1
|
|
}
|
|
if opts.PageSize == 0 {
|
|
opts.PageSize = constants.DefaultPageSize
|
|
}
|
|
|
|
filters := make(map[string]interface{})
|
|
if req.Status != nil {
|
|
filters["status"] = *req.Status
|
|
}
|
|
if req.CardCondition != nil {
|
|
filters["card_condition"] = *req.CardCondition
|
|
}
|
|
if req.CardCategory != nil {
|
|
filters["card_category"] = *req.CardCategory
|
|
}
|
|
if req.CarrierID != nil {
|
|
filters["carrier_id"] = *req.CarrierID
|
|
}
|
|
if req.ConfigName != nil {
|
|
filters["config_name"] = *req.ConfigName
|
|
}
|
|
|
|
configs, total, err := s.configStore.List(ctx, opts, filters)
|
|
if err != nil {
|
|
return nil, 0, errors.Wrap(errors.CodeInternalError, err, "查询轮询配置列表失败")
|
|
}
|
|
|
|
responses := make([]*dto.PollingConfigResponse, len(configs))
|
|
for i, c := range configs {
|
|
responses[i] = s.toResponse(c)
|
|
}
|
|
|
|
return responses, total, nil
|
|
}
|
|
|
|
// UpdateStatus 更新配置状态（启用/禁用）
|
|
func (s *ConfigService) UpdateStatus(ctx context.Context, id uint, status int16) error {
|
|
currentUserID := middleware.GetUserIDFromContext(ctx)
|
|
if currentUserID == 0 {
|
|
return errors.New(errors.CodeUnauthorized, "未授权访问")
|
|
}
|
|
|
|
_, err := s.configStore.GetByID(ctx, id)
|
|
if err != nil {
|
|
if err == gorm.ErrRecordNotFound {
|
|
return errors.New(errors.CodePollingConfigNotFound, "轮询配置不存在")
|
|
}
|
|
return errors.Wrap(errors.CodeInternalError, err, "获取轮询配置失败")
|
|
}
|
|
|
|
if err := s.configStore.UpdateStatus(ctx, id, status, currentUserID); err != nil {
|
|
return errors.Wrap(errors.CodeInternalError, err, "更新轮询配置状态失败")
|
|
}
|
|
|
|
s.notifyConfigChanged(ctx, "updated")
|
|
return nil
|
|
}
|
|
|
|
// ListEnabled 获取所有启用的配置
|
|
func (s *ConfigService) ListEnabled(ctx context.Context) ([]*dto.PollingConfigResponse, error) {
|
|
configs, err := s.configStore.ListEnabled(ctx)
|
|
if err != nil {
|
|
return nil, errors.Wrap(errors.CodeInternalError, err, "获取启用配置失败")
|
|
}
|
|
|
|
responses := make([]*dto.PollingConfigResponse, len(configs))
|
|
for i, c := range configs {
|
|
responses[i] = s.toResponse(c)
|
|
}
|
|
|
|
return responses, nil
|
|
}
|
|
|
|
// toResponse 转换为响应 DTO
|
|
func (s *ConfigService) toResponse(c *model.PollingConfig) *dto.PollingConfigResponse {
|
|
return &dto.PollingConfigResponse{
|
|
ID: c.ID,
|
|
ConfigName: c.ConfigName,
|
|
CardCondition: c.CardCondition,
|
|
CardCategory: c.CardCategory,
|
|
CarrierID: c.CarrierID,
|
|
Priority: c.Priority,
|
|
RealnameCheckInterval: c.RealnameCheckInterval,
|
|
CarddataCheckInterval: c.CarddataCheckInterval,
|
|
PackageCheckInterval: c.PackageCheckInterval,
|
|
ProtectCheckInterval: c.ProtectCheckInterval,
|
|
CardStatusCheckInterval: c.CardStatusCheckInterval,
|
|
Status: c.Status,
|
|
Description: c.Description,
|
|
CreatedAt: c.CreatedAt.Format(time.RFC3339),
|
|
UpdatedAt: c.UpdatedAt.Format(time.RFC3339),
|
|
}
|
|
}
|