affiliate_dash/backend/internal/service/rock_delivery_log.go

package service

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"net/http"
	"strings"
	"sync/atomic"
	"time"

	"affiliate_dash/internal/model"
	"gorm.io/gorm"
)

const (
	// 发货报文日志保留天数：按业务要求原样明文保存 180 天。
	rockDeliveryLogRetentionDays = 180
	// 单个报文字段最大长度，避免异常大响应拖垮数据库；超出部分截断并标注。
	rockDeliveryLogMaxBodyBytes = 512 * 1024
	// 上游 110001 频控后的全局冷却时间。
	rockRateLimitCooldown = 60 * time.Second
)

// rockUpstreamRateLimitedUntil 记录上游频控解禁时间（UnixNano）。
// 买家扫码轮询与后台确认任务共用，避免把 0666.pro 打成持续 110001。
var rockUpstreamRateLimitedUntil atomic.Int64

func markRockUpstreamRateLimited() {
	rockUpstreamRateLimitedUntil.Store(time.Now().Add(rockRateLimitCooldown).UnixNano())
}

func rockUpstreamRateLimitActive() bool {
	until := rockUpstreamRateLimitedUntil.Load()
	return until > 0 && time.Now().UnixNano() < until
}

func isRockRateLimitError(err error) bool {
	if err == nil {
		return false
	}
	msg := err.Error()
	return strings.Contains(msg, "110001") || strings.Contains(msg, "请求过于频繁")
}

// RockDeliveryRecorder 负责把一次发货链路的每一步写入 rock_delivery_logs。
//
// 设计要点：
//   - 上游调用采用「预写日志」：请求前先落 started，拿到响应后更新为最终状态，
//     即使进程中途重启也能看到未完成的调用；
//   - 入站请求（本系统收到的参数）与内部状态迁移也各落一条，前端按 trace_id 串起来；
//   - 写入为 best-effort，失败只打日志，绝不阻断发货主流程。
type RockDeliveryRecorder struct {
	db              *gorm.DB
	TraceID         string
	TokenID         uint
	UpstreamTokenID string
	OrderNo         string
	ActorUserID     *uint
	ActorIP         string
	UserAgent       string
}

// NewDeliveryRecorder 创建一次发货链路的记录器。
func (s *RockService) NewDeliveryRecorder(orderNo string, actorUserID *uint, actorIP, userAgent string) *RockDeliveryRecorder {
	return &RockDeliveryRecorder{
		db:          s.db,
		TraceID:     fmt.Sprintf("rd-%d-%d", time.Now().UnixNano(), time.Now().UnixNano()%100000),
		OrderNo:     strings.TrimSpace(orderNo),
		ActorUserID: actorUserID,
		ActorIP:     strings.TrimSpace(actorIP),
		UserAgent:   strings.TrimSpace(userAgent),
	}
}

// AttachToken 把 Token 身份补到后续日志上（Token 在订单内查询后才能确定）。
func (r *RockDeliveryRecorder) AttachToken(token *model.RockToken) {
	if r == nil || token == nil {
		return
	}
	r.TokenID = token.ID
	r.UpstreamTokenID = strings.TrimSpace(token.UpstreamTokenID)
}

// Inbound 记录本系统收到的一次业务请求（买家提交发货等）。
func (r *RockDeliveryRecorder) Inbound(ctx context.Context, step string, payload interface{}) {
	r.write(ctx, &model.RockDeliveryLog{
		TraceID:         r.traceID(),
		TokenID:         r.TokenID,
		UpstreamTokenID: r.UpstreamTokenID,
		OrderNo:         r.OrderNo,
		Step:            step,
		Direction:       "inbound",
		Status:          "success",
		RequestHeaders:  "",
		RequestBody:     rockJSONText(payload),
		ResponseBody:    "",
		ActorUserID:     r.ActorUserID,
		ActorIP:         r.ActorIP,
		UserAgent:       r.UserAgent,
	})
}

// Internal 记录一个内部步骤（状态迁移、订单更新、回调等）。
func (r *RockDeliveryRecorder) Internal(ctx context.Context, step, status, message string, payload interface{}) {
	if r == nil {
		return
	}
	entry := &model.RockDeliveryLog{
		TraceID:         r.traceID(),
		TokenID:         r.TokenID,
		UpstreamTokenID: r.UpstreamTokenID,
		OrderNo:         r.OrderNo,
		Step:            step,
		Direction:       "internal",
		Status:          status,
		RequestBody:     rockJSONText(payload),
		ErrorMessage:    strings.TrimSpace(message),
		ActorUserID:     r.ActorUserID,
		ActorIP:         r.ActorIP,
		UserAgent:       r.UserAgent,
	}
	now := time.Now()
	entry.FinishedAt = &now
	r.write(ctx, entry)
}

// StartUpstream 在调用上游前预写一条 started 日志，返回的 entry 用于回填响应。
func (r *RockDeliveryRecorder) StartUpstream(ctx context.Context, step, method, reqURL string, headers http.Header, reqBody []byte) *model.RockDeliveryLog {
	if r == nil {
		return nil
	}
	entry := &model.RockDeliveryLog{
		TraceID:         r.traceID(),
		TokenID:         r.TokenID,
		UpstreamTokenID: r.UpstreamTokenID,
		OrderNo:         r.OrderNo,
		Step:            step,
		Direction:       "outbound",
		Status:          "started",
		HTTPMethod:      method,
		HTTPURL:         reqURL,
		RequestHeaders:  rockHeadersText(headers),
		RequestBody:     truncateRockDeliveryBody(reqBody),
		ActorUserID:     r.ActorUserID,
		ActorIP:         r.ActorIP,
		UserAgent:       r.UserAgent,
	}
	if err := r.db.WithContext(ctx).Create(entry).Error; err != nil {
		log.Printf("[Rock] 写入发货请求日志失败 trace=%s step=%s: %v", r.TraceID, step, err)
		return nil
	}
	return entry
}

// FinishUpstream 回填上游响应报文与耗时。
func (r *RockDeliveryRecorder) FinishUpstream(ctx context.Context, entry *model.RockDeliveryLog, status string, resp *http.Response, respBody []byte, callErr error, duration time.Duration) {
	if r == nil || entry == nil {
		return
	}
	updates := map[string]interface{}{
		"status":        status,
		"duration_ms":   int(duration.Milliseconds()),
		"response_body": truncateRockDeliveryBody(respBody),
		"finished_at":   time.Now(),
	}
	if resp != nil {
		updates["http_status"] = resp.StatusCode
		updates["response_headers"] = rockHeadersText(resp.Header)
	}
	if callErr != nil {
		updates["error_message"] = callErr.Error()
	}
	if err := r.db.WithContext(ctx).Model(&model.RockDeliveryLog{}).
		Where("id = ?", entry.ID).
		Updates(updates).Error; err != nil {
		log.Printf("[Rock] 回填发货响应日志失败 trace=%s id=%d: %v", r.TraceID, entry.ID, err)
	}
}

func (r *RockDeliveryRecorder) traceID() string {
	if r == nil || strings.TrimSpace(r.TraceID) == "" {
		return "rd-unknown"
	}
	return r.TraceID
}

func (r *RockDeliveryRecorder) write(ctx context.Context, entry *model.RockDeliveryLog) {
	if r == nil || r.db == nil || entry == nil {
		return
	}
	entry.TraceID = r.traceID()
	if entry.TokenID == 0 {
		entry.TokenID = r.TokenID
	}
	if entry.UpstreamTokenID == "" {
		entry.UpstreamTokenID = r.UpstreamTokenID
	}
	if entry.OrderNo == "" {
		entry.OrderNo = r.OrderNo
	}
	if err := r.db.WithContext(ctx).Create(entry).Error; err != nil {
		log.Printf("[Rock] 写入发货链路日志失败 trace=%s step=%s: %v", entry.TraceID, entry.Step, err)
	}
}

type rockDeliveryRecorderCtxKey struct{}

func withRockDeliveryRecorder(ctx context.Context, recorder *RockDeliveryRecorder) context.Context {
	if recorder == nil {
		return ctx
	}
	return context.WithValue(ctx, rockDeliveryRecorderCtxKey{}, recorder)
}

func rockDeliveryRecorderFrom(ctx context.Context) *RockDeliveryRecorder {
	if ctx == nil {
		return nil
	}
	if recorder, ok := ctx.Value(rockDeliveryRecorderCtxKey{}).(*RockDeliveryRecorder); ok {
		return recorder
	}
	return nil
}

// ListRockDeliveryLogs 返回指定 Token 的完整发货链路（按时间正序，便于前端时间线展示）。
func (s *RockService) ListRockDeliveryLogs(ctx context.Context, tokenID uint) ([]model.RockDeliveryLog, error) {
	var logs []model.RockDeliveryLog
	if err := s.db.WithContext(ctx).
		Where("token_id = ?", tokenID).
		Order("id ASC").
		Limit(500).
		Find(&logs).Error; err != nil {
		return nil, err
	}
	return logs, nil
}

// CleanupRockDeliveryLogs 清理超过保留期的发货报文日志。
func (s *RockService) CleanupRockDeliveryLogs(ctx context.Context) {
	cutoff := time.Now().AddDate(0, 0, -rockDeliveryLogRetentionDays)
	if err := s.db.WithContext(ctx).
		Where("created_at < ?", cutoff).
		Delete(&model.RockDeliveryLog{}).Error; err != nil {
		log.Printf("[Rock] 清理发货报文日志失败: %v", err)
	}
}

func rockHeadersText(headers http.Header) string {
	if len(headers) == 0 {
		return ""
	}
	encoded, err := json.Marshal(headers)
	if err != nil {
		return ""
	}
	return string(encoded)
}

func rockJSONText(value interface{}) string {
	if value == nil {
		return ""
	}
	encoded, err := json.Marshal(value)
	if err != nil {
		return fmt.Sprintf("%v", value)
	}
	return string(encoded)
}

func truncateRockDeliveryBody(body []byte) string {
	if len(body) == 0 {
		return ""
	}
	if len(body) <= rockDeliveryLogMaxBodyBytes {
		return string(body)
	}
	return string(body[:rockDeliveryLogMaxBodyBytes]) + "\n...[truncated]"
}