465 lines
16 KiB
Go
465 lines
16 KiB
Go
package service
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
"time"
|
|
|
|
"affiliate_dash/internal/model"
|
|
|
|
"gorm.io/gorm"
|
|
"gorm.io/gorm/clause"
|
|
)
|
|
|
|
type RockIncidentListResult struct {
|
|
List []model.RockDeliveryIncident `json:"list"`
|
|
Total int64 `json:"total"`
|
|
}
|
|
|
|
type RockIncidentRetryPlan struct {
|
|
Incident model.RockDeliveryIncident
|
|
Token model.RockToken
|
|
Order model.FulfillmentOrder
|
|
}
|
|
|
|
func rockIncidentRetryable(category string) bool {
|
|
switch category {
|
|
case model.RockIncidentCategoryUpstreamTimeout, model.RockIncidentCategoryRateLimited:
|
|
return true
|
|
default:
|
|
return false
|
|
}
|
|
}
|
|
|
|
func classifyRockDeliveryError(err error) (category string, httpStatus int, response string) {
|
|
if err == nil {
|
|
return model.RockIncidentCategoryOther, 0, ""
|
|
}
|
|
var statusErr *RockUpstreamStatusError
|
|
if errors.As(err, &statusErr) {
|
|
httpStatus = statusErr.StatusCode
|
|
response = statusErr.Body
|
|
}
|
|
body := strings.ToLower(strings.TrimSpace(response + " " + err.Error()))
|
|
compactBody := strings.NewReplacer(" ", "", "\n", "", "\r", "", "\t", "").Replace(body)
|
|
switch {
|
|
case strings.Contains(body, "用户不存在可以兑换的可用权益"):
|
|
return model.RockIncidentCategoryNoEntitlement, httpStatus, response
|
|
case strings.Contains(body, "同一角色无法重复兑换已兑换过的道具"):
|
|
return model.RockIncidentCategoryDuplicateItem, httpStatus, response
|
|
case strings.Contains(body, "礼包已发放完毕"):
|
|
return model.RockIncidentCategoryGiftExhausted, httpStatus, response
|
|
case strings.Contains(compactBody, "该token已兑换"):
|
|
return model.RockIncidentCategoryTokenRedeemed, httpStatus, response
|
|
case strings.Contains(compactBody, "该token暂不可兑换"):
|
|
return model.RockIncidentCategoryTokenNotRedeemable, httpStatus, response
|
|
case strings.Contains(compactBody, "token不存在"):
|
|
return model.RockIncidentCategoryTokenNotFound, httpStatus, response
|
|
case strings.Contains(body, "请求过于频繁") || strings.Contains(body, "110001"):
|
|
return model.RockIncidentCategoryRateLimited, httpStatus, response
|
|
case strings.Contains(body, "context deadline exceeded") || strings.Contains(body, "timeout"):
|
|
return model.RockIncidentCategoryUpstreamTimeout, httpStatus, response
|
|
case strings.Contains(body, "发货超时"):
|
|
return model.RockIncidentCategoryShipTimeout, httpStatus, response
|
|
case strings.Contains(body, "uid") || strings.Contains(body, "玩家编号") || strings.Contains(body, "绑定账号"):
|
|
return model.RockIncidentCategoryBindMismatch, httpStatus, response
|
|
default:
|
|
return model.RockIncidentCategoryOther, httpStatus, response
|
|
}
|
|
}
|
|
|
|
// RecordRockDeliveryIncident 持久化一次需要人工处理的发货异常，并同步 Token 错误摘要。
|
|
func (s *RockService) RecordRockDeliveryIncident(ctx context.Context, token *model.RockToken, orderNo string, cause error) (*model.RockDeliveryIncident, error) {
|
|
if token == nil || token.ID == 0 {
|
|
return nil, errors.New("Token 不存在")
|
|
}
|
|
category, httpStatus, response := classifyRockDeliveryError(cause)
|
|
message := ""
|
|
if cause != nil {
|
|
message = strings.TrimSpace(cause.Error())
|
|
}
|
|
if message == "" {
|
|
message = "上游发货失败"
|
|
}
|
|
return s.RecordRockIncident(ctx, token, orderNo, category, message, httpStatus, response)
|
|
}
|
|
|
|
func (s *RockService) RecordRockIncident(
|
|
ctx context.Context,
|
|
token *model.RockToken,
|
|
orderNo, category, message string,
|
|
httpStatus int,
|
|
upstreamResponse string,
|
|
) (*model.RockDeliveryIncident, error) {
|
|
if token == nil || token.ID == 0 {
|
|
return nil, errors.New("Token 不存在")
|
|
}
|
|
category = strings.TrimSpace(category)
|
|
if category == "" {
|
|
category = model.RockIncidentCategoryOther
|
|
}
|
|
orderNo = strings.TrimSpace(orderNo)
|
|
message = strings.TrimSpace(message)
|
|
upstreamResponse = strings.TrimSpace(upstreamResponse)
|
|
now := time.Now()
|
|
|
|
var incident model.RockDeliveryIncident
|
|
err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
|
var current model.RockToken
|
|
if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(¤t, token.ID).Error; err != nil {
|
|
return err
|
|
}
|
|
err := tx.Where(
|
|
"token_id = ? AND order_no = ? AND category = ? AND status = ?",
|
|
token.ID, orderNo, category, model.RockIncidentStatusOpen,
|
|
).First(&incident).Error
|
|
switch {
|
|
case err == nil:
|
|
if err := tx.Model(&incident).Updates(map[string]interface{}{
|
|
"occurrences": incident.Occurrences + 1,
|
|
"blocking_retry": true,
|
|
"message": message,
|
|
"upstream_http_status": httpStatus,
|
|
"upstream_response": upstreamResponse,
|
|
"last_seen_at": now,
|
|
"updated_at": now,
|
|
}).Error; err != nil {
|
|
return err
|
|
}
|
|
incident.Occurrences++
|
|
incident.BlockingRetry = true
|
|
incident.Message = message
|
|
incident.UpstreamHTTPStatus = httpStatus
|
|
incident.UpstreamResponse = upstreamResponse
|
|
incident.LastSeenAt = now
|
|
case errors.Is(err, gorm.ErrRecordNotFound):
|
|
incident = model.RockDeliveryIncident{
|
|
TokenID: current.ID,
|
|
OrderNo: orderNo,
|
|
UpstreamTokenID: strings.TrimSpace(current.UpstreamTokenID),
|
|
Category: category,
|
|
Status: model.RockIncidentStatusOpen,
|
|
BlockingRetry: true,
|
|
Message: message,
|
|
UpstreamHTTPStatus: httpStatus,
|
|
UpstreamResponse: upstreamResponse,
|
|
Occurrences: 1,
|
|
FirstSeenAt: now,
|
|
LastSeenAt: now,
|
|
}
|
|
if err := tx.Create(&incident).Error; err != nil {
|
|
return err
|
|
}
|
|
default:
|
|
return err
|
|
}
|
|
|
|
if err := tx.Model(¤t).Updates(map[string]interface{}{
|
|
"redeem_last_error": message,
|
|
"remark": appendRockRemark(current.Remark, message),
|
|
"updated_at": now,
|
|
}).Error; err != nil {
|
|
return err
|
|
}
|
|
writeRockTokenStatusLog(tx, ¤t, rockTokenEventIncident, current.Status, current.Status,
|
|
rockTokenSourceIncident, fmt.Sprintf("%s: %s", category, message), false, nil)
|
|
return writeAudit(tx, nil, nil, nil, "rock_delivery_incident.open", "rock_token",
|
|
fmt.Sprintf("%d", current.ID), map[string]interface{}{
|
|
"incident_id": incident.ID,
|
|
"order_no": orderNo,
|
|
"category": category,
|
|
"http_status": httpStatus,
|
|
})
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return &incident, nil
|
|
}
|
|
|
|
func (s *RockService) HasBlockingRockIncident(ctx context.Context, orderNo string) (*model.RockDeliveryIncident, error) {
|
|
var incident model.RockDeliveryIncident
|
|
err := s.db.WithContext(ctx).
|
|
Where("order_no = ? AND status = ? AND blocking_retry = TRUE", strings.TrimSpace(orderNo), model.RockIncidentStatusOpen).
|
|
Order("last_seen_at DESC").
|
|
First(&incident).Error
|
|
if errors.Is(err, gorm.ErrRecordNotFound) {
|
|
return nil, nil
|
|
}
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return &incident, nil
|
|
}
|
|
|
|
func (s *RockService) ListRockTokenIncidents(ctx context.Context, tokenID uint) ([]model.RockDeliveryIncident, error) {
|
|
var list []model.RockDeliveryIncident
|
|
err := s.db.WithContext(ctx).
|
|
Where("token_id = ?", tokenID).
|
|
Order("id DESC").
|
|
Limit(200).
|
|
Find(&list).Error
|
|
return list, err
|
|
}
|
|
|
|
func (s *RockService) ListRockDeliveryIncidents(ctx context.Context, status, category string, page, pageSize int) (*RockIncidentListResult, error) {
|
|
if page <= 0 {
|
|
page = 1
|
|
}
|
|
if pageSize <= 0 || pageSize > 100 {
|
|
pageSize = 20
|
|
}
|
|
db := s.db.WithContext(ctx).Model(&model.RockDeliveryIncident{})
|
|
if v := strings.TrimSpace(status); v != "" {
|
|
db = db.Where("status = ?", v)
|
|
}
|
|
if v := strings.TrimSpace(category); v != "" {
|
|
db = db.Where("category = ?", v)
|
|
}
|
|
var total int64
|
|
if err := db.Count(&total).Error; err != nil {
|
|
return nil, err
|
|
}
|
|
var list []model.RockDeliveryIncident
|
|
err := db.Order("last_seen_at DESC, id DESC").
|
|
Offset((page - 1) * pageSize).
|
|
Limit(pageSize).
|
|
Find(&list).Error
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return &RockIncidentListResult{List: list, Total: total}, nil
|
|
}
|
|
|
|
// ResolveRockDeliveryIncident 只关闭异常事件，不改变 Token 或订单主状态。
|
|
func (s *RockService) ResolveRockDeliveryIncident(ctx context.Context, id uint, note string, actorUserID *uint) error {
|
|
if id == 0 {
|
|
return errors.New("异常记录不存在")
|
|
}
|
|
now := time.Now()
|
|
return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
|
var incident model.RockDeliveryIncident
|
|
if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).
|
|
Where("id = ? AND status = ?", id, model.RockIncidentStatusOpen).
|
|
First(&incident).Error; err != nil {
|
|
if errors.Is(err, gorm.ErrRecordNotFound) {
|
|
return errors.New("异常记录不存在或已处理")
|
|
}
|
|
return err
|
|
}
|
|
if err := tx.Model(&incident).Updates(map[string]interface{}{
|
|
"status": model.RockIncidentStatusResolved,
|
|
"blocking_retry": false,
|
|
"resolved_at": now,
|
|
"resolved_by": actorUserID,
|
|
"resolution_action": "resolve",
|
|
"resolution_note": strings.TrimSpace(note),
|
|
"updated_at": now,
|
|
}).Error; err != nil {
|
|
return err
|
|
}
|
|
return writeAudit(tx, nil, actorUserID, nil, "rock_delivery_incident.resolve", "rock_delivery_incident",
|
|
fmt.Sprintf("%d", id), map[string]interface{}{
|
|
"token_id": incident.TokenID,
|
|
"order_no": incident.OrderNo,
|
|
"category": incident.Category,
|
|
"note": strings.TrimSpace(note),
|
|
})
|
|
})
|
|
}
|
|
|
|
// ApproveRockIncidentRetry 只对超时/频控异常临时解除重试阻塞。
|
|
// 实际发货结束后必须调用 Complete 或重新记录异常，恢复最终状态。
|
|
func (s *RockService) ApproveRockIncidentRetry(
|
|
ctx context.Context,
|
|
id uint,
|
|
note string,
|
|
actorUserID *uint,
|
|
) (*RockIncidentRetryPlan, error) {
|
|
note = strings.TrimSpace(note)
|
|
if note == "" {
|
|
return nil, errors.New("请填写人工重试说明")
|
|
}
|
|
now := time.Now()
|
|
var plan RockIncidentRetryPlan
|
|
err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
|
if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).
|
|
Where("id = ? AND status = ?", id, model.RockIncidentStatusOpen).
|
|
First(&plan.Incident).Error; err != nil {
|
|
if errors.Is(err, gorm.ErrRecordNotFound) {
|
|
return errors.New("异常记录不存在或已处理")
|
|
}
|
|
return err
|
|
}
|
|
if !rockIncidentRetryable(plan.Incident.Category) {
|
|
return fmt.Errorf("%s 类型异常不允许直接重试，请先核对上游状态", plan.Incident.Category)
|
|
}
|
|
if err := tx.First(&plan.Token, plan.Incident.TokenID).Error; err != nil {
|
|
return errors.New("异常关联的 Token 不存在")
|
|
}
|
|
if plan.Token.Status != model.RockStatusBound ||
|
|
plan.Token.AssignedOrderNo != plan.Incident.OrderNo ||
|
|
plan.Token.RedeemedAt != nil ||
|
|
rockTokenRedeemPending(&plan.Token) {
|
|
return errors.New("Token 当前状态不允许人工重试，请刷新后核对")
|
|
}
|
|
if strings.TrimSpace(plan.Token.ExpectedUID) == "" {
|
|
return errors.New("Token 缺少预期 UID，不能人工重试")
|
|
}
|
|
if err := tx.Where("order_no = ?", plan.Incident.OrderNo).First(&plan.Order).Error; err != nil {
|
|
return errors.New("异常关联的订单不存在")
|
|
}
|
|
if plan.Order.OrderStatus != model.OrderStatusShipFailed {
|
|
return fmt.Errorf("订单当前状态为 %s，不允许人工重试", plan.Order.OrderStatus)
|
|
}
|
|
var blockingNonRetryable model.RockDeliveryIncident
|
|
if err := tx.Where(
|
|
"token_id = ? AND order_no = ? AND status = ? AND category NOT IN ?",
|
|
plan.Token.ID,
|
|
plan.Incident.OrderNo,
|
|
model.RockIncidentStatusOpen,
|
|
[]string{model.RockIncidentCategoryUpstreamTimeout, model.RockIncidentCategoryRateLimited},
|
|
).First(&blockingNonRetryable).Error; err == nil {
|
|
return fmt.Errorf("该订单同时存在不可重试异常（%s），请先人工核对", blockingNonRetryable.Category)
|
|
} else if !errors.Is(err, gorm.ErrRecordNotFound) {
|
|
return err
|
|
}
|
|
if err := tx.Model(&model.RockDeliveryIncident{}).
|
|
Where("token_id = ? AND order_no = ? AND status = ?",
|
|
plan.Token.ID, plan.Incident.OrderNo, model.RockIncidentStatusOpen).
|
|
Updates(map[string]interface{}{
|
|
"blocking_retry": false,
|
|
"updated_at": now,
|
|
}).Error; err != nil {
|
|
return err
|
|
}
|
|
return writeAudit(tx, nil, actorUserID, nil, "rock_delivery_incident.approve_retry",
|
|
"rock_delivery_incident", fmt.Sprintf("%d", plan.Incident.ID), map[string]interface{}{
|
|
"token_id": plan.Token.ID,
|
|
"order_no": plan.Incident.OrderNo,
|
|
"category": plan.Incident.Category,
|
|
"note": note,
|
|
})
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return &plan, nil
|
|
}
|
|
|
|
// CompleteRockIncidentRetry 在上游已受理或已确认发货后，关闭同一订单的开放异常。
|
|
func (s *RockService) CompleteRockIncidentRetry(
|
|
ctx context.Context,
|
|
tokenID uint,
|
|
orderNo, note string,
|
|
actorUserID *uint,
|
|
) error {
|
|
now := time.Now()
|
|
note = strings.TrimSpace(note)
|
|
return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
|
result := tx.Model(&model.RockDeliveryIncident{}).
|
|
Where("token_id = ? AND order_no = ? AND status = ?",
|
|
tokenID, strings.TrimSpace(orderNo), model.RockIncidentStatusOpen).
|
|
Updates(map[string]interface{}{
|
|
"status": model.RockIncidentStatusResolved,
|
|
"blocking_retry": false,
|
|
"resolved_at": now,
|
|
"resolved_by": actorUserID,
|
|
"resolution_action": "approve_retry",
|
|
"resolution_note": note,
|
|
"updated_at": now,
|
|
})
|
|
if result.Error != nil {
|
|
return result.Error
|
|
}
|
|
if result.RowsAffected == 0 {
|
|
return errors.New("没有可关闭的开放异常记录")
|
|
}
|
|
return writeAudit(tx, nil, actorUserID, nil, "rock_delivery_incident.retry_complete",
|
|
"rock_token", fmt.Sprintf("%d", tokenID), map[string]interface{}{
|
|
"order_no": orderNo,
|
|
"note": note,
|
|
})
|
|
})
|
|
}
|
|
|
|
func resolveOpenRockIncidents(tx *gorm.DB, tokenID uint, action, note string, actorUserID *uint) error {
|
|
now := time.Now()
|
|
return tx.Model(&model.RockDeliveryIncident{}).
|
|
Where("token_id = ? AND status = ?", tokenID, model.RockIncidentStatusOpen).
|
|
Updates(map[string]interface{}{
|
|
"status": model.RockIncidentStatusResolved,
|
|
"blocking_retry": false,
|
|
"resolved_at": now,
|
|
"resolved_by": actorUserID,
|
|
"resolution_action": action,
|
|
"resolution_note": strings.TrimSpace(note),
|
|
"updated_at": now,
|
|
}).Error
|
|
}
|
|
|
|
func rockTokenIncidentFilter(db *gorm.DB, category string) *gorm.DB {
|
|
if category == model.RockIncidentCategoryNoEntitlement {
|
|
return db.Where(
|
|
"EXISTS (SELECT 1 FROM rock_delivery_incidents i WHERE i.token_id = rock_tokens.id AND i.status = 'open' AND i.category = ?)",
|
|
model.RockIncidentCategoryNoEntitlement,
|
|
)
|
|
}
|
|
return db.Where(
|
|
"EXISTS (SELECT 1 FROM rock_delivery_incidents i WHERE i.token_id = rock_tokens.id AND i.status = 'open' AND i.category <> ?)",
|
|
model.RockIncidentCategoryNoEntitlement,
|
|
)
|
|
}
|
|
|
|
func (s *RockService) fillRockTokenIncidentFlags(ctx context.Context, tokens []model.RockToken) {
|
|
if len(tokens) == 0 {
|
|
return
|
|
}
|
|
ids := make([]uint, 0, len(tokens))
|
|
for i := range tokens {
|
|
if tokens[i].ID != 0 {
|
|
ids = append(ids, tokens[i].ID)
|
|
}
|
|
}
|
|
if len(ids) == 0 {
|
|
return
|
|
}
|
|
type row struct {
|
|
TokenID uint
|
|
Category string
|
|
Message string
|
|
LastSeenAt time.Time
|
|
IncidentCount int64
|
|
}
|
|
var rows []row
|
|
err := s.db.WithContext(ctx).Raw(`
|
|
SELECT DISTINCT ON (token_id)
|
|
token_id, category, message, last_seen_at,
|
|
COUNT(*) OVER (PARTITION BY token_id) AS incident_count
|
|
FROM rock_delivery_incidents
|
|
WHERE token_id IN ? AND status = 'open'
|
|
ORDER BY token_id, (category = 'no_entitlement') DESC, last_seen_at DESC, id DESC
|
|
`, ids).Scan(&rows).Error
|
|
if err != nil {
|
|
return
|
|
}
|
|
byID := make(map[uint]row, len(rows))
|
|
for _, item := range rows {
|
|
byID[item.TokenID] = item
|
|
}
|
|
for i := range tokens {
|
|
item, ok := byID[tokens[i].ID]
|
|
if !ok {
|
|
continue
|
|
}
|
|
at := item.LastSeenAt
|
|
tokens[i].HasOpenIncident = true
|
|
tokens[i].IncidentCategory = item.Category
|
|
tokens[i].IncidentMessage = item.Message
|
|
tokens[i].IncidentAt = &at
|
|
tokens[i].IncidentCount = item.IncidentCount
|
|
}
|
|
}
|