0e514ed80b
- 为 GetActiveSubscription 添加 ristretto L1 缓存 + singleflight 防击穿 - 合并 ValidateSubscription + CheckUsageLimits 为纯内存 ValidateAndCheckLimits - 窗口维护操作（激活/重置）异步化，不再阻塞首字节 - 缓存返回浅拷贝，避免并发 data race 和缓存污染 - 所有管理操作（分配/续期/撤销/扩展/窗口重置）同步失效 L1 缓存 - 新增 SubscriptionCacheConfig 可配置 L1 缓存大小/TTL/抖动 Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
878 lines
27 KiB
Go
878 lines
27 KiB
Go
package service
|
||
|
||
import (
|
||
"context"
|
||
"fmt"
|
||
"log"
|
||
"math/rand/v2"
|
||
"strconv"
|
||
"time"
|
||
|
||
"github.com/Wei-Shaw/sub2api/internal/config"
|
||
infraerrors "github.com/Wei-Shaw/sub2api/internal/pkg/errors"
|
||
"github.com/Wei-Shaw/sub2api/internal/pkg/pagination"
|
||
"github.com/dgraph-io/ristretto"
|
||
"golang.org/x/sync/singleflight"
|
||
)
|
||
|
||
// MaxExpiresAt is the maximum allowed expiration date (year 2099)
|
||
// This prevents time.Time JSON serialization errors (RFC 3339 requires year <= 9999)
|
||
var MaxExpiresAt = time.Date(2099, 12, 31, 23, 59, 59, 0, time.UTC)
|
||
|
||
// MaxValidityDays is the maximum allowed validity days for subscriptions (100 years)
|
||
const MaxValidityDays = 36500
|
||
|
||
var (
|
||
ErrSubscriptionNotFound = infraerrors.NotFound("SUBSCRIPTION_NOT_FOUND", "subscription not found")
|
||
ErrSubscriptionExpired = infraerrors.Forbidden("SUBSCRIPTION_EXPIRED", "subscription has expired")
|
||
ErrSubscriptionSuspended = infraerrors.Forbidden("SUBSCRIPTION_SUSPENDED", "subscription is suspended")
|
||
ErrSubscriptionAlreadyExists = infraerrors.Conflict("SUBSCRIPTION_ALREADY_EXISTS", "subscription already exists for this user and group")
|
||
ErrGroupNotSubscriptionType = infraerrors.BadRequest("GROUP_NOT_SUBSCRIPTION_TYPE", "group is not a subscription type")
|
||
ErrDailyLimitExceeded = infraerrors.TooManyRequests("DAILY_LIMIT_EXCEEDED", "daily usage limit exceeded")
|
||
ErrWeeklyLimitExceeded = infraerrors.TooManyRequests("WEEKLY_LIMIT_EXCEEDED", "weekly usage limit exceeded")
|
||
ErrMonthlyLimitExceeded = infraerrors.TooManyRequests("MONTHLY_LIMIT_EXCEEDED", "monthly usage limit exceeded")
|
||
ErrSubscriptionNilInput = infraerrors.BadRequest("SUBSCRIPTION_NIL_INPUT", "subscription input cannot be nil")
|
||
ErrAdjustWouldExpire = infraerrors.BadRequest("ADJUST_WOULD_EXPIRE", "adjustment would result in expired subscription (remaining days must be > 0)")
|
||
)
|
||
|
||
// SubscriptionService 订阅服务
|
||
type SubscriptionService struct {
|
||
groupRepo GroupRepository
|
||
userSubRepo UserSubscriptionRepository
|
||
billingCacheService *BillingCacheService
|
||
|
||
// L1 缓存：加速中间件热路径的订阅查询
|
||
subCacheL1 *ristretto.Cache
|
||
subCacheGroup singleflight.Group
|
||
subCacheTTL time.Duration
|
||
subCacheJitter int // 抖动百分比
|
||
}
|
||
|
||
// NewSubscriptionService 创建订阅服务
|
||
func NewSubscriptionService(groupRepo GroupRepository, userSubRepo UserSubscriptionRepository, billingCacheService *BillingCacheService, cfg *config.Config) *SubscriptionService {
|
||
svc := &SubscriptionService{
|
||
groupRepo: groupRepo,
|
||
userSubRepo: userSubRepo,
|
||
billingCacheService: billingCacheService,
|
||
}
|
||
svc.initSubCache(cfg)
|
||
return svc
|
||
}
|
||
|
||
// initSubCache 初始化订阅 L1 缓存
|
||
func (s *SubscriptionService) initSubCache(cfg *config.Config) {
|
||
if cfg == nil {
|
||
return
|
||
}
|
||
sc := cfg.SubscriptionCache
|
||
if sc.L1Size <= 0 || sc.L1TTLSeconds <= 0 {
|
||
return
|
||
}
|
||
cache, err := ristretto.NewCache(&ristretto.Config{
|
||
NumCounters: int64(sc.L1Size) * 10,
|
||
MaxCost: int64(sc.L1Size),
|
||
BufferItems: 64,
|
||
})
|
||
if err != nil {
|
||
log.Printf("Warning: failed to init subscription L1 cache: %v", err)
|
||
return
|
||
}
|
||
s.subCacheL1 = cache
|
||
s.subCacheTTL = time.Duration(sc.L1TTLSeconds) * time.Second
|
||
s.subCacheJitter = sc.JitterPercent
|
||
}
|
||
|
||
// subCacheKey 生成订阅缓存 key（热路径，避免 fmt.Sprintf 开销）
|
||
func subCacheKey(userID, groupID int64) string {
|
||
return "sub:" + strconv.FormatInt(userID, 10) + ":" + strconv.FormatInt(groupID, 10)
|
||
}
|
||
|
||
// jitteredTTL 为 TTL 添加抖动，避免集中过期
|
||
func (s *SubscriptionService) jitteredTTL(ttl time.Duration) time.Duration {
|
||
if ttl <= 0 || s.subCacheJitter <= 0 {
|
||
return ttl
|
||
}
|
||
pct := s.subCacheJitter
|
||
if pct > 100 {
|
||
pct = 100
|
||
}
|
||
delta := float64(pct) / 100
|
||
factor := 1 - delta + rand.Float64()*(2*delta)
|
||
if factor <= 0 {
|
||
return ttl
|
||
}
|
||
return time.Duration(float64(ttl) * factor)
|
||
}
|
||
|
||
// InvalidateSubCache 失效指定用户+分组的订阅 L1 缓存
|
||
func (s *SubscriptionService) InvalidateSubCache(userID, groupID int64) {
|
||
if s.subCacheL1 == nil {
|
||
return
|
||
}
|
||
s.subCacheL1.Del(subCacheKey(userID, groupID))
|
||
}
|
||
|
||
// AssignSubscriptionInput 分配订阅输入
|
||
type AssignSubscriptionInput struct {
|
||
UserID int64
|
||
GroupID int64
|
||
ValidityDays int
|
||
AssignedBy int64
|
||
Notes string
|
||
}
|
||
|
||
// AssignSubscription 分配订阅给用户（不允许重复分配）
|
||
func (s *SubscriptionService) AssignSubscription(ctx context.Context, input *AssignSubscriptionInput) (*UserSubscription, error) {
|
||
// 检查分组是否存在且为订阅类型
|
||
group, err := s.groupRepo.GetByID(ctx, input.GroupID)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("group not found: %w", err)
|
||
}
|
||
if !group.IsSubscriptionType() {
|
||
return nil, ErrGroupNotSubscriptionType
|
||
}
|
||
|
||
// 检查是否已存在订阅
|
||
exists, err := s.userSubRepo.ExistsByUserIDAndGroupID(ctx, input.UserID, input.GroupID)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if exists {
|
||
return nil, ErrSubscriptionAlreadyExists
|
||
}
|
||
|
||
sub, err := s.createSubscription(ctx, input)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
// 失效订阅缓存
|
||
s.InvalidateSubCache(input.UserID, input.GroupID)
|
||
if s.billingCacheService != nil {
|
||
userID, groupID := input.UserID, input.GroupID
|
||
go func() {
|
||
cacheCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||
defer cancel()
|
||
_ = s.billingCacheService.InvalidateSubscription(cacheCtx, userID, groupID)
|
||
}()
|
||
}
|
||
|
||
return sub, nil
|
||
}
|
||
|
||
// AssignOrExtendSubscription 分配或续期订阅（用于兑换码等场景）
|
||
// 如果用户已有同分组的订阅：
|
||
// - 未过期：从当前过期时间累加天数
|
||
// - 已过期：从当前时间开始计算新的过期时间，并激活订阅
|
||
//
|
||
// 如果没有订阅：创建新订阅
|
||
func (s *SubscriptionService) AssignOrExtendSubscription(ctx context.Context, input *AssignSubscriptionInput) (*UserSubscription, bool, error) {
|
||
// 检查分组是否存在且为订阅类型
|
||
group, err := s.groupRepo.GetByID(ctx, input.GroupID)
|
||
if err != nil {
|
||
return nil, false, fmt.Errorf("group not found: %w", err)
|
||
}
|
||
if !group.IsSubscriptionType() {
|
||
return nil, false, ErrGroupNotSubscriptionType
|
||
}
|
||
|
||
// 查询是否已有订阅
|
||
existingSub, err := s.userSubRepo.GetByUserIDAndGroupID(ctx, input.UserID, input.GroupID)
|
||
if err != nil {
|
||
// 不存在记录是正常情况，其他错误需要返回
|
||
existingSub = nil
|
||
}
|
||
|
||
validityDays := input.ValidityDays
|
||
if validityDays <= 0 {
|
||
validityDays = 30
|
||
}
|
||
if validityDays > MaxValidityDays {
|
||
validityDays = MaxValidityDays
|
||
}
|
||
|
||
// 已有订阅，执行续期
|
||
if existingSub != nil {
|
||
now := time.Now()
|
||
var newExpiresAt time.Time
|
||
|
||
if existingSub.ExpiresAt.After(now) {
|
||
// 未过期：从当前过期时间累加
|
||
newExpiresAt = existingSub.ExpiresAt.AddDate(0, 0, validityDays)
|
||
} else {
|
||
// 已过期：从当前时间开始计算
|
||
newExpiresAt = now.AddDate(0, 0, validityDays)
|
||
}
|
||
|
||
// 确保不超过最大过期时间
|
||
if newExpiresAt.After(MaxExpiresAt) {
|
||
newExpiresAt = MaxExpiresAt
|
||
}
|
||
|
||
// 更新过期时间
|
||
if err := s.userSubRepo.ExtendExpiry(ctx, existingSub.ID, newExpiresAt); err != nil {
|
||
return nil, false, fmt.Errorf("extend subscription: %w", err)
|
||
}
|
||
|
||
// 如果订阅已过期或被暂停，恢复为active状态
|
||
if existingSub.Status != SubscriptionStatusActive {
|
||
if err := s.userSubRepo.UpdateStatus(ctx, existingSub.ID, SubscriptionStatusActive); err != nil {
|
||
return nil, false, fmt.Errorf("update subscription status: %w", err)
|
||
}
|
||
}
|
||
|
||
// 追加备注
|
||
if input.Notes != "" {
|
||
newNotes := existingSub.Notes
|
||
if newNotes != "" {
|
||
newNotes += "\n"
|
||
}
|
||
newNotes += input.Notes
|
||
if err := s.userSubRepo.UpdateNotes(ctx, existingSub.ID, newNotes); err != nil {
|
||
log.Printf("update subscription notes failed: sub_id=%d err=%v", existingSub.ID, err)
|
||
}
|
||
}
|
||
|
||
// 失效订阅缓存
|
||
s.InvalidateSubCache(input.UserID, input.GroupID)
|
||
if s.billingCacheService != nil {
|
||
userID, groupID := input.UserID, input.GroupID
|
||
go func() {
|
||
cacheCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||
defer cancel()
|
||
_ = s.billingCacheService.InvalidateSubscription(cacheCtx, userID, groupID)
|
||
}()
|
||
}
|
||
|
||
// 返回更新后的订阅
|
||
sub, err := s.userSubRepo.GetByID(ctx, existingSub.ID)
|
||
return sub, true, err // true 表示是续期
|
||
}
|
||
|
||
// 没有订阅，创建新订阅
|
||
sub, err := s.createSubscription(ctx, input)
|
||
if err != nil {
|
||
return nil, false, err
|
||
}
|
||
|
||
// 失效订阅缓存
|
||
s.InvalidateSubCache(input.UserID, input.GroupID)
|
||
if s.billingCacheService != nil {
|
||
userID, groupID := input.UserID, input.GroupID
|
||
go func() {
|
||
cacheCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||
defer cancel()
|
||
_ = s.billingCacheService.InvalidateSubscription(cacheCtx, userID, groupID)
|
||
}()
|
||
}
|
||
|
||
return sub, false, nil // false 表示是新建
|
||
}
|
||
|
||
// createSubscription 创建新订阅（内部方法）
|
||
func (s *SubscriptionService) createSubscription(ctx context.Context, input *AssignSubscriptionInput) (*UserSubscription, error) {
|
||
validityDays := input.ValidityDays
|
||
if validityDays <= 0 {
|
||
validityDays = 30
|
||
}
|
||
if validityDays > MaxValidityDays {
|
||
validityDays = MaxValidityDays
|
||
}
|
||
|
||
now := time.Now()
|
||
expiresAt := now.AddDate(0, 0, validityDays)
|
||
if expiresAt.After(MaxExpiresAt) {
|
||
expiresAt = MaxExpiresAt
|
||
}
|
||
|
||
sub := &UserSubscription{
|
||
UserID: input.UserID,
|
||
GroupID: input.GroupID,
|
||
StartsAt: now,
|
||
ExpiresAt: expiresAt,
|
||
Status: SubscriptionStatusActive,
|
||
AssignedAt: now,
|
||
Notes: input.Notes,
|
||
CreatedAt: now,
|
||
UpdatedAt: now,
|
||
}
|
||
// 只有当 AssignedBy > 0 时才设置（0 表示系统分配，如兑换码）
|
||
if input.AssignedBy > 0 {
|
||
sub.AssignedBy = &input.AssignedBy
|
||
}
|
||
|
||
if err := s.userSubRepo.Create(ctx, sub); err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
// 重新获取完整订阅信息（包含关联）
|
||
return s.userSubRepo.GetByID(ctx, sub.ID)
|
||
}
|
||
|
||
// BulkAssignSubscriptionInput 批量分配订阅输入
|
||
type BulkAssignSubscriptionInput struct {
|
||
UserIDs []int64
|
||
GroupID int64
|
||
ValidityDays int
|
||
AssignedBy int64
|
||
Notes string
|
||
}
|
||
|
||
// BulkAssignResult 批量分配结果
|
||
type BulkAssignResult struct {
|
||
SuccessCount int
|
||
FailedCount int
|
||
Subscriptions []UserSubscription
|
||
Errors []string
|
||
}
|
||
|
||
// BulkAssignSubscription 批量分配订阅
|
||
func (s *SubscriptionService) BulkAssignSubscription(ctx context.Context, input *BulkAssignSubscriptionInput) (*BulkAssignResult, error) {
|
||
result := &BulkAssignResult{
|
||
Subscriptions: make([]UserSubscription, 0),
|
||
Errors: make([]string, 0),
|
||
}
|
||
|
||
for _, userID := range input.UserIDs {
|
||
sub, err := s.AssignSubscription(ctx, &AssignSubscriptionInput{
|
||
UserID: userID,
|
||
GroupID: input.GroupID,
|
||
ValidityDays: input.ValidityDays,
|
||
AssignedBy: input.AssignedBy,
|
||
Notes: input.Notes,
|
||
})
|
||
if err != nil {
|
||
result.FailedCount++
|
||
result.Errors = append(result.Errors, fmt.Sprintf("user %d: %v", userID, err))
|
||
} else {
|
||
result.SuccessCount++
|
||
result.Subscriptions = append(result.Subscriptions, *sub)
|
||
}
|
||
}
|
||
|
||
return result, nil
|
||
}
|
||
|
||
// RevokeSubscription 撤销订阅
|
||
func (s *SubscriptionService) RevokeSubscription(ctx context.Context, subscriptionID int64) error {
|
||
// 先获取订阅信息用于失效缓存
|
||
sub, err := s.userSubRepo.GetByID(ctx, subscriptionID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
if err := s.userSubRepo.Delete(ctx, subscriptionID); err != nil {
|
||
return err
|
||
}
|
||
|
||
// 失效订阅缓存
|
||
s.InvalidateSubCache(sub.UserID, sub.GroupID)
|
||
if s.billingCacheService != nil {
|
||
userID, groupID := sub.UserID, sub.GroupID
|
||
go func() {
|
||
cacheCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||
defer cancel()
|
||
_ = s.billingCacheService.InvalidateSubscription(cacheCtx, userID, groupID)
|
||
}()
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
// ExtendSubscription 调整订阅时长（正数延长，负数缩短）
|
||
func (s *SubscriptionService) ExtendSubscription(ctx context.Context, subscriptionID int64, days int) (*UserSubscription, error) {
|
||
sub, err := s.userSubRepo.GetByID(ctx, subscriptionID)
|
||
if err != nil {
|
||
return nil, ErrSubscriptionNotFound
|
||
}
|
||
|
||
// 限制调整天数范围
|
||
if days > MaxValidityDays {
|
||
days = MaxValidityDays
|
||
}
|
||
if days < -MaxValidityDays {
|
||
days = -MaxValidityDays
|
||
}
|
||
|
||
now := time.Now()
|
||
isExpired := !sub.ExpiresAt.After(now)
|
||
|
||
// 如果订阅已过期，不允许负向调整
|
||
if isExpired && days < 0 {
|
||
return nil, infraerrors.BadRequest("CANNOT_SHORTEN_EXPIRED", "cannot shorten an expired subscription")
|
||
}
|
||
|
||
// 计算新的过期时间
|
||
var newExpiresAt time.Time
|
||
if isExpired {
|
||
// 已过期：从当前时间开始增加天数
|
||
newExpiresAt = now.AddDate(0, 0, days)
|
||
} else {
|
||
// 未过期：从原过期时间增加/减少天数
|
||
newExpiresAt = sub.ExpiresAt.AddDate(0, 0, days)
|
||
}
|
||
|
||
if newExpiresAt.After(MaxExpiresAt) {
|
||
newExpiresAt = MaxExpiresAt
|
||
}
|
||
|
||
// 检查新的过期时间必须大于当前时间
|
||
if !newExpiresAt.After(now) {
|
||
return nil, ErrAdjustWouldExpire
|
||
}
|
||
|
||
if err := s.userSubRepo.ExtendExpiry(ctx, subscriptionID, newExpiresAt); err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
// 如果订阅已过期，恢复为active状态
|
||
if sub.Status == SubscriptionStatusExpired {
|
||
if err := s.userSubRepo.UpdateStatus(ctx, subscriptionID, SubscriptionStatusActive); err != nil {
|
||
return nil, err
|
||
}
|
||
}
|
||
|
||
// 失效订阅缓存
|
||
s.InvalidateSubCache(sub.UserID, sub.GroupID)
|
||
if s.billingCacheService != nil {
|
||
userID, groupID := sub.UserID, sub.GroupID
|
||
go func() {
|
||
cacheCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||
defer cancel()
|
||
_ = s.billingCacheService.InvalidateSubscription(cacheCtx, userID, groupID)
|
||
}()
|
||
}
|
||
|
||
return s.userSubRepo.GetByID(ctx, subscriptionID)
|
||
}
|
||
|
||
// GetByID 根据ID获取订阅
|
||
func (s *SubscriptionService) GetByID(ctx context.Context, id int64) (*UserSubscription, error) {
|
||
return s.userSubRepo.GetByID(ctx, id)
|
||
}
|
||
|
||
// GetActiveSubscription 获取用户对特定分组的有效订阅
|
||
// 使用 L1 缓存 + singleflight 加速中间件热路径。
|
||
// 返回缓存对象的浅拷贝，调用方可安全修改字段而不会污染缓存或触发 data race。
|
||
func (s *SubscriptionService) GetActiveSubscription(ctx context.Context, userID, groupID int64) (*UserSubscription, error) {
|
||
key := subCacheKey(userID, groupID)
|
||
|
||
// L1 缓存命中：返回浅拷贝
|
||
if s.subCacheL1 != nil {
|
||
if v, ok := s.subCacheL1.Get(key); ok {
|
||
if sub, ok := v.(*UserSubscription); ok {
|
||
cp := *sub
|
||
return &cp, nil
|
||
}
|
||
}
|
||
}
|
||
|
||
// singleflight 防止并发击穿
|
||
value, err, _ := s.subCacheGroup.Do(key, func() (any, error) {
|
||
sub, err := s.userSubRepo.GetActiveByUserIDAndGroupID(ctx, userID, groupID)
|
||
if err != nil {
|
||
return nil, ErrSubscriptionNotFound
|
||
}
|
||
// 写入 L1 缓存
|
||
if s.subCacheL1 != nil {
|
||
_ = s.subCacheL1.SetWithTTL(key, sub, 1, s.jitteredTTL(s.subCacheTTL))
|
||
}
|
||
return sub, nil
|
||
})
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
// singleflight 返回的也是缓存指针，需要浅拷贝
|
||
cp := *value.(*UserSubscription)
|
||
return &cp, nil
|
||
}
|
||
|
||
// ListUserSubscriptions 获取用户的所有订阅
|
||
func (s *SubscriptionService) ListUserSubscriptions(ctx context.Context, userID int64) ([]UserSubscription, error) {
|
||
subs, err := s.userSubRepo.ListByUserID(ctx, userID)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
normalizeExpiredWindows(subs)
|
||
normalizeSubscriptionStatus(subs)
|
||
return subs, nil
|
||
}
|
||
|
||
// ListActiveUserSubscriptions 获取用户的所有有效订阅
|
||
func (s *SubscriptionService) ListActiveUserSubscriptions(ctx context.Context, userID int64) ([]UserSubscription, error) {
|
||
subs, err := s.userSubRepo.ListActiveByUserID(ctx, userID)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
normalizeExpiredWindows(subs)
|
||
return subs, nil
|
||
}
|
||
|
||
// ListGroupSubscriptions 获取分组的所有订阅
|
||
func (s *SubscriptionService) ListGroupSubscriptions(ctx context.Context, groupID int64, page, pageSize int) ([]UserSubscription, *pagination.PaginationResult, error) {
|
||
params := pagination.PaginationParams{Page: page, PageSize: pageSize}
|
||
subs, pag, err := s.userSubRepo.ListByGroupID(ctx, groupID, params)
|
||
if err != nil {
|
||
return nil, nil, err
|
||
}
|
||
normalizeExpiredWindows(subs)
|
||
normalizeSubscriptionStatus(subs)
|
||
return subs, pag, nil
|
||
}
|
||
|
||
// List 获取所有订阅（分页，支持筛选和排序）
|
||
func (s *SubscriptionService) List(ctx context.Context, page, pageSize int, userID, groupID *int64, status, sortBy, sortOrder string) ([]UserSubscription, *pagination.PaginationResult, error) {
|
||
params := pagination.PaginationParams{Page: page, PageSize: pageSize}
|
||
subs, pag, err := s.userSubRepo.List(ctx, params, userID, groupID, status, sortBy, sortOrder)
|
||
if err != nil {
|
||
return nil, nil, err
|
||
}
|
||
normalizeExpiredWindows(subs)
|
||
normalizeSubscriptionStatus(subs)
|
||
return subs, pag, nil
|
||
}
|
||
|
||
// normalizeExpiredWindows 将已过期窗口的数据清零（仅影响返回数据，不影响数据库）
|
||
// 这确保前端显示正确的当前窗口状态，而不是过期窗口的历史数据
|
||
func normalizeExpiredWindows(subs []UserSubscription) {
|
||
for i := range subs {
|
||
sub := &subs[i]
|
||
// 日窗口过期：清零展示数据
|
||
if sub.NeedsDailyReset() {
|
||
sub.DailyWindowStart = nil
|
||
sub.DailyUsageUSD = 0
|
||
}
|
||
// 周窗口过期：清零展示数据
|
||
if sub.NeedsWeeklyReset() {
|
||
sub.WeeklyWindowStart = nil
|
||
sub.WeeklyUsageUSD = 0
|
||
}
|
||
// 月窗口过期：清零展示数据
|
||
if sub.NeedsMonthlyReset() {
|
||
sub.MonthlyWindowStart = nil
|
||
sub.MonthlyUsageUSD = 0
|
||
}
|
||
}
|
||
}
|
||
|
||
// normalizeSubscriptionStatus 根据实际过期时间修正状态（仅影响返回数据，不影响数据库）
|
||
// 这确保前端显示正确的状态，即使定时任务尚未更新数据库
|
||
func normalizeSubscriptionStatus(subs []UserSubscription) {
|
||
now := time.Now()
|
||
for i := range subs {
|
||
sub := &subs[i]
|
||
if sub.Status == SubscriptionStatusActive && !sub.ExpiresAt.After(now) {
|
||
sub.Status = SubscriptionStatusExpired
|
||
}
|
||
}
|
||
}
|
||
|
||
// startOfDay 返回给定时间所在日期的零点（保持原时区）
|
||
func startOfDay(t time.Time) time.Time {
|
||
return time.Date(t.Year(), t.Month(), t.Day(), 0, 0, 0, 0, t.Location())
|
||
}
|
||
|
||
// CheckAndActivateWindow 检查并激活窗口（首次使用时）
|
||
func (s *SubscriptionService) CheckAndActivateWindow(ctx context.Context, sub *UserSubscription) error {
|
||
if sub.IsWindowActivated() {
|
||
return nil
|
||
}
|
||
|
||
// 使用当天零点作为窗口起始时间
|
||
windowStart := startOfDay(time.Now())
|
||
return s.userSubRepo.ActivateWindows(ctx, sub.ID, windowStart)
|
||
}
|
||
|
||
// CheckAndResetWindows 检查并重置过期的窗口
|
||
func (s *SubscriptionService) CheckAndResetWindows(ctx context.Context, sub *UserSubscription) error {
|
||
// 使用当天零点作为新窗口起始时间
|
||
windowStart := startOfDay(time.Now())
|
||
needsInvalidateCache := false
|
||
|
||
// 日窗口重置（24小时）
|
||
if sub.NeedsDailyReset() {
|
||
if err := s.userSubRepo.ResetDailyUsage(ctx, sub.ID, windowStart); err != nil {
|
||
return err
|
||
}
|
||
sub.DailyWindowStart = &windowStart
|
||
sub.DailyUsageUSD = 0
|
||
needsInvalidateCache = true
|
||
}
|
||
|
||
// 周窗口重置（7天）
|
||
if sub.NeedsWeeklyReset() {
|
||
if err := s.userSubRepo.ResetWeeklyUsage(ctx, sub.ID, windowStart); err != nil {
|
||
return err
|
||
}
|
||
sub.WeeklyWindowStart = &windowStart
|
||
sub.WeeklyUsageUSD = 0
|
||
needsInvalidateCache = true
|
||
}
|
||
|
||
// 月窗口重置（30天）
|
||
if sub.NeedsMonthlyReset() {
|
||
if err := s.userSubRepo.ResetMonthlyUsage(ctx, sub.ID, windowStart); err != nil {
|
||
return err
|
||
}
|
||
sub.MonthlyWindowStart = &windowStart
|
||
sub.MonthlyUsageUSD = 0
|
||
needsInvalidateCache = true
|
||
}
|
||
|
||
// 如果有窗口被重置，失效缓存以保持一致性
|
||
if needsInvalidateCache {
|
||
s.InvalidateSubCache(sub.UserID, sub.GroupID)
|
||
if s.billingCacheService != nil {
|
||
_ = s.billingCacheService.InvalidateSubscription(ctx, sub.UserID, sub.GroupID)
|
||
}
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
// CheckUsageLimits 检查使用限额（返回错误如果超限）
|
||
// 用于中间件的快速预检查，additionalCost 通常为 0
|
||
func (s *SubscriptionService) CheckUsageLimits(ctx context.Context, sub *UserSubscription, group *Group, additionalCost float64) error {
|
||
if !sub.CheckDailyLimit(group, additionalCost) {
|
||
return ErrDailyLimitExceeded
|
||
}
|
||
if !sub.CheckWeeklyLimit(group, additionalCost) {
|
||
return ErrWeeklyLimitExceeded
|
||
}
|
||
if !sub.CheckMonthlyLimit(group, additionalCost) {
|
||
return ErrMonthlyLimitExceeded
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// ValidateAndCheckLimits 合并验证+限额检查（中间件热路径专用）
|
||
// 仅做内存检查，不触发 DB 写入。窗口重置的 DB 写入由 DoWindowMaintenance 异步完成。
|
||
// 返回 needsMaintenance 表示是否需要异步执行窗口维护。
|
||
func (s *SubscriptionService) ValidateAndCheckLimits(sub *UserSubscription, group *Group) (needsMaintenance bool, err error) {
|
||
// 1. 验证订阅状态
|
||
if sub.Status == SubscriptionStatusExpired {
|
||
return false, ErrSubscriptionExpired
|
||
}
|
||
if sub.Status == SubscriptionStatusSuspended {
|
||
return false, ErrSubscriptionSuspended
|
||
}
|
||
if sub.IsExpired() {
|
||
return false, ErrSubscriptionExpired
|
||
}
|
||
|
||
// 2. 内存中修正过期窗口的用量，确保 CheckUsageLimits 不会误拒绝用户
|
||
// 实际的 DB 窗口重置由 DoWindowMaintenance 异步完成
|
||
if sub.NeedsDailyReset() {
|
||
sub.DailyUsageUSD = 0
|
||
needsMaintenance = true
|
||
}
|
||
if sub.NeedsWeeklyReset() {
|
||
sub.WeeklyUsageUSD = 0
|
||
needsMaintenance = true
|
||
}
|
||
if sub.NeedsMonthlyReset() {
|
||
sub.MonthlyUsageUSD = 0
|
||
needsMaintenance = true
|
||
}
|
||
if !sub.IsWindowActivated() {
|
||
needsMaintenance = true
|
||
}
|
||
|
||
// 3. 检查用量限额
|
||
if !sub.CheckDailyLimit(group, 0) {
|
||
return needsMaintenance, ErrDailyLimitExceeded
|
||
}
|
||
if !sub.CheckWeeklyLimit(group, 0) {
|
||
return needsMaintenance, ErrWeeklyLimitExceeded
|
||
}
|
||
if !sub.CheckMonthlyLimit(group, 0) {
|
||
return needsMaintenance, ErrMonthlyLimitExceeded
|
||
}
|
||
|
||
return needsMaintenance, nil
|
||
}
|
||
|
||
// DoWindowMaintenance 异步执行窗口维护（激活+重置）
|
||
// 使用独立 context，不受请求取消影响。
|
||
// 注意：此方法仅在 ValidateAndCheckLimits 返回 needsMaintenance=true 时调用，
|
||
// 而 IsExpired()=true 的订阅在 ValidateAndCheckLimits 中已被拦截返回错误，
|
||
// 因此进入此方法的订阅一定未过期，无需处理过期状态同步。
|
||
func (s *SubscriptionService) DoWindowMaintenance(sub *UserSubscription) {
|
||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||
defer cancel()
|
||
|
||
// 激活窗口（首次使用时）
|
||
if !sub.IsWindowActivated() {
|
||
if err := s.CheckAndActivateWindow(ctx, sub); err != nil {
|
||
log.Printf("Failed to activate subscription windows: %v", err)
|
||
}
|
||
}
|
||
|
||
// 重置过期窗口
|
||
if err := s.CheckAndResetWindows(ctx, sub); err != nil {
|
||
log.Printf("Failed to reset subscription windows: %v", err)
|
||
}
|
||
|
||
// 失效 L1 缓存，确保后续请求拿到更新后的数据
|
||
s.InvalidateSubCache(sub.UserID, sub.GroupID)
|
||
}
|
||
|
||
// RecordUsage 记录使用量到订阅
|
||
func (s *SubscriptionService) RecordUsage(ctx context.Context, subscriptionID int64, costUSD float64) error {
|
||
return s.userSubRepo.IncrementUsage(ctx, subscriptionID, costUSD)
|
||
}
|
||
|
||
// SubscriptionProgress 订阅进度
|
||
type SubscriptionProgress struct {
|
||
ID int64 `json:"id"`
|
||
GroupName string `json:"group_name"`
|
||
ExpiresAt time.Time `json:"expires_at"`
|
||
ExpiresInDays int `json:"expires_in_days"`
|
||
Daily *UsageWindowProgress `json:"daily,omitempty"`
|
||
Weekly *UsageWindowProgress `json:"weekly,omitempty"`
|
||
Monthly *UsageWindowProgress `json:"monthly,omitempty"`
|
||
}
|
||
|
||
// UsageWindowProgress 使用窗口进度
|
||
type UsageWindowProgress struct {
|
||
LimitUSD float64 `json:"limit_usd"`
|
||
UsedUSD float64 `json:"used_usd"`
|
||
RemainingUSD float64 `json:"remaining_usd"`
|
||
Percentage float64 `json:"percentage"`
|
||
WindowStart time.Time `json:"window_start"`
|
||
ResetsAt time.Time `json:"resets_at"`
|
||
ResetsInSeconds int64 `json:"resets_in_seconds"`
|
||
}
|
||
|
||
// GetSubscriptionProgress 获取订阅使用进度
|
||
func (s *SubscriptionService) GetSubscriptionProgress(ctx context.Context, subscriptionID int64) (*SubscriptionProgress, error) {
|
||
sub, err := s.userSubRepo.GetByID(ctx, subscriptionID)
|
||
if err != nil {
|
||
return nil, ErrSubscriptionNotFound
|
||
}
|
||
|
||
group := sub.Group
|
||
if group == nil {
|
||
group, err = s.groupRepo.GetByID(ctx, sub.GroupID)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
}
|
||
|
||
progress := &SubscriptionProgress{
|
||
ID: sub.ID,
|
||
GroupName: group.Name,
|
||
ExpiresAt: sub.ExpiresAt,
|
||
ExpiresInDays: sub.DaysRemaining(),
|
||
}
|
||
|
||
// 日进度
|
||
if group.HasDailyLimit() && sub.DailyWindowStart != nil {
|
||
limit := *group.DailyLimitUSD
|
||
resetsAt := sub.DailyWindowStart.Add(24 * time.Hour)
|
||
progress.Daily = &UsageWindowProgress{
|
||
LimitUSD: limit,
|
||
UsedUSD: sub.DailyUsageUSD,
|
||
RemainingUSD: limit - sub.DailyUsageUSD,
|
||
Percentage: (sub.DailyUsageUSD / limit) * 100,
|
||
WindowStart: *sub.DailyWindowStart,
|
||
ResetsAt: resetsAt,
|
||
ResetsInSeconds: int64(time.Until(resetsAt).Seconds()),
|
||
}
|
||
if progress.Daily.RemainingUSD < 0 {
|
||
progress.Daily.RemainingUSD = 0
|
||
}
|
||
if progress.Daily.Percentage > 100 {
|
||
progress.Daily.Percentage = 100
|
||
}
|
||
if progress.Daily.ResetsInSeconds < 0 {
|
||
progress.Daily.ResetsInSeconds = 0
|
||
}
|
||
}
|
||
|
||
// 周进度
|
||
if group.HasWeeklyLimit() && sub.WeeklyWindowStart != nil {
|
||
limit := *group.WeeklyLimitUSD
|
||
resetsAt := sub.WeeklyWindowStart.Add(7 * 24 * time.Hour)
|
||
progress.Weekly = &UsageWindowProgress{
|
||
LimitUSD: limit,
|
||
UsedUSD: sub.WeeklyUsageUSD,
|
||
RemainingUSD: limit - sub.WeeklyUsageUSD,
|
||
Percentage: (sub.WeeklyUsageUSD / limit) * 100,
|
||
WindowStart: *sub.WeeklyWindowStart,
|
||
ResetsAt: resetsAt,
|
||
ResetsInSeconds: int64(time.Until(resetsAt).Seconds()),
|
||
}
|
||
if progress.Weekly.RemainingUSD < 0 {
|
||
progress.Weekly.RemainingUSD = 0
|
||
}
|
||
if progress.Weekly.Percentage > 100 {
|
||
progress.Weekly.Percentage = 100
|
||
}
|
||
if progress.Weekly.ResetsInSeconds < 0 {
|
||
progress.Weekly.ResetsInSeconds = 0
|
||
}
|
||
}
|
||
|
||
// 月进度
|
||
if group.HasMonthlyLimit() && sub.MonthlyWindowStart != nil {
|
||
limit := *group.MonthlyLimitUSD
|
||
resetsAt := sub.MonthlyWindowStart.Add(30 * 24 * time.Hour)
|
||
progress.Monthly = &UsageWindowProgress{
|
||
LimitUSD: limit,
|
||
UsedUSD: sub.MonthlyUsageUSD,
|
||
RemainingUSD: limit - sub.MonthlyUsageUSD,
|
||
Percentage: (sub.MonthlyUsageUSD / limit) * 100,
|
||
WindowStart: *sub.MonthlyWindowStart,
|
||
ResetsAt: resetsAt,
|
||
ResetsInSeconds: int64(time.Until(resetsAt).Seconds()),
|
||
}
|
||
if progress.Monthly.RemainingUSD < 0 {
|
||
progress.Monthly.RemainingUSD = 0
|
||
}
|
||
if progress.Monthly.Percentage > 100 {
|
||
progress.Monthly.Percentage = 100
|
||
}
|
||
if progress.Monthly.ResetsInSeconds < 0 {
|
||
progress.Monthly.ResetsInSeconds = 0
|
||
}
|
||
}
|
||
|
||
return progress, nil
|
||
}
|
||
|
||
// GetUserSubscriptionsWithProgress 获取用户所有订阅及进度
|
||
func (s *SubscriptionService) GetUserSubscriptionsWithProgress(ctx context.Context, userID int64) ([]SubscriptionProgress, error) {
|
||
subs, err := s.userSubRepo.ListActiveByUserID(ctx, userID)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
progresses := make([]SubscriptionProgress, 0, len(subs))
|
||
for _, sub := range subs {
|
||
progress, err := s.GetSubscriptionProgress(ctx, sub.ID)
|
||
if err != nil {
|
||
continue
|
||
}
|
||
progresses = append(progresses, *progress)
|
||
}
|
||
|
||
return progresses, nil
|
||
}
|
||
|
||
// ValidateSubscription 验证订阅是否有效
|
||
func (s *SubscriptionService) ValidateSubscription(ctx context.Context, sub *UserSubscription) error {
|
||
if sub.Status == SubscriptionStatusExpired {
|
||
return ErrSubscriptionExpired
|
||
}
|
||
if sub.Status == SubscriptionStatusSuspended {
|
||
return ErrSubscriptionSuspended
|
||
}
|
||
if sub.IsExpired() {
|
||
// 更新状态
|
||
_ = s.userSubRepo.UpdateStatus(ctx, sub.ID, SubscriptionStatusExpired)
|
||
return ErrSubscriptionExpired
|
||
}
|
||
return nil
|
||
}
|