458 lines
15 KiB
Go
458 lines
15 KiB
Go
package service
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"log"
|
||
"strconv"
|
||
"strings"
|
||
"time"
|
||
|
||
"affiliate_dash/internal/model"
|
||
"gorm.io/gorm"
|
||
)
|
||
|
||
const (
|
||
rockPurchaseNoteAutoPrefix = "补货"
|
||
rockPurchaseNoteManualPrefix = "手动采购"
|
||
)
|
||
|
||
// PurchaseTokens 执行采购并入库。
|
||
// 流程：claim 批量提取 → 创建上游打包记录 → 拉取包详情 → 幂等导入本地（新记录 available）。
|
||
// 后续任一步失败都会用 claim 数据兜底入库，保证已扣积分的 Token 不会丢失。
|
||
func (s *RockService) PurchaseTokens(ctx context.Context, count int, isAuto bool) ([]model.RockToken, error) {
|
||
s.mu.Lock()
|
||
defer s.mu.Unlock()
|
||
|
||
if count <= 0 {
|
||
return nil, fmt.Errorf("采购数量必须大于 0")
|
||
}
|
||
|
||
cfg, err := s.GetConfig(ctx)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if strings.TrimSpace(cfg.BearerToken) == "" {
|
||
return nil, fmt.Errorf("尚未配置上游 Bearer Token，无法执行采购")
|
||
}
|
||
|
||
now := time.Now()
|
||
if isAuto {
|
||
// 冷却以最近一次“尝试”为准：失败也生效，避免上游故障时每个触发点反复请求。
|
||
last := cfg.LastAttemptAt
|
||
if last == nil {
|
||
last = cfg.LastPurchasedAt
|
||
}
|
||
if last != nil && cfg.PurchaseCooldownMinutes > 0 {
|
||
cooldown := time.Duration(cfg.PurchaseCooldownMinutes) * time.Minute
|
||
if now.Sub(*last) < cooldown {
|
||
return nil, fmt.Errorf("自动采购处于冷却中(冷却时间 %d 分钟)", cfg.PurchaseCooldownMinutes)
|
||
}
|
||
}
|
||
}
|
||
s.updatePurchaseAttempt(ctx, cfg.ID, now)
|
||
|
||
// 采购前先刷新上游 dashboard：校验可用库存，同时同步最新采购单价。
|
||
dashboard, dashErr := s.refreshDashboardWithConfig(ctx, cfg)
|
||
if dashErr != nil {
|
||
wrapped := fmt.Errorf("获取上游库存失败，已取消本次采购: %w", dashErr)
|
||
s.recordPurchaseFailure(ctx, cfg.ID, wrapped.Error())
|
||
return nil, wrapped
|
||
}
|
||
if dashboard.AvailableBlack < count {
|
||
err := fmt.Errorf("上游可用库存不足：当前 %d 个，本次需要 %d 个", dashboard.AvailableBlack, count)
|
||
s.recordPurchaseFailure(ctx, cfg.ID, err.Error())
|
||
return nil, err
|
||
}
|
||
|
||
claimed, err := s.client.ClaimTokens(ctx, cfg.BearerToken, model.RockColorBlack, count)
|
||
if err != nil {
|
||
wrapped := fmt.Errorf("调用上游采购接口失败: %w", err)
|
||
s.recordPurchaseFailure(ctx, cfg.ID, wrapped.Error())
|
||
return nil, wrapped
|
||
}
|
||
if len(claimed) == 0 {
|
||
err := errors.New("上游未返回任何有效 Token")
|
||
s.recordPurchaseFailure(ctx, cfg.ID, err.Error())
|
||
return nil, err
|
||
}
|
||
|
||
// 积分已经扣除，立即刷新最后采购时间，后续打包失败也不会被重复采购。
|
||
s.markPurchased(ctx, cfg.ID, now)
|
||
s.refreshPointsAsync(cfg.ID, cfg.BearerToken)
|
||
|
||
ids := make([]string, 0, len(claimed))
|
||
claimedSet := make(map[string]struct{}, len(claimed))
|
||
for _, t := range claimed {
|
||
if t.ID <= 0 {
|
||
continue
|
||
}
|
||
upstreamID := strconv.FormatInt(t.ID, 10)
|
||
ids = append(ids, upstreamID)
|
||
claimedSet[upstreamID] = struct{}{}
|
||
}
|
||
if len(ids) == 0 {
|
||
err := errors.New("上游返回的 Token ID 均无效")
|
||
s.recordPurchaseFailure(ctx, cfg.ID, err.Error())
|
||
return nil, err
|
||
}
|
||
|
||
// 1. 创建上游打包记录
|
||
note := purchaseNote(isAuto, now)
|
||
pkgResp, pkgErr := s.client.CreatePackage(ctx, cfg.BearerToken, ids, note)
|
||
var packageID int64
|
||
shareCode := ""
|
||
if pkgErr == nil && pkgResp != nil {
|
||
packageID, shareCode = pkgResp.ResolvePackage()
|
||
if shareCode == "" {
|
||
pkgErr = errors.New("上游未返回 package.share_code")
|
||
}
|
||
} else if pkgErr == nil {
|
||
pkgErr = errors.New("上游打包接口未返回内容")
|
||
}
|
||
if pkgErr != nil {
|
||
return s.fallbackClaimedTokens(ctx, cfg.ID, claimed, 0, "", "purchase", now,
|
||
"创建上游打包记录失败: "+pkgErr.Error())
|
||
}
|
||
|
||
// 2. 拉取包详情（带重试）
|
||
detail, detailErr := s.getPackageDetailWithRetry(ctx, shareCode)
|
||
if detailErr != nil {
|
||
return s.fallbackClaimedTokens(ctx, cfg.ID, claimed, packageID, shareCode, "package", now,
|
||
fmt.Sprintf("包 %s 详情拉取失败: %v", shareCode, detailErr))
|
||
}
|
||
|
||
// 3. 导入包内 Token：本批刚 claim 的 Token 强制 available，其余沿用上游状态
|
||
inserted, updated, skipped, importErr := s.importPackageDetail(ctx, detail, claimedSet)
|
||
if importErr != nil {
|
||
return s.fallbackClaimedTokens(ctx, cfg.ID, claimed, packageID, shareCode, "package", now,
|
||
fmt.Sprintf("包 %s 导入失败: %v", shareCode, importErr))
|
||
}
|
||
if inserted+updated == 0 {
|
||
return s.fallbackClaimedTokens(ctx, cfg.ID, claimed, packageID, shareCode, "package", now,
|
||
fmt.Sprintf("包 %s 未返回有效 Token", shareCode))
|
||
}
|
||
|
||
// 4. 极端情况下详情缺 Token：用 claim 数据补齐，保证每个已扣积分的 Token 都在本地
|
||
ensureErr := s.ensureClaimedTokens(ctx, claimed, packageID, shareCode, "package", now)
|
||
if ensureErr != nil {
|
||
s.recordPurchaseFailure(ctx, cfg.ID, "补录缺失 Token 失败: "+ensureErr.Error())
|
||
}
|
||
|
||
dbTokens, findErr := s.findTokensByUpstreamIDs(ctx, ids)
|
||
if findErr != nil || len(dbTokens) == 0 {
|
||
if findErr != nil {
|
||
log.Printf("[RockService] 回查采购 Token 失败: %v", findErr)
|
||
}
|
||
dbTokens = s.buildClaimedTokens(claimed, packageID, shareCode, "package", now)
|
||
}
|
||
|
||
if ensureErr == nil {
|
||
s.recordPurchaseSuccess(ctx, cfg.ID)
|
||
}
|
||
log.Printf("[RockService] 采购打包完成: 包=%s 新增=%d 更新=%d 跳过=%d 实际入库=%d",
|
||
shareCode, inserted, updated, skipped, len(dbTokens))
|
||
return dbTokens, nil
|
||
}
|
||
|
||
// fallbackClaimedTokens 在打包/详情/导入失败时，用 claim 返回的数据直接入库，避免积分白花。
|
||
func (s *RockService) fallbackClaimedTokens(ctx context.Context, cfgID uint, claimed []RockClaimedToken, packageID int64, shareCode, source string, at time.Time, reason string) ([]model.RockToken, error) {
|
||
tokens, err := s.insertClaimedTokens(ctx, claimed, packageID, shareCode, source, at)
|
||
if err != nil {
|
||
s.recordPurchaseFailure(ctx, cfgID, reason+"；兜底入库失败: "+err.Error())
|
||
return nil, fmt.Errorf("%s；兜底入库失败: %w", reason, err)
|
||
}
|
||
s.recordPurchaseFailure(ctx, cfgID, fmt.Sprintf("%s；已兜底入库 %d 个 Token", reason, len(tokens)))
|
||
return tokens, nil
|
||
}
|
||
|
||
// insertClaimedTokens 把 claim 结果直接写入库存表。
|
||
func (s *RockService) insertClaimedTokens(ctx context.Context, claimed []RockClaimedToken, packageID int64, shareCode, source string, at time.Time) ([]model.RockToken, error) {
|
||
tokens := s.buildClaimedTokens(claimed, packageID, shareCode, source, at)
|
||
if len(tokens) == 0 {
|
||
return nil, errors.New("没有可入库的 Token")
|
||
}
|
||
if err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||
return tx.Create(&tokens).Error
|
||
}); err != nil {
|
||
return nil, err
|
||
}
|
||
return tokens, nil
|
||
}
|
||
|
||
// ensureClaimedTokens 补录本地缺失的 claim Token（详情返回不完整时使用）。
|
||
func (s *RockService) ensureClaimedTokens(ctx context.Context, claimed []RockClaimedToken, packageID int64, shareCode, source string, at time.Time) error {
|
||
tokens := s.buildClaimedTokens(claimed, packageID, shareCode, source, at)
|
||
if len(tokens) == 0 {
|
||
return nil
|
||
}
|
||
ids := make([]string, 0, len(tokens))
|
||
for _, t := range tokens {
|
||
ids = append(ids, t.UpstreamTokenID)
|
||
}
|
||
|
||
var existing []string
|
||
if err := s.db.WithContext(ctx).Model(&model.RockToken{}).
|
||
Where("upstream_token_id IN ?", ids).
|
||
Pluck("upstream_token_id", &existing).Error; err != nil {
|
||
return err
|
||
}
|
||
seen := make(map[string]bool, len(existing))
|
||
for _, id := range existing {
|
||
seen[id] = true
|
||
}
|
||
missing := make([]model.RockToken, 0, len(tokens))
|
||
for _, t := range tokens {
|
||
if !seen[t.UpstreamTokenID] {
|
||
missing = append(missing, t)
|
||
}
|
||
}
|
||
if len(missing) == 0 {
|
||
return nil
|
||
}
|
||
return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||
return tx.Create(&missing).Error
|
||
})
|
||
}
|
||
|
||
func (s *RockService) buildClaimedTokens(claimed []RockClaimedToken, packageID int64, shareCode, source string, at time.Time) []model.RockToken {
|
||
out := make([]model.RockToken, 0, len(claimed))
|
||
for _, t := range claimed {
|
||
if t.ID <= 0 {
|
||
continue
|
||
}
|
||
raw, err := json.Marshal(t)
|
||
if err != nil {
|
||
raw = []byte(`{}`)
|
||
}
|
||
rec := model.RockToken{
|
||
UpstreamTokenID: strconv.FormatInt(t.ID, 10),
|
||
Token: strings.TrimSpace(t.Token),
|
||
RedeemCode: strings.TrimSpace(t.RedeemCode),
|
||
Color: defaultValue(strings.TrimSpace(t.Color), model.RockColorBlack),
|
||
Region: defaultValue(strings.TrimSpace(t.Region), "Q/V可用"),
|
||
Status: model.RockStatusAvailable,
|
||
RawData: string(raw),
|
||
Source: source,
|
||
ClaimedAt: &at,
|
||
}
|
||
if packageID > 0 {
|
||
pid := packageID
|
||
rec.PackageID = &pid
|
||
rec.PackageShareCode = strings.TrimSpace(shareCode)
|
||
}
|
||
out = append(out, rec)
|
||
}
|
||
return out
|
||
}
|
||
|
||
func (s *RockService) findTokensByUpstreamIDs(ctx context.Context, ids []string) ([]model.RockToken, error) {
|
||
if len(ids) == 0 {
|
||
return nil, nil
|
||
}
|
||
var tokens []model.RockToken
|
||
err := s.db.WithContext(ctx).
|
||
Where("upstream_token_id IN ?", ids).
|
||
Order("id ASC").
|
||
Find(&tokens).Error
|
||
return tokens, err
|
||
}
|
||
|
||
// getPackageDetailWithRetry 拉取包详情，最多 3 次（1s/2s 退避）。
|
||
func (s *RockService) getPackageDetailWithRetry(ctx context.Context, shareCode string) (*RockPackageDetailResponse, error) {
|
||
var lastErr error
|
||
for attempt := 1; attempt <= 3; attempt++ {
|
||
detail, err := s.client.GetPackageDetail(ctx, shareCode)
|
||
if err == nil && detail != nil && detail.Package.ID != 0 {
|
||
return detail, nil
|
||
}
|
||
if err == nil {
|
||
err = errors.New("上游未返回 package.id")
|
||
}
|
||
lastErr = err
|
||
if attempt < 3 {
|
||
select {
|
||
case <-ctx.Done():
|
||
return nil, ctx.Err()
|
||
case <-time.After(time.Duration(attempt) * time.Second):
|
||
}
|
||
}
|
||
}
|
||
return nil, lastErr
|
||
}
|
||
|
||
func purchaseNote(isAuto bool, at time.Time) string {
|
||
prefix := rockPurchaseNoteManualPrefix
|
||
if isAuto {
|
||
prefix = rockPurchaseNoteAutoPrefix
|
||
}
|
||
return prefix + "-" + at.Format("20060102-150405")
|
||
}
|
||
|
||
func (s *RockService) updatePurchaseAttempt(ctx context.Context, cfgID uint, at time.Time) {
|
||
if err := s.db.WithContext(ctx).Model(&model.RockConfig{}).
|
||
Where("id = ?", cfgID).
|
||
Update("last_attempt_at", at).Error; err != nil {
|
||
log.Printf("[RockService] 记录采购尝试时间失败: %v", err)
|
||
}
|
||
}
|
||
|
||
func (s *RockService) markPurchased(ctx context.Context, cfgID uint, at time.Time) {
|
||
if err := s.db.WithContext(ctx).Model(&model.RockConfig{}).
|
||
Where("id = ?", cfgID).
|
||
Update("last_purchased_at", at).Error; err != nil {
|
||
log.Printf("[RockService] 记录最后采购时间失败: %v", err)
|
||
}
|
||
}
|
||
|
||
func (s *RockService) recordPurchaseFailure(ctx context.Context, cfgID uint, msg string) {
|
||
msg = truncatePurchaseError(msg)
|
||
log.Printf("[RockService][ALERT] 补货异常: %s", msg)
|
||
if err := s.db.WithContext(ctx).Model(&model.RockConfig{}).
|
||
Where("id = ?", cfgID).
|
||
Updates(map[string]interface{}{
|
||
"last_error": msg,
|
||
"last_error_at": time.Now(),
|
||
}).Error; err != nil {
|
||
log.Printf("[RockService] 写入补货错误失败: %v", err)
|
||
}
|
||
}
|
||
|
||
func (s *RockService) recordPurchaseSuccess(ctx context.Context, cfgID uint) {
|
||
if err := s.db.WithContext(ctx).Model(&model.RockConfig{}).
|
||
Where("id = ?", cfgID).
|
||
Updates(map[string]interface{}{
|
||
"last_error": "",
|
||
"last_error_at": nil,
|
||
}).Error; err != nil {
|
||
log.Printf("[RockService] 清理补货错误失败: %v", err)
|
||
}
|
||
}
|
||
|
||
func (s *RockService) refreshPointsAsync(cfgID uint, bearerToken string) {
|
||
go func() {
|
||
bgCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||
defer cancel()
|
||
me, meErr := s.client.GetMe(bgCtx, bearerToken)
|
||
if meErr == nil && me != nil {
|
||
_ = s.db.Model(&model.RockConfig{}).Where("id = ?", cfgID).Update("points", me.User.Points.Float64()).Error
|
||
}
|
||
}()
|
||
}
|
||
|
||
func truncatePurchaseError(msg string) string {
|
||
msg = strings.TrimSpace(msg)
|
||
runes := []rune(msg)
|
||
if len(runes) <= 500 {
|
||
return msg
|
||
}
|
||
return string(runes[:500]) + "..."
|
||
}
|
||
|
||
// CheckStockAndAutoPurchase 水位检查与自动采购触发
|
||
func (s *RockService) CheckStockAndAutoPurchase(ctx context.Context) {
|
||
cfg, err := s.GetConfig(ctx)
|
||
if err != nil {
|
||
return
|
||
}
|
||
|
||
var count int64
|
||
if err := s.db.WithContext(ctx).Model(&model.RockToken{}).
|
||
Where("status = ? AND color = ?", model.RockStatusAvailable, model.RockColorBlack).
|
||
Count(&count).Error; err != nil {
|
||
log.Printf("[RockService] 统计可用库存失败: %v", err)
|
||
return
|
||
}
|
||
|
||
if count > int64(cfg.TokenWarningStock) {
|
||
return
|
||
}
|
||
log.Printf("[RockService] 可用库存预警: 当前仅存 %d 个 (预警阈值 %d)", count, cfg.TokenWarningStock)
|
||
if !cfg.AutoPurchaseEnabled {
|
||
return
|
||
}
|
||
|
||
buyCount := cfg.AutoPurchaseCount
|
||
if buyCount <= 0 {
|
||
buyCount = 50
|
||
}
|
||
log.Printf("[RockService] 触发自动补货，采购数量: %d", buyCount)
|
||
tokens, err := s.PurchaseTokens(ctx, buyCount, true)
|
||
if err != nil {
|
||
log.Printf("[RockService] 自动补货失败: %v", err)
|
||
return
|
||
}
|
||
log.Printf("[RockService] 自动补货成功，实际入库 %d 个 Token", len(tokens))
|
||
}
|
||
|
||
// ListTokens 分页查询 Token 列表
|
||
func (s *RockService) ListTokens(ctx context.Context, page, pageSize int, status, search string) ([]model.RockToken, int64, error) {
|
||
if page <= 0 {
|
||
page = 1
|
||
}
|
||
if pageSize <= 0 {
|
||
pageSize = 20
|
||
}
|
||
|
||
db := s.db.WithContext(ctx).Model(&model.RockToken{})
|
||
if strings.TrimSpace(status) != "" {
|
||
db = db.Where("status = ?", strings.TrimSpace(status))
|
||
}
|
||
if strings.TrimSpace(search) != "" {
|
||
term := "%" + strings.TrimSpace(search) + "%"
|
||
db = db.Where("CAST(id AS TEXT) = ? OR upstream_token_id = ? OR upstream_token_id LIKE ? OR assigned_order_no LIKE ? OR expected_uid LIKE ? OR actual_role_name LIKE ? OR actual_role_id LIKE ? OR redeem_code LIKE ? OR remark LIKE ? OR CAST(package_id AS TEXT) LIKE ? OR token LIKE ? OR assigned_order_no IN (SELECT order_no FROM fulfillment_orders WHERE client_order_no LIKE ?)",
|
||
strings.TrimSpace(search), strings.TrimSpace(search), term, term, term, term, term, term, term, term, term, term)
|
||
}
|
||
|
||
var total int64
|
||
if err := db.Count(&total).Error; err != nil {
|
||
return nil, 0, err
|
||
}
|
||
|
||
var tokens []model.RockToken
|
||
// 按包 ID 倒序（未打包的手工采购记录排在最后），同包内按本地 ID 倒序，保证分页稳定
|
||
err := db.Order("package_id DESC NULLS LAST").
|
||
Order("id DESC").
|
||
Offset((page - 1) * pageSize).
|
||
Limit(pageSize).
|
||
Find(&tokens).Error
|
||
if err != nil {
|
||
return nil, 0, err
|
||
}
|
||
for i := range tokens {
|
||
tokens[i].TokenCode = tokens[i].Token
|
||
}
|
||
|
||
// 批量反查关联订单的商户单号（创建订单时的 ClientOrderNo）
|
||
orderNos := make([]string, 0, len(tokens))
|
||
for _, t := range tokens {
|
||
if t.AssignedOrderNo != "" {
|
||
orderNos = append(orderNos, t.AssignedOrderNo)
|
||
}
|
||
}
|
||
if len(orderNos) > 0 {
|
||
var orders []struct {
|
||
OrderNo string `gorm:"column:order_no"`
|
||
ClientOrderNo string `gorm:"column:client_order_no"`
|
||
}
|
||
if scanErr := s.db.WithContext(ctx).Table("fulfillment_orders").
|
||
Select("order_no, client_order_no").
|
||
Where("order_no IN ?", orderNos).
|
||
Scan(&orders).Error; scanErr == nil {
|
||
clientMap := make(map[string]string, len(orders))
|
||
for _, o := range orders {
|
||
clientMap[o.OrderNo] = o.ClientOrderNo
|
||
}
|
||
for i := range tokens {
|
||
if cNo, ok := clientMap[tokens[i].AssignedOrderNo]; ok && cNo != "" {
|
||
tokens[i].ClientOrderNo = cNo
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
return tokens, total, nil
|
||
}
|