goauto/server/app/goauto/purchase/order_writeback_worker.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 rebuilds a SYB client from the cached session
// only. It never logs in, never triggers OCR and never deletes the cached
// session (that stays the exclusive responsibility of sybimport.Connect's
// login/refresh path) — it only reports whether the cached cookies still
// work, via CheckSession, so the caller can classify the failure (#330).
func restoreOrderWritebackClient(ctx context.Context, db *gorm.DB) (OrderNumberClient, error) {
	cfg := config.ExtConfig.SYB.Resolved()
	session, err := sybclient.NewSessionStore(db).Load(ctx, cfg.Username, time.Now())
	if err != nil {
		return nil, err
	}
	if session.UserID <= 0 {
		return nil, errSessionUserIDMissing
	}
	c, err := sybclient.New(cfg.BaseURL)
	if err != nil {
		return nil, err
	}
	if err = c.ImportCookiesJSON(session.CookiesJSON); err != nil {
		return nil, err
	}
	// Active probe (#330 修订1): without this, a remotely-expired cookie jar
	// imports cleanly and only fails later inside read(), which would record
	// it as SYB_READ_FAILED instead of the retryable session-class outcome.
	// Any error here — ErrSessionInvalid or network/format — is treated as
	// session-class; only ErrSessionInvalid is a confirmed logout, but a
	// network/format error is not confirmed-valid either, so it is still
	// retried rather than attempted as a write.
	if err = c.CheckSession(ctx, session.UserID, cfg.Username); err != nil {
		return nil, err
	}
	return c, nil
}

// 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()
		}
		return db.Model(&models.PurchaseOrderWriteback{}).Where("id = ? AND status = 'running' AND lease_owner = ?", item.ID, owner).Updates(updates).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
		}
		return db.Model(&models.PurchaseOrderWriteback{}).Where("id = ? AND status = 'running' AND lease_owner = ?", item.ID, owner).Updates(updates).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", "写入后尚未回读确认，请人工核对后重试")
}