243 lines
7.3 KiB
Go
243 lines
7.3 KiB
Go
package service
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"log"
|
|
"strings"
|
|
"time"
|
|
|
|
"affiliate_dash/internal/model"
|
|
"gorm.io/gorm"
|
|
)
|
|
|
|
// rockPurchaseStageMaxBytes 单个阶段原始响应/请求的最大保存体积。
|
|
// 超过后截断并标记 truncated，避免异常大响应撑爆数据库。
|
|
const rockPurchaseStageMaxBytes = 512 * 1024
|
|
|
|
// PurchaseSummary 是最近一次采购的可展示摘要，供手动采购接口回显降级结果。
|
|
type PurchaseSummary struct {
|
|
LogID uint `json:"log_id"`
|
|
TriggerType string `json:"trigger_type"`
|
|
Status string `json:"status"`
|
|
RequestedCount int `json:"requested_count"`
|
|
PlannedCount int `json:"planned_count"`
|
|
ClaimedCount int `json:"claimed_count"`
|
|
ImportedCount int `json:"imported_count"`
|
|
LimitReason string `json:"limit_reason"`
|
|
ErrorMessage string `json:"error_message"`
|
|
}
|
|
|
|
// purchaseRecorder 负责一次采购的日志生命周期：
|
|
// 开始时插入主记录（running），每个上游调用写一条阶段记录，结束时更新主记录。
|
|
// 所有写入失败都只记日志，绝不阻断采购主流程。
|
|
type purchaseRecorder struct {
|
|
db *gorm.DB
|
|
logID uint
|
|
seq int
|
|
requested int
|
|
planned int
|
|
claimed int
|
|
imported int
|
|
limitReason string
|
|
pointsSource string
|
|
stage string
|
|
errMessage string
|
|
degraded bool
|
|
|
|
upstreamAvailable int
|
|
pointsBefore float64
|
|
unitCost float64
|
|
maxByPoints int
|
|
maxByStock int
|
|
packageID *int64
|
|
shareCode string
|
|
}
|
|
|
|
func newPurchaseRecorder(db *gorm.DB, triggerType string, requested int, startedAt time.Time) *purchaseRecorder {
|
|
rec := &purchaseRecorder{db: db, requested: requested}
|
|
entry := model.RockPurchaseLog{
|
|
TriggerType: triggerType,
|
|
Status: "running",
|
|
RequestedCount: requested,
|
|
StartedAt: startedAt,
|
|
CreatedAt: startedAt,
|
|
}
|
|
if err := db.WithContext(context.Background()).Create(&entry).Error; err != nil {
|
|
log.Printf("[RockService] 创建采购日志失败，本次采购不记录流水: %v", err)
|
|
return rec
|
|
}
|
|
rec.logID = entry.ID
|
|
return rec
|
|
}
|
|
|
|
// addStage 记录一次上游调用的原始请求与响应。
|
|
func (r *purchaseRecorder) addStage(stage, method, url string, reqBody []byte, status int, respBody []byte, dur time.Duration, callErr error) {
|
|
if r == nil || r.logID == 0 {
|
|
return
|
|
}
|
|
r.seq++
|
|
responseText, truncated := truncateStageBody(respBody)
|
|
requestText, _ := truncateStageBody(reqBody)
|
|
|
|
errText := ""
|
|
if callErr != nil {
|
|
errText = callErr.Error()
|
|
}
|
|
if errText == "" && status >= 400 {
|
|
errText = fmt.Sprintf("上游返回 HTTP %d", status)
|
|
}
|
|
if errText != "" && r.stage == "" {
|
|
r.stage = stage
|
|
r.errMessage = truncatePurchaseError(errText)
|
|
}
|
|
|
|
entry := model.RockPurchaseStage{
|
|
LogID: r.logID,
|
|
Seq: r.seq,
|
|
Stage: stage,
|
|
Method: method,
|
|
URL: url,
|
|
RequestBody: requestText,
|
|
HTTPStatus: status,
|
|
ResponseBody: responseText,
|
|
Truncated: truncated,
|
|
DurationMs: int(dur.Milliseconds()),
|
|
ErrorMessage: truncatePurchaseError(errText),
|
|
CreatedAt: time.Now(),
|
|
}
|
|
if err := r.db.WithContext(context.Background()).Create(&entry).Error; err != nil {
|
|
log.Printf("[RockService] 写入采购阶段日志失败 stage=%s: %v", stage, err)
|
|
}
|
|
}
|
|
|
|
// markDegraded 标记本次采购虽有产出但存在降级/兜底行为，最终状态记为 partial。
|
|
func (r *purchaseRecorder) markDegraded(reason string) {
|
|
if r == nil {
|
|
return
|
|
}
|
|
r.degraded = true
|
|
if r.errMessage == "" {
|
|
r.errMessage = truncatePurchaseError(reason)
|
|
}
|
|
}
|
|
|
|
// finish 更新主记录并返回摘要。
|
|
func (r *purchaseRecorder) finish(tokens []model.RockToken, runErr error, duration time.Duration) *PurchaseSummary {
|
|
if r == nil {
|
|
return nil
|
|
}
|
|
if r.imported == 0 && len(tokens) > 0 {
|
|
r.imported = len(tokens)
|
|
}
|
|
status := "success"
|
|
switch {
|
|
case r.planned <= 0:
|
|
status = "skipped"
|
|
case runErr != nil:
|
|
status = "failed"
|
|
case r.degraded || r.planned < r.requested || (r.claimed > 0 && r.imported < r.claimed):
|
|
status = "partial"
|
|
}
|
|
if runErr != nil && r.errMessage == "" {
|
|
r.errMessage = truncatePurchaseError(runErr.Error())
|
|
}
|
|
|
|
finished := time.Now()
|
|
summary := &PurchaseSummary{
|
|
LogID: r.logID,
|
|
Status: status,
|
|
RequestedCount: r.requested,
|
|
PlannedCount: r.planned,
|
|
ClaimedCount: r.claimed,
|
|
ImportedCount: r.imported,
|
|
LimitReason: r.limitReason,
|
|
ErrorMessage: r.errMessage,
|
|
}
|
|
if r.logID == 0 {
|
|
return summary
|
|
}
|
|
updates := map[string]interface{}{
|
|
"status": status,
|
|
"planned_count": r.planned,
|
|
"claimed_count": r.claimed,
|
|
"imported_count": r.imported,
|
|
"upstream_available": r.upstreamAvailable,
|
|
"points_before": r.pointsBefore,
|
|
"unit_cost": r.unitCost,
|
|
"max_by_points": r.maxByPoints,
|
|
"max_by_stock": r.maxByStock,
|
|
"limit_reason": r.limitReason,
|
|
"points_source": r.pointsSource,
|
|
"stage": r.stage,
|
|
"error_message": r.errMessage,
|
|
"package_id": r.packageID,
|
|
"share_code": r.shareCode,
|
|
"duration_ms": int(duration.Milliseconds()),
|
|
"finished_at": finished,
|
|
}
|
|
if err := r.db.WithContext(context.Background()).Model(&model.RockPurchaseLog{}).
|
|
Where("id = ?", r.logID).Updates(updates).Error; err != nil {
|
|
log.Printf("[RockService] 更新采购日志失败 id=%d: %v", r.logID, err)
|
|
}
|
|
return summary
|
|
}
|
|
|
|
// truncateStageBody 把原始请求/响应限制在 512KB 内，保证截断后仍是合法 UTF-8。
|
|
func truncateStageBody(raw []byte) (string, bool) {
|
|
if len(raw) == 0 {
|
|
return "", false
|
|
}
|
|
if len(raw) <= rockPurchaseStageMaxBytes {
|
|
return string(raw), false
|
|
}
|
|
cut := raw[:rockPurchaseStageMaxBytes]
|
|
return strings.ToValidUTF8(string(cut), "?"), true
|
|
}
|
|
|
|
// LastPurchaseSummary 返回最近一次采购的摘要（包含降级原因），供手动采购接口回显。
|
|
func (s *RockService) LastPurchaseSummary() *PurchaseSummary {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
return s.lastSummary
|
|
}
|
|
|
|
// ListPurchaseLogs 分页查询采购流水（不返回阶段大字段）。
|
|
func (s *RockService) ListPurchaseLogs(ctx context.Context, page, pageSize int, triggerType, status string) ([]model.RockPurchaseLog, int64, error) {
|
|
if page <= 0 {
|
|
page = 1
|
|
}
|
|
if pageSize <= 0 || pageSize > 200 {
|
|
pageSize = 20
|
|
}
|
|
db := s.db.WithContext(ctx).Model(&model.RockPurchaseLog{})
|
|
if t := strings.TrimSpace(triggerType); t != "" {
|
|
db = db.Where("trigger_type = ?", t)
|
|
}
|
|
if st := strings.TrimSpace(status); st != "" {
|
|
db = db.Where("status = ?", st)
|
|
}
|
|
var total int64
|
|
if err := db.Count(&total).Error; err != nil {
|
|
return nil, 0, err
|
|
}
|
|
var list []model.RockPurchaseLog
|
|
if err := db.Order("id DESC").Offset((page - 1) * pageSize).Limit(pageSize).Find(&list).Error; err != nil {
|
|
return nil, 0, err
|
|
}
|
|
return list, total, nil
|
|
}
|
|
|
|
// GetPurchaseLog 查询单次采购详情，包含全部阶段原始报文。
|
|
func (s *RockService) GetPurchaseLog(ctx context.Context, id uint) (*model.RockPurchaseLog, []model.RockPurchaseStage, error) {
|
|
var entry model.RockPurchaseLog
|
|
if err := s.db.WithContext(ctx).First(&entry, id).Error; err != nil {
|
|
return nil, nil, err
|
|
}
|
|
var stages []model.RockPurchaseStage
|
|
if err := s.db.WithContext(ctx).Where("log_id = ?", id).Order("seq ASC, id ASC").Find(&stages).Error; err != nil {
|
|
return nil, nil, err
|
|
}
|
|
return &entry, stages, nil
|
|
}
|