package service import ( "context" "errors" "fmt" "strings" "sync" "time" "affiliate_dash/internal/model" "gorm.io/gorm" ) // RockService 负责管理 Rock 平台的配置、库存池、采购、UID强一致风控与状态机 type RockService struct { db *gorm.DB client *RockClient mu sync.Mutex refreshCache sync.Map // key: upstreamTokenID string, value: rockRefreshCacheItem bindLinkCache sync.Map // key: token ID uint, value: rockBindLinkCacheItem } func NewRockService(db *gorm.DB, client *RockClient) *RockService { return &RockService{ db: db, client: client, } } // RockStatsResponse 看板与库存统计返回 type RockStatsResponse struct { AvailableStock int64 `json:"available_stock"` UsingStock int64 `json:"using_stock"` BoundStock int64 `json:"bound_stock"` RedeemedStock int64 `json:"redeemed_stock"` MismatchStock int64 `json:"mismatch_stock"` // 绑定错误(待排查) AbnormalStock int64 `json:"abnormal_stock"` UpstreamPoints float64 `json:"upstream_points"` AutoDeliveryEnabled bool `json:"auto_delivery_enabled"` TokenWarningStock int `json:"token_warning_stock"` AutoPurchaseEnabled bool `json:"auto_purchase_enabled"` Username string `json:"username"` LastHeartbeatAt string `json:"last_heartbeat_at,omitempty"` } type UpdateRockConfigReq struct { AutoDeliveryEnabled *bool `json:"auto_delivery_enabled"` BaseURL *string `json:"base_url"` BearerToken *string `json:"bearer_token"` MinPointsWarning *float64 `json:"min_points_warning"` TokenWarningStock *int `json:"token_warning_stock"` AutoPurchaseEnabled *bool `json:"auto_purchase_enabled"` AutoPurchaseCount *int `json:"auto_purchase_count"` PurchaseCooldownMinutes *int `json:"purchase_cooldown_minutes"` } // GetConfig 获取当前配置,若无则初始化 func (s *RockService) GetConfig(ctx context.Context) (*model.RockConfig, error) { var cfg model.RockConfig err := s.db.WithContext(ctx).First(&cfg, 1).Error if errors.Is(err, gorm.ErrRecordNotFound) { cfg = model.RockConfig{ ID: 1, AutoDeliveryEnabled: true, BaseURL: model.RockDefaultBaseURL, BearerToken: "", MinPointsWarning: 1050.00, TokenWarningStock: 10, AutoPurchaseEnabled: false, AutoPurchaseCount: 50, PurchaseCooldownMinutes: 10, } if createErr := s.db.WithContext(ctx).Create(&cfg).Error; createErr != nil { return nil, createErr } return &cfg, nil } if err != nil { return nil, err } return &cfg, nil } // ApplyConfigToClient 把数据库中的上游地址同步到 HTTP 客户端。 // // 必须在进程启动时调用一次:否则重启后客户端会回落到内置默认地址, // 后台改过的 base_url 就会表现为「保存时生效、重启即失效」。 func (s *RockService) ApplyConfigToClient(ctx context.Context) error { cfg, err := s.GetConfig(ctx) if err != nil { return err } s.client.SetBaseURL(cfg.BaseURL) return nil } // UpdateConfig 更新系统配置 func (s *RockService) UpdateConfig(ctx context.Context, req UpdateRockConfigReq) (*model.RockConfig, error) { cfg, err := s.GetConfig(ctx) if err != nil { return nil, err } updates := make(map[string]interface{}) if req.AutoDeliveryEnabled != nil { updates["auto_delivery_enabled"] = *req.AutoDeliveryEnabled } if req.BaseURL != nil && strings.TrimSpace(*req.BaseURL) != "" { updates["base_url"] = strings.TrimRight(strings.TrimSpace(*req.BaseURL), "/") // 立即生效,避免还要等下一次重启或配置读取;客户端内部已加锁。 s.client.SetBaseURL(updates["base_url"].(string)) } if req.BearerToken != nil { updates["bearer_token"] = strings.TrimSpace(*req.BearerToken) } if req.MinPointsWarning != nil { updates["min_points_warning"] = *req.MinPointsWarning } if req.TokenWarningStock != nil { updates["token_warning_stock"] = *req.TokenWarningStock } if req.AutoPurchaseEnabled != nil { updates["auto_purchase_enabled"] = *req.AutoPurchaseEnabled } if req.AutoPurchaseCount != nil { updates["auto_purchase_count"] = *req.AutoPurchaseCount } if req.PurchaseCooldownMinutes != nil { updates["purchase_cooldown_minutes"] = *req.PurchaseCooldownMinutes } if len(updates) > 0 { updates["updated_at"] = time.Now() if err := s.db.WithContext(ctx).Model(&model.RockConfig{}).Where("id = ?", cfg.ID).Updates(updates).Error; err != nil { return nil, err } } return s.GetConfig(ctx) } // TestConnection 测试上游 Token 连通性并同步最新用户信息与积分 func (s *RockService) TestConnection(ctx context.Context, bearerToken string) (*RockMeResponse, error) { token := strings.TrimSpace(bearerToken) if token == "" { cfg, err := s.GetConfig(ctx) if err != nil { return nil, err } token = cfg.BearerToken } if token == "" { return nil, fmt.Errorf("请先输入或配置上游 Bearer Token") } me, err := s.client.GetMe(ctx, token) if err != nil { return nil, err } // 连通成功,顺便异步更新数据库快照 now := time.Now() _ = s.db.WithContext(ctx).Model(&model.RockConfig{}).Where("id = 1").Updates(map[string]interface{}{ "username": me.User.Username, "points": me.User.Points.Float64(), "last_heartbeat_at": now, }).Error return me, nil } // GetStats 获取概览统计 func (s *RockService) GetStats(ctx context.Context) (*RockStatsResponse, error) { cfg, err := s.GetConfig(ctx) if err != nil { return nil, err } var stats RockStatsResponse stats.AutoDeliveryEnabled = cfg.AutoDeliveryEnabled stats.TokenWarningStock = cfg.TokenWarningStock stats.AutoPurchaseEnabled = cfg.AutoPurchaseEnabled stats.UpstreamPoints = cfg.Points stats.Username = cfg.Username if cfg.LastHeartbeatAt != nil { stats.LastHeartbeatAt = cfg.LastHeartbeatAt.Format("2006-01-02 15:04:05") } type CountResult struct { Status string Total int64 } var results []CountResult err = s.db.WithContext(ctx).Model(&model.RockToken{}). Select("status, count(*) as total"). Group("status"). Scan(&results).Error if err != nil { return nil, err } for _, r := range results { switch r.Status { case model.RockStatusAvailable: stats.AvailableStock = r.Total case model.RockStatusUsing: stats.UsingStock = r.Total case model.RockStatusBound: stats.BoundStock = r.Total case model.RockStatusRedeemed: stats.RedeemedStock = r.Total case model.RockStatusMismatch: stats.MismatchStock = r.Total case model.RockStatusAbnormal: stats.AbnormalStock = r.Total } } return &stats, nil }