epusdt/src/mq/worker.go

package mq

import (
	"errors"
	"fmt"
	"net/http"
	"strconv"
	"strings"
	"time"

	"github.com/GMWalletApp/epusdt/config"
	"github.com/GMWalletApp/epusdt/model/dao"
	"github.com/GMWalletApp/epusdt/model/data"
	"github.com/GMWalletApp/epusdt/model/mdb"
	"github.com/GMWalletApp/epusdt/model/response"
	"github.com/GMWalletApp/epusdt/util/http_client"
	"github.com/GMWalletApp/epusdt/util/log"
	"github.com/GMWalletApp/epusdt/util/sign"
)

// resolveOrderApiKey returns the api_keys row that signed the order.
// Every order must carry an ApiKeyID (the default key is always seeded
// at bootstrap, and the middleware/EPAY inline flow always stamps it).
// If the row is missing or disabled, we return an error — using any
// other secret would produce a signature the merchant can't verify.
// The admin can resend the callback after fixing the key.
func resolveOrderApiKey(order *mdb.Orders) (*mdb.ApiKey, error) {
	if order.ApiKeyID == 0 {
		return nil, fmt.Errorf("order trade_id=%s has no api_key_id", order.TradeId)
	}
	row, err := data.GetApiKeyByID(order.ApiKeyID)
	if err != nil {
		return nil, fmt.Errorf("lookup api_key_id=%d: %w", order.ApiKeyID, err)
	}
	if row.ID == 0 {
		return nil, fmt.Errorf("api_key_id=%d not found (deleted?)", order.ApiKeyID)
	}
	if row.Status != mdb.ApiKeyStatusEnable {
		return nil, fmt.Errorf("api_key_id=%d is disabled", order.ApiKeyID)
	}
	return row, nil
}

const batchSize = 100

const sqliteBusyRetryAttempts = 3

type expirableOrder struct {
	ID             uint64  `gorm:"column:id"`
	TradeId        string  `gorm:"column:trade_id"`
	Network        string  `gorm:"column:network"`
	ReceiveAddress string  `gorm:"column:receive_address"`
	Token          string  `gorm:"column:token"`
	ActualAmount   float64 `gorm:"column:actual_amount"`
}

func runOrderExpirationLoop() {
	runLoop("order_expiration", processExpiredOrders)
}

func runOrderCallbackLoop() {
	runLoop("order_callback", dispatchPendingCallbacks)
}

func runTransactionLockCleanupLoop() {
	runLoop("transaction_lock_cleanup", cleanupExpiredTransactionLocks)
}

func runLoop(name string, fn func()) {
	safeRun(name, fn)
	ticker := time.NewTicker(config.GetQueuePollInterval())
	defer ticker.Stop()

	for range ticker.C {
		safeRun(name, fn)
	}
}

func safeRun(name string, fn func()) {
	defer func() {
		if err := recover(); err != nil {
			log.Sugar.Errorf("[mq] %s panic: %v", name, err)
		}
	}()
	fn()
}

func processExpiredOrders() {
	expirationCutoff := time.Now().Add(-config.GetOrderExpirationTimeDuration())
	for {
		var orders []expirableOrder
		err := withSQLiteBusyRetry(func() error {
			return dao.Mdb.Model(&mdb.Orders{}).
				Select("id", "trade_id", "network", "receive_address", "token", "actual_amount").
				Where("status = ?", mdb.StatusWaitPay).
				Where("created_at <= ?", expirationCutoff).
				Order("id asc").
				Limit(batchSize).
				Find(&orders).Error
		})
		if err != nil {
			log.Sugar.Errorf("[mq] query expired orders failed: %v", err)
			return
		}
		if len(orders) == 0 {
			return
		}

		for _, order := range orders {
			expired, err := data.UpdateOrderIsExpirationById(order.ID, expirationCutoff)
			if err != nil {
				log.Sugar.Errorf("[mq] expire order failed, trade_id=%s, err=%v", order.TradeId, err)
				continue
			}
			if !expired {
				continue
			}
			if err = data.UnLockTransaction(order.Network, order.ReceiveAddress, order.Token, order.ActualAmount); err != nil {
				log.Sugar.Warnf("[mq] release expired transaction lock failed, trade_id=%s, err=%v", order.TradeId, err)
			}
		}

		if len(orders) < batchSize {
			return
		}
	}
}

func dispatchPendingCallbacks() {
	maxRetry := config.GetOrderNoticeMaxRetry()
	var orders []data.PendingCallbackOrder
	err := withSQLiteBusyRetry(func() error {
		var innerErr error
		orders, innerErr = data.GetPendingCallbackOrders(maxRetry, batchSize)
		return innerErr
	})
	if err != nil {
		log.Sugar.Errorf("[mq] query callback orders failed: %v", err)
		return
	}

	now := time.Now()
	for _, order := range orders {
		if !isCallbackDue(&order, now, maxRetry) {
			continue
		}
		tradeID := order.TradeId
		if _, loaded := callbackInflight.LoadOrStore(tradeID, struct{}{}); loaded {
			continue
		}

		select {
		case callbackLimiter <- struct{}{}:
			go processCallback(tradeID)
		default:
			callbackInflight.Delete(tradeID)
			return
		}
	}
}

func processCallback(tradeID string) {
	defer func() {
		<-callbackLimiter
		callbackInflight.Delete(tradeID)
	}()

	freshOrder, err := data.GetOrderInfoByTradeId(tradeID)
	if err != nil {
		log.Sugar.Errorf("[mq] reload callback order failed, trade_id=%s, err=%v", tradeID, err)
		return
	}
	if freshOrder.ID <= 0 || freshOrder.Status != mdb.StatusPaySuccess || freshOrder.CallBackConfirm != mdb.CallBackConfirmNo {
		return
	}

	if err = sendOrderCallback(freshOrder); err != nil {
		log.Sugar.Warnf("[mq] callback request failed, trade_id=%s, attempt=%d, err=%v", freshOrder.TradeId, freshOrder.CallbackNum+1, err)
		freshOrder.CallBackConfirm = mdb.CallBackConfirmNo
	} else {
		freshOrder.CallBackConfirm = mdb.CallBackConfirmOk
	}

	if err = data.SaveCallBackOrdersResp(freshOrder); err != nil {
		log.Sugar.Errorf("[mq] save callback result failed, trade_id=%s, err=%v", freshOrder.TradeId, err)
	}
}

func sendOrderCallback(order *mdb.Orders) error {
	apiKeyRow, err := resolveOrderApiKey(order)
	if err != nil || apiKeyRow == nil || apiKeyRow.ID == 0 {
		return errors.New("no api key row available for callback")
	}

	switch order.PaymentType {
	case mdb.PaymentTypeEpay:
		// EPAY uses pid (integer) and secret_key as "key".
		pidInt, convErr := strconv.Atoi(apiKeyRow.Pid)
		if convErr != nil {
			return fmt.Errorf("epay pid not numeric: %s", apiKeyRow.Pid)
		}
		notifyData := response.OrderNotifyResponseEpay{
			PID:         pidInt,
			TradeNo:     order.TradeId,
			OutTradeNo:  order.OrderId,
			Type:        "alipay",
			Name:        order.Name,
			Money:       fmt.Sprintf("%.4f", order.Amount),
			TradeStatus: "TRADE_SUCCESS",
		}

		signstr2, err := sign.Get(notifyData, apiKeyRow.SecretKey)
		if err != nil {
			return err
		}

		formData := map[string]string{
			"pid":          fmt.Sprintf("%d", notifyData.PID),
			"trade_no":     notifyData.TradeNo,
			"out_trade_no": notifyData.OutTradeNo,
			"type":         notifyData.Type,
			"name":         notifyData.Name,
			"money":        notifyData.Money,
			"trade_status": notifyData.TradeStatus,
			"sign":         signstr2,
			"sign_type":    "MD5",
		}

		epayResp, err := http_client.GetHttpClient().R().SetQueryParams(formData).Get(order.NotifyUrl)
		if err != nil {
			return err
		}
		log.Sugar.Infof("[mq] epay notify_url response status: %d, body: %s", epayResp.StatusCode(), string(epayResp.Body()))
		if epayResp.StatusCode() != http.StatusOK {
			return errors.New(epayResp.Status())
		}
		if !isCallbackAck(epayResp.Body()) {
			return errors.New("not ok")
		}

	default:

		client := http_client.GetHttpClient()
		orderResp := response.OrderNotifyResponse{
			Pid:                apiKeyRow.Pid,
			TradeId:            order.TradeId,
			OrderId:            order.OrderId,
			Amount:             order.Amount,
			ActualAmount:       order.ActualAmount,
			ReceiveAddress:     order.ReceiveAddress,
			Token:              order.Token,
			BlockTransactionId: order.BlockTransactionId,
			Status:             mdb.StatusPaySuccess,
		}
		signature, err := sign.Get(orderResp, apiKeyRow.SecretKey)
		if err != nil {
			return err
		}
		orderResp.Signature = signature

		resp, err := client.R().
			SetHeader("powered-by", "Epusdt(https://github.com/GMwalletApp/epusdt)").
			SetBody(orderResp).
			Post(order.NotifyUrl)
		if err != nil {
			return err
		}
		if resp.StatusCode() != http.StatusOK {
			return errors.New(resp.Status())
		}
		if !isCallbackAck(resp.Body()) {
			return errors.New("not ok")
		}
	}

	return nil
}

func isCallbackAck(body []byte) bool {
	ack := strings.ToLower(strings.TrimSpace(string(body)))
	return ack == "ok" || ack == "success"
}

func cleanupExpiredTransactionLocks() {
	if err := data.CleanupExpiredTransactionLocks(); err != nil {
		log.Sugar.Errorf("[mq] cleanup expired transaction locks failed: %v", err)
	}
}

func withSQLiteBusyRetry(fn func() error) error {
	var err error
	for attempt := 1; attempt <= sqliteBusyRetryAttempts; attempt++ {
		err = fn()
		if err == nil {
			return nil
		}
		if !isSQLiteBusyError(err) || attempt == sqliteBusyRetryAttempts {
			return err
		}
		time.Sleep(time.Duration(attempt*25) * time.Millisecond)
	}
	return err
}

func isSQLiteBusyError(err error) bool {
	if err == nil {
		return false
	}
	msg := strings.ToLower(err.Error())
	return strings.Contains(msg, "database is locked") || strings.Contains(msg, "sqlite_busy")
}

func isCallbackDue(order *data.PendingCallbackOrder, now time.Time, maxRetry int) bool {
	if order.CallBackConfirm != mdb.CallBackConfirmNo {
		return false
	}
	if order.CallbackNum > maxRetry {
		return false
	}
	if order.CallbackNum == 0 {
		return true
	}
	nextRunAt := order.UpdatedAt.StdTime().Add(callbackRetryDelay(order.CallbackNum))
	return !nextRunAt.After(now)
}

func callbackRetryDelay(attempts int) time.Duration {
	if attempts <= 0 {
		return 0
	}

	delay := config.GetCallbackRetryBaseDuration()
	maxDelay := 5 * time.Minute
	for i := 1; i < attempts; i++ {
		if delay >= maxDelay/2 {
			return maxDelay
		}
		delay *= 2
	}
	if delay > maxDelay {
		return maxDelay
	}
	return delay
}