337 lines
9.2 KiB
Go
337 lines
9.2 KiB
Go
package service
|
||
|
||
import (
|
||
"context"
|
||
"log"
|
||
"sort"
|
||
"strings"
|
||
"sync"
|
||
"time"
|
||
|
||
"github.com/Wei-Shaw/sub2api/internal/model"
|
||
)
|
||
|
||
// ErrorPassthroughRepository 定义错误透传规则的数据访问接口
|
||
type ErrorPassthroughRepository interface {
|
||
// List 获取所有规则
|
||
List(ctx context.Context) ([]*model.ErrorPassthroughRule, error)
|
||
// GetByID 根据 ID 获取规则
|
||
GetByID(ctx context.Context, id int64) (*model.ErrorPassthroughRule, error)
|
||
// Create 创建规则
|
||
Create(ctx context.Context, rule *model.ErrorPassthroughRule) (*model.ErrorPassthroughRule, error)
|
||
// Update 更新规则
|
||
Update(ctx context.Context, rule *model.ErrorPassthroughRule) (*model.ErrorPassthroughRule, error)
|
||
// Delete 删除规则
|
||
Delete(ctx context.Context, id int64) error
|
||
}
|
||
|
||
// ErrorPassthroughCache 定义错误透传规则的缓存接口
|
||
type ErrorPassthroughCache interface {
|
||
// Get 从缓存获取规则列表
|
||
Get(ctx context.Context) ([]*model.ErrorPassthroughRule, bool)
|
||
// Set 设置缓存
|
||
Set(ctx context.Context, rules []*model.ErrorPassthroughRule) error
|
||
// Invalidate 使缓存失效
|
||
Invalidate(ctx context.Context) error
|
||
// NotifyUpdate 通知其他实例刷新缓存
|
||
NotifyUpdate(ctx context.Context) error
|
||
// SubscribeUpdates 订阅缓存更新通知
|
||
SubscribeUpdates(ctx context.Context, handler func())
|
||
}
|
||
|
||
// ErrorPassthroughService 错误透传规则服务
|
||
type ErrorPassthroughService struct {
|
||
repo ErrorPassthroughRepository
|
||
cache ErrorPassthroughCache
|
||
|
||
// 本地内存缓存，用于快速匹配
|
||
localCache []*model.ErrorPassthroughRule
|
||
localCacheMu sync.RWMutex
|
||
}
|
||
|
||
// NewErrorPassthroughService 创建错误透传规则服务
|
||
func NewErrorPassthroughService(
|
||
repo ErrorPassthroughRepository,
|
||
cache ErrorPassthroughCache,
|
||
) *ErrorPassthroughService {
|
||
svc := &ErrorPassthroughService{
|
||
repo: repo,
|
||
cache: cache,
|
||
}
|
||
|
||
// 启动时加载规则到本地缓存
|
||
ctx := context.Background()
|
||
if err := svc.reloadRulesFromDB(ctx); err != nil {
|
||
log.Printf("[ErrorPassthroughService] Failed to load rules from DB on startup: %v", err)
|
||
if fallbackErr := svc.refreshLocalCache(ctx); fallbackErr != nil {
|
||
log.Printf("[ErrorPassthroughService] Failed to load rules from cache fallback on startup: %v", fallbackErr)
|
||
}
|
||
}
|
||
|
||
// 订阅缓存更新通知
|
||
if cache != nil {
|
||
cache.SubscribeUpdates(ctx, func() {
|
||
if err := svc.refreshLocalCache(context.Background()); err != nil {
|
||
log.Printf("[ErrorPassthroughService] Failed to refresh cache on notification: %v", err)
|
||
}
|
||
})
|
||
}
|
||
|
||
return svc
|
||
}
|
||
|
||
// List 获取所有规则
|
||
func (s *ErrorPassthroughService) List(ctx context.Context) ([]*model.ErrorPassthroughRule, error) {
|
||
return s.repo.List(ctx)
|
||
}
|
||
|
||
// GetByID 根据 ID 获取规则
|
||
func (s *ErrorPassthroughService) GetByID(ctx context.Context, id int64) (*model.ErrorPassthroughRule, error) {
|
||
return s.repo.GetByID(ctx, id)
|
||
}
|
||
|
||
// Create 创建规则
|
||
func (s *ErrorPassthroughService) Create(ctx context.Context, rule *model.ErrorPassthroughRule) (*model.ErrorPassthroughRule, error) {
|
||
if err := rule.Validate(); err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
created, err := s.repo.Create(ctx, rule)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
// 刷新缓存
|
||
refreshCtx, cancel := s.newCacheRefreshContext()
|
||
defer cancel()
|
||
s.invalidateAndNotify(refreshCtx)
|
||
|
||
return created, nil
|
||
}
|
||
|
||
// Update 更新规则
|
||
func (s *ErrorPassthroughService) Update(ctx context.Context, rule *model.ErrorPassthroughRule) (*model.ErrorPassthroughRule, error) {
|
||
if err := rule.Validate(); err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
updated, err := s.repo.Update(ctx, rule)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
// 刷新缓存
|
||
refreshCtx, cancel := s.newCacheRefreshContext()
|
||
defer cancel()
|
||
s.invalidateAndNotify(refreshCtx)
|
||
|
||
return updated, nil
|
||
}
|
||
|
||
// Delete 删除规则
|
||
func (s *ErrorPassthroughService) Delete(ctx context.Context, id int64) error {
|
||
if err := s.repo.Delete(ctx, id); err != nil {
|
||
return err
|
||
}
|
||
|
||
// 刷新缓存
|
||
refreshCtx, cancel := s.newCacheRefreshContext()
|
||
defer cancel()
|
||
s.invalidateAndNotify(refreshCtx)
|
||
|
||
return nil
|
||
}
|
||
|
||
// MatchRule 匹配透传规则
|
||
// 返回第一个匹配的规则，如果没有匹配则返回 nil
|
||
func (s *ErrorPassthroughService) MatchRule(platform string, statusCode int, body []byte) *model.ErrorPassthroughRule {
|
||
rules := s.getCachedRules()
|
||
if len(rules) == 0 {
|
||
return nil
|
||
}
|
||
|
||
bodyStr := strings.ToLower(string(body))
|
||
|
||
for _, rule := range rules {
|
||
if !rule.Enabled {
|
||
continue
|
||
}
|
||
if !s.platformMatches(rule, platform) {
|
||
continue
|
||
}
|
||
if s.ruleMatches(rule, statusCode, bodyStr) {
|
||
return rule
|
||
}
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
// getCachedRules 获取缓存的规则列表（按优先级排序）
|
||
func (s *ErrorPassthroughService) getCachedRules() []*model.ErrorPassthroughRule {
|
||
s.localCacheMu.RLock()
|
||
rules := s.localCache
|
||
s.localCacheMu.RUnlock()
|
||
|
||
if rules != nil {
|
||
return rules
|
||
}
|
||
|
||
// 如果本地缓存为空，尝试刷新
|
||
ctx := context.Background()
|
||
if err := s.refreshLocalCache(ctx); err != nil {
|
||
log.Printf("[ErrorPassthroughService] Failed to refresh cache: %v", err)
|
||
return nil
|
||
}
|
||
|
||
s.localCacheMu.RLock()
|
||
defer s.localCacheMu.RUnlock()
|
||
return s.localCache
|
||
}
|
||
|
||
// refreshLocalCache 刷新本地缓存
|
||
func (s *ErrorPassthroughService) refreshLocalCache(ctx context.Context) error {
|
||
// 先尝试从 Redis 缓存获取
|
||
if s.cache != nil {
|
||
if rules, ok := s.cache.Get(ctx); ok {
|
||
s.setLocalCache(rules)
|
||
return nil
|
||
}
|
||
}
|
||
|
||
return s.reloadRulesFromDB(ctx)
|
||
}
|
||
|
||
// 从数据库加载（repo.List 已按 priority 排序）
|
||
// 注意：该方法会绕过 cache.Get，确保拿到数据库最新值。
|
||
func (s *ErrorPassthroughService) reloadRulesFromDB(ctx context.Context) error {
|
||
rules, err := s.repo.List(ctx)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
// 更新 Redis 缓存
|
||
if s.cache != nil {
|
||
if err := s.cache.Set(ctx, rules); err != nil {
|
||
log.Printf("[ErrorPassthroughService] Failed to set cache: %v", err)
|
||
}
|
||
}
|
||
|
||
// 更新本地缓存（setLocalCache 内部会确保排序）
|
||
s.setLocalCache(rules)
|
||
|
||
return nil
|
||
}
|
||
|
||
// setLocalCache 设置本地缓存
|
||
func (s *ErrorPassthroughService) setLocalCache(rules []*model.ErrorPassthroughRule) {
|
||
// 按优先级排序
|
||
sorted := make([]*model.ErrorPassthroughRule, len(rules))
|
||
copy(sorted, rules)
|
||
sort.Slice(sorted, func(i, j int) bool {
|
||
return sorted[i].Priority < sorted[j].Priority
|
||
})
|
||
|
||
s.localCacheMu.Lock()
|
||
s.localCache = sorted
|
||
s.localCacheMu.Unlock()
|
||
}
|
||
|
||
// clearLocalCache 清空本地缓存，避免刷新失败时继续命中陈旧规则。
|
||
func (s *ErrorPassthroughService) clearLocalCache() {
|
||
s.localCacheMu.Lock()
|
||
s.localCache = nil
|
||
s.localCacheMu.Unlock()
|
||
}
|
||
|
||
// newCacheRefreshContext 为写路径缓存同步创建独立上下文，避免受请求取消影响。
|
||
func (s *ErrorPassthroughService) newCacheRefreshContext() (context.Context, context.CancelFunc) {
|
||
return context.WithTimeout(context.Background(), 3*time.Second)
|
||
}
|
||
|
||
// invalidateAndNotify 使缓存失效并通知其他实例
|
||
func (s *ErrorPassthroughService) invalidateAndNotify(ctx context.Context) {
|
||
// 先失效缓存，避免后续刷新读到陈旧规则。
|
||
if s.cache != nil {
|
||
if err := s.cache.Invalidate(ctx); err != nil {
|
||
log.Printf("[ErrorPassthroughService] Failed to invalidate cache: %v", err)
|
||
}
|
||
}
|
||
|
||
// 刷新本地缓存
|
||
if err := s.reloadRulesFromDB(ctx); err != nil {
|
||
log.Printf("[ErrorPassthroughService] Failed to refresh local cache: %v", err)
|
||
// 刷新失败时清空本地缓存，避免继续使用陈旧规则。
|
||
s.clearLocalCache()
|
||
}
|
||
|
||
// 通知其他实例
|
||
if s.cache != nil {
|
||
if err := s.cache.NotifyUpdate(ctx); err != nil {
|
||
log.Printf("[ErrorPassthroughService] Failed to notify cache update: %v", err)
|
||
}
|
||
}
|
||
}
|
||
|
||
// platformMatches 检查平台是否匹配
|
||
func (s *ErrorPassthroughService) platformMatches(rule *model.ErrorPassthroughRule, platform string) bool {
|
||
// 如果没有配置平台限制，则匹配所有平台
|
||
if len(rule.Platforms) == 0 {
|
||
return true
|
||
}
|
||
|
||
platform = strings.ToLower(platform)
|
||
for _, p := range rule.Platforms {
|
||
if strings.ToLower(p) == platform {
|
||
return true
|
||
}
|
||
}
|
||
|
||
return false
|
||
}
|
||
|
||
// ruleMatches 检查规则是否匹配
|
||
func (s *ErrorPassthroughService) ruleMatches(rule *model.ErrorPassthroughRule, statusCode int, bodyLower string) bool {
|
||
hasErrorCodes := len(rule.ErrorCodes) > 0
|
||
hasKeywords := len(rule.Keywords) > 0
|
||
|
||
// 如果没有配置任何条件，不匹配
|
||
if !hasErrorCodes && !hasKeywords {
|
||
return false
|
||
}
|
||
|
||
codeMatch := !hasErrorCodes || s.containsInt(rule.ErrorCodes, statusCode)
|
||
keywordMatch := !hasKeywords || s.containsAnyKeyword(bodyLower, rule.Keywords)
|
||
|
||
if rule.MatchMode == model.MatchModeAll {
|
||
// "all" 模式：所有配置的条件都必须满足
|
||
return codeMatch && keywordMatch
|
||
}
|
||
|
||
// "any" 模式：任一条件满足即可
|
||
if hasErrorCodes && hasKeywords {
|
||
return codeMatch || keywordMatch
|
||
}
|
||
return codeMatch && keywordMatch
|
||
}
|
||
|
||
// containsInt 检查切片是否包含指定整数
|
||
func (s *ErrorPassthroughService) containsInt(slice []int, val int) bool {
|
||
for _, v := range slice {
|
||
if v == val {
|
||
return true
|
||
}
|
||
}
|
||
return false
|
||
}
|
||
|
||
// containsAnyKeyword 检查字符串是否包含任一关键词（不区分大小写）
|
||
func (s *ErrorPassthroughService) containsAnyKeyword(bodyLower string, keywords []string) bool {
|
||
for _, kw := range keywords {
|
||
if strings.Contains(bodyLower, strings.ToLower(kw)) {
|
||
return true
|
||
}
|
||
}
|
||
return false
|
||
}
|