311 lines
13 KiB
Go
311 lines
13 KiB
Go
package purchase
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"time"
|
|
|
|
"github.com/google/uuid"
|
|
"go-admin/app/goauto/models"
|
|
"go-admin/app/goauto/sybclient"
|
|
"go-admin/config"
|
|
"gorm.io/gorm"
|
|
"gorm.io/gorm/clause"
|
|
"gorm.io/gorm/logger"
|
|
)
|
|
|
|
type OrderNumberClient interface {
|
|
DetailListByStock(context.Context, []int64) ([]sybclient.StockDetail, error)
|
|
UpdateDetailPurchaseCode(context.Context, int64, int64, string) error
|
|
}
|
|
type OrderWritebackWorker struct {
|
|
DB *gorm.DB
|
|
Now func() time.Time
|
|
Factory func(context.Context, *gorm.DB) (OrderNumberClient, error)
|
|
}
|
|
|
|
// errSessionUserIDMissing marks a cached session whose UserID column is not a
|
|
// positive SYB account id. SessionStore.Save (session.go) rejects UserID<=0
|
|
// before it is ever persisted, so this should be unreachable in practice; it
|
|
// exists so a corrupted/legacy row fails loudly and safely instead of calling
|
|
// CheckSession with id=0 (#330 修订1).
|
|
var errSessionUserIDMissing = errors.New("SYB 会话记录缺少有效 user id")
|
|
|
|
// Bounded auto-retry for session-class writeback failures (#330). A session
|
|
// outage self-heals once GoAutoSYBHourlySync refreshes syb_session, but that
|
|
// refresh only happens once per hour (at :05) and only fires the run *after*
|
|
// the session is found dead — so the wait from failure to refresh can be
|
|
// close to a full hour. The backoff schedule below sums to ~90 minutes
|
|
// (5+10+15+30+30) across maxSessionRetryAttempts=6 attempts, deliberately
|
|
// longer than one hourly sync period so a session recovered by "the next"
|
|
// hourly run is still caught automatically instead of exhausting attempts
|
|
// first. maxSessionRetryAttempts caps the automatic attempts so a session
|
|
// that never recovers still lands back in "failed" for a human instead of
|
|
// retrying forever.
|
|
const maxSessionRetryAttempts = 6
|
|
|
|
var sessionRetryBackoff = []time.Duration{
|
|
5 * time.Minute,
|
|
10 * time.Minute,
|
|
15 * time.Minute,
|
|
30 * time.Minute,
|
|
30 * time.Minute,
|
|
}
|
|
|
|
// sessionRetryDelay returns the backoff before the next automatic attempt,
|
|
// given the attempt number (1-based, i.e. the count already recorded for the
|
|
// attempt that just failed).
|
|
func sessionRetryDelay(attempt int) time.Duration {
|
|
idx := attempt - 1
|
|
if idx < 0 {
|
|
idx = 0
|
|
}
|
|
if idx >= len(sessionRetryBackoff) {
|
|
idx = len(sessionRetryBackoff) - 1
|
|
}
|
|
return sessionRetryBackoff[idx]
|
|
}
|
|
|
|
// sessionUnavailableMessage classifies why the cached SYB session could not
|
|
// be used, without ever including cookies, tokens or other credential
|
|
// material (#330 修订1点3). The category — not the raw error text — is what
|
|
// gets persisted to error_message, wrapped in a fixed, actionable template
|
|
// that stays well under the 300-char column limit.
|
|
func sessionUnavailableMessage(err error) string {
|
|
category := "会话恢复失败（网络/其他）"
|
|
switch {
|
|
case errors.Is(err, sybclient.ErrNoSession):
|
|
category = "会话缺失/已过期"
|
|
case errors.Is(err, errSessionUserIDMissing):
|
|
category = "会话记录异常，缺少 user id"
|
|
case errors.Is(err, sybclient.ErrSessionInvalid):
|
|
category = "会话校验失效"
|
|
}
|
|
return "SYB会话不可用（" + category + "），将自动重试；如持续失败请恢复登录后重试"
|
|
}
|
|
|
|
// restoreOrderWritebackClient uses the same session acquisition path as SYB
|
|
// sync. A valid cached session is reused; an explicitly invalid session is
|
|
// refreshed under the shared database auth lease so concurrent workers do not
|
|
// request multiple captcha codes.
|
|
func restoreOrderWritebackClient(ctx context.Context, db *gorm.DB) (OrderNumberClient, error) {
|
|
cfg := config.ExtConfig.SYB.Resolved()
|
|
return sybclient.AcquireSession(ctx, db, sybclient.LoginConfig{
|
|
BaseURL: cfg.BaseURL, Username: cfg.Username, Password: cfg.Password,
|
|
OcrURL: cfg.OcrURL, OcrMaxAttempts: cfg.OcrMaxAttempts,
|
|
})
|
|
}
|
|
|
|
// One short-lived claim at a time across processes. No business writes occur
|
|
// during startup itself; only explicitly persisted pending records are handled.
|
|
func RecoverOrderWritebacks(db *gorm.DB) {
|
|
go func() {
|
|
w := OrderWritebackWorker{DB: db, Now: func() time.Time { return time.Now().UTC() }, Factory: restoreOrderWritebackClient}
|
|
ticker := time.NewTicker(3 * time.Second)
|
|
defer ticker.Stop()
|
|
for range ticker.C {
|
|
_, _ = w.RunOnce(context.Background())
|
|
}
|
|
}()
|
|
}
|
|
|
|
func (w *OrderWritebackWorker) RunOnce(ctx context.Context) (bool, error) {
|
|
if w.Now == nil {
|
|
w.Now = func() time.Time { return time.Now().UTC() }
|
|
}
|
|
if w.Factory == nil {
|
|
w.Factory = restoreOrderWritebackClient
|
|
}
|
|
db := w.DB.WithContext(ctx).Session(&gorm.Session{Logger: logger.Default.LogMode(logger.Silent)})
|
|
if err := db.Clauses(clause.OnConflict{DoNothing: true}).Create(&models.PurchaseOrderWritebackLease{ID: 1}).Error; err != nil {
|
|
return false, err
|
|
}
|
|
owner := uuid.NewString()
|
|
now := w.Now()
|
|
expires := now.Add(2 * time.Minute)
|
|
claim := db.Model(&models.PurchaseOrderWritebackLease{}).Where("id = 1 AND (expires_at IS NULL OR expires_at <= ?)", now).Updates(map[string]any{"owner": owner, "expires_at": expires})
|
|
if claim.Error != nil {
|
|
return false, claim.Error
|
|
}
|
|
if claim.RowsAffected != 1 {
|
|
return false, nil
|
|
}
|
|
defer db.Model(&models.PurchaseOrderWritebackLease{}).Where("id = 1 AND owner = ?", owner).Updates(map[string]any{"owner": "", "expires_at": nil})
|
|
var item models.PurchaseOrderWriteback
|
|
recovering := false
|
|
err := db.Transaction(func(tx *gorm.DB) error {
|
|
if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where(
|
|
"status = ? OR (status = ? AND lease_expires_at <= ?) OR (status = ? AND error_code = ? AND lease_expires_at IS NOT NULL AND lease_expires_at <= ? AND attempt_count < ?)",
|
|
"pending", "running", now, "failed", "SYB_SESSION_UNAVAILABLE", now, maxSessionRetryAttempts,
|
|
).Order("id").First(&item).Error; err != nil {
|
|
return err
|
|
}
|
|
recovering = item.Status == "running"
|
|
return tx.Model(&item).Updates(map[string]any{"status": "running", "attempt_count": gorm.Expr("attempt_count + 1"), "lease_owner": owner, "lease_expires_at": expires}).Error
|
|
})
|
|
if errors.Is(err, gorm.ErrRecordNotFound) {
|
|
return false, nil
|
|
}
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
// tx.Model(&item).Updates used gorm.Expr("attempt_count + 1") above, which
|
|
// GORM does not read back into the struct; sync it here so downstream
|
|
// bounded-retry math (finishSessionUnavailable) sees the true post-claim
|
|
// count instead of being off by one.
|
|
item.AttemptCount++
|
|
finish := func(status, code, message string) error {
|
|
updates := map[string]any{"status": status, "error_code": code, "error_message": message, "lease_owner": ""}
|
|
if status != "unknown" {
|
|
updates["lease_expires_at"] = nil
|
|
}
|
|
if status == "succeeded" {
|
|
updates["completed_at"] = w.Now()
|
|
}
|
|
if err := db.Model(&models.PurchaseOrderWriteback{}).Where("id = ? AND status = 'running' AND lease_owner = ?", item.ID, owner).Updates(updates).Error; err != nil {
|
|
return err
|
|
}
|
|
// Keep the admin task state aligned with the authoritative writeback
|
|
// record. Conflicts and unknown outcomes remain failed until a human
|
|
// resolves them; they must never appear as successful.
|
|
taskStatus := models.PurchaseWritebackStatusFailed
|
|
if status == "succeeded" {
|
|
taskStatus = models.PurchaseWritebackStatusSucceeded
|
|
}
|
|
return db.Session(&gorm.Session{SkipHooks: true}).Model(&models.PurchaseTask{}).Where("id = ?", item.PurchaseTaskID).Updates(map[string]any{
|
|
"writeback_status": taskStatus,
|
|
"writeback_at": gorm.Expr("CASE WHEN ? = 'succeeded' THEN ? ELSE writeback_at END", status, w.Now()),
|
|
}).Error
|
|
}
|
|
// finishSessionUnavailable is the bounded-retry counterpart of finish for
|
|
// SYB_SESSION_UNAVAILABLE: instead of clearing the lease, it schedules the
|
|
// next automatic attempt (item.AttemptCount was already incremented by the
|
|
// claim above) until maxSessionRetryAttempts is reached, at which point it
|
|
// behaves like finish("failed", ...) and stops retrying (#330).
|
|
finishSessionUnavailable := func(err error) error {
|
|
updates := map[string]any{
|
|
"status": "failed", "error_code": "SYB_SESSION_UNAVAILABLE",
|
|
"error_message": sessionUnavailableMessage(err), "lease_owner": "",
|
|
}
|
|
if item.AttemptCount < maxSessionRetryAttempts {
|
|
updates["lease_expires_at"] = w.Now().Add(sessionRetryDelay(item.AttemptCount))
|
|
} else {
|
|
updates["lease_expires_at"] = nil
|
|
}
|
|
if err := db.Model(&models.PurchaseOrderWriteback{}).Where("id = ? AND status = 'running' AND lease_owner = ?", item.ID, owner).Updates(updates).Error; err != nil {
|
|
return err
|
|
}
|
|
return db.Session(&gorm.Session{SkipHooks: true}).Model(&models.PurchaseTask{}).Where("id = ?", item.PurchaseTaskID).Update("writeback_status", models.PurchaseWritebackStatusFailed).Error
|
|
}
|
|
var task models.PurchaseTask
|
|
if err = db.First(&task, item.PurchaseTaskID).Error; err != nil {
|
|
return true, finish("failed", "TASK_UNAVAILABLE", "采购任务不可用，请人工核对")
|
|
}
|
|
var syb models.SYBProduct
|
|
if !orderWritebackEligible(task) || *task.PDDOrderNo != item.OrderNo {
|
|
return true, finish("conflict", "ORDER_FACT_CHANGED", "采购订单事实已变化，请人工核对")
|
|
}
|
|
if err = db.First(&syb, *task.SYBProductID).Error; err != nil || !validOrderWritebackTarget(syb) || int64(syb.StockID) != item.StockID || int64(syb.DetailID) != item.DetailID {
|
|
return true, finish("conflict", "SYB_TARGET_CHANGED", "SYB商品明细关联已变化")
|
|
}
|
|
callCtx, cancel := context.WithTimeout(ctx, 25*time.Second)
|
|
client, err := w.Factory(callCtx, db)
|
|
cancel()
|
|
if err != nil {
|
|
return true, finishSessionUnavailable(err)
|
|
}
|
|
read := func() (string, string, error) {
|
|
readCtx, stop := context.WithTimeout(ctx, 20*time.Second)
|
|
defer stop()
|
|
rows, e := client.DetailListByStock(readCtx, []int64{item.StockID})
|
|
if e != nil {
|
|
return "", "", e
|
|
}
|
|
found := 0
|
|
code, platform := "", ""
|
|
for _, stock := range rows {
|
|
if stock.ID != item.StockID {
|
|
continue
|
|
}
|
|
for _, detail := range stock.Details {
|
|
if detail.ID == item.DetailID {
|
|
found++
|
|
var ok bool
|
|
code, ok = detail.Raw["purchaseCode"].(string)
|
|
if !ok && detail.Raw["purchaseCode"] != nil {
|
|
return "", "", errors.New("invalid purchase code")
|
|
}
|
|
platform, ok = detail.Raw["purchasePlatform"].(string)
|
|
if !ok && detail.Raw["purchasePlatform"] != nil {
|
|
return "", "", errors.New("invalid platform")
|
|
}
|
|
}
|
|
}
|
|
}
|
|
if found != 1 {
|
|
return "", "", errors.New("ambiguous remote detail")
|
|
}
|
|
return code, platform, nil
|
|
}
|
|
code, platform, err := read()
|
|
if err != nil {
|
|
status := "failed"
|
|
if recovering || item.WriteStarted {
|
|
status = "unknown"
|
|
}
|
|
return true, finish(status, "SYB_READ_FAILED", "无法回读SYB目标明细，请恢复连接后重试")
|
|
}
|
|
if code == item.OrderNo && platform == "pdd" {
|
|
return true, finish("succeeded", "", "")
|
|
}
|
|
if code != "" || (platform != "" && platform != "pdd") {
|
|
return true, finish("conflict", "SYB_ORDER_CONFLICT", "SYB已有不同单号或平台，未覆盖，请人工核对")
|
|
}
|
|
if recovering || item.WriteStarted {
|
|
return true, finish("unknown", "SYB_WRITE_UNCONFIRMED", "上次写入结果未确认，未自动重发，请人工核对后重试")
|
|
}
|
|
var blocked int64
|
|
if err = db.Model(&models.PurchaseOrderWriteback{}).Where("id <> ? AND stock_id = ? AND detail_id = ? AND write_started = ? AND status IN ?", item.ID, item.StockID, item.DetailID, true, []string{"running", "unknown"}).Count(&blocked).Error; err != nil {
|
|
return true, err
|
|
}
|
|
if blocked > 0 {
|
|
return true, finish("unknown", "SYB_TARGET_IN_FLIGHT", "同一SYB明细有其他未确认写入，请先核对该记录")
|
|
}
|
|
// Fence immediately before the sole external write. A recovered owner never
|
|
// writes; a durable marker survives crashes between request and acknowledgment.
|
|
err = db.Transaction(func(tx *gorm.DB) error {
|
|
var lease models.PurchaseOrderWritebackLease
|
|
if e := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&lease, 1).Error; e != nil {
|
|
return e
|
|
}
|
|
if lease.Owner != owner || lease.ExpiresAt == nil || !lease.ExpiresAt.After(w.Now().Add(25*time.Second)) {
|
|
return errors.New("write lease lost")
|
|
}
|
|
r := tx.Model(&models.PurchaseOrderWriteback{}).Where("id = ? AND lease_owner = ? AND status = 'running'", item.ID, owner).Update("write_started", true)
|
|
if r.Error != nil {
|
|
return r.Error
|
|
}
|
|
if r.RowsAffected != 1 {
|
|
return errors.New("item lease lost")
|
|
}
|
|
return nil
|
|
})
|
|
if err != nil {
|
|
return true, err
|
|
}
|
|
writeCtx, stop := context.WithTimeout(ctx, 20*time.Second)
|
|
writeErr := client.UpdateDetailPurchaseCode(writeCtx, item.StockID, item.DetailID, item.OrderNo)
|
|
stop()
|
|
code, platform, err = read()
|
|
if err == nil && code == item.OrderNo && platform == "pdd" {
|
|
return true, finish("succeeded", "", "")
|
|
}
|
|
if err == nil && (code != "" || (platform != "" && platform != "pdd")) {
|
|
return true, finish("conflict", "SYB_ORDER_CONFLICT", "SYB单号或平台与目标不一致，未覆盖")
|
|
}
|
|
if writeErr != nil && !errors.Is(writeErr, sybclient.ErrWriteResultUnknown) {
|
|
return true, finish("failed", "SYB_WRITE_REJECTED", "SYB拒绝回填，请检查会话与明细后重试")
|
|
}
|
|
return true, finish("unknown", "SYB_WRITE_UNCONFIRMED", "写入后尚未回读确认，请人工核对后重试")
|
|
}
|